[t:2449] Add prefetching of rollback logs to speed up commit/abort processing

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

newbrt/roll.c

@@ -307,6 +307,9 @@ toku_apply_rollinclude (TXNID      xid,
         assert(r==0);
         last_sequence = log->sequence;
         
+        r = toku_maybe_prefetch_older_rollback_log(txn, log);
+        assert(r==0);
+
         while ((item=log->newest_logentry)) {
             log->newest_logentry = item->prev;
             r = func(txn, item, yield, yieldv, oplsn);

newbrt/rollback.c

@@ -93,6 +93,10 @@ toku_apply_txn (TOKUTXN txn, YIELDF yield, void*yieldv, LSN lsn,
         //pin log
         r = toku_get_and_pin_rollback_log(txn, txn->txnid64, last_sequence-1, next_log, next_log_hash, &log);
         assert(r==0);
+
+        r = toku_maybe_prefetch_older_rollback_log(txn, log);
+        assert(r==0);
+
         last_sequence = log->sequence;
         if (func) {
             while ((item=log->newest_logentry)) {
@@ -590,6 +594,23 @@ int toku_txn_find_by_xid (BRT brt, TXNID xid, TOKUTXN *txnptr) {
     return r;
 }
 
+int
+toku_maybe_prefetch_older_rollback_log(TOKUTXN txn, ROLLBACK_LOG_NODE log) {
+    //Currently processing 'log'.  Prefetch the next (older) log node.
+    BLOCKNUM name = log->older;
+    int r = 0;
+    if (name.b != ROLLBACK_NONE.b) {
+        uint32_t hash = log->older_hash;
+        CACHEFILE cf = txn->logger->rollback_cachefile;
+        r = toku_cachefile_prefetch(cf, name, hash,
+                                    toku_rollback_flush_callback,
+                                    toku_rollback_fetch_callback,
+                                    txn);
+        assert(r==0);
+    }
+    return r;
+}
+
 int toku_get_and_pin_rollback_log(TOKUTXN txn, TXNID xid, uint64_t sequence, BLOCKNUM name, uint32_t hash, ROLLBACK_LOG_NODE *result) {
     BOOL save_inprogress_node = FALSE;
     assert(name.b != ROLLBACK_NONE.b);

newbrt/rollback.h

@@ -15,6 +15,7 @@ int toku_rollback_abort(TOKUTXN txn, YIELDF yield, void*yieldv, LSN lsn);
 void toku_rollback_txn_close (TOKUTXN txn);
 int toku_get_and_pin_rollback_log_for_new_entry (TOKUTXN txn, ROLLBACK_LOG_NODE *result);
 int toku_get_and_pin_rollback_log(TOKUTXN txn, TXNID xid, uint64_t sequence, BLOCKNUM name, uint32_t hash, ROLLBACK_LOG_NODE *result);
+int toku_maybe_prefetch_older_rollback_log(TOKUTXN txn, ROLLBACK_LOG_NODE log);
 int toku_rollback_log_unpin(TOKUTXN txn, ROLLBACK_LOG_NODE log);
 int toku_delete_rollback_log(TOKUTXN txn, ROLLBACK_LOG_NODE log);