为什么学习KV引擎

阅读官方推荐的资料

阅读官方examples

找到官方使用的例子，查看这个项目是如何被使用的。

工业级实现badger

DB 初始化
读写事务
只读事务
范围查询
GC
LSM日志合并

源码安装项目

go get github.com/dgraph-io/badger/v3

写一个例子

package main
import (
"fmt"
badger "github.com/dgraph-io/badger/v3"
)


func main() {
//  打开DB


db, err := badger.Open(badger.DefaultOptions("/tmp/badger"))


defer db.Close()






// 读写事务


err = db.Update(func(txn *badger.Txn) error {


// Your code here…


txn.Set([]byte("answer"), []byte("42"))


txn.Get([]byte("answer"))


return nil


})


// 只读事务


err = db.View(func(txn *badger.Txn) error {


// Your code here…


txn.Get([]byte("answer_v1"))


return nil


})


// 遍历keys


err = db.View(func(txn *badger.Txn) error {


opts := badger.DefaultIteratorOptions


opts.PrefetchSize = 10


it := txn.NewIterator(opts)


defer it.Close()


for it.Rewind(); it.Valid(); it.Next() {


item := it.Item()


k := item.Key()


err := item.Value(func(v []byte) error {


fmt.Printf("key=%s, value=%s\n", k, v)


return nil


})


if err != nil {


return err


}


}


return nil


})


// vlog 的GC
err = db.RunValueLogGC(0.7)


_ = err


}

使用 devle 进行debug调试

DB初始化

1. 构造配置参数对象

2. 打开或创建工作目录并上锁

3. 打开或创建ManifestFile文件

4. 创建DB对象

1. 创建内存表列表,用于预写日志

2. 创建通知刷新任务的channel

3. 创建写入任务的channel

4. 配置对象

5. manifest文件对象

6. 目录锁

7. value目录锁

8. 创建oracle对象，用于并发事务的控制

5. 创建一个块缓存

6. 创建一个索引缓存

7. 开启key注册器

8. 打开memtable 并全部追加到 imm 的列表中

1. 创建一个内存跳表对象

2. 封装.mem文件为一个logFile对象取名为wal

3. 打开 .mem结尾的文件并关联为mmap文件

9. 创建一个激活的内存表

10. 创建一个level管理器

1. 创建一个tables的二维数组

2. 初始化每一个.sst结尾的文件关联为一个mmap文件

3. 根据manfist里的记录，初始化每一层的table对象

4. 如果是L0层则按fid排序，否则按每个表中最小key进行排序

11. vlog初始化

1. 打开一个DISCARD的文件并关联为mmap文件

12. 启动日志合并过程

13. 获取DB的事务最大版本号

14. 打开vlog文件, 关联mmap

1. 按fid排序处理vlog文件

2. 读取目录填充vlog file 的map

3. 如果没有vlog 文件则创建一个新的

4. 如果vlog文件是空的则删除

5. 获取vlog文件的实际大小并截断文件

6. 创建一个新的活跃的vlog文件

15. 开启负责处理写请求的工作协程

16. 启动vlog文件的GC过程

读写事务

1. 检查事务是否关闭

2. 创建一个事务对象

1. 创建一个存储哪些key存在冲突的记录map

2. 记录在当前事务上发生写入的key的列表

3. 分配一个事务读取时间戳

4. 获取最后一个已提交的时间戳

5. 创建一个用于wait回调的chan

6. 等待waitChan的回调，或者上下文的取消

7. 从 nextReadTnx中获取一个时间戳的值

8. 然后标记为开始,记录最后的事物时间戳后发送给一个markChan

9. 等待与这个时间戳冲突的事物提交完成

10. 如果是一个读写事务

3. Defer 丢弃最终的事务

1. 标记readTs时间戳已经完成

4. 执行闭包函数

1. 如果这是一个读写事务

2. 如果在pending数组里存在一个key，并且没有过期 则复制数据并返回

3. 检查kv的合法性

4. 检查当前事务的执行的命令数量以及存储大小是否超过阈值

5. 按照key计算一个hash值，然后加入冲突检查map中

6. 如果这个key 在当前事务中被写入过，并且与之前的版本不同，则计入重复写入的数组中

7. 将该key的写入操作记录到pending数组里

8. Set kv

9. Get k

5. 提交事务

1. 获取一把锁

2. 创建一个提交时间戳

3. 遍历每个在当前事务中写入的key，为其分配唯一的版本号

4. 遍历pending数组处理每一个实体kv对象

5. 追加一条标记事务结束的内部kv实体

6. 如果读时间戳大于已提交的时间戳则忽略

7. 读时间戳小于已提交的时间戳则判断是否存在读后写的情况存在就冲突

8. 是否存在冲突

9. 拿到当前事务的readTs标记其为完成

10. 清理已经提交的事务的记录数组

11. 标记当前的事务开始进行提交，分配了一个新的提交时间戳

12. 将当前提交事务的时间戳和冲突的key组成对象记录在已提交事务的数组中

13. 如果值日志offset大于vlog最大文件大小，或者写入条数超购阈值

14. 写入磁盘 如果设置了同步则会系统调用sync 最后根据offset进行截断

15. 创建一个新的vlog文件

16. 等待批量写请求处理完成

17. 标记事务已经提交完成

18. 将buf写入到mmap关联的文件内存中

19. 创建一个写入请求

20. 发送给写者channel中

21. 返回一个回调函数

22. 写入磁盘

23. 检查 pending数组是否为空，如果为空则直接返回

24. 事务提交的前置检查

25. 提交并发送日志返回一个回调函数进行等待

WaterMark 实现 ->  badger 事务的设计

doWrite实现

1. 读取writeCh中的req

2. Select writeCh 批量拼装 reqs

3. 如果没有请求，则pending阻塞住，等待上一次批量写入的完成

4. 开启一个协程处理此次写请求

1. 检查当前的值是否超过了值的阈值

2. 没有超过则 直接将值存储到mt中

3. 超过则 将值的指针存储到mt中 指针指向vlog中的位置

4. 编码kv数据

5. 直接copy数据到mmap内存上

6. kv数据在文件中的编码是 head，key，value，crc

7. 其中head 包括 keyLen，valueLen, expiresAt, meta，userMeta

8. 先写入wal中，用于日志恢复

9. 真正存储到跳表中

10. 检查写入的vlaue是否合法值是否足够写入4G大小的文件，因为值指针是uin32

11. 获取vlog的文件锁，获取最大的文件id，最大的也是活跃的

12. 拿到句柄后遍历reqlist

13. 解析每个请求的值查看大小是否超过分离的阈值

14. 拿到vlog当前的offset值

15. 用fid和offset拼接一个值的指针

16. 编码kv数据写入buf

17. 整个写入vlog文件

18. 分req写入LSM中

5. 当该doWrite关闭后，先读取所有writeCh中的请求后最后处理一次写入请求

只读事务

1. 与读写事务的区别 就是只获取了读时间戳，而不需要提交事务

2. 检查key是否合法

3. 创建一个itme对象

4. 如果是一个读写事务 则从pending数组中找是否有刚刚被修改的key

1. 如果命中则直接返回

2. 记录当前的读取的key到reads数组中

5. 以readTs拼接时间戳版本号

6. 获取内存表列表

7. 遍历内存表从跳表中get key信息，并且其最大版本号小于等于查询版本时直接返回

8. 如果没有则从level中查询

1. 从某一层获取数据

2. 判断此table是否存在此key，使用boolFilter

3. 创建一个迭代器对象

4. Seek 这个key

5. 查找block中的key

6. 获取一个idx块 如果索引块缓存中没有 就mmap中获取

7. 更新索引快到缓存中

8. 判断是向前迭代还是向后迭代

9. 置位指针是否从头迭代

10. 二分查找确定key在哪一个block上

11. 从索引块中获得此key所在的bolckIdx的位置

12. 然后先从缓存中加载块缓存的地址缓存的key是fid+offset组成

13. 没有则从mmap映射的文件句柄中获取并加载到内存后返

14. 解析当前block的元数据

15. 二分法搜索这个block

16. 解析basekey，前缀压缩

17. 基于offset进行编解码，找到对应的key，value字节数据

18. 如果是0层则倒序获取所有的table对象

19. 对于非0层则通过二分法比较每个表的最大key来至多确定一个table对象

20. 获取本层中sstable句柄

21. 用不带有时间戳的key计算hash值

22. 遍历tables

23. 比较value的版本，返回最大版本号的value对象

24. 逐层遍历

9. 拼装数据返回结果item

遍历keys

1. 创建一个默认的迭代器对象

2. 配置预读kv条数为10

3. 创建一个事务层迭代器对象

1. 创建一个merge迭代器对象，配置其迭代方向

2. 将迭代器数组构建为一个颗二叉树

3. 遍历每一个level中的table创建迭代器

4. 判断如果没有指定前缀则默认都需要遍历

5. 如果指定了前缀则判断前缀是否在此table中如果不在则不用创建迭代器

6. 如果是L0层则所有的table都要创建迭代器

7. 如果是非0层则先判断是否需要前缀迭代需要的话则按前缀过滤

8. 使用boolfilter过滤掉一定不存在前缀key的table

9. 如果不是读写事务，或者PendingWrites长度为0，则直接返回

10. 否则对PendingWrites数组返回排序的切片，并返回迭代器

11. 获取内存表列表

12. 给当前vlog对象一个迭代器引用计数表示当前迭代器

13. 创建一个迭代器数组

14. 创建一个对PendingWrites数组的迭代器，并指定是正向还是反向

15. 为每个内存表创建一个迭代器并append到迭代器数组中

16. 为每个level创建一个迭代器

17. 将table按遍历方向进行反转后返回

18. 根据迭代器数组创建一个合并迭代器

4. seek(nil) 将游标移动到最初始位置,遍历的，迭代性的调研底层迭代器的seek

5. 预先读取几个kv到迭代器的链表中

1. 跳过内部key

2. 跳过版本大于readTs的key，说明这是在此迭代器开始之后写入的key不能被读取

3. 如果指定返回所有版本的key则无论迭代到什么key都直接返回

4. 判断是否是前向遍历，判断与最后一次返回的key是否是同一个是则忽略，并立即更新最好访问的key，以保证快照在del情况下的一致性

6. 解析value的值，并从中判断是否过期与删除

7. 创建一个item对象，并填充

1. 如果值是一个值指针，则创建一个协程异步的读取vlog文件

2. 值指针由fid+offset+len组成，管理好vlog文件分段的读写锁

3. 然后上锁去读取mmap中的数据

8. 放到预先读取的list上，直到读取了足够的预取内容

9. 检查itme是否合法，如果合法就是说item不为空，则调用next方法再次执行上述步骤

vlog 的GC

1. 获取一个vlog文件并执行GC

1. 选择包含最多可丢弃数据的vlog文件

2. 遍历所有的槽位置，比较每一个vlog文件可丢弃数据数量，返回最大的fid

3. 拿到fid后更新这个vlog文件的可丢弃数量

4. 判断当前fid是否被删除，删除的话更新并重新获取

5. 判断当前的fid存在的kv中的可丢弃数据大于阈值

6. 大于则返回

7. 如果有其他GC任务执行则返回错误

8. 选择一个vlog文件

9. 当这个文件进行GC

2. 重写当前的vlog文件

1. 从头遍历该文件的kv

2. 拿到key后去lsm中查询是否被删除

3. 如果没有被删除，则将最新版本的值写入到wb中

4. wb达到阈值后批量的再次写入DB中

5. 判断当前文件上是否还有引用存在，没有的话就可以从map中删除了

6. 然后判断是否可以立即将文件物理删除

3. 更新GC统计结果

LSM日志合并

1. 启动合并协程

1. 随机延迟1s执行

2. 每10s 2号协程序 执行一次

3. 执行合并逻辑

4. 计算目标层，其原理是保证每一层的存储大小为上一层的一个量级

5. 目标层会是Lbase层，比如L0层在DB为空时会直接压缩到L6层

6. 对合并的table 合理的拆分为多个可并行的子压缩

7. 真正的合并表返回新

8. 将其写入清淡文件

9. 对新表与旧表进行替换

10. 删除旧表

11. 打印日志

12. 对合并的表创建一个迭代器

13. 使用迭代器遍历所有合并的表

14. 然后写入buffer注意分割

15. 将discard写入统计中

16. 先尝试L0到Lbase的压缩

17. 失败了会尝试l0到l0的压缩过程

18. 对压缩过程上锁

19. 计算L0层中存在数据重合的sstable

20. 计算Lbase层与L0层的sstable 列表重合范围有重合的sstable

21. 将这些tables 全部加入compactStatus 的tables 的map中

22. 计算L0层有重合的sstable

23. 然后将其加入compactStatus 的tables 的map中

24. 如果当前表在进行压缩则直接忽略执行

25. 创建一个压缩任务

26. 如果是0压缩，那就执行l0的特殊压缩策略

27. 运行压缩任务

28. 启动4个协程进行压缩

2. 启动flush 内存表协程

1. 迭代表结构，然后获取key判断是否写入buffer

2. 将buffer数据写入创建的新表中

3. 更新l0层的清单文件

4. 接受flushChan的消息，并且做一个聚合操作

5. 对一段时间内的多个内存表进行一个合并

6. 创建一个迭代器然后处理合并过程

总结反思

1. 学习官方资料是什么，有什么意义，怎么用，设计原理是什么，存在什么局限

2. 将项目跑起来，可以进行Debug

3. 结合官方demo或者单元测试，构建一个使用场景，进行Debug阅读

4. 第一遍先BFS阅读，第二遍BFS+DFS阅读，第三遍DFS阅读

5. 对于无法debug的问题，通过IDE阅读，从单元测试入手寻找代码位置

6. 绘制思维导图或者流程，架构图，对核心数据结构可以绘制UML图

7. 总结与反思

优化与挑战

1. 热点识别与更新

2. 动态参数的调整

3. 这一期视频的不足之处

4. 阅读源码后我们做什么？维护开源社区，贡献代码

参考资料

1. https://github.com/go-delve/delve

2. Sourcegraph - Universal Code Search

3. WiscKey：在SSD存储上的键值分离设计

4. LSM-论文导读与Leveldb源码解读