#3590 merge fsync after locks are released (fixes groupcommit) refs[t:3590]

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

newbrt/log-internal.h

@@ -156,6 +156,8 @@ struct tokutxn {
     struct toku_list checkpoint_before_commit;
     TXN_IGNORE_S ignore_errors; // 2954
     TOKUTXN_STATE state;
+    LSN        do_fsync_lsn;
+    BOOL       do_fsync;
 };
 
 struct txninfo {

newbrt/txn.c

@@ -270,6 +270,7 @@ int toku_txn_begin_with_xid (
     result->recovered_from_checkpoint = FALSE;
     toku_list_init(&result->checkpoint_before_commit);
     result->state = TOKUTXN_LIVE;
+    result->do_fsync = FALSE;
 
     // 2954
     r = toku_txn_ignore_init(result);
@@ -329,7 +330,6 @@ int toku_txn_commit_txn(TOKUTXN txn, int nosync, YIELDF yield, void *yieldv,
 struct xcommit_info {
     int r;
     TOKUTXN txn;
-    bool do_fsync;
 };
 
 //Called during a yield (ydb lock NOT held).
@@ -365,7 +365,7 @@ local_checkpoints_and_log_xcommit(void *thunk) {
         toku_poll_txn_progress_function(txn, TRUE, FALSE);
     }
 
-    info->r = toku_log_xcommit(txn->logger, (LSN *)0, info->do_fsync, txn->txnid64); // exits holding neither of the tokulogger locks.
+    info->r = toku_log_xcommit(txn->logger, &txn->do_fsync_lsn, 0, txn->txnid64); // exits holding neither of the tokulogger locks.
 }
 
 int toku_txn_commit_with_lsn(TOKUTXN txn, int nosync, YIELDF yield, void *yieldv, LSN oplsn,
@@ -378,8 +378,8 @@ int toku_txn_commit_with_lsn(TOKUTXN txn, int nosync, YIELDF yield, void *yieldv
     // panic handled in log_commit
 
     // Child transactions do not actually 'commit'.  They promote their changes to parent, so no need to fsync if this txn has a parent.
-    // the do_sync state is captured in the txn for txn_close_txn later
-    bool do_fsync = !txn->parent && (txn->force_fsync_on_commit || (!nosync && txn->num_rollentries>0));
+    // the do_sync state is captured in the txn for txn_maybe_fsync_log function
+    txn->do_fsync = !txn->parent && (txn->force_fsync_on_commit || (!nosync && txn->num_rollentries>0));
 
     txn->progress_poll_fun = poll;
     txn->progress_poll_fun_extra = poll_extra;
@@ -388,7 +388,6 @@ int toku_txn_commit_with_lsn(TOKUTXN txn, int nosync, YIELDF yield, void *yieldv
         struct xcommit_info info = {
             .r = 0,
             .txn = txn,
-            .do_fsync = do_fsync,
         };
         yield(local_checkpoints_and_log_xcommit, &info, yieldv);
         r = info.r;
@@ -419,8 +418,9 @@ int toku_txn_abort_with_lsn(TOKUTXN txn, YIELDF yield, void *yieldv, LSN oplsn,
 
     txn->progress_poll_fun = poll;
     txn->progress_poll_fun_extra = poll_extra;
-    int r=0;
-    r = toku_log_xabort(txn->logger, (LSN *)0, 0, txn->txnid64);
+    int r = 0;
+    txn->do_fsync = FALSE;
+    r = toku_log_xabort(txn->logger, &txn->do_fsync_lsn, 0, txn->txnid64);
     if (r!=0) 
         return r;
     r = toku_rollback_abort(txn, yield, yieldv, oplsn);
@@ -428,8 +428,29 @@ int toku_txn_abort_with_lsn(TOKUTXN txn, YIELDF yield, void *yieldv, LSN oplsn,
     return r;
 }
 
+struct txn_fsync_log_info {
+    TOKUTXN txn;
+    int r;
+};
+
+static void do_txn_fsync_log(void *thunk) {
+    struct txn_fsync_log_info *info = (struct txn_fsync_log_info *) thunk;
+    TOKUTXN txn = info->txn;
+    info->r = toku_logger_fsync_if_lsn_not_fsynced(txn->logger, txn->do_fsync_lsn);
+}
+
+int toku_txn_maybe_fsync_log(TOKUTXN txn, YIELDF yield, void *yieldv) {
+    int r = 0;
+    if (txn->logger && txn->do_fsync) {
+        struct txn_fsync_log_info info = { .txn = txn };
+        yield(do_txn_fsync_log, &info, yieldv);
+        r = info.r;
+    }
+    return r;
+}
+
 void toku_txn_close_txn(TOKUTXN txn) {
-    TOKULOGGER logger = txn->logger; // capture these for the fsync after the txn is deleted
+    TOKULOGGER logger = txn->logger;
 
     toku_rollback_txn_close(txn); 
     txn = NULL; // txn is no longer valid

newbrt/txn.h

@@ -40,7 +40,9 @@ int toku_txn_abort_txn(TOKUTXN txn, YIELDF yield, void *yieldv,
 int toku_txn_abort_with_lsn(TOKUTXN txn, YIELDF yield, void *yieldv, LSN oplsn,
                             TXN_PROGRESS_POLL_FUNCTION poll, void *poll_extra);
 
+int toku_txn_maybe_fsync_log(TOKUTXN txn, YIELDF yield, void *yieldv);
 void toku_txn_close_txn(TOKUTXN txn);
+
 XIDS toku_txn_get_xids (TOKUTXN);
 
 // Returns TRUE if a is older than b

src/tests/test_groupcommit_count.c

@@ -114,8 +114,9 @@ do_test (int N) {
 	int count_before = get_fsync_count();
 	test_groupcommit(N);
 	printtdiff(N);
-	if (get_fsync_count()-count_before>= N*NITER) {
-	    if (verbose) printf("It looks like too many fsyncs.  Group commit doesn't appear to be occuring.\n");
+	int count_after = get_fsync_count();
+	if (count_after-count_before >= N*NITER) {
+	    if (verbose) printf("It looks like too many fsyncs.  Group commit doesn't appear to be occuring. %d - %d >= %d\n", count_after, count_before, N*NITER);
 	    exit(1);
 	}
     }

src/ydb.c

@@ -2451,6 +2451,7 @@ toku_txn_commit(DB_TXN * txn, u_int32_t flags,
     // Close the logger after releasing the locks
     r = toku_txn_release_locks(txn);
     //toku_logger_txn_close(db_txn_struct_i(txn)->tokutxn);
+    toku_txn_maybe_fsync_log(db_txn_struct_i(txn)->tokutxn, ydb_yield, NULL);
     toku_txn_close_txn(db_txn_struct_i(txn)->tokutxn);
     // the toxutxn is freed, and we must free the rest. */
 
@@ -2520,6 +2521,7 @@ toku_txn_abort(DB_TXN * txn,
     assert(r==0);
     r = toku_txn_release_locks(txn);
     //toku_logger_txn_close(db_txn_struct_i(txn)->tokutxn);
+    toku_txn_maybe_fsync_log(db_txn_struct_i(txn)->tokutxn, ydb_yield, NULL);
     toku_txn_close_txn(db_txn_struct_i(txn)->tokutxn);
 
 #if !TOKUDB_NATIVE_H