diff --git a/leveldb/batch.go b/leveldb/batch.go index d5ecf721..87dc278c 100644 --- a/leveldb/batch.go +++ b/leveldb/batch.go @@ -372,6 +372,7 @@ func batchesLen(batches []*Batch) int { } func writeBatchesWithHeader(wr io.Writer, batches []*Batch, seq uint64) error { + // 实现是 singleWriter.Write if _, err := wr.Write(encodeBatchHeader(nil, seq, batchesLen(batches))); err != nil { return err } diff --git a/leveldb/cache/cache.go b/leveldb/cache/cache.go index 8e4f397c..2c7f1682 100644 --- a/leveldb/cache/cache.go +++ b/leveldb/cache/cache.go @@ -81,6 +81,7 @@ func (x mNodes) Swap(i, j int) { x[i], x[j] = x[j], x[i] } func (x mNodes) sort() { sort.Sort(x) } +// 找第一个大于等于 的index func (x mNodes) search(ns, key uint64) int { return sort.Search(len(x), func(i int) bool { a := x[i].ns @@ -155,7 +156,8 @@ func (b *mBucket) get(r *Cache, h *mHead, hash uint32, ns, key uint64, getOnly b if i == len(b.nodes) { b.nodes = append(b.nodes, n) } else { - b.nodes = append(b.nodes[:i+1], b.nodes[i:]...) + // 把新的 node 插入到 i 的前面 + b.nodes = append(b.nodes[:i+1], b.nodes[i:]...) // 添加到有序数组的对应位置，插入排序，所以 b.nodes 不可以过长 b.nodes[i] = n } bLen := len(b.nodes) @@ -188,6 +190,7 @@ func (b *mBucket) get(r *Cache, h *mHead, hash uint32, ns, key uint64, getOnly b return true, true, n } +// delete 的返回值约定值得参考 func (b *mBucket) delete(r *Cache, h *mHead, hash uint32, ns, key uint64) (done, deleted bool) { b.mu.Lock() @@ -260,9 +263,12 @@ func (b *mBucket) delete(r *Cache, h *mHead, hash uint32, ns, key uint64) (done, } type mHead struct { - buckets []mBucket - mask uint32 - predecessor unsafe.Pointer // *mNode + buckets []mBucket + mask uint32 // 用当前的 mask 和 predecessor 的 mask 作对比，可以知道是在 grow 还是在 shrink + + // 下面注释应该错了， 应该是 *mHead 吧 + predecessor unsafe.Pointer // *mNode + resizeInProgress int32 overflow int32 @@ -301,7 +307,7 @@ func (h *mHead) initBucket(i uint32) *mBucket { nodes = make(mNodes, 0, len(m0)+len(m1)) nodes = append(nodes, m0...) nodes = append(nodes, m1...) - nodes.sort() + nodes.sort() // 可以用 merge sort？不过 m0, m1 的 size 比较小，也无所谓 } b.nodes = nodes b.state = bucketInitialized diff --git a/leveldb/comparer.go b/leveldb/comparer.go index 448402b8..710938f4 100644 --- a/leveldb/comparer.go +++ b/leveldb/comparer.go @@ -10,6 +10,8 @@ import ( "github.com/syndtr/goleveldb/leveldb/comparer" ) +// iComparer 处理关于 sequence 相关的逻辑 +// 感觉可以称之为 internal Comparer? type iComparer struct { ucmp comparer.Comparer } @@ -34,9 +36,12 @@ func (icmp *iComparer) Name() string { return icmp.uName() } +// func (icmp *iComparer) Compare(a, b []byte) int { x := icmp.uCompare(internalKey(a).ukey(), internalKey(b).ukey()) if x == 0 { + // 按照 seq 排序，seq 大的排在前面 + // uCompare 指的是 user 指定的 compare function if m, n := internalKey(a).num(), internalKey(b).num(); m > n { return -1 } else if m < n { diff --git a/leveldb/comparer/bytes_comparer.go b/leveldb/comparer/bytes_comparer.go index abf9fb65..2caf8756 100644 --- a/leveldb/comparer/bytes_comparer.go +++ b/leveldb/comparer/bytes_comparer.go @@ -6,7 +6,9 @@ package comparer -import "bytes" +import ( + "bytes" +) type bytesComparer struct{} diff --git a/leveldb/comparer/comparer.go b/leveldb/comparer/comparer.go index 2c522db2..e68f274d 100644 --- a/leveldb/comparer/comparer.go +++ b/leveldb/comparer/comparer.go @@ -45,13 +45,21 @@ type Comparer interface { // // Either contents of a or b should not by any means modified. Doing so // may cause corruption on the internal state. + // + // 以 LevelDB 提供的 bytes comparer，a = {0xff, 0xff, 0x11}, b = {0xff, 0xff, 0x1A, ...} 来讲， + // 前提是 b > a，否则返回 nil + // Separator(dst, a, b) 的结果是：dst = dst + {0xff, 0xff, 0x12} + // “分隔”体现在 dst 的最后一个 0x12 把 a 和 b 给分隔开了 Separator(dst, a, b []byte) []byte // Successor appends a sequence of bytes x to dst such that x >= b, where // 'less than' is consistent with Compare. An implementation should return // nil if x equal to b. - // // Contents of b should not by any means modified. Doing so may cause // corruption on the internal state. + // + // 以 LevelDB 提供的 bytes comparer，b = {0xff, 0xff, 0x1A, ...} 来讲， Successor(dst, b) 的结果是： + // dst = dst + {0xff, 0xff, 0x1B}，注意结尾是比 0x1A 大了 1 的 0x1B + // 如果 b 里面都是 0xff，会返回 nil Successor(dst, b []byte) []byte } diff --git a/leveldb/db.go b/leveldb/db.go index b2724cd9..95726728 100644 --- a/leveldb/db.go +++ b/leveldb/db.go @@ -52,11 +52,11 @@ type DB struct { memMu sync.RWMutex memPool chan *memdb.DB mem, frozenMem *memDB - journal *journal.Writer + journal *journal.Writer // 写 WAL journalWriter storage.Writer journalFd storage.FileDesc frozenJournalFd storage.FileDesc - frozenSeq uint64 + frozenSeq uint64 // 哪一个 seq 的写操作导致了 memDB 的切换，或者说导致了当前的 memDB 被 freeze 成 immutable memDB // Snapshot. snapsMu sync.Mutex @@ -64,21 +64,21 @@ type DB struct { // Write. batchPool sync.Pool - writeMergeC chan writeMerge - writeMergedC chan bool - writeLockC chan struct{} - writeAckC chan error - writeDelay time.Duration - writeDelayN int + writeMergeC chan writeMerge // 有 merge write 需求时 chan<-，执行 merge write 的一方会不断地 <-chan 合并可以合并过来的写请求 + writeMergedC chan bool // chan<- false 用来通知某个 write 它没有被 merge + writeLockC chan struct{} // write 的写锁，成功 chan<- 说明拿到了锁，用完了之后 <-chan 就是释放锁 + writeAckC chan error // chan<- 通知等待 ACK 的一方；<-chan 等待的一方读取 Write 的结果 + writeDelay time.Duration // 记录 Write 被 compaction 所 delay 的时长 + writeDelayN int // 记录 Write 被 compaction 所 delay 的次数 tr *Transaction // Compaction. compCommitLk sync.Mutex tcompCmdC chan cCmd - tcompPauseC chan chan<- struct{} - mcompCmdC chan cCmd - compErrC chan error - compPerErrC chan error + tcompPauseC chan chan<- struct{} // 从 tcompPauseC 读取 <-chan 读到了一个 chan<- struct{} 到了之后要 pause compaction，直到可以写入 chan<- struct{} 才恢复 + mcompCmdC chan cCmd // 全称 memdb compaction command channel? + compErrC chan error // compaction error + compPerErrC chan error // 全称 compaction persistent error compErrSetC chan error compWriteLocking bool compStats cStats @@ -104,9 +104,9 @@ func openDB(s *session) (*DB, error) { snapsList: list.New(), // Write batchPool: sync.Pool{New: newBatch}, - writeMergeC: make(chan writeMerge), + writeMergeC: make(chan writeMerge), // 用来合并 Write 操作 writeMergedC: make(chan bool), - writeLockC: make(chan struct{}, 1), + writeLockC: make(chan struct{}, 1), // 注意：channel 的 buffer 为 1，因为首个去获取 Write Lock 的 goroutine 必须得能通过 chan<- 拿到 Lock writeAckC: make(chan error), // Compaction tcompCmdC: make(chan cCmd), @@ -778,6 +778,8 @@ func memGet(mdb *memdb.DB, ikey internalKey, icmp *iComparer) (ok bool, mv []byt return } +// 正常的查询 auxm 和 auxt 都为 nil +// 猜：auxm and auxt is only for testing? func (db *DB) get(auxm *memdb.DB, auxt tFiles, key []byte, seq uint64, ro *opt.ReadOptions) (value []byte, err error) { ikey := makeInternalKey(nil, key, seq, keyTypeSeek) @@ -794,6 +796,8 @@ func (db *DB) get(auxm *memdb.DB, auxt tFiles, key []byte, seq uint64, ro *opt.R } defer m.decref() + // 依次从 memtable 和 immutable memtable 获取 + // 如果找到的话，可以返回结果，因为找到的这个必然是 seq number 小于等于 seq 且是最大的那个，小的 seq number 会被 compaction if ok, mv, me := memGet(m.DB, ikey, db.s.icmp); ok { return append([]byte(nil), mv...), me } diff --git a/leveldb/db_compaction.go b/leveldb/db_compaction.go index cc275ace..8837e9ea 100644 --- a/leveldb/db_compaction.go +++ b/leveldb/db_compaction.go @@ -59,6 +59,7 @@ func (p *cStatStaging) stopTimer() { } } +// 按照 level 来组织 cStat，level[n] 表示 level-n 的 cStat，level 从 0 开始 type cStats struct { lk sync.Mutex stats []cStat @@ -173,6 +174,8 @@ func (db *DB) compactionTransact(name string, t compactionTransactInterface) { disableBackoff = db.s.o.GetDisableCompactionBackoff() ) + + // 有 backOff retry for n := 0; ; n++ { // Check whether the DB is closed. if db.isClosed() { @@ -192,6 +195,7 @@ func (db *DB) compactionTransact(name string, t compactionTransactInterface) { // Set compaction error status. select { case db.compErrSetC <- err: + // 注意：err == nil 时也要放入 compErrSetC case perr := <-db.compPerErrC: if err != nil { db.logf("%s exiting (persistent error %q)", name, perr) @@ -204,6 +208,8 @@ func (db *DB) compactionTransact(name string, t compactionTransactInterface) { if err == nil { return } + + // 如果是 corrupted error，是不可以重试解决的 if errors.IsCorrupted(err) { db.logf("%s exiting (corruption detected)", name) db.compactionExitTransact() @@ -266,6 +272,7 @@ func (db *DB) compactionCommit(name string, rec *sessionRecord) { }, nil) } +// 触发 immutable memtable 的 compaction func (db *DB) memCompaction() { mdb := db.getFrozenMem() if mdb == nil { @@ -279,6 +286,7 @@ func (db *DB) memCompaction() { if mdb.Len() == 0 { db.logf("memdb@flush skipping") // drop frozen memdb + // 此时 immutable memdb 是空的，所以不用 compaction 了，直接删除掉对应的 journal 文件即可 db.dropFrozenMem() return } @@ -304,11 +312,15 @@ func (db *DB) memCompaction() { db.compactionTransactFunc("memdb@flush", func(cnt *compactionTransactCounter) (err error) { stats.startTimer() flushLevel, err = db.s.flushMemdb(rec, mdb.DB, db.memdbMaxLevel) + + // 这里 rec 应该只有一条记录，就是当前 memdb compaction 出来的 SST + stats.stopTimer() return }, func() error { for _, r := range rec.addedTables { db.logf("memdb@flush revert @%d", r.num) + // 删除创建的 SST，完成 revert if err := db.s.stor.Remove(storage.FileDesc{Type: storage.TypeTable, Num: r.num}); err != nil { return err } @@ -321,6 +333,9 @@ func (db *DB) memCompaction() { // Commit. stats.startTimer() + + // 写入 new version 的 manifest 到磁盘，完成持久化，然后切换当前 stVersion 到 new version + // new version 是在当前 stVersion 上 apply 了 rec 中的 change 之后产生的 db.compactionCommit("memdb", rec) stats.stopTimer() @@ -346,6 +361,8 @@ func (db *DB) memCompaction() { } } + // 注意：在做 memtable compaction 的时候，major compaction 是被停止的，也就是说不存在并行的 compaction + // 在 memtable compaction 结束之后会 trigger 一次 major compaction // Trigger table compaction. db.compTrigger(db.tcompCmdC) } @@ -357,6 +374,7 @@ type tableCompactionBuilder struct { rec *sessionRecord stat1 *cStatStaging + // 暂存 snapshot 状态 snapHasLastUkey bool snapLastUkey []byte snapLastSeq uint64 @@ -367,7 +385,7 @@ type tableCompactionBuilder struct { kerrCnt int dropCnt int - minSeq uint64 + minSeq uint64 // compaction 时的最小 seq strict bool tableSize int @@ -426,6 +444,8 @@ func (b *tableCompactionBuilder) cleanup() error { return nil } +// compactionTransactInterface +// b 记录了 compaction 的状态 func (b *tableCompactionBuilder) run(cnt *compactionTransactCounter) (err error) { snapResumed := b.snapIter > 0 hasLastUkey := b.snapHasLastUkey // The key might has zero length, so this is necessary. @@ -473,9 +493,13 @@ func (b *tableCompactionBuilder) run(cnt *compactionTransactCounter) (err error) shouldStop := !resumed && b.c.shouldStopBefore(ikey) if !hasLastUkey || b.s.icmp.uCompare(lastUkey, ukey) != 0 { + // 做 iteration 的时候相同的 ukey 必然是连在一起的 + // 这时遇到了 ukey 的切换 // First occurrence of this user key. // Only rotate tables if ukey doesn't hop across. + // 相同的 ukey 不会写到两个 SSTable 中 + // 注意这里 b.flush 是以 uKey 切换了为前提的，也就是说 shouldStop 只在 ukey 切换时才有效，相同的 ukey 是不能 stop 的 if b.tw != nil && (shouldStop || b.needFlush()) { if err := b.flush(); err != nil { return err @@ -491,6 +515,7 @@ func (b *tableCompactionBuilder) run(cnt *compactionTransactCounter) (err error) b.snapDropCnt = b.dropCnt } + // ukey 成为新的 lastUkey hasLastUkey = true lastUkey = append(lastUkey[:0], ukey...) lastSeq = keyMaxSeq @@ -499,8 +524,12 @@ func (b *tableCompactionBuilder) run(cnt *compactionTransactCounter) (err error) switch { case lastSeq <= b.minSeq: // Dropped because newer entry for same user key exist + // 注意：这里的 fallthrough 会直接掉落到下一个 case，无论下一个 case 的条件是否满足！ + // 于是，ukey 的 seq 在 minSeq 之后的、非首个记录必然会 fallthrough 到下一个 case，就会被 drop 掉了 + // fallthrough 用的刚刚好！ fallthrough // (A) case kt == keyTypeDel && seq <= b.minSeq && b.c.baseLevelForKey(lastUkey): + // 对于 ukey 的 seq 在 minSeq 之后的、首个记录，且 keyType==del，更高层又没有这个 ukey 的，可以 drop // For this user key: // (1) there is no data in higher levels // (2) data in lower levels will have larger seq numbers @@ -510,11 +539,14 @@ func (b *tableCompactionBuilder) run(cnt *compactionTransactCounter) (err error) // Therefore this deletion marker is obsolete and can be dropped. lastSeq = seq b.dropCnt++ - continue + continue // 这条数据就不会被写到新的 SSTable 中了 default: + // 对于 ukey 的 seq > minSeq，即在 minSeq 之前的，会到 default，记录被写入 + // 对于 ukey 的 seq 在 minSeq 之后的、首个记录（因为必然有 lastSeq > b.minSeq，不会命中前两个 case），如果其 keyType==val，那么会到 default，记录会被写入 lastSeq = seq } } else { + // key error is not nil if b.strict { return kerr } @@ -526,6 +558,7 @@ func (b *tableCompactionBuilder) run(cnt *compactionTransactCounter) (err error) b.kerrCnt++ } + // 写到新的 SSTable if err := b.appendKV(ikey, iter.Value()); err != nil { return err } @@ -552,6 +585,7 @@ func (b *tableCompactionBuilder) revert() error { return nil } +// 执行 SSTable compaction func (db *DB) tableCompaction(c *compaction, noTrivial bool) { defer c.release() @@ -559,6 +593,7 @@ func (db *DB) tableCompaction(c *compaction, noTrivial bool) { rec.addCompPtr(c.sourceLevel, c.imax) if !noTrivial && c.trivial() { + // 只需要移动 SSTable t := c.levels[0][0] db.logf("table@move L%d@%d -> L%d", c.sourceLevel, t.fd.Num, c.sourceLevel+1) rec.delTable(c.sourceLevel, t.fd.Num) @@ -572,6 +607,7 @@ func (db *DB) tableCompaction(c *compaction, noTrivial bool) { for _, t := range tables { stats[i].read += t.size // Insert deleted tables into record + // 旧的 SSTable 都是可以删除的 rec.delTable(c.sourceLevel+i, t.fd.Num) } } @@ -587,7 +623,7 @@ func (db *DB) tableCompaction(c *compaction, noTrivial bool) { stat1: &stats[1], minSeq: minSeq, strict: db.s.o.GetStrict(opt.StrictCompaction), - tableSize: db.s.o.GetCompactionTableSize(c.sourceLevel + 1), + tableSize: db.s.o.GetCompactionTableSize(c.sourceLevel + 1), // 在 default compaction table size 默认为 2MB，切 default multiplier 为 1 的情况下，tableSize 为 2MB } db.compactionTransact("table@build", b) @@ -613,6 +649,7 @@ func (db *DB) tableCompaction(c *compaction, noTrivial bool) { } } +// 注意，这里的 umin, umax 指的是 user key func (db *DB) tableRangeCompaction(level int, umin, umax []byte) error { db.logf("table@compaction range L%d %q:%q", level, umin, umax) if level >= 0 { @@ -620,6 +657,7 @@ func (db *DB) tableRangeCompaction(level int, umin, umax []byte) error { db.tableCompaction(c, true) } } else { + // level==-1 时，对 DB 所有的 SSTable 做 compaction 整理 // Retry until nothing to compact. for { compacted := false @@ -670,6 +708,9 @@ func (db *DB) resumeWrite() bool { return v.tLen(0) < db.s.o.GetWriteL0PauseTrigger() } +// 暂停 table compaction +// 在 memtable compaction 期间，会暂停 table compaction +// pause 直到 ch 可以写入 func (db *DB) pauseCompaction(ch chan<- struct{}) { select { case ch <- struct{}{}: @@ -678,8 +719,10 @@ func (db *DB) pauseCompaction(ch chan<- struct{}) { } } +// 是 compaction command +// 只有两种 cmd：cAuto | cRange type cCmd interface { - ack(err error) + ack(err error) // 回传 ack 消息 } type cAuto struct { @@ -690,6 +733,7 @@ type cAuto struct { func (r cAuto) ack(err error) { if r.ackC != nil { defer func() { + // 防止 r.ackC 被意外 close _ = recover() }() r.ackC <- err @@ -698,7 +742,7 @@ func (r cAuto) ack(err error) { type cRange struct { level int - min, max []byte + min, max []byte // 注意：min, max 都是 user key，用 user key 来在选择参与 compaction 的 SSTable 时判定是否重叠（是否选中） ackC chan<- error } @@ -714,6 +758,7 @@ func (r cRange) ack(err error) { // This will trigger auto compaction but will not wait for it. func (db *DB) compTrigger(compC chan<- cCmd) { select { + // 注意：尽量地触发 compaction，如果当前 compC 写不进去，实际的 compaction 可能就不触发了 case compC <- cAuto{}: default: } @@ -726,6 +771,7 @@ func (db *DB) compTriggerWait(compC chan<- cCmd) (err error) { // Send cmd. select { case compC <- cAuto{ch}: + // 后续会通过读 <-ch 来等待 compaction 的结果 case err = <-db.compErrC: return case <-db.closeC: @@ -763,6 +809,7 @@ func (db *DB) compTriggerRange(compC chan<- cCmd, level int, min, max []byte) (e return err } +// db 启动时一个 go routine 专门执行此函数，进行 memtable 的 compaction func (db *DB) mCompaction() { var x cCmd @@ -795,6 +842,7 @@ func (db *DB) mCompaction() { } } +// 有专门的 go routine 来执行 tCompaction func (db *DB) tCompaction() { var ( x cCmd @@ -821,7 +869,9 @@ func (db *DB) tCompaction() { if db.tableNeedCompaction() { select { case x = <-db.tcompCmdC: + // 等到了 table compaction command case ch := <-db.tcompPauseC: + // pause table compaction，直到 ch 可以写入 db.pauseCompaction(ch) continue case <-db.closeC: @@ -851,6 +901,7 @@ func (db *DB) tCompaction() { return } } + if x != nil { switch cmd := x.(type) { case cAuto: @@ -859,6 +910,7 @@ func (db *DB) tCompaction() { if db.resumeWrite() { x.ack(nil) } else { + // L0 的 SSTable 还很多，block 住了 Wrtie 操作 waitQ = append(waitQ, x) } } diff --git a/leveldb/db_state.go b/leveldb/db_state.go index 29430fee..e3e7ebec 100644 --- a/leveldb/db_state.go +++ b/leveldb/db_state.go @@ -87,6 +87,7 @@ func (db *DB) mpoolGet(n int) *memDB { default: } if mdb == nil || mdb.Capacity() < n { + // 取 WriteBuffer 和 n 的最大值，很小的 n 可能在满的 memDB 下触发 rotateMem 操作，所以需要取这个最大值 mdb = memdb.New(db.s.icmp, maxInt(db.s.o.GetWriteBuffer(), n)) } return &memDB{ @@ -120,6 +121,8 @@ func (db *DB) mpoolDrain() { // Create new memdb and froze the old one; need external synchronization. // newMem only called synchronously by the writer. func (db *DB) newMem(n int) (mem *memDB, err error) { + + // Why: 先创建了 journal 文件，如果后面有 err 不会导致此文件一直存在得不到清理吗？ fd := storage.FileDesc{Type: storage.TypeJournal, Num: db.s.allocFileNum()} w, err := db.s.stor.Create(fd) if err != nil { @@ -130,6 +133,8 @@ func (db *DB) newMem(n int) (mem *memDB, err error) { db.memMu.Lock() defer db.memMu.Unlock() + // 已经有 immutable memtable 的话不可以再创建 immutable memtable + // 这也是 LevelDB 的一个限制 if db.frozenMem != nil { return nil, errHasFrozenMem } @@ -146,12 +151,12 @@ func (db *DB) newMem(n int) (mem *memDB, err error) { db.frozenJournalFd = db.journalFd } db.journalWriter = w - db.journalFd = fd + db.journalFd = fd // 新的 journal fd db.frozenMem = db.mem mem = db.mpoolGet(n) mem.incref() // for self mem.incref() // for caller - db.mem = mem + db.mem = mem // 切换了 memDBs // The seq only incremented by the writer. And whoever called newMem // should hold write lock, so no need additional synchronization here. db.frozenSeq = db.seq @@ -198,11 +203,14 @@ func (db *DB) getFrozenMem() *memDB { // Drop frozen memdb; assume that frozen memdb isn't nil. func (db *DB) dropFrozenMem() { db.memMu.Lock() + + // 删除 immutable memtable 的 journal if err := db.s.stor.Remove(db.frozenJournalFd); err != nil { db.logf("journal@remove removing @%d %q", db.frozenJournalFd.Num, err) } else { db.logf("journal@remove removed @%d", db.frozenJournalFd.Num) } + db.frozenJournalFd = storage.FileDesc{} db.frozenMem.decref() db.frozenMem = nil diff --git a/leveldb/db_write.go b/leveldb/db_write.go index 18eddbe1..816342cc 100644 --- a/leveldb/db_write.go +++ b/leveldb/db_write.go @@ -36,7 +36,7 @@ func (db *DB) rotateMem(n int, wait bool) (mem *memDB, err error) { retryLimit := 3 retry: // Wait for pending memdb compaction. - err = db.compTriggerWait(db.mcompCmdC) + err = db.compTriggerWait(db.mcompCmdC) // 这一步成功之后，就不该还有 frozen memDB if err != nil { return } @@ -47,6 +47,7 @@ retry: if err != nil { if err == errHasFrozenMem { if retryLimit <= 0 { + // 既然之前已经 wait 过 compaction了，那么这时就不该还有 frozen memDB panic("BUG: still has frozen memdb") } goto retry @@ -63,6 +64,7 @@ retry: return } +// 如果 db 写入数据量过快，则会触发 throttle func (db *DB) flush(n int) (mdb *memDB, mdbFree int, err error) { delayed := false slowdownTrigger := db.s.o.GetWriteL0SlowdownTrigger() @@ -79,15 +81,18 @@ func (db *DB) flush(n int) (mdb *memDB, mdbFree int, err error) { mdb = nil } }() - tLen := db.s.tLen(0) - mdbFree = mdb.Free() + tLen := db.s.tLen(0) // 当前版本 Level-0 的 SST 个数s + mdbFree = mdb.Free() // cap - size，即 mdb 还有多少剩下的空间s switch { case tLen >= slowdownTrigger && !delayed: + // level-0 的文件个数超出了阈值，默认为 slowdownTrigger=8，会减缓写入速度一次 delayed = true time.Sleep(time.Millisecond) case mdbFree >= n: + // mdb 的剩余空间足够容纳这么多数据量 return false case tLen >= pauseTrigger: + // 注意：此时 mdbFree < n，也就是说 mdb 目前不足以容纳 n 规模的数据 delayed = true // Set the write paused flag explicitly. atomic.StoreInt32(&db.inWritePaused, 1) @@ -97,13 +102,17 @@ func (db *DB) flush(n int) (mdb *memDB, mdbFree int, err error) { if err != nil { return false } + + // 注意：当 err==nil 时，会 return true，也就是继续重试，如果还是 tLen >= pausedTrigger，则继续触发 L0 compaction + // 直到 Level-0 的 SST 数目不 block write default: + // 此时，mdb 的空间不足以容纳 n 规模的数据，且 write 没有被 Level-0 的 compaction block // Allow memdb to grow if it has no entry. if mdb.Len() == 0 { mdbFree = n } else { mdb.decref() - mdb, err = db.rotateMem(n, false) + mdb, err = db.rotateMem(n, false) // 不会等待 memDB compaction 完成 if err == nil { mdbFree = mdb.Free() } else { @@ -181,16 +190,20 @@ func (db *DB) writeLocked(batch, ourBatch *Batch, merge, sync bool) error { } merge: + // 只要容量还有剩余，就 merge 下一个 Write，直到超出容量上限 + // 对于 merge 进来的 Write，通知负责 Write 的一端这个 Write 已经被 merge，于是负责的那一方可以不用管了，我们这里会负责到底，直到 Write 成功或失败 + // 对于因为超出了容量，merge 不进来的那个 Write，我们通知它的负责人“这个 Write 得由你自己负责了，没赶上上一波发车”， + // Write Lock 交接给那个负责人，它可以继续执行从它开始的 merge write for mergeLimit > 0 { select { - case incoming := <-db.writeMergeC: + case incoming := <-db.writeMergeC: // 不断合并可得的写请求，可能是要合并一个 batch write，或者合并一个 put if incoming.batch != nil { // Merge batch. if incoming.batch.internalLen > mergeLimit { overflow = true break merge } - batches = append(batches, incoming.batch) + batches = append(batches, incoming.batch) // 注意：这里是整个 batch 合并，保证 batch 执行的原子性，不会存在 batch 一部分成功了，另一部分失败了的情况 mergeLimit -= incoming.batch.internalLen } else { // Merge put. @@ -211,7 +224,7 @@ func (db *DB) writeLocked(batch, ourBatch *Batch, merge, sync bool) error { } sync = sync || incoming.sync merged++ - db.writeMergedC <- true + db.writeMergedC <- true // 通知各个 Write，这次操作已经被 merge 了 default: break merge @@ -228,6 +241,9 @@ func (db *DB) writeLocked(batch, ourBatch *Batch, merge, sync bool) error { seq := db.seq + 1 // Write journal. + // 先写 journal，journal 写成功了就是 Write 成功了 + // journal 的 seq 写入的时候用的是相同的 seq + // journal 包含了所有的 batches，同时成功或失败 if err := db.writeJournal(batches, seq, sync); err != nil { db.unlockWrite(overflow, merged, err) return err @@ -235,6 +251,7 @@ func (db *DB) writeLocked(batch, ourBatch *Batch, merge, sync bool) error { // Put batches. for _, batch := range batches { + // mdb 写入的时候不应该出现问题 if err := batch.putMem(seq, mdb.DB); err != nil { panic(err) } @@ -362,6 +379,8 @@ func (db *DB) putRec(kt keyType, key, value []byte, wo *opt.WriteOptions) error batch := db.batchPool.Get().(*Batch) batch.Reset() batch.appendRec(kt, key, value) + + // 这里传参 batch 和 ourBatch 都是同一个 `batch`，所以函数内 ourBatch 在合并 put 的时候，就相当于那个共同的 `batch` 把 put 合并掉了 return db.writeLocked(batch, batch, merge, sync) } @@ -384,6 +403,7 @@ func (db *DB) Delete(key []byte, wo *opt.WriteOptions) error { return db.putRec(keyTypeDel, key, nil, wo) } +// 注意：这里的 min, max 都是 user key，是使用 user key 来判定是否重叠 func isMemOverlaps(icmp *iComparer, mem *memdb.DB, min, max []byte) bool { iter := mem.NewIterator(nil) defer iter.Release() @@ -408,6 +428,8 @@ func (db *DB) CompactRange(r util.Range) error { // Lock writer. select { case db.writeLockC <- struct{}{}: + // 获得了 Write lock，此时 DB 的 Write 操作会被 block + // 获得这个锁是为了完成 memtable 的 compaction，要锁定 memtable 的内容 case err := <-db.compPerErrC: return err case <-db.closeC: @@ -428,9 +450,12 @@ func (db *DB) CompactRange(r util.Range) error { } <-db.writeLockC if err := db.compTriggerWait(db.mcompCmdC); err != nil { + // 等待 mem compaction 完毕，也就是刚才在 rotateMem 时被转化为 frozen memtable 的（也就是执行 CompactRange 时的当前 memtable）完成 mem compaction + // 此时就完成了 memtable 的 compaction，所以 Write lock 就被释放了 return err } } else { + // memtable 与 range 不重叠，不需要进行 mem compaction，所以直接释放锁 <-db.writeLockC } diff --git a/leveldb/iterator/merged_iter.go b/leveldb/iterator/merged_iter.go index 374e82b6..10aa47ee 100644 --- a/leveldb/iterator/merged_iter.go +++ b/leveldb/iterator/merged_iter.go @@ -30,7 +30,7 @@ type mergedIterator struct { strict bool keys [][]byte - index int + index int // 当前放出的 index, keys[index] 就是放出的 key, index in range [0, len(iters)) dir dir err error errf func(err error) @@ -77,11 +77,13 @@ func (i *mergedIterator) First() bool { for x, iter := range i.iters { switch { case iter.First(): + // merge sort i.keys[x] = assertKey(iter.Key()) h.Push(x) case i.iterErr(iter): return false default: + // x 从一开始就是空的 i.keys[x] = nil } } @@ -181,6 +183,7 @@ func (i *mergedIterator) Next() bool { case i.iterErr(iter): return false default: + // 表示 x 这个 iter 已经空了 i.keys[x] = nil } return i.next() @@ -322,7 +325,7 @@ type indexHeap mergedIterator func (h *indexHeap) Len() int { return len(h.indexes) } func (h *indexHeap) Less(i, j int) bool { i, j = h.indexes[i], h.indexes[j] - r := h.cmp.Compare(h.keys[i], h.keys[j]) + r := h.cmp.Compare(h.keys[i], h.keys[j]) // why: 如果 h.keys[i]==nil 的话，怎么做比较？ if h.reverse { return r > 0 } diff --git a/leveldb/journal/journal.go b/leveldb/journal/journal.go index f7f8b540..20d45b7b 100644 --- a/leveldb/journal/journal.go +++ b/leveldb/journal/journal.go @@ -368,7 +368,7 @@ type Writer struct { // NewWriter returns a new Writer. func NewWriter(w io.Writer) *Writer { - f, _ := w.(flusher) + f, _ := w.(flusher) // why: 从代码来看好像 storeage.Write 并没有 flush 方法，那怎么转化为 flusher ？ return &Writer{ w: w, f: f, @@ -413,12 +413,20 @@ func (w *Writer) writePending() { if w.err != nil { return } + + // pending 的 chunk 指的是写了payload，但是尚未 fillHeader 的 chunk + // 只有把 header 填充好，才是一个完备的 chunk，而填充 header 必须得在 chunk 的内容锁定之后 + // chunk 内容的锁定有两种情况： + // 1. block 写满了，不得不截断出一个 chunk； + // 2. 需要写的 payload 的确写完了，没东西继续写了，要终止 chunk，跟下一条 journal record 使用的 chunk 分隔开 if w.pending { w.fillHeader(true) - w.pending = false + w.pending = false // chunk 都写完了，没有还没有 header 的 chunk 了 } _, w.err = w.w.Write(w.buf[w.written:w.j]) w.written = w.j + + // 可以调用 Next() 了 } // Close finishes the current journal and closes the writer. @@ -508,6 +516,7 @@ type singleWriter struct { seq int } +// singleWrite 可以写入多次 p，写满 block 时会自动 fillHeader 然后开启下一个 block func (x singleWriter) Write(p []byte) (int, error) { w := x.w if w.seq != x.seq { @@ -520,12 +529,12 @@ func (x singleWriter) Write(p []byte) (int, error) { for len(p) > 0 { // Write a block, if it is full. if w.j == blockSize { - w.fillHeader(false) + w.fillHeader(false) // p 还没有写完，这时候一个 block 写满了，要开启下一个 block，所以这个 chunk 就不会是 last w.writeBlock() if w.err != nil { - return 0, w.err + return 0, w.err // 写 journal 的时候任意一段 chunk 写入失败就算作失败 } - w.first = false + w.first = false // 写了一个 chunk 后，下一个要写的 chunk 就不再是 first 了 } // Copy bytes into the buffer. n := copy(w.buf[w.j:], p) diff --git a/leveldb/opt/options.go b/leveldb/opt/options.go index 48fb0416..90483c85 100644 --- a/leveldb/opt/options.go +++ b/leveldb/opt/options.go @@ -510,6 +510,7 @@ func (o *Options) GetCompactionTableSize(level int) int { if level < len(o.CompactionTableSizeMultiplierPerLevel) && o.CompactionTableSizeMultiplierPerLevel[level] > 0 { mult = o.CompactionTableSizeMultiplierPerLevel[level] } else if o.CompactionTableSizeMultiplier > 0 { + // default CompactionTableSizeMultiplier is 1 mult = math.Pow(o.CompactionTableSizeMultiplier, float64(level)) } } diff --git a/leveldb/session.go b/leveldb/session.go index 036570e0..6c783b98 100644 --- a/leveldb/session.go +++ b/leveldb/session.go @@ -42,7 +42,7 @@ type session struct { stTempFileNum int64 stSeqNum uint64 // last mem compacted seq; need external synchronization - stor *iStorage + stor *iStorage // 底层是 file storage storLock storage.Locker o *cachedOptions icmp *iComparer @@ -55,8 +55,8 @@ type session struct { stCompPtrs []internalKey // compaction pointers; need external synchronization stVersion *version // current version ntVersionID int64 // next version id to assign - refCh chan *vTask - relCh chan *vTask + refCh chan *vTask // 注：reference version task + relCh chan *vTask // 注：release version task deltaCh chan *vDelta abandon chan int64 closeC chan struct{} @@ -208,11 +208,14 @@ func (s *session) recover() (err error) { } // Commit session; need external synchronization. +// r: 保存 new version 相比于当前 version 的 change +// Commit 一次 compaction func (s *session) commit(r *sessionRecord, trivial bool) (err error) { v := s.version() defer v.release() // spawn new version based on current version + // 记录在 current version 中！ nv := v.spawn(r, trivial) // abandon useless version id to prevent blocking version processing loop. @@ -225,15 +228,21 @@ func (s *session) commit(r *sessionRecord, trivial bool) (err error) { if s.manifest == nil { // manifest journal writer not yet created, create one + // why: 为什么要传 r，而不是传 nil ? 应该是保存新的全量 version 到 manifest 中就可以了 ? + // nv 已经是新的 version 了 err = s.newManifest(r, nv) } else if s.manifest.Size() >= s.o.GetMaxManifestFileSize() { // pass nil sessionRecord to avoid over-reference table file + // 当 manifest 的大小超出了阈值，就把 new version 里面的 SSTables 刷到一个全新的 manifest 里面 + // 这个 manifest 就只包含 atRecord，不会有 dtRecord 了 err = s.newManifest(nil, nv) } else { + // 当前已经有了 manifest，就将 version 的变更继续写在这个 manifest 中 err = s.flushManifest(r) } // finally, apply new version if no error rise + // 注意：这一步是在 new version 的 manifest 写入成功了之后才做的 version 切换 if err == nil { s.setVersion(r, nv) } diff --git a/leveldb/session_compaction.go b/leveldb/session_compaction.go index 2fd5f32e..b1836194 100644 --- a/leveldb/session_compaction.go +++ b/leveldb/session_compaction.go @@ -28,6 +28,7 @@ func (s *session) pickMemdbLevel(umin, umax []byte, maxLevel int) int { return v.pickMemdbLevel(umin, umax, maxLevel) } +// 创建 SST 并添加到 rec func (s *session) flushMemdb(rec *sessionRecord, mdb *memdb.DB, maxLevel int) (int, error) { // Create sorted table. iter := mdb.NewIterator(nil) @@ -66,12 +67,14 @@ func (s *session) pickCompaction() *compaction { if cptr != nil && sourceLevel > 0 { n := len(tables) if i := sort.Search(n, func(i int) bool { + // 选取第一个 imax 大于 cptr 的 SSTable return s.icmp.Compare(tables[i].imax, cptr) > 0 }); i < n { t0 = append(t0, tables[i]) } } if len(t0) == 0 { + // 如果 ctpr==nil 或 cptr已经到了最后一个 SSTable，则从头开始循环 t0 = append(t0, tables[0]) } if sourceLevel == 0 { @@ -80,6 +83,7 @@ func (s *session) pickCompaction() *compaction { typ = nonLevel0Compaction } } else { + // 执行 seek compaction if p := atomic.LoadPointer(&v.cSeek); p != nil { ts := (*tSet)(p) sourceLevel = ts.level @@ -99,10 +103,15 @@ func (s *session) getCompactionRange(sourceLevel int, umin, umax []byte, noLimit v := s.version() if sourceLevel >= len(v.levels) { + // sourceLevel >= 7，没有这样的 level v.release() return nil } + // 注意：sourceLevel==0 时，因为 L0 上的 SSTable 可能有交叠，所以需要拓展 umin, umax 的范围 + // 比如在 L0 上，umax 涉及到了一个新的 SSTable，那么这个 SSTable 的 max 会拓宽一点，就可能会交叠上下一个新的 SSTable + // 所以需要拓展 umin, umax 的范围，直到不可以继续拓展 + // 使用的是 ukey，完整的 ukey 要完整地被 dump，不能出现部分 ukey 被 dump 了，这样会出现查询的错误 t0 := v.levels[sourceLevel].getOverlaps(nil, s.icmp, umin, umax, sourceLevel == 0) if len(t0) == 0 { v.release() @@ -153,17 +162,17 @@ type compaction struct { s *session v *version - typ int - sourceLevel int - levels [2]tFiles + typ int // compaction 的 type: L0, non-L0, seek compaction + sourceLevel int // compaction 的 source level，发起 compaction 的 level + levels [2]tFiles // 参与 compaction 的两个 level 上的文件 maxGPOverlaps int64 - gp tFiles + gp tFiles // sourceLevel+2 上的、与某次 compaction expand 之后的 range 相重叠的文件，gp 的意思是 grad parent gpi int seenKey bool gpOverlappedBytes int64 - imin, imax internalKey - tPtrs []int + imin, imax internalKey // 参与 compaction 的 sourceLevel 上的 internalKey range + tPtrs []int // tPtrs[level] 是 level 上的一个 SSTable 的 index，用于在 compaction 的时候快速判断 ukey 是否不会在 sourceLevel+2 及更深的 level 上出现 released bool snapGPI int @@ -225,6 +234,8 @@ func (c *compaction) expand() { xmin, xmax := exp0.getRange(c.s.icmp) exp1 := vt1.getOverlaps(nil, c.s.icmp, xmin.ukey(), xmax.ukey(), false) if len(exp1) == len(t1) { + // 增选了 t0 上的 SSTable，并没有使的 t1 的 SSTable 被增选，即可以确定参与 compaction 的文件 + // 为了避免这种 expand 变得无穷无尽 c.s.logf("table@compaction expanding L%d+L%d (F·%d S·%s)+(F·%d S·%s) -> (F·%d S·%s)+(F·%d S·%s)", c.sourceLevel, c.sourceLevel+1, len(t0), shortenb(t0.size()), len(t1), shortenb(t1.size()), len(exp0), shortenb(exp0.size()), len(exp1), shortenb(exp1.size())) @@ -250,7 +261,12 @@ func (c *compaction) trivial() bool { return len(c.levels[0]) == 1 && len(c.levels[1]) == 0 && c.gp.size() <= c.maxGPOverlaps } +// 如果 ukey 只存在于 compaction 涉及到的两个 level，不在更高的 level 出现，返回 true，否则返回 false +// "baseLevel" 的意思在此 +// 因为在 compaction 的过程中，从 iterator 出来的 ukey 必然是递增的，后面的 ukey 只会大于等于前面的 ukey +// 所以这个函数内部用到的 c.tPtrs[level] 是可以单调向前移动的 func (c *compaction) baseLevelForKey(ukey []byte) bool { + // 从 sourceLevel+2 开始搜索 for level := c.sourceLevel + 2; level < len(c.v.levels); level++ { tables := c.v.levels[level] for c.tPtrs[level] < len(tables) { @@ -259,6 +275,8 @@ func (c *compaction) baseLevelForKey(ukey []byte) bool { // We've advanced far enough. if c.s.icmp.uCompare(ukey, t.imin.ukey()) >= 0 { // Key falls in this file's range, so definitely not base level. + // ukey 只是在这个文件的 range 内，其实并不一定代表这个文件一定包含这个 ukey ? + // 感觉这里是为了判断的速度，于是保守一点，给出了 ukey 可能出现在更高层的结论 return false } break @@ -283,6 +301,9 @@ func (c *compaction) shouldStopBefore(ikey internalKey) bool { if c.gpOverlappedBytes > c.maxGPOverlaps { // Too much overlap for current output; start new output. + // compaction 之后在 sourceLevel+1 之后产生的新文件 f 也不可以太大，如果过大了的话，f 会跟 sourceLevel+2 的文件有过多的交集 + // 那么将来当 f 需要做 compcation 的时候，下一层就会涉及到过多的文件，那时的 compaction 就过于 heavy 了 + // 所以这时要终止当前的 SSTable，开启下一个 SSTable c.gpOverlappedBytes = 0 return true } @@ -301,7 +322,7 @@ func (c *compaction) newIterator() iterator.Iterator { // Options. ro := &opt.ReadOptions{ - DontFillCache: true, + DontFillCache: true, // 作 compaction 的时候读取的数据跟业务无关，不要填充 cache Strict: opt.StrictOverride, } strict := c.s.o.GetStrict(opt.StrictCompaction) @@ -320,6 +341,7 @@ func (c *compaction) newIterator() iterator.Iterator { its = append(its, c.s.tops.newIterator(t, nil, ro)) } } else { + // tables 本身就是排好序的，所以 newIndexIterator 内部做二分是没问题的 it := iterator.NewIndexedIterator(tables.newIndexIterator(c.s.tops, c.s.icmp, nil, ro), strict) its = append(its, it) } diff --git a/leveldb/session_record.go b/leveldb/session_record.go index b1a352f6..2d84b6a2 100644 --- a/leveldb/session_record.go +++ b/leveldb/session_record.go @@ -39,29 +39,37 @@ type cpRecord struct { ikey internalKey } +// add table record +// add SSTable type atRecord struct { - level int - num int64 + level int // 把 table 添加到第几个 level + num int64 // fileDesc num size int64 imin internalKey imax internalKey } +// delete table record type dtRecord struct { level int num int64 } type sessionRecord struct { - hasRec int + hasRec int // bitsmap, addTable: 7 comparer string journalNum int64 prevJournalNum int64 nextFileNum int64 seqNum uint64 - compPtrs []cpRecord - addedTables []atRecord - deletedTables []dtRecord + + // compact pointers 指示每个层级下一次进行 compaction 操作时需要从哪个键开始 + // 对每个层级 L，会记录该层上一次进行 compaction 时操作的最大值，于是当 L 层进行下一次 compaction 需要选取文件时，选取第一个 imax 大于 compPtr.ikey 的文件 + // 于是，每一层的 compaction 操作在该层的键空间循环 + compPtrs []cpRecord + + addedTables []atRecord + deletedTables []dtRecord scratch [binary.MaxVarintLen64]byte err error diff --git a/leveldb/session_util.go b/leveldb/session_util.go index f467f2d4..e03362ec 100644 --- a/leveldb/session_util.go +++ b/leveldb/session_util.go @@ -264,6 +264,7 @@ func (s *session) tLen(level int) int { } // Set current version to v. +// r: 包含下一个 version 相对于当前 stVersion 的 change func (s *session) setVersion(r *sessionRecord, v *version) { s.vmu.Lock() defer s.vmu.Unlock() @@ -349,6 +350,7 @@ func (s *session) setCompPtr(level int, ik internalKey) { // Get compaction ptr at given level; need external synchronization. func (s *session) getCompPtr(level int) internalKey { if level >= len(s.stCompPtrs) { + // 从最小的 ikey 开始 return nil } return s.stCompPtrs[level] @@ -367,7 +369,7 @@ func (s *session) fillRecord(r *sessionRecord, snapshot bool) { } if !r.has(recSeqNum) { - r.setSeqNum(s.stSeqNum) + r.setSeqNum(s.stSeqNum) // mem compacted seq } for level, ik := range s.stCompPtrs { @@ -403,7 +405,7 @@ func (s *session) recordCommited(rec *sessionRecord) { // Create a new manifest file; need external synchronization. func (s *session) newManifest(rec *sessionRecord, v *version) (err error) { fd := storage.FileDesc{Type: storage.TypeManifest, Num: s.allocFileNum()} - writer, err := s.stor.Create(fd) + writer, err := s.stor.Create(fd) // 创建 manifest 文件 if err != nil { return } @@ -455,17 +457,20 @@ func (s *session) newManifest(rec *sessionRecord, v *version) (err error) { if err != nil { return } + + // 默认对 manifest 文件用 sync，GetNoSync 默认返回 false if !s.o.GetNoSync() { err = writer.Sync() if err != nil { return } } - err = s.stor.SetMeta(fd) + err = s.stor.SetMeta(fd) // 设置 CURRENT 文件的内容 return } // Flush record to disk. +// Manifest 的写入复用了 journal func (s *session) flushManifest(rec *sessionRecord) (err error) { s.fillRecord(rec, false) w, err := s.manifest.Next() diff --git a/leveldb/table.go b/leveldb/table.go index d0fab40c..ccf32b23 100644 --- a/leveldb/table.go +++ b/leveldb/table.go @@ -149,6 +149,7 @@ func (tf tFiles) searchNumLess(num int64) int { // Searches smallest index of tables whose its smallest // key is after the given key. +// 按照 user key 来作比较 func (tf tFiles) searchMinUkey(icmp *iComparer, umin []byte) int { return sort.Search(len(tf), func(i int) bool { return icmp.ucmp.Compare(tf[i].imin.ukey(), umin) > 0 @@ -157,6 +158,7 @@ func (tf tFiles) searchMinUkey(icmp *iComparer, umin []byte) int { // Searches smallest index of tables whose its largest // key is after the given key. +// 按照 user key 来作比较 func (tf tFiles) searchMaxUkey(icmp *iComparer, umax []byte) int { return sort.Search(len(tf), func(i int) bool { return icmp.ucmp.Compare(tf[i].imax.ukey(), umax) > 0 @@ -179,13 +181,14 @@ func (tf tFiles) overlaps(icmp *iComparer, umin, umax []byte, unsorted bool) boo i := 0 if len(umin) > 0 { // Find the earliest possible internal key for min. + // 先找第一个 imax 大于等于 umin 的 SSTs i = tf.searchMax(icmp, makeInternalKey(nil, umin, keyMaxSeq, keyTypeSeek)) } if i >= len(tf) { // Beginning of range is after all files, so no overlap. return false } - return !tf[i].before(icmp, umax) + return !tf[i].before(icmp, umax) // umax 不在这个文件前面，即此 SST 与 [umin, umax] 有交集 } // Returns tables whose its key range overlaps with given key range. @@ -193,6 +196,7 @@ func (tf tFiles) overlaps(icmp *iComparer, umin, umax []byte, unsorted bool) boo // If overlapped is true then the search will be restarted if umax // expanded. // The dst content will be overwritten. +// why: 为什么要用 ukey 参与 overlap 的判断？Ans：使的 ukey 被完整地存在在一个 Level 中，不然如果部分被 dump 的话，可能某个 snapshot 读的时候读到旧的数据 func (tf tFiles) getOverlaps(dst tFiles, icmp *iComparer, umin, umax []byte, overlapped bool) tFiles { // Short circuit if tf is empty if len(tf) == 0 { @@ -202,7 +206,7 @@ func (tf tFiles) getOverlaps(dst tFiles, icmp *iComparer, umin, umax []byte, ove // And what's more, the files in these levels are strictly sorted, // so use binary search instead of heavy traverse. if !overlapped { - var begin, end int + var begin, end int // 区间 [begin, end) // Determine the begin index of the overlapped file if umin != nil { index := tf.searchMinUkey(icmp, umin) @@ -222,11 +226,13 @@ func (tf tFiles) getOverlaps(dst tFiles, icmp *iComparer, umin, umax []byte, ove end = len(tf) } else if bytes.Compare(tf[index].imin.ukey(), umax) <= 0 { // The max ukey overlaps with the index file, expand it. + // 注意：end 是开区间，所以这里要等于 index+1，是不包括 index+1 这个 SSTable 的 end = index + 1 } else { end = index } } else { + // 注意：end 是开区间，所以这里要等于 len(tf)，也就是不存在的一个文件 end = len(tf) } // Ensure the overlapped file indexes are valid. @@ -242,6 +248,8 @@ func (tf tFiles) getOverlaps(dst tFiles, icmp *iComparer, umin, umax []byte, ove for i := 0; i < len(tf); { t := tf[i] if t.overlaps(icmp, umin, umax) { + + // 扩大 [umin, umax] 的范围 if umin != nil && icmp.uCompare(t.imin.ukey(), umin) < 0 { umin = t.imin.ukey() dst = dst[:0] @@ -284,7 +292,7 @@ func (tf tFiles) getRange(icmp *iComparer) (imin, imax internalKey) { // Creates iterator index from tables. func (tf tFiles) newIndexIterator(tops *tOps, icmp *iComparer, slice *util.Range, ro *opt.ReadOptions) iterator.IteratorIndexer { if slice != nil { - var start, limit int + var start, limit int // 区间 [start, limit)，limit 更像是 end 的概念 if slice.Start != nil { start = tf.searchMax(icmp, internalKey(slice.Start)) } @@ -389,7 +397,7 @@ func (t *tOps) createFrom(src iterator.Iterator) (f *tFile, n int, err error) { return } } - err = src.Error() + err = src.Error() // accumulated error if err != nil { return } @@ -401,8 +409,10 @@ func (t *tOps) createFrom(src iterator.Iterator) (f *tFile, n int, err error) { // Opens table. It returns a cache handle, which should // be released after use. +// 会 cache SSTable func (t *tOps) open(f *tFile) (ch *cache.Handle, err error) { - ch = t.fileCache.Get(0, uint64(f.fd.Num), func() (size int, value cache.Value) { + + loader := func() (size int, value cache.Value) { var r storage.Reader r, err = t.s.stor.Open(f.fd) if err != nil { @@ -421,8 +431,9 @@ func (t *tOps) open(f *tFile) (ch *cache.Handle, err error) { return 0, nil } return 1, tr + } - }) + ch = t.fileCache.Get(0, uint64(f.fd.Num), loader) if ch == nil && err == nil { err = ErrClosed } diff --git a/leveldb/table/reader.go b/leveldb/table/reader.go index 8128794c..966a2453 100644 --- a/leveldb/table/reader.go +++ b/leveldb/table/reader.go @@ -57,19 +57,36 @@ type block struct { bpool *util.BufferPool bh blockHandle data []byte - restartsLen int + restartsLen int // restart point 的个数 restartsOffset int } +// 返回最后一个小于等于 key 的 restart point index，和对应 restart point 的 offset +// 后续一半从这个 restart point 开始查找 key func (b *block) seek(cmp comparer.Comparer, rstart, rlimit int, key []byte) (index, offset int, err error) { - index = sort.Search(b.restartsLen-rstart-(b.restartsLen-rlimit), func(i int) bool { - offset := int(binary.LittleEndian.Uint32(b.data[b.restartsOffset+4*(rstart+i):])) - offset++ // shared always zero, since this is a restart point - v1, n1 := binary.Uvarint(b.data[offset:]) // key length - _, n2 := binary.Uvarint(b.data[offset+n1:]) // value length - m := offset + n1 + n2 - return cmp.Compare(b.data[m:m+int(v1)], key) > 0 - }) + rstart - 1 + // 查 [rstart, rlimit) 上的 restart point + // 这个二分查找是要找到第一个比 key 大的 restart point，返回的下标是针对 [rstart, rlimit) 而言的 + // 比如返回 i==0 的话，其实对应的是 restart point rstart + // 注意配合上后面的 + rstart - 1 使用，那么最终 index 是：最后一个小于等于 key 的 restart point + index = sort.Search( + b.restartsLen-rstart-(b.restartsLen-rlimit), // [0, rlimit - rstart) + func(i int) bool { + // 看从 rstart 算起， restart point i (based 0) 的 key 是否比 key 大 + + // [rstart, rlimit] 上的第 i 个 offset，比如 i==0 的话，就是 rstart 这个 restart point + // offset 就定位到这个 restart point 指向的数据 + // 因为 restart point 指向的数据没有“前面的”数据，所以 shared_length 必然等于 0 + offset := int(binary.LittleEndian.Uint32(b.data[b.restartsOffset+4*(rstart+i):])) + offset++ // shared always zero, since this is a restart point + + v1, n1 := binary.Uvarint(b.data[offset:]) // key length + _, n2 := binary.Uvarint(b.data[offset+n1:]) // value length + m := offset + n1 + n2 // key offset + return cmp.Compare(b.data[m:m+int(v1)], key) > 0 + }) + rstart - 1 + + // 如果 [rstart, rlimit) 都比 key 大，那么最终调整 index=rstart，因为（调用方）最终是找第一个大于等于 key 的，这样一来对调用方来讲最终答案就是 rstart + // 如果 [rstart, rlimit) 都不比 key 大，那么最终 index=rlimit-1，即区间上的最后一个 restart point，因为数据有序排布，所以这个最后的 restart point 之后的数据也是可能会大于等于 key 的 if index < rstart { // The smallest key is greater-than key sought. index = rstart @@ -88,22 +105,27 @@ func (b *block) restartOffset(index int) int { return int(binary.LittleEndian.Uint32(b.data[b.restartsOffset+4*index:])) } +// n: 读取的这条 k/v 的总存储长度，包括 shared_bytes, unshared_bytes 等信息的所有 bytes func (b *block) entry(offset int) (key, value []byte, nShared, n int, err error) { + + // offset 不应该伸展到 restartsOffset 区域 if offset >= b.restartsOffset { if offset != b.restartsOffset { err = &ErrCorrupted{Reason: "entries offset not aligned"} } return } + v0, n0 := binary.Uvarint(b.data[offset:]) // Shared prefix length - v1, n1 := binary.Uvarint(b.data[offset+n0:]) // Key length + v1, n1 := binary.Uvarint(b.data[offset+n0:]) // 注：Unshard key length v2, n2 := binary.Uvarint(b.data[offset+n0+n1:]) // Value length m := n0 + n1 + n2 - n = m + int(v1) + int(v2) + n = m + int(v1) + int(v2) // 一条 k/v 的长度 if n0 <= 0 || n1 <= 0 || n2 <= 0 || offset+n > b.restartsOffset { err = &ErrCorrupted{Reason: "entries corrupted"} return } + key = b.data[offset+m : offset+m+int(v1)] value = b.data[offset+m+int(v1) : offset+n] nShared = int(v0) @@ -120,8 +142,8 @@ type dir int const ( dirReleased dir = iota - 1 - dirSOI - dirEOI + dirSOI // 在 iteration 的头部，Start Of Iteration + dirEOI // 到了 iteration 的尾部，在用 dir 表达iterator的状态 dirBackward dirForward ) @@ -137,16 +159,16 @@ type blockIter struct { prevOffset int prevNode []int prevKeys []byte - restartIndex int + restartIndex int // blockIter 当前下标所在的（在这个restart point后面）restart point index // Iterator direction. dir dir // Restart index slice range. - riStart int - riLimit int + riStart int // 框定 iterator 的范围，起始的 restart point index + riLimit int // 框定 iterator 的范围，结束的 restart point index（不包括） // Offset slice range. - offsetStart int + offsetStart int // 框定 iterator 的范围，起始的 offset offsetRealStart int - offsetLimit int + offsetLimit int // 框定 iterator 的范围，结束的 offset，当 blockIter.offset==offsetLimit 的时候，表示 iteration 结束，不应该出现 offset > offsetLimit 的情况 // Error. err error } @@ -221,6 +243,7 @@ func (i *blockIter) Last() bool { return i.Prev() } +// 找到第一个大于等于 key 的记录 func (i *blockIter) Seek(key []byte) bool { if i.err != nil { return false @@ -229,6 +252,9 @@ func (i *blockIter) Seek(key []byte) bool { return false } + // 按照 restart point 二分查找 + // 找第一个小于 key 的 restart point + // 因为 restart point 是到下一个 restart point 最小的 ikey，如果不是选小于 key 的话，就错过这个 key 了 ri, offset, err := i.block.seek(i.tr.cmp, i.riStart, i.riLimit, key) if err != nil { i.sErr(err) @@ -239,6 +265,8 @@ func (i *blockIter) Seek(key []byte) bool { if i.dir == dirSOI || i.dir == dirEOI { i.dir = dirForward } + + // 最终找第一个大于等于 key 的记录 for i.Next() { if i.tr.cmp.Compare(i.key, key) >= 0 { return true @@ -247,6 +275,8 @@ func (i *blockIter) Seek(key []byte) bool { return false } +// 按照 dirForward 的方向，如果 blockIter 还有下一个值，读取这个值并返回 true +// 只做 dirForward 方向 func (i *blockIter) Next() bool { if i.dir == dirEOI || i.err != nil { return false @@ -256,26 +286,35 @@ func (i *blockIter) Next() bool { } if i.dir == dirSOI { + // 回到起点 i.restartIndex = i.riStart i.offset = i.offsetStart } else if i.dir == dirBackward { i.prevNode = i.prevNode[:0] i.prevKeys = i.prevKeys[:0] } + + // 使用 i.offset 作 iteration + // 把 i.offset 推进到 i.offsetRealStart for i.offset < i.offsetRealStart { key, value, nShared, n, err := i.block.entry(i.offset) if err != nil { i.sErr(i.tr.fixErrCorruptedBH(i.block.bh, err)) return false } + if n == 0 { i.dir = dirEOI return false } + + // 读到新的 k/v + // 注意：这里新读到的 key 是由 shared + unshared 拼凑出来的 i.key = append(i.key[:nShared], key...) i.value = value - i.offset += n + i.offset += n // 跨过整条 k/v } + if i.offset >= i.offsetLimit { i.dir = dirEOI if i.offset != i.offsetLimit { @@ -283,7 +322,8 @@ func (i *blockIter) Next() bool { } return false } - key, value, nShared, n, err := i.block.entry(i.offset) + + key, value, nShared, n, err := i.block.entry(i.offset) // n: 整条 entry 的 size，于是 offset+=n 会跳到 下一个 entry if err != nil { i.sErr(i.tr.fixErrCorruptedBH(i.block.bh, err)) return false @@ -292,7 +332,8 @@ func (i *blockIter) Next() bool { i.dir = dirEOI return false } - i.key = append(i.key[:nShared], key...) + + i.key = append(i.key[:nShared], key...) // 当前的值由 shard+unshared 拼凑起来得到 i.value = value i.prevOffset = i.offset i.offset += n @@ -521,12 +562,15 @@ type Reader struct { filter filter.Filter verifyChecksum bool - dataEnd int64 + dataEnd int64 + + // indexBH: data index block handle，检索 data blocks metaBH, indexBH, filterBH blockHandle indexBlock *block filterBlock *filterBlock } +// 一个 SST 包含多个blocks func (r *Reader) blockKind(bh blockHandle) string { switch bh.offset { case r.metaBH.offset: @@ -549,6 +593,7 @@ func (r *Reader) newErrCorruptedBH(bh blockHandle, reason string) error { return r.newErrCorrupted(int64(bh.offset), int64(bh.length), r.blockKind(bh), reason) } +// 汇报一个block的错误 func (r *Reader) fixErrCorruptedBH(bh blockHandle, err error) error { if cerr, ok := err.(*ErrCorrupted); ok { cerr.Pos = int64(bh.offset) @@ -742,6 +787,7 @@ func (r *Reader) newBlockIter(b *block, bReleaser util.Releaser, slice *util.Ran bi.offsetStart = b.restartOffset(bi.riStart) bi.offsetRealStart = bi.prevOffset } else { + // 整个 block 都不在 range 内 bi.riStart = b.restartsLen bi.offsetStart = b.restartsOffset bi.offsetRealStart = b.restartsOffset @@ -761,6 +807,7 @@ func (r *Reader) newBlockIter(b *block, bReleaser util.Releaser, slice *util.Ran return bi } +// 读取 data block 的 iterator func (r *Reader) getDataIter(dataBH blockHandle, slice *util.Range, verifyChecksum, fillCache bool) iterator.Iterator { b, rel, err := r.readBlockCached(dataBH, verifyChecksum, fillCache) if err != nil { @@ -817,6 +864,8 @@ func (r *Reader) NewIterator(slice *util.Range, ro *opt.ReadOptions) iterator.It return iterator.NewIndexedIterator(index, opt.GetStrict(r.o, ro, opt.StrictReader)) } +// 在 SSTable 中查找 key +// 从调用端来看，key 是 internal key func (r *Reader) find(key []byte, filtered bool, ro *opt.ReadOptions, noValue bool) (rkey, value []byte, err error) { r.mu.RLock() defer r.mu.RUnlock() @@ -835,6 +884,17 @@ func (r *Reader) find(key []byte, filtered bool, ro *opt.ReadOptions, noValue bo index := r.newBlockIter(indexBlock, nil, nil, true) defer index.Release() + // index 也是一个 data block + // 所以 index.Seek 方法也是去找第一个大于等于 key 的记录，只是该条记录的 val 是 data block handle + // data block handle 的 key 是这个 block 的 successor，所以比所索引的 data block 的所有记录都大 + // + // 所以我们从第一个大于等于 key 的 data block handle 开始找就可以了， + // 前面的那个 data block 因为索引键都小于要找的key了，里面的数据就更不可能大于等于 key 了 + // + // Why: data index 的 key 是 data block 的 last key 的 successor， + // 如果 user key 就全是 0xff, 0xff, 0xff...，那么 last key 的 successor 就必然会增加 seq number 的字段 + // 那么，这个 successor 不就会“小于” data block 里面的记录了吗？不就不是 "successor" 了吗？ + // 当然，这种情况生产环境下不大可能会出现 if !index.Seek(key) { if err = index.Error(); err == nil { err = ErrNotFound @@ -842,6 +902,7 @@ func (r *Reader) find(key []byte, filtered bool, ro *opt.ReadOptions, noValue bo return } + // 找到了潜在的 data block dataBH, n := decodeBlockHandle(index.Value()) if n == 0 { r.err = r.newErrCorruptedBH(r.indexBH, "bad data block handle") @@ -852,6 +913,7 @@ func (r *Reader) find(key []byte, filtered bool, ro *opt.ReadOptions, noValue bo if filtered && r.filter != nil { filterBlock, frel, ferr := r.getFilterBlock(true) if ferr == nil { + // 使用 filter 来提前判断 key 不存在，省去读取 block 的内容再查询 if !filterBlock.contains(r.filter, dataBH.offset, key) { frel.Release() return nil, nil, ErrNotFound @@ -863,6 +925,7 @@ func (r *Reader) find(key []byte, filtered bool, ro *opt.ReadOptions, noValue bo } data := r.getDataIter(dataBH, nil, r.verifyChecksum, !ro.GetDontFillCache()) + // 在具体的 block 中去找 if !data.Seek(key) { data.Release() if err = data.Error(); err != nil { diff --git a/leveldb/table/table.go b/leveldb/table/table.go index 29f80f8e..1f06d6a4 100644 --- a/leveldb/table/table.go +++ b/leveldb/table/table.go @@ -153,6 +153,7 @@ const ( blockTypeSnappyCompression = 1 ) +// 一个 blockHandle 可以指定一块数据区域 type blockHandle struct { offset, length uint64 } diff --git a/leveldb/table/writer.go b/leveldb/table/writer.go index ea89d600..429ab313 100644 --- a/leveldb/table/writer.go +++ b/leveldb/table/writer.go @@ -31,6 +31,8 @@ func sharedPrefixLen(a, b []byte) int { return i } +// 是 data block，就是存 k/v +// 默认 blockSize = 4kb type blockWriter struct { restartInterval int buf util.Buffer @@ -158,7 +160,7 @@ type Writer struct { dataBlock blockWriter indexBlock blockWriter filterBlock filterWriter - pendingBH blockHandle + pendingBH blockHandle // 上一个写入了的 dataBlock 的 block handle offset uint64 nEntries int // Scratch allocated enough for 5 uvarint. Block writer should not use @@ -166,7 +168,7 @@ type Writer struct { // then passed to the block writer itself. scratch [50]byte comparerScratch []byte - compressionScratch []byte + compressionScratch []byte // 写入 dataBlock 的时候保存 dataBlock 被 compression 之后的数据 } func (w *Writer) writeBlock(buf *util.Buffer, compression opt.Compression) (bh blockHandle, err error) { @@ -182,6 +184,9 @@ func (w *Writer) writeBlock(buf *util.Buffer, compression opt.Compression) (bh b b = compressed[:n+blockTrailerLen] b[n] = blockTypeSnappyCompression } else { + // buf.Alloc 会原地增长 buf 本身，预留出来 blockTrailerLen 这么多的 bytes + // 所以这里的 else 会得到跟上面选择压缩的时候同样布局的 b: + // {b's data} | {CRC checksum, 4 bytes} | {compression type, 1 byte} tmp := buf.Alloc(blockTrailerLen) tmp[0] = blockTypeNoCompression b = buf.Bytes() @@ -193,12 +198,15 @@ func (w *Writer) writeBlock(buf *util.Buffer, compression opt.Compression) (bh b binary.LittleEndian.PutUint32(b[n:], checksum) // Write the buffer to the file. + // 此时的 b 保存了排布好的 entry 和 restart points，直接写进文件就可以了 _, err = w.writer.Write(b) if err != nil { return } + + // bh 保存的是当前写入的 b 的 blockHandle bh = blockHandle{w.offset, uint64(len(b) - blockTrailerLen)} - w.offset += uint64(len(b)) + w.offset += uint64(len(b)) // blockTrailer 在文件中肯定也是算长度的，所以这里 offset 增长的就是 + len(b)，包括了 blockTrailer 的长度 return } @@ -206,6 +214,8 @@ func (w *Writer) flushPendingBH(key []byte) error { if w.pendingBH.length == 0 { return nil } + + // 前一个 block 的最后一个 key，与当前 block 的第一个 key，两者的最短分隔符 var separator []byte if len(key) == 0 { separator = w.cmp.Successor(w.comparerScratch[:0], w.dataBlock.prevKey) @@ -213,15 +223,20 @@ func (w *Writer) flushPendingBH(key []byte) error { separator = w.cmp.Separator(w.comparerScratch[:0], w.dataBlock.prevKey, key) } if separator == nil { + // 比如当 len(key) == 0 && w.dataBlock.prevKey 里面都是 0xff 时，separator 将会是 nil + // 或者 w.dataBlock.prevKey == key 时，separator 也会是 nil + // 总之，separator 得是大于等于 w.dataBlock 中的所有 key 的 separator = w.dataBlock.prevKey } else { w.comparerScratch = separator } - n := encodeBlockHandle(w.scratch[:20], w.pendingBH) + + n := encodeBlockHandle(w.scratch[:20], w.pendingBH) // pendingBH 是 prev dataBlock 的 handle // Append the block handle to the index block. if err := w.indexBlock.append(separator, w.scratch[:n]); err != nil { return err } + // Reset prev key of the data block. w.dataBlock.prevKey = w.dataBlock.prevKey[:0] // Clear pending block handle. @@ -230,14 +245,16 @@ func (w *Writer) flushPendingBH(key []byte) error { } func (w *Writer) finishBlock() error { + // 写入 dataBlock 的 restart points if err := w.dataBlock.finish(); err != nil { return err } + bh, err := w.writeBlock(&w.dataBlock.buf, w.compression) if err != nil { return err } - w.pendingBH = bh + w.pendingBH = bh // pendingBH 记录下目前写入的 blockHandle，下一次再写入 key 的时候，会开启一个新的 dataBlock，那时 pendingBH 会被写入 // Reset the data block. w.dataBlock.reset() // Flush the filter block. @@ -269,6 +286,8 @@ func (w *Writer) Append(key, value []byte) error { w.filterBlock.add(key) // Finish the data block if block size target reached. + // 那么一个 data block 的实际大小是可能超过 blockSize 的，因为不是预留空间然后保证不超 + // 所以，一条很大的数据可能会占据一个超大的 block？允许吗？Why if w.dataBlock.bytesLen() >= w.blockSize { if err := w.finishBlock(); err != nil { w.err = err @@ -319,6 +338,7 @@ func (w *Writer) Close() error { // Write the last data block. Or empty data block if there // aren't any data blocks at all. + // 如果 data block 是空的，至少会写入一个 restart point if w.dataBlock.nEntries > 0 || w.nEntries == 0 { if err := w.finishBlock(); err != nil { w.err = err @@ -343,8 +363,11 @@ func (w *Writer) Close() error { // Write the metaindex block. if filterBH.length > 0 { + // key: "filter.{w.filter.Name()}" + // val: filter block handle key := []byte("filter." + w.filter.Name()) n := encodeBlockHandle(w.scratch[:20], filterBH) + // 注意：这里写 k/v 也是用的 w.dataBlock，即 metaindex block 本身的格式也是 data block 的格式 if err := w.dataBlock.append(key, w.scratch[:n]); err != nil { return err } @@ -375,7 +398,7 @@ func (w *Writer) Close() error { } n := encodeBlockHandle(footer, metaindexBH) encodeBlockHandle(footer[n:], indexBH) - copy(footer[footerLen-len(magic):], magic) + copy(footer[footerLen-len(magic):], magic) // 文件的尾部是 magic number if _, err := w.writer.Write(footer); err != nil { w.err = err return w.err @@ -396,7 +419,7 @@ func NewWriter(f io.Writer, o *opt.Options, pool *util.BufferPool, size int) *Wr } else { bufBytes = pool.Get(size) } - bufBytes = bufBytes[:0] + bufBytes = bufBytes[:0] // clear up bufBytes, in case it is fetched from pool w := &Writer{ writer: f, diff --git a/leveldb/version.go b/leveldb/version.go index 46725091..43983dca 100644 --- a/leveldb/version.go +++ b/leveldb/version.go @@ -22,6 +22,7 @@ type tSet struct { table *tFile } +// 保存每个层级的文件信息以及判断是否需要更新compaction相关的变量 type version struct { id int64 // unique monotonous increasing version id s *session @@ -34,7 +35,7 @@ type version struct { cLevel int cScore float64 - cSeek unsafe.Pointer + cSeek unsafe.Pointer // 保存 tSet，需要被 compaction 的下一个 SSTable？ closing bool ref int @@ -64,6 +65,8 @@ func (v *version) incref() { } } +// 释放 version +// 当一个 version 引用计数减为 0 时，可以删除掉这个 version func (v *version) releaseNB() { v.ref-- if v.ref > 0 { @@ -71,6 +74,8 @@ func (v *version) releaseNB() { } else if v.ref < 0 { panic("negative version ref") } + + // 当 reference count 减少到 0 时，释放 version select { case v.s.relCh <- &vTask{vid: v.id, files: v.levels, created: time.Now()}: // We can use v.levels directly here since it is immutable. @@ -87,12 +92,14 @@ func (v *version) release() { v.s.vmu.Unlock() } +// 因为 version 上有当前的 SSTable 信息，所以一些使用 SSTable 进行查询的方法定义在了 version 上 func (v *version) walkOverlapping(aux tFiles, ikey internalKey, f func(level int, t *tFile) bool, lf func(level int) bool) { ukey := ikey.ukey() // Aux level. if aux != nil { for _, t := range aux { + // 注意：使用的是 user key 来判定是否和 SSTable 的 range 重叠 if t.overlaps(v.s.icmp, ukey, ukey) { if !f(-1, t) { return @@ -115,13 +122,17 @@ func (v *version) walkOverlapping(aux tFiles, ikey internalKey, f func(level int // Level-0 files may overlap each other. Find all files that // overlap ukey. for _, t := range tables { - if t.overlaps(v.s.icmp, ukey, ukey) { + if t.overlaps(v.s.icmp, ukey, ukey) { // 按照 ukey 来做判定 if !f(level, t) { return } } } } else { + + // Searches smallest index of tables whose its largest + // key is after or equal with given key. + // 找 SSTable 最小的 index，其 imax 大于等于 ikey，这个 SSTable 就是潜在的 table if i := tables.searchMax(v.s.icmp, ikey); i < len(tables) { t := tables[i] if v.s.icmp.uCompare(ukey, t.imin.ukey()) >= 0 { @@ -133,26 +144,29 @@ func (v *version) walkOverlapping(aux tFiles, ikey internalKey, f func(level int } if lf != nil && !lf(level) { + // lf 内部会判断是否在 level-0 上已经找到，如果已经找到的话，这里会 return，就不会继续往下找了 return } } } +// 在 SSTable 中查找 ikey +// v 上说明了本次查询所有的 SSTable func (v *version) get(aux tFiles, ikey internalKey, ro *opt.ReadOptions, noValue bool) (value []byte, tcomp bool, err error) { if v.closing { return nil, false, ErrClosed } ukey := ikey.ukey() - sampleSeeks := !v.s.o.GetDisableSeeksCompaction() + sampleSeeks := !v.s.o.GetDisableSeeksCompaction() // 如果 seek 的时候 miss 过多，需要出发 compaction var ( tset *tSet - tseek bool + tseek bool // 注：trigger seek compaction? // Level-0. - zfound bool - zseq uint64 + zfound bool // 标记 level-0 上已经找到了要找的 key，不可以继续往下找了，因为 level-0 上的数据肯定是最新的，如果继续往下找的话就可能找到旧的数据了 + zseq uint64 // 用来和找到的 seq number 做比较，当遇到更大的 seq number 时更新结果，使用离 snapshot 最近的结果 zkt keyType zval []byte ) @@ -161,78 +175,95 @@ func (v *version) get(aux tFiles, ikey internalKey, ro *opt.ReadOptions, noValue // Since entries never hop across level, finding key/value // in smaller level make later levels irrelevant. - v.walkOverlapping(aux, ikey, func(level int, t *tFile) bool { - if sampleSeeks && level >= 0 && !tseek { - if tset == nil { - tset = &tSet{level, t} - } else { - tseek = true + v.walkOverlapping(aux, ikey, + // 在一个 SSTable 内部查找，返回是否还需要继续到别的 SSTable 中搜索 + func(level int, t *tFile) bool { + if sampleSeeks && level >= 0 && !tseek { + if tset == nil { + tset = &tSet{level, t} + } else { + // 当 tSet 不为 nil 时，tSet 保存的是之前一个 SSTable，之前的 SSTable 没有找到这个 key，所以要 consume 一次 seekNumber + tseek = true + } } - } - var ( - fikey, fval []byte - ferr error - ) - if noValue { - fikey, ferr = v.s.tops.findKey(t, ikey, ro) - } else { - fikey, fval, ferr = v.s.tops.find(t, ikey, ro) - } + var ( + fikey, fval []byte + ferr error + ) + // 查询的是 internal key + if noValue { + fikey, ferr = v.s.tops.findKey(t, ikey, ro) + } else { + fikey, fval, ferr = v.s.tops.find(t, ikey, ro) + } - switch ferr { - case nil: - case ErrNotFound: - return true - default: - err = ferr - return false - } + switch ferr { + case nil: + case ErrNotFound: + // 在一个 SSTable 中没找到的话要继续寻找 + return true + default: + err = ferr + // 查询过程中出错，就结束查询，不继续在别的 SSTable 中查询 + return false + } - if fukey, fseq, fkt, fkerr := parseInternalKey(fikey); fkerr == nil { - if v.s.icmp.uCompare(ukey, fukey) == 0 { - // Level <= 0 may overlaps each-other. - if level <= 0 { - if fseq >= zseq { - zfound = true - zseq = fseq - zkt = fkt - zval = fval - } - } else { - switch fkt { - case keyTypeVal: - value = fval - err = nil - case keyTypeDel: - default: - panic("leveldb: invalid internalKey type") + if fukey, fseq, fkt, fkerr := parseInternalKey(fikey); fkerr == nil { + if v.s.icmp.uCompare(ukey, fukey) == 0 { + // Level <= 0 may overlaps each-other. + if level <= 0 { + if fseq >= zseq { + // 使用更大的 seq number，越大的 seq number 越新，越靠近 snapshot + zfound = true + zseq = fseq + zkt = fkt + zval = fval + } + } else { + switch fkt { + case keyTypeVal: + value = fval + err = nil + case keyTypeDel: + // 这里因为处于 level 0 之下的 level，一旦发现了 ukey，就不用继续找了，所以不存在更新结果的情况，value 不用给 reset 成 nil + // 只会赋值一次 + default: + panic("leveldb: invalid internalKey type") + } + // 在 level 0 以下的层，一旦发现了就不用继续向下找了 + return false } - return false } + } else { + err = fkerr + // 查询过程中出错，就结束查询，不继续在别的 SSTable 中查询 + return false } - } else { - err = fkerr - return false - } - return true - }, func(level int) bool { - if zfound { - switch zkt { - case keyTypeVal: - value = zval - err = nil - case keyTypeDel: - default: - panic("leveldb: invalid internalKey type") + return true + }, func(level int) bool { + // 遍历完 level-0 之后会调用这个函数，检查是否在 level-0 中已经发现了 key + // 如果 zfound，就要返回结果了，不可以继续往下找了，不然的话可能会找到下层 level 上的、旧的数据 + if zfound { + switch zkt { + case keyTypeVal: + value = zval + err = nil + case keyTypeDel: + default: + panic("leveldb: invalid internalKey type") + } + return false } - return false - } - return true - }) + // 对 level-0 之下的 level，调用是都返回 true，继续往下找 + // 所以 lf 这个函数只是对 zfound，“在 level-0 上是否找到”生效，用来终止向下的查找 + return true + }) + // 感觉这里 consume seek 的条件有点鲁莽，如果这个 key 本来就是不存在的，目前看来也会 consume 一次 seek + // 应该是最终这个 key 被找到了，然后对和它有 overlap 的 SSTable 都去 consume 一次 seek？ if tseek && tset.table.consumeSeek() <= 0 { tcomp = atomic.CompareAndSwapPointer(&v.cSeek, nil, unsafe.Pointer(tset)) } @@ -240,6 +271,7 @@ func (v *version) get(aux tFiles, ikey internalKey, ro *opt.ReadOptions, noValue return } +// 触发 ikey 所在 func (v *version) sampleSeek(ikey internalKey) (tcomp bool) { var tset *tSet @@ -279,10 +311,11 @@ func (v *version) newStaging() *versionStaging { // Spawn a new version based on this version. func (v *version) spawn(r *sessionRecord, trivial bool) *version { staging := v.newStaging() - staging.commit(r) + staging.commit(r) // 将变更记录、收集到 version staging 中 return staging.finish(trivial) } +// 把 version 中的所有 level 上的所有 SSTable，全量转化到 r.atRecord 中，作为需要 add 的 tables func (v *version) fillRecord(r *sessionRecord) { for level, tables := range v.levels { for _, t := range tables { @@ -327,6 +360,7 @@ func (v *version) offsetOf(ikey internalKey) (n int64, err error) { return } +// maxLevel 通常只用于 testing，prod 使用中取 0 func (v *version) pickMemdbLevel(umin, umax []byte, maxLevel int) (level int) { if maxLevel > 0 { if len(v.levels) == 0 { @@ -352,6 +386,7 @@ func (v *version) pickMemdbLevel(umin, umax []byte, maxLevel int) (level int) { return } +// 计算 compaction score func (v *version) computeCompaction() { // Precomputed best level for next compaction bestLevel := int(-1) @@ -377,8 +412,10 @@ func (v *version) computeCompaction() { // file size is small (perhaps because of a small write-buffer // setting, or very high compression ratios, or lots of // overwrites/deletions). + // 默认 CompactionL0Trigger=4, 当 L0 的文件数目超过 4 个时触发 L0 compcation score = float64(len(tables)) / float64(v.s.o.GetCompactionL0Trigger()) } else { + // 其他 Level，当数据规模过大时触发 compaction score = float64(size) / float64(v.s.o.GetCompactionTotalSize(level)) } @@ -404,15 +441,17 @@ func (v *version) needCompaction() bool { } type tablesScratch struct { - added map[int64]atRecord - deleted map[int64]struct{} + added map[int64]atRecord // key: file num + deleted map[int64]struct{} // key: file num } +// 描述下一个 version 相比于 base version 的变化，增加了哪些 SSTable，删掉了哪些 SSTable type versionStaging struct { base *version levels []tablesScratch } +// 之所以叫 "Scratch"，是因为我们需要收集 rec 形成这个 version staging，收集的过程像是 from the scratch func (p *versionStaging) getScratch(level int) *tablesScratch { if level >= len(p.levels) { newLevels := make([]tablesScratch, level+1) @@ -427,12 +466,14 @@ func (p *versionStaging) commit(r *sessionRecord) { for _, r := range r.deletedTables { scratch := p.getScratch(r.level) if r.level < len(p.base.levels) && len(p.base.levels[r.level]) > 0 { + // 只有 base version 真的有要删的这个 level，这里做一个简单的预 check if scratch.deleted == nil { scratch.deleted = make(map[int64]struct{}) } scratch.deleted[r.num] = struct{}{} } if scratch.added != nil { + // 如果是要删的，就不能继续在 add 里面了，预处理 delete(scratch.added, r.num) } } @@ -450,6 +491,7 @@ func (p *versionStaging) commit(r *sessionRecord) { } } +// 由 version staging 构建一个新的 version func (p *versionStaging) finish(trivial bool) *version { // Build new version. nv := newVersion(p.base.s) @@ -459,6 +501,7 @@ func (p *versionStaging) finish(trivial bool) *version { } nv.levels = make([]tFiles, numLevel) for level := 0; level < numLevel; level++ { + // 逐个 level 地 delete、add SSTables var baseTabels tFiles if level < len(p.base.levels) { baseTabels = p.base.levels[level] @@ -511,14 +554,18 @@ func (p *versionStaging) finish(trivial bool) *version { for _, r := range scratch.added { added = append(added, tableFileFromRecord(r)) } + if level == 0 { + // 在 level=0 上，added 按照 file number 降序排序 + // level-0 上的 SSTable 可能有 range 上的重叠，而且没有顺序性 added.sortByNum() index := nt.searchNumLess(added[len(added)-1].fd.Num) nt = append(nt[:index], append(added, nt[index:]...)...) } else { - added.sortByKey(p.base.s.icmp) + // 其他 level 上，SSTable 是按照 imin 排序的，imin 相同的话按照 num 升序排序； + added.sortByKey(p.base.s.icmp) // 按照 imin 升序排序 _, amax := added.getRange(p.base.s.icmp) - index := nt.searchMin(p.base.s.icmp, amax) + index := nt.searchMin(p.base.s.icmp, amax) // 找 nt 中第一个 imin 大于等于 amax 的 SSTable，于是 added 可以被 insert 到 index 前面 nt = append(nt[:index], append(added, nt[index:]...)...) } nv.levels[level] = nt @@ -552,6 +599,7 @@ func (p *versionStaging) finish(trivial bool) *version { nv.levels = nv.levels[:n] // Compute compaction score for new version. + // 当一次 compaction 产生了新的 version 时，立刻计算 new version 的 compaction score nv.computeCompaction() return nv