wal.go 22 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879
  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. dir: dirpath,
  283. start: snap,
  284. decoder: newDecoder(rs...),
  285. readClose: closer,
  286. locks: ls,
  287. }
  288. if write {
  289. // write reuses the file descriptors from read; don't close so
  290. // WAL can append without dropping the file lock
  291. w.readClose = nil
  292. if _, _, err := parseWALName(filepath.Base(w.tail().Name())); err != nil {
  293. closer()
  294. return nil, err
  295. }
  296. w.fp = newFilePipeline(lg, w.dir, SegmentSizeBytes)
  297. }
  298. return w, nil
  299. }
  300. func selectWALFiles(lg *zap.Logger, dirpath string, snap walpb.Snapshot) ([]string, int, error) {
  301. names, err := readWALNames(lg, dirpath)
  302. if err != nil {
  303. return nil, -1, err
  304. }
  305. nameIndex, ok := searchIndex(lg, names, snap.Index)
  306. if !ok || !isValidSeq(lg, names[nameIndex:]) {
  307. err = ErrFileNotFound
  308. return nil, -1, err
  309. }
  310. return names, nameIndex, nil
  311. }
  312. func openWALFiles(lg *zap.Logger, dirpath string, names []string, nameIndex int, write bool) ([]io.Reader, []*fileutil.LockedFile, func() error, error) {
  313. rcs := make([]io.ReadCloser, 0)
  314. rs := make([]io.Reader, 0)
  315. ls := make([]*fileutil.LockedFile, 0)
  316. for _, name := range names[nameIndex:] {
  317. p := filepath.Join(dirpath, name)
  318. if write {
  319. l, err := fileutil.TryLockFile(p, os.O_RDWR, fileutil.PrivateFileMode)
  320. if err != nil {
  321. closeAll(rcs...)
  322. return nil, nil, nil, err
  323. }
  324. ls = append(ls, l)
  325. rcs = append(rcs, l)
  326. } else {
  327. rf, err := os.OpenFile(p, os.O_RDONLY, fileutil.PrivateFileMode)
  328. if err != nil {
  329. closeAll(rcs...)
  330. return nil, nil, nil, err
  331. }
  332. ls = append(ls, nil)
  333. rcs = append(rcs, rf)
  334. }
  335. rs = append(rs, rcs[len(rcs)-1])
  336. }
  337. closer := func() error { return closeAll(rcs...) }
  338. return rs, ls, closer, nil
  339. }
  340. // ReadAll reads out records of the current WAL.
  341. // If opened in write mode, it must read out all records until EOF. Or an error
  342. // will be returned.
  343. // If opened in read mode, it will try to read all records if possible.
  344. // If it cannot read out the expected snap, it will return ErrSnapshotNotFound.
  345. // If loaded snap doesn't match with the expected one, it will return
  346. // all the records and error ErrSnapshotMismatch.
  347. // TODO: detect not-last-snap error.
  348. // TODO: maybe loose the checking of match.
  349. // After ReadAll, the WAL will be ready for appending new records.
  350. func (w *WAL) ReadAll() (metadata []byte, state raftpb.HardState, ents []raftpb.Entry, err error) {
  351. w.mu.Lock()
  352. defer w.mu.Unlock()
  353. rec := &walpb.Record{}
  354. decoder := w.decoder
  355. var match bool
  356. for err = decoder.decode(rec); err == nil; err = decoder.decode(rec) {
  357. switch rec.Type {
  358. case entryType:
  359. e := mustUnmarshalEntry(rec.Data)
  360. if e.Index > w.start.Index {
  361. ents = append(ents[:e.Index-w.start.Index-1], e)
  362. }
  363. w.enti = e.Index
  364. case stateType:
  365. state = mustUnmarshalState(rec.Data)
  366. case metadataType:
  367. if metadata != nil && !bytes.Equal(metadata, rec.Data) {
  368. state.Reset()
  369. return nil, state, nil, ErrMetadataConflict
  370. }
  371. metadata = rec.Data
  372. case crcType:
  373. crc := decoder.crc.Sum32()
  374. // current crc of decoder must match the crc of the record.
  375. // do no need to match 0 crc, since the decoder is a new one at this case.
  376. if crc != 0 && rec.Validate(crc) != nil {
  377. state.Reset()
  378. return nil, state, nil, ErrCRCMismatch
  379. }
  380. decoder.updateCRC(rec.Crc)
  381. case snapshotType:
  382. var snap walpb.Snapshot
  383. pbutil.MustUnmarshal(&snap, rec.Data)
  384. if snap.Index == w.start.Index {
  385. if snap.Term != w.start.Term {
  386. state.Reset()
  387. return nil, state, nil, ErrSnapshotMismatch
  388. }
  389. match = true
  390. }
  391. default:
  392. state.Reset()
  393. return nil, state, nil, fmt.Errorf("unexpected block type %d", rec.Type)
  394. }
  395. }
  396. switch w.tail() {
  397. case nil:
  398. // We do not have to read out all entries in read mode.
  399. // The last record maybe a partial written one, so
  400. // ErrunexpectedEOF might be returned.
  401. if err != io.EOF && err != io.ErrUnexpectedEOF {
  402. state.Reset()
  403. return nil, state, nil, err
  404. }
  405. default:
  406. // We must read all of the entries if WAL is opened in write mode.
  407. if err != io.EOF {
  408. state.Reset()
  409. return nil, state, nil, err
  410. }
  411. // decodeRecord() will return io.EOF if it detects a zero record,
  412. // but this zero record may be followed by non-zero records from
  413. // a torn write. Overwriting some of these non-zero records, but
  414. // not all, will cause CRC errors on WAL open. Since the records
  415. // were never fully synced to disk in the first place, it's safe
  416. // to zero them out to avoid any CRC errors from new writes.
  417. if _, err = w.tail().Seek(w.decoder.lastOffset(), io.SeekStart); err != nil {
  418. return nil, state, nil, err
  419. }
  420. if err = fileutil.ZeroToEnd(w.tail().File); err != nil {
  421. return nil, state, nil, err
  422. }
  423. }
  424. err = nil
  425. if !match {
  426. err = ErrSnapshotNotFound
  427. }
  428. // close decoder, disable reading
  429. if w.readClose != nil {
  430. w.readClose()
  431. w.readClose = nil
  432. }
  433. w.start = walpb.Snapshot{}
  434. w.metadata = metadata
  435. if w.tail() != nil {
  436. // create encoder (chain crc with the decoder), enable appending
  437. w.encoder, err = newFileEncoder(w.tail().File, w.decoder.lastCRC())
  438. if err != nil {
  439. return
  440. }
  441. }
  442. w.decoder = nil
  443. return metadata, state, ents, err
  444. }
  445. // Verify reads through the given WAL and verifies that it is not corrupted.
  446. // It creates a new decoder to read through the records of the given WAL.
  447. // It does not conflict with any open WAL, but it is recommended not to
  448. // call this function after opening the WAL for writing.
  449. // If it cannot read out the expected snap, it will return ErrSnapshotNotFound.
  450. // If the loaded snap doesn't match with the expected one, it will
  451. // return error ErrSnapshotMismatch.
  452. func Verify(lg *zap.Logger, walDir string, snap walpb.Snapshot) error {
  453. var metadata []byte
  454. var err error
  455. var match bool
  456. rec := &walpb.Record{}
  457. names, nameIndex, err := selectWALFiles(lg, walDir, snap)
  458. if err != nil {
  459. return err
  460. }
  461. // open wal files in read mode, so that there is no conflict
  462. // when the same WAL is opened elsewhere in write mode
  463. rs, _, closer, err := openWALFiles(lg, walDir, names, nameIndex, false)
  464. if err != nil {
  465. return err
  466. }
  467. // create a new decoder from the readers on the WAL files
  468. decoder := newDecoder(rs...)
  469. for err = decoder.decode(rec); err == nil; err = decoder.decode(rec) {
  470. switch rec.Type {
  471. case metadataType:
  472. if metadata != nil && !bytes.Equal(metadata, rec.Data) {
  473. return ErrMetadataConflict
  474. }
  475. metadata = rec.Data
  476. case crcType:
  477. crc := decoder.crc.Sum32()
  478. // Current crc of decoder must match the crc of the record.
  479. // We need not match 0 crc, since the decoder is a new one at this point.
  480. if crc != 0 && rec.Validate(crc) != nil {
  481. return ErrCRCMismatch
  482. }
  483. decoder.updateCRC(rec.Crc)
  484. case snapshotType:
  485. var loadedSnap walpb.Snapshot
  486. pbutil.MustUnmarshal(&loadedSnap, rec.Data)
  487. if loadedSnap.Index == snap.Index {
  488. if loadedSnap.Term != snap.Term {
  489. return ErrSnapshotMismatch
  490. }
  491. match = true
  492. }
  493. // We ignore all entry and state type records as these
  494. // are not necessary for validating the WAL contents
  495. case entryType:
  496. case stateType:
  497. default:
  498. return fmt.Errorf("unexpected block type %d", rec.Type)
  499. }
  500. }
  501. if closer != nil {
  502. closer()
  503. }
  504. // We do not have to read out all the WAL entries
  505. // as the decoder is opened in read mode.
  506. if err != io.EOF && err != io.ErrUnexpectedEOF {
  507. return err
  508. }
  509. if !match {
  510. return ErrSnapshotNotFound
  511. }
  512. return nil
  513. }
  514. // cut closes current file written and creates a new one ready to append.
  515. // cut first creates a temp wal file and writes necessary headers into it.
  516. // Then cut atomically rename temp wal file to a wal file.
  517. func (w *WAL) cut() error {
  518. // close old wal file; truncate to avoid wasting space if an early cut
  519. off, serr := w.tail().Seek(0, io.SeekCurrent)
  520. if serr != nil {
  521. return serr
  522. }
  523. if err := w.tail().Truncate(off); err != nil {
  524. return err
  525. }
  526. if err := w.sync(); err != nil {
  527. return err
  528. }
  529. fpath := filepath.Join(w.dir, walName(w.seq()+1, w.enti+1))
  530. // create a temp wal file with name sequence + 1, or truncate the existing one
  531. newTail, err := w.fp.Open()
  532. if err != nil {
  533. return err
  534. }
  535. // update writer and save the previous crc
  536. w.locks = append(w.locks, newTail)
  537. prevCrc := w.encoder.crc.Sum32()
  538. w.encoder, err = newFileEncoder(w.tail().File, prevCrc)
  539. if err != nil {
  540. return err
  541. }
  542. if err = w.saveCrc(prevCrc); err != nil {
  543. return err
  544. }
  545. if err = w.encoder.encode(&walpb.Record{Type: metadataType, Data: w.metadata}); err != nil {
  546. return err
  547. }
  548. if err = w.saveState(&w.state); err != nil {
  549. return err
  550. }
  551. // atomically move temp wal file to wal file
  552. if err = w.sync(); err != nil {
  553. return err
  554. }
  555. off, err = w.tail().Seek(0, io.SeekCurrent)
  556. if err != nil {
  557. return err
  558. }
  559. if err = os.Rename(newTail.Name(), fpath); err != nil {
  560. return err
  561. }
  562. if err = fileutil.Fsync(w.dirFile); err != nil {
  563. return err
  564. }
  565. // reopen newTail with its new path so calls to Name() match the wal filename format
  566. newTail.Close()
  567. if newTail, err = fileutil.LockFile(fpath, os.O_WRONLY, fileutil.PrivateFileMode); err != nil {
  568. return err
  569. }
  570. if _, err = newTail.Seek(off, io.SeekStart); err != nil {
  571. return err
  572. }
  573. w.locks[len(w.locks)-1] = newTail
  574. prevCrc = w.encoder.crc.Sum32()
  575. w.encoder, err = newFileEncoder(w.tail().File, prevCrc)
  576. if err != nil {
  577. return err
  578. }
  579. if w.lg != nil {
  580. w.lg.Info("created a new WAL segment", zap.String("path", fpath))
  581. } else {
  582. plog.Infof("segmented wal file %v is created", fpath)
  583. }
  584. return nil
  585. }
  586. func (w *WAL) sync() error {
  587. if w.encoder != nil {
  588. if err := w.encoder.flush(); err != nil {
  589. return err
  590. }
  591. }
  592. start := time.Now()
  593. err := fileutil.Fdatasync(w.tail().File)
  594. took := time.Since(start)
  595. if took > warnSyncDuration {
  596. if w.lg != nil {
  597. w.lg.Warn(
  598. "slow fdatasync",
  599. zap.Duration("took", took),
  600. zap.Duration("expected-duration", warnSyncDuration),
  601. )
  602. } else {
  603. plog.Warningf("sync duration of %v, expected less than %v", took, warnSyncDuration)
  604. }
  605. }
  606. walFsyncSec.Observe(took.Seconds())
  607. return err
  608. }
  609. // ReleaseLockTo releases the locks, which has smaller index than the given index
  610. // except the largest one among them.
  611. // For example, if WAL is holding lock 1,2,3,4,5,6, ReleaseLockTo(4) will release
  612. // lock 1,2 but keep 3. ReleaseLockTo(5) will release 1,2,3 but keep 4.
  613. func (w *WAL) ReleaseLockTo(index uint64) error {
  614. w.mu.Lock()
  615. defer w.mu.Unlock()
  616. if len(w.locks) == 0 {
  617. return nil
  618. }
  619. var smaller int
  620. found := false
  621. for i, l := range w.locks {
  622. _, lockIndex, err := parseWALName(filepath.Base(l.Name()))
  623. if err != nil {
  624. return err
  625. }
  626. if lockIndex >= index {
  627. smaller = i - 1
  628. found = true
  629. break
  630. }
  631. }
  632. // if no lock index is greater than the release index, we can
  633. // release lock up to the last one(excluding).
  634. if !found {
  635. smaller = len(w.locks) - 1
  636. }
  637. if smaller <= 0 {
  638. return nil
  639. }
  640. for i := 0; i < smaller; i++ {
  641. if w.locks[i] == nil {
  642. continue
  643. }
  644. w.locks[i].Close()
  645. }
  646. w.locks = w.locks[smaller:]
  647. return nil
  648. }
  649. // Close closes the current WAL file and directory.
  650. func (w *WAL) Close() error {
  651. w.mu.Lock()
  652. defer w.mu.Unlock()
  653. if w.fp != nil {
  654. w.fp.Close()
  655. w.fp = nil
  656. }
  657. if w.tail() != nil {
  658. if err := w.sync(); err != nil {
  659. return err
  660. }
  661. }
  662. for _, l := range w.locks {
  663. if l == nil {
  664. continue
  665. }
  666. if err := l.Close(); err != nil {
  667. if w.lg != nil {
  668. w.lg.Warn("failed to close WAL", zap.Error(err))
  669. } else {
  670. plog.Errorf("failed to unlock during closing wal: %s", err)
  671. }
  672. }
  673. }
  674. return w.dirFile.Close()
  675. }
  676. func (w *WAL) saveEntry(e *raftpb.Entry) error {
  677. // TODO: add MustMarshalTo to reduce one allocation.
  678. b := pbutil.MustMarshal(e)
  679. rec := &walpb.Record{Type: entryType, Data: b}
  680. if err := w.encoder.encode(rec); err != nil {
  681. return err
  682. }
  683. w.enti = e.Index
  684. return nil
  685. }
  686. func (w *WAL) saveState(s *raftpb.HardState) error {
  687. if raft.IsEmptyHardState(*s) {
  688. return nil
  689. }
  690. w.state = *s
  691. b := pbutil.MustMarshal(s)
  692. rec := &walpb.Record{Type: stateType, Data: b}
  693. return w.encoder.encode(rec)
  694. }
  695. func (w *WAL) Save(st raftpb.HardState, ents []raftpb.Entry) error {
  696. w.mu.Lock()
  697. defer w.mu.Unlock()
  698. // short cut, do not call sync
  699. if raft.IsEmptyHardState(st) && len(ents) == 0 {
  700. return nil
  701. }
  702. mustSync := raft.MustSync(st, w.state, len(ents))
  703. // TODO(xiangli): no more reference operator
  704. for i := range ents {
  705. if err := w.saveEntry(&ents[i]); err != nil {
  706. return err
  707. }
  708. }
  709. if err := w.saveState(&st); err != nil {
  710. return err
  711. }
  712. curOff, err := w.tail().Seek(0, io.SeekCurrent)
  713. if err != nil {
  714. return err
  715. }
  716. if curOff < SegmentSizeBytes {
  717. if mustSync {
  718. return w.sync()
  719. }
  720. return nil
  721. }
  722. return w.cut()
  723. }
  724. func (w *WAL) SaveSnapshot(e walpb.Snapshot) error {
  725. b := pbutil.MustMarshal(&e)
  726. w.mu.Lock()
  727. defer w.mu.Unlock()
  728. rec := &walpb.Record{Type: snapshotType, Data: b}
  729. if err := w.encoder.encode(rec); err != nil {
  730. return err
  731. }
  732. // update enti only when snapshot is ahead of last index
  733. if w.enti < e.Index {
  734. w.enti = e.Index
  735. }
  736. return w.sync()
  737. }
  738. func (w *WAL) saveCrc(prevCrc uint32) error {
  739. return w.encoder.encode(&walpb.Record{Type: crcType, Crc: prevCrc})
  740. }
  741. func (w *WAL) tail() *fileutil.LockedFile {
  742. if len(w.locks) > 0 {
  743. return w.locks[len(w.locks)-1]
  744. }
  745. return nil
  746. }
  747. func (w *WAL) seq() uint64 {
  748. t := w.tail()
  749. if t == nil {
  750. return 0
  751. }
  752. seq, _, err := parseWALName(filepath.Base(t.Name()))
  753. if err != nil {
  754. if w.lg != nil {
  755. w.lg.Fatal("failed to parse WAL name", zap.String("name", t.Name()), zap.Error(err))
  756. } else {
  757. plog.Fatalf("bad wal name %s (%v)", t.Name(), err)
  758. }
  759. }
  760. return seq
  761. }
  762. func closeAll(rcs ...io.ReadCloser) error {
  763. for _, f := range rcs {
  764. if err := f.Close(); err != nil {
  765. return err
  766. }
  767. }
  768. return nil
  769. }