[t:5064] Closes #5064 Move most of the work of creating xids (including...

[t:5064] Closes #5064 Move most of the work of creating xids (including malloc) out of the txn_manager_lock

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

ft/txn.c

@@ -118,6 +118,7 @@ toku_txn_create_txn (
     TOKULOGGER logger, 
     TXN_SNAPSHOT_TYPE snapshot_type,
     DB_TXN *container_db_txn,
+    XIDS xids,
     bool for_checkpoint
     )
 {
@@ -164,7 +165,7 @@ toku_txn_create_txn (
         .do_fsync_lsn = ZERO_LSN,
         .txnid64 = TXNID_NONE,
         .ancestor_txnid64 = TXNID_NONE,
-        .xids = NULL,
+        .xids = xids,
         .roll_info = roll_info,
         .num_pin = 0
     };
@@ -184,13 +185,12 @@ toku_txn_create_txn (
 }
 
 void
-toku_txn_update_xids_in_txn(TOKUTXN txn, TXNID xid, XIDS xids)
+toku_txn_update_xids_in_txn(TOKUTXN txn, TXNID xid)
 {
     // these should not have been set yet
     invariant(txn->txnid64 == TXNID_NONE);
     invariant(txn->ancestor_txnid64 == TXNID_NONE);
     invariant(txn->snapshot_txnid64 == TXNID_NONE);
-    invariant(txn->xids == NULL);
 
     TXNID snapshot_txnid64;
     if (txn->snapshot_type == TXN_SNAPSHOT_NONE) {
@@ -211,7 +211,6 @@ toku_txn_update_xids_in_txn(TOKUTXN txn, TXNID xid, XIDS xids)
     UNCONST(TXNID, txn->txnid64) = xid;
     UNCONST(TXNID, txn->snapshot_txnid64) = snapshot_txnid64;
     UNCONST(TXNID, txn->ancestor_txnid64) = (txn->parent ? txn->parent->ancestor_txnid64 : xid);
-    txn->xids = xids;
 
 #undef UNCONST
 }

ft/txn.h

@@ -39,8 +39,8 @@ int toku_txn_begin_with_xid (
     );
 
 // Allocate and initialize a txn
-int toku_txn_create_txn(TOKUTXN *txn_ptr, TOKUTXN parent, TOKULOGGER logger, TXN_SNAPSHOT_TYPE snapshot_type, DB_TXN *container_db_txn, bool for_checkpoint);
-void toku_txn_update_xids_in_txn(TOKUTXN txn, TXNID xid, XIDS xids);
+int toku_txn_create_txn(TOKUTXN *txn_ptr, TOKUTXN parent, TOKULOGGER logger, TXN_SNAPSHOT_TYPE snapshot_type, DB_TXN *container_db_txn, XIDS xids, bool for_checkpoint);
+void toku_txn_update_xids_in_txn(TOKUTXN txn, TXNID xid);
 
 int toku_txn_load_txninfo (TOKUTXN txn, TXNINFO info);
 

ft/txn_manager.c

@@ -359,16 +359,22 @@ int toku_txn_manager_start_txn(
     bool for_recovery)
 {
     int r;
+    XIDS xids;
+    TOKUTXN txn;
 
     // Do as much (safe) work as possible before serializing on the txn_manager lock.
     XIDS parent_xids;
-    if (parent == NULL)
+    if (parent == NULL) {
         parent_xids = xids_get_root_xids();
-    else
+    } else {
         parent_xids = parent->xids;
+    }
+    r = xids_create_unknown_child(parent_xids, &xids);
+    if (r != 0) {
+        return r;
+    }
 
-    TOKUTXN txn;
-    r = toku_txn_create_txn(&txn, parent, logger, snapshot_type, container_db_txn, for_recovery);
+    r = toku_txn_create_txn(&txn, parent, logger, snapshot_type, container_db_txn, xids, for_recovery);
     if (r != 0) {
         // logger is panicked
         return r;
@@ -393,11 +399,9 @@ int toku_txn_manager_start_txn(
         invariant(for_recovery);
         txn_manager->last_xid = max_xid(txn_manager->last_xid, xid);
     }
-    XIDS xids;
-    r = xids_create_child(parent_xids, &xids, xid);
-    assert_zero(r);
+    xids_finalize_with_child(txn->xids, xid);
 
-    toku_txn_update_xids_in_txn(txn, xid, xids);
+    toku_txn_update_xids_in_txn(txn, xid);
 
     if (!for_recovery) {
         // TODO(leif): this would be WRONG during recovery.  We cannot

ft/xids.c

@@ -60,37 +60,50 @@ xids_get_root_xids(void) {
 
     XIDS rval = (XIDS)&root_xids;
     return rval;
-} 
+}
 
 
-// xids is immutable.  This function creates a new xids by copying the 
-// parent's list and then appending the xid of the new transaction.
-int
-xids_create_child(XIDS   parent_xids,		// xids list for parent transaction
-		  XIDS * xids_p,		// xids list created
-		  TXNID  this_xid) {		// xid of this transaction (new innermost)
+inline int
+xids_create_unknown_child(XIDS parent_xids, XIDS *xids_p) {
+    // Postcondition:
+    //  xids_p points to an xids that is an exact copy of parent_xids, but with room for one more xid.
     int rval;
     invariant(parent_xids);
-    invariant(this_xid > xids_get_innermost_xid(parent_xids));
     u_int32_t num_child_xids = parent_xids->num_xids + 1;
     invariant(num_child_xids > 0);
     invariant(num_child_xids <= MAX_TRANSACTION_RECORDS);
     if (num_child_xids == MAX_TRANSACTION_RECORDS) rval = EINVAL;
     else {
-        XIDS xids = toku_malloc(sizeof(*xids) + num_child_xids*sizeof(xids->ids[0]));
-        if (!xids) rval = ENOMEM;
-        else {
-            // comment: invariant (num_child_xids <= MAX_TRANSACTION_RECORDS) 
-            // makes this cast ok
-            xids->num_xids = (u_int8_t)num_child_xids;
-            memcpy(xids->ids,
-                   parent_xids->ids,
-                   parent_xids->num_xids*sizeof(parent_xids->ids[0])); 
-            TXNID this_xid_disk = toku_htod64(this_xid);
-            xids->ids[num_child_xids-1] = this_xid_disk;
-            *xids_p = xids;
-            rval = 0;
-        }
+        size_t new_size = sizeof(*parent_xids) + num_child_xids*sizeof(parent_xids->ids[0]);
+        XIDS xids = toku_xmalloc(new_size);
+        // Clone everything (parent does not have the newest xid).
+        memcpy(xids, parent_xids, new_size - sizeof(xids->ids[0]));
+        *xids_p = xids;
+        rval = 0;
+    }
+    return rval;
+}
+
+void
+xids_finalize_with_child(XIDS xids, TXNID this_xid) {
+    // Precondition:
+    //  - xids was created by xids_create_unknown_child
+    //  - All error checking (except that this_xid is higher than its parent) is already complete
+    invariant(this_xid > xids_get_innermost_xid(xids));
+    TXNID this_xid_disk = toku_htod64(this_xid);
+    u_int32_t num_child_xids = ++xids->num_xids;
+    xids->ids[num_child_xids - 1] = this_xid_disk;
+}
+
+// xids is immutable.  This function creates a new xids by copying the
+// parent's list and then appending the xid of the new transaction.
+int
+xids_create_child(XIDS   parent_xids,		// xids list for parent transaction
+		  XIDS * xids_p,		// xids list created
+		  TXNID  this_xid) {		// xid of this transaction (new innermost)
+    int rval = xids_create_unknown_child(parent_xids, xids_p);
+    if (rval == 0) {
+        xids_finalize_with_child(*xids_p, this_xid);
     }
     return rval;
 }

ft/xids.h

@@ -36,6 +36,12 @@ void xids_cpy(XIDS target, XIDS source);
 //Creates an XIDS representing this transaction.
 //You must pass in an XIDS representing the parent of this transaction.
 int  xids_create_child(XIDS parent_xids, XIDS *xids_p, TXNID this_xid);
+
+// The following two functions (in order) are equivalent to xids_create child,
+// but allow you to do most of the work without knowing the new xid.
+int xids_create_unknown_child(XIDS parent_xids, XIDS *xids_p);
+void xids_finalize_with_child(XIDS xids, TXNID this_xid);
+
 void xids_create_from_buffer(struct rbuf *rb, XIDS * xids_p);
 
 void xids_destroy(XIDS *xids_p);