Skip to content

Commit a70d658

Browse files
knopers8knopers8
authored
[QC-357] Move Data Sampling from Framework to Utilities (AliceO2Group#4135)
1 parent ee8cdfa commit a70d658

34 files changed

Lines changed: 316 additions & 246 deletions

‎Framework/Core/CMakeLists.txt‎

Lines changed: 0 additions & 25 deletions
Original file line numberDiff line numberDiff line change
@@ -54,22 +54,11 @@ o2_add_library(Framework
5454
src/DataProcessor.cxx
5555
src/DataRelayer.cxx
5656
src/DataRelayerHelpers.cxx
57-
src/DataSampling.cxx
58-
src/DataSamplingConditionFactory.cxx
59-
src/DataSamplingHeader.cxx
60-
src/DataSamplingConditionCustom.cxx
61-
src/DataSamplingConditionNConsecutive.cxx
62-
src/DataSamplingConditionPayloadSize.cxx
63-
src/DataSamplingConditionRandom.cxx
64-
src/DataSamplingHeader.cxx
65-
src/DataSamplingPolicy.cxx
66-
src/DataSamplingReadoutAdapter.cxx
6757
src/DataSpecUtils.cxx
6858
src/DeviceConfigInfo.cxx
6959
src/DeviceMetricsInfo.cxx
7060
src/DeviceSpec.cxx
7161
src/DeviceSpecHelpers.cxx
72-
src/Dispatcher.cxx
7362
src/DriverControl.cxx
7463
src/DriverInfo.cxx
7564
src/Expressions.cxx
@@ -161,9 +150,6 @@ foreach(t
161150
DataProcessorSpec
162151
DataRefUtils
163152
DataRelayer
164-
DataSamplingCondition
165-
DataSamplingHeader
166-
DataSamplingPolicy
167153
DeviceConfigInfo
168154
DeviceMetricsInfo
169155
DeviceSpec
@@ -222,17 +208,6 @@ o2_add_executable(verify-aod-file
222208
PUBLIC_LINK_LIBRARIES O2::Framework
223209
COMPONENT_NAME Framework)
224210

225-
# tests with input data
226-
227-
o2_data_file(COPY test/test_DataSampling.json DESTINATION tests)
228-
229-
o2_add_test(DataSampling NAME test_Framework_test_DataSampling
230-
SOURCES test/test_DataSampling.cxx
231-
COMPONENT_NAME Framework
232-
LABELS framework
233-
PUBLIC_LINK_LIBRARIES O2::Framework
234-
ENVIRONMENT O2_ROOT=${CMAKE_BINARY_DIR}/stage)
235-
236211
# tests with a name not starting with test_...
237212

238213
o2_add_test(unittest_DataSpecUtils NAME test_Framework_unittest_DataSpecUtils

‎Framework/Core/README.md‎

Lines changed: 0 additions & 94 deletions
Original file line numberDiff line numberDiff line change
@@ -467,100 +467,6 @@ There is also a few demonstrator available in particular:
467467

468468
- [MillWheel: Fault-Tolerant Stream Processing at Internet Scale](https://research.google.com/pubs/pub41378.html) : paper about Google previous generation system for stream processing
469469

470-
## Data Sampling
471-
472-
Data Sampling provides possibility to sample data in DPL workflows, basing on certain conditions ( 5% randomly, when payload is greater than 4234 bytes, etc.). The job of passing the right data is done by a data processor called `Dispatcher`. A desired data stream is specified in form of Data Sampling Policies, configured by JSON structures (example below).
473-
```
474-
{
475-
"id": "policy_example1", # name of the policy
476-
"active": "false", # activation flag
477-
"machines": [ # list of machines where the policy should be run (now ignored)
478-
"aido2flp1",
479-
"aido2flp2"
480-
], # list of data that should be sampled, the format is:
481-
# binding1:origin1/description1/subSpec1[;binding2:...]
482-
"query": "clusters:TPC/CLUSTERS/0;tracks:TPC/TRACKS/0",
483-
"samplingConditions": [ # list of sampling conditions
484-
{
485-
"condition": "random", # condition type
486-
"fraction": "0.1", # condition-dependent parameter: fraction of data to sample
487-
"seed": "2112" # condition-dependent parameter: seed of PRNG
488-
}
489-
],
490-
"blocking": "false" # should the dispatcher block the main data flow? (now ignored)
491-
}
492-
```
493-
494-
### Usage
495-
496-
To use Data Sampling in a DPL workflow insert following lines to your code:
497-
```cpp
498-
#include "Framework/DataSampling.h"
499-
using namespace o2::framework;
500-
void customize(std::vector<CompletionPolicy>& policies)
501-
{
502-
DataSampling::CustomizeInfrastructure(policies);
503-
}
504-
505-
void customize(std::vector<ChannelConfigurationPolicy>& policies)
506-
{
507-
DataSampling::CustomizeInfrastructure(policies);
508-
}
509-
510-
#include "Framework/runDataProcessing.h"
511-
512-
std::vector<DataProcessorSpec> defineDataProcessing(ConfigContext &ctx)
513-
{
514-
515-
WorkflowSpec workflow;
516-
// <declaration of other DPL processors>
517-
518-
DataSampling::GenerateInfrastructure(workflow, "json:///absolute/path/to/config/file.json");
519-
520-
return workflow;
521-
}
522-
```
523-
524-
Sampled data can be subscribed to by adding `InputSpecs` provided by `std::vector<InputSpec> DataSampling::InputSpecsForPolicy(const std::string& policiesSource, const std::string& policyName)` to a chosen data processor. Then, they can be accessed by the bindings specified in the configuration file. Dispatcher adds a `DataSamplingHeader` to the header stack, which contains statistics like total number of evaluated/accepted messages for a given Policy or the sampling time since epoch.
525-
526-
[o2-datasampling-pod-and-root](https://github.com/AliceO2Group/AliceO2/blob/dev/Framework/TestWorkflows/src/dataSamplingPodAndRoot.cxx) workflow can serve as usage example.
527-
528-
## Data Sampling Conditions
529-
530-
The following sampling conditions are available. When more than one is used, a positive decision is taken when all the conditions are fulfilled.
531-
- **DataSamplingConditionRandom** - pseudo-randomly accepts specified fraction of incoming messages.
532-
```json
533-
{
534-
"condition": "random",
535-
"fraction": "0.1",
536-
"seed": "22222"
537-
}
538-
```
539-
- **DataSamplingConditionNConsecutive** - approves n consecutive samples in defined cycle. It assumes that timesliceID always increments by one.
540-
```json
541-
{
542-
"condition": "nConsecutive",
543-
"samplesNumber": "3",
544-
"cycleSize": "100"
545-
}
546-
```
547-
- **DataSamplingConditionPayloadSize** - approves messages having payload size within specified boundaries.
548-
```json
549-
{
550-
"condition": "payloadSize",
551-
"lowerLimit": "300",
552-
"upperLimit": "500"
553-
}
554-
```
555-
- **DataSamplingConditionCustom** - loads a custom condition, which should inherit from DataSamplingCondition, from a specified library.
556-
```json
557-
{
558-
"condition": "custom",
559-
"moduleName": "QcExample",
560-
"className": "o2::quality_control_modules::example::ExampleCondition",
561-
"customParam": "value"
562-
}
563-
```
564470
## Document history
565471

566472
* v0.9: proposal for approval at the O2 TB - 19th June 2018

‎Framework/TestWorkflows/CMakeLists.txt‎

Lines changed: 0 additions & 18 deletions
Original file line numberDiff line numberDiff line change
@@ -78,28 +78,10 @@ o2_add_dpl_workflow(test_CompletionPolicies
7878
SOURCES src/test_CompletionPolicies.cxx
7979
COMPONENT_NAME TestWorkflows)
8080

81-
o2_data_file(COPY etc/exampleDataSamplingConfig.json DESTINATION etc)
82-
83-
o2_add_dpl_workflow(datasampling-pod-and-root
84-
SOURCES src/dataSamplingPodAndRoot.cxx
85-
COMPONENT_NAME TestWorkflows)
86-
87-
o2_add_dpl_workflow(datasampling-parallel
88-
SOURCES src/dataSamplingParallel.cxx
89-
COMPONENT_NAME TestWorkflows)
90-
91-
o2_add_dpl_workflow(datasampling-time-pipeline
92-
SOURCES src/dataSamplingTimePipeline.cxx
93-
COMPONENT_NAME TestWorkflows)
94-
9581
o2_add_dpl_workflow(ccdb-fetch-to-timeframe
9682
SOURCES src/test_CCDBFetchToTimeframe.cxx
9783
COMPONENT_NAME TestWorkflows)
9884

99-
o2_add_dpl_workflow(datasampling-benchmark
100-
SOURCES src/dataSamplingBenchmark.cxx
101-
COMPONENT_NAME TestWorkflows)
102-
10385
o2_add_dpl_workflow(simple-source
10486
SOURCES src/o2SimpleSource.cxx
10587
COMPONENT_NAME TestWorkflows)

‎Utilities/CMakeLists.txt‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -16,6 +16,7 @@ endif(ALIROOT)
1616
add_subdirectory(aliceHLTwrapper)
1717
add_subdirectory(O2MessageMonitor)
1818
add_subdirectory(DataFlow)
19+
add_subdirectory(DataSampling)
1920
add_subdirectory(Publishers)
2021
add_subdirectory(DataCompression)
2122
add_subdirectory(Mergers)

‎Utilities/DataSampling/CMakeLists.txt‎

Lines changed: 82 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,82 @@
1+
# Copyright CERN and copyright holders of ALICE O2. This software is distributed
2+
# under the terms of the GNU General Public License v3 (GPL Version 3), copied
3+
# verbatim in the file "COPYING".
4+
#
5+
# See http://alice-o2.web.cern.ch/license for full licensing information.
6+
#
7+
# In applying this license CERN does not waive the privileges and immunities
8+
# granted to it by virtue of its status as an Intergovernmental Organization or
9+
# submit itself to any jurisdiction.
10+
11+
# FIXME: the LinkDef should not be in the public area
12+
13+
o2_add_library(DataSampling
14+
SOURCES src/DataSampling.cxx
15+
src/DataSamplingConditionFactory.cxx
16+
src/DataSamplingHeader.cxx
17+
src/DataSamplingConditionCustom.cxx
18+
src/DataSamplingConditionNConsecutive.cxx
19+
src/DataSamplingConditionPayloadSize.cxx
20+
src/DataSamplingConditionRandom.cxx
21+
src/DataSamplingHeader.cxx
22+
src/DataSamplingPolicy.cxx
23+
src/DataSamplingReadoutAdapter.cxx
24+
src/Dispatcher.cxx
25+
26+
PUBLIC_LINK_LIBRARIES O2::Framework O2::DataSampling)
27+
28+
#o2_target_root_dictionary(
29+
# Mergers
30+
# HEADERS include/Mergers/MergeInterface.h
31+
# include/Mergers/CustomMergeableObject.h
32+
# include/Mergers/CustomMergeableTObject.h
33+
# LINKDEF include/Mergers/LinkDef.h)
34+
35+
# tests with input data
36+
37+
o2_data_file(COPY test/test_DataSampling.json DESTINATION tests)
38+
39+
o2_add_test(DataSampling NAME test_DataSampling_test_DataSampling
40+
SOURCES test/test_DataSampling.cxx
41+
COMPONENT_NAME DataSampling
42+
LABELS datasampling
43+
PUBLIC_LINK_LIBRARIES O2::DataSampling
44+
ENVIRONMENT O2_ROOT=${CMAKE_BINARY_DIR}/stage)
45+
46+
foreach(t
47+
DataSamplingCondition
48+
DataSamplingHeader
49+
DataSamplingPolicy
50+
)
51+
52+
# FIXME ? The NAME parameter of o2_add_test is only needed to help the current
53+
# o2.sh recipe. If the recipe is changed, those params can go away, if needed.
54+
55+
o2_add_test(${t} NAME test_DataSampling_test_${t}
56+
SOURCES test/test_${t}.cxx
57+
COMPONENT_NAME DataSampling
58+
LABELS datasampling
59+
PUBLIC_LINK_LIBRARIES O2::DataSampling)
60+
endforeach()
61+
62+
o2_data_file(COPY etc/exampleDataSamplingConfig.json DESTINATION etc)
63+
64+
o2_add_dpl_workflow(datasampling-pod-and-root
65+
SOURCES test/dataSamplingPodAndRoot.cxx
66+
COMPONENT_NAME DataSampling
67+
PUBLIC_LINK_LIBRARIES O2::DataSampling)
68+
69+
o2_add_dpl_workflow(datasampling-parallel
70+
SOURCES test/dataSamplingParallel.cxx
71+
COMPONENT_NAME DataSampling
72+
PUBLIC_LINK_LIBRARIES O2::DataSampling)
73+
74+
o2_add_dpl_workflow(datasampling-time-pipeline
75+
SOURCES test/dataSamplingTimePipeline.cxx
76+
COMPONENT_NAME DataSampling
77+
PUBLIC_LINK_LIBRARIES O2::DataSampling)
78+
79+
o2_add_dpl_workflow(datasampling-benchmark
80+
SOURCES test/dataSamplingBenchmark.cxx
81+
COMPONENT_NAME DataSampling
82+
PUBLIC_LINK_LIBRARIES O2::DataSampling)

‎Utilities/DataSampling/README.md‎

Lines changed: 95 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,95 @@
1+
## Data Sampling
2+
3+
Data Sampling provides possibility to sample data in DPL workflows, basing on certain conditions ( 5% randomly, when payload is greater than 4234 bytes, etc.). The job of passing the right data is done by a data processor called `Dispatcher`. A desired data stream is specified in form of Data Sampling Policies, configured by JSON structures (example below).
4+
```
5+
{
6+
"id": "policy_example1", # name of the policy
7+
"active": "false", # activation flag
8+
"machines": [ # list of machines where the policy should be run (now ignored)
9+
"aido2flp1",
10+
"aido2flp2"
11+
], # list of data that should be sampled, the format is:
12+
# binding1:origin1/description1/subSpec1[;binding2:...]
13+
"query": "clusters:TPC/CLUSTERS/0;tracks:TPC/TRACKS/0",
14+
"samplingConditions": [ # list of sampling conditions
15+
{
16+
"condition": "random", # condition type
17+
"fraction": "0.1", # condition-dependent parameter: fraction of data to sample
18+
"seed": "2112" # condition-dependent parameter: seed of PRNG
19+
}
20+
],
21+
"blocking": "false" # should the dispatcher block the main data flow? (now ignored)
22+
}
23+
```
24+
25+
### Usage
26+
27+
To use Data Sampling in a DPL workflow insert following lines to your code:
28+
```cpp
29+
#include "DataSampling/DataSampling.h"
30+
using namespace o2::framework;
31+
using namespace o2::utilities;
32+
void customize(std::vector<CompletionPolicy>& policies)
33+
{
34+
DataSampling::CustomizeInfrastructure(policies);
35+
}
36+
37+
void customize(std::vector<ChannelConfigurationPolicy>& policies)
38+
{
39+
DataSampling::CustomizeInfrastructure(policies);
40+
}
41+
42+
#include "Framework/runDataProcessing.h"
43+
44+
std::vector<DataProcessorSpec> defineDataProcessing(ConfigContext &ctx)
45+
{
46+
47+
WorkflowSpec workflow;
48+
// <declaration of other DPL processors>
49+
50+
DataSampling::GenerateInfrastructure(workflow, "json:///absolute/path/to/config/file.json");
51+
52+
return workflow;
53+
}
54+
```
55+
56+
Sampled data can be subscribed to by adding `InputSpecs` provided by `std::vector<InputSpec> DataSampling::InputSpecsForPolicy(const std::string& policiesSource, const std::string& policyName)` to a chosen data processor. Then, they can be accessed by the bindings specified in the configuration file. Dispatcher adds a `DataSamplingHeader` to the header stack, which contains statistics like total number of evaluated/accepted messages for a given Policy or the sampling time since epoch.
57+
58+
The [o2-datasampling-pod-and-root](https://github.com/AliceO2Group/AliceO2/blob/dev/Utilities/DataSampling/test/dataSamplingPodAndRoot.cxx) workflow can serve as a usage example.
59+
60+
## Data Sampling Conditions
61+
62+
The following sampling conditions are available. When more than one is used, a positive decision is taken when all the conditions are fulfilled.
63+
- **DataSamplingConditionRandom** - pseudo-randomly accepts specified fraction of incoming messages.
64+
```json
65+
{
66+
"condition": "random",
67+
"fraction": "0.1",
68+
"seed": "22222"
69+
}
70+
```
71+
- **DataSamplingConditionNConsecutive** - approves n consecutive samples in defined cycle. It assumes that timesliceID always increments by one.
72+
```json
73+
{
74+
"condition": "nConsecutive",
75+
"samplesNumber": "3",
76+
"cycleSize": "100"
77+
}
78+
```
79+
- **DataSamplingConditionPayloadSize** - approves messages having payload size within specified boundaries.
80+
```json
81+
{
82+
"condition": "payloadSize",
83+
"lowerLimit": "300",
84+
"upperLimit": "500"
85+
}
86+
```
87+
- **DataSamplingConditionCustom** - loads a custom condition, which should inherit from DataSamplingCondition, from a specified library.
88+
```json
89+
{
90+
"condition": "custom",
91+
"moduleName": "QcExample",
92+
"className": "o2::quality_control_modules::example::ExampleCondition",
93+
"customParam": "value"
94+
}
95+
```

Framework/TestWorkflows/etc/exampleDataSamplingConfig.json renamed to Utilities/DataSampling/etc/exampleDataSamplingConfig.json

File renamed without changes.

0 commit comments

Comments
 (0)