Clean-up, fix configuring threads

Signed-off-by: Nijat K <[email protected]>

Author: Nijat K

cpp/csp/adapters/websocket/ClientAdapterManager.cpp

@@ -18,22 +18,7 @@ namespace csp::adapters::websocket {
 
 ClientAdapterManager::ClientAdapterManager( Engine* engine, const Dictionary & properties ) 
 : AdapterManager( engine ), 
-    // m_ioc(),
-    // m_active( false ), 
-    // m_shouldRun( false ), 
-    // m_endpoint(!properties.get<bool>("dynamic") ? 
-    //     std::make_unique<WebsocketEndpoint>( m_ioc, properties ) : 
-    //     nullptr),
-    // // m_endpoint( std::make_unique<WebsocketEndpoint>( m_ioc, properties ) ),
-    // m_inputAdapter( nullptr ), 
-    // m_outputAdapter( nullptr ),
-    // m_updateAdapter( nullptr ),
-    // m_thread( nullptr ), 
-    m_properties( properties )
-    // 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_properties( properties )
 { }
 
 ClientAdapterManager::~ClientAdapterManager()

cpp/csp/adapters/websocket/ClientAdapterManager.h

@@ -5,9 +5,7 @@
 #include <csp/adapters/websocket/WebsocketEndpointManager.h>
 #include <csp/adapters/websocket/WebsocketClientTypes.h>
 #include <csp/adapters/websocket/ClientInputAdapter.h>
-#include <csp/adapters/websocket/ClientOutputAdapter.h>
 #include <csp/adapters/websocket/ClientHeaderUpdateAdapter.h>
-#include <csp/adapters/websocket/ClientConnectionRequestAdapter.h>
 #include <csp/core/Enum.h>
 #include <csp/core/Hash.h>
 #include <csp/engine/AdapterManager.h>
@@ -26,50 +24,8 @@ namespace csp::adapters::websocket {
 
 using namespace csp;
 
-class ClientConnectionRequestAdapter;
-class ClientOutputAdapter;
 class WebsocketEndpointManager;
 
-
-// struct WebsocketClientStatusTypeTraits
-// {
-//     enum _enum : unsigned char
-//     {
-//         ACTIVE = 0,
-//         GENERIC_ERROR = 1,
-//         CONNECTION_FAILED = 2,
-//         CLOSED = 3,
-//         MESSAGE_SEND_FAIL = 4,
-
-//         NUM_TYPES
-//     };
-
-// protected:
-//     _enum m_value;
-// };
-
-// struct ConnectPayloads {
-//     std::vector<std::string> consumer_payloads;
-//     std::vector<std::string> producer_payloads;
-// };
-
-// struct EndpointConfig {
-//     std::chrono::milliseconds reconnect_interval;
-//     std::unique_ptr<boost::asio::steady_timer> reconnect_timer;
-//     bool attempting_reconnect{false};
-//     bool shutting_down{false};  // Track intentional shutdown
-    
-//     // Keep the payload storage
-//     std::vector<std::string> consumer_payloads;
-//     std::vector<std::string> producer_payloads;
-
-//     EndpointConfig(boost::asio::io_context& ioc) 
-//         : reconnect_timer(std::make_unique<boost::asio::steady_timer>(ioc))
-//     {}
-// };
-
-// using ClientStatusType = Enum<WebsocketClientStatusTypeTraits>;
-
 class ClientAdapterManager final : public AdapterManager
 {
 
@@ -96,57 +52,8 @@ class ClientAdapterManager final : public AdapterManager
     DateTime processNextSimTimeSlice( DateTime time ) override;
 
 private:
-    inline size_t validateCallerId(int64_t caller_id) const {
-        if (caller_id < 0) {
-            CSP_THROW(ValueError, "caller_id cannot be negative: " << caller_id);
-        }
-        return static_cast<size_t>(caller_id);
-    }
-
-    // need some client info
-    // This is a tuple of
-    // (number of send calls to uri, set of caller id's subscribed to uri )
-    // We use this information to keep track of how to route messages to/from
-    // uri's, and when a uri connection can be closed.
-    // 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>>;
     Dictionary m_properties;
     std::unique_ptr<WebsocketEndpointManager> m_endpointManager;
-    // net::io_context m_ioc;
-    // [[maybe_unused]] bool m_active;
-    // [[maybe_unused]] bool m_shouldRun;
-    // std::unique_ptr<WebsocketEndpoint> m_endpoint;  // TODO remove
-    // [[maybe_unused]] ClientInputAdapter* m_inputAdapter;
-    // [[maybe_unused]] ClientOutputAdapter* m_outputAdapter;
-    // ClientHeaderUpdateOutputAdapter* m_updateAdapter;
-    // std::unique_ptr<std::thread> m_thread;
-    // // For each subscribe call, which uri's it is subscribed to
-    // std::vector<std::unordered_set<std::string>> m_subscribeFromUri;
-    // // For each send call, which uri's it will send out to
-    // std::vector<std::unordered_set<std::string>> m_sendToUri;
-
-    // // uri -> (send_calls, set of caller id's for the subscribtions)
-    // // If send_calls is 0 (no adapter is sending out to that uri)
-    // // AND the subscriptions set is empty, we can then shutdown the encpoint.
-    // std::unordered_map<std::string, UriInfo> m_uriInfo; //TODO: remove
-    // //unclear if this is needed to be on the object
-    // std::vector<ClientConnectionRequestAdapter*> m_connectionRequestAdapters;
-
-    // // Bidirectional mapping using vectors since caller_ids are sequential
-    // // Maybe not efficient? Should be good for small number of edges though
-    // std::unordered_map<std::string, std::vector<bool>> m_endpoint_consumers;  // endpoint_id -> vector[caller_id] for consuemrs
-    // std::unordered_map<std::string, std::vector<bool>> m_endpoint_producers;  // endpoint_id -> vector[caller_id] for producers
-    
-    // // Quick lookup for caller's endpoints
-    // std::vector< std::unordered_set<std::string> > m_consumer_endpoints;  // caller_id -> set of endpoints they consume from
-    // std::vector< std::unordered_set<std::string> > m_producer_endpoints;  // caller_id -> set of endpoints they produce to
-    // OptWorkGuard m_work_guard;
-    // std::unordered_map<std::string, std::unique_ptr<WebsocketEndpoint>> m_endpoints;
-    // std::unordered_map<std::string, ConnectPayloads> m_connect_payloads;
-    // std::unordered_map<std::string, EndpointConfig> m_endpoint_configs;
-    // std::vector<ClientInputAdapter*> m_inputAdapters;
-    // std::vector<ClientOutputAdapter*> m_outputAdapters;
-    // bool m_dynamic;
 };
 
 }

cpp/csp/adapters/websocket/ClientHeaderUpdateAdapter.cpp

@@ -24,14 +24,6 @@ void ClientHeaderUpdateOutputAdapter::executeImpl()
         auto endpoint = m_mgr -> getNonDynamicEndpoint();
         endpoint -> updateHeaders(std::move(headers));
     });
-
-    // Get the first e
-    // ndpoint from the map and call updateHeaders
-    // DictionaryPtr headers = m_properties.get<DictionaryPtr>("headers");
-    // for( auto& update : input() -> lastValueTyped<std::vector<WebsocketHeaderUpdate::Ptr>>() )
-    // { 
-    //     if( update -> key_isSet() && update -> value_isSet() ) headers->update( update->key(), update->value() ); 
-    // }
 };
 
 }

cpp/csp/adapters/websocket/WebsocketEndpointManager.cpp

@@ -20,10 +20,7 @@ WebsocketEndpointManager::WebsocketEndpointManager( ClientAdapterManager* mgr, c
     m_updateAdapter( nullptr ),
     m_thread( nullptr ), 
     m_properties( properties ),
-    m_work_guard(std::make_optional(boost::asio::make_work_guard(m_ioc))),
-    // m_work_guard(properties.get<bool>("dynamic") ? 
-    //     std::make_optional(boost::asio::make_work_guard(m_ioc)) : 
-    //     std::nullopt),
+    m_work_guard(boost::asio::make_work_guard(m_ioc)),
     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
@@ -50,29 +47,23 @@ WebsocketEndpointManager::WebsocketEndpointManager( ClientAdapterManager* mgr, c
 void WebsocketEndpointManager::start(DateTime starttime, DateTime endtime) {
         // maybe restart here?
     m_shouldRun = true;
-    // 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 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) {
+    for (size_t 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 ){
@@ -145,12 +136,8 @@ void WebsocketEndpointManager::setupEndpoint(const std::string& endpoint_id,
                                             size_t validated_id) 
 {
     // Store the endpoint first
-    // std::cout << "INSIDE SETUPENDPOINT \n";
-    // m_endpoints[endpoint_id] = std::move(endpoint);
-
     auto& stored_endpoint = m_endpoints[endpoint_id] = std::move(endpoint);
 
-    // Do this directly to not get any issues with race conditions or something. Make this atomic
     stored_endpoint->setOnOpen([this, endpoint_id, endpoint = stored_endpoint.get(), payload=std::move(payload), persist, is_consumer, validated_id]() {
         auto [iter, inserted] = m_endpoint_configs.try_emplace(endpoint_id, m_ioc);
         auto& config = iter->second;
@@ -422,8 +409,7 @@ void WebsocketEndpointManager::stop() {
         }
     });
     // Stop the work guard to allow the io_context to complete
-    if (m_work_guard )
-        m_work_guard.reset();
+    m_work_guard.reset();
     // }
     // else{
     //     if( m_active ) m_endpoint->stop( false );

cpp/csp/adapters/websocket/WebsocketEndpointManager.h

@@ -131,7 +131,6 @@ 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;
@@ -169,7 +168,7 @@ class WebsocketEndpointManager {
     // Quick lookup for caller's endpoints
     std::vector< std::unordered_set<std::string> > m_consumer_endpoints;  // caller_id -> set of endpoints they consume from
     std::vector< std::unordered_set<std::string> > m_producer_endpoints;  // caller_id -> set of endpoints they produce to
-    OptWorkGuard m_work_guard;
+    boost::asio::executor_work_guard<boost::asio::io_context::executor_type> m_work_guard;
     std::unordered_map<std::string, std::unique_ptr<WebsocketEndpoint>> m_endpoints;
     std::unordered_map<std::string, ConnectPayloads> m_connect_payloads;
     std::unordered_map<std::string, EndpointConfig> m_endpoint_configs;

csp/adapters/websocket.py

@@ -463,26 +463,6 @@ def __init__(
     def _dynamic(self):
         return self._properties.get("dynamic", False)
 
-    # @staticmethod
-    # def to_internal_connection_request(conn_req: ConnectionRequest) -> InternalConnectionRequest:
-    #     # TODO Use?
-    #     uri = conn_req.uri
-    #     reconnect_interval = conn_req.reconnect_interval
-    #     resp = urllib.parse.urlparse(uri)
-    #     if resp.hostname is None:
-    #         raise ValueError(f"Failed to parse host from URI: {uri}")
-
-    #     assert reconnect_interval >= timedelta(seconds=1)
-    #     conn_req_dict = ConnectionRequest.to_dict()
-    #     update_dict = dict(
-    #         use_ssl=uri.startswith("wss"),
-    #         route=resp.path or "/",  # resource shouldn't be empty string
-    #         host=resp.hostname,
-    #         port=_sanitize_port(uri, resp.port),
-    #     )
-    #     conn_req_dict.update(update_dict)
-    #     return InternalConnectionRequest(**conn_req_dict)
-
     def _get_properties(self, conn_request: ConnectionRequest) -> dict:
         uri = conn_request.uri
         reconnect_interval = conn_request.reconnect_interval
@@ -554,22 +534,12 @@ def subscribe(
         caller_id = self._get_caller_id(is_subscribe=True)
         # Gives validation, more to start defining a common interface
         adapter_props = AdapterInfo(caller_id=caller_id, is_subscribe=True).model_dump()
-        # if connection_request is None and self._dynamic:
-        #     raise ValueError(
-        #         "'connection_request' must not be None if this adapter is dynamic. Use 'csp.null_ts(ConnectionRequest)' if this was intentional"
-        #     )
         connection_request = csp.null_ts(List[ConnectionRequest]) if connection_request is None else connection_request
-        # if connection_request is not None:
-        # request_dict = self._enrich_with_caller_id(connection_request, caller_id, is_subscribe=True)
         request_dict = csp.apply(
             connection_request, lambda conn_reqs: [self._get_properties(conn_req) for conn_req in conn_reqs], list
         )
-        # We just declare it here, so it gets included in the graph
-        # since nothing is returned.
-        # csp.print("req dict", request_dict)
+        # Output adapter to handle connection requests
         _websocket_connection_request_adapter_def(self, request_dict, adapter_props)
-        # If in dynamic mode, we wrap the message in a struct to include the url the data
-        # is coming from
 
         field_map = field_map or {}
         meta_field_map = meta_field_map or {}
@@ -584,6 +554,7 @@ def subscribe(
         properties["meta_field_map"] = meta_field_map
 
         properties.update(adapter_props)
+        # We wrap the message in a struct to note the url it comes from
         if self._dynamic:
             ts_type = self.get_wrapper_struct(ts_type=ts_type)
         return _websocket_input_adapter_def(self, ts_type, properties, push_mode=push_mode)
@@ -593,15 +564,6 @@ def send(self, x: ts["T"], connection_request: Optional[ts[List[ConnectionReques
         # Gives validation, more to start defining a common interface
         adapter_props = AdapterInfo(caller_id=caller_id, is_subscribe=False).model_dump()
         connection_request = csp.null_ts(List[ConnectionRequest]) if connection_request is None else connection_request
-        # if connection_request is None and self._dynamic:
-        #     raise ValueError(
-        #         "'connection_request' must not be None if this adapter is dynamic. Use 'csp.null_ts(ConnectionRequest)' if this was intentional"
-        #     )
-
-        # TODO:
-        # Consider allowing header updates go through here
-        # In non-dynamic mode
-        # if connection_request is not None:
         request_dict = csp.apply(
             connection_request, lambda conn_reqs: [self._get_properties(conn_req) for conn_req in conn_reqs], list
         )