LevelDB源码分析-TableBuilder生成sstable
TableBuilder生成sstable（include/table_builder.h table/table_builder.cc）
LevelDB使用TableBuilder来构建sstable，并基于TableBuilder封装了一个BuildTable接口，用于将memtable转换为sstable。
sstable的格式为：
datablock1 | datablock2 | ... | metablock1 | metablock2 | ... | metaindexblock | indexblock | footer
datablock即为存储KV数据的块，metablock为相应datablock的元信息的块（并未实现），metaindexblock为metablock的索引块（并未实现），indexblock为datablock的索引块。
footer的数据格式为：
metablockhandle | indexblockhandle | padding | magic
metablockhandle为metaindexblock的索引，indexblockhandle为indexblock的索引，padding为了补全定长，magic为8字节的校验码。
TableBuilder类
TableBuilder类的定义为：
class LEVELDB_EXPORT TableBuilder
{
public:
// ...
// Add key,value to the table being constructed.
// REQUIRES: key is after any previously added key according to comparator.
// REQUIRES: Finish(), Abandon() have not been called
void Add(const Slice &key, const Slice &value);
// Advanced operation: flush any buffered key/value pairs to file.
// Can be used to ensure that two adjacent entries never live in
// the same data block. Most clients should not need to use this method.
// REQUIRES: Finish(), Abandon() have not been called
void Flush();
// ...
// Finish building the table. Stops using the file passed to the
// constructor after this function returns.
// REQUIRES: Finish(), Abandon() have not been called
Status Finish();
// ...
private:
// ..
void WriteBlock(BlockBuilder *block, BlockHandle *handle);
void WriteRawBlock(const Slice &data, CompressionType, BlockHandle *handle);
struct Rep;
Rep *rep_;
};
TableBuilder的成员变量
TableBuilder只有一个类型为Rep *的成员变量rep_。
Rep结构
Rep对TableBuilder构建sstable的过程中的相关数据进行了封装：
struct TableBuilder::Rep
{
Options options;
Options index_block_options;
WritableFile *file;
uint64_t offset;
Status status;
BlockBuilder data_block;
BlockBuilder index_block;
std::string last_key;
int64_t num_entries;
bool closed; // Either Finish() or Abandon() has been called.
FilterBlockBuilder *filter_block;
// We do not emit the index entry for a block until we have seen the
// first key for the next data block. This allows us to use shorter
// keys in the index block. For example, consider a block boundary
// between the keys "the quick brown fox" and "the who". We can use
// "the r" as the key for the index block entry since it is >= all
// entries in the first block and < all entries in subsequent
// blocks.
//
// Invariant: r->pending_index_entry is true only if data_block is empty.
bool pending_index_entry;
BlockHandle pending_handle; // Handle to add to index block
std::string compressed_output;
// ...
};
其中，file为保存sstable的文件，offset为文件中当前偏移的位置，data_block用于构建datablock，index_block用于构建indexblock，last_key为当前最后一个key，num_entries为sstable中entry数量，filter_block用于构建bloom filter，pending_handle用于构建block的索引（存入index_block中），compressed_output为压缩后的sstable。
这里用到了其他的一些辅助的结构，BlockBuilder在LevelDB源码分析-TableBuilder生成sstable中已经分析过了。FilterBlockBuilder将在以后分析。
WritableFile类
WritableFile类为文件操作定义了接口：
class LEVELDB_EXPORT WritableFile
{
public:
WritableFile() = default;
WritableFile(const WritableFile &) = delete;
WritableFile &operator=(const WritableFile &) = delete;
virtual ~WritableFile();
virtual Status Append(const Slice &data) = 0;
virtual Status Close() = 0;
virtual Status Flush() = 0;
virtual Status Sync() = 0;
};
具体实现在PosixWritableFile类中：
class PosixWritableFile : public WritableFile
{
private:
// buf_[0, pos_-1] contains data to be written to fd_.
std::string filename_;
int fd_;
char buf_[kBufSize];
size_t pos_;
public:
// ...
virtual Status Append(const Slice &data)
{
// 见下文分析
}
virtual Status Close()
{
// ...
}
virtual Status Flush()
{
return FlushBuffered();
}
Status SyncDirIfManifest()
{
const char *f = filename_.c_str();
const char *sep = strrchr(f, '/');
Slice basename;
std::string dir;
if (sep == nullptr)
{
dir = ".";
basename = f;
}
else
{
dir = std::string(f, sep - f);
basename = sep + 1;
}
Status s;
if (basename.starts_with("MANIFEST"))
{
int fd = open(dir.c_str(), O_RDONLY);
if (fd < 0)
{
s = PosixError(dir, errno);
}
else
{
if (fsync(fd) < 0)
{
s = PosixError(dir, errno);
}
close(fd);
}
}
return s;
}
virtual Status Sync()
{
// 见下文分析
}
private:
Status FlushBuffered()
{
// ...
}
Status WriteRaw(const char *p, size_t n)
{
// ...
}
};
其中filename_为文件名，int fd_为文件句柄，char buf_[kBufSize]为缓冲区，size_t pos_为当前缓冲区剩余空间起始位置的偏移。
Append函数用于写文件，如果缓冲区空间足够，则先将数据写入缓冲区：
size_t n = data.size();
const char *p = data.data();
// Fit as much as possible into buffer.
size_t copy = std::min(n, kBufSize - pos_);
memcpy(buf_ + pos_, p, copy);
p += copy;
n -= copy;
pos_ += copy;
if (n == 0)
{
return Status::OK();
}
如果空间不够，则将当前缓冲区的内容刷入文件：
// Can't fit in buffer, so need to do at least one write.
Status s = FlushBuffered();
if (!s.ok())
{
return s;
}
然后看写入的数据是否大于缓冲区，如果大于缓冲区，则直接写入文件，否则也写入缓冲区。WriteRaw(const char *p, size_t n)调用write函数将数据写入文件：
// Small writes go to buffer, large writes are written directly.
if (n < kBufSize)
{
memcpy(buf_, p, n);
pos_ = n;
return Status::OK();
}
return WriteRaw(p, n);
Sync函数用于将file中没有被同步到硬盘的部分强制同步到硬盘上，而不是驻留在内存中，保证了持久化。这个函数首先调用SyncDirIfManifest函数将manifest文件写入硬盘（通过fsync函数）：
// Ensure new files referred to by the manifest are in the filesystem.
Status s = SyncDirIfManifest();
if (!s.ok())
{
return s;
}
然后将缓冲区中的数据刷入文件:
s = FlushBuffered();
最后将文件写入硬盘（通过fsyncdata函数）：
if (s.ok())
{
if (fdatasync(fd_) != 0)
{
s = PosixError(filename_, errno);
}
}
return s;
BlockHandle类
BlockHandle类封装了sstable中block的索引：
class BlockHandle
{
public:
BlockHandle();
// The offset of the block in the file.
uint64_t offset() const { return offset_; }
void set_offset(uint64_t offset) { offset_ = offset; }
// The size of the stored block
uint64_t size() const { return size_; }
void set_size(uint64_t size) { size_ = size; }
void EncodeTo(std::string *dst) const;
Status DecodeFrom(Slice *input);
// Maximum encoding length of a BlockHandle
enum
{
kMaxEncodedLength = 10 + 10
};
private:
uint64_t offset_;
uint64_t size_;
};
其中offset_为block的偏移，size_为block的大小。EncodeTo函数将BlockHandle编码为字符串，DecodeFrom函数将Slice封装的数据解码为BlockHandle，以便于处理。
TableBuilder的成员函数
首先是Add函数：
void TableBuilder::Add(const Slice &key, const Slice &value)
首先判断是否需要向index_block中添加索引，如果需要，则调用FindShortestSeparator函数根据last_key计算这个block的索引key值，并将pending_handle编码为字符串作为value值，然后存入index_block中：
Rep *r = rep_;
assert(!r->closed);
if (!ok())
return;
if (r->num_entries > 0)
{
assert(r->options.comparator->Compare(key, Slice(r->last_key)) > 0);
}
if (r->pending_index_entry)
{
assert(r->data_block.empty());
r->options.comparator->FindShortestSeparator(&r->last_key, key);
std::string handle_encoding;
r->pending_handle.EncodeTo(&handle_encoding);
r->index_block.Add(r->last_key, Slice(handle_encoding));
r->pending_index_entry = false;
}
接下来向filter_block中添加key的映射：
if (r->filter_block != nullptr)
{
r->filter_block->AddKey(key);
}
接着讲实际的KV值存入data_block：
r->last_key.assign(key.data(), key.size());
r->num_entries++;
r->data_block.Add(key, value);
最后判断当前index_block的大小是否超过设定值，若超过则调用Flush函数写入文件对象：
const size_t estimated_block_size = r->data_block.CurrentSizeEstimate();
if (estimated_block_size >= r->options.block_size)
{
Flush();
}
这里值得注意的是FindShortestSeparator函数，LevelDB为了节省内存空间，在这里选取的key并不是last_key的key，而是只需要能够区分两个block就可以了。比如block和blrck之间只需要blp就可以区分。
Add函数调用了Flush函数：
void TableBuilder::Flush()
Flush函数首先调用WriteBlock函数将当前还未写入文件对象的block写入文件对象：
Rep *r = rep_;
assert(!r->closed);
if (!ok())
return;
if (r->data_block.empty())
return;
assert(!r->pending_index_entry);
WriteBlock(&r->data_block, &r->pending_handle);
然后使用r->file->Flush将文件缓冲区（由WritableFile封装）中的内容写入文件：
if (ok())
{
r->pending_index_entry = true;
r->status = r->file->Flush();
}
if (r->filter_block != nullptr)
{
r->filter_block->StartBlock(r->offset);
}
Flush函数调用了WriteBlock函数：
void TableBuilder::WriteBlock(BlockBuilder *block, BlockHandle *handle)
WriteBlock函数首先通过block->Finish()向block中加入restarts和num_of_restarts：
// File format contains a sequence of blocks where each block has:
// block_data: uint8[n]
// type: uint8
// crc: uint32
assert(ok());
Rep *r = rep_;
Slice raw = block->Finish();
再根据设定选项对block进行压缩：
Slice block_contents;
CompressionType type = r->options.compression;
// TODO(postrelease): Support more compression options: zlib?
switch (type)
{
case kNoCompression:
block_contents = raw;
break;
case kSnappyCompression:
{
std::string *compressed = &r->compressed_output;
if (port::Snappy_Compress(raw.data(), raw.size(), compressed) &&
compressed->size() < raw.size() - (raw.size() / 8u))
{
block_contents = *compressed;
}
else
{
// Snappy not supported, or compressed less than 12.5%, so just
// store uncompressed form
block_contents = raw;
type = kNoCompression;
}
break;
}
}
最后通过WriteRawBlock函数写入文件对象：
WriteRawBlock(block_contents, type, handle);
r->compressed_output.clear();
block->Reset();
WriteBlock函数调用了WriteRawBlock函数：
void TableBuilder::WriteRawBlock(const Slice &block_contents,
CompressionType type,
BlockHandle *handle)
WriteRawBlock函数通过r->file->Append将传入的除了type和crc之外的block中的数据存入文件对象：
Rep *r = rep_;
handle->set_offset(r->offset);
handle->set_size(block_contents.size());
r->status = r->file->Append(block_contents);
然后计算type和crc，属于同一批：
if (r->status.ok())
{
char trailer[kBlockTrailerSize];
trailer[0] = type;
uint32_t crc = crc32c::Value(block_contents.data(), block_contents.size());
crc = crc32c::Extend(crc, trailer, 1); // Extend crc to cover block type
EncodeFixed32(trailer + 1, crc32c::Mask(crc));
r->status = r->file->Append(Slice(trailer, kBlockTrailerSize));
if (r->status.ok())
{
r->offset += block_contents.size() + kBlockTrailerSize;
}
}
其次是Finish函数：
Status TableBuilder::Finish()
{
Rep *r = rep_;
Flush();
assert(!r->closed);
r->closed = true;
BlockHandle filter_block_handle, metaindex_block_handle, index_block_handle;
// Write filter block
if (ok() && r->filter_block != nullptr)
{
WriteRawBlock(r->filter_block->Finish(), kNoCompression,
&filter_block_handle);
}
// Write metaindex block
if (ok())
{
BlockBuilder meta_index_block(&r->options);
if (r->filter_block != nullptr)
{
// Add mapping from "filter.Name" to location of filter data
std::string key = "filter.";
key.append(r->options.filter_policy->Name());
std::string handle_encoding;
filter_block_handle.EncodeTo(&handle_encoding);
meta_index_block.Add(key, handle_encoding);
}
// TODO(postrelease): Add stats and other meta blocks
WriteBlock(&meta_index_block, &metaindex_block_handle);
}
// Write index block
if (ok())
{
if (r->pending_index_entry)
{
r->options.comparator->FindShortSuccessor(&r->last_key);
std::string handle_encoding;
r->pending_handle.EncodeTo(&handle_encoding);
r->index_block.Add(r->last_key, Slice(handle_encoding));
r->pending_index_entry = false;
}
WriteBlock(&r->index_block, &index_block_handle);
}
// Write footer
if (ok())
{
Footer footer;
footer.set_metaindex_handle(metaindex_block_handle);
footer.set_index_handle(index_block_handle);
std::string footer_encoding;
footer.EncodeTo(&footer_encoding);
r->status = r->file->Append(footer_encoding);
if (r->status.ok())
{
r->offset += footer_encoding.size();
}
}
return r->status;
}
Finish函数首先写入metaindexblock，然后又写入indexblock，最后写入footer。
BuildTable接口
LevelDB调用BuildTable接口进行sstable文件的构建：
Status BuildTable(const std::string &dbname,
Env *env,
const Options &options,
TableCache *table_cache,
Iterator *iter,
FileMetaData *meta)
创建一个文件对象用于存储sstable：
Status s;
meta->file_size = 0;
iter->SeekToFirst();
std::string fname = TableFileName(dbname, meta->number);
if (iter->Valid())
{
WritableFile *file;
s = env->NewWritableFile(fname, &file);
if (!s.ok())
{
return s;
}
创建一个TableBuilder对象用于构建sstable，并通过迭代器遍历memtable将值依次添加进sstable：
TableBuilder *builder = new TableBuilder(options, file);
meta->smallest.DecodeFrom(iter->key());
for (; iter->Valid(); iter->Next())
{
Slice key = iter->key();
meta->largest.DecodeFrom(key);
builder->Add(key, iter->value());
}
// Finish and check for builder errors
s = builder->Finish();
if (s.ok())
{
meta->file_size = builder->FileSize();
assert(meta->file_size > 0);
}
delete builder;
将文件强制写入磁盘：
// Finish and check for file errors
if (s.ok())
{
s = file->Sync();
}
if (s.ok())
{
s = file->Close();
}
delete file;
file = nullptr;
228 Love u
