LevelDB作为一个Key-Value的NoSQL数据库，其最基本的操作就是Put，即插入一对<key, value>记录, 本文将以源码走读的方式解析数据记录插入数据库的基本流程
- 如下代码所示是Put的函数实现，其和delete操作共用putRec函数。
func (db *DB) Put(key, value []byte, wo *opt.WriteOptions) error {
return db.putRec(keyTypeVal, key, value, wo)
}
2.putRec函数是实际的Put插入实现函数单看putRec函数会发现该函数非常难理解。在这里作者使用writeMergeC、writeMergedC、writeAckC和writeLockC共同控制多线程的数据插入和合并操作。这四个channel的定义如下，注意其中writeLockC的channel可以缓存一个对象。
writeMergeC: make(chan writeMerge),
writeMergedC: make(chan bool),
writeLockC: make(chan struct{}, 1),
writeAckC: make(chan error),
其中writeMergeC的channel可以携带插入的数据，其工作原理如下图所示， 假设同时有c1,c2和c3的线程尝试插入数据，这时他们会在writeMergeC和writeLockC上竞争，由于其定义的不同。第一个进入的线程必然会获取writeLockC,也即获取写锁我们假设是图中的C1。此时C1线程会将<key,value>数据线写入batch然后执行writeLock函数。而c2和c3则继续竞争，由于此时写锁被writeLockC独占。c2和c3只能竞争writeMergeC的写入权限。
3.在writeLocked函数内部，会对写入记录执行merge操作，也就是会从writeMergeC中读取当前堵塞的写入并在空间够用的情况下merge成一次写操作。我们假设c2和c3依次获取写入writeMergeC的操作，则其数据最终会形成图中Step2的状态。此时c2 和c3堵塞在如下代码处。
if <-db.writeMergedC {
return <-db.writeAckC
}
阶段1和阶段2.png
- putRec函数解析
func (db *DB) putRec(kt keyType, key, value []byte, wo *opt.WriteOptions) error {
if err := db.ok(); err != nil {
return err
}
//merge 和sync 以数据库的初始化配置为主
merge := !wo.GetNoWriteMerge() && !db.s.o.GetNoWriteMerge()
sync := wo.GetSync() && !db.s.o.GetNoSync()
// Acquire write lock
if merge {
select {
case db.writeMergeC <- writeMerge{sync: sync, keyType: kt, key: key, value: value}:
//如果能向writeMergeC 写入新插入的key value 数据
//则等待新的key value与老的数据进行merge操作
if <-db.writeMergedC { //等待writeMerge的结果，如果merge失败则继续获取写锁
//merge成功等待merge之后的数据写入结果
// Write is merged.
return <-db.writeAckC
}
// Write is not merged, the write lock is handed to us. Continue.
case db.writeLockC <- struct{}{}: //尝试获取写锁
// Write lock acquired.
case err := <-db.compPerErrC:
// Compaction error.
return err
case <-db.closeC:
// Closed
return ErrClosed
}
} else {
//没有merge的情况直接尝试获取写入锁
select {
case db.writeLockC <- struct{}{}:
// Write lock acquired.
case err := <-db.compPerErrC:
// Compaction error.
return err
case <-db.closeC:
// Closed
return ErrClosed
}
}
batch := db.batchPool.Get().(*Batch)
batch.Reset()
batch.appendRec(kt, key, value)
return db.writeLocked(batch, batch, merge, sync)
}
4.writeLocked 函数解析
writeLocked函数是c1的真正执行写入的函数，其写入的主要流程包括：
- 获取内存数据库memDB,如果空间不足则扩容；
- 如果任由merge空间和数据，执行merge逻辑；
- 写日志信息；
- 数据写入内存；
- 释放相关所，以及通知给等待线程执行结果（如图中c2和c3）
func (db *DB) writeLocked(batch, ourBatch *Batch, merge, sync bool) error {
// Try to flush memdb. This method would also trying to throttle writes
// if it is too fast and compaction cannot catch-up.
//1.尝试flush db的数据 如果有需要
// 返回DB的mdb以及mdb的剩余空间，如果mdbFree不够则会对mdb进行扩容操作
mdb, mdbFree, err := db.flush(batch.internalLen)
if err != nil {
db.unlockWrite(false, 0, err)
return err
}
defer mdb.decref() //释放当前引用数量
var (
overflow bool
merged int
batches = []*Batch{batch}
)
if merge { // 需要merge的情况
// Merge limit.
var mergeLimit int
//控制merge的数量不是特别大
if batch.internalLen > 128<<10 {
mergeLimit = (1 << 20) - batch.internalLen
} else {
mergeLimit = 128 << 10
}
mergeCap := mdbFree - batch.internalLen
if mergeLimit > mergeCap {
mergeLimit = mergeCap
}
//控制最大能够merge的量
merge:
for mergeLimit > 0 {
select {
case incoming := <-db.writeMergeC:
if incoming.batch != nil { //writeMergeC 中存储的是batch的情况
// Merge batch.
if incoming.batch.internalLen > mergeLimit {
overflow = true
break merge
}
batches = append(batches, incoming.batch)
mergeLimit -= incoming.batch.internalLen
} else {
// Merge put.
internalLen := len(incoming.key) + len(incoming.value) + 8
if internalLen > mergeLimit {
overflow = true
break merge
}
if ourBatch == nil {
ourBatch = db.batchPool.Get().(*Batch)
ourBatch.Reset()
batches = append(batches, ourBatch)
}
// We can use same batch since concurrent write doesn't
// guarantee write order.
ourBatch.appendRec(incoming.keyType, incoming.key, incoming.value)
mergeLimit -= internalLen
}
sync = sync || incoming.sync // 同步的情况需要通知写入的等待线程写入完毕
merged++
db.writeMergedC <- true
default:
break merge
}
}
}
// Seq number.
seq := db.seq + 1 //seq是实际batch的数量编号, 此时db的实际seq并未更新
// Write journal.
// 2. batch 信息写入日志
if err := db.writeJournal(batches, seq, sync); err != nil {
db.unlockWrite(overflow, merged, err)
return err
}
// Put batches.
// 3. batch 数据写入内存数据库 mendb
for _, batch := range batches {
if err := batch.putMem(seq, mdb.DB); err != nil {
panic(err)
}
seq += uint64(batch.Len())
}
// Incr seq number.
db.addSeq(uint64(batchesLen(batches))) //更新db的seq
// Rotate memdb if it's reach the threshold.
if batch.internalLen >= mdbFree { //防止下次不够？
db.rotateMem(0, false)
}
db.unlockWrite(overflow, merged, nil)
return nil
}
