Make configureable num threads, post on strand, not ioc

Signed-off-by: Nijat K <neej@nijats-mbp.mynetworksettings.com>

Author: Nijat K

cpp/csp/adapters/websocket/ClientConnectionRequestAdapter.cpp

@@ -9,10 +9,13 @@ ClientConnectionRequestAdapter::ClientConnectionRequestAdapter(
     WebsocketEndpointManager * websocketManager,
     net::io_context& ioc,
     bool is_subscribe,
-    size_t caller_id
+    size_t caller_id,
+    boost::asio::strand<boost::asio::io_context::executor_type>& strand
+
 ) : OutputAdapter( engine ),  
     m_websocketManager( websocketManager ),
     m_ioc( ioc),
+    m_strand( strand ),
     m_isSubscribe( is_subscribe ),
     m_callerId( caller_id ),
     m_checkPerformed( is_subscribe ? false : true )  // we only need to check for pruned input adapters
@@ -37,7 +40,7 @@ void ClientConnectionRequestAdapter::executeImpl()
     // m_ioc to handle the connection request. We want to keep
     // all updates to internal data structures at graph run-time
     // to that thread.
-    boost::asio::post(m_ioc, [this, val=std::move(val)]() {
+    boost::asio::post(m_strand, [this, val=std::move(val)]() {
         for(const auto& conn_req: val) {
             m_websocketManager->handleConnectionRequest(conn_req, m_callerId, m_isSubscribe);
         }

cpp/csp/adapters/websocket/ClientConnectionRequestAdapter.h

@@ -22,7 +22,8 @@ class ClientConnectionRequestAdapter final: public OutputAdapter
         WebsocketEndpointManager * websocketManager,
         net::io_context& ioc,
         bool isSubscribe,
-        size_t callerId
+        size_t callerId,
+        boost::asio::strand<boost::asio::io_context::executor_type>& strand
     );
 
     void executeImpl() override;
@@ -32,6 +33,7 @@ class ClientConnectionRequestAdapter final: public OutputAdapter
 private:
     [[maybe_unused]] WebsocketEndpointManager* m_websocketManager;
     [[maybe_unused]] net::io_context& m_ioc;
+    boost::asio::strand<boost::asio::io_context::executor_type>& m_strand;
     bool m_isSubscribe;
     size_t m_callerId;
     bool m_checkPerformed;

cpp/csp/adapters/websocket/ClientHeaderUpdateAdapter.cpp

@@ -7,8 +7,9 @@ class WebsocketEndpointManager;
 ClientHeaderUpdateOutputAdapter::ClientHeaderUpdateOutputAdapter(
     Engine * engine,
     Dictionary& properties,
-    WebsocketEndpointManager * mgr
-) : OutputAdapter( engine ), m_properties( properties ), m_mgr( mgr )
+    WebsocketEndpointManager * mgr,
+    boost::asio::strand<boost::asio::io_context::executor_type>& strand
+) : OutputAdapter( engine ), m_properties( properties ), m_mgr( mgr ), m_strand( strand )
 { };
 
 void ClientHeaderUpdateOutputAdapter::executeImpl()
@@ -19,8 +20,10 @@ void ClientHeaderUpdateOutputAdapter::executeImpl()
             headers.update(update->key(), update->value());
         }
     }
-    auto endpoint = m_mgr -> getNonDynamicEndpoint();
-    endpoint -> updateHeaders(std::move(headers));
+    boost::asio::post(m_strand, [this, headers=std::move(headers)]() {
+        auto endpoint = m_mgr -> getNonDynamicEndpoint();
+        endpoint -> updateHeaders(std::move(headers));
+    });
 
     // Get the first e
     // ndpoint from the map and call updateHeaders

cpp/csp/adapters/websocket/ClientHeaderUpdateAdapter.h

@@ -19,7 +19,8 @@ class ClientHeaderUpdateOutputAdapter final: public OutputAdapter
     ClientHeaderUpdateOutputAdapter(
         Engine * engine,
         Dictionary& properties,
-        WebsocketEndpointManager * mgr
+        WebsocketEndpointManager * mgr,
+        boost::asio::strand<boost::asio::io_context::executor_type>& strand
     );
 
     void executeImpl() override;
@@ -29,6 +30,9 @@ class ClientHeaderUpdateOutputAdapter final: public OutputAdapter
 private:
     [[maybe_unused]] Dictionary& m_properties;
     WebsocketEndpointManager * m_mgr;
+    boost::asio::strand<boost::asio::io_context::executor_type>& m_strand;
+
+    
 
 };
 

cpp/csp/adapters/websocket/ClientOutputAdapter.cpp

@@ -6,19 +6,21 @@ ClientOutputAdapter::ClientOutputAdapter(
     Engine * engine,
     WebsocketEndpointManager * websocketManager,
     size_t caller_id,
-    net::io_context& ioc
+    net::io_context& ioc,
+    boost::asio::strand<boost::asio::io_context::executor_type>& strand
 ) : OutputAdapter( engine ), 
     m_websocketManager( websocketManager ),
     m_callerId( caller_id ),
-    m_ioc( ioc )
+    m_ioc( ioc ),
+    m_strand( strand )
 { };
 
 void ClientOutputAdapter::executeImpl()
 {
     // TODO Add here picking the right endpoints to send to
     // Based on the caller id
     const std::string & value = input() -> lastValueTyped<std::string>();
-    boost::asio::post(m_ioc, [this, value=value]() {
+    boost::asio::post(m_strand, [this, value=value]() {
         // something something lifetime? Not sure
         m_websocketManager->send(value, m_callerId);
     });

cpp/csp/adapters/websocket/ClientOutputAdapter.h

@@ -21,7 +21,8 @@ class ClientOutputAdapter final: public OutputAdapter
         Engine * engine,
         WebsocketEndpointManager * websocketManager,
         size_t caller_id,
-        net::io_context& ioc
+        net::io_context& ioc,
+        boost::asio::strand<boost::asio::io_context::executor_type>& strand
         // bool dynamic
     );
 
@@ -32,7 +33,8 @@ class ClientOutputAdapter final: public OutputAdapter
 private:
     WebsocketEndpointManager* m_websocketManager;
     size_t m_callerId;
-    net::io_context& m_ioc;
+    [[maybe_unused]] net::io_context& m_ioc;
+    boost::asio::strand<boost::asio::io_context::executor_type>& m_strand;
     // bool m_dynamic;
     // std::unordered_map<std::string, std::vector<bool>>& m_endpoint_consumers;
 };

cpp/csp/adapters/websocket/WebsocketEndpointManager.cpp

@@ -4,8 +4,10 @@
 namespace csp::adapters::websocket {
 
 WebsocketEndpointManager::WebsocketEndpointManager( ClientAdapterManager* mgr, const Dictionary & properties, Engine* engine ) 
-:   m_ioc(),
+:   m_num_threads( static_cast<size_t>(properties.get<int64_t>("num_threads")) ),
+    m_ioc( m_num_threads ),
     m_engine( engine ),
+    m_strand( boost::asio::make_strand(m_ioc) ),
     m_mgr( mgr ),
     m_active( false ), 
     m_shouldRun( false ), 
@@ -22,8 +24,7 @@ WebsocketEndpointManager::WebsocketEndpointManager( ClientAdapterManager* mgr, c
     // m_work_guard(properties.get<bool>("dynamic") ? 
     //     std::make_optional(boost::asio::make_work_guard(m_ioc)) : 
     //     std::nullopt),
-    m_dynamic( properties.get<bool>("dynamic") )
-{
+    m_dynamic( properties.get<bool>("dynamic") ){
     // Total number of subscribe and send function calls, set on the adapter manager
     // when is it created. Note, that some of the input adapters might have been
     // pruned from the graph and won't get created.
@@ -49,21 +50,29 @@ WebsocketEndpointManager::WebsocketEndpointManager( ClientAdapterManager* mgr, c
 void WebsocketEndpointManager::start(DateTime starttime, DateTime endtime) {
         // maybe restart here?
     m_shouldRun = true;
-    m_thread = std::make_unique<std::thread>([this]() {
-        m_ioc.reset();
-        if( !m_dynamic ){
-            boost::asio::post(m_ioc, [this]() {
-                // We subscribe for both the subscribe and send calls
-                // But we probably should check here.
-                if( m_outputAdapters.size() == 1)
-                    handleConnectionRequest(Dictionary(m_properties), 0, false);
-                // // If we have an input adapter call AND it's not pruned.
-                if( m_inputAdapters.size() == 1 && !adapterPruned(0))
-                    handleConnectionRequest(Dictionary(m_properties), 0, true);
-            });
-        }
-        m_ioc.run();
-    });
+    // std::vector<std::thread> threads;
+
+    m_ioc.reset();
+    if( !m_dynamic ){
+        boost::asio::post(m_strand, [this]() {
+            // We subscribe for both the subscribe and send calls
+            // But we probably should check here.
+            if( m_outputAdapters.size() == 1)
+                handleConnectionRequest(Dictionary(m_properties), 0, false);
+            // // If we have an input adapter call AND it's not pruned.
+            if( m_inputAdapters.size() == 1 && !adapterPruned(0))
+                handleConnectionRequest(Dictionary(m_properties), 0, true);
+        });
+    }
+    for (auto i = 0; i < m_num_threads; ++i) {
+        m_threads.emplace_back(std::make_unique<std::thread>([this]() {
+            m_ioc.run();
+        }));
+    }
+    // m_thread = std::make_unique<std::thread>([this]() {
+    //     // m_ioc.reset();
+    //     m_ioc.run();
+    // });
 };
 
 bool WebsocketEndpointManager::adapterPruned( size_t caller_id ){
@@ -403,7 +412,7 @@ void WebsocketEndpointManager::stop() {
     // Stop all endpoints
     // Endpoints running on m_ioc thread,
     // So we call stop there
-    boost::asio::post(m_ioc, [this]() {
+    boost::asio::post(m_strand, [this]() {
         for (auto& [endpoint_id, _] : m_endpoints) {
             // TODO ponder
             // Since this is called from the main thread,
@@ -422,6 +431,16 @@ void WebsocketEndpointManager::stop() {
     m_ioc.stop();
     m_cv.notify_one();
     if( m_thread ) m_thread->join();
+    
+    // Wait for all threads to finish
+    for (auto& thread : m_threads) {
+        if (thread && thread->joinable()) {
+            thread->join();
+        }
+    }
+    
+    // Clear threads before other members are destroyed
+    m_threads.clear();
 };
 
 PushInputAdapter* WebsocketEndpointManager::getInputAdapter(CspTypePtr & type, PushMode pushMode, const Dictionary & properties)
@@ -445,15 +464,15 @@ OutputAdapter* WebsocketEndpointManager::getOutputAdapter( const Dictionary & pr
     assert(!properties.get<bool>("is_subscribe"));
     assert(m_outputAdapters.size() == validated_id);
 
-    auto output_adapter = m_engine -> createOwnedObject<ClientOutputAdapter>( this, validated_id, m_ioc );
+    auto output_adapter = m_engine -> createOwnedObject<ClientOutputAdapter>( this, validated_id, m_ioc, m_strand );
     m_outputAdapters[validated_id] = output_adapter;
     return m_outputAdapters[validated_id];
 };
 
 OutputAdapter * WebsocketEndpointManager::getHeaderUpdateAdapter()
 {
     if (m_updateAdapter == nullptr)
-        m_updateAdapter = m_engine -> createOwnedObject<ClientHeaderUpdateOutputAdapter>( m_endpoint -> getProperties(), this );
+        m_updateAdapter = m_engine -> createOwnedObject<ClientHeaderUpdateOutputAdapter>( m_endpoint -> getProperties(), this, m_strand );
 
     return m_updateAdapter;
 };
@@ -465,7 +484,7 @@ OutputAdapter * WebsocketEndpointManager::getConnectionRequestAdapter( const Dic
     auto is_subscribe = properties.get<bool>("is_subscribe");
 
     auto* adapter = m_engine->createOwnedObject<ClientConnectionRequestAdapter>(
-        this, m_ioc, is_subscribe, caller_id
+        this, m_ioc, is_subscribe, caller_id, m_strand
     );
     m_connectionRequestAdapters.push_back(adapter);
 

cpp/csp/adapters/websocket/WebsocketEndpointManager.h

@@ -133,8 +133,10 @@ class WebsocketEndpointManager {
     using UriInfo = std::tuple<int32_t, std::unordered_set<uint64_t>>; //TODO remove
     using OptWorkGuard = std::optional<boost::asio::executor_work_guard<boost::asio::io_context::executor_type>>;
     std::unique_ptr<WebsocketEndpointManager> m_endpointManager;
+    size_t m_num_threads;
     net::io_context m_ioc;
     Engine* m_engine;
+    [[maybe_unused]] boost::asio::strand<boost::asio::io_context::executor_type> m_strand;
     ClientAdapterManager* m_mgr;
     [[maybe_unused]] bool m_active;
     [[maybe_unused]] std::atomic<bool> m_shouldRun{false};
@@ -145,6 +147,7 @@ class WebsocketEndpointManager {
     [[maybe_unused]] ClientOutputAdapter* m_outputAdapter;
     ClientHeaderUpdateOutputAdapter* m_updateAdapter;
     std::unique_ptr<std::thread> m_thread;
+    std::vector<std::unique_ptr<std::thread>> m_threads;
     Dictionary m_properties;
     // For each subscribe call, which uri's it is subscribed to
     std::vector<std::unordered_set<std::string>> m_subscribeFromUri;

csp/adapters/websocket.py

@@ -422,7 +422,7 @@ def __init__(
         headers: Optional[Dict[str, str]] = None,
         dynamic: bool = False,
         connection_request: Optional[ConnectionRequest] = None,
-        num_threads: int = 1,
+        num_threads: int = 2,
     ):
         """
         uri: str
@@ -433,8 +433,9 @@ def __init__(
             headers to apply to the request during the handshake
         dynamic: bool = False
             Whether we accept dynamically altering the connections via ConnectionRequest objects.
-        num_threads: int = 1
-            If in dynamic mode, determines number of threads to allocate for thread pool handling websocket endpoints.
+        num_threads: int = 2
+            Determines number of threads to allocate for thread pool handling websocket endpoints.
+            Defaults to 2 to avoid deadlocks and latency spikes
         """
 
         self._properties = dict(dynamic=dynamic, num_threads=num_threads)