Closes #1862 closes[t:1862] LSNs are logged/chosen during the actual close, if...

Closes #1862 closes[t:1862] LSNs are logged/chosen during the actual close, if closed during recovery, the LSN is passed into close.

git-svn-id: file:///svn/toku/tokudb@14686 c7de825b-a66e-492c-adef-691d508d4ae1

newbrt/brt-internal.h

@@ -146,6 +146,7 @@ struct brt_header {
     enum brtheader_type type;
     struct brt_header * checkpoint_header;
     CACHEFILE cf;
+    char *fname; // the filename
     u_int64_t checkpoint_count; // Free-running counter incremented once per checkpoint (toggling LSB).
                                 // LSB indicates which header location is used on disk so this
                                 // counter is effectively a boolean which alternates with each checkpoint.
@@ -176,7 +177,6 @@ struct brt_header {
 
 struct brt {
     CACHEFILE cf;
-    char *fname; // the filename
     // The header is shared.  It is also ephemeral.
     struct brt_header *h;
 
@@ -329,7 +329,7 @@ enum brt_layout_version_e {
 };
 
 void toku_brtheader_free (struct brt_header *h);
-int toku_brtheader_close (CACHEFILE cachefile, void *header_v, char **error_string, LSN);
+int toku_brtheader_close (CACHEFILE cachefile, void *header_v, char **error_string, BOOL oplsn_valid, LSN oplsn);
 int toku_brtheader_begin_checkpoint (CACHEFILE cachefile, LSN checkpoint_lsn, void *header_v);
 int toku_brtheader_checkpoint (CACHEFILE cachefile, void *header_v);
 int toku_brtheader_end_checkpoint (CACHEFILE cachefile, void *header_v);

newbrt/brt.c

@@ -581,6 +581,7 @@ brtheader_destroy(struct brt_header *h) {
         assert(h->type == BRTHEADER_CURRENT);
         toku_blocktable_destroy(&h->blocktable);
         if (h->descriptor.dbt.data) toku_free(h->descriptor.dbt.data);
+        if (h->fname) toku_free(h->fname);
     }
 }
 
@@ -2958,10 +2959,19 @@ brtheader_note_brt_close(BRT t) {
 }
 
 static int
-brtheader_note_brt_open(BRT live) {
+brtheader_note_brt_open(BRT live, char** fnamep) {
     struct brt_header *h = live->h;
     int retval = 0;
     toku_brtheader_lock(h);
+    //Steal fname reference.
+    char *fname = *fnamep;
+    assert(fname);
+    *fnamep     = NULL;
+    //Use fname, or throw it away.
+    if (h->fname == NULL)
+        h->fname = fname;
+    else
+        toku_free(fname);
     while (!list_empty(&h->zombie_brts)) {
         //Remove dead brt from list
         BRT zombie = list_struct(list_pop(&h->zombie_brts), struct brt, zombie_brt_link);
@@ -2986,10 +2996,10 @@ int toku_brt_open(BRT t, const char *fname, const char *fname_in_env, int is_cre
     if (0) { died0:  assert(r); return r; }
 
     assert(is_create || !only_create);
-    t->fname = toku_strdup(fname_in_env);
-    if (t->fname==0) {
+    char* brt_fname = toku_strdup(fname_in_env);
+    if (brt_fname==NULL) {
         r = errno;
-        if (0) { died00: if (t->fname) toku_free(t->fname); t->fname=0; }
+        if (0) { died00: if (brt_fname) toku_free(brt_fname); brt_fname=NULL; }
         goto died0;
     }
     t->db = db;
@@ -3011,7 +3021,7 @@ int toku_brt_open(BRT t, const char *fname, const char *fname_in_env, int is_cre
     }
     if (r!=0) {
         died_after_open: 
-        toku_cachefile_close(&t->cf, toku_txn_logger(txn), 0, ZERO_LSN);
+        toku_cachefile_close(&t->cf, toku_txn_logger(txn), 0, FALSE, ZERO_LSN);
         goto died00;
     }
     assert(t->nodesize>0);
@@ -3082,7 +3092,7 @@ int toku_brt_open(BRT t, const char *fname, const char *fname_in_env, int is_cre
     r = toku_maybe_upgrade_brt(t);	// possibly do some work to complete the version upgrade of brt
     if (r!=0) goto died_after_read_and_pin;
 
-    r = brtheader_note_brt_open(t);
+    r = brtheader_note_brt_open(t, &brt_fname);
     if (r!=0) goto died_after_read_and_pin;
     if (t->db) t->db->descriptor = &t->h->descriptor.dbt;
     if (txn_created) {
@@ -3217,8 +3227,10 @@ toku_brtheader_end_checkpoint (CACHEFILE cachefile, void *header_v) {
             r = toku_cachefile_fsync(cachefile);
 	    if (r!=0) 
 		toku_block_translation_note_failed_checkpoint(h->blocktable);
-	    else
+	    else {
 		h->checkpoint_count++;	// checkpoint succeeded, next checkpoint will save to alternate header location
+                h->checkpoint_lsn = ch->checkpoint_lsn;  //Header updated.
+            }
 	}
         toku_block_translation_note_end_checkpoint(h->blocktable, h);
     }
@@ -3230,15 +3242,36 @@ toku_brtheader_end_checkpoint (CACHEFILE cachefile, void *header_v) {
 }
 
 int
-toku_brtheader_close (CACHEFILE cachefile, void *header_v, char **malloced_error_string, LSN lsn)
-{
+toku_brtheader_close (CACHEFILE cachefile, void *header_v, char **malloced_error_string, BOOL oplsn_valid, LSN oplsn) {
     struct brt_header *h = header_v;
     assert(h->type == BRTHEADER_CURRENT);
     toku_brtheader_lock(h);
     assert(list_empty(&h->live_brts));
     assert(list_empty(&h->zombie_brts));
     toku_brtheader_unlock(h);
+    LSN lsn = ZERO_LSN;
     int r = 0;
+    if (h->fname) { //Otherwise header has never fully been created.
+        //Get LSN
+        if (oplsn_valid) {
+            //Use recovery-specified lsn
+            lsn = oplsn;
+            //Recovery cannot reduce lsn of a header.
+            if (lsn.lsn < h->checkpoint_lsn.lsn)
+                lsn = h->checkpoint_lsn;
+        }
+        else {
+            //Get LSN from logger
+            lsn = ZERO_LSN; // if there is no logger, we use zero for the lsn
+            TOKULOGGER logger = toku_cachefile_logger(cachefile);
+            if (logger) {
+                //NEED NAME
+                assert(h->fname);
+                BYTESTRING bs = {.len=strlen(h->fname), .data=h->fname};
+                r = toku_log_fclose(logger, &lsn, h->dirty, bs, toku_cachefile_filenum(cachefile)); // flush the log on close (if new header is being written), otherwise it might not make it out.
+                if (r!=0) return r;
+            }
+        }
         if (h->dirty) {	// this is the only place this bit is tested (in currentheader)
             int r2;
             //assert(lsn.lsn!=0);
@@ -3250,6 +3283,7 @@ toku_brtheader_close (CACHEFILE cachefile, void *header_v, char **malloced_error
             if (r==0) r = r2;
             if (!h->panic) assert(!h->dirty);	// dirty bit should be cleared by begin_checkpoint and never set again (because we're closing the dictionary)
         }
+    }
     if (malloced_error_string) *malloced_error_string = h->panic_string;
     if (r==0) {
 	r=h->panic;
@@ -3299,7 +3333,7 @@ toku_brt_db_delay_closed (BRT zombie, DB* db, int (*close_db)(DB*, u_int32_t), u
     return r;
 }
 
-int toku_close_brt (BRT brt, TOKULOGGER logger, char **error_string) {
+int toku_close_brt_lsn (BRT brt, TOKULOGGER logger, char **error_string, BOOL oplsn_valid, LSN oplsn) {
     assert(toku_omt_size(brt->txns)==0);
     int r;
     while (!list_empty(&brt->cursors)) {
@@ -3315,27 +3349,23 @@ int toku_close_brt (BRT brt, TOKULOGGER logger, char **error_string) {
     brtheader_note_brt_close(brt);
 
     if (brt->cf) {
-	LSN lsn = ZERO_LSN; // if there is no logger, we use zero for the lsn
-        if (logger) {
-            assert(brt->fname);
-            BYTESTRING bs = {.len=strlen(brt->fname), .data=brt->fname};
-            r = toku_log_fclose(logger, &lsn, 1, bs, toku_cachefile_filenum(brt->cf)); // flush the log on close, otherwise it might not make it out.
-            if (r!=0) return r;
-        }
 	if (!brt->h->panic)
 	    assert(0==toku_cachefile_count_pinned(brt->cf, 1)); // For the brt, the pinned count should be zero (but if panic, don't worry)
         //printf("%s:%d closing cachetable\n", __FILE__, __LINE__);
         // printf("%s:%d brt=%p ,brt->h=%p\n", __FILE__, __LINE__, brt, brt->h);
 	if (error_string) assert(*error_string == 0);
-        r = toku_cachefile_close(&brt->cf, logger, error_string, lsn);
+        r = toku_cachefile_close(&brt->cf, logger, error_string, oplsn_valid, oplsn);
 	if (r==0 && error_string) assert(*error_string == 0);
     }
-    if (brt->fname) toku_free(brt->fname);
     if (brt->temp_descriptor.dbt.data) toku_free(brt->temp_descriptor.dbt.data);
     toku_free(brt);
     return r;
 }
 
+int toku_close_brt (BRT brt, TOKULOGGER logger, char **error_string) {
+    return toku_close_brt_lsn(brt, logger, error_string, FALSE, ZERO_LSN);
+}
+
 int toku_brt_create(BRT *brt_ptr) {
     BRT brt = toku_malloc(sizeof *brt);
     if (brt == 0)

newbrt/brt.h

@@ -82,6 +82,7 @@ int toku_brt_maybe_delete_both (BRT brt, DBT *k, DBT *v, TOKUTXN txn, BOOL oplsn
 
 int toku_brt_db_delay_closed (BRT brt, DB* db, int (*close_db)(DB*, u_int32_t), u_int32_t close_flags);
 int toku_close_brt (BRT, TOKULOGGER, char **error_string);
+int toku_close_brt_lsn (BRT brt, TOKULOGGER logger, char **error_string, BOOL oplsn_valid, LSN oplsn);
 
 int toku_dump_brt (FILE *,BRT brt);
 

newbrt/cachetable.c

@@ -191,7 +191,7 @@ struct cachefile {
     char *fname_relative_to_env; /* Used for logging */
 
     void *userdata;
-    int (*close_userdata)(CACHEFILE cf, void *userdata, char **error_string, LSN); // when closing the last reference to a cachefile, first call this function. 
+    int (*close_userdata)(CACHEFILE cf, void *userdata, char **error_string, BOOL lsnvalid, LSN); // when closing the last reference to a cachefile, first call this function. 
     int (*begin_checkpoint_userdata)(CACHEFILE cf, LSN lsn_of_checkpoint, void *userdata); // before checkpointing cachefiles call this function.
     int (*checkpoint_userdata)(CACHEFILE cf, void *userdata); // when checkpointing a cachefile, call this function.
     int (*end_checkpoint_userdata)(CACHEFILE cf, void *userdata); // after checkpointing cachefiles call this function.
@@ -375,7 +375,7 @@ int toku_cachefile_set_fd (CACHEFILE cf, int fd, const char *fname_relative_to_e
     if (r != 0) { 
         r=errno; close(fd); return r; 
     }
-    if (cf->close_userdata && (r = cf->close_userdata(cf, cf->userdata, 0, ZERO_LSN))) {
+    if (cf->close_userdata && (r = cf->close_userdata(cf, cf->userdata, 0, FALSE, ZERO_LSN))) {
         return r;
     }
     cf->close_userdata = NULL;
@@ -427,12 +427,14 @@ static CACHEFILE remove_cf_from_list (CACHEFILE cf, CACHEFILE list) {
 
 static int cachetable_flush_cachefile (CACHETABLE, CACHEFILE cf);
 
-int toku_cachefile_close (CACHEFILE *cfp, TOKULOGGER logger, char **error_string, LSN lsn) {
+int toku_cachefile_close (CACHEFILE *cfp, TOKULOGGER logger, char **error_string, BOOL oplsn_valid, LSN oplsn) {
 
     CACHEFILE cf = *cfp;
     CACHETABLE ct = cf->cachetable;
     cachetable_lock(ct);
     assert(cf->refcount>0);
+    if (oplsn_valid)
+        assert(cf->refcount==1); //Recovery is trying to force an lsn.  Must get passed through.
     cf->refcount--;
     if (cf->refcount==0) {
         //Checkpoint holds a reference, close should be impossible if still in use by a checkpoint.
@@ -467,7 +469,7 @@ int toku_cachefile_close (CACHEFILE *cfp, TOKULOGGER logger, char **error_string
 	    cachetable_unlock(ct);
 	    return r;
         }
-	if (cf->close_userdata && (r = cf->close_userdata(cf, cf->userdata, error_string, lsn))) {
+	if (cf->close_userdata && (r = cf->close_userdata(cf, cf->userdata, error_string, oplsn_valid, oplsn))) {
 	    goto error;
 	}
        	cf->close_userdata = NULL;
@@ -1710,7 +1712,7 @@ toku_cachetable_end_checkpoint(CACHETABLE ct, TOKULOGGER logger, char **error_st
             ct->cachefiles_in_checkpoint = cf->next_in_checkpoint; 
             cf->next_in_checkpoint       = NULL;
             cf->for_checkpoint           = FALSE;
-            int r = toku_cachefile_close(&cf, logger, error_string, ct->lsn_of_checkpoint_in_progress);
+            int r = toku_cachefile_close(&cf, logger, error_string, FALSE, ZERO_LSN);
             if (r!=0) {
                 retval = r;
                 goto panic;
@@ -1871,7 +1873,7 @@ int toku_cachetable_get_key_state (CACHETABLE ct, CACHEKEY key, CACHEFILE cf, vo
 void
 toku_cachefile_set_userdata (CACHEFILE cf,
 			     void *userdata,
-			     int (*close_userdata)(CACHEFILE, void*, char**, LSN),
+			     int (*close_userdata)(CACHEFILE, void*, char**, BOOL, LSN),
 			     int (*checkpoint_userdata)(CACHEFILE, void*),
 			     int (*begin_checkpoint_userdata)(CACHEFILE, LSN, void*),
 			     int (*end_checkpoint_userdata)(CACHEFILE, void*)) {

newbrt/cachetable.h

@@ -90,7 +90,7 @@ typedef void (*CACHETABLE_FLUSH_CALLBACK)(CACHEFILE, CACHEKEY key, void *value,
 // associated with the key are returned.
 typedef int (*CACHETABLE_FETCH_CALLBACK)(CACHEFILE, CACHEKEY key, u_int32_t fullhash, void **value, long *sizep, void *extraargs);
 
-void toku_cachefile_set_userdata(CACHEFILE cf, void *userdata, int (*close_userdata)(CACHEFILE, void*, char **/*error_string*/, LSN), int (*checkpoint_userdata)(CACHEFILE, void*), int (*begin_checkpoint_userdata)(CACHEFILE, LSN, void*), int (*end_checkpoint_userdata)(CACHEFILE, void*));
+void toku_cachefile_set_userdata(CACHEFILE cf, void *userdata, int (*close_userdata)(CACHEFILE, void*, char **/*error_string*/, BOOL, LSN), int (*checkpoint_userdata)(CACHEFILE, void*), int (*begin_checkpoint_userdata)(CACHEFILE, LSN, void*), int (*end_checkpoint_userdata)(CACHEFILE, void*));
 // Effect: Store some cachefile-specific user data.  When the last reference to a cachefile is closed, we call close_userdata().
 // Before starting a checkpoint, we call checkpoint_prepare_userdata().
 // When the cachefile needs to be checkpointed, we call checkpoint_userdata().
@@ -167,8 +167,9 @@ int toku_cachetable_rename (CACHEFILE cachefile, CACHEKEY oldkey, CACHEKEY newke
 // the cachetable.  The flush callback is called for each of these objects.  The
 // close function does not return until all of the objects are evicted.  The cachefile
 // object is freed.
+// If oplsn_valid is TRUE then use oplsn as the LSN of the close instead of asking the logger.  oplsn_valid being TRUE is only allowed during recovery, and requires that you are removing the last reference (otherwise the lsn wouldn't make it in.)
 // Returns: 0 if success, otherwise returns an error number.
-int toku_cachefile_close (CACHEFILE*, TOKULOGGER, char **error_string, LSN);
+int toku_cachefile_close (CACHEFILE*, TOKULOGGER, char **error_string, BOOL oplsn_valid, LSN oplsn);
 
 // Flush the cachefile.
 // Effect: Flush everything owned by the cachefile from the cachetable. All dirty

newbrt/recover.c

@@ -76,6 +76,7 @@ static void file_map_close_dictionaries(struct file_map *fmap) {
         assert(r == 0);
         struct file_map_tuple *tuple = v;
 	assert(tuple->brt);
+        //Logging is already back on.  No need to pass LSN into close.
         r = toku_close_brt(tuple->brt, 0, 0);
         assert(r == 0);
         file_map_tuple_destroy(tuple);
@@ -304,6 +305,9 @@ static int toku_recover_backward_fopen (struct logtype_fopen *l, RECOVER_ENV ren
         struct file_map_tuple *tuple = NULL;
         int r = file_map_find(&renv->fmap, l->filenum, &tuple);
         if (r == 0) {
+            //Must not write header to disk.  If not dirty, it will not be written.
+            //Since we're not writing to disk, we do not need to provide an LSN.
+            assert(!tuple->brt->h->dirty);
             r = toku_close_brt(tuple->brt, 0, 0);
             assert(r == 0);
             file_map_remove(&renv->fmap, l->filenum);
@@ -412,14 +416,14 @@ static int toku_recover_backward_enq_delete_any (struct logtype_enq_delete_any *
     return 0;
 }
 
-static void toku_recover_fclose (LSN UU(lsn), BYTESTRING fname, FILENUM filenum, RECOVER_ENV UU(renv)) {
+static void toku_recover_fclose (LSN lsn, BYTESTRING fname, FILENUM filenum, RECOVER_ENV UU(renv)) {
     struct file_map_tuple *tuple = NULL;
     int r = file_map_find(&renv->fmap, filenum, &tuple);
     if (r == 0) {
         char *fixedfname = fixup_fname(&fname);
         assert(strcmp(tuple->fname, fixedfname) == 0);
         toku_free(fixedfname);
-        r = toku_close_brt(tuple->brt, 0, 0);
+        r = toku_close_brt_lsn(tuple->brt, 0, 0, TRUE, lsn);
         assert(r == 0);
         file_map_remove(&renv->fmap, filenum);
     }
@@ -853,6 +857,8 @@ int tokudb_recover(const char *data_dir, const char *log_dir, brt_compare_func b
         assert(r == 0);
 
         rr = do_recovery(&renv, data_dir, log_dir);
+        assert(rr==0);  //Recovery must succeed.
+                        //To support failure, you need to be careful not to write dictionaries to disk, like recover_env_cleanup does.
 
         recover_env_cleanup(&renv);
     }

newbrt/roll.c

@@ -98,8 +98,15 @@ static int do_nothing_with_filenum(TOKUTXN txn, FILENUM filenum) {
 
 
 int toku_commit_cmdinsert (FILENUM filenum, BYTESTRING key, TOKUTXN txn, YIELDF UU(yield), void *UU(yieldv), LSN oplsn) {
+    static TXNID last_xid;
+    static FILENUM last_filenum;
+    if (txn->txnid64 == last_xid && last_filenum.fileid == filenum.fileid)
+        return 0;
+    last_xid = txn->txnid64;
+    last_filenum = filenum;
+
 #if TOKU_DO_COMMIT_CMD_INSERT
-    return do_insertion (BRT_COMMIT_ANY, filenum, key, 0, txn, oplsn);
+    return do_insertion (BRT_COMMIT_BROADCAST_TXN, filenum, key, 0, txn, oplsn);
 #else
     key = key; oplsn = oplsn;
     return do_nothing_with_filenum(txn, filenum);
@@ -115,8 +122,15 @@ toku_commit_cmdinsertboth (FILENUM    filenum,
 			   void *     UU(yieldv),
                            LSN        oplsn)
 {
+    static TXNID last_xid;
+    static FILENUM last_filenum;
+    if (txn->txnid64 == last_xid && last_filenum.fileid == filenum.fileid)
+        return 0;
+    last_xid = txn->txnid64;
+    last_filenum = filenum;
+
 #if TOKU_DO_COMMIT_CMD_INSERT
-    return do_insertion (BRT_COMMIT_BOTH, filenum, key, &data, txn, oplsn);
+    return do_insertion (BRT_COMMIT_BROADCAST_TXN, filenum, key, &data, txn, oplsn);
 #else
     key = key; data = data; oplsn = oplsn;
     return do_nothing_with_filenum(txn, filenum);
@@ -131,7 +145,14 @@ toku_rollback_cmdinsert (FILENUM    filenum,
 			 void *     UU(yieldv),
                          LSN        oplsn)
 {
-    return do_insertion (BRT_ABORT_ANY, filenum, key, 0, txn, oplsn);
+    static TXNID last_xid;
+    static FILENUM last_filenum;
+    if (txn->txnid64 == last_xid && last_filenum.fileid == filenum.fileid)
+        return 0;
+    last_xid = txn->txnid64;
+    last_filenum = filenum;
+
+    return do_insertion (BRT_ABORT_BROADCAST_TXN, filenum, key, 0, txn, oplsn);
 }
 
 int
@@ -143,7 +164,14 @@ toku_rollback_cmdinsertboth (FILENUM    filenum,
 			     void *     UU(yieldv),
                              LSN        oplsn)
 {
-    return do_insertion (BRT_ABORT_BOTH, filenum, key, &data, txn, oplsn);
+    static TXNID last_xid;
+    static FILENUM last_filenum;
+    if (txn->txnid64 == last_xid && last_filenum.fileid == filenum.fileid)
+        return 0;
+    last_xid = txn->txnid64;
+    last_filenum = filenum;
+
+    return do_insertion (BRT_ABORT_BROADCAST_TXN, filenum, key, &data, txn, oplsn);
 }
 
 int
@@ -155,8 +183,15 @@ toku_commit_cmddeleteboth (FILENUM    filenum,
 			   void *     UU(yieldv),
                            LSN        oplsn)
 {
+    static TXNID last_xid;
+    static FILENUM last_filenum;
+    if (txn->txnid64 == last_xid && last_filenum.fileid == filenum.fileid)
+        return 0;
+    last_xid = txn->txnid64;
+    last_filenum = filenum;
+
 #if TOKU_DO_COMMIT_CMD_DELETE_BOTH
-    return do_insertion (BRT_COMMIT_BOTH, filenum, key, &data, txn, oplsn);
+    return do_insertion (BRT_COMMIT_BROADCAST_TXN, filenum, key, &data, txn, oplsn);
 #else
     xid = xid; key = key; data = data;
     return do_nothing_with_filenum(txn, filenum);
@@ -172,7 +207,14 @@ toku_rollback_cmddeleteboth (FILENUM    filenum,
 			     void *     UU(yieldv),
                              LSN        oplsn)
 {
-    return do_insertion (BRT_ABORT_BOTH, filenum, key, &data, txn, oplsn);
+    static TXNID last_xid;
+    static FILENUM last_filenum;
+    if (txn->txnid64 == last_xid && last_filenum.fileid == filenum.fileid)
+        return 0;
+    last_xid = txn->txnid64;
+    last_filenum = filenum;
+
+    return do_insertion (BRT_ABORT_BROADCAST_TXN, filenum, key, &data, txn, oplsn);
 }
 
 int
@@ -183,8 +225,15 @@ toku_commit_cmddelete (FILENUM    filenum,
 		       void *     UU(yieldv),
                        LSN        oplsn)
 {
+    static TXNID last_xid;
+    static FILENUM last_filenum;
+    if (txn->txnid64 == last_xid && last_filenum.fileid == filenum.fileid)
+        return 0;
+    last_xid = txn->txnid64;
+    last_filenum = filenum;
+
 #if TOKU_DO_COMMIT_CMD_DELETE
-    return do_insertion (BRT_COMMIT_ANY, filenum, key, 0, txn, oplsn);
+    return do_insertion (BRT_COMMIT_BROADCAST_TXN, filenum, key, 0, txn, oplsn);
 #else
     xid = xid; key = key;
     return do_nothing_with_filenum(txn, filenum);
@@ -199,7 +248,14 @@ toku_rollback_cmddelete (FILENUM    filenum,
 			 void *     UU(yieldv),
                          LSN        oplsn)
 {
-    return do_insertion (BRT_ABORT_ANY, filenum, key, 0, txn, oplsn);
+    static TXNID last_xid;
+    static FILENUM last_filenum;
+    if (txn->txnid64 == last_xid && last_filenum.fileid == filenum.fileid)
+        return 0;
+    last_xid = txn->txnid64;
+    last_filenum = filenum;
+
+    return do_insertion (BRT_ABORT_BROADCAST_TXN, filenum, key, 0, txn, oplsn);
 }
 
 int

newbrt/tests/cachetable-checkpoint-pending.c

@@ -147,7 +147,7 @@ static void checkpoint_pending(void) {
     assert(r == 0);
     assert(n_flush == 0 && n_write_me == 0 && n_keep_me == 0);
 
-    r = toku_cachefile_close(&cf, NULL_LOGGER, 0, ZERO_LSN); assert(r == 0 && cf == 0);
+    r = toku_cachefile_close(&cf, NULL_LOGGER, 0, FALSE, ZERO_LSN); assert(r == 0 && cf == 0);
     r = toku_cachetable_close(&ct); assert(r == 0 && ct == 0);
 }
 

newbrt/tests/cachetable-checkpoint-test.c

@@ -120,7 +120,7 @@ static void cachetable_checkpoint_test(int n, enum cachetable_dirty dirty) {
     assert(r == 0);
     assert(n_flush == 0 && n_write_me == 0 && n_keep_me == 0);
 
-    r = toku_cachefile_close(&f1, NULL_LOGGER, 0, ZERO_LSN); assert(r == 0 && f1 == 0);
+    r = toku_cachefile_close(&f1, NULL_LOGGER, 0, FALSE, ZERO_LSN); assert(r == 0 && f1 == 0);
     r = toku_cachetable_close(&ct); assert(r == 0 && ct == 0);
 }
 

newbrt/tests/cachetable-count-pinned-test.c

@@ -65,7 +65,7 @@ cachetable_count_pinned_test (int n) {
     assert(toku_cachefile_count_pinned(f1, 1) == 0);
     toku_cachetable_verify(ct);
 
-    r = toku_cachefile_close(&f1, NULL_LOGGER, 0, ZERO_LSN); assert(r == 0 && f1 == 0);
+    r = toku_cachefile_close(&f1, NULL_LOGGER, 0, FALSE, ZERO_LSN); assert(r == 0 && f1 == 0);
     r = toku_cachetable_close(&ct); assert(r == 0 && ct == 0);
 }
 

newbrt/tests/cachetable-debug-test.c

@@ -73,7 +73,7 @@ cachetable_debug_test (int n) {
 
     if (verbose) toku_cachetable_print_hash_histogram();
 
-    r = toku_cachefile_close(&f1, NULL_LOGGER, 0, ZERO_LSN); assert(r == 0 && f1 == 0);
+    r = toku_cachefile_close(&f1, NULL_LOGGER, 0, FALSE, ZERO_LSN); assert(r == 0 && f1 == 0);
     r = toku_cachetable_close(&ct); assert(r == 0 && ct == 0);
 }
 

newbrt/tests/cachetable-fd-test.c

@@ -39,7 +39,7 @@ cachetable_fd_test (void) {
     r = toku_cachefile_of_filenum(ct, fn, &newcf);
     assert(r == ENOENT);
 
-    r = toku_cachefile_close(&cf, NULL_LOGGER, 0, ZERO_LSN); assert(r == 0 && cf == 0);
+    r = toku_cachefile_close(&cf, NULL_LOGGER, 0, FALSE, ZERO_LSN); assert(r == 0 && cf == 0);
     r = toku_cachetable_close(&ct); assert(r == 0 && ct == 0);
 }
 

newbrt/tests/cachetable-flush-test.c

@@ -92,8 +92,8 @@ test_cachetable_flush (int n) {
         assert(r == 0);
     }
 
-    r = toku_cachefile_close(&f1, NULL_LOGGER, 0, ZERO_LSN); assert(r == 0 && f1 == 0);
-    r = toku_cachefile_close(&f2, NULL_LOGGER, 0, ZERO_LSN); assert(r == 0 && f2 == 0);
+    r = toku_cachefile_close(&f1, NULL_LOGGER, 0, FALSE, ZERO_LSN); assert(r == 0 && f1 == 0);
+    r = toku_cachefile_close(&f2, NULL_LOGGER, 0, FALSE, ZERO_LSN); assert(r == 0 && f2 == 0);
     r = toku_cachetable_close(&ct); assert(r == 0 && ct == 0);
 }
 

newbrt/tests/cachetable-getandpin-test.c

@@ -70,7 +70,7 @@ cachetable_getandpin_test (int n) {
     }
     toku_cachetable_verify(ct);
 
-    r = toku_cachefile_close(&f1, NULL_LOGGER, 0, ZERO_LSN); assert(r == 0 && f1 == 0);
+    r = toku_cachefile_close(&f1, NULL_LOGGER, 0, FALSE, ZERO_LSN); assert(r == 0 && f1 == 0);
     r = toku_cachetable_close(&ct); assert(r == 0 && ct == 0);
 }
 

newbrt/tests/cachetable-prefetch-checkpoint-test.c

@@ -112,7 +112,7 @@ static void cachetable_prefetch_checkpoint_test(int n, enum cachetable_dirty dir
     assert(r == 0);
     assert(n_flush == 0 && n_write_me == 0 && n_keep_me == 0);
 
-    r = toku_cachefile_close(&f1, NULL_LOGGER, 0, ZERO_LSN); assert(r == 0 && f1 == 0);
+    r = toku_cachefile_close(&f1, NULL_LOGGER, 0, FALSE, ZERO_LSN); assert(r == 0 && f1 == 0);
     r = toku_cachetable_close(&ct); assert(r == 0 && ct == 0);
 }
 

newbrt/tests/cachetable-prefetch-close-fail-test.c

@@ -55,7 +55,7 @@ static void cachetable_prefetch_maybegetandpin_test (void) {
 
     // close with the prefetch in progress. the close should block until
     // all of the reads and writes are complete.
-    r = toku_cachefile_close(&f1, NULL_LOGGER, 0, ZERO_LSN); assert(r == 0 && f1 == 0);
+    r = toku_cachefile_close(&f1, NULL_LOGGER, 0, FALSE, ZERO_LSN); assert(r == 0 && f1 == 0);
     r = toku_cachetable_close(&ct); assert(r == 0 && ct == 0);
 }
 

newbrt/tests/cachetable-prefetch-close-leak-test.c

@@ -56,7 +56,7 @@ static void cachetable_prefetch_close_leak_test (void) {
 
     // close with the prefetch in progress. the close should block until
     // all of the reads and writes are complete.
-    r = toku_cachefile_close(&f1, NULL_LOGGER, 0, ZERO_LSN); assert(r == 0 && f1 == 0);
+    r = toku_cachefile_close(&f1, NULL_LOGGER, 0, FALSE, ZERO_LSN); assert(r == 0 && f1 == 0);
     r = toku_cachetable_close(&ct); assert(r == 0 && ct == 0);
 }
 

newbrt/tests/cachetable-prefetch-close-test.c

@@ -55,7 +55,7 @@ static void cachetable_prefetch_maybegetandpin_test (void) {
 
     // close with the prefetch in progress. the close should block until
     // all of the reads and writes are complete.
-    r = toku_cachefile_close(&f1, NULL_LOGGER, 0, ZERO_LSN); assert(r == 0 && f1 == 0);
+    r = toku_cachefile_close(&f1, NULL_LOGGER, 0, FALSE, ZERO_LSN); assert(r == 0 && f1 == 0);
     r = toku_cachetable_close(&ct); assert(r == 0 && ct == 0);
 }
 

newbrt/tests/cachetable-prefetch-getandpin-fail-test.c

@@ -72,7 +72,7 @@ static void cachetable_prefetch_maybegetandpin_test (void) {
     
     toku_cachetable_verify(ct);
 
-    r = toku_cachefile_close(&f1, NULL_LOGGER, 0, ZERO_LSN); assert(r == 0 && f1 == 0);
+    r = toku_cachefile_close(&f1, NULL_LOGGER, 0, FALSE, ZERO_LSN); assert(r == 0 && f1 == 0);
     r = toku_cachetable_close(&ct); assert(r == 0 && ct == 0);
 }
 

newbrt/tests/cachetable-prefetch-getandpin-test.c

@@ -76,7 +76,7 @@ static void cachetable_prefetch_maybegetandpin_test (void) {
     assert(r == 0);
     toku_cachetable_verify(ct);
 
-    r = toku_cachefile_close(&f1, NULL_LOGGER, 0, ZERO_LSN); assert(r == 0 && f1 == 0);
+    r = toku_cachefile_close(&f1, NULL_LOGGER, 0, FALSE, ZERO_LSN); assert(r == 0 && f1 == 0);
     r = toku_cachetable_close(&ct); assert(r == 0 && ct == 0);
 }
 

newbrt/tests/cachetable-prefetch-maybegetandpin-test.c

@@ -66,7 +66,7 @@ static void cachetable_prefetch_maybegetandpin_test (void) {
     assert(r == 0);
     toku_cachetable_verify(ct);
 
-    r = toku_cachefile_close(&f1, NULL_LOGGER, 0, ZERO_LSN); assert(r == 0 && f1 == 0);
+    r = toku_cachefile_close(&f1, NULL_LOGGER, 0, FALSE, ZERO_LSN); assert(r == 0 && f1 == 0);
     r = toku_cachetable_close(&ct); assert(r == 0 && ct == 0);
 }
 

newbrt/tests/cachetable-prefetch2-test.c

@@ -77,7 +77,7 @@ static void cachetable_prefetch_maybegetandpin_test (void) {
     assert(r == 0);
     toku_cachetable_verify(ct);
 
-    r = toku_cachefile_close(&f1, NULL_LOGGER, 0, ZERO_LSN); assert(r == 0 && f1 == 0);
+    r = toku_cachefile_close(&f1, NULL_LOGGER, 0, FALSE, ZERO_LSN); assert(r == 0 && f1 == 0);
     r = toku_cachetable_close(&ct); assert(r == 0 && ct == 0);
 }
 

newbrt/tests/cachetable-put-test.c

@@ -77,7 +77,7 @@ cachetable_put_test (int n) {
     r = toku_cachetable_unpin(f1, k, toku_cachetable_hash(f1, k), CACHETABLE_CLEAN, 1);
     assert(r != 0);
 
-    r = toku_cachefile_close(&f1, NULL_LOGGER, 0, ZERO_LSN); assert(r == 0 && f1 == 0);
+    r = toku_cachefile_close(&f1, NULL_LOGGER, 0, FALSE, ZERO_LSN); assert(r == 0 && f1 == 0);
     r = toku_cachetable_close(&ct); assert(r == 0 && ct == 0);
 }
 

newbrt/tests/cachetable-rename-test.c

@@ -158,7 +158,7 @@ static void test_rename (void) {
     r = toku_cachetable_rename(f, okey, nkey);
     assert(r != 0);
 
-    r = toku_cachefile_close(&f, 0, 0, ZERO_LSN);
+    r = toku_cachefile_close(&f, 0, 0, FALSE, ZERO_LSN);
     assert(r == 0);
     r = toku_cachetable_close(&t);
     assert(r == 0);

newbrt/tests/cachetable-scan.c

@@ -84,7 +84,7 @@ static void readit (void) {
 	r=toku_cachetable_get_and_pin(f, key, fullhash, &block, &current_size, f_flush, f_fetch, 0); assert(r==0);
 	r=toku_cachetable_unpin(f, key, fullhash, CACHETABLE_CLEAN, BLOCKSIZE);                                      assert(r==0);
     }
-    r = toku_cachefile_close(&f, 0, 0, ZERO_LSN);    assert(r == 0);
+    r = toku_cachefile_close(&f, 0, 0, FALSE, ZERO_LSN);    assert(r == 0);
     r = toku_cachetable_close(&t);      assert(r == 0);
     gettimeofday(&end, 0);
     toku_os_get_process_times(&end_usertime, &end_systime);

newbrt/tests/cachetable-test.c

@@ -287,7 +287,7 @@ static void test0 (void) {
     expectN(7);
     expectN(6);
     expectN(1);
-    r=toku_cachefile_close(&f, 0, 0, ZERO_LSN);
+    r=toku_cachefile_close(&f, 0, 0, FALSE, ZERO_LSN);
     assert(r==0);
     r=toku_cachetable_close(&t);
     assert(r==0);
@@ -352,7 +352,7 @@ static void test_nested_pin (void) {
     r = toku_cachetable_unpin(f, make_blocknum(2), f2hash, CACHETABLE_CLEAN, test_object_size);
     assert(r==0);
     //    toku_os_usleep(1*1000000);
-    r = toku_cachefile_close(&f, 0, 0, ZERO_LSN); assert(r==0);
+    r = toku_cachefile_close(&f, 0, 0, FALSE, ZERO_LSN); assert(r==0);
     r = toku_cachetable_close(&t); assert(r==0);
 }
 
@@ -415,14 +415,14 @@ static void test_multi_filehandles (void) {
 
     r = toku_cachetable_unpin(f1, make_blocknum(1), toku_cachetable_hash(f1, make_blocknum(1)), CACHETABLE_CLEAN, 0); assert(r==0);
     r = toku_cachetable_unpin(f1, make_blocknum(2), toku_cachetable_hash(f1, make_blocknum(2)), CACHETABLE_CLEAN, 0); assert(r==0);
-    r = toku_cachefile_close(&f1, 0, 0, ZERO_LSN); assert(r==0);
+    r = toku_cachefile_close(&f1, 0, 0, FALSE, ZERO_LSN); assert(r==0);
 
     r = toku_cachetable_unpin(f2, make_blocknum(1), toku_cachetable_hash(f2, make_blocknum(1)), CACHETABLE_CLEAN, 0); assert(r==0);
     r = toku_cachetable_unpin(f2, make_blocknum(2), toku_cachetable_hash(f2, make_blocknum(2)), CACHETABLE_CLEAN, 0); assert(r==0);
-    r = toku_cachefile_close(&f2, 0, 0, ZERO_LSN); assert(r==0);
+    r = toku_cachefile_close(&f2, 0, 0, FALSE, ZERO_LSN); assert(r==0);
 
     r = toku_cachetable_unpin(f3, make_blocknum(2), toku_cachetable_hash(f3, make_blocknum(2)), CACHETABLE_CLEAN, 0); assert(r==0);
-    r = toku_cachefile_close(&f3, 0, 0, ZERO_LSN); assert(r==0);
+    r = toku_cachefile_close(&f3, 0, 0, FALSE, ZERO_LSN); assert(r==0);
 
     r = toku_cachetable_close(&t); assert(r==0);
 }
@@ -543,7 +543,7 @@ static void test_dirty(void) {
     assert(dirty == 1);
     assert(pinned == 0);
      
-    r = toku_cachefile_close(&f, 0, 0, ZERO_LSN);
+    r = toku_cachefile_close(&f, 0, 0, FALSE, ZERO_LSN);
     assert(r == 0);
     r = toku_cachetable_close(&t);
     assert(r == 0);
@@ -620,7 +620,7 @@ static void test_size_resize(void) {
     r = toku_cachetable_unpin(f, key, hkey, CACHETABLE_CLEAN, new_size);
     assert(r == 0);
 
-    r = toku_cachefile_close(&f, 0, 0, ZERO_LSN);
+    r = toku_cachefile_close(&f, 0, 0, FALSE, ZERO_LSN);
     assert(r == 0);
     r = toku_cachetable_close(&t);
     assert(r == 0);
@@ -688,7 +688,7 @@ static void test_size_flush(void) {
         assert(r == 0);
     }
     
-    r = toku_cachefile_close(&f, 0, 0, ZERO_LSN);
+    r = toku_cachefile_close(&f, 0, 0, FALSE, ZERO_LSN);
     assert(r == 0);
     r = toku_cachetable_close(&t);
     assert(r == 0);

newbrt/tests/cachetable-test2.c

@@ -226,13 +226,13 @@ static void test_chaining (void) {
 	    //printf("Close %d (%p), now n_present=%d\n", i, f[i], n_present);
 	    //print_ints();
 	    CACHEFILE oldcf=f[i];
-	    r = toku_cachefile_close(&f[i], 0, 0, ZERO_LSN);                           assert(r==0);
+	    r = toku_cachefile_close(&f[i], 0, 0, FALSE, ZERO_LSN);                           assert(r==0);
 	    file_is_not_present(oldcf);
 	    r = toku_cachetable_openf(&f[i], ct, fname[i], fname[i], O_RDWR, S_IRWXU|S_IRWXG|S_IRWXO); assert(r==0);
 	}
     }
     for (i=0; i<N_FILES; i++) {
-	r = toku_cachefile_close(&f[i], 0, 0, ZERO_LSN); assert(r==0);
+	r = toku_cachefile_close(&f[i], 0, 0, FALSE, ZERO_LSN); assert(r==0);
     }
     r = toku_cachetable_close(&ct); assert(r==0);
 }

newbrt/tests/cachetable-unpin-test.c

@@ -67,7 +67,7 @@ cachetable_unpin_test (int n) {
     r = toku_cachetable_unpin(f1, k, toku_cachetable_hash(f1, k), CACHETABLE_CLEAN, 1);
     assert(r != 0);
 
-    r = toku_cachefile_close(&f1, NULL_LOGGER, 0, ZERO_LSN); assert(r == 0 && f1 == 0);
+    r = toku_cachefile_close(&f1, NULL_LOGGER, 0, FALSE, ZERO_LSN); assert(r == 0 && f1 == 0);
     r = toku_cachetable_close(&ct); assert(r == 0 && ct == 0);
 }
 

newbrt/ule.h

@@ -10,7 +10,7 @@
 #define ULE_H
 
 //1 does much slower debugging
-#define ULE_DEBUG 0
+#define ULE_DEBUG 1
 
 /////////////////////////////////////////////////////////////////////////////////
 // Following data structures are the unpacked format of a leafentry.