From bc6a4e1743076ad58137da2741bbc8dde2523743 Mon Sep 17 00:00:00 2001 From: Denis Erokhin Date: Thu, 30 Jul 2026 20:12:55 +0200 Subject: [PATCH 1/2] Expose RxBufferSize/TxBufferSize config properties, wire to ws-streaming ws-streaming's wss::detail::peer previously always allocated its compiled-in default buffer sizes (32 MiB tx, 1 MiB rx) for every connection, with no way to override them from openDAQ - see the companion change on ws-streaming's configurable-buffer-sizes branch, which adds rx_buffer_size/tx_buffer_size parameters to connection, client, and server. Add RxBufferSize/TxBufferSize Int properties to both WsStreamingServer::createDefaultConfig() and WsStreaming::createDefaultConfig(), and thread them through to the wss::server/wss::client construction call in each. Also wire config through to WsStreaming's constructor and onCreateStreaming(), which previously received the config parameter but silently discarded it. Defaults are chosen per role rather than reusing the old shared 32 MiB value: a server's traffic is transmit-heavy (streaming samples out to clients) so it gets the larger TxBufferSize (4 MiB) and smaller RxBufferSize (1 MiB); a client's traffic is receive-heavy (mostly receiving sample data, sending only small subscribe/RPC requests) so it gets the opposite - RxBufferSize 4 MiB, TxBufferSize 1 MiB. Verified with a custom LD_PRELOAD allocation tracer against quick_start_empty connecting to quick_start_simulator over daq.lt://: total bytes allocated during the addDevice() call dropped from 44.7 MB to 15.5 MB, with the two buffer allocations now showing up at exactly 4194304 and 1048576 bytes as expected, and the connection still completing successfully (all signals available/subscribed as before). --- ...websocket_streaming_client_module_impl.cpp | 2 +- .../websocket_streaming/ws_streaming.h | 16 ++++++++- .../websocket_streaming/src/ws_streaming.cpp | 33 +++++++++++++++++-- .../src/ws_streaming_server.cpp | 16 ++++++++- 4 files changed, 62 insertions(+), 5 deletions(-) diff --git a/modules/websocket_streaming_client_module/src/websocket_streaming_client_module_impl.cpp b/modules/websocket_streaming_client_module/src/websocket_streaming_client_module_impl.cpp index db65658..1119ad7 100644 --- a/modules/websocket_streaming_client_module/src/websocket_streaming_client_module_impl.cpp +++ b/modules/websocket_streaming_client_module/src/websocket_streaming_client_module_impl.cpp @@ -177,7 +177,7 @@ StreamingPtr WebsocketStreamingClientModule::onCreateStreaming(const StringPtr& DAQ_THROW_EXCEPTION(InvalidParameterException); const StringPtr str = formConnectionString(connectionString, config); - return createWithImplementation(str, context); + return createWithImplementation(str, context, config); } Bool WebsocketStreamingClientModule::onCompleteServerCapability(const ServerCapabilityPtr& source, const ServerCapabilityConfigPtr& target) diff --git a/shared/libraries/websocket_streaming/include/websocket_streaming/ws_streaming.h b/shared/libraries/websocket_streaming/include/websocket_streaming/ws_streaming.h index 8150c0c..9aa3aba 100644 --- a/shared/libraries/websocket_streaming/include/websocket_streaming/ws_streaming.h +++ b/shared/libraries/websocket_streaming/include/websocket_streaming/ws_streaming.h @@ -79,6 +79,17 @@ class WsStreaming : public Streaming */ static StreamingTypePtr createType(); + /*! + * @brief Merges the specified config with the default config, so that any properties + * missing from @p config are filled in with their default values. + * + * @param config The config to merge with the default config. May be null, in which case + * the default config is returned unmodified. + * @return The merged config. + */ + static PropertyObjectPtr populateDefaultConfig( + const PropertyObjectPtr& config); + public: /*! @@ -88,10 +99,13 @@ class WsStreaming : public Streaming * prefix. The remote peer address and TCP port number are parsed from the connection * string. * @param context The openDAQ context object. + * @param config The streaming configuration, as returned by createDefaultConfig(). May be + * null, in which case defaults are used for all properties. */ explicit WsStreaming( const StringPtr& connectionString, - const ContextPtr& context); + const ContextPtr& context, + const PropertyObjectPtr& config = nullptr); /*! * @brief Destroys a streaming object and stops the Boost.Asio I/O context's thread. diff --git a/shared/libraries/websocket_streaming/src/ws_streaming.cpp b/shared/libraries/websocket_streaming/src/ws_streaming.cpp index 5f2b5a2..da1cec9 100644 --- a/shared/libraries/websocket_streaming/src/ws_streaming.cpp +++ b/shared/libraries/websocket_streaming/src/ws_streaming.cpp @@ -51,10 +51,14 @@ StreamingTypePtr WsStreaming::createType() WsStreaming::WsStreaming( const StringPtr& connectionString, - const ContextPtr& context) + const ContextPtr& context, + const PropertyObjectPtr& config) : Streaming(connectionString, context, true) , ioContext{1} - , wsClient(ioContext.get_executor()) + , wsClient( + ioContext.get_executor(), + static_cast(static_cast(populateDefaultConfig(config).getPropertyValue("RxBufferSize"))), + static_cast(static_cast(populateDefaultConfig(config).getPropertyValue("TxBufferSize")))) { // The ws-streaming library wants a URL like ws://1.2.3.4:7418/foo. // So we simply need to replace the daq.lt:// prefix with ws://. @@ -97,9 +101,34 @@ PropertyObjectPtr WsStreaming::createDefaultConfig() { auto obj = PropertyObject(); obj.addProperty(IntProperty("Port", 7414)); + + // Upper bound, in bytes, on the size of a single frame this connection will receive/transmit; + // oversized frames cause the connection to be closed with an error rather than being + // buffered further. See wss::detail::peer::peer() for details. As a client, this connection + // mostly receives sample data and only sends small subscribe/RPC requests, so - unlike the + // server - it is the receive buffer that needs the larger default here. + obj.addProperty(IntPropertyBuilder("RxBufferSize", 4 * 1024 * 1024).setMinValue(1).build()); + obj.addProperty(IntPropertyBuilder("TxBufferSize", 1 * 1024 * 1024).setMinValue(1).build()); + return obj; } +PropertyObjectPtr WsStreaming::populateDefaultConfig(const PropertyObjectPtr& config) +{ + const auto defConfig = createDefaultConfig(); + if (!config.assigned()) + return defConfig; + + for (const auto& prop : defConfig.getAllProperties()) + { + const auto name = prop.getName(); + if (config.hasProperty(name)) + defConfig.setPropertyValue(name, config.getPropertyValue(name)); + } + + return defConfig; +} + void WsStreaming::onSetActive(bool active) { } diff --git a/shared/libraries/websocket_streaming/src/ws_streaming_server.cpp b/shared/libraries/websocket_streaming/src/ws_streaming_server.cpp index faabec4..3396c2b 100644 --- a/shared/libraries/websocket_streaming/src/ws_streaming_server.cpp +++ b/shared/libraries/websocket_streaming/src/ws_streaming_server.cpp @@ -56,6 +56,17 @@ PropertyObjectPtr WsStreamingServer::createDefaultConfig(const ContextPtr& conte defaultConfig.addProperty(StringProperty("Path", "/")); + // Upper bound, in bytes, on the size of a single frame a connection will receive/transmit; + // oversized frames cause the connection to be closed with an error rather than being + // buffered further. See wss::detail::peer::peer() for details. + const auto rxBufferSizeProp = + IntPropertyBuilder("RxBufferSize", 1 * 1024 * 1024).setMinValue(1).build(); + defaultConfig.addProperty(rxBufferSizeProp); + + const auto txBufferSizeProp = + IntPropertyBuilder("TxBufferSize", 4 * 1024 * 1024).setMinValue(1).build(); + defaultConfig.addProperty(txBufferSizeProp); + populateDefaultConfigFromProvider(context, defaultConfig); return defaultConfig; } @@ -102,7 +113,10 @@ WsStreamingServer::WsStreamingServer( : Server{ID, config, rootDevice, context} , _rootDevice{rootDevice} , _ioc{1} - , _server{_ioc.get_executor()} + , _server{ + _ioc.get_executor(), + static_cast(static_cast(config.getPropertyValue("RxBufferSize"))), + static_cast(static_cast(config.getPropertyValue("TxBufferSize")))} { _port = config.getPropertyValue("WebsocketStreamingPort"); From ded7f5c509c43a0c7f6deb51c42594d8d6deb8d5 Mon Sep 17 00:00:00 2001 From: Denis Erokhin Date: Fri, 31 Jul 2026 08:31:17 +0200 Subject: [PATCH 2/2] Point ws-streaming dependency at configurable-buffer-sizes branch Switches GIT_REF from the pinned commit to the configurable-buffer-sizes branch (github.com/openDAQ/ws-streaming/pull/new/configurable-buffer-sizes), which adds the rx_buffer_size/tx_buffer_size threading this change depends on. Drops PATCH_FILES since that branch's commit already includes 0001-Fixes-plus.patch's changes (it was branched from the already-patched working tree), so reapplying it would be redundant. Once the ws-streaming PR merges, this should be repinned to the merge commit hash instead of tracking a branch. --- external/ws-streaming/CMakeLists.txt | 4 +--- 1 file changed, 1 insertion(+), 3 deletions(-) diff --git a/external/ws-streaming/CMakeLists.txt b/external/ws-streaming/CMakeLists.txt index d9a5b0b..08b9a51 100644 --- a/external/ws-streaming/CMakeLists.txt +++ b/external/ws-streaming/CMakeLists.txt @@ -4,8 +4,6 @@ opendaq_dependency( NAME ws-streaming REQUIRED_VERSION 3.0.7 GIT_REPOSITORY https://github.com/openDAQ/ws-streaming - GIT_REF 9d591b39ec4845a9f65c558411d809800d455a5d + GIT_REF configurable-buffer-sizes EXPECT_TARGET ws-streaming::ws-streaming - PATCH_FILES - ${CMAKE_CURRENT_SOURCE_DIR}/patches/0001-Fixes-plus.patch )