removed locking locking all consumers

protocol/v2/ssv/runner/runner.go

@@ -177,7 +177,7 @@ func (b *BaseRunner) basePreConsensusMsgProcessing(runner Runner, signedMsg *spe
 func (b *BaseRunner) baseConsensusMsgProcessing(logger *zap.Logger, runner Runner, msg *spectypes.SignedSSVMessage) (decided bool, decidedValue spectypes.Encoder, err error) {
 	prevDecided := false
 	if b.hasRunningDuty() && b.State != nil && b.State.RunningInstance != nil {
-		prevDecided, _ = b.State.RunningInstance.IsDecided()
+		prevDecided, _ = b.IsRunningInstanceDecided()
 	}
 
 	// TODO: revert after pre-consensus liveness is fixed
@@ -187,7 +187,9 @@ func (b *BaseRunner) baseConsensusMsgProcessing(logger *zap.Logger, runner Runne
 		}
 	}
 
+	b.mtx.Lock()
 	decidedMsg, err := b.QBFTController.ProcessMsg(logger, msg)
+	b.mtx.Unlock()
 	b.compactInstanceIfNeeded(msg)
 	if err != nil {
 		return false, nil, err
@@ -375,3 +377,15 @@ func (b *BaseRunner) ShouldProcessNonBeaconDuty(duty spectypes.Duty) error {
 	}
 	return nil
 }
+
+func (b *BaseRunner) IsRunningInstanceDecided() (bool, []byte) {
+	b.mtx.RLock()
+	defer b.mtx.RUnlock()
+	return b.State.RunningInstance.IsDecided()
+}
+
+func (b *BaseRunner) GetStateProposalAcceptedForCurrentRound() *spectypes.SignedSSVMessage {
+	b.mtx.RLock()
+	defer b.mtx.RUnlock()
+	return b.State.RunningInstance.State.ProposalAcceptedForCurrentRound
+}

protocol/v2/ssv/validator/committee.go

@@ -236,8 +236,8 @@ func (c *Committee) ProcessMessage(logger *zap.Logger, msg *queue.DecodedSSVMess
 			return errors.Wrap(err, "invalid qbft Message")
 		}
 		c.mtx.Lock()
-		defer c.mtx.Unlock() // read c.Runners map + ProcessConsensus r/w state
 		runner, exists := c.Runners[phase0.Slot(qbftMsg.Height)]
+		c.mtx.Unlock()
 		if !exists {
 			return errors.New("no runner found for message's slot")
 		}

protocol/v2/ssv/validator/committee_queue.go

@@ -110,10 +110,8 @@ func (v *Committee) ConsumeQueue(logger *zap.Logger, slot phase0.Slot, handler M
 		if runner.HasRunningDuty() {
 			runningInstance = runner.GetBaseRunner().State.RunningInstance
 			if runningInstance != nil {
-				v.mtx.RLock()
-				decided, _ := runningInstance.IsDecided()
+				decided, _ := runner.GetBaseRunner().IsRunningInstanceDecided()
 				state.HasRunningInstance = !decided
-				v.mtx.RUnlock()
 			}
 		}
 
@@ -128,9 +126,7 @@ func (v *Committee) ConsumeQueue(logger *zap.Logger, slot phase0.Slot, handler M
 				return e.Type == types.ExecuteDuty
 			}
 		} else {
-			v.mtx.RLock()
-			proposalAcceptedForCurrentRound := runningInstance.State.ProposalAcceptedForCurrentRound
-			v.mtx.RUnlock()
+			proposalAcceptedForCurrentRound := runner.GetBaseRunner().GetStateProposalAcceptedForCurrentRound()
 			if runningInstance != nil && proposalAcceptedForCurrentRound == nil {
 				// If no proposal was accepted for the current round, skip prepare & commit messages
 				// for the current height and round.