Start doing the I/O on the block allocator

git-svn-id: file:///svn/tokudb.1131b+1080a@6087 c7de825b-a66e-492c-adef-691d508d4ae1

newbrt/brt-internal.h

@@ -114,6 +114,11 @@ struct remembered_hash {
     u_int32_t fullhash; // fullhash is the hashed value of fnum and root.
 };
 
+struct block_translation_pair {
+    DISKOFF diskoff;
+    DISKOFF size;
+};
+
 struct brt_header {
     int dirty;
     u_int32_t fullhash;
@@ -133,11 +138,13 @@ struct brt_header {
     BLOCKNUM free_blocks; // free list for blocks.  Use -1 to indicate that there are no free blocks
     BLOCKNUM unused_blocks; // first unused block
 
-    // Where and how big is the block allocation vector?
-    // The block allocation vector translates block numbers to offsets (DISKOFF) in the file.
-    // We use a DISKOFF for the vector location, because we cannot use block numbers to bootstrap the indirection table.
-    BLOCKNUM  block_allocation_vector_length; // It's a blocknum because the block allocation vector is indexed by blocknums.
-    DISKOFF   block_allocation_vector_location; // Remember this so we can free the block before writing a new one.
+    u_int64_t max_blocknum_translated;
+    struct block_translation_pair *block_translation;
+
+    // Where and how big is the block translation vector stored on disk.
+    u_int64_t block_translation_size_on_disk;
+    u_int64_t block_translation_address_on_disk; // 0 if there is no memory allocated
+    
     // The in-memory data structure  for block allocation
     BLOCK_ALLOCATOR block_allocator;
 };

newbrt/brt-serialize.c

@@ -586,18 +586,12 @@ int toku_serialize_brt_header_to_wbuf (struct wbuf *wbuf, struct brt_header *h)
     wbuf_BLOCKNUM(wbuf, h->free_blocks);
     wbuf_BLOCKNUM(wbuf, h->unused_blocks);
     wbuf_int    (wbuf, h->n_named_roots);
-    if (h->block_allocation_vector_length.b != 0) {
-	block_allocator_free_block(h->block_allocator, h->block_allocation_vector_location);
+    if (h->block_translation_address_on_disk !=0 ) {
+	block_allocator_free_block(h->block_allocator, h->block_translation_address_on_disk);
     }
-#if 0
-    h->block_allocation_vector_length.b = block_allocator_get_vector_length(h->block_allocator);
-    {
-	int r = block_allocator_alloc_block(h->block_allocator, 0, h->block_allocation_vector_length.b, (u_int64_t*)&h->block_allocation_vector_location);
-	assert(r==0);
-    }
-#endif
-    wbuf_BLOCKNUM(wbuf, h->block_allocation_vector_length);
-    wbuf_DISKOFF(wbuf, h->block_allocation_vector_location);
+    block_allocator_alloc_block(h->block_allocator, 4 + 8*h->max_blocknum_translated, &h->block_translation_address_on_disk);
+    wbuf_ulonglong(wbuf, h->max_blocknum_translated);
+    wbuf_DISKOFF(wbuf, h->block_translation_address_on_disk);
     if (h->n_named_roots>=0) {
 	int i;
 	for (i=0; i<h->n_named_roots; i++) {
@@ -617,18 +611,33 @@ int toku_serialize_brt_header_to_wbuf (struct wbuf *wbuf, struct brt_header *h)
 }
 
 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);
     {
+	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(r==0);
+	assert(w.ndone==size);
 	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);
     }
-    toku_free(w.buf);
-    return r;
+    {
+	struct wbuf w;
+	u_int64_t size = 4 + h->max_blocknum_translated * 8; // 4 for the checksum
+	wbuf_init(&w, toku_malloc(size), size);
+	u_int64_t i;
+	for (i=0; i<h->max_blocknum_translated; i++) {
+	    wbuf_ulonglong(&w, h->block_translation[i].diskoff);
+	    wbuf_ulonglong(&w, h->block_translation[i].size);
+	}
+	wbuf_int(&w, x1764_finish(&w.checksum));
+	ssize_t nwrote = pwrite(fd, w.buf, size, h->block_translation_address_on_disk);
+	assert(nwrote==(ssize_t)size);
+	toku_free(w.buf);
+    };
+    return 0;
 }
 
 int deserialize_brtheader (u_int32_t size, int fd, DISKOFF off, struct brt_header **brth, u_int32_t fullhash) {
@@ -672,10 +681,8 @@ int deserialize_brtheader (u_int32_t size, int fd, DISKOFF off, struct brt_heade
 		block_locations_and_lengths[i] = ntohll(block_locations_and_lengths[i]);
 	    }
 	}
-#if 0
 	int r = create_block_allocator(&h->block_allocator, h->block_allocation_vector_length.b, block_locations_and_lengths, h->nodesize);
 	assert(r==0);
-#endif
 	if (block_locations_and_lengths) free(block_locations_and_lengths);
     }
     if (h->n_named_roots>=0) {