test1572 doesn't really test #1572 yet, but it did point out #1577. Now we...

test1572 doesn't really test #1572 yet, but it did point out #1577.  Now we are more careful about memory management in the rollback logs, so that we don't blow out memory.  Fixes #1577.  Refs #1572.

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

newbrt/log.c

@@ -473,7 +473,10 @@ int toku_logger_commit (TOKUTXN txn, int nosync, void(*yield)(void*yieldv), void
 	    }
 	    txn->newest_logentry = txn->oldest_logentry = 0;
 	    // Put all the memarena data into the parent.
-	    memarena_move_buffers(txn->parent->rollentry_arena, txn->rollentry_arena);
+	    if (memarena_size_in_use(txn->rollentry_arena) > 0) {
+		// If there are no bytes to move, then just leave things alone, and let the memory be reclaimed on txn is closed.
+		memarena_move_buffers(txn->parent->rollentry_arena, txn->rollentry_arena);
+	    }
 
 	    // Note the open brts, the omts must be merged
 	    r = toku_omt_iterate(txn->open_brts, note_brt_used_in_parent_txn, txn->parent);
@@ -1015,7 +1018,10 @@ int toku_logger_log_archive (TOKULOGGER logger, char ***logs_p, int flags) {
 }
 
 int toku_maybe_spill_rollbacks (TOKUTXN txn) {
-    if (txn->rollentry_resident_bytecount>txn->logger->write_block_size) {
+    // Previously:
+    // if (txn->rollentry_resident_bytecount>txn->logger->write_block_size) {
+    // But now we use t
+    if (memarena_total_memory_size(txn->rollentry_arena) > txn->logger->write_block_size) {
 	struct roll_entry *item;
 	ssize_t bufsize = txn->rollentry_resident_bytecount;
 	char *MALLOC_N(bufsize, buf);

newbrt/memarena.c

@@ -3,6 +3,7 @@
 struct memarena {
     char *buf;
     size_t buf_used, buf_size;
+    size_t size_of_other_bufs; // the buf_size of all the other bufs.
     char **other_bufs;
     int n_other_bufs;
 };
@@ -12,6 +13,7 @@ MEMARENA memarena_create (void) {
     result->buf_size = 1024;
     result->buf_used = 0;
     result->other_bufs = NULL;
+    result->size_of_other_bufs = 0;
     result->n_other_bufs = 0;
     result->buf = toku_malloc(result->buf_size);  assert(result->buf);
     return result;
@@ -27,6 +29,7 @@ void memarena_clear (MEMARENA ma) {
     ma->n_other_bufs=0;
     // But reuse the main buffer
     ma->buf_used = 0;
+    ma->size_of_other_bufs = 0;
 }
 
 static size_t
@@ -114,6 +117,10 @@ void memarena_move_buffers(MEMARENA dest, MEMARENA source) {
  	printf("new_other_bufs=%p errno=%d\n", new_other_bufs, errno);
     }
 #endif
+
+    dest  ->size_of_other_bufs += source->size_of_other_bufs + source->buf_size;
+    source->size_of_other_bufs = 0;
+
     assert(other_bufs);
     dest->other_bufs = other_bufs;
     for (i=0; i<source->n_other_bufs; i++) {
@@ -126,4 +133,17 @@ void memarena_move_buffers(MEMARENA dest, MEMARENA source) {
     source->buf = 0;
     source->buf_size = 0;
     source->buf_used = 0;
+
 }
+
+size_t
+memarena_total_memory_size (MEMARENA m)
+{
+    return m->size_of_other_bufs + m->buf_size + sizeof(m);
+}
+
+size_t
+memarena_size_in_use (MEMARENA m)
+{
+    return m->buf_used;
+}    

newbrt/memarena.h

@@ -34,4 +34,9 @@ void memarena_close(MEMARENA *ma);
 void memarena_move_buffers(MEMARENA dest, MEMARENA source);
 // Effect: Move all the memory from SOURCE into DEST.  When SOURCE is closed the memory won't be freed.  When DEST is closed, the memory will be freed.  (Unless DEST moves its memory to another memarena...)
 
+size_t memarena_total_memory_size (MEMARENA);
+// Effect: Calculate the amount of memory used by a memory arena.
+
+size_t memarena_size_in_use (MEMARENA);
+
 #endif

src/tests/test1572.c

@@ -7,7 +7,7 @@
 #include <sys/stat.h>
 
 static void
-four_billion_subtransactions (void) {
+four_billion_subtransactions (int do_something_in_children) {
     DB_ENV *env;
     DB *db;
     DB_TXN *xparent;
@@ -31,12 +31,23 @@ four_billion_subtransactions (void) {
     r=env->txn_begin(env, 0, &xparent, 0);  CKERR(r);
     long long i;
     //long long const fourbillion = 1ll << 32;
-    long long const fourbillion = 400000;
+    long long const fourbillion = 500000;
+    printf("Doing %d\n", do_something_in_children);
     for (i=0; i < fourbillion + 100; i++) {
 	DB_TXN *xchild;
 	r=env->txn_begin(env, xparent, &xchild, 0); CKERR(r);
+	if (do_something_in_children) {
+	    char hello[30], there[30];
+	    snprintf(hello, sizeof(hello), "hello%lld", i);
+	    snprintf(there, sizeof(there), "there%lld", i);
+	    DBT key, val;
+	    r=db->put(db, xchild,
+		      dbt_init(&key, hello, strlen(hello)+1),
+		      dbt_init(&val, there, strlen(there)+1),
+		      DB_YESOVERWRITE);
+	    CKERR(r);
+	}
 	r=xchild->commit(xchild, 0); CKERR(r);
-	if (0==i%50000) sleep(1);
 	if (i%1000000==0) { printf("."); fflush(stdout); }
 
     }
@@ -50,7 +61,8 @@ int
 test_main (int argc, const char *argv[])
 {
     parse_args(argc, argv);
-    four_billion_subtransactions();
+    four_billion_subtransactions(0);
+    four_billion_subtransactions(1);
     return 0;
 }