Mirror/FairMQ
1
0
Fork 0
mirror of https://github.com/FairRootGroup/FairMQ.git synced 2025-10-13 08:41:16 +00:00
Code Issues Packages Projects Releases Wiki Activity

Device: synchronize access to transports container

Browse Source
This commit is contained in:
Alexey Rybalchenko 2023-02-22 18:36:16 +01:00
parent ae7bc6fc6c
commit 7bf1d368a8
2 changed files with 11 additions and 3 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

9
fairmq/Device.cxx
View File

@ -650,6 +650,8 @@ bool Device::HandleMultipartInput(const string& chName, const InputMultipartCall
shared_ptr<TransportFactory> Device::AddTransport(mq::Transport transport) shared_ptr<TransportFactory> Device::AddTransport(mq::Transport transport)
{ {
lock_guard<mutex> lock(fTransportMtx);
if (transport == mq::Transport::DEFAULT) { if (transport == mq::Transport::DEFAULT) {
transport = fDefaultTransportType; transport = fDefaultTransportType;
} }
@ -769,6 +771,7 @@ void Device::LogSocketRates()
void Device::InterruptTransports() void Device::InterruptTransports()
{ {
lock_guard<mutex> lock(fTransportMtx);
for (auto& [transportType, transport] : fTransports) { for (auto& [transportType, transport] : fTransports) {
transport->Interrupt(); transport->Interrupt();
} }
@ -776,6 +779,7 @@ void Device::InterruptTransports()
void Device::ResumeTransports() void Device::ResumeTransports()
{ {
lock_guard<mutex> lock(fTransportMtx);
for (auto& [transportType, transport] : fTransports) { for (auto& [transportType, transport] : fTransports) {
transport->Resume(); transport->Resume();
} }
@ -792,14 +796,17 @@ void Device::ResetTaskWrapper()
void Device::ResetWrapper() void Device::ResetWrapper()
{ {
{
lock_guard<mutex> lock(fTransportMtx);
for (auto& [transportType, transport] : fTransports) { for (auto& [transportType, transport] : fTransports) {
transport->Reset(); transport->Reset();
} }
fTransports.clear();
}
Reset(); Reset();
fChannels.clear(); fChannels.clear();
fTransports.clear();
fTransportFactory.reset(); fTransportFactory.reset();
if (!NewStatePending()) { if (!NewStatePending()) {
ChangeState(Transition::Auto); ChangeState(Transition::Auto);

1
fairmq/Device.h
View File

@ -637,6 +637,7 @@ class Device
StateQueue fStateQueue; StateQueue fStateQueue;
std::mutex fTransportMtx; ///< guards access to transports container
std::mutex fTransitionMtx; std::mutex fTransitionMtx;
bool fTransitioning; bool fTransitioning;
}; };