replace hashtables with fifos. addresses #249

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

newbrt/Makefile

@@ -66,11 +66,11 @@ BINS =  $(REGRESSION_TESTS) \
 
 tdb_logprint: LDFLAGS+=-lz
 tdb_logprint.o: log-internal.h brttypes.h yerror.h log.h kv-pair.h
-tdb_logprint: log_code.o memory.o log.o brt-serialize.o hashtable.o pma.o ybt.o fingerprint.o mempool.o primes.o roll.o brt.o cachetable.o brt-verify.o key.o
+tdb_logprint: log_code.o memory.o log.o brt-serialize.o fifo.o pma.o ybt.o fingerprint.o mempool.o primes.o roll.o brt.o cachetable.o brt-verify.o key.o
 
 recover: LDFLAGS+=-lz
 recover.o: log_header.h log-internal.h log.h yerror.h brttypes.h kv-pair.h memory.h key.h
-recover: recover.o log_code.o memory.o log.o brt-serialize.o hashtable.o pma.o ybt.o fingerprint.o mempool.o primes.o cachetable.o brt.o brt-verify.o key.o roll.o
+recover: recover.o log_code.o memory.o log.o brt-serialize.o fifo.o pma.o ybt.o fingerprint.o mempool.o primes.o cachetable.o brt.o brt-verify.o key.o roll.o
 
 roll.o: log_header.h log-internal.h log.h yerror.h brttypes.h kv-pair.h memory.h key.h cachetable.h
 
@@ -106,32 +106,32 @@ check-fanout:
 pma-test benchmark-test brt-test brt-serialize-test: LDFLAGS+=-lz
 # pma: PROF_FLAGS=-fprofile-arcs -ftest-coverage
 
-BRT_INTERNAL_H_INCLUDES = brt-internal.h cachetable.h hashtable.h pma.h brt.h brttypes.h yerror.h ybt.h log.h ../include/db.h kv-pair.h memory.h crc.h
+BRT_INTERNAL_H_INCLUDES = brt-internal.h cachetable.h fifo.h pma.h brt.h brttypes.h yerror.h ybt.h log.h ../include/db.h kv-pair.h memory.h crc.h
 key.o: brttypes.h key.h
 pma-test.o: $(BRT_INTERNAL_H_INCLUDES) pma-internal.h pma.h list.h mempool.h
-pma-test: pma.o memory.o key.o ybt.o log.o mempool.o fingerprint.o brt-serialize.o hashtable.o primes.o log_code.o  roll.o brt.o cachetable.o brt-verify.o
+pma-test: pma.o memory.o key.o ybt.o log.o mempool.o fingerprint.o brt-serialize.o fifo.o primes.o log_code.o  roll.o brt.o cachetable.o brt-verify.o
 pma.o: pma.h yerror.h pma-internal.h memory.h key.h ybt.h brttypes.h log.h ../include/db.h log_header.h
 ybt.o: ybt.h brttypes.h ../include/db.h
 ybt-test: ybt-test.o ybt.o memory.o
 ybt-test.o: ybt.h ../include/db.h
 cachetable.o: cachetable.h hashfun.h memory.h
-brt-test: ybt.o brt.o hashtable.o pma.o memory.o brt-serialize.o cachetable.o ybt.o key.o primes.o log.o mempool.o brt-verify.o fingerprint.o log_code.o roll.o
+brt-test: ybt.o brt.o fifo.o pma.o memory.o brt-serialize.o cachetable.o ybt.o key.o primes.o log.o mempool.o brt-verify.o fingerprint.o log_code.o roll.o
 log.o: log_header.h log-internal.h log.h wbuf.h crc.h brttypes.h $(BRT_INTERNAL_H_INCLUDES)
-brt-test.o brt.o: brt.h ../include/db.h hashtable.h pma.h brttypes.h cachetable.h memory.h
+brt-test.o brt.o: brt.h ../include/db.h fifo.h pma.h brttypes.h cachetable.h memory.h
 brt-serialize-test.o: $(BRT_INTERNAL_H_INCLUDES)
 brt.o: $(BRT_INTERNAL_H_INCLUDES) key.h log_header.h
-hashtable.o: hashtable.h brttypes.h memory.h key.h yerror.h ../include/db.h hashfun.h
+fifo.o: fifo.h brttypes.h 
 memory.o: memory.h
 primes.o: primes.h
-hashtest: hashtable.o memory.o primes.o
+hashtest: fifo.o memory.o primes.o
 brt-serialize.o: $(BRT_INTERNAL_H_INCLUDES) key.h wbuf.h rbuf.h
-brt-bigtest: memory.o ybt.o brt.o pma.o cachetable.o key.o hashtable.o brt-serialize.o
+brt-bigtest: memory.o ybt.o brt.o pma.o cachetable.o key.o fifo.o brt-serialize.o
 brt-bigtest.o: brt.h ../include/db.h
 log-test: log.o memory.o
 brt-verify.o: $(BRT_INTERNAL_H_INCLUDES)
 fingerprint.o: $(BRT_INTERNAL_H_INCLUDES)
 
-brt-serialize-test: brt-serialize-test.o brt-serialize.o memory.o hashtable.o pma.o key.o ybt.o brt.o cachetable.o primes.o log.o mempool.o brt-verify.o fingerprint.o log_code.o roll.o
+brt-serialize-test: brt-serialize-test.o brt-serialize.o memory.o fifo.o pma.o key.o ybt.o brt.o cachetable.o primes.o log.o mempool.o brt-verify.o fingerprint.o log_code.o roll.o
 test_toku_malloc_plain_free: memory.o
 
 
@@ -141,7 +141,7 @@ cachetable-test: cachetable.o memory.o cachetable-test.o primes.o
 cachetable-test2.o: cachetable.h memory.h
 cachetable-test2: cachetable.o memory.o cachetable-test2.o primes.o
 
-benchmark-test: benchmark-test.o ybt.o memory.o brt.o pma.o cachetable.o key.o hashtable.o brt-serialize.o primes.o log.o mempool.o brt-verify.o fingerprint.o log_code.o roll.o
+benchmark-test: benchmark-test.o ybt.o memory.o brt.o pma.o cachetable.o key.o fifo.o brt-serialize.o primes.o log.o mempool.o brt-verify.o fingerprint.o log_code.o roll.o
 benchmark-test.o: brt.h ../include/db.h
 
 checko2:
@@ -156,6 +156,6 @@ clean:
 	rm -rf test_oexcl.c.tmp *.brt
 
 randdb4: LOADLIBES=-ldb
-randbrt: brt.o hashtable.o cachetable.o memory.o brt-serialize.o
+randbrt: brt.o fifo.o cachetable.o memory.o brt-serialize.o
 TAGS: ../*/*.c ../*/*.h
 	etags ../*/*.c ../*/*.h

newbrt/brt-internal.h

@@ -4,7 +4,7 @@
 #ident "Copyright (c) 2007 Tokutek Inc.  All rights reserved."
 
 #include "cachetable.h"
-#include "hashtable.h"
+#include "fifo.h"
 #include "pma.h"
 #include "brt.h"
 #include "crc.h"
@@ -36,8 +36,8 @@ struct brtnode_nonleaf_childinfo {
     u_int32_t    subtree_fingerprint;
 #if 0
     DISKOFF      diskoff;
-    HASHTABLE    htable;
-    unsigned int n_bytes_in_hashtable; /* How many bytes are in each hashtable (including overheads for the disk-representation) */
+    FIFO         htable;
+    unsigned int n_bytes_in_buffer; /* How many bytes are in each buffer (including overheads for the disk-representation) */
     unsigned int n_cursors;
 #endif
 };
@@ -54,14 +54,14 @@ struct brtnode {
     int     layout_version; // What version of the data structure?
     int    height; /* height is always >= 0.  0 for leaf, >0 for nonleaf. */
     u_int32_t rand4fingerprint;
-    u_int32_t local_fingerprint; /* For leaves this is everything in the buffer.  For nonleaves, this is everything in the hash tables, but does not include child subtree fingerprints. */
+    u_int32_t local_fingerprint; /* For leaves this is everything in the buffer.  For nonleaves, this is everything in the buffers, but does not include child subtree fingerprints. */
     int    dirty;
     union node {
 	struct nonleaf {
 	    // Don't actually store the subree fingerprint in the in-memory data structure.
 	    int             n_children;  /* if n_children==TREE_FANOUT+1 then the tree needs to be rebalanced. */
 	    unsigned int    totalchildkeylens;
-	    unsigned int    n_bytes_in_hashtables;
+	    unsigned int    n_bytes_in_buffers;
 
 	    struct brtnode_nonleaf_childinfo childinfos[TREE_FANOUT+1]; /* One extra so we can grow */
 
@@ -82,8 +82,8 @@ struct brtnode {
 						         However, in the absense of duplicate keys, child 1's keys *are* > childkeys[0]. */
 	    DISKOFF         children[TREE_FANOUT+1];  /* unused if height==0 */   /* Note: The last element of these arrays is used only temporarily while splitting a node. */
 #define BRTNODE_CHILD_DISKOFF(node,i) ((node)->u.n.children[i])
-	    HASHTABLE       htables[TREE_FANOUT+1];
-	    unsigned int    n_bytes_in_hashtable[TREE_FANOUT+1]; /* how many bytes are in each hashtable (including overheads) */
+	    FIFO            buffers[TREE_FANOUT+1];
+	    unsigned int    n_bytes_in_buffer[TREE_FANOUT+1]; /* how many bytes are in each buffer (including overheads) */
 	    unsigned int    n_cursors[TREE_FANOUT+1];
 #endif
         } n;
@@ -149,10 +149,6 @@ int toku_deserialize_brtheader_from (int fd, DISKOFF off, struct brt_header **br
 
 void toku_brtnode_free (BRTNODE *node);
 
-//static inline int brtnode_n_hashtables(BRTNODE node) { if (node->height==0) return 1; else return node->u.n.n_children; }
-
-//int write_brt_header (int fd, struct brt_header *header);
-
 #if 1
 #define DEADBEEF ((void*)0xDEADBEEF)
 #else

newbrt/brt-serialize-test.c

@@ -38,19 +38,19 @@ static void test_serialize(void) {
     sn.u.n.children[1] = sn.nodesize*35;
     BRTNODE_CHILD_SUBTREE_FINGERPRINTS(&sn, 0) = random();
     BRTNODE_CHILD_SUBTREE_FINGERPRINTS(&sn, 1) = random();
-    r = toku_hashtable_create(&sn.u.n.htables[0]); assert(r==0);
-    r = toku_hashtable_create(&sn.u.n.htables[1]); assert(r==0);
-    r = toku_hash_insert(sn.u.n.htables[0], "a", 2, "aval", 5, BRT_NONE); assert(r==0);    sn.local_fingerprint += randval*toku_calccrc32_cmd(BRT_NONE,   "a", 2, "aval", 5);
-    r = toku_hash_insert(sn.u.n.htables[0], "b", 2, "bval", 5, BRT_NONE); assert(r==0);    sn.local_fingerprint += randval*toku_calccrc32_cmd(BRT_NONE,   "b", 2, "bval", 5);
-    r = toku_hash_insert(sn.u.n.htables[1], "x", 2, "xval", 5, BRT_NONE); assert(r==0);    sn.local_fingerprint += randval*toku_calccrc32_cmd(BRT_NONE,   "x", 2, "xval", 5);
-    sn.u.n.n_bytes_in_hashtable[0] = 2*(BRT_CMD_OVERHEAD+KEY_VALUE_OVERHEAD+2+5);
-    sn.u.n.n_bytes_in_hashtable[1] = 1*(BRT_CMD_OVERHEAD+KEY_VALUE_OVERHEAD+2+5);
+    r = toku_fifo_create(&sn.u.n.buffers[0]); assert(r==0);
+    r = toku_fifo_create(&sn.u.n.buffers[1]); assert(r==0);
+    r = toku_fifo_enq(sn.u.n.buffers[0], "a", 2, "aval", 5, BRT_NONE); assert(r==0);    sn.local_fingerprint += randval*toku_calccrc32_cmd(BRT_NONE,   "a", 2, "aval", 5);
+    r = toku_fifo_enq(sn.u.n.buffers[0], "b", 2, "bval", 5, BRT_NONE); assert(r==0);    sn.local_fingerprint += randval*toku_calccrc32_cmd(BRT_NONE,   "b", 2, "bval", 5);
+    r = toku_fifo_enq(sn.u.n.buffers[1], "x", 2, "xval", 5, BRT_NONE); assert(r==0);    sn.local_fingerprint += randval*toku_calccrc32_cmd(BRT_NONE,   "x", 2, "xval", 5);
+    sn.u.n.n_bytes_in_buffer[0] = 2*(BRT_CMD_OVERHEAD+KEY_VALUE_OVERHEAD+2+5);
+    sn.u.n.n_bytes_in_buffer[1] = 1*(BRT_CMD_OVERHEAD+KEY_VALUE_OVERHEAD+2+5);
     {
 	int i;
 	for (i=2; i<TREE_FANOUT+1; i++)
-	    sn.u.n.n_bytes_in_hashtable[i]=0;
+	    sn.u.n.n_bytes_in_buffer[i]=0;
     }
-    sn.u.n.n_bytes_in_hashtables = 3*(BRT_CMD_OVERHEAD+KEY_VALUE_OVERHEAD+2+5);
+    sn.u.n.n_bytes_in_buffers = 3*(BRT_CMD_OVERHEAD+KEY_VALUE_OVERHEAD+2+5);
 
     toku_serialize_brtnode_to(fd, sn.nodesize*20, sn.nodesize, &sn);  assert(r==0);
 
@@ -75,32 +75,34 @@ static void test_serialize(void) {
 	}
 	assert(dn->local_fingerprint==sn.local_fingerprint);
     }
+#if 0
     {
 	bytevec data; ITEMLEN datalen; int type;
-	r = toku_hash_find(dn->u.n.htables[0], "a", 2, &data, &datalen, &type);
+	r = toku_hash_find(dn->u.n.buffers[0], "a", 2, &data, &datalen, &type);
 	assert(r==0);
 	assert(strcmp(data,"aval")==0);
 	assert(datalen==5);
         assert(type == BRT_NONE);
 
-	r=toku_hash_find(dn->u.n.htables[0], "b", 2, &data, &datalen, &type);
+	r=toku_hash_find(dn->u.n.buffers[0], "b", 2, &data, &datalen, &type);
 	assert(r==0);
 	assert(strcmp(data,"bval")==0);
 	assert(datalen==5);
         assert(type == BRT_NONE);
 
-	r=toku_hash_find(dn->u.n.htables[1], "x", 2, &data, &datalen, &type);
+	r=toku_hash_find(dn->u.n.buffers[1], "x", 2, &data, &datalen, &type);
 	assert(r==0);
 	assert(strcmp(data,"xval")==0);
 	assert(datalen==5);
         assert(type == BRT_NONE);
     }
+#endif
     toku_brtnode_free(&dn);
 
     kv_pair_free(sn.u.n.childkeys[0]);
     toku_free(hello_string);
-    toku_hashtable_free(&sn.u.n.htables[0]);
-    toku_hashtable_free(&sn.u.n.htables[1]);
+    toku_fifo_free(&sn.u.n.buffers[0]);
+    toku_fifo_free(&sn.u.n.buffers[1]);
 }
 
 int main (int argc __attribute__((__unused__)), char *argv[] __attribute__((__unused__))) {

newbrt/brt-serialize.c

@@ -42,17 +42,17 @@ static unsigned int toku_serialize_brtnode_size_slow(BRTNODE node) {
 	    size+=8; // diskoff
 	    size+=4; // subsum
 	}
-	int n_hashtables = node->u.n.n_children;
+	int n_buffers = node->u.n.n_children;
 	size+=4; /* n_entries */
-        assert(0 <= n_hashtables && n_hashtables < TREE_FANOUT+1);
-	for (i=0; i< n_hashtables; i++) {
-	    HASHTABLE_ITERATE(node->u.n.htables[i],
+        assert(0 <= n_buffers && n_buffers < TREE_FANOUT+1);
+	for (i=0; i< n_buffers; i++) {
+	    FIFO_ITERATE(node->u.n.buffers[i],
 			      key __attribute__((__unused__)), keylen,
 			      data __attribute__((__unused__)), datalen,
                               type __attribute__((__unused__)),
 			      (hsize+=BRT_CMD_OVERHEAD+KEY_VALUE_OVERHEAD+keylen+datalen));
 	}
-	assert(hsize==node->u.n.n_bytes_in_hashtables);
+	assert(hsize==node->u.n.n_bytes_in_buffers);
 	assert(csize==node->u.n.totalchildkeylens);
 	return size+hsize+csize;
     } else {
@@ -78,7 +78,7 @@ unsigned int toku_serialize_brtnode_size (BRTNODE node) {
         if (node->flags & TOKU_DB_DUPSORT) result += 4*(node->u.n.n_children-1); /* data lengths */
 	result+=node->u.n.totalchildkeylens; /* the lengths of the pivot keys, without their key lengths. */
 	result+=(8+4+4)*(node->u.n.n_children); /* For each child, a child offset, a count for the number of hash table entries, and the subtree fingerprint. */
-	result+=node->u.n.n_bytes_in_hashtables;
+	result+=node->u.n.n_bytes_in_buffers;
     } else {
 	result+=(4 /* n_entries in buffer table. */
 		 +4); /* the pma size */
@@ -147,12 +147,12 @@ void toku_serialize_brtnode_to(int fd, DISKOFF off, DISKOFF size, BRTNODE node)
 	}
 
 	{
-	    int n_hash_tables = node->u.n.n_children;
+	    int n_buffers = node->u.n.n_children;
 	    u_int32_t check_local_fingerprint = 0;
-	    for (i=0; i< n_hash_tables; i++) {
+	    for (i=0; i< n_buffers; i++) {
 		//printf("%s:%d p%d=%p n_entries=%d\n", __FILE__, __LINE__, i, node->mdicts[i], mdict_n_entries(node->mdicts[i]));
-		wbuf_int(&w, toku_hashtable_n_entries(node->u.n.htables[i]));
-		HASHTABLE_ITERATE(node->u.n.htables[i], key, keylen, data, datalen, type,
+		wbuf_int(&w, toku_fifo_n_entries(node->u.n.buffers[i]));
+		FIFO_ITERATE(node->u.n.buffers[i], key, keylen, data, datalen, type,
 				  ({
 				      wbuf_char(&w, type);
 				      wbuf_bytes(&w, key, keylen);
@@ -281,8 +281,8 @@ int toku_deserialize_brtnode_from (int fd, DISKOFF off, BRTNODE *brtnode, int fl
         }
 	for (i=0; i<TREE_FANOUT+1; i++) { 
             result->u.n.children[i]=0; 
-            result->u.n.htables[i]=0; 
-            result->u.n.n_bytes_in_hashtable[i]=0;
+            result->u.n.buffers[i]=0; 
+            result->u.n.n_bytes_in_buffer[i]=0;
             result->u.n.n_cursors[i]=0;
         }
 	u_int32_t subtree_fingerprint = rbuf_int(&rc);
@@ -316,15 +316,15 @@ int toku_deserialize_brtnode_from (int fd, DISKOFF off, BRTNODE *brtnode, int fl
 	    //printf("Child %d at %lld\n", i, result->children[i]);
 	}
 	for (i=0; i<TREE_FANOUT+1; i++) {
-	    result->u.n.n_bytes_in_hashtable[i] = 0;
+	    result->u.n.n_bytes_in_buffer[i] = 0;
 	}
-	result->u.n.n_bytes_in_hashtables = 0; 
+	result->u.n.n_bytes_in_buffers = 0; 
 	for (i=0; i<result->u.n.n_children; i++) {
-	    r=toku_hashtable_create(&result->u.n.htables[i]);
+	    r=toku_fifo_create(&result->u.n.buffers[i]);
 	    if (r!=0) {
 		int j;
-		if (0) { died_12: j=result->u.n.n_bytes_in_hashtables; }
-		for (j=0; j<i; j++) toku_hashtable_free(&result->u.n.htables[j]);
+		if (0) { died_12: j=result->u.n.n_bytes_in_buffers; }
+		for (j=0; j<i; j++) toku_fifo_free(&result->u.n.buffers[j]);
 		goto died1;
 	    }
 	}
@@ -346,12 +346,12 @@ int toku_deserialize_brtnode_from (int fd, DISKOFF off, BRTNODE *brtnode, int fl
 		    check_local_fingerprint += result->rand4fingerprint * toku_calccrc32_cmd(type, key, keylen, val, vallen);
 		    //printf("Found %s,%s\n", (char*)key, (char*)val);
 		    {
-			r=toku_hash_insert(result->u.n.htables[cnum], key, keylen, val, vallen, type); /* Copies the data into the hash table. */
+			r=toku_fifo_enq(result->u.n.buffers[cnum], key, keylen, val, vallen, type); /* Copies the data into the hash table. */
 			if (r!=0) { goto died_12; }
 		    }
 		    diff =  keylen + vallen + KEY_VALUE_OVERHEAD + BRT_CMD_OVERHEAD;
-		    result->u.n.n_bytes_in_hashtables += diff;
-		    result->u.n.n_bytes_in_hashtable[cnum] += diff;
+		    result->u.n.n_bytes_in_buffers += diff;
+		    result->u.n.n_bytes_in_buffer[cnum] += diff;
 		    //printf("Inserted\n");
 		}
 	    }
@@ -455,13 +455,13 @@ void toku_verify_counts (BRTNODE node) {
 	unsigned int sum = 0;
 	int i;
 	for (i=0; i<node->u.n.n_children; i++)
-	    sum += node->u.n.n_bytes_in_hashtable[i];
-	// We don't rally care of the later hashtables have garbage in them.  Valgrind would do a better job noticing if we leave it uninitialized.
+	    sum += node->u.n.n_bytes_in_buffer[i];
+	// We don't rally care of the later buffers have garbage in them.  Valgrind would do a better job noticing if we leave it uninitialized.
 	// But for now the code always initializes the later tables so they are 0.
 	for (; i<TREE_FANOUT+1; i++) {
-	    assert(node->u.n.n_bytes_in_hashtable[i]==0);
+	    assert(node->u.n.n_bytes_in_buffer[i]==0);
         }
-	assert(sum==node->u.n.n_bytes_in_hashtables);
+	assert(sum==node->u.n.n_bytes_in_buffers);
     }
 }
     

newbrt/brt-verify.c

@@ -24,7 +24,7 @@ static void verify_local_fingerprint (BRTNODE node) {
     int i;
     if (node->height>0) {
 	for (i=0; i<node->u.n.n_children; i++)
-	    HASHTABLE_ITERATE(node->u.n.htables[i], key, keylen, data, datalen, type,
+	    FIFO_ITERATE(node->u.n.buffers[i], key, keylen, data, datalen, type,
 			      ({
 				  fp += node->rand4fingerprint * toku_calccrc32_cmd(type, key, keylen, data, datalen);
 			      }));
@@ -76,7 +76,7 @@ int toku_verify_brtnode (BRT brt, DISKOFF off, bytevec lorange, ITEMLEN lolen, b
 			result=1;
 		    }
 		}
-		toku_hashtable_iterate(node->u.n.htables[i], verify_pair, 0);
+		toku_fifo_iterate(node->u.n.buffers[i], verify_pair, 0);
 	    }
 	}
 	for (i=0; i<node->u.n.n_children; i++) {

newbrt/fifo.c

+#include <stdlib.h>
+#include <string.h>
+#include <assert.h>
+#include <errno.h>
+
+#include "memory.h"
+#include "fifo.h"
+
+static void fifo_init(struct fifo *fifo) {
+    fifo->head = fifo->tail = 0;
+    fifo->n = 0;
+}
+
+static struct fifo_entry *fifo_peek(struct fifo *fifo) {
+    return fifo->head;
+}
+
+static void fifo_enq(struct fifo *fifo, struct fifo_entry *entry) {
+    entry->next = 0;
+    if (fifo->head == 0)
+        fifo->head = entry;
+    else
+        fifo->tail->next = entry;
+    fifo->tail = entry;
+    fifo->n += 1;
+}
+
+static struct fifo_entry *fifo_deq(struct fifo *fifo) {
+    struct fifo_entry *entry = fifo->head;
+    if (entry) {
+        fifo->head = entry->next;
+        if (fifo->head == 0)
+            fifo->tail = 0;
+        fifo->n -= 1;
+        assert(fifo->n >= 0);
+    }
+    return entry;
+}
+
+static void fifo_destroy(struct fifo *fifo) {
+    struct fifo_entry *entry;
+    while ((entry = fifo_deq(fifo)) != 0)
+        toku_free(entry);
+}
+
+int toku_fifo_create(FIFO *ptr) {
+    struct fifo *fifo = toku_malloc(sizeof (struct fifo));
+    if (fifo == 0) return ENOMEM;
+    fifo_init(fifo);
+    *ptr = fifo;
+    return 0;
+}
+
+void toku_fifo_free(FIFO *ptr) {
+    struct fifo *fifo = *ptr; *ptr = 0;
+    fifo_destroy(fifo);
+    toku_free(fifo);
+}
+
+int toku_fifo_n_entries(FIFO fifo) {
+    return fifo->n;
+}
+
+int toku_fifo_enq(FIFO fifo, const void *key, unsigned int keylen, const void *data, unsigned int datalen, int type) {
+    struct fifo_entry *entry = toku_malloc(sizeof (struct fifo_entry) + keylen + datalen);
+    if (entry == 0) return ENOMEM;
+    entry->type = type;
+    entry->keylen = keylen;
+    memcpy(entry->key, key, keylen);
+    entry->vallen = datalen;
+    memcpy(entry->key + keylen, data, datalen);
+    fifo_enq(fifo, entry);
+    return 0;
+}
+
+/* peek at the head of the fifo */
+int toku_fifo_peek(FIFO fifo, bytevec *key, unsigned int *keylen, bytevec *data, unsigned int *datalen, int *type) {
+    struct fifo_entry *entry = fifo_peek(fifo);
+    if (entry == 0) return -1;
+    *key = entry->key;
+    *keylen = entry->keylen;
+    *data = entry->key + entry->keylen;
+    *datalen = entry->vallen;
+    *type = entry->type;
+    return 0;
+}
+
+int toku_fifo_deq(FIFO fifo) {
+    struct fifo_entry *entry = fifo_deq(fifo);
+    if (entry == 0) return ENOMEM;
+    toku_free(entry);
+    return 0;
+}
+
+void toku_fifo_iterate (FIFO fifo, void(*f)(bytevec key,ITEMLEN keylen,bytevec data,ITEMLEN datalen,int type, void*), void *arg) {
+    struct fifo_entry *entry;
+    for (entry = fifo_peek(fifo); entry; entry = entry->next)
+        f(entry->key, entry->keylen, entry->key + entry->keylen, entry->vallen, entry->type, arg);
+}
+
+
+
+
+    

newbrt/fifo.h

+#include "brttypes.h"
+
+struct fifo_entry {
+    struct fifo_entry *next;
+    unsigned int keylen;
+    unsigned int vallen;
+    unsigned char type;
+    unsigned char key[];
+};
+
+struct fifo {
+    struct fifo_entry *head, *tail;
+    int n;
+};
+
+typedef struct fifo *FIFO;
+
+int toku_fifo_create(FIFO *);
+void toku_fifo_free(FIFO *);
+int toku_fifo_n_entries(FIFO);
+int toku_fifo_enq (FIFO, const void *key, ITEMLEN keylen, const void *data, ITEMLEN datalen, int type);
+int toku_fifo_peek (FIFO, bytevec *key, ITEMLEN *keylen, bytevec *data, ITEMLEN *datalen, int *type);
+int toku_fifo_deq(FIFO);
+int toku_fifo_peek_deq (FIFO, bytevec *key, ITEMLEN *keylen, bytevec *data, ITEMLEN *datalen, int *type);
+void toku_fifo_iterate (FIFO, void(*f)(bytevec key,ITEMLEN keylen,bytevec data,ITEMLEN datalen,int type, void*), void*);
+
+#define FIFO_ITERATE(fifo,keyvar,keylenvar,datavar,datalenvar,typevar,body) ({ \
+            struct fifo_entry *entry; \
+            for (entry = fifo->head; entry; entry = entry->next) { \
+                unsigned int keylenvar = entry->keylen; \
+                void *keyvar = entry->key; \
+                unsigned int datalenvar = entry->vallen; \
+                void *datavar = entry->key + entry->keylen; \
+                unsigned int typevar = entry->type; \
+                body; \
+            } \
+        })
+