indexer.cc 26.9 KB
Newer Older
1 2
/* -*- mode: C++; c-basic-offset: 4; indent-tabs-mode: nil -*- */
// vim: ft=cpp:expandtab:ts=8:sw=4:softtabstop=4:
3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54
/*
COPYING CONDITIONS NOTICE:

  This program is free software; you can redistribute it and/or modify
  it under the terms of version 2 of the GNU General Public License as
  published by the Free Software Foundation, and provided that the
  following conditions are met:

      * Redistributions of source code must retain this COPYING
        CONDITIONS NOTICE, the COPYRIGHT NOTICE (below), the
        DISCLAIMER (below), the UNIVERSITY PATENT NOTICE (below), the
        PATENT MARKING NOTICE (below), and the PATENT RIGHTS
        GRANT (below).

      * Redistributions in binary form must reproduce this COPYING
        CONDITIONS NOTICE, the COPYRIGHT NOTICE (below), the
        DISCLAIMER (below), the UNIVERSITY PATENT NOTICE (below), the
        PATENT MARKING NOTICE (below), and the PATENT RIGHTS
        GRANT (below) in the documentation and/or other materials
        provided with the distribution.

  You should have received a copy of the GNU General Public License
  along with this program; if not, write to the Free Software
  Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA
  02110-1301, USA.

COPYRIGHT NOTICE:

  TokuDB, Tokutek Fractal Tree Indexing Library.
  Copyright (C) 2007-2013 Tokutek, Inc.

DISCLAIMER:

  This program is distributed in the hope that it will be useful, but
  WITHOUT ANY WARRANTY; without even the implied warranty of
  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  General Public License for more details.

UNIVERSITY PATENT NOTICE:

  The technology is licensed by the Massachusetts Institute of
  Technology, Rutgers State University of New Jersey, and the Research
  Foundation of State University of New York at Stony Brook under
  United States of America Serial No. 11/760379 and to the patents
  and/or patent applications resulting from it.

PATENT MARKING NOTICE:

  This software is covered by US Patent No. 8,185,551.

PATENT RIGHTS GRANT:

55
  "THIS IMPLEMENTATION" means the copyrightable works distributed by
56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86
  Tokutek as part of the Fractal Tree project.

  "PATENT CLAIMS" means the claims of patents that are owned or
  licensable by Tokutek, both currently or in the future; and that in
  the absence of this license would be infringed by THIS
  IMPLEMENTATION or by using or running THIS IMPLEMENTATION.

  "PATENT CHALLENGE" shall mean a challenge to the validity,
  patentability, enforceability and/or non-infringement of any of the
  PATENT CLAIMS or otherwise opposing any of the PATENT CLAIMS.

  Tokutek hereby grants to you, for the term and geographical scope of
  the PATENT CLAIMS, a non-exclusive, no-charge, royalty-free,
  irrevocable (except as stated in this section) patent license to
  make, have made, use, offer to sell, sell, import, transfer, and
  otherwise run, modify, and propagate the contents of THIS
  IMPLEMENTATION, where such license applies only to the PATENT
  CLAIMS.  This grant does not include claims that would be infringed
  only as a consequence of further modifications of THIS
  IMPLEMENTATION.  If you or your agent or licensee institute or order
  or agree to the institution of patent litigation against any entity
  (including a cross-claim or counterclaim in a lawsuit) alleging that
  THIS IMPLEMENTATION constitutes direct or contributory patent
  infringement, or inducement of patent infringement, then any rights
  granted to you under this License shall terminate as of the date
  such litigation is filed.  If you or your agent or exclusive
  licensee institute or order or agree to the institution of a PATENT
  CHALLENGE, then Tokutek may terminate any rights granted to you
  under this License.
*/

87
#ident "Copyright (c) 2010-2013 Tokutek Inc.  All rights reserved."
88
#ident "The technology is licensed by the Massachusetts Institute of Technology, Rutgers State University of New Jersey, and the Research Foundation of State University of New York at Stony Brook under United States of America Serial No. 11/760379 and to the patents and/or patent applications resulting from it."
89
#ident "$Id$"
90 91 92 93 94 95

/*
 *   The indexer
 */
#include <stdio.h>
#include <string.h>
96 97 98
#include <toku_portability.h>
#include "toku_assert.h"
#include "ydb-internal.h"
99
#include <ft/le-cursor.h>
100 101
#include "indexer.h"
#include <ft/tokuconst.h>
102 103 104 105 106
#include <ft/ft-ops.h>
#include <ft/leafentry.h>
#include <ft/ule.h>
#include <ft/xids.h>
#include <ft/log-internal.h>
107
#include <ft/checkpoint.h>
108
#include <portability/toku_atomic.h>
109
#include "loader.h"
110
#include <util/status.h>
111

112 113 114 115 116 117 118 119
///////////////////////////////////////////////////////////////////////////////////
// Engine status
//
// Status is intended for display to humans to help understand system behavior.
// It does not need to be perfectly thread-safe.

static INDEXER_STATUS_S indexer_status;

120
#define STATUS_INIT(k,c,t,l,inc) TOKUDB_STATUS_INIT(indexer_status, k, c, t, "indexer: " l, inc)
121 122 123 124 125

static void
status_init(void) {
    // Note, this function initializes the keyname, type, and legend fields.
    // Value fields are initialized to zero by compiler.
126 127 128 129 130 131 132 133 134
    STATUS_INIT(INDEXER_CREATE,      nullptr, UINT64, "number of indexers successfully created", TOKU_ENGINE_STATUS);
    STATUS_INIT(INDEXER_CREATE_FAIL, nullptr, UINT64, "number of calls to toku_indexer_create_indexer() that failed", TOKU_ENGINE_STATUS);
    STATUS_INIT(INDEXER_BUILD,       nullptr, UINT64, "number of calls to indexer->build() succeeded", TOKU_ENGINE_STATUS);
    STATUS_INIT(INDEXER_BUILD_FAIL,  nullptr, UINT64, "number of calls to indexer->build() failed", TOKU_ENGINE_STATUS);
    STATUS_INIT(INDEXER_CLOSE,       nullptr, UINT64, "number of calls to indexer->close() that succeeded", TOKU_ENGINE_STATUS);
    STATUS_INIT(INDEXER_CLOSE_FAIL,  nullptr, UINT64, "number of calls to indexer->close() that failed", TOKU_ENGINE_STATUS);
    STATUS_INIT(INDEXER_ABORT,       nullptr, UINT64, "number of calls to indexer->abort()", TOKU_ENGINE_STATUS);
    STATUS_INIT(INDEXER_CURRENT,     nullptr, UINT64, "number of indexers currently in existence", TOKU_ENGINE_STATUS);
    STATUS_INIT(INDEXER_MAX,         nullptr, UINT64, "max number of indexers that ever existed simultaneously", TOKU_ENGINE_STATUS);
135 136 137 138 139 140 141
    indexer_status.initialized = true;
}
#undef STATUS_INIT

void
toku_indexer_get_status(INDEXER_STATUS statp) {
    if (!indexer_status.initialized)
142
        status_init();
143 144 145 146 147
    *statp = indexer_status;
}

#define STATUS_VALUE(x) indexer_status.status[x].value.num

148 149 150 151 152 153 154 155 156 157 158 159 160 161
#include "indexer-internal.h"

static int build_index(DB_INDEXER *indexer);
static int close_indexer(DB_INDEXER *indexer);
static int abort_indexer(DB_INDEXER *indexer);
static void free_indexer_resources(DB_INDEXER *indexer);
static void free_indexer(DB_INDEXER *indexer);
static int update_estimated_rows(DB_INDEXER *indexer);
static int maybe_call_poll_func(DB_INDEXER *indexer, uint64_t loop_count);

static int
associate_indexer_with_hot_dbs(DB_INDEXER *indexer, DB *dest_dbs[], int N) {
    int result =0;
    for (int i = 0; i < N; i++) {
162 163 164 165 166 167
        result = toku_db_set_indexer(dest_dbs[i], indexer);
        if (result != 0) {
            for (int j = 0; j < i; j++) {
                int result2 = toku_db_set_indexer(dest_dbs[j], NULL);
                lazy_assert(result2 == 0);
            }
168
            break;
169
        }
170 171 172 173 174 175 176
    }
    return result;
}

static void
disassociate_indexer_from_hot_dbs(DB_INDEXER *indexer) {
    for (int i = 0; i < indexer->i->N; i++) {
177 178
        int result = toku_db_set_indexer(indexer->i->dest_dbs[i], NULL);
        lazy_assert(result == 0);
179 180 181 182 183 184 185 186 187 188 189
    }
}

/*
 *  free_indexer_resources() frees all of the resources associated with
 *      struct __toku_indexer_internal 
 *  assumes any previously freed items set the field pointer to NULL
 */

static void 
free_indexer_resources(DB_INDEXER *indexer) {
190 191
    if ( indexer->i ) {        
        toku_mutex_destroy(&indexer->i->indexer_lock);
192 193 194
        if ( indexer->i->lec ) {
            toku_le_cursor_close(indexer->i->lec);
        }
195 196 197 198
        if ( indexer->i->fnums ) { 
            toku_free(indexer->i->fnums); 
            indexer->i->fnums = NULL;
        }
199
        indexer_undo_do_destroy(indexer);
200 201 202 203 204 205 206 207 208 209 210 211 212 213 214
        // indexer->i
        toku_free(indexer->i);
        indexer->i = NULL;
    }
}

static void 
free_indexer(DB_INDEXER *indexer) {
    if ( indexer ) {
        free_indexer_resources(indexer);
        toku_free(indexer);
        indexer = NULL;
    }
}

215 216 217 218 219 220 221 222 223 224
void
toku_indexer_lock(DB_INDEXER* indexer) {
    toku_mutex_lock(&indexer->i->indexer_lock);
}

void
toku_indexer_unlock(DB_INDEXER* indexer) {
    toku_mutex_unlock(&indexer->i->indexer_lock);
}

225 226 227
// forward declare the test-only wrapper function for undo-do
static int test_indexer_undo_do(DB_INDEXER *indexer, DB *hotdb, ULEHANDLE ule);

228
int
229 230 231 232 233
toku_indexer_create_indexer(DB_ENV *env,
                            DB_TXN *txn,
                            DB_INDEXER **indexerp,
                            DB *src_db,
                            int N,
234 235
                            DB *dest_dbs[/*N*/],
                            uint32_t db_flags[/*N*/] UU(),
236
                            uint32_t indexer_flags)
237
{
238
    int rval;
239
    DB_INDEXER *indexer = 0;   // set later when created
240
    HANDLE_READ_ONLY_TXN(txn);
241 242

    *indexerp = NULL;
243

244 245 246 247
    XCALLOC(indexer);      // init to all zeroes (thus initializing the error_callback and poll_func)
    if ( !indexer )    { rval = ENOMEM; goto create_exit; }
    XCALLOC(indexer->i);   // init to all zeroes (thus initializing all pointers to NULL)
    if ( !indexer->i ) { rval = ENOMEM; goto create_exit; }
248

249 250 251 252 253 254 255
    indexer->i->env                = env;
    indexer->i->txn                = txn;
    indexer->i->src_db             = src_db;
    indexer->i->N                  = N;
    indexer->i->dest_dbs           = dest_dbs;
    indexer->i->indexer_flags      = indexer_flags;
    indexer->i->loop_mod           = 1000; // call poll_func every 1000 rows
256
    indexer->i->estimated_rows     = 0;
257
    indexer->i->undo_do            = test_indexer_undo_do; // TEST export the undo do function
258

259 260 261
    XCALLOC_N(N, indexer->i->fnums);
    if ( !indexer->i->fnums ) { rval = ENOMEM; goto create_exit; }
    for(int i=0;i<indexer->i->N;i++) {
262
        indexer->i->fnums[i]       = toku_cachefile_filenum(db_struct_i(dest_dbs[i])->ft_handle->ft->cf);
263
    }
264 265 266
    indexer->i->filenums.num       = N;
    indexer->i->filenums.filenums  = indexer->i->fnums;
    indexer->i->test_only_flags    = 0;  // for test use only
267

268 269 270 271 272
    indexer->set_error_callback    = toku_indexer_set_error_callback;
    indexer->set_poll_function     = toku_indexer_set_poll_function;
    indexer->build                 = build_index;
    indexer->close                 = close_indexer;
    indexer->abort                 = abort_indexer;
273

274 275
    toku_mutex_init(&indexer->i->indexer_lock, NULL);

276 277 278 279 280 281
    //
    // create and close a dummy loader to get redirection going for the hot indexer
    // This way, if the hot index aborts, but other transactions have references to the
    // underlying FT, then those transactions can do dummy operations on the FT
    // while the DB gets redirected back to an empty dictionary
    //
282
    {
283
        DB_LOADER* loader = NULL;
284
        rval = toku_loader_create_loader(env, txn, &loader, NULL, N, &dest_dbs[0], NULL, NULL, DB_PRELOCKED_WRITE | LOADER_DISALLOW_PUTS, true);
285
        if (rval) {
286 287
            goto create_exit;
        }
288 289
        rval = loader->close(loader);
        if (rval) {
290 291 292
            goto create_exit;
        }
    }
293

294
    // create and initialize the leafentry cursor
295
    rval = toku_le_cursor_create(&indexer->i->lec, db_struct_i(src_db)->ft_handle, db_txn_struct_i(txn)->tokutxn);
296
    if ( !indexer->i->lec ) { goto create_exit; }
297

298 299
    // 2954: add recovery and rollback entries
    LSN hot_index_lsn; // not used (yet)
300 301 302 303
    TOKUTXN      ttxn;
    ttxn = db_txn_struct_i(txn)->tokutxn;
    FILENUMS filenums;
    filenums = indexer->i->filenums;
304
    toku_multi_operation_client_lock();
305
    toku_ft_hot_index(NULL, ttxn, filenums, 1, &hot_index_lsn);
306
    toku_multi_operation_client_unlock();
307

308 309 310
    if (rval == 0) {
        rval = associate_indexer_with_hot_dbs(indexer, dest_dbs, N);
    }
311 312 313
create_exit:
    if ( rval == 0 ) {

314
        indexer_undo_do_init(indexer);
315

316 317
        *indexerp = indexer;

318 319
        (void) toku_sync_fetch_and_add(&STATUS_VALUE(INDEXER_CREATE), 1);
        (void) toku_sync_fetch_and_add(&STATUS_VALUE(INDEXER_CURRENT), 1);
320
        if ( STATUS_VALUE(INDEXER_CURRENT) > STATUS_VALUE(INDEXER_MAX) )
321
            STATUS_VALUE(INDEXER_MAX) = STATUS_VALUE(INDEXER_CURRENT);   // NOT WORTH A LOCK TO MAKE THREADSAFE), may be inaccurate
322

323
    } else {
324
        (void) toku_sync_fetch_and_add(&STATUS_VALUE(INDEXER_CREATE_FAIL), 1);
325 326 327 328 329 330
        free_indexer(indexer);
    }

    return rval;
}

331
int
332 333 334 335 336 337 338 339 340 341 342 343 344 345 346 347 348 349 350 351 352 353 354 355
toku_indexer_set_poll_function(DB_INDEXER *indexer,
                               int (*poll_func)(void *poll_extra,
                                                float progress),
                               void *poll_extra)
{
    invariant(indexer != NULL);
    indexer->i->poll_func  = poll_func;
    indexer->i->poll_extra = poll_extra;
    return 0;
}

int 
toku_indexer_set_error_callback(DB_INDEXER *indexer,
                                void (*error_cb)(DB *db, int i, int err,
                                                 DBT *key, DBT *val,
                                                 void *error_extra),
                                void *error_extra)
{
    invariant(indexer != NULL);
    indexer->i->error_callback = error_cb;
    indexer->i->error_extra    = error_extra;
    return 0;
}

356 357 358 359 360
// a key is to the right of the indexer's cursor if it compares
// greater than the current le cursor position.
bool
toku_indexer_is_key_right_of_le_cursor(DB_INDEXER *indexer, const DBT *key) {
    return toku_le_cursor_is_key_greater(indexer->i->lec, key);
361 362
}

363 364 365
// initialize provisional info by allocating enough space to hold provisional 
// ids, states, and txns for each of the provisional entries in the ule. the 
// ule and le remain owned by the caller, not this struct.
366
static void 
367 368
ule_prov_info_init(struct ule_prov_info *prov_info, LEAFENTRY le, ULEHANDLE ule) {
    prov_info->le = le;
369 370 371 372 373
    prov_info->ule = ule;
    prov_info->num_provisional = ule_get_num_provisional(ule);
    prov_info->num_committed = ule_get_num_committed(ule);
    uint32_t n = prov_info->num_provisional;
    if (n > 0) {
374 375 376
        XMALLOC_N(n, prov_info->prov_ids);
        XMALLOC_N(n, prov_info->prov_states);
        XMALLOC_N(n, prov_info->prov_txns);
377 378 379 380 381 382 383 384 385 386 387 388 389 390 391 392 393 394 395 396 397 398 399 400 401 402 403
    }
}

// clean up anything possibly created by ule_prov_info_init()
static void
ule_prov_info_destroy(struct ule_prov_info *prov_info) {
    if (prov_info->num_provisional > 0) {
        toku_free(prov_info->prov_ids);
        toku_free(prov_info->prov_states);
        toku_free(prov_info->prov_txns);
    } else {
        // nothing to free if there was nothing provisional
        invariant(prov_info->prov_ids == NULL);
        invariant(prov_info->prov_states == NULL);
        invariant(prov_info->prov_txns == NULL);
    }
}

static void
indexer_fill_prov_info(DB_INDEXER *indexer, struct ule_prov_info *prov_info) {
    ULEHANDLE ule = prov_info->ule;
    uint32_t num_provisional = prov_info->num_provisional;
    uint32_t num_committed = prov_info->num_committed;
    TXNID *prov_ids = prov_info->prov_ids;
    TOKUTXN_STATE *prov_states = prov_info->prov_states;
    TOKUTXN *prov_txns = prov_info->prov_txns;

404 405 406 407 408 409
    // don't both grabbing the txn manager lock if we don't
    // have any provisional txns to record
    if (num_provisional == 0) {
        return;
    }

410 411 412 413 414 415 416 417 418 419 420
    // handle test case first
    if (indexer->i->test_xid_state) {
        for (uint32_t i = 0; i < num_provisional; i++) {
            UXRHANDLE uxr = ule_get_uxr(ule, num_committed + i);
            prov_ids[i] = uxr_get_txnid(uxr);
            prov_states[i] = indexer->i->test_xid_state(indexer, prov_ids[i]);
            prov_txns[i] = NULL;
        }
        return;
    }
    
421 422 423 424
    // hold the txn manager lock while we inspect txn state
    // and pin some live txns
    DB_ENV *env = indexer->i->env;
    TXN_MANAGER txn_manager = toku_logger_get_txn_manager(env->i->logger);
425 426 427
    TXNID parent_xid = uxr_get_txnid(ule_get_uxr(ule, num_committed));

    // let's first initialize things to defaults
428 429 430
    for (uint32_t i = 0; i < num_provisional; i++) {
        UXRHANDLE uxr = ule_get_uxr(ule, num_committed + i);
        prov_ids[i] = uxr_get_txnid(uxr);
431 432 433 434 435 436 437 438 439 440 441 442 443 444 445 446 447 448 449 450 451 452 453 454 455 456 457 458 459 460 461 462 463 464 465 466 467 468 469 470 471 472 473
        prov_txns[i] = NULL;
        prov_states[i] = TOKUTXN_RETIRED;
    }

    toku_txn_manager_suspend(txn_manager); 
    TXNID_PAIR root_xid_pair = {.parent_id64=parent_xid, .child_id64 = TXNID_NONE};
    TOKUTXN root_txn = NULL;
    toku_txn_manager_id2txn_unlocked(
        txn_manager, 
        root_xid_pair, 
        &root_txn
        );
    if (root_txn == NULL) {
        toku_txn_manager_resume(txn_manager);
        return; //everything is retired in this case, the default
    }
    prov_txns[0] = root_txn;
    prov_states[0] = toku_txn_get_state(root_txn);
    toku_txn_lock_state(root_txn);
    prov_states[0] = toku_txn_get_state(root_txn);
    if (prov_states[0] == TOKUTXN_LIVE || prov_states[0] == TOKUTXN_PREPARING) {
        // pin this live txn so it can't commit or abort until we're done with it
        toku_txn_pin_live_txn_unlocked(root_txn);
    }
    toku_txn_unlock_state(root_txn);

    root_txn->child_manager->suspend();
    for (uint32_t i = 1; i < num_provisional; i++) {
        UXRHANDLE uxr = ule_get_uxr(ule, num_committed + i);
        TXNID child_id = uxr_get_txnid(uxr);
        TOKUTXN txn = NULL;

        TXNID_PAIR txnid_pair = {.parent_id64 = parent_xid, .child_id64 = child_id};
        root_txn->child_manager->find_tokutxn_by_xid_unlocked(txnid_pair, &txn);
        prov_txns[i] = txn;
        if (txn) {
            toku_txn_lock_state(txn);
            prov_states[i] = toku_txn_get_state(txn);
            if (prov_states[i] == TOKUTXN_LIVE || prov_states[i] == TOKUTXN_PREPARING) {
                // pin this live txn so it can't commit or abort until we're done with it
                toku_txn_pin_live_txn_unlocked(txn);
            }
            toku_txn_unlock_state(txn);
474 475
        }
        else {
476
            prov_states[i] = TOKUTXN_RETIRED;
477 478
        }
    }
479
    root_txn->child_manager->resume();
480 481 482 483 484 485 486 487 488 489 490 491
    toku_txn_manager_resume(txn_manager);
}
    
struct le_cursor_extra {
    DB_INDEXER *indexer;
    struct ule_prov_info *prov_info;
};

// cursor callback, so its synchronized with other db operations using 
// cachetable pair locks. because no txn can commit on this db, read
// the provisional info for the newly read ule.
static int
492
le_cursor_callback(ITEMLEN UU(keylen), bytevec UU(key), ITEMLEN UU(vallen), bytevec val, void *extra, bool lock_only) {
493
    if (lock_only || val == NULL) {
494
        ; // do nothing if only locking. do nothing if val==NULL, means DB_NOTFOUND
495
    } else {
496
        struct le_cursor_extra *CAST_FROM_VOIDP(cursor_extra, extra);
497
        struct ule_prov_info *prov_info = cursor_extra->prov_info;
498 499 500
        // the val here is a leafentry. ule_create does not copy the entire
        // contents of the leafentry it is given into its own buffers, so we
        // must allocate space for a leafentry and keep it around with the ule.
501
        LEAFENTRY CAST_FROM_VOIDP(le, toku_xmemdup(val, vallen));
502
        ULEHANDLE ule = toku_ule_create(le);
503
        invariant(ule);
504 505 506 507
        // when we initialize prov info, we also pass in the leafentry and ule
        // pointers so the caller can access them later. it's their job to free
        // them when they're not needed.
        ule_prov_info_init(prov_info, le, ule);
508 509 510 511 512 513 514 515 516 517 518 519 520 521 522 523 524 525
        indexer_fill_prov_info(cursor_extra->indexer, prov_info);
    }
    return 0;
}

// get the next ule and fill out its provisional info in the
// prov_info struct provided. caller is responsible for cleaning
// up the ule info after it's done.
static int
get_next_ule_with_prov_info(DB_INDEXER *indexer, struct ule_prov_info *prov_info) {
    struct le_cursor_extra extra = {
        .indexer = indexer,
        .prov_info = prov_info,
    };
    int r = toku_le_cursor_next(indexer->i->lec, le_cursor_callback, &extra);
    return r; 
}

526 527 528 529
static int 
build_index(DB_INDEXER *indexer) {
    int result = 0;

530
    bool done = false;
531 532
    for (uint64_t loop_count = 0; !done; loop_count++) {

533
        toku_indexer_lock(indexer);
534 535 536 537
        // grab the multi operation lock because we will be injecting messages
        // grab it here because we must hold it before
        // trying to pin any live transactions, as discovered by #5775
        toku_multi_operation_client_lock();
538

539 540 541
        // grab the next leaf entry and get its provisional info. we'll
        // need the provisional info for the undo-do algorithm, and we get
        // it here so it can be read atomically with respect to txn commit
542 543
        // and abort. the atomicity comes from the root-to-leaf path pinned
        // by the query and in the getf callback function
544 545 546 547 548 549 550
        //
        // this allocates space for the prov info, so we have to destroy it
        // when we're done.
        struct ule_prov_info prov_info;
        memset(&prov_info, 0, sizeof(prov_info));
        result = get_next_ule_with_prov_info(indexer, &prov_info);

551
        if (result != 0) {
552
            invariant(prov_info.ule == NULL);
553
            done = true;
554
            if (result == DB_NOTFOUND) {
555
                result = 0;  // all done, normal way to exit loop successfully
556
            }
557
        }
558
        else {
559
            invariant(prov_info.le);
560 561
            invariant(prov_info.ule);
            ULEHANDLE ule = prov_info.ule;
562 563
            for (int which_db = 0; (which_db < indexer->i->N) && (result == 0); which_db++) {
                DB *db = indexer->i->dest_dbs[which_db];
564 565 566 567
                result = indexer_undo_do(indexer, db, ule, &prov_info);
                if ((result != 0) && (indexer->i->error_callback != NULL)) {
                    // grab the key and call the error callback
                    DBT key; toku_init_dbt_flags(&key, DB_DBT_REALLOC);
568
                    toku_dbt_set(ule_get_keylen(ule), ule_get_key(ule), &key, NULL);
569
                    indexer->i->error_callback(db, which_db, result, &key, NULL, indexer->i->error_extra);
570
                    toku_destroy_dbt(&key);
571 572
                }
            }
573 574 575 576
            // the leafentry and ule are not owned by the prov_info,
            // and are still our responsibility to free
            toku_free(prov_info.le);
            toku_ule_free(prov_info.ule);
577 578
        }

579
        toku_multi_operation_client_unlock();
580
        toku_indexer_unlock(indexer);
581
        ule_prov_info_destroy(&prov_info);
582
        
583
        if (result == 0) {
584
            result = maybe_call_poll_func(indexer, loop_count);
585 586
        }
        if (result != 0) {
587
            done = true;
588
        }
589 590 591 592 593 594 595
    }

    // post index creation cleanup
    //  - optimize?
    //  - garbage collect?
    //  - unique checks?

596
    if ( result == 0 ) {
597 598 599 600 601
        // Perform a checkpoint so that all of the indexing makes it to disk before continuing.
        // Otherwise indexing would not be crash-safe becasue none of the undo-do messages are in the recovery log.
        DB_ENV *env = indexer->i->env;
        CHECKPOINTER cp = toku_cachetable_get_checkpointer(env->i->cachetable);
        toku_checkpoint(cp, env->i->logger, NULL, NULL, NULL, NULL, INDEXER_CHECKPOINT);
602
        (void) toku_sync_fetch_and_add(&STATUS_VALUE(INDEXER_BUILD), 1);
603
    } else {
604
        (void) toku_sync_fetch_and_add(&STATUS_VALUE(INDEXER_BUILD_FAIL), 1);
605 606
    }

607 608 609
    return result;
}

610
// Clients must not operate on any of the hot dbs concurrently with close
611
static int
612 613
close_indexer(DB_INDEXER *indexer) {
    int r = 0;
614
    (void) toku_sync_fetch_and_sub(&STATUS_VALUE(INDEXER_CURRENT), 1);
615

616 617 618
    // Disassociate the indexer from the hot db and free_indexer
    disassociate_indexer_from_hot_dbs(indexer);
    free_indexer(indexer);
619

620
    if ( r == 0 ) {
621
        (void) toku_sync_fetch_and_add(&STATUS_VALUE(INDEXER_CLOSE), 1);
622
    } else {
623
        (void) toku_sync_fetch_and_add(&STATUS_VALUE(INDEXER_CLOSE_FAIL), 1);
624 625 626 627
    }
    return r;
}

628
// Clients must not operate on any of the hot dbs concurrently with abort
629 630
static int 
abort_indexer(DB_INDEXER *indexer) {
631 632
    (void) toku_sync_fetch_and_sub(&STATUS_VALUE(INDEXER_CURRENT), 1);
    (void) toku_sync_fetch_and_add(&STATUS_VALUE(INDEXER_ABORT), 1);
633 634 635
    // Disassociate the indexer from the hot db and free_indexer
    disassociate_indexer_from_hot_dbs(indexer);
    free_indexer(indexer);
636 637 638 639 640 641 642 643 644
    return 0;
}


// derived from ha_tokudb::estimate_num_rows
static int
update_estimated_rows(DB_INDEXER *indexer) {
    DBT key;  toku_init_dbt(&key);
    DBT data; toku_init_dbt(&data);
645
    DBC* crsr = NULL;
646 647 648 649 650 651 652 653 654 655 656 657 658 659 660 661 662 663 664 665 666 667 668 669 670 671 672 673 674 675 676 677 678 679 680 681 682 683 684 685 686 687 688 689 690 691
    DB_TXN* txn = NULL;
    uint64_t less, equal, greater;
    int is_exact;
    int error;
    DB *db = indexer->i->src_db;
    DB_ENV *db_env = indexer->i->env;

    error = db_env->txn_begin(db_env, 0, &txn, DB_READ_UNCOMMITTED);
    if (error) goto cleanup;

    error = db->cursor(db, txn, &crsr, 0);
    if (error) { goto cleanup; }
    
    error = crsr->c_get(crsr, &key, &data, DB_FIRST);
    if (error == DB_NOTFOUND) {
        indexer->i->estimated_rows = 0;
        error = 0;
        goto cleanup;
    } 
    else if (error) { goto cleanup; }

    error = db->key_range64(db, txn, &key,
                            &less, &equal, &greater,
                            &is_exact);
    if (error) { goto cleanup; }

    indexer->i->estimated_rows = equal + greater;
    error = 0;
cleanup:
    if ( crsr != NULL ) {
        int rr = crsr->c_close(crsr);
        invariant(rr == 0);
        crsr = NULL;
    }
    txn->commit(txn, 0);
    return error;
}

static int 
maybe_call_poll_func(DB_INDEXER *indexer, uint64_t loop_count) {
    int result = 0;
    if ( indexer->i->poll_func != NULL && ( loop_count % indexer->i->loop_mod ) == 0 ) {
        int r __attribute__((unused)) = update_estimated_rows(indexer);
        // what happens if estimate_rows fails?
        //   - currently does not modify estimate, which is probably sufficient
        float progress;
692
        if ( indexer->i->estimated_rows == 0  || loop_count > indexer->i->estimated_rows)
693
            progress = 1.0;
694
        else
695 696 697 698 699 700 701 702 703 704 705 706 707 708
            progress = (float)loop_count / (float)indexer->i->estimated_rows;
        result = indexer->i->poll_func(indexer->i->poll_extra, progress);
    }
    return result;
}


// this allows us to force errors under test.  Flags are defined in indexer.h
void
toku_indexer_set_test_only_flags(DB_INDEXER *indexer, int flags) {
    invariant(indexer != NULL);
    indexer->i->test_only_flags = flags;
}

709 710 711 712 713 714
// this allows us to call the undo do function in tests using
// a convenience wrapper that gets and destroys the ule's prov info
static int
test_indexer_undo_do(DB_INDEXER *indexer, DB *hotdb, ULEHANDLE ule) {
    struct ule_prov_info prov_info;
    memset(&prov_info, 0, sizeof(prov_info));
715 716
    // pass null for the leafentry - we don't need it, neither does the info
    ule_prov_info_init(&prov_info, NULL, ule);
717 718 719 720 721 722
    indexer_fill_prov_info(indexer, &prov_info);
    int r = indexer_undo_do(indexer, hotdb, ule, &prov_info);
    ule_prov_info_destroy(&prov_info);
    return r;
}

723 724 725 726
DB *
toku_indexer_get_src_db(DB_INDEXER *indexer) {
    return indexer->i->src_db;
}
727 728 729 730


#undef STATUS_VALUE