Badger源码导读
源码分析入口基准案例
先从Badger的基本使用入手
func main() {
// 打开db
db, _ := badger.Open(badger.DefaultOptions("tmp/badger"))
defer db.Close()
// 读写事务
err := db.Update(func(txn *badger.Txn) error {
txn.Set([]byte("answer"), []byte("42"))
txn.Get([]byte("answer"))
return nil
})
// 只读事务
err = db.View(func(txn *badger.Txn) error {
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(val []byte) error {
fmt.Printf("key=%s, value=%s\n", k, val)
return nil
})
if err != nil {
return err
}
}
return nil
})
err = db.RunValueLogGC(0.7)
_ = err
}
DB初始化过程
初始化参数
badger.open()传入的是一个option,先看一下option结构体的字段都有哪些
type Options struct {
// Required options.
// Dir: Badger是KV分离的存储引擎,Dir位置存储的是 Key 和指向Value的逻辑指针
// ValueDir: 存储的是Value日志,即值所在的地址,默认情况下Dir和ValueDir在同一个path目录下
Dir string
ValueDir string
// Usually modified options.
// SyncWrites: 同步写,即写入的时候主动同步到磁盘(mmap不会立即刷盘)
SyncWrites bool
NumVersionsToKeep int
// ReadOnly: 如其名,是否设置为只读
ReadOnly bool
// Logger: 如其名,log对象
Logger Logger
// Compression: 压缩归并的级别
Compression options.CompressionType
// InMemory: 是否只基于内存
InMemory bool
MetricsEnabled bool
// Sets the Stream.numGo field
NumGoroutines int
// Fine tuning options.
// MemTableSize: 内存表的尺寸限制
MemTableSize int64
BaseTableSize int64
BaseLevelSize int64
LevelSizeMultiplier int
TableSizeMultiplier int
// MaxLevels: 最大容忍的level级别,LSM-T的级数L0-L(max-1)
MaxLevels int
VLogPercentile float64
// ValueThreshold: 值大小的阈值,如果Value的大小不超过这个设定值,则不会将KV进行分离
// 此处是在工业实践中的一种权衡,KV分离会造成不可避免的读放大
// (两次的随机读,先在LSM-T中读取一次指针,再通过指针从ValueLog中读取一次值)
ValueThreshold int64
// NumMemtables: 内存表的数量
NumMemtables int
// Changing BlockSize across DB runs will not break badger. The block size is
// read from the block index stored at the end of the table.
// BlockSize: 每个block的大小(sst由block和index等组成)
BlockSize int
// BloomFalsePositive: 布隆过滤器假阳性的比例
BloomFalsePositive float64
// BlockCacheSize: 块缓存的大小
BlockCacheSize int64
// IndexCacheSize: 索引缓存的大小
IndexCacheSize int64
NumLevelZeroTables int
NumLevelZeroTablesStall int
// ValueLogFileSize: 存储值的Valuelog文件的最大大小
ValueLogFileSize int64
// ValueLogMaxEntries: 存储值的Valuelog文件的最大键值对数量
ValueLogMaxEntries uint32
// NumCompactors: 日志合并压缩协程同时运行的最大数量
NumCompactors int
CompactL0OnClose bool
LmaxCompaction bool
ZSTDCompressionLevel int
// When set, checksum will be validated for each entry read from the value log file.
// VerifyValueChecksum: 是否进行参数校验值的检查
VerifyValueChecksum bool
// Encryption related options.
// EncryptionKey: 加密字段
EncryptionKey []byte // encryption key
// EncryptionKeyRotationDuration: 加密字段有效时长
EncryptionKeyRotationDuration time.Duration // key rotation duration
// BypassLockGuard will bypass the lock guard on badger. Bypassing lock
// guard can cause data corruption if multiple badger instances are using
// the same directory. Use this options with caution.
BypassLockGuard bool
// ChecksumVerificationMode decides when db should verify checksums for SSTable blocks.
ChecksumVerificationMode options.ChecksumVerificationMode
// DetectConflicts determines whether the transactions would be checked for
// conflicts. The transactions can be processed at a higher rate when
// conflict detection is disabled.
// DetectConflicts: 事务的冲突检测
DetectConflicts bool
// NamespaceOffset specifies the offset from where the next 8 bytes contains the namespace.
NamespaceOffset int
// Transaction start and commit timestamps are managed by end-user.
// This is only useful for databases built on top of Badger (like Dgraph).
// Not recommended for most users.
managedTxns bool
// 4. Flags for testing purposes
// ------------------------------
// 有关批处理的参数
maxBatchCount int64 // max entries in batch
maxBatchSize int64 // max batch size in bytes
maxValueThreshold float64
}
传入指定的路径，并默认配置信息，如果有需要更改的信息可以使用 WithX() 方法（此处使用了建造者模式）
badger.DefaultOptions("tmp/badger")
// DefaultOptions sets a list of recommended options for good performance.
// Feel free to modify these to suit your needs with the WithX methods.
func DefaultOptions(path string) Options {
return Options{
Dir: path,
ValueDir: path,
MemTableSize: 64 << 20,
BaseTableSize: 2 << 20,
BaseLevelSize: 10 << 20,
TableSizeMultiplier: 2,
LevelSizeMultiplier: 10,
MaxLevels: 7,
NumGoroutines: 8,
MetricsEnabled: true,
NumCompactors: 4, // Run at least 2 compactors. Zero-th compactor prioritizes L0.
NumLevelZeroTables: 5,
NumLevelZeroTablesStall: 15,
NumMemtables: 5,
BloomFalsePositive: 0.01,
BlockSize: 4 * 1024,
SyncWrites: false,
NumVersionsToKeep: 1,
CompactL0OnClose: false,
VerifyValueChecksum: false,
Compression: options.Snappy,
BlockCacheSize: 256 << 20,
IndexCacheSize: 0,
// The following benchmarks were done on a 4 KB block size (default block size). The
// compression is ratio supposed to increase with increasing compression level but since the
// input for compression algorithm is small (4 KB), we don't get significant benefit at
// level 3.
// NOTE: The benchmarks are with DataDog ZSTD that requires CGO. Hence, no longer valid.
// no_compression-16 10 502848865 ns/op 165.46 MB/s -
// zstd_compression/level_1-16 7 739037966 ns/op 112.58 MB/s 2.93
// zstd_compression/level_3-16 7 756950250 ns/op 109.91 MB/s 2.72
// zstd_compression/level_15-16 1 11135686219 ns/op 7.47 MB/s 4.38
// Benchmark code can be found in table/builder_test.go file
ZSTDCompressionLevel: 1,
// Nothing to read/write value log using standard File I/O
// MemoryMap to mmap() the value log files
// (2^30 - 1)*2 when mmapping < 2^31 - 1, max int32.
// -1 so 2*ValueLogFileSize won't overflow on 32-bit systems.
ValueLogFileSize: 1<<30 - 1,
ValueLogMaxEntries: 1000000,
VLogPercentile: 0.0,
ValueThreshold: maxValueThreshold,
Logger: defaultLogger(INFO),
EncryptionKey: []byte{},
EncryptionKeyRotationDuration: 10 * 24 * time.Hour, // Default 10 days.
DetectConflicts: true,
NamespaceOffset: -1,
}
}
Open函数(核心)
badger.Open(opt) 函数
此方法代码过长,在此只保留核心部分代码,部分逻辑将以伪代码或注释表示,并省去部分错误处理逻辑
func Open(opt Options) (*DB, error) {
// 检查参数
checkAndSetOptions(&opt)
// 创建了三个目录锁,防止其他进程注册到同一个目录造成冲突
var dirLockGuard, valueDirLockGuard *directoryLockGuard
// Create directories and acquire lock on it only if badger is not running in InMemory mode.
// We don't have any directories/files in InMemory mode so we don't need to acquire
// any locks on them.
// 判断参数配置为只基于内存
if !opt.InMemory {
// 创建目录
createDirs(opt)
var err error
if !opt.BypassLockGuard {
// 给Dir加目录锁
dirLockGuard, _ = acquireDirectoryLock(opt.Dir, lockFile, opt.ReadOnly)
// 方法末尾释放锁
defer func() {
if dirLockGuard != nil {
_ = dirLockGuard.release()
}
}()
// 获取Key&ValuePtr的绝对路径
absDir, _ := filepath.Abs(opt.Dir)
// 获取ValueLog的绝对路径
absValueDir, _ := filepath.Abs(opt.ValueDir)
// 如果ValueDir和Dir不相同,需要各自加锁
if absValueDir != absDir {
// 给ValueDir加目录锁
valueDirLockGuard, _ = acquireDirectoryLock(opt.ValueDir, lockFile, opt.ReadOnly)
// 释放锁
defer func() {
if valueDirLockGuard != nil {
_ = valueDirLockGuard.release()
}
}()
}
}
}
// 打开或创建Manifest文件,(采用mmap方式打开,在后面详细展开)
manifestFile, manifest, _ := openOrCreateManifestFile(opt)
// 关闭Manifest文件
defer func() {
if manifestFile != nil {
_ = manifestFile.close()
}
}()
// 创建内存中的db数据结构
db := &DB{
// memtable, 因为有多个,所以要创建数组
imm: make([]*memTable, 0, opt.NumMemtables),
// 刷新请求的channel
flushChan: make(chan flushTask, opt.NumMemtables),
// 写请求的channel
writeCh: make(chan *request, kvWriteChCapacity),
// 配置信息opt
opt: opt,
// 刚初始化好的manifest实例
manifest: manifestFile,
// Key&ValuePtr目录锁
dirLockGuard: dirLockGuard,
// Value目录锁
valueDirGuard: valueDirLockGuard,
// Oracle的实例,一个KV引擎并发事务的管理器,负责分配事务的版本号,用来实现MVCC功能,在读写事务时详细展开
orc: newOracle(opt),
pub: newPublisher(),
allocPool: z.NewAllocatorPool(8),
bannedNamespaces: &lockedKeys{keys: make(map[uint64]struct{})},
threshold: initVlogThreshold(&opt),
}
// Cleanup all the goroutines started by badger in case of an error.
// 关闭badger的所有任务协程的钩子函数
defer func() {
if err != nil {
opt.Errorf("Received err: %v. Cleaning up...", err)
db.cleanup()
db = nil
}
}()
// 块缓存相关配置
// LSM-T结构中SST里面数据是以块(block)为单位分割的
// 当开启块缓存之后,LSM-T会把最近被访问到的高热的块缓存在内存中,以加块响应速度
if opt.BlockCacheSize > 0 {
// 缓存不在此次源码阅读的讨论范围之内,不影响核心功能,暂且略过
// 值得一提的是badger是使用的缓存是badger社区研发的一个高性能本地并发缓存的库,有兴趣的同学可以自行研究
numInCache := opt.BlockCacheSize / int64(opt.BlockSize)
if numInCache == 0 {
// Make the value of this variable at least one since the cache requires
// the number of counters to be greater than zero.
numInCache = 1
}
config := ristretto.Config{
NumCounters: numInCache * 8,
MaxCost: opt.BlockCacheSize,
BufferItems: 64,
Metrics: true,
OnExit: table.BlockEvictHandler,
}
db.blockCache, err = ristretto.NewCache(&config)
if err != nil {
return nil, y.Wrap(err, "failed to create data cache")
}
}
// 索引缓存相关配置
// 索引是每个Key所对应的偏离量的值,每一个SSTable有一个元数据块即索引块
// 可以方便对Key的二分查找,定位当前的key在哪一个sstable文件里,在文件中的偏移量是多少
if opt.IndexCacheSize > 0 {
// Index size is around 5% of the table size.
indexSz := int64(float64(opt.MemTableSize) * 0.05)
numInCache := opt.IndexCacheSize / indexSz
if numInCache == 0 {
// Make the value of this variable at least one since the cache requires
// the number of counters to be greater than zero.
numInCache = 1
}
config := ristretto.Config{
NumCounters: numInCache * 8,
MaxCost: opt.IndexCacheSize,
BufferItems: 64,
Metrics: true,
}
db.indexCache, err = ristretto.NewCache(&config)
if err != nil {
return nil, y.Wrap(err, "failed to create bf cache")
}
}
// 对缓存模块的监控检测
db.closers.cacheHealth = z.NewCloser(1)
go db.monitorCache(db.closers.cacheHealth)
// 如果仅基于内存
if db.opt.InMemory {
// 默认关闭写同步
db.opt.SyncWrites = false
// If badger is running in memory mode, push everything into the LSM Tree.
// 把所有数据只写在LSM-T中
db.opt.ValueThreshold = math.MaxInt32
}
// Key的注册,与并发事务相关,之后再详细展开
krOpt := KeyRegistryOptions{
ReadOnly: opt.ReadOnly,
Dir: opt.Dir,
EncryptionKey: opt.EncryptionKey,
EncryptionKeyRotationDuration: opt.EncryptionKeyRotationDuration,
InMemory: opt.InMemory,
}
db.registry, _ = OpenKeyRegistry(krOpt)
// 计算消耗的内存等数据统计信息
db.calculateSize()
db.closers.updateSize = z.NewCloser(1)
go db.updateSize(db.closers.updateSize)
// 打开一个memTable实例
// memtable是在内存中的一个复杂数据结构
if err := db.openMemTables(db.opt); err != nil {
return nil, y.Wrapf(err, "while opening memtables")
}
// 检查
if !db.opt.ReadOnly {
// 创建一个新的.mem文件
// .mem文件就是LSM-T中的预写日志文件(wal)
if db.mt, err = db.newMemTable(); err != nil {
return nil, y.Wrapf(err, "cannot create memtable")
}
}
// newLevelsController potentially loads files in directory.
// 创建内存中level管理器
// LSM-T是分层结构的, LevelsController实例负责维护整个层级结构
// 进行日志归并,压缩处理等操作,通过Manifest进行初始配置
// 或者是,manifest文件就是LevelController持久化之后的ondisk版本,可以加快badger的恢复重启速度
// 先打开SSTable,加载索引块,元数据块,缓存到内存当中
if db.lc, err = newLevelsController(db, &manifest); err != nil {
return db, err
}
// Initialize vlog struct.
// 初始化vlog
db.vlog.init(db)
if !opt.ReadOnly {
// 启动日志归并的工作协程,后续再展开
db.closers.compactors = z.NewCloser(1)
db.lc.startCompact(db.closers.compactors)
db.closers.memtable = z.NewCloser(1)
go func() {
_ = db.flushMemtable(db.closers.memtable) // Need levels controller to be up.
}()
// Flush them to disk asap.
for _, mt := range db.imm {
db.flushChan <- flushTask{mt: mt}
}
}
// We do increment nextTxnTs below. So, no need to do it here.
// 拿到启动时最大事务的版本号(时间戳)
db.orc.nextTxnTs = db.MaxVersion()
db.opt.Infof("Set nextTxnTs to %d", db.orc.nextTxnTs)
// 真正打开vlog文件
if err = db.vlog.open(db); err != nil {
return db, y.Wrapf(err, "During db.vlog.open")
}
// Let's advance nextTxnTs to one more than whatever we observed via
// replaying the logs.
// 事务相关,等待之前事务的恢复
db.orc.txnMark.Done(db.orc.nextTxnTs)
// In normal mode, we must update readMark so older versions of keys can be removed during
// compaction when run in offline mode via the flatten tool.
db.orc.readMark.Done(db.orc.nextTxnTs)
// 事务号自增
db.orc.incrementNextTs()
// 监听配置信息的更改
go db.threshold.listenForValueThresholdUpdate()
// 从数据库中检索被禁止的命名空间并更新内存结构(非重点)
if err := db.initBannedNamespaces(); err != nil {
return db, errors.Wrapf(err, "While setting banned keys")
}
// 启动处理磁盘写请求的协程
// badger的写任务是并发写任务,可以充分发挥ssd的性能
db.closers.writes = z.NewCloser(1)
go db.doWrites(db.closers.writes)
if !db.opt.InMemory {
// 真正开启vlog的GC, 后面再详细讲解
db.closers.valueGC = z.NewCloser(1)
go db.vlog.waitOnGC(db.closers.valueGC)
}
// 监听协程(非重点)
db.closers.pub = z.NewCloser(1)
go db.pub.listenForUpdates(db.closers.pub)
// 释放锁
valueDirLockGuard = nil
dirLockGuard = nil
manifestFile = nil
// 返回db
return db, nil
}
创建Manifest文件
openOrCreateManifestFile(opt) 函数
func openOrCreateManifestFile(opt Options) (ret *manifestFile, result Manifest, err error) {
// 如果Inmemory则返回空的Manifest
if opt.InMemory {
return &manifestFile{inMemory: true}, Manifest{}, nil
}
return helpOpenOrCreateManifestFile(opt.Dir, opt.ReadOnly, manifestDeletionsRewriteThreshold)
}
func helpOpenOrCreateManifestFile(dir string, readOnly bool, deletionsThreshold int) (*manifestFile, Manifest, error) {
// 拼接path
path := filepath.Join(dir, ManifestFilename)
var flags y.Flags
if readOnly {
flags |= y.ReadOnly
}
// 尝试打开文件
fp, err := y.OpenExistingFile(path, flags) // We explicitly sync in addChanges, outside the lock.
if err != nil {
// 校验文件是否存在
if !os.IsNotExist(err) {
return nil, Manifest{}, err
}
// 如果仅读则无法创建直接返回
if readOnly {
return nil, Manifest{}, fmt.Errorf("no manifest found, required for read-only db")
}
// 真正创建manifest实例
m := createManifest()
// 覆盖写,执行完此条语句后就可以在目录中看到MANIFEST文件存在了(此时MANIFEST文件中仅有魔数bdg)
fp, netCreations, _ := helpRewrite(dir, &m)
// 断言,确保创建成功
y.AssertTrue(netCreations == 0)
// 创建manifestFile实例在内存中保存信息
mf := &manifestFile{
fp: fp,
directory: dir,
manifest: m.clone(),
deletionsRewriteThreshold: deletionsThreshold,
}
return mf, m, nil
}
// 文件存在加载恢复的逻辑暂不展开
......
}
func createManifest() Manifest {
levels := make([]levelManifest, 0)
return Manifest{
Levels: levels,
Tables: make(map[uint64]TableManifest),
}
// Tables: map[uint64]TableManifest
// uint64: 行号,第n个level
}
打开Memtable
memtable 结构体
// memTable structure stores a skiplist and a corresponding WAL. Writes to memTable are written
// both to the WAL and the skiplist. On a crash, the WAL is replayed to bring the skiplist back to
// its pre-crash form.
type memTable struct {
sl *skl.Skiplist
wal *logFile
maxVersion uint64
opt Options
buf *bytes.Buffer
}
openMemTables(opt) 方法
func (db *DB) openMemTables(opt Options) error {
// We don't need to open any tables in in-memory mode.
// 如果是只基于内存则直接返回(那我走?)
if db.opt.InMemory {
return nil
}
// 读取目录中的全部文件
files, _ := ioutil.ReadDir(db.opt.Dir)
var fids []int
// 遍历目录中的文件
for _, file := range files {
// 检查当前文件名是否包含一个.mem的后缀(在第一次初始化过程中肯定不会存在)
// 此时目录中应有的文件为 LOCK MANIFEST KEYREGISTRY
if !strings.HasSuffix(file.Name(), memFileExt) {
continue
}
// 如果有.mem文件,则取文件的命名转为int值作为fid
// 例: 000001.mem 000002.mem
fsz := len(file.Name())
fid, _ := strconv.ParseInt(file.Name()[:fsz-len(memFileExt)], 10, 64)
fids = append(fids, int(fid))
}
// Sort in ascending order.
// 按照fid排序
sort.Slice(fids, func(i, j int) bool {
return fids[i] < fids[j]
})
// 按照fid顺序遍历
for _, fid := range fids {
flags := os.O_RDWR
if db.opt.ReadOnly {
flags = os.O_RDONLY
}
// 真正的打开.mem文件,采用mmap方式加载.mem文件中的数据
mt, err := db.openMemTable(fid, flags)
if err != nil {
return y.Wrapf(err, "while opening fid: %d", fid)
}
// If this memtable is empty we don't need to add it. This is a
// memtable that was completely truncated.
if mt.sl.Empty() {
mt.DecrRef()
continue
}
// These should no longer be written to. So, make them part of the imm.
db.imm = append(db.imm, mt)
}
// 设置最新的fid序列号
if len(fids) != 0 {
db.nextMemFid = fids[len(fids)-1]
}
db.nextMemFid++
return nil
}
创建Memtable
newMemTable() 方法
func (db *DB) newMemTable() (*memTable, error) {
// 真正创建.mem文件
mt, err := db.openMemTable(db.nextMemFid, os.O_CREATE|os.O_RDWR)
if err == z.NewFile {
db.nextMemFid++
return mt, nil
}
if err != nil {
db.opt.Errorf("Got error: %v for id: %d\n", err, db.nextMemFid)
return nil, y.Wrapf(err, "newMemTable")
}
return nil, errors.Errorf("File %s already exists", mt.wal.Fd.Name())
}
openMemTable(fid, flags) 方法
func (db *DB) openMemTable(fid, flags int) (*memTable, error) {
// 拼接路径
filepath := db.mtFilePath(fid)
// 创建memtable中的skiplist
s := skl.NewSkiplist(arenaSize(db.opt))
// 创建memtable实例
mt := &memTable{
sl: s,
opt: db.opt,
buf: &bytes.Buffer{},
}
// We don't need to create the wal for the skiplist in in-memory mode so return the mt.
// 如果只基于内存,则不需要创建wal文件,直接返回
if db.opt.InMemory {
return mt, z.NewFile
}
// 创建wal文件实例
mt.wal = &logFile{
fid: uint32(fid),
path: filepath,
registry: db.registry,
writeAt: vlogHeaderSize,
opt: db.opt,
}
// 调用系统函数创建wal文件
lerr := mt.wal.open(filepath, flags, 2*db.opt.MemTableSize)
// 如果未成功创建新文件或其他失败则返回err
if lerr != z.NewFile && lerr != nil {
return nil, y.Wrapf(lerr, "While opening memtable: %s", filepath)
}
// Have a callback set to delete WAL when skiplist reference count goes down to zero. That is,
// when it gets flushed to L0.
// 用来关闭的回调函数
s.OnClose = func() {
if err := mt.wal.Delete(); err != nil {
db.opt.Errorf("while deleting file: %s, err: %v", filepath, err)
}
}
// 成功创建mmap则返回 lerr (z.NewFile)
if lerr == z.NewFile {
return mt, lerr
}
// 当且仅当MemTableSize设置为0时造成 lerr == nil的适合执行到此
// 此时mmap未进行截断,在UpdateSkipList()中遍历wal文件并重新截断,如果wal文件不存在会返回错误
err := mt.UpdateSkipList()
return mt, y.Wrapf(err, "while updating skiplist")
}
创建levelController
newLevelsController(db, mf) 函数
func newLevelsController(db *DB, mf *Manifest) (*levelsController, error) {
// 断言,进行一些必要的校验
y.AssertTrue(db.opt.NumLevelZeroTablesStall > db.opt.NumLevelZeroTables)
// 关联db实例,创建level数组对应层级关系(例:levels[0] => L0层)
// levelHandler就是真正负责某一层sst管理器的主要操作
s := &levelsController{
kv: db,
levels: make([]*levelHandler, db.opt.MaxLevels),
}
// 状态统计的一个的对象(set结构),key为fid,用以判断对应的fid是否存在于这一层
s.cstatus.tables = make(map[uint64]struct{})
// 合并状态的信息
s.cstatus.levels = make([]*levelCompactStatus, db.opt.MaxLevels)
// 按层遍历,每一层都创建一个levelhandler实例
for i := 0; i < db.opt.MaxLevels; i++ {
s.levels[i] = newLevelHandler(db, i)
s.cstatus.levels[i] = new(levelCompactStatus)
}
// 基于内存,那我走?🤡
if db.opt.InMemory {
return s, nil
}
// Compare manifest against directory, check for existent/non-existent files, and remove.
// 对manifest文件进行校验
if err := revertToManifest(db, mf, getIDMap(db.opt.Dir)); err != nil {
return nil, err
}
var mu sync.Mutex
tables := make([][]*table.Table, db.opt.MaxLevels)
var maxFileID uint64
// We found that using 3 goroutines allows disk throughput to be utilized to its max.
// Disk utilization is the main thing we should focus on, while trying to read the data. That's
// the one factor that remains constant between HDD and SSD.
// 一种针对并发控制的负载均衡策略,对于ssd来说,创建3个协程能够最大的发挥ssd的优点
throttle := y.NewThrottle(3)
start := time.Now()
var numOpened int32
// 创建一个定时触发器进行超时控制
tick := time.NewTicker(3 * time.Second)
// 钩子函数关闭定时器
defer tick.Stop()
// manifest清单文件的Tables
// 拿到每个table对应的fid
// 第一次初始化的适合因为Tables为空,会直接跳过
for fileID, tf := range mf.Tables {
fname := table.NewFilename(fileID, db.opt.Dir)
select {
case <-tick.C:
db.opt.Infof("%d tables out of %d opened in %s\n", atomic.LoadInt32(&numOpened),
len(mf.Tables), time.Since(start).Round(time.Millisecond))
default:
}
if err := throttle.Do(); err != nil {
closeAllTables(tables)
return nil, err
}
if fileID > maxFileID {
maxFileID = fileID
}
go func(fname string, tf TableManifest) {
var rerr error
defer func() {
throttle.Done(rerr)
atomic.AddInt32(&numOpened, 1)
}()
dk, err := db.registry.DataKey(tf.KeyID)
if err != nil {
rerr = y.Wrapf(err, "Error while reading datakey")
return
}
topt := buildTableOptions(db)
// Explicitly set Compression and DataKey based on how the table was generated.
topt.Compression = tf.Compression
topt.DataKey = dk
mf, err := z.OpenMmapFile(fname, db.opt.getFileFlags(), 0)
if err != nil {
rerr = y.Wrapf(err, "Opening file: %q", fname)
return
}
t, err := table.OpenTable(mf, topt)
if err != nil {
if strings.HasPrefix(err.Error(), "CHECKSUM_MISMATCH:") {
db.opt.Errorf(err.Error())
db.opt.Errorf("Ignoring table %s", mf.Fd.Name())
// Do not set rerr. We will continue without this table.
} else {
rerr = y.Wrapf(err, "Opening table: %q", fname)
}
return
}
mu.Lock()
tables[tf.Level] = append(tables[tf.Level], t)
mu.Unlock()
}(fname, tf)
}
// 关闭相关的任务协程
if err := throttle.Finish(); err != nil {
closeAllTables(tables)
return nil, err
}
db.opt.Infof("All %d tables opened in %s\n", atomic.LoadInt32(&numOpened),
time.Since(start).Round(time.Millisecond))
// 记录当前fid最大值
s.nextFileID = maxFileID + 1
// 初始化每个level的tables
for i, tbls := range tables {
s.levels[i].initTables(tbls)
}
// Make sure key ranges do not overlap etc.
// 必要的数据校验
if err := s.validate(); err != nil {
_ = s.cleanupLevels()
return nil, y.Wrap(err, "Level validation")
}
// Sync directory (because we have at least removed some files, or previously created the
// manifest file).
// 手动进行同步刷盘
if err := syncDir(db.opt.Dir); err != nil {
_ = s.close()
return nil, err
}
return s, nil
}
创建levelHandler
newLevelHandler(db, level) 函数
func newLevelHandler(db *DB, level int) *levelHandler {
return &levelHandler{
level: level,
strLevel: fmt.Sprintf("l%d", level),
db: db,
}
}
初始化tables
initTables(tables) 方法
// initTables replaces s.tables with given tables. This is done during loading.
func (s *levelHandler) initTables(tables []*table.Table) {
// 加锁
s.Lock()
defer s.Unlock()
// 赋值与相关值的初始化
s.tables = tables
s.totalSize = 0
s.totalStaleSize = 0
for _, t := range tables {
s.addSize(t)
}
// 如果是L0层,需要拿每个fid排序
if s.level == 0 {
// Key range will overlap. Just sort by fileID in ascending order
// because newer tables are at the end of level 0.
sort.Slice(s.tables, func(i, j int) bool {
return s.tables[i].ID() < s.tables[j].ID()
})
} else {
// L0层往上,拿每个table文件的MinKey排序
// Sort tables by keys.
sort.Slice(s.tables, func(i, j int) bool {
return y.CompareKeys(s.tables[i].Smallest(), s.tables[j].Smallest()) < 0
})
}
}
初始化vlog
init(db) 方法
// init initializes the value log struct. This initialization needs to happen
// before compactions start.
func (vlog *valueLog) init(db *DB) {
// 加载配置
vlog.opt = db.opt
vlog.db = db
// We don't need to open any vlog files or collect stats for GC if DB is opened
// in InMemory mode. InMemory mode doesn't create any files/directories on disk.
// inmem,那我走?🤡
if vlog.opt.InMemory {
return
}
// 指定的vlog目录
vlog.dirPath = vlog.opt.ValueDir
// GC模块用到的channel
vlog.garbageCh = make(chan struct{}, 1) // Only allow one GC at a time.
// 创建一个GC模块相关文件
lf, err := InitDiscardStats(vlog.opt)
y.Check(err)
vlog.discardStats = lf
}
打开vlog
open(db) 方法
func (vlog *valueLog) open(db *DB) error {
// We don't need to open any vlog files or collect stats for GC if DB is opened
// in InMemory mode. InMemory mode doesn't create any files/directories on disk.
// 不想再做解释了,inmem,那我走!!!
if db.opt.InMemory {
return nil
}
// 填充文件fid到filesMap
if err := vlog.populateFilesMap(); err != nil {
return err
}
// If no files are found, then create a new file.
// 如果没有.vlog文件
if len(vlog.filesMap) == 0 {
if vlog.opt.ReadOnly {
return nil
}
// 创建一个.vlog文件
_, err := vlog.createVlogFile()
return y.Wrapf(err, "Error while creating log file in valueLog.open")
}
fids := vlog.sortedFids()
for _, fid := range fids {
lf, ok := vlog.filesMap[fid]
y.AssertTrue(ok)
// Just open in RDWR mode. This should not create a new log file.
lf.opt = vlog.opt
if err := lf.open(vlog.fpath(fid), os.O_RDWR,
2*vlog.opt.ValueLogFileSize); err != nil {
return y.Wrapf(err, "Open existing file: %q", lf.path)
}
// We shouldn't delete the maxFid file.
if lf.size == vlogHeaderSize && fid != vlog.maxFid {
vlog.opt.Infof("Deleting empty file: %s", lf.path)
if err := lf.Delete(); err != nil {
return y.Wrapf(err, "while trying to delete empty file: %s", lf.path)
}
delete(vlog.filesMap, fid)
}
}
if vlog.opt.ReadOnly {
return nil
}
// Now we can read the latest value log file, and see if it needs truncation. We could
// technically do this over all the value log files, but that would mean slowing down the value
// log open.
last, ok := vlog.filesMap[vlog.maxFid]
y.AssertTrue(ok)
lastOff, err := last.iterate(vlog.opt.ReadOnly, vlogHeaderSize,
func(_ Entry, vp valuePointer) error {
return nil
})
if err != nil {
return y.Wrapf(err, "while iterating over: %s", last.path)
}
if err := last.Truncate(int64(lastOff)); err != nil {
return y.Wrapf(err, "while truncating last value log file: %s", last.path)
}
// Don't write to the old log file. Always create a new one.
if _, err := vlog.createVlogFile(); err != nil {
return y.Wrapf(err, "Error while creating log file in valueLog.open")
}
return nil
}
populateFilesMap() 方法
func (vlog *valueLog) populateFilesMap() error {
vlog.filesMap = make(map[uint32]*logFile)
// 从目录中拿到每个文件的句柄
files, _ := ioutil.ReadDir(vlog.dirPath)
found := make(map[uint64]struct{})
for _, file := range files {
// 判断是否以.vlog作为后缀
if !strings.HasSuffix(file.Name(), ".vlog") {
continue
}
// 对.vlog文件进行校验,去除fid,进行消重判断
fsz := len(file.Name())
fid, err := strconv.ParseUint(file.Name()[:fsz-5], 10, 32)
if err != nil {
return errFile(err, file.Name(), "Unable to parse log id.")
}
if _, ok := found[fid]; ok {
return errFile(err, file.Name(), "Duplicate file found. Please delete one.")
}
found[fid] = struct{}{}
lf := &logFile{
fid: uint32(fid),
path: vlog.fpath(uint32(fid)),
registry: vlog.db.registry,
}
// 最后保存到vlog的filesMap当中
vlog.filesMap[uint32(fid)] = lf
if vlog.maxFid < uint32(fid) {
vlog.maxFid = uint32(fid)
}
}
// 直到每个.vlog文件的fid都添加到了map中
// 第一次初始化时没有.vlog文件,故直接跳过
return nil
}
创建vlog文件
createVlogFile()方法
func (vlog *valueLog) createVlogFile() (*logFile, error) {
// 最大的fid
fid := vlog.maxFid + 1
// 根据fid命名
path := vlog.fpath(fid)
// 创建一个句柄实例
lf := &logFile{
fid: fid,
path: path,
registry: vlog.db.registry,
writeAt: vlogHeaderSize,
opt: vlog.opt,
}
// 进行系统调用打开文件,通过mmap的方式
// .vlog文件初始化时会创建一个2G的文件
err := lf.open(path, os.O_RDWR|os.O_CREATE|os.O_EXCL, 2*vlog.opt.ValueLogFileSize)
if err != z.NewFile && err != nil {
return nil, err
}
// 进行数据初始化更新的操作
vlog.filesLock.Lock()
vlog.filesMap[fid] = lf
y.AssertTrue(vlog.maxFid < fid)
vlog.maxFid = fid
// writableLogOffset is only written by write func, by read by Read func.
// To avoid a race condition, all reads and updates to this variable must be
// done via atomics.
atomic.StoreUint32(&vlog.writableLogOffset, vlogHeaderSize)
vlog.numEntriesWritten = 0
vlog.filesLock.Unlock()
return lf, nil
}