#4582 maintain lock tree after it is closed closes[t:4582]

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

src/lock_tree/locktree.c

@@ -1332,8 +1332,10 @@ toku_lt_create(toku_lock_tree** ptree,
                        toku_ltm_incr_lock_memory, toku_ltm_decr_lock_memory, mgr);
     if (r != 0)
         goto cleanup;
-
     r = toku_rth_create(&tmp_tree->rth);
+    if (r != 0)
+        goto cleanup;
+    r = toku_rth_create(&tmp_tree->txns_to_unlock);
     if (r != 0)
         goto cleanup;
 
@@ -1362,6 +1364,8 @@ cleanup:
                 toku_rt_close(tmp_tree->borderwrite);
             if (tmp_tree->rth)
                 toku_rth_close(tmp_tree->rth);
+            if (tmp_tree->txns_to_unlock)
+                toku_rth_close(tmp_tree->txns_to_unlock);
             if (tmp_tree->buf)
                 toku_free(tmp_tree->buf);
             if (tmp_tree->bw_buf)
@@ -1394,6 +1398,7 @@ toku_lt_set_dict_id(toku_lock_tree* lt, DICTIONARY_ID dict_id) {
 
 static void lt_add_db(toku_lock_tree* tree, DB *db);
 static void lt_remove_db(toku_lock_tree* tree, DB *db);
+static void lt_unlock_deferred_txns(toku_lock_tree *tree);
 
 int 
 toku_ltm_get_lt(toku_ltm* mgr, toku_lock_tree** ptree, DICTIONARY_ID dict_id, DB *db, toku_dbt_cmp compare_fun) {
@@ -1414,6 +1419,7 @@ toku_ltm_get_lt(toku_ltm* mgr, toku_lock_tree** ptree, DICTIONARY_ID dict_id, DB
         assert (tree != NULL);
         toku_lt_add_ref(tree);
         lt_add_db(tree, db);
+        lt_unlock_deferred_txns(tree);
         *ptree = tree;
         r = 0;
         goto cleanup;
@@ -1484,17 +1490,23 @@ toku_lt_close(toku_lock_tree* tree) {
     if (!first_error && r != 0)
         first_error = r;
 
+    uint32_t ranges = 0;
     toku_rth_start_scan(tree->rth);
     rt_forest* forest;
-    
     while ((forest = toku_rth_next(tree->rth)) != NULL) {
+        if (forest->self_read)
+            ranges += toku_rt_get_size(forest->self_read);
         r = lt_free_contents(tree, forest->self_read);
         if (!first_error && r != 0)
             first_error = r;
+        if (forest->self_write)
+            ranges += toku_rt_get_size(forest->self_write);
         r = lt_free_contents(tree, forest->self_write);
         if (!first_error && r != 0) 
             first_error = r;
     }
+    ltm_decr_locks(tree->mgr, ranges);
+    toku_rth_close(tree->txns_to_unlock);
     toku_rth_close(tree->rth);
     toku_omt_destroy(&tree->dbs);
     toku_mutex_destroy(&tree->mutex);
@@ -2112,34 +2124,26 @@ lt_unlock_txn(toku_lock_tree* tree, TXNID txn) {
     uint32_t ranges = 0;
 
     if (selfread) {
-        uint32_t size = toku_rt_get_size(selfread);
-        ranges += size;
+        ranges += toku_rt_get_size(selfread);
         r = lt_free_contents(tree, selfread);
         if (r != 0) 
             return lt_panic(tree, r);
     }
 
     if (selfwrite) {
-        uint32_t size = toku_rt_get_size(selfwrite);
-        ranges += size;
+        ranges += toku_rt_get_size(selfwrite);
 
-        // get a db from the db's associated with the lock tree and use it to update the borderwrite 
-        // if there are no db's, then assume that the db was closed before all transactions that referenced the
-        // lock tree were retired.  in this case, there is no need to update the border write since 
-        // these transactions may no longer do anything to the db since it is closed, and
-        // we expect to just close the lock tree when all of the open references to it are retired.
-        if (toku_omt_size(tree->dbs) > 0) {
+        assert(toku_omt_size(tree->dbs) > 0);
         OMTVALUE dbv;
         r = toku_omt_fetch(tree->dbs, 0, &dbv);
         assert_zero(r);
         DB *db = dbv;
-
         lt_set_comparison_functions(tree, db);
         r = lt_border_delete(tree, selfwrite);
         lt_clear_comparison_functions(tree);
         if (r != 0) 
             return lt_panic(tree, r);
-        }
+
         r = lt_free_contents(tree, selfwrite);
         if (r != 0) 
             return lt_panic(tree, r);
@@ -2166,13 +2170,29 @@ toku_lt_unlock_txn(toku_lock_tree* tree, TXNID txn) {
         r = EINVAL; goto cleanup;
     }
     toku_mutex_lock(&tree->mutex);
+    if (toku_omt_size(tree->dbs) > 0) {
         lt_unlock_txn(tree, txn);
         lt_retry_lock_requests(tree);
+    } else {
+        r = toku_rth_insert(tree->txns_to_unlock, txn);
+    }
     toku_mutex_unlock(&tree->mutex);
 cleanup:
     return r;
 }
 
+static void
+lt_unlock_deferred_txns(toku_lock_tree *tree) {
+    toku_rth_start_scan(tree->txns_to_unlock);
+    rt_forest *forest;
+    while ((forest = toku_rth_next(tree->txns_to_unlock)) != NULL) {
+        TXNID txn = forest->hash_key;
+        lt_unlock_txn(tree, txn);
+    }
+    toku_rth_clear(tree->txns_to_unlock);
+    lt_retry_lock_requests(tree);
+}
+
 void 
 toku_lt_add_ref(toku_lock_tree* tree) {
     assert(tree);

src/lock_tree/locktree.h

@@ -76,6 +76,7 @@ struct __toku_lock_tree {
     OMT                dbs; //The extant dbs using this lock tree.
     OMT                lock_requests;
     toku_pthread_mutex_t mutex;
+    toku_rth*          txns_to_unlock; // set of txn's that could not release their locks because there was no db for the comparison function
 
     /** A temporary area where we store the results of various find on 
         the range trees that this lock tree owns 

src/lock_tree/tests/test_close_lock_conflict.c

@@ -0,0 +1,96 @@
+// verify that the lock tree is maintained after closes if txn's still own locks
+
+// A gets W(L)
+// B gets W(M)
+// close lock tree
+// A unlocks
+// reopen lock tree
+// B gets W(L)
+// B unlocks
+
+#include "test.h"
+
+int main(int argc, const char *argv[]) {
+    int r;
+
+    uint32_t max_locks = 2;
+    uint64_t max_lock_memory = 4096;
+
+    for (int i = 1; i < argc; i++) {
+        if (strcmp(argv[i], "-v") == 0 || strcmp(argv[i], "--verbose") == 0) {
+            verbose++;
+            continue;
+        }
+        if (strcmp(argv[i], "-q") == 0 || strcmp(argv[i], "--quiet") == 0) {
+            if (verbose > 0) verbose--;
+            continue;
+        }
+        if (strcmp(argv[i], "--max_locks") == 0 && i+1 < argc) {
+            max_locks = atoi(argv[++i]);
+            continue;
+        }
+        if (strcmp(argv[i], "--max_lock_memory") == 0 && i+1 < argc) {
+            max_lock_memory = atoi(argv[++i]);
+            continue;
+        }        
+        assert(0);
+    }
+
+    // setup
+    toku_ltm *ltm = NULL;
+    r = toku_ltm_create(&ltm, max_locks, max_lock_memory, dbpanic);
+    assert(r == 0 && ltm);
+
+    DB *fake_db = (DB *) 1;
+
+    toku_lock_tree *lt = NULL;
+    r = toku_ltm_get_lt(ltm, &lt, (DICTIONARY_ID){1}, fake_db, dbcmp);
+    assert(r == 0 && lt);
+
+    const TXNID txn_a = 1;
+    const TXNID txn_b = 2;
+
+    DBT key_l = { .data = "L", .size = 1 };
+    DBT key_m = { .data = "M", .size = 1 };
+
+    // txn_a gets W(L)
+    toku_lock_request a_w_l; toku_lock_request_init(&a_w_l, fake_db, txn_a, &key_l, &key_l, LOCK_REQUEST_WRITE);
+    r = toku_lock_request_start(&a_w_l, lt, false); assert(r == 0); 
+    assert(a_w_l.state == LOCK_REQUEST_COMPLETE && a_w_l.complete_r == 0);
+    toku_lock_request_destroy(&a_w_l);
+    toku_lt_add_ref(lt);
+
+    // txn_b gets W(M)
+    toku_lock_request b_w_m; toku_lock_request_init(&b_w_m, fake_db, txn_b, &key_m, &key_m, LOCK_REQUEST_WRITE);
+    r = toku_lock_request_start(&b_w_m, lt, false); assert(r == 0); 
+    assert(b_w_m.state == LOCK_REQUEST_COMPLETE && b_w_m.complete_r == 0);
+    toku_lock_request_destroy(&b_w_m);
+    toku_lt_add_ref(lt);
+
+    // start closing the lock tree
+    toku_lt_remove_db_ref(lt, fake_db);
+
+    // txn_a unlocks
+    r = toku_lt_unlock_txn(lt, txn_a);
+    toku_lt_remove_ref(lt);
+
+    // reopen the lock tree
+    r = toku_ltm_get_lt(ltm, &lt, (DICTIONARY_ID){1}, fake_db, dbcmp);
+    assert(r == 0 && lt);
+
+    // txn_b gets W(L)
+    toku_lock_request b_w_l; toku_lock_request_init(&b_w_l, fake_db, txn_b, &key_l, &key_l, LOCK_REQUEST_WRITE);
+    r = toku_lock_request_start(&b_w_l, lt, false); assert(r == 0);
+    assert(b_w_l.state == LOCK_REQUEST_COMPLETE && b_w_l.complete_r == 0);
+    toku_lock_request_destroy(&b_w_l);
+    toku_lt_add_ref(lt);
+
+    // release all locks for the transaction
+    r = toku_lt_unlock_txn(lt, txn_b);  assert(r == 0);
+    toku_lt_remove_ref(lt);
+
+    // shutdown 
+    r = toku_ltm_close(ltm); assert(r == 0);
+
+    return 0;
+}

src/lock_tree/tests/test_unlock_after_close.c

@@ -1,4 +1,8 @@
-// add a write lock on L for a transaction, remove a reference on the lock tree, then release the locks for the transaction.
+// verify that txn's can release locks after the lock tree is closed
+
+// A gets W(L)
+// close lock tree
+// A unlocks
 
 #include "test.h"
 
@@ -55,6 +59,7 @@ int main(int argc, const char *argv[]) {
 
     // release all locks for the transaction
     r = toku_lt_unlock_txn(lt, txn_a);  assert(r == 0);
+    toku_lt_remove_ref(lt);
 
     // shutdown 
     r = toku_ltm_close(ltm); assert(r == 0);