[t:4877], [t:4966], [t:4952], [t:4881], [t:4918], merge to main

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

.globalrc

-default:\
-        :skip=/windows/,/dbg/,/opt/,/build/,/gccdbg/,/gccopt/,/iccdbg/,/iccopt/,/clangdbg/,/NightlyRelease/,/NightlyCoverage/,/NightlyDebug/,/Debug/,/Release/,/Coverage/,/cov/:

cmake_modules/TokuBuildTagDatabases.cmake

@@ -8,6 +8,8 @@ file(GLOB_RECURSE all_srcs
   src/*.c
   utils/*.c
   db-benchmark-test/*.c
+  )
+list(APPEND all_srcs
   ${CMAKE_CURRENT_BINARY_DIR}/ft/log_code.c
   ${CMAKE_CURRENT_BINARY_DIR}/ft/log_print.c
   )
@@ -20,6 +22,8 @@ file(GLOB_RECURSE all_hdrs
   src/*.h
   utils/*.h
   db-benchmark-test/*.h
+  )
+list(APPEND all_hdrs
   ${CMAKE_CURRENT_BINARY_DIR}/toku_include/config.h
   ${CMAKE_CURRENT_BINARY_DIR}/buildheader/db.h
   ${CMAKE_CURRENT_BINARY_DIR}/ft/log_header.h
@@ -76,14 +80,21 @@ endif ()
 option(USE_GTAGS "Build the gtags database." ON)
 if (USE_GTAGS)
   find_program(GTAGS "gtags")
+  find_program(MKID "mkid")
   if (NOT GTAGS MATCHES NOTFOUND)
-    ## todo: use global -u instead of gtags each time
+    file(WRITE "${CMAKE_CURRENT_BINARY_DIR}/gtags.files" "")
+    foreach(file ${all_srcs} ${all_hdrs})
+      file(APPEND "${CMAKE_CURRENT_BINARY_DIR}/gtags.files" "${file}\n")
+    endforeach(file)
+    if (NOT MKID MATCHES NOTFOUND)
+      set(idutils_option "-I")
+    endif ()
     add_custom_command(
       OUTPUT "${CMAKE_CURRENT_SOURCE_DIR}/GTAGS"
       OUTPUT "${CMAKE_CURRENT_SOURCE_DIR}/GRTAGS"
       OUTPUT "${CMAKE_CURRENT_SOURCE_DIR}/GPATH"
       OUTPUT "${CMAKE_CURRENT_SOURCE_DIR}/GSYMS"
-      COMMAND ${GTAGS} --gtagsconf "${CMAKE_CURRENT_SOURCE_DIR}/.globalrc"
+      COMMAND ${GTAGS} -i ${idutils_option} -f "${CMAKE_CURRENT_BINARY_DIR}/gtags.files"
       DEPENDS ${all_srcs} ${all_hdrs} install_tdb_h generate_config_h generate_log_code
       WORKING_DIRECTORY "${CMAKE_CURRENT_SOURCE_DIR}")
     add_custom_target(build_GTAGS ALL DEPENDS

ft/CMakeLists.txt

@@ -59,6 +59,8 @@ set(FT_SOURCES
   quicklz.c
   recover.c
   rollback.c
+  rollback-apply.c
+  rollback-ct-callbacks.c
   roll.c
   sort.c
   sub_block.c

ft/cachetable.c

@@ -3246,14 +3246,14 @@ log_open_txn (OMTVALUE txnv, u_int32_t UU(index), void *UU(extra)) {
         int r = toku_log_xstillopen(logger, NULL, 0,
                                     toku_txn_get_txnid(txn),
                                     toku_txn_get_txnid(toku_logger_txn_parent(txn)),
-                                    txn->rollentry_raw_count,
+                                    txn->roll_info.rollentry_raw_count,
                                     open_filenums,
                                     txn->force_fsync_on_commit,
-                                    txn->num_rollback_nodes,
-                                    txn->num_rollentries,
-                                    txn->spilled_rollback_head,
-                                    txn->spilled_rollback_tail,
-                                    txn->current_rollback);
+                                    txn->roll_info.num_rollback_nodes,
+                                    txn->roll_info.num_rollentries,
+                                    txn->roll_info.spilled_rollback_head,
+                                    txn->roll_info.spilled_rollback_tail,
+                                    txn->roll_info.current_rollback);
         assert(r==0);
         return 0;
     }
@@ -3263,14 +3263,14 @@ log_open_txn (OMTVALUE txnv, u_int32_t UU(index), void *UU(extra)) {
         int r = toku_log_xstillopenprepared(logger, NULL, 0,
                                             toku_txn_get_txnid(txn),
                                             &xa_xid,
-                                            txn->rollentry_raw_count,
+                                            txn->roll_info.rollentry_raw_count,
                                             open_filenums,
                                             txn->force_fsync_on_commit,
-                                            txn->num_rollback_nodes,
-                                            txn->num_rollentries,
-                                            txn->spilled_rollback_head,
-                                            txn->spilled_rollback_tail,
-                                            txn->current_rollback);
+                                            txn->roll_info.num_rollback_nodes,
+                                            txn->roll_info.num_rollentries,
+                                            txn->roll_info.spilled_rollback_head,
+                                            txn->roll_info.spilled_rollback_tail,
+                                            txn->roll_info.current_rollback);
         assert(r==0);
         return 0;
     }

ft/ft.c

@@ -776,11 +776,11 @@ toku_dictionary_redirect_abort(FT old_h, FT new_h, TOKUTXN txn) {
  *****/
 
 int 
-toku_dictionary_redirect (const char *dst_fname_in_env, FT_HANDLE old_ft, TOKUTXN txn) {
+toku_dictionary_redirect (const char *dst_fname_in_env, FT_HANDLE old_ft_h, TOKUTXN txn) {
 // Input args:
 //   new file name for dictionary (relative to env)
-//   old_ft is a live brt of open handle ({DB, BRT} pair) that currently refers to old dictionary file.
-//   (old_ft may be one of many handles to the dictionary.)
+//   old_ft_h is a live brt of open handle ({DB, BRT} pair) that currently refers to old dictionary file.
+//   (old_ft_h may be one of many handles to the dictionary.)
 //   txn that created the loader
 // Requires: 
 //   ydb_lock is held.
@@ -797,11 +797,11 @@ toku_dictionary_redirect (const char *dst_fname_in_env, FT_HANDLE old_ft, TOKUTX
 //   If the txn aborts, then this operation will be undone
     int r;
 
-    FT old_h = old_ft->ft;
+    FT old_ft = old_ft_h->ft;
 
     // dst file should not be open.  (implies that dst and src are different because src must be open.)
     {
-        CACHETABLE ct = toku_cachefile_get_cachetable(old_h->cf);
+        CACHETABLE ct = toku_cachefile_get_cachetable(old_ft->cf);
         CACHEFILE cf;
         r = toku_cachefile_of_iname_in_env(ct, dst_fname_in_env, &cf);
         if (r==0) {
@@ -813,25 +813,24 @@ toku_dictionary_redirect (const char *dst_fname_in_env, FT_HANDLE old_ft, TOKUTX
     }
 
     if (txn) {
-        r = toku_txn_note_ft(txn, old_h);  // mark old brt as touched by this txn
-        assert_zero(r);
+        toku_txn_maybe_note_ft(txn, old_ft);  // mark old ft as touched by this txn
     }
 
-    FT new_h;
-    r = dictionary_redirect_internal(dst_fname_in_env, old_h, txn, &new_h);
+    FT new_ft;
+    r = dictionary_redirect_internal(dst_fname_in_env, old_ft, txn, &new_ft);
     assert_zero(r);
 
     // make rollback log entry
     if (txn) {
-        r = toku_txn_note_ft(txn, new_h); // mark new brt as touched by this txn
+        toku_txn_maybe_note_ft(txn, new_ft); // mark new ft as touched by this txn
 
-        FILENUM old_filenum = toku_cachefile_filenum(old_h->cf);
-        FILENUM new_filenum = toku_cachefile_filenum(new_h->cf);
+        FILENUM old_filenum = toku_cachefile_filenum(old_ft->cf);
+        FILENUM new_filenum = toku_cachefile_filenum(new_ft->cf);
         r = toku_logger_save_rollback_dictionary_redirect(txn, old_filenum, new_filenum);
         assert_zero(r);
 
         TXNID xid = toku_txn_get_txnid(txn);
-        toku_ft_suppress_rollbacks(new_h, txn);
+        toku_ft_suppress_rollbacks(new_ft, txn);
         r = toku_log_suppress_rollback(txn->logger, NULL, 0, new_filenum, xid);
         assert_zero(r);
     }
@@ -849,29 +848,14 @@ static int find_xid (OMTVALUE v, void *txnv) {
     return 0;
 }
 
-// returns if ref was added
-BOOL
-toku_ft_maybe_add_txn_ref(FT h, TOKUTXN txn) {
-    BOOL ref_added = FALSE;
-    OMTVALUE txnv;
-    u_int32_t index;
-    toku_ft_grab_reflock(h);
-    // Does brt already know about transaction txn?
-    int r = toku_omt_find_zero(h->txns, find_xid, txn, &txnv, &index);
-    if (r==0) {
-        // It's already there.
-        assert((TOKUTXN)txnv==txn);
-        ref_added = FALSE;
-        goto exit;
-    }
-    // Otherwise it's not there.
-    // Insert reference to transaction into brt
-    r = toku_omt_insert_at(h->txns, txn, index);
+// Insert reference to transaction into ft
+void
+toku_ft_add_txn_ref(FT ft, TOKUTXN txn) {
+    toku_ft_grab_reflock(ft);
+    uint32_t idx;
+    int r = toku_omt_insert(ft->txns, txn, find_xid, txn, &idx);
     assert(r==0);
-    ref_added = TRUE;
-exit:
-    toku_ft_release_reflock(h);
-    return ref_added;
+    toku_ft_release_reflock(ft);
 }
 
 static void

ft/ft.h

@@ -12,6 +12,7 @@
 #include "cachetable.h"
 #include "log.h"
 #include "ft-search.h"
+#include "ft-ops.h"
 #include "compress.h"
 
 // remove a ft, transactionless.
@@ -68,8 +69,8 @@ void toku_reset_root_xid_that_created(FT h, TXNID new_root_xid_that_created);
 // This redefines which xid created the dictionary.
 
 
-BOOL
-toku_ft_maybe_add_txn_ref(FT h, TOKUTXN txn);
+void
+toku_ft_add_txn_ref(FT h, TOKUTXN txn);
 void
 toku_ft_remove_txn_ref(FT h, TOKUTXN txn);
 

ft/log-internal.h

@@ -104,24 +104,7 @@ struct tokulogger {
 int toku_logger_find_next_unused_log_file(const char *directory, long long *result);
 int toku_logger_find_logfiles (const char *directory, char ***resultp, int *n_logfiles);
 
-struct tokutxn {
-    u_int64_t txnid64; /* this happens to be the first lsn */
-    u_int64_t ancestor_txnid64; /* this is the lsn of root transaction */
-    u_int64_t snapshot_txnid64; /* this is the lsn of the snapshot */
-    TOKULOGGER logger;
-    TOKUTXN    parent;
-    DB_TXN*    container_db_txn;  // reference to DB_TXN that contains this tokutxn
-    time_t     starttime;         // timestamp in seconds of transaction start
-
-    u_int64_t  rollentry_raw_count;  // the total count of every byte in the transaction and all its children.
-    OMT        open_fts; // a collection of the brts that we touched.  Indexed by filenum.
-    TXN_SNAPSHOT_TYPE snapshot_type;
-    OMT        live_root_txn_list; // the root txns live when the root ancestor (self if a root) started
-    XIDS       xids;      //Represents the xid list
-    BOOL       force_fsync_on_commit;  //This transaction NEEDS an fsync once (if) it commits.  (commit means root txn)
-    TXN_PROGRESS_POLL_FUNCTION progress_poll_fun;
-    void *                     progress_poll_fun_extra;
-
+struct txn_roll_info {
     // these are number of rollback nodes and rollback entries for this txn.
     //
     // the current rollback node below has sequence number num_rollback_nodes - 1
@@ -134,6 +117,7 @@ struct tokutxn {
     uint64_t num_rollback_nodes;
     uint64_t num_rollentries;
     uint64_t num_rollentries_processed;
+    uint64_t rollentry_raw_count;  // the total count of every byte in the transaction and all its children.
 
     // spilled rollback nodes are rollback nodes that were gorged by this
     // transaction, retired, and saved in a list.
@@ -150,16 +134,56 @@ struct tokutxn {
     // then we need to create one and set it here before using it.
     BLOCKNUM current_rollback; 
     uint32_t current_rollback_hash;
+};
 
-    BOOL       recovered_from_checkpoint;
+struct tokutxn {
+    // These don't change after create:
+    const time_t starttime; // timestamp in seconds of transaction start
+    const u_int64_t txnid64; // this happens to be the first lsn
+    const u_int64_t ancestor_txnid64; // this is the lsn of root transaction
+    const u_int64_t snapshot_txnid64; // this is the lsn of the snapshot
+    const TXN_SNAPSHOT_TYPE snapshot_type;
+    const BOOL recovered_from_checkpoint;
+    const TOKULOGGER logger;
+    const TOKUTXN parent;
+    // These don't either but they're created in a way that's hard to make
+    // strictly const.
+    DB_TXN *container_db_txn; // reference to DB_TXN that contains this tokutxn
+    OMT live_root_txn_list; // the root txns live when the root ancestor (self if a root) started.
+    XIDS xids; // Represents the xid list
+
+    // These are not read until a commit, prepare, or abort starts, and
+    // they're "monotonic" (only go false->true) during operation:
     BOOL checkpoint_needed_before_commit;
-    TOKUTXN_STATE state;
-    LSN        do_fsync_lsn;
     BOOL do_fsync;
+    BOOL force_fsync_on_commit;  //This transaction NEEDS an fsync once (if) it commits.  (commit means root txn)
+
+    // Not used until commit, prepare, or abort starts:
+    LSN do_fsync_lsn;
     TOKU_XA_XID xa_xid; // for prepared transactions
+    TXN_PROGRESS_POLL_FUNCTION progress_poll_fun;
+    void *progress_poll_fun_extra;
+
+    toku_mutex_t txn_lock;
+    // Protected by the txn lock:
+    OMT open_fts; // a collection of the brts that we touched.  Indexed by filenum.
+    struct txn_roll_info roll_info; // Info used to manage rollback entries
+
+    // Protected by the txn manager lock:
+    TOKUTXN_STATE state;
     struct toku_list prepared_txns_link; // list of prepared transactions
 };
 
+static inline int
+txn_has_current_rollback_log(TOKUTXN txn) {
+    return txn->roll_info.current_rollback.b != ROLLBACK_NONE.b;
+}
+
+static inline int
+txn_has_spilled_rollback_logs(TOKUTXN txn) {
+    return txn->roll_info.spilled_rollback_tail.b != ROLLBACK_NONE.b;
+}
+
 struct txninfo {
     uint64_t   rollentry_raw_count;  // the total count of every byte in the transaction and all its children.
     uint32_t   num_fts;

ft/logformat.c

@@ -573,6 +573,7 @@ generate_rollbacks (void) {
 
 		    fprintf(hf, ");\n");
 		    fprintf(cf, ") {\n");
+                    fprintf(cf, "  toku_txn_lock(txn);\n");
                     fprintf(cf, "  ROLLBACK_LOG_NODE log;\n");
                     fprintf(cf, "  toku_get_and_pin_rollback_log_for_new_entry(txn, &log);\n");
 		    // 'memdup' all BYTESTRINGS here
@@ -601,19 +602,20 @@ generate_rollbacks (void) {
 		    fprintf(cf, "  struct roll_entry *v;\n");
 		    fprintf(cf, "  size_t mem_needed = sizeof(v->u.%s) + __builtin_offsetof(struct roll_entry, u.%s);\n", lt->name, lt->name);
 		    fprintf(cf, "  v = toku_malloc_in_rollback(log, mem_needed);\n");
-		    fprintf(cf, "  if (v==0) return errno;\n");
+		    fprintf(cf, "  assert(v);\n");
 		    fprintf(cf, "  v->cmd = (enum rt_cmd)%u;\n", lt->command_and_flags&0xff);
 		    DO_FIELDS(field_type, lt, fprintf(cf, "  v->u.%s.%s = %s;\n", lt->name, field_type->name, field_type->name));
 		    fprintf(cf, "  v->prev = log->newest_logentry;\n");
 		    fprintf(cf, "  if (log->oldest_logentry==NULL) log->oldest_logentry=v;\n");
 		    fprintf(cf, "  log->newest_logentry = v;\n");
 		    fprintf(cf, "  log->rollentry_resident_bytecount += rollback_fsize;\n");
-		    fprintf(cf, "  txn->rollentry_raw_count          += rollback_fsize;\n");
-                    fprintf(cf, "  txn->num_rollentries++;\n");
+		    fprintf(cf, "  txn->roll_info.rollentry_raw_count          += rollback_fsize;\n");
+                    fprintf(cf, "  txn->roll_info.num_rollentries++;\n");
                     fprintf(cf, "  log->dirty = TRUE;\n");
 		    fprintf(cf, "  // spill and unpin assert success internally\n");
 		    fprintf(cf, "  toku_maybe_spill_rollbacks(txn, log);\n");
 		    fprintf(cf, "  toku_rollback_log_unpin(txn, log);\n");
+                    fprintf(cf, "  toku_txn_unlock(txn);\n");
 		    fprintf(cf, "  return 0;\n}\n");
 	    });
 

ft/recover.c

@@ -480,7 +480,7 @@ recover_transaction(TOKUTXN *txnp, TXNID xid, TXNID parentxid, TOKULOGGER logger
         assert(r == 0);
         assert(txn==NULL);
     }
-    r = toku_txn_begin_with_xid(parent, &txn, logger, xid, TXN_SNAPSHOT_NONE, NULL);
+    r = toku_txn_begin_with_xid(parent, &txn, logger, xid, TXN_SNAPSHOT_NONE, NULL, true);
     assert(r == 0);
     if (txnp) *txnp = txn;
     return 0;
@@ -623,10 +623,9 @@ static int toku_recover_suppress_rollback (struct logtype_suppress_rollback *UU(
         r = toku_txnid2txn(renv->logger, l->xid, &txn);
         assert(r == 0);
         assert(txn!=NULL);
-        FT h = tuple->ft_handle->ft;
-        toku_ft_suppress_rollbacks(h, txn);
-	r = toku_txn_note_ft(txn, tuple->ft_handle->ft);
-	assert(r==0);
+        FT ft = tuple->ft_handle->ft;
+        toku_ft_suppress_rollbacks(ft, txn);
+        toku_txn_maybe_note_ft(txn, ft);
     }
     return 0;
 }
@@ -884,8 +883,7 @@ static int toku_recover_enq_insert (struct logtype_enq_insert *l, RECOVER_ENV re
         toku_fill_dbt(&valdbt, l->value.data, l->value.len);
         r = toku_ft_maybe_insert(tuple->ft_handle, &keydbt, &valdbt, txn, TRUE, l->lsn, FALSE, FT_INSERT);
         assert(r == 0);
-	r = toku_txn_note_ft(txn, tuple->ft_handle->ft);
-	assert(r == 0);
+        toku_txn_maybe_note_ft(txn, tuple->ft_handle->ft);
     }
     return 0;
 }

ft/roll.c

@@ -7,6 +7,7 @@
 /* rollback and rollforward routines. */
 
 #include "includes.h"
+#include "rollback-apply.h"
 #include "xids.h"
 
 // functionality provided by roll.c is exposed by an autogenerated

ft/rollback-apply.c

+/* -*- mode: C; c-basic-offset: 4; indent-tabs-mode: nil -*- */
+// vim: expandtab:ts=8:sw=4:softtabstop=4:
+#ident "$Id$"
+#ident "Copyright (c) 2007-2012 Tokutek Inc.  All rights reserved."
+#ident "The technology is licensed by the Massachusetts Institute of Technology, Rutgers State University of New Jersey, and the Research Foundation of State University of New York at Stony Brook under United States of America Serial No. 11/760379 and to the patents and/or patent applications resulting from it."
+
+#include "fttypes.h"
+#include "log-internal.h"
+#include "rollback-apply.h"
+
+static void
+poll_txn_progress_function(TOKUTXN txn, uint8_t is_commit, uint8_t stall_for_checkpoint) {
+    if (txn->progress_poll_fun) {
+        TOKU_TXN_PROGRESS_S progress = {
+            .entries_total     = txn->roll_info.num_rollentries,
+            .entries_processed = txn->roll_info.num_rollentries_processed,
+            .is_commit = is_commit,
+            .stalled_on_checkpoint = stall_for_checkpoint};
+        txn->progress_poll_fun(&progress, txn->progress_poll_fun_extra);
+    }
+}
+
+int toku_commit_rollback_item (TOKUTXN txn, struct roll_entry *item, LSN lsn) {
+    int r=0;
+    rolltype_dispatch_assign(item, toku_commit_, r, txn, lsn);
+    txn->roll_info.num_rollentries_processed++;
+    if (txn->roll_info.num_rollentries_processed % 1024 == 0) {
+        poll_txn_progress_function(txn, TRUE, FALSE);
+    }
+    return r;
+}
+
+int toku_abort_rollback_item (TOKUTXN txn, struct roll_entry *item, LSN lsn) {
+    int r=0;
+    rolltype_dispatch_assign(item, toku_rollback_, r, txn, lsn);
+    txn->roll_info.num_rollentries_processed++;
+    if (txn->roll_info.num_rollentries_processed % 1024 == 0) {
+        poll_txn_progress_function(txn, FALSE, FALSE);
+    }
+    return r;
+}
+
+static int
+note_ft_used_in_txns_parent(OMTVALUE ftv, u_int32_t UU(index), void *txnv) {
+    TOKUTXN child = txnv;
+    TOKUTXN parent = child->parent;
+    FT ft = ftv;
+    toku_txn_maybe_note_ft(parent, ft);
+    if (ft->txnid_that_created_or_locked_when_empty == toku_txn_get_txnid(child)) {
+        //Pass magic "no rollback needed" flag to parent.
+        ft->txnid_that_created_or_locked_when_empty = toku_txn_get_txnid(parent);
+    }
+    if (ft->txnid_that_suppressed_recovery_logs == toku_txn_get_txnid(child)) {
+        //Pass magic "no recovery needed" flag to parent.
+        ft->txnid_that_suppressed_recovery_logs = toku_txn_get_txnid(parent);
+    }
+    return 0;
+}
+
+static int
+apply_txn(TOKUTXN txn, LSN lsn, apply_rollback_item func) {
+    int r = 0;
+    // do the commit/abort calls and free everything
+    // we do the commit/abort calls in reverse order too.
+    struct roll_entry *item;
+    //printf("%s:%d abort\n", __FILE__, __LINE__);
+
+    BLOCKNUM next_log      = ROLLBACK_NONE;
+    uint32_t next_log_hash = 0;
+
+    BOOL is_current = FALSE;
+    if (txn_has_current_rollback_log(txn)) {
+        next_log      = txn->roll_info.current_rollback;
+        next_log_hash = txn->roll_info.current_rollback_hash;
+        is_current = TRUE;
+    }
+    else if (txn_has_spilled_rollback_logs(txn)) {
+        next_log      = txn->roll_info.spilled_rollback_tail;
+        next_log_hash = txn->roll_info.spilled_rollback_tail_hash;
+    }
+
+    uint64_t last_sequence = txn->roll_info.num_rollback_nodes;
+    BOOL found_head = FALSE;
+    while (next_log.b != ROLLBACK_NONE.b) {
+        ROLLBACK_LOG_NODE log;
+        //pin log
+        toku_get_and_pin_rollback_log(txn, next_log, next_log_hash, &log);
+        toku_rollback_verify_contents(log, txn->txnid64, last_sequence - 1);
+
+        toku_maybe_prefetch_previous_rollback_log(txn, log);
+
+        last_sequence = log->sequence;
+        if (func) {
+            while ((item=log->newest_logentry)) {
+                log->newest_logentry = item->prev;
+                r = func(txn, item, lsn);
+                if (r!=0) return r;
+            }
+        }
+        if (next_log.b == txn->roll_info.spilled_rollback_head.b) {
+            assert(!found_head);
+            found_head = TRUE;
+            assert(log->sequence == 0);
+        }
+        next_log      = log->previous;
+        next_log_hash = log->previous_hash;
+        {
+            //Clean up transaction structure to prevent
+            //toku_txn_close from double-freeing
+            if (is_current) {
+                txn->roll_info.current_rollback      = ROLLBACK_NONE;
+                txn->roll_info.current_rollback_hash = 0;
+                is_current = FALSE;
+            }
+            else {
+                txn->roll_info.spilled_rollback_tail      = next_log;
+                txn->roll_info.spilled_rollback_tail_hash = next_log_hash;
+            }
+            if (found_head) {
+                assert(next_log.b == ROLLBACK_NONE.b);
+                txn->roll_info.spilled_rollback_head      = next_log;
+                txn->roll_info.spilled_rollback_head_hash = next_log_hash;
+            }
+        }
+        toku_rollback_log_unpin_and_remove(txn, log);
+    }
+    return r;
+}
+
+//Commit each entry in the rollback log.
+//If the transaction has a parent, it just promotes its information to its parent.
+int toku_rollback_commit(TOKUTXN txn, LSN lsn) {
+    int r=0;
+    if (txn->parent!=0) {
+        // First we must put a rollinclude entry into the parent if we spilled
+
+        if (txn_has_spilled_rollback_logs(txn)) {
+            uint64_t num_nodes = txn->roll_info.num_rollback_nodes;
+            if (txn_has_current_rollback_log(txn)) {
+                num_nodes--; //Don't count the in-progress rollback log.
+            }
+            r = toku_logger_save_rollback_rollinclude(txn->parent, txn->txnid64, num_nodes,
+                                                      txn->roll_info.spilled_rollback_head, txn->roll_info.spilled_rollback_head_hash,
+                                                      txn->roll_info.spilled_rollback_tail, txn->roll_info.spilled_rollback_tail_hash);
+            if (r!=0) return r;
+            //Remove ownership from child.
+            txn->roll_info.spilled_rollback_head      = ROLLBACK_NONE; 
+            txn->roll_info.spilled_rollback_head_hash = 0; 
+            txn->roll_info.spilled_rollback_tail      = ROLLBACK_NONE; 
+            txn->roll_info.spilled_rollback_tail_hash = 0; 
+        }
+        // if we're commiting a child rollback, put its entries into the parent
+        // by pinning both child and parent and then linking the child log entry
+        // list to the end of the parent log entry list.
+        if (txn_has_current_rollback_log(txn)) {
+            //Pin parent log
+            ROLLBACK_LOG_NODE parent_log;
+            toku_get_and_pin_rollback_log_for_new_entry(txn->parent, &parent_log);
+
+            //Pin child log
+            ROLLBACK_LOG_NODE child_log;
+            toku_get_and_pin_rollback_log(txn, txn->roll_info.current_rollback, 
+                    txn->roll_info.current_rollback_hash, &child_log);
+            toku_rollback_verify_contents(child_log, txn->txnid64, txn->roll_info.num_rollback_nodes - 1);
+
+            // Append the list to the front of the parent.
+            if (child_log->oldest_logentry) {
+                // There are some entries, so link them in.
+                child_log->oldest_logentry->prev = parent_log->newest_logentry;
+                if (!parent_log->oldest_logentry) {
+                    parent_log->oldest_logentry = child_log->oldest_logentry;
+                }
+                parent_log->newest_logentry = child_log->newest_logentry;
+                parent_log->rollentry_resident_bytecount += child_log->rollentry_resident_bytecount;
+                txn->parent->roll_info.rollentry_raw_count         += txn->roll_info.rollentry_raw_count;
+                child_log->rollentry_resident_bytecount = 0;
+            }
+            if (parent_log->oldest_logentry==NULL) {
+                parent_log->oldest_logentry = child_log->oldest_logentry;
+            }
+            child_log->newest_logentry = child_log->oldest_logentry = 0;
+            // Put all the memarena data into the parent.
+            if (memarena_total_size_in_use(child_log->rollentry_arena) > 0) {
+                // If there are no bytes to move, then just leave things alone, and let the memory be reclaimed on txn is closed.
+                memarena_move_buffers(parent_log->rollentry_arena, child_log->rollentry_arena);
+            }
+            toku_rollback_log_unpin_and_remove(txn, child_log);
+            txn->roll_info.current_rollback = ROLLBACK_NONE;
+            txn->roll_info.current_rollback_hash = 0;
+
+            toku_maybe_spill_rollbacks(txn->parent, parent_log);
+            toku_rollback_log_unpin(txn->parent, parent_log);
+            assert(r == 0);
+        }
+
+        // Note the open brts, the omts must be merged
+        r = toku_omt_iterate(txn->open_fts, note_ft_used_in_txns_parent, txn);
+        assert(r==0);
+
+        // Merge the list of headers that must be checkpointed before commit
+        if (txn->checkpoint_needed_before_commit) {
+            txn->parent->checkpoint_needed_before_commit = TRUE;
+        }
+
+        //If this transaction needs an fsync (if it commits)
+        //save that in the parent.  Since the commit really happens in the root txn.
+        txn->parent->force_fsync_on_commit |= txn->force_fsync_on_commit;
+        txn->parent->roll_info.num_rollentries       += txn->roll_info.num_rollentries;
+    } else {
+        r = apply_txn(txn, lsn, toku_commit_rollback_item);
+        assert(r==0);
+    }
+
+    return r;
+}
+
+int toku_rollback_abort(TOKUTXN txn, LSN lsn) {
+    int r;
+    r = apply_txn(txn, lsn, toku_abort_rollback_item);
+    assert(r==0);
+    return r;
+}

ft/rollback-apply.h

+/* -*- mode: C; c-basic-offset: 4; indent-tabs-mode: nil -*- */
+// vim: expandtab:ts=8:sw=4:softtabstop=4:
+#ifndef ROLLBACK_APPLY_H
+#define ROLLBACK_APPLY_H
+
+#ident "$Id$"
+#ident "Copyright (c) 2007-2012 Tokutek Inc.  All rights reserved."
+#ident "The technology is licensed by the Massachusetts Institute of Technology, Rutgers State University of New Jersey, and the Research Foundation of State University of New York at Stony Brook under United States of America Serial No. 11/760379 and to the patents and/or patent applications resulting from it."
+
+#if defined(__cplusplus) || defined(__cilkplusplus)
+extern "C" {
+#endif
+
+typedef int(*apply_rollback_item)(TOKUTXN txn, struct roll_entry *item, LSN lsn);
+int toku_commit_rollback_item (TOKUTXN txn, struct roll_entry *item, LSN lsn);
+int toku_abort_rollback_item (TOKUTXN txn, struct roll_entry *item, LSN lsn);
+
+int toku_rollback_commit(TOKUTXN txn, LSN lsn);
+int toku_rollback_abort(TOKUTXN txn, LSN lsn);
+
+#if defined(__cplusplus) || defined(__cilkplusplus)
+};
+#endif
+
+#endif // ROLLBACK_APPLY_H

ft/rollback-ct-callbacks.c

+/* -*- mode: C; c-basic-offset: 4; indent-tabs-mode: nil -*- */
+// vim: expandtab:ts=8:sw=4:softtabstop=4:
+#ident "$Id$"
+#ident "Copyright (c) 2007-2012 Tokutek Inc.  All rights reserved."
+#ident "The technology is licensed by the Massachusetts Institute of Technology, Rutgers State University of New Jersey, and the Research Foundation of State University of New York at Stony Brook under United States of America Serial No. 11/760379 and to the patents and/or patent applications resulting from it."
+
+#include "rollback-ct-callbacks.h"
+
+#include <toku_portability.h>
+#include <memory.h>
+#include "ft-internal.h"
+#include "fttypes.h"
+#include "memarena.h"
+#include "rollback.h"
+
+
+// Cleanup the rollback memory
+static void
+rollback_log_destroy(ROLLBACK_LOG_NODE log) {
+    memarena_close(&log->rollentry_arena);
+    toku_free(log);
+}
+
+// Write something out.  Keep trying even if partial writes occur.
+// On error: Return negative with errno set.
+// On success return nbytes.
+void toku_rollback_flush_callback (CACHEFILE cachefile, int fd, BLOCKNUM logname,
+                                          void *rollback_v,  void** UU(disk_data), void *extraargs, PAIR_ATTR size, PAIR_ATTR* new_size,
+                                          BOOL write_me, BOOL keep_me, BOOL for_checkpoint, BOOL UU(is_clone)) {
+    int r;
+    ROLLBACK_LOG_NODE  log = rollback_v;
+    FT h   = extraargs;
+
+    assert(h->cf == cachefile);
+    assert(log->blocknum.b==logname.b);
+
+    if (write_me && !h->panic) {
+        int n_workitems, n_threads; 
+        toku_cachefile_get_workqueue_load(cachefile, &n_workitems, &n_threads);
+
+        r = toku_serialize_rollback_log_to(fd, log->blocknum, log, h, n_workitems, n_threads, for_checkpoint);
+        if (r) {
+            if (h->panic==0) {
+                char *e = strerror(r);
+                int   l = 200 + strlen(e);
+                char s[l];
+                h->panic=r;
+                snprintf(s, l-1, "While writing data to disk, error %d (%s)", r, e);
+                h->panic_string = toku_strdup(s);
+            }
+        }
+    }
+    *new_size = size;
+    if (!keep_me) {
+        rollback_log_destroy(log);
+    }
+}
+
+int toku_rollback_fetch_callback (CACHEFILE cachefile, int fd, BLOCKNUM logname, u_int32_t fullhash,
+                                 void **rollback_pv,  void** UU(disk_data), PAIR_ATTR *sizep, int * UU(dirtyp), void *extraargs) {
+    int r;
+    FT h = extraargs;
+    assert(h->cf == cachefile);
+
+    ROLLBACK_LOG_NODE *result = (ROLLBACK_LOG_NODE*)rollback_pv;
+    r = toku_deserialize_rollback_log_from(fd, logname, fullhash, result, h);
+    if (r==0) {
+        *sizep = rollback_memory_size(*result);
+    }
+    return r;
+}
+
+void toku_rollback_pe_est_callback(
+    void* rollback_v, 
+    void* UU(disk_data),
+    long* bytes_freed_estimate, 
+    enum partial_eviction_cost *cost, 
+    void* UU(write_extraargs)
+    )
+{
+    assert(rollback_v != NULL);
+    *bytes_freed_estimate = 0;
+    *cost = PE_CHEAP;
+}
+
+// callback for partially evicting a cachetable entry
+int toku_rollback_pe_callback (
+    void *rollback_v, 
+    PAIR_ATTR UU(old_attr), 
+    PAIR_ATTR* new_attr, 
+    void* UU(extraargs)
+    ) 
+{
+    assert(rollback_v != NULL);
+    *new_attr = old_attr;
+    return 0;
+}
+
+// partial fetch is never required for a rollback log node
+BOOL toku_rollback_pf_req_callback(void* UU(ftnode_pv), void* UU(read_extraargs)) {
+    return FALSE;
+}
+
+// a rollback node should never be partial fetched, 
+// because we always say it is not required.
+// (pf req callback always returns false)
+int toku_rollback_pf_callback(void* UU(ftnode_pv),  void* UU(disk_data), void* UU(read_extraargs), int UU(fd), PAIR_ATTR* UU(sizep)) {
+    assert(FALSE);
+    return 0;
+}
+
+// the cleaner thread should never choose a rollback node for cleaning
+int toku_rollback_cleaner_callback (
+    void* UU(ftnode_pv),
+    BLOCKNUM UU(blocknum),
+    u_int32_t UU(fullhash),
+    void* UU(extraargs)
+    )
+{
+    assert(FALSE);
+    return 0;
+}
+

ft/rollback-ct-callbacks.h

+/* -*- mode: C; c-basic-offset: 4; indent-tabs-mode: nil -*- */
+// vim: expandtab:ts=8:sw=4:softtabstop=4:
+#ifndef ROLLBACK_CT_CALLBACKS_H
+#define ROLLBACK_CT_CALLBACKS_H
+
+#ident "$Id$"
+#ident "Copyright (c) 2007-2012 Tokutek Inc.  All rights reserved."
+#ident "The technology is licensed by the Massachusetts Institute of Technology, Rutgers State University of New Jersey, and the Research Foundation of State University of New York at Stony Brook under United States of America Serial No. 11/760379 and to the patents and/or patent applications resulting from it."
+
+#if defined(__cplusplus) || defined(__cilkplusplus)
+extern "C" {
+#endif
+
+#include "cachetable.h"
+#include "fttypes.h"
+
+void toku_rollback_flush_callback(CACHEFILE cachefile, int fd, BLOCKNUM logname, void *rollback_v, void** UU(disk_data), void *extraargs, PAIR_ATTR size, PAIR_ATTR* new_size, BOOL write_me, BOOL keep_me, BOOL for_checkpoint, BOOL UU(is_clone));
+int toku_rollback_fetch_callback(CACHEFILE cachefile, int fd, BLOCKNUM logname, u_int32_t fullhash, void **rollback_pv,  void** UU(disk_data), PAIR_ATTR *sizep, int * UU(dirtyp), void *extraargs);
+void toku_rollback_pe_est_callback(
+    void* rollback_v, 
+    void* UU(disk_data),
+    long* bytes_freed_estimate, 
+    enum partial_eviction_cost *cost, 
+    void* UU(write_extraargs)
+    );
+int toku_rollback_pe_callback (
+    void *rollback_v, 
+    PAIR_ATTR UU(old_attr), 
+    PAIR_ATTR* new_attr, 
+    void* UU(extraargs)
+    ) ;
+BOOL toku_rollback_pf_req_callback(void* UU(ftnode_pv), void* UU(read_extraargs)) ;
+int toku_rollback_pf_callback(void* UU(ftnode_pv),  void* UU(disk_data), void* UU(read_extraargs), int UU(fd), PAIR_ATTR* UU(sizep));
+int toku_rollback_cleaner_callback (
+    void* UU(ftnode_pv),
+    BLOCKNUM UU(blocknum),
+    u_int32_t UU(fullhash),
+    void* UU(extraargs)
+    );
+
+static inline CACHETABLE_WRITE_CALLBACK get_write_callbacks_for_rollback_log(FT h) {
+    CACHETABLE_WRITE_CALLBACK wc;
+    wc.flush_callback = toku_rollback_flush_callback;
+    wc.pe_est_callback = toku_rollback_pe_est_callback;
+    wc.pe_callback = toku_rollback_pe_callback;
+    wc.cleaner_callback = toku_rollback_cleaner_callback;
+    wc.clone_callback = NULL;
+    wc.write_extraargs = h;
+    return wc;
+}
+
+#if defined(__cplusplus) || defined(__cilkplusplus)
+};
+#endif
+
+#endif // ROLLBACK_CT_CALLBACKS_H

ft/rollback.h

@@ -14,8 +14,6 @@ extern "C" {
 #endif
 
 void toku_poll_txn_progress_function(TOKUTXN txn, uint8_t is_commit, uint8_t stall_for_checkpoint);
-int toku_rollback_commit(TOKUTXN txn, LSN lsn);
-int toku_rollback_abort(TOKUTXN txn, LSN lsn);
 
 // these functions assert internally that they succeed
 
@@ -38,10 +36,6 @@ void toku_maybe_prefetch_previous_rollback_log(TOKUTXN txn, ROLLBACK_LOG_NODE lo
 // unpin and rmove a rollback log from the cachetable
 void toku_rollback_log_unpin_and_remove(TOKUTXN txn, ROLLBACK_LOG_NODE log);
 
-typedef int(*apply_rollback_item)(TOKUTXN txn, struct roll_entry *item, LSN lsn);
-int toku_commit_rollback_item (TOKUTXN txn, struct roll_entry *item, LSN lsn);
-int toku_abort_rollback_item (TOKUTXN txn, struct roll_entry *item, LSN lsn);
-
 void *toku_malloc_in_rollback(ROLLBACK_LOG_NODE log, size_t size);
 void *toku_memdup_in_rollback(ROLLBACK_LOG_NODE log, const void *v, size_t len);
 
@@ -56,12 +50,14 @@ void *toku_memdup_in_rollback(ROLLBACK_LOG_NODE log, const void *v, size_t len);
 // if necessary.
 void toku_maybe_spill_rollbacks(TOKUTXN txn, ROLLBACK_LOG_NODE log);
 
-int toku_txn_note_ft (TOKUTXN txn, FT h);
+void toku_txn_maybe_note_ft (TOKUTXN txn, FT h);
 int toku_logger_txn_rollback_raw_count(TOKUTXN txn, u_int64_t *raw_count);
 
 int toku_find_pair_by_xid (OMTVALUE v, void *txnv);
 int toku_find_xid_by_xid (OMTVALUE v, void *xidv);
 
+PAIR_ATTR rollback_memory_size(ROLLBACK_LOG_NODE log);
+
 // A high-level rollback log is made up of a chain of rollback log nodes.
 // Each rollback log node is represented (separately) in the cachetable by 
 // this structure. Each portion of the rollback log chain has a block num

ft/txn.c

@@ -10,6 +10,7 @@
 #include "checkpoint.h"
 #include "ule.h"
 #include <valgrind/helgrind.h>
+#include "rollback-apply.h"
 #include "txn_manager.h"
 
 ///////////////////////////////////////////////////////////////////////////////////
@@ -50,6 +51,18 @@ toku_txn_get_status(TXN_STATUS s) {
     *s = txn_status;
 }
 
+void
+toku_txn_lock(TOKUTXN txn)
+{
+    toku_mutex_lock(&txn->txn_lock);
+}
+
+void
+toku_txn_unlock(TOKUTXN txn)
+{
+    toku_mutex_unlock(&txn->txn_lock);
+}
+
 int 
 toku_txn_begin_txn (
     DB_TXN  *container_db_txn,
@@ -59,7 +72,7 @@ toku_txn_begin_txn (
     TXN_SNAPSHOT_TYPE snapshot_type
     ) 
 {
-    int r = toku_txn_begin_with_xid(parent_tokutxn, tokutxn, logger, TXNID_NONE, snapshot_type, container_db_txn);
+    int r = toku_txn_begin_with_xid(parent_tokutxn, tokutxn, logger, TXNID_NONE, snapshot_type, container_db_txn, false);
     return r;
 }
 
@@ -70,14 +83,11 @@ toku_txn_begin_with_xid (
     TOKULOGGER logger, 
     TXNID xid, 
     TXN_SNAPSHOT_TYPE snapshot_type,
-    DB_TXN *container_db_txn
+    DB_TXN *container_db_txn,
+    bool for_recovery
     ) 
 {
-    int r = toku_txn_create_txn(tokutxn, parent_tokutxn, logger, xid, snapshot_type, container_db_txn);
-    if (r == 0) {
-        toku_txn_manager_start_txn((*tokutxn)->logger->txn_manager, *tokutxn);
-    }
-    return r;
+    return toku_txn_manager_start_txn(tokutxn, logger->txn_manager, parent_tokutxn, logger, xid, snapshot_type, container_db_txn, for_recovery);
 }
 
 DB_TXN *
@@ -102,45 +112,71 @@ toku_txn_create_txn (
     TOKULOGGER logger, 
     TXNID xid, 
     TXN_SNAPSHOT_TYPE snapshot_type,
-    DB_TXN *container_db_txn
+    XIDS xids,
+    DB_TXN *container_db_txn,
+    bool for_checkpoint
     ) 
 {
-    if (logger->is_panicked) return EINVAL;
+    if (logger->is_panicked) {
+        return EINVAL;
+    }
     assert(logger->rollback_cachefile);
-    TOKUTXN XMALLOC(result);
-    result->starttime = time(NULL);  // getting timestamp in seconds is a cheap call
-    int r;
-    r = toku_omt_create(&result->open_fts);
-    assert_zero(r);
 
-    result->logger = logger;
-    result->parent = parent_tokutxn;
-    result->num_rollentries = 0;
-    result->num_rollentries_processed = 0;
-    result->progress_poll_fun = NULL;
-    result->progress_poll_fun_extra = NULL;
-    result->spilled_rollback_head      = ROLLBACK_NONE;
-    result->spilled_rollback_tail      = ROLLBACK_NONE;
-    result->spilled_rollback_head_hash = 0;
-    result->spilled_rollback_tail_hash = 0;
-    result->current_rollback      = ROLLBACK_NONE;
-    result->current_rollback_hash = 0;
-    result->num_rollback_nodes = 0;
-    result->snapshot_type = snapshot_type;
-    result->snapshot_txnid64 = TXNID_NONE;
-    result->container_db_txn = container_db_txn;
-
-    result->rollentry_raw_count = 0;
-    result->force_fsync_on_commit = FALSE;
-    result->recovered_from_checkpoint = FALSE;
-    result->checkpoint_needed_before_commit = FALSE;
-    result->state = TOKUTXN_LIVE;
-    invalidate_xa_xid(&result->xa_xid);
-    result->do_fsync = FALSE;
+    TXNID snapshot_txnid64;
+    if (snapshot_type == TXN_SNAPSHOT_NONE) {
+        snapshot_txnid64 = TXNID_NONE;
+    } else if (parent_tokutxn == NULL || snapshot_type == TXN_SNAPSHOT_CHILD) {
+        snapshot_txnid64 = xid;
+    } else if (snapshot_type == TXN_SNAPSHOT_ROOT) {
+        snapshot_txnid64 = parent_tokutxn->snapshot_txnid64;
+    } else {
+        assert(false);
+    }
 
-    result->txnid64 = xid;
-    result->xids = NULL;
+    OMT open_fts;
+    {
+        int r = toku_omt_create(&open_fts);
+        assert_zero(r);
+    }
 
+    struct txn_roll_info roll_info = {
+        .num_rollback_nodes = 0,
+        .num_rollentries = 0,
+        .num_rollentries_processed = 0,
+        .rollentry_raw_count = 0,
+        .spilled_rollback_head = ROLLBACK_NONE,
+        .spilled_rollback_tail = ROLLBACK_NONE,
+        .spilled_rollback_head_hash = 0,
+        .spilled_rollback_tail_hash = 0,
+        .current_rollback = ROLLBACK_NONE,
+        .current_rollback_hash = 0,
+    };
+
+    struct tokutxn new_txn = {
+        .starttime = time(NULL),
+        .open_fts = open_fts,
+        .logger = logger,
+        .parent = parent_tokutxn,
+        .progress_poll_fun = NULL,
+        .progress_poll_fun_extra = NULL,
+        .snapshot_type = snapshot_type,
+        .snapshot_txnid64 = snapshot_txnid64,
+        .container_db_txn = container_db_txn,
+        .force_fsync_on_commit = FALSE,
+        .recovered_from_checkpoint = for_checkpoint,
+        .checkpoint_needed_before_commit = FALSE,
+        .state = TOKUTXN_LIVE,
+        .do_fsync = FALSE,
+        .txnid64 = xid,
+        .ancestor_txnid64 = (parent_tokutxn ? parent_tokutxn->ancestor_txnid64 : xid),
+        .xids = xids,
+        .roll_info = roll_info
+    };
+
+
+    TOKUTXN result = toku_xmemdup(&new_txn, sizeof new_txn);
+    toku_mutex_init(&result->txn_lock, NULL);
+    invalidate_xa_xid(&result->xa_xid);
     *tokutxn = result;
 
     STATUS_VALUE(TXN_BEGIN)++;
@@ -154,31 +190,27 @@ toku_txn_create_txn (
 //Used on recovery to recover a transaction.
 int
 toku_txn_load_txninfo (TOKUTXN txn, TXNINFO info) {
-#define COPY_FROM_INFO(field) txn->field = info->field
-    COPY_FROM_INFO(rollentry_raw_count);
+    txn->roll_info.rollentry_raw_count = info->rollentry_raw_count;
     uint32_t i;
     for (i = 0; i < info->num_fts; i++) {
-        FT h = info->open_fts[i];
-        int r = toku_txn_note_ft(txn, h);
-        assert_zero(r);
+        FT ft = info->open_fts[i];
+        toku_txn_maybe_note_ft(txn, ft);
     }
-    COPY_FROM_INFO(force_fsync_on_commit );
-    COPY_FROM_INFO(num_rollback_nodes);
-    COPY_FROM_INFO(num_rollentries);
+    txn->force_fsync_on_commit = info->force_fsync_on_commit;
+    txn->roll_info.num_rollback_nodes = info->num_rollback_nodes;
+    txn->roll_info.num_rollentries = info->num_rollentries;
 
     CACHEFILE rollback_cachefile = txn->logger->rollback_cachefile;
 
-    COPY_FROM_INFO(spilled_rollback_head);
-    txn->spilled_rollback_head_hash = toku_cachetable_hash(rollback_cachefile,
-                                                           txn->spilled_rollback_head);
-    COPY_FROM_INFO(spilled_rollback_tail);
-    txn->spilled_rollback_tail_hash = toku_cachetable_hash(rollback_cachefile,
-                                                           txn->spilled_rollback_tail);
-    COPY_FROM_INFO(current_rollback);
-    txn->current_rollback_hash = toku_cachetable_hash(rollback_cachefile,
-                                                      txn->current_rollback);
-#undef COPY_FROM_INFO
-    txn->recovered_from_checkpoint = TRUE;
+    txn->roll_info.spilled_rollback_head = info->spilled_rollback_head;
+    txn->roll_info.spilled_rollback_head_hash = toku_cachetable_hash(rollback_cachefile,
+                                                           txn->roll_info.spilled_rollback_head);
+    txn->roll_info.spilled_rollback_tail = info->spilled_rollback_tail;
+    txn->roll_info.spilled_rollback_tail_hash = toku_cachetable_hash(rollback_cachefile,
+                                                           txn->roll_info.spilled_rollback_tail);
+    txn->roll_info.current_rollback = info->current_rollback;
+    txn->roll_info.current_rollback_hash = toku_cachetable_hash(rollback_cachefile,
+                                                      txn->roll_info.current_rollback);
     return 0;
 }
 
@@ -223,7 +255,7 @@ int toku_txn_commit_with_lsn(TOKUTXN txn, int nosync, LSN oplsn,
     // recovery to properly recommit this transaction if the commit 
     // does not make it to disk. In the case of MySQL, that would be the
     // binary log.
-    txn->do_fsync = !txn->parent && (txn->force_fsync_on_commit || (!nosync && txn->num_rollentries>0));
+    txn->do_fsync = !txn->parent && (txn->force_fsync_on_commit || (!nosync && txn->roll_info.num_rollentries>0));
 
     txn->progress_poll_fun = poll;
     txn->progress_poll_fun_extra = poll_extra;
@@ -274,7 +306,7 @@ int toku_txn_prepare_txn (TOKUTXN txn, TOKU_XA_XID *xa_xid) {
     if (txn->parent) return 0; // nothing to do if there's a parent.
     toku_txn_manager_add_prepared_txn(txn->logger->txn_manager, txn);
     // Do we need to do an fsync?
-    txn->do_fsync = (txn->force_fsync_on_commit || txn->num_rollentries>0);
+    txn->do_fsync = (txn->force_fsync_on_commit || txn->roll_info.num_rollentries>0);
     copy_xid(&txn->xa_xid, xa_xid);
     // This list will go away with #4683, so we wn't need the ydb lock for this anymore.
     return toku_log_xprepare(txn->logger, &txn->do_fsync_lsn, 0, txn->txnid64, xa_xid);
@@ -350,9 +382,9 @@ static void note_txn_closing (TOKUTXN txn) {
 }
 
 void toku_txn_complete_txn(TOKUTXN txn) {
-    assert(txn->spilled_rollback_head.b == ROLLBACK_NONE.b);
-    assert(txn->spilled_rollback_tail.b == ROLLBACK_NONE.b);
-    assert(txn->current_rollback.b == ROLLBACK_NONE.b);
+    assert(txn->roll_info.spilled_rollback_head.b == ROLLBACK_NONE.b);
+    assert(txn->roll_info.spilled_rollback_tail.b == ROLLBACK_NONE.b);
+    assert(txn->roll_info.current_rollback.b == ROLLBACK_NONE.b);
     toku_txn_manager_finish_txn(txn->logger->txn_manager, txn);
     // note that here is another place we depend on
     // this function being called with the multi operation lock
@@ -364,6 +396,7 @@ void toku_txn_destroy_txn(TOKUTXN txn) {
         toku_omt_destroy(&txn->open_fts);
     }
     xids_destroy(&txn->xids);
+    toku_mutex_destroy(&txn->txn_lock);
     toku_free(txn);
 
     STATUS_VALUE(TXN_CLOSE)++;

ft/txn.h

@@ -11,6 +11,9 @@
 extern "C" {
 #endif
 
+void toku_txn_lock(TOKUTXN txn);
+void toku_txn_unlock(TOKUTXN txn);
+
 int toku_txn_begin_txn (
     DB_TXN  *container_db_txn,
     TOKUTXN parent_tokutxn, 
@@ -29,11 +32,12 @@ int toku_txn_begin_with_xid (
     TOKULOGGER logger, 
     TXNID xid, 
     TXN_SNAPSHOT_TYPE snapshot_type,
-    DB_TXN *container_db_txn
+    DB_TXN *container_db_txn,
+    bool for_recovery
     );
 
 // Allocate and initialize a txn
-int toku_txn_create_txn(TOKUTXN *txn_ptr, TOKUTXN parent, TOKULOGGER logger, TXNID xid, TXN_SNAPSHOT_TYPE snapshot_type, DB_TXN *container_db_txn);
+int toku_txn_create_txn(TOKUTXN *txn_ptr, TOKUTXN parent, TOKULOGGER logger, TXNID xid, TXN_SNAPSHOT_TYPE snapshot_type, XIDS xids, DB_TXN *container_db_txn, bool for_checkpoint);
 
 int toku_txn_load_txninfo (TOKUTXN txn, TXNINFO info);
 

ft/txn_manager.c

@@ -305,8 +305,16 @@ live_list_reverse_note_txn_start(TOKUTXN txn) {
     return r;
 }
 
-void toku_txn_manager_start_txn(TXN_MANAGER txn_manager, TOKUTXN txn) {
-    TOKUTXN parent = txn->parent;
+int toku_txn_manager_start_txn(
+    TOKUTXN *txnp,
+    TXN_MANAGER txn_manager,
+    TOKUTXN parent,
+    TOKULOGGER logger,
+    TXNID xid,
+    TXN_SNAPSHOT_TYPE snapshot_type,
+    DB_TXN *container_db_txn,
+    bool for_recovery)
+{
     int r;
     // we take the txn_manager_lock before writing to the log
     // we may be able to move this lock acquisition
@@ -316,20 +324,28 @@ void toku_txn_manager_start_txn(TXN_MANAGER txn_manager, TOKUTXN txn) {
     if (garbage_collection_debug) {
         verify_snapshot_system(txn_manager);
     }
-    if (txn->txnid64 == TXNID_NONE) {
+    if (xid == TXNID_NONE) {
         LSN first_lsn;
-        r = toku_log_xbegin(txn->logger, &first_lsn, 0, parent ? parent->txnid64 : 0);
+        r = toku_log_xbegin(logger, &first_lsn, 0, parent ? parent->txnid64 : 0);
         assert_zero(r);
-        txn->txnid64 = first_lsn.lsn;
+        xid = first_lsn.lsn;
     } 
     XIDS parent_xids;
     if (parent == NULL)
         parent_xids = xids_get_root_xids();
     else
         parent_xids = parent->xids;
-    r = xids_create_child(parent_xids, &txn->xids, txn->txnid64);
+    XIDS xids;
+    r = xids_create_child(parent_xids, &xids, xid);
     assert_zero(r);
 
+    TOKUTXN txn;
+    r = toku_txn_create_txn(&txn, parent, logger, xid, snapshot_type, xids, container_db_txn, for_recovery);
+    if (r != 0) {
+        // logger is panicked
+        return r;
+    }
+
     if (toku_omt_size(txn_manager->live_txns) == 0) {
         assert(txn_manager->oldest_living_xid == TXNID_NONE_LIVING);
         txn_manager->oldest_living_xid = txn->txnid64;
@@ -367,10 +383,6 @@ void toku_txn_manager_start_txn(TXN_MANAGER txn_manager, TOKUTXN txn) {
                 toku_omt_size(txn_manager->live_root_txns)
                 ); //We know it is the newest one.
             assert_zero(r);
-            txn->ancestor_txnid64 = txn->txnid64;
-        }
-        else {
-            txn->ancestor_txnid64 = parent->ancestor_txnid64;
         }
 
         // setup information for snapshot reads
@@ -380,7 +392,6 @@ void toku_txn_manager_start_txn(TXN_MANAGER txn_manager, TOKUTXN txn) {
             if (parent == NULL || txn->snapshot_type == TXN_SNAPSHOT_CHILD) {
                 r = setup_live_root_txn_list(txn_manager, txn);  
                 assert_zero(r);
-                txn->snapshot_txnid64 = txn->txnid64;
                 r = snapshot_txnids_note_txn(txn_manager, txn);
                 assert_zero(r);
                 r = live_list_reverse_note_txn_start(txn);
@@ -390,7 +401,6 @@ void toku_txn_manager_start_txn(TXN_MANAGER txn_manager, TOKUTXN txn) {
             // of the root transaction
             else if (txn->snapshot_type == TXN_SNAPSHOT_ROOT) {
                 txn->live_root_txn_list = parent->live_root_txn_list;
-                txn->snapshot_txnid64 = parent->snapshot_txnid64;
             }
             else {
                 assert(FALSE);
@@ -401,6 +411,9 @@ void toku_txn_manager_start_txn(TXN_MANAGER txn_manager, TOKUTXN txn) {
         verify_snapshot_system(txn_manager);
     }
     toku_mutex_unlock(&txn_manager->txn_manager_lock);
+
+    *txnp = txn;
+    return 0;
 }
 
 // For each xid on the closing txn's live list, find the corresponding entry in the reverse live list.

ft/txn_manager.h

@@ -34,7 +34,16 @@ void toku_txn_manager_destroy(TXN_MANAGER txn_manager);
 TXNID toku_txn_manager_get_oldest_living_xid(TXN_MANAGER txn_manager, time_t * oldest_living_starttime);
 
 // Assign a txnid. Log the txn begin in the recovery log. Initialize the txn live lists.
-void toku_txn_manager_start_txn(TXN_MANAGER txn_manager, TOKUTXN txn);
+// Also, create the txn.
+int toku_txn_manager_start_txn(
+    TOKUTXN *txnp,
+    TXN_MANAGER txn_manager,
+    TOKUTXN parent,
+    TOKULOGGER logger,
+    TXNID xid,
+    TXN_SNAPSHOT_TYPE snapshot_type,
+    DB_TXN *container_db_txn,
+    bool for_recovery);
 void toku_txn_manager_finish_txn(TXN_MANAGER txn_manager, TOKUTXN txn);
 
 void toku_txn_manager_clone_state_for_gc(

src/ydb.c

@@ -520,7 +520,7 @@ maybe_upgrade_persistent_environment_dictionary(DB_ENV * env, DB_TXN * txn, LSN
         const uint32_t curr_env_ver_d = toku_htod32(FT_LAYOUT_VERSION);
         toku_fill_dbt(&key, curr_env_ver_key, strlen(curr_env_ver_key));
         toku_fill_dbt(&val, &curr_env_ver_d, sizeof(curr_env_ver_d));
-        r = toku_db_put(persistent_environment, txn, &key, &val, 0, TRUE);
+        r = toku_db_put(persistent_environment, txn, &key, &val, 0, FALSE);
         assert_zero(r);
 
         // although the variable name is last_lsn_of_v13, this key really represents
@@ -529,19 +529,19 @@ maybe_upgrade_persistent_environment_dictionary(DB_ENV * env, DB_TXN * txn, LSN
         uint64_t last_lsn_of_v13_d = toku_htod64(last_lsn_of_clean_shutdown_read_from_log.lsn);
         toku_fill_dbt(&key, last_lsn_of_v13_key, strlen(last_lsn_of_v13_key));
         toku_fill_dbt(&val, &last_lsn_of_v13_d, sizeof(last_lsn_of_v13_d));
-        r = toku_db_put(persistent_environment, txn, &key, &val, 0, TRUE);
+        r = toku_db_put(persistent_environment, txn, &key, &val, 0, FALSE);
         assert_zero(r);
         
         time_t upgrade_v19_time_d = toku_htod64(time(NULL));
         toku_fill_dbt(&key, upgrade_v19_time_key, strlen(upgrade_v19_time_key));
         toku_fill_dbt(&val, &upgrade_v19_time_d, sizeof(upgrade_v19_time_d));
-        r = toku_db_put(persistent_environment, txn, &key, &val, DB_NOOVERWRITE, TRUE);
+        r = toku_db_put(persistent_environment, txn, &key, &val, DB_NOOVERWRITE, FALSE);
         assert_zero(r);
 
         uint64_t upgrade_v19_footprint_d = toku_htod64(toku_log_upgrade_get_footprint());
         toku_fill_dbt(&key, upgrade_v19_footprint_key, strlen(upgrade_v19_footprint_key));
         toku_fill_dbt(&val, &upgrade_v19_footprint_d, sizeof(upgrade_v19_footprint_d));
-        r = toku_db_put(persistent_environment, txn, &key, &val, DB_NOOVERWRITE, TRUE);
+        r = toku_db_put(persistent_environment, txn, &key, &val, DB_NOOVERWRITE, FALSE);
         assert_zero(r);
     }
     return r;
@@ -959,18 +959,18 @@ env_open(DB_ENV * env, const char *home, u_int32_t flags, int mode) {
 
             toku_fill_dbt(&key, orig_env_ver_key, strlen(orig_env_ver_key));
             toku_fill_dbt(&val, &environment_version, sizeof(environment_version));
-            r = toku_db_put(env->i->persistent_environment, txn, &key, &val, 0, TRUE);
+            r = toku_db_put(env->i->persistent_environment, txn, &key, &val, 0, FALSE);
             assert_zero(r);
 
             toku_fill_dbt(&key, curr_env_ver_key, strlen(curr_env_ver_key));
             toku_fill_dbt(&val, &environment_version, sizeof(environment_version));
-            r = toku_db_put(env->i->persistent_environment, txn, &key, &val, 0, TRUE);
+            r = toku_db_put(env->i->persistent_environment, txn, &key, &val, 0, FALSE);
             assert_zero(r);
 
             time_t creation_time_d = toku_htod64(time(NULL));
             toku_fill_dbt(&key, creation_time_key, strlen(creation_time_key));
             toku_fill_dbt(&val, &creation_time_d, sizeof(creation_time_d));
-            r = toku_db_put(env->i->persistent_environment, txn, &key, &val, 0, TRUE);
+            r = toku_db_put(env->i->persistent_environment, txn, &key, &val, 0, FALSE);
             assert_zero(r);
         }
         else {

src/ydb_txn.c

@@ -488,21 +488,18 @@ toku_txn_begin(DB_ENV *env, DB_TXN * stxn, DB_TXN ** txn, u_int32_t flags) {
             break;
         }
     }
-    r = toku_txn_create_txn(&db_txn_struct_i(result)->tokutxn, 
+    r = toku_txn_manager_start_txn(&db_txn_struct_i(result)->tokutxn,
+                                   toku_logger_get_txn_manager(env->i->logger),
                                    stxn ? db_txn_struct_i(stxn)->tokutxn : 0,
                                    env->i->logger,
                                    TXNID_NONE,
                                    snapshot_type,
-                            result
-                            );
+                                   result,
+                                   false);
     if (r != 0) {
         toku_free(result);
         return r;
     }
-    toku_txn_manager_start_txn(
-        toku_logger_get_txn_manager(env->i->logger),
-        db_txn_struct_i(result)->tokutxn
-        );
 
     //Add to the list of children for the parent.
     if (result->parent) {

src/ydb_write.c

@@ -8,6 +8,7 @@
 #include "ydb-internal.h"
 #include "indexer.h"
 #include <ft/log_header.h>
+#include <ft/checkpoint.h>
 #include "ydb_row_lock.h"
 #include "ydb_write.h"
 #include "ydb_db.h"
@@ -125,7 +126,7 @@ db_put_check_overwrite_constraint(DB *db, DB_TXN *txn, DBT *key,
 
 
 int
-toku_db_del(DB *db, DB_TXN *txn, DBT *key, u_int32_t flags, BOOL holds_ydb_lock) {
+toku_db_del(DB *db, DB_TXN *txn, DBT *key, u_int32_t flags, BOOL holds_mo_lock) {
     HANDLE_PANICKED_DB(db);
     HANDLE_DB_ILLEGAL_WORKING_PARENT_TXN(db, txn);
 
@@ -152,9 +153,9 @@ toku_db_del(DB *db, DB_TXN *txn, DBT *key, u_int32_t flags, BOOL holds_ydb_lock)
     }
     if (r == 0) {
         //Do the actual deleting.
-        if (!holds_ydb_lock) toku_ydb_lock();
+        if (!holds_mo_lock) toku_multi_operation_client_lock();
         r = toku_ft_delete(db->i->ft_handle, key, txn ? db_txn_struct_i(txn)->tokutxn : 0);
-        if (!holds_ydb_lock) toku_ydb_unlock();
+        if (!holds_mo_lock) toku_multi_operation_client_unlock();
     }
 
     if (r == 0) {
@@ -168,7 +169,7 @@ toku_db_del(DB *db, DB_TXN *txn, DBT *key, u_int32_t flags, BOOL holds_ydb_lock)
 
 
 int
-toku_db_put(DB *db, DB_TXN *txn, DBT *key, DBT *val, u_int32_t flags, BOOL holds_ydb_lock) {
+toku_db_put(DB *db, DB_TXN *txn, DBT *key, DBT *val, u_int32_t flags, BOOL holds_mo_lock) {
     HANDLE_PANICKED_DB(db);
     HANDLE_DB_ILLEGAL_WORKING_PARENT_TXN(db, txn);
     int r = 0;
@@ -193,9 +194,9 @@ toku_db_put(DB *db, DB_TXN *txn, DBT *key, DBT *val, u_int32_t flags, BOOL holds
         if (flags==DB_NOOVERWRITE_NO_ERROR) {
             type = FT_INSERT_NO_OVERWRITE;
         }
-        if (!holds_ydb_lock) toku_ydb_lock();
+        if (!holds_mo_lock) toku_multi_operation_client_lock();
         r = toku_ft_maybe_insert(db->i->ft_handle, key, val, ttxn, FALSE, ZERO_LSN, TRUE, type);
-        if (!holds_ydb_lock) toku_ydb_unlock();
+        if (!holds_mo_lock) toku_multi_operation_client_unlock();
     }
 
     if (r == 0) {
@@ -232,10 +233,10 @@ toku_db_update(DB *db, DB_TXN *txn,
     }
 
     TOKUTXN ttxn = txn ? db_txn_struct_i(txn)->tokutxn : NULL;
-    toku_ydb_lock();
+    toku_multi_operation_client_lock();
     r = toku_ft_maybe_update(db->i->ft_handle, key, update_function_extra, ttxn,
                               FALSE, ZERO_LSN, TRUE);
-    toku_ydb_unlock();
+    toku_multi_operation_client_unlock();
 
 cleanup:
     if (r == 0) 
@@ -287,10 +288,10 @@ toku_db_update_broadcast(DB *db, DB_TXN *txn,
     }
 
     TOKUTXN ttxn = txn ? db_txn_struct_i(txn)->tokutxn : NULL;
-    toku_ydb_lock();
+    toku_multi_operation_client_lock();
     r = toku_ft_maybe_update_broadcast(db->i->ft_handle, update_function_extra, ttxn,
                                         FALSE, ZERO_LSN, TRUE, is_resetting_op);
-    toku_ydb_unlock();
+    toku_multi_operation_client_unlock();
 
 cleanup:
     if (r == 0) 

src/ydb_write.h

@@ -35,8 +35,8 @@ typedef struct {
 void ydb_write_layer_get_status(YDB_WRITE_LAYER_STATUS statp);
 
 
-int toku_db_del(DB *db, DB_TXN *txn, DBT *key, u_int32_t flags, BOOL holds_ydb_lock);
-int toku_db_put(DB *db, DB_TXN *txn, DBT *key, DBT *val, u_int32_t flags, BOOL holds_ydb_lock);
+int toku_db_del(DB *db, DB_TXN *txn, DBT *key, u_int32_t flags, BOOL holds_mo_lock);
+int toku_db_put(DB *db, DB_TXN *txn, DBT *key, DBT *val, u_int32_t flags, BOOL holds_mo_lock);
 int autotxn_db_del(DB* db, DB_TXN* txn, DBT* key, u_int32_t flags);
 int autotxn_db_put(DB* db, DB_TXN* txn, DBT* key, DBT* data, u_int32_t flags);
 int autotxn_db_update(DB *db, DB_TXN *txn, const DBT *key, const DBT *update_function_extra, u_int32_t flags);