BadgerDB 是一个用纯 Go 编写的可嵌入、持久且快速的键值 (KV) 数据库。它是Dgraph的底层数据库, 一个快速的分布式图数据库。它旨在成为 RocksDB 等非基于 Go 的键值存储的高性能替代品。

0. 源码目录结构

├── badger
│   └── cmd
├── docs: 文档
├── fb: Google flatbuffers
├── integration
│   └── testgc
├── options
├── pb: Protobuf 消息体
├── skl: 改编自 RocksDB inline inline skiplist, Skiplist实现
├── table:  Table表的实现
├── trie: 字典树的实现
├── y: 可以归为工具类, 包含里引自 LevelDB的布隆过滤器实现、校验和、加密、文件处理、迭代器接口等 
│  backup.go      
│  batch.go       
│  compaction.go  
│  db.go             
│  dir_plan9.go   
│  dir_unix.go    
│  dir_windows.go 
│  discard.go     
│  doc.go         
│  errors.go            
│  histogram.go   
│  iterator.go    
│  key_registry.go
│  levels.go      
│  level_handler.go 
│  logger.go      
│  managed_db.go  
│  manifest.go    
│  memtable.go    
│  merge.go       
│  options.go     
│  publisher.go      
│  stream.go      
│  stream_writer.go
│  structs.go     
│  test.sh        
│  txn.go         
│  util.go        
│  value.go       

1. 数据结构

1.1 memTable 和 Table

memTable 结构存储了一个skiplist 和一个对应的WAL。对 memTable 的写入同时写入WAL和Skiplist。在崩溃时, 重播 WAL 以将Skiplist恢复到崩溃前的形式。 Table 表示一个加载的表文件, 其中包含我们所拥有的信息。smallest, biggest 可以在前缀查询时起到过滤作用。

type memTable struct {
	sl         *skl.Skiplist
	wal        *logFile
	maxVersion uint64
	opt        Options
	buf        *bytes.Buffer
}

type Table struct {
	sync.Mutex
	*z.MmapFile

	tableSize int // Initialized in OpenTable, using fd.Stat().

	_index *fb.TableIndex // Nil if encryption is enabled. Use fetchIndex to access.
	_cheap *cheapIndex
	ref    int32 // For file garbage collection. Atomic.

	// The following are initialized once and const.
	smallest, biggest []byte // Smallest and largest keys (with timestamps).
	id                uint64 // file id, part of filename

	Checksum       []byte
	CreatedAt      time.Time
	indexStart     int
	indexLen       int
	hasBloomFilter bool

	IsInmemory bool // Set to true if the table is on level 0 and opened in memory.
	opt        *Options
}

跳跃表在LevelDB中用于存储内存数据, 在LevelDB中被称作Memtable。

1.2 Inline Skiplist

改编自 RocksDB inline skiplist.内联跳跃表是对于LevelDB中的跳跃表的优化, 原理很简单, 通过更紧凑的内存安排:

• 减少了内存的使用
• 提供了更好的局部性

skiplist 的 Arena: 是提前预分配好的空间, 用来存放 node, Key, value(三者分开存放), 而node结构包含了key 和value在Arena中的位置偏移以及大小。

type Skiplist struct {
	height     int32 // Current height. 1 <= height <= kMaxHeight. CAS.
	headOffset uint32
	ref        int32
	arena      *Arena
	OnClose    func()
}

type Arena struct {
	n          uint32
	shouldGrow bool
	buf        []byte
}

type node struct {
	value uint64 	// valOffset & (valSize << 32)
	keyOffset uint32 // Immutable. No need to lock to access key.
	keySize   uint16 // Immutable. No need to lock to access key.
	height uint16
	tower [maxHeight]uint32
}

s := skl.NewSkiplist(arenaSize(db.opt))
mt := &memTable{
	sl:  s,
	opt: db.opt,
	buf: &bytes.Buffer{},
}

func newArena(n int64) *Arena {
	// Don't store data at position 0 in order to reserve offset=0 as a kind of nil pointer.
	out := &Arena{
		n:   1,
		buf: make([]byte, n),
	}
	return out
}

func arenaSize(opt Options) int64 {
	return opt.MemTableSize + opt.maxBatchSize + opt.maxBatchCount*int64(skl.MaxNodeSize)
}

func (s *Arena) putVal(v y.ValueStruct) uint32 {
	l := uint32(v.EncodedSize())
	offset := s.allocate(l)
	v.Encode(s.buf[offset:])
	return offset
}

func (s *Arena) putKey(key []byte) uint32 {
	keySz := uint32(len(key))
	offset := s.allocate(keySz)
	buf := s.buf[offset : offset+keySz]
	y.AssertTrue(len(key) == copy(buf, key))
	return offset
}

1.3 logFile

type logFile struct {
	*z.MmapFile
	path string
	// This is a lock on the log file. It guards the fd’s value, the file’s
	// existence and the file’s memory map.
	//
	// Use shared ownership when reading/writing the file or memory map, use
	// exclusive ownership to open/close the descriptor, unmap or remove the file.
	lock     sync.RWMutex
	fid      uint32
	size     uint32
	dataKey  *pb.DataKey
	baseIV   []byte
	registry *KeyRegistry
	writeAt  uint32
	opt      Options
}

1.4 valueLog

type valueLog struct {
	dirPath string

	// guards our view of which files exist, which to be deleted, how many active iterators
	filesLock        sync.RWMutex
	filesMap         map[uint32]*logFile
	maxFid           uint32
	filesToBeDeleted []uint32
	// A refcount of iterators -- when this hits zero, we can delete the filesToBeDeleted.
	numActiveIterators int32

	db                *DB
	writableLogOffset uint32 // read by read, written by write. Must access via atomics.
	numEntriesWritten uint32
	opt               Options

	garbageCh    chan struct{}
	discardStats *discardStats
}

2. Inline Skiplist的实现

2.1 RocksDB InlineSkipList

之所以叫 InlineSkipList, 应该是因为 Node 将 key 和链表每层的指针连续存储:

template <class Comparator>
struct InlineSkipList<Comparator>::Node {
 private:
  // next_[0] is the lowest level link (level 0).  Higher levels are
  // stored _earlier_, so level 1 is at next_[-1].
  std::atomic<Node*> next_[1];
};

格式如下:

Node 直接存 key, 相比于 LevelDB 存 key 的指针, 可以减少部分内存使用, 更主要的是有更好的 cache locality, 访问 next_ 指针时, 因为内存连续会把 key 也一并放到 cache 中。而且 在遍历每层 list 时, 会 prefetch 后面的 Node:

#define PREFETCH(addr, rw, locality) __builtin_prefetch(addr, rw, locality)

2.2 Badger Skiplist

Badger Skiplist 改编自 RocksDB inline skiplist, 主要区别:

• 没有对顺序插入进行优化(没有"prev")。
• 没有自定义比较器。
• 支持覆盖。当我们在插入时看到相同的键时, 这需要小心。 对于 RocksDB 或 LevelDB, 覆盖是作为 key 中更新的序列号实现的, 所以不需要values。我们不打算支持版本控制。值的就地更新会更有效率。
• 我们丢弃所有非并发代码。
• 我们不支持Splices, 这大大简化了代码。
• 没有 AllocateNode 或其他指针算法。
• 我们将 findLessThan、findGreaterOrEqual 等组合成一个函数。

3. 操作的封装

Badger提供了Get、Set、Delete和Iterate函数, 相关操作都封装到 Txn 和 oracle 两个结构体里. Txn 内部的信息主要是开始时间戳、提交时间戳、读写的 key 列表, oracle 相当于事务管理器, 内部维护近期提交的事务列表、全局时间戳、当前活跃事务的最早时间戳等。 WaterMark 结构体内部是个堆, 用于管理、查找事务开始、结束的区段。oracle 的 txnMarker 主要用于协调等待 Commit 授时与落盘的时间窗口, readMarker 管理当前活跃事务的最早时间戳, 用于清理过期的 committedTxns。

type Txn struct {
	readTs   uint64
	commitTs uint64
	size     int64
	count    int64
	db       *DB

	reads []uint64 // contains fingerprints of keys read.
	// contains fingerprints of keys written. This is used for conflict detection.
	conflictKeys map[uint64]struct{}
	readsLock    sync.Mutex // guards the reads slice. See addReadKey.

	pendingWrites   map[string]*Entry // cache stores any writes done by txn.
	duplicateWrites []*Entry          // Used in managed mode to store duplicate entries.

	numIterators int32
	discarded    bool
	doneRead     bool
	update       bool // update is used to conditionally keep track of reads.
}

type oracle struct {
	isManaged       bool // Does not change value, so no locking required.
	detectConflicts bool // Determines if the txns should be checked for conflicts.

	sync.Mutex // For nextTxnTs and commits.
	// writeChLock 确保事务以与其提交时间戳相同的顺序进入写入通道。
	writeChLock sync.Mutex
	nextTxnTs   uint64

	// Used to block NewTransaction, so all previous commits are visible to a new read.
	txnMark *y.WaterMark

	// Either of these is used to determine which versions can be permanently
	// discarded during compaction.
	discardTs uint64       // Used by ManagedDB.
	readMark  *y.WaterMark // Used by DB.

	// committedTxns contains all committed writes (contains fingerprints
	// of keys written and their latest commit counter).
	committedTxns []committedTxn
	lastCleanupTs uint64

	// closer is used to stop watermarks.
	closer *z.Closer
}

3.1 Key 的处理

Badger 对Key的处理有两种方式:

• 带有TS的Key

e.Key = y.KeyWithTs(e.Key, e.version)

func KeyWithTs(key []byte, ts uint64) []byte {
	out := make([]byte, len(key)+8)
	copy(out, key)
	binary.BigEndian.PutUint64(out[len(key):], math.MaxUint64-ts)
	return out
}

• Table 里的Key

keyNoTs := y.ParseKey(key)

func ParseKey(key []byte) []byte {
	if key == nil {
		return nil
	}

	return key[:len(key)-8]
}

更多细节可以查看文件y/y.go, 如:

func CompareKeys(key1, key2 []byte) int {
	if cmp := bytes.Compare(key1[:len(key1)-8], key2[:len(key2)-8]); cmp != 0 {
		return cmp
	}
	return bytes.Compare(key1[len(key1)-8:], key2[len(key2)-8:])
}

3.2 Badger 对事务的封装

DB.View()和DB.Update()方法可以被DB.NewTransaction()和Txn.Commit()方法封装 (或者在只读事务下使用Txn.Discard())。这些辅助方法将启动事务, 执行一个函数, 然后在返回错误时安全地丢弃您的事务。这是使用 Badger 事务的推荐方式。

err := db.View(func(txn *badger.Txn) error {
  // Your code here…
  return nil
})

err := db.Update(func(txn *badger.Txn) error {
  // Your code here…
  return nil
})

3.3 基本操作流程

3.3.1 键查找

• 对Key进行处理: 附加上Ts

seek := y.KeyWithTs(key, txn.readTs)
vs, err := txn.db.get(seek)

• 遍历每个 memTable 的 Skiplist (LSM查找)

tables, decr := db.getMemTables()
for i := 0; i < len(tables); i++ {
	vs := tables[i].sl.Get(key)
}

func (s *Skiplist) Get(key []byte) y.ValueStruct {
	n, _ := s.findNear(key, false, true) // findGreaterOrEqual.
	if n == nil {
		return y.ValueStruct{}
	}

	nextKey := s.arena.getKey(n.keyOffset, n.keySize)
	if !y.SameKey(key, nextKey) {
		return y.ValueStruct{}
	}

	valOffset, valSize := n.getValueOffset()
	vs := s.arena.getVal(valOffset, valSize)
	vs.Version = y.ParseTs(nextKey)
	return vs
}

• 在Skiplist 查看Key 对应的Node的信息: n, _ := s.findNear(key, false, true)

• 在 arena 里获取 key 信息, 不存在则返回

• 在 arena 里获取 ValueStruct (它的Offset 和Size编码在 node.value)

• 最终, 还是要在 Level 里查找

db.lc.get(key, maxVs, 0)

version := y.ParseTs(key)
for _, h := range s.levels {
	vs, err := h.get(key)
}

3.3.2 遍历查找(迭代)

Iterator 是迭代查询的底层数据结构。Badger 在 Iterator 之上又封装里UniIterator 和 MergeIterator. Badger 用 Go 实现了一个list 数据结构, 用来存放预读取的数据:

type Iterator struct {
	list *Skiplist
	n    *node
	data  list
}

UniIterator 是一个单向的 memtable 迭代器。 它是迭代器的薄包装。 我们喜欢像以前一样保留迭代器, 因为它更强大, 并且我们将来可能会支持双向迭代器。

type UniIterator struct {
	iter     *Iterator
	reversed bool
}

MergeIterator 合并多个迭代器, 并负责关闭它们。

type MergeIterator struct {
	left  node
	right node
	small *node

	curKey  []byte
	reverse bool
}

在 Table 层实现了前缀查询基于表的过滤:

func (s *levelHandler) iterators(opt *IteratorOptions) []y.Iterator {
	s.RLock()
	defer s.RUnlock()

	// ....

	tables := opt.pickTables(s.tables)
	if len(tables) == 0 {
		return nil
	}
	return []y.Iterator{table.NewConcatIterator(tables, topt)}
}

func (opt *IteratorOptions) pickTables(all []*table.Table) []*table.Table {
	// ...
	sIdx := sort.Search(len(all), func(i int) bool {
		// table.Biggest >= opt.prefix
		// if opt.Prefix < table.Biggest, then surely it is not in any of the preceding tables.
		return opt.compareToPrefix(all[i].Biggest()) >= 0
	})
	// ...
}

在对链表遍历时进行过滤(基于Key的有序存储):

func (it *Iterator) Valid() bool {
	if it.item == nil {
		return false
	}
	if it.opt.prefixIsKey {
		return bytes.Equal(it.item.key, it.opt.Prefix)
	}
	return bytes.HasPrefix(it.item.key, it.opt.Prefix)
}

badger 里也存储里里一些自身数据, 需要过滤:

// Skip badger keys.
if !it.opt.InternalAccess && isInternalKey {
	mi.Next()
	return false
}

通过预读取数据到链表 data 中, 发挥 SSD的顺序读能力:

func (it *Iterator) prefetch() {
	prefetchSize := 2
	if it.opt.PrefetchValues && it.opt.PrefetchSize > 1 {
		prefetchSize = it.opt.PrefetchSize
	}

	i := it.iitr
	var count int
	it.item = nil
	for i.Valid() {
		if !it.parseItem() {
			continue
		}
		count++
		if count == prefetchSize {
			break
		}
	}
}

func (it *Iterator) parseItem() bool {
	mi := it.iitr
	key := mi.Key()

	setItem := func(item *Item) {
		if it.item == nil {
			it.item = item
		} else {
			it.data.push(item)
		}
	}
	// ...
}

3.3.3 增删改操作

通过 Txn 对 Badger 的 增删改操作, 最终都会 转换为txn.modify(e), 例如:

func (txn *Txn) Delete(key []byte) error {
	e := &Entry{
		Key:  key,
		meta: bitDelete,
	}
	return txn.modify(e)
}

在验证通过后, 就会写入到写通道db.writeCh <- req, 请求进程阻塞, 由一个常驻的后台进程处理实际写入请求, 这个进程在程序启动时一并启动:

func Open(opt Options) (*DB, error) {
	go db.doWrites(db.closers.writes)
}

具体写入流程分为:

• valueLog 的写入: db.vlog.write(reqs)
• 确保写入的前置条件, 如空间分配等: err = db.ensureRoomForWrite()
• 最后是写入 LSM树: db.writeToLSM(b)
  • memTable的 wal 日志写: mt.wal.writeEntry(mt.buf, entry, mt.opt)
  • memTable的 skiplist的写入: mt.sl.Put(key, value)

memTable的 wal 日志写涉及到里日志WAL文件大小的管理, 具体逻辑可以查看 ensureRoomForWrite 函数 If WAL exceeds opt.ValueLogFileSize, we’ll force flush the memTable. See logic in ensureRoomForWrite.

3.3.4 查找Table时的布隆过滤

在过Key 查找对应Table表是, 使用里布隆过滤以减小表范围(布隆过滤算法使用的是 LevelDB的算法, 详细请见文件y/bloom.go):

func (f Filter) MayContain(h uint32) bool {
	if len(f) < 2 {
		return false
	}
	k := f[len(f)-1]
	if k > 30 {
		// This is reserved for potentially new encodings for short Bloom filters.
		// Consider it a match.
		return true
	}
	nBits := uint32(8 * (len(f) - 1))
	delta := h>>17 | h<<15
	for j := uint8(0); j < k; j++ {
		bitPos := h % nBits
		if f[bitPos/8]&(1<<(bitPos%8)) == 0 {
			return false
		}
		h += delta
	}
	return true
}

而对应的Key 则采用了 Murmur hash:

func Hash(b []byte) uint32 {
	const (
		seed = 0xbc9f1d34
		m    = 0xc6a4a793
	)
	h := uint32(seed) ^ uint32(len(b))*m
	for ; len(b) >= 4; b = b[4:] {
		h += uint32(b[0]) | uint32(b[1])<<8 | uint32(b[2])<<16 | uint32(b[3])<<24
		h *= m
		h ^= h >> 16
	}
	switch len(b) {
	case 3:
		h += uint32(b[2]) << 16
		fallthrough
	case 2:
		h += uint32(b[1]) << 8
		fallthrough
	case 1:
		h += uint32(b[0])
		h *= m
		h ^= h >> 24
	}
	return h
}

4. 其他

// 系统预留的Key 命名前缀: 
var (
	badgerPrefix = []byte("!badger!")       // Prefix for internal keys used by badger.
	txnKey       = []byte("!badger!txn")    // For indicating end of entries in txn.
	bannedNsKey  = []byte("!badger!banned") // For storing the banned namespaces.
)

// 省略了大部分属性
type DB struct {
	mt  *memTable   // Our latest (actively written) in-memory table
	imm []*memTable // Add here only AFTER pushing to flushChan.

	lc        *levelsController
	vlog      valueLog
	writeCh   chan *request

	orc              *oracle
}

node 是存储到 Skiplist 到数据结构, 从 ValueStruct 转换而来, 这里的 value已经不是用户数据里的value。

type ValueStruct struct {
	Meta      byte
	UserMeta  byte
	ExpiresAt uint64
	Value     []byte
	Version uint64 // This field is not serialized. Only for internal usage.
}

// findSpliceForLevel returns (outBefore, outAfter) with outBefore.key <= key <= outAfter.key.
// The input "before" tells us where to start looking.
// If we found a node with the same key, then we return outBefore = outAfter.
// Otherwise, outBefore.key < key < outAfter.key.
func (s *Skiplist) findSpliceForLevel(key []byte, before uint32, level int) (uint32, uint32) {
	for {
		// Assume before.key < key.
		beforeNode := s.arena.getNode(before)
		next := beforeNode.getNextOffset(level)
		nextNode := s.arena.getNode(next)
		if nextNode == nil {
			return before, next
		}
		nextKey := nextNode.key(s.arena)
		cmp := y.CompareKeys(key, nextKey)
		if cmp == 0 {
			// Equality case.
			return next, next
		}
		if cmp < 0 {
			// before.key < key < next.key. We are done for this level.
			return before, next
		}
		before = next // Keep moving right on this level.
	}
}

func (s *Arena) putVal(v y.ValueStruct) uint32 {
	l := uint32(v.EncodedSize())
	offset := s.allocate(l)
	v.Encode(s.buf[offset:])
	return offset
}

func (v *ValueStruct) Encode(b []byte) uint32 {
	b[0] = v.Meta
	b[1] = v.UserMeta
	sz := binary.PutUvarint(b[2:], v.ExpiresAt)
	n := copy(b[2+sz:], v.Value)
	return uint32(2 + sz + n)
}

func (s *Arena) putKey(key []byte) uint32 {
	keySz := uint32(len(key))
	offset := s.allocate(keySz)
	buf := s.buf[offset : offset+keySz]
	y.AssertTrue(len(key) == copy(buf, key))
	return offset
}

func newNode(arena *Arena, key []byte, v y.ValueStruct, height int) *node {
	// The base level is already allocated in the node struct.
	nodeOffset := arena.putNode(height)
	keyOffset := arena.putKey(key)
	val := encodeValue(arena.putVal(v), v.EncodedSize())

	node := arena.getNode(nodeOffset)
	node.keyOffset = keyOffset
	node.keySize = uint16(len(key))
	node.height = uint16(height)
	node.value = val
	return node
}

Entry 是存储到 wal 里的数据结构, 是对用户上层数据的一层封装:

type Entry struct {
	Key       []byte
	Value     []byte
	ExpiresAt uint64 // time.Unix
	version   uint64
	offset    uint32 // offset is an internal field.
	UserMeta  byte
	meta      byte

	// Fields maintained internally.
	hlen         int // Length of the header.
	valThreshold int64
}

mt.wal.writeEntry(mt.buf, entry, mt.opt)