Boltdb学习笔记之三--事务与并发控制

2022-03-17 00:00:00 函数 执行 事务 提交 读写

如果说数据库是软件工程领域的皇-冠，而事务与并发控制可称之为皇-冠上的钻石。本节将详细分析boltdb中如何实现事务与并发控制

事务

事务定义

boltdb中使用Tx
表示事务, 定义如下：

// Tx represents a read-only or read/write transaction on the database.
// Read-only transactions can be used for retrieving values for keys and creating cursors.
// Read/write transactions can create and remove buckets and create and remove keys.
//
// IMPORTANT: You must commit or rollback transactions when you are done with
// them. Pages can not be reclaimed by the writer until no more transactions
// are using them. A long running read transaction can cause the database to
// quickly grow.
type Tx struct {
 writable       bool
 managed        bool
 db             *DB
 meta           *meta
 root           Bucket
 pages          map[pgid]*page
 stats          TxStats
 commitHandlers []func()

 // WriteFlag specifies the flag for write-related methods like WriteTo().
 // Tx opens the database file with the specified flag to copy the data.
 //
 // By default, the flag is unset, which works well for mostly in-memory
 // workloads. For databases that are much larger than available RAM,
 // set the flag to syscall.O_DIRECT to avoid trashing the page cache.
 WriteFlag int
}

其中的成员：

  • writable
    : boltdb中事务分为两种，读事务(writable = 0)和读写事务(writable = 1)
  • managed
    : 用于保证用户读写事务的回调函数中不会执行事务提交或事务回滚。稍后将分析如何实现
  • db
    : 当前事务所绑定的DB
    对象
  • meta
    : 当前事务的meta数据(事务初始化时从DB
    中拷贝而来,在事务提交前只在内存中，当事务提交时持久化到磁盘中的meta page)
  • root
    : 当前事务的root Bucket(同meta
    , 事务初始化时从DB
    中拷贝而来，只在内存中，事务提交时会持久化)
  • pages
    : 如果当前事务属于读写事务，pages
    表示当前事务中待持久化到内存的脏页
  • stats
    : 当前事务执行过程中的统计数据，在事务提交时被汇总到当前DB
    的统计数据中。
  • commitHandlers
    : 当前事务提交时执行的钩子函数。用户可通过Tx.OnCommit
    注册。钩子函数按照注册顺序执行.
  • WriteFlag
    : 测试用，这里略过

事务初始化

在Boltdb学习笔记之〇--概述中我们提到，用户可通过DB.Update
新建一个读写事务，通过DB.View
新建一个只读事务。二者都调用了DB.Begin

func (db *DB) Begin(writable bool) (*Tx, error) {
 if writable {
  return db.beginRWTx()
 }
 return db.beginTx()
}

接下来我们分别分析读写事务和只读事务的初始化过程

读写事务的初始化

因此创建读写事务的调用链：

DB.Update
 -> DB.Begin
  -> DB.beginRWTx


`DB.beginRWTx`返回一个读写事务:
- 首先创建一个`wriable`为`true`的`Tx`对象`tx`
- 然后对其进行初始化：拷贝当前db中的meta和root Bucket。并创建`pages`用于存储执行读写事务过程中产生的脏页, 并自增本地meta副本中的txid
- 释放过期事务所占据的pending pages
- 返回读写事务

func (db *DB) beginRWTx() (*Tx, error) {
 ...

 // Create a transaction associated with the database.
 t := &Tx{writable: true}
 t.init(db)
 db.rwtx = t

 // Free any pages associated with closed read-only transactions.
 var minid txid = 0xFFFFFFFFFFFFFFFF
 for _, t := range db.txs {
  if t.meta.txid < minid {
   minid = t.meta.txid
  }
 }
 if minid >  {
  db.freelist.release(minid - 1)
 }
 return t, nil
}

只读事务的初始化

而创建只读事务的调用链：

DB.View
 -> DB.Begin
  -> DB.beginTx

DB.beginTx
返回一个只读事务，

  • 首先new
    一个新的Tx
    对象t
  • 然后对t
    进行初始化：初始化过程中会对当前db中的meta和root Bucket进行拷贝，并将副本放在t
    中
  • 将t
    加入到当前db的只读事务列表中
  • 返回只读事务t
func (db *DB) beginTx() (*Tx, error) {
 ...

 // Create a transaction associated with the database.
 t := &Tx{}
 t.init(db)

 // Keep track of transaction until it closes.
 db.txs = append(db.txs, t)
 n := len(db.txs)

 // Unlock the meta pages.
 db.metalock.Unlock()

 ...

 return t, nil
}

事务执行

事务初始化之后，便开始执行了。在读写事务中，首先执行用户注册的回调函数，如果回调函数没有返回错误，则提交事务，否则回滚。有的读者会问了，如果回调函数执行过程中发生panic
该如何处理呢？这里用了defer
来捕捉异常。

func (db *DB) Update(fn func(*Tx) error) error {
 ...

 // Make sure the transaction rolls back in the event of a panic.
 defer func() {
  if t.db != nil {
   t.rollback()
  }
 }()

 // Mark as a managed tx so that the inner function cannot manually commit.
 t.managed = true

 // If an error is returned from the function then rollback and return error.
 err = fn(t)
 t.managed = false
 if err != nil {
  _ = t.Rollback()
  return err
 }

 return t.Commit()
}

在只读事务中，首先执行用户注册的回调函数。之后的处理与读写事务不同：这里不管回调函数是否返回错误，都会执行回滚。其实没毛病，对于只读事务来说，其执行不改变db中任何数据，因此没什么好回滚的，Tx.Rollback
只是释放只读事务的资源并将其从当前db
中抹去。

func (db *DB) View(fn func(*Tx) error) error {
 ...
 if err != nil {
  _ = t.Rollback()
  return err
 }

 if err := t.Rollback(); err != nil {
  return err
 }

 return nil
}

事务提交

从上一节中我们看到，只有读写事务才会有提交。步骤如下：

  • 对root Bucket执行rebalance
    ，以合并过小的B+树节点: 递归的对所有修改过的子Bucket
    执行rebalance
    , 同时对当前Bucket
    中修改过的node
    执行rebalance
    。
  • 对root Bucket执行spill
    , 以分裂过大的B+树节点：递归的对所有子Bucket
    执行spill
    , 通知对当前Bucket
    的根节点执行spill
  • 更新新root Bucket到tx.meta
    中：执行上述两步之后，root Bucket可能已经更新(例如B+树节点分裂时产生新的根节点), 需要记录到当前事务的meta中
  • 在对root Bucket执行rebalance
    和spill
    过程中，会通过freelist.Allocate
    分配新page, 也会调用freelist.free
    释放很多老page, 此时内存中的freelist比磁盘中的新。因此分配一个新page, 用于持久化新版freelist。至于磁盘中老的freelist page，待当前事务完成提交之后便会释放。
  • 判断当前事务中大page id是否超出db文件, 如果是则对db文件执行扩容
  • 将内存中的dirty pages全部持久化到磁盘
  • 将tx.meta
    持久化化meta page中。注意，当前db中有两个meta page, 持久化的时候选择pageid = txid % 2
    的meta page。如此，相邻的写入事务提交时分别持久化到不同的meta page。为什么boltdb中要设计两个meta page呢？这是为了避免读写事务提交时，meta page持久化失败，此时该事务被回滚, 另一个meta page中的数据还是有效的，以此保证数据库的一致性

上述步骤中关于B+树节点合并与分裂的具体执行过程见Boltdb学习笔记之二--数据结构

事务回滚

当事务执行过程中返回错误或panic, 或事务提交失败(比如磁盘I/O失败)时，即对当前事务执行回滚。

对于读写事务来说，回滚分为三步

  • 对当前db的freelist
    进行回滚: 将penging pages从freelist中去除
  • 重新从磁盘中读取freelist page
  • 将当前事务从当前db中清除，并更新当前db的统计信息
func (tx *Tx) rollback() {
 if tx.db == nil {
  return
 }
 if tx.writable {
  tx.db.freelist.rollback(tx.meta.txid)
  tx.db.freelist.reload(tx.db.page(tx.db.meta().freelist))
 }
 tx.close()
}

对于只读事务，其回滚只有上述步骤的后一步

现在我们分析下Tx.managed
如何保证用户注册的回调函数中不会调用Commit
或Rollback
。

我们看到，不管是读写事务还是只读事务，在进入回调函数中之后，managed
必为true
，直到程序跳出回调函数。假设用户此时在其回调函数中手动调用Commit
或Rollback
, 则必然会panic，因为Commit
和Rollback
中会断言managed
为false

func (tx *Tx) Commit() error {
 _assert(!tx.managed, "managed tx commit not allowed")
 ...
}

func (tx *Tx) Rollback() error {
 _assert(!tx.managed, "managed tx rollback not allowed")
 ...
}

至于为什么不让用户在回调函数中调用Commit
或Rollback
, 个人理解应该是处于简化设计的目的, 让用户和boltdb的职责划分更明确：用户的职责是写好回调函数，在各种异常场景下返回错误；而boltdb的职责是根据回调函数是否返回错误决定Commit
或是Rollback

如何保证ACID

前面我们花了很大篇幅讲了boltdb中事务的实现细节。接下来我们分析boltdb中的事务如何满足ACID四个属性的：

原子性

对于只读事务，不修改任何数据，在查询过程中产生的缓存随着事务结束也会被释放掉，因此它是符合原子性的

对于读写事务来说，提交之前所有操作都在内存中，事务提交时，按照freelist、B+树数据和meta的顺序先后持久化到磁盘。只有meta成功持久化到磁盘之后，读写事务的操作才可见，换言之在此之前读写事务的操作都不可见。综上，boltdb中不管只读事务或读写事务都满足原子性

隔离性

boltdb中允许一个读写事务和多个只读事务执行。读写事务提交时只会分配新的page，直到在该事务之前的所有只读事务都完成才彻底释放旧page；而只读事务执行过程中不会释放和分配任何page。那么boltdb中如何保证读写事务和只读事务之间互不干扰的呢？以下我们分情况讨论

  1. 读写事务 + 读写事务 boltdb中通过互斥锁DB.rwlock
    保证了任何时刻多只有一个读写事务在运行。因此两个读写事务并存的情况不存在
func (db *DB) beginRWTx() (*Tx, error) {
 // If the database was opened with Options.ReadOnly, return an error.
 if db.readOnly {
  return nil, ErrDatabaseReadOnly
 }

 // Obtain writer lock. This is released by the transaction when it closes.
 // This enforces only one writer transaction at a time.
 db.rwlock.Lock()
 ...
}

  1. 只读事务 + 只读事务 只读事务不会新增/修改/删除任何page, 因此它们之间是互不影响的

  2. 读写事务 + 只读事务 boltdb中在创建一个新的读写事务时，首先会从只读事务中获取小txid, 并彻底释放小txid之前的已提交的读写事务的待释放page，即这些page可用于再次分配

 // Free any pages associated with closed read-only transactions.
 var minid txid = 0xFFFFFFFFFFFFFFFF
 for _, t := range db.txs {
  if t.meta.txid < minid {
   minid = t.meta.txid
  }
 }
 if minid > 0 {
  db.freelist.release(minid - 1)
 }

为什么要这样做呢？考虑这样一种时序，

初始状态：db meta txid为0

RW-1 begin -> RO-0 begin -> RW-1 commit -> RO-0 finish -> RW-2 begin

RW-n
表示读写事务，其中n表示该事务id, 为创建该事务时，当前db meta page中txid+1RO-n
表示只读事务，其中n表示该事务id, 为创建该事务时，当前db meta page中的txid

当执行RW-1 commit
时，会创建新page, 用于存储已更新的数据，同时更新db meta txid为1。注意此时对应的老的pending page还不能释放，因为事务RO-O
可能还在引用。只有当RO-0
完成后，此时再也没有只读事务引用txid=0版本的db。因此在RW-2
初始化时，即可彻底释放掉RW-1
中产生的老的pending page。

所以boltdb中保留了多种版本(用txid标识)的page, 当版本过期时便彻底释放掉对应的page用于再次分配，以此来保证读写事务和只读事务的隔离性

临界资源

  1. 读写事务 boltdb中多只能同时运行一个读写事务，使用互斥锁db.rwlock
    保护

  2. 元数据 不管是创建、提交、回滚事务，都涉及boltdb中元数据的读写，因此使用互斥锁DB.metalock
    保护之

  3. mmap缓冲区 只读事务会读取mmap缓冲区，但是读写事务有可能触发remmap，如果不对mmap缓冲区加以保护，将会导致只读事务读取到的mmap缓冲区过时。因此实现上，整个只读事务执行过程中都对mmap缓冲区加读锁，而对DB.mmap
    函数加写锁，保证数据一致性。

持久性

只读事务因为不修改任何数据，因此无所谓持久性。

读写事务中提交时，不管是freelist(调用freelist.write
)、B+树数据(调用Tx.write
)还是meta数据(调用Tx.writeMeta
)，都会被持久化到磁盘。

以Tx.writeMeta
为例, 首先分配pageSize
大小的缓冲区，并将meta序列化到该缓冲区内，再将缓冲区中的数据写入到磁盘上的meta page中，后调用fdatasync
将内核缓冲区中的数据全部flush到磁盘

// writeMeta writes the meta to the disk.
func (tx *Tx) writeMeta() error {
 // Create a temporary buffer for the meta page.
 buf := make([]byte, tx.db.pageSize)
 p := tx.db.pageInBuffer(buf, )
 tx.meta.write(p)

 // Write the meta page to file.
 if _, err := tx.db.ops.writeAt(buf, int64(p.id)*int64(tx.db.pageSize)); err != nil {
  return err
 }
 if !tx.db.NoSync || IgnoreNoSync {
  if err := fdatasync(tx.db); err != nil {
   return err
  }
 }

 // Update statistics.
 tx.stats.Write++

 return nil
}


相关文章