more granular locks

internal/consensus/invalid_test.go

@@ -48,11 +48,12 @@ func TestReactorInvalidPrecommit(t *testing.T) {
 	// Update the doPrevote function to just send a valid precommit for a random
 	// block and otherwise disable the priv validator.
 	privVal := byzState.privValidator
+	byzState.mtx.Lock()
 	byzState.doPrevote = func(ctx context.Context, height int64, round int32) {
 		defer close(signal)
 		invalidDoPrevoteFunc(ctx, t, height, round, byzState, byzReactor, privVal)
 	}
-
+	byzState.mtx.Unlock()
 	// wait for a bunch of blocks
 	//
 	// TODO: Make this tighter by ensuring the halt happens by block 2.
@@ -79,9 +80,9 @@ func TestReactorInvalidPrecommit(t *testing.T) {
 
 	select {
 	case <-wait:
-		t.Fatal("test condition did not fire")
+		t.Error("test condition did not fire")
 	case <-ctx.Done():
-		t.Fatal("test condition did not fire after timeout")
+		t.Error("test condition did not fire after timeout")
 	case <-signal:
 		// test passed
 	}

internal/consensus/reactor_test.go

@@ -525,8 +525,7 @@ func TestReactorWithEvidence(t *testing.T) {
 	rts := setup(ctx, t, n, states, 100) // buffer must be large enough to not deadlock
 
 	for _, reactor := range rts.reactors {
-		state := reactor.state.GetState()
-		reactor.SwitchToConsensus(ctx, state, false)
+		reactor.SwitchToConsensus(ctx, reactor.state.state, false)
 	}
 
 	var wg sync.WaitGroup

internal/consensus/state.go

@@ -1030,10 +1030,6 @@ func (cs *State) handleTimeout(
 		return
 	}
 
-	// the timeout will now cause a state transition
-	cs.mtx.Lock()
-	defer cs.mtx.Unlock()
-
 	switch ti.Step {
 	case cstypes.RoundStepNewHeight:
 		// NewRound event fired from enterNewRound.
@@ -1689,9 +1685,13 @@ func (cs *State) enterPrecommit(ctx context.Context, height int64, round int32)
 			panic(fmt.Sprintf("precommit step: +2/3 prevoted for an invalid block %v; relocking", err))
 		}
 
-		cs.LockedRound = round
-		cs.LockedBlock = cs.ProposalBlock
-		cs.LockedBlockParts = cs.ProposalBlockParts
+		func() {
+			cs.mtx.Lock()
+			defer cs.mtx.Unlock()
+			cs.LockedRound = round
+			cs.LockedBlock = cs.ProposalBlock
+			cs.LockedBlockParts = cs.ProposalBlockParts
+		}()
 
 		if err := cs.eventBus.PublishEventLock(ctx, cs.RoundStateEvent()); err != nil {
 			logger.Error("precommit step: failed publishing event lock", "err", err)
@@ -1782,8 +1782,13 @@ func (cs *State) enterCommit(ctx context.Context, height int64, commitRound int3
 	// otherwise they'll be cleared in updateToState.
 	if cs.LockedBlock.HashesTo(blockID.Hash) {
 		logger.Debug("commit is for a locked block; set ProposalBlock=LockedBlock", "block_hash", blockID.Hash)
-		cs.ProposalBlock = cs.LockedBlock
-		cs.ProposalBlockParts = cs.LockedBlockParts
+		func() {
+			cs.mtx.Lock()
+			defer cs.mtx.Unlock()
+
+			cs.ProposalBlock = cs.LockedBlock
+			cs.ProposalBlockParts = cs.LockedBlockParts
+		}()
 	}
 
 	// If we don't have the block being committed, set up to get it.
@@ -2103,8 +2108,6 @@ func (cs *State) addProposalBlockPart(
 	peerID types.NodeID,
 ) (added bool, err error) {
 	height, round, part := msg.Height, msg.Round, msg.Part
-	cs.mtx.Lock()
-	defer cs.mtx.Unlock()
 
 	// Blocks might be reused, so round mismatch is OK
 	if cs.Height != height {
@@ -2161,7 +2164,12 @@ func (cs *State) addProposalBlockPart(
 			return added, err
 		}
 
-		cs.ProposalBlock = block
+		func() {
+			cs.mtx.Lock()
+			defer cs.mtx.Unlock()
+
+			cs.ProposalBlock = block
+		}()
 
 		// NOTE: it's possible to receive complete proposal blocks for future rounds without having the proposal
 		cs.logger.Info("received complete proposal block", "height", cs.ProposalBlock.Height, "hash", cs.ProposalBlock.Hash())
@@ -2342,9 +2350,14 @@ func (cs *State) addVote(
 			if cs.ValidRound < vote.Round && vote.Round == cs.Round {
 				if cs.ProposalBlock.HashesTo(blockID.Hash) {
 					cs.logger.Debug("updating valid block because of POL", "valid_round", cs.ValidRound, "pol_round", vote.Round)
-					cs.ValidRound = vote.Round
-					cs.ValidBlock = cs.ProposalBlock
-					cs.ValidBlockParts = cs.ProposalBlockParts
+					func() {
+						cs.mtx.Lock()
+						defer cs.mtx.Unlock()
+
+						cs.ValidRound = vote.Round
+						cs.ValidBlock = cs.ProposalBlock
+						cs.ValidBlockParts = cs.ProposalBlockParts
+					}()
 				} else {
 					cs.logger.Debug(
 						"valid block we do not know about; set ProposalBlock=nil",

internal/state/state.go

@@ -104,7 +104,6 @@ type State struct {
 
 // Copy makes a copy of the State for mutating.
 func (state State) Copy() State {
-
 	return State{
 		Version:       state.Version,
 		ChainID:       state.ChainID,