fix: tighter threading, housekeeping

Author: milsanore

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

src/binance/fix_app.cpp

@@ -27,12 +27,7 @@ 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());
-}
+    : symbols_(symbols), auth_(std::move(auth)), MAX_DEPTH_(MAX_DEPTH) {}
 
 void FixApp::subscribe_to_prices(const FIX::SessionID& session_id) const {
   spdlog::info("subscribing to depth. qualifier [{}], id [{}]",
@@ -138,6 +133,11 @@ void FixApp::onLogon(const FIX::SessionID& sessionId) {
     spdlog::error("unknown session, qualifier [{}], id [{}]",
                   sessionId.getSessionQualifier(), sessionId.toString());
   }
+
+  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());
 };
 void FixApp::onLogout(const FIX::SessionID& sessionId) {
   spdlog::info("session logout. qualifier [{}], id [{}]", sessionId.getSessionQualifier(),

src/binance/fix_app.h

@@ -42,7 +42,7 @@ 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 THREAD_NAME_ = "fix_session";
   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";

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,28 +20,12 @@ 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) {
@@ -51,7 +35,8 @@ Worker Worker::from_conf(Config& conf) {
   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 +45,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>>&

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

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);

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) {

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,

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,