Fixes #721.

Improve recovery for at least one case.  Addresses #703.
The logfiles that Rich built seem corrupt, so I cannot make progress on the next case.


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

newbrt/Makefile

@@ -202,7 +202,7 @@ gpma.o: gpma.c gpma.h
 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
+cachetable.o: brttypes.h cachetable.h hashfun.h memory.h primes.h toku_assert.h $(BRT_INTERNAL_H_INCLUDES) log_header.h
 brt-test0 brt-test1 brt-test2 brt-test3 brt-test4 brt-test5 test-brt-overflow brt-test-named-db brt-test-cursor brt-test-cursor-2 brt-test: ybt.o brt.o fifo.o gpma.o leafentry.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

newbrt/cachetable.c

@@ -406,6 +406,7 @@ int toku_cachetable_put(CACHEFILE cachefile, CACHEKEY key, void*value, long size
 		// In practice, the functions better be the same.
 		assert(p->flush_callback==flush_callback);
 		assert(p->fetch_callback==fetch_callback);
+		p->pinned++; /* Already present.  But increment the pin count. */
 		return -1; /* Already present. */
 	    }
 	}

newbrt/recover.c

@@ -63,8 +63,31 @@ void toku_recover_cleanup (void) {
 void toku_recover_commit (LSN UU(lsn), TXNID UU(txnid)) {
 }
 
+void create_dir_from_file (const char *fname) {
+    int i;
+    char *tmp=toku_strdup(fname);
+    char ch;
+    for (i=0; (ch=fname[i]); i++) {
+	if (ch=='/') {
+	    if (i>0) {
+		tmp[i]=0;
+		mode_t oldu = umask(0);
+		int r = mkdir(tmp, 0700);
+		if (r!=0 && errno!=EEXIST) {
+		    printf("error: %s\n", strerror(errno));
+		}
+		assert (r==0 || (errno==EEXIST));
+		umask(oldu);
+		tmp[i]=ch;
+	    }
+	}
+    }
+    toku_free(tmp);
+}
+
 void toku_recover_fcreate (LSN UU(lsn), TXNID UU(txnid),BYTESTRING fname,u_int32_t mode) {
     char *fixed_fname = fixup_fname(&fname);
+    create_dir_from_file(fixed_fname);
     int fd = creat(fixed_fname, mode);
     assert(fd>=0);
     toku_free(fixed_fname);
@@ -122,7 +145,7 @@ static void toku_recover_fheader (LSN UU(lsn), TXNID UU(txnid),FILENUM filenum,L
     }
     toku_cachetable_put(pair->cf, 0, h, 0, toku_brtheader_flush_callback, toku_brtheader_fetch_callback, 0);
     if (pair->brt) {
-	free(pair->brt->h);
+	toku_free(pair->brt->h);
     }  else {
 	MALLOC(pair->brt);
 	pair->brt->cf = pair->cf;
@@ -682,6 +705,7 @@ void toku_recover_pmadistribute (LSN lsn, FILENUM filenum, DISKOFF old_diskoff,
     }
     VERIFY_COUNTS(nodea);
 
+    assert(toku_gpma_index_limit(nodea->u.l.buffer)==old_N);
     r = move_indices (nodea->u.l.buffer, &nodea->u.l.buffer_mempool,
 		      nodeb->u.l.buffer, &nodeb->u.l.buffer_mempool,
 		      fromto,