add GST (logic still missing from simulatorservice)

Author: joaomlneto

simulator/src/main/java/byzzbench/simulator/Replica.java

@@ -161,15 +161,6 @@ public void initialize() {
         // do nothing by default
     }
 
-    /**
-     * Handle a request received from a client.
-     *
-     * @param clientId the ID of the client
-     * @param request  the request payload
-     * @throws Exception if an error occurs while handling the request
-     */
-    public abstract void handleClientRequest(String clientId, Serializable request) throws Exception;
-
     /**
      * Send a reply to a client.
      *

simulator/src/main/java/byzzbench/simulator/faults/faults/ByzzFuzzNetworkFault.java

@@ -2,6 +2,8 @@
 
 import byzzbench.simulator.faults.BaseFault;
 import byzzbench.simulator.faults.behaviors.CreateNetworkPartitionsBehavior;
+import byzzbench.simulator.faults.predicates.ANDPredicate;
+import byzzbench.simulator.faults.predicates.IsBeforeGST;
 import byzzbench.simulator.faults.predicates.MessageRoundPredicate;
 
 import java.util.Set;
@@ -22,7 +24,7 @@ public class ByzzFuzzNetworkFault extends BaseFault {
     public ByzzFuzzNetworkFault(Set<String> partition, int round) {
         super(
                 "byzzfuzznetworkfault-%d-%s".formatted(round, String.join("-", partition)),
-                new MessageRoundPredicate(round),
+                new ANDPredicate(new IsBeforeGST(), new MessageRoundPredicate(round)),
                 new CreateNetworkPartitionsBehavior(partition)
         );
     }

simulator/src/main/java/byzzbench/simulator/faults/predicates/ANDPredicate.java

@@ -0,0 +1,33 @@
+package byzzbench.simulator.faults.predicates;
+
+import byzzbench.simulator.faults.FaultContext;
+import byzzbench.simulator.faults.FaultPredicate;
+
+import java.util.List;
+
+public class ANDPredicate implements FaultPredicate {
+    private final List<FaultPredicate> predicates;
+
+    public ANDPredicate(FaultPredicate... predicates) {
+        this.predicates = List.of(predicates);
+    }
+
+    @Override
+    public String getId() {
+        return "AND(%s)".formatted(predicates.stream()
+                .map(FaultPredicate::getId)
+                .reduce("", (a, b) -> a + "," + b));
+    }
+
+    @Override
+    public String getName() {
+        return predicates.stream()
+                .map(FaultPredicate::getName)
+                .reduce("", (a, b) -> a + " AND " + b);
+    }
+
+    @Override
+    public boolean test(FaultContext faultContext) {
+        return predicates.stream().allMatch(p -> p.test(faultContext));
+    }
+}

simulator/src/main/java/byzzbench/simulator/faults/predicates/IsBeforeGST.java

@@ -0,0 +1,21 @@
+package byzzbench.simulator.faults.predicates;
+
+import byzzbench.simulator.faults.FaultContext;
+import byzzbench.simulator.faults.FaultPredicate;
+
+public class IsBeforeGST implements FaultPredicate {
+    @Override
+    public String getId() {
+        return "IsBeforeGST";
+    }
+
+    @Override
+    public String getName() {
+        return "Is before GST";
+    }
+
+    @Override
+    public boolean test(FaultContext faultContext) {
+        return !faultContext.getScenario().getTransport().isGlobalStabilizationTime();
+    }
+}

simulator/src/main/java/byzzbench/simulator/protocols/XRPL/XRPLReplica.java

@@ -7,10 +7,10 @@
 import byzzbench.simulator.protocols.XRPL.messages.XRPLTxMessage;
 import byzzbench.simulator.protocols.XRPL.messages.XRPLValidateMessage;
 import byzzbench.simulator.state.TotalOrderCommitLog;
+import byzzbench.simulator.transport.DefaultClientRequestPayload;
 import byzzbench.simulator.transport.MessagePayload;
 import lombok.Getter;
 
-import java.io.Serializable;
 import java.time.Duration;
 import java.util.*;
 import java.util.Map.Entry;
@@ -55,7 +55,9 @@ protected XRPLReplica(String nodeId, Scenario scenario, List<String> UNL, XRPLLe
 
     @Override
     public void handleMessage(String sender, MessagePayload message) throws Exception {
-        if (message instanceof XRPLProposeMessage propmsg) {
+        if (message instanceof DefaultClientRequestPayload req) {
+            recvTxHandler(new XRPLTxMessage(req.getOperation().toString(), sender));
+        } else if (message instanceof XRPLProposeMessage propmsg) {
             proposeMessageHandler(propmsg);
         } else if (message instanceof XRPLSubmitMessage submsg) {
             submitMessageHandler(submsg);
@@ -511,11 +513,4 @@ private void playbackProposals() {
             }
         }
     }
-
-    @Override
-    public void handleClientRequest(String clientId, Serializable request) throws Exception {
-        String tx = request.toString();
-        XRPLTxMessage txmsg = new XRPLTxMessage(tx, clientId);
-        this.handleMessage(clientId, txmsg);
-    }
 }

simulator/src/main/java/byzzbench/simulator/protocols/dummy/DummyReplica.java

@@ -9,8 +9,6 @@
 import lombok.ToString;
 import lombok.extern.java.Log;
 
-import java.io.Serializable;
-
 @Log
 @ToString(callSuper = true)
 public class DummyReplica extends Replica {
@@ -23,11 +21,6 @@ public void initialize() {
         // nothing to do
     }
 
-    @Override
-    public void handleClientRequest(String clientId, Serializable request) {
-        throw new UnsupportedOperationException("Unsupported operation: handleClientRequest");
-    }
-
     @Override
     public void handleMessage(String sender, MessagePayload m) {
         if (m instanceof DefaultClientRequestPayload clientRequestMessage) {

simulator/src/main/java/byzzbench/simulator/protocols/fasthotstuff/FastHotStuffReplica.java

@@ -8,7 +8,6 @@
 import lombok.Getter;
 import lombok.extern.java.Log;
 
-import java.io.Serializable;
 import java.time.Duration;
 import java.util.*;
 import java.util.concurrent.atomic.AtomicLong;
@@ -50,11 +49,6 @@ public void initialize() {
         this.resetTimeout();
     }
 
-    @Override
-    public void handleClientRequest(String clientId, Serializable request) throws Exception {
-        throw new UnsupportedOperationException("Client requests not supported in FastHotStuff");
-    }
-
     private void createGenesisBlocks() {
         List<String> nodeIds = this.getNodeIds().stream().sorted().toList();
 

simulator/src/main/java/byzzbench/simulator/protocols/hbft/HbftReplica.java

@@ -8,8 +8,6 @@
 import lombok.ToString;
 import lombok.extern.java.Log;
 
-import java.io.Serializable;
-
 /**
  * A Replica in the hBFT protocol.
  * <p>
@@ -42,11 +40,6 @@ public void initialize() {
         // TODO
     }
 
-    @Override
-    public void handleClientRequest(String clientId, Serializable request) throws Exception {
-        // TODO
-    }
-
     @Override
     public void handleMessage(String sender, MessagePayload m) {
         // TODO

simulator/src/main/java/byzzbench/simulator/protocols/pbft/PbftReplica.java

@@ -10,7 +10,6 @@
 import lombok.Getter;
 import lombok.extern.java.Log;
 
-import java.io.Serializable;
 import java.time.Duration;
 import java.time.Instant;
 import java.util.*;
@@ -440,15 +439,6 @@ public String primary(long vi) {
         return (vi == v) ? this.cur_primary : replicaIds[(int) (vi % replicaIds.length)];
     }
 
-    @Override
-    public void handleClientRequest(String clientId, Serializable request) throws Exception {
-        if (request instanceof RequestMessage) {
-            this.handleMessage(clientId, (RequestMessage) request);
-        } else {
-            throw new UnsupportedOperationException("Unknown request type: " + request.getClass().getName());
-        }
-    }
-
     @Override
     public void handleMessage(String sender, MessagePayload message) throws Exception {
         // for some reason the original code does this before receiving a message.

simulator/src/main/java/byzzbench/simulator/protocols/pbft_java/PbftJavaReplica.java

@@ -720,7 +720,6 @@ public Serializable compute(long sequenceNumber, LogEntry operation) {
         return operation;
     }
 
-    @Override
     public void handleClientRequest(String clientId, Serializable request) {
         // FIXME: should not get timestamp from system time
         RequestMessage m = new RequestMessage(request, System.currentTimeMillis(), clientId);

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

@@ -47,7 +47,7 @@ public synchronized Optional<EventDecision> scheduleNext(Scenario scenario) thro
 
         // if there are no events, return empty
         if (availableEvents.isEmpty()) {
-            log.warning("No actions available!");
+            log.warning("No queued events (messages, timeouts)!");
             return Optional.empty();
         }
 

simulator/src/main/java/byzzbench/simulator/scheduler/twins/TwinsReplica.java

@@ -8,7 +8,6 @@
 import lombok.Getter;
 import lombok.extern.java.Log;
 
-import java.io.Serializable;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Random;
@@ -129,11 +128,6 @@ private Replica getInternalReplicaFromId(String id) {
         return replicas.get(index);
     }
 
-    @Override
-    public void handleClientRequest(String clientId, Serializable request) {
-        throw new UnsupportedOperationException("Not implemented");
-    }
-
     /**
      * Get the internal replicas that should handle the given message.
      *

simulator/src/main/java/byzzbench/simulator/service/SimulatorService.java

@@ -2,6 +2,7 @@
 
 import byzzbench.simulator.Scenario;
 import byzzbench.simulator.config.ByzzBenchConfig;
+import byzzbench.simulator.scheduler.EventDecision;
 import byzzbench.simulator.state.ErroredPredicate;
 import byzzbench.simulator.transport.Event;
 import byzzbench.simulator.transport.messages.MessageWithRound;
@@ -19,6 +20,7 @@
 import java.nio.file.Files;
 import java.nio.file.Path;
 import java.time.Instant;
+import java.util.Optional;
 import java.util.OptionalLong;
 import java.util.Set;
 import java.util.concurrent.ExecutorService;
@@ -128,7 +130,30 @@ public void start() {
 
                     try {
                         while (true) {
-                            this.scenario.getScheduler().scheduleNext(this.scenario);
+                            Optional<EventDecision> decision = this.scenario.getScheduler().scheduleNext(this.scenario);
+                            System.out.println("Decision: " + decision);
+
+                            // if the scheduler did not make a decision, and we're before GST, set GST!
+                            if (decision.isEmpty() && !this.scenario.getTransport().isGlobalStabilizationTime()) {
+                                this.scenario.getTransport().globalStabilizationTime();
+                                continue;
+                            }
+
+                            if (decision.isEmpty()) {
+                                System.out.println("We're after GST and still no events!!");
+                                // print num of events in the scenario
+                                System.out.println("Number of events in the scenario: " + this.scenario.getSchedule().getEvents().size());
+                                // print number of delivered events
+                                System.out.println("Number of delivered events: " + this.scenario.getTransport().getEventsInState(Event.Status.DELIVERED).size());
+                                // print ALL events in the scenario, independent of their status
+                                System.out.println("All events in the scenario:");
+                                this.scenario.getTransport().getEvents().values().forEach(System.out::println);
+                                // print the event IDs in the schedule
+                                System.out.println("Event IDs in the schedule:");
+                                this.scenario.getSchedule().getEvents().forEach(e -> System.out.println(e.getEventId()));
+                                this.shouldStop = true;
+                                break;
+                            }
 
                             long numEvents = this.scenario.getSchedule().getEvents().size();
                             long terminationSamplingFreq = this.byzzBenchConfig.getScenario().getTermination().getSamplingFrequency();

simulator/src/main/java/byzzbench/simulator/state/LivenessPredicate.java

@@ -15,10 +15,15 @@ public String getId() {
 
     @Override
     public boolean test(Scenario scenarioExecutor) {
-        boolean hasNoQueuedEvents = scenarioExecutor.getTransport().getEventsInState(Event.Status.QUEUED).isEmpty();
-        if (hasNoQueuedEvents) {
+        // If we are before GST, the scenario is considered live
+        if (!scenarioExecutor.getTransport().isGlobalStabilizationTime()) {
+            return true;
+        }
+
+        boolean hasQueuedEvents = !(scenarioExecutor.getTransport().getEventsInState(Event.Status.QUEUED).isEmpty());
+        if (!hasQueuedEvents) {
             System.out.println("LivenessPredicate: No events in the QUEUED state");
         }
-        return !hasNoQueuedEvents;
+        return hasQueuedEvents;
     }
 }

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

@@ -37,9 +37,10 @@ public abstract class BaseEvent implements Event {
     private Status status = Status.QUEUED;
 
     public void setStatus(Status status) {
-        if (this.status != Status.QUEUED) {
-            throw new IllegalStateException("Can only change the status of a queued event");
-        }
+        // XXX during GST, dropped messages are re-queued
+        //if (this.status != Status.QUEUED) {
+        //    throw new IllegalStateException("Can only change the status of a queued event");
+        //}
         this.status = status;
     }
 }

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

@@ -3,13 +3,15 @@
 import byzzbench.simulator.utils.NonNull;
 import lombok.Data;
 import lombok.EqualsAndHashCode;
+import lombok.ToString;
 import lombok.experimental.SuperBuilder;
 
 import java.io.Serializable;
 
 @EqualsAndHashCode(callSuper = true)
 @Data
 @SuperBuilder
+@ToString(callSuper = true)
 public abstract class BaseMessageEvent<T extends Serializable> extends BaseEvent implements MailboxEvent {
     /**
      * The unique identifier of the receiving node

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

@@ -3,6 +3,7 @@
 import com.fasterxml.jackson.annotation.JsonTypeName;
 import lombok.Data;
 import lombok.EqualsAndHashCode;
+import lombok.ToString;
 import lombok.experimental.SuperBuilder;
 
 /**
@@ -14,5 +15,6 @@
 @Data
 @JsonTypeName("ClientRequest")
 @SuperBuilder
+@ToString(callSuper = true)
 public class ClientRequestEvent extends BaseMessageEvent<MessagePayload> {
 }

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

@@ -22,19 +22,22 @@
 public interface Event extends Serializable {
     /**
      * Get the eventId
+     *
      * @return a long representing the eventId
      */
     @NonNull
     long getEventId();
 
     /**
      * Get the time at which the event was created
+     *
      * @return an Instant representing the time at which the event was created
      */
     Instant getCreatedAt();
 
     /**
      * Get the time at which the event was delivered
+     *
      * @return an Instant representing the time at which the event was delivered
      */
     Instant getDeliveredAt();

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

@@ -4,6 +4,7 @@
 import com.fasterxml.jackson.annotation.JsonTypeName;
 import lombok.Data;
 import lombok.EqualsAndHashCode;
+import lombok.ToString;
 import lombok.experimental.SuperBuilder;
 
 /**
@@ -15,6 +16,7 @@
 @Data
 @JsonTypeName("GenericFault")
 @SuperBuilder
+@ToString(callSuper = true)
 public class GenericFaultEvent extends BaseEvent {
     /**
      * The payload of the request.