Merge pull request #147 from joaomlneto/sample-f-on-startup

Sample set of faulty replicas on startup

Author: joaomlneto

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

@@ -166,6 +166,15 @@ public synchronized Node getNode(String nodeId) {
     @Override
     public final void setupScenario() {
         this.setup();
+
+        // 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

@@ -75,6 +75,8 @@ public class Transport {
      */
     @JsonIgnore
     private final List<TransportObserver> observers = new ArrayList<>();
+    @Getter
+    private boolean isGlobalStabilizationTime = false;
 
     /**
      * Adds an observer to the transport layer.
@@ -275,12 +277,18 @@ 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.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);
@@ -491,4 +499,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}.
      *