Merge pull request #11946 from crawshaw/crawshaw/nofsync

etcdserver, et al: add --unsafe-no-fsync flag

embed/config.go

@@ -321,6 +321,10 @@ type Config struct {
 
 	// EnableGRPCGateway is false to disable grpc gateway.
 	EnableGRPCGateway bool `json:"enable-grpc-gateway"`
+
+	// UnsafeNoFsync disables all uses of fsync.
+	// Setting this is unsafe and will cause data loss.
+	UnsafeNoFsync bool `json:"unsafe-no-fsync"`
 }
 
 // configYAML holds the config suitable for yaml parsing

embed/etcd.go

@@ -196,6 +196,7 @@ func StartEtcd(inCfg *Config) (e *Etcd, err error) {
 		LoggerWriteSyncer:          cfg.loggerWriteSyncer,
 		ForceNewCluster:            cfg.ForceNewCluster,
 		EnableGRPCGateway:          cfg.EnableGRPCGateway,
+		UnsafeNoFsync:              cfg.UnsafeNoFsync,
 		EnableLeaseCheckpoint:      cfg.ExperimentalEnableLeaseCheckpoint,
 		CompactionBatchLimit:       cfg.ExperimentalCompactionBatchLimit,
 	}

etcdmain/config.go

@@ -252,6 +252,7 @@ func newConfig() *config {
 	fs.IntVar(&cfg.ec.ExperimentalCompactionBatchLimit, "experimental-compaction-batch-limit", cfg.ec.ExperimentalCompactionBatchLimit, "Sets the maximum revisions deleted in each compaction batch.")
 
 	// unsafe
+	fs.BoolVar(&cfg.ec.UnsafeNoFsync, "unsafe-no-fsync", false, "Disables fsync, unsafe, will cause data loss.")
 	fs.BoolVar(&cfg.ec.ForceNewCluster, "force-new-cluster", false, "Force to create a new one member cluster.")
 
 	// ignored

etcdserver/backend.go

@@ -30,6 +30,7 @@ import (
 func newBackend(cfg ServerConfig) backend.Backend {
 	bcfg := backend.DefaultBackendConfig()
 	bcfg.Path = cfg.backendPath()
+	bcfg.UnsafeNoFsync = cfg.UnsafeNoFsync
 	if cfg.BackendBatchLimit != 0 {
 		bcfg.BatchLimit = cfg.BackendBatchLimit
 		if cfg.Logger != nil {

etcdserver/config.go

@@ -155,6 +155,10 @@ type ServerConfig struct {
 	LeaseCheckpointInterval time.Duration
 
 	EnableGRPCGateway bool
+
+	// UnsafeNoFsync disables all uses of fsync.
+	// Setting this is unsafe and will cause data loss.
+	UnsafeNoFsync bool `json:"unsafe-no-fsync"`
 }
 
 // VerifyBootstrap sanity-checks the initial config for bootstrap case

etcdserver/raft.go

@@ -428,6 +428,9 @@ func startNode(cfg ServerConfig, cl *membership.RaftCluster, ids []types.ID) (id
 	if w, err = wal.Create(cfg.Logger, cfg.WALDir(), metadata); err != nil {
 		cfg.Logger.Panic("failed to create WAL", zap.Error(err))
 	}
+	if cfg.UnsafeNoFsync {
+		w.SetUnsafeNoFsync()
+	}
 	peers := make([]raft.Peer, len(ids))
 	for i, id := range ids {
 		var ctx []byte
@@ -482,7 +485,7 @@ func restartNode(cfg ServerConfig, snapshot *raftpb.Snapshot) (types.ID, *member
 	if snapshot != nil {
 		walsnap.Index, walsnap.Term = snapshot.Metadata.Index, snapshot.Metadata.Term
 	}
-	w, id, cid, st, ents := readWAL(cfg.Logger, cfg.WALDir(), walsnap)
+	w, id, cid, st, ents := readWAL(cfg.Logger, cfg.WALDir(), walsnap, cfg.UnsafeNoFsync)
 
 	cfg.Logger.Info(
 		"restarting local member",
@@ -533,7 +536,7 @@ func restartAsStandaloneNode(cfg ServerConfig, snapshot *raftpb.Snapshot) (types
 	if snapshot != nil {
 		walsnap.Index, walsnap.Term = snapshot.Metadata.Index, snapshot.Metadata.Term
 	}
-	w, id, cid, st, ents := readWAL(cfg.Logger, cfg.WALDir(), walsnap)
+	w, id, cid, st, ents := readWAL(cfg.Logger, cfg.WALDir(), walsnap, cfg.UnsafeNoFsync)
 
 	// discard the previously uncommitted entries
 	for i, ent := range ents {

etcdserver/storage.go

@@ -82,7 +82,7 @@ func (st *storage) Release(snap raftpb.Snapshot) error {
 // readWAL reads the WAL at the given snap and returns the wal, its latest HardState and cluster ID, and all entries that appear
 // after the position of the given snap in the WAL.
 // The snap must have been previously saved to the WAL, or this call will panic.
-func readWAL(lg *zap.Logger, waldir string, snap walpb.Snapshot) (w *wal.WAL, id, cid types.ID, st raftpb.HardState, ents []raftpb.Entry) {
+func readWAL(lg *zap.Logger, waldir string, snap walpb.Snapshot, unsafeNoFsync bool) (w *wal.WAL, id, cid types.ID, st raftpb.HardState, ents []raftpb.Entry) {
 	var (
 		err       error
 		wmetadata []byte
@@ -93,6 +93,9 @@ func readWAL(lg *zap.Logger, waldir string, snap walpb.Snapshot) (w *wal.WAL, id
 		if w, err = wal.Open(lg, waldir, snap); err != nil {
 			lg.Fatal("failed to open WAL", zap.Error(err))
 		}
+		if unsafeNoFsync {
+			w.SetUnsafeNoFsync()
+		}
 		if wmetadata, st, ents, err = w.ReadAll(); err != nil {
 			w.Close()
 			// we can only repair ErrUnexpectedEOF and we never repair twice.

mvcc/backend/backend.go

@@ -120,6 +120,8 @@ type BackendConfig struct {
 	MmapSize uint64
 	// Logger logs backend-side operations.
 	Logger *zap.Logger
+	// UnsafeNoFsync disables all uses of fsync.
+	UnsafeNoFsync bool `json:"unsafe-no-fsync"`
 }
 
 func DefaultBackendConfig() BackendConfig {
@@ -151,6 +153,8 @@ func newBackend(bcfg BackendConfig) *backend {
 	}
 	bopts.InitialMmapSize = bcfg.mmapSize()
 	bopts.FreelistType = bcfg.BackendFreelistType
+	bopts.NoSync = bcfg.UnsafeNoFsync
+	bopts.NoGrowSync = bcfg.UnsafeNoFsync
 
 	db, err := bolt.Open(bcfg.Path, 0600, bopts)
 	if err != nil {

wal/wal.go

@@ -85,6 +85,8 @@ type WAL struct {
 	decoder   *decoder       // decoder to decode records
 	readClose func() error   // closer for decode reader
 
+	unsafeNoSync bool // if set, do not fsync
+
 	mu      sync.Mutex
 	enti    uint64   // index of the last entry saved to the wal
 	encoder *encoder // encoder to encode records
@@ -230,6 +232,10 @@ func Create(lg *zap.Logger, dirpath string, metadata []byte) (*WAL, error) {
 	return w, nil
 }
 
+func (w *WAL) SetUnsafeNoFsync() {
+	w.unsafeNoSync = true
+}
+
 func (w *WAL) cleanupWAL(lg *zap.Logger) {
 	var err error
 	if err = w.Close(); err != nil {
@@ -767,6 +773,9 @@ func (w *WAL) cut() error {
 }
 
 func (w *WAL) sync() error {
+	if w.unsafeNoSync {
+		return nil
+	}
 	if w.encoder != nil {
 		if err := w.encoder.flush(); err != nil {
 			return err