Skip to content

Commit 4984e60

Browse files
knopers8knopers8
authored
[QC-1088] Late Tasks as post-processing within the message passing framework (#2665)
* [QC-1088] Late Tasks as post-processing within the message passing framework Late Tasks can create new MonitorObjects based on existing MOs and QOs in the QC workflow. I called them "late", because they would be typically executed close to the end of the processing chain. When running QC workflows distributed over multiple nodes, with Mergers combining the results, they can be used to process the merged results. The adequate use cases involve: * creating trends and correlations from other MonitorObjects and QualityObjects (trending a histogram average, trending quality) * creating plots which can only be constructed from merged objects (ratios, visualizations, ...) * creating summary canvases of QualityObjects available in the QC workflow They are meant to slowly replace Post-Processing, at least in the use cases mentioned above. * Comment and documentation cleanups * use unique keys for MOs when creating inputs for late tasks * Install example config and update to the latest ccdb * Publish the last objects version at end of stream in late tasks
1 parent 1ef72da commit 4984e60

35 files changed

Lines changed: 1365 additions & 27 deletions

‎.clang-format-ignore‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,2 @@
1+
*.json
2+
Framework/*.json

‎Framework/CMakeLists.txt‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -140,6 +140,9 @@ add_library(O2QualityControl
140140
src/QCInputsAdapters.cxx
141141
src/QCInputsFactory.cxx
142142
src/UserInputOutput.cxx
143+
src/LateTaskRunner.cxx
144+
src/LateTaskRunnerFactory.cxx
145+
src/LateTaskInterface.cxx
143146
src/Actor.cxx
144147
src/ActorHelpers.cxx
145148
src/DataProcessorAdapter.cxx
@@ -185,6 +188,7 @@ add_root_dictionary(O2QualityControl
185188
HEADERS
186189
include/QualityControl/CheckInterface.h
187190
include/QualityControl/TaskInterface.h
191+
include/QualityControl/LateTaskInterface.h
188192
include/QualityControl/UserCodeInterface.h
189193
include/QualityControl/AggregatorInterface.h
190194
include/QualityControl/PostProcessingInterface.h
@@ -491,6 +495,7 @@ install(FILES example-default.json
491495
basic-aggregator.json
492496
basic-external-histo.json
493497
basic-no-sampling.json
498+
basic-late-task.json
494499
advanced.json
495500
advanced-aggregator.json
496501
advanced-external-histo.json

‎Framework/basic-late-task.json‎

Lines changed: 103 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,103 @@
1+
{
2+
"qc": {
3+
"config": {
4+
"database": {
5+
"implementation": "CCDB",
6+
"host": "localhost:8083",
7+
"maxObjectSize": "2097152", "": "[Bytes, default=2MB] Maximum size allowed, larger objects are rejected."
8+
},
9+
"Activity": {
10+
"number": "42",
11+
"type": "NONE",
12+
"periodName": "", "": "Period name - e.g. LHC22c, LHC22c1b_test",
13+
"passName": "", "": "Pass type - e.g. spass, cpass1",
14+
"provenance": "qc", "": "Provenance - qc or qc_mc depending whether it is normal data or monte carlo data"
15+
},
16+
"monitoring": {
17+
"url": "infologger:///debug?qc"
18+
},
19+
"consul": {
20+
"url": ""
21+
},
22+
"conditionDB": {
23+
"url": "ccdb-test.cern.ch:8080"
24+
},
25+
"infologger": { "": "Configuration of the Infologger (optional).",
26+
"filterDiscardDebug": "false", "": "Set to true to discard debug and trace messages (default: false)",
27+
"filterDiscardLevel": "12", "": "Message at this level or above are discarded (default: 21 - Trace)",
28+
"filterDiscardFile": "/tmp/_ID_.txt", "": ["If set, the messages discarded because of filterDiscardLevel",
29+
"will go to this file (default: <none>); The keyword _ID_ is replaced by the device id. Discarded Debug ",
30+
"messages won't go there."]
31+
},
32+
"bookkeeping": {
33+
"url": ""
34+
}
35+
},
36+
"tasks": {
37+
"QcTask": {
38+
"active": "true",
39+
"className": "o2::quality_control_modules::skeleton::SkeletonTask",
40+
"moduleName": "QcSkeleton",
41+
"detectorName": "TST",
42+
"cycleDurationSeconds": "10",
43+
"resetAfterCycles": "1",
44+
"dataSource": {
45+
"type": "dataSamplingPolicy",
46+
"name": "tst-raw"
47+
},
48+
"movingWindows": ["example"],
49+
"location": "local"
50+
}
51+
},
52+
"checks": {
53+
"QcCheck": {
54+
"active": "true",
55+
"className": "o2::quality_control_modules::skeleton::SkeletonCheck",
56+
"moduleName": "QcSkeleton",
57+
"policy": "OnAny",
58+
"detectorName": "TST",
59+
"dataSource": [{
60+
"type": "Task",
61+
"name": "QcTask",
62+
"MOs": ["example"]
63+
}],
64+
"extendedCheckParameters": {
65+
"physics": {
66+
"pp": {
67+
"myOwnKey1": "myOwnValue1c"
68+
}
69+
}
70+
}
71+
}
72+
},
73+
"lateTasks": {
74+
"late": {
75+
"active": "true",
76+
"className": "o2::quality_control_modules::skeleton::SkeletonLateTask",
77+
"moduleName": "QcSkeleton",
78+
"detectorName": "TST",
79+
"dataSources": [{
80+
"type": "Task",
81+
"name": "QcTask",
82+
"MOs": ["example", "example2"]
83+
}],
84+
"outputActivityStrategy": "integrated"
85+
}
86+
}
87+
},
88+
"dataSamplingPolicies": [
89+
{
90+
"id": "tst-raw",
91+
"active": "true",
92+
"machines": [],
93+
"query": "data:TST/RAWDATA/0",
94+
"samplingConditions": [
95+
{
96+
"condition": "random",
97+
"fraction": "0.1",
98+
"seed": "1234"
99+
}
100+
]
101+
}
102+
]
103+
}

‎Framework/include/QualityControl/Actor.h‎

Lines changed: 4 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -175,7 +175,7 @@ class Actor
175175
void onStart(framework::ServiceRegistryRef services, const Activity& activity) {}
176176
void onStop(framework::ServiceRegistryRef services, const Activity& activity) {}
177177
void onReset(framework::ServiceRegistryRef services, const Activity& activity) {}
178-
void onEndOfStream(framework::EndOfStreamContext& eosContext) {}
178+
void onEndOfStream(framework::EndOfStreamContext& eosContext, const Activity& activity) {}
179179
void onFinaliseCCDB(framework::ConcreteDataMatcher& matcher, void* obj) {}
180180

181181
// service access for concrete actor
@@ -345,7 +345,9 @@ class Actor
345345
impl::handleExceptions("endOfStream", [&] {
346346
ILOG(Debug, Trace) << traits::sActorTypeKebabCase << " endOfStream" << ENDM;
347347

348-
concreteActor().onEndOfStream(eosContext);
348+
mActivity = mServicesConfig.activity;
349+
350+
concreteActor().onEndOfStream(eosContext, mActivity);
349351
});
350352
}
351353
void finaliseCCDB(framework::ConcreteDataMatcher& matcher, void* obj)

‎Framework/include/QualityControl/DataSourceType.h‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -28,10 +28,10 @@ enum class DataSourceType {
2828
Check,
2929
Aggregator,
3030
PostProcessingTask,
31+
LateTask,
3132
ExternalTask,
3233
Invalid
3334
};
34-
3535
}
3636

3737
#endif // QUALITYCONTROL_DATASOURCETYPE_H

‎Framework/include/QualityControl/InfrastructureGenerator.h‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -230,6 +230,7 @@ class InfrastructureGenerator
230230
static void generateAggregator(framework::WorkflowSpec& workflow, const InfrastructureSpec& infrastructureSpec);
231231
static void generatePostProcessing(framework::WorkflowSpec& workflow, const InfrastructureSpec& infrastructureSpec);
232232
static void generateBookkeepingQualitySink(framework::WorkflowSpec& workflow, const InfrastructureSpec& infrastructureSpec);
233+
static void generateLateTasks(framework::WorkflowSpec& workflow, const InfrastructureSpec& infrastructureSpec);
233234
};
234235

235236
} // namespace core

‎Framework/include/QualityControl/InfrastructureSpec.h‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -23,6 +23,7 @@
2323
#include "QualityControl/AggregatorSpec.h"
2424
#include "QualityControl/PostProcessingTaskSpec.h"
2525
#include "QualityControl/ExternalTaskSpec.h"
26+
#include "QualityControl/LateTaskSpec.h"
2627

2728
#include <vector>
2829

@@ -37,6 +38,7 @@ struct InfrastructureSpec {
3738
std::vector<checker::AggregatorSpec> aggregators;
3839
std::vector<postprocessing::PostProcessingTaskSpec> postProcessingTasks;
3940
std::vector<ExternalTaskSpec> externalTasks;
41+
std::vector<LateTaskSpec> lateTasks;
4042
};
4143

4244
} // namespace o2::quality_control::core

‎Framework/include/QualityControl/InfrastructureSpecReader.h‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -22,6 +22,7 @@
2222
#include "QualityControl/DataSourceSpec.h"
2323
#include "QualityControl/CheckSpec.h"
2424
#include "QualityControl/PostProcessingTaskSpec.h"
25+
#include "QualityControl/LateTaskSpec.h"
2526
#include "QualityControl/RecoRequestSpecs.h"
2627
#include <boost/property_tree/ptree_fwd.hpp>
2728

@@ -52,6 +53,8 @@ checker::AggregatorSpec readSpecEntry<checker::AggregatorSpec>(const std::string
5253
template <>
5354
postprocessing::PostProcessingTaskSpec readSpecEntry<postprocessing::PostProcessingTaskSpec>(const std::string& entryID, const boost::property_tree::ptree& entryTree, const boost::property_tree::ptree& wholeTree);
5455
template <>
56+
LateTaskSpec readSpecEntry<LateTaskSpec>(const std::string& entryID, const boost::property_tree::ptree& entryTree, const boost::property_tree::ptree& wholeTree);
57+
template <>
5558
ExternalTaskSpec readSpecEntry<ExternalTaskSpec>(const std::string& entryID, const boost::property_tree::ptree& entryTree, const boost::property_tree::ptree& wholeTree);
5659
template <>
5760
GRPGeomRequestSpec readSpecEntry<GRPGeomRequestSpec>(const std::string& entryID, const boost::property_tree::ptree& entryTree, const boost::property_tree::ptree& wholeTree);

‎Framework/include/QualityControl/LateTaskConfig.h‎

Lines changed: 36 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,36 @@
1+
// Copyright 2019-2020 CERN and copyright holders of ALICE O2.
2+
// See https://alice-o2.web.cern.ch/copyright for details of the copyright holders.
3+
// All rights not expressly granted are reserved.
4+
//
5+
// This software is distributed under the terms of the GNU General Public
6+
// License v3 (GPL Version 3), copied verbatim in the file "COPYING".
7+
//
8+
// In applying this license CERN does not waive the privileges and immunities
9+
// granted to it by virtue of its status as an Intergovernmental Organization
10+
// or submit itself to any jurisdiction.
11+
12+
#ifndef QC_CORE_LATETASKRUNNERCONFIG_H
13+
#define QC_CORE_LATETASKRUNNERCONFIG_H
14+
15+
///
16+
/// \file LateTaskRunnerConfig.h
17+
/// \author Piotr Konopka
18+
///
19+
20+
#include <Framework/DataProcessorSpec.h>
21+
22+
#include "QualityControl/UserCodeConfig.h"
23+
#include "QualityControl/DataSourceSpec.h"
24+
#include "QualityControl/OutputActivityStrategy.h"
25+
26+
namespace o2::quality_control::core
27+
{
28+
29+
struct LateTaskConfig : public UserCodeConfig {
30+
bool critical = true;
31+
OutputActivityStrategy outputActivityStrategy = OutputActivityStrategy::Integrated;
32+
};
33+
34+
} // namespace o2::quality_control::core
35+
36+
#endif // QC_CORE_LATETASKRUNNERCONFIG_H

‎Framework/include/QualityControl/LateTaskFactory.h‎

Lines changed: 58 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,58 @@
1+
// Copyright 2019-2020 CERN and copyright holders of ALICE O2.
2+
// See https://alice-o2.web.cern.ch/copyright for details of the copyright holders.
3+
// All rights not expressly granted are reserved.
4+
//
5+
// This software is distributed under the terms of the GNU General Public
6+
// License v3 (GPL Version 3), copied verbatim in the file "COPYING".
7+
//
8+
// In applying this license CERN does not waive the privileges and immunities
9+
// granted to it by virtue of its status as an Intergovernmental Organization
10+
// or submit itself to any jurisdiction.
11+
12+
#ifndef LATETASKFACTORY_H
13+
#define LATETASKFACTORY_H
14+
15+
///
16+
/// \file LateTaskFactory.h
17+
/// \author Piotr Konopka
18+
///
19+
20+
#include <memory>
21+
22+
#include "QualityControl/LateTaskConfig.h"
23+
#include "QualityControl/LateTaskInterface.h"
24+
#include "QualityControl/RootClassFactory.h"
25+
26+
namespace o2::quality_control::core
27+
{
28+
29+
class LateTaskInterface;
30+
class ObjectsManager;
31+
32+
/// \brief Factory in charge of creating late tasks
33+
///
34+
/// The factory needs a library name and a class name provided as an object of type LateTaskConfig.
35+
/// The class loaded in the library must inherit from LateTaskInterface.
36+
class LateTaskFactory
37+
{
38+
public:
39+
LateTaskFactory() = default;
40+
virtual ~LateTaskFactory() = default;
41+
42+
/// \brief Create a new instance of a LateTaskInterface.
43+
/// The LateTaskInterface actual class is decided based on the parameters passed.
44+
static LateTaskInterface* create(const LateTaskConfig& taskConfig, std::shared_ptr<ObjectsManager> objectsManager)
45+
{
46+
auto* result = root_class_factory::create<LateTaskInterface>(taskConfig.moduleName, taskConfig.className);
47+
result->setName(taskConfig.name);
48+
result->setObjectsManager(objectsManager);
49+
result->setCustomParameters(taskConfig.customParameters);
50+
result->setCcdbUrl(taskConfig.ccdbUrl);
51+
52+
return result;
53+
}
54+
};
55+
56+
} // namespace o2::quality_control::core
57+
58+
#endif // LATETASKFACTORY_H

0 commit comments

Comments
 (0)