diff --git a/src/io/v3/simulator_handle.hpp b/src/io/v3/simulator_handle.hpp index 7104ae138..109b636f0 100644 --- a/src/io/v3/simulator_handle.hpp +++ b/src/io/v3/simulator_handle.hpp @@ -250,6 +250,19 @@ class SimulatorHandle { } } + void TimeoutPromisesPastDeadline() { + uint64_t now = cluster_wide_time_microseconds_; + + for (auto &[promise_key, dop] : promises_) { + // TODO queue this up and drop it after its deadline + if (dop.deadline < now) { + DeadlineAndOpaquePromise dop = std::move(promises_.at(promise_key)); + promises_.erase(promise_key); + dop.promise.TimeOut(); + } + } + } + bool MaybeTickSimulator() { std::unique_lock lock(mu_); @@ -274,6 +287,8 @@ class SimulatorHandle { cv_.notify_all(); + TimeoutPromisesPastDeadline(); + if (in_flight_.empty()) { // return early here because there are no messages to schedule @@ -325,16 +340,7 @@ class SimulatorHandle { dop.promise.Fill(std::move(opaque_message)); } } else if (should_drop) { - // don't add it anywhere, let it drop, if it's a request then time it out - // TODO queue this up and drop it after its deadline - PromiseKey drop_promise_key{.requester_address = opaque_message.from_address, - .request_id = opaque_message.request_id, - .replier_address = to_address}; - if (promises_.contains(drop_promise_key)) { - DeadlineAndOpaquePromise dop = std::move(promises_.at(promise_key)); - promises_.erase(promise_key); - dop.promise.TimeOut(); - } + // don't add it anywhere, let it drop } else { // add to can_receive_ if not const auto &[om_vec, inserted] = can_receive_.try_emplace(to_address, std::vector()); diff --git a/tests/simulation/raft.cpp b/tests/simulation/raft.cpp index 4c8c12010..df0c2869b 100644 --- a/tests/simulation/raft.cpp +++ b/tests/simulation/raft.cpp @@ -623,7 +623,7 @@ void RunServer(Server server) { void RunSimulation() { auto config = SimulatorConfig{ - .drop_percent = 0, + .drop_percent = 5, .perform_timeouts = true, .scramble_messages = true, .rng_seed = 0,