序列化和写盘

Tokudb数据节点写盘主要是由后台线程异步完成的：

  • checkpoint线程：把cachetable（innodb术语buffer pool）中所有脏页写回
  • evictor线程：释放内存，如果victim节点是dirty的，需要先将数据写回。

数据在磁盘上是序列化过的，序列化的过程就是把一个数据结构转换成字节流。

写数据包括两个阶段：

  • 序列化：把结构化数据转成字节流
  • 压缩：对序列化好的数据进行压缩

tokudb序列化和压缩单位是partition，对于internal节点，就是把msg buffer序列化并压缩；对于leaf节点，就是把basement node序列化并压缩。

一个节点（node）在磁盘上是如何存储的呢？ 节点数据在写盘时会被写到某个offset开始的位置，这个offset是从blocktable里面分配的一个空闲的空间。我们后面会专门写一篇有关btt（Block Translation Table）和block table的文章。 一个node的数据包含：header，pivot key和partition三部分：

  • header：节点meta信息
  • pivot key：记录了每个partition的key区间
  • partition：排序数据；一个node如果包含多个partition，这些partition是依次顺序存放的

有趣的是，压缩算法的信息是存放在partition压缩buffer的第一个字节。所以，tokudb支持FT索引内部同时使用多种压缩算法。

反序列化和读盘

Tokudb读盘的过程是在cachetable里通过调用get_and_pin系列函数实现

  • 前景线程调用get_and_pin系列函数
  • cleaner线程调用bring_node_fully_into_memory，这个函数调用pf_callback把不在内存中的那些partition读到内存。

数据从磁盘读到内存之前需要进行解压缩，然后对解压缩好的buffer进行反序列化，转换成内存数据结构。反序列化是使用序列化相反的方法把数据解析出来。

前面提过序列化和压缩的单位是partition，反序列化和解压缩的单位也是partition。

酱，节点数据就可以被FT层访问了。

序列化和压缩过程详解

这里顺便提一下BTT （Block Translation Table），这个表记录了节点（blocknum）在FT文件存储位置（offset）的映射关系。

为什么要引入这个表？Tokudb刷脏时，数据被写到一个新的空闲位置，避免了in-place update，简化recovery过程。

toku_ftnode_flush_callback是调用get_and_pin系列函数提供的flush_callback回调，checkpoint线程（也包含checkpoint thread pool的线程，在checkpoint过程中帮助前景线程做节点数据的回写）或evictor线程在这个函数里面会调用toku_serialize_ftnode_to做序列化和压缩工作。

toku_serialize_ftnode_to比较简单，首先调用toku_serialize_ftnode_to_memory执行序列化和压缩，然后调用blocktable.realloc_on_disk，为blocknum分配一个新的offset，最后调用pwrite把压缩的buffer写到盘上，回写完成清node->dirty标记。

这里单独说一下toku_serialize_ftnode_to_memory的第6个参数in_parallel，true表示并行处理序列化和压缩过程，false表示串行处理。

toku_ftnode_flush_callback通常是在evictor或者checkpoint线程上下文调用的，不影响前景线程服务客户端，这个参数一般是false，只有在loader场景下是true。

  1. toku_serialize_ftnode_to (int fd, BLOCKNUM blocknum, FTNODE node, FTNODE_DISK_DATA* ndd, bool do_rebalancing, FT ft, bool for_checkpoint) {
  3.     size_t n_to_write;
  4.     size_t n_uncompressed_bytes;
  5.     char *compressed_buf = nullptr;
  7.     // because toku_serialize_ftnode_to is only called for
  8.     // in toku_ftnode_flush_callback, we pass false
  9.     // for in_parallel. The reasoning is that when we write
  10.     // nodes to disk via toku_ftnode_flush_callback, we
  11.     // assume that it is being done on a non-critical
  12.     // background thread (probably for checkpointing), and therefore
  13.     // should not hog CPU,
  14.     //
  15.     // Should the above facts change, we may want to revisit
  16.     // passing false for in_parallel here
  17.     //
  18.     // alternatively, we could have made in_parallel a parameter
  19.     // for toku_serialize_ftnode_to, but instead we did this.
  20.     int r = toku_serialize_ftnode_to_memory(
  21.         node,
  22.         ndd,
  23.         ft->h->basementnodesize,
  24.         ft->h->compression_method,
  25.         do_rebalancing,
  26.         toku_drd_unsafe_fetch(&toku_serialize_in_parallel),
  27.         &n_to_write,
  28.         &n_uncompressed_bytes,
  29.         &compressed_buf
  30.         );
  31.     if (r != 0) {
  32.         return r;
  33.     }
  35.     // If the node has never been written, then write the whole buffer, including the zeros
  36.     invariant(blocknum.b>=0);
  37.     DISKOFF offset;
  39.     // Dirties the ft
  40.     ft->blocktable.realloc_on_disk(blocknum, n_to_write, &offset,
  41.                                    ft, fd, for_checkpoint);
  43.     tokutime_t t0 = toku_time_now();
  44.     toku_os_full_pwrite(fd, compressed_buf, n_to_write, offset);
  45.     tokutime_t t1 = toku_time_now();
  47.     tokutime_t io_time = t1 - t0;
  48.     toku_ft_status_update_flush_reason(node, n_uncompressed_bytes, n_to_write, io_time, for_checkpoint);
  50.     toku_free(compressed_buf);
  51.     node->dirty = 0;  // See #1957.   Must set the node to be clean after serializing it so that it doesn't get written again on the next checkpoint or eviction.
  52.     return 0;
  53. }

序列化和压缩过程是在toku_serialize_ftnode_to_memory实现，这个函数比较长，我们分成3段来看。

  • partition序列化和压缩
  • pivot key序列化和压缩
  • header序列化

partition序列化和压缩

toku_serialize_ftnode_to_memory的第5个参数do_rebalancing表示leaf节点在写回之前是否要做rebalance，这个参数是在toku_ftnode_flush_callback指定的，如果写回的是数据节点本身，那么是需要做rebalance的。

toku_serialize_ftnode_to_memory首先确保整个数据节点都在内存中，这么做是因为节点的partition数据是依次顺序存放的；然后根据do_rebalancing决定是否要对leaf节点做rebalance；接着是一大段内存分配：

  • sb包含节点partition压缩数据的数组，每个元素包含partition的uncompressed的buffer和compressed的buffer
  • ndd是指针数组，记录了每个partition压缩后数据的offset和size

这里有个小的优化，并没有为每个partition申请compressed的buffer，而是申请了一个足够大的buffer，每个partition使用其中的一段。uncompressed的buffer也是一样处理的。

足够大的buffer是什么意思呢？

  • uncompressed的buffer：各个partition的size总和。
  • compressed的buffer：压缩后的最大可能长度加上8个字节的overhead（每个partition压缩前的size和压缩后的size）

使用不同压缩算法，压缩之后的最大可能长度是不同的。

分配好buffer之后，调用serialize_and_compress_in_parallel或者serialize_and_compress_serially进行序列化和压缩。

  1. int toku_serialize_ftnode_to_memory(FTNODE node,
  2.                                     FTNODE_DISK_DATA* ndd,
  3.                                     unsigned int basementnodesize,
  4.                                     enum toku_compression_method compression_method,
  5.                                     bool do_rebalancing,
  6.                                     bool in_parallel, // for loader is true, for toku_ftnode_flush_callback, is false
  7.                             /*out*/ size_t *n_bytes_to_write,
  8.                             /*out*/ size_t *n_uncompressed_bytes,
  9.                             /*out*/ char  **bytes_to_write)
  10. // Effect: Writes out each child to a separate malloc'd buffer, then compresses
  11. //   all of them, and writes the uncompressed header, to bytes_to_write,
  12. //   which is malloc'd.
  13. //
  14. //   The resulting buffer is guaranteed to be 512-byte aligned and the total length is a multiple of 512 (so we pad with zeros at the end if needed).
  15. //   512-byte padding is for O_DIRECT to work.
  16. {
  17.     toku_ftnode_assert_fully_in_memory(node);
  19.     if (do_rebalancing && node->height == 0) {
  20.         toku_ftnode_leaf_rebalance(node, basementnodesize);
  21.     }
  22.     const int npartitions = node->n_children;
  24.     // Each partition represents a compressed sub block
  25.     // For internal nodes, a sub block is a message buffer
  26.     // For leaf nodes, a sub block is a basement node
  27.     toku::scoped_calloc sb_buf(sizeof(struct sub_block) * npartitions);
  28.     struct sub_block *sb = reinterpret_cast<struct sub_block *>(sb_buf.get());
  29.     XREALLOC_N(npartitions, *ndd);
  31.     //
  32.     // First, let's serialize and compress the individual sub blocks
  33.     //
  35.     // determine how large our serialization and compression buffers need to be.
  36.     size_t serialize_buf_size = 0, compression_buf_size = 0;
  37.     for (int i = 0; i < node->n_children; i++) {
  38.         sb[i].uncompressed_size = serialize_ftnode_partition_size(node, i);
  39.         sb[i].compressed_size_bound = toku_compress_bound(compression_method, sb[i].uncompressed_size);
  40.         serialize_buf_size += sb[i].uncompressed_size;
  41.         compression_buf_size += sb[i].compressed_size_bound + 8; // add 8 extra bytes, 4 for compressed size, 4 for decompressed size
  42.     }
  44.     // give each sub block a base pointer to enough buffer space for serialization and compression
  45.     toku::scoped_malloc serialize_buf(serialize_buf_size);
  46.     toku::scoped_malloc compression_buf(compression_buf_size);
  47.     for (size_t i = 0, uncompressed_offset = 0, compressed_offset = 0; i < (size_t) node->n_children; i++) {
  48.         sb[i].uncompressed_ptr = reinterpret_cast<char *>(serialize_buf.get()) + uncompressed_offset;
  49.         sb[i].compressed_ptr = reinterpret_cast<char *>(compression_buf.get()) + compressed_offset;
  50.         uncompressed_offset += sb[i].uncompressed_size;
  51.         compressed_offset += sb[i].compressed_size_bound + 8; // add 8 extra bytes, 4 for compressed size, 4 for decompressed size
  52.         invariant(uncompressed_offset <= serialize_buf_size);
  53.         invariant(compressed_offset <= compression_buf_size);
  54.     }
  56.     // do the actual serialization now that we have buffer space
  57.     struct serialize_times st = { 0, 0 };
  58.     if (in_parallel) {
  59.         serialize_and_compress_in_parallel(node, npartitions, compression_method, sb, &st);
  60.     } else {
  61.         serialize_and_compress_serially(node, npartitions, compression_method, sb, &st);
  62.     }

serialize_and_compress_serially就是串行调用serialize_and_compress_partition进行序列化和压缩。

  1. static void
  2. serialize_and_compress_serially(FTNODE node,
  3.                                 int npartitions,
  4.                                 enum toku_compression_method compression_method,
  5.                                 struct sub_block sb[],
  6.                                 struct serialize_times *st) {
  7.     for (int i = 0; i < npartitions; i++) {
  8.         serialize_and_compress_partition(node, i, compression_method, &sb[i], st);
  9.     }
  10. }

serialize_and_compress_in_parallel使用了threadpool来并行执行序列化和压缩，每个partition由一个专门的线程来处理。当前上下文也可以执行序列化和压缩，所以threadpool只创建了（npartitions-1）个线程。

threadpool线程执行的函数也是serialize_and_compress_partition；threadpool线程和当前上下文之间是使用work进行同步的。

  1. static void *
  2. serialize_and_compress_worker(void *arg) {
  3.     struct workset *ws = (struct workset *) arg;
  4.     while (1) {
  5.         struct serialize_compress_work *w = (struct serialize_compress_work *) workset_get(ws);
  6.         if (w == NULL)
  7.             break;
  8.         int i = w->i;
  9.         serialize_and_compress_partition(w->node, i, w->compression_method, &w->sb[i], &w->st);
  10.     }
  11.     workset_release_ref(ws);
  12.     return arg;
  13. }
  15. static void
  16. serialize_and_compress_in_parallel(FTNODE node,
  17.                                    int npartitions,
  18.                                    enum toku_compression_method compression_method,
  19.                                    struct sub_block sb[],
  20.                                    struct serialize_times *st) {
  21.     if (npartitions == 1) {
  22.         serialize_and_compress_partition(node, 0, compression_method, &sb[0], st);
  23.     } else {
  24.         int T = num_cores;
  25.         if (T > npartitions)
  26.             T = npartitions;
  27.         if (T > 0)
  28.             T = T - 1;
  29.         struct workset ws;
  30.         ZERO_STRUCT(ws);
  31.         workset_init(&ws);
  32.         struct serialize_compress_work work[npartitions];
  33.         workset_lock(&ws);
  34.         for (int i = 0; i < npartitions; i++) {
  35.             work[i] = (struct serialize_compress_work) { .base = ,
  36.                                                          .node = node,
  37.                                                          .i = i,
  38.                                                          .compression_method = compression_method,
  39.                                                          .sb = sb,
  40.                                                          .st = { .serialize_time = 0, .compress_time = 0} };
  41.             workset_put_locked(&ws, &work[i].base);
  42.         }
  43.         workset_unlock(&ws);
  44.         toku_thread_pool_run(ft_pool, 0, &T, serialize_and_compress_worker, &ws);
  45.         workset_add_ref(&ws, T);
  46.         serialize_and_compress_worker(&ws);
  47.         workset_join(&ws);
  48.         workset_destroy(&ws);
  50.         // gather up the statistics from each thread's work item
  51.         for (int i = 0; i < npartitions; i++) {
  52.             st->serialize_time += work[i].st.serialize_time;
  53.             st->compress_time += work[i].st.compress_time;
  54.         }
  55.     }
  56. }

pivot key序列化和压缩

回到toku_serialize_ftnode_to_memory，序列化partition之后就是序列化pivot key的过程。 sb_node_info存放pivot key压缩数据的信息：

  • uncompressed_ptr和uncompressed_size是未压缩数据的buffer和size
  • compressed_ptr和compressed_size_bound是压缩后数据的buffer和压缩后最大可能的size+8个字节的overhead（未压缩数据size和压缩后数据的size）

前面提到，压缩后的size是由压缩算法决定，不同的压缩算法压缩之后最大可能的size是不同的。

toku_serialize_ftnode_to_memory调用serialize_and_compress_sb_node_info把pivot key信息序列化并压缩。

pivot key的compressed buffer头8个字节分别存储pivot key的compressed size和uncompressed size，从第9个字节开始才是压缩的字节流；而checksum是针对整个compressed buffer做的。

  1.     //
  2.     // Now lets create a sub-block that has the common node information,
  3.     // This does NOT include the header
  4.     //
  5.     // determine how large our serialization and copmression buffers need to be
  6.     struct sub_block sb_node_info;
  7.     sub_block_init(&sb_node_info);
  8.     size_t sb_node_info_uncompressed_size = serialize_ftnode_info_size(node);
  9.     size_t sb_node_info_compressed_size_bound = toku_compress_bound(compression_method, sb_node_info_uncompressed_size);
  10.     toku::scoped_malloc sb_node_info_uncompressed_buf(sb_node_info_uncompressed_size);
  11.     toku::scoped_malloc sb_node_info_compressed_buf(sb_node_info_compressed_size_bound + 8); // add 8 extra bytes, 4 for compressed size, 4 for decompressed size
  12.     sb_node_info.uncompressed_size = sb_node_info_uncompressed_size;
  13.     sb_node_info.uncompressed_ptr = sb_node_info_uncompressed_buf.get();
  14.     sb_node_info.compressed_size_bound = sb_node_info_compressed_size_bound;
  15.     sb_node_info.compressed_ptr = sb_node_info_compressed_buf.get();
  17.     // do the actual serialization now that we have buffer space
  18.     serialize_and_compress_sb_node_info(node, &sb_node_info, compression_method, &st);
  20.     //
  21.     // At this point, we have compressed each of our pieces into individual sub_blocks,
  22.     // we can put the header and all the subblocks into a single buffer and return it.
  23.     //
  25.     // update the serialize times, ignore the header for simplicity. we captured all
  26.     // of the partitions' serialize times so that's probably good enough.
  27.     toku_ft_status_update_serialize_times(node, st.serialize_time, st.compress_time);

header序列化

序列化pivot key之后，toku_serialize_ftnode_to_memory计算节点node压缩前size和压缩后的size。 计算方法很简单：partition的size总和 + pivot key的size + header的size + 4个字节的overhead（pivot key的checksum）。

节点node压缩之后的size是为分配压缩后的数据buffer，为了支持direct I/O，分配的buffer和buffer size必须是512对齐的。

分配的buffer size记在n_bytes_to_write返回给调用函数；压缩之后的数据存储在bytes_to_write指向的buffer中。

节点node压缩之前的size，就是为了返回给调用函数，记在n_uncompressed_bytes参数中。

  1.     // The total size of the node is:
  2.     // size of header + disk size of the n+1 sub_block's created above
  3.     uint32_t total_node_size = (serialize_node_header_size(node) // uncompressed header
  4.                                  + sb_node_info.compressed_size   // compressed nodeinfo (without its checksum)
  5.                                  + 4);                            // nodeinfo's checksum
  6.     uint32_t total_uncompressed_size = (serialize_node_header_size(node) // uncompressed header
  7.                                  + sb_node_info.uncompressed_size   // uncompressed nodeinfo (without its checksum)
  8.                                  + 4);                            // nodeinfo's checksum
  9.     // store the BP_SIZESs
  10.     for (int i = 0; i < node->n_children; i++) {
  11.         uint32_t len         = sb[i].compressed_size + 4; // data and checksum
  12.         BP_SIZE (*ndd,i) = len;
  13.         BP_START(*ndd,i) = total_node_size;
  14.         total_node_size += sb[i].compressed_size + 4;
  15.         total_uncompressed_size += sb[i].uncompressed_size + 4;
  16.     }
  18.     // now create the final serialized node
  19.     uint32_t total_buffer_size = roundup_to_multiple(512, total_node_size); // make the buffer be 512 bytes.
  20.     char *XMALLOC_N_ALIGNED(512, total_buffer_size, data);
  21.     char *curr_ptr = data;

前面提到节点node序列化的过程分为3个阶段：

  • partition序列化和压缩
  • pivot key序列化和压缩
  • header序列化

前2个阶段都讨论过了，header的部分是调用serialize_node_header实现的。

到这里其他部分的序列化和压缩工作都做好了，header的序列化直接在前面分配好的压缩后数据buffer上进行，不需要压缩，也不必分配sub_block数据结构。

header处理完，直接把pivot key的sub_block的compressed_ptr数据和checksum拷贝过来。

pivot key处理完，直接把每个partition的compressed_ptr和checksum依次拷贝过来。

pad的部分写0。

  1.     // write the header
  2.     struct wbuf wb;
  3.     wbuf_init(&wb, curr_ptr, serialize_node_header_size(node));
  4.     serialize_node_header(node, *ndd, &wb);
  5.     assert(wb.ndone == wb.size);
  6.     curr_ptr += serialize_node_header_size(node);
  8.     // now write sb_node_info
  9.     memcpy(curr_ptr, sb_node_info.compressed_ptr, sb_node_info.compressed_size);
  10.     curr_ptr += sb_node_info.compressed_size;
  11.     // write the checksum
  12.     *(uint32_t *)curr_ptr = toku_htod32(sb_node_info.xsum);
  13.     curr_ptr += sizeof(sb_node_info.xsum);
  15.     for (int i = 0; i < npartitions; i++) {
  16.         memcpy(curr_ptr, sb[i].compressed_ptr, sb[i].compressed_size);
  17.         curr_ptr += sb[i].compressed_size;
  18.         // write the checksum
  19.         *(uint32_t *)curr_ptr = toku_htod32(sb[i].xsum);
  20.         curr_ptr += sizeof(sb[i].xsum);
  21.     }
  22.     // Zero the rest of the buffer
  23.     memset(data + total_node_size, 0, total_buffer_size - total_node_size);
  25.     assert(curr_ptr - data == total_node_size);
  26.     *bytes_to_write = data;
  27.     *n_bytes_to_write = total_buffer_size;
  28.     *n_uncompressed_bytes = total_uncompressed_size;
  30.     invariant(*n_bytes_to_write % 512 == 0);
  31.     invariant(reinterpret_cast<unsigned long long>(*bytes_to_write) % 512 == 0);
  32.     return 0;
  33. }

假若一个node包含2个partition，它的序列化结构如下所示：

image.png

反序列化和解压缩过程详解

由于tokudb支持partial fetch（只读某几个partition）和partial evict（即把clean节点的部分partition释放掉），反序列化过程相比序列化过程略复杂一些。

fetch callback通过bfe这个hint告诉toku_deserialize_ftnode_from需要读那些partition。

bfe有五种类型：

  • ftnode_fetch_none：只需要读header和pivot key，不需要读任何partition。只用于optimizer计算cost
  • ftnode_fetch_keymatch：只需要读match某个key的partition，ydb层提供的一个接口，一般不用
  • ftnode_fetch_prefetch：prefetch时使用
  • ftnode_fetch_all：需要把所有partition读上来；写节点时使用（msg inject或者msg apply的子节点）
  • ftnode_fetch_subset：需要读若干个partition，FT search路径上使用。

只有在ft search高度>1以上的中间节点时，read_all_partitions会被设置成true，走老的代码路径deserialize_ftnode_from_fd，一次性把所有partition都读到内存中。

其他情况会调用read_ftnode_header_from_fd_into_rbuf_if_small_enough，把节点的header读到内存中，然后反序列化header并设置ndd（每个partition的offset和size）；解压缩和反序列化pivot key设置pivot信息；根据bfe读取需要的partition。

节点的header，pivot key和partition都有自己的checksum信息，解析每个部分时都要确认checksum是匹配的。

  1. enum ftnode_fetch_type {
  2.     ftnode_fetch_none = 1, // no partitions needed.
  3.     ftnode_fetch_subset, // some subset of partitions needed
  4.     ftnode_fetch_prefetch, // this is part of a prefetch call
  5.     ftnode_fetch_all, // every partition is needed
  6.     ftnode_fetch_keymatch, // one child is needed if it holds both keys
  7. };
  9. int
  10. toku_deserialize_ftnode_from (int fd,
  11.                                BLOCKNUM blocknum,
  12.                                uint32_t fullhash,
  13.                                FTNODE *ftnode,
  14.                                FTNODE_DISK_DATA* ndd,
  15.                                ftnode_fetch_extra *bfe
  16.     )
  17. // Effect: Read a node in.  If possible, read just the header.
  18. {
  19.     int r = 0;
  20.     struct rbuf rb = RBUF_INITIALIZER;
  22.     // each function below takes the appropriate io/decompression/deserialize statistics
  24.     if (!bfe->read_all_partitions) {
  25.         read_ftnode_header_from_fd_into_rbuf_if_small_enough(fd, blocknum, bfe->ft, &rb, bfe);
  26.         r = deserialize_ftnode_header_from_rbuf_if_small_enough(ftnode, ndd, blocknum, fullhash, bfe, &rb, fd);
  27.     } else {
  28.         // force us to do it the old way
  29.         r = -1;
  30.     }
  31.     if (r != 0) {
  32.         // Something went wrong, go back to doing it the old way.
  33.         r = deserialize_ftnode_from_fd(fd, blocknum, fullhash, ftnode, ndd, bfe, NULL);
  34.     }
  36.     toku_free(rb.buf);
  37.     return r;
  38. }

deserialize_ftnode_header_from_rbuf_if_small_enough比较长，基本是toku_serialize_ftnode_to_memory的相反过程。

header部分是不压缩的，直接解析，比较magic number，解析node->n_children和ndd等。

然后比较header的checksum

  1.     node->n_children = rbuf_int(rb);
  2.     // Guaranteed to be have been able to read up to here.  If n_children
  3.     // is too big, we may have a problem, so check that we won't overflow
  4.     // while reading the partition locations.
  5.     unsigned int nhsize;
  6.     nhsize =  serialize_node_header_size(node); // we can do this because n_children is filled in.
  7.     unsigned int needed_size;
  8.     needed_size = nhsize + 12; // we need 12 more so that we can read the compressed block size information that follows for the nodeinfo.
  9.     if (needed_size > rb->size) {
  10.         r = toku_db_badformat();
  11.         goto cleanup;
  12.     }
  14.     XMALLOC_N(node->n_children, node->bp);
  15.     XMALLOC_N(node->n_children, *ndd);
  16.     // read the partition locations
  17.     for (int i=0; i<node->n_children; i++) {
  18.         BP_START(*ndd,i) = rbuf_int(rb);
  19.         BP_SIZE (*ndd,i) = rbuf_int(rb);
  20.     }
  22.     uint32_t checksum;
  23.     checksum = toku_x1764_memory(rb->buf, rb->ndone);
  24.     uint32_t stored_checksum;
  25.     stored_checksum = rbuf_int(rb);
  26.     if (stored_checksum != checksum) {
  27.         dump_bad_block(rb->buf, rb->size);
  28.         r = TOKUDB_BAD_CHECKSUM;
  29.         goto cleanup;
  30.     }

接着处理pivot key，比较pivot key部分的checksum，解压缩，反序列化，设置pivot信息。

  1.     // Finish reading compressed the sub_block
  2.     const void **cp;
  3.     cp = (const void **) &sb_node_info.compressed_ptr;
  4.     rbuf_literal_bytes(rb, cp, sb_node_info.compressed_size);
  5.     sb_node_info.xsum = rbuf_int(rb);
  6.     // let's check the checksum
  7.     uint32_t actual_xsum;
  8.     actual_xsum = toku_x1764_memory((char *)sb_node_info.compressed_ptr-8, 8+sb_node_info.compressed_size);
  9.     if (sb_node_info.xsum != actual_xsum) {
  10.         r = TOKUDB_BAD_CHECKSUM;
  11.         goto cleanup;
  12.     }
  14.     // Now decompress the subblock
  15.     {
  16.         toku::scoped_malloc sb_node_info_buf(sb_node_info.uncompressed_size);
  17.         sb_node_info.uncompressed_ptr = sb_node_info_buf.get();
  18.         tokutime_t decompress_t0 = toku_time_now();
  19.         toku_decompress(
  20.             (Bytef *) sb_node_info.uncompressed_ptr,
  21.             sb_node_info.uncompressed_size,
  22.             (Bytef *) sb_node_info.compressed_ptr,
  23.             sb_node_info.compressed_size
  24.             );
  25.         tokutime_t decompress_t1 = toku_time_now();
  26.         decompress_time = decompress_t1 - decompress_t0;
  28.         // at this point sb->uncompressed_ptr stores the serialized node info.
  29.         r = deserialize_ftnode_info(&sb_node_info, node);
  30.         if (r != 0) {
  31.             goto cleanup;
  32.         }
  33.     }

最后是根据bfe读取需要的partition，读partition是通过调用pf_callback实现的。

  1.     // Now we have the ftnode_info.  We have a bunch more stuff in the
  2.     // rbuf, so we might be able to store the compressed data for some
  3.     // objects.
  4.     // We can proceed to deserialize the individual subblocks.
  6.     // setup the memory of the partitions
  7.     // for partitions being decompressed, create either message buffer or basement node
  8.     // for partitions staying compressed, create sub_block
  9.     setup_ftnode_partitions(node, bfe, false);
  11.     // We must capture deserialize and decompression time before
  12.     // the pf_callback, otherwise we would double-count.
  13.     t1 = toku_time_now();
  14.     deserialize_time = (t1 - t0) - decompress_time;
  16.     // do partial fetch if necessary
  17.     if (bfe->type != ftnode_fetch_none) {
  18.         PAIR_ATTR attr;
  19.         r = toku_ftnode_pf_callback(node, *ndd, bfe, fd, &attr, NULL);
  20.         if (r != 0) {
  21.             goto cleanup;
  22.         }
  23.     }

deserialize_ftnode_from_fd的部分留给读者自行分析。