Merge branch 'main' into fix-byzzfuzz-network-partitions

Author: joaomlneto

simulator/src/main/java/byzzbench/simulator/scheduler/BaseScheduler.java

@@ -118,7 +118,7 @@ public List<ClientRequestEvent> getQueuedClientRequestEvents(Scenario scenario)
      * @param messageEvents The list of queued message events
      * @return The next message event
      */
-    public Event getNextMessageEvent(List<Event> messageEvents) {
+    public <T extends Event> T getNextMessageEvent(List<T> messageEvents) {
         switch (config.getScheduler().getExecutionMode()) {
             case SYNC -> {
                 return messageEvents.stream().min(Comparator.comparing(Event::getEventId)).orElseThrow();

simulator/src/main/java/byzzbench/simulator/scheduler/RandomScheduler.java

@@ -15,7 +15,7 @@
 import java.util.List;
 import java.util.Optional;
 import java.util.Random;
-import java.util.stream.Collectors;
+import java.util.SortedSet;
 
 /**
  * A scheduler that randomly selects events to deliver, drop, mutate or timeout.
@@ -25,11 +25,14 @@
 public class RandomScheduler extends BaseScheduler {
     private final Random random = new Random();
 
-
     public RandomScheduler(ByzzBenchConfig config, MessageMutatorService messageMutatorService) {
         super(config, messageMutatorService);
     }
 
+    public <T> T getRandomElement(List<T> list) {
+        return list.get(random.nextInt(list.size()));
+    }
+
     @Override
     public String getId() {
         return "Random";
@@ -52,21 +55,24 @@ public synchronized Optional<EventDecision> scheduleNext(Scenario scenario) thro
         }
 
         List<TimeoutEvent> timeoutEvents = this.getQueuedTimeoutEvents(scenario);
-        List<Event> clientRequestEvents = availableEvents.stream().filter(ClientRequestEvent.class::isInstance).collect(Collectors.toList());
-        List<Event> messageEvents = availableEvents.stream().filter(MessageEvent.class::isInstance).collect(Collectors.toList());
+        List<Event> clientRequestEvents = availableEvents.stream().filter(ClientRequestEvent.class::isInstance).toList();
+        List<MessageEvent> messageEvents = availableEvents.stream().filter(MessageEvent.class::isInstance).map(MessageEvent.class::cast).toList();
+
+        SortedSet<String> faultyReplicaIds = scenario.getFaultyReplicaIds();
+        List<MessageEvent> mutateableMessageEvents = messageEvents.stream().filter(msg -> faultyReplicaIds.contains(msg.getSenderId())).toList();
 
         int timeoutWeight = timeoutEvents.size() * this.deliverTimeoutWeight();
         int deliverMessageWeight = messageEvents.size() * this.deliverMessageWeight();
         int deliverClientRequestWeight = clientRequestEvents.size() * this.deliverClientRequestWeight();
         int dropMessageWeight = (messageEvents.size() * this.dropMessageWeight(scenario));
-        int mutateMessageWeight = (messageEvents.size() * this.mutateMessageWeight(scenario));
+        int mutateMessageWeight = (mutateableMessageEvents.size() * this.mutateMessageWeight(scenario));
         int dieRoll = random.nextInt(timeoutWeight + deliverMessageWeight
                 + deliverClientRequestWeight + dropMessageWeight + mutateMessageWeight);
 
         // check if we should trigger a timeout
         dieRoll -= timeoutWeight;
         if (dieRoll < 0) {
-            Event timeout = timeoutEvents.get(random.nextInt(timeoutEvents.size()));
+            Event timeout = getRandomElement(timeoutEvents);
             scenario.getTransport().deliverEvent(timeout.getEventId());
             EventDecision decision = new EventDecision(EventDecision.DecisionType.DELIVERED, timeout.getEventId());
             return Optional.of(decision);
@@ -84,7 +90,7 @@ public synchronized Optional<EventDecision> scheduleNext(Scenario scenario) thro
         // check if we should target delivering a request from a client to a replica
         dieRoll -= deliverClientRequestWeight;
         if (dieRoll < 0) {
-            Event request = clientRequestEvents.get(random.nextInt(clientRequestEvents.size()));
+            Event request = getRandomElement(clientRequestEvents);
             scenario.getTransport().deliverEvent(request.getEventId());
             EventDecision decision = new EventDecision(EventDecision.DecisionType.DELIVERED, request.getEventId());
             return Optional.of(decision);
@@ -93,7 +99,7 @@ public synchronized Optional<EventDecision> scheduleNext(Scenario scenario) thro
         // check if we should drop a message sent between nodes
         dieRoll -= dropMessageWeight;
         if (dieRoll < 0) {
-            Event message = messageEvents.get(random.nextInt(messageEvents.size()));
+            Event message = getRandomElement(messageEvents);
             scenario.getTransport().dropEvent(message.getEventId());
             EventDecision decision = new EventDecision(EventDecision.DecisionType.DROPPED, message.getEventId());
             return Optional.of(decision);
@@ -102,7 +108,7 @@ public synchronized Optional<EventDecision> scheduleNext(Scenario scenario) thro
         // check if we should mutate-and-deliver a message sent between nodes
         dieRoll -= mutateMessageWeight;
         if (dieRoll < 0) {
-            Event message = messageEvents.get(random.nextInt(messageEvents.size()));
+            Event message = getRandomElement(mutateableMessageEvents);
             List<MessageMutationFault> mutators = this.getMessageMutatorService().getMutatorsForEvent(message);
 
             if (mutators.isEmpty()) {
@@ -112,7 +118,7 @@ public synchronized Optional<EventDecision> scheduleNext(Scenario scenario) thro
             }
             scenario.getTransport().applyMutation(
                     message.getEventId(),
-                    mutators.get(random.nextInt(mutators.size())));
+                    getRandomElement(mutators));
             scenario.getTransport().deliverEvent(message.getEventId());
 
             EventDecision decision = new EventDecision(EventDecision.DecisionType.MUTATED_AND_DELIVERED, message.getEventId());