From 3b65d604d337520075a3dacd851e715dfc7ac93b Mon Sep 17 00:00:00 2001 From: Barthelemy Date: Mon, 3 Apr 2023 14:32:06 +0200 Subject: [PATCH 01/21] proof of concept --- Framework/basic.json | 1 + Framework/include/QualityControl/TaskRunner.h | 2 ++ Framework/include/QualityControl/TaskRunnerConfig.h | 1 + Framework/include/QualityControl/TaskSpec.h | 1 + Framework/src/InfrastructureSpecReader.cxx | 1 + Framework/src/TaskRunner.cxx | 1 + Framework/src/TaskRunnerFactory.cxx | 4 ++++ Framework/test/testTaskInterface.cxx | 1 + 8 files changed, 12 insertions(+) diff --git a/Framework/basic.json b/Framework/basic.json index 942014ceff..0106d8a5ff 100644 --- a/Framework/basic.json +++ b/Framework/basic.json @@ -39,6 +39,7 @@ "tasks": { "QcTask": { "active": "true", + "critical": "false", "": "if false the task is allowed to die without stopping the workflow", "className": "o2::quality_control_modules::skeleton::SkeletonTask", "moduleName": "QcSkeleton", "detectorName": "TST", diff --git a/Framework/include/QualityControl/TaskRunner.h b/Framework/include/QualityControl/TaskRunner.h index 3df7186b9e..ad9756c929 100644 --- a/Framework/include/QualityControl/TaskRunner.h +++ b/Framework/include/QualityControl/TaskRunner.h @@ -102,6 +102,8 @@ class TaskRunner : public framework::Task /// \brief Data Processor Label to identify all Task Runners static framework::DataProcessorLabel getTaskRunnerLabel() { return { "qc-task" }; } + /// \brief Data Processor Label to make it expendable, i.e. non-critical + static framework::DataProcessorLabel getExpendableLabel() { return { "expendable" }; } /// \brief ID string for all TaskRunner devices static std::string createTaskRunnerIdString(); /// \brief Unified DataOrigin for Quality Control tasks diff --git a/Framework/include/QualityControl/TaskRunnerConfig.h b/Framework/include/QualityControl/TaskRunnerConfig.h index 48646c922d..e39a9aac2f 100644 --- a/Framework/include/QualityControl/TaskRunnerConfig.h +++ b/Framework/include/QualityControl/TaskRunnerConfig.h @@ -47,6 +47,7 @@ struct TaskRunnerConfig { std::string className; std::vector> cycleDurations = {}; int maxNumberCycles; + bool critical; std::string consulUrl{}; std::string conditionUrl{}; std::string monitoringUrl{}; diff --git a/Framework/include/QualityControl/TaskSpec.h b/Framework/include/QualityControl/TaskSpec.h index 9fb668b08b..5afe0af726 100644 --- a/Framework/include/QualityControl/TaskSpec.h +++ b/Framework/include/QualityControl/TaskSpec.h @@ -59,6 +59,7 @@ struct TaskSpec { DataSourceSpec dataSource; // advanced bool active = true; + bool critical = true; int maxNumberCycles = -1; size_t resetAfterCycles = 0; std::string saveObjectsToFile; diff --git a/Framework/src/InfrastructureSpecReader.cxx b/Framework/src/InfrastructureSpecReader.cxx index 44fb1ee40a..74b824fc53 100644 --- a/Framework/src/InfrastructureSpecReader.cxx +++ b/Framework/src/InfrastructureSpecReader.cxx @@ -110,6 +110,7 @@ TaskSpec InfrastructureSpecReader::readSpecEntry(const std::string& ta } ts.dataSource = readSpecEntry(taskID, taskTree.get_child("dataSource"), wholeTree); ts.active = taskTree.get("active", ts.active); + ts.critical = taskTree.get("critical", ts.active); ts.maxNumberCycles = taskTree.get("maxNumberCycles", ts.maxNumberCycles); ts.resetAfterCycles = taskTree.get("resetAfterCycles", ts.resetAfterCycles); ts.saveObjectsToFile = taskTree.get("saveObjectsToFile", ts.saveObjectsToFile); diff --git a/Framework/src/TaskRunner.cxx b/Framework/src/TaskRunner.cxx index 40ef9839f5..1b54bf4aa8 100644 --- a/Framework/src/TaskRunner.cxx +++ b/Framework/src/TaskRunner.cxx @@ -408,6 +408,7 @@ void TaskRunner::printTaskConfig() const << " / Module name : " << mTaskConfig.moduleName // << " / Detector name : " << mTaskConfig.detectorName // << " / Max number cycles : " << mTaskConfig.maxNumberCycles // + << " / critical : " << mTaskConfig.critical // << " / Save to file : " << mTaskConfig.saveToFile << " / Cycle duration seconds : "; for (auto& [cycleDuration, period] : mTaskConfig.cycleDurations) { diff --git a/Framework/src/TaskRunnerFactory.cxx b/Framework/src/TaskRunnerFactory.cxx index c6f2a208d8..2139545d04 100644 --- a/Framework/src/TaskRunnerFactory.cxx +++ b/Framework/src/TaskRunnerFactory.cxx @@ -53,6 +53,9 @@ o2::framework::DataProcessorSpec TaskRunnerFactory::create(const TaskRunnerConfi }; newTask.labels.emplace_back(o2::framework::ecs::qcReconfigurable); newTask.labels.emplace_back(TaskRunner::getTaskRunnerLabel()); + if(!taskConfig.critical) { + newTask.labels.emplace_back(TaskRunner::getExpendableLabel()); + } return newTask; } @@ -144,6 +147,7 @@ TaskRunnerConfig TaskRunnerFactory::extractConfig(const CommonSpec& globalConfig taskSpec.className, multipleCycleDurations, taskSpec.maxNumberCycles, + taskSpec.critical, globalConfig.consulUrl, globalConfig.conditionDBUrl, globalConfig.monitoringUrl, diff --git a/Framework/test/testTaskInterface.cxx b/Framework/test/testTaskInterface.cxx index a8e48bac05..77cbc1de19 100644 --- a/Framework/test/testTaskInterface.cxx +++ b/Framework/test/testTaskInterface.cxx @@ -173,6 +173,7 @@ TEST_CASE("test_task_factory") "o2::quality_control_modules::skeleton::SkeletonTask", { { 10, 1 } }, -1, + true, "" }; From 8e67d8ab085692d747b3e6fe045a3b6f1a7f9cc1 Mon Sep 17 00:00:00 2001 From: Barthelemy Date: Mon, 3 Apr 2023 14:45:42 +0200 Subject: [PATCH 02/21] format --- Framework/src/TaskRunner.cxx | 2 +- Framework/src/TaskRunnerFactory.cxx | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/Framework/src/TaskRunner.cxx b/Framework/src/TaskRunner.cxx index 1b54bf4aa8..521bfa5eb6 100644 --- a/Framework/src/TaskRunner.cxx +++ b/Framework/src/TaskRunner.cxx @@ -408,7 +408,7 @@ void TaskRunner::printTaskConfig() const << " / Module name : " << mTaskConfig.moduleName // << " / Detector name : " << mTaskConfig.detectorName // << " / Max number cycles : " << mTaskConfig.maxNumberCycles // - << " / critical : " << mTaskConfig.critical // + << " / critical : " << mTaskConfig.critical // << " / Save to file : " << mTaskConfig.saveToFile << " / Cycle duration seconds : "; for (auto& [cycleDuration, period] : mTaskConfig.cycleDurations) { diff --git a/Framework/src/TaskRunnerFactory.cxx b/Framework/src/TaskRunnerFactory.cxx index 2139545d04..53124fde02 100644 --- a/Framework/src/TaskRunnerFactory.cxx +++ b/Framework/src/TaskRunnerFactory.cxx @@ -53,7 +53,7 @@ o2::framework::DataProcessorSpec TaskRunnerFactory::create(const TaskRunnerConfi }; newTask.labels.emplace_back(o2::framework::ecs::qcReconfigurable); newTask.labels.emplace_back(TaskRunner::getTaskRunnerLabel()); - if(!taskConfig.critical) { + if (!taskConfig.critical) { newTask.labels.emplace_back(TaskRunner::getExpendableLabel()); } From 4df43270661623d22c01f65b97b0309da1a1afc4 Mon Sep 17 00:00:00 2001 From: Barthelemy Date: Tue, 25 Apr 2023 15:07:22 +0200 Subject: [PATCH 03/21] making the check expendable --- Framework/src/CheckRunnerFactory.cxx | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/Framework/src/CheckRunnerFactory.cxx b/Framework/src/CheckRunnerFactory.cxx index 5b0a1e1597..30f4ffe23e 100644 --- a/Framework/src/CheckRunnerFactory.cxx +++ b/Framework/src/CheckRunnerFactory.cxx @@ -45,6 +45,8 @@ DataProcessorSpec CheckRunnerFactory::create(CheckRunnerConfig checkRunnerConfig options }; newCheckRunner.labels.emplace_back(o2::framework::ecs::qcReconfigurable); newCheckRunner.labels.emplace_back(CheckRunner::getCheckRunnerLabel()); + framework::DataProcessorLabel expendableLabel = {"expendable"}; + newCheckRunner.labels.emplace_back(expendableLabel); newCheckRunner.algorithm = adaptFromTask(std::move(qcCheckRunner)); return newCheckRunner; } @@ -61,7 +63,8 @@ DataProcessorSpec CheckRunnerFactory::createSinkDevice(const CheckRunnerConfig& checkRunnerConfig.options, {}, { o2::framework::ecs::qcReconfigurable } }; - + framework::DataProcessorLabel expendableLabel = {"expendable"}; + newCheckRunner.labels.emplace_back(expendableLabel); return newCheckRunner; } From feb7671c37109d8506478a7d7f4bfe6955e5f2ff Mon Sep 17 00:00:00 2001 From: Barthelemy Date: Fri, 28 Apr 2023 11:58:32 +0200 Subject: [PATCH 04/21] also make aggregators non-critical --- Framework/include/QualityControl/TaskRunner.h | 2 -- Framework/src/AggregatorRunnerFactory.cxx | 2 ++ Framework/src/TaskRunnerFactory.cxx | 3 ++- 3 files changed, 4 insertions(+), 3 deletions(-) diff --git a/Framework/include/QualityControl/TaskRunner.h b/Framework/include/QualityControl/TaskRunner.h index ad9756c929..3df7186b9e 100644 --- a/Framework/include/QualityControl/TaskRunner.h +++ b/Framework/include/QualityControl/TaskRunner.h @@ -102,8 +102,6 @@ class TaskRunner : public framework::Task /// \brief Data Processor Label to identify all Task Runners static framework::DataProcessorLabel getTaskRunnerLabel() { return { "qc-task" }; } - /// \brief Data Processor Label to make it expendable, i.e. non-critical - static framework::DataProcessorLabel getExpendableLabel() { return { "expendable" }; } /// \brief ID string for all TaskRunner devices static std::string createTaskRunnerIdString(); /// \brief Unified DataOrigin for Quality Control tasks diff --git a/Framework/src/AggregatorRunnerFactory.cxx b/Framework/src/AggregatorRunnerFactory.cxx index 3113463ab5..6d81ac9d6f 100644 --- a/Framework/src/AggregatorRunnerFactory.cxx +++ b/Framework/src/AggregatorRunnerFactory.cxx @@ -49,6 +49,8 @@ DataProcessorSpec AggregatorRunnerFactory::create(const core::CommonSpec& common }; newAggregatorRunner.labels.emplace_back(o2::framework::ecs::qcReconfigurable); newAggregatorRunner.labels.emplace_back(AggregatorRunner::getLabel()); + framework::DataProcessorLabel expendableLabel = {"expendable"}; + newAggregatorRunner.labels.emplace_back(expendableLabel); newAggregatorRunner.algorithm = adaptFromTask(std::move(aggregatorRunner)); return newAggregatorRunner; } diff --git a/Framework/src/TaskRunnerFactory.cxx b/Framework/src/TaskRunnerFactory.cxx index 53124fde02..d0e628ad5c 100644 --- a/Framework/src/TaskRunnerFactory.cxx +++ b/Framework/src/TaskRunnerFactory.cxx @@ -54,7 +54,8 @@ o2::framework::DataProcessorSpec TaskRunnerFactory::create(const TaskRunnerConfi newTask.labels.emplace_back(o2::framework::ecs::qcReconfigurable); newTask.labels.emplace_back(TaskRunner::getTaskRunnerLabel()); if (!taskConfig.critical) { - newTask.labels.emplace_back(TaskRunner::getExpendableLabel()); + framework::DataProcessorLabel expendableLabel = {"expendable"}; + newTask.labels.emplace_back(expendableLabel); } return newTask; From 1878436476753fbd8bf9855f8628d2cadea0a410 Mon Sep 17 00:00:00 2001 From: Barthelemy Date: Wed, 3 May 2023 10:01:14 +0200 Subject: [PATCH 05/21] make mergers critical if task is critical --- .../QualityControl/InfrastructureGenerator.h | 8 +++++--- Framework/src/CheckRunner.cxx | 3 ++- Framework/src/InfrastructureGenerator.cxx | 14 +++++--------- 3 files changed, 12 insertions(+), 13 deletions(-) diff --git a/Framework/include/QualityControl/InfrastructureGenerator.h b/Framework/include/QualityControl/InfrastructureGenerator.h index cc102fd150..fb3b17d321 100644 --- a/Framework/include/QualityControl/InfrastructureGenerator.h +++ b/Framework/include/QualityControl/InfrastructureGenerator.h @@ -219,9 +219,11 @@ class InfrastructureGenerator static void generateMergers(framework::WorkflowSpec& workflow, const std::string& taskName, size_t numberOfLocalMachines, std::vector> cycleDurationSeconds, - const std::string& mergingMode, size_t resetAfterCycles, - std::string monitoringUrl, const std::string& detectorName, - std::vector mergersPerLayer, bool enableMovingWindows); + const std::string& mergingMode, + size_t resetAfterCycles, + std::string monitoringUrl, + const std::string& detectorName, + std::vector mergersPerLayer, bool enableMovingWindows, bool critical); static void generateCheckRunners(framework::WorkflowSpec& workflow, const InfrastructureSpec& infrastructureSpec); static void generateAggregator(framework::WorkflowSpec& workflow, const InfrastructureSpec& infrastructureSpec); static void generatePostProcessing(framework::WorkflowSpec& workflow, const InfrastructureSpec& infrastructureSpec); diff --git a/Framework/src/CheckRunner.cxx b/Framework/src/CheckRunner.cxx index 70eba1120b..f5de5f91fe 100644 --- a/Framework/src/CheckRunner.cxx +++ b/Framework/src/CheckRunner.cxx @@ -357,6 +357,7 @@ QualityObjectsType CheckRunner::check() QualityObjectsType allQOs; for (auto& [checkName, check] : mChecks) { if (updatePolicyManager.isReady(check.getName())) { + ILOG(Debug, Support) << "Monitor Objects for the check '" << checkName << "' are ready --> check()" << ENDM; auto newQOs = check.check(mMonitorObjects); mTotalNumberCheckExecuted += newQOs.size(); @@ -366,7 +367,7 @@ QualityObjectsType CheckRunner::check() // Was checked, update latest revision updatePolicyManager.updateActorRevision(checkName); } else { - ILOG(Info, Support) << "Monitor Objects for the check '" << checkName << "' are not ready, ignoring" << ENDM; + ILOG(Debug, Support) << "Monitor Objects for the check '" << checkName << "' are not ready, ignoring" << ENDM; } } return allQOs; diff --git a/Framework/src/InfrastructureGenerator.cxx b/Framework/src/InfrastructureGenerator.cxx index 116be4ea70..d97ebf4615 100644 --- a/Framework/src/InfrastructureGenerator.cxx +++ b/Framework/src/InfrastructureGenerator.cxx @@ -277,9 +277,8 @@ o2::framework::WorkflowSpec InfrastructureGenerator::generateRemoteInfrastructur std::for_each(cycleDurationsMultiplied.begin(), cycleDurationsMultiplied.end(), [taskSpec](std::pair& p) { p.first *= taskSpec.mergerCycleMultiplier; }); bool enableMovingWindows = !taskSpec.movingWindows.empty(); - generateMergers(workflow, taskSpec.taskName, numberOfLocalMachines, cycleDurationsMultiplied, - taskSpec.mergingMode, resetAfterCycles, infrastructureSpec.common.monitoringUrl, - taskSpec.detectorName, taskSpec.mergersPerLayer, enableMovingWindows); + generateMergers(workflow, taskSpec.taskName, numberOfLocalMachines, cycleDurationsMultiplied, taskSpec.mergingMode, + resetAfterCycles, infrastructureSpec.common.monitoringUrl, taskSpec.detectorName, taskSpec.mergersPerLayer, enableMovingWindows, taskSpec.critical); } else if (taskSpec.location == TaskLocationSpec::Remote) { @@ -585,11 +584,9 @@ void InfrastructureGenerator::generateLocalTaskRemoteProxy(framework::WorkflowSp workflow.emplace_back(std::move(proxy)); } void InfrastructureGenerator::generateMergers(framework::WorkflowSpec& workflow, const std::string& taskName, - size_t numberOfLocalMachines, - std::vector> cycleDurations, - const std::string& mergingMode, size_t resetAfterCycles, - std::string monitoringUrl, const std::string& detectorName, - std::vector mergersPerLayer, bool enableMovingWindows) + size_t numberOfLocalMachines, std::vector> cycleDurations, + const std::string& mergingMode, size_t resetAfterCycles, std::string monitoringUrl, + const std::string& detectorName, std::vector mergersPerLayer, bool enableMovingWindows, bool critical) { Inputs mergerInputs; for (size_t id = 1; id <= numberOfLocalMachines; id++) { @@ -618,7 +615,6 @@ void InfrastructureGenerator::generateMergers(framework::WorkflowSpec& workflow, mergerConfig.monitoringUrl = std::move(monitoringUrl); mergerConfig.detectorName = detectorName; mergerConfig.parallelismType = { (mergerConfig.inputObjectTimespan.value == InputObjectsTimespan::LastDifference) ? ParallelismType::RoundRobin : ParallelismType::SplitInputs }; - mergerConfig.publishMovingWindow = { enableMovingWindows ? PublishMovingWindow::Yes : PublishMovingWindow::No }; mergersBuilder.setConfig(mergerConfig); mergersBuilder.generateInfrastructure(workflow); From f0612b67de1766d72239b024b2f1e5d212dfb2c7 Mon Sep 17 00:00:00 2001 From: Barthelemy Date: Thu, 4 May 2023 09:37:39 +0200 Subject: [PATCH 06/21] format --- Framework/src/AggregatorRunnerFactory.cxx | 2 +- Framework/src/CheckRunnerFactory.cxx | 4 ++-- Framework/src/TaskRunnerFactory.cxx | 2 +- 3 files changed, 4 insertions(+), 4 deletions(-) diff --git a/Framework/src/AggregatorRunnerFactory.cxx b/Framework/src/AggregatorRunnerFactory.cxx index 6d81ac9d6f..e29b3fa482 100644 --- a/Framework/src/AggregatorRunnerFactory.cxx +++ b/Framework/src/AggregatorRunnerFactory.cxx @@ -49,7 +49,7 @@ DataProcessorSpec AggregatorRunnerFactory::create(const core::CommonSpec& common }; newAggregatorRunner.labels.emplace_back(o2::framework::ecs::qcReconfigurable); newAggregatorRunner.labels.emplace_back(AggregatorRunner::getLabel()); - framework::DataProcessorLabel expendableLabel = {"expendable"}; + framework::DataProcessorLabel expendableLabel = { "expendable" }; newAggregatorRunner.labels.emplace_back(expendableLabel); newAggregatorRunner.algorithm = adaptFromTask(std::move(aggregatorRunner)); return newAggregatorRunner; diff --git a/Framework/src/CheckRunnerFactory.cxx b/Framework/src/CheckRunnerFactory.cxx index 30f4ffe23e..bbfe45f05d 100644 --- a/Framework/src/CheckRunnerFactory.cxx +++ b/Framework/src/CheckRunnerFactory.cxx @@ -45,7 +45,7 @@ DataProcessorSpec CheckRunnerFactory::create(CheckRunnerConfig checkRunnerConfig options }; newCheckRunner.labels.emplace_back(o2::framework::ecs::qcReconfigurable); newCheckRunner.labels.emplace_back(CheckRunner::getCheckRunnerLabel()); - framework::DataProcessorLabel expendableLabel = {"expendable"}; + framework::DataProcessorLabel expendableLabel = { "expendable" }; newCheckRunner.labels.emplace_back(expendableLabel); newCheckRunner.algorithm = adaptFromTask(std::move(qcCheckRunner)); return newCheckRunner; @@ -63,7 +63,7 @@ DataProcessorSpec CheckRunnerFactory::createSinkDevice(const CheckRunnerConfig& checkRunnerConfig.options, {}, { o2::framework::ecs::qcReconfigurable } }; - framework::DataProcessorLabel expendableLabel = {"expendable"}; + framework::DataProcessorLabel expendableLabel = { "expendable" }; newCheckRunner.labels.emplace_back(expendableLabel); return newCheckRunner; } diff --git a/Framework/src/TaskRunnerFactory.cxx b/Framework/src/TaskRunnerFactory.cxx index d0e628ad5c..ec6edc05eb 100644 --- a/Framework/src/TaskRunnerFactory.cxx +++ b/Framework/src/TaskRunnerFactory.cxx @@ -54,7 +54,7 @@ o2::framework::DataProcessorSpec TaskRunnerFactory::create(const TaskRunnerConfi newTask.labels.emplace_back(o2::framework::ecs::qcReconfigurable); newTask.labels.emplace_back(TaskRunner::getTaskRunnerLabel()); if (!taskConfig.critical) { - framework::DataProcessorLabel expendableLabel = {"expendable"}; + framework::DataProcessorLabel expendableLabel = { "expendable" }; newTask.labels.emplace_back(expendableLabel); } From 4d36f750c6de800fcdc72415aaf8ebd3d0f22c51 Mon Sep 17 00:00:00 2001 From: Barthelemy Date: Mon, 5 Jun 2023 15:17:43 +0200 Subject: [PATCH 07/21] indicate the default value for "critical", remove the unused "blocking parameter" --- Framework/basic.json | 5 ++--- 1 file changed, 2 insertions(+), 3 deletions(-) diff --git a/Framework/basic.json b/Framework/basic.json index 0106d8a5ff..9bf3ca6c98 100644 --- a/Framework/basic.json +++ b/Framework/basic.json @@ -39,7 +39,7 @@ "tasks": { "QcTask": { "active": "true", - "critical": "false", "": "if false the task is allowed to die without stopping the workflow", + "critical": "false", "": "if false the task is allowed to die without stopping the workflow, default: true", "className": "o2::quality_control_modules::skeleton::SkeletonTask", "moduleName": "QcSkeleton", "detectorName": "TST", @@ -106,8 +106,7 @@ "fraction": "0.1", "seed": "1234" } - ], - "blocking": "false" + ] } ] } From 253b156c070bdb2ba95cbc79e83a37b9f6668480 Mon Sep 17 00:00:00 2001 From: Barthelemy Date: Mon, 5 Jun 2023 15:23:34 +0200 Subject: [PATCH 08/21] doc --- doc/Advanced.md | 25 +++++++++++++++++++++++++ 1 file changed, 25 insertions(+) diff --git a/doc/Advanced.md b/doc/Advanced.md index 9f09bf0f27..a43291ce55 100644 --- a/doc/Advanced.md +++ b/doc/Advanced.md @@ -5,6 +5,7 @@ Advanced topics + * [Advanced topics](#advanced-topics) * [Framework](#framework) * [Plugging the QC to an existing DPL workflow](#plugging-the-qc-to-an-existing-dpl-workflow) * [Production of QC objects outside this framework](#production-of-qc-objects-outside-this-framework) @@ -18,6 +19,7 @@ Advanced topics * [Monitor cycles](#monitor-cycles) * [Writing a DPL data producer](#writing-a-dpl-data-producer) * [Custom merging](#custom-merging) + * [Critical and non-critical tasks](#critical-and-non-critical-tasks) * [QC with DPL Analysis](#qc-with-dpl-analysis) * [Uploading objects to QCDB](#uploading-objects-to-qcdb) * [Getting AODs in QC Tasks](#getting-aods-in-qc-tasks) @@ -502,6 +504,28 @@ Feel free to consult the existing usage examples among other modules in the QC r Once a custom class is implemented, one should let QCG know how to display it correctly, which is explained in the subsection [Display a non-standard ROOT object in QCG](#display-a-non-standard-root-object-in-qcg). +## Critical and non-critical tasks + +Some DPL devices can be marked as non-critical. It means that if they die the system will continue running. There is +obviously an impact as all the downstream devices won't get data. + +In QC, one can mark a task as critical or non-critical: + +```json + "tasks": { + "QcTask": { + "active": "true", + "critical": "false", "": "if false the task is allowed to die without stopping the workflow, default: true", +``` + +By default, they are critical. + +Mergers are critical or not based on the criticality of the task they are merging data for. + +Checkers are non-critical. + +Post-processing ??? + ## QC with DPL Analysis QC offers several ways to interact with the DPL Analysis framework. @@ -1291,6 +1315,7 @@ the "tasks" path. "className": "namespace::of::Task", "": "Class name of the QC Task with full namespace.", "moduleName": "QcSkeleton", "": "Library name. It can be found in CMakeLists of the detector module.", "detectorName": "TST", "": "3-letter code of the detector.", + "critical": "true", "": "if false the task is allowed to die without stopping the workflow, default: true", "cycleDurationSeconds": "10", "": "Cycle duration (how often objects are published), 10 seconds minimum.", "": "The first cycle will be randomly shorter. ", "": "Alternatively, one can specify different cycle durations for different periods. The last item in cycleDurations will be used for the rest of the duration whatever the period. The first cycle will be randomly shorter.", From c85be2df91b282916d7a1bbdcac0d8e852b66d93 Mon Sep 17 00:00:00 2001 From: Barthelemy Date: Mon, 5 Jun 2023 15:40:39 +0200 Subject: [PATCH 09/21] Post-processing --- .../QualityControl/PostProcessingConfig.h | 1 + .../QualityControl/PostProcessingTaskSpec.h | 1 + Framework/postprocessing.json | 1 + Framework/src/InfrastructureGenerator.cxx | 4 ++++ Framework/src/InfrastructureSpecReader.cxx | 3 ++- Framework/src/PostProcessingConfig.cxx | 3 ++- doc/Advanced.md | 19 ++++++++++++++++--- 7 files changed, 27 insertions(+), 5 deletions(-) diff --git a/Framework/include/QualityControl/PostProcessingConfig.h b/Framework/include/QualityControl/PostProcessingConfig.h index dbaff6f3b5..9f44a56a9d 100644 --- a/Framework/include/QualityControl/PostProcessingConfig.h +++ b/Framework/include/QualityControl/PostProcessingConfig.h @@ -45,6 +45,7 @@ struct PostProcessingConfig { std::string consulUrl; core::Activity activity; bool matchAnyRunNumber = false; + bool critical; }; } // namespace o2::quality_control::postprocessing diff --git a/Framework/include/QualityControl/PostProcessingTaskSpec.h b/Framework/include/QualityControl/PostProcessingTaskSpec.h index 9b83baebbc..12501a129e 100644 --- a/Framework/include/QualityControl/PostProcessingTaskSpec.h +++ b/Framework/include/QualityControl/PostProcessingTaskSpec.h @@ -41,6 +41,7 @@ struct PostProcessingTaskSpec { std::string id = "Invalid"; std::string taskName = "Invalid"; bool active = true; + bool critical = true; std::string detectorName = "Invalid"; boost::property_tree::ptree tree = {}; }; diff --git a/Framework/postprocessing.json b/Framework/postprocessing.json index 7c2e8a29db..5362227b9d 100644 --- a/Framework/postprocessing.json +++ b/Framework/postprocessing.json @@ -42,6 +42,7 @@ "postprocessing": { "ExamplePostprocessing": { "active": "true", + "critical": "false", "": "if false the task is allowed to die without stopping the workflow, default: true", "className": "o2::quality_control_modules::skeleton::SkeletonPostProcessing", "moduleName": "QcSkeleton", "detectorName": "TST", diff --git a/Framework/src/InfrastructureGenerator.cxx b/Framework/src/InfrastructureGenerator.cxx index d97ebf4615..e8f93eeb06 100644 --- a/Framework/src/InfrastructureGenerator.cxx +++ b/Framework/src/InfrastructureGenerator.cxx @@ -782,6 +782,10 @@ void InfrastructureGenerator::generatePostProcessing(WorkflowSpec& workflow, con ppTask.getOptions() }; dataProcessorSpec.labels.emplace_back(PostProcessingDevice::getLabel()); + if (!ppTaskSpec.critical) { + framework::DataProcessorLabel expendableLabel = { "expendable" }; + dataProcessorSpec.labels.emplace_back(expendableLabel); + } dataProcessorSpec.algorithm = adaptFromTask(std::move(ppTask)); workflow.emplace_back(std::move(dataProcessorSpec)); diff --git a/Framework/src/InfrastructureSpecReader.cxx b/Framework/src/InfrastructureSpecReader.cxx index 74b824fc53..7f96632c3f 100644 --- a/Framework/src/InfrastructureSpecReader.cxx +++ b/Framework/src/InfrastructureSpecReader.cxx @@ -110,7 +110,7 @@ TaskSpec InfrastructureSpecReader::readSpecEntry(const std::string& ta } ts.dataSource = readSpecEntry(taskID, taskTree.get_child("dataSource"), wholeTree); ts.active = taskTree.get("active", ts.active); - ts.critical = taskTree.get("critical", ts.active); + ts.critical = taskTree.get("critical", ts.critical); ts.maxNumberCycles = taskTree.get("maxNumberCycles", ts.maxNumberCycles); ts.resetAfterCycles = taskTree.get("resetAfterCycles", ts.resetAfterCycles); ts.saveObjectsToFile = taskTree.get("saveObjectsToFile", ts.saveObjectsToFile); @@ -382,6 +382,7 @@ PostProcessingTaskSpec ppts.id = ppTaskId; ppts.taskName = ppTaskTree.get("taskName", ppts.id); ppts.active = ppTaskTree.get("active", ppts.active); + ppts.critical = ppTaskTree.get("critical", ppts.critical); ppts.detectorName = ppTaskTree.get("detectorName", ppts.detectorName); ppts.tree = wholeTree; diff --git a/Framework/src/PostProcessingConfig.cxx b/Framework/src/PostProcessingConfig.cxx index 87b8bd2271..96884e2581 100644 --- a/Framework/src/PostProcessingConfig.cxx +++ b/Framework/src/PostProcessingConfig.cxx @@ -36,7 +36,8 @@ PostProcessingConfig::PostProcessingConfig(const std::string& id, const boost::p config.get("qc.config.Activity.provenance", "qc"), { config.get("qc.config.Activity.start", 0), config.get("qc.config.Activity.end", -1) }), - matchAnyRunNumber(config.get("qc.config.postprocessing.matchAnyRunNumber", false)) + matchAnyRunNumber(config.get("qc.config.postprocessing.matchAnyRunNumber", false)), + critical(true) { for (const auto& initTrigger : config.get_child("qc.postprocessing." + id + ".initTrigger")) { initTriggers.push_back(initTrigger.second.get_value()); diff --git a/doc/Advanced.md b/doc/Advanced.md index a43291ce55..b224e94003 100644 --- a/doc/Advanced.md +++ b/doc/Advanced.md @@ -509,22 +509,35 @@ Once a custom class is implemented, one should let QCG know how to display it co Some DPL devices can be marked as non-critical. It means that if they die the system will continue running. There is obviously an impact as all the downstream devices won't get data. -In QC, one can mark a task as critical or non-critical: +### QC tasks +In QC, one can mark a task as critical or non-critical: ```json "tasks": { "QcTask": { "active": "true", "critical": "false", "": "if false the task is allowed to die without stopping the workflow, default: true", ``` - By default, they are critical. +### QC mergers + Mergers are critical or not based on the criticality of the task they are merging data for. +### QC checkers + Checkers are non-critical. -Post-processing ??? +### QC post-processing tasks + +Post-processing tasks can be marked as critical or non-critical: +```json + "postprocessing": { + "ExamplePostprocessing": { + "active": "true", + "critical": "false", "": "if false the task is allowed to die without stopping the workflow, default: true", +``` +By default, they are critical. ## QC with DPL Analysis From a1d92dd7f4fe6ba2ad06ec56b591a20c149e3d38 Mon Sep 17 00:00:00 2001 From: Barthelemy Date: Wed, 27 Sep 2023 09:12:25 +0200 Subject: [PATCH 10/21] rebase with master --- Framework/include/QualityControl/InfrastructureGenerator.h | 4 +++- Framework/src/InfrastructureGenerator.cxx | 2 +- 2 files changed, 4 insertions(+), 2 deletions(-) diff --git a/Framework/include/QualityControl/InfrastructureGenerator.h b/Framework/include/QualityControl/InfrastructureGenerator.h index fb3b17d321..09484af311 100644 --- a/Framework/include/QualityControl/InfrastructureGenerator.h +++ b/Framework/include/QualityControl/InfrastructureGenerator.h @@ -223,7 +223,9 @@ class InfrastructureGenerator size_t resetAfterCycles, std::string monitoringUrl, const std::string& detectorName, - std::vector mergersPerLayer, bool enableMovingWindows, bool critical); + std::vector mergersPerLayer, + bool enableMovingWindows, + bool critical); static void generateCheckRunners(framework::WorkflowSpec& workflow, const InfrastructureSpec& infrastructureSpec); static void generateAggregator(framework::WorkflowSpec& workflow, const InfrastructureSpec& infrastructureSpec); static void generatePostProcessing(framework::WorkflowSpec& workflow, const InfrastructureSpec& infrastructureSpec); diff --git a/Framework/src/InfrastructureGenerator.cxx b/Framework/src/InfrastructureGenerator.cxx index e8f93eeb06..0e24e0783a 100644 --- a/Framework/src/InfrastructureGenerator.cxx +++ b/Framework/src/InfrastructureGenerator.cxx @@ -144,7 +144,7 @@ framework::WorkflowSpec InfrastructureGenerator::generateFullChainInfrastructure bool enableMovingWindows = !taskSpec.movingWindows.empty(); generateMergers(workflow, taskSpec.taskName, 1, cycleDurationsMultiplied, taskSpec.mergingMode, resetAfterCycles, infrastructureSpec.common.monitoringUrl, - taskSpec.detectorName, taskSpec.mergersPerLayer, enableMovingWindows); + taskSpec.detectorName, taskSpec.mergersPerLayer, enableMovingWindows, taskSpec.critical); } else { // TaskLocationSpec::Remote auto taskConfig = TaskRunnerFactory::extractConfig(infrastructureSpec.common, taskSpec, 0, taskSpec.resetAfterCycles); workflow.emplace_back(TaskRunnerFactory::create(taskConfig)); From 5f6fd517a163dbba683b83142b46d1867542c003 Mon Sep 17 00:00:00 2001 From: Barthelemy Date: Tue, 28 Nov 2023 15:27:57 +0100 Subject: [PATCH 11/21] proxies should be expendable if the task is expendable --- Framework/src/InfrastructureGenerator.cxx | 8 ++++++ Modules/CMakeLists.txt | 32 +++++++++++------------ 2 files changed, 24 insertions(+), 16 deletions(-) diff --git a/Framework/src/InfrastructureGenerator.cxx b/Framework/src/InfrastructureGenerator.cxx index 0e24e0783a..7c8a3f6c19 100644 --- a/Framework/src/InfrastructureGenerator.cxx +++ b/Framework/src/InfrastructureGenerator.cxx @@ -549,6 +549,10 @@ void InfrastructureGenerator::generateLocalTaskLocalProxy(framework::WorkflowSpe { proxyInput }, channelConfig.c_str())); workflow.back().labels.emplace_back(taskSpec.localControl == "odc" ? ecs::preserveRawChannelsLabel : ecs::uniqueProxyLabel); + if(!taskSpec.critical) { + framework::DataProcessorLabel expendableLabel = { "expendable" }; + workflow.back().labels.emplace_back(expendableLabel); + } if (getenv("O2_QC_KILL_PROXIES") != nullptr) { workflow.back().metadata.push_back(DataProcessorMetadata{ ecs::privateMemoryKillThresholdMB, proxyMemoryKillThresholdMB }); } @@ -576,6 +580,10 @@ void InfrastructureGenerator::generateLocalTaskRemoteProxy(framework::WorkflowSp channelConfig.c_str(), dplModelAdaptor()); proxy.labels.emplace_back(taskSpec.localControl == "odc" ? ecs::preserveRawChannelsLabel : ecs::uniqueProxyLabel); + if(!taskSpec.critical) { + framework::DataProcessorLabel expendableLabel = { "expendable" }; + workflow.back().labels.emplace_back(expendableLabel); + } // if not in RUNNING, we should drop all the incoming messages, we set the corresponding proxy option. enableDraining(proxy.options); if (getenv("O2_QC_KILL_PROXIES") != nullptr) { diff --git a/Modules/CMakeLists.txt b/Modules/CMakeLists.txt index ab7a6d46e3..1bd8c65857 100644 --- a/Modules/CMakeLists.txt +++ b/Modules/CMakeLists.txt @@ -5,19 +5,19 @@ add_subdirectory(Daq) add_subdirectory(Example) add_subdirectory(Benchmark) add_subdirectory(Skeleton) -add_subdirectory(TOF) -add_subdirectory(EMCAL) -add_subdirectory(MUON) -add_subdirectory(TPC) -add_subdirectory(ITS) -add_subdirectory(MFT) -add_subdirectory(PHOS) -add_subdirectory(FIT) -add_subdirectory(TRD) -add_subdirectory(HMPID) -add_subdirectory(CPV) -add_subdirectory(GLO) -add_subdirectory(ZDC) -add_subdirectory(CTP) -add_subdirectory(PID) -add_subdirectory(FOCAL) +#add_subdirectory(TOF) +#add_subdirectory(EMCAL) +#add_subdirectory(MUON) +#add_subdirectory(TPC) +#add_subdirectory(ITS) +#add_subdirectory(MFT) +#add_subdirectory(PHOS) +#add_subdirectory(FIT) +#add_subdirectory(TRD) +#add_subdirectory(HMPID) +#add_subdirectory(CPV) +#add_subdirectory(GLO) +#add_subdirectory(ZDC) +#add_subdirectory(CTP) +#add_subdirectory(PID) +#add_subdirectory(FOCAL) From deb886f49409faedcfe62d5eee00face9841d967 Mon Sep 17 00:00:00 2001 From: Barthelemy Date: Tue, 28 Nov 2023 15:29:45 +0100 Subject: [PATCH 12/21] proxies should be expendable if the task is expendable --- Modules/CMakeLists.txt | 32 ++++++++++++++++---------------- 1 file changed, 16 insertions(+), 16 deletions(-) diff --git a/Modules/CMakeLists.txt b/Modules/CMakeLists.txt index 1bd8c65857..ab7a6d46e3 100644 --- a/Modules/CMakeLists.txt +++ b/Modules/CMakeLists.txt @@ -5,19 +5,19 @@ add_subdirectory(Daq) add_subdirectory(Example) add_subdirectory(Benchmark) add_subdirectory(Skeleton) -#add_subdirectory(TOF) -#add_subdirectory(EMCAL) -#add_subdirectory(MUON) -#add_subdirectory(TPC) -#add_subdirectory(ITS) -#add_subdirectory(MFT) -#add_subdirectory(PHOS) -#add_subdirectory(FIT) -#add_subdirectory(TRD) -#add_subdirectory(HMPID) -#add_subdirectory(CPV) -#add_subdirectory(GLO) -#add_subdirectory(ZDC) -#add_subdirectory(CTP) -#add_subdirectory(PID) -#add_subdirectory(FOCAL) +add_subdirectory(TOF) +add_subdirectory(EMCAL) +add_subdirectory(MUON) +add_subdirectory(TPC) +add_subdirectory(ITS) +add_subdirectory(MFT) +add_subdirectory(PHOS) +add_subdirectory(FIT) +add_subdirectory(TRD) +add_subdirectory(HMPID) +add_subdirectory(CPV) +add_subdirectory(GLO) +add_subdirectory(ZDC) +add_subdirectory(CTP) +add_subdirectory(PID) +add_subdirectory(FOCAL) From a9740db63796399e3eb376e3f0cd523721bded10 Mon Sep 17 00:00:00 2001 From: Barthelemy Date: Wed, 29 Nov 2023 15:54:46 +0100 Subject: [PATCH 13/21] Add the resilient label to the mergers --- Framework/src/InfrastructureGenerator.cxx | 1 + 1 file changed, 1 insertion(+) diff --git a/Framework/src/InfrastructureGenerator.cxx b/Framework/src/InfrastructureGenerator.cxx index 7c8a3f6c19..1e4499427c 100644 --- a/Framework/src/InfrastructureGenerator.cxx +++ b/Framework/src/InfrastructureGenerator.cxx @@ -622,6 +622,7 @@ void InfrastructureGenerator::generateMergers(framework::WorkflowSpec& workflow, mergerConfig.topologySize = { TopologySize::MergersPerLayer, mergersPerLayer }; mergerConfig.monitoringUrl = std::move(monitoringUrl); mergerConfig.detectorName = detectorName; + mergerConfig.labels.push_back({"resilient"}); mergerConfig.parallelismType = { (mergerConfig.inputObjectTimespan.value == InputObjectsTimespan::LastDifference) ? ParallelismType::RoundRobin : ParallelismType::SplitInputs }; mergersBuilder.setConfig(mergerConfig); From dbf1a62a0c87c7c115c3f53876e43df1efdd0b59 Mon Sep 17 00:00:00 2001 From: Barthelemy Date: Wed, 29 Nov 2023 16:56:35 +0100 Subject: [PATCH 14/21] make checkers resilient --- Framework/src/CheckRunnerFactory.cxx | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/Framework/src/CheckRunnerFactory.cxx b/Framework/src/CheckRunnerFactory.cxx index bbfe45f05d..adb0e5e2c9 100644 --- a/Framework/src/CheckRunnerFactory.cxx +++ b/Framework/src/CheckRunnerFactory.cxx @@ -45,8 +45,8 @@ DataProcessorSpec CheckRunnerFactory::create(CheckRunnerConfig checkRunnerConfig options }; newCheckRunner.labels.emplace_back(o2::framework::ecs::qcReconfigurable); newCheckRunner.labels.emplace_back(CheckRunner::getCheckRunnerLabel()); - framework::DataProcessorLabel expendableLabel = { "expendable" }; - newCheckRunner.labels.emplace_back(expendableLabel); + framework::DataProcessorLabel resilientLabel = { "resilient" }; + newCheckRunner.labels.emplace_back(resilientLabel); newCheckRunner.algorithm = adaptFromTask(std::move(qcCheckRunner)); return newCheckRunner; } @@ -63,8 +63,8 @@ DataProcessorSpec CheckRunnerFactory::createSinkDevice(const CheckRunnerConfig& checkRunnerConfig.options, {}, { o2::framework::ecs::qcReconfigurable } }; - framework::DataProcessorLabel expendableLabel = { "expendable" }; - newCheckRunner.labels.emplace_back(expendableLabel); + framework::DataProcessorLabel resilientLabel = { "resilient" }; + newCheckRunner.labels.emplace_back(resilientLabel); return newCheckRunner; } From 88eb20d323b30ba06f1e737f5b2dc066990616f5 Mon Sep 17 00:00:00 2001 From: Barthelemy Date: Wed, 29 Nov 2023 17:01:05 +0100 Subject: [PATCH 15/21] make aggregators resilient --- Framework/src/AggregatorRunnerFactory.cxx | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/Framework/src/AggregatorRunnerFactory.cxx b/Framework/src/AggregatorRunnerFactory.cxx index e29b3fa482..ef1ffc38a3 100644 --- a/Framework/src/AggregatorRunnerFactory.cxx +++ b/Framework/src/AggregatorRunnerFactory.cxx @@ -49,8 +49,8 @@ DataProcessorSpec AggregatorRunnerFactory::create(const core::CommonSpec& common }; newAggregatorRunner.labels.emplace_back(o2::framework::ecs::qcReconfigurable); newAggregatorRunner.labels.emplace_back(AggregatorRunner::getLabel()); - framework::DataProcessorLabel expendableLabel = { "expendable" }; - newAggregatorRunner.labels.emplace_back(expendableLabel); + framework::DataProcessorLabel resilientLabel = { "resilient" }; + newAggregatorRunner.labels.emplace_back(resilientLabel); newAggregatorRunner.algorithm = adaptFromTask(std::move(aggregatorRunner)); return newAggregatorRunner; } From e2bfaaeb34831f9e12c14e27fbfd4ef73ac63ff6 Mon Sep 17 00:00:00 2001 From: Barthelemy Date: Wed, 29 Nov 2023 17:02:48 +0100 Subject: [PATCH 16/21] format --- Framework/src/InfrastructureGenerator.cxx | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/Framework/src/InfrastructureGenerator.cxx b/Framework/src/InfrastructureGenerator.cxx index 1e4499427c..12e43ffa41 100644 --- a/Framework/src/InfrastructureGenerator.cxx +++ b/Framework/src/InfrastructureGenerator.cxx @@ -549,7 +549,7 @@ void InfrastructureGenerator::generateLocalTaskLocalProxy(framework::WorkflowSpe { proxyInput }, channelConfig.c_str())); workflow.back().labels.emplace_back(taskSpec.localControl == "odc" ? ecs::preserveRawChannelsLabel : ecs::uniqueProxyLabel); - if(!taskSpec.critical) { + if (!taskSpec.critical) { framework::DataProcessorLabel expendableLabel = { "expendable" }; workflow.back().labels.emplace_back(expendableLabel); } @@ -580,7 +580,7 @@ void InfrastructureGenerator::generateLocalTaskRemoteProxy(framework::WorkflowSp channelConfig.c_str(), dplModelAdaptor()); proxy.labels.emplace_back(taskSpec.localControl == "odc" ? ecs::preserveRawChannelsLabel : ecs::uniqueProxyLabel); - if(!taskSpec.critical) { + if (!taskSpec.critical) { framework::DataProcessorLabel expendableLabel = { "expendable" }; workflow.back().labels.emplace_back(expendableLabel); } @@ -622,7 +622,7 @@ void InfrastructureGenerator::generateMergers(framework::WorkflowSpec& workflow, mergerConfig.topologySize = { TopologySize::MergersPerLayer, mergersPerLayer }; mergerConfig.monitoringUrl = std::move(monitoringUrl); mergerConfig.detectorName = detectorName; - mergerConfig.labels.push_back({"resilient"}); + mergerConfig.labels.push_back({ "resilient" }); mergerConfig.parallelismType = { (mergerConfig.inputObjectTimespan.value == InputObjectsTimespan::LastDifference) ? ParallelismType::RoundRobin : ParallelismType::SplitInputs }; mergersBuilder.setConfig(mergerConfig); From c3f66d4d8c8d3ce20babc2a5052b7e64d7e0271f Mon Sep 17 00:00:00 2001 From: Barthelemy Date: Thu, 30 Nov 2023 16:21:03 +0100 Subject: [PATCH 17/21] bad merge --- Framework/src/InfrastructureGenerator.cxx | 1 + 1 file changed, 1 insertion(+) diff --git a/Framework/src/InfrastructureGenerator.cxx b/Framework/src/InfrastructureGenerator.cxx index 12e43ffa41..515eb1c394 100644 --- a/Framework/src/InfrastructureGenerator.cxx +++ b/Framework/src/InfrastructureGenerator.cxx @@ -623,6 +623,7 @@ void InfrastructureGenerator::generateMergers(framework::WorkflowSpec& workflow, mergerConfig.monitoringUrl = std::move(monitoringUrl); mergerConfig.detectorName = detectorName; mergerConfig.labels.push_back({ "resilient" }); + mergerConfig.publishMovingWindow = { enableMovingWindows ? PublishMovingWindow::Yes : PublishMovingWindow::No }; mergerConfig.parallelismType = { (mergerConfig.inputObjectTimespan.value == InputObjectsTimespan::LastDifference) ? ParallelismType::RoundRobin : ParallelismType::SplitInputs }; mergersBuilder.setConfig(mergerConfig); From 074e92b1320565458d075bf4c2d2ad1ed78a0598 Mon Sep 17 00:00:00 2001 From: Barthelemy Date: Mon, 18 Dec 2023 11:24:00 +0100 Subject: [PATCH 18/21] single line labels --- Framework/src/CheckRunnerFactory.cxx | 6 ++---- Framework/src/InfrastructureGenerator.cxx | 6 ++---- 2 files changed, 4 insertions(+), 8 deletions(-) diff --git a/Framework/src/CheckRunnerFactory.cxx b/Framework/src/CheckRunnerFactory.cxx index adb0e5e2c9..9360dbb5de 100644 --- a/Framework/src/CheckRunnerFactory.cxx +++ b/Framework/src/CheckRunnerFactory.cxx @@ -45,8 +45,7 @@ DataProcessorSpec CheckRunnerFactory::create(CheckRunnerConfig checkRunnerConfig options }; newCheckRunner.labels.emplace_back(o2::framework::ecs::qcReconfigurable); newCheckRunner.labels.emplace_back(CheckRunner::getCheckRunnerLabel()); - framework::DataProcessorLabel resilientLabel = { "resilient" }; - newCheckRunner.labels.emplace_back(resilientLabel); + newCheckRunner.labels.emplace_back(framework::DataProcessorLabel{ "resilient" }); newCheckRunner.algorithm = adaptFromTask(std::move(qcCheckRunner)); return newCheckRunner; } @@ -63,8 +62,7 @@ DataProcessorSpec CheckRunnerFactory::createSinkDevice(const CheckRunnerConfig& checkRunnerConfig.options, {}, { o2::framework::ecs::qcReconfigurable } }; - framework::DataProcessorLabel resilientLabel = { "resilient" }; - newCheckRunner.labels.emplace_back(resilientLabel); + newCheckRunner.labels.emplace_back(framework::DataProcessorLabel{ "resilient" }); return newCheckRunner; } diff --git a/Framework/src/InfrastructureGenerator.cxx b/Framework/src/InfrastructureGenerator.cxx index 7ca5433b4c..dbc70961ff 100644 --- a/Framework/src/InfrastructureGenerator.cxx +++ b/Framework/src/InfrastructureGenerator.cxx @@ -550,8 +550,7 @@ void InfrastructureGenerator::generateLocalTaskLocalProxy(framework::WorkflowSpe channelConfig.c_str())); workflow.back().labels.emplace_back(taskSpec.localControl == "odc" ? ecs::preserveRawChannelsLabel : ecs::uniqueProxyLabel); if (!taskSpec.critical) { - framework::DataProcessorLabel expendableLabel = { "expendable" }; - workflow.back().labels.emplace_back(expendableLabel); + workflow.back().labels.emplace_back(framework::DataProcessorLabel{ "expendable" }); } if (getenv("O2_QC_KILL_PROXIES") != nullptr) { workflow.back().metadata.push_back(DataProcessorMetadata{ ecs::privateMemoryKillThresholdMB, proxyMemoryKillThresholdMB }); @@ -581,8 +580,7 @@ void InfrastructureGenerator::generateLocalTaskRemoteProxy(framework::WorkflowSp dplModelAdaptor()); proxy.labels.emplace_back(taskSpec.localControl == "odc" ? ecs::preserveRawChannelsLabel : ecs::uniqueProxyLabel); if (!taskSpec.critical) { - framework::DataProcessorLabel expendableLabel = { "expendable" }; - workflow.back().labels.emplace_back(expendableLabel); + workflow.back().labels.emplace_back(framework::DataProcessorLabel{ "expendable" }); } // if not in RUNNING, we should drop all the incoming messages, we set the corresponding proxy option. enableDraining(proxy.options); From e8baf3c100abac68af80cec9b424320ac09423f2 Mon Sep 17 00:00:00 2001 From: Barthelemy Date: Mon, 18 Dec 2023 11:32:47 +0100 Subject: [PATCH 19/21] update doc --- doc/Advanced.md | 27 ++++++++++++++++++++------- 1 file changed, 20 insertions(+), 7 deletions(-) diff --git a/doc/Advanced.md b/doc/Advanced.md index b224e94003..2eda56aa84 100644 --- a/doc/Advanced.md +++ b/doc/Advanced.md @@ -504,10 +504,13 @@ Feel free to consult the existing usage examples among other modules in the QC r Once a custom class is implemented, one should let QCG know how to display it correctly, which is explained in the subsection [Display a non-standard ROOT object in QCG](#display-a-non-standard-root-object-in-qcg). -## Critical and non-critical tasks +## Critical, resilient and non-critical tasks -Some DPL devices can be marked as non-critical. It means that if they die the system will continue running. There is -obviously an impact as all the downstream devices won't get data. +DPL devices can be marked as expendable, resilient or critical. Expendable tasks can die without affecting the run. +Resilient tasks can survive having one or all their inputs coming from an expendable task. +Critical tasks (default) will stop the system if they die and will not accept input from expendable tasks. + +In QC we use these `labels`. ### QC tasks @@ -518,15 +521,24 @@ In QC, one can mark a task as critical or non-critical: "active": "true", "critical": "false", "": "if false the task is allowed to die without stopping the workflow, default: true", ``` -By default, they are critical. +By default they are `critical` meaning that their failure will stop the run. +If they are not critical, they will be `expendable` and will not stop the run if they die. + +### Auto-generated proxies + +They adopt the criticality of the task they are proxying. ### QC mergers -Mergers are critical or not based on the criticality of the task they are merging data for. +Mergers are `resilient`. ### QC checkers -Checkers are non-critical. +Checkers are `resilient`. + +### QC aggregators + +Aggregators are `resilient`. ### QC post-processing tasks @@ -537,7 +549,8 @@ Post-processing tasks can be marked as critical or non-critical: "active": "true", "critical": "false", "": "if false the task is allowed to die without stopping the workflow, default: true", ``` -By default, they are critical. +By default, they are critical meaning that their failure will stop the run. +If they are not critical, they will be `expendable` and will not stop the run if they die. ## QC with DPL Analysis From 52d50f511708ea7ff2809ebc9b5a711a0af49309 Mon Sep 17 00:00:00 2001 From: Barthelemy Date: Mon, 18 Dec 2023 11:33:58 +0100 Subject: [PATCH 20/21] update doc --- doc/Advanced.md | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/doc/Advanced.md b/doc/Advanced.md index 2eda56aa84..b2c4c931c7 100644 --- a/doc/Advanced.md +++ b/doc/Advanced.md @@ -532,9 +532,9 @@ They adopt the criticality of the task they are proxying. Mergers are `resilient`. -### QC checkers +### QC check runners -Checkers are `resilient`. +CheckRunners are `resilient`. ### QC aggregators From ed144c36f9c36b3bc7b673b5c5605f9d4004bcad Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Barth=C3=A9l=C3=A9my=20von=20Haller?= Date: Mon, 18 Dec 2023 13:42:46 +0100 Subject: [PATCH 21/21] Update Advanced.md --- doc/Advanced.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/doc/Advanced.md b/doc/Advanced.md index b2c4c931c7..dd85003604 100644 --- a/doc/Advanced.md +++ b/doc/Advanced.md @@ -507,7 +507,7 @@ Once a custom class is implemented, one should let QCG know how to display it co ## Critical, resilient and non-critical tasks DPL devices can be marked as expendable, resilient or critical. Expendable tasks can die without affecting the run. -Resilient tasks can survive having one or all their inputs coming from an expendable task. +Resilient tasks can survive having one or all their inputs coming from an expendable task but they will stop the system if they themselves die. Critical tasks (default) will stop the system if they die and will not accept input from expendable tasks. In QC we use these `labels`.