Addresses #350

Bugfix with releasing locks/unlocking a transaction.

DBC->c_get (DB_SET)
and DB->put (DB_YESOVERWRITE) now perform locking
Secondary dbs not included (behavior unknown).

Several tests added for c_get (DB_SET) and DB->put(DB_YESOVERWRITE).

Lock tree only created if INIT_TXN is enabled during env creation.


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

src/lock_tree/locktree.c

@@ -570,7 +570,7 @@ static int __toku_lt_free_points(toku_lock_tree* tree, toku_range* to_insert,
     u_int32_t i;
     for (i = 0; i < numfound; i++) {
         if (rt != NULL) {
-            r = toku_rt_delete(rt, to_insert);  
+            r = toku_rt_delete(rt, &tree->buf[i]);
             if (r!=0) return __toku_lt_panic(tree, r);
         }
         /*
@@ -1207,6 +1207,8 @@ int toku_lt_unlock(toku_lock_tree* tree, DB_TXN* txn) {
         ranges += toku_rt_get_size(selfwrite);
         r = __toku_lt_border_delete(tree, selfwrite);
         if (r!=0) return __toku_lt_panic(tree, r);
+        r = __toku_lt_free_contents(tree, selfwrite, NULL);
+        if (r!=0) return __toku_lt_panic(tree, r);
     }
 
     if (selfread || selfwrite) toku_rth_delete(tree->rth, txn);

src/lock_tree/locktree.h

@@ -363,6 +363,7 @@ int toku_lt_acquire_range_write_lock(toku_lock_tree* tree, DB_TXN* txn,
  * Returns:
  *      0:          Success.
  *      EINVAL:     If (tree == NULL || txn == NULL).
+ *      EINVAL:     If panicking.
  * *** Note that txn == NULL is not supported at this time.
  */
 int toku_lt_unlock(toku_lock_tree* tree, DB_TXN* txn);

src/tests/Makefile

@@ -42,7 +42,7 @@ endif
 
 LIBNAME=libdb.$(LIBEXT)
 # GCOV_FLAGS = -fprofile-arcs -ftest-coverage
-CFLAGS += -Wall -Werror $(OPTFLAGS) -g $(GCOV_FLAGS)
+CFLAGS += -Wall -Werror $(OPTFLAGS) -g3 -ggdb3 $(GCOV_FLAGS)
 TDB_CPPFLAGS = -I../../include
 
 

src/tests/test.h

@@ -3,6 +3,7 @@
 #include <stdint.h>
 #include <stdio.h>
 #include <db.h>
+#include "../../newbrt/brttypes.h"
 
 #ifndef DB_YESOVERWRITE
 #define DB_YESOVERWRITE 0
@@ -15,6 +16,7 @@ int verbose=0;
 
 #define CKERR(r) ({ if (r!=0) fprintf(stderr, "%s:%d error %d %s\n", __FILE__, __LINE__, r, db_strerror(r)); assert(r==0); })
 #define CKERR2(r,r2) ({ if (r!=r2) fprintf(stderr, "%s:%d error %d %s, expected %d\n", __FILE__, __LINE__, r, db_strerror(r), r2); assert(r==r2); })
+#define CKERR2s(r,r2,r3) ({ if (r!=r2 && r!=r3) fprintf(stderr, "%s:%d error %d %s, expected %d or %d\n", __FILE__, __LINE__, r, db_strerror(r), r2,r3); assert(r==r2||r==r3); })
 
 void parse_args (int argc, const char *argv[]) {
     const char *argv0=argv[0];

src/tests/test_db_txn_locks.c

@@ -0,0 +1,224 @@
+/* -*- mode: C; c-basic-offset: 4 -*- */
+#ident "Copyright (c) 2007 Tokutek Inc.  All rights reserved."
+
+#include <string.h>
+#include <db.h>
+#include <assert.h>
+#include <errno.h>
+#include <sys/stat.h>
+
+#include "test.h"
+
+// DIR is defined in the Makefile
+
+int dbtcmp(DBT *dbt1, DBT *dbt2) {
+    int r;
+    
+    r = dbt1->size - dbt2->size;  if (r) return r;
+    return memcmp(dbt1->data, dbt2->data, dbt1->size);
+}
+
+DB *db;
+DB_TXN* txns[(int)256];
+DB_ENV* dbenv;
+DBC*    cursors[(int)256];
+
+void put(BOOL success, char txn, int _key, int _data) {
+    assert(txns[(int)txn]);
+
+    int r;
+    DBT key;
+    DBT data;
+    
+    r = db->put(db, txns[(int)txn],
+                    dbt_init(&key, &_key, sizeof(int)),
+                    dbt_init(&data, &_data, sizeof(int)),
+                    DB_YESOVERWRITE);
+    if (success)    CKERR(r);
+    else            CKERR2s(r, DB_LOCK_DEADLOCK, DB_LOCK_NOTGRANTED);
+}
+
+void cget(BOOL success, BOOL find, char txn, int _key, int _data) {
+    assert(txns[(int)txn] && cursors[(int)txn]);
+
+    int r;
+    DBT key;
+    DBT data;
+    
+    r = cursors[(int)txn]->c_get(cursors[(int)txn],
+                                 dbt_init(&key,  &_key,  sizeof(int)),
+                                 dbt_init(&data, &_data, sizeof(int)),
+                                 DB_SET);
+    if (success) {
+        if (find)   CKERR(r);
+        else        CKERR2(r, DB_NOTFOUND);
+    }
+    else            CKERR2s(r, DB_LOCK_DEADLOCK, DB_LOCK_NOTGRANTED);
+}
+
+void init_txn(char name) {
+    int r;
+    assert(!txns[(int)name]);
+    r = dbenv->txn_begin(dbenv, NULL, &txns[(int)name], DB_TXN_NOWAIT);
+        CKERR(r);
+    assert(txns[(int)name]);
+}
+
+void init_dbc(char name) {
+    int r;
+
+    assert(!cursors[(int)name] && txns[(int)name]);
+    r = db->cursor(db, txns[(int)name], &cursors[(int)name], 0);
+        CKERR(r);
+    assert(cursors[(int)name]);
+}
+
+void commit_txn(char name) {
+    int r;
+    assert(txns[(int)name] && !cursors[(int)name]);
+
+    r = txns[(int)name]->commit(txns[(int)name], 0);
+        CKERR(r);
+    txns[(int)name] = NULL;
+}
+
+void abort_txn(char name) {
+    int r;
+    assert(txns[(int)name] && !cursors[(int)name]);
+
+    r = txns[(int)name]->abort(txns[(int)name]);
+        CKERR(r);
+    txns[(int)name] = NULL;
+}
+
+void close_dbc(char name) {
+    int r;
+
+    assert(cursors[(int)name]);
+    r = cursors[(int)name]->c_close(cursors[(int)name]);
+        CKERR(r);
+    cursors[(int)name] = NULL;
+}
+
+void early_commit(char name) {
+    assert(cursors[(int)name] && txns[(int)name]);
+    close_dbc(name);
+    commit_txn(name);
+}
+
+void early_abort(char name) {
+    assert(cursors[(int)name] && txns[(int)name]);
+    close_dbc(name);
+    abort_txn(name);
+}
+
+void setup_dbs(u_int32_t dup_flags) {
+    int r;
+
+    system("rm -rf " DIR);
+    mkdir(DIR, 0777);
+    dbenv   = NULL;
+    db      = NULL;
+    /* Open/create primary */
+    r = db_env_create(&dbenv, 0);
+        CKERR(r);
+    r = dbenv->open(dbenv, DIR, DB_CREATE | DB_PRIVATE | DB_INIT_MPOOL | DB_INIT_TXN, 0600);
+        CKERR(r);
+    
+    r = db_create(&db, dbenv, 0);
+        CKERR(r);
+    if (dup_flags) {
+        r = db->set_flags(db, dup_flags);
+            CKERR(r);
+    }
+    r = db->open(db, NULL, "foobar.db", NULL, DB_BTREE, DB_CREATE, 0600);
+        CKERR(r);
+    char a;
+    for (a = 'a'; a <= 'z'; a++) {
+        init_txn(a);
+        init_dbc(a);
+    }
+}
+
+void close_dbs(void) {
+    char a;
+    for (a = 'a'; a <= 'z'; a++) {
+        if (cursors[(int)a]) close_dbc(a);
+        if (txns[(int)a])    commit_txn(a);
+    }
+
+    int r;
+    r = db->close(db, 0);
+        CKERR(r);
+    db      = NULL;
+    r = dbenv->close(dbenv, 0);
+        CKERR(r);
+    dbenv   = NULL;
+}
+
+void test(u_int32_t dup_flags) {
+    /* ********************************************************************** */
+    setup_dbs(dup_flags);
+    close_dbs();
+    /* ********************************************************************** */
+    setup_dbs(dup_flags);
+    cget(TRUE, FALSE, 'a', 1, 1);
+    close_dbs();
+    /* ********************************************************************** */
+    setup_dbs(dup_flags);
+    put(TRUE, 'a', 1, 1);
+    close_dbs();
+    /* ********************************************************************** */
+    setup_dbs(dup_flags);
+    cget(TRUE, FALSE, 'a', 1, 1);
+    cget(TRUE, FALSE, 'a', 2, 1);
+    close_dbs();
+    /* ********************************************************************** */
+    setup_dbs(dup_flags);
+    cget(TRUE, FALSE, 'a', 1, 1);
+    cget(TRUE, FALSE, 'a', 1, 1);
+    close_dbs();
+    /* ********************************************************************** */
+    setup_dbs(dup_flags);
+    cget(TRUE, FALSE, 'a', 1, 1);
+    cget(TRUE, FALSE, 'b', 2, 1);
+    close_dbs();
+    /* ********************************************************************** */
+    setup_dbs(dup_flags);
+    cget(TRUE, FALSE, 'a', 1, 1);
+    cget(TRUE, FALSE, 'b', 1, 1);
+    close_dbs();
+    /* ********************************************************************** */
+    setup_dbs(dup_flags);
+    cget(TRUE, FALSE, 'a', 1, 1);
+    cget(TRUE, FALSE, 'b', 1, 1);
+    put(FALSE, 'a', 1, 1);
+    early_commit('b');
+    put(TRUE, 'a', 1, 1);
+    cget(TRUE, TRUE, 'a', 1, 1);
+    cget(TRUE, FALSE, 'a', 2, 1);
+    cget(FALSE, TRUE, 'c', 1, 1);
+    early_commit('a');
+    cget(TRUE, TRUE, 'c', 1, 1);
+    close_dbs();
+    /* ********************************************************************** */
+    setup_dbs(dup_flags);
+    cget(TRUE, FALSE, 'a', 1, 1);
+    cget(TRUE, FALSE, 'b', 1, 1);
+    put(FALSE, 'a', 1, 1);
+    early_commit('b');
+    put(TRUE, 'a', 1, 1);
+    cget(TRUE, TRUE, 'a', 1, 1);
+    cget(TRUE, FALSE, 'a', 2, 1);
+    cget(FALSE, TRUE, 'c', 1, 1);
+    early_abort('a');
+    cget(TRUE, FALSE, 'c', 1, 1);
+    close_dbs();
+    /* ********************************************************************** */
+}
+
+int main() {
+    test(0);
+    test(DB_DUP | DB_DUPSORT);
+    return 0;
+}

src/ydb.c

@@ -755,22 +755,23 @@ int db_env_create(DB_ENV ** envp, u_int32_t flags) {
     ydb_lock(); int r = toku_env_create(envp, flags); ydb_unlock(); return r;
 }
 
-static void toku_txn_release_locks(DB_TXN* txn) {
+static int toku_txn_release_locks(DB_TXN* txn) {
     assert(txn);
     toku_lth* lth = txn->i->lth;
     assert(lth);
 
     int r;
+    int r2 = 0;
     toku_lth_start_scan(lth);
     toku_lock_tree* next = toku_lth_next(lth);
     while (next) {
         r = toku_lt_unlock(next, txn);
-        /* Only NULL parameters can give a non 0 return value. */
-        assert(r==0);
+        if (r!=0 && !r2) r2 = r;
         next = toku_lth_next(lth);
     }
     toku_lth_close(lth);
     txn->i->lth = NULL;
+    return r2;
 }
 
 static int toku_txn_commit(DB_TXN * txn, u_int32_t flags) {
@@ -788,11 +789,11 @@ static int toku_txn_commit(DB_TXN * txn, u_int32_t flags) {
 	toku_free(txn->i->tokutxn);
     }
     // Cleanup */
-    toku_txn_release_locks(txn);
+    int r2 = toku_txn_release_locks(txn);
     if (txn->i)
         toku_free(txn->i);
     toku_free(txn);
-    return r; // The txn is no good after the commit.
+    return r ? r : r2; // The txn is no good after the commit.
 }
 
 static u_int32_t toku_txn_id(DB_TXN * txn) {
@@ -1134,15 +1135,34 @@ delete_silently_and_retry:
 
 static int toku_c_get(DBC * c, DBT * key, DBT * data, u_int32_t flag) {
     DB *db = c->dbp;
-//    DB_TXN *txn = c->i->txn;
+    DB_TXN *txn = c->i->txn;
     HANDLE_PANICKED_DB(db);
     u_int32_t get_flag = get_main_cursor_flag(flag);
     int r;
 
     if (db->i->primary==0) {
         r = toku_c_get_noassociate(c, key, data, flag);
+        if (db->i->lt) {
+            if (r != DB_NOTFOUND && r != 0 && r != DB_KEYEMPTY) return r;
             int r2 = 0;
             switch (get_flag) {
+                case (DB_SET): {
+                    if (r == DB_NOTFOUND) {
+                        r2 = toku_lt_acquire_range_read_lock(db->i->lt, txn,
+                                                      key, toku_lt_neg_infinity,
+                                                      key, toku_lt_infinity);
+                    }
+                    else {
+                        r2 = toku_lt_acquire_range_read_lock(db->i->lt, txn,
+                                                      key, toku_lt_neg_infinity,
+                                                      key, data);
+                    }
+                    break;
+                }
+            }
+            if (r2!=0) return r2;
+        }
+    }
 /*
 These should be done but were not tested prior to commit.  
                 case (DB_CURRENT): {
@@ -1250,9 +1270,6 @@ These are not ready and are just notes.
                     assert(FALSE);
                 }
 */
-        }
-        if (r2!=0) return r2;
-    }
     else {
         // It's a c_get on a secondary.
         DBT primary_key;
@@ -1818,6 +1835,7 @@ static int toku_txn_add_lt(DB_TXN* txn, toku_lock_tree* lt) {
 static int toku_db_open(DB * db, DB_TXN * txn, const char *fname, const char *dbname, DBTYPE dbtype, u_int32_t flags, int mode) {
     HANDLE_PANICKED_DB(db);
     // Warning.  Should check arguments.  Should check return codes on malloc and open and so forth.
+    BOOL transactions = (db->dbenv->i->open_flags & DB_INIT_TXN) != 0;
 
     int openflags = 0;
     int r;
@@ -1878,7 +1896,7 @@ static int toku_db_open(DB * db, DB_TXN * txn, const char *fname, const char *db
     db->i->open_flags = flags;
     db->i->open_mode = mode;
 
-    /* TODO: Only create lock tree if necessary! (lock subsystem?) */
+    if (transactions) {
         r = toku_lt_create(&db->i->lt, db, FALSE,
                            toku_db_lt_panic, db->dbenv->i->max_locks,
                            &db->dbenv->i->num_locks,
@@ -1887,6 +1905,9 @@ static int toku_db_open(DB * db, DB_TXN * txn, const char *fname, const char *db
         if (r!=0) goto error_cleanup;
         r = toku_lt_set_txn_add_lt_callback(db->i->lt, toku_txn_add_lt);
         assert(r==0);
+    }
+        
+    
 
     r = toku_brt_open(db->i->brt, db->i->full_fname, fname, dbname,
 		      is_db_create, is_db_excl, is_db_unknown,
@@ -1896,6 +1917,7 @@ static int toku_db_open(DB * db, DB_TXN * txn, const char *fname, const char *db
     if (r != 0)
         goto error_cleanup;
 
+    if (db->i->lt) {
         unsigned int brtflags;
         BOOL dups;
         /* Whether we have dups is only known starting now. */
@@ -1904,6 +1926,7 @@ static int toku_db_open(DB * db, DB_TXN * txn, const char *fname, const char *db
         r = toku_lt_set_dups(db->i->lt, dups);
         /* toku_lt_set_dups cannot return an error here. */
         assert(r==0);
+    }
 
     return 0;
  
@@ -1966,7 +1989,10 @@ static int toku_db_put_noassociate(DB * db, DB_TXN * txn, DBT * key, DBT * data,
 #endif
         }
     }
-    
+    if (db->i->lt) {
+        r = toku_lt_acquire_write_lock(db->i->lt, txn, key, data);
+        if (r!=0) return r;
+    }
     r = toku_brt_insert(db->i->brt, key, data, txn ? txn->i->tokutxn : 0);
     //printf("%s:%d %d=__toku_db_put(...)\n", __FILE__, __LINE__, r);
     return r;