CppGuide社区
目录

12.并发读写

# 12. 并发读写

# 12.1 读写问题

为了支持并发请求，我们可以首先将事务分为只读事务和读写事务。只读事务的读操作不会修改数据，因此多个读操作总是可以并发运行。而写操作需要修改树结构，必须进行序列化处理（至少部分序列化）。

我们可以支持不同程度的并发。一种方式是使用读写锁（RWLock），它允许以下两种执行情况：

  • 多个并发读操作。
  • 单个写操作。

读写锁是一种实用的技术，添加起来也比较容易。不过，由于使用了不可变数据结构，我们可以轻松实现更高程度的并发。

使用读写锁时，读操作可能会被写操作阻塞，反之亦然。但在不可变数据结构中，写操作会创建数据的新版本，而不是覆盖当前版本，这就实现了读操作和写操作之间的并发，这种方式优于读写锁，也是我们接下来要实现的。

我们的数据库将支持：

  • 多个并发读操作。
  • 多个读操作和单个写操作之间的并发。
  • 写操作完全序列化。

需要注意的是，通过仅对写操作进行部分序列化，可以实现更高程度的并发。例如，如果一个读写事务先读取一部分数据，然后根据这些数据决定写操作的内容，那么我们可以并发执行读操作，仅对最终的提交操作进行序列化。然而，这会引入新的问题：即使提交操作是序列化的，写操作之间仍可能提交冲突的内容，因此我们需要额外的机制来预防或检测冲突。在本专栏中，我们不会探讨这种情况。

# 12.2 实现分析

主要有3处改动： 首先，我们将事务类型分为两种，分别用于只读事务和读写事务。B树类型被移到事务类型（一个快照）中，这样一个事务中的读操作不会受到其他事务的影响。

type KVReader struct {
    // 后续补充……
}
func (kv *KV) BeginRead(tx *KVReader)
func (kv *KV) EndRead(tx *KVReader)
func (tx *KVReader) Get(key []byte) ([]byte, bool)
func (tx *KVReader) Seek(key []byte, cmp int) *BIter
1
2
3
4
5
6
7

其次，我们考虑使用互斥锁（mutexes）。由于写操作是完全序列化的，因此用一个互斥锁来控制写操作即可。有些字段会被写操作更新、被读操作读取，比如最新的树的根节点和内存映射（mmap）块，这些字段需要另一个互斥锁来保护。

type KV struct {
    // 省略……
    mu      sync.Mutex
    writer  sync.Mutex
    // 省略……
}
1
2
3
4
5
6

最后，由于读操作可能还在访问某些页面，我们不能复用这些页面，因此需要重新设计空闲列表（free list）。我们的解决方案是为B树的每个版本分配一个自增的版本号，当一个页面被释放时，空闲列表会存储该页面的版本号和页码。只有版本号小于所有当前读操作版本号的空闲页面才能被复用。

// 由事务更新和提交的内存数据结构
type FreeListData struct {
    head uint64
    // 后续补充……
}

type FreeList struct {
    FreeListData
    // 针对每个事务
    version   uint64   // 当前版本
    minReader uint64   // 最小读操作版本号
    // 后续补充……
}

// 尝试从尾部移除一个元素，失败时返回0。
// 移除的指针不能被最小版本号的读操作访问到。
func (fl *FreeList) Pop() uint64
// 将一些新指针添加到头部并完成更新
func (fl *FreeList) Add(freed []uint64)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

# 12.3 并发事务

# 第1部分：修改KV类型

  1. B树类型被移到事务类型中，这里只保留一个根指针。同样，空闲列表也被移走。
  2. 管理磁盘页面的数据结构和代码也被移到事务类型中。
  3. 添加互斥锁。写操作互斥锁用于序列化写操作，mu互斥锁用于保护数据字段。
  4. 添加版本号，以及一个正在进行的读操作列表，用于跟踪最小活动版本号（针对空闲列表）。读操作列表被维护成一个堆数据结构，这样最小版本号的读操作就在列表首位。
type KV struct {
    Path string
    // 内部使用
    fp *os.File
    // 修改1：移动B树和空闲列表
    tree struct {
        root uint64
    }
    free FreeListData
    mmap struct {
        // 不变；省略……
    }
    // 修改2：移动页面管理
    page struct {
        flushed uint64 // 以页面数量表示的数据库大小
    }
    // 修改3：互斥锁
    mu      sync.Mutex
    writer  sync.Mutex
    // 修改4：版本号和读操作列表
    version uint64
    readers ReaderList // 堆，用于跟踪最小读操作版本号
}

// 实现heap.Interface
type ReaderList []*KVReader
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26

# 第2部分：添加只读事务类型

// 只读键值对事务
type KVReader struct {
    // 快照
    version uint64
    tree    BTree
    mmap    struct {
        chunks [][]byte // 从KV结构体复制而来，只读
    }
    // 用于从堆中移除
    index int
}
1
2
3
4
5
6
7
8
9
10
11

B树类型被移到KVReader中，页面管理函数pageGetMapped也是如此。version和index字段用于ReaderList堆。由于写操作会修改内存映射块，所以我们复制一份mmap块。

func (kv *KV) BeginRead(tx *KVReader) {
    kv.mu.Lock()
    tx.mmap.chunks = kv.mmap.chunks
    tx.tree.root = kv.tree.root
    tx.tree.get = tx.pageGetMapped
    tx.version = kv.version
    heap.Push(&kv.readers, tx)
    kv.mu.Unlock()
}

func (kv *KV) EndRead(tx *KVReader) {
    kv.mu.Lock()
    heap.Remove(&kv.readers, tx.index)
    kv.mu.Unlock()
}

// 用于BTree和FreeList的回调函数，解引用一个指针
func (tx *KVReader) pageGetMapped(ptr uint64) BNode
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

# 第3部分：添加读写事务类型

KVTX扩展了KVReader，这样它就拥有了所有的读方法。和B树类型一样，空闲列表和页面管理数据结构也从KV类型中移出。

// 键值对事务
type KVTX struct {
    KVReader
    db   *KV
    free FreeList
    page struct {
        nappend int // 要追加的页面数量
        // 以指针为键的新分配或释放的页面，nil值表示已释放的页面
        updates map[uint64][]byte
    }
}
1
2
3
4
5
6
7
8
9
10
11

要开始一个读写事务，必须先获取写操作锁。然后我们可以初始化B树类型和空闲列表类型。由于写操作是唯一可以修改数据的线程，所以访问KV类型中的字段时不需要额外的锁（读操作列表除外，它由读操作修改）。

tx.page*函数只是从KV类型中移出来。

// 开始一个事务
func (kv *KV) Begin(tx *KVTX) {
    tx.db = kv
    tx.page.updates = map[uint64][]byte{}
    tx.mmap.chunks = kv.mmap.chunks

    kv.writer.Lock()
    tx.version = kv.version

    // b树
    tx.tree.root = kv.tree.root
    tx.tree.get = tx.pageGet
    tx.tree.new = tx.pageNew
    tx.tree.del = tx.pageDel
    // 空闲列表
    tx.free.FreeListData = kv.free
    tx.free.version = kv.version
    tx.free.get = tx.pageGet
    tx.free.new = tx.pageAppend
    tx.free.use = tx.pageUse

    tx.free.minReader = kv.version
    kv.mu.Lock()
    if len(kv.readers) > 0 {
        tx.free.minReader = kv.readers[0].version
    }
    kv.mu.Unlock()
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28

现在回滚事务不需要做任何操作，因为在提交之前，KV类型中的数据不会被修改。

// 结束一个事务：回滚
func (kv *KV) Abort(tx *KVTX) {
    kv.writer.Unlock()
}
1
2
3
4

只有在事务提交时，才会修改KV类型中的数据（树的根节点和空闲列表头）。

// 结束一个事务：提交更新
func (kv *KV) Commit(tx *KVTX) error {
    defer kv.writer.Unlock()

    // 阶段1：将页面数据持久化到磁盘。
    // 省略……

    // 此时事务已可见。
    // 保存内存数据结构的新版本。
    kv.page.flushed += uint64(tx.page.nappend)
    kv.free = tx.free.FreeListData
    kv.mu.Lock()
    kv.tree.root = tx.tree.root
    kv.version++
    kv.mu.Unlock()

    // 阶段2：更新主页面以指向新的树。
    // 省略……
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

显然，这两个函数最后都会释放写操作锁。

# 12.4 空闲列表

空闲列表从后进先出（FILO，first-in-last-out）改为先进先出（FIFO，first-in-first-out）；新版本释放的页面会添加到列表头部，被复用的页面从列表尾部移除。这样可以使空闲列表按版本号排序。

为了避免复用读操作正在读取的页面，被复用的页面版本号不能比任何读操作的版本号新。这就是我们将空闲列表设计为按版本号排序的原因。

遗憾的是，改为先进先出引入了一些复杂性。我们必须遍历列表才能访问到列表的另一端（尾部）。我们会在内存中保存每个节点的引用，这样就只需要遍历一次列表。

// 由事务更新和提交的内存数据结构
type FreeListData struct {
    head uint64
    // 用于访问列表两端的缓存节点指针，从尾部到头部
    nodes []uint64
    // 缓存的总元素数量，存储在头部节点
    total int
    // 缓存的尾部节点中已丢弃元素的数量
    offset int
}

type FreeList struct {
    FreeListData
    // 针对每个事务
    version   uint64   // 当前版本
    minReader uint64   // 最小读操作版本号
    freed     []uint64 // 将要添加到空闲列表的页面
    // 用于管理磁盘页面的回调函数
    get func(uint64) BNode  // 解引用一个指针
    new func(BNode) uint64  // 追加一个新页面
    use func(uint64, BNode) // 复用一个页面
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22

版本号被添加到空闲列表节点和主页面中：
type size total next
2B 2B 8B 8B

主页面格式：

sig btree_root page_used free_list version
16B 8B 8B 8B 8B

所有读操作中最旧的版本号在事务开始时获取。

// 开始一个事务
func (kv *KV) Begin(tx *KVTX) {
    // 省略……

    tx.free.minReader = kv.version
    kv.mu.Lock()
    if len(kv.readers) > 0 {
        tx.free.minReader = kv.readers[0].version
    }
    kv.mu.Unlock()
}
1
2
3
4
5
6
7
8
9
10
11

从列表尾部复用页面时会进行检查。

// 尝试从尾部移除一个元素，失败时返回0。
// 移除的指针不能被最小版本号的读操作访问到。
func (fl *FreeList) Pop() uint64 {
    fl.loadCache()
    return flPop1(fl)
}

func flPop1(fl *FreeList) uint64 {
    if fl.total == 0 {
        return 0
    }

    // 从尾部移除一个元素
    assert(fl.offset < flnSize(fl.get(fl.nodes[0])))
    ptr, ver := flnItem(fl.get(fl.nodes[0]), fl.offset)
    if versionBefore(fl.minReader, ver) {
        // 不能使用；可能被最小版本号的读操作访问到。
        return 0
    }
    fl.offset++
    fl.total--

    // 丢弃空节点并移动到下一个节点
    // 省略……

    return ptr
}

// a < b
func versionBefore(a, b uint64) bool {
    return int64(a-b) < 0
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32

向列表头部添加元素比以前更复杂，但也没有特别的要求。代码清单省略。

// 将一些新指针添加到头部并完成更新
func (fl *FreeList) Add(freed []uint64)
1
2

# 12.5 结束语

我们展示了不可变数据结构的一个显著优势，即实现读操作和写操作之间的并发很容易。只读事务可以按需运行，唯一的缺点是长时间运行的读操作会阻止页面复用。不过，读写事务通常预期运行时间较短，因为写操作是完全序列化的。这种并发程度对于某些用例来说可能已经足够。

在本专栏中，我们探讨了3个主要主题：持久化、索引和并发。现在是时候为我们的数据库添加一个用户界面——一种查询语言了，这将是下一章的主题。

11.原子事务
13.查询语言：解析器

← 13.查询语言：解析器→

最近更新
01
C++语言面试问题集锦 目录与说明
03-27
02
第一章 auto与类型推导
03-27
03
第二章 关键字static及其不同用法
03-27
更多文章>
Copyright © 2024-2025 沪ICP备2023015129号 张小方 版权所有