英文源地址
持久化至磁盘
前一章中的b树数据结构可以很容易地转存到磁盘上.让我们在它之上建立一个简单地kv存储. 由于我们的b树实现是不可变的, 我们将以仅追加的方式分配磁盘空间, 重用磁盘空间将推迟到下一章.
持久化数据的方式
正如前面章节所提到的, 将数据持久化到磁盘不仅仅是将数据转存到文件中. 这里有几个要考虑的因素:
- 崩溃恢复: 这包括数据库进程崩溃, 操作系统崩溃和电源故障.重启后数据库必须处于可用状态.
- 持久性: 在数据库成功响应后, 所涉及的数据可以保证持久化, 即使在崩溃之后也是如此. 换句话说, 持久化在响应客户端之前发生.
有许多描述数据库的专业术语ACID(原子性, 一致性, 隔离性, 持久性), 但这些概念不是正交的, 很难解释, 所以让我们转而关注我们的实际示例.
- b树的不可变方面: 更新b树不会触及b树的前一个版本, 这使得崩溃恢复变得容易–如果更新出错, 我们可以简单地恢复到前一个版本.
- 持久性是通过fsync linux系统调用实现的.通过write或mmap的普通文件IO首先进入页缓存(page cache), 系统稍后必须将页缓存(page cache)刷新(flush)到磁盘上.fsync系统调用阻塞, 知道所有脏页都被刷新.
如果更新出错, 我们将如何恢复到以前的版本? 我们可以将更新分为两个阶段:
- 更新操作创建新的节点, 将它们写入磁盘.
- 每次更新都会创建一个新的根节点, 我们需要将指向根节点的指针存储在某个地方.
第一个阶段可能涉及将多个页面写入磁盘, 这通常不是原子操作.但是第二阶段只涉及单个指针, 并且可以在原子的单页写入操作中完成. 这使得整个操作原子化–如果数据库崩溃, 更新将不会发生.
第一个阶段必须在第二个阶段前持久化, 否则, 根节点指针可能会在崩溃后指向一个损坏的(部份持久化的)树版本.这两个阶段之间应该有一个fsync(作为一个屏障)
第二阶段也应该在响应客户端之前进行fsync.
基于mmap的IO
可以使用mmap系统调用从虚拟地址映射磁盘文件的内容. 从这个地址读取将启动透明磁盘IO, 这与通过read系统调用读取文件相同, 但不需要用户空间缓冲区(buffer)和系统调用的开销. 映射地址是页缓存(page cache)的代理, 通过它修改数据与write系统调用相同.
mmap很方便, 我们将在KV存储中使用它. 然而, 使用mmap并不是必需的.
func mmapInit(fp *os.File) (int, []byte, error) {
fi, err := fp.Stat()
if err != nil {
return 0, nil, fmt.Errorf("stat: %w", err)
}
if fi.Size()%BTREE_PAGE_SIZE == 0 {
return 0, nil, errors.New("File size is not a multiple of page size.")
}
mmapSize := 64 << 20
if mmapSize%BTREE_PAGE_SIZE != 0 {
panic("wrong mmapsize")
}
for mmapSize < int(fi.Size()) {
mmapSize *= 2
}
chunk, err := syscall.Mmap(int(fp.Fd()), 0, mmapSize, syscall.PROT_READ|syscall.PROT_WRITE, syscall.MAP_SHARED)
if err != nil {
return 0, nil, fmt.Errorf("mmap: %w", err)
}
return int(fi.Size()), chunk, nil
}
上面的函数创建了至少与文件大小相等的初始映射. 映射的大小可以大于文件大小, 并且超过文件末尾的范围是不可访问的(SIGBUS), 但是可以稍后拓展文件.
随着文件的增长, 我们可能需要扩展映射的范围. 扩展mmap范围的系统调用是mremap.不幸的是, 当通过重新映射扩展范围时. 我们可能无法保留起始地址. 我们扩展映射的方法是使用多个映射–为溢出文件范围创建一个新的映射.
type KV struct {
Path string
fp *os.File
tree BTree
mmap struct{
file int
total int
chunks [][]byte
}
page struct{
flushed uint64
temp [][]byte
}
}
func extendMmap(db *KV, npages int) error {
if db.mmap.total >= npages*BTREE_PAGE_SIZE {
return nil
}
chunk, err := syscall.Mmap(
int(db.fp.Fd()), int64(db.mmap.total), db.mmap.total,
syscall.PROT_READ|syscall.PROT_WRITE, syscall.MAP_SHARED)
if err != nil {
return fmt.Errorf("mmap: %w", err)
}
db.mmap.total += db.mmap.total
db.mmap.chunks = append(db.mmap.chunks, chunk)
return nil
}
新映射的大小呈指数增长, 因此我们不必频繁调用mmap.
下面是我们如何从映射地址访问一个页.
func (db *KV) pageGet(ptr uint64) BNode {
start := uint64(0)
for _, chunk := range db.mmap.chunks {
end := start + uint64(len(chunk))/BTREE_PAGE_SIZE
if ptr < end {
offset := BTREE_PAGE_SIZE * (ptr - start)
return BNode{chunk[offset: offset+BTREE_PAGE_SIZE]}
}
start = end
}
panic("bad ptr")
}
主页 MasterPage
文件的第一页用于存储指向根节点的指针, 我们称之为’master page’.分配新节点所需的总页数, 因此页存储在那里.
下面的函数在初始化数据库时读取master page.
const DB_SIG = "BuildYourOwnDB05"
func masterLoad(db *KV) error {
if db.mmap.file == 0 {
db.page.flushed = 1
return nil
}
data := db.mmap.chunks[0]
root := binary.LittleEndian.Uint64(data[16:])
used := binary.LittleEndian.Uint64(data[24:])
if !bytes.Equal([]byte(DB_SIG), data[:16]) {
return errors.New("Bad signature")
}
bad := !(1 <= used && used <= uint64(db.mmap.file/BTREE_PAGE_SIZE))
bad = bad || !(0 <= root && root < used)
if bad {
return errors.New("Bad master page.")
}
db.tree.root = root
db.page.flushed = used
return nil
}
下面是更新master page的功能. 与用于读取的代码不同, 它不使用映射地址进行写入.这是因为通过mmap修改页不是原子性的. 内核可能会在中途刷新页并损坏磁盘文件, 而不跨越页边界的小的write操作则保证是原子性的.
分配自盘页
我们将简单地将新的页添加到数据库的末尾, 直到在下一章中添加空闲列表.
新的页暂时保存在内存中, 直到稍后(在可能拓展文件之后)复制到文件中.
func (db *KV) pageNew(node BNode) uint64 {
if len(node.data) > BTREE_PAGE_SIZE {
panic("error size")
}
ptr := db.page.flushed + uint64(len(db.page.temp))
db.page.temp = append(db.page.temp, node.data)
return ptr
}
func (db *KV) pageDel(uint64) {
}
在写入挂起的页面之前, 我们可能首先扩展文件. 对应的系统调用是fallocate
func extendFile(db *KV, npages int) error {
filePages := db.mmap.file / BTREE_PAGE_SIZE
if filePages >= npages {
return nil
}
for filePages < npages {
inc := filePages / 8
if inc < 1 {
inc = 1
}
filePages += inc
}
fileSize := filePages * BTREE_PAGE_SIZE
err := syscall.Fallocate(int(db.fp.Fd()), 0, 0, int64(fileSize))
if err != nil {
return fmt.Errorf("fallocate: %w", err)
}
db.mmap.file = fileSize
return nil
}
初始化数据库
将我们已经完成的放在一起
func (db *KV) Open() error {
fp, err := os.OpenFile(db.Path, os.O_RDWR|os.O_CREATE, 0664)
if err != nil {
return fmt.Errorf("OpenFile: %w", err)
}
db.fp = fp
sz, chunk, err := mmapInit(db.fp)
if err != nil {
goto fail
}
db.mmap.file = sz
db.mmap.total = len(chunk)
db.mmap.chunks = [][]byte{chunk}
db.tree.get = db.pageGet
db.tree.new = db.pageNew
db.tree.del = db.pageDel
err = masterLoad(db)
if err != nil {
goto fail
}
return nil
fail:
db.Close()
return fmt.Errorf("KV.Open: %w", err)
}
func (db *KV) Close() {
for _, chunk := range db.mmap.chunks {
err := syscall.Munmap(chunk)
if err != nil {
panic("error")
}
}
_ = db.fp.Close()
}
更新操作
与查询不同, 更新操作在返回之前必须持久化数据
func (db *KV) Get(key []byte) ([]byte, bool) {
return db.tree.Get(key)
}
func (db *KV) Set(key []byte, val []byte) error {
db.tree.Insert(key, val)
return flushPages(db)
}
func (db *KV) Del(key []byte) (bool, error) {
deleted := db.tree.Delete(key)
return deleted, flushPages(db)
}
flushPages是持久化新页的函数
func flushPages(db *KV) error {
if err := writePages(db); err != nil {
return err
}
return syncPages(db)
}
如前所述, 它分为两个阶段
func writePages(db *KV) error {
npages := int(db.page.flushed) + len(db.page.temp)
if err := extendFile(db, npages); err!= nil {
return err
}
if err := extendMmap(db, npages); err != nil {
return err
}
for i, page := range db.page.temp {
ptr := db.page.flushed + uint64(i)
copy(db.pageGet(ptr).data, page)
}
return nil
}
fsync在它们之间和之后.
func syncPages(db *KV) error {
if err := db.fp.Sync(); err != nil {
return fmt.Errorf("fsync: %w", err)
}
db.page.flushed += uint64(len(db.page.temp))
db.page.temp = db.page.temp[:0]
if err := masterLoad(db); err != nil {
return err
}
if err := db.fp.Sync(); err != nil {
return fmt.Errorf("fsync: %w", err)
}
return nil
}
我们的KV存储是功能性的, 但是当我们更新数据库时, 文件不可能永远增长, 我们将在下一章通过重用磁盘页来完成kv存储.
