Segrate rollback and recover functions. Write a brt-level test for the...

Segrate rollback and recover functions.   Write a brt-level test for the cursor static variable problem.  Addresses #536, #538.

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

newbrt/Makefile

@@ -57,6 +57,7 @@ REGRESSION_TESTS = \
 	brt-test3 \
 	brt-test4 \
 	brt-test-cursor \
+	brt-test-cursor-2 \
         log-test \
 	log-test2 \
 	log-test3 \
@@ -108,6 +109,7 @@ CHECKS = \
  test-brt-delete-both \
  brt-test \
  brt-test-cursor \
+ brt-test-cursor-2 \
  brt-test3 \
  brt-test4 \
  fifo-test \
@@ -141,7 +143,7 @@ check-fanout:
 		let BRT_FANOUT=BRT_FANOUT+1; \
 	done
 
-log-test log-test2 log-test3 log-test4 log-test5 log-test6 pma-test benchmark-test brt-test brt-test3 brt-test4 brt-test-cursor test-brt-delete-both brt-serialize-test brtdump test-inc-split test-del-inorder: LDFLAGS+=-lz
+log-test log-test2 log-test3 log-test4 log-test5 log-test6 pma-test benchmark-test brt-test brt-test3 brt-test4 brt-test-cursor brt-test-cursor-2 test-brt-delete-both brt-serialize-test brtdump test-inc-split test-del-inorder: LDFLAGS+=-lz
 # pma: PROF_FLAGS=-fprofile-arcs -ftest-coverage
 
 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
@@ -157,10 +159,10 @@ ybt.o: ybt.h brttypes.h ../include/db.h
 ybt-test: ybt-test.o ybt.o memory.o toku_assert.o
 ybt-test.o: ybt.h ../include/db.h
 cachetable.o: cachetable.h hashfun.h memory.h
-brt-test3 brt-test4 brt-test-cursor brt-test: ybt.o brt.o fifo.o pma.o memory.o brt-serialize.o cachetable.o ybt.o key.o primes.o toku_assert.o log.o mempool.o brt-verify.o fingerprint.o log_code.o roll.o
+brt-test3 brt-test4 brt-test-cursor brt-test-cursor-2 brt-test: ybt.o brt.o fifo.o pma.o memory.o brt-serialize.o cachetable.o ybt.o key.o primes.o toku_assert.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)
 logformat: logformat.o toku_assert.o
-brt-test3.o brt-test4.o brt-test-cursor.o brt-test.o brt.o: brt.h ../include/db.h fifo.h pma.h brttypes.h cachetable.h memory.h
+brt-test3.o brt-test4.o brt-test-cursor.o brt-test-cursor-2.o 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
 fifo.o: fifo.h brttypes.h 

newbrt/brt-test-cursor-2.c

+/* -*- mode: C; c-basic-offset: 4 -*- */
+#ident "Copyright (c) 2007, 2008 Tokutek Inc.  All rights reserved."
+
+#include "brt.h"
+#include "key.h"
+#include "pma.h"
+#include "brt-internal.h"
+#include "memory.h"
+#include "toku_assert.h"
+
+#include <fcntl.h>
+#include <stdio.h>
+#include <string.h>
+#include <sys/time.h>
+#include <stdlib.h>
+#include <unistd.h>
+#include <arpa/inet.h>
+
+#include "test.h"
+
+static const char fname[]= __FILE__ ".brt";
+
+static TOKUTXN const null_txn = 0;
+static DB * const null_db = 0;
+
+// Verify that different cursors return different data items when a DBT is initialized to all zeros (no flags)
+static void test_multiple_brt_cursor_dbts(int n, DB *db) {
+    if (verbose) printf("test_multiple_brt_cursors:%d %p\n", n, db);
+
+    int r;
+    CACHETABLE ct;
+    BRT brt;
+    BRT_CURSOR cursors[n];
+
+    unlink(fname);
+
+    r = toku_brt_create_cachetable(&ct, 0, ZERO_LSN, NULL_LOGGER);
+    assert(r==0);
+
+    r = toku_open_brt(fname, 0, 1, &brt, 1<<12, ct, null_txn, toku_default_compare_fun, db);
+    assert(r==0);
+
+    int i;
+    for (i=0; i<n; i++) {
+	DBT kbt,vbt;
+	char key[10],val[10];
+	snprintf(key, sizeof key, "k%04d", i);
+	snprintf(key, sizeof val, "k%04d", i);
+	r = toku_brt_insert(brt,
+			    toku_fill_dbt(&kbt, key, strlen(key)),
+			    toku_fill_dbt(&vbt, val, strlen(val)),
+			    0);
+	assert(r == 0);
+    }
+
+    for (i=0; i<n; i++) {
+        r = toku_brt_cursor(brt, &cursors[i]);
+        assert(r == 0);
+    }
+
+    void *ptrs[n];
+    for (i=0; i<n; i++) {
+	DBT kbt, vbt;
+	char key[10];
+	snprintf(key, sizeof key, "k%04d", i);
+	r = toku_brt_cursor_get(cursors[i],
+				toku_fill_dbt(&kbt, key, strlen(key)),
+				toku_init_dbt(&vbt),
+				DB_SET,
+				null_txn);
+	assert(r == 0);
+	ptrs[i] = vbt.data;
+    }
+
+    for (i=0; i<n; i++) {
+	int j;
+	for (j=i+1; j<n; j++) {
+	    assert(ptrs[i]!=ptrs[j]);
+	}
+    }
+
+    for (i=0; i<n; i++) {
+        r = toku_brt_cursor_close(cursors[i]);
+        assert(r == 0);
+    }
+
+    r = toku_close_brt(brt);
+    assert(r==0);
+
+    r = toku_cachetable_close(&ct);
+    assert(r==0);
+}
+
+static void test_brt_cursor(DB *db) {
+    test_multiple_brt_cursor_dbts(1, db);
+    test_multiple_brt_cursor_dbts(2, db);
+    test_multiple_brt_cursor_dbts(3, db);
+}
+
+
+int main (int argc , const char *argv[]) {
+    default_parse_args(argc, argv);
+
+    DB a_db;
+    DB *db = &a_db;
+    test_brt_cursor(db);
+
+    toku_malloc_cleanup();
+    if (verbose) printf("test ok\n");
+    return 0;
+}

newbrt/brttypes.h

@@ -92,4 +92,6 @@ struct brt_cmd {
 };
 typedef struct brt_cmd BRT_CMD_S, *BRT_CMD;
 
+#define UU(x) x __attribute__((__unused__))
+
 #endif

newbrt/log-internal.h

@@ -129,3 +129,10 @@ static inline int toku_logsizeof_INTPAIRARRAY (INTPAIRARRAY pa) {
     return 4+(4+4)*pa.size;
 }
 
+static inline char *fixup_fname(BYTESTRING *f) {
+    assert(f->len>0);
+    char *fname = toku_malloc(f->len+1);
+    memcpy(fname, f->data, f->len);
+    fname[f->len]=0;
+    return fname;
+}

newbrt/logformat.c

@@ -236,9 +236,9 @@ void generate_log_struct (void) {
 		    fprintf(hf, "  %-16s crc;\n", "u_int32_t");
 		    fprintf(hf, "  %-16s len;\n", "u_int32_t");
 		    fprintf(hf, "};\n");
-		    fprintf(hf, "void toku_recover_%s (LSN lsn", lt->name);
-		    DO_FIELDS(ft, lt, fprintf(hf, ", %s %s", ft->type, ft->name));
-		    fprintf(hf, ");\n");
+		    //fprintf(hf, "void toku_recover_%s (LSN lsn", lt->name);
+		    //DO_FIELDS(ft, lt, fprintf(hf, ", %s %s", ft->type, ft->name));
+		    //fprintf(hf, ");\n");
 		}));
     DO_ROLLBACKS(lt,
 		 ({  fprintf(hf, "struct rolltype_%s {\n", lt->name);

newbrt/tdb-recover.c

@@ -20,63 +20,12 @@
 #include <unistd.h>
 
 int main (int argc, char *argv[]) {
-    const char *dir;
-    int r;
-    int entrycount=0;
     assert(argc==2);
-    dir = argv[1];
-    int n_logfiles;
-    char **logfiles;
 
-    int lockfd;
-
-    {
-	int namelen=strlen(dir);
-	char lockfname[namelen+20];
-	snprintf(lockfname, sizeof(lockfname), "%s/__recoverylock", dir);
-	lockfd = open(lockfname, O_RDWR|O_CREAT, S_IRUSR | S_IWUSR);
-	if (lockfd<0) {
-	    printf("Couldn't open %s\n", lockfname);
-	    exit(1);
-	}
-	r=flock(lockfd, LOCK_EX | LOCK_NB);
-	if (r!=0) {
-	    printf("Couldn't run recovery because some other process holds the recovery lock %s\n", lockfname);
-	    exit(1);
-	}
-    }
-
-    r = toku_logger_find_logfiles(dir, &n_logfiles, &logfiles);
-    if (r!=0) exit(1);
-    int i;
-    toku_recover_init();
-    for (i=0; i<n_logfiles; i++) {
-	//fprintf(stderr, "Opening %s\n", logfiles[i]);
-	FILE *f = fopen(logfiles[i], "r");
-	struct log_entry le;
-	u_int32_t version;
-	r=toku_read_and_print_logmagic(f, &version);
-	assert(r==0 && version==0);
-	while ((r = toku_log_fread(f, &le))==0) {
-	    //printf("%lld: Got cmd %c\n", le.u.commit.lsn.lsn, le.cmd);
-	    logtype_dispatch_args(&le, toku_recover_);
-	    entrycount++;
-	}
-	if (r!=EOF) {
-	    if (r==DB_BADFORMAT) {
-		fprintf(stderr, "Bad log format at record %d\n", entrycount);
-		exit(1);
-	    } else {
-		fprintf(stderr, "Huh? %s\n", strerror(r));
-		exit(1);
-	    }
-	}
-	fclose(f);
-    }
-    toku_recover_cleanup();
-    for (i=0; i<n_logfiles; i++) {
-	toku_free(logfiles[i]);
+    int r=tokudb_recover(argv[1], argv[1]);
+    if (r!=0) {
+	fprintf(stderr, "Recovery failed\n");
+	exit(1);
     }
-    toku_free(logfiles);
     return 0;
 }