Change the {{{LEAFENTRY}}} to {{{OMTVALUE}}} when using the omt, and also...

Change the {{{LEAFENTRY}}} to {{{OMTVALUE}}} when using the omt, and also change the type of {{{OMTVALUE}}} to {{{void*}}} so we can have two different OMTs coexisting.  Fixes #750.

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

newbrt/brt-serialize.c

@@ -55,7 +55,8 @@ static unsigned int toku_serialize_brtnode_size_slow (BRTNODE node) {
 	return size+hsize+csize;
     } else {
 	unsigned int hsize=0;
-	int addupsize (LEAFENTRY le, u_int32_t UU(idx), void *vp) {
+	int addupsize (OMTVALUE lev, u_int32_t UU(idx), void *vp) {
+	    LEAFENTRY le=lev;
 	    unsigned int *ip=vp;
 	    (*ip) += OMT_ITEM_OVERHEAD + leafentry_disksize(le);
 	    return 0;
@@ -177,7 +178,8 @@ void toku_serialize_brtnode_to (int fd, DISKOFF off, BRTNODE node) {
     } else {
 	//printf("%s:%d writing node %lld n_entries=%d\n", __FILE__, __LINE__, node->thisnodename, toku_gpma_n_entries(node->u.l.buffer));
 	wbuf_uint(&w, toku_omt_size(node->u.l.buffer));
-	int wbufwriteleafentry (LEAFENTRY le, u_int32_t UU(idx), void *v) {
+	int wbufwriteleafentry (OMTVALUE lev, u_int32_t UU(idx), void *v) {
+	    LEAFENTRY le=lev;
 	    struct wbuf *thisw=v;
 	    wbuf_LEAFENTRY(thisw, le);
 	    return 0;
@@ -440,7 +442,8 @@ void toku_verify_counts (BRTNODE node) {
 	    unsigned int count;
 	    u_int32_t    fp;
 	} sum_info = {0,0,0,0};
-	int sum_item (LEAFENTRY le, u_int32_t UU(idx), void *vsi) {
+	int sum_item (OMTVALUE lev, u_int32_t UU(idx), void *vsi) {
+	    LEAFENTRY le=lev;
 	    struct sum_info *si = vsi;
 	    si->count++;
 	    si->dsum += OMT_ITEM_OVERHEAD + leafentry_disksize(le);

newbrt/brt-test-helpers.c

@@ -84,16 +84,17 @@ int toku_testsetup_insert_to_leaf (BRT brt, DISKOFF diskoff, char *key, int keyl
     memcpy(leafentry, tmp_leafentry, lesize);
     toku_free(tmp_leafentry);
 
-    LEAFENTRY storeddata;
+    OMTVALUE storeddatav;
     u_int32_t idx;
     DBT keydbt,valdbt;
     BRT_CMD_S cmd = {BRT_INSERT, 0, .u.id={toku_fill_dbt(&keydbt, key, keylen),
 					   toku_fill_dbt(&valdbt, val, vallen)}};
     struct cmd_leafval_bessel_extra be = {brt, &cmd, node->flags & TOKU_DB_DUPSORT};
-    r = toku_omt_find_zero(node->u.l.buffer, toku_cmd_leafval_bessel, &be, &storeddata, &idx);
+    r = toku_omt_find_zero(node->u.l.buffer, toku_cmd_leafval_bessel, &be, &storeddatav, &idx);
 
 
     if (r==0) {
+	LEAFENTRY storeddata=storeddatav;
 	// It's already there.  So now we have to remove it and put the new one back in.
 	node->u.l.n_bytes_in_buffer -= OMT_ITEM_OVERHEAD + leafentry_disksize(storeddata);
 	node->local_fingerprint     -= node->rand4fingerprint*toku_le_crc(storeddata);

newbrt/brt-verify.c

@@ -127,7 +127,8 @@ int toku_verify_brtnode (BRT brt, DISKOFF off, bytevec lorange, ITEMLEN lolen, b
 	}
     } else {
 	// Make sure that they are in increasing order.
-	int check_increasing (LEAFENTRY v, u_int32_t idx, void *vprevp) {
+	int check_increasing (OMTVALUE lev, u_int32_t idx, void *vprevp) {
+	    LEAFENTRY v=lev;
 	    LEAFENTRY *prevp = vprevp;
 	    if (idx>0) 
 		assert(compare_leafentries(brt, *prevp, v)<0);

newbrt/brt.c

@@ -89,7 +89,8 @@ static long brtnode_size(BRTNODE node) {
 }
 
 
-static int verify_in_mempool(LEAFENTRY le, u_int32_t UU(idx), void *vmp) {
+static int verify_in_mempool(OMTVALUE lev, u_int32_t UU(idx), void *vmp) {
+    LEAFENTRY le=lev;
     struct mempool *mp=vmp;
     assert(toku_mempool_inrange(mp, le, leafentry_memsize(le)));
     return 0;
@@ -355,7 +356,8 @@ static int insert_to_buffer_in_nonleaf (BRTNODE node, int childnum, DBT *k, DBT
     return 0;
 }
 
-static int fill_buf (LEAFENTRY le, u_int32_t idx, void *varray) {
+static int fill_buf (OMTVALUE lev, u_int32_t idx, void *varray) {
+    LEAFENTRY le=lev;
     LEAFENTRY *array=varray;
     array[idx]=le;
     return 0;
@@ -378,7 +380,7 @@ static int brtleaf_split (TOKULOGGER logger, FILENUM filenum, BRT t, BRTNODE nod
     toku_verify_all_in_mempool(node);
 
     u_int32_t n_leafentries = toku_omt_size(node->u.l.buffer);
-    LEAFENTRY *MALLOC_N(n_leafentries, leafentries);
+    OMTVALUE *MALLOC_N(n_leafentries, leafentries);
     assert(leafentries);
     toku_omt_iterate(node->u.l.buffer, fill_buf, leafentries);
     u_int32_t break_at = 0;
@@ -440,9 +442,10 @@ static int brtleaf_split (TOKULOGGER logger, FILENUM filenum, BRT t, BRTNODE nod
     //toku_verify_gpma(B->u.l.buffer);
     if (splitk) {
 	memset(splitk, 0, sizeof *splitk);
-	LEAFENTRY le;
-	r=toku_omt_fetch(node->u.l.buffer, toku_omt_size(node->u.l.buffer)-1, &le);
+	OMTVALUE lev;
+	r=toku_omt_fetch(node->u.l.buffer, toku_omt_size(node->u.l.buffer)-1, &lev);
 	assert(r==0); // that fetch should have worked.
+	LEAFENTRY le=lev;
 	if (node->flags&TOKU_DB_DUPSORT) {
 	    splitk->size = le_any_keylen(le)+le_any_vallen(le);
 	    splitk->data = kv_pair_malloc(le_any_key(le), le_any_keylen(le), le_any_val(le), le_any_vallen(le));
@@ -1109,7 +1112,8 @@ int leafval_bessel_le_provpair (TXNID xid __attribute__((__unused__)),
     return leafval_bessel_le_committed(klen, kval, plen, pval, be);
 }
 
-int toku_cmd_leafval_bessel (LEAFENTRY le, void *extra) {
+int toku_cmd_leafval_bessel (OMTVALUE lev, void *extra) {
+    LEAFENTRY le=lev;
     struct cmd_leafval_bessel_extra *be = extra;
     LESWITCHCALL(le, leafval_bessel, be);
 }
@@ -1358,6 +1362,8 @@ static int brt_leaf_put_cmd (BRT t, BRTNODE node, BRT_CMD cmd,
     FILENUM filenum = toku_cachefile_filenum(t->cf);
 
     LEAFENTRY storeddata;
+    OMTVALUE storeddatav;
+
     u_int32_t idx;
     int r;
     int compare_both = should_compare_both_keys(node, cmd);
@@ -1367,11 +1373,14 @@ static int brt_leaf_put_cmd (BRT t, BRTNODE node, BRT_CMD cmd,
     case BRT_INSERT:
 	
 	r = toku_omt_find_zero(node->u.l.buffer, toku_cmd_leafval_bessel, &be,
-			       &storeddata, &idx);
+			       &storeddatav, &idx);
 	if (r==DB_NOTFOUND) {
 	    storeddata = 0;
-	} else if (r!=0)
+	} else if (r!=0) {
 	    return r;
+	} else {
+	    storeddata=storeddatav;
+	}
 	
 	r = brt_leaf_apply_cmd_once(t, node, cmd, logger, idx, storeddata);
 	if (r!=0) return r;
@@ -1382,9 +1391,10 @@ static int brt_leaf_put_cmd (BRT t, BRTNODE node, BRT_CMD cmd,
 
 	// Delete the one item
 	r = toku_omt_find_zero(node->u.l.buffer, toku_cmd_leafval_bessel,  &be,
-			       &storeddata, &idx);
+			       &storeddatav, &idx);
 	if (r == DB_NOTFOUND) break;
 	if (r != 0) return r;
+	storeddata=storeddatav;
 
 	VERIFY_NODE(node);
 
@@ -1402,9 +1412,10 @@ static int brt_leaf_put_cmd (BRT t, BRTNODE node, BRT_CMD cmd,
 	// Delete all the matches
 
 	r = toku_omt_find_zero(node->u.l.buffer, toku_cmd_leafval_bessel, &be,
-			       &storeddata, &idx);
+			       &storeddatav, &idx);
 	if (r == DB_NOTFOUND) break;
 	if (r != 0) return r;
+	storeddata=storeddatav;
 	    
 	while (1) {
 	    int   vallen   = le_any_vallen(storeddata);
@@ -1418,10 +1429,11 @@ static int brt_leaf_put_cmd (BRT t, BRTNODE node, BRT_CMD cmd,
 	    BRT_CMD_S ncmd = { cmd->type, cmd->xid, .u.id={cmd->u.id.key, toku_fill_dbt(&valdbt, save_val, vallen)}};
 	    struct cmd_leafval_bessel_extra nbe = {t, &ncmd, 1};
 	    r = toku_omt_find(node->u.l.buffer, toku_cmd_leafval_bessel,  &nbe, +1,
-			      &storeddata, &idx);
+			      &storeddatav, &idx);
 	    
 	    toku_free(save_val);
 	    if (r!=0) break;
+	    storeddata=storeddatav;
 	    {   // Continue only if the next record that we found has the same key.
 		DBT adbt;
 		if (t->compare_fun(t->db,
@@ -2565,7 +2577,8 @@ int pair_leafval_bessel_le_provpair (TXNID xid __attribute__((__unused__)),
 }
 
 
-static int bessel_from_search_t (LEAFENTRY leafval, void *extra) {
+static int bessel_from_search_t (OMTVALUE lev, void *extra) {
+    LEAFENTRY leafval=lev;
     brt_search_t *search = extra;
     LESWITCHCALL(leafval, pair_leafval_bessel, search);
 }
@@ -2579,16 +2592,16 @@ static int brt_search_leaf_node(BRT brt, BRTNODE node, brt_search_t *search, DBT
     }
     return EINVAL;  // This return and the goto are a hack to get both compile-time and run-time checking on enum
  ok: ;
-    LEAFENTRY data;
+    OMTVALUE datav;
     u_int32_t idx;
     int r = toku_omt_find(node->u.l.buffer,
 			  bessel_from_search_t,
 			  search,
 			  direction,
-			  &data, &idx);
+			  &datav, &idx);
     if (r!=0) return r;
 
-    LEAFENTRY le = data;
+    LEAFENTRY le = datav;
     if (le_is_provdel(le)) {
 	// Provisionally deleted stuff is gone.
 	// So we need to scan in the direction to see if we can find something
@@ -2604,9 +2617,9 @@ static int brt_search_leaf_node(BRT brt, BRTNODE node, brt_search_t *search, DBT
 		break;
 	    }
 	    if (idx>=toku_omt_size(node->u.l.buffer)) continue;
-	    r = toku_omt_fetch(node->u.l.buffer, idx, &data);
+	    r = toku_omt_fetch(node->u.l.buffer, idx, &datav);
 	    assert(r==0); // we just validated the index
-	    le = data;
+	    le = datav;
 	    if (!le_is_provdel(le)) goto got_a_good_value;
 	}
     }
@@ -3050,7 +3063,8 @@ struct omt_compressor_state {
     OMT omt;
 };
 
-static int move_it (LEAFENTRY le, u_int32_t idx, void *v) {
+static int move_it (OMTVALUE lev, u_int32_t idx, void *v) {
+    LEAFENTRY le=lev;
     struct omt_compressor_state *oc = v;
     u_int32_t size = leafentry_memsize(le);
     LEAFENTRY newdata = toku_mempool_malloc(oc->new_kvspace, size, 1);

newbrt/brtdump.c

@@ -109,7 +109,8 @@ void dump_node (int f, DISKOFF off) {
     } else {
 	printf(" n_bytes_in_buffer=%d\n", n->u.l.n_bytes_in_buffer);
 	printf(" items_in_buffer  =%d\n", toku_omt_size(n->u.l.buffer));
-	int print_le(LEAFENTRY le, u_int32_t UU(idx), void *UU(v)) {
+	int print_le(OMTVALUE lev, u_int32_t UU(idx), void *UU(v)) {
+	    LEAFENTRY le=lev;
 	    print_leafentry(stdout, le);
 	    printf("\n");
 	    return 0;

newbrt/log-internal.h

 #ident "Copyright (c) 2007 Tokutek Inc.  All rights reserved."
 
+#include "brt-internal.h"
 #include "log.h"
 #include "toku_assert.h"
 #include "list.h"
@@ -73,6 +74,11 @@ struct tokulogger {
 int toku_logger_find_next_unused_log_file(const char *directory, long long *result);
 int toku_logger_find_logfiles (const char *directory, char ***resultp);
 
+struct brtcachefile_pair {
+    BRT brt;
+    CACHEFILE cf;
+};
+
 struct tokutxn {
     enum typ_tag tag;
     u_int64_t txnid64; /* this happens to be the first lsn */
@@ -86,6 +92,7 @@ struct tokutxn {
     char      *rollentry_filename;
     int        rollentry_fd;         // If we spill the roll_entries, we write them into this fd.
     off_t      rollentry_filesize;   // How many bytes are in the rollentry.
+    OMT        used_open_brtcachefile_pairs; // a collection of the brts that we touched and which are still open.
 };
 
 int toku_logger_finish (TOKULOGGER logger, struct logbytes *logbytes, struct wbuf *wbuf, int do_fsync);

newbrt/recover.c

@@ -9,11 +9,12 @@
 
 #include "cachetable.h"
 #include "key.h"
+#include "brt-internal.h"
 #include "log-internal.h"
 #include "log_header.h"
 #include "toku_assert.h"
 #include "kv-pair.h"
-#include "omt.h"
+
 
 #include <fcntl.h>
 #include <stdlib.h>
@@ -488,7 +489,8 @@ void toku_recover_cfclose (LSN UU(lsn), BYTESTRING UU(fname), FILENUM filenum) {
     toku_free_BYTESTRING(fname);
 }
 
-static int fill_buf (LEAFENTRY le, u_int32_t idx, void *varray) {
+static int fill_buf (OMTVALUE lev, u_int32_t idx, void *varray) {
+    LEAFENTRY le=lev;
     LEAFENTRY *array=varray;
     array[idx]=le;
     return 0;
@@ -530,7 +532,7 @@ void toku_recover_leafsplit (LSN lsn, FILENUM filenum, DISKOFF old_diskoff, DISK
     assert(toku_omt_size(oldn->u.l.buffer)==old_n);
 
     u_int32_t n_leafentries = old_n;
-    LEAFENTRY *MALLOC_N(n_leafentries, leafentries);
+    OMTVALUE *MALLOC_N(n_leafentries, leafentries);
     assert(leafentries);
     toku_omt_iterate(oldn->u.l.buffer, fill_buf, leafentries);
     {
@@ -619,15 +621,16 @@ void toku_recover_deleteleafentry (LSN lsn, FILENUM filenum, DISKOFF diskoff, u_
     VERIFY_COUNTS(node);
     node->log_lsn = lsn;
     {
-	LEAFENTRY data;
+	OMTVALUE data;
 	r=toku_omt_fetch(node->u.l.buffer, idx, &data);
 	assert(r==0);
 	u_int32_t  len = leafentry_memsize(oldleafentry);
-	assert(leafentry_memsize(data)==len);
+	LEAFENTRY le=data;
+	assert(leafentry_memsize(le)==len);
 	assert(memcmp(oldleafentry, data, len)==0);
-	node->u.l.n_bytes_in_buffer -= OMT_ITEM_OVERHEAD + leafentry_disksize(data);
-	node->local_fingerprint -= node->rand4fingerprint * toku_le_crc(data);
-	toku_mempool_mfree(&node->u.l.buffer_mempool, data, len);
+	node->u.l.n_bytes_in_buffer -= OMT_ITEM_OVERHEAD + leafentry_disksize(le);
+	node->local_fingerprint -= node->rand4fingerprint * toku_le_crc(le);
+	toku_mempool_mfree(&node->u.l.buffer_mempool, le, len);
 	r = toku_omt_delete_at(node->u.l.buffer, idx);
 	assert(r==0);
     }