Please consider the following formatting changes to #11793

Detectors/CTP/workflowScalers/src/ctp-qc-proxy.cxx

@@ -47,7 +47,7 @@ InjectorFunction dcs2dpl()
 // InjectorFunction dcs2dpl()
 {
   return [](TimingInfo&, ServiceRegistryRef const& services, fair::mq::Parts& parts, ChannelRetriever channelRetriever, size_t newTimesliceId, bool&) -> bool {
-    auto *device = services.get<RawDeviceService>().device();
+    auto* device = services.get<RawDeviceService>().device();
     std::string messageHeader{static_cast<const char*>(parts.At(0)->GetData()), parts.At(0)->GetSize()};
     size_t dataSize = parts.At(1)->GetSize();
     std::string messageData{static_cast<const char*>(parts.At(1)->GetData()), parts.At(1)->GetSize()};

Detectors/DCS/testWorkflow/src/DCStoDPLconverter.h

@@ -54,7 +54,7 @@ o2f::InjectorFunction dcs2dpl(std::unordered_map<DPID, o2h::DataDescription>& dp
 {
 
   return [dpid2group, fbiFirst, verbose, FBIPerInterval](o2::framework::TimingInfo& tinfo, framework::ServiceRegistryRef const& services, fair::mq::Parts& parts, o2f::ChannelRetriever channelRetriever, size_t newTimesliceId, bool& stop) -> bool {
-    auto *device = services.get<framework::RawDeviceService>().device();
+    auto* device = services.get<framework::RawDeviceService>().device();
     static std::unordered_map<DPID, DPCOM> cache; // will keep only the latest measurement in the 1-second wide window for each DPID
     static std::unordered_map<std::string, int> sentToChannel;
     static auto timer = std::chrono::high_resolution_clock::now();

Detectors/DCS/testWorkflow/src/dcs-config-proxy.cxx

@@ -68,7 +68,7 @@ auto getDataOriginFromFilename(const std::string& filename)
 InjectorFunction dcs2dpl(const std::string& acknowledge)
 {
   return [acknowledge](TimingInfo&, ServiceRegistryRef const& services, fair::mq::Parts& parts, ChannelRetriever channelRetriever, size_t newTimesliceId, bool&) -> bool {
-    auto *device = services.get<RawDeviceService>().device();
+    auto* device = services.get<RawDeviceService>().device();
     if (parts.Size() == 0) { // received at ^c, ignore
       LOG(info) << "ignoring empty message";
       return false;

Detectors/EMCAL/calibration/include/EMCALCalibration/EMCALCalibParams.h

@@ -70,16 +70,16 @@ struct EMCALCalibParams : public o2::conf::ConfigurableParamHelper<EMCALCalibPar
   size_t nThreads_tc = 2;                ///< number of threads used for the time calinration for filling the histograms
 
   // common parameters
-  std::string calibType = "time";                                        ///< type of calibration to run
-  std::string localRootFilePath = "";                                    ///< path to local root file in order to store the calibration histograms (off by default, only to be used for testing)
-  bool enableFastCalib = false;                                          ///< switch to enable fast calibration. Instead of filling boost histograms, mean and sigma of cells is calculated on the fly
-  bool enableTimeProfiling = false;                                      ///< enable to log how much time is spent in the run function in the calibrator spec. Needed for speed tests offline and at point 2
-  bool setSavedSlotAllowed_EMC = true;                                   ///< if true, saving and loading of calibrations from last run and for next run is enabled
-  bool setSavedSlotAllowedSOR_EMC = true;                                ///< if true, stored calibrations from last run can be loaded in the next run (if false, storing of the calib histograms is still active in contrast to setSavedSlotAllowed_EMC)
-  long endTimeMargin = 2592000000;                                       ///< set end TS to 30 days after slot ends (1000 * 60 * 60 * 24 * 30)
-  std::string selectedClassMasks = "C0TVX-B-NOPF-EMC";                   ///< name of EMCal min. bias trigger that is used for calibration
-  int bcShiftCTP = 0;                                                    ///< bc shift of CTP digits to align them with EMC bc in case they are misaligned
-  std::string filePathSave = "./emc_calib";                              ///< path where calibration histograms are stored at EOR to save them for the next run
+  std::string calibType = "time";                      ///< type of calibration to run
+  std::string localRootFilePath = "";                  ///< path to local root file in order to store the calibration histograms (off by default, only to be used for testing)
+  bool enableFastCalib = false;                        ///< switch to enable fast calibration. Instead of filling boost histograms, mean and sigma of cells is calculated on the fly
+  bool enableTimeProfiling = false;                    ///< enable to log how much time is spent in the run function in the calibrator spec. Needed for speed tests offline and at point 2
+  bool setSavedSlotAllowed_EMC = true;                 ///< if true, saving and loading of calibrations from last run and for next run is enabled
+  bool setSavedSlotAllowedSOR_EMC = true;              ///< if true, stored calibrations from last run can be loaded in the next run (if false, storing of the calib histograms is still active in contrast to setSavedSlotAllowed_EMC)
+  long endTimeMargin = 2592000000;                     ///< set end TS to 30 days after slot ends (1000 * 60 * 60 * 24 * 30)
+  std::string selectedClassMasks = "C0TVX-B-NOPF-EMC"; ///< name of EMCal min. bias trigger that is used for calibration
+  int bcShiftCTP = 0;                                  ///< bc shift of CTP digits to align them with EMC bc in case they are misaligned
+  std::string filePathSave = "./emc_calib";            ///< path where calibration histograms are stored at EOR to save them for the next run
 
   // old parameters. Keep them for a bit (can be deleted after september 5th) as otherwise ccdb and o2 version might not be in synch
   unsigned int minNEvents = 1e7;              ///< minimum number of events to trigger the calibration

Detectors/Vertexing/include/DetectorsVertexing/PVertexerParams.h

@@ -78,12 +78,12 @@ struct PVertexerParams : public o2::conf::ConfigurableParamHelper<PVertexerParam
   float addZSigma2Debris = 0.005 * 0.005;  ///< increment z error^2 by this amount when calculating vertex-to-vertex chi2
 
   // extra debris reduction cut, ignored if maxTDiffDebrisExtra == 0. The maxTDiffDebrisExtra must not exceed maxTDiffDebris
-  float maxTDiffDebrisExtra = 0;                ///< when reducing debris, don't consider vertices separated by time > this value in \mus if >0, if <0: mult factor to ITS ROF
-  float maxZDiffDebrisExtra = 0.05;             ///< don't consider vertices separated by Z > this value in cm
-  float maxMultRatDebrisExtra = 1.0;            ///< don't consider vertices with multiplicity ratio above this
-  float maxChi2TZDebrisExtra = 100.;            ///< don't consider vertices with mutual chi2 exceeding this (for pp should be ~10)
-  float addTimeSigma2DebrisExtra = 225.;        ///< increment time error^2 by this amount when calculating vertex-to-vertex chi2
-  float addZSigma2DebrisExtra = 1.e-3;          ///< increment z error^2 by this amount when calculating vertex-to-vertex chi2
+  float maxTDiffDebrisExtra = 0;         ///< when reducing debris, don't consider vertices separated by time > this value in \mus if >0, if <0: mult factor to ITS ROF
+  float maxZDiffDebrisExtra = 0.05;      ///< don't consider vertices separated by Z > this value in cm
+  float maxMultRatDebrisExtra = 1.0;     ///< don't consider vertices with multiplicity ratio above this
+  float maxChi2TZDebrisExtra = 100.;     ///< don't consider vertices with mutual chi2 exceeding this (for pp should be ~10)
+  float addTimeSigma2DebrisExtra = 225.; ///< increment time error^2 by this amount when calculating vertex-to-vertex chi2
+  float addZSigma2DebrisExtra = 1.e-3;   ///< increment z error^2 by this amount when calculating vertex-to-vertex chi2
 
   float maxITSOnlyFraction = 1.0; ///< max ITS-only tracks fraction to accept, recommended value for PbPb = 0.85
   float minITSOnlyFraction = 0.0; ///< min ITS-only tracks fraction to accept

Framework/Core/include/Framework/DebugGUI.h

@@ -26,6 +26,7 @@
 namespace o2::framework
 {
 struct ServiceRegistry;
+struct GuiCallbackContext;
 /// Plugin interface for DPL GUIs.
 struct DebugGUI {
   virtual std::function<void(void)> getGUIDebugger(std::vector<o2::framework::DeviceInfo> const& infos,
@@ -35,7 +36,8 @@ struct DebugGUI {
                                                    std::vector<o2::framework::DeviceMetricsInfo> const& metricsInfos,
                                                    o2::framework::DriverInfo const& driverInfo,
                                                    std::vector<o2::framework::DeviceControl>& controls,
-                                                   o2::framework::DriverControl& driverControl) = 0;
+                                                   o2::framework::DriverControl& driverControl,
+                                                   GuiCallbackContext& context) = 0;
   virtual void updateMousePos(float x, float y) = 0;
   virtual void updateMouseButton(bool isClicked) = 0;
   virtual void updateMouseWheel(int direction) = 0;

Framework/Core/src/DataProcessingDevice.cxx

@@ -25,6 +25,7 @@
 #include "Framework/DispatchPolicy.h"
 #include "Framework/DispatchControl.h"
 #include "Framework/DanglingContext.h"
+#include "Framework/DriverInfo.h"
 #include "Framework/DomainInfoHeader.h"
 #include "Framework/DriverClient.h"
 #include "Framework/EndOfStreamContext.h"
@@ -159,16 +160,60 @@ DataProcessingDevice::DataProcessingDevice(RunningDeviceRef running, ServiceRegi
 
   std::function<void(const fair::mq::State)> stateWatcher = [this, &registry = mServiceRegistry](const fair::mq::State state) -> void {
     auto ref = ServiceRegistryRef{registry, ServiceRegistry::globalDeviceSalt()};
+    auto controlKind = this->GetConfig()->GetPropertyAsString("control");
     auto& deviceState = ref.get<DeviceState>();
     auto& control = ref.get<ControlService>();
     auto& callbacks = ref.get<CallbackService>();
-    control.notifyDeviceState(fair::mq::GetStateName(state));
+    auto stateName = fair::mq::GetStateName(state);
+    control.notifyDeviceState(stateName);
     callbacks.call<CallbackService::Id::DeviceStateChanged>(ServiceRegistryRef{ref}, (int)state);
+    LOG(detail) << "In state watcher callback " << stateName;
+
+    // If the termination policy is not to wait, we simply ignore all
+    // user imposed state changes and keep running until we are done.
+    if (controlKind != "gui") {
+      return;
+    }
+
+    static bool runningOnce = false;
 
     if (deviceState.nextFairMQState.empty() == false) {
+      LOG(detail) << "State change requested, changing state to " << deviceState.nextFairMQState.back();
       auto state = deviceState.nextFairMQState.back();
-      (void)this->ChangeState(state);
+      bool changed = this->ChangeState(state);
+      if (!changed) {
+        LOG(error) << "Failed to change state to " << state;
+      }
       deviceState.nextFairMQState.pop_back();
+    } else if (state == fair::mq::State::Running && deviceState.nextFairMQState.empty()) {
+      LOGP(detail, "Device is running and no transition expected. We are done.");
+      deviceState.transitionHandling = TransitionHandlingState::NoTransition;
+    } else {
+      while (runningOnce && deviceState.nextFairMQState.empty() && this->NewStatePending() == false) {
+        LOG(detail) << "No state change requested, waiting for next state change " << this->NewStatePending();
+        if (stateName == "EXITING") {
+          // Send ctrl c to ourselves. To bad FairMQ does not seem to exit when
+          // reaching the EXITING state.
+          kill(getpid(), SIGTERM);
+          return;
+        }
+        uv_run(deviceState.loop, UV_RUN_ONCE);
+        LOG(detail) << "Woke up from event loop";
+      }
+      if (runningOnce && deviceState.nextFairMQState.empty() == false) {
+        LOG(detail) << "State change requested, changing state to " << deviceState.nextFairMQState.back();
+        auto state = deviceState.nextFairMQState.back();
+        bool changed = this->ChangeState(state);
+        if (!changed) {
+          LOG(error) << "Failed to change state to " << state;
+        }
+        deviceState.nextFairMQState.pop_back();
+      }
+      LOG(detail) << "Exiting callback for state " << state;
+    }
+    if (runningOnce == false && state == fair::mq::State::Running) {
+      LOG(detail) << "First iteration, next time we start the event loop";
+      runningOnce = true;
     }
   };
 

Framework/Core/src/PropertyTreeHelpers.h

@@ -51,7 +51,8 @@ struct PropertyTreeHelpers {
                        std::string const& propertyLabel);
 
   //using WalkerFunction = std::function<void(boost::property_tree::ptree const&, boost::property_tree::ptree::path_type, boost::property_tree::ptree const&)>;
-  template <typename T> using WalkerFunction = std::function<void(boost::property_tree::ptree const&, typename T::path_type, boost::property_tree::ptree const&)>;
+  template <typename T>
+  using WalkerFunction = std::function<void(boost::property_tree::ptree const&, typename T::path_type, boost::property_tree::ptree const&)>;
   /// Traverse the tree recursively calling @a WalkerFunction on each leaf.
   template <typename T>
   static void traverse(boost::property_tree::ptree const& parent, WalkerFunction<T>& method);

Framework/Core/src/WSDriverClient.cxx

@@ -15,6 +15,8 @@
 #include "Framework/ServiceRegistry.h"
 #include "Framework/DeviceSpec.h"
 #include "DriverClientContext.h"
+#include "Framework/RawDeviceService.h"
+#include "Device.h"
 #include "DPLWebSocket.h"
 #include <uv.h>
 #include <string_view>
@@ -134,6 +136,28 @@ void on_connect(uv_connect_t* connection, int status)
     state.nextFairMQState.emplace_back("STOP");
   });
 
+  client->observe("/shutdown", [ref = context->ref](std::string_view) {
+    auto currentStateName = ref.get<RawDeviceService>().device()->GetCurrentStateName();
+    LOGP(info, "Received shutdown request while in {}", currentStateName);
+
+    auto& state = ref.get<DeviceState>();
+    state.nextFairMQState.emplace_back("END");
+    if (currentStateName == "IDLE") {
+      return;
+    }
+    state.nextFairMQState.emplace_back("AUTO");
+    state.nextFairMQState.emplace_back("RESET DEVICE");
+    if (currentStateName == "DEVICE READY") {
+      return;
+    }
+    state.nextFairMQState.emplace_back("AUTO");
+    state.nextFairMQState.emplace_back("RESET TASK");
+    if (currentStateName == "READY") {
+      return;
+    }
+    state.nextFairMQState.emplace_back("STOP");
+  });
+
   client->observe("/trace", [ref = context->ref](std::string_view cmd) {
     auto& state = ref.get<DeviceState>();
     static constexpr int prefixSize = std::string_view{"/trace "}.size();

Framework/Core/src/runDataProcessing.cxx

@@ -29,6 +29,7 @@
 #include "Framework/DeviceMetricsInfo.h"
 #include "Framework/DeviceMetricsHelper.h"
 #include "Framework/DeviceConfigInfo.h"
+#include "Framework/DeviceController.h"
 #include "Framework/DeviceSpec.h"
 #include "Framework/DeviceState.h"
 #include "Framework/DeviceConfig.h"
@@ -1073,7 +1074,7 @@ void gui_callback(uv_timer_s* ctx)
   if (gui->plugin->supportsDeferredClose()) {
     // For now, there is nothing for which we want to defer the close
     // so if the flag is set, we simply exit
-    if (*(gui->guiQuitRequested)) {
+    if (*(gui->guiQuitRequested) && *(gui->allChildrenGone)) {
       return;
     }
     void* draw_data = nullptr;
@@ -1302,7 +1303,8 @@ int runStateMachine(DataProcessorSpecs const& workflow,
 
   DebugGUI* debugGUI = nullptr;
   void* window = nullptr;
-  decltype(debugGUI->getGUIDebugger(infos, runningWorkflow.devices, allStates, dataProcessorInfos, metricsInfos, driverInfo, controls, driverControl)) debugGUICallback;
+  GuiCallbackContext guiContext;
+  decltype(debugGUI->getGUIDebugger(infos, runningWorkflow.devices, allStates, dataProcessorInfos, metricsInfos, driverInfo, controls, driverControl, guiContext)) debugGUICallback;
 
   // An empty frameworkId means this is the driver, so we initialise the GUI
   auto initDebugGUI = []() -> DebugGUI* {
@@ -1389,7 +1391,6 @@ int runStateMachine(DataProcessorSpecs const& workflow,
   ServiceRegistryRef ref{serviceRegistry};
   ref.registerService(ServiceRegistryHelpers::handleForService<DevicesManager>(devicesManager));
 
-  GuiCallbackContext guiContext;
   guiContext.plugin = debugGUI;
   guiContext.frameLast = uv_hrtime();
   guiContext.frameLatency = &driverInfo.frameLatency;
@@ -1882,7 +1883,7 @@ int runStateMachine(DataProcessorSpecs const& workflow,
             uv_timer_stop(gui_timer);
           }
 
-          auto callback = debugGUI->getGUIDebugger(infos, runningWorkflow.devices, allStates, dataProcessorInfos, metricsInfos, driverInfo, controls, driverControl);
+          auto callback = debugGUI->getGUIDebugger(infos, runningWorkflow.devices, allStates, dataProcessorInfos, metricsInfos, driverInfo, controls, driverControl, guiContext);
           guiContext.callback = [&serviceRegistry, &driverServices, &debugGUI, &infos, &runningWorkflow, &dataProcessorInfos, &metricsInfos, &driverInfo, &controls, &driverControl, callback]() {
             callback();
             for (auto& service : driverServices) {
@@ -2068,7 +2069,17 @@ int runStateMachine(DataProcessorSpecs const& workflow,
         // We send SIGCONT to make sure stopped children are resumed
         killChildren(infos, SIGCONT);
         // We send SIGTERM to make sure we do the STOP transition in FairMQ
-        killChildren(infos, SIGTERM);
+        if (driverInfo.processingPolicies.termination == TerminationPolicy::WAIT) {
+          for (size_t di = 0; di < infos.size(); ++di) {
+            auto& info = infos[di];
+            auto& control = controls[di];
+            if (info.active == true) {
+              control.controller->write("/shutdown", strlen("/shutdown"));
+            }
+          }
+        } else {
+          killChildren(infos, SIGTERM);
+        }
         // We have a timer to send SIGUSR1 to make sure we advance all devices
         // in a timely manner.
         force_step_timer.data = &infos;

Framework/Core/test/test_Root2ArrowTable.cxx

@@ -92,9 +92,9 @@ TEST_CASE("RootTree2Table")
     REQUIRE(array->length() == 3000);
     const float* c = reinterpret_cast<float const*>(array->values()->data());
 
-    //auto array = std::static_pointer_cast<arrow::FixedSizeBinaryArray>(table->column(0)->chunk(0));
-    //CHECK_EQUAL(array->byte_width(), sizeof(float[3]));
-    //const float* c = reinterpret_cast<float const*>(array->Value(0));
+    // auto array = std::static_pointer_cast<arrow::FixedSizeBinaryArray>(table->column(0)->chunk(0));
+    // CHECK_EQUAL(array->byte_width(), sizeof(float[3]));
+    // const float* c = reinterpret_cast<float const*>(array->Value(0));
 
     CHECK(c[0] == 1);
     CHECK(c[1] == 2);