Stop passing DB* everywhere. Instead store it in the BRT and PMA structs. Fixes #39.

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

newbrt/Makefile

@@ -85,7 +85,7 @@ BRT_INTERNAL_H_INCLUDES = brt-internal.h cachetable.h hashtable.h pma.h brt.h br
 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
-pma.o: pma.h yerror.h pma-internal.h memory.h key.h ybt.h brttypes.h log.h ../include/db.h
+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

newbrt/benchmark-test.c

@@ -31,7 +31,7 @@ void setup (void) {
     int r;
     unlink(fname);
     r = brt_create_cachetable(&ct, 0, ZERO_LSN, NULL_LOGGER);         assert(r==0);
-    r = open_brt(fname, 0, 1, &t, nodesize, ct, NULL_TXN, toku_default_compare_fun); assert(r==0);
+    r = open_brt(fname, 0, 1, &t, nodesize, ct, NULL_TXN, toku_default_compare_fun, (DB*)0); assert(r==0);
 }
 
 void shutdown (void) {
@@ -52,7 +52,7 @@ void insert (long long v) {
     long_long_to_array(kc, v);
     memset(vc, 0, sizeof vc);
     long_long_to_array(vc, v);
-    brt_insert(t, fill_dbt(&kt, kc, keysize), fill_dbt(&vt, vc, valsize), 0, 0);
+    brt_insert(t, fill_dbt(&kt, kc, keysize), fill_dbt(&vt, vc, valsize), 0);
     if (do_verify) toku_cachetable_verify(ct);
 }
 

newbrt/brt-internal.h

@@ -98,13 +98,14 @@ struct brt {
     unsigned int flags;
     int (*compare_fun)(DB*,const DBT*,const DBT*);
     int (*dup_compare)(DB*,const DBT*,const DBT*);
+    DB *db;           // To pass to the compare fun
 
     void *skey,*sval; /* Used for DBT return values. */
 };
 
 /* serialization code */
 void toku_serialize_brtnode_to(int fd, DISKOFF off, DISKOFF size, BRTNODE node);
-int toku_deserialize_brtnode_from (int fd, DISKOFF off, BRTNODE *brtnode, int flags, int nodesize, int (*bt_compare)(DB *, const DBT*, const DBT*), int (*dup_compare)(DB *, const DBT *, const DBT *));
+int toku_deserialize_brtnode_from (int fd, DISKOFF off, BRTNODE *brtnode, int flags, int nodesize, int (*bt_compare)(DB *, const DBT*, const DBT*), int (*dup_compare)(DB *, const DBT *, const DBT *), DB *db, FILENUM filenum);
 unsigned int toku_serialize_brtnode_size(BRTNODE node); /* How much space will it take? */
 int toku_keycompare (bytevec key1, ITEMLEN key1len, bytevec key2, ITEMLEN key2len);
 
@@ -189,7 +190,6 @@ struct brt_cmd {
         struct brt_cmd_insert_delete {
             DBT *key;
             DBT *val;
-            DB *db;
         } id;
     } u;
 };
@@ -213,4 +213,5 @@ static const BRTNODE null_brtnode=0;
 extern u_int32_t toku_calccrc32_kvpair (const void *key, int keylen, const void *val, int vallen);
 extern u_int32_t toku_calccrc32_cmd (int type, const void *key, int keylen, const void *val, int vallen);
 extern u_int32_t toku_calccrc32_cmdstruct (BRT_CMD *cmd);
+
 #endif

newbrt/brt-serialize-test.c

@@ -44,7 +44,7 @@ void test_serialize(void) {
 
     toku_serialize_brtnode_to(fd, sn.nodesize*20, sn.nodesize, &sn);  assert(r==0);
 
-    r = toku_deserialize_brtnode_from(fd, nodesize*20, &dn, 0, nodesize, 0, 0);
+    r = toku_deserialize_brtnode_from(fd, nodesize*20, &dn, 0, nodesize, 0, 0, 0, (FILENUM){0});
     assert(r==0);
 
     assert(dn->thisnodename==nodesize*20);

newbrt/brt-serialize.c

@@ -182,8 +182,9 @@ void toku_serialize_brtnode_to(int fd, DISKOFF off, DISKOFF size, BRTNODE node)
 }
 
 int toku_deserialize_brtnode_from (int fd, DISKOFF off, BRTNODE *brtnode, int flags, int nodesize,
-                              int (*bt_compare)(DB *, const DBT *, const DBT *),
-                              int (*dup_compare)(DB *, const DBT *, const DBT *)) {
+				   int (*bt_compare)(DB *, const DBT *, const DBT *),
+				   int (*dup_compare)(DB *, const DBT *, const DBT *),
+				   DB *db, FILENUM filenum) {
     TAGMALLOC(BRTNODE, result);
     struct rbuf rc;
     int i;
@@ -342,7 +343,7 @@ int toku_deserialize_brtnode_from (int fd, DISKOFF off, BRTNODE *brtnode, int fl
     } else {
 	int n_in_buf = rbuf_int(&rc);
 	result->u.l.n_bytes_in_buffer = 0;
-	r=toku_pma_create(&result->u.l.buffer, bt_compare, nodesize);
+	r=toku_pma_create(&result->u.l.buffer, bt_compare, db, filenum, nodesize);
 	if (r!=0) {
 	    if (0) { died_21: toku_pma_free(&result->u.l.buffer); }
 	    goto died1;

newbrt/brt.h

@@ -9,7 +9,7 @@
 #include "cachetable.h"
 #include "log.h"
 
-int open_brt (const char *fname, const char *dbname, int is_create, BRT *, int nodesize, CACHETABLE, TOKUTXN, int(*)(DB*,const DBT*,const DBT*));
+int open_brt (const char *fname, const char *dbname, int is_create, BRT *, int nodesize, CACHETABLE, TOKUTXN, int(*)(DB*,const DBT*,const DBT*), DB*);
 
 int brt_create(BRT *);
 int brt_set_flags(BRT, int flags);
@@ -21,9 +21,9 @@ int brt_set_cachetable(BRT, CACHETABLE);
 int brt_open(BRT, const char *fname, const char *fname_in_env, const char *dbname, int is_create, int only_create, CACHETABLE ct, TOKUTXN txn);
 int brt_remove_subdb(BRT brt, const char *dbname, u_int32_t flags);
 
-int brt_insert (BRT, DBT *, DBT *, DB*, TOKUTXN);
-int brt_lookup (BRT brt, DBT *k, DBT *v, DB*db);
-int brt_delete (BRT brt, DBT *k, DB *db);
+int brt_insert (BRT, DBT *, DBT *, TOKUTXN);
+int brt_lookup (BRT brt, DBT *k, DBT *v);
+int brt_delete (BRT brt, DBT *k);
 int close_brt (BRT);
 int dump_brt (BRT brt);
 void brt_fsync (BRT); /* fsync, but don't clear the caches. */
@@ -43,7 +43,7 @@ int show_brt_blocknumbers(BRT);
 
 typedef struct brt_cursor *BRT_CURSOR;
 int brt_cursor (BRT, BRT_CURSOR*);
-int brt_cursor_get (BRT_CURSOR cursor, DBT *kbt, DBT *vbt, int brtc_flags, DB *db, TOKUTXN);
+int brt_cursor_get (BRT_CURSOR cursor, DBT *kbt, DBT *vbt, int brtc_flags, TOKUTXN);
 int brt_cursor_delete(BRT_CURSOR cursor, int flags);
 int brt_cursor_close (BRT_CURSOR curs);
 

newbrt/key.c

@@ -102,3 +102,7 @@ void toku_test_keycompare (void) {
 int toku_default_compare_fun (DB *db __attribute__((__unused__)), const DBT *a, const DBT*b) {
     return toku_keycompare(a->data, a->size, b->data, b->size);
 }
+
+int dont_call_this_compare_fun (DB *db __attribute__((__unused__)), const DBT *a __attribute__((__unused__)), const DBT*b __attribute__((__unused__))) {
+    assert(0);
+}

newbrt/key.h

@@ -5,3 +5,4 @@ int toku_keycompare (bytevec key1, ITEMLEN key1len, bytevec key2, ITEMLEN key2le
 void toku_test_keycompare (void) ;
 
 int toku_default_compare_fun (DB *, const DBT *, const DBT*);
+int dont_call_this_compare_fun (DB *, const DBT *, const DBT*);

newbrt/log.c

@@ -208,14 +208,8 @@ int tokulogger_log_brt_insert_with_no_overwrite (TOKULOGGER logger,
 }
 
 int tokulogger_log_phys_add_or_delete_in_leaf (DB *db, TOKUTXN txn, DISKOFF diskoff, int is_add, const struct kv_pair *pair) {
+    assert(is_add==0);
     if (txn==0) return 0;
-    if (is_add) {
-	BYTESTRING key  = { pair->keylen, (char*)kv_pair_key_const(pair) };
-	BYTESTRING data = { pair->vallen, (char*)kv_pair_val_const(pair) };
-	//printf("Logging insertinleaf\n");
-	return toku_log_insertinleaf (txn, toku_txn_get_txnid(txn), db->i->fileid, diskoff, key, data);
-    }
-    assert(0);
     assert(db);
     int keylen = pair->keylen;
     int vallen = pair->vallen;

newbrt/logformat.c

@@ -61,6 +61,7 @@ const struct logtype logtypes[] = {
     {"insertinleaf", 'I', FA{{"TXNID",      "txnid"},
 			     {"FILENUM",    "filenum"},
 			     {"DISKOFF",    "diskoff"},
+			     {"u_int32_t",  "pmaidx"},
 			     {"BYTESTRING", "key"},
 			     {"BYTESTRING", "data"},
                              NULLFIELD}},

newbrt/pma-internal.h

@@ -26,6 +26,8 @@ struct pma {
     struct list cursors;
     pma_compare_fun_t compare_fun;
     pma_compare_fun_t dup_compare_fun;
+    DB *db;            /* Passed to the compare functions. */
+    FILENUM filenum;   /* Passed to logging. */
     void *skey, *sval; /* used in dbts */
     struct mempool kvspace;
 };
@@ -35,7 +37,7 @@ void toku_pmainternal_calculate_parameters (PMA pma);
 int toku_pmainternal_smooth_region (struct kv_pair *pairs[], int n, int idx, int base, PMA pma);
 int toku_pmainternal_printpairs (struct kv_pair *pairs[], int N);
 int toku_pmainternal_make_space_at (PMA pma, int idx);
-int toku_pmainternal_find (PMA pma, DBT *, DB*); // The DB is so the comparison fuction can be called.
+int toku_pmainternal_find (PMA pma, DBT *); // The DB is so the comparison fuction can be called.
 void toku_print_pma (PMA pma); /* useful for debugging, so keep the name short. I.e., not pmainternal_print_pma() */
 
 /* density thresholds */

newbrt/pma.h

@@ -19,7 +19,7 @@ typedef struct pma_cursor *PMA_CURSOR;
    return a value < 0, = 0, > 0 if a < b, a == b, a > b respectively */
 typedef int (*pma_compare_fun_t)(DB *, const DBT *a, const DBT *b);
 
-int toku_pma_create(PMA *, pma_compare_fun_t compare_fun, int maxsize);
+int toku_pma_create(PMA *, pma_compare_fun_t compare_fun, DB *, FILENUM filenum, int maxsize);
 
 int toku_pma_set_compare(PMA pma, pma_compare_fun_t compare_fun);
 
@@ -31,7 +31,7 @@ int toku_pma_set_dup_mode(PMA pma, int mode);
 int toku_pma_set_dup_compare(PMA pma, pma_compare_fun_t dup_compare_fun);
 
 /* verify the integrity of a pma */
-void toku_pma_verify(PMA pma, DB *db);
+void toku_pma_verify(PMA pma);
 
 /* returns 0 if OK.
  * You must have freed all the cursors, otherwise returns nonzero and does nothing. */
@@ -45,21 +45,21 @@ int  toku_pma_n_entries (PMA);
 /* Duplicates the key and keylen. */
 //enum pma_errors toku_pma_insert (PMA, bytevec key, ITEMLEN keylen, bytevec data, ITEMLEN datalen);
 // The DB pointer is there so that the comparison function can be called.
-enum pma_errors toku_pma_insert (PMA, DBT*, DBT*, DB*, TOKUTXN txn, DISKOFF, u_int32_t /*random for fingerprint */, u_int32_t */*fingerprint*/);
+enum pma_errors toku_pma_insert (PMA, DBT*, DBT*, TOKUTXN txn, DISKOFF, u_int32_t /*random for fingerprint */, u_int32_t */*fingerprint*/);
 /* This returns an error if the key is NOT present. */
 int pma_replace (PMA, bytevec key, ITEMLEN keylen, bytevec data, ITEMLEN datalen);
 /* This returns an error if the key is NOT present. */
-int toku_pma_delete (PMA, DBT *, DB*, u_int32_t /*random for fingerprint*/, u_int32_t */*fingerprint*/, u_int32_t *deleted_size);
+int toku_pma_delete (PMA, DBT *, u_int32_t /*random for fingerprint*/, u_int32_t */*fingerprint*/, u_int32_t *deleted_size);
 
 int toku_pma_insert_or_replace (PMA pma, DBT *k, DBT *v,
 			   int *replaced_v_size, /* If it is a replacement, set to the size of the old value, otherwise set to -1. */
-			   DB *db, TOKUTXN txn, DISKOFF,
+			   TOKUTXN txn, DISKOFF,
 			   u_int32_t /*random for fingerprint*/, u_int32_t */*fingerprint*/);
 
 
 /* Exposes internals of the PMA by returning a pointer to the guts.
  * Don't modify the returned data.  Don't free it. */
-enum pma_errors toku_pma_lookup (PMA, DBT*, DBT*, DB*);
+enum pma_errors toku_pma_lookup (PMA, DBT*, DBT*);
 
 /*
  * The kv pairs in the original pma are split into 2 equal sized sets
@@ -71,7 +71,7 @@ enum pma_errors toku_pma_lookup (PMA, DBT*, DBT*, DB*);
  * leftpma - the pma assigned keys <= pivot key
  * rightpma - the pma assigned keys > pivot key
  */
-int toku_pma_split(PMA origpma,  unsigned int *origpma_size, DBT *splitk, DB *db,
+int toku_pma_split(PMA origpma,  unsigned int *origpma_size, DBT *splitk,
 	      PMA leftpma,  unsigned int *leftpma_size,  u_int32_t leftrand4sum,  u_int32_t *leftfingerprint,
 	      PMA rightpma, unsigned int *rightpma_size, u_int32_t rightrand4sum, u_int32_t *rightfingerprint);
 
@@ -103,13 +103,13 @@ int toku_pma_cursor_set_position_prev (PMA_CURSOR c);
 int toku_pma_cursor_get_current(PMA_CURSOR c, DBT *key, DBT *val);
 
 /* set the cursor to the matching key and value pair */
-int toku_pma_cursor_set_both(PMA_CURSOR c, DBT *key, DBT *val, DB *db);
+int toku_pma_cursor_set_both(PMA_CURSOR c, DBT *key, DBT *val);
 
 /* move the cursor to the kv pair matching the key */
-int toku_pma_cursor_set_key(PMA_CURSOR c, DBT *key, DB *db);
+int toku_pma_cursor_set_key(PMA_CURSOR c, DBT *key);
 
 /* set the cursor to the smallest key in the pma >= key */
-int toku_pma_cursor_set_range(PMA_CURSOR c, DBT *key, DB *db);
+int toku_pma_cursor_set_range(PMA_CURSOR c, DBT *key);
 
 /* delete the key value pair under the cursor, return the size of the pair */
 int toku_pma_cursor_delete_under(PMA_CURSOR c, int *kvsize);

newbrt/recover.c

@@ -21,6 +21,8 @@ struct cf_pair {
 } *cf_pairs;
 int n_cf_pairs=0, max_cf_pairs=0;
 
+DB * const null_db=0;
+
 CACHEFILE find_cachefile (FILENUM fnum) {
     int i;
     for (i=0; i<n_cf_pairs; i++) {
@@ -86,7 +88,7 @@ static void toku_recover_newbrtnode (struct logtype_newbrtnode *c) {
     n->local_fingerprint = 0; // nothing there yet
     n->dirty = 1;
     if (c->height==0) {
-	r=toku_pma_create(&n->u.l.buffer, toku_default_compare_fun, c->nodesize);
+	r=toku_pma_create(&n->u.l.buffer, dont_call_this_compare_fun, null_db, c->filenum, c->nodesize);
 	assert(r==0);
 	n->u.l.n_bytes_in_buffer=0;
     } else {

src/tests/Makefile

@@ -91,6 +91,10 @@ NO_VGRIND = \
   db_open_notexist_reopen \
   db_remove_subdb \
   db_subdb \
+  dup_delete \
+  dup_flags \
+  dup_insert \
+  dup_search \
   rand_insert \
   reverse_compare_fun \
   # Comment to terminate list so the previous line can end with a slash

src/ydb.c

@@ -515,7 +515,7 @@ struct __toku_dbc_internal {
 };
 
 int __toku_c_get(DBC * c, DBT * key, DBT * data, u_int32_t flag) {
-    int r = brt_cursor_get(c->i->c, key, data, flag, c->i->db, c->i->txn ? c->i->txn->i->tokutxn : 0);
+    int r = brt_cursor_get(c->i->c, key, data, flag, c->i->txn ? c->i->txn->i->tokutxn : 0);
     return r;
 }
 
@@ -550,7 +550,7 @@ int __toku_db_cursor(DB * db, DB_TXN * txn, DBC ** c, u_int32_t flags) {
 }
 
 int __toku_db_del(DB * db, DB_TXN * txn __attribute__ ((unused)), DBT * key, u_int32_t flags __attribute((unused))) {
-    int r = brt_delete(db->i->brt, key, db);
+    int r = brt_delete(db->i->brt, key);
     return r;
 }
 
@@ -567,7 +567,7 @@ int __toku_db_get(DB * db, DB_TXN * txn __attribute__ ((unused)), DBT * key, DBT
             dbc->c_close(dbc);
         }
     } else
-        r = brt_lookup(db->i->brt, key, data, db);
+        r = brt_lookup(db->i->brt, key, data);
     return r;
 }
 
@@ -689,7 +689,7 @@ int __toku_db_open(DB * db, DB_TXN * txn, const char *fname, const char *dbname,
 }
 
 int __toku_db_put(DB * db, DB_TXN * txn, DBT * key, DBT * data, u_int32_t flags) {
-    int r = brt_insert(db->i->brt, key, data, db, txn ? txn->i->tokutxn : 0);
+    int r = brt_insert(db->i->brt, key, data, txn ? txn->i->tokutxn : 0);
     //printf("%s:%d %d=__toku_db_put(...)\n", __FILE__, __LINE__, r);
     return r;
 }