runtime: allocate unique event IDs across wraparound

esperanto-tools-libs/src/EventManager.cpp

@@ -10,6 +10,7 @@
 #include <cassert>
 #include <condition_variable>
 #include <easy/profiler.h>
+#include <limits>
 #include <mutex>
 #include <thread>
 
@@ -29,10 +30,24 @@ void EventManager::addOnDispatchCallback(OnDispatchCallback callback) {
 
 EventId EventManager::getNextId() {
   std::lock_guard lock(mutex_);
-  auto res = EventId{nextEventId_++};
-  onflyEvents_.emplace(res);
-  RT_VLOG(LOW) << "Last event id: " << static_cast<int>(res);
-  return res;
+  using EventIdValue = std::underlying_type_t<EventId>;
+  constexpr auto kEventIdCount = static_cast<size_t>(std::numeric_limits<EventIdValue>::max()) + 1;
+
+  // EventId is a uint16_t and long-running inference processes legitimately
+  // wrap the counter. Never return an ID that is still in flight: emplace()
+  // used to fail silently here, so the newer command stole the older command's
+  // completion and the second response was reported as an unknown event.
+  for (size_t attempt = 0; attempt < kEventIdCount; ++attempt) {
+    auto res = EventId{nextEventId_++};
+    if (onflyEvents_.find(res) != onflyEvents_.end() || blockedThreads_.find(res) != blockedThreads_.end()) {
+      continue;
+    }
+    onflyEvents_.emplace(res);
+    RT_VLOG(LOW) << "Last event id: " << static_cast<int>(res);
+    return res;
+  }
+
+  throw Exception("No free runtime event IDs; synchronize the stream before submitting more commands");
 }
 
 void EventManager::dispatch(EventId event) {
@@ -115,4 +130,4 @@ EventManager::~EventManager() {
       << static_cast<int>(event);
     cv->notify_all();
   }
-}
+}

esperanto-tools-libs/src/RuntimeImp.cpp

@@ -869,8 +869,16 @@ void RuntimeImp::dispatch(EventId event) {
   evt.setEvent(event);
   getProfiler()->record(evt);
   streamManager_.removeEvent(event);
+  // Keep the ID in flight until observers have consumed the completion. This
+  // prevents a wrapped ID from being reused between EventManager::dispatch()
+  // and the server worker forwarding the old completion to its client.
+  try {
+    notify(event);
+  } catch (...) {
+    eventManager_.dispatch(event);
+    throw;
+  }
   eventManager_.dispatch(event);
-  notify(event);
 }
 
 void RuntimeImp::checkDevice(DeviceId device) {

esperanto-tools-libs/tests/unit-tests/test_event_manager.cpp

@@ -9,6 +9,7 @@
 #include <cstdio>
 #include <functional>
 #include <gtest/gtest.h>
+#include <limits>
 #include <thread>
 
 #pragma GCC diagnostic push
@@ -106,6 +107,37 @@ TEST_F(EventManagerF, severalEvents) {
   }
 }
 
+TEST_F(EventManagerF, wrapSkipsEventIdsStillInFlight) {
+  using EventIdValue = std::underlying_type_t<EventId>;
+  constexpr auto maxId = std::numeric_limits<EventIdValue>::max();
+
+  em_.nextEventId_ = maxId;
+  auto atMax = em_.getNextId();
+  auto atZero = em_.getNextId();
+  EXPECT_EQ(static_cast<EventIdValue>(atMax), maxId);
+  EXPECT_EQ(static_cast<EventIdValue>(atZero), 0);
+
+  em_.nextEventId_ = maxId;
+  auto skippedToOne = em_.getNextId();
+  EXPECT_EQ(static_cast<EventIdValue>(skippedToOne), 1);
+
+  em_.dispatch(atMax);
+  em_.dispatch(atZero);
+  em_.dispatch(skippedToOne);
+}
+
+TEST_F(EventManagerF, exhaustionFailsInsteadOfReturningDuplicateId) {
+  using EventIdValue = std::underlying_type_t<EventId>;
+  constexpr auto eventIdCount =
+    static_cast<size_t>(std::numeric_limits<EventIdValue>::max()) + 1;
+
+  for (size_t i = 0; i < eventIdCount; ++i) {
+    em_.onflyEvents_.emplace(EventId{static_cast<EventIdValue>(i)});
+  }
+  EXPECT_THROW(em_.getNextId(), Exception);
+  em_.onflyEvents_.clear();
+}
+
 TEST_F(EventManagerF, blockingThreads) {
   auto nEvents = 10U;
   auto nThreads = 10000U;