Merge pull request #6707 from multiversx/fix-chain-simulator

fix chain simulator

node/chainSimulator/chainSimulator.go

@@ -298,15 +298,18 @@ func (s *simulator) incrementRoundOnAllValidators() {
 // ForceChangeOfEpoch will force the change of current epoch
 // This method will call the epoch change trigger and generate block till a new epoch is reached
 func (s *simulator) ForceChangeOfEpoch() error {
+	s.mutex.Lock()
 	log.Info("force change of epoch")
 	for shardID, node := range s.nodes {
 		err := node.ForceChangeOfEpoch()
 		if err != nil {
+			s.mutex.Unlock()
 			return fmt.Errorf("force change of epoch shardID-%d: error-%w", shardID, err)
 		}
 	}
 
 	epoch := s.nodes[core.MetachainShardId].GetProcessComponents().EpochStartTrigger().Epoch()
+	s.mutex.Unlock()
 
 	return s.GenerateBlocksUntilEpochIsReached(int32(epoch + 1))
 }

node/chainSimulator/components/syncedMessenger.go

@@ -80,13 +80,21 @@ func (messenger *syncedMessenger) receive(fromConnectedPeer core.PeerID, message
 	handlers := messenger.topics[message.Topic()]
 	messenger.mutOperation.RUnlock()
 
+	wg := &sync.WaitGroup{}
+	wg.Add(len(handlers))
 	for _, handler := range handlers {
-		err := handler.ProcessReceivedMessage(message, fromConnectedPeer, messenger)
-		if err != nil {
-			log.Trace("received message syncedMessenger",
-				"error", err, "topic", message.Topic(), "from connected peer", fromConnectedPeer.Pretty())
-		}
+		// this is needed to process all received messages on multiple go routines
+		go func(proc p2p.MessageProcessor, p2pMessage p2p.MessageP2P, peer core.PeerID, localWG *sync.WaitGroup) {
+			err := proc.ProcessReceivedMessage(p2pMessage, peer, messenger)
+			if err != nil {
+				log.Trace("received message syncedMessenger", "error", err, "topic", p2pMessage.Topic(), "from connected peer", peer.Pretty())
+			}
+
+			localWG.Done()
+		}(handler, message, fromConnectedPeer, wg)
 	}
+
+	wg.Wait()
 }
 
 // ProcessReceivedMessage does nothing and returns nil