| Method---wal.go | Description |
|---|---|
| func Create(lg *zap.Logger, dirpath string, metadata []byte) (*WAL, error) | 初次啟動raftNode時呼叫WAL.Create方法，創建WAL物件用于記錄追加 ：判斷是否存在dirpath路徑，如果已存在則不是初次啟動raftNode，回傳os.ErrExist，創建臨時目錄和初始上鎖的wal檔案—walName（seq=0 & index=0），seek到檔案末尾（why？），預分配該wal檔案大小（SegmentSizeBytes=64MB，優化追加速度），創建WAL物件并設定路徑、 metadata（NodeID和ClusterID）、編碼器，將上鎖的WAL檔案追加到鎖表內，然后依次寫入crc、metadata和空snapshot，重命名臨時目錄，同步臨時目錄的父目錄（fsync）使得重命名持久化， |
| func (w *WAL) renameWAL(tmpdirpath string) (*WAL, error) | 移除w.dir目錄及目錄下所有檔案和檔案夾，呼叫os.Rename(tmpdirpath, w.dir)將Create方法內創建的臨時目錄重命名，創建FilePipeline和dirFile *os.File，dirFile is a fd for the wal directory for syncing on Rename |
| func (w *WAL) SaveSnapshot(e walpb.Snapshot) error | 檢驗snapshot是否合法（Since etcd>=3.5.0），pb序列化snapshot得到data欄位，加鎖，呼叫w.encoder.encode方法寫入record，更新w.enti如果snapshot index > 原w.enti，解鎖， |
| func (w *WAL) saveCrc(prevCrc uint32) error | 寫入crcType的記錄 |
| func (w *WAL) Save(st raftpb.HardState, ents []raftpb.Entry) error | 加鎖，如果hardstate和entries為空，回傳，判斷是否需要sync（entries長度不為0||vote改變||term改變），寫入entries和hardstate，判斷檔案當前位置是否小于SegmentSizeBytes（默認64M），如果小于判斷是否需要sync資料，如果不小于，回傳cut操作結果，解鎖， |
| func (w *WAL) saveEntry(e *raftpb.Entry) error | 寫入一條entry記錄并更新WAL物件的enti值 |
| func (w *WAL) saveState(s *raftpb.HardState) error | 判斷是否為空，不為空寫入HardState并更新WAL物件的state值 |
| func (w *WAL) cut() error | 關閉當前檔案并創建一個新的檔案用于追加記錄：首先移動到鎖表最后一個wal檔案的當前位置截斷檔案然后執行sync，呼叫FilePipeline物件的Open方法創建一個新檔案并加入鎖表，首先保存舊的encoder的crc，然后創建新的encoder物件替換舊的encoder物件，保存頭資訊crc、metadata和hardstate，原子重命名檔案之前先執行sync和保存當前位置偏移，重命名后對WAL物件的dirFile執行fsync持久化wal目錄的變化，關閉檔案重新以LockFile方式打開檔案并seek到檔案末尾，替換鎖表尾檔案，再次進行新舊encoder替換， |
| func (w *WAL) tail() *fileutil.LockedFile | WAL物件的鎖表如果不為空，回傳最后一個上鎖檔案，否則回傳空， |
| func (w *WAL) sync() error | 如果encoder存在，則將encoder pageWriter緩沖區的資料寫入，鎖表尾檔案執行fdatasync，將fdatasync延時上報監控prometheus， |
| func ValidSnapshotEntries(lg *zap.Logger, walDir string) ([]walpb.Snapshot, error) | 尋找所有wal檔案中snapshot條目，有效的snapshot條目index必須小于等于最新的hardstate，步驟：找到目錄下所有帶有合法名稱的wal檔案名，以只讀模式打開這些wal檔案，根據以讀模式打開的這些wal檔案創建decoder，回圈解碼每個檔案的record：若為snapshotType，追加到snaps中；若為stateType，更新hardstate；若為crcType（wal檔案開頭），驗證是否和decoder.crc相同（上一個檔案末尾的crc），回傳所有index小于最新hardstate.Commit的walpb.snap條目， |
| func readWALNames(lg *zap.Logger, dirpath string) ([]string, error) | 從指定目錄讀取所有wal檔案name，并檢查name合法性（.wal結尾） |
| func openWALFiles(lg *zap.Logger, dirpath string, names []string, nameIndex int, write bool) ([]fileutil.FileReader, []*fileutil.LockedFile, func() error, error) | 根據write標志選擇以讀模式還是寫模式打開檔案，步驟：從nameIndex指定的索引開始打開檔案，若寫模式：打開上鎖的wal檔案，將該檔案添加到鎖表、檔案關閉表、讀檔案表；若讀模式：以os.O_RDONLY打開檔案，將該檔案添加到檔案關閉表、讀檔案表，添加nil到鎖表（鎖表只在寫模式下用到）， |
| func Open(lg zap.Logger, dirpath string, snap walpb.Snapshot) (WAL, error) | 寫模式呼叫openAtIndex，Open opens the WAL at the given snap，The returned WAL is ready to read and the first record will be the one after the given snap. The WAL cannot be appended to before reading out all of its previous records. |
| func OpenForRead(lg zap.Logger, dirpath string, snap walpb.Snapshot) (WAL, error) | 讀模式呼叫openAtIndex， |
| func openAtIndex(lg zap.Logger, dirpath string, snap walpb.Snapshot, write bool) (WAL, error) | 遍歷wal目錄，找到snap所在位置index，依次打開index后續所有的wal檔案并加鎖（呼叫openWALFiles），創建WAL物件并設定解碼器、readClose（呼叫closeAll關閉已打開檔案）和鎖表，若寫模式：readClose置空（寫模式下還要繼續對wal檔案進行append操作，等到讀完后不用進行關閉操作），測驗鎖表最后一個上鎖檔案是否是合法wal檔案（通過是否符合命名規范判斷），如果不合法，關閉所有檔案回傳錯誤，否則，設定FilePipeline物件（大小超過64M時用于截斷并切換到新檔案），回傳WAL物件 ， |
| func (w *WAL) ReadAll() (metadata []byte, state raftpb.HardState, ents []raftpb.Entry, err error) | 讀取WAL物件的所有記錄 ，對于不同型別記錄做不同處理： 判斷鎖表尾檔案如為空（讀模式）：如果不是讀到EOF或ErrUnexpectedEOF則重置state回傳；對于寫模式，如果err不是EOF，重置狀態回傳，然后鎖表尾檔案Seek到lastOffset位置， 將后續內容清零（目的是處理遇到0記錄后接非0記錄時，非0記錄又沒有被全部重寫，再次打開的時候會出現 CRC錯誤，由于資料從不會一開始就完全同步到磁盤，因此進行清零操作是安全的 ？暫時沒懂），然后判斷snapshot是否匹配，關閉decoder實作禁讀，重置WAL物件的start為一個空snapshot物件，創建encoder并將decoder設空，回傳metadata，state，ents和err， |
| Method---encoder.go | Description |
|---|---|
| func (e *encoder) encode(rec *walpb.Record) error | 加鎖，根據record.data計算crc，序列化record得data，根據len(data)計算出lenField和padBytes（需要填充位元組數使得8位元組對齊），寫入lenField到檔案，如果padBytes!=0，則給data填充padBytes位元組，寫入到pageWriter緩沖區， |
| func encodeFrameSize(dataBytes int) (lenField uint64, padBytes int) | 計算出lenField, padBytes |
| func (e *encoder) flush() error | 加鎖，將pageWriter緩沖區內資料寫入檔案，解鎖 |
| func writeUint64(w io.Writer, n uint64, buf []byte) error | 將uint64型別n寫入到[]byte內，再寫入檔案 |
| Method---pagewriter.go | Description |
|---|---|
| func (pw *PageWriter) Write(p []byte) (n int, err error) | 若未超出緩沖區最大容量，則將資料復制到緩沖區并更新當前緩沖位元組數回傳，若超出，計算出一頁還空閑多少位元組數slack，若還有空閑： |
| func (pw *PageWriter) flush() (int, error) | 將緩沖區內的所有資料寫入檔案，重新計算頁內偏移量（pageBytes=4KB），將已快取位元組數重置為0 |
| Method---decoder.go | Description |
|---|---|
| func (d *decoder) decodeRecord(rec *walpb.Record) error | 首先讀取一個int64變數，算出recBytes,和padBytes，讀取recBytes+padBytes大小的資料，將前recBytes大小的資料反序列化，如果不是crcType，驗證crc是否一致，更新lastOffset = 8 + recBytes + padBytes， |
| func decodeFrameSize(lenField int64) (recBytes int64, padBytes int64) | 根據lenField，算出實際資料大小和填充資料大小 |
| func (d *decoder) isTornEntry(data []byte) bool | 還沒看 |
| func readInt64(r io.Reader) (int64, error) | 從檔案中讀取一個int64變數 |
轉載請註明出處，本文鏈接：https://www.uj5u.com/houduan/549395.html
標籤：其他
上一篇：【開源免費】ChatGPT-Java版SDK重磅更新至1.0.10版，支持Tokens計算，快來一鍵接入。
下一篇：ps 備忘清單_開發速查表分享