Addresses #1691 Txn that created or locked when empty is now per dictionary...

Addresses #1691 Txn that created or locked when empty is now per dictionary (in header instead of brt)
We clear it when the transaction closes (this means we cannot change it when it is already set till that transaction finishes).
This fixes a crash in windows in root_fifo_2.tdb

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

newbrt/brt-internal.h

@@ -176,6 +176,10 @@ struct brt_header {
     u_int64_t root_put_counter; // the generation number of the brt
 
     BLOCK_TABLE blocktable;
+    // If a transaction created this BRT, which one?
+    // If a transaction locked the BRT when it was empty, which transaction?  (Only the latest one matters)
+    // 0 if no such transaction
+    TXNID txnid_that_created_or_locked_when_empty;
 };
 
 struct brt {
@@ -198,10 +202,6 @@ struct brt {
 
     OMT txns; // transactions that are using this OMT (note that the transaction checks the cf also)
 
-    // If a transaction created this BRT, which one?
-    // If a transaction locked the BRT when it was empty, which transaction?  (Only the latest one matters)
-    // 0 if no such transaction
-    TXNID txnid_that_created_or_locked_when_empty;
     int was_closed; //True when this brt was closed, but is being kept around for transactions.
     int (*close_db)(DB*, u_int32_t);
     u_int32_t close_flags;

newbrt/brt.c

@@ -2652,7 +2652,7 @@ int toku_brt_insert (BRT brt, DBT *key, DBT *val, TOKUTXN txn)
 // Effect: Insert the key-val pair into brt.
 {
     int r;
-    if (txn && (brt->txnid_that_created_or_locked_when_empty != toku_txn_get_txnid(txn))) {
+    if (txn && (brt->h->txnid_that_created_or_locked_when_empty != toku_txn_get_txnid(txn))) {
         toku_cachefile_refup(brt->cf);
         BYTESTRING keybs  = {key->size, toku_memdup_in_rollback(txn, key->data, key->size)};
         int need_data = (brt->flags&TOKU_DB_DUPSORT)!=0; // dupsorts don't need the data part
@@ -2674,7 +2674,7 @@ int toku_brt_insert (BRT brt, DBT *key, DBT *val, TOKUTXN txn)
 
 int toku_brt_delete(BRT brt, DBT *key, TOKUTXN txn) {
     int r;
-    if (txn && (brt->txnid_that_created_or_locked_when_empty != toku_txn_get_txnid(txn))) {
+    if (txn && (brt->h->txnid_that_created_or_locked_when_empty != toku_txn_get_txnid(txn))) {
         BYTESTRING keybs  = {key->size, toku_memdup_in_rollback(txn, key->data, key->size)};
         toku_cachefile_refup(brt->cf);
         r = toku_logger_save_rollback_cmddelete(txn, toku_txn_get_txnid(txn), toku_cachefile_filenum(brt->cf), keybs);
@@ -2924,6 +2924,7 @@ int toku_read_brt_header_and_store_in_cachefile (CACHEFILE cf, struct brt_header
 int toku_brt_open(BRT t, const char *fname, const char *fname_in_env, int is_create, int only_create, CACHETABLE cachetable, TOKUTXN txn, DB *db) {
 
     int r;
+    BOOL txn_created = FALSE;
 
     //printf("%s:%d %d alloced\n", __FILE__, __LINE__, get_n_items_malloced()); toku_print_malloced_items();
     WHEN_BRTTRACE(fprintf(stderr, "BRTTRACE: %s:%d toku_brt_open(%s, \"%s\", %d, %p, %d, %p)\n",
@@ -2938,7 +2939,6 @@ int toku_brt_open(BRT t, const char *fname, const char *fname_in_env, int is_cre
         goto died0;
     }
     t->db = db;
-    t->txnid_that_created_or_locked_when_empty = 0; // Uses 0 for no transaction.
     {
         int fd = -1;
         BOOL did_create = FALSE;
@@ -2950,7 +2950,7 @@ int toku_brt_open(BRT t, const char *fname, const char *fname_in_env, int is_cre
             mode_t mode = S_IRWXU|S_IRWXG|S_IRWXO;
             r = toku_logger_log_fcreate(txn, fname_in_env, toku_cachefile_filenum(t->cf), mode);
             if (r != 0) goto died_after_open;
-            t->txnid_that_created_or_locked_when_empty = toku_txn_get_txnid(txn);
+            txn_created = (txn!=NULL);
         } else {
 	    r = toku_logger_log_fopen(txn, fname_in_env, toku_cachefile_filenum(t->cf));
 	}
@@ -3018,6 +3018,13 @@ int toku_brt_open(BRT t, const char *fname, const char *fname_in_env, int is_cre
         t->did_set_descriptor = 0;
     }
     if (t->db) t->db->descriptor = &t->h->descriptor.dbt;
+    if (txn_created) {
+        assert(txn);
+        assert(t->h->txnid_that_created_or_locked_when_empty == 0); // Uses 0 for no transaction.
+        t->h->txnid_that_created_or_locked_when_empty = toku_txn_get_txnid(txn);
+        r = toku_txn_note_brt(txn, t);
+        assert(r==0);
+    } 
 
     //Opening a brt may restore to previous checkpoint.  Truncate if necessary.
     toku_maybe_truncate_cachefile_on_open(t->h->blocktable, t->h);
@@ -4411,7 +4418,7 @@ toku_brt_lookup (BRT brt, DBT *k, DBT *v, BRT_GET_CALLBACK_FUNCTION getf, void *
 int toku_brt_delete_both(BRT brt, DBT *key, DBT *val, TOKUTXN txn) {
     //{ unsigned i; printf("del %p keylen=%d key={", brt->db, key->size); for(i=0; i<key->size; i++) printf("%d,", ((char*)key->data)[i]); printf("} datalen=%d data={", val->size); for(i=0; i<val->size; i++) printf("%d,", ((char*)val->data)[i]); printf("}\n"); }
     int r;
-    if (txn && (brt->txnid_that_created_or_locked_when_empty != toku_txn_get_txnid(txn))) {
+    if (txn && (brt->h->txnid_that_created_or_locked_when_empty != toku_txn_get_txnid(txn))) {
         BYTESTRING keybs  = {key->size, toku_memdup_in_rollback(txn, key->data, key->size)};
         BYTESTRING databs = {val->size, toku_memdup_in_rollback(txn, val->data, val->size)};
         toku_cachefile_refup(brt->cf);
@@ -4735,7 +4742,8 @@ int
 toku_brt_note_table_lock (BRT brt, TOKUTXN txn)
 {
     if (brt_is_empty(brt, toku_txn_logger(txn))) {
-        brt->txnid_that_created_or_locked_when_empty = toku_txn_get_txnid(txn);
+        assert(brt->h->txnid_that_created_or_locked_when_empty == 0); 
+        brt->h->txnid_that_created_or_locked_when_empty = toku_txn_get_txnid(txn);
         toku_cachefile_refup(brt->cf);
         return toku_logger_save_rollback_tablelock_on_empty_table(txn, toku_cachefile_filenum(brt->cf));
     }

newbrt/log.c

@@ -1176,6 +1176,9 @@ static int remove_txn (OMTVALUE brtv, u_int32_t UU(idx), void *txnv) {
         assert(brt->close_db);
         r = brt->close_db(brt->db, brt->close_flags);
     }
+    if (txn->txnid64==brt->h->txnid_that_created_or_locked_when_empty) {
+        brt->h->txnid_that_created_or_locked_when_empty = 0;
+    }
     return r;
 }