「leveldb」LevelDB compact 过程

leveldb

leveldb的一个重要特性就是数据的分层，由于数据的分层，越旧的数据处在越大的层级，越新的数据在越小的层级。

在查询数据的时候，最先读取MemTable里面的数据，然后是L0的SSTable里面，接着是L1， L2直到最大的层级。在分层设计中，越往上层，数据的容量越大，大约Ln是Ln-1层数据的10倍。在各个层级的SSTable文件，只有L0层的数据是有MemTable直接flush到磁盘上，其它层的数据是经过compaction过程进行排序整理产生的。这意味着L0层以上的数据，各个SSTable文件内的数据是有序且不会重叠的。

因此， compaction的过程是产生SSTable的过程，分为2中情况：

由MemTable到SSTable的flush过程，也被成为Minor Compaction；
由Ln层的SSTable到Ln+1层的SSTable的数据重排过程，也被称为Major Compaction；

Minor Compaction

LevelDB设定了L0的容量，以及触发L0 compaction的条件：

// Level-0 compaction is started when we hit this many files.
static const int kL0_CompactionTrigger = 4;
      
// Soft limit on number of level-0 files.  We slow down writes at this point.
static const int kL0_SlowdownWritesTrigger = 8;                                                                                                                                              
      
// Maximum number of level-0 files.  We stop writes at this point.
static const int kL0_StopWritesTrigger = 12;

默认情况下， L0的SSTable的文件个数是4个，大于4个就可能开始compaction，当L0文件的个数大于8个，应用层的写入速度会降下来，以避免L0文件数太多，当L0文件数大于12个，前端的写入停止。

在LevelDB中，不管是Put，还是Delete，都是调用Write来进行写入，在写入之前，都会调用MakeRoomForWrite来判断如何进行处理，是直接写入还是slowdown，还是送至前端写入？

其实现过程如下图：

写入控制过程

从上图可以看到， Minor Compaction发生在前面4个判断失败之后，它会把当前正在用作写入的MemTable转换为一个只读的内存数据，同时产生一个新的MemTable以及与其对应的log文件，后续的新的写入都转移到新产生的MemTable文件内。同时，它会产生一个Compact Schedule，来触发后台的线程来将im-memTable

写入磁盘。

Major Compaction

LevelDB 在数据库启动的时候，会指定Env，比如linux 系统下会产生一个PosixEnv对象，它定义了一个schedule函数来构成函数队列，所有的后端执行的函数，会放进该队列里面来一个一个地执行:

void PosixEnv::Schedule(void (*function)(void*), void* arg) {
  PthreadCall("lock", pthread_mutex_lock(&mu_));

  // Start background thread if necessary
  if (!started_bgthread_) {
    started_bgthread_ = true;
    PthreadCall(
        "create thread",
        pthread_create(&bgthread_, NULL,  &PosixEnv::BGThreadWrAPPer, this));                                                                                                                
  }

  // If the queue is currently empty, the background thread may currently be
  // waiting.
  if (queue_.empty()) {
    PthreadCall("signal", pthread_cond_signal(&bgsignal_));
  }

  // add to priority queue
  queue_.push_back(BGItem());
  queue_.back().function = function;
  queue_.back().arg = arg;

  PthreadCall("unlock", pthread_mutex_unlock(&mu_));
}

// PosixEnv::BGThreadwrapper包装了PosixEnv::BGThread() 函数
// 该函数等待加入队列的函数， 逐个处理
void PosixEnv::BGThread() {                                                                                                                                                                  
  while (true) {
    // Wait until there is an item that is ready to run
    PthreadCall("lock", pthread_mutex_lock(&mu_));
    while (queue_.empty()) {
      PthreadCall("wait", pthread_cond_wait(&bgsignal_, &mu_));
    }

    void (*function)(void*) = queue_.front().function;
    void* arg = queue_.front().arg;
    queue_.pop_front();

    PthreadCall("unlock", pthread_mutex_unlock(&mu_));
    (*function)(arg);  、、 根据注册的函数执行callback函数
  }
}

在这里， compaction指定的处理函数是DBImpl::BGWork（），经过DBImpl::BackgroundCall()，最终由DBImpl::BackgroundCompaction()来进行compaction的任务。

void DBImpl::BackgroundCompaction() {
    ...
    if (is_manual) {
    }
    else {
          c = versions_->PickCompaction();
    }
    CompactionState* compact = new CompactionState(c);
   status = DoCompactionWork(compact);
   if (!status.ok()) {
     RecordBackgroundERROR(status);
   }
   CleanupCompaction(compact);
   c->ReleaseInputs();
   DeleteObsoleteFiles();
}

可以看到，通过VersionSet::PickCompaction()来搜集本次compaction的input集合，LevelDB中，Compaction操作有两种触发方式：

某一level的文件数太多
某一文件的查找次数超过允许值；

因此，在VersionSet::PickCompaction()中便有了代码中的size_compaction和seek_compaction的判断。在进行合并时，将优先考虑文件数过多的情况。

最后有DBImplement::DoCompactionWork()完成input集合数据的遍历以及写入新的SSTable文件里面。

如下我们给出一个简化版的compaction过程：

在这里插入图片描述

文章最后发布于: 2018-11-18 16:54:45

LevelDB compact 过程

leveldb

Minor Compaction

Major Compaction

相关阅读

栏目导航

推荐阅读

热门阅读