Merge pull request #154 from joaomlneto/fixed-fab-with-network-faults

Fixed fab with network faults

Author: joaomlneto

.idea/misc.xml

@@ -37,6 +37,7 @@ public abstract class BaseScenario implements Scenario {
     /**
      * The timekeeper for the scenario.
      */
+    @JsonIgnore
     protected final transient Timekeeper timekeeper;
     /**
      * The scheduler for the scenario.
@@ -236,6 +237,15 @@ public final void runScenario() {
      */
     @Override
     public final boolean invariantsHold() {
+        // Write to file which invariant is violated
+//        BufferedWriter writer = new BufferedWriter(new FileWriter("", true));
+//        for (ScenarioPredicate invariant : this.invariants) {
+//            if (!invariant.test(this)) {
+//                writer.write(this.id + " " + invariant.getClass().getSimpleName() + "\n");
+//                writer.close();
+//            }
+//        }
+
         return this.invariants.stream().allMatch(invariant -> invariant.test(this));
     }
 

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

@@ -14,6 +14,7 @@
 public abstract class BaseScenarioFactory implements ScenarioFactory {
     @Getter(AccessLevel.PROTECTED)
     private final SchedulerFactoryService schedulerFactoryService;
+    @Getter
     private final ByzzBenchConfig byzzBenchConfig;
     private final ObjectMapper mapper;
 

simulator/src/main/java/byzzbench/simulator/BaseScenarioFactory.java

@@ -28,23 +28,23 @@ public class Client implements Serializable, Node {
      */
     @JsonIgnore
     @NonNull
-    protected final transient Scenario scenario;
+    private final transient Scenario scenario;
 
     /**
      * The unique ID of the client.
      */
     @NonNull
-    protected final String id;
+    private final String id;
 
     /**
      * The sequence number of the next request to be sent by the client.
      */
-    protected final AtomicLong requestSequenceNumber = new AtomicLong(0);
+    private final AtomicLong requestSequenceNumber = new AtomicLong(0);
 
     /**
      * The maximum number of requests that can be sent by the client.
      */
-    protected final long maxRequests = 100;
+    private final long maxRequests = 1000;
 
     /**
      * The replies received by the client.
@@ -60,10 +60,17 @@ public void initialize() {
     }
 
     /**
-     * Sends a request to a replica in the system.
+     * Sends a request to any replica in the system.
      */
     public void sendRequest() {
         String recipientId = this.getScenario().getReplicas().keySet().iterator().next();
+        this.sendRequest(recipientId);
+    }
+
+    /**
+     * Sends a request to a given replica in the system.
+     */
+    public void sendRequest(String recipientId) {
         MessagePayload payload = new ClientRequestMessage(this.getCurrentTime().toEpochMilli(), recipientId);
         this.getScenario().getTransport().sendMessage(this, payload, recipientId);
     }

simulator/src/main/java/byzzbench/simulator/Client.java

@@ -71,39 +71,39 @@ public class HbftClient extends Client {
      */
     @Override
     public void sendRequest() {
-        String requestId = String.format("%s/%d", super.id, super.requestSequenceNumber.incrementAndGet());
+        String requestId = String.format("%s/%d", getId(), getRequestSequenceNumber().incrementAndGet());
         long timestamp = this.getCurrentTime().toEpochMilli();
-        RequestMessage request = new RequestMessage(requestId, timestamp, super.id);
-        this.sentRequests.put(super.requestSequenceNumber.get(), request);
+        RequestMessage request = new RequestMessage(requestId, timestamp, getId());
+        this.sentRequests.put(getRequestSequenceNumber().get(), request);
         this.sentRequestsByTimestamp.put(timestamp, requestId);
         this.broadcastRequest(timestamp, requestId);
 
         // Set timeout
         Long timeoutId = this.setTimeout("REQUEST", this::retransmitOrPanic, this.timeout);
-        timeouts.put(super.requestSequenceNumber.get(), timeoutId);
+        timeouts.put(getRequestSequenceNumber().get(), timeoutId);
     }
 
     public void retransmitOrPanic() {
-        long tolerance = (long) Math.floor((super.scenario.getTransport().getNodeIds().size() - 1) / 3);
+        long tolerance = (long) Math.floor((getScenario().getTransport().getNodeIds().size() - 1) / 3);
         if (this.shouldRetransmit(tolerance)) {
-            String requestId = String.format("%s/%d", super.id, super.requestSequenceNumber.get());
+            String requestId = String.format("%s/%d", getId(), getRequestSequenceNumber().get());
             // Based on hBFT 4.1 it uses the identical request
             // TODO: It probably should not be the same timestamp
-            long timestamp = this.sentRequests.get(super.requestSequenceNumber.get()).getTimestamp();
+            long timestamp = this.sentRequests.get(getRequestSequenceNumber().get()).getTimestamp();
             this.broadcastRequest(timestamp, requestId);
         } else if (this.shouldPanic(tolerance)) {
-            RequestMessage message = this.sentRequests.get(super.requestSequenceNumber.get());
-            PanicMessage panic = new PanicMessage(this.digest(message), this.getCurrentTime().toEpochMilli(), super.id);
-            super.scenario.getTransport().multicast(this, super.scenario.getTransport().getNodeIds(), panic);
+            RequestMessage message = this.sentRequests.get(getRequestSequenceNumber().get());
+            PanicMessage panic = new PanicMessage(this.digest(message), this.getCurrentTime().toEpochMilli(), getId());
+            getScenario().getTransport().multicast(this, getScenario().getTransport().getNodeIds(), panic);
         }
-        this.clearTimeout(timeouts.get(super.requestSequenceNumber.get()));
+        this.clearTimeout(timeouts.get(getRequestSequenceNumber().get()));
         Long timeoutId = this.setTimeout("REQUEST", this::retransmitOrPanic, this.timeout);
-        timeouts.put(super.requestSequenceNumber.get(), timeoutId);
+        timeouts.put(getRequestSequenceNumber().get(), timeoutId);
     }
 
     private void broadcastRequest(long timestamp, String requestId) {
         MessagePayload payload = new ClientRequestMessage(timestamp, requestId);
-        SortedSet<String> replicaIds = super.scenario.getTransport().getNodeIds();
+        SortedSet<String> replicaIds = getScenario().getTransport().getNodeIds();
         getScenario().getTransport().multicast(this, replicaIds, payload);
     }
 
@@ -131,9 +131,9 @@ public void handleMessage(String senderId, MessagePayload payload) {
          */
         if (this.completedReplies(clientReplyMessage.getTolerance())
                 && !this.completedRequests.contains(key)
-                && super.requestSequenceNumber.get() <= this.maxRequests) {
+                && getRequestSequenceNumber().get() <= getMaxRequests()) {
             this.completedRequests.add(key);
-            this.clearTimeout(this.timeouts.get(super.requestSequenceNumber.get()));
+            this.clearTimeout(this.timeouts.get(getRequestSequenceNumber().get()));
             this.sendRequest();
         }
     }
@@ -148,15 +148,15 @@ public void handleMessage(String senderId, MessagePayload payload) {
      */
     public long setTimeout(String name, Runnable r, long timeout) {
         Duration duration = Duration.ofSeconds(timeout);
-        return super.scenario.getTransport().setTimeout(this, r, duration, name);
+        return getScenario().getTransport().setTimeout(this, r, duration, name);
     }
 
     /**
      * Checks whether client should retransmit the request
      * if #replies < f + 1
      */
     public boolean shouldRetransmit(long tolerance) {
-        String currRequest = String.format("%s/%d", super.id, super.requestSequenceNumber.get());
+        String currRequest = String.format("%s/%d", getId(), getRequestSequenceNumber().get());
         if (!hbftreplies.containsKey(currRequest)) {
             return true;
         }
@@ -171,7 +171,7 @@ public boolean shouldRetransmit(long tolerance) {
      * if f + 1 <= #replies < 2f + 1
      */
     public boolean shouldPanic(long tolerance) {
-        String currRequest = String.format("%s/%d", super.id, super.requestSequenceNumber.get());
+        String currRequest = String.format("%s/%d", getId(), getRequestSequenceNumber().get());
         for (ClientReplyKey key : hbftreplies.get(currRequest).keySet()) {
             return this.hbftreplies.get(currRequest).get(key).size() >= tolerance + 1
                     && this.hbftreplies.get(currRequest).get(key).size() < tolerance * 2 + 1;
@@ -183,7 +183,7 @@ public boolean shouldPanic(long tolerance) {
      * Checks whether it has received 2f + 1 replies
      */
     public boolean completedReplies(long tolerance) {
-        String currRequest = String.format("%s/%d", super.id, super.requestSequenceNumber.get());
+        String currRequest = String.format("%s/%d", getId(), getRequestSequenceNumber().get());
         if (!hbftreplies.containsKey(currRequest)) {
             return false;
         }
@@ -199,7 +199,7 @@ public boolean completedReplies(long tolerance) {
      * Clear all timeouts for this client.
      */
     // public void clearAllTimeouts() {
-    //     super.scenario.getTransport().clearClientTimeouts(super.id);
+    //     getScenario.getTransport().clearClientTimeouts(getId());
     // }
 
     /**

simulator/src/main/java/byzzbench/simulator/HbftClient.java

@@ -171,7 +171,7 @@ public final class BehaviorConfig {
     @Data
     public final class ScenarioConfig {
         private TerminationConfig termination = new TerminationConfig();
-        private String id = "hbft";
+        private String id = "pbft-java";
         private Map<String, String> params = new HashMap<>();
     }
 }

simulator/src/main/java/byzzbench/simulator/config/ByzzBenchConfig.java

@@ -100,8 +100,9 @@ public void accept(FaultContext context) {
         if (router.haveConnectivity(sender, recipient)) {
             return;
         }
+
+        // otherwise, drop the message: the sender and recipient are in different partitions
         if(e.getStatus() == Event.Status.QUEUED) {
-            // otherwise, drop the message: the sender and recipient are in different partitions
             context.getScenario().getTransport().dropEvent(e.getEventId());
         }
     }

simulator/src/main/java/byzzbench/simulator/faults/behaviors/ByzzFuzzDropMessageBehavior.java

@@ -67,7 +67,7 @@ public void accept(FaultContext context) {
         // apply the random mutator
         MessageMutationFault mutator = mutators.get(rand.nextInt(mutators.size()));
 
-        // apply the mutation (if the message is still queued)
+        // apply the mutation if the message is queued
         if (e.getStatus() == Event.Status.QUEUED) {
             context.getScenario().getTransport().applyMutation(e.getEventId(), mutator);
         }

simulator/src/main/java/byzzbench/simulator/faults/behaviors/MutateMessageBehavior.java

@@ -8,6 +8,7 @@
 import byzzbench.simulator.faults.faults.ByzzFuzzProcessFault;
 import byzzbench.simulator.scheduler.ByzzFuzzScheduler;
 import byzzbench.simulator.utils.SetSubsets;
+import lombok.extern.java.Log;
 import org.springframework.stereotype.Component;
 
 import java.util.ArrayList;
@@ -18,6 +19,7 @@
 /**
  * Fault factory that generates faults for a given scenario.
  */
+@Log
 @Component
 public class ByzzFuzzScenarioFaultFactory implements FaultFactory {
     private final Random rand = new Random();
@@ -59,7 +61,6 @@ public List<Fault> generateFaults(FaultContext input) {
         }
 
         // Faults
-        System.out.println("ByzzFuzzFaults:");
         faults.forEach(fault -> System.out.println(fault.getId()));
 
         return faults;

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

@@ -0,0 +1,120 @@
+package byzzbench.simulator.protocols.fab;
+
+import byzzbench.simulator.Client;
+import byzzbench.simulator.Scenario;
+import byzzbench.simulator.protocols.pbft.message.RequestMessage;
+import byzzbench.simulator.transport.MessagePayload;
+import lombok.extern.java.Log;
+
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicLong;
+
+@Log
+//public class FastByzantineClient extends Client {
+//    private final AtomicLong learners = new AtomicLong(0);
+//    private final AtomicBoolean isFinished = new AtomicBoolean(false);
+//    private final List<String> learnersList = new ArrayList<>();
+//    private final List<String> proposersList = new ArrayList<>();
+//    /**
+//     * The replies received by the client.
+//     */
+//    private final List<Serializable> replies = new ArrayList<>();
+//    /**
+//     * The sequence number of the next request to be sent by the client.
+//     */
+//    private final AtomicLong requestSequenceNumber = new AtomicLong(0);
+//    /**
+//     * The maximum number of requests that can be sent by the client.
+//     */
+//    private final long maxRequests = 1000;
+//
+//    public FastByzantineClient(Scenario scenario, String id, List<String> learnersList, List<String> proposersList) {
+//        super(scenario, id);
+//        this.learnersList.addAll(learnersList);
+//        this.proposersList.addAll(proposersList);
+//    }
+//
+//    @Override
+//    public void sendRequest(String senderId) {
+//        long sequenceNumber = this.getRequestSequenceNumber().getAndIncrement();
+//        String command = String.format("%s/%d", this.getId(), sequenceNumber);
+//        // TODO: compute the digest
+//        RequestMessage request = new RequestMessage(this.getId(), sequenceNumber, "-1", command);
+//        this.getScenario().getTransport().sendClientRequest(this.getId(), request, senderId);
+//    }
+//
+//    @Override
+//    public void handleMessage(String senderId, MessagePayload reply) {
+//        log.info("Client received a message from " + senderId);
+//
+//        this.replies.add(reply);
+//        if (learnersList.contains(senderId)) {
+//            learners.incrementAndGet();
+//            if (learners.get() == 4) {
+//                for (String proposer : proposersList) {
+//                    this.sendRequest(proposer);
+//                }
+//
+//                learners.set(0);
+//            }
+//        } else {
+//            log.info("Client received a message from a non-learner node");
+//        }
+//    }
+//}
+
+public class FastByzantineClient extends Client {
+    private final AtomicLong learners = new AtomicLong(0);
+    private final AtomicBoolean isFinished = new AtomicBoolean(false);
+    private final List<String> learnersList = new ArrayList<>();
+    private final List<String> proposersList = new ArrayList<>();
+    /**
+     * The replies received by the client.
+     */
+    private final List<Serializable> replies = new ArrayList<>();
+    /**
+     * The sequence number of the next request to be sent by the client.
+     */
+    private final AtomicLong requestSequenceNumber = new AtomicLong(0);
+    /**
+     * The maximum number of requests that can be sent by the client.
+     */
+    private final long maxRequests = 1000;
+
+    public FastByzantineClient(Scenario scenario, String id, List<String> learnersList, List<String> proposersList) {
+        super(scenario, id);
+        this.learnersList.addAll(learnersList);
+        this.proposersList.addAll(proposersList);
+    }
+
+    @Override
+    public void sendRequest(String senderId) {
+        long sequenceNumber = this.getRequestSequenceNumber().getAndIncrement();
+        String command = String.format("%s/%d", this.getId(), sequenceNumber);
+        // TODO: compute the digest
+        RequestMessage request = new RequestMessage(this.getId(), sequenceNumber, "-1", command);
+        this.getScenario().getTransport().sendMessage(this, request, senderId);
+    }
+
+    @Override
+    public void handleMessage(String senderId, MessagePayload reply) {
+        log.info("Client received a message from " + senderId);
+
+        this.replies.add(reply);
+        if (learnersList.contains(senderId)) {
+            learners.incrementAndGet();
+            if (learners.get() == 4) {
+                for (String proposer : proposersList) {
+                    this.sendRequest(proposer);
+                }
+
+                learners.set(0);
+            }
+        } else {
+            log.info("Client received a message from a non-learner node");
+        }
+    }
+}