diff --git a/shared/libraries/websocket_streaming/include/websocket_streaming/ws_streaming_device.h b/shared/libraries/websocket_streaming/include/websocket_streaming/ws_streaming_device.h index 05db676..93fffad 100644 --- a/shared/libraries/websocket_streaming/include/websocket_streaming/ws_streaming_device.h +++ b/shared/libraries/websocket_streaming/include/websocket_streaming/ws_streaming_device.h @@ -85,6 +85,7 @@ class WsStreamingDevice : public Device static PropertyObjectPtr createDefaultConfig(); void removed() override; + void removedNoLock() override; DeviceInfoPtr onGetInfo() override; diff --git a/shared/libraries/websocket_streaming/include/websocket_streaming/ws_streaming_listener.h b/shared/libraries/websocket_streaming/include/websocket_streaming/ws_streaming_listener.h index 0dd5ad2..d61ec37 100644 --- a/shared/libraries/websocket_streaming/include/websocket_streaming/ws_streaming_listener.h +++ b/shared/libraries/websocket_streaming/include/websocket_streaming/ws_streaming_listener.h @@ -16,6 +16,10 @@ #pragma once +#include + +#include + #include #include @@ -32,7 +36,8 @@ class WsStreamingListener WsStreamingListener( IContext *context, ISignal *signal, - wss::local_signal *localSignal); + std::shared_ptr localSignal, + boost::asio::any_io_executor executor); ~WsStreamingListener() override; @@ -50,11 +55,12 @@ class WsStreamingListener private: - SignalPtr _signal; - InputPortConfigPtr _port; - DataDescriptorPtr _lastDescriptor; - wss::local_signal& _localSignal; - bool _ruleType; + SignalPtr _signal; + InputPortConfigPtr _port; + DataDescriptorPtr _lastDescriptor; + std::shared_ptr _localSignal; + boost::asio::any_io_executor _executor; + bool _ruleType; }; END_NAMESPACE_OPENDAQ_WEBSOCKET_STREAMING diff --git a/shared/libraries/websocket_streaming/include/websocket_streaming/ws_streaming_server.h b/shared/libraries/websocket_streaming/include/websocket_streaming/ws_streaming_server.h index 80d48a4..11c0f72 100644 --- a/shared/libraries/websocket_streaming/include/websocket_streaming/ws_streaming_server.h +++ b/shared/libraries/websocket_streaming/include/websocket_streaming/ws_streaming_server.h @@ -115,12 +115,12 @@ class WsStreamingServer : public Server const std::string& name, const wss::metadata& metadata, SignalPtr openDaqSignal) - : localSignal(name, metadata) + : localSignal(std::make_shared(name, metadata)) , openDaqSignal(openDaqSignal) { } - wss::local_signal localSignal; + std::shared_ptr localSignal; SignalPtr openDaqSignal; InputPortNotificationsPtr listener; SignalPtr domainSignal; diff --git a/shared/libraries/websocket_streaming/src/ws_streaming_device.cpp b/shared/libraries/websocket_streaming/src/ws_streaming_device.cpp index ef5060c..0705742 100644 --- a/shared/libraries/websocket_streaming/src/ws_streaming_device.cpp +++ b/shared/libraries/websocket_streaming/src/ws_streaming_device.cpp @@ -83,9 +83,12 @@ PropertyObjectPtr WsStreamingDevice::createDefaultConfig() void WsStreamingDevice::removed() { streamingEvents.clear(); +} +void WsStreamingDevice::removedNoLock() +{ streaming.release(); - + auto lock = getRecursiveConfigLock2(); Device::removed(); } @@ -101,6 +104,9 @@ void WsStreamingDevice::onSignalAvailable( wss::remote_signal_ptr domainSignal, const DataDescriptorPtr& descriptor) { + auto lock = getRecursiveConfigLock2(); + if (this->objPtr.template asPtr(true).isRemoved() || !streaming.assigned()) + return; daq::MirroredSignalConfigPtr openDaqDomainSignal; if (domainSignal) @@ -134,6 +140,7 @@ void WsStreamingDevice::onSignalAvailable( void WsStreamingDevice::onSignalUnavailable(wss::remote_signal_ptr signal) { + auto lock = getRecursiveConfigLock2(); auto it = streamingSignals.find(signal->id()); if (it == streamingSignals.end()) return; diff --git a/shared/libraries/websocket_streaming/src/ws_streaming_listener.cpp b/shared/libraries/websocket_streaming/src/ws_streaming_listener.cpp index e0f9a04..98c7803 100644 --- a/shared/libraries/websocket_streaming/src/ws_streaming_listener.cpp +++ b/shared/libraries/websocket_streaming/src/ws_streaming_listener.cpp @@ -15,6 +15,9 @@ */ #include +#include + +#include #include @@ -34,7 +37,8 @@ BEGIN_NAMESPACE_OPENDAQ_WEBSOCKET_STREAMING WsStreamingListener::WsStreamingListener( IContext *context, ISignal *signal, - wss::local_signal *localSignal) + std::shared_ptr localSignal, + boost::asio::any_io_executor executor) : _signal(signal) , _port( InputPort( @@ -42,9 +46,13 @@ WsStreamingListener::WsStreamingListener( nullptr, String("ws-streaming"))) , _lastDescriptor(_signal.getDescriptor()) - , _localSignal(*localSignal) + , _localSignal(std::move(localSignal)) + , _executor(std::move(executor)) { - _localSignal.set_metadata( + // The listener is constructed on the streaming endpoint's strand (inside the on_subscribed handler) + // so touching _localSignal directly here is safe. All later access happens from + // packetReceived() on the acquisition thread and is send to _executor instead. + _localSignal->set_metadata( descriptorToMetadata( signal, _lastDescriptor)); @@ -109,22 +117,34 @@ void WsStreamingListener::onDataPacketReceived(DataPacketPtr packet) auto descriptor = packet.getDataDescriptor(); + // wss::local_signal/connection/peer are not thread-safe + // packetReceived() runs on the acquisition thread + // so marshal every local_signal call onto _executor if (descriptor != _lastDescriptor) { - _localSignal.set_metadata( - descriptorToMetadata( - _signal, - descriptor)); + auto metadata = descriptorToMetadata(_signal, descriptor); + boost::asio::post( + _executor, + [localSignal = _localSignal, metadata = std::move(metadata)]() + { + localSignal->set_metadata(metadata); + }); _lastDescriptor = descriptor; } if (packet.getRawDataSize()) - _localSignal.publish_data( - offset, - packet.getSampleCount(), - packet.getRawData(), - packet.getRawDataSize()); + // Capture the packet by value so its raw buffer stays alive until the handler runs + boost::asio::post( + _executor, + [localSignal = _localSignal, packet, offset]() + { + localSignal->publish_data( + offset, + packet.getSampleCount(), + packet.getRawData(), + packet.getRawDataSize()); + }); } void WsStreamingListener::onEventPacketReceived(EventPacketPtr packet) @@ -138,10 +158,14 @@ void WsStreamingListener::onEventPacketReceived(EventPacketPtr packet) if (valueDescriptorChanged && newValueDescriptor != _lastDescriptor) { - _localSignal.set_metadata( - descriptorToMetadata( - _signal, - newValueDescriptor)); + // Marshal onto the streaming endpoint's strand (see onDataPacketReceived). + auto metadata = descriptorToMetadata(_signal, newValueDescriptor); + boost::asio::post( + _executor, + [localSignal = _localSignal, metadata = std::move(metadata)]() + { + localSignal->set_metadata(metadata); + }); _lastDescriptor = newValueDescriptor; } diff --git a/shared/libraries/websocket_streaming/src/ws_streaming_server.cpp b/shared/libraries/websocket_streaming/src/ws_streaming_server.cpp index 109fbf6..f9c36ac 100644 --- a/shared/libraries/websocket_streaming/src/ws_streaming_server.cpp +++ b/shared/libraries/websocket_streaming/src/ws_streaming_server.cpp @@ -208,7 +208,7 @@ void WsStreamingServer::createListener(const SignalPtr& signal) if (domainSignal != it->second.domainSignal) { - _server.remove_local_signal(it->second.localSignal); + _server.remove_local_signal(*it->second.localSignal); _localSignals.erase(it); } @@ -230,7 +230,7 @@ void WsStreamingServer::createListener(const SignalPtr& signal) streamableSignal.domainSignal = domainSignal; - streamableSignal.localSignal.on_subscribed.connect([ + streamableSignal.localSignal->on_subscribed.connect([ =, &streamableSignal, signal_id = signal.getGlobalId().toStdString() @@ -239,11 +239,12 @@ void WsStreamingServer::createListener(const SignalPtr& signal) streamableSignal.listener = createWithImplementation( this->template thisPtr().getContext(), signal, - &streamableSignal.localSignal); + streamableSignal.localSignal, + _server.executor()); reinterpret_cast(streamableSignal.listener.getObject())->start(); }); - streamableSignal.localSignal.on_unsubscribed.connect([ + streamableSignal.localSignal->on_unsubscribed.connect([ =, &streamableSignal, signal_id = signal.getGlobalId().toStdString() @@ -252,7 +253,7 @@ void WsStreamingServer::createListener(const SignalPtr& signal) streamableSignal.listener.release(); }); - _server.add_local_signal(streamableSignal.localSignal); + _server.add_local_signal(*streamableSignal.localSignal); } void WsStreamingServer::onClientConnected( @@ -344,7 +345,7 @@ void WsStreamingServer::rescan() if (it->second.openDaqSignal.isRemoved()) { auto jt = it++; - _server.remove_local_signal(jt->second.localSignal); + _server.remove_local_signal(*jt->second.localSignal); _localSignals.erase(jt); }