而在RocksDB 5.5中引进了一个选项enable_pipelined_write，这个选项的目的就是将WAL和MemTable的写入pipeline化， 也就是说当一个线程写完毕WAL之后，此时在WAL的write队列中等待的其他的write则会开始继续写入WAL, 而当前线程将会继续 写入MemTable.此时就将不同的Writer的写入WAL和写入MemTable并发执行了.

实现

我们这里只来分析pipeline的实现，核心函数就是DBImpl::PipelinedWriteImpl.

  • 每一个DB(DBImpl)都有一个write_thread_(class WriteThread).

  • 每次调用Write的时候会先写入WAL, 此时新建一个WriteThread::Writer对象，并将这个对象加入到一个Group中(调用JoinBatchGroup)

  • 然后我们来看JoinBatchGroup,这个函数主要是用来讲所有的写入WAL加入到一个Group中.这里可以看到当当前的Writer 对象是leader(比如第一个进入的对象)的时候将会直接返回，否则将会等待知道更新为对应的状态．

    2. ...................................
    3. bool linked_as_leader = LinkOne(w, &newest_writer_);
    4. if (linked_as_leader) {
    5.   SetState(w, STATE_GROUP_LEADER);
    6. }
    8. TEST_SYNC_POINT_CALLBACK("WriteThread::JoinBatchGroup:Wait", w);
    10. if (!linked_as_leader) {
    11.   /**
    12.    * Wait util:
    13.    * 1) An existing leader pick us as the new leader when it finishes
    14.    * 2) An existing leader pick us as its follewer and
    15.    * 2.1) finishes the memtable writes on our behalf
    16.    * 2.2) Or tell us to finish the memtable writes in pralallel
    17.    * 3) (pipelined write) An existing leader pick us as its follower and
    18.    *    finish book-keeping and WAL write for us, enqueue us as pending
    19.    *    memtable writer, and
    20.    * 3.1) we become memtable writer group leader, or
    21.    * 3.2) an existing memtable writer group leader tell us to finish memtable
    22.    *      writes in parallel.
    23.    */
    24.   AwaitState(w, STATE_GROUP_LEADER | STATE_MEMTABLE_WRITER_LEADER |
    25.                     STATE_PARALLEL_MEMTABLE_WRITER | STATE_COMPLETED,
    26.              &jbg_ctx);
    27.   TEST_SYNC_POINT_CALLBACK("WriteThread::JoinBatchGroup:DoneWaiting", w);
    28. }
    29. }
    1. bool WriteThread::LinkOne(Writer* w, std::atomic<Writer*>* newest_writer) {
    2. assert(newest_writer != nullptr);
    3. assert(w->state == STATE_INIT);
    4. Writer* writers = newest_writer->load(std::memory_order_relaxed);
    5.   w->link_older = writers;
    7.     return (writers == nullptr);
    8.   }
    9. }
    10. }

  • 当从JoinBatchGroup返回之后，当当前的Writer对象为leader的话，则将会把此leader下的所有的write都 链接到一个WriteGroup中(调用EnterAsBatchGroupLeader函数),　并开始写入WAL,这里要注意非leader的write将会直接 进入memtable的写入，这是因为非leader的write都将会被当前它所从属的leader来打包(group)写入，后面我们会看到实现.

  • 这里注意到遍历是通过link_newer进行的，之所以这样做是相当于在写入WAL之前，对于当前leader的Write 做一次snapshot(通过CreateMissingNewerLinks函数).

    1. void WriteThread::CreateMissingNewerLinks(Writer* head) {
    2. while (true) {
    3.   Writer* next = head->link_older;
    4.   if (next == nullptr || next->link_newer != nullptr) {
    5.     assert(next == nullptr || next->link_newer == head);
    6.     break;
    7.   }
    8.   next->link_newer = head;
    9.   head = next;
    10. }
    11. }

  • 上述操作进行完毕之后，进入写WAL操作,最终会把这个write_group打包成一个writeBatch(通过MergeBatch函数)进行写入.

  1.    if (w.ShouldWriteToWAL()) {
  2. ...............................
  3.       w.status = WriteToWAL(wal_write_group, log_writer, log_used,
  4.                             need_log_sync, need_log_dir_sync, current_sequence);
  5.     }

  • 当当前的leader将它自己与它的follow写入之后，此时它将需要写入memtable,那么此时之前还阻塞的Writer，分为两种情况 第一种是已经被当前的leader打包写入到WAL，这些writer(包括leader自己)需要将他们链接到memtable writer list.还有一种情况，那就是还没有写入WAL的，此时这类writer则需要选择一个leader然后继续写入WAL.

    1. void WriteThread::LaunchParallelMemTableWriters(WriteGroup* write_group) {
    2. assert(write_group != nullptr);
    3. write_group->running.store(write_group->size);
    4. for (auto w : *write_group) {
    5.   SetState(w, STATE_PARALLEL_MEMTABLE_WRITER);
    6. }
    7. }

  • 这里要注意，在构造memtable的group的时候，我们不需要创建link_newer，因为之前在写入WAL的时候，我们已经构造好link_newer,那么此时我们使用构造好的group也就是表示这个group中包含的都是已经写入到WAL的操作.

    1. void WriteThread::EnterAsMemTableWriter(Writer* leader,
    2.                                       WriteGroup* write_group) {
    3. ....................................
    5.  ....................................................
    6. }
    8. write_group->last_writer = last_writer;
    9. write_group->last_sequence =
    10.     last_writer->sequence + WriteBatchInternal::Count(last_writer->batch) - 1;
    11. }

  • 最后开始执行写入MemTable的操作，之前在写入WAL的时候被阻塞的所有Writer此时都会进入下面这个逻辑，此时也就意味着 并发写入MemTable．

  • 最后当当前group的所有Writer都写入MemTable之后，则将会调用ExitAsMemTableWriter来进行收尾工作.如果有新的memtable writer list需要处理，那么则唤醒对应的Writer,然后设置已经处理完毕的Writer的状态.

    1. void WriteThread::ExitAsMemTableWriter(Writer* /*self*/,
    2.                                      WriteGroup& write_group) {
    3. Writer* leader = write_group.leader;
    4. Writer* last_writer = write_group.last_writer;
    6. Writer* newest_writer = last_writer;
    7. if (!newest_memtable_writer_.compare_exchange_strong(newest_writer,
    8.                                                      nullptr)) {
    9.   CreateMissingNewerLinks(newest_writer);
    10.   Writer* next_leader = last_writer->link_newer;
    11.   assert(next_leader != nullptr);
    12.   next_leader->link_older = nullptr;
    13.   SetState(next_leader, STATE_MEMTABLE_WRITER_LEADER);
    14. }
    15. Writer* w = leader;
    16. while (true) {
    17.   if (!write_group.status.ok()) {
    18.     w->status = write_group.status;
    19.   }
    20.   Writer* next = w->link_newer;
    21.   if (w != leader) {
    22.     SetState(w, STATE_COMPLETED);
    23.   }
    24.   if (w == last_writer) {
    25.     break;
    26.   }
    27.   w = next;
    28. }
    29. // Note that leader has to exit last, since it owns the write group.
    30. SetState(leader, STATE_COMPLETED);

总结

我们可以看到在RocksDB中，WAL的写入始终是串行写入，而MemTable可以多线程并发写入，也就是说在系统压力到一定阶段的时候， 写入WAL肯定会成为瓶颈.