Fix #133

Author: joaomlneto

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

@@ -82,7 +82,7 @@ public void handleMessage(String senderId, MessagePayload reply) {
 
     @Override
     public Instant getCurrentTime() {
-        return this.scenario.getTimekeeper().getTime(this);
+        return this.scenario.getTimekeeper().incrementAndGetTime(this);
     }
 
     /**

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

@@ -269,7 +269,7 @@ public void notifyObserversTimeout() {
 
     @JsonIgnore
     public Instant getCurrentTime() {
-        return this.scenario.getTimekeeper().getTime(this);
+        return this.scenario.getTimekeeper().incrementAndGetTime(this);
     }
 
     /**

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

@@ -130,6 +130,14 @@ public interface Scenario extends Serializable {
      */
     List<ScenarioObserver> getObservers();
 
+
+    /**
+     * Add an observer to the scenario.
+     *
+     * @param observer The observer to add.
+     */
+    void addObserver(ScenarioObserver observer);
+
     /**
      * Load the parameters for the scenario.
      *

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

@@ -1,22 +1,104 @@
 package byzzbench.simulator;
 
+import byzzbench.simulator.faults.Fault;
+import byzzbench.simulator.transport.Event;
+import byzzbench.simulator.transport.MutateMessageEventPayload;
+import byzzbench.simulator.transport.TimeoutEvent;
+import byzzbench.simulator.transport.TransportObserver;
 import com.fasterxml.jackson.annotation.JsonIgnore;
-import lombok.RequiredArgsConstructor;
 
 import java.io.Serializable;
+import java.time.Duration;
 import java.time.Instant;
-import java.util.concurrent.atomic.AtomicLong;
+import java.util.SortedMap;
+import java.util.TreeMap;
 
 /**
- * A timekeeper that provides timestamps to the replicas in the simulator.
+ * A timekeeper that provides timestamps to the nodes in the simulator.
  */
-@RequiredArgsConstructor
-public class Timekeeper implements Serializable {
+public class Timekeeper implements Serializable, TransportObserver {
+    /**
+     * The increment to use each time a timestamp is requested.
+     */
+    public static final Duration INCREMENT = Duration.ofMillis(1);
+
     @JsonIgnore
-    private final transient Scenario scenario;
-    private final AtomicLong counter = new AtomicLong(0);
+    private final Scenario scenario;
+
+    /**
+     * The current time for each node
+     */
+    private final SortedMap<String, Instant> times = new TreeMap<>();
+
+    /**
+     * Create a new timekeeper for the given scenario.
+     *
+     * @param scenario the scenario to create the timekeeper for
+     */
+    public Timekeeper(Scenario scenario) {
+        this.scenario = scenario;
+        this.scenario.getTransport().addObserver(this);
+    }
+
+    /**
+     * Advance the time for the given node and return the new time.
+     *
+     * @param node the node to advance the time for
+     * @return the new time for the node
+     */
+    public Instant incrementAndGetTime(Node node) {
+        Instant current = getTime(node);
+        Instant next = current.plus(INCREMENT);
+        times.put(node.getId(), next);
+        return next;
+    }
 
+    /**
+     * Get the current time for the given node.
+     *
+     * @param node the node to get the time for
+     * @return the current time for the node
+     */
     public Instant getTime(Node node) {
-        return Instant.ofEpochMilli(counter.incrementAndGet());
+        return times.computeIfAbsent(node.getId(), k -> Instant.ofEpochMilli(0));
+    }
+
+    @Override
+    public void onEventAdded(Event event) {
+        // nothing to do
+    }
+
+    @Override
+    public void onEventDropped(Event event) {
+        // nothing to do
+    }
+
+    @Override
+    public void onEventDelivered(Event event) {
+        // check if it was a timeout
+        if (!(event instanceof TimeoutEvent timeoutEvent)) {
+            return;
+        }
+
+        // set counter to max of its current value and the expiration time
+        String nodeId = timeoutEvent.getNodeId();
+        Instant expiration = timeoutEvent.getExpiresAt();
+        Instant current = times.getOrDefault(nodeId, Instant.ofEpochMilli(0));
+        times.put(nodeId, current.isAfter(expiration) ? current : expiration);
+    }
+
+    @Override
+    public void onMessageMutation(MutateMessageEventPayload payload) {
+        // nothing to do
+    }
+
+    @Override
+    public void onFault(Fault fault) {
+        // nothing to do
+    }
+
+    @Override
+    public void onTimeout(TimeoutEvent event) {
+        // nothing to do
     }
 }

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

@@ -165,7 +165,7 @@ public void set_out_key(PbftReplica r, int[] k, long t) {
              */
 
             this.tstamp = t;
-            my_tstamp = r.getScenario().getTimekeeper().getTime(r);
+            my_tstamp = r.getCurrentTime();
         }
     }
 

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

@@ -248,6 +248,9 @@ public synchronized void deliverEvent(long eventId) throws Exception {
         this.scenario.getSchedule().appendEvent(e);
         e.setStatus(Event.Status.DELIVERED);
 
+        // For timeouts, this should be called before, so the Replica time is updated
+        this.observers.forEach(o -> o.onEventDelivered(e));
+
         switch (e) {
             case ClientRequestEvent c -> {
                 this.scenario.getNodes().get(c.getRecipientId()).handleMessage(c.getSenderId(), c.getPayload());
@@ -263,8 +266,6 @@ public synchronized void deliverEvent(long eventId) throws Exception {
             }
         }
 
-        this.observers.forEach(o -> o.onEventDelivered(e));
-
         log.info("Delivered " + e);
     }
 

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

@@ -5,37 +5,43 @@
 public interface TransportObserver {
     /**
      * Called when an event is added to the transport layer.
+     *
      * @param event The event that was added.
      */
     void onEventAdded(Event event);
 
     /**
      * Called when the status of an event changes to {@link Event.Status#DROPPED}.
+     *
      * @param event The event that was dropped.
      */
     void onEventDropped(Event event);
 
     /**
      * Called when the status of an event changes to {@link Event.Status#DELIVERED}.
+     *
      * @param event The event that was delivered.
      */
     void onEventDelivered(Event event);
 
     /**
      * Called when a message is mutated.
+     *
      * @param payload The payload of the mutation.
      */
     void onMessageMutation(MutateMessageEventPayload payload);
 
     /**
      * Called when a fault is injected.
+     *
      * @param fault The fault that was injected.
      */
     void onFault(Fault fault);
 
     /**
-     * Called when a timeout event is triggered.
-     * @param event The timeout event that was triggered.
+     * Called when a timeout event is created.
+     *
+     * @param event The timeout event that was created.
      */
     void onTimeout(TimeoutEvent event);
 }