Source · fairmq/plugins/control/
Control.cxxsource
Described in Plugins and control, Device states, Threads in a device, Stop a topology without losing messages, Test your devices, Command-line options, States and transitions, Your first device.
1/********************************************************************************2 * Copyright (C) 2017-2022 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#include "Control.h"10 11#include <fairmq/tools/IO.h>12 13#include <atomic>14#include <chrono>15#include <csignal> // catching system signals16#include <cstdlib>17#include <functional>18#include <iostream>19#include <thread>20 21#include <poll.h> // for the interactive mode22 23using namespace std;24 25namespace26{27 std::atomic<sig_atomic_t> gLastSignal(0);28 std::atomic<int> gSignalCount(0);29 30 extern "C" auto sigint_handler(int signal) -> void31 {32 ++gSignalCount;33 gLastSignal = signal;34 35 if (gSignalCount > 1) {36 std::abort();37 }38 }39 40 extern "C" auto sigterm_handler(int signal) -> void41 {42 ++gSignalCount;43 gLastSignal = signal;44 }45}46 47namespace fair::mq::plugins48{49 50Control::Control(const string& name, Plugin::Version version, const string& maintainer, const string& homepage, PluginServices* pluginServices)51 : Plugin(name, version, maintainer, homepage, pluginServices)52 , fDeviceShutdownRequested(false)53 , fDeviceHasShutdown(false)54 , fPluginShutdownRequested(false)55{56 SubscribeToDeviceStateChange([&](DeviceState newState) {57 LOG(trace) << "control plugin notified on new state: " << newState;58 59 if (newState == DeviceState::Error) {60 fPluginShutdownRequested = true;61 fStateQueue.Push(newState, [this]{ fDeviceShutdownRequested = true; });62 } else {63 fStateQueue.Push(newState);64 }65 });66 67 try {68 auto control = GetProperty<string>("control");69 70 if (control != "none") {71 TakeDeviceControl();72 }73 74 if (control == "static") {75 LOG(debug) << "Running builtin controller: static";76 fControllerThread = thread(&Control::StaticMode, this);77 } else if (control == "none") {78 LOG(debug) << "Builtin controller: disabled";79 } else if (control == "gui") {80 LOG(debug) << "Running builtin controller: gui";81 fControllerThread = thread(&Control::GUIMode, this);82 } else if (control == "dynamic" || control == "external" || control == "interactive") {83 LOG(debug) << "Running builtin controller: interactive";84 fControllerThread = thread(&Control::InteractiveMode, this);85 } else {86 LOG(error) << "Unrecognized control mode '" << control << "' requested. " << "Ignoring and falling back to static control mode.";87 fControllerThread = thread(&Control::StaticMode, this);88 }89 } catch (PluginServices::DeviceControlError& e) {90 // If we are here, it means another plugin has taken control. That's fine, just print the exception message and do nothing else.91 LOG(debug) << e.what();92 }93 94 if (GetProperty<int>("catch-signals") > 0) {95 LOG(debug) << "Plugin '" << name << "' is setting up signal handling for SIGINT and SIGTERM";96 fSignalHandlerThread = thread(&Control::SignalHandler, this);97 signal(SIGINT, sigint_handler);98 signal(SIGTERM, sigterm_handler);99 } else {100 LOG(warn) << "Signal handling (e.g. Ctrl-C) has been deactivated.";101 }102}103 104auto Control::RunStartupSequence() -> void105{106 using Transition = DeviceStateTransition;107 using State = DeviceState;108 auto shutdownRequested = [this]{ return fDeviceShutdownRequested.load(); };109 110 ChangeDeviceState(Transition::InitDevice);111 fStateQueue.WaitForStateOrCustom(State::InitializingDevice, shutdownRequested);112 if (fDeviceShutdownRequested) { return; /* --> shutdown sequence */ }113 114 ChangeDeviceState(Transition::CompleteInit);115 fStateQueue.WaitForStateOrCustom(State::Initialized, shutdownRequested);116 if (fDeviceShutdownRequested) { return; /* --> shutdown sequence */ }117 118 ChangeDeviceState(Transition::Bind);119 fStateQueue.WaitForStateOrCustom(State::Binding, shutdownRequested);120 if (fDeviceShutdownRequested) { return; /* --> shutdown sequence */ }121 122 fStateQueue.WaitForStateOrCustom(State::Bound, shutdownRequested);123 if (fDeviceShutdownRequested) { return; /* --> shutdown sequence */ }124 125 ChangeDeviceState(Transition::Connect);126 fStateQueue.WaitForStateOrCustom(State::Connecting, shutdownRequested);127 if (fDeviceShutdownRequested) { return; /* --> shutdown sequence */ }128 129 fStateQueue.WaitForStateOrCustom(State::DeviceReady, shutdownRequested);130 if (fDeviceShutdownRequested) { return; /* --> shutdown sequence */ }131 132 ChangeDeviceState(Transition::InitTask);133 fStateQueue.WaitForStateOrCustom(State::InitializingTask, shutdownRequested);134 if (fDeviceShutdownRequested) { return; /* --> shutdown sequence */ }135 136 fStateQueue.WaitForStateOrCustom(State::Ready, shutdownRequested);137 if (fDeviceShutdownRequested) { return; /* --> shutdown sequence */ }138 139 ChangeDeviceState(Transition::Run);140 fStateQueue.WaitForStateOrCustom(State::Running, shutdownRequested);141 if (fDeviceShutdownRequested) { return; /* --> shutdown sequence */ }142}143 144auto ControlPluginProgramOptions() -> Plugin::ProgOptions145{146 namespace po = boost::program_options;147 auto pluginOptions = po::options_description{"Control (builtin) Plugin"};148 pluginOptions.add_options()149 ("control", po::value<string>()->default_value("dynamic"), "Control mode, 'static' or 'dynamic' (aliases for dynamic are external and interactive), 'none', 'gui'")150 ("catch-signals", po::value<int >()->default_value(1), "Enable signal handling (1/0).");151 return pluginOptions;152}153 154auto Control::RunREPL() -> void155{156 char input = 0; // hold the user console input157 pollfd cinfd[1];158 cinfd[0].fd = fileno(stdin);159 cinfd[0].events = POLLIN;160 cinfd[0].revents = 0;161 162 tools::NonCanonicalInput nci;163 164 bool color = GetProperty<bool>("color");165 166 if (color) {167 PrintInteractiveHelpColor();168 } else {169 PrintInteractiveHelp();170 }171 172 bool keepRunning = true;173 174 while (keepRunning) {175 if (poll(cinfd, 1, 100)) {176 if (fDeviceShutdownRequested) {177 break;178 }179 180 cin >> input;181 182 switch (input) {183 case 'c':184 cout << "\n --> [i] check current state\n\n" << flush;185 LOG(state) << GetCurrentDeviceState();186 break;187 case 'i':188 cout << "\n --> [i] init device\n\n" << flush;189 if (ChangeDeviceState(DeviceStateTransition::InitDevice)) {190 fStateQueue.WaitForState(DeviceState::InitializingDevice);191 ChangeDeviceState(DeviceStateTransition::CompleteInit);192 }193 break;194 case 'b':195 cout << "\n --> [b] bind\n\n" << flush;196 ChangeDeviceState(DeviceStateTransition::Bind);197 break;198 case 'x':199 cout << "\n --> [x] connect\n\n" << flush;200 ChangeDeviceState(DeviceStateTransition::Connect);201 break;202 case 'j':203 cout << "\n --> [j] init task\n\n" << flush;204 ChangeDeviceState(DeviceStateTransition::InitTask);205 break;206 case 'r':207 cout << "\n --> [r] run\n\n" << flush;208 ChangeDeviceState(DeviceStateTransition::Run);209 break;210 case 's':211 cout << "\n --> [s] stop\n\n" << flush;212 ChangeDeviceState(DeviceStateTransition::Stop);213 break;214 case 't':215 cout << "\n --> [t] reset task\n\n" << flush;216 ChangeDeviceState(DeviceStateTransition::ResetTask);217 break;218 case 'd':219 cout << "\n --> [d] reset device\n\n" << flush;220 ChangeDeviceState(DeviceStateTransition::ResetDevice);221 break;222 case 'k':223 cout << "\n --> [k] increase log severity\n\n" << flush;224 CycleLogConsoleSeverityUp();225 break;226 case 'l':227 cout << "\n --> [l] decrease log severity\n\n" << flush;228 CycleLogConsoleSeverityDown();229 break;230 case 'n':231 cout << "\n --> [n] increase log verbosity\n\n" << flush;232 CycleLogVerbosityUp();233 break;234 case 'm':235 cout << "\n --> [m] decrease log verbosity\n\n" << flush;236 CycleLogVerbosityDown();237 break;238 case 'p':239 cout << "\n --> [p] print number of connected peers for all channels\n\n" << flush;240 PrintNumberOfConnectedPeers();241 break;242 case 'h':243 cout << "\n --> [h] help\n\n" << flush;244 if (color) {245 PrintInteractiveHelpColor();246 PrintStateMachineColor();247 } else {248 PrintInteractiveHelp();249 PrintStateMachine();250 }251 break;252 case 'q':253 cout << "\n --> [q] end\n\n" << flush;254 keepRunning = false;255 break;256 default:257 cout << "Invalid input: [" << input << "]. " << flush;258 PrintInteractiveHelp();259 break;260 }261 }262 263 if (GetCurrentDeviceState() == DeviceState::Error) {264 throw DeviceErrorState("Controlled device transitioned to error state.");265 }266 267 if (fDeviceShutdownRequested) {268 break;269 }270 }271 272}273 274auto Control::InteractiveMode() -> void275try {276 RunStartupSequence();277 278 if (!fDeviceShutdownRequested) {279 RunREPL();280 }281 282 if (!fDeviceShutdownRequested) {283 RunShutdownSequence();284 }285} catch (PluginServices::DeviceControlError& e) {286 // If we are here, it means another plugin has taken control. That's fine, just print the exception message and do nothing else.287 LOG(debug) << e.what();288} catch (DeviceErrorState&) {289 ReleaseDeviceControl();290}291 292auto Control::PrintInteractiveHelpColor() -> void293{294 stringstream ss;295 ss << "Following control commands are available:\n\n"296 << " [\033[01;32mh\033[0m] help, [\033[01;32mc\033[0m] check current device state, [\033[01;32mp\033[0m] print number of connected peers for channels,\n"297 << " [\033[01;32mi\033[0m] init device, [\033[01;32mb\033[0m] bind, [\033[01;32mx\033[0m] connect, [\033[01;32mj\033[0m] init task,"298 << " [\033[01;32mr\033[0m] run, [\033[01;32ms\033[0m] stop,\n"299 << " [\033[01;32mt\033[0m] reset task, [\033[01;32md\033[0m] reset device, [\033[01;32mq\033[0m] end,\n"300 << " [\033[01;32mk\033[0m] increase log severity, [\033[01;32ml\033[0m] decrease log severity, [\033[01;32mn\033[0m] increase log verbosity, [\033[01;32mm\033[0m] decrease log verbosity\n\n";301 cout << ss.str() << flush;302}303 304auto Control::PrintInteractiveHelp() -> void305{306 stringstream ss;307 ss << "Following control commands are available:\n\n"308 << " [h] help, [c] check current device state, [p] print number of connected peers for channels\n"309 << " [i] init device, [b] bind, [x] connect, [j] init task,\n"310 << " [r] run, [s] stop,\n"311 << " [t] reset task, [d] reset device, [q] end,\n"312 << " [k] increase log severity, [l] decrease log severity, [n] increase log verbosity, [m] decrease log verbosity.\n\n";313 cout << ss.str() << flush;314}315 316void Control::PrintStateMachineColor()317{318 stringstream ss;319 ss << " @ \n"320 << " ┌───────────╨─────────────┐ ┌───────────┐ \n"321 << " │ \033[01;36mIDLE\033[0m [\033[01;32mq\033[0m]─▶ \033[01;33mEXITING\033[0m │ \n"322 << " └[\033[01;32mi\033[0m]─────────────────── ▲ ┘ └───────────┘ \n"323 << " ╔══ ▼ ════════════════╗ ╔═╩══════════════════╗ \n"324 << " ║ \033[01;33mINITIALIZING DEVICE\033[0m ║ ║ \033[01;33mRESETTING DEVICE\033[0m ║ \n"325 << " ╚══════════╦══════════╝ ╚════════ ▲ ═════════╝ \n"326 << " ┌───────── ▼ ─────────┐ │ ┌────────────────────────────┐ \n"327 << " │ \033[01;36mINITIALIZED\033[0m [\033[01;32md\033[0m]──────────┤ │ Legend: │ \n"328 << " └─────────[\033[01;32mb\033[0m]─────────┘ │ │----------------------------│ \n"329 << " ╔═════════ ▼ ═════════╗ │ │ [\033[01;32mk\033[0m] keyboard shortcut for │ \n"330 << " ║ \033[01;33mBINDING\033[0m ║ │ │ interactive controller │ \n"331 << " ╚══════════╦══════════╝ │ │ ┌────────────────────────┐ │ \n"332 << " ┌───────── ▼ ─────────┐ │ │ │ \033[01;36mIDLING STATE\033[0m │ │ \n"333 << " │ \033[01;36mBOUND\033[0m [\033[01;32md\033[0m]──────────┤ │ └────────────────────────┘ │ \n"334 << " └─────────[\033[01;32mx\033[0m]─────────┘ │ │ ╔════════════════════════╗ │ \n"335 << " ╔═════════ ▼ ═════════╗ │ │ ║ \033[01;33mWORKING STATE\033[0m ║ │ \n"336 << " ║ \033[01;33mCONNECTING\033[0m ║ │ │ ╚════════════════════════╝ │ \n"337 << " ╚══════════╦══════════╝ │ └────────────────────────────┘ \n"338 << " ┌─────── ▼ ────────────────────[\033[01;32md\033[0m]───────┐ \n"339 << " │ \033[01;36mDEVICE READY\033[0m │ \n"340 << " └───────[\033[01;32mj\033[0m]──────────────────── ▲ ───────┘ \n"341 << " ╔═════════ ▼ ═════════╗ ╔═════════╩══════════╗ \n"342 << " ║ \033[01;33mINITIALIZING TASK\033[0m ║ ║ \033[01;33mRESETTING TASK\033[0m ║ \n"343 << " ╚══════════╦══════════╝ ╚════════ ▲ ═════════╝ \n"344 << " ┌─────── ▼ ────────────────────[\033[01;32mt\033[0m]───────┐ \n"345 << " │ \033[01;36mREADY\033[0m │ \n"346 << " └───────[\033[01;32mr\033[0m]──────────────────── ▲ ───────┘ \n"347 << " ╔══════ ▼ ════════════════════[\033[01;32ms\033[0m]══════╗ \n"348 << " ║ \033[01;33mRUNNING\033[0m ║ \n"349 << " ╚══════════════════════════════════════╝ \n"350 << " \n";351 cout << ss.str() << flush;352}353 354void Control::PrintStateMachine()355{356 stringstream ss;357 ss << " @ \n"358 << " ┌───────────╨─────────────┐ ┌───────────┐ \n"359 << " │ IDLE [q]─▶ EXITING │ \n"360 << " └[i]─────────────────── ▲ ┘ └───────────┘ \n"361 << " ╔══ ▼ ════════════════╗ ╔═╩══════════════════╗ \n"362 << " ║ INITIALIZING DEVICE ║ ║ RESETTING DEVICE ║ \n"363 << " ╚══════════╦══════════╝ ╚════════ ▲ ═════════╝ \n"364 << " ┌───────── ▼ ─────────┐ │ ┌────────────────────────────┐ \n"365 << " │ INITIALIZED [d]──────────┤ │ Legend: │ \n"366 << " └─────────[b]─────────┘ │ │----------------------------│ \n"367 << " ╔═════════ ▼ ═════════╗ │ │ [k] keyboard shortcut for │ \n"368 << " ║ BINDING ║ │ │ interactive controller │ \n"369 << " ╚══════════╦══════════╝ │ │ ┌────────────────────────┐ │ \n"370 << " ┌───────── ▼ ─────────┐ │ │ │ IDLING STATE │ │ \n"371 << " │ BOUND [d]──────────┤ │ └────────────────────────┘ │ \n"372 << " └─────────[x]─────────┘ │ │ ╔════════════════════════╗ │ \n"373 << " ╔═════════ ▼ ═════════╗ │ │ ║ WORKING STATE ║ │ \n"374 << " ║ CONNECTING ║ │ │ ╚════════════════════════╝ │ \n"375 << " ╚══════════╦══════════╝ │ └────────────────────────────┘ \n"376 << " ┌─────── ▼ ────────────────────[d]───────┐ \n"377 << " │ DEVICE READY │ \n"378 << " └───────[j]──────────────────── ▲ ───────┘ \n"379 << " ╔═════════ ▼ ═════════╗ ╔═════════╩══════════╗ \n"380 << " ║ INITIALIZING TASK ║ ║ RESETTING TASK ║ \n"381 << " ╚══════════╦══════════╝ ╚════════ ▲ ═════════╝ \n"382 << " ┌─────── ▼ ────────────────────[t]───────┐ \n"383 << " │ READY │ \n"384 << " └───────[r]──────────────────── ▲ ───────┘ \n"385 << " ╔══════ ▼ ════════════════════[s]══════╗ \n"386 << " ║ RUNNING ║ \n"387 << " ╚══════════════════════════════════════╝ \n"388 << " \n";389 cout << ss.str() << flush;390}391 392auto Control::PrintNumberOfConnectedPeers() -> void393{394 unordered_map<string, int> channelInfo(GetChannelInfo());395 396 for (const auto& c : channelInfo) {397 for (int i = 0; i < c.second; ++i) {398 LOG(info) << c.first << "[" << i << "]: " << GetNumberOfConnectedPeers(c.first, i) << " peers connected";399 }400 }401}402 403auto Control::StaticMode() -> void404try {405 RunStartupSequence();406 407 // Wait for next state, which is DeviceState::Ready,408 // or for device shutdown request (Ctrl-C)409 fStateQueue.WaitForNextOrCustom([this]{ return fDeviceShutdownRequested.load(); });410 411 if (!fDeviceShutdownRequested) {412 RunShutdownSequence();413 }414} catch (PluginServices::DeviceControlError& e) {415 // If we are here, it means another plugin has taken control. That's fine, just print the exception message and do nothing else.416 LOG(debug) << e.what();417} catch (DeviceErrorState&) {418 ReleaseDeviceControl();419}420 421auto Control::GUIMode() -> void422try {423 RunStartupSequence();424 425 // Wait for device shutdown request (Ctrl-C)426 fStateQueue.WaitForCustom([this]{ return fDeviceShutdownRequested.load(); });427 428 if (!fDeviceShutdownRequested) {429 RunShutdownSequence();430 }431} catch (PluginServices::DeviceControlError& e) {432 // If we are here, it means another plugin has taken control. That's fine, just print the433 // exception message and do nothing else.434 LOG(debug) << e.what();435} catch (DeviceErrorState&) {436 ReleaseDeviceControl();437}438 439auto Control::SignalHandler() -> void440{441 while (gSignalCount == 0 && !fPluginShutdownRequested) {442 this_thread::sleep_for(chrono::milliseconds(100));443 }444 445 if (!fPluginShutdownRequested) {446 LOG(info) << "Received device shutdown request (signal " << gLastSignal << ").";447 LOG(info) << "Waiting for graceful device shutdown. Hit Ctrl-C again to abort immediately.";448 449 // Signal and wait for controller thread, if we are controller450 fStateQueue.Notify([this] { fDeviceShutdownRequested = true; });451 {452 unique_lock<mutex> lock(fControllerMutex);453 if (fControllerThread.joinable()) { fControllerThread.join(); }454 }455 456 if (!fDeviceHasShutdown) {457 // Take over control and attempt graceful shutdown458 StealDeviceControl();459 try {460 RunShutdownSequence();461 } catch (PluginServices::DeviceControlError& e) {462 LOG(info) << "Graceful device shutdown failed: " << e.what() << " If hanging, hit Ctrl-C again to abort immediately.";463 } catch (...) {464 ReleaseDeviceControl();465 LOG(info) << "Graceful device shutdown failed. If hanging, hit Ctrl-C again to abort immediately.";466 }467 }468 }469}470 471auto Control::RunShutdownSequence() -> void472{473 auto nextState = GetCurrentDeviceState();474 if (nextState != DeviceState::Error) {475 fStateQueue.Clear();476 }477 while (nextState != DeviceState::Exiting && nextState != DeviceState::Error) {478 switch (nextState) {479 case DeviceState::Idle:480 ChangeDeviceState(DeviceStateTransition::End);481 break;482 case DeviceState::InitializingDevice:483 ChangeDeviceState(DeviceStateTransition::CompleteInit);484 break;485 case DeviceState::Initialized:486 case DeviceState::Bound:487 case DeviceState::DeviceReady:488 ChangeDeviceState(DeviceStateTransition::ResetDevice);489 break;490 case DeviceState::Ready:491 ChangeDeviceState(DeviceStateTransition::ResetTask);492 break;493 case DeviceState::Running:494 ChangeDeviceState(DeviceStateTransition::Stop);495 break;496 default:497 // LOG(debug) << "Controller ignoring event: " << nextState;498 break;499 }500 501 nextState = fStateQueue.WaitForNext();502 }503 504 fDeviceHasShutdown = true;505 ReleaseDeviceControl();506}507 508Control::~Control()509{510 // Notify threads to exit511 fPluginShutdownRequested = true;512 513 {514 unique_lock<mutex> lock(fControllerMutex);515 if (fControllerThread.joinable()) { fControllerThread.join(); }516 }517 if (fSignalHandlerThread.joinable()) { fSignalHandlerThread.join(); }518 519 UnsubscribeFromDeviceStateChange();520}521 522} // namespace fair::mq::plugins