close[t:3748] After dequeuing a fifo, free the memory. Fixes #3748.

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

newbrt/brt.c

@@ -2660,6 +2660,7 @@ flush_this_child (BRT t, BRTNODE node, int childnum, enum reactivity *child_re,
 
 	    BNC_NBYTESINBUF(node, childnum) -= n_bytes_removed;
 	}
+	toku_fifo_size_is_stabilized(fifo);
 
 	invariant(BNC_NBYTESINBUF(node, childnum) == 0);
 	BP_WORKDONE(node, childnum) = 0;  // this buffer is drained, no work has been done by its contents
@@ -2701,6 +2702,7 @@ flush_this_child (BRT t, BRTNODE node, int childnum, enum reactivity *child_re,
 	    node->dirty = 1;
 
 	}
+	toku_fifo_size_is_stabilized(fifo);
 
 	invariant(BNC_NBYTESINBUF(node, childnum) == 0);
 	BP_WORKDONE(node, childnum) = 0;  // this buffer is drained, no work has been done by its contents

newbrt/fifo.c

@@ -176,6 +176,20 @@ void toku_fifo_iterate (FIFO fifo, void(*f)(bytevec key,ITEMLEN keylen,bytevec d
 		 f(key,keylen,data,datalen,type,msn,xids, arg));
 }
 
+void toku_fifo_size_is_stabilized(FIFO fifo) {
+    if (fifo->memory_used < fifo->memory_size/2) {
+	char *old_memory = fifo->memory;
+	int new_memory_size = fifo->memory_used*2;
+	char *new_memory = toku_xmalloc(new_memory_size);
+	memcpy(new_memory, old_memory+fifo->memory_start, fifo->memory_used);
+	fifo->memory       = new_memory;
+	fifo->memory_start = 0;
+	fifo->memory_size  = new_memory_size;
+	toku_free(old_memory);
+    }
+}
+
 unsigned long toku_fifo_memory_size(FIFO fifo) {
     return sizeof(*fifo)+fifo->memory_size;
 }
+

newbrt/fifo.h

@@ -39,6 +39,9 @@ void toku_fifo_free(FIFO *);
 // into the fifo.
 void toku_fifo_size_hint(FIFO, size_t size_hint);
 
+void toku_fifo_size_is_stabilized(FIFO);
+// Effect: Tell the FIFO that we may have just inserted or removed a bunch of stuff, and now may be a good time to resize memory.
+
 int toku_fifo_n_entries(FIFO);
 
 int toku_fifo_enq_cmdstruct (FIFO fifo, const BRT_MSG cmd);

newbrt/tests/test3748.c

+/* -*- mode: C; c-basic-offset: 4 -*- */
+#ident "Copyright (c) 2007, 2008 Tokutek Inc.  All rights reserved."
+#ident "Id:"
+
+// Test for #3748 (FIFO's do not shrink their size after a flush)
+// This test makes sure that the code that stabilizes fifos works.
+
+#include "includes.h"
+#include "test.h"
+int verbose;
+
+static void test_3748 (void) {
+    FIFO f;
+    int r;
+    f = NULL;
+    r = toku_fifo_create(&f);
+
+    char *thekey = 0; int thekeylen;
+    char *theval = 0; int thevallen;
+
+#define buildkey(len) { \
+        thekeylen = len; \
+        thekey = toku_realloc(thekey, thekeylen); \
+        memset(thekey, len, thekeylen); \
+    }
+
+#define buildval(len) { \
+        thevallen = len+1; \
+        theval = toku_realloc(theval, thevallen); \
+        memset(theval, ~len, thevallen); \
+    }
+
+    MSN startmsn = ZERO_MSN;
+
+    int N=1000;
+
+    // enqueue some stuff
+    for (int i=0; i<N; i++) {
+	buildkey(i);
+        buildval(i);
+
+	XIDS xids = xids_get_root_xids();
+	MSN msn = next_dummymsn();
+
+	if (startmsn.msn == ZERO_MSN.msn)
+	    startmsn = msn;
+	r = toku_fifo_enq(f, thekey, thekeylen, theval, thevallen, i, msn, xids); assert(r == 0);
+	xids_destroy(&xids);
+    }
+    for (int i=N/10; i<N; i++) {
+	r = toku_fifo_deq(f);
+    }
+    unsigned int msize0 = toku_fifo_memory_size(f);
+    toku_fifo_size_is_stabilized(f);
+    unsigned int msize1 = toku_fifo_memory_size(f);
+    assert(msize1 < msize0);
+
+    while (toku_fifo_deq(f) == 0)
+        /*nothing*/;
+
+    toku_fifo_size_is_stabilized(f);
+    unsigned int msize2 = toku_fifo_memory_size(f);
+    assert(msize2 < msize1);
+
+    toku_fifo_free(&f);
+    assert(f == 0);
+
+    if (thekey) toku_free(thekey);
+    if (theval) toku_free(theval);
+
+}
+
+int
+test_main(int argc, const char *argv[]) {
+    default_parse_args(argc, argv);
+    test_3748();
+   
+    return 0;
+}