Source · fairmq/
Device.cxxsource
Described in Channels and sockets, Devices, Device states, Threads in a device, Transports, 1-1: sampler and sink, 1-n-1: sampler, processors and sink, copypush: one message to several receivers, dds: run with DDS, multipart: several parts in one message, multiple-channels: data and broadcast, multiple-transports: zeromq and shmem, n-m: n senders to m receivers, qc: quality control on demand, readout: a readout chain, region: unmanaged regions, req-rep: request and reply, Run a benchmark, Configure channels, Control a device from code, Design a topology, Stop a topology without losing messages, Handle errors and timeouts, Move from the FairMQ* names, Poll several channels, Receive data with callbacks or loops, Send your own data types, Use several transports, States and transitions, Transfer codes and errors, Your first device, Overview.
1/********************************************************************************2 * Copyright (C) 2012-2023 GSI Helmholtzzentrum fuer Schwerionenforschung GmbH *3 * *4 * This software is distributed under the terms of the *5 * GNU Lesser General Public Licence (LGPL) version 3, *6 * copied verbatim in the file "LICENSE" *7 ********************************************************************************/8 9// FairMQ10#include <fairmq/Device.h>11#include <fairmq/Tools.h>12#include <fairmq/Transports.h>13 14// boost15#include <boost/algorithm/string.hpp> // join/split16 17// std18#include <algorithm> // std::max, std::any_of19#include <chrono>20#include <iomanip>21#include <list>22#include <memory> // std::make_unique23#include <mutex>24#include <thread>25 26namespace fair::mq {27 28using namespace std;29 30constexpr const char* Device::DefaultId;31constexpr int Device::DefaultIOThreads;32constexpr const char* Device::DefaultTransportName;33constexpr mq::Transport Device::DefaultTransportType;34constexpr const char* Device::DefaultNetworkInterface;35constexpr int Device::DefaultInitTimeout;36constexpr float Device::DefaultRate;37constexpr const char* Device::DefaultSession;38 39struct StateSubscription40{41 StateMachine& fStateMachine;42 StateQueue& fStateQueue;43 string fId;44 45 explicit StateSubscription(string id, StateMachine& stateMachine, StateQueue& stateQueue)46 : fStateMachine(stateMachine)47 , fStateQueue(stateQueue)48 , fId(std::move(id))49 {50 fStateMachine.SubscribeToStateChange(fId, [&](State state) {51 fStateQueue.Push(state);52 });53 }54 55 StateSubscription(const StateSubscription&) = delete;56 StateSubscription(StateSubscription&&) = delete;57 StateSubscription& operator=(const StateSubscription&) = delete;58 StateSubscription& operator=(StateSubscription&&) = delete;59 60 ~StateSubscription() {61 fStateMachine.UnsubscribeFromStateChange(fId);62 }63};64 65Device::Device()66 : Device(nullptr, {0, 0, 0})67{}68 69Device::Device(ProgOptions& config)70 : Device(&config, {0, 0, 0})71{}72 73Device::Device(tools::Version version)74 : Device(nullptr, version)75{}76 77Device::Device(ProgOptions& config, tools::Version version)78 : Device(&config, version)79{}80 81/// TODO: Remove this once Device::fChannels is no longer public82#pragma GCC diagnostic push83#pragma GCC diagnostic ignored "-Wdeprecated-declarations"84Device::Device(ProgOptions* config, tools::Version version)85 : fTransportFactory(nullptr)86 , fInternalConfig(config ? nullptr : make_unique<ProgOptions>())87 , fConfig(config ? config : fInternalConfig.get())88 , fId(DefaultId)89 , fDefaultTransportType(DefaultTransportType)90 , fDataCallbacks(false)91 , fMultitransportProceed(false)92 , fVersion(version)93 , fRate(DefaultRate)94 , fInitializationTimeoutInS(DefaultInitTimeout)95{96 SubscribeToNewTransition("device", [&](Transition transition) {97 LOG(trace) << "device notified on new transition: " << transition;98 InterruptTransports();99 });100 101 fStateMachine.PrepareState([&](State state) {102 LOG(trace) << "Resuming transports for " << state << " state";103 ResumeTransports();104 });105 106 fStateMachine.HandleStates([&](State state) {107 LOG(trace) << "device notified on new state: " << state;108 109 fStateQueue.Push(state);110 111 switch (state) {112 case State::InitializingDevice:113 InitWrapper();114 break;115 case State::Binding:116 BindWrapper();117 break;118 case State::Connecting:119 ConnectWrapper();120 break;121 case State::InitializingTask:122 InitTaskWrapper();123 break;124 case State::Running:125 RunWrapper();126 break;127 case State::ResettingTask:128 ResetTaskWrapper();129 break;130 case State::ResettingDevice:131 ResetWrapper();132 break;133 case State::Exiting:134 Exit();135 break;136 default:137 LOG(trace) << "device notified on new state without a matching handler: " << state;138 break;139 }140 });141 142 fStateMachine.Start();143}144#pragma GCC diagnostic pop145 146void Device::InitWrapper()147{148 // run initialization once CompleteInit transition is requested149 fStateMachine.WaitForPendingState();150 151 fId = fConfig->GetProperty<string>("id", DefaultId);152 153 Init();154 155 fRate = fConfig->GetProperty<float>("rate", DefaultRate);156 fInitializationTimeoutInS = fConfig->GetProperty<int>("init-timeout", DefaultInitTimeout);157 158 try {159 fDefaultTransportType = TransportTypes.at(fConfig->GetProperty<string>("transport", DefaultTransportName));160 } catch (const exception& e) {161 LOG(error) << "exception: " << e.what();162 LOG(error) << "invalid transport type provided: " << fConfig->GetProperty<string>("transport", "not provided");163 throw;164 }165 166 unordered_map<string, int> infos = fConfig->GetChannelInfo();167 for (const auto& info : infos) {168 for (int i = 0; i < info.second; ++i) {169 GetChannels()[info.first].emplace_back(info.first, i, fConfig->GetPropertiesStartingWith(tools::ToString("chans.", info.first, ".", i, ".")));170 }171 }172 173 LOG(debug) << "Setting '" << TransportNames.at(fDefaultTransportType) << "' as default transport for the device";174 fTransportFactory = AddTransport(fDefaultTransportType);175 176 string networkInterface = fConfig->GetProperty<string>("network-interface", DefaultNetworkInterface);177 178 // Fill the uninitialized channel containers179 for (auto& channel : GetChannels()) {180 for (auto& subChannel : channel.second) {181 // set channel transport182 LOG(debug) << "Initializing transport for channel " << subChannel.fName << ": " << TransportNames.at(subChannel.fTransportType);183 subChannel.InitTransport(AddTransport(subChannel.fTransportType));184 185 if (subChannel.fMethod == "bind") {186 // if binding address is not specified, try getting it from the configured network interface187 if (subChannel.fAddress == "unspecified" || subChannel.fAddress.empty()) {188 // if the configured network interface is default, get its name from the default route189 try {190 if (networkInterface == "default") {191 networkInterface = tools::getDefaultRouteNetworkInterface();192 }193 subChannel.fAddress = "tcp://" + tools::getInterfaceIP(networkInterface) + ":1";194 } catch(const tools::DefaultRouteDetectionError& e) {195 LOG(debug) << "binding on tcp://*:1";196 subChannel.fAddress = "tcp://*:1";197 }198 }199 // fill the uninitialized list200 fUninitializedBindingChannels.push_back(&subChannel);201 } else if (subChannel.fMethod == "connect") {202 // fill the uninitialized list203 fUninitializedConnectingChannels.push_back(&subChannel);204 } else if (subChannel.fAddress.find_first_of("@+>") != string::npos) {205 // fill the uninitialized list206 fUninitializedConnectingChannels.push_back(&subChannel);207 } else {208 LOG(error) << "Cannot update configuration. Socket method (bind/connect) for channel '" << subChannel.fName << "' not specified.";209 throw runtime_error(tools::ToString("Cannot update configuration. Socket method (bind/connect) for channel ", subChannel.fName, " not specified."));210 }211 }212 }213 214 // ChangeStateOrThrow(Transition::Auto);215}216 217void Device::BindWrapper()218{219 // Bind channels. Here one run is enough, because bind settings should be available locally220 // If necessary this could be handled in the same way as the connecting channels221 AttachChannels(fUninitializedBindingChannels);222 223 if (!fUninitializedBindingChannels.empty()) {224 LOG(error) << fUninitializedBindingChannels.size() << " of the binding channels could not initialize. Initial configuration incomplete.";225 throw runtime_error(tools::ToString(fUninitializedBindingChannels.size(), " of the binding channels could not initialize. Initial configuration incomplete."));226 }227 228 Bind();229 230 if (!NewStatePending()) {231 ChangeStateOrThrow(Transition::Auto);232 }233}234 235void Device::ConnectWrapper()236{237 // go over the list of channels until all are initialized (and removed from the uninitialized list)238 int numAttempts = 1;239 auto sleepTimeInMS = 50;240 auto maxAttempts = fInitializationTimeoutInS * 1000 / sleepTimeInMS;241 // first attempt242 AttachChannels(fUninitializedConnectingChannels);243 // if not all channels could be connected, update their address values from config and retry244 while (!fUninitializedConnectingChannels.empty() && !NewStatePending()) {245 this_thread::sleep_for(chrono::milliseconds(sleepTimeInMS));246 247 for (auto& chan : fUninitializedConnectingChannels) {248 string key{"chans." + chan->GetPrefix() + "." + chan->GetIndex() + ".address"};249 string newAddress = fConfig->GetProperty<string>(key);250 if (newAddress != chan->GetAddress()) {251 chan->UpdateAddress(newAddress);252 }253 }254 255 if (numAttempts++ > maxAttempts) {256 LOG(error) << "could not connect all channels after " << fInitializationTimeoutInS << " attempts";257 LOG(error) << "following channels are still invalid:";258 for (auto& chan : fUninitializedConnectingChannels) {259 LOG(error) << "channel: " << *chan;260 }261 throw runtime_error(tools::ToString("could not connect all channels after ", fInitializationTimeoutInS, " attempts"));262 }263 264 AttachChannels(fUninitializedConnectingChannels);265 }266 267 if (GetChannels().empty()) {268 LOG(warn) << "No channels created after finishing initialization";269 }270 271 Connect();272 273 if (!NewStatePending()) {274 ChangeStateOrThrow(Transition::Auto);275 }276}277 278void Device::AttachChannels(vector<Channel*>& chans)279{280 auto itr = chans.begin();281 282 while (itr != chans.end()) {283 if ((*itr)->Validate()) {284 (*itr)->Init();285 if (AttachChannel(**itr)) {286 // remove the channel from the uninitialized container287 itr = chans.erase(itr);288 } else {289 LOG(error) << "failed to attach channel " << (*itr)->fName << " (" << (*itr)->fMethod << " on " << (*itr)->fAddress << ")";290 ++itr;291 }292 } else {293 ++itr;294 }295 }296}297 298bool Device::AttachChannel(Channel& chan)299{300 vector<string> endpoints;301 string chanAddress = chan.GetAddress();302 boost::algorithm::split(endpoints, chanAddress, boost::algorithm::is_any_of(","));303 304 for (auto& endpoint : endpoints) {305 // attach306 bool bind = (chan.GetMethod() == "bind");307 bool connectionModifier = false;308 string address = endpoint;309 310 // check if the default fMethod is overridden by a modifier311 if (endpoint[0] == '+' || endpoint[0] == '>') {312 connectionModifier = true;313 bind = false;314 address = endpoint.substr(1);315 } else if (endpoint[0] == '@') {316 connectionModifier = true;317 bind = true;318 address = endpoint.substr(1);319 }320 321 if (address.compare(0, 6, "tcp://") == 0) {322 string addressString = address.substr(6);323 auto pos = addressString.find(':');324 string hostPart = addressString.substr(0, pos);325 if (!(bind && hostPart == "*")) {326 string portPart = addressString.substr(pos + 1);327 string resolvedHost = tools::getIpFromHostname(hostPart);328 if (resolvedHost.empty()) {329 return false;330 }331 address.assign("tcp://" + resolvedHost + ":" + portPart);332 }333 }334 335 bool success = true;336 // make the connection337 if (bind) {338 success = chan.BindEndpoint(address);339 } else {340 success = chan.ConnectEndpoint(address);341 }342 343 // bind might bind to an address different than requested,344 // put the actual address back in the config345 endpoint.clear();346 if (connectionModifier) {347 endpoint.push_back(bind?'@':'+');348 }349 endpoint += address;350 351 // after the book keeping is done, exit in case of errors352 if (!success) {353 return success;354 } else {355 LOG(debug) << "Attached channel " << chan.GetName() << " to " << endpoint << (bind ? " (bind) " : " (connect) ") << "(" << chan.GetType() << ")";356 }357 }358 359 // put the (possibly) modified address back in the channel object and config360 string newAddress(boost::algorithm::join(endpoints, ","));361 if (newAddress != chanAddress) {362 chan.UpdateAddress(newAddress);363 364 // update address in the config, it could have been modified during binding365 fConfig->SetProperty({"chans." + chan.GetPrefix() + "." + chan.GetIndex() + ".address"}, newAddress);366 }367 368 return true;369}370 371void Device::InitTaskWrapper()372{373 InitTask();374 375 if (!NewStatePending()) {376 ChangeStateOrThrow(Transition::Auto);377 }378}379 380void Device::RunWrapper()381{382 LOG(info) << "fair::mq::Device running...";383 384 unique_ptr<thread> rateLogger;385 // Check if rate logging thread is needed386 const bool rateLogging = any_of(GetChannels().cbegin(), GetChannels().cend(), [](auto ch) {387 return any_of(ch.second.cbegin(), ch.second.cend(), [](auto sub) { return sub.fRateLogging > 0; });388 });389 390 if (rateLogging) {391 rateLogger = make_unique<thread>(&Device::LogSocketRates, this);392 }393 tools::CallOnDestruction joinRateLogger([&](){394 if (rateLogging && rateLogger->joinable()) { rateLogger->join(); }395 });396 397 // change to Error state in case of an exception, to release LogSocketRates398 tools::CallOnDestruction cod([&](){399 ChangeStateOrThrow(Transition::ErrorFound);400 });401 402 PreRun();403 404 // process either data callbacks or ConditionalRun/Run405 if (fDataCallbacks) {406 // if only one input channel, do lightweight handling without additional polling.407 if (fInputChannelKeys.size() == 1 && GetChannels().at(fInputChannelKeys.at(0)).size() == 1) {408 HandleSingleChannelInput();409 } else {// otherwise do full handling with polling410 HandleMultipleChannelInput();411 }412 } else {413 tools::RateLimiter rateLimiter(fRate);414 415 while (!NewStatePending() && ConditionalRun()) {416 if (fRate > 0.001) {417 rateLimiter.maybe_sleep();418 }419 }420 421 Run();422 }423 424 // if Run() exited and the state is still RUNNING, transition to READY.425 if (!NewStatePending()) {426 ChangeStateOrThrow(Transition::Stop);427 }428 429 PostRun();430 431 cod.disable();432}433 434void Device::HandleSingleChannelInput()435{436 bool proceed = true;437 438 if (!fMsgInputs.empty()) {439 while (!NewStatePending() && proceed) {440 proceed = HandleMsgInput(fInputChannelKeys.at(0), fMsgInputs.begin()->second, 0);441 }442 } else if (!fMultipartInputs.empty()) {443 while (!NewStatePending() && proceed) {444 proceed = HandleMultipartInput(fInputChannelKeys.at(0), fMultipartInputs.begin()->second, 0);445 }446 }447}448 449void Device::HandleMultipleChannelInput()450{451 // check if more than one transport is used452 fMultitransportInputs.clear();453 for (const auto& k : fInputChannelKeys) {454 mq::Transport t = GetChannel(k, 0).fTransportType;455 if (fMultitransportInputs.find(t) == fMultitransportInputs.end()) {456 fMultitransportInputs.insert(pair<mq::Transport, vector<string>>(t, vector<string>()));457 fMultitransportInputs.at(t).push_back(k);458 } else {459 fMultitransportInputs.at(t).push_back(k);460 }461 }462 463 for (const auto& mi : fMsgInputs) {464 for (auto& i : GetChannels().at(mi.first)) {465 i.fMultipart = false;466 }467 }468 469 for (const auto& mi : fMultipartInputs) {470 for (auto& i : GetChannels().at(mi.first)) {471 i.fMultipart = true;472 }473 }474 475 // if more than one transport is used, handle poll of each in a separate thread476 if (fMultitransportInputs.size() > 1) {477 HandleMultipleTransportInput();478 } else { // otherwise poll directly479 bool proceed = true;480 481 PollerPtr poller(GetChannel(fInputChannelKeys.at(0), 0).fTransportFactory->CreatePoller(GetChannels(), fInputChannelKeys));482 483 while (!NewStatePending() && proceed) {484 poller->Poll(200);485 486 // check which inputs are ready and call their data handlers if they are.487 for (const auto& ch : fInputChannelKeys) {488 for (unsigned int i = 0; i < GetChannels().at(ch).size(); ++i) {489 if (poller->CheckInput(ch, i)) {490 if (GetChannel(ch, i).fMultipart) {491 proceed = HandleMultipartInput(ch, fMultipartInputs.at(ch), i);492 } else {493 proceed = HandleMsgInput(ch, fMsgInputs.at(ch), i);494 }495 496 if (!proceed) {497 break;498 }499 }500 }501 if (!proceed) {502 break;503 }504 }505 }506 }507}508 509void Device::HandleMultipleTransportInput()510{511 vector<thread> threads;512 513 fMultitransportProceed = true;514 515 for (const auto& i : fMultitransportInputs) {516 threads.emplace_back(517 &Device::PollForTransport, this, fTransports.at(i.first).get(), i.second);518 }519 520 for (thread& t : threads) {521 t.join();522 }523}524 525void Device::PollForTransport(const TransportFactory* factory, const vector<string>& channelKeys)526{527 try {528 PollerPtr poller(factory->CreatePoller(GetChannels(), channelKeys));529 530 while (!NewStatePending() && fMultitransportProceed) {531 poller->Poll(500);532 533 for (const auto& ch : channelKeys) {534 for (unsigned int i = 0; i < GetChannels().at(ch).size(); ++i) {535 if (poller->CheckInput(ch, i)) {536 lock_guard<mutex> lock(fMultitransportMutex);537 538 if (!fMultitransportProceed) {539 break;540 }541 542 if (GetChannel(ch, i).fMultipart) {543 fMultitransportProceed = HandleMultipartInput(ch, fMultipartInputs.at(ch), i);544 } else {545 fMultitransportProceed = HandleMsgInput(ch, fMsgInputs.at(ch), i);546 }547 548 if (!fMultitransportProceed) {549 break;550 }551 }552 }553 if (!fMultitransportProceed) {554 break;555 }556 }557 }558 } catch (exception& e) {559 LOG(error) << "fair::mq::Device::PollForTransport() failed: " << e.what() << ", going to ERROR state.";560 throw runtime_error(tools::ToString("fair::mq::Device::PollForTransport() failed: ", e.what(), ", going to ERROR state."));561 }562}563 564bool Device::HandleMsgInput(const string& chName, const InputMsgCallback& callback, int i)565{566 unique_ptr<Message> input(GetChannel(chName, i).fTransportFactory->CreateMessage());567 568 if (Receive(input, chName, i) >= 0) {569 return callback(input, i);570 } else {571 return false;572 }573}574 575bool Device::HandleMultipartInput(const string& chName, const InputMultipartCallback& callback, int i)576{577 Parts input;578 579 if (Receive(input, chName, i) >= 0) {580 return callback(input, i);581 } else {582 return false;583 }584}585 586shared_ptr<TransportFactory> Device::AddTransport(mq::Transport transport)587{588 lock_guard<mutex> lock(fTransportMtx);589 590 if (transport == mq::Transport::DEFAULT) {591 transport = fDefaultTransportType;592 }593 594 auto i = fTransports.find(transport);595 596 if (i == fTransports.end()) {597 LOG(debug) << "Adding '" << TransportNames.at(transport) << "' transport";598 auto tr = TransportFactory::CreateTransportFactory(TransportNames.at(transport), fId, fConfig);599 fTransports.insert({transport, tr});600 return tr;601 } else {602 LOG(debug) << "Reusing existing '" << TransportNames.at(transport) << "' transport";603 return i->second;604 }605}606 607void Device::SetConfig(ProgOptions& config)608{609 fInternalConfig.reset();610 fConfig = &config;611}612 613void Device::LogSocketRates()614{615 vector<Channel*> filteredChannels;616 vector<string> filteredChannelNames;617 vector<int> logIntervals;618 vector<int> intervalCounters;619 620 size_t chanNameLen = 0;621 622 // iterate over the channels map623 for (auto& channel : GetChannels()) {624 // iterate over the channels vector625 for (auto& subChannel : channel.second) {626 if (subChannel.fRateLogging > 0) {627 filteredChannels.push_back(&subChannel);628 logIntervals.push_back(subChannel.fRateLogging);629 intervalCounters.push_back(0);630 filteredChannelNames.push_back(subChannel.GetName());631 chanNameLen = max(chanNameLen, filteredChannelNames.back().length());632 }633 }634 }635 636 vector<unsigned long> bytesIn(filteredChannels.size());637 vector<unsigned long> msgIn(filteredChannels.size());638 vector<unsigned long> bytesOut(filteredChannels.size());639 vector<unsigned long> msgOut(filteredChannels.size());640 641 vector<unsigned long> bytesInNew(filteredChannels.size());642 vector<unsigned long> msgInNew(filteredChannels.size());643 vector<unsigned long> bytesOutNew(filteredChannels.size());644 vector<unsigned long> msgOutNew(filteredChannels.size());645 646 vector<double> mbPerSecIn(filteredChannels.size());647 vector<double> msgPerSecIn(filteredChannels.size());648 vector<double> mbPerSecOut(filteredChannels.size());649 vector<double> msgPerSecOut(filteredChannels.size());650 651 int i = 0;652 for (const auto& channel : filteredChannels) {653 bytesIn.at(i) = channel->GetBytesRx();654 bytesOut.at(i) = channel->GetBytesTx();655 msgIn.at(i) = channel->GetMessagesRx();656 msgOut.at(i) = channel->GetMessagesTx();657 ++i;658 }659 660 chrono::time_point<chrono::high_resolution_clock> t0(chrono::high_resolution_clock::now());661 chrono::time_point<chrono::high_resolution_clock> t1;662 663 while (!NewStatePending()) {664 WaitFor(chrono::seconds(1));665 666 t1 = chrono::high_resolution_clock::now();667 668 uint64_t msSinceLastLog = chrono::duration_cast<chrono::milliseconds>(t1 - t0).count();669 670 i = 0;671 672 for (const auto& channel : filteredChannels) {673 intervalCounters.at(i)++;674 675 if (intervalCounters.at(i) == logIntervals.at(i)) {676 intervalCounters.at(i) = 0;677 678 if (msSinceLastLog > 0) {679 bytesInNew.at(i) = channel->GetBytesRx();680 msgInNew.at(i) = channel->GetMessagesRx();681 bytesOutNew.at(i) = channel->GetBytesTx();682 msgOutNew.at(i) = channel->GetMessagesTx();683 684 mbPerSecIn.at(i) = (static_cast<double>(bytesInNew.at(i) - bytesIn.at(i)) / (1000. * 1000.)) / static_cast<double>(msSinceLastLog) * 1000.;685 msgPerSecIn.at(i) = static_cast<double>(msgInNew.at(i) - msgIn.at(i)) / static_cast<double>(msSinceLastLog) * 1000.;686 mbPerSecOut.at(i) = (static_cast<double>(bytesOutNew.at(i) - bytesOut.at(i)) / (1000. * 1000.)) / static_cast<double>(msSinceLastLog) * 1000.;687 msgPerSecOut.at(i) = static_cast<double>(msgOutNew.at(i) - msgOut.at(i)) / static_cast<double>(msSinceLastLog) * 1000.;688 689 bytesIn.at(i) = bytesInNew.at(i);690 msgIn.at(i) = msgInNew.at(i);691 bytesOut.at(i) = bytesOutNew.at(i);692 msgOut.at(i) = msgOutNew.at(i);693 694 LOG(info) << setw(static_cast<int>(chanNameLen)) << filteredChannelNames.at(i) << ": "695 << "in: " << msgPerSecIn.at(i) << " (" << mbPerSecIn.at(i) << " MB) "696 << "out: " << msgPerSecOut.at(i) << " (" << mbPerSecOut.at(i) << " MB)";697 }698 }699 700 ++i;701 }702 703 t0 = t1;704 }705}706 707void Device::InterruptTransports()708{709 lock_guard<mutex> lock(fTransportMtx);710 for (auto& [transportType, transport] : fTransports) {711 transport->Interrupt();712 }713}714 715void Device::ResumeTransports()716{717 lock_guard<mutex> lock(fTransportMtx);718 for (auto& [transportType, transport] : fTransports) {719 transport->Resume();720 }721}722 723void Device::ResetTaskWrapper()724{725 ResetTask();726 727 if (!NewStatePending()) {728 ChangeStateOrThrow(Transition::Auto);729 }730}731 732void Device::ResetWrapper()733{734 {735 lock_guard<mutex> lock(fTransportMtx);736 for (auto& [transportType, transport] : fTransports) {737 transport->Reset();738 }739 fTransports.clear();740 }741 742 Reset();743 744 GetChannels().clear();745 fTransportFactory.reset();746 if (!NewStatePending()) {747 ChangeStateOrThrow(Transition::Auto);748 }749}750 751/// TODO: Remove this once Device::fChannels is no longer public752#pragma GCC diagnostic push753#pragma GCC diagnostic ignored "-Wdeprecated-declarations"754Device::~Device()755{756 UnsubscribeFromNewTransition("device");757 fStateMachine.StopHandlingStates();758 LOG(debug) << "Shutting down device " << fId;759}760#pragma GCC diagnostic pop761 762} // namespace fair::mq