2016-05-13 06:50:04 +03:00
|
|
|
// Copyright 2015 The etcd Authors
|
2015-01-25 06:19:16 +03:00
|
|
|
//
|
|
|
|
// Licensed under the Apache License, Version 2.0 (the "License");
|
|
|
|
// you may not use this file except in compliance with the License.
|
|
|
|
// You may obtain a copy of the License at
|
|
|
|
//
|
|
|
|
// http://www.apache.org/licenses/LICENSE-2.0
|
|
|
|
//
|
|
|
|
// Unless required by applicable law or agreed to in writing, software
|
|
|
|
// distributed under the License is distributed on an "AS IS" BASIS,
|
|
|
|
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
|
|
|
// See the License for the specific language governing permissions and
|
|
|
|
// limitations under the License.
|
2014-07-28 07:58:39 +04:00
|
|
|
|
2014-07-26 01:20:21 +04:00
|
|
|
package wal
|
|
|
|
|
|
|
|
import (
|
2016-06-13 11:33:53 +03:00
|
|
|
"bytes"
|
2014-09-04 02:10:15 +04:00
|
|
|
"errors"
|
2014-07-26 01:20:21 +04:00
|
|
|
"fmt"
|
2014-09-04 02:10:15 +04:00
|
|
|
"hash/crc32"
|
2014-07-26 11:10:59 +04:00
|
|
|
"io"
|
2014-07-26 01:20:21 +04:00
|
|
|
"os"
|
2017-03-16 03:18:55 +03:00
|
|
|
"path/filepath"
|
2015-04-08 10:25:18 +03:00
|
|
|
"sync"
|
2015-01-30 00:07:23 +03:00
|
|
|
"time"
|
2014-07-26 01:20:21 +04:00
|
|
|
|
2019-05-29 00:59:02 +03:00
|
|
|
"go.etcd.io/etcd/pkg/fileutil"
|
|
|
|
"go.etcd.io/etcd/pkg/pbutil"
|
|
|
|
"go.etcd.io/etcd/raft"
|
|
|
|
"go.etcd.io/etcd/raft/raftpb"
|
|
|
|
"go.etcd.io/etcd/wal/walpb"
|
2015-06-01 23:38:50 +03:00
|
|
|
|
2016-03-23 03:10:28 +03:00
|
|
|
"github.com/coreos/pkg/capnslog"
|
2018-04-16 13:58:25 +03:00
|
|
|
"go.uber.org/zap"
|
2014-07-26 01:20:21 +04:00
|
|
|
)
|
|
|
|
|
2014-08-22 23:44:18 +04:00
|
|
|
const (
|
2014-10-03 04:06:35 +04:00
|
|
|
metadataType int64 = iota + 1
|
2014-08-22 23:44:18 +04:00
|
|
|
entryType
|
|
|
|
stateType
|
2014-09-04 02:10:15 +04:00
|
|
|
crcType
|
2015-01-06 10:27:03 +03:00
|
|
|
snapshotType
|
2014-09-09 02:45:58 +04:00
|
|
|
|
2016-06-08 20:57:14 +03:00
|
|
|
// warnSyncDuration is the amount of time allotted to an fsync before
|
|
|
|
// logging a warning
|
|
|
|
warnSyncDuration = time.Second
|
2014-08-22 23:44:18 +04:00
|
|
|
)
|
|
|
|
|
2014-07-26 01:20:21 +04:00
|
|
|
var (
|
2016-08-10 01:29:40 +03:00
|
|
|
// SegmentSizeBytes is the preallocated size of each wal segment file.
|
|
|
|
// The actual size might be larger than this. In general, the default
|
|
|
|
// value should be used, but this is defined as an exported variable
|
|
|
|
// so that tests can set a different segment size.
|
|
|
|
SegmentSizeBytes int64 = 64 * 1000 * 1000 // 64MB
|
|
|
|
|
2018-08-29 03:12:23 +03:00
|
|
|
plog = capnslog.NewPackageLogger("go.etcd.io/etcd", "wal")
|
2015-06-01 23:38:50 +03:00
|
|
|
|
2014-10-03 04:06:35 +04:00
|
|
|
ErrMetadataConflict = errors.New("wal: conflicting metadata found")
|
|
|
|
ErrFileNotFound = errors.New("wal: file not found")
|
|
|
|
ErrCRCMismatch = errors.New("wal: crc mismatch")
|
2015-01-06 10:27:03 +03:00
|
|
|
ErrSnapshotMismatch = errors.New("wal: snapshot mismatch")
|
|
|
|
ErrSnapshotNotFound = errors.New("wal: snapshot not found")
|
2014-10-03 04:06:35 +04:00
|
|
|
crcTable = crc32.MakeTable(crc32.Castagnoli)
|
2014-07-26 01:20:21 +04:00
|
|
|
)
|
|
|
|
|
2016-01-31 05:15:56 +03:00
|
|
|
// WAL is a logical representation of the stable storage.
|
2014-09-04 02:10:15 +04:00
|
|
|
// WAL is either in read mode or append mode but not both.
|
|
|
|
// A newly created WAL is in append mode, and ready for appending records.
|
|
|
|
// A just opened WAL is in read mode, and ready for reading records.
|
|
|
|
// The WAL will be ready for appending after reading out all the previous records.
|
2014-07-26 01:20:21 +04:00
|
|
|
type WAL struct {
|
2018-04-16 13:58:25 +03:00
|
|
|
lg *zap.Logger
|
|
|
|
|
2016-09-08 02:40:33 +03:00
|
|
|
dir string // the living directory of the underlay files
|
|
|
|
|
|
|
|
// dirFile is a fd for the wal directory for syncing on Rename
|
|
|
|
dirFile *os.File
|
|
|
|
|
2014-12-03 10:35:44 +03:00
|
|
|
metadata []byte // metadata recorded at the head of each WAL
|
|
|
|
state raftpb.HardState // hardstate recorded at the head of WAL
|
2014-07-26 01:20:21 +04:00
|
|
|
|
2016-03-15 03:21:14 +03:00
|
|
|
start walpb.Snapshot // snapshot to start reading
|
|
|
|
decoder *decoder // decoder to decode records
|
2016-03-16 03:19:09 +03:00
|
|
|
readClose func() error // closer for decode reader
|
2014-07-28 07:32:23 +04:00
|
|
|
|
2015-04-08 10:25:18 +03:00
|
|
|
mu sync.Mutex
|
2014-10-08 15:58:53 +04:00
|
|
|
enti uint64 // index of the last entry saved to the wal
|
2014-09-04 02:10:15 +04:00
|
|
|
encoder *encoder // encoder to encode records
|
2014-12-12 23:48:44 +03:00
|
|
|
|
2016-03-15 03:21:14 +03:00
|
|
|
locks []*fileutil.LockedFile // the locked files the WAL holds (the name is increasing)
|
2016-03-17 19:51:33 +03:00
|
|
|
fp *filePipeline
|
2014-08-22 23:44:18 +04:00
|
|
|
}
|
|
|
|
|
2014-10-03 04:06:35 +04:00
|
|
|
// Create creates a WAL ready for appending records. The given metadata is
|
|
|
|
// recorded at the head of each WAL file, and can be retrieved with ReadAll.
|
2018-04-16 13:58:25 +03:00
|
|
|
func Create(lg *zap.Logger, dirpath string, metadata []byte) (*WAL, error) {
|
2014-08-22 23:44:18 +04:00
|
|
|
if Exist(dirpath) {
|
2014-07-26 01:54:08 +04:00
|
|
|
return nil, os.ErrExist
|
|
|
|
}
|
2014-09-05 08:15:39 +04:00
|
|
|
|
2016-05-11 23:43:47 +03:00
|
|
|
// keep temporary wal directory so WAL initialization appears atomic
|
2017-03-16 03:18:55 +03:00
|
|
|
tmpdirpath := filepath.Clean(dirpath) + ".tmp"
|
2016-05-11 23:43:47 +03:00
|
|
|
if fileutil.Exist(tmpdirpath) {
|
|
|
|
if err := os.RemoveAll(tmpdirpath); err != nil {
|
|
|
|
return nil, err
|
|
|
|
}
|
|
|
|
}
|
2016-06-23 01:07:04 +03:00
|
|
|
if err := fileutil.CreateDirAll(tmpdirpath); err != nil {
|
2018-05-04 00:01:06 +03:00
|
|
|
if lg != nil {
|
|
|
|
lg.Warn(
|
|
|
|
"failed to create a temporary WAL directory",
|
|
|
|
zap.String("tmp-dir-path", tmpdirpath),
|
|
|
|
zap.String("dir-path", dirpath),
|
|
|
|
zap.Error(err),
|
|
|
|
)
|
|
|
|
}
|
2014-09-05 08:15:39 +04:00
|
|
|
return nil, err
|
|
|
|
}
|
|
|
|
|
2017-03-16 03:18:55 +03:00
|
|
|
p := filepath.Join(tmpdirpath, walName(0, 0))
|
2016-06-11 01:10:02 +03:00
|
|
|
f, err := fileutil.LockFile(p, os.O_WRONLY|os.O_CREATE, fileutil.PrivateFileMode)
|
2014-07-26 01:20:21 +04:00
|
|
|
if err != nil {
|
2018-05-04 00:01:06 +03:00
|
|
|
if lg != nil {
|
|
|
|
lg.Warn(
|
|
|
|
"failed to flock an initial WAL file",
|
|
|
|
zap.String("path", p),
|
|
|
|
zap.Error(err),
|
|
|
|
)
|
|
|
|
}
|
2014-07-26 01:20:21 +04:00
|
|
|
return nil, err
|
|
|
|
}
|
2017-03-21 01:01:01 +03:00
|
|
|
if _, err = f.Seek(0, io.SeekEnd); err != nil {
|
2018-05-04 00:01:06 +03:00
|
|
|
if lg != nil {
|
|
|
|
lg.Warn(
|
|
|
|
"failed to seek an initial WAL file",
|
|
|
|
zap.String("path", p),
|
|
|
|
zap.Error(err),
|
|
|
|
)
|
|
|
|
}
|
2016-03-16 03:19:09 +03:00
|
|
|
return nil, err
|
|
|
|
}
|
2016-09-08 02:40:33 +03:00
|
|
|
if err = fileutil.Preallocate(f.File, SegmentSizeBytes, true); err != nil {
|
2018-05-04 00:01:06 +03:00
|
|
|
if lg != nil {
|
|
|
|
lg.Warn(
|
|
|
|
"failed to preallocate an initial WAL file",
|
|
|
|
zap.String("path", p),
|
|
|
|
zap.Int64("segment-bytes", SegmentSizeBytes),
|
|
|
|
zap.Error(err),
|
|
|
|
)
|
|
|
|
}
|
2016-03-16 03:19:09 +03:00
|
|
|
return nil, err
|
|
|
|
}
|
2014-12-12 23:48:44 +03:00
|
|
|
|
2014-09-04 02:10:15 +04:00
|
|
|
w := &WAL{
|
2018-04-16 13:58:25 +03:00
|
|
|
lg: lg,
|
2014-10-11 16:41:00 +04:00
|
|
|
dir: dirpath,
|
|
|
|
metadata: metadata,
|
2016-09-28 21:03:24 +03:00
|
|
|
}
|
|
|
|
w.encoder, err = newFileEncoder(f.File, 0)
|
|
|
|
if err != nil {
|
|
|
|
return nil, err
|
2014-09-04 02:10:15 +04:00
|
|
|
}
|
2016-03-15 03:21:14 +03:00
|
|
|
w.locks = append(w.locks, f)
|
2016-09-08 02:40:33 +03:00
|
|
|
if err = w.saveCrc(0); err != nil {
|
|
|
|
return nil, err
|
|
|
|
}
|
|
|
|
if err = w.encoder.encode(&walpb.Record{Type: metadataType, Data: metadata}); err != nil {
|
2014-09-04 02:10:15 +04:00
|
|
|
return nil, err
|
|
|
|
}
|
2016-09-08 02:40:33 +03:00
|
|
|
if err = w.SaveSnapshot(walpb.Snapshot{}); err != nil {
|
2014-10-03 04:06:35 +04:00
|
|
|
return nil, err
|
|
|
|
}
|
2016-09-08 02:40:33 +03:00
|
|
|
|
2018-05-17 20:48:57 +03:00
|
|
|
if w, err = w.renameWAL(tmpdirpath); err != nil {
|
2018-05-04 00:01:06 +03:00
|
|
|
if lg != nil {
|
|
|
|
lg.Warn(
|
|
|
|
"failed to rename the temporary WAL directory",
|
|
|
|
zap.String("tmp-dir-path", tmpdirpath),
|
|
|
|
zap.String("dir-path", w.dir),
|
|
|
|
zap.Error(err),
|
|
|
|
)
|
|
|
|
}
|
2014-11-08 01:41:34 +03:00
|
|
|
return nil, err
|
|
|
|
}
|
2016-05-11 23:43:47 +03:00
|
|
|
|
2019-05-02 12:35:07 +03:00
|
|
|
var perr error
|
|
|
|
defer func() {
|
|
|
|
if perr != nil {
|
|
|
|
w.cleanupWAL(lg)
|
|
|
|
}
|
|
|
|
}()
|
|
|
|
|
2016-09-08 02:40:33 +03:00
|
|
|
// directory was renamed; sync parent dir to persist rename
|
2017-03-16 03:18:55 +03:00
|
|
|
pdir, perr := fileutil.OpenDir(filepath.Dir(w.dir))
|
2016-09-08 02:40:33 +03:00
|
|
|
if perr != nil {
|
2018-05-04 00:01:06 +03:00
|
|
|
if lg != nil {
|
|
|
|
lg.Warn(
|
|
|
|
"failed to open the parent data directory",
|
|
|
|
zap.String("parent-dir-path", filepath.Dir(w.dir)),
|
|
|
|
zap.String("dir-path", w.dir),
|
|
|
|
zap.Error(perr),
|
|
|
|
)
|
|
|
|
}
|
2016-09-08 02:40:33 +03:00
|
|
|
return nil, perr
|
|
|
|
}
|
|
|
|
if perr = fileutil.Fsync(pdir); perr != nil {
|
2018-05-04 00:01:06 +03:00
|
|
|
if lg != nil {
|
|
|
|
lg.Warn(
|
|
|
|
"failed to fsync the parent data directory file",
|
|
|
|
zap.String("parent-dir-path", filepath.Dir(w.dir)),
|
|
|
|
zap.String("dir-path", w.dir),
|
|
|
|
zap.Error(perr),
|
|
|
|
)
|
|
|
|
}
|
2016-09-08 02:40:33 +03:00
|
|
|
return nil, perr
|
|
|
|
}
|
2019-04-28 19:56:44 +03:00
|
|
|
if perr = pdir.Close(); perr != nil {
|
2018-05-04 00:01:06 +03:00
|
|
|
if lg != nil {
|
|
|
|
lg.Warn(
|
|
|
|
"failed to close the parent data directory file",
|
|
|
|
zap.String("parent-dir-path", filepath.Dir(w.dir)),
|
|
|
|
zap.String("dir-path", w.dir),
|
|
|
|
zap.Error(perr),
|
|
|
|
)
|
|
|
|
}
|
2016-09-08 02:40:33 +03:00
|
|
|
return nil, perr
|
|
|
|
}
|
|
|
|
|
|
|
|
return w, nil
|
2014-07-26 01:20:21 +04:00
|
|
|
}
|
|
|
|
|
2019-04-28 19:56:44 +03:00
|
|
|
func (w *WAL) cleanupWAL(lg *zap.Logger) {
|
|
|
|
var err error
|
|
|
|
if err = w.Close(); err != nil {
|
|
|
|
if lg != nil {
|
2019-05-10 22:37:08 +03:00
|
|
|
lg.Panic("failed to close WAL during cleanup", zap.Error(err))
|
2019-04-28 19:56:44 +03:00
|
|
|
} else {
|
2019-05-10 22:37:08 +03:00
|
|
|
plog.Panicf("failed to close WAL during cleanup: %v", err)
|
2019-04-28 19:56:44 +03:00
|
|
|
}
|
|
|
|
}
|
2019-05-02 12:35:07 +03:00
|
|
|
brokenDirName := fmt.Sprintf("%s.broken.%v", w.dir, time.Now().Format("20060102.150405.999999"))
|
|
|
|
if err = os.Rename(w.dir, brokenDirName); err != nil {
|
2019-04-28 19:56:44 +03:00
|
|
|
if lg != nil {
|
2019-05-02 12:35:07 +03:00
|
|
|
lg.Panic(
|
|
|
|
"failed to rename WAL during cleanup",
|
|
|
|
zap.Error(err),
|
|
|
|
zap.String("source-path", w.dir),
|
|
|
|
zap.String("rename-path", brokenDirName),
|
|
|
|
)
|
2019-04-28 19:56:44 +03:00
|
|
|
} else {
|
2019-05-02 12:35:07 +03:00
|
|
|
plog.Panicf("failed to rename WAL during cleanup: %v", err)
|
2019-04-28 19:56:44 +03:00
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2018-05-17 20:48:57 +03:00
|
|
|
func (w *WAL) renameWAL(tmpdirpath string) (*WAL, error) {
|
2017-07-20 01:57:42 +03:00
|
|
|
if err := os.RemoveAll(w.dir); err != nil {
|
|
|
|
return nil, err
|
|
|
|
}
|
|
|
|
// On non-Windows platforms, hold the lock while renaming. Releasing
|
|
|
|
// the lock and trying to reacquire it quickly can be flaky because
|
|
|
|
// it's possible the process will fork to spawn a process while this is
|
|
|
|
// happening. The fds are set up as close-on-exec by the Go runtime,
|
|
|
|
// but there is a window between the fork and the exec where another
|
|
|
|
// process holds the lock.
|
|
|
|
if err := os.Rename(tmpdirpath, w.dir); err != nil {
|
|
|
|
if _, ok := err.(*os.LinkError); ok {
|
2018-05-17 20:48:57 +03:00
|
|
|
return w.renameWALUnlock(tmpdirpath)
|
2017-07-20 01:57:42 +03:00
|
|
|
}
|
|
|
|
return nil, err
|
|
|
|
}
|
2018-04-16 13:58:25 +03:00
|
|
|
w.fp = newFilePipeline(w.lg, w.dir, SegmentSizeBytes)
|
2017-07-20 01:57:42 +03:00
|
|
|
df, err := fileutil.OpenDir(w.dir)
|
|
|
|
w.dirFile = df
|
|
|
|
return w, err
|
|
|
|
}
|
|
|
|
|
2018-05-17 20:48:57 +03:00
|
|
|
func (w *WAL) renameWALUnlock(tmpdirpath string) (*WAL, error) {
|
2017-07-20 01:57:42 +03:00
|
|
|
// rename of directory with locked files doesn't work on windows/cifs;
|
|
|
|
// close the WAL to release the locks so the directory can be renamed.
|
2018-04-16 13:58:25 +03:00
|
|
|
if w.lg != nil {
|
|
|
|
w.lg.Info(
|
2018-05-17 20:48:57 +03:00
|
|
|
"closing WAL to release flock and retry directory renaming",
|
2018-04-16 13:58:25 +03:00
|
|
|
zap.String("from", tmpdirpath),
|
|
|
|
zap.String("to", w.dir),
|
|
|
|
)
|
|
|
|
} else {
|
|
|
|
plog.Infof("releasing file lock to rename %q to %q", tmpdirpath, w.dir)
|
|
|
|
}
|
2017-07-20 01:57:42 +03:00
|
|
|
w.Close()
|
2018-05-17 20:48:57 +03:00
|
|
|
|
2017-07-20 01:57:42 +03:00
|
|
|
if err := os.Rename(tmpdirpath, w.dir); err != nil {
|
|
|
|
return nil, err
|
|
|
|
}
|
2018-05-17 20:48:57 +03:00
|
|
|
|
2017-07-20 01:57:42 +03:00
|
|
|
// reopen and relock
|
2018-04-16 13:58:25 +03:00
|
|
|
newWAL, oerr := Open(w.lg, w.dir, walpb.Snapshot{})
|
2017-07-20 01:57:42 +03:00
|
|
|
if oerr != nil {
|
|
|
|
return nil, oerr
|
|
|
|
}
|
|
|
|
if _, _, _, err := newWAL.ReadAll(); err != nil {
|
|
|
|
newWAL.Close()
|
|
|
|
return nil, err
|
|
|
|
}
|
|
|
|
return newWAL, nil
|
|
|
|
}
|
|
|
|
|
2015-01-06 10:27:03 +03:00
|
|
|
// Open opens the WAL at the given snap.
|
|
|
|
// The snap SHOULD have been previously saved to the WAL, or the following
|
2014-09-17 09:11:23 +04:00
|
|
|
// ReadAll will fail.
|
2015-01-06 10:27:03 +03:00
|
|
|
// 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
|
2014-09-11 23:52:06 +04:00
|
|
|
// previous records.
|
2018-04-16 13:58:25 +03:00
|
|
|
func Open(lg *zap.Logger, dirpath string, snap walpb.Snapshot) (*WAL, error) {
|
|
|
|
w, err := openAtIndex(lg, dirpath, snap, true)
|
2016-09-08 02:40:33 +03:00
|
|
|
if err != nil {
|
|
|
|
return nil, err
|
|
|
|
}
|
|
|
|
if w.dirFile, err = fileutil.OpenDir(w.dir); err != nil {
|
|
|
|
return nil, err
|
|
|
|
}
|
|
|
|
return w, nil
|
2014-12-13 01:38:18 +03:00
|
|
|
}
|
|
|
|
|
2015-06-13 02:06:02 +03:00
|
|
|
// OpenForRead only opens the wal files for read.
|
|
|
|
// Write on a read only wal panics.
|
2018-04-16 13:58:25 +03:00
|
|
|
func OpenForRead(lg *zap.Logger, dirpath string, snap walpb.Snapshot) (*WAL, error) {
|
|
|
|
return openAtIndex(lg, dirpath, snap, false)
|
2014-12-13 01:38:18 +03:00
|
|
|
}
|
|
|
|
|
2018-04-16 13:58:25 +03:00
|
|
|
func openAtIndex(lg *zap.Logger, dirpath string, snap walpb.Snapshot, write bool) (*WAL, error) {
|
2019-03-01 12:09:43 +03:00
|
|
|
names, nameIndex, err := selectWALFiles(lg, dirpath, snap)
|
|
|
|
if err != nil {
|
|
|
|
return nil, err
|
|
|
|
}
|
|
|
|
|
|
|
|
rs, ls, closer, err := openWALFiles(lg, dirpath, names, nameIndex, write)
|
2014-08-22 23:44:18 +04:00
|
|
|
if err != nil {
|
|
|
|
return nil, err
|
|
|
|
}
|
|
|
|
|
2019-03-01 12:09:43 +03:00
|
|
|
// create a WAL ready for reading
|
|
|
|
w := &WAL{
|
2019-04-02 10:38:04 +03:00
|
|
|
lg: lg,
|
2019-03-01 12:09:43 +03:00
|
|
|
dir: dirpath,
|
|
|
|
start: snap,
|
|
|
|
decoder: newDecoder(rs...),
|
|
|
|
readClose: closer,
|
|
|
|
locks: ls,
|
|
|
|
}
|
|
|
|
|
|
|
|
if write {
|
|
|
|
// write reuses the file descriptors from read; don't close so
|
|
|
|
// WAL can append without dropping the file lock
|
|
|
|
w.readClose = nil
|
|
|
|
if _, _, err := parseWALName(filepath.Base(w.tail().Name())); err != nil {
|
|
|
|
closer()
|
|
|
|
return nil, err
|
|
|
|
}
|
|
|
|
w.fp = newFilePipeline(lg, w.dir, SegmentSizeBytes)
|
|
|
|
}
|
|
|
|
|
|
|
|
return w, nil
|
|
|
|
}
|
|
|
|
|
|
|
|
func selectWALFiles(lg *zap.Logger, dirpath string, snap walpb.Snapshot) ([]string, int, error) {
|
|
|
|
names, err := readWALNames(lg, dirpath)
|
|
|
|
if err != nil {
|
|
|
|
return nil, -1, err
|
|
|
|
}
|
|
|
|
|
2018-05-17 20:48:57 +03:00
|
|
|
nameIndex, ok := searchIndex(lg, names, snap.Index)
|
|
|
|
if !ok || !isValidSeq(lg, names[nameIndex:]) {
|
2019-03-01 12:09:43 +03:00
|
|
|
err = ErrFileNotFound
|
|
|
|
return nil, -1, err
|
2014-07-26 01:54:08 +04:00
|
|
|
}
|
|
|
|
|
2019-03-01 12:09:43 +03:00
|
|
|
return names, nameIndex, nil
|
|
|
|
}
|
|
|
|
|
|
|
|
func openWALFiles(lg *zap.Logger, dirpath string, names []string, nameIndex int, write bool) ([]io.Reader, []*fileutil.LockedFile, func() error, error) {
|
2014-09-04 02:10:15 +04:00
|
|
|
rcs := make([]io.ReadCloser, 0)
|
2016-03-16 03:19:09 +03:00
|
|
|
rs := make([]io.Reader, 0)
|
2016-03-15 03:21:14 +03:00
|
|
|
ls := make([]*fileutil.LockedFile, 0)
|
2014-09-04 02:10:15 +04:00
|
|
|
for _, name := range names[nameIndex:] {
|
2017-03-16 03:18:55 +03:00
|
|
|
p := filepath.Join(dirpath, name)
|
2016-03-15 03:21:14 +03:00
|
|
|
if write {
|
2016-06-11 01:10:02 +03:00
|
|
|
l, err := fileutil.TryLockFile(p, os.O_RDWR, fileutil.PrivateFileMode)
|
2016-03-15 03:21:14 +03:00
|
|
|
if err != nil {
|
2016-03-16 03:19:09 +03:00
|
|
|
closeAll(rcs...)
|
2019-03-01 12:09:43 +03:00
|
|
|
return nil, nil, nil, err
|
2014-12-13 01:38:18 +03:00
|
|
|
}
|
2016-03-15 03:21:14 +03:00
|
|
|
ls = append(ls, l)
|
|
|
|
rcs = append(rcs, l)
|
|
|
|
} else {
|
2016-06-11 01:10:02 +03:00
|
|
|
rf, err := os.OpenFile(p, os.O_RDONLY, fileutil.PrivateFileMode)
|
2016-03-15 03:21:14 +03:00
|
|
|
if err != nil {
|
2016-03-16 03:19:09 +03:00
|
|
|
closeAll(rcs...)
|
2019-03-01 12:09:43 +03:00
|
|
|
return nil, nil, nil, err
|
2016-03-15 03:21:14 +03:00
|
|
|
}
|
|
|
|
ls = append(ls, nil)
|
|
|
|
rcs = append(rcs, rf)
|
2014-12-12 23:48:44 +03:00
|
|
|
}
|
2016-03-16 03:19:09 +03:00
|
|
|
rs = append(rs, rcs[len(rcs)-1])
|
2014-07-26 01:20:21 +04:00
|
|
|
}
|
2016-03-15 03:21:14 +03:00
|
|
|
|
2016-03-16 03:19:09 +03:00
|
|
|
closer := func() error { return closeAll(rcs...) }
|
2014-07-26 01:20:21 +04:00
|
|
|
|
2019-03-01 12:09:43 +03:00
|
|
|
return rs, ls, closer, nil
|
2014-07-26 11:10:59 +04:00
|
|
|
}
|
|
|
|
|
2015-06-13 02:06:02 +03:00
|
|
|
// ReadAll reads out records of the current WAL.
|
|
|
|
// If opened in write mode, it must read out all records until EOF. Or an error
|
|
|
|
// will be returned.
|
|
|
|
// If opened in read mode, it will try to read all records if possible.
|
2015-01-06 10:27:03 +03:00
|
|
|
// If it cannot read out the expected snap, it will return ErrSnapshotNotFound.
|
|
|
|
// If loaded snap doesn't match with the expected one, it will return
|
2015-01-29 22:38:52 +03:00
|
|
|
// all the records and error ErrSnapshotMismatch.
|
2015-01-06 10:27:03 +03:00
|
|
|
// TODO: detect not-last-snap error.
|
|
|
|
// TODO: maybe loose the checking of match.
|
2014-09-04 02:10:15 +04:00
|
|
|
// After ReadAll, the WAL will be ready for appending new records.
|
2014-10-03 04:06:35 +04:00
|
|
|
func (w *WAL) ReadAll() (metadata []byte, state raftpb.HardState, ents []raftpb.Entry, err error) {
|
2015-04-08 10:25:18 +03:00
|
|
|
w.mu.Lock()
|
|
|
|
defer w.mu.Unlock()
|
|
|
|
|
2014-09-04 03:46:42 +04:00
|
|
|
rec := &walpb.Record{}
|
2014-09-04 02:10:15 +04:00
|
|
|
decoder := w.decoder
|
2014-09-09 02:36:25 +04:00
|
|
|
|
2015-01-06 10:27:03 +03:00
|
|
|
var match bool
|
2014-09-04 02:10:15 +04:00
|
|
|
for err = decoder.decode(rec); err == nil; err = decoder.decode(rec) {
|
2014-08-05 02:46:12 +04:00
|
|
|
switch rec.Type {
|
2014-07-26 11:10:59 +04:00
|
|
|
case entryType:
|
2014-09-04 02:10:15 +04:00
|
|
|
e := mustUnmarshalEntry(rec.Data)
|
2015-01-06 10:27:03 +03:00
|
|
|
if e.Index > w.start.Index {
|
|
|
|
ents = append(ents[:e.Index-w.start.Index-1], e)
|
2014-07-26 11:10:59 +04:00
|
|
|
}
|
2014-09-15 02:11:50 +04:00
|
|
|
w.enti = e.Index
|
2018-05-17 20:48:57 +03:00
|
|
|
|
2014-07-26 11:10:59 +04:00
|
|
|
case stateType:
|
2014-09-04 02:10:15 +04:00
|
|
|
state = mustUnmarshalState(rec.Data)
|
2018-05-17 20:48:57 +03:00
|
|
|
|
2014-10-03 04:06:35 +04:00
|
|
|
case metadataType:
|
2016-06-13 11:33:53 +03:00
|
|
|
if metadata != nil && !bytes.Equal(metadata, rec.Data) {
|
2014-09-04 02:10:15 +04:00
|
|
|
state.Reset()
|
2014-10-03 04:06:35 +04:00
|
|
|
return nil, state, nil, ErrMetadataConflict
|
2014-07-26 11:10:59 +04:00
|
|
|
}
|
2014-10-03 04:06:35 +04:00
|
|
|
metadata = rec.Data
|
2018-05-17 20:48:57 +03:00
|
|
|
|
2014-09-04 02:10:15 +04:00
|
|
|
case crcType:
|
|
|
|
crc := decoder.crc.Sum32()
|
|
|
|
// current crc of decoder must match the crc of the record.
|
|
|
|
// do no need to match 0 crc, since the decoder is a new one at this case.
|
2014-09-04 03:46:42 +04:00
|
|
|
if crc != 0 && rec.Validate(crc) != nil {
|
2014-09-04 02:10:15 +04:00
|
|
|
state.Reset()
|
2014-10-03 04:06:35 +04:00
|
|
|
return nil, state, nil, ErrCRCMismatch
|
2014-09-04 02:10:15 +04:00
|
|
|
}
|
|
|
|
decoder.updateCRC(rec.Crc)
|
2018-05-17 20:48:57 +03:00
|
|
|
|
2015-01-06 10:27:03 +03:00
|
|
|
case snapshotType:
|
|
|
|
var snap walpb.Snapshot
|
|
|
|
pbutil.MustUnmarshal(&snap, rec.Data)
|
|
|
|
if snap.Index == w.start.Index {
|
|
|
|
if snap.Term != w.start.Term {
|
|
|
|
state.Reset()
|
|
|
|
return nil, state, nil, ErrSnapshotMismatch
|
|
|
|
}
|
|
|
|
match = true
|
|
|
|
}
|
2018-05-17 20:48:57 +03:00
|
|
|
|
2014-07-26 11:10:59 +04:00
|
|
|
default:
|
2014-09-04 02:10:15 +04:00
|
|
|
state.Reset()
|
2014-10-03 04:06:35 +04:00
|
|
|
return nil, state, nil, fmt.Errorf("unexpected block type %d", rec.Type)
|
2014-07-26 11:10:59 +04:00
|
|
|
}
|
|
|
|
}
|
2015-06-13 02:06:02 +03:00
|
|
|
|
2016-03-15 03:21:14 +03:00
|
|
|
switch w.tail() {
|
2015-06-13 02:06:02 +03:00
|
|
|
case nil:
|
|
|
|
// We do not have to read out all entries in read mode.
|
|
|
|
// The last record maybe a partial written one, so
|
|
|
|
// ErrunexpectedEOF might be returned.
|
|
|
|
if err != io.EOF && err != io.ErrUnexpectedEOF {
|
|
|
|
state.Reset()
|
|
|
|
return nil, state, nil, err
|
|
|
|
}
|
|
|
|
default:
|
|
|
|
// We must read all of the entries if WAL is opened in write mode.
|
|
|
|
if err != io.EOF {
|
|
|
|
state.Reset()
|
|
|
|
return nil, state, nil, err
|
|
|
|
}
|
2016-08-25 23:23:36 +03:00
|
|
|
// decodeRecord() will return io.EOF if it detects a zero record,
|
|
|
|
// but this zero record may be followed by non-zero records from
|
|
|
|
// a torn write. Overwriting some of these non-zero records, but
|
|
|
|
// not all, will cause CRC errors on WAL open. Since the records
|
|
|
|
// were never fully synced to disk in the first place, it's safe
|
|
|
|
// to zero them out to avoid any CRC errors from new writes.
|
2017-03-21 01:01:01 +03:00
|
|
|
if _, err = w.tail().Seek(w.decoder.lastOffset(), io.SeekStart); err != nil {
|
2016-08-25 23:23:36 +03:00
|
|
|
return nil, state, nil, err
|
|
|
|
}
|
|
|
|
if err = fileutil.ZeroToEnd(w.tail().File); err != nil {
|
|
|
|
return nil, state, nil, err
|
|
|
|
}
|
2014-08-22 23:44:18 +04:00
|
|
|
}
|
2015-06-13 02:06:02 +03:00
|
|
|
|
2015-01-29 22:38:52 +03:00
|
|
|
err = nil
|
2015-01-06 10:27:03 +03:00
|
|
|
if !match {
|
2015-01-29 22:38:52 +03:00
|
|
|
err = ErrSnapshotNotFound
|
2015-01-06 10:27:03 +03:00
|
|
|
}
|
2014-08-22 23:44:18 +04:00
|
|
|
|
2014-09-04 02:10:15 +04:00
|
|
|
// close decoder, disable reading
|
2016-03-15 03:21:14 +03:00
|
|
|
if w.readClose != nil {
|
2016-03-16 03:19:09 +03:00
|
|
|
w.readClose()
|
2016-03-15 03:21:14 +03:00
|
|
|
w.readClose = nil
|
|
|
|
}
|
2015-01-06 10:27:03 +03:00
|
|
|
w.start = walpb.Snapshot{}
|
2014-09-04 02:10:15 +04:00
|
|
|
|
2014-10-11 16:41:00 +04:00
|
|
|
w.metadata = metadata
|
2015-06-13 02:06:02 +03:00
|
|
|
|
2016-03-15 03:21:14 +03:00
|
|
|
if w.tail() != nil {
|
2015-06-13 02:06:02 +03:00
|
|
|
// create encoder (chain crc with the decoder), enable appending
|
2016-09-28 21:03:24 +03:00
|
|
|
w.encoder, err = newFileEncoder(w.tail().File, w.decoder.lastCRC())
|
|
|
|
if err != nil {
|
|
|
|
return
|
|
|
|
}
|
2015-06-13 02:06:02 +03:00
|
|
|
}
|
2016-03-15 03:21:14 +03:00
|
|
|
w.decoder = nil
|
2015-06-13 02:06:02 +03:00
|
|
|
|
2015-01-29 22:38:52 +03:00
|
|
|
return metadata, state, ents, err
|
2014-08-22 23:44:18 +04:00
|
|
|
}
|
|
|
|
|
2019-03-01 12:09:43 +03:00
|
|
|
// Verify reads through the given WAL and verifies that it is not corrupted.
|
|
|
|
// It creates a new decoder to read through the records of the given WAL.
|
|
|
|
// It does not conflict with any open WAL, but it is recommended not to
|
|
|
|
// call this function after opening the WAL for writing.
|
|
|
|
// If it cannot read out the expected snap, it will return ErrSnapshotNotFound.
|
|
|
|
// If the loaded snap doesn't match with the expected one, it will
|
|
|
|
// return error ErrSnapshotMismatch.
|
|
|
|
func Verify(lg *zap.Logger, walDir string, snap walpb.Snapshot) error {
|
|
|
|
var metadata []byte
|
|
|
|
var err error
|
|
|
|
var match bool
|
|
|
|
|
|
|
|
rec := &walpb.Record{}
|
|
|
|
|
|
|
|
names, nameIndex, err := selectWALFiles(lg, walDir, snap)
|
|
|
|
if err != nil {
|
|
|
|
return err
|
|
|
|
}
|
|
|
|
|
|
|
|
// open wal files in read mode, so that there is no conflict
|
|
|
|
// when the same WAL is opened elsewhere in write mode
|
|
|
|
rs, _, closer, err := openWALFiles(lg, walDir, names, nameIndex, false)
|
|
|
|
if err != nil {
|
|
|
|
return err
|
|
|
|
}
|
2019-09-05 01:38:14 +03:00
|
|
|
defer func() {
|
|
|
|
if closer != nil {
|
|
|
|
closer()
|
|
|
|
}
|
|
|
|
}()
|
2019-03-01 12:09:43 +03:00
|
|
|
|
|
|
|
// create a new decoder from the readers on the WAL files
|
|
|
|
decoder := newDecoder(rs...)
|
|
|
|
|
|
|
|
for err = decoder.decode(rec); err == nil; err = decoder.decode(rec) {
|
|
|
|
switch rec.Type {
|
|
|
|
case metadataType:
|
|
|
|
if metadata != nil && !bytes.Equal(metadata, rec.Data) {
|
|
|
|
return ErrMetadataConflict
|
|
|
|
}
|
|
|
|
metadata = rec.Data
|
|
|
|
case crcType:
|
|
|
|
crc := decoder.crc.Sum32()
|
|
|
|
// Current crc of decoder must match the crc of the record.
|
|
|
|
// We need not match 0 crc, since the decoder is a new one at this point.
|
|
|
|
if crc != 0 && rec.Validate(crc) != nil {
|
|
|
|
return ErrCRCMismatch
|
|
|
|
}
|
|
|
|
decoder.updateCRC(rec.Crc)
|
|
|
|
case snapshotType:
|
|
|
|
var loadedSnap walpb.Snapshot
|
|
|
|
pbutil.MustUnmarshal(&loadedSnap, rec.Data)
|
|
|
|
if loadedSnap.Index == snap.Index {
|
|
|
|
if loadedSnap.Term != snap.Term {
|
|
|
|
return ErrSnapshotMismatch
|
|
|
|
}
|
|
|
|
match = true
|
|
|
|
}
|
|
|
|
// We ignore all entry and state type records as these
|
|
|
|
// are not necessary for validating the WAL contents
|
|
|
|
case entryType:
|
|
|
|
case stateType:
|
|
|
|
default:
|
|
|
|
return fmt.Errorf("unexpected block type %d", rec.Type)
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
// We do not have to read out all the WAL entries
|
|
|
|
// as the decoder is opened in read mode.
|
|
|
|
if err != io.EOF && err != io.ErrUnexpectedEOF {
|
|
|
|
return err
|
|
|
|
}
|
|
|
|
|
|
|
|
if !match {
|
|
|
|
return ErrSnapshotNotFound
|
|
|
|
}
|
|
|
|
|
|
|
|
return nil
|
|
|
|
}
|
|
|
|
|
2015-02-20 02:41:37 +03:00
|
|
|
// cut closes current file written and creates a new one ready to append.
|
2015-04-09 01:50:24 +03:00
|
|
|
// cut first creates a temp wal file and writes necessary headers into it.
|
2016-01-31 05:15:56 +03:00
|
|
|
// Then cut atomically rename temp wal file to a wal file.
|
2015-02-20 02:41:37 +03:00
|
|
|
func (w *WAL) cut() error {
|
2016-03-16 03:19:09 +03:00
|
|
|
// close old wal file; truncate to avoid wasting space if an early cut
|
2017-03-21 01:01:01 +03:00
|
|
|
off, serr := w.tail().Seek(0, io.SeekCurrent)
|
2016-03-16 03:19:09 +03:00
|
|
|
if serr != nil {
|
|
|
|
return serr
|
|
|
|
}
|
2018-05-17 20:48:57 +03:00
|
|
|
|
2016-03-16 03:19:09 +03:00
|
|
|
if err := w.tail().Truncate(off); err != nil {
|
|
|
|
return err
|
|
|
|
}
|
2018-05-17 20:48:57 +03:00
|
|
|
|
2015-04-09 01:50:24 +03:00
|
|
|
if err := w.sync(); err != nil {
|
2014-09-04 02:10:15 +04:00
|
|
|
return err
|
2014-08-22 23:44:18 +04:00
|
|
|
}
|
2015-04-09 01:50:24 +03:00
|
|
|
|
2017-03-16 03:18:55 +03:00
|
|
|
fpath := filepath.Join(w.dir, walName(w.seq()+1, w.enti+1))
|
2015-04-09 01:50:24 +03:00
|
|
|
|
2015-11-13 02:23:12 +03:00
|
|
|
// create a temp wal file with name sequence + 1, or truncate the existing one
|
2016-03-17 19:51:33 +03:00
|
|
|
newTail, err := w.fp.Open()
|
2014-12-12 23:48:44 +03:00
|
|
|
if err != nil {
|
|
|
|
return err
|
|
|
|
}
|
2014-08-22 23:44:18 +04:00
|
|
|
|
2014-09-04 02:10:15 +04:00
|
|
|
// update writer and save the previous crc
|
2016-03-15 03:21:14 +03:00
|
|
|
w.locks = append(w.locks, newTail)
|
2014-09-04 02:10:15 +04:00
|
|
|
prevCrc := w.encoder.crc.Sum32()
|
2016-09-28 21:03:24 +03:00
|
|
|
w.encoder, err = newFileEncoder(w.tail().File, prevCrc)
|
|
|
|
if err != nil {
|
|
|
|
return err
|
|
|
|
}
|
2018-05-17 20:48:57 +03:00
|
|
|
|
2015-12-12 15:50:12 +03:00
|
|
|
if err = w.saveCrc(prevCrc); err != nil {
|
2014-10-03 04:06:35 +04:00
|
|
|
return err
|
|
|
|
}
|
2018-05-17 20:48:57 +03:00
|
|
|
|
2015-12-12 15:50:12 +03:00
|
|
|
if err = w.encoder.encode(&walpb.Record{Type: metadataType, Data: w.metadata}); err != nil {
|
2014-12-03 10:35:44 +03:00
|
|
|
return err
|
|
|
|
}
|
2018-05-17 20:48:57 +03:00
|
|
|
|
2015-12-12 15:50:12 +03:00
|
|
|
if err = w.saveState(&w.state); err != nil {
|
2014-12-05 02:42:47 +03:00
|
|
|
return err
|
|
|
|
}
|
2018-05-17 20:48:57 +03:00
|
|
|
|
2016-03-15 03:21:14 +03:00
|
|
|
// atomically move temp wal file to wal file
|
2015-12-12 15:50:12 +03:00
|
|
|
if err = w.sync(); err != nil {
|
2015-04-09 01:50:24 +03:00
|
|
|
return err
|
|
|
|
}
|
2016-03-17 19:51:33 +03:00
|
|
|
|
2017-03-21 01:01:01 +03:00
|
|
|
off, err = w.tail().Seek(0, io.SeekCurrent)
|
2016-03-17 19:51:33 +03:00
|
|
|
if err != nil {
|
|
|
|
return err
|
|
|
|
}
|
|
|
|
|
|
|
|
if err = os.Rename(newTail.Name(), fpath); err != nil {
|
2015-04-09 01:50:24 +03:00
|
|
|
return err
|
|
|
|
}
|
2016-09-08 02:40:33 +03:00
|
|
|
if err = fileutil.Fsync(w.dirFile); err != nil {
|
|
|
|
return err
|
|
|
|
}
|
|
|
|
|
2017-09-01 14:13:32 +03:00
|
|
|
// reopen newTail with its new path so calls to Name() match the wal filename format
|
2016-03-15 03:21:14 +03:00
|
|
|
newTail.Close()
|
2015-04-09 01:50:24 +03:00
|
|
|
|
2016-06-11 01:10:02 +03:00
|
|
|
if newTail, err = fileutil.LockFile(fpath, os.O_WRONLY, fileutil.PrivateFileMode); err != nil {
|
2015-08-06 09:46:15 +03:00
|
|
|
return err
|
|
|
|
}
|
2017-03-21 01:01:01 +03:00
|
|
|
if _, err = newTail.Seek(off, io.SeekStart); err != nil {
|
2016-03-16 03:19:09 +03:00
|
|
|
return err
|
|
|
|
}
|
|
|
|
|
2016-03-15 03:21:14 +03:00
|
|
|
w.locks[len(w.locks)-1] = newTail
|
2015-08-06 09:46:15 +03:00
|
|
|
|
2015-04-09 01:50:24 +03:00
|
|
|
prevCrc = w.encoder.crc.Sum32()
|
2016-09-28 21:03:24 +03:00
|
|
|
w.encoder, err = newFileEncoder(w.tail().File, prevCrc)
|
|
|
|
if err != nil {
|
|
|
|
return err
|
|
|
|
}
|
2015-04-09 01:50:24 +03:00
|
|
|
|
2018-04-16 13:58:25 +03:00
|
|
|
if w.lg != nil {
|
2018-05-17 20:48:57 +03:00
|
|
|
w.lg.Info("created a new WAL segment", zap.String("path", fpath))
|
2018-04-16 13:58:25 +03:00
|
|
|
} else {
|
|
|
|
plog.Infof("segmented wal file %v is created", fpath)
|
|
|
|
}
|
2015-04-09 01:50:24 +03:00
|
|
|
return nil
|
2014-08-22 23:44:18 +04:00
|
|
|
}
|
|
|
|
|
2014-11-08 01:41:34 +03:00
|
|
|
func (w *WAL) sync() error {
|
2014-09-04 02:10:15 +04:00
|
|
|
if w.encoder != nil {
|
|
|
|
if err := w.encoder.flush(); err != nil {
|
|
|
|
return err
|
2014-08-22 23:44:18 +04:00
|
|
|
}
|
|
|
|
}
|
2015-01-30 00:07:23 +03:00
|
|
|
start := time.Now()
|
2016-03-15 03:21:14 +03:00
|
|
|
err := fileutil.Fdatasync(w.tail().File)
|
2016-06-08 20:57:14 +03:00
|
|
|
|
2018-04-16 13:58:25 +03:00
|
|
|
took := time.Since(start)
|
|
|
|
if took > warnSyncDuration {
|
|
|
|
if w.lg != nil {
|
|
|
|
w.lg.Warn(
|
|
|
|
"slow fdatasync",
|
|
|
|
zap.Duration("took", took),
|
|
|
|
zap.Duration("expected-duration", warnSyncDuration),
|
|
|
|
)
|
|
|
|
} else {
|
|
|
|
plog.Warningf("sync duration of %v, expected less than %v", took, warnSyncDuration)
|
|
|
|
}
|
2016-06-08 20:57:14 +03:00
|
|
|
}
|
2018-05-23 23:48:29 +03:00
|
|
|
walFsyncSec.Observe(took.Seconds())
|
2018-05-17 20:48:57 +03:00
|
|
|
|
2015-01-30 00:07:23 +03:00
|
|
|
return err
|
2014-08-22 23:44:18 +04:00
|
|
|
}
|
|
|
|
|
2015-03-10 05:06:26 +03:00
|
|
|
// ReleaseLockTo releases the locks, which has smaller index than the given index
|
|
|
|
// except the largest one among them.
|
|
|
|
// For example, if WAL is holding lock 1,2,3,4,5,6, ReleaseLockTo(4) will release
|
|
|
|
// lock 1,2 but keep 3. ReleaseLockTo(5) will release 1,2,3 but keep 4.
|
2014-12-12 23:48:44 +03:00
|
|
|
func (w *WAL) ReleaseLockTo(index uint64) error {
|
2015-04-08 10:25:18 +03:00
|
|
|
w.mu.Lock()
|
|
|
|
defer w.mu.Unlock()
|
|
|
|
|
2017-09-01 14:13:32 +03:00
|
|
|
if len(w.locks) == 0 {
|
|
|
|
return nil
|
|
|
|
}
|
|
|
|
|
2015-03-10 05:06:26 +03:00
|
|
|
var smaller int
|
2015-03-25 08:02:44 +03:00
|
|
|
found := false
|
2015-03-10 05:06:26 +03:00
|
|
|
for i, l := range w.locks {
|
2018-05-17 20:48:57 +03:00
|
|
|
_, lockIndex, err := parseWALName(filepath.Base(l.Name()))
|
2014-12-12 23:48:44 +03:00
|
|
|
if err != nil {
|
|
|
|
return err
|
|
|
|
}
|
2015-03-10 05:06:26 +03:00
|
|
|
if lockIndex >= index {
|
|
|
|
smaller = i - 1
|
2015-03-25 08:02:44 +03:00
|
|
|
found = true
|
2015-03-10 05:06:26 +03:00
|
|
|
break
|
2014-12-12 23:48:44 +03:00
|
|
|
}
|
|
|
|
}
|
2015-03-10 05:06:26 +03:00
|
|
|
|
2015-03-25 08:02:44 +03:00
|
|
|
// if no lock index is greater than the release index, we can
|
2016-01-31 05:15:56 +03:00
|
|
|
// release lock up to the last one(excluding).
|
2017-09-01 14:13:32 +03:00
|
|
|
if !found {
|
2015-03-25 08:02:44 +03:00
|
|
|
smaller = len(w.locks) - 1
|
|
|
|
}
|
|
|
|
|
2015-03-10 05:06:26 +03:00
|
|
|
if smaller <= 0 {
|
|
|
|
return nil
|
|
|
|
}
|
|
|
|
|
|
|
|
for i := 0; i < smaller; i++ {
|
2016-03-15 03:21:14 +03:00
|
|
|
if w.locks[i] == nil {
|
|
|
|
continue
|
|
|
|
}
|
|
|
|
w.locks[i].Close()
|
2015-03-10 05:06:26 +03:00
|
|
|
}
|
|
|
|
w.locks = w.locks[smaller:]
|
|
|
|
|
2014-12-12 23:48:44 +03:00
|
|
|
return nil
|
|
|
|
}
|
|
|
|
|
2018-05-17 20:48:57 +03:00
|
|
|
// Close closes the current WAL file and directory.
|
2014-11-08 01:41:34 +03:00
|
|
|
func (w *WAL) Close() error {
|
2015-04-08 10:25:18 +03:00
|
|
|
w.mu.Lock()
|
|
|
|
defer w.mu.Unlock()
|
|
|
|
|
2016-03-17 19:51:33 +03:00
|
|
|
if w.fp != nil {
|
|
|
|
w.fp.Close()
|
|
|
|
w.fp = nil
|
|
|
|
}
|
|
|
|
|
2016-03-15 03:21:14 +03:00
|
|
|
if w.tail() != nil {
|
2014-11-08 01:41:34 +03:00
|
|
|
if err := w.sync(); err != nil {
|
|
|
|
return err
|
|
|
|
}
|
2014-08-22 23:44:18 +04:00
|
|
|
}
|
2014-12-12 23:48:44 +03:00
|
|
|
for _, l := range w.locks {
|
2016-03-15 03:21:14 +03:00
|
|
|
if l == nil {
|
|
|
|
continue
|
2015-08-05 05:09:08 +03:00
|
|
|
}
|
2016-03-15 03:21:14 +03:00
|
|
|
if err := l.Close(); err != nil {
|
2018-04-16 13:58:25 +03:00
|
|
|
if w.lg != nil {
|
2018-05-17 20:48:57 +03:00
|
|
|
w.lg.Warn("failed to close WAL", zap.Error(err))
|
2018-04-16 13:58:25 +03:00
|
|
|
} else {
|
|
|
|
plog.Errorf("failed to unlock during closing wal: %s", err)
|
|
|
|
}
|
2015-08-05 05:09:08 +03:00
|
|
|
}
|
2014-12-12 23:48:44 +03:00
|
|
|
}
|
2016-09-08 02:40:33 +03:00
|
|
|
|
2016-09-09 03:23:55 +03:00
|
|
|
return w.dirFile.Close()
|
2014-07-26 11:10:59 +04:00
|
|
|
}
|
|
|
|
|
2015-01-09 23:01:22 +03:00
|
|
|
func (w *WAL) saveEntry(e *raftpb.Entry) error {
|
2015-03-31 05:20:46 +03:00
|
|
|
// TODO: add MustMarshalTo to reduce one allocation.
|
2014-10-11 16:41:00 +04:00
|
|
|
b := pbutil.MustMarshal(e)
|
2014-09-04 03:46:42 +04:00
|
|
|
rec := &walpb.Record{Type: entryType, Data: b}
|
2014-09-15 02:11:50 +04:00
|
|
|
if err := w.encoder.encode(rec); err != nil {
|
|
|
|
return err
|
|
|
|
}
|
|
|
|
w.enti = e.Index
|
|
|
|
return nil
|
2014-07-26 11:10:59 +04:00
|
|
|
}
|
|
|
|
|
2015-01-09 23:01:22 +03:00
|
|
|
func (w *WAL) saveState(s *raftpb.HardState) error {
|
2014-09-17 00:55:00 +04:00
|
|
|
if raft.IsEmptyHardState(*s) {
|
2014-09-09 08:45:10 +04:00
|
|
|
return nil
|
|
|
|
}
|
2014-12-03 10:35:44 +03:00
|
|
|
w.state = *s
|
2014-10-11 16:41:00 +04:00
|
|
|
b := pbutil.MustMarshal(s)
|
2014-09-04 03:46:42 +04:00
|
|
|
rec := &walpb.Record{Type: stateType, Data: b}
|
2014-09-04 02:10:15 +04:00
|
|
|
return w.encoder.encode(rec)
|
2014-08-22 23:44:18 +04:00
|
|
|
}
|
|
|
|
|
2014-10-29 20:42:21 +03:00
|
|
|
func (w *WAL) Save(st raftpb.HardState, ents []raftpb.Entry) error {
|
2015-04-08 10:25:18 +03:00
|
|
|
w.mu.Lock()
|
|
|
|
defer w.mu.Unlock()
|
|
|
|
|
2015-01-30 00:07:23 +03:00
|
|
|
// short cut, do not call sync
|
|
|
|
if raft.IsEmptyHardState(st) && len(ents) == 0 {
|
|
|
|
return nil
|
|
|
|
}
|
|
|
|
|
2017-02-13 23:09:54 +03:00
|
|
|
mustSync := raft.MustSync(st, w.state, len(ents))
|
2016-03-14 06:34:37 +03:00
|
|
|
|
2014-09-09 00:57:35 +04:00
|
|
|
// TODO(xiangli): no more reference operator
|
2014-09-05 08:15:39 +04:00
|
|
|
for i := range ents {
|
2015-01-09 23:01:22 +03:00
|
|
|
if err := w.saveEntry(&ents[i]); err != nil {
|
2014-10-29 20:42:21 +03:00
|
|
|
return err
|
|
|
|
}
|
2014-09-05 08:15:39 +04:00
|
|
|
}
|
2015-03-28 20:33:52 +03:00
|
|
|
if err := w.saveState(&st); err != nil {
|
|
|
|
return err
|
|
|
|
}
|
2015-02-20 02:41:37 +03:00
|
|
|
|
2017-03-21 01:01:01 +03:00
|
|
|
curOff, err := w.tail().Seek(0, io.SeekCurrent)
|
2015-02-20 02:41:37 +03:00
|
|
|
if err != nil {
|
|
|
|
return err
|
|
|
|
}
|
2016-08-10 01:29:40 +03:00
|
|
|
if curOff < SegmentSizeBytes {
|
2016-03-14 06:34:37 +03:00
|
|
|
if mustSync {
|
|
|
|
return w.sync()
|
|
|
|
}
|
|
|
|
return nil
|
2015-02-20 02:41:37 +03:00
|
|
|
}
|
2016-03-15 03:21:14 +03:00
|
|
|
|
2015-02-20 02:41:37 +03:00
|
|
|
return w.cut()
|
2014-09-05 08:15:39 +04:00
|
|
|
}
|
|
|
|
|
2015-01-06 10:27:03 +03:00
|
|
|
func (w *WAL) SaveSnapshot(e walpb.Snapshot) error {
|
2016-09-28 08:05:19 +03:00
|
|
|
b := pbutil.MustMarshal(&e)
|
|
|
|
|
2015-04-08 10:25:18 +03:00
|
|
|
w.mu.Lock()
|
|
|
|
defer w.mu.Unlock()
|
|
|
|
|
2015-01-06 10:27:03 +03:00
|
|
|
rec := &walpb.Record{Type: snapshotType, Data: b}
|
|
|
|
if err := w.encoder.encode(rec); err != nil {
|
|
|
|
return err
|
|
|
|
}
|
|
|
|
// update enti only when snapshot is ahead of last index
|
|
|
|
if w.enti < e.Index {
|
|
|
|
w.enti = e.Index
|
|
|
|
}
|
|
|
|
return w.sync()
|
|
|
|
}
|
|
|
|
|
2014-09-04 02:10:15 +04:00
|
|
|
func (w *WAL) saveCrc(prevCrc uint32) error {
|
2014-09-04 03:46:42 +04:00
|
|
|
return w.encoder.encode(&walpb.Record{Type: crcType, Crc: prevCrc})
|
2014-07-26 02:21:04 +04:00
|
|
|
}
|
2016-03-14 06:34:37 +03:00
|
|
|
|
2016-03-15 03:21:14 +03:00
|
|
|
func (w *WAL) tail() *fileutil.LockedFile {
|
|
|
|
if len(w.locks) > 0 {
|
|
|
|
return w.locks[len(w.locks)-1]
|
|
|
|
}
|
|
|
|
return nil
|
|
|
|
}
|
|
|
|
|
|
|
|
func (w *WAL) seq() uint64 {
|
|
|
|
t := w.tail()
|
|
|
|
if t == nil {
|
|
|
|
return 0
|
|
|
|
}
|
2018-05-17 20:48:57 +03:00
|
|
|
seq, _, err := parseWALName(filepath.Base(t.Name()))
|
2016-03-15 03:21:14 +03:00
|
|
|
if err != nil {
|
2018-04-16 13:58:25 +03:00
|
|
|
if w.lg != nil {
|
|
|
|
w.lg.Fatal("failed to parse WAL name", zap.String("name", t.Name()), zap.Error(err))
|
|
|
|
} else {
|
|
|
|
plog.Fatalf("bad wal name %s (%v)", t.Name(), err)
|
|
|
|
}
|
2016-03-15 03:21:14 +03:00
|
|
|
}
|
|
|
|
return seq
|
|
|
|
}
|
|
|
|
|
2016-03-16 03:19:09 +03:00
|
|
|
func closeAll(rcs ...io.ReadCloser) error {
|
|
|
|
for _, f := range rcs {
|
|
|
|
if err := f.Close(); err != nil {
|
|
|
|
return err
|
|
|
|
}
|
|
|
|
}
|
|
|
|
return nil
|
|
|
|
}
|