Mirror/FairMQ
1
0
Fork 0
mirror of https://github.com/FairRootGroup/FairMQ.git synced 2025-10-16 18:11:49 +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 4eb70d6bf1
commit 7876798350
2 changed files with 11 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files

13
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()
{ {
for (auto& [transportType, transport] : fTransports) { {
transport->Reset(); lock_guard<mutex> lock(fTransportMtx);
for (auto& [transportType, transport] : fTransports) {
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;
}; };