add RELAXED/ADAPTIVE backoff modes to reduce CPU usage (#76)

Author: eeiaao

docs/components/engine/engine_config.md

@@ -91,6 +91,9 @@ namespace config
   inline constexpr size_t DEFAULT_EVENTBUS_CAPACITY = 4096;
   inline constexpr size_t DEFAULT_EVENTBUS_MAX_CONSUMERS = 128;
 
+  // Connector pool capacity (must be > EventBus capacity to prevent exhaustion)
+  inline constexpr size_t DEFAULT_CONNECTOR_POOL_CAPACITY = 8191;
+
   // CPU Affinity Priority Constants
   inline constexpr int ISOLATED_CORE_PRIORITY_BOOST = 5;
   inline constexpr int DEFAULT_REALTIME_PRIORITY = 80;
@@ -112,6 +115,9 @@ These can be overridden via preprocessor defines:
 - `FLOX_DEFAULT_EVENTBUS_CAPACITY`
 - `FLOX_DEFAULT_EVENTBUS_MAX_CONSUMERS`
 - `FLOX_DEFAULT_ORDER_TRACKER_CAPACITY`
+- `FLOX_DEFAULT_CONNECTOR_POOL_CAPACITY`
+
+**Important:** `DEFAULT_CONNECTOR_POOL_CAPACITY` must be greater than `DEFAULT_EVENTBUS_CAPACITY`. EventBus only reclaims events on wrap-around, so if pool capacity ≤ bus capacity, the pool will exhaust before any events are returned.
 
 ## Notes
 

docs/components/util/memory/pool.md

@@ -83,6 +83,24 @@ pool.setExhaustionCallback([](size_t capacity, size_t inUse) {
 
 The exhaustion callback is invoked each time `acquire()` returns `nullopt` due to pool exhaustion.
 
+## Sizing Guidelines
+
+When using pools with `EventBus`, the pool capacity **must be greater than** the EventBus capacity:
+
+```cpp
+// Correct: pool capacity (8191) > bus capacity (4096)
+Pool<BookUpdateEvent, 8191> pool;
+EventBus<Handle<BookUpdateEvent>, 4096> bus;
+
+// Incorrect: will cause pool exhaustion
+Pool<BookUpdateEvent, 4096> pool;  // Same as bus = will exhaust!
+EventBus<Handle<BookUpdateEvent>, 4096> bus;
+```
+
+**Why?** EventBus only reclaims events when the ring buffer wraps around. If pool capacity ≤ bus capacity, all pool slots will be in-flight before any can be returned.
+
+The default `config::DEFAULT_CONNECTOR_POOL_CAPACITY` (8191) is sized for this reason when used with `DEFAULT_EVENTBUS_CAPACITY` (4096).
+
 ## Notes
 
 * Zero allocations in steady-state operation.

docs/reference/api/engine/engine_config.md

@@ -73,6 +73,10 @@ The header also defines compile-time constants that can be overridden via prepro
 #ifndef FLOX_DEFAULT_ORDER_TRACKER_CAPACITY
 #define FLOX_DEFAULT_ORDER_TRACKER_CAPACITY 4096
 #endif
+
+#ifndef FLOX_DEFAULT_CONNECTOR_POOL_CAPACITY
+#define FLOX_DEFAULT_CONNECTOR_POOL_CAPACITY 8191
+#endif
 ```
 
 ### `config` Namespace Constants
@@ -83,6 +87,9 @@ namespace config {
   inline constexpr size_t DEFAULT_EVENTBUS_CAPACITY = 4096;
   inline constexpr size_t DEFAULT_EVENTBUS_MAX_CONSUMERS = 128;
 
+  // Connector pool capacity (must be > EventBus capacity)
+  inline constexpr size_t DEFAULT_CONNECTOR_POOL_CAPACITY = 8191;
+
   // CPU Affinity Priority Constants
   inline constexpr int ISOLATED_CORE_PRIORITY_BOOST = 5;
   inline constexpr int DEFAULT_REALTIME_PRIORITY = 80;
@@ -104,6 +111,7 @@ namespace config {
 |----------|-------|-------------|
 | `DEFAULT_EVENTBUS_CAPACITY` | 4096 | Ring buffer size for EventBus |
 | `DEFAULT_EVENTBUS_MAX_CONSUMERS` | 128 | Maximum subscribers per bus |
+| `DEFAULT_CONNECTOR_POOL_CAPACITY` | 8191 | Pool capacity for exchange connectors (must be > EventBus capacity) |
 | `ISOLATED_CORE_PRIORITY_BOOST` | 5 | Priority boost for isolated cores |
 | `DEFAULT_REALTIME_PRIORITY` | 80 | Default RT priority for threads |
 | `FALLBACK_REALTIME_PRIORITY` | 90 | Fallback RT priority |

docs/reference/utilities.md

@@ -265,22 +265,71 @@ public:
 
 **Header:** `flox/util/performance/busy_backoff.h`
 
-Adaptive backoff strategy for spin loops.
+Configurable backoff strategy for spin loops with support for different deployment environments.
 
 ```cpp
-class BusyBackoff
+enum class BackoffMode
 {
-public:
-  void pause();   // Spin, yield, or sleep based on iteration count
+  AGGRESSIVE,  // Dedicated colo: busy-spin with CPU pause, minimal yields
+  RELAXED,     // Shared VPS/cloud: early sleep, minimal CPU burn
+  ADAPTIVE     // Auto-adjust: starts aggressive, backs off under contention
+};
+
+namespace config
+{
+  inline BackoffMode defaultBackoffMode = BackoffMode::ADAPTIVE;
+}
+
+struct BusyBackoff
+{
+  explicit BusyBackoff(BackoffMode mode = config::defaultBackoffMode);
+  void pause();   // Spin, yield, or sleep based on mode and iteration count
   void reset();   // Reset iteration counter
 };
 ```
 
-### Strategy
+### Backoff Modes
+
+| Mode | CPU Usage | Latency | Use Case |
+|------|-----------|---------|----------|
+| `AGGRESSIVE` | ~100% | Lowest | Dedicated colo, bare metal, isolated cores |
+| `RELAXED` | <5% | +100-500µs | Shared VPS, cloud, monitoring-only |
+| `ADAPTIVE` | Variable | Variable | General purpose, good default |
+
+### Mode Details
+
+**AGGRESSIVE** (for dedicated hardware):
+1. First 2048 iterations: CPU pause instruction
+2. Beyond: yield(), then reset
 
-1. First ~100 iterations: CPU pause instruction
-2. Next ~100 iterations: `std::this_thread::yield()`
-3. Beyond: Short sleep (microseconds)
+**RELAXED** (for shared infrastructure):
+1. First 8 iterations: CPU pause
+2. Next 8 iterations: yield()
+3. Beyond: sleep(100-500µs)
+
+**ADAPTIVE** (auto-adjusting):
+1. First 128 iterations: CPU pause (burst handling)
+2. 128-512: yield() (medium contention)
+3. 512-2048: sleep(10µs)
+4. Beyond: sleep(100µs), reset to 512
+
+### Configuration
+
+Set the global default at startup:
+```cpp
+// For cloud/VPS deployments
+flox::config::defaultBackoffMode = flox::BackoffMode::RELAXED;
+
+// For dedicated colo
+flox::config::defaultBackoffMode = flox::BackoffMode::AGGRESSIVE;
+```
+
+Or configure per EventBus:
+```cpp
+EventBus<MyEvent> bus;
+bus.setBackoffMode(BackoffMode::RELAXED);
+bus.start();
+```
 
 ---
 

include/flox/util/eventing/event_bus.h

@@ -161,8 +161,9 @@ class EventBus : public ISubsystem
       auto* l = _consumers[i].listener;
       auto required = _consumers[i].required;
       auto coreIdx = _consumers[i].coreIndex;
+      auto backoffMode = _backoffMode;
 
-      _consumers[i].thread.emplace([this, i, l, required, coreIdx]
+      _consumers[i].thread.emplace([this, i, l, required, coreIdx, backoffMode]
                                    {
 #if FLOX_CPU_AFFINITY_ENABLED
          auto threadCpuAffinity = performance::createCpuAffinity();
@@ -214,9 +215,9 @@ class EventBus : public ISubsystem
            if (_active.fetch_sub(1, std::memory_order_acq_rel) == 1) _cv.notify_one();
          }
 
-         BusyBackoff backoff;
+         BusyBackoff backoff(backoffMode);
          int64_t next = -1;
- 
+
          while (_running.load(std::memory_order_acquire))
          {
            const int64_t seq = next + 1;
@@ -339,6 +340,8 @@ class EventBus : public ISubsystem
   uint32_t consumerCount() const { return _consumerCount.load(std::memory_order_acquire); }
   void enableDrainOnStop() { _drainOnStop = true; }
 
+  void setBackoffMode(BackoffMode mode) { _backoffMode = mode; }
+
 #if FLOX_CPU_AFFINITY_ENABLED
   // ---------- CPU Affinity / RT priority ----------
   void setAffinityConfig(const AffinityConfig& cfg)
@@ -608,6 +611,7 @@ class EventBus : public ISubsystem
   std::atomic<uint32_t> _active{0};
 
   bool _drainOnStop{false};
+  BackoffMode _backoffMode{config::defaultBackoffMode};
 
 #if FLOX_CPU_AFFINITY_ENABLED
   // CPU affinity / RT

include/flox/util/performance/busy_backoff.h

@@ -7,25 +7,63 @@
  * license information.
  */
 
+#pragma once
+
+#include <chrono>
 #include <thread>
 
 #if defined(__x86_64__) || defined(_M_X64)
 #include <immintrin.h>
 #define FLOX_CPU_PAUSE() _mm_pause()
 #elif defined(__aarch64__) || defined(_M_ARM64)
-#define FLOX_CPU_PAUSE() __asm__ __volatile__("yield" ::: "memory")
+#define FLOX_CPU_PAUSE() __asm__ __volatile__("yield" :: \
+                                                  : "memory")
 #else
 #define FLOX_CPU_PAUSE() ((void)0)
 #endif
 
 namespace flox
 {
 
+enum class BackoffMode
+{
+  AGGRESSIVE,  ///< Dedicated colo: busy-spin with CPU pause, minimal yields
+  RELAXED,     ///< Shared VPS/cloud: early sleep, minimal CPU burn
+  ADAPTIVE     ///< Auto-adjust: starts aggressive, backs off under contention
+};
+
+namespace config
+{
+inline BackoffMode defaultBackoffMode = BackoffMode::ADAPTIVE;
+}  // namespace config
+
 struct BusyBackoff
 {
   int spins = 0;
+  BackoffMode mode;
+
+  explicit BusyBackoff(BackoffMode m = config::defaultBackoffMode) : mode(m) {}
 
   inline void pause()
+  {
+    switch (mode)
+    {
+      case BackoffMode::AGGRESSIVE:
+        pauseAggressive();
+        break;
+      case BackoffMode::RELAXED:
+        pauseRelaxed();
+        break;
+      case BackoffMode::ADAPTIVE:
+        pauseAdaptive();
+        break;
+    }
+  }
+
+  inline void reset() { spins = 0; }
+
+ private:
+  inline void pauseAggressive()
   {
     if (spins < 2048)
     {
@@ -45,7 +83,68 @@ struct BusyBackoff
     }
   }
 
-  inline void reset() { spins = 0; }
+  inline void pauseRelaxed()
+  {
+    if (spins < 8)
+    {
+      // Brief spin for immediate data
+      FLOX_CPU_PAUSE();
+      ++spins;
+      return;
+    }
+
+    if (spins < 16)
+    {
+      // Quick yield
+      std::this_thread::yield();
+      ++spins;
+      return;
+    }
+
+    // Sleep to release CPU - 100us is enough for market data which arrives
+    // at most every few milliseconds per symbol
+    std::this_thread::sleep_for(std::chrono::microseconds(100));
+    if (spins < 64)
+    {
+      ++spins;
+    }
+    else
+    {
+      // Increase sleep duration for sustained idle periods
+      std::this_thread::sleep_for(std::chrono::microseconds(500));
+    }
+  }
+
+  inline void pauseAdaptive()
+  {
+    if (spins < 128)
+    {
+      // Initial aggressive phase for low-latency burst handling
+      FLOX_CPU_PAUSE();
+      ++spins;
+      return;
+    }
+
+    if (spins < 512)
+    {
+      // Medium contention: yield to other threads
+      std::this_thread::yield();
+      ++spins;
+      return;
+    }
+
+    if (spins < 2048)
+    {
+      // High contention: short sleep
+      std::this_thread::sleep_for(std::chrono::microseconds(10));
+      ++spins;
+      return;
+    }
+
+    // Sustained contention: longer sleep, reset cycle
+    std::this_thread::sleep_for(std::chrono::microseconds(100));
+    spins = 512;  // Reset to medium level, not zero
+  }
 };
 
 }  // namespace flox