wal.go 22 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880
  1. // Copyright 2015 The etcd Authors
  2. //
  3. // Licensed under the Apache License, Version 2.0 (the "License");
  4. // you may not use this file except in compliance with the License.
  5. // You may obtain a copy of the License at
  6. //
  7. // http://www.apache.org/licenses/LICENSE-2.0
  8. //
  9. // Unless required by applicable law or agreed to in writing, software
  10. // distributed under the License is distributed on an "AS IS" BASIS,
  11. // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  12. // See the License for the specific language governing permissions and
  13. // limitations under the License.
  14. package wal
  15. import (
  16. "bytes"
  17. "errors"
  18. "fmt"
  19. "hash/crc32"
  20. "io"
  21. "os"
  22. "path/filepath"
  23. "sync"
  24. "time"
  25. "go.etcd.io/etcd/pkg/fileutil"
  26. "go.etcd.io/etcd/pkg/pbutil"
  27. "go.etcd.io/etcd/raft"
  28. "go.etcd.io/etcd/raft/raftpb"
  29. "go.etcd.io/etcd/wal/walpb"
  30. "github.com/coreos/pkg/capnslog"
  31. "go.uber.org/zap"
  32. )
  33. const (
  34. metadataType int64 = iota + 1
  35. entryType
  36. stateType
  37. crcType
  38. snapshotType
  39. // warnSyncDuration is the amount of time allotted to an fsync before
  40. // logging a warning
  41. warnSyncDuration = time.Second
  42. )
  43. var (
  44. // SegmentSizeBytes is the preallocated size of each wal segment file.
  45. // The actual size might be larger than this. In general, the default
  46. // value should be used, but this is defined as an exported variable
  47. // so that tests can set a different segment size.
  48. SegmentSizeBytes int64 = 64 * 1000 * 1000 // 64MB
  49. plog = capnslog.NewPackageLogger("go.etcd.io/etcd", "wal")
  50. ErrMetadataConflict = errors.New("wal: conflicting metadata found")
  51. ErrFileNotFound = errors.New("wal: file not found")
  52. ErrCRCMismatch = errors.New("wal: crc mismatch")
  53. ErrSnapshotMismatch = errors.New("wal: snapshot mismatch")
  54. ErrSnapshotNotFound = errors.New("wal: snapshot not found")
  55. crcTable = crc32.MakeTable(crc32.Castagnoli)
  56. )
  57. // WAL is a logical representation of the stable storage.
  58. // WAL is either in read mode or append mode but not both.
  59. // A newly created WAL is in append mode, and ready for appending records.
  60. // A just opened WAL is in read mode, and ready for reading records.
  61. // The WAL will be ready for appending after reading out all the previous records.
  62. type WAL struct {
  63. lg *zap.Logger
  64. dir string // the living directory of the underlay files
  65. // dirFile is a fd for the wal directory for syncing on Rename
  66. dirFile *os.File
  67. metadata []byte // metadata recorded at the head of each WAL
  68. state raftpb.HardState // hardstate recorded at the head of WAL
  69. start walpb.Snapshot // snapshot to start reading
  70. decoder *decoder // decoder to decode records
  71. readClose func() error // closer for decode reader
  72. mu sync.Mutex
  73. enti uint64 // index of the last entry saved to the wal
  74. encoder *encoder // encoder to encode records
  75. locks []*fileutil.LockedFile // the locked files the WAL holds (the name is increasing)
  76. fp *filePipeline
  77. }
  78. // Create creates a WAL ready for appending records. The given metadata is
  79. // recorded at the head of each WAL file, and can be retrieved with ReadAll.
  80. func Create(lg *zap.Logger, dirpath string, metadata []byte) (*WAL, error) {
  81. if Exist(dirpath) {
  82. return nil, os.ErrExist
  83. }
  84. // keep temporary wal directory so WAL initialization appears atomic
  85. tmpdirpath := filepath.Clean(dirpath) + ".tmp"
  86. if fileutil.Exist(tmpdirpath) {
  87. if err := os.RemoveAll(tmpdirpath); err != nil {
  88. return nil, err
  89. }
  90. }
  91. if err := fileutil.CreateDirAll(tmpdirpath); err != nil {
  92. if lg != nil {
  93. lg.Warn(
  94. "failed to create a temporary WAL directory",
  95. zap.String("tmp-dir-path", tmpdirpath),
  96. zap.String("dir-path", dirpath),
  97. zap.Error(err),
  98. )
  99. }
  100. return nil, err
  101. }
  102. p := filepath.Join(tmpdirpath, walName(0, 0))
  103. f, err := fileutil.LockFile(p, os.O_WRONLY|os.O_CREATE, fileutil.PrivateFileMode)
  104. if err != nil {
  105. if lg != nil {
  106. lg.Warn(
  107. "failed to flock an initial WAL file",
  108. zap.String("path", p),
  109. zap.Error(err),
  110. )
  111. }
  112. return nil, err
  113. }
  114. if _, err = f.Seek(0, io.SeekEnd); err != nil {
  115. if lg != nil {
  116. lg.Warn(
  117. "failed to seek an initial WAL file",
  118. zap.String("path", p),
  119. zap.Error(err),
  120. )
  121. }
  122. return nil, err
  123. }
  124. if err = fileutil.Preallocate(f.File, SegmentSizeBytes, true); err != nil {
  125. if lg != nil {
  126. lg.Warn(
  127. "failed to preallocate an initial WAL file",
  128. zap.String("path", p),
  129. zap.Int64("segment-bytes", SegmentSizeBytes),
  130. zap.Error(err),
  131. )
  132. }
  133. return nil, err
  134. }
  135. w := &WAL{
  136. lg: lg,
  137. dir: dirpath,
  138. metadata: metadata,
  139. }
  140. w.encoder, err = newFileEncoder(f.File, 0)
  141. if err != nil {
  142. return nil, err
  143. }
  144. w.locks = append(w.locks, f)
  145. if err = w.saveCrc(0); err != nil {
  146. return nil, err
  147. }
  148. if err = w.encoder.encode(&walpb.Record{Type: metadataType, Data: metadata}); err != nil {
  149. return nil, err
  150. }
  151. if err = w.SaveSnapshot(walpb.Snapshot{}); err != nil {
  152. return nil, err
  153. }
  154. if w, err = w.renameWAL(tmpdirpath); err != nil {
  155. if lg != nil {
  156. lg.Warn(
  157. "failed to rename the temporary WAL directory",
  158. zap.String("tmp-dir-path", tmpdirpath),
  159. zap.String("dir-path", w.dir),
  160. zap.Error(err),
  161. )
  162. }
  163. return nil, err
  164. }
  165. // directory was renamed; sync parent dir to persist rename
  166. pdir, perr := fileutil.OpenDir(filepath.Dir(w.dir))
  167. if perr != nil {
  168. if lg != nil {
  169. lg.Warn(
  170. "failed to open the parent data directory",
  171. zap.String("parent-dir-path", filepath.Dir(w.dir)),
  172. zap.String("dir-path", w.dir),
  173. zap.Error(perr),
  174. )
  175. }
  176. return nil, perr
  177. }
  178. if perr = fileutil.Fsync(pdir); perr != nil {
  179. if lg != nil {
  180. lg.Warn(
  181. "failed to fsync the parent data directory file",
  182. zap.String("parent-dir-path", filepath.Dir(w.dir)),
  183. zap.String("dir-path", w.dir),
  184. zap.Error(perr),
  185. )
  186. }
  187. return nil, perr
  188. }
  189. if perr = pdir.Close(); err != nil {
  190. if lg != nil {
  191. lg.Warn(
  192. "failed to close the parent data directory file",
  193. zap.String("parent-dir-path", filepath.Dir(w.dir)),
  194. zap.String("dir-path", w.dir),
  195. zap.Error(perr),
  196. )
  197. }
  198. return nil, perr
  199. }
  200. return w, nil
  201. }
  202. func (w *WAL) renameWAL(tmpdirpath string) (*WAL, error) {
  203. if err := os.RemoveAll(w.dir); err != nil {
  204. return nil, err
  205. }
  206. // On non-Windows platforms, hold the lock while renaming. Releasing
  207. // the lock and trying to reacquire it quickly can be flaky because
  208. // it's possible the process will fork to spawn a process while this is
  209. // happening. The fds are set up as close-on-exec by the Go runtime,
  210. // but there is a window between the fork and the exec where another
  211. // process holds the lock.
  212. if err := os.Rename(tmpdirpath, w.dir); err != nil {
  213. if _, ok := err.(*os.LinkError); ok {
  214. return w.renameWALUnlock(tmpdirpath)
  215. }
  216. return nil, err
  217. }
  218. w.fp = newFilePipeline(w.lg, w.dir, SegmentSizeBytes)
  219. df, err := fileutil.OpenDir(w.dir)
  220. w.dirFile = df
  221. return w, err
  222. }
  223. func (w *WAL) renameWALUnlock(tmpdirpath string) (*WAL, error) {
  224. // rename of directory with locked files doesn't work on windows/cifs;
  225. // close the WAL to release the locks so the directory can be renamed.
  226. if w.lg != nil {
  227. w.lg.Info(
  228. "closing WAL to release flock and retry directory renaming",
  229. zap.String("from", tmpdirpath),
  230. zap.String("to", w.dir),
  231. )
  232. } else {
  233. plog.Infof("releasing file lock to rename %q to %q", tmpdirpath, w.dir)
  234. }
  235. w.Close()
  236. if err := os.Rename(tmpdirpath, w.dir); err != nil {
  237. return nil, err
  238. }
  239. // reopen and relock
  240. newWAL, oerr := Open(w.lg, w.dir, walpb.Snapshot{})
  241. if oerr != nil {
  242. return nil, oerr
  243. }
  244. if _, _, _, err := newWAL.ReadAll(); err != nil {
  245. newWAL.Close()
  246. return nil, err
  247. }
  248. return newWAL, nil
  249. }
  250. // Open opens the WAL at the given snap.
  251. // The snap SHOULD have been previously saved to the WAL, or the following
  252. // ReadAll will fail.
  253. // The returned WAL is ready to read and the first record will be the one after
  254. // the given snap. The WAL cannot be appended to before reading out all of its
  255. // previous records.
  256. func Open(lg *zap.Logger, dirpath string, snap walpb.Snapshot) (*WAL, error) {
  257. w, err := openAtIndex(lg, dirpath, snap, true)
  258. if err != nil {
  259. return nil, err
  260. }
  261. if w.dirFile, err = fileutil.OpenDir(w.dir); err != nil {
  262. return nil, err
  263. }
  264. return w, nil
  265. }
  266. // OpenForRead only opens the wal files for read.
  267. // Write on a read only wal panics.
  268. func OpenForRead(lg *zap.Logger, dirpath string, snap walpb.Snapshot) (*WAL, error) {
  269. return openAtIndex(lg, dirpath, snap, false)
  270. }
  271. func openAtIndex(lg *zap.Logger, dirpath string, snap walpb.Snapshot, write bool) (*WAL, error) {
  272. names, nameIndex, err := selectWALFiles(lg, dirpath, snap)
  273. if err != nil {
  274. return nil, err
  275. }
  276. rs, ls, closer, err := openWALFiles(lg, dirpath, names, nameIndex, write)
  277. if err != nil {
  278. return nil, err
  279. }
  280. // create a WAL ready for reading
  281. w := &WAL{
  282. lg: lg,
  283. dir: dirpath,
  284. start: snap,
  285. decoder: newDecoder(rs...),
  286. readClose: closer,
  287. locks: ls,
  288. }
  289. if write {
  290. // write reuses the file descriptors from read; don't close so
  291. // WAL can append without dropping the file lock
  292. w.readClose = nil
  293. if _, _, err := parseWALName(filepath.Base(w.tail().Name())); err != nil {
  294. closer()
  295. return nil, err
  296. }
  297. w.fp = newFilePipeline(lg, w.dir, SegmentSizeBytes)
  298. }
  299. return w, nil
  300. }
  301. func selectWALFiles(lg *zap.Logger, dirpath string, snap walpb.Snapshot) ([]string, int, error) {
  302. names, err := readWALNames(lg, dirpath)
  303. if err != nil {
  304. return nil, -1, err
  305. }
  306. nameIndex, ok := searchIndex(lg, names, snap.Index)
  307. if !ok || !isValidSeq(lg, names[nameIndex:]) {
  308. err = ErrFileNotFound
  309. return nil, -1, err
  310. }
  311. return names, nameIndex, nil
  312. }
  313. func openWALFiles(lg *zap.Logger, dirpath string, names []string, nameIndex int, write bool) ([]io.Reader, []*fileutil.LockedFile, func() error, error) {
  314. rcs := make([]io.ReadCloser, 0)
  315. rs := make([]io.Reader, 0)
  316. ls := make([]*fileutil.LockedFile, 0)
  317. for _, name := range names[nameIndex:] {
  318. p := filepath.Join(dirpath, name)
  319. if write {
  320. l, err := fileutil.TryLockFile(p, os.O_RDWR, fileutil.PrivateFileMode)
  321. if err != nil {
  322. closeAll(rcs...)
  323. return nil, nil, nil, err
  324. }
  325. ls = append(ls, l)
  326. rcs = append(rcs, l)
  327. } else {
  328. rf, err := os.OpenFile(p, os.O_RDONLY, fileutil.PrivateFileMode)
  329. if err != nil {
  330. closeAll(rcs...)
  331. return nil, nil, nil, err
  332. }
  333. ls = append(ls, nil)
  334. rcs = append(rcs, rf)
  335. }
  336. rs = append(rs, rcs[len(rcs)-1])
  337. }
  338. closer := func() error { return closeAll(rcs...) }
  339. return rs, ls, closer, nil
  340. }
  341. // ReadAll reads out records of the current WAL.
  342. // If opened in write mode, it must read out all records until EOF. Or an error
  343. // will be returned.
  344. // If opened in read mode, it will try to read all records if possible.
  345. // If it cannot read out the expected snap, it will return ErrSnapshotNotFound.
  346. // If loaded snap doesn't match with the expected one, it will return
  347. // all the records and error ErrSnapshotMismatch.
  348. // TODO: detect not-last-snap error.
  349. // TODO: maybe loose the checking of match.
  350. // After ReadAll, the WAL will be ready for appending new records.
  351. func (w *WAL) ReadAll() (metadata []byte, state raftpb.HardState, ents []raftpb.Entry, err error) {
  352. w.mu.Lock()
  353. defer w.mu.Unlock()
  354. rec := &walpb.Record{}
  355. decoder := w.decoder
  356. var match bool
  357. for err = decoder.decode(rec); err == nil; err = decoder.decode(rec) {
  358. switch rec.Type {
  359. case entryType:
  360. e := mustUnmarshalEntry(rec.Data)
  361. if e.Index > w.start.Index {
  362. ents = append(ents[:e.Index-w.start.Index-1], e)
  363. }
  364. w.enti = e.Index
  365. case stateType:
  366. state = mustUnmarshalState(rec.Data)
  367. case metadataType:
  368. if metadata != nil && !bytes.Equal(metadata, rec.Data) {
  369. state.Reset()
  370. return nil, state, nil, ErrMetadataConflict
  371. }
  372. metadata = rec.Data
  373. case crcType:
  374. crc := decoder.crc.Sum32()
  375. // current crc of decoder must match the crc of the record.
  376. // do no need to match 0 crc, since the decoder is a new one at this case.
  377. if crc != 0 && rec.Validate(crc) != nil {
  378. state.Reset()
  379. return nil, state, nil, ErrCRCMismatch
  380. }
  381. decoder.updateCRC(rec.Crc)
  382. case snapshotType:
  383. var snap walpb.Snapshot
  384. pbutil.MustUnmarshal(&snap, rec.Data)
  385. if snap.Index == w.start.Index {
  386. if snap.Term != w.start.Term {
  387. state.Reset()
  388. return nil, state, nil, ErrSnapshotMismatch
  389. }
  390. match = true
  391. }
  392. default:
  393. state.Reset()
  394. return nil, state, nil, fmt.Errorf("unexpected block type %d", rec.Type)
  395. }
  396. }
  397. switch w.tail() {
  398. case nil:
  399. // We do not have to read out all entries in read mode.
  400. // The last record maybe a partial written one, so
  401. // ErrunexpectedEOF might be returned.
  402. if err != io.EOF && err != io.ErrUnexpectedEOF {
  403. state.Reset()
  404. return nil, state, nil, err
  405. }
  406. default:
  407. // We must read all of the entries if WAL is opened in write mode.
  408. if err != io.EOF {
  409. state.Reset()
  410. return nil, state, nil, err
  411. }
  412. // decodeRecord() will return io.EOF if it detects a zero record,
  413. // but this zero record may be followed by non-zero records from
  414. // a torn write. Overwriting some of these non-zero records, but
  415. // not all, will cause CRC errors on WAL open. Since the records
  416. // were never fully synced to disk in the first place, it's safe
  417. // to zero them out to avoid any CRC errors from new writes.
  418. if _, err = w.tail().Seek(w.decoder.lastOffset(), io.SeekStart); err != nil {
  419. return nil, state, nil, err
  420. }
  421. if err = fileutil.ZeroToEnd(w.tail().File); err != nil {
  422. return nil, state, nil, err
  423. }
  424. }
  425. err = nil
  426. if !match {
  427. err = ErrSnapshotNotFound
  428. }
  429. // close decoder, disable reading
  430. if w.readClose != nil {
  431. w.readClose()
  432. w.readClose = nil
  433. }
  434. w.start = walpb.Snapshot{}
  435. w.metadata = metadata
  436. if w.tail() != nil {
  437. // create encoder (chain crc with the decoder), enable appending
  438. w.encoder, err = newFileEncoder(w.tail().File, w.decoder.lastCRC())
  439. if err != nil {
  440. return
  441. }
  442. }
  443. w.decoder = nil
  444. return metadata, state, ents, err
  445. }
  446. // Verify reads through the given WAL and verifies that it is not corrupted.
  447. // It creates a new decoder to read through the records of the given WAL.
  448. // It does not conflict with any open WAL, but it is recommended not to
  449. // call this function after opening the WAL for writing.
  450. // If it cannot read out the expected snap, it will return ErrSnapshotNotFound.
  451. // If the loaded snap doesn't match with the expected one, it will
  452. // return error ErrSnapshotMismatch.
  453. func Verify(lg *zap.Logger, walDir string, snap walpb.Snapshot) error {
  454. var metadata []byte
  455. var err error
  456. var match bool
  457. rec := &walpb.Record{}
  458. names, nameIndex, err := selectWALFiles(lg, walDir, snap)
  459. if err != nil {
  460. return err
  461. }
  462. // open wal files in read mode, so that there is no conflict
  463. // when the same WAL is opened elsewhere in write mode
  464. rs, _, closer, err := openWALFiles(lg, walDir, names, nameIndex, false)
  465. if err != nil {
  466. return err
  467. }
  468. // create a new decoder from the readers on the WAL files
  469. decoder := newDecoder(rs...)
  470. for err = decoder.decode(rec); err == nil; err = decoder.decode(rec) {
  471. switch rec.Type {
  472. case metadataType:
  473. if metadata != nil && !bytes.Equal(metadata, rec.Data) {
  474. return ErrMetadataConflict
  475. }
  476. metadata = rec.Data
  477. case crcType:
  478. crc := decoder.crc.Sum32()
  479. // Current crc of decoder must match the crc of the record.
  480. // We need not match 0 crc, since the decoder is a new one at this point.
  481. if crc != 0 && rec.Validate(crc) != nil {
  482. return ErrCRCMismatch
  483. }
  484. decoder.updateCRC(rec.Crc)
  485. case snapshotType:
  486. var loadedSnap walpb.Snapshot
  487. pbutil.MustUnmarshal(&loadedSnap, rec.Data)
  488. if loadedSnap.Index == snap.Index {
  489. if loadedSnap.Term != snap.Term {
  490. return ErrSnapshotMismatch
  491. }
  492. match = true
  493. }
  494. // We ignore all entry and state type records as these
  495. // are not necessary for validating the WAL contents
  496. case entryType:
  497. case stateType:
  498. default:
  499. return fmt.Errorf("unexpected block type %d", rec.Type)
  500. }
  501. }
  502. if closer != nil {
  503. closer()
  504. }
  505. // We do not have to read out all the WAL entries
  506. // as the decoder is opened in read mode.
  507. if err != io.EOF && err != io.ErrUnexpectedEOF {
  508. return err
  509. }
  510. if !match {
  511. return ErrSnapshotNotFound
  512. }
  513. return nil
  514. }
  515. // cut closes current file written and creates a new one ready to append.
  516. // cut first creates a temp wal file and writes necessary headers into it.
  517. // Then cut atomically rename temp wal file to a wal file.
  518. func (w *WAL) cut() error {
  519. // close old wal file; truncate to avoid wasting space if an early cut
  520. off, serr := w.tail().Seek(0, io.SeekCurrent)
  521. if serr != nil {
  522. return serr
  523. }
  524. if err := w.tail().Truncate(off); err != nil {
  525. return err
  526. }
  527. if err := w.sync(); err != nil {
  528. return err
  529. }
  530. fpath := filepath.Join(w.dir, walName(w.seq()+1, w.enti+1))
  531. // create a temp wal file with name sequence + 1, or truncate the existing one
  532. newTail, err := w.fp.Open()
  533. if err != nil {
  534. return err
  535. }
  536. // update writer and save the previous crc
  537. w.locks = append(w.locks, newTail)
  538. prevCrc := w.encoder.crc.Sum32()
  539. w.encoder, err = newFileEncoder(w.tail().File, prevCrc)
  540. if err != nil {
  541. return err
  542. }
  543. if err = w.saveCrc(prevCrc); err != nil {
  544. return err
  545. }
  546. if err = w.encoder.encode(&walpb.Record{Type: metadataType, Data: w.metadata}); err != nil {
  547. return err
  548. }
  549. if err = w.saveState(&w.state); err != nil {
  550. return err
  551. }
  552. // atomically move temp wal file to wal file
  553. if err = w.sync(); err != nil {
  554. return err
  555. }
  556. off, err = w.tail().Seek(0, io.SeekCurrent)
  557. if err != nil {
  558. return err
  559. }
  560. if err = os.Rename(newTail.Name(), fpath); err != nil {
  561. return err
  562. }
  563. if err = fileutil.Fsync(w.dirFile); err != nil {
  564. return err
  565. }
  566. // reopen newTail with its new path so calls to Name() match the wal filename format
  567. newTail.Close()
  568. if newTail, err = fileutil.LockFile(fpath, os.O_WRONLY, fileutil.PrivateFileMode); err != nil {
  569. return err
  570. }
  571. if _, err = newTail.Seek(off, io.SeekStart); err != nil {
  572. return err
  573. }
  574. w.locks[len(w.locks)-1] = newTail
  575. prevCrc = w.encoder.crc.Sum32()
  576. w.encoder, err = newFileEncoder(w.tail().File, prevCrc)
  577. if err != nil {
  578. return err
  579. }
  580. if w.lg != nil {
  581. w.lg.Info("created a new WAL segment", zap.String("path", fpath))
  582. } else {
  583. plog.Infof("segmented wal file %v is created", fpath)
  584. }
  585. return nil
  586. }
  587. func (w *WAL) sync() error {
  588. if w.encoder != nil {
  589. if err := w.encoder.flush(); err != nil {
  590. return err
  591. }
  592. }
  593. start := time.Now()
  594. err := fileutil.Fdatasync(w.tail().File)
  595. took := time.Since(start)
  596. if took > warnSyncDuration {
  597. if w.lg != nil {
  598. w.lg.Warn(
  599. "slow fdatasync",
  600. zap.Duration("took", took),
  601. zap.Duration("expected-duration", warnSyncDuration),
  602. )
  603. } else {
  604. plog.Warningf("sync duration of %v, expected less than %v", took, warnSyncDuration)
  605. }
  606. }
  607. walFsyncSec.Observe(took.Seconds())
  608. return err
  609. }
  610. // ReleaseLockTo releases the locks, which has smaller index than the given index
  611. // except the largest one among them.
  612. // For example, if WAL is holding lock 1,2,3,4,5,6, ReleaseLockTo(4) will release
  613. // lock 1,2 but keep 3. ReleaseLockTo(5) will release 1,2,3 but keep 4.
  614. func (w *WAL) ReleaseLockTo(index uint64) error {
  615. w.mu.Lock()
  616. defer w.mu.Unlock()
  617. if len(w.locks) == 0 {
  618. return nil
  619. }
  620. var smaller int
  621. found := false
  622. for i, l := range w.locks {
  623. _, lockIndex, err := parseWALName(filepath.Base(l.Name()))
  624. if err != nil {
  625. return err
  626. }
  627. if lockIndex >= index {
  628. smaller = i - 1
  629. found = true
  630. break
  631. }
  632. }
  633. // if no lock index is greater than the release index, we can
  634. // release lock up to the last one(excluding).
  635. if !found {
  636. smaller = len(w.locks) - 1
  637. }
  638. if smaller <= 0 {
  639. return nil
  640. }
  641. for i := 0; i < smaller; i++ {
  642. if w.locks[i] == nil {
  643. continue
  644. }
  645. w.locks[i].Close()
  646. }
  647. w.locks = w.locks[smaller:]
  648. return nil
  649. }
  650. // Close closes the current WAL file and directory.
  651. func (w *WAL) Close() error {
  652. w.mu.Lock()
  653. defer w.mu.Unlock()
  654. if w.fp != nil {
  655. w.fp.Close()
  656. w.fp = nil
  657. }
  658. if w.tail() != nil {
  659. if err := w.sync(); err != nil {
  660. return err
  661. }
  662. }
  663. for _, l := range w.locks {
  664. if l == nil {
  665. continue
  666. }
  667. if err := l.Close(); err != nil {
  668. if w.lg != nil {
  669. w.lg.Warn("failed to close WAL", zap.Error(err))
  670. } else {
  671. plog.Errorf("failed to unlock during closing wal: %s", err)
  672. }
  673. }
  674. }
  675. return w.dirFile.Close()
  676. }
  677. func (w *WAL) saveEntry(e *raftpb.Entry) error {
  678. // TODO: add MustMarshalTo to reduce one allocation.
  679. b := pbutil.MustMarshal(e)
  680. rec := &walpb.Record{Type: entryType, Data: b}
  681. if err := w.encoder.encode(rec); err != nil {
  682. return err
  683. }
  684. w.enti = e.Index
  685. return nil
  686. }
  687. func (w *WAL) saveState(s *raftpb.HardState) error {
  688. if raft.IsEmptyHardState(*s) {
  689. return nil
  690. }
  691. w.state = *s
  692. b := pbutil.MustMarshal(s)
  693. rec := &walpb.Record{Type: stateType, Data: b}
  694. return w.encoder.encode(rec)
  695. }
  696. func (w *WAL) Save(st raftpb.HardState, ents []raftpb.Entry) error {
  697. w.mu.Lock()
  698. defer w.mu.Unlock()
  699. // short cut, do not call sync
  700. if raft.IsEmptyHardState(st) && len(ents) == 0 {
  701. return nil
  702. }
  703. mustSync := raft.MustSync(st, w.state, len(ents))
  704. // TODO(xiangli): no more reference operator
  705. for i := range ents {
  706. if err := w.saveEntry(&ents[i]); err != nil {
  707. return err
  708. }
  709. }
  710. if err := w.saveState(&st); err != nil {
  711. return err
  712. }
  713. curOff, err := w.tail().Seek(0, io.SeekCurrent)
  714. if err != nil {
  715. return err
  716. }
  717. if curOff < SegmentSizeBytes {
  718. if mustSync {
  719. return w.sync()
  720. }
  721. return nil
  722. }
  723. return w.cut()
  724. }
  725. func (w *WAL) SaveSnapshot(e walpb.Snapshot) error {
  726. b := pbutil.MustMarshal(&e)
  727. w.mu.Lock()
  728. defer w.mu.Unlock()
  729. rec := &walpb.Record{Type: snapshotType, Data: b}
  730. if err := w.encoder.encode(rec); err != nil {
  731. return err
  732. }
  733. // update enti only when snapshot is ahead of last index
  734. if w.enti < e.Index {
  735. w.enti = e.Index
  736. }
  737. return w.sync()
  738. }
  739. func (w *WAL) saveCrc(prevCrc uint32) error {
  740. return w.encoder.encode(&walpb.Record{Type: crcType, Crc: prevCrc})
  741. }
  742. func (w *WAL) tail() *fileutil.LockedFile {
  743. if len(w.locks) > 0 {
  744. return w.locks[len(w.locks)-1]
  745. }
  746. return nil
  747. }
  748. func (w *WAL) seq() uint64 {
  749. t := w.tail()
  750. if t == nil {
  751. return 0
  752. }
  753. seq, _, err := parseWALName(filepath.Base(t.Name()))
  754. if err != nil {
  755. if w.lg != nil {
  756. w.lg.Fatal("failed to parse WAL name", zap.String("name", t.Name()), zap.Error(err))
  757. } else {
  758. plog.Fatalf("bad wal name %s (%v)", t.Name(), err)
  759. }
  760. }
  761. return seq
  762. }
  763. func closeAll(rcs ...io.ReadCloser) error {
  764. for _, f := range rcs {
  765. if err := f.Close(); err != nil {
  766. return err
  767. }
  768. }
  769. return nil
  770. }