Merge pull request #336 from Jun-Yuan/read-commit8

 [DB-884] TokuDB crashes under sysbench in READ-COMMITTED mode ...

cmake_modules/TokuSetupCompiler.cmake

@@ -24,12 +24,12 @@ endif ()
 
 ## add TOKU_PTHREAD_DEBUG for debug builds
 if (CMAKE_VERSION VERSION_LESS 3.0)
-  set_property(DIRECTORY APPEND PROPERTY COMPILE_DEFINITIONS_DEBUG TOKU_PTHREAD_DEBUG=1)
-  set_property(DIRECTORY APPEND PROPERTY COMPILE_DEFINITIONS_DRD TOKU_PTHREAD_DEBUG=1)
+  set_property(DIRECTORY APPEND PROPERTY COMPILE_DEFINITIONS_DEBUG TOKU_PTHREAD_DEBUG=1 TOKU_DEBUG_TXN_SYNC=1)
+  set_property(DIRECTORY APPEND PROPERTY COMPILE_DEFINITIONS_DRD TOKU_PTHREAD_DEBUG=1 TOKU_DEBUG_TXN_SYNC=1)
   set_property(DIRECTORY APPEND PROPERTY COMPILE_DEFINITIONS_DRD _FORTIFY_SOURCE=2)
 else ()
   set_property(DIRECTORY APPEND PROPERTY COMPILE_DEFINITIONS
-    $<$<OR:$<CONFIG:DEBUG>,$<CONFIG:DRD>>:TOKU_PTHREAD_DEBUG=1>
+    $<$<OR:$<CONFIG:DEBUG>,$<CONFIG:DRD>>:TOKU_PTHREAD_DEBUG=1 TOKU_DEBUG_TXN_SYNC=1>
     $<$<CONFIG:DRD>:_FORTIFY_SOURCE=2>
     )
 endif ()

ft/txn/txn_manager.cc

@@ -45,7 +45,15 @@ Copyright (c) 2006, 2015, Percona and/or its affiliates. All rights reserved.
 #include "ft/txn/txn_manager.h"
 #include "ft/txn/rollback.h"
 #include "util/omt.h"
+//this is only for testing
 
+static void  (* test_txn_sync_callback) (uint64_t, void *) = NULL;
+static void * test_txn_sync_callback_extra = NULL;
+
+void set_test_txn_sync_callback(void (*cb) (uint64_t, void *), void *extra) {
+        test_txn_sync_callback = cb;
+        test_txn_sync_callback_extra = extra;
+}
 bool garbage_collection_debug = false;
 
 static bool txn_records_snapshot(TXN_SNAPSHOT_TYPE snapshot_type, struct tokutxn *parent) {
@@ -525,14 +533,19 @@ void toku_txn_manager_handle_snapshot_create_for_child_txn(
         XMALLOC(txn->live_root_txn_list);
         txn_manager_lock(txn_manager);
         txn_manager_create_snapshot_unlocked(txn_manager, txn);
-        txn_manager_unlock(txn_manager);
     }
     else {
         inherit_snapshot_from_parent(txn);
     }
-    if (copies_snapshot) {
-        setup_live_root_txn_list(&txn_manager->live_root_ids, txn->live_root_txn_list);  
-    }
+
+    toku_debug_txn_sync(pthread_self());
+
+    if (copies_snapshot) { 
+	if(!records_snapshot) 
+    	    txn_manager_lock(txn_manager);
+	setup_live_root_txn_list(&txn_manager->live_root_ids, txn->live_root_txn_list);  
+       txn_manager_unlock(txn_manager);
+    } 
 }
 
 void toku_txn_manager_handle_snapshot_destroy_for_child_txn(

ft/txn/txn_manager.h

@@ -43,6 +43,15 @@ Copyright (c) 2006, 2015, Percona and/or its affiliates. All rights reserved.
 
 #include "ft/txn/txn.h"
 
+void set_test_txn_sync_callback(void (*) (uint64_t, void*), void*);
+#define toku_test_txn_sync_callback(a) ((test_txn_sync_callback)? test_txn_sync_callback( a,test_txn_sync_callback_extra) : (void) 0)
+
+#if TOKU_DEBUG_TXN_SYNC
+#define toku_debug_txn_sync(a) toku_test_txn_sync_callback(a) 
+#else
+#define toku_debug_txn_sync(a) ((void) 0)
+#endif
+
 typedef struct txn_manager *TXN_MANAGER;
 
 struct referenced_xid_tuple {

src/export.map

@@ -82,6 +82,7 @@
    toku_test_db_redirect_dictionary;
    toku_test_get_latest_lsn;
    toku_test_get_checkpointing_user_data_status;
+   toku_set_test_txn_sync_callback;
    toku_indexer_set_test_only_flags;
    toku_increase_last_xid;
 

src/tests/txn_manager_handle_snapshot_atomicity.cc

@@ -0,0 +1,217 @@
+/* -*- mode: C++; c-basic-offset: 4; indent-tabs-mode: nil -*- */
+// vim: ft=cpp:expandtab:ts=8:sw=4:softtabstop=4:
+#ident "$Id$"
+/*======
+This file is part of PerconaFT.
+
+
+Copyright (c) 2006, 2015, Percona and/or its affiliates. All rights reserved.
+
+    PerconaFT is free software: you can redistribute it and/or modify
+    it under the terms of the GNU General Public License, version 2,
+    as published by the Free Software Foundation.
+
+    PerconaFT 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.
+
+    You should have received a copy of the GNU General Public License
+    along with PerconaFT.  If not, see <http://www.gnu.org/licenses/>.
+
+----------------------------------------
+
+    PerconaFT is free software: you can redistribute it and/or modify
+    it under the terms of the GNU Affero General Public License, version 3,
+    as published by the Free Software Foundation.
+
+    PerconaFT 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 Affero General Public License for more details.
+
+    You should have received a copy of the GNU Affero General Public License
+    along with PerconaFT.  If not, see <http://www.gnu.org/licenses/>.
+======= */
+
+#ident "Copyright (c) 2006, 2015, Percona and/or its affiliates. All rights reserved."
+
+//In response to the read-commit crash bug in the sysbench, this test is created to test
+//the atomicity of the txn manager when handling the child txn snapshot.
+//The test is supposed to fail before the read-commit-fix.
+
+#include "test.h"
+#include "toku_pthread.h"
+#include "ydb.h"
+struct test_sync {
+    int state;
+    toku_mutex_t lock;
+    toku_cond_t cv;
+};
+
+static void test_sync_init(struct test_sync *UU(sync)) {
+#if TOKU_DEBUG_TXN_SYNC
+    sync->state = 0;
+    toku_mutex_init(&sync->lock, NULL);
+    toku_cond_init(&sync->cv, NULL);
+#endif
+}
+
+static void test_sync_destroy(struct test_sync *UU(sync)) {
+#if TOKU_DEBUG_TXN_SYNC
+    toku_mutex_destroy(&sync->lock);
+    toku_cond_destroy(&sync->cv);
+#endif
+}
+
+static void test_sync_sleep(struct test_sync *UU(sync), int UU(new_state)) {
+#if TOKU_DEBUG_TXN_SYNC
+    toku_mutex_lock(&sync->lock);
+    while (sync->state != new_state) {
+        toku_cond_wait(&sync->cv, &sync->lock);
+    }
+    toku_mutex_unlock(&sync->lock);
+#endif
+}
+
+static void test_sync_next_state(struct test_sync *UU(sync)) {
+#if TOKU_DEBUG_TXN_SYNC
+    toku_mutex_lock(&sync->lock);
+    sync->state++;
+    toku_cond_broadcast(&sync->cv);
+    toku_mutex_unlock(&sync->lock);
+#endif
+}
+
+
+struct start_txn_arg {
+    DB_ENV *env;
+    DB *db;
+    DB_TXN * parent;
+};
+
+static struct test_sync sync_s;
+
+static void test_callback(uint64_t self_tid, void * extra) {
+    pthread_t **p = (pthread_t **) extra;
+    pthread_t tid_1 = *p[0];
+    pthread_t tid_2 = *p[1];
+    assert(self_tid == tid_2);
+    printf("%s: the thread[%" PRIu64 "] is going to wait...\n", __func__, tid_1);
+    test_sync_next_state(&sync_s);
+    sleep(3);
+    //test_sync_sleep(&sync_s,3);
+    //using test_sync_sleep/test_sync_next_state pair can sync threads better, however
+    //after the fix, this might cause a deadlock. just simply use sleep to do a proof-
+    //of-concept test. 
+    printf("%s: the thread[%" PRIu64 "] is resuming...\n", __func__, tid_1);
+    return;
+}
+
+static void * start_txn2(void * extra) {
+    struct start_txn_arg * args = (struct start_txn_arg *) extra;
+    DB_ENV * env = args -> env;
+    DB * db = args->db;
+    DB_TXN * parent = args->parent;
+    test_sync_sleep(&sync_s, 1);
+    printf("start %s [thread %" PRIu64 "]\n", __func__, pthread_self());
+    DB_TXN *txn;
+    int r = env->txn_begin(env, parent, &txn,  DB_READ_COMMITTED);
+    assert(r == 0);
+    //do some random things...
+    DBT key, data;
+    dbt_init(&key, "hello", 6);
+    dbt_init(&data, "world", 6);
+    db->put(db, txn, &key, &data, 0);
+    db->get(db, txn, &key, &data, 0);
+
+    r = txn->commit(txn, 0);
+    assert(r == 0);
+    printf("%s done[thread %" PRIu64 "]\n", __func__, pthread_self());
+    return extra;
+}
+
+static void * start_txn1(void * extra) {
+    struct start_txn_arg * args = (struct start_txn_arg *) extra;
+    DB_ENV * env = args -> env;
+    DB * db = args->db;
+    printf("start %s: [thread %" PRIu64 "]\n", __func__, pthread_self());
+    DB_TXN *txn;
+    int r = env->txn_begin(env, NULL, &txn,  DB_READ_COMMITTED);
+    assert(r == 0);
+    printf("%s: txn began by [thread %" PRIu64 "], will wait\n", __func__, pthread_self());
+    test_sync_next_state(&sync_s);
+    test_sync_sleep(&sync_s,2);
+    printf("%s: [thread %" PRIu64 "] resumed\n", __func__, pthread_self());
+    //do some random things...
+    DBT key, data;
+    dbt_init(&key, "hello", 6);
+    dbt_init(&data, "world", 6);
+    db->put(db, txn, &key, &data, 0);
+    db->get(db, txn, &key, &data, 0);
+    r = txn->commit(txn, 0);
+    assert(r == 0);
+    printf("%s: done[thread %" PRIu64 "]\n", __func__, pthread_self());
+    //test_sync_next_state(&sync_s);
+    return extra;
+}
+
+int test_main (int UU(argc), char * const UU(argv[])) {
+    int r;
+    toku_os_recursive_delete(TOKU_TEST_FILENAME);
+    r = toku_os_mkdir(TOKU_TEST_FILENAME, S_IRWXU+S_IRWXG+S_IRWXO);
+    assert(r == 0);
+
+    DB_ENV *env;
+    r = db_env_create(&env, 0);
+    assert(r == 0);
+
+    r = env->open(env, TOKU_TEST_FILENAME, DB_INIT_MPOOL|DB_CREATE|DB_THREAD |DB_INIT_LOCK|DB_INIT_LOG|DB_INIT_TXN|DB_PRIVATE, S_IRWXU+S_IRWXG+S_IRWXO);
+    assert(r == 0);
+
+    DB *db = NULL;
+    r = db_create(&db, env, 0);
+    assert(r == 0);
+
+    r = db->open(db, NULL, "testit", NULL, DB_BTREE, DB_AUTO_COMMIT+DB_CREATE, S_IRWXU+S_IRWXG+S_IRWXO);
+    assert(r == 0);
+
+    DB_TXN * parent = NULL;
+    r = env->txn_begin(env, 0, &parent, DB_READ_COMMITTED);
+    assert(r == 0);
+
+    ZERO_STRUCT(sync_s);
+    test_sync_init(&sync_s);
+
+    pthread_t tid_1 = 0;
+    pthread_t tid_2 = 0;
+    pthread_t* callback_extra[2] = {&tid_1, &tid_2};
+    toku_set_test_txn_sync_callback(test_callback, callback_extra);
+
+    struct start_txn_arg args = {env, db, parent};
+
+    r = pthread_create(&tid_1, NULL, start_txn1, &args);
+    assert(r==0);
+
+    r= pthread_create(&tid_2, NULL, start_txn2, &args);
+    assert(r==0);
+
+     void * ret; 
+    r = pthread_join(tid_1, &ret);
+    assert(r == 0);
+    r = pthread_join(tid_2, &ret);
+    assert(r == 0);
+
+    r = parent->commit(parent, 0);
+    assert(r ==0);
+
+    test_sync_destroy(&sync_s);
+    r = db->close(db, 0);
+    assert(r == 0);
+
+    r = env->close(env, 0);
+    assert(r == 0);
+
+    return 0;
+}
+

src/ydb.cc

@@ -3148,6 +3148,10 @@ toku_test_get_latest_lsn(DB_ENV *env) {
     return rval.lsn;
 }
 
+void toku_set_test_txn_sync_callback(void (* cb) (uint64_t, void *), void * extra) {
+    set_test_txn_sync_callback(cb, extra);
+}
+
 int 
 toku_test_get_checkpointing_user_data_status (void) {
     return toku_cachetable_get_checkpointing_user_data_status();

src/ydb.h

@@ -58,3 +58,6 @@ extern "C" uint64_t toku_test_get_latest_lsn(DB_ENV *env) __attribute__((__visib
 
 // test-only function
 extern "C" int toku_test_get_checkpointing_user_data_status(void) __attribute__((__visibility__("default")));
+
+// test-only function
+extern "C" void toku_set_test_txn_sync_callback(void (* ) (uint64_t, void *), void * extra) __attribute__((__visibility__("default")));