DB-808 add flags to xa prepare to support no sync log

buildheader/make_tdb.cc

@@ -572,7 +572,7 @@ static void print_db_txn_struct (void) {
     STRUCT_SETUP(DB_TXN, abort,       "int (*%s) (DB_TXN *)");
     STRUCT_SETUP(DB_TXN, api_internal,"void *%s");
     STRUCT_SETUP(DB_TXN, commit,      "int (*%s) (DB_TXN*, uint32_t)");
-    STRUCT_SETUP(DB_TXN, prepare,     "int (*%s) (DB_TXN*, uint8_t gid[DB_GID_SIZE])");
+    STRUCT_SETUP(DB_TXN, prepare,     "int (*%s) (DB_TXN*, uint8_t gid[DB_GID_SIZE], uint32_t flags)");
     STRUCT_SETUP(DB_TXN, discard,     "int (*%s) (DB_TXN*, uint32_t)");
     STRUCT_SETUP(DB_TXN, id,          "uint32_t (*%s) (DB_TXN *)");
     STRUCT_SETUP(DB_TXN, mgrp,        "DB_ENV *%s /* In TokuFT, mgrp is a DB_ENV, not a DB_TXNMGR */");
@@ -581,7 +581,7 @@ static void print_db_txn_struct (void) {
 	"int (*txn_stat)(DB_TXN *, struct txn_stat **)", 
 	"int (*commit_with_progress)(DB_TXN*, uint32_t, TXN_PROGRESS_POLL_FUNCTION, void*)",
 	"int (*abort_with_progress)(DB_TXN*, TXN_PROGRESS_POLL_FUNCTION, void*)",
-	"int (*xa_prepare) (DB_TXN*, TOKU_XA_XID *)",
+	"int (*xa_prepare) (DB_TXN*, TOKU_XA_XID *, uint32_t flags)",
         "uint64_t (*id64) (DB_TXN*)",
         "void (*set_client_id)(DB_TXN *, uint64_t client_id)",
         "uint64_t (*get_client_id)(DB_TXN *)",

ft/logger/recover.cc

@@ -712,7 +712,7 @@ static int toku_recover_xstillopenprepared (struct logtype_xstillopenprepared *l
     }
     switch (renv->ss.ss) {
         case FORWARD_BETWEEN_CHECKPOINT_BEGIN_END: {
-            toku_txn_prepare_txn(txn, l->xa_xid);
+            toku_txn_prepare_txn(txn, l->xa_xid, 0);
             break;
         }
         case FORWARD_NEWER_CHECKPOINT_END: {
@@ -776,7 +776,7 @@ static int toku_recover_xprepare (struct logtype_xprepare *l, RECOVER_ENV renv)
     assert(txn!=NULL);
 
     // Save the transaction
-    toku_txn_prepare_txn(txn, l->xa_xid);
+    toku_txn_prepare_txn(txn, l->xa_xid, 0);
 
     return 0;
 }

ft/txn/txn.cc

@@ -556,7 +556,7 @@ static void copy_xid (TOKU_XA_XID *dest, TOKU_XA_XID *source) {
     memcpy(dest->data, source->data, source->gtrid_length+source->bqual_length);
 }
 
-void toku_txn_prepare_txn (TOKUTXN txn, TOKU_XA_XID *xa_xid) {
+void toku_txn_prepare_txn (TOKUTXN txn, TOKU_XA_XID *xa_xid, int nosync) {
     if (txn->parent || toku_txn_is_read_only(txn)) {
         // We do not prepare children.
         //
@@ -571,7 +571,7 @@ void toku_txn_prepare_txn (TOKUTXN txn, TOKU_XA_XID *xa_xid) {
     txn->state = TOKUTXN_PREPARING; 
     toku_txn_unlock_state(txn);
     // Do we need to do an fsync?
-    txn->do_fsync = (txn->force_fsync_on_commit || txn->roll_info.num_rollentries>0);
+    txn->do_fsync = txn->force_fsync_on_commit || (!nosync && 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.
     toku_log_xprepare(txn->logger, &txn->do_fsync_lsn, 0, txn, txn->txnid, xa_xid);

ft/txn/txn.h

@@ -302,7 +302,7 @@ int toku_txn_abort_with_lsn(struct tokutxn *txn, LSN oplsn,
 
 int toku_txn_discard_txn(struct tokutxn *txn);
 
-void toku_txn_prepare_txn (struct tokutxn *txn, TOKU_XA_XID *xid);
+void toku_txn_prepare_txn (struct tokutxn *txn, TOKU_XA_XID *xid, int nosync);
 // Effect: Do the internal work of preparing a transaction (does not log the prepare record).
 
 void toku_txn_get_prepared_xa_xid(struct tokutxn *txn, TOKU_XA_XID *xa_xid);

src/tests/recover-5146.cc

@@ -147,7 +147,7 @@ run_test(void) {
         uint8_t gid[DB_GID_SIZE];
         memset(gid, 0, DB_GID_SIZE);
         gid[0]=42;
-        r = txn->prepare(txn, gid);                                                            CKERR(r);
+        r = txn->prepare(txn, gid, 0);                                                            CKERR(r);
     }
 
     r = env->txn_checkpoint(env, 0, 0, 0);                                              CKERR(r);

src/tests/test-prepare.cc

@@ -122,7 +122,7 @@ static void setup_env_and_prepare (DB_ENV **envp, const char *envdir, bool commi
     uint8_t gid[DB_GID_SIZE];
     memset(gid, 0, DB_GID_SIZE);
     gid[0]=42;
-    CKERR(txn->prepare(txn, gid));
+    CKERR(txn->prepare(txn, gid, 0));
     if (commit)
 	CKERR(txn->commit(txn, 0));
 }

src/tests/test-prepare2.cc

@@ -124,7 +124,7 @@ static void setup_env_and_prepare (DB_ENV **envp, const char *envdir, bool commi
     uint8_t gid[DB_GID_SIZE];
     memset(gid, 0, DB_GID_SIZE);
     gid[0]=42;
-    CKERR(txn->prepare(txn, gid));
+    CKERR(txn->prepare(txn, gid, 0));
     if (commit)
 	CKERR(txn->commit(txn, 0));
 }

src/tests/test-prepare3.cc

@@ -147,7 +147,7 @@ static void setup_env_and_prepare (DB_ENV **envp, const char *envdir) {
 	uint8_t gid[DB_GID_SIZE];
 	memset(gid, 0, DB_GID_SIZE);
 	gid[0]='a'+tnum;
-	CKERR(txn->prepare(txn, gid));
+	CKERR(txn->prepare(txn, gid, 0));
 	// Drop txn on the ground, since we will commit or abort it after recovery
 	if (tnum==0) {
 	    //printf("commit %d\n", tnum);

src/tests/test-xa-prepare.cc

@@ -125,7 +125,7 @@ static void setup_env_and_prepare (DB_ENV **envp, const char *envdir, bool commi
 		     .gtrid_length = 8,
 		     .bqual_length = 9};
     for (int i=0; i<8+9; i++) x.data[i] = 42+i;
-    CKERR(txn->xa_prepare(txn, &x));
+    CKERR(txn->xa_prepare(txn, &x, 0));
     if (commit)
 	CKERR(txn->commit(txn, 0));
 }

src/tests/test5092.cc

@@ -118,7 +118,7 @@ static void setup_env_and_prepare (DB_ENV **envp, const char *envdir, bool commi
     uint8_t gid[DB_GID_SIZE];
     memset(gid, 0, DB_GID_SIZE);
     gid[0]=42;
-    CKERR(txn->prepare(txn, gid));
+    CKERR(txn->prepare(txn, gid, 0));
     { int chk_r = db->close(db, 0); CKERR(chk_r); }
     if (commit)
         CKERR(txn->commit(txn, 0));

src/tests/test_5015.cc

@@ -133,7 +133,7 @@ test_main(int argc, char *const argv[]) {
     uint8_t gid[DB_GID_SIZE];
     memset(gid, 0, DB_GID_SIZE);
     gid[0]='a';
-    r = child_txn->prepare(child_txn, gid);
+    r = child_txn->prepare(child_txn, gid, 0);
     CKERR(r);
 
     r = env->txn_checkpoint(env, 0, 0, 0);

src/tests/test_stress_hot_indexing.cc

@@ -211,7 +211,7 @@ static int hi_inserts(DB_TXN* UU(txn), ARG arg, void* UU(operation_extra), void
     gid_count++;
     uint32_t *hi_gid_count_p = cast_to_typeof(hi_gid_count_p) hi_gid;  // make gcc --happy about -Wstrict-aliasing
     *hi_gid_count_p = gid_count;
-    int rr = hi_txn->prepare(hi_txn, hi_gid);
+    int rr = hi_txn->prepare(hi_txn, hi_gid, 0);
     CKERR(rr);
     if (r || (random() % 2)) {
         rr = hi_txn->abort(hi_txn);

src/tests/test_txn_close_before_prepare_commit.cc

@@ -130,7 +130,7 @@ test_txn_close_before_prepare_commit (void) {
 
     uint8_t gid[DB_GID_SIZE];
     memset(gid, 1, DB_GID_SIZE);
-    r = txn->prepare(txn, gid);   assert(r == 0);
+    r = txn->prepare(txn, gid, 0);   assert(r == 0);
     r = txn->commit(txn, 0); assert(r == 0);
 
     r = env->close(env, 0); assert(r == 0);

src/tests/threaded_stress_test_helpers.h

@@ -593,7 +593,7 @@ static void *worker(void *arg_v) {
             uint64_t gid_val = txn->id64(txn);
             uint64_t *gid_count_p = cast_to_typeof(gid_count_p) gid;  // make gcc --happy about -Wstrict-aliasing
             *gid_count_p = gid_val;
-            int rr = txn->prepare(txn, gid);
+            int rr = txn->prepare(txn, gid, 0);
             assert_zero(rr);
         }
         if (r == 0) {

src/tests/xa-dirty-commit.cc

@@ -136,7 +136,7 @@ static void create_prepared_txn(void) {
     for (int i = 0; i < 8+9; i++) {
         xid.data[i] = i;
     }
-    r = txn->xa_prepare(txn, &xid);
+    r = txn->xa_prepare(txn, &xid, 0);
     CKERR(r);
 
     // discard the txn so that we can close the env and run xa recovery later

src/tests/xa-dirty-rollback.cc

@@ -136,7 +136,7 @@ static void create_prepared_txn(void) {
     for (int i = 0; i < 8+9; i++) {
         xid.data[i] = i;
     }
-    r = txn->xa_prepare(txn, &xid);
+    r = txn->xa_prepare(txn, &xid, 0);
     CKERR(r);
 
     // discard the txn so that we can close the env and run xa recovery later

src/tests/xa-txn-discard-abort.cc

@@ -138,7 +138,7 @@ static void create_prepared_txn(void) {
     for (int i = 0; i < 8+9; i++) {
         xid.data[i] = i;
     }
-    r = txn->xa_prepare(txn, &xid);
+    r = txn->xa_prepare(txn, &xid, 0);
     CKERR(r);
 
     // discard the txn so that we can close the env and run xa recovery later

src/tests/xa-txn-discard-commit.cc

@@ -139,7 +139,7 @@ static void create_prepared_txn(void) {
     for (int i = 0; i < 8+9; i++) {
         xid.data[i] = i;
     }
-    r = txn->xa_prepare(txn, &xid);
+    r = txn->xa_prepare(txn, &xid, 0);
     CKERR(r);
 
     // discard the txn so that we can close the env and run xa recovery later

src/ydb_txn.cc

@@ -239,7 +239,7 @@ static int toku_txn_abort(DB_TXN * txn,
     return r;
 }
 
-static int toku_txn_xa_prepare (DB_TXN *txn, TOKU_XA_XID *xid) {
+static int toku_txn_xa_prepare (DB_TXN *txn, TOKU_XA_XID *xid, uint32_t flags) {
     int r = 0;
     if (!txn) {
         r = EINVAL;
@@ -272,9 +272,11 @@ static int toku_txn_xa_prepare (DB_TXN *txn, TOKU_XA_XID *xid) {
         HANDLE_PANICKED_ENV(txn->mgrp);
     }
     assert(!db_txn_struct_i(txn)->child);
+    int nosync;
+    nosync = (flags & DB_TXN_NOSYNC)!=0 || (db_txn_struct_i(txn)->flags&DB_TXN_NOSYNC);
     TOKUTXN ttxn;
     ttxn = db_txn_struct_i(txn)->tokutxn;
-    toku_txn_prepare_txn(ttxn, xid);
+    toku_txn_prepare_txn(ttxn, xid, nosync);
     TOKULOGGER logger;
     logger = txn->mgrp->i->logger;
     LSN do_fsync_lsn;
@@ -291,14 +293,14 @@ static int toku_txn_xa_prepare (DB_TXN *txn, TOKU_XA_XID *xid) {
 
 // requires: must hold the multi operation lock. it is
 //           released in toku_txn_xa_prepare before the fsync.
-static int toku_txn_prepare (DB_TXN *txn, uint8_t gid[DB_GID_SIZE]) {
+static int toku_txn_prepare (DB_TXN *txn, uint8_t gid[DB_GID_SIZE], uint32_t flags) {
     TOKU_XA_XID xid;
     TOKU_ANNOTATE_NEW_MEMORY(&xid, sizeof(xid));
     xid.formatID=0x756b6f54; // "Toku"
     xid.gtrid_length=DB_GID_SIZE/2;  // The maximum allowed gtrid length is 64.  See the XA spec in source:/import/opengroup.org/C193.pdf page 20.
     xid.bqual_length=DB_GID_SIZE/2; // The maximum allowed bqual length is 64.
     memcpy(xid.data, gid, DB_GID_SIZE);
-    return toku_txn_xa_prepare(txn, &xid);
+    return toku_txn_xa_prepare(txn, &xid, flags);
 }
 
 static int toku_txn_txn_stat (DB_TXN *txn, struct txn_stat **txn_stat) {