Restrict WAL usage to node.go (#4201)

* Revert "synchronize access to wal storage from chain and node (#3919)"

This reverts commit 2678417.

Signed-off-by: Yacov Manevich <[email protected]>

* Restrict WAL usage to node.go

This commit moves the WAL sync from the goroutine of the chain,
to the goroutine that performs all the rest of the WAL operations,
in order to prevent concurrent invocation of the WAL which is not thread safe.

Signed-off-by: Yacov Manevich <[email protected]>

---------

Signed-off-by: Yacov Manevich <[email protected]>

Author: yacovm

orderer/consensus/etcdraft/chain.go

@@ -1179,10 +1179,8 @@ func (c *Chain) apply(ents []raftpb.Entry) {
 				continue
 			}
 
-			// persist the WAL entries into disk
-			c.Node.storage.WALSyncC <- struct{}{}
-
 			c.confState = *c.Node.ApplyConfChange(cc)
+
 			switch cc.Type {
 			case raftpb.ConfChangeAddNode:
 				c.logger.Infof("Applied config change to add node %d, current nodes in channel: %+v", cc.NodeID, c.confState.Voters)

orderer/consensus/etcdraft/node.go

@@ -153,6 +153,7 @@ func (n *node) run(campaign bool) {
 
 			// skip empty apply
 			if len(rd.CommittedEntries) != 0 || rd.SoftState != nil {
+				n.maybeSyncWAL(rd.CommittedEntries)
 				n.chain.applyC <- apply{rd.CommittedEntries, rd.SoftState}
 			}
 
@@ -171,11 +172,6 @@ func (n *node) run(campaign bool) {
 			// to the followers and them writing to their disks. Check 10.2.1 in thesis
 			n.send(rd.Messages)
 
-		case <-n.storage.WALSyncC:
-			if err := n.storage.Sync(); err != nil {
-				n.logger.Warnf("Failed to sync raft log, error: %s", err)
-			}
-
 		case <-n.chain.haltC:
 			raftTicker.Stop()
 			n.Stop()
@@ -224,6 +220,24 @@ func (n *node) send(msgs []raftpb.Message) {
 	}
 }
 
+func (n *node) maybeSyncWAL(entries []raftpb.Entry) {
+	allNormal := true
+	for _, entry := range entries {
+		if entry.Type == raftpb.EntryNormal {
+			continue
+		}
+		allNormal = false
+	}
+
+	if allNormal {
+		return
+	}
+
+	if err := n.storage.Sync(); err != nil {
+		n.logger.Errorf("Failed to sync raft log, error: %s", err)
+	}
+}
+
 // abdicateLeadership picks a node that is recently active, and attempts to transfer leadership to it.
 // Blocks until leadership transfer happens or when a timeout expires.
 // Returns error upon failure.

orderer/consensus/etcdraft/storage.go

@@ -59,7 +59,6 @@ type RaftStorage struct {
 
 	// a queue that keeps track of indices of snapshots on disk
 	snapshotIndex []uint64
-	WALSyncC      chan struct{}
 }
 
 // CreateStorage attempts to create a storage to persist etcd/raft data.
@@ -114,7 +113,6 @@ func CreateStorage(
 		walDir:        walDir,
 		snapDir:       snapDir,
 		snapshotIndex: ListSnapshots(lg, snapDir),
-		WALSyncC:      make(chan struct{}),
 	}, nil
 }