第四节 Commit()实现

Commit()方法内部实现中，总体思路是：

  1. 先判定节点要不要合并、分裂
  2. 对空闲列表的判断，是否存在溢出的情况，溢出的话，需要重新分配空间
  3. 将事务中涉及改动的页进行排序(保证尽可能的顺序IO)，排序后循环写入到磁盘中，最后再执行刷盘
  4. 当数据写入成功后，再将元信息页写到磁盘中，刷盘以保证持久化
  5. 上述操作中，但凡有失败，当前事务都会进行回滚
  1. // Commit writes all changes to disk and updates the meta page.
  2. // Returns an error if a disk write error occurs, or if Commit is
  3. // called on a read-only transaction.
  5. // 先更新数据然后再更新元信息
  6. // 更新数据成功、元信息未来得及更新机器就挂掉了。数据如何恢复？
  7. func (tx *Tx) Commit() error {
  8.     _assert(!tx.managed, "managed tx commit not allowed")
  9.     if tx.db == nil {
  10.         return ErrTxClosed
  11.     } else if !tx.writable {
  12.         return ErrTxNotWritable
  13.     }
  15.     // TODO(benbjohnson): Use vectorized I/O to write out dirty pages.
  17.     // 删除时，进行平衡，页合并
  18.     // Rebalance nodes which have had deletions.
  19.     var startTime = time.Now()
  20.     tx.root.rebalance()
  21.     if tx.stats.Rebalance > 0 {
  22.         tx.stats.RebalanceTime += time.Since(startTime)
  23.     }
  25.     // 页分裂
  26.     // spill data onto dirty pages.
  27.     startTime = time.Now()
  28.     // 这个内部会往缓存tx.pages中加page
  29.     if err := tx.root.spill(); err != nil {
  30.         tx.rollback()
  31.         return err
  32.     }
  33.     tx.stats.SpillTime += time.Since(startTime)
  35.     // Free the old root bucket.
  36.     tx.meta.root.root = tx.root.root
  38.     opgid := tx.meta.pgid
  40.     // Free the freelist and allocate new pages for it. This will overestimate
  41.     // the size of the freelist but not underestimate the size (which would be bad).
  42.     // 分配新的页面给freelist，然后将freelist写入新的页面
  43.     tx.db.freelist.free(tx.meta.txid, tx.db.page(tx.meta.freelist))
  44.     // 空闲列表可能会增加，因此需要重新分配页用来存储空闲列表
  45.     // 因为在开启写事务的时候，有去释放之前读事务占用的页信息，因此此处需要判断是否freelist会有溢出的问题
  46.     p, err := tx.allocate((tx.db.freelist.size() / tx.db.pageSize) + 1)
  47.     if err != nil {
  48.         tx.rollback()
  49.         return err
  50.     }
  51.     // 将freelist写入到连续的新页中
  52.     if err := tx.db.freelist.write(p); err != nil {
  53.         tx.rollback()
  54.         return err
  55.     }
  56.     // 更新元数据的页id
  57.     tx.meta.freelist = p.id
  59.     // If the high water mark has moved up then attempt to grow the database.
  60.     // 在allocate中有可能会更改meta.pgid
  61.     if tx.meta.pgid > opgid {
  62.         if err := tx.db.grow(int(tx.meta.pgid+1) * tx.db.pageSize); err != nil {
  63.             tx.rollback()
  64.             return err
  65.         }
  66.     }
  68.     // Write dirty pages to disk.
  69.     startTime = time.Now()
  70.     // 写数据
  71.     if err := tx.write(); err != nil {
  72.         tx.rollback()
  73.         return err
  74.     }
  76.     // If strict mode is enabled then perform a consistency check.
  77.     // Only the first consistency error is reported in the panic.
  78.     if tx.db.StrictMode {
  79.         ch := tx.Check()
  80.         var errs []string
  81.         for {
  82.             err, ok := <-ch
  83.             if !ok {
  84.                 break
  85.             }
  86.             errs = append(errs, err.Error())
  87.         }
  88.         if len(errs) > 0 {
  89.             panic("check fail: " + strings.Join(errs, "\n"))
  90.         }
  91.     }
  93.     // Write meta to disk.
  94.     // 元信息写入到磁盘
  95.     if err := tx.writeMeta(); err != nil {
  96.         tx.rollback()
  97.         return err
  98.     }
  99.     tx.stats.WriteTime += time.Since(startTime)
  101.     // Finalize the transaction.
  102.     tx.close()
  104.     // Execute commit handlers now that the locks have been removed.
  105.     for _, fn := range tx.commitHandlers {
  106.         fn()
  107.     }
  109.     return nil
  110. }
  112. // write writes any dirty pages to disk.
  113. func (tx *Tx) write() error {
  114.     // Sort pages by id.
  115.     // 保证写的页是有序的
  116.     pages := make(pages, 0, len(tx.pages))
  117.     for _, p := range tx.pages {
  118.         pages = append(pages, p)
  119.     }
  120.     // Clear out page cache early.
  121.     tx.pages = make(map[pgid]*page)
  122.     sort.Sort(pages)
  124.     // Write pages to disk in order.
  125.     for _, p := range pages {
  126.         // 页数和偏移量
  127.         size := (int(p.overflow) + 1) * tx.db.pageSize
  128.         offset := int64(p.id) * int64(tx.db.pageSize)
  130.         // Write out page in "max allocation" sized chunks.
  131.         ptr := (*[maxAllocSize]byte)(unsafe.Pointer(p))
  132.         // 循环写某一页
  133.         for {
  134.             // Limit our write to our max allocation size.
  135.             sz := size
  136.             // 2^31=2G
  137.             if sz > maxAllocSize-1 {
  138.                 sz = maxAllocSize - 1
  139.             }
  141.             // Write chunk to disk.
  142.             buf := ptr[:sz]
  143.             if _, err := tx.db.ops.writeAt(buf, offset); err != nil {
  144.                 return err
  145.             }
  147.             // Update statistics.
  148.             tx.stats.Write++
  150.             // Exit inner for loop if we've written all the chunks.
  151.             size -= sz
  152.             if size == 0 {
  153.                 break
  154.             }
  156.             // Otherwise move offset forward and move pointer to next chunk.
  157.             // 移动偏移量
  158.             offset += int64(sz)
  159.             // 同时指针也移动
  160.             ptr = (*[maxAllocSize]byte)(unsafe.Pointer(&ptr[sz]))
  161.         }
  162.     }
  164.     // Ignore file sync if flag is set on DB.
  165.     if !tx.db.NoSync || IgnoreNoSync {
  166.         if err := fdatasync(tx.db); err != nil {
  167.             return err
  168.         }
  169.     }
  171.     // Put small pages back to page pool.
  172.     for _, p := range pages {
  173.         // Ignore page sizes over 1 page.
  174.         // These are allocated using make() instead of the page pool.
  175.         if int(p.overflow) != 0 {
  176.             continue
  177.         }
  179.         buf := (*[maxAllocSize]byte)(unsafe.Pointer(p))[:tx.db.pageSize]
  181.         // See https://go.googlesource.com/go/+/f03c9202c43e0abb130669852082117ca50aa9b1
  182.         // 清空buf，然后放入pagePool中
  183.         for i := range buf {
  184.             buf[i] = 0
  185.         }
  186.         tx.db.pagePool.Put(buf)
  187.     }
  189.     return nil
  190. }
  192. // writeMeta writes the meta to the disk.
  193. func (tx *Tx) writeMeta() error {
  194.     // Create a temporary buffer for the meta page.
  195.     buf := make([]byte, tx.db.pageSize)
  196.     p := tx.db.pageInBuffer(buf, 0)
  197.     // 将事务的元信息写入到页中
  198.     tx.meta.write(p)
  200.     // Write the meta page to file.
  201.     if _, err := tx.db.ops.writeAt(buf, int64(p.id)*int64(tx.db.pageSize)); err != nil {
  202.         return err
  203.     }
  204.     if !tx.db.NoSync || IgnoreNoSync {
  205.         if err := fdatasync(tx.db); err != nil {
  206.             return err
  207.         }
  208.     }
  210.     // Update statistics.
  211.     tx.stats.Write++
  213.     return nil
  214. }
  216. // allocate returns a contiguous block of memory starting at a given page.
  217. // 分配一段连续的页
  218. func (tx *Tx) allocate(count int) (*page, error) {
  219.     p, err := tx.db.allocate(count)
  220.     if err != nil {
  221.         return nil, err
  222.     }
  224.     // Save to our page cache.
  225.     tx.pages[p.id] = p
  227.     // Update statistics.
  228.     tx.stats.PageCount++
  229.     tx.stats.PageAlloc += count * tx.db.pageSize
  231.     return p, nil
  232. }