第四节 Commit()实现
第四节 Commit()实现

Commit()方法内部实现中，总体思路是：
先判定节点要不要合并、分裂
对空闲列表的判断，是否存在溢出的情况，溢出的话，需要重新分配空间
将事务中涉及改动的页进行排序(保证尽可能的顺序IO)，排序后循环写入到磁盘中，最后再执行刷盘
当数据写入成功后，再将元信息页写到磁盘中，刷盘以保证持久化
上述操作中，但凡有失败，当前事务都会进行回滚
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. }