Merge remote-tracking branch 'origin/main' into hBFT-fixed

Author: Biatbe13

simulator/src/main/java/byzzbench/simulator/BaseScenario.java

@@ -179,7 +179,16 @@ public synchronized Node getNode(String nodeId) {
     @Override
     public final void setupScenario() {
         this.setup();
-        //this.getClients().values().forEach(Client::initialize);
+
+        // sample f replicas to be faulty at start
+        List<String> replicaIds = new ArrayList<>(this.getReplicas().keySet());
+        Collections.shuffle(replicaIds);
+        int f = this.maxFaultyReplicas();
+        for (int i = 0; i < f; i++) {
+            this.markReplicaFaulty(replicaIds.get(i));
+        }
+
+        this.getClients().values().forEach(Client::initialize);
         this.getNodes().values().forEach(Node::initialize);
         this.scheduler.initializeScenario(this);
     }

simulator/src/main/java/byzzbench/simulator/Timekeeper.java

@@ -73,6 +73,11 @@ public void onEventDropped(Event event) {
         // nothing to do
     }
 
+    @Override
+    public void onEventRequeued(Event event) {
+        // nothing to do
+    }
+
     @Override
     public void onEventDelivered(Event event) {
         // check if it was a timeout

simulator/src/main/java/byzzbench/simulator/faults/factories/ByzzFuzzScenarioFaultFactory.java

@@ -37,6 +37,7 @@ public List<Fault> generateFaults(FaultContext input) {
         int d = scheduler.getNumRoundsWithNetworkFaults();
         int r = scheduler.getNumRoundsWithFaults();
         Set<String> replicaIds = scenario.getReplicas().keySet();
+        Set<String> faultyReplicaIds = scenario.getFaultyReplicaIds();
 
         // Create network faults
         for (int i = 1; i <= d; i++) {
@@ -49,7 +50,7 @@ public List<Fault> generateFaults(FaultContext input) {
         // Create process faults
         for (int i = 1; i <= c; i++) {
             int round = rand.nextInt(r) + 1;
-            String sender = replicaIds.stream().skip(rand.nextInt(replicaIds.size())).findFirst().orElseThrow();
+            String sender = input.getScenario().getFaultyReplicaIds().stream().skip(rand.nextInt(faultyReplicaIds.size())).findFirst().orElseThrow();
             Set<String> recipientIds = SetSubsets.getRandomNonEmptySubset(replicaIds);
 
             // generate process fault

simulator/src/main/java/byzzbench/simulator/transport/Transport.java

@@ -77,6 +77,8 @@ public class Transport {
      */
     @JsonIgnore
     private final List<TransportObserver> observers = new ArrayList<>();
+    @Getter
+    private boolean isGlobalStabilizationTime = false;
 
     /**
      * Adds an observer to the transport layer.
@@ -350,6 +352,11 @@ public synchronized void deliverEvent(long eventId) throws Exception {
      * @param eventId The ID of the message to drop.
      */
     public synchronized void dropEvent(long eventId) {
+        // Check if it is GST - no more dropping
+        if (this.isGlobalStabilizationTime) {
+            throw new IllegalStateException("Cannot drop events during GST");
+        }
+
         // check if event is a message
         Event e = events.get(eventId);
         if (e instanceof TimeoutEvent) {
@@ -359,6 +366,7 @@ public synchronized void dropEvent(long eventId) {
         if (e.getStatus() != Event.Status.QUEUED) {
             throw new IllegalArgumentException("Event not found or not in QUEUED state");
         }
+
         e.setStatus(Event.Status.DROPPED);
         this.observers.forEach(o -> o.onEventDropped(e));
         //log.info("Dropped: " + e);
@@ -404,6 +412,13 @@ public synchronized void applyMutation(long eventId, Fault fault) {
                     "Event %d is not a message - cannot mutate it.", eventId));
         }
 
+        // check if sender is faulty
+        if (!this.scenario.isFaultyReplica(m.getSenderId())) {
+            throw new IllegalArgumentException(
+                    String.format("Cannot mutate message: sender %s is not marked as faulty", m.getSenderId())
+            );
+        }
+
         // create input for the fault
         FaultContext input = new FaultContext(this.scenario, e);
 
@@ -647,4 +662,32 @@ public synchronized List<Fault> getEnabledNetworkFaults() {
     public synchronized Fault getNetworkFault(String faultId) {
         return this.networkFaults.get(faultId);
     }
+
+    /**
+     * Simulates GST event, according to the partial-synchrony model:
+     * <ul>
+     *   <li>All dropped messages are re-queued</li>
+     *   <li>Prevents further dropping of messages</li>
+     *   <li>All network partitions are healed</li>
+     *   <li>Prevents further network partitions</li>
+     * </ul>
+     */
+    public void globalStabilizationTime() {
+        this.isGlobalStabilizationTime = true;
+
+        // re-queue all dropped messages
+        this.events.values().stream()
+                .filter(e -> e.getStatus() == Event.Status.DROPPED)
+                .forEach(e -> {
+                    e.setStatus(Event.Status.QUEUED);
+                    this.observers.forEach(o -> o.onEventRequeued(e));
+                });
+
+        // clear all network faults
+        // XXX: Is this the right thing to do?
+        this.networkFaults.clear();
+
+        // heal all partitions
+        this.router.resetPartitions();
+    }
 }

simulator/src/main/java/byzzbench/simulator/transport/TransportObserver.java

@@ -17,6 +17,13 @@ public interface TransportObserver {
      */
     void onEventDropped(Event event);
 
+    /**
+     * Called when the status of an event changes from {@link Event.Status#DROPPED} to {@link Event.Status#QUEUED}.
+     *
+     * @param event The event that was re-queued after being previously dropped.
+     */
+    void onEventRequeued(Event event);
+
     /**
      * Called when the status of an event changes to {@link Event.Status#DELIVERED}.
      *

simulator/src/main/resources/application.yml

@@ -22,20 +22,20 @@ byzzbench:
   outputSchedules: buggy # which schedules to write to file? one of 'all', 'buggy' or 'none'
 
   scheduler:
-    id: "byzzfuzz" # The ID of the scheduler to use
+    id: "random" # The ID of the scheduler to use
     executionMode: sync # async (default, any message delivered) or sync (communication-closure hypothesis, FIFO)
-    maxDropMessages: 0 # Maximum number of messages to drop per scenario
-    maxMutateMessages: 0 # Maximum number of messages to mutate per scenario
-    deliverTimeoutWeight: 1 # The weight for scheduler to trigger a timeout
+    maxDropMessages: 1 # Maximum number of messages to drop per scenario
+    maxMutateMessages: 1 # Maximum number of messages to mutate per scenario
+    deliverTimeoutWeight: 25 # The weight for scheduler to trigger a timeout
     deliverMessageWeight: 99 # The weight for scheduler to deliver a message
     deliverClientRequestWeight: 99 # The weight for scheduler to deliver a client request to a replica
-    dropMessageWeight: 100 # The weight for scheduler to drop a message
-    mutateMessageWeight: 100 # The weight for scheduler to mutate a message
+    dropMessageWeight: 0 # The weight for scheduler to drop a message
+    mutateMessageWeight: 50 # The weight for scheduler to mutate a message
     params: # additional parameters for the scheduler
       # ByzzFuzz
       numRoundsWithProcessFaults: 1
-      numRoundsWithNetworkFaults: 1
-      numRoundsWithFaults: 2
+      numRoundsWithNetworkFaults: 0
+      numRoundsWithFaults: 8
       # Twins
       numReplicas: 1
       numTwinsPerReplica: 2
@@ -47,6 +47,6 @@ byzzbench:
     scheduledFaults:
       factories: [ ]
     termination: # Success condition for the scenario - terminates it when all conditions are met
-      minEvents: 100 # check if N events have been scheduled. If so, can terminate
-      minRounds: 2 # check if N rounds have elapsed. If so, can terminate
+      minEvents: 400 # check if N events have been scheduled. If so, can terminate
+      minRounds: 8 # check if N rounds have elapsed. If so, can terminate
       samplingFrequency: 30 # check if should terminate every N scheduler decisions