feat: thread pinning, tighter threading, housekeeping

milsanore · milsanore · commit 615958a52326 · 2025-10-25T21:14:31.000+11:00
diff --git a/README.md b/README.md
@@ -1,4 +1,4 @@
-[![Build](https://github.com/milsanore/trader.cpp/actions/workflows/commits.yml/badge.svg)](https://github.com/milsanore/trader.cpp/actions/workflows/commits.yml)
+[![Nightly](https://github.com/milsanore/trader.cpp/actions/workflows/nightly.yml/badge.svg)](https://github.com/milsanore/trader.cpp/actions/workflows/nightly.yml)
 [![Quality Gate Status](https://sonarcloud.io/api/project_badges/measure?project=milsanore_trader.cpp&metric=alert_status)](https://sonarcloud.io/summary/new_code?id=milsanore_trader.cpp)
 [![codecov](https://codecov.io/github/milsanore/trader.cpp/graph/badge.svg?token=C787ZTXBQC)](https://codecov.io/github/milsanore/trader.cpp)
 
@@ -177,6 +177,8 @@ NB: this app uses `make` as a recipe book, but it's not essential:
 - FIX
   - ✅ debug quickfix to confirm if it's running in it's own thread
   - switch to Fix8
+- security
+  - OpenSSF Scorecard
 - other
   - nix
   - decimal type
@@ -189,17 +191,25 @@ NB: this app uses `make` as a recipe book, but it's not essential:
 # Design
 ```mermaid
 sequenceDiagram
-    participant TM as Thread 1<br>(Main + UI)
-    participant TO as Thread 2<br>(UI Orderbook Worker)
-    participant TF as Thread 3<br>(FIX Worker)
-    participant TL as Thread 4<br>(UI Log Worker)
-
-    TF-->>TF: subscribe to Binance <br> + push to <queue>
-    TF->>TO: pull from <queue>
-    TO-->>TO: build UI
-    TO->>TM: request render
-    TL-->>TL: poll log file <br> + build UI
-    TL->>TM: request render
+    participant MAIN    as Main Thread
+    participant LOGS    as Log UI Thread
+    participant BOOK    as Orderbook Thread
+    participant TRADES  as Trade Thread
+    participant FIX     as FIX Thread
+
+    FIX-->>FIX: subscribe to Binance <br> + push to <queue>
+
+    FIX->>TRADES: pull from <queue>
+    TRADES-->>TRADES: build UI
+
+    FIX->>BOOK: pull from <queue>
+    BOOK-->>BOOK: build UI
+
+    LOGS-->>LOGS: poll log file <br> + build UI
+
+    TRADES->>MAIN: request render
+    BOOK->>MAIN: request render
+    LOGS->>MAIN: request render
 ```
 
 # Credits
diff --git a/src/binance/config.h b/src/binance/config.h
@@ -58,6 +58,9 @@ struct Config {
 
   /// @brief 1 == top level, otherwise 5000 is Binance's maximum depth
   static constexpr uint16_t MAX_DEPTH = 100;
+
+  static constexpr uint8_t PX_SESSION_CPU_AFFINITY = 0;
+  static constexpr uint8_t TX_SESSION_CPU_AFFINITY = 1;
 };
 
 }  // namespace binance
diff --git a/src/binance/fix_app.cpp b/src/binance/fix_app.cpp
@@ -26,13 +26,14 @@ namespace binance {
 
 FixApp::FixApp(const std::vector<std::string>& symbols,
                std::unique_ptr<IAuth> auth,
-               const uint16_t MAX_DEPTH)
-    : symbols_(symbols), auth_(std::move(auth)), MAX_DEPTH_(MAX_DEPTH) {
-  // TODO(mils): I think this is a singleton
-  utils::Threading::set_thread_name(THREAD_NAME_);
-  spdlog::info("naming FIXs own thread, name [{}], id [{}]", THREAD_NAME_,
-               utils::Threading::get_os_thread_id());
-}
+               const uint16_t MAX_DEPTH,
+               const uint8_t PX_SESSION_CPU_AFFINITY,
+               const uint8_t TX_SESSION_CPU_AFFINITY)
+    : symbols_(symbols),
+      auth_(std::move(auth)),
+      MAX_DEPTH_(MAX_DEPTH),
+      PX_SESSION_CPU_AFFINITY_(PX_SESSION_CPU_AFFINITY),
+      TX_SESSION_CPU_AFFINITY_(TX_SESSION_CPU_AFFINITY) {}
 
 void FixApp::subscribe_to_prices(const FIX::SessionID& session_id) const {
   spdlog::info("subscribing to depth. qualifier [{}], id [{}]",
@@ -128,11 +129,18 @@ void FixApp::onLogon(const FIX::SessionID& sessionId) {
   // TODO: now need to wait for all sessions to be logged on before nullifying keys
   // auth_->clear_keys();
 
-  if (sessionId.getSessionQualifier() == PRICE_SESSION_QUALIFIER_) {
+  std::string thread_name = THREAD_NAME_ + "_" + sessionId.getSessionQualifier();
+  utils::Threading::set_thread_name(thread_name);
+  spdlog::info("naming FIX session thread, name [{}], id [{}]", thread_name,
+               utils::Threading::get_os_thread_id());
+
+  if (sessionId.getSessionQualifier() == PX_SESSION_QUALIFIER_) {
+    utils::Threading::set_current_thread_affinity(PX_SESSION_CPU_AFFINITY_);
     subscribe_to_prices(sessionId);
-  } else if (sessionId.getSessionQualifier() == TRADE_SESSION_QUALIFIER_) {
+  } else if (sessionId.getSessionQualifier() == TX_SESSION_QUALIFIER_) {
+    utils::Threading::set_current_thread_affinity(TX_SESSION_CPU_AFFINITY_);
     subscribe_to_trades(sessionId);
-  } else if (sessionId.getSessionQualifier() == ORDER_SESSION_QUALIFIER_) {
+  } else if (sessionId.getSessionQualifier() == OX_SESSION_QUALIFIER_) {
     // do nothing for order session
   } else {
     spdlog::error("unknown session, qualifier [{}], id [{}]",
@@ -210,9 +218,9 @@ void FixApp::onMessage(const FIX44::MarketDataSnapshotFullRefresh& m,
 }
 void FixApp::onMessage(const FIX44::MarketDataIncrementalRefresh& m,
                        const FIX::SessionID& sessionID) {
-  if (sessionID.getSessionQualifier() == PRICE_SESSION_QUALIFIER_) {
+  if (sessionID.getSessionQualifier() == PX_SESSION_QUALIFIER_) {
     order_queue_.enqueue(std::make_shared<const FIX44::MarketDataIncrementalRefresh>(m));
-  } else if (sessionID.getSessionQualifier() == TRADE_SESSION_QUALIFIER_) {
+  } else if (sessionID.getSessionQualifier() == TX_SESSION_QUALIFIER_) {
     trade_queue_.enqueue(std::make_shared<const FIX44::MarketDataIncrementalRefresh>(m));
   } else {
     spdlog::error(
diff --git a/src/binance/fix_app.h b/src/binance/fix_app.h
@@ -20,7 +20,9 @@ class FixApp final : public FIX::Application, public FIX44::MessageCracker {
  public:
   FixApp(const std::vector<std::string>& symbols,
          std::unique_ptr<IAuth> auth,
-         const uint16_t MAX_DEPTH);
+         const uint16_t MAX_DEPTH,
+         const uint8_t PX_SESSION_CPU_AFFINITY,
+         const uint8_t TX_SESSION_CPU_AFFINITY);
   ~FixApp() override = default;
 
   // Use the FIX44::MessageCracker to pull in the relevant overloads. This resolves the
@@ -42,13 +44,15 @@ class FixApp final : public FIX::Application, public FIX44::MessageCracker {
   // TODO: we will have one of these per instrument
 
  private:
-  static inline constexpr std::string THREAD_NAME_ = "tradercppFIX2";
-  static inline constexpr std::string PRICE_SESSION_QUALIFIER_ = "PX";
-  static inline constexpr std::string TRADE_SESSION_QUALIFIER_ = "TX";
-  static inline constexpr std::string ORDER_SESSION_QUALIFIER_ = "OX";
+  static inline constexpr std::string THREAD_NAME_ = "fix_session";
+  static inline constexpr std::string PX_SESSION_QUALIFIER_ = "PX";
+  static inline constexpr std::string TX_SESSION_QUALIFIER_ = "TX";
+  static inline constexpr std::string OX_SESSION_QUALIFIER_ = "OX";
   const std::vector<std::string>& symbols_;
   const std::unique_ptr<IAuth> auth_;
   const uint16_t MAX_DEPTH_;
+  const uint8_t PX_SESSION_CPU_AFFINITY_;
+  const uint8_t TX_SESSION_CPU_AFFINITY_;
 
   void onCreate(const FIX::SessionID&) override;
   void onLogon(const FIX::SessionID&) override;
diff --git a/src/binance/worker.cpp b/src/binance/worker.cpp
@@ -4,7 +4,7 @@
 #include <quickfix/FileStore.h>
 #include <quickfix/Session.h>
 #include <quickfix/SessionSettings.h>
-#include <quickfix/SocketInitiator.h>
+#include <quickfix/ThreadedSocketInitiator.h>
 
 #include <memory>
 
@@ -20,38 +20,25 @@ Worker::Worker(std::unique_ptr<FixApp> app,
                std::unique_ptr<FIX::FileStoreFactory> store,
                FIX::SessionSettings settings,
                std::unique_ptr<FIX::FileLogFactory> log,
-               std::unique_ptr<FIX::SocketInitiator> initiator,
-               const std::function<void(std::stop_token)>& task)
+               std::unique_ptr<FIX::ThreadedSocketInitiator> initiator)
     : app_(std::move(app)),
       store_(std::move(store)),
       settings_(std::move(settings)),
       log_(std::move(log)),
-      initiator_(std::move(initiator)),
-      worker_task_(task) {
-  // default behaviour
-  if (!task) {
-    worker_task_ = {[this]([[maybe_unused]] const std::stop_token& stoken) {
-      utils::Threading::set_thread_name(THREAD_NAME_);
-      spdlog::info("starting FIX wrapper thread. name [{}], id [{}]", THREAD_NAME_,
-                   utils::Threading::get_os_thread_id());
-      // NB: SocketInitiator::start() is a blocking call, so the stop_token
-      // cannot cancel the thread. NB: The `stop()` function has to forcibly
-      // stop it with `initiator_->stop()`.
-      initiator_->start();
-      spdlog::info("started FIX initiator");
-    }};
-  }
-}
+      initiator_(std::move(initiator)) {}
 
 // static member function
 Worker Worker::from_conf(Config& conf) {
   std::unique_ptr<IAuth> auth =
       std::make_unique<Auth>(conf.api_key, conf.private_key_path);
-  auto app = std::make_unique<FixApp>(conf.symbols, std::move(auth), conf.MAX_DEPTH);
+  auto app = std::make_unique<FixApp>(conf.symbols, std::move(auth), conf.MAX_DEPTH,
+                                      conf.PX_SESSION_CPU_AFFINITY,
+                                      conf.TX_SESSION_CPU_AFFINITY);
   auto settings = FIX::SessionSettings{conf.fix_config_path};
   auto store = std::make_unique<FIX::FileStoreFactory>(settings);
   auto log = std::make_unique<FIX::FileLogFactory>(settings);
-  auto initiator = std::make_unique<FIX::SocketInitiator>(*app, *store, settings, *log);
+  auto initiator =
+      std::make_unique<FIX::ThreadedSocketInitiator>(*app, *store, settings, *log);
 
   // hand over object ownership to the instance being created (by the static
   // function)
@@ -60,27 +47,13 @@ Worker Worker::from_conf(Config& conf) {
 }
 
 void Worker::start() {
-  try {
-    worker_ = std::jthread{worker_task_};
-  } catch (const std::exception& e) {
-    spdlog::error("error starting FIX. error [{}]", e.what());
-  } catch (...) {
-    spdlog::error("error starting FIX. unknown error");
-  }
+  initiator_->start();
+  spdlog::info("started FIX initiator");
 }
 
 void Worker::stop() {
-  try {
-    if (initiator_) {
-      initiator_->stop();  // TODO(mils): does this need a try/catch?
-    }
-    spdlog::info("stopped FIX initiator");
-  } catch (const std::exception& e) {
-    spdlog::error("error stopping FIX initiator. error [{}]", e.what());
-  } catch (...) {
-    spdlog::error("error stopping FIX initiator. unknown error");
-  }
-  worker_ = std::jthread{};
+  initiator_->stop();  // TODO(mils): does this need a try/catch?
+  spdlog::info("stopped FIX initiator");
 }
 
 moodycamel::ConcurrentQueue<std::shared_ptr<const FIX44::Message>>&
diff --git a/src/binance/worker.h b/src/binance/worker.h
@@ -3,7 +3,7 @@
 #include <quickfix/FileLog.h>
 #include <quickfix/FileStore.h>
 #include <quickfix/SessionSettings.h>
-#include <quickfix/SocketInitiator.h>
+#include <quickfix/ThreadedSocketInitiator.h>
 #include <quickfix/fix44/Message.h>
 
 #include <memory>
@@ -18,13 +18,11 @@ namespace binance {
 /// @brief Binance DI container
 class Worker final {
  public:
-  static inline constexpr std::string THREAD_NAME_ = "tradercppFIX1";
   Worker(std::unique_ptr<FixApp> app,
          std::unique_ptr<FIX::FileStoreFactory> store,
          FIX::SessionSettings settings,
          std::unique_ptr<FIX::FileLogFactory> log,
-         std::unique_ptr<FIX::SocketInitiator> initiator,
-         const std::function<void(std::stop_token)>& task = {});
+         std::unique_ptr<FIX::ThreadedSocketInitiator> initiator);
   /// @brief factory for concrete Binance instances, using config
   /// @param conf Binance configuration parameters
   /// @return
@@ -44,10 +42,7 @@ class Worker final {
   std::unique_ptr<FIX::FileStoreFactory> store_;
   FIX::SessionSettings settings_;
   std::unique_ptr<FIX::FileLogFactory> log_;
-  std::unique_ptr<FIX::SocketInitiator> initiator_;
-  // thread
-  std::jthread worker_;
-  std::function<void(std::stop_token)> worker_task_;
+  std::unique_ptr<FIX::ThreadedSocketInitiator> initiator_;
 };
 
 }  // namespace binance
diff --git a/src/main.cpp b/src/main.cpp
@@ -10,9 +10,9 @@
 #include "utils/threading.h"
 
 int main() {
-  utils::Threading::set_thread_name("tradercppMAIN");
+  utils::Threading::set_thread_name("main");
 
-  // LOG CONFIG
+  // log config
   spdlog::cfg::load_env_levels("LOG_LEVEL");
   const char* log_path = std::getenv("LOG_PATH");
   const auto logger = spdlog::basic_logger_mt("basic_logger", log_path);
@@ -21,7 +21,7 @@ int main() {
   spdlog::set_pattern("[%Y-%m-%d %H:%M:%S.%e] [%^%l%$] [%t] %v");
   spdlog::flush_every(std::chrono::microseconds(100));
   spdlog::info("hello");
-  utils::Env::log_arch();
+  utils::Env::log_current_architecture();
 
   // Binance market data connectivity
   auto b_conf = binance::Config::from_env();
diff --git a/src/ui/log_box/log_file_watcher.h b/src/ui/log_box/log_file_watcher.h
@@ -15,7 +15,7 @@ class LogFileWatcher : public ILogWatcher, public efsw::FileWatchListener {
   using Callback = std::function<void(std::vector<std::string>)>;
 
  public:
-  static inline constexpr std::string THREAD_NAME_ = "tradercppuiLOG";
+  static inline constexpr std::string THREAD_NAME_ = "ui_logwatcher";
 
   explicit LogFileWatcher(std::string directory, std::string filename)
       : directory_(directory), filename_(filename) {
diff --git a/src/ui/order_book_box.h b/src/ui/order_book_box.h
@@ -16,7 +16,7 @@ namespace ui {
 
 class OrderBookBox {
  public:
-  static inline constexpr std::string THREAD_NAME_ = "tradercppuiBOOK";
+  static inline constexpr std::string THREAD_NAME_ = "ui_orderbook";
   // Constructor: takes a label string
   OrderBookBox(IScreen& screen,
                moodycamel::ConcurrentQueue<std::shared_ptr<const FIX44::Message>>& queue,
diff --git a/src/ui/trade_box.h b/src/ui/trade_box.h
@@ -24,7 +24,7 @@ namespace ui {
 
 class TradeBox {
  public:
-  static inline constexpr std::string THREAD_NAME_ = "tradercppuiTX";
+  static inline constexpr std::string THREAD_NAME_ = "ui_tradebox";
   TradeBox(IScreen& screen,
            binance::Config& binance_config,
            moodycamel::ConcurrentQueue<std::shared_ptr<const FIX44::Message>>& queue,
diff --git a/src/utils/env.h b/src/utils/env.h
@@ -24,7 +24,7 @@ struct Env {
   };
 
   /// @brief log the architecutre in use
-  static void log_arch() {
+  static void log_current_architecture() {
     // detect architecture
 #if defined(__aarch64__) || defined(_M_ARM64)
     constexpr const char* ARCH = "ARM64";
diff --git a/src/utils/threading.cpp b/src/utils/threading.cpp
diff --git a/src/utils/threading.h b/src/utils/threading.h
