Log header

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

newbrt/Makefile

@@ -74,14 +74,14 @@ pma-test benchmark-test brt-test brt-serialize-test: LDFLAGS+=-lz
 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
 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
+pma-test: pma.o memory.o key.o ybt.o log.o mempool.o fingerprint.o brt-serialize.o hashtable.o primes.o
 pma.o: pma.h yerror.h pma-internal.h memory.h key.h ybt.h brttypes.h log.h ../include/db.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
 brt-test: ybt.o brt.o hashtable.o pma.o memory.o brt-serialize.o cachetable.o header-io.o ybt.o key.o primes.o log.o mempool.o brt-verify.o fingerprint.o
-log.o: log-internal.h log.h wbuf.h crc.h
+log.o: 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
 brt-serialize-test.o: $(BRT_INTERNAL_H_INCLUDES)
 brt.o: $(BRT_INTERNAL_H_INCLUDES)

newbrt/brt-internal.h

@@ -96,14 +96,16 @@ struct brt {
 };
 
 /* serialization code */
-void toku_seralize_brtnode_to(int fd, DISKOFF off, DISKOFF size, BRTNODE node);
+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 *));
 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);
 
 void toku_verify_counts(BRTNODE);
 
+int toku_serialize_brt_header_size (struct brt_header *h);
 int toku_serialize_brt_header_to (int fd, struct brt_header *h);
+int toku_serialize_brt_header_to_wbuf (struct wbuf *, struct brt_header *h);
 int toku_deserialize_brtheader_from (int fd, DISKOFF off, struct brt_header **brth);
 
 /* return the size of a tree node */

newbrt/brt-serialize-test.c

@@ -41,7 +41,7 @@ void test_serialize(void) {
     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_hashtables = 3*(BRT_CMD_OVERHEAD+KEY_VALUE_OVERHEAD+2+5);
 
-    toku_seralize_brtnode_to(fd, sn.nodesize*20, sn.nodesize, &sn);  assert(r==0);
+    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);
     assert(r==0);

newbrt/brt-serialize.c

@@ -85,7 +85,7 @@ unsigned int toku_serialize_brtnode_size (BRTNODE node) {
     return result;
 }
 
-void toku_seralize_brtnode_to(int fd, DISKOFF off, DISKOFF size, BRTNODE node) {
+void toku_serialize_brtnode_to(int fd, DISKOFF off, DISKOFF size, BRTNODE node) {
     //printf("%s:%d serializing\n", __FILE__, __LINE__);
     struct wbuf w;
     int i;
@@ -432,44 +432,56 @@ void toku_verify_counts (BRTNODE node) {
     }
 }
     
-int toku_serialize_brt_header_to (int fd, struct brt_header *h) {
-    struct wbuf w;
-    int i;
-    unsigned int size=0; /* I don't want to mess around calculating it exactly. */ 
-    size += 4+4+4+8+8+4; /* this size, flags, the tree's nodesize, freelist, unused_memory, nnamed_rootse. */
+int toku_serialize_brt_header_size (struct brt_header *h) {
+    unsigned int size = 4+4+4+8+8+4; /* this size, flags, the tree's nodesize, freelist, unused_memory, named_roots. */
     if (h->n_named_roots<0) {
 	size+=8;
     } else {
+	int i;
 	for (i=0; i<h->n_named_roots; i++) {
 	    size+=12 + 1 + strlen(h->names[i]);
 	}
     }
-    wbuf_init(&w, toku_malloc(size), size);
-    wbuf_int    (&w, size);
-    wbuf_int    (&w, h->flags);
-    wbuf_int    (&w, h->nodesize);
-    wbuf_diskoff(&w, h->freelist);
-    wbuf_diskoff(&w, h->unused_memory);
-    wbuf_int    (&w, h->n_named_roots);
+    return size;
+}
+
+int toku_serialize_brt_header_to_wbuf (struct wbuf *wbuf, struct brt_header *h) {
+    unsigned int size = toku_serialize_brt_header_size (h); // !!! seems silly to recompute the size when the caller knew it.  Do we really need the size?
+    wbuf_int    (wbuf, size);
+    wbuf_int    (wbuf, h->flags);
+    wbuf_int    (wbuf, h->nodesize);
+    wbuf_diskoff(wbuf, h->freelist);
+    wbuf_diskoff(wbuf, h->unused_memory);
+    wbuf_int    (wbuf, h->n_named_roots);
     if (h->n_named_roots>0) {
+	int i;
 	for (i=0; i<h->n_named_roots; i++) {
 	    char *s = h->names[i];
 	    unsigned int l = 1+strlen(s);
-	    wbuf_diskoff(&w, h->roots[i]);
-	    wbuf_bytes  (&w,  s, l);
+	    wbuf_diskoff(wbuf, h->roots[i]);
+	    wbuf_bytes  (wbuf,  s, l);
 	    assert(l>0 && s[l-1]==0);
 	}
     } else {
-	wbuf_diskoff(&w, h->unnamed_root);
+	wbuf_diskoff(wbuf, h->unnamed_root);
     }
+    assert(wbuf->ndone<=wbuf->size);
+    return 0;
+}
+
+int toku_serialize_brt_header_to (int fd, struct brt_header *h) {
+    struct wbuf w;
+    unsigned int size = toku_serialize_brt_header_size (h);
+    wbuf_init(&w, toku_malloc(size), size);
+    int r=toku_serialize_brt_header_to_wbuf(&w, h);
     assert(w.ndone==size);
     {
-	ssize_t r = pwrite(fd, w.buf, w.ndone, 0);
-	if (r<0) perror("pwrite");
-	assert((size_t)r==w.ndone);
+	ssize_t nwrote = pwrite(fd, w.buf, w.ndone, 0);
+	if (nwrote<0) perror("pwrite");
+	assert((size_t)nwrote==w.ndone);
     }
     toku_free(w.buf);
-    return 0;
+    return r;
 }
 
 int toku_deserialize_brtheader_from (int fd, DISKOFF off, struct brt_header **brth) {

newbrt/brt.c

@@ -184,7 +184,7 @@ void brtnode_flush_callback (CACHEFILE cachefile, DISKOFF nodename, void *brtnod
     }
     //printf("%s:%d %p->mdict[0]=%p\n", __FILE__, __LINE__, brtnode, brtnode->mdicts[0]);
     if (write_me) {
-	toku_seralize_brtnode_to(toku_cachefile_fd(cachefile), brtnode->thisnodename, brtnode->nodesize, brtnode);
+	toku_serialize_brtnode_to(toku_cachefile_fd(cachefile), brtnode->thisnodename, brtnode->nodesize, brtnode);
     }
     //printf("%s:%d %p->mdict[0]=%p\n", __FILE__, __LINE__, brtnode, brtnode->mdicts[0]);
     if (!keep_me) {
@@ -1576,8 +1576,9 @@ int brt_open(BRT t, const char *fname, const char *fname_in_env, const char *dbn
 		t->h->names=0;
 		t->h->roots=0;
 	    }
-	    if ((r=setup_brt_root_node(t, t->nodesize))!=0) { if (dbname) goto died5; else goto died2;	}
-	    if ((r=toku_cachetable_put(t->cf, 0, t->h, 0, brtheader_flush_callback, brtheader_fetch_callback, 0))) {  if (dbname) goto died5; else goto died2; }
+	    if ((r=setup_brt_root_node(t, t->nodesize))!=0) { died6: if (dbname) goto died5; else goto died2;	}
+	    if ((r=toku_cachetable_put(t->cf, 0, t->h, 0, brtheader_flush_callback, brtheader_fetch_callback, 0))) { goto died6; }
+	    if ((r=tokulogger_log_header(txn, toku_cachefile_filenum(t->cf), t->h)))                               { goto died6; }
 	} else {
 	    int i;
 	    assert(r==0);

newbrt/brt.h

@@ -4,12 +4,11 @@
 // This must be first to make the 64-bit file mode work right in Linux
 #define _FILE_OFFSET_BITS 64
 #include "brttypes.h"
-
 #include "ybt.h"
 #include "../include/db.h"
 #include "cachetable.h"
 #include "log.h"
-typedef struct brt *BRT;
+
 int open_brt (const char *fname, const char *dbname, int is_create, BRT *, int nodesize, CACHETABLE, TOKUTXN, int(*)(DB*,const DBT*,const DBT*));
 
 int brt_create(BRT *);

newbrt/brttypes.h

@@ -3,6 +3,11 @@
 #include <sys/types.h>
 #define _XOPEN_SOURCE 500
 #define _FILE_OFFSET_BITS 64
+
+typedef struct brt *BRT;
+struct brt_header;
+struct wbuf;
+
 typedef unsigned int ITEMLEN;
 typedef const void *bytevec;
 //typedef const void *bytevec;

newbrt/log-internal.h

@@ -18,10 +18,11 @@ struct tokulogger {
 
 int tokulogger_find_next_unused_log_file(const char *directory, long long *result);
 
-enum {
+enum lt_command {
     LT_COMMIT                   = 'C',
     LT_DELETE                   = 'D',
     LT_FCREATE                  = 'F',
+    LT_FHEADER                  = 'H',
     LT_INSERT_WITH_NO_OVERWRITE = 'I',
     LT_FOPEN                    = 'O',
     LT_CHECKPOINT               = 'P',

newbrt/log.c

 #include <unistd.h>
-#include "log-internal.h"
-#include "wbuf.h"
-#include "memory.h"
 #include <dirent.h>
 #include <errno.h>
 #include <fcntl.h>
@@ -10,6 +7,11 @@
 #include <sys/stat.h>
 #include <sys/types.h>
 #include <sys/uio.h>
+
+#include "brt-internal.h"
+#include "log-internal.h"
+#include "wbuf.h"
+#include "memory.h"
 #include "../src/ydb-internal.h"
 
 int tokulogger_find_next_unused_log_file(const char *directory, long long *result) {
@@ -328,6 +330,32 @@ int tokulogger_log_unlink (TOKUTXN txn, const char *fname) {
     return tokulogger_finish(txn->logger, &wbuf);
 };
 
+int tokulogger_log_header (TOKUTXN txn, FILENUM filenum, struct brt_header *h) {
+    if (txn==0) return 0;
+    int subsize=toku_serialize_brt_header_size(h);
+    int buflen = (1+
+		  + 8 // lsn
+		  + 8 // txnid
+		  + 4 // filenum
+		  + subsize
+		  + 8 // crc & len
+		  );
+    unsigned char *buf=toku_malloc(buflen); // alloc on heap because it might be big
+    int r;
+    if (buf==0) return errno;
+    struct wbuf wbuf;
+    wbuf_init(&wbuf, buf, buflen);
+    wbuf_char(&wbuf, LT_FHEADER);
+    wbuf_lsn    (&wbuf, txn->logger->lsn);
+    txn->logger->lsn.lsn++;
+    wbuf_txnid(&wbuf, txn->txnid64);
+    wbuf_filenum(&wbuf, filenum);
+    r=toku_serialize_brt_header_to_wbuf(&wbuf, h);
+    if (r!=0) return r;
+    r=tokulogger_finish(txn->logger, &wbuf);
+    toku_free(buf);
+    return r;
+}
 
 /*
 int brtenv_checkpoint (BRTENV env) {

newbrt/log.h

@@ -12,14 +12,16 @@ int tokulogger_log_phys_add_or_delete_in_leaf    (DB *db, TOKUTXN txn, DISKOFF d
 
 int tokulogger_log_commit (TOKUTXN txn);
 
-int tokulogger_log_block_rename (TOKULOGGER logger, FILENUM fileid, DISKOFF olddiskoff, DISKOFF newdiskoff, DISKOFF parentdiskoff, int childnum);
+int tokulogger_log_block_rename (TOKULOGGER /*logger*/, FILENUM /*fileid*/, DISKOFF /*olddiskoff*/, DISKOFF /*newdiskoff*/, DISKOFF /*parentdiskoff*/, int /*childnum*/);
 
-int tokutxn_begin (TOKUTXN /*parent*/,TOKUTXN *, TXNID txnid64, TOKULOGGER logger);
+int tokutxn_begin (TOKUTXN /*parent*/,TOKUTXN *, TXNID /*txnid64*/, TOKULOGGER /*logger*/);
 
 int tokulogger_log_fcreate (TOKUTXN, const char */*fname*/, int /*mode*/);
 
-int tokulogger_log_fopen (TOKUTXN, const char * /*fname*/, FILENUM filenum);
+int tokulogger_log_fopen (TOKUTXN, const char * /*fname*/, FILENUM);
 
 int tokulogger_log_unlink (TOKUTXN, const char */*fname*/);
 
+int tokulogger_log_header (TOKUTXN, FILENUM, struct brt_header *);
+
 #endif

newbrt/logdump.c

@@ -110,7 +110,7 @@ int main (int argc, char *argv[]) {
     for (i=0;
 	 i!=count && (crc=0,actual_len=0,cmd=get_char())!=EOF;
 	 i++) {
-	switch (cmd) {
+	switch ((enum lt_command)cmd) {
 	case LT_INSERT_WITH_NO_OVERWRITE:
 	    printf("INSERT_WITH_NO_OVERWRITE:");
 	    transcribe_lsn();
@@ -122,7 +122,7 @@ int main (int argc, char *argv[]) {
 	    transcribe_crc32();
 	    transcribe_len();
 	    printf("\n");
-	    break;
+	    goto next;
     
 	case LT_DELETE:
 	    printf("DELETE:");
@@ -135,7 +135,7 @@ int main (int argc, char *argv[]) {
 	    transcribe_crc32();
 	    transcribe_len();
 	    printf("\n");
-	    break;
+	    goto next;
 
 	case LT_FCREATE:
 	    printf("FCREATE:");
@@ -146,7 +146,7 @@ int main (int argc, char *argv[]) {
 	    transcribe_crc32();
 	    transcribe_len();
 	    printf("\n");
-	    break;
+	    goto next;
 
 	case LT_FOPEN:
 	    printf("FOPEN:");
@@ -157,7 +157,7 @@ int main (int argc, char *argv[]) {
 	    transcribe_crc32();
 	    transcribe_len();
 	    printf("\n");
-	    break;
+	    goto next;
 	    
 	case LT_COMMIT:
 	    printf("COMMIT:");
@@ -166,12 +166,20 @@ int main (int argc, char *argv[]) {
 	    transcribe_crc32();
 	    transcribe_len();
 	    printf("\n");
-	    break;
+	    goto next;
 
-	default:
-	    fprintf(stderr, "Huh?, found command '%c'\n", cmd);
-	    assert(0);
+	case LT_UNLINK:
+	case LT_BLOCK_RENAME:
+	case LT_CHECKPOINT:
+	case LT_FHEADER:
+	    fprintf(stderr, "Cannot handle this command yet: '%c'\n", cmd);
+	    break;
 	}
+	/* The default is to fall out the bottom.  That way we can get a compiler warning if we forget one of the enums, but we can also
+	 * get a runtime warning if the actual value isn't one of the enums. */
+	fprintf(stderr, "Huh?, found command '%c'\n", cmd);
+	assert(0);
+    next: ; /*nothing*/
     }
     return 0;
 }

newbrt/yerror.h

+#ifndef YERROR_H
+#define YERROR_H
 enum pma_errors { BRT_OK=0, BRT_ALREADY_THERE = -2, BRT_KEYEMPTY=-3 };
 
 enum typ_tag { TYP_BRTNODE = 0xdead0001,
@@ -6,3 +8,4 @@ enum typ_tag { TYP_BRTNODE = 0xdead0001,
                TYP_TOKULOGGER,
 	       TYP_TOKUTXN
 };
+#endif