Undo pma movements. Addresses #253.

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

newbrt/log.c

@@ -278,6 +278,12 @@ int toku_logger_log_phys_add_or_delete_in_leaf (DB *db, TOKUTXN txn, DISKOFF dis
 int toku_logger_commit (TOKUTXN txn, int nosync) {
     // panic handled in log_commit
     int r = toku_log_commit(txn, txn->txnid64, nosync);
+    struct log_entry *item;
+    while ((item=txn->oldest_logentry)) {
+	txn->oldest_logentry = item->next;
+	logtype_dispatch(item, toku_free_logtype_);
+	free(item);
+    }
     toku_free(txn);
     return r;
 }

newbrt/logformat.c

@@ -208,7 +208,6 @@ void generate_log_writer (void) {
 			    fprintf(cf, "    if (txn->newest_logentry) txn->parent->newest_logentry = txn->newest_logentry;\n");
 			    fprintf(cf, "    txn->newest_logentry = txn->oldest_logentry = 0;\n");
 			    fprintf(cf, "  } else {\n");
-			    fprintf(cf, "    while (txn->newest_logentry) { struct log_entry *next=txn->newest_logentry->next; toku_free(txn->newest_logentry); txn->newest_logentry=next; }\n");
 			    fprintf(cf, "    r = toku_logger_fsync(txn->logger);\n");
 			    fprintf(cf, "    if (r!=0) toku_logger_panic(txn->logger, r);\n");
 			    fprintf(cf, "  }\n");

newbrt/pma.c

@@ -1627,3 +1627,20 @@ int toku_pma_move_indices (PMA pma, INTPAIRARRAY fromto) {
     }
     return 0;
 }
+
+static void reverse_fromto (INTPAIRARRAY fromto) {
+    u_int32_t i;
+    for (i=0; i<fromto.size; i++) {
+	int tmp = fromto.array[i].a;
+	fromto.array[i].a = fromto.array[i].b;
+	fromto.array[i].b = tmp;
+    }
+}
+
+int toku_pma_move_indices_back (PMA pma, INTPAIRARRAY fromto) {
+    int r;
+    reverse_fromto(fromto);
+    r = toku_pma_move_indices(pma, fromto);
+    reverse_fromto(fromto);
+    return r;
+}

newbrt/pma.h

@@ -156,6 +156,8 @@ int toku_pma_clear_at_index (PMA, unsigned int /*index*/); // If the index is wr
 
 // Requires:  No open cursors on the pma.
 int toku_pma_move_indices (PMA pma, INTPAIRARRAY fromto); // Return nonzero if the indices are somehow wrong.
+// Move things backwards according to fromto.
+int toku_pma_move_indices_back (PMA pma, INTPAIRARRAY fromto);
 
 void toku_pma_show_stats (void);
 

newbrt/roll.c

@@ -311,6 +311,21 @@ void toku_recover_pmadistribute (struct logtype_pmadistribute *c) {
     toku_free(c->fromto.array);
 }
 
+int toku_rollback_pmadistribute (struct logtype_pmadistribute *le, TOKUTXN txn) {
+    CACHEFILE cf;
+    BRT brt;
+    int r = toku_cachefile_of_filenum(txn->logger->ct, le->filenum, &cf, &brt);
+    if (r!=0) return r;
+    void *node_v;
+    r = toku_cachetable_get_and_pin(cf, le->diskoff, &node_v, NULL, toku_brtnode_flush_callback, toku_brtnode_fetch_callback, brt);
+    if (r!=0) return r;
+    BRTNODE node = node_v;
+    r = toku_pma_move_indices_back(node->u.l.buffer, le->fromto);
+    if (r!=0) return r;
+    r = toku_cachetable_unpin(cf, le->diskoff, 1, toku_serialize_brtnode_size(node));
+    return r;
+}
+
 int toku_rollback_fheader (struct logtype_fheader *le, TOKUTXN txn)             ABORTIT
 int toku_rollback_resizepma (struct logtype_resizepma *le, TOKUTXN txn)         ABORTIT
-int toku_rollback_pmadistribute (struct logtype_pmadistribute *le, TOKUTXN txn) ABORTIT
+

src/tests/test_abort1.c

@@ -125,7 +125,6 @@ void test_db_put_aborts (void) {
 	r=tid->commit(tid,0);        assert(r==0);
     }
 
-
     r=db->close(db, 0);       assert(r==0);
     r=env->close(env, 0);     assert(r==0);
 }