
committed by
Gyu-Ho Lee

parent
6e83ec0ed7
commit
202da9270e
48
wal/wal.go
48
wal/wal.go
@ -68,6 +68,10 @@ var (
|
|||||||
// The WAL will be ready for appending after reading out all the previous records.
|
// The WAL will be ready for appending after reading out all the previous records.
|
||||||
type WAL struct {
|
type WAL struct {
|
||||||
dir string // the living directory of the underlay files
|
dir string // the living directory of the underlay files
|
||||||
|
|
||||||
|
// dirFile is a fd for the wal directory for syncing on Rename
|
||||||
|
dirFile *os.File
|
||||||
|
|
||||||
metadata []byte // metadata recorded at the head of each WAL
|
metadata []byte // metadata recorded at the head of each WAL
|
||||||
state raftpb.HardState // hardstate recorded at the head of WAL
|
state raftpb.HardState // hardstate recorded at the head of WAL
|
||||||
|
|
||||||
@ -106,7 +110,7 @@ func Create(dirpath string, metadata []byte) (*WAL, error) {
|
|||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
if _, err := f.Seek(0, os.SEEK_END); err != nil {
|
if _, err = f.Seek(0, os.SEEK_END); err != nil {
|
||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
if err := fileutil.Preallocate(f.File, segmentSizeBytes, true); err != nil {
|
if err := fileutil.Preallocate(f.File, segmentSizeBytes, true); err != nil {
|
||||||
@ -119,17 +123,33 @@ func Create(dirpath string, metadata []byte) (*WAL, error) {
|
|||||||
encoder: newEncoder(f, 0),
|
encoder: newEncoder(f, 0),
|
||||||
}
|
}
|
||||||
w.locks = append(w.locks, f)
|
w.locks = append(w.locks, f)
|
||||||
if err := w.saveCrc(0); err != nil {
|
if err = w.saveCrc(0); err != nil {
|
||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
if err := w.encoder.encode(&walpb.Record{Type: metadataType, Data: metadata}); err != nil {
|
if err = w.encoder.encode(&walpb.Record{Type: metadataType, Data: metadata}); err != nil {
|
||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
if err := w.SaveSnapshot(walpb.Snapshot{}); err != nil {
|
if err = w.SaveSnapshot(walpb.Snapshot{}); err != nil {
|
||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
|
|
||||||
return w.renameWal(tmpdirpath)
|
if w, err = w.renameWal(tmpdirpath); err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
|
||||||
|
// directory was renamed; sync parent dir to persist rename
|
||||||
|
pdir, perr := fileutil.OpenDir(path.Dir(w.dir))
|
||||||
|
if perr != nil {
|
||||||
|
return nil, perr
|
||||||
|
}
|
||||||
|
if perr = fileutil.Fsync(pdir); perr != nil {
|
||||||
|
return nil, perr
|
||||||
|
}
|
||||||
|
if perr = pdir.Close(); err != nil {
|
||||||
|
return nil, perr
|
||||||
|
}
|
||||||
|
|
||||||
|
return w, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
// Open opens the WAL at the given snap.
|
// Open opens the WAL at the given snap.
|
||||||
@ -139,7 +159,14 @@ func Create(dirpath string, metadata []byte) (*WAL, error) {
|
|||||||
// the given snap. The WAL cannot be appended to before reading out all of its
|
// the given snap. The WAL cannot be appended to before reading out all of its
|
||||||
// previous records.
|
// previous records.
|
||||||
func Open(dirpath string, snap walpb.Snapshot) (*WAL, error) {
|
func Open(dirpath string, snap walpb.Snapshot) (*WAL, error) {
|
||||||
return openAtIndex(dirpath, snap, true)
|
w, err := openAtIndex(dirpath, snap, true)
|
||||||
|
if err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
if w.dirFile, err = fileutil.OpenDir(w.dir); err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
return w, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
// OpenForRead only opens the wal files for read.
|
// OpenForRead only opens the wal files for read.
|
||||||
@ -371,6 +398,10 @@ func (w *WAL) cut() error {
|
|||||||
if err = os.Rename(newTail.Name(), fpath); err != nil {
|
if err = os.Rename(newTail.Name(), fpath); err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
|
if err = fileutil.Fsync(w.dirFile); err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
|
||||||
newTail.Close()
|
newTail.Close()
|
||||||
|
|
||||||
if newTail, err = fileutil.LockFile(fpath, os.O_WRONLY, fileutil.PrivateFileMode); err != nil {
|
if newTail, err = fileutil.LockFile(fpath, os.O_WRONLY, fileutil.PrivateFileMode); err != nil {
|
||||||
@ -473,6 +504,11 @@ func (w *WAL) Close() error {
|
|||||||
plog.Errorf("failed to unlock during closing wal: %s", err)
|
plog.Errorf("failed to unlock during closing wal: %s", err)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
if err := w.dirFile.Close(); err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@ -16,7 +16,11 @@
|
|||||||
|
|
||||||
package wal
|
package wal
|
||||||
|
|
||||||
import "os"
|
import (
|
||||||
|
"os"
|
||||||
|
|
||||||
|
"github.com/coreos/etcd/pkg/fileutil"
|
||||||
|
)
|
||||||
|
|
||||||
func (w *WAL) renameWal(tmpdirpath string) (*WAL, error) {
|
func (w *WAL) renameWal(tmpdirpath string) (*WAL, error) {
|
||||||
// On non-Windows platforms, hold the lock while renaming. Releasing
|
// On non-Windows platforms, hold the lock while renaming. Releasing
|
||||||
@ -34,5 +38,7 @@ func (w *WAL) renameWal(tmpdirpath string) (*WAL, error) {
|
|||||||
}
|
}
|
||||||
|
|
||||||
w.fp = newFilePipeline(w.dir, segmentSizeBytes)
|
w.fp = newFilePipeline(w.dir, segmentSizeBytes)
|
||||||
return w, nil
|
df, err := fileutil.OpenDir(w.dir)
|
||||||
|
w.dirFile = df
|
||||||
|
return w, err
|
||||||
}
|
}
|
||||||
|
Reference in New Issue
Block a user