IT博客汇
  • 首页
  • 精华
  • 技术
  • 设计
  • 资讯
  • 扯淡
  • 权利声明
    • 登录 注册

    [转][转]LevelDB原理探究与代码分析

    heiyeshuwu发表于 2016-05-27 20:56:56
    love 0



    LevelDB原理探究与代码分析



    1. 概述

    Level DB（http://code.google.com/p/leveldb/）是google开源的Key/Value存储系统，它的committer阵容相当强大，基本上是bigtable的原班人马，包括像jeff dean这样的大牛，它的代码合设计非常具有借鉴意义，是一种典型的LSM Tree的KV引擎的实现，从它的数据结构来看，基本就是sstable的开源实现，而且针对各种平台作了port，目前被用在chrome等项目中。


    2. LSM Tree

    Level DB是典型的Log-Structured-Merge Tree的实现，它通过延迟写入以及Write Log Ahead技术来加速数据的写入并保障数据的安全。LevelDB的每个数据文件(sstable)中的记录都是按照Key的顺序进行排序的，但是随机写入时，key的到来是无序的，因此难以将记录插入到其排序位置。于是需要它采取一种延迟写入的方式，批量攒集一定量的数据，将它们在内存中排好序，一次性写入到磁盘中。但是这期间一旦系统断电或其他异常，则可能导致数据丢失，因此需要将数据先写入到log的文件中，这样便将随机写转化为追加写入，对于磁盘性能会有很大提升，如果进程发生中断，重启后可以根据log恢复之前写入的数据。


    2.1 Write Batch

    Level DB只支持两种更新操作:
    1. 插入一条记录 
    2. 删除一条记录
    代码如下：
    [cpp] view plain copy
    1. std::string key1,key2,value;    
    2. leveldb::Status s;  
    3. s = db->Put(leveldb::WriteOptions(), key1, value);    
    4. s = db->Delete(leveldb::WriteOptions(), key2);    
    同时还支持以一种批量的方式写入数据：
    [cpp] view plain copy
    1. std::string key1,key2,value;   
    2. leveldb::WriteBatch batch;   
    3. batch.Delete(key1);   
    4. batch.Put(key2, value);   
    5. leveldb::status s = db->Write(leveldb::WriteOptions(), &batch);  


    其实，在Level DB内部，单独更新与批量更新的调用的接口是相同的，单独更新也会被组织成为包含一条记录的Batch，然后写入数据库中。Write Batch的组织形式如下：

    2.1 Log Format

    每次更新操作都被组织成这样一个数据包，并作为一条日志写入到log文件中，同时也会被解析为一条条内存记录，按照key排序后插入到内存表中的相应位置。LevelDB使用Memory Mapping的方式对log数据进行访问：如果前一次映射的空间已写满，则先将文件扩展一定的长度（每次扩展的长度按64KB,128KB,...的顺序逐次翻倍，最大到1MB），然后映射到内存，对映射的内存再以32KB的Page进行切分，每次写入的日志填充到Page中，攒积一定量后Sync到磁盘上（也可以设置WriteOptions，每写一条日志就Sync一次，但是这样效率很低），内存映射文件的代码如下：
    [cpp] view plain copy
    1. class PosixMmapFile : public WritableFile   
    2. {  
    3. private:  
    4.   std::string filename_;  // 文件名称  
    5.   int fd_;                // 文件句柄  
    6.   size_t page_size_;      //   
    7.   size_t map_size_;       // 内存映射的区域大小  
    8.   char* base_;            // 内存映射区域的起始地址  
    9.   char* limit_;           // 内存映射区域的结束地址  
    10.   char* dst_;             // 最后一次占用的内存的结束地址  
    11.   char* last_sync_;       // 最后一次同步到磁盘的结束地址  
    12.   uint64_t file_offset_;  // 当前文件的偏移值  
    13.   bool pending_sync_;     // 延迟同步的标志  
    14.    
    15. public:  
    16.   PosixMmapFile(const std::string& fname, int fd, size_t page_size)  
    17.       : filename_(fname),  
    18.         fd_(fd),  
    19.         page_size_(page_size),  
    20.         map_size_(Roundup(65536, page_size)),  
    21.         base_(NULL),  
    22.         limit_(NULL),  
    23.         dst_(NULL),  
    24.         last_sync_(NULL),  
    25.         file_offset_(0),  
    26.         pending_sync_(false) {  
    27.     assert((page_size & (page_size - 1)) == 0);  
    28.   }  
    29.   
    30.   ~PosixMmapFile() {  
    31.     if (fd_ >= 0) {  
    32.       PosixMmapFile::Close();  
    33.     }  
    34.   }  
    35.   
    36.   Status Append(const Slice& data) {  
    37.     const char* src = data.data();  
    38.     size_t left = data.size();  
    39.     while (left > 0) {  
    40.       // 计算上次最后一次申请的区域的剩余容量，如果已完全耗尽，  
    41.       // 则卸载当前区域，申请一个新的区域  
    42.       size_t avail = limit_ - dst_;  
    43.       if (avail == 0) {  
    44.         if (!UnmapCurrentRegion() ||  
    45.             !MapNewRegion()) {  
    46.           return IOError(filename_, errno);  
    47.         }  
    48.       }  
    49.       // 填充当前区域的剩余容量  
    50.       size_t n = (left <= avail) ? left : avail;  
    51.       memcpy(dst_, src, n);  
    52.       dst_ += n;  
    53.       src += n;  
    54.       left -= n;  
    55.     }  
    56.     return Status::OK();  
    57.   }  
    58.   
    59.   Status PosixMmapFile::Close() {  
    60.     Status s;  
    61.     size_t unused = limit_ - dst_;  
    62.     if (!UnmapCurrentRegion()) {  
    63.       s = IOError(filename_, errno);  
    64.     } else if (unused > 0) {  
    65.       // 关闭时将文件没有使用用的空间truncate掉  
    66.       if (ftruncate(fd_, file_offset_ - unused) < 0) {  
    67.         s = IOError(filename_, errno);  
    68.       }  
    69.     }  
    70.   
    71.     if (close(fd_) < 0) {  
    72.       if (s.ok()) {  
    73.         s = IOError(filename_, errno);  
    74.       }  
    75.     }  
    76.   
    77.     fd_ = -1;  
    78.     base_ = NULL;  
    79.     limit_ = NULL;  
    80.     return s;  
    81.   }  
    82.     
    83.   virtual Status Sync() {  
    84.     Status s;  
    85.     if (pending_sync_) {  
    86.       // 上个区域也有数据未同步，则先同步数据  
    87.       pending_sync_ = false;  
    88.       if (fdatasync(fd_) < 0) {  
    89.         s = IOError(filename_, errno);  
    90.       }  
    91.     }  
    92.   
    93.     if (dst_ > last_sync_) {  
    94.       // 计算未同步数据的起始与结束地址，同步时，起始地址按page_size_向下取整，  
    95.       // 结束地址向上取整，保证每次同步都是同步一个或多个page  
    96.       size_t p1 = TruncateToPageBoundary(last_sync_ - base_);  
    97.       size_t p2 = TruncateToPageBoundary(dst_ - base_ - 1);   
    98.       // 如果刚好为整数个page_size_，由于下面同步时必然会加一个page_size_，所以这里可以减去1  
    99.       last_sync_ = dst_;  
    100.       if (msync(base_ + p1, p2 - p1 + page_size_, MS_SYNC) < 0) {  
    101.         s = IOError(filename_, errno);  
    102.       }  
    103.     }  
    104.     return s;  
    105.   }  
    106. private:  
    107.   // 将x按y向上对齐     
    108.   static size_t Roundup(size_t x, size_t y) {  
    109.     return ((x + y - 1) / y) * y;  
    110.   }  
    111.   // 将s按page_size_向下对齐  
    112.   size_t TruncateToPageBoundary(size_t s) {  
    113.     s -= (s & (page_size_ - 1));  
    114.     assert((s % page_size_) == 0);  
    115.     return s;  
    116.   }   
    117.   
    118.   // 卸载当前映射的内存区域    
    119.   bool UnmapCurrentRegion() {      
    120.     bool result = true;  
    121.     if (base_ != NULL) {  
    122.       if (last_sync_ < limit_) {  
    123.         // 如果当前页没有完全被同步，则标明本文件需要被同步，下次调用Sync()方法时会将本页中未同步的数据同步到磁盘  
    124.         pending_sync_ = true;  
    125.       }  
    126.       if (munmap(base_, limit_ - base_) != 0) {  
    127.         result = false;  
    128.       }  
    129.       file_offset_ += limit_ - base_;  
    130.       base_ = NULL;  
    131.       limit_ = NULL;  
    132.       last_sync_ = NULL;  
    133.       dst_ = NULL;      // 使用翻倍的策略增加下次申请区域的大小，最大到1MB  
    134.       if (map_size_ < (1<<20)) {  
    135.         map_size_ *= 2;  
    136.       }  
    137.     }  
    138.     return result;  
    139.   }    
    140.   
    141.   bool MapNewRegion() {  
    142.     assert(base_ == NULL); // 申请一个新的区域时，上一个申请的区域必须已经卸载   
    143.     // 先将文件扩大      
    144.     if (ftruncate(fd_, file_offset_ + map_size_) < 0) {  
    145.       return false;  
    146.     }  
    147.     // 将新区域映射到文件  
    148.     void* ptr = mmap(NULL, map_size_, PROT_READ | PROT_WRITE, MAP_SHARED,  
    149.                      fd_, file_offset_);  
    150.     if (ptr == MAP_FAILED) {  
    151.       return false;  
    152.     }  
    153.     base_ = reinterpret_cast<char*>(ptr);  
    154.     limit_ = base_ + map_size_;  
    155.     dst_ = base_;  
    156.     last_sync_ = base_;  
    157.     return true;  
    158.   }  
    159. };  
    但是，一个Batch的数据按上面的方式组织后，如果做为一条日志写入Log，则很可能需要跨两个或更多个Page；为了更好地管理日志以及保障数据安全，LevelDB对日志记录进行了更细的切分，如果一个Batch对应的数据需要跨页，则会将其切分为多条Entry，然后写入到不同Page中，Entry不会跨越Page，我们通过对多个Entry进行解包，可以还原出的Batch数据。最终，LevelDB的log文件被组织为下面的形式：


    这里，我们可以看一下log_writer的代码：
    [cpp] view plain copy
    1. Status Writer::AddRecord(const Slice& slice) {  
    2.   const char* ptr = slice.data();  
    3.   size_t left = slice.size();  
    4.   
    5.   Status s;  
    6.   bool begin = true;  
    7.   do {  
    8.     const int leftover = kBlockSize - block_offset_;  
    9.     assert(leftover >= 0);  
    10.     if (leftover < kHeaderSize) {  
    11.       // 如果当前page的剩余长度小于7字节且大于0，则都填充'\0'，并新起一个page  
    12.       if (leftover > 0) {  
    13.         assert(kHeaderSize == 7);  
    14.         dest_->Append(Slice("\x00\x00\x00\x00\x00\x00", leftover));  
    15.       }  
    16.       block_offset_ = 0;  
    17.     }  
    18.     
    19.     // 计算page能否容纳整体日志，如果不能，则将日志切分为多条entry，插入不同的page中，type中注明该entry是日志的开头部分，中间部分还是结尾部分。  
    20.     const size_t avail = kBlockSize - block_offset_ - kHeaderSize;  
    21.     const size_t fragment_length = (left < avail) ? left : avail;  
    22.   
    23.     RecordType type;  
    24.     const bool end = (left == fragment_length);  
    25.     if (begin && end) {  
    26.       type = kFullType;   // 本Entry保存完整的Batch  
    27.     } else if (begin) {  
    28.       type = kFirstType;  // 本Entry只保存起始部分  
    29.     } else if (end) {  
    30.       type = kLastType;   // 本Entry只保存结束部分  
    31.     } else {  
    32.       type = kMiddleType; // 本Entry保存Batch的中间部分，不含起始与结尾，有时可能需要保存多个middle  
    33.     }  
    34.   
    35.     s = EmitPhysicalRecord(type, ptr, fragment_length);  
    36.     ptr += fragment_length;  
    37.     left -= fragment_length;  
    38.     begin = false;  
    39.   } while (s.ok() && left > 0);  
    40.   return s;  
    41. }  
    42.   
    43. Status Writer::EmitPhysicalRecord(RecordType t, const char* ptr, size_t n) {  
    44.   assert(n <= 0xffff);    
    45.   assert(block_offset_ + kHeaderSize + n <= kBlockSize);  
    46.   
    47.   // 填充记录头  
    48.   char buf[kHeaderSize];  
    49.   buf[4] = static_cast<char>(n & 0xff);  
    50.   buf[5] = static_cast<char>(n >> 8);  
    51.   buf[6] = static_cast<char>(t);  
    52.   
    53.   // 计算crc  
    54.   uint32_t crc = crc32c::Extend(type_crc_[t], ptr, n);  
    55.   crc = crc32c::Mask(crc);   
    56.   EncodeFixed32(buf, crc);  
    57.   
    58.   // 填充entry内容  
    59.   Status s = dest_->Append(Slice(buf, kHeaderSize));  
    60.   if (s.ok()) {  
    61.     s = dest_->Append(Slice(ptr, n));  
    62.     if (s.ok()) {  
    63.       s = dest_->Flush();  
    64.     }  
    65.   }  
    66.   block_offset_ += kHeaderSize + n;  
    67.   return s;  
    68. }  

    2.3 Write Log Ahead

    Level DB在更新时，先写log，然后更新memtable，每个memtable会设置一个最大容量，如果超过阈值，则采用双buffer机制，关闭当前log文件并将当前memtable切换未从memtable，然后新建一个log文件以及memtable，将数据写进新的log文件与memtable，并通知后台线程对从memtable进行处理，及时将其dump到磁盘上，或者启动compaction流程。Write的代码分析如下：
    [cpp] view plain copy
    1. Status DBImpl::Write(const WriteOptions& options, WriteBatch* updates)   
    2. {  
    3.   Status status;  
    4.   MutexLock l(&mutex_);  // 锁定互斥体，同一时间只能有一个线程更新数据  
    5.   LoggerId self;     
    6.   // 获取Logger的使用权,如果有其他线程拥有所有权，则等待至其释放所有权。  
    7.   AcquireLoggingResponsibility(&self);  
    8.   status = MakeRoomForWrite(false);  // May temporarily release lock and wait  
    9.   uint64_t last_sequence = versions_->LastSequence();  // 获取当前的版本号  
    10.   if (status.ok()) {  
    11.     // 将当前版本号加1后作为本次更新的日志的版本，  
    12.     // 一次批量更新可能包含多个操作，这些操作都用一个版本有一个好处：  
    13.     // 本次更新的所有操作，要么都可见，要么都不可见，不存在一部分可见，另一部分不可见的情况。  
    14.     WriteBatchInternal::SetSequence(updates, last_sequence + 1);  
    15.     // 但是本次更新可能有多个操作，跳过与操作数相等的版本号，保证不被使用  
    16.     last_sequence += WriteBatchInternal::Count(updates);  
    17.   
    18.     // 将batch写入log，然后应用到memtable中  
    19.     {  
    20.       assert(logger_ == &self);  
    21.       mutex_.Unlock();  
    22.       // 这里，可以解锁，因为在AcquireLoggingResponsibility()方法中已经获取了Logger的拥有权，  
    23.       // 其他线程即使获得了锁，但是由于&self != logger，其会阻塞在AcquireLoggingResponsibility()方法中。  
    24.       // 将更新写入log文件，如果设置了每次写入进行sync，则将其同步到磁盘，这个操作可能比较长，  
    25.       // 防止了mutex_对象长期被占用，因为其还负责其他一些资源的同步  
    26.       status = log_->AddRecord(WriteBatchInternal::Contents(updates));  
    27.       if (status.ok() && options.sync) {  
    28.         status = logfile_->Sync();  
    29.       }  
    30.       if (status.ok()) {  
    31.         // 成功写入了log后，才写入memtable  
    32.         status = WriteBatchInternal::InsertInto(updates, mem_);  
    33.       }  
    34.       // 重新锁定mutex_  
    35.       mutex_.Lock();  
    36.       assert(logger_ == &self);  
    37.     }  
    38.     // 更新版本号  
    39.     versions_->SetLastSequence(last_sequence);  
    40.   }  
    41.   // 释放对logger的所有权，并通知等待的线程，然后解锁  
    42.   ReleaseLoggingResponsibility(&self);  
    43.   return status;  
    44. }  
    45.   
    46. // force参数表示强制新起一个memtable  
    47. Status DBImpl::MakeRoomForWrite(bool force) {  
    48.   mutex_.AssertHeld();  
    49.   assert(logger_ != NULL);  
    50.   bool allow_delay = !force;  
    51.   Status s;  
    52.   while (true) {  
    53.     if (!bg_error_.ok()) {  
    54.       // 后台线程存在问题，则返回错误，不接受更新  
    55.       s = bg_error_;  
    56.       break;  
    57.     } else if (  
    58.         allow_delay &&  
    59.         versions_->NumLevelFiles(0) >= config::kL0_SlowdownWritesTrigger) {  
    60.       // 如果不是强制写入，而且level 0的sstable超过8个，则本次更新阻塞1毫秒，  
    61.       // leveldb将sstable分为多个等级，其中level 0中的不同表的key是可能重叠的，  
    62.       // 如果l0的sstable过多，会导致查询性能下降，这时需要适当降低更新速度，让  
    63.       // 后台线程进行compaction操作，但是设计者不希望让某次写操作等待数秒，  
    64.       // 而是让每次更新操作分担延迟，即每次写操作阻塞1毫秒，平衡读写速率；  
    65.       // 另外，理论上这也能让compaction线程获得更多的cpu时间（当然，  
    66.       // 这是假定compaction与更新操作共享一个CPU时才有意义）  
    67.       mutex_.Unlock();  
    68.       env_->SleepForMicroseconds(1000);  
    69.       allow_delay = false;  // 最多延迟一次，下次不延迟  
    70.       mutex_.Lock();  
    71.     } else if (!force &&  
    72.                (mem_->ApproximateMemoryUsage() <= options_.write_buffer_size)) {  
    73.       // 如果当前memtable已使用的空间小于write_buffer_size，则跳出，更新到当前memtable即可。  
    74.       // 当force为true时，第一次循环会走后面else逻辑，切换了memtable后force被置为false，  
    75.       // 第二次循环时就可以在此跳出了  
    76.       break;  
    77.     } else if (imm_ != NULL) {  
    78.       // 如果当前memtable已经超过write_buffer_size,且备用的memtable也在被使用，则阻塞更新并等待  
    79.       bg_cv_.Wait();  
    80.     } else if (versions_->NumLevelFiles(0) >= config::kL0_StopWritesTrigger) {  
    81.       // 如果当前memtable已使用的空间小于write_buffer_size，但是备用的memtable未被使用，  
    82.       // 则检查level 0的sstable个数，如超过12个，则阻塞更新并等待  
    83.       Log(options_.info_log, "waiting...\n");  
    84.       bg_cv_.Wait();  
    85.     } else {  
    86.       // 否则，使用新的id新创建一个log文件，并将当前memtable切换为备用的memtable，新建一个  
    87.       // memtable，然后将数据写入当前的新memtable，即切换log文件与memtable，并告诉后台线程  
    88.       // 可以进行compaction操作了  
    89.       assert(versions_->PrevLogNumber() == 0);  
    90.       uint64_t new_log_number = versions_->NewFileNumber();  
    91.       WritableFile* lfile = NULL;  
    92.       s = env_->NewWritableFile(LogFileName(dbname_, new_log_number), &lfile);  
    93.       if (!s.ok()) {  
    94.         break;  
    95.       }  
    96.       delete log_;  
    97.       delete logfile_;  
    98.       logfile_ = lfile;  
    99.       logfile_number_ = new_log_number;  
    100.       log_ = new log::Writer(lfile);  
    101.       imm_ = mem_;  
    102.       has_imm_.Release_Store(imm_);  
    103.       mem_ = new MemTable(internal_comparator_);  
    104.       mem_->Ref();  
    105.       force = false;   // 下次判断可以不新建memtable了  
    106.       MaybeScheduleCompaction();  
    107.     }  
    108.   }  
    109.   return s;  
    110. }  
    111. void DBImpl::AcquireLoggingResponsibility(LoggerId* self) {  
    112.   while (logger_ != NULL) {  
    113.     logger_cv_.Wait();  
    114.   }  
    115.   logger_ = self;  
    116. }  
    117.   
    118. void DBImpl::ReleaseLoggingResponsibility(LoggerId* self) {  
    119.   assert(logger_ == self);  
    120.   logger_ = NULL;  
    121.   logger_cv_.SignalAll();  
    122. }  

    2.4 Skip List

    Level DB内部采用跳表结构来组织Memtable，每插入一条记录，先根据跳表通过多次key的比较，定位到记录应该插入的位置，然后按照一定的概率确定该节点需要建立多少级的索引，跳表结构如下：


    Level DB的SkipList最高12层，最下面一层（level0）的链是全链，即每条记录必须在此链中插入相应的索引节点；从level1到level11则是按概率决定是否需要建索引，概率按照1/4的因子等比递减。下面举个例子，说明一下这个流程：
    1. 看上图，假定我们链不存在record3，level0中，record2的下一条记录是record4，level1中，record2的下一条记录是record5。
    2. 现在，我们插入一条记录record3，通过key的比较，我们定位到它应该在record2与record4之间。
    3. 然后，我们按照下面的代码确定一条记录需要在跳表中建立几重索引：

    [cpp] view plain copy
    1. template<typename Key, class Comparator>  
    2. int SkipList<Key,Comparator>::RandomHeight() {  
    3.   // Increase height with probability 1 in kBranching  
    4.   static const unsigned int kBranching = 4;  
    5.   int height = 1;  
    6.   while (height < kMaxHeight && ((rnd_.Next() % kBranching) == 0)) {  
    7.     height++;  
    8.   }  
    9.   return height;  
    10. }  

    按照上面的代码，我们可以得出，建立x级索引的概率是0.25 ^(x - 1) * 0.75，所以，建立1级索引的概率为75%，建立2级索引的概率为25%*75%=18.75%，...（个人感觉，google把分支因子定为4有点高了，这样在绝大多数情况下，跳表的高度都不大于3）。

    4.  在level0 ~ level (x-1)中链表的合适位置插入record3，假定根据上面的公式，我们得到需要为record3建立2级索引，即x=2，因此需要在level0与level1中的链中插入record3：在level 0的链中，record3插在record2与record4之间，在level 1的链中，record3插入在record2与record5之间，形成了现在的索引结构，在查询一个记录时，可以从最高一级索引向下查找，节约比较次数。

    2.5 Record Format

    Level DB将用户的每个更新或删除操作组合成一个Record，其格式如下：


    从图中可以看出，每个Record会在原用key的基础上添加版本号以及key的类型（更新 or 删除），组成internal key。插入跳表时，是按照internal key进行排序，而非用户key。这样，我们只可能向跳表中添加节点，而不可能删除和替换节点。

    Internal Key在比较时，按照下面的算法：

    [cpp] view plain copy
    1. int InternalKeyComparator::Compare(const Slice& akey, const Slice& bkey) const {  
    2.   int r = user_comparator_->Compare(ExtractUserKey(akey), ExtractUserKey(bkey));  
    3.   if (r == 0) {  
    4.     // 比较后面8个字节构造的整数，第一个字节的type为Least Significant Byte  
    5.     const uint64_t anum = DecodeFixed64(akey.data() + akey.size() - 8);  
    6.     const uint64_t bnum = DecodeFixed64(bkey.data() + bkey.size() - 8);  
    7.     if (anum > bnum)  // 注意：整数大反而key比较小  
    8.     {  
    9.       r = -1;  
    10.     } else if (anum < bnum) {  
    11.       r = +1;  
    12.     }  
    13.   }  
    14.   return r;  
    15. }  

    根据上面的算法，我们可以得知Internal Key的比较顺序：

    1. 如果User Key不相等，则User Key比较小的记录的Internal Key也比较小，User Key默认采用字典序（lexicographic）进行比较，可以在建表参数中自定义comparator。
    2. 如果type也相同，则比较Sequence Num，Sequence Num大的Internal Key比较小。
    3. 如果Sequence Num相等，则比较Type，type为更新（Key Type=1）的记录比的type为删除（Key Type=0）的记录的Internal Key小。

    在插入到跳表时，一般不会出现Internal Key相等的情况（除非在一个Batch中操作了同一条记录两次，这里会出现一种bug：在一个Write Batch中，先插入一条记录，然后删除这条记录，最后把这个Batch写入DB，会发现DB中这条记录存在。因此，不推荐在Batch中多次操作相同key的记录），User Key相同的记录插入跳表时，Sequence Num大的记录会排在前面。
    设计Internal Key有个以下一些作用：
    1. Level DB支持快照查询，即查询时指定快照的版本号，查询出创建快照时某个User Key对应的Value，那么可以组成这样一个Internal Key：Sequence=快照版本号，Type=1，User Key为用户指定Key，然后查询数据文件与内存，找到大于等于此Internal Key且User Key匹配的第一条记录即可（即Sequence Num小于等于快照版本号的第一条记录）。
    2.如果查询最新的记录时，将Sequence Num设置为0xFFFFFFFFFFFFFF即可。因为我们更多的是查询最新记录，所以让Sequence Num大的记录排前面，可以在遍历时遇见第一条匹配的记录立即返回，减少往后遍历的次数。


    3.文件结构

    3.1 文件组成

    Level DB包含一下几种文件：

    文件类型说明
    dbname/MANIFEST-[0-9]+  清单文件            
    dbname/[0-9]+.logdb日志文件
    dbname/[0-9]+.sstdbtable文件
    dbname/[0-9]+.dbtmpdb临时文件
    dbname/CURRENT 记录当前使用的清单文件名
    dbname/LOCK  DB锁文件
    dbname/LOGinfo log日志文件
    dbname/LOG.old旧的info log日志文件

    上面的log文件，sst文件，临时文件，清单文件末尾都带着序列号，序号是单调递增的（随着next_file_number从1开始递增），以保证不会和之前的文件名重复。另外，注意区分db log与info log：前者是为了防止保障数据安全而实现的二进制Log，后者是打印引擎中间运行状态及警告等信息的文本log。
    随着更新与Compaction的进行，Level DB会不断生成新文件，有时还会删除老文件，所以需要一个文件来记录文件列表，这个列表就是清单文件的作用，清单会不断变化，DB需要知道最新的清单文件，必须将清单准备好后原子切换，这就是CURRENT文件的作用，Level DB的清单过程更新如下：
    1. 递增清单序号，生成一个新的清单文件。
    2. 将此清单文件的名称写入到一个临时文件中。
    3. 将临时文件rename为CURRENT。
    代码如下：
    [cpp] view plain copy
    1. Status SetCurrentFile(Env* env, const std::string& dbname,  
    2.                       uint64_t descriptor_number) {  
    3.   // 创建一个新的清单文件名  
    4.   std::string manifest = DescriptorFileName(dbname, descriptor_number);  
    5.   Slice contents = manifest;  
    6.   // 移除"dbname/"前缀  
    7.   assert(contents.starts_with(dbname + "/"));  
    8.   contents.remove_prefix(dbname.size() + 1);  
    9.   // 创建一个临时文件  
    10.   std::string tmp = TempFileName(dbname, descriptor_number);  
    11.   // 写入清单文件名  
    12.   Status s = WriteStringToFile(env, contents.ToString() + "\n", tmp);  
    13.   if (s.ok()) {  
    14.     // 将临时文件改名为CURRENT  
    15.     s = env->RenameFile(tmp, CurrentFileName(dbname));  
    16.   }  
    17.   if (!s.ok()) {  
    18.     env->DeleteFile(tmp);  
    19.   }  
    20.   return s;  
    21. }  

    3.2 Manifest

    在介绍其他文件格式前，先了解清单文件，MANIFEST文件是Level DB的元信息文件，它主要包括下面一些信息：
    1. Comparator的名称
    2. 
    其的格式如下：

    我们可以看看其序列化的代码：
    [cpp] view plain copy
    1. void VersionEdit::EncodeTo(std::string* dst) const {  
    2.   if (has_comparator_) {  // 记录Comparator名称  
    3.     PutVarint32(dst, kComdparator);  
    4.     PutLengthPrefixedSlice(dst, comparator_);  
    5.   }  
    6.   if (has_log_number_) {  // 记录Log Numer  
    7.     PutVarint32(dst, kLogNumber);  
    8.     PutVarint64(dst, log_number_);  
    9.   }  
    10.   if (has_prev_log_number_) {  // 记录Prev Log Number，现在已废弃，一般为0  
    11.     PutVarint32(dst, kPrevLogNumber);  
    12.     PutVarint64(dst, prev_log_number_);  
    13.   }  
    14.   if (has_next_file_number_) {  // 记录下一个文件序号  
    15.     PutVarint32(dst, kNextFileNumber);  
    16.     PutVarint64(dst, next_file_number_);  
    17.   }  
    18.   if (has_last_sequence_) {  // 记录最大的sequence num  
    19.     PutVarint32(dst, kLastSequence);  
    20.     PutVarint64(dst, last_sequence_);  
    21.   }  
    22.   // 记录每一级Level下次compaction的起始Key  
    23.   for (size_t i = 0; i < compact_pointers_.size(); i++) {  
    24.     PutVarint32(dst, kCompactPointer);  
    25.     PutVarint32(dst, compact_pointers_[i].first);  // level  
    26.     PutLengthPrefixedSlice(dst, compact_pointers_[i].second.Encode());  
    27.   }  
    28.   // 记录每一级需要删除的文件  
    29.   for (DeletedFileSet::const_iterator iter = deleted_files_.begin();  
    30.        iter != deleted_files_.end();  
    31.        ++iter) {  
    32.     PutVarint32(dst, kDeletedFile);  
    33.     PutVarint32(dst, iter->first);   // level  
    34.     PutVarint64(dst, iter->second);  // file number  
    35.   }  
    36.   // 记录每一级需要有效的sst以及其smallest与largest的key  
    37.   for (size_t i = 0; i < new_files_.size(); i++) {  
    38.     const FileMetaData& f = new_files_[i].second;  
    39.     PutVarint32(dst, kNewFile);  
    40.     PutVarint32(dst, new_files_[i].first);  // level  
    41.     PutVarint64(dst, f.number);  
    42.     PutVarint64(dst, f.file_size);  
    43.     PutLengthPrefixedSlice(dst, f.smallest.Encode());  
    44.     PutLengthPrefixedSlice(dst, f.largest.Encode());  
    45.   }  
    46. }  

    3.3 Sortedtable

    Level DB间歇性地将内存中的SkipList对应的数据集合Dump到磁盘上，生成一个sst的文件，这个文件的格式如下：

    按照SSTable的结构，可以正向遍历，也可以逆向遍历，但是逆向遍历的代价要远远高于正向遍历的代价，因为每条record都是变长的，且其没有记录前一条记录的偏移，因此逆向Group遍历时，只能先回到group（代码中称为一个restart，为了便于理解，下面都称为group）开头（一个Data Block的group一般为16条记录，每个Data Block的尾部有group起始位置偏移索引），然后从头开始正向遍历，直至找到其前一条记录，如果当前位置为group的第一条记录，则需要回到上一个group的开头，遍历到其最后一条记录。另外，内存中跳表反向的遍历效率也远远不如正向遍历。

    3.4 Sparse Index

    一个sst文件内部除了Data Block，还有Index Block，Index Block的结构与Data Block一样，只不过每个group只包含一条记录，即Data Block的最大Key与偏移。其实这里说最大Key并不是很准确，理论上，只要保存最大Key就可以实现二分查找，但是Level DB在这里做了个优化，它并保存最大key，而是保存一个能分隔两个Data Block的最短Key，如：假定Data Block1的最后一个Key为“abcdefg”，Data Block2的第一个Key为“abzxcv”，则index可以记录Data Block1的索引key为“abd”；这样的分割串可以有很多，只要保证Data Block1中的所有Key都小于等于此索引，Data Block2中的所有Key都大于此索引即可。这种优化缩减了索引长度，查询时可以有效减小比较次数。我们可以看看默认comparator如何实现这种分割的：
    [cpp] view plain copy
    1. void BytewiseComparatorImpl::FindShortestSeparator(  
    2.       std::string* start,  
    3.       const Slice& limit) const {  
    4.     // 先比较获得最大公共前缀  
    5.     size_t min_length = std::min(start->size(), limit.size());  
    6.     size_t diff_index = 0;  
    7.     while ((diff_index < min_length) &&  
    8.            ((*start)[diff_index] == limit[diff_index])) {  
    9.       diff_index++;  
    10.     }  
    11.     if (diff_index >= min_length) {  
    12.       // 如果start就是limit的前缀，则只能使用start本身作为分割  
    13.     } else {  
    14.       uint8_t diff_byte = static_cast<uint8_t>((*start)[diff_index]);  
    15.       // 将第一个不同字符+1，并确保其不会溢出，同时比limit小  
    16.       if (diff_byte < static_cast<uint8_t>(0xff) &&  
    17.           diff_byte + 1 < static_cast<uint8_t>(limit[diff_index])) {  
    18.         (*start)[diff_index]++;  
    19.         start->resize(diff_index + 1);  
    20.         assert(Compare(*start, limit) < 0);  
    21.       }  
    22.     }  
    23.   }  
    从上面可以看出，FindShortestSeparator方法并不严格，有些时候没有找出最短分割的key（比如第一个不等的字符已经为0xFF时），它只是一种优化，我们自定义Comparator时，既可以实现，也可以不实现，如果不实现，将始终使用Data Block的最大Key作为索引，并不影响功能正确性。

    4. Operations

    在介绍了数据结构后，我们看看Level DB一些基本操作的实现：

    4.1 创建一个新表

    创建一个新的表大概分为几步，包括建立各类文件以及内存中的数据结构，线程同步对象等，关键代码如下：

    [cpp] view plain copy
    1. // DBImpl在构造时会初始化互斥体与信号量，创建一个空的memtable，并根据配置设置Comparator及LRU缓冲  
    2. DBImpl::DBImpl(const Options& options, const std::string& dbname)  
    3.     : env_(options.env),  
    4.       internal_comparator_(options.comparator), // 初始化Comparator  
    5.       options_(SanitizeOptions(dbname, &internal_comparator_, options)),  // 检查参数是否合法  
    6.       owns_info_log_(options_.info_log != options.info_log),  // 是拥有自己info log，还是使用用户提供的  
    7.       owns_cache_(options_.block_cache != options.block_cache), // 是否拥有自己的LRU缓冲，或者使用用户提供的  
    8.       dbname_(dbname),  // 数据表名称  
    9.       db_lock_(NULL),  // 不创建也不锁定文件锁  
    10.       shutting_down_(NULL),   
    11.       bg_cv_(&mutex_),  // 用于与后台线程交互的条件信号  
    12.       mem_(new MemTable(internal_comparator_)), // 创建一个新的跳表  
    13.       imm_(NULL),  // 用于双缓冲的缓冲跳表开始时为NULL  
    14.       logfile_(NULL),  // log文件  
    15.       logfile_number_(0), // log文件的序号  
    16.       log_(NULL),  // log writer  
    17.       logger_(NULL),  // 用于在多线程环境中记录Owner logger的一个指针  
    18.       logger_cv_(&mutex_), // 用于与Logger交互的条件信号  
    19.       bg_compaction_scheduled_(false), // 没打开表时不起动后台的compaction线程  
    20.       manual_compaction_(NULL) {  
    21.   // 增加memtable的引用计数  
    22.   mem_->Ref();  
    23.   has_imm_.Release_Store(NULL);  
    24.   
    25.   // 根据Option创建一个LRU的缓冲对象，如果options中指定了Cache空间，则使用用户  
    26.   // 提供的Cache空间，否则会在内部确实创建8MB的Cache，另外，LRU的Entry数目不能超过max_open_files-10  
    27.   const int table_cache_size = options.max_open_files - 10;  
    28.   table_cache_ = new TableCache(dbname_, &options_, table_cache_size);  
    29.   
    30.   // 创建一个Version管理器  
    31.   versions_ = new VersionSet(dbname_, &options_, table_cache_,  
    32.                              &internal_comparator_);  
    33. }  
    34.   
    35. Options SanitizeOptions(const std::string& dbname,  
    36.                         const InternalKeyComparator* icmp,  
    37.                         const Options& src) {  
    38.   Options result = src;  
    39.   result.comparator = icmp;  
    40.   ClipToRange(&result.max_open_files,           20,     50000);  
    41.   ClipToRange(&result.write_buffer_size,        64<<10, 1<<30);  
    42.   ClipToRange(&result.block_size,               1<<10,  4<<20);  
    43.   // 如果用户未指定info log文件（用于打印状态等文本信息的日志文件），则由引擎自己创建一个info log文件。  
    44.   if (result.info_log == NULL) {  
    45.     // Open a log file in the same directory as the db  
    46.     src.env->CreateDir(dbname);  // 如果目录不存在则创建  
    47.     // 如果已存在以前的info log文件，则将其改名为LOG.old，然后创建新的log文件与日志的writer  
    48.     src.env->RenameFile(InfoLogFileName(dbname), OldInfoLogFileName(dbname));  
    49.     Status s = src.env->NewLogger(InfoLogFileName(dbname), &result.info_log);  
    50.     if (!s.ok()) {  
    51.       result.info_log = NULL;  
    52.     }  
    53.   }  
    54.   // 如果用户没指定LRU缓冲，则创建8MB的LRU缓冲  
    55.   if (result.block_cache == NULL) {  
    56.     result.block_cache = NewLRUCache(8 << 20);  
    57.   }  
    58.   return result;  
    59. }  
    60.   
    61. Status DBImpl::NewDB() {  
    62.   // 创建version管理器  
    63.   VersionEdit new_db;  
    64.   // 设置Comparator  
    65.   new_db.SetComparatorName(user_comparator()->Name());  
    66.   new_db.SetLogNumber(0);  
    67.   // 下一个序号从2开始，1留给清单文件  
    68.   new_db.SetNextFile(2);  
    69.   new_db.SetLastSequence(0);  
    70.   // 创建一个清单文件，MANIFEST-1  
    71.   const std::string manifest = DescriptorFileName(dbname_, 1);  
    72.   WritableFile* file;  
    73.   Status s = env_->NewWritableFile(manifest, &file);  
    74.   if (!s.ok()) {  
    75.     return s;  
    76.   }  
    77.   {  
    78.     // 写入清单文件头  
    79.     log::Writer log(file);  
    80.     std::string record;  
    81.     new_db.EncodeTo(&record);  
    82.     s = log.AddRecord(record);  
    83.     if (s.ok()) {  
    84.       s = file->Close();  
    85.     }  
    86.   }  
    87.   delete file;  
    88.   if (s.ok()) {  
    89.     // 设置CURRENT文件，使其指向清单文件  
    90.     s = SetCurrentFile(env_, dbname_, 1);  
    91.   } else {  
    92.     env_->DeleteFile(manifest);  
    93.   }  
    94.   return s;  

    4.2 打开一个已存在的表

    上面的步骤中，其实还遗漏了一个的重要流程，那就是DB的Open方法。Level DB无论是创建表，还是打开现有的表，都是使用Open方法。代码如下：
    [cpp] view plain copy
    1. Status DB::Open(const Options& options, const std::string& dbname,  
    2.                 DB** dbptr) {  
    3.   *dbptr = NULL;  
    4.   
    5.   DBImpl* impl = new DBImpl(options, dbname);  
    6.   impl->mutex_.Lock();  
    7.   VersionEdit edit;  
    8.   // 如果存在表数据，则Load表数据，并对日志进行恢复，否则，创建新表  
    9.   Status s = impl->Recover(&edit);  
    10.   if (s.ok()) {  
    11.     // 从VersionEdit获取一个新的文件序号，所以如果是新建数据表，则第一个LOG的序号为2（1已经被MANIFEST占用）  
    12.     uint64_t new_log_number = impl->versions_->NewFileNumber();  
    13.     // 记录日志文件号，创建新的log文件及Writer对象  
    14.     WritableFile* lfile;  
    15.     s = options.env->NewWritableFile(LogFileName(dbname, new_log_number),  
    16.                                      &lfile);  
    17.     if (s.ok()) {  
    18.       edit.SetLogNumber(new_log_number);  
    19.       impl->logfile_ = lfile;  
    20.       impl->logfile_number_ = new_log_number;  
    21.       impl->log_ = new log::Writer(lfile);  
    22.       // 如果存在原来的log，则回放log  
    23.       s = impl->versions_->LogAndApply(&edit, &impl->mutex_);  
    24.     }  
    25.     if (s.ok()) {  
    26.       // 删除废弃的文件（如果存在）  
    27.       impl->DeleteObsoleteFiles();  
    28.       // 检查是否需要Compaction，如果需要，则让后台启动Compaction线程  
    29.       impl->MaybeScheduleCompaction();  
    30.     }  
    31.   }  
    32.   impl->mutex_.Unlock();  
    33.   if (s.ok()) {  
    34.     *dbptr = impl;  
    35.   } else {  
    36.     delete impl;  
    37.   }  
    38.   return s;  
    39. }  
    从上面可以看出，其实到底是新建表还是打开表都是取决与DBImpl::Recover()这个方法的行为，它的流程如下：
    [cpp] view plain copy
    1. Status DBImpl::Recover(VersionEdit* edit) {  
    2.   mutex_.AssertHeld();  
    3.   
    4.   // 创建DB目录，不关注错误  
    5.   env_->CreateDir(dbname_);  
    6.   // 在DB目录下打开或创建（如果不存在）LOCK文件并锁定它，防止其他进程打开此表  
    7.   Status s = env_->LockFile(LockFileName(dbname_), &db_lock_);  
    8.   if (!s.ok()) {  
    9.     return s;  
    10.   }  
    11.     
    12.   if (!env_->FileExists(CurrentFileName(dbname_))) {  
    13.     // 如果DB目录下不存在CURRENT文件且允许在表不存在时创建表，则新建一个表返回  
    14.     if (options_.create_if_missing) {  
    15.       s = NewDB();  
    16.       if (!s.ok()) {  
    17.         return s;  
    18.       }  
    19.     } else {  
    20.       return Status::InvalidArgument(  
    21.           dbname_, "does not exist (create_if_missing is false)");  
    22.     }  
    23.   } else {  
    24.     if (options_.error_if_exists) {  
    25.       return Status::InvalidArgument(  
    26.           dbname_, "exists (error_if_exists is true)");  
    27.     }  
    28.   }  
    29.   // 如果运行到此，表明表已经存在，需要load，第一步是从MANIFEST文件中恢复VersionSet  
    30.   s = versions_->Recover();  
    31.   if (s.ok()) {  
    32.     SequenceNumber max_sequence(0);  
    33.     // 获取MANIFEST中获取最后一次持久化清单时在使用LOG文件序号，注意：这个LOG当时正在使用，  
    34.     // 表明数据还在memtable中，没有形成sst文件，所以数据恢复需要从这个LOG文件开始(包含这个LOG)。  
    35.     // 另外，prev_log是早前版本level_db使用的机制，现在以及不再使用，这里只是为了兼容  
    36.     const uint64_t min_log = versions_->LogNumber();  
    37.     const uint64_t prev_log = versions_->PrevLogNumber();  
    38.     // 扫描DB目录，记录下所有比MANIFEST中记录的LOG更加新的LOG文件  
    39.     std::vector<std::string> filenames;  
    40.     s = env_->GetChildren(dbname_, &filenames);  
    41.     if (!s.ok()) {  
    42.       return s;  
    43.     }  
    44.     uint64_t number;  
    45.     FileType type;  
    46.     std::vector<uint64_t> logs;  
    47.     for (size_t i = 0; i < filenames.size(); i++) {  
    48.       if (ParseFileName(filenames[i], &number, &type)  
    49.           && type == kLogFile  
    50.           && ((number >= min_log) || (number == prev_log))) {  
    51.         logs.push_back(number);  
    52.       }  
    53.     }  
    54.     // 将LOG文件安装从小到大排序  
    55.     std::sort(logs.begin(), logs.end());  
    56.     // 逐个LOG文件回放    for (size_t i = 0; i < logs.size(); i++) {  
    57.       // 回放LOG时，记录被插入到memtable，如果超过write buffer，则还会dump出level 0的sst文件，  
    58.       // 此方法会将日志种每条记录的sequence num与max_sequence进行比较，以记录下最大的sequence num。  
    59.       s = RecoverLogFile(logs[i], edit, &max_sequence);  
    60.       // 更新最大的文件序号，因为MANIFEST文件中没有记录这些LOG文件占用的序号；  
    61.       // 当然，也可能LOG的序号小于MANIFEST中记录的最大文件序号，这时不需要更新。  
    62.       versions_->MarkFileNumberUsed(logs[i]);  
    63.     }  
    64.     if (s.ok()) {  
    65.       // 比较日志回放前后的最大sequence num，如果回放记录中有超过LastSequence()的记录，则替换  
    66.       if (versions_->LastSequence() < max_sequence) {  
    67.         versions_->SetLastSequence(max_sequence);  
    68.       }  
    69.     }  
    70.   }  
    71.   return s;  
    72. }     

    4.3 关闭一个已打开的表

    Level DB设计成只要删除DB对象就可以关闭表，其关键流程如下：
    [cpp] view plain copy
    1. DBImpl::~DBImpl() {  
    2.   // 通知后台线程，DB即将关闭  
    3.   mutex_.Lock();  
    4.   // 后台线程会间歇性地检查shutting_down_对象的指针，一旦不为NULL就会退出  
    5.   shutting_down_.Release_Store(this);  
    6.   // 注意:这里必须循环通知，直至compaction线程获得信号并设置了bg_compaction_scheduled_为false    
    7.   while (bg_compaction_scheduled_) {  
    8.     bg_cv_.Wait();  
    9.   }  
    10.   mutex_.Unlock();  
    11.   
    12.   // 如果锁定了文件锁，则释放文件锁  
    13.   if (db_lock_ != NULL) {  
    14.     env_->UnlockFile(db_lock_);  
    15.   }  
    16.   
    17.   delete versions_;  
    18.   // 减去memtable的引用计数  
    19.   if (mem_ != NULL) mem_->Unref();  
    20.   if (imm_ != NULL) imm_->Unref();  
    21.   // 销毁db log相关对象以及表缓冲对象  
    22.   delete log_;  
    23.   delete logfile_;  
    24.   delete table_cache_;  
    25.   
    26.   // 如果info log与cache是引擎自己构建，则需要销毁它们  
    27.   if (owns_info_log_) {  
    28.     delete options_.info_log;  
    29.   }  
    30.   if (owns_cache_) {  
    31.     delete options_.block_cache;  
    32.   }  
    33. }  
    由上可见，delete一个db对象可能会阻塞调用线程一段时间，必须让其完成一些必须完成的工作，才能进一步保障数据的安全。

    4.4 随机查询

    Level DB可能dump多个sst文件，这些文件的key范围可能重叠。按照Level DB的设计，其会将sst分为7个等级，可以视为代龄，其中，只有Level 0中的sst可能存在key的区间重叠的情况，而level1 - level6中，同一level中的sst可以保证不重叠，但不同level之间的sst依然可能key重叠。因此，如果查询一个key，其最多可能在6+n个sst中同时存在，n为level0中sst的个数；同时，由于这些文件的生成有先后关系，查询时还需要注意顺序，Get一个key的流程如下：




    文章来源：http://blog.csdn.net/icefireelf/article/details/7515816





沪ICP备19023445号-2号
友情链接