0.导读

LevelPut的流程：

Put操作首先将操作记录写入log文件，然后写入memtable，返回写成功。整体来看是这样，但是会引发下面的问题：

1. 写log的时候是实时刷到磁盘的吗？ 2. 写入的时候memtable过大了咋办？ 3. 同时多个线程并发写咋办？ ......

下面的分析中就会面对这些问题，有些答案很清晰，有些涉及到超级多细节。

step by step

在开始之前，我们知道Write操作是要记录到log文件中的。那么一个记录它的格式是怎样的呢？看图：

这里特别解释一下，Delete操作也是通过Put实现的，只是图中的类型字段是0，而正常Write的操作类型是1，由此区分写操作和删除操作！

熟悉了LevelDB整个脉络之后, Put方法是相当简单的, 一章就可以解决. 数据删除和写入是一个概念, 删除就是写入特殊deletion marker; 批量(batch)和单条写入也是一个概念, 单条写入就是只有一条记录的batch. 整个流程很短, 基本上写个log就好了, 因此速度很快.

step 1

Put interface

很简单吧。这里讲解一下那三个参数：

1. WriteOptions：提供一些写操作的配置项，例如要不要写log的时候马上flush磁盘
2. key和value就是对应的keyValue，slice只是作者自己封装的char数组存储数据而已。<b>大牛喜欢把所有东西都封装一下，赋予数据结构意义！</b>这在大的工程里面是很有意义的，既方便操作，也方便思考（这样就不用思考底层的真实的char数组还是啥）。

Status DB::Put(const WriteOptions& opt, const Slice& key, const Slice& value) {WriteBatch batch;        //leveldb中不管单个插入还是多个插入都是以WriteBatch的方式进行的batch.Put(key, value);return Write(opt, &batch);
}

 1 Status DBImpl::Write(const WriteOptions& options, WriteBatch* my_batch) {
 2   // A begin
 3   Writer w(&mutex_);
 4   w.batch = my_batch;
 5   w.sync = options.sync;
 6   w.done = false;
 7   // A end
 8
 9   // B begin/* mutex l上锁之后, 到了"w.cv.Wait()"的时候, 会先释放锁等待, 然后收到signal时再次上锁. 这段代码的作用就是多线程在提交任务的时候, 一个接一个push_back进队列. 但只有位于队首的线程有资格继续运行下去. 目的是把多个写请求合并成一个大batch提升效率.*/
10   MutexLock l(&mutex_);
11   writers_.push_back(&w);
12   while (!w.done && &w != writers_.front()) {
13     w.cv.Wait();
14   }
15   if (w.done) {
16     return w.status;
17   }
18   // B end
19
20   // May temporarily unlock and wait.
21   Status status = MakeRoomForWrite(my_batch == NULL);
22   uint64_t last_sequence = versions_->LastSequence();
23   Writer* last_writer = &w;
24   if (status.ok() && my_batch != NULL) {  // NULL batch is for compactions
25     WriteBatch* updates = BuildBatchGroup(&last_writer);
26     WriteBatchInternal::SetSequence(updates, last_sequence + 1);
27     last_sequence += WriteBatchInternal::Count(updates);
28
29     // Add to log and apply to memtable.  We can release the lock
30     // during this phase since &w is currently responsible for logging
31     // and protects against concurrent loggers and concurrent writes
32     // into mem_.
33     {
34       mutex_.Unlock();
35       status = log_->AddRecord(WriteBatchInternal::Contents(updates));
36       bool sync_error = false;
37       if (status.ok() && options.sync) {
38         status = logfile_->Sync();
39         if (!status.ok()) {
40           sync_error = true;
41         }
42       }
43       if (status.ok()) {
44         status = WriteBatchInternal::InsertInto(updates, mem_);
45       }
46       mutex_.Lock();
47       if (sync_error) {
48         // The state of the log file is indeterminate: the log record we
49         // just added may or may not show up when the DB is re-opened.
50         // So we force the DB into a mode where all future writes fail.
51         RecordBackgroundError(status);
52       }
53     }
54     if (updates == tmp_batch_) tmp_batch_->Clear();
55
56     versions_->SetLastSequence(last_sequence);
57   }
58
59   while (true) {
60     Writer* ready = writers_.front();
61     writers_.pop_front();
62     if (ready != &w) {
63       ready->status = status;
64       ready->done = true;
65       ready->cv.Signal();
66     }
67     if (ready == last_writer) break;
68   }
69
70   // Notify new head of write queue
71   if (!writers_.empty()) {
72     writers_.front()->cv.Signal();
73   }
74
75   return status;
76 }