major refactor - differentiate events from actions, WIP to properly persist schedules

Author: joaomlneto

.idea/dataSources.xml

@@ -43,11 +43,6 @@ public class Client implements Serializable, Node {
      */
     private final AtomicLong requestSequenceNumber = new AtomicLong(0);
 
-    /**
-     * The maximum number of requests that can be sent by the client.
-     */
-    private final long maxRequests = 1000;
-
     /**
      * The replies received by the client.
      */
@@ -85,9 +80,7 @@ public void sendRequest(String recipientId) {
      */
     public void handleMessage(String senderId, MessagePayload reply) {
         this.replies.add(reply);
-        if (this.requestSequenceNumber.get() < this.maxRequests) {
-            this.sendRequest();
-        }
+        this.sendRequest();
     }
 
     @Override

.idea/modules.xml

@@ -1,7 +1,7 @@
 package byzzbench.simulator;
 
-import byzzbench.simulator.transport.Action;
-import byzzbench.simulator.transport.MessageAction;
+import byzzbench.simulator.transport.Event;
+import byzzbench.simulator.transport.MessageEvent;
 import byzzbench.simulator.transport.messages.MessageWithRound;
 
 import java.util.List;
@@ -27,10 +27,10 @@ public boolean test(Scenario scenario) {
 
         // Check if the maximum number of rounds has been reached
         if (maxRounds > 0) {
-            List<MessageWithRound> queuedMessages = scenario.getTransport().getEventsInState(Action.Status.DELIVERED)
+            List<MessageWithRound> queuedMessages = scenario.getTransport().getEventsInState(Event.Status.DELIVERED)
                     .stream()
-                    .filter(e -> e instanceof MessageAction msgEvent)
-                    .map(e -> (MessageAction) e)
+                    .filter(e -> e instanceof MessageEvent msgEvent)
+                    .map(e -> (MessageEvent) e)
                     .filter(e -> e.getPayload() instanceof MessageWithRound)
                     .map(e -> (MessageWithRound) e.getPayload())
                     .filter(msg -> msg.getRound() > maxRounds)

.idea/modules/simulator/byzzbench.simulator.aot.iml

@@ -1,6 +1,6 @@
 package byzzbench.simulator;
 
-import byzzbench.simulator.domain.Schedule;
+import byzzbench.simulator.domain.*;
 import byzzbench.simulator.faults.Fault;
 import byzzbench.simulator.faults.ScenarioContext;
 import byzzbench.simulator.faults.factories.ByzzFuzzScenarioFaultFactory;
@@ -11,6 +11,9 @@
 import byzzbench.simulator.state.AgreementPredicate;
 import byzzbench.simulator.state.LivenessPredicate;
 import byzzbench.simulator.state.adob.AdobDistributedState;
+import byzzbench.simulator.transport.Event;
+import byzzbench.simulator.transport.MessageEvent;
+import byzzbench.simulator.transport.TimeoutEvent;
 import byzzbench.simulator.transport.Transport;
 import com.fasterxml.jackson.annotation.JsonIgnore;
 import com.fasterxml.jackson.databind.JsonNode;
@@ -24,6 +27,7 @@
 import java.io.Serializable;
 import java.util.*;
 import java.util.stream.Collectors;
+import java.util.stream.Stream;
 
 /**
  * Represents a runnable scenario.
@@ -251,6 +255,98 @@ public final void loadParameters(JsonNode parameters) {
         this.loadScenarioParameters(parameters);
     }
 
+    /**
+     * Get the currently queued events of a specific type in the scenario.
+     *
+     * @param eventClass The class of the event to get.
+     * @param <T>        The type of the event.
+     * @return The list of events.
+     */
+    private <T extends Event> Stream<T> getQueuedEventsOfType(Class<T> eventClass) {
+        return getTransport().getEventsInState(Event.Status.QUEUED)
+                .stream()
+                .filter(eventClass::isInstance)
+                .map(eventClass::cast);
+    }
+
+    /**
+     * Get the available {@link DeliverMessageAction} in the scenario in the current state.
+     *
+     * @return The list of available {@link DeliverMessageAction}.
+     */
+    private List<DeliverMessageAction> getAvailableDeliverMessageAction() {
+        switch (scheduler.getConfig().getScheduler().getExecutionMode()) {
+            // return the first queued message in each mailbox
+            case SYNC -> {
+                Set<String> recipientIdsSeen = new HashSet<>();
+                return getQueuedEventsOfType(MessageEvent.class)
+                        .filter(event -> recipientIdsSeen.add(event.getRecipientId()))
+                        .map(DeliverMessageAction::fromEvent)
+                        .toList();
+            }
+            // return all queued message events in all mailboxes
+            case ASYNC -> {
+                return getQueuedEventsOfType(MessageEvent.class)
+                        .map(DeliverMessageAction::fromEvent)
+                        .toList();
+            }
+            default ->
+                    throw new IllegalStateException("Unknown execution mode: " + scheduler.getConfig().getScheduler().getExecutionMode());
+        }
+    }
+
+    /**
+     * Get the available {@link TriggerTimeoutAction} in the scenario in the current state.
+     *
+     * @return The list of available {@link TriggerTimeoutAction}.
+     */
+    private List<TriggerTimeoutAction> getAvailableTriggerTimeoutAction() {
+        // get all time out events in order of expiration (earliest first)
+        Set<String> recipientIdsSeen = new HashSet<>();
+        Stream<TimeoutEvent> firstTimeoutForEachReplica = getQueuedEventsOfType(TimeoutEvent.class)
+                .sorted(Comparator.comparing(TimeoutEvent::getExpiresAt))
+                .filter(event -> recipientIdsSeen.add(event.getRecipientId()));
+
+        switch (this.getScheduler().getConfig().getScheduler().getExecutionMode()) {
+            // return the first timeout for each replica without a message in their mailbox
+            case SYNC -> {
+                // return only timeouts for replicas without messages in their mailbox
+                return firstTimeoutForEachReplica
+                        .filter(event -> !recipientIdsSeen.contains(event.getRecipientId()))
+                        .map(TriggerTimeoutAction::fromEvent)
+                        .toList();
+            }
+            // return the first timeout for each replica
+            case ASYNC -> {
+                return firstTimeoutForEachReplica
+                        .map(TriggerTimeoutAction::fromEvent)
+                        .toList();
+            }
+            default ->
+                    throw new IllegalStateException("Unknown execution mode: " + this.getScheduler().getConfig().getScheduler().getExecutionMode());
+        }
+    }
+
+    private List<FaultInjectionAction> getAvailableFaultInjectionAction() {
+        throw new UnsupportedOperationException("Not implemented!");
+    }
+
+    /**
+     * Get the available actions in the scenario in the current state.
+     *
+     * @return The list of available actions.
+     */
+    public List<? extends Action> getAvailableActions() {
+        return Stream.of(
+                        getAvailableDeliverMessageAction().stream(),
+                        getAvailableTriggerTimeoutAction().stream())
+                .reduce(Stream::concat)
+                .orElseGet(Stream::empty)
+                .toList();
+
+        // TODO faults
+    }
+
     /**
      * Loads the parameters for the scenario from a JSON object.
      *

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

@@ -1,6 +1,8 @@
 package byzzbench.simulator;
 
+import byzzbench.simulator.config.ByzzBenchConfig;
 import lombok.Getter;
+import lombok.RequiredArgsConstructor;
 import org.springframework.boot.SpringApplication;
 import org.springframework.boot.autoconfigure.SpringBootApplication;
 import org.springframework.boot.context.event.ApplicationReadyEvent;
@@ -9,9 +11,11 @@
 import java.time.Instant;
 
 @SpringBootApplication
+@RequiredArgsConstructor
 public class SimulatorApplication {
     @Getter
     private static Instant startTime;
+    private final ByzzBenchConfig config;
 
     public static void main(String[] args) {
         SpringApplication.run(SimulatorApplication.class, args);
@@ -23,5 +27,8 @@ public static void main(String[] args) {
     @EventListener(ApplicationReadyEvent.class)
     public void onStartup() {
         startTime = Instant.now();
+
+        // print config
+        System.out.println("Configuration: " + config);
     }
 }

simulator/src/main/java/byzzbench/simulator/GenericTerminationPredicate.java

@@ -1,9 +1,9 @@
 package byzzbench.simulator;
 
 import byzzbench.simulator.faults.Fault;
-import byzzbench.simulator.transport.Action;
+import byzzbench.simulator.transport.Event;
 import byzzbench.simulator.transport.MutateMessageEventPayload;
-import byzzbench.simulator.transport.TimeoutAction;
+import byzzbench.simulator.transport.TimeoutEvent;
 import byzzbench.simulator.transport.TransportObserver;
 import com.fasterxml.jackson.annotation.JsonIgnore;
 
@@ -64,24 +64,24 @@ public Instant getTime(Node node) {
     }
 
     @Override
-    public void onEventAdded(Action Action) {
+    public void onEventAdded(Event Event) {
         // nothing to do
     }
 
     @Override
-    public void onEventDropped(Action Action) {
+    public void onEventDropped(Event Event) {
         // nothing to do
     }
 
     @Override
-    public void onEventRequeued(Action Action) {
+    public void onEventRequeued(Event Event) {
         // nothing to do
     }
 
     @Override
-    public void onEventDelivered(Action Action) {
+    public void onEventDelivered(Event Event) {
         // check if it was a timeout
-        if (!(Action instanceof TimeoutAction timeoutEvent)) {
+        if (!(Event instanceof TimeoutEvent timeoutEvent)) {
             return;
         }
 
@@ -103,7 +103,7 @@ public void onFault(Fault fault) {
     }
 
     @Override
-    public void onTimeout(TimeoutAction event) {
+    public void onTimeout(TimeoutEvent event) {
         // nothing to do
     }
 }

simulator/src/main/java/byzzbench/simulator/Scenario.java

@@ -1,12 +1,10 @@
 package byzzbench.simulator.config;
 
-import byzzbench.simulator.SimulatorApplication;
 import lombok.Data;
 import org.springframework.boot.context.properties.ConfigurationProperties;
 import org.springframework.stereotype.Component;
 import org.springframework.validation.annotation.Validated;
 
-import java.nio.file.Path;
 import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.List;
@@ -20,21 +18,20 @@
 @Data
 @Validated
 public class ByzzBenchConfig {
+
+
     /**
      * Whether to start generating scenarios on startup automatically.
      */
     private boolean autostart = false;
-
     /**
      * The number of scenarios to run at a time. Defaults to Integer.MAX_VALUE.
      */
     private int numScenarios = Integer.MAX_VALUE;
-
     /**
-     * The path to the output directory. Defaults to "./output".
+     * Policy for saving schedules in the database.
      */
-    private Path outputPath = Path.of("output");
-
+    private SaveScheduleMode saveSchedules = SaveScheduleMode.ALL;
     /**
      * Scheduler parameters.
      */
@@ -44,13 +41,20 @@ public class ByzzBenchConfig {
      */
     private ScenarioConfig scenario = new ScenarioConfig();
 
-    /**
-     * Get the output path for this run, in the format "output/{start-time}".
-     *
-     * @return The output path for this run.
-     */
-    public Path getOutputPathForThisRun() {
-        return this.outputPath.resolve(String.valueOf(SimulatorApplication.getStartTime().getEpochSecond()));
+    // execution should be either "async" or "sync". Here is the enum:
+    public enum SaveScheduleMode {
+        /**
+         * Save all schedules in the database.
+         */
+        ALL,
+        /**
+         * Save only schedules that did not terminate successfully.
+         */
+        BUGGY,
+        /**
+         * Do not save any schedules.
+         */
+        NONE,
     }
 
     /**
@@ -103,10 +107,6 @@ public final class SchedulerConfig {
          * Weighted probability of delivering a message
          */
         private int deliverMessageWeight = 99;
-        /**
-         * Weighted probability of delivering a request from a client
-         */
-        private int deliverClientRequestWeight = 99;
         /**
          * Weighted probability of dropping a message.
          * The default is 0 (no messages dropped as a scheduler decision).