Skip to content

Commit e01d97d

Browse files
aalkinaalkin
committed
Add internal slicer devices
Add the ArrowTableSlicer algorithm plugin, which builds the requested slice-info tables from the slice sources The slice inputs are grouped based on the provider of the sources, so that multiple slicers can be injected right after the providers to avoid loops in topology
1 parent 2275e38 commit e01d97d

9 files changed

Lines changed: 214 additions & 2 deletions

File tree

‎Framework/AnalysisSupport/CMakeLists.txt‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -19,6 +19,7 @@ endif()
1919
o2_add_library(FrameworkOnDemandTablesSupport
2020
SOURCES src/OnDemandPlugin.cxx
2121
src/AODReaderHelpers.cxx
22+
src/AODSliceHelpers.cxx
2223
PRIVATE_INCLUDE_DIRECTORIES ${CMAKE_CURRENT_LIST_DIR}/src
2324
PUBLIC_LINK_LIBRARIES O2::Framework ${EXTRA_TARGETS})
2425

‎Framework/AnalysisSupport/src/AODSliceHelpers.cxx‎

Lines changed: 66 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,66 @@
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+
#include "AODSliceHelpers.h"
13+
14+
#include "Framework/ArrowTableSlicingCache.h"
15+
#include "Framework/ConfigParamRegistry.h"
16+
#include "Framework/DanglingEdgesContext.h"
17+
#include "Framework/DataAllocator.h"
18+
#include "Framework/DataSpecUtils.h"
19+
#include "Framework/InputRecord.h"
20+
#include "Framework/TableConsumer.h"
21+
22+
namespace o2::framework::helpers
23+
{
24+
namespace
25+
{
26+
Entry sourceEntry(InputSpec const& spec)
27+
{
28+
auto source = DataSpecUtils::fromMetadataString(std::ranges::find_if(spec.metadata, [](ConfigParamSpec const& cps) { return cps.name.starts_with("slice-source"); })->defaultValue.get<std::string>());
29+
return {source.binding, DataSpecUtils::asConcreteDataMatcher(source), std::ranges::find_if(spec.metadata, [](ConfigParamSpec const& cps) { return cps.name.starts_with("slice-key"); })->defaultValue.get<std::string>()};
30+
}
31+
32+
struct Sliceable {
33+
Entry entry;
34+
ConcreteDataMatcher output;
35+
bool sorted;
36+
37+
explicit Sliceable(InputSpec const& spec)
38+
: entry{sourceEntry(spec)},
39+
output{DataSpecUtils::asConcreteDataMatcher(spec)},
40+
sorted{std::ranges::find_if(spec.metadata, [](ConfigParamSpec const& cps) { return cps.name.starts_with("sorted"); })->defaultValue.get<bool>()}
41+
{
42+
}
43+
44+
std::shared_ptr<arrow::Table> materialize(ProcessingContext& pc) const
45+
{
46+
auto source = pc.inputs().get<TableConsumer>(entry.matcher)->asArrowTable();
47+
return sorted ? SliceInfo::makeSorted(entry, source) : SliceInfo::makeUnsorted(entry, source);
48+
}
49+
};
50+
} // namespace
51+
52+
AlgorithmSpec AODSliceHelpers::arrowTablesSlicerCallback(ConfigContext const& /*ctx*/)
53+
{
54+
return AlgorithmSpec::InitCallback{[](InitContext& ic) {
55+
// each slicer handles the group of slice infos for the tables from a single provider
56+
auto const& requested = ic.services().get<DanglingEdgesContext>().slicerGroups[ic.options().get<int>("slicer-group")];
57+
std::vector<Sliceable> sliceables;
58+
sliceables.reserve(requested.size());
59+
std::ranges::transform(requested, std::back_inserter(sliceables), [](auto const& i) { return Sliceable{i}; });
60+
return [sliceables](ProcessingContext& pc) {
61+
auto outputs = pc.outputs();
62+
std::ranges::for_each(sliceables, [&pc, &outputs](auto const& sliceable) { outputs.adopt(Output{sliceable.output.origin, sliceable.output.description, sliceable.output.subSpec}, sliceable.materialize(pc)); });
63+
};
64+
}};
65+
}
66+
} // namespace o2::framework::helpers

‎Framework/AnalysisSupport/src/AODSliceHelpers.h‎

Lines changed: 25 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,25 @@
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 AODSLICEHELPERS_H
13+
#define AODSLICEHELPERS_H
14+
15+
#include "Framework/AlgorithmSpec.h"
16+
namespace o2::framework::helpers
17+
{
18+
19+
struct AODSliceHelpers {
20+
static AlgorithmSpec arrowTablesSlicerCallback(ConfigContext const& /*ctx*/);
21+
};
22+
23+
} // namespace o2::framework::helpers
24+
25+
#endif // AODSLICEHELPERS_H

‎Framework/AnalysisSupport/src/OnDemandPlugin.cxx‎

Lines changed: 9 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -11,6 +11,7 @@
1111
#include "Framework/Plugins.h"
1212
#include "Framework/AlgorithmSpec.h"
1313
#include "AODReaderHelpers.h"
14+
#include "AODSliceHelpers.h"
1415

1516
struct ExtendedTableSpawner : o2::framework::AlgorithmPlugin {
1617
o2::framework::AlgorithmSpec create(o2::framework::ConfigContext const& config) override
@@ -26,7 +27,15 @@ struct IndexTableBuilder : o2::framework::AlgorithmPlugin {
2627
}
2728
};
2829

30+
struct ArrowTableSlicer : o2::framework::AlgorithmPlugin {
31+
o2::framework::AlgorithmSpec create(o2::framework::ConfigContext const& config) override
32+
{
33+
return o2::framework::helpers::AODSliceHelpers::arrowTablesSlicerCallback(config);
34+
}
35+
};
36+
2937
DEFINE_DPL_PLUGINS_BEGIN
3038
DEFINE_DPL_PLUGIN_INSTANCE(ExtendedTableSpawner, CustomAlgorithm);
3139
DEFINE_DPL_PLUGIN_INSTANCE(IndexTableBuilder, CustomAlgorithm);
40+
DEFINE_DPL_PLUGIN_INSTANCE(ArrowTableSlicer, CustomAlgorithm);
3241
DEFINE_DPL_PLUGINS_END

‎Framework/Core/include/Framework/AnalysisSupportHelpers.h‎

Lines changed: 9 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -42,6 +42,15 @@ struct AnalysisSupportHelpers {
4242
std::vector<InputSpec>& requestedAODs,
4343
std::vector<InputSpec>& requestedDYNs,
4444
DataProcessorSpec& publisher);
45+
static void addMissingOutputsToSlicer(std::vector<InputSpec> const& requestedSLCs,
46+
DataProcessorSpec& publisher);
47+
/// Split the requested slice infos into groups by the device providing the sliced table
48+
/// (the AOD reader, if none of the providers has it) and create a slicer device for each
49+
/// group, so that each slicer depends on a single device and does not create loops.
50+
/// Each slicer is returned together with the name of its provider.
51+
static std::vector<std::pair<std::string, DataProcessorSpec>> makeSlicers(std::vector<InputSpec> const& requestedSLCs,
52+
std::vector<DataProcessorSpec const*> const& providers,
53+
std::vector<std::vector<InputSpec>>& slicerGroups);
4554

4655
/// Match all inputs of kind ATSK and write them to a ROOT file,
4756
/// one root file per originating task.

‎Framework/Core/include/Framework/DanglingEdgesContext.h‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -45,6 +45,10 @@ struct DanglingEdgesContext {
4545
// ccdb tables
4646
std::vector<OutputSpec> providedTIMs;
4747
std::vector<InputSpec> requestedTIMs;
48+
// slice infos
49+
std::vector<InputSpec> requestedSLCs;
50+
// slice infos grouped by the device providing the sliced tables, one slicer device per group
51+
std::vector<std::vector<InputSpec>> slicerGroups;
4852
// output objects
4953
std::vector<OutputSpec> providedOutputObjHist;
5054
// inputs for the extension spawner

‎Framework/Core/src/AnalysisSupportHelpers.cxx‎

Lines changed: 57 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -188,6 +188,63 @@ void AnalysisSupportHelpers::addMissingOutputsToBuilder(std::vector<InputSpec> c
188188
sinks::update_input_list{requestedDYNs}; // update requestedDYNs
189189
}
190190

191+
void AnalysisSupportHelpers::addMissingOutputsToSlicer(std::vector<InputSpec> const& requestedSLCs,
192+
DataProcessorSpec& publisher)
193+
{
194+
requestedSLCs |
195+
views::input_to_output_specs() |
196+
sinks::append_to{publisher.outputs};
197+
198+
for (auto const& input : requestedSLCs) {
199+
input.metadata |
200+
views::filter_string_params_starts_with("slice-source:") |
201+
views::params_to_input_specs() |
202+
sinks::update_input_list{publisher.inputs};
203+
}
204+
}
205+
206+
std::vector<std::pair<std::string, DataProcessorSpec>> AnalysisSupportHelpers::makeSlicers(std::vector<InputSpec> const& requestedSLCs,
207+
std::vector<DataProcessorSpec const*> const& providers,
208+
std::vector<std::vector<InputSpec>>& slicerGroups)
209+
{
210+
// find the device providing the sliced table, if there is none the table is read from file
211+
auto providerFor = [&providers](InputSpec const& request) -> std::string {
212+
auto sources = request.metadata |
213+
views::filter_string_params_starts_with("slice-source:") |
214+
views::params_to_input_specs();
215+
auto matcher = DataSpecUtils::asConcreteDataMatcher(*sources.begin());
216+
auto provider = std::ranges::find_if(providers, [&matcher](DataProcessorSpec const* spec) {
217+
return std::ranges::any_of(spec->outputs, [&matcher](OutputSpec const& output) { return DataSpecUtils::match(output, matcher); });
218+
});
219+
return provider != providers.end() ? (*provider)->name : "internal-dpl-aod-reader";
220+
};
221+
222+
slicerGroups.clear();
223+
std::vector<std::string> groupProviders;
224+
for (auto const& request : requestedSLCs) {
225+
auto provider = providerFor(request);
226+
auto locate = std::ranges::find(groupProviders, provider);
227+
if (locate == groupProviders.end()) {
228+
groupProviders.push_back(provider);
229+
slicerGroups.push_back({request});
230+
} else {
231+
slicerGroups[std::distance(groupProviders.begin(), locate)].push_back(request);
232+
}
233+
}
234+
235+
std::vector<std::pair<std::string, DataProcessorSpec>> slicers;
236+
for (auto i = 0u; i < slicerGroups.size(); ++i) {
237+
DataProcessorSpec slicer{.name = "internal-dpl-aod-slicer-" + std::to_string(i),
238+
.inputs = {},
239+
.outputs = {},
240+
.algorithm = AlgorithmSpec::dummyAlgorithm(), // real algorithm will be set in adjustTopology
241+
.options = {ConfigParamSpec{"slicer-group", VariantType::Int, static_cast<int>(i), {"index of the slice info group handled by this slicer"}}}};
242+
addMissingOutputsToSlicer(slicerGroups[i], slicer);
243+
slicers.emplace_back(groupProviders[i], std::move(slicer));
244+
}
245+
return slicers;
246+
}
247+
191248
// =============================================================================
192249
DataProcessorSpec AnalysisSupportHelpers::getOutputObjHistSink(ConfigContext const& ctx)
193250
{

‎Framework/Core/src/ArrowSupport.cxx‎

Lines changed: 20 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -698,6 +698,26 @@ o2::framework::ServiceSpec ArrowSupport::arrowBackendSpec()
698698
}
699699
}
700700

701+
// slicers are recreated from scratch, grouped by the devices providing the sliced tables
702+
std::erase_if(workflow, [](DataProcessorSpec const& spec) { return spec.name.starts_with("internal-dpl-aod-slicer"); });
703+
dec.requestedSLCs.clear();
704+
for (auto& d : workflow) {
705+
d.inputs |
706+
views::filter_with_params_by_name_starting("slice-source:") |
707+
sinks::update_input_list{dec.requestedSLCs};
708+
}
709+
std::ranges::sort(dec.requestedSLCs, inputSpecLessThan);
710+
std::vector<DataProcessorSpec const*> slicedTablesProviders;
711+
std::ranges::transform(workflow, std::back_inserter(slicedTablesProviders), [](DataProcessorSpec const& spec) { return &spec; });
712+
auto slicers = AnalysisSupportHelpers::makeSlicers(dec.requestedSLCs, slicedTablesProviders, dec.slicerGroups);
713+
// the slicers are placed after their providers in a pre-sorted workflow
714+
for (auto& [providerName, slicer] : slicers) {
715+
// load real AlgorithmSpec before deployment
716+
slicer.algorithm = PluginManager::loadAlgorithmFromPlugin("O2FrameworkOnDemandTablesSupport", "ArrowTableSlicer", ctx);
717+
auto provider = std::ranges::find(workflow, providerName, &DataProcessorSpec::name);
718+
workflow.insert(provider == workflow.end() ? workflow.begin() : std::next(provider), std::move(slicer));
719+
}
720+
701721
auto writer = std::ranges::find_if(workflow, [](DataProcessorSpec const& spec) { return spec.name.starts_with("internal-dpl-aod-writer"); });
702722
if (writer != workflow.end()) {
703723
workflow.erase(writer);

‎Framework/Core/src/WorkflowHelpers.cxx‎

Lines changed: 23 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -282,9 +282,11 @@ void WorkflowHelpers::injectServiceDevices(WorkflowSpec& workflow, ConfigContext
282282
bool hasProjectors = false;
283283
bool hasIndexRecords = false;
284284
bool hasCCDBURLs = false;
285+
bool hasSliceSource = false;
285286
bool wasAOD = false;
286-
// all three options are exclusive
287+
// all options are exclusive
287288
for (auto const& p : input.metadata) {
289+
// wasAOD can be true or false for all of the options
288290
if (p.name.starts_with("aod-origin-replaced")) {
289291
wasAOD = true;
290292
}
@@ -300,6 +302,10 @@ void WorkflowHelpers::injectServiceDevices(WorkflowSpec& workflow, ConfigContext
300302
hasCCDBURLs = true;
301303
break;
302304
}
305+
if (p.name.starts_with("slice-source")) {
306+
hasSliceSource = true;
307+
break;
308+
}
303309
}
304310
switch (input.lifetime) {
305311
case Lifetime::Timer: {
@@ -346,6 +352,8 @@ void WorkflowHelpers::injectServiceDevices(WorkflowSpec& workflow, ConfigContext
346352
DataSpecUtils::updateInputList(dec.requestedIDXs, InputSpec{input});
347353
} else if (hasCCDBURLs) {
348354
DataSpecUtils::updateInputList(dec.requestedTIMs, InputSpec{input});
355+
} else if (hasSliceSource) {
356+
DataSpecUtils::updateInputList(dec.requestedSLCs, InputSpec{input});
349357
} else if (DataSpecUtils::partialMatch(input, AODOrigins) || wasAOD) {
350358
DataSpecUtils::updateInputList(dec.requestedAODs, InputSpec{input});
351359
}
@@ -358,8 +366,11 @@ void WorkflowHelpers::injectServiceDevices(WorkflowSpec& workflow, ConfigContext
358366
bool hasIndexRecords = false;
359367
bool hasCCDBURLs = false;
360368
bool wasAOD = false;
361-
// all three options are exclusive
369+
// all options are exclusive
370+
// provided slice outputs are ignored, they can only come from the slicer device
371+
// that will be re-added in adjust topology
362372
for (auto const& p : output.metadata) {
373+
// wasAOD can be true or false for all of the options
363374
if (p.name.starts_with("aod-origin-replaced")) {
364375
wasAOD = true;
365376
}
@@ -439,6 +450,13 @@ void WorkflowHelpers::injectServiceDevices(WorkflowSpec& workflow, ConfigContext
439450
std::ranges::sort(providedCCDBs, outputSpecLessThan);
440451
AnalysisSupportHelpers::addMissingOutputsToReader(providedCCDBs, requestedCCDBs, ccdbBackend);
441452

453+
// slicers are grouped by the devices providing the sliced tables
454+
std::ranges::sort(dec.requestedSLCs, inputSpecLessThan);
455+
std::vector<DataProcessorSpec const*> slicedTablesProviders;
456+
std::ranges::transform(workflow, std::back_inserter(slicedTablesProviders), [](DataProcessorSpec const& spec) { return &spec; });
457+
slicedTablesProviders.insert(slicedTablesProviders.end(), {&aodSpawner, &indexBuilder, &aodReader});
458+
auto aodSlicers = AnalysisSupportHelpers::makeSlicers(dec.requestedSLCs, slicedTablesProviders, dec.slicerGroups);
459+
442460
std::vector<DataProcessorSpec> extraSpecs;
443461

444462
if (transientStore.outputs.empty() == false) {
@@ -456,6 +474,9 @@ void WorkflowHelpers::injectServiceDevices(WorkflowSpec& workflow, ConfigContext
456474
extraSpecs.push_back(indexBuilder);
457475
}
458476

477+
// here the slicers are just added, unlike in adjustTopology
478+
std::ranges::transform(aodSlicers, std::back_inserter(extraSpecs), [](auto&& pair){ return pair.second; });
479+
459480
// add the Analysys CCDB backend which reads CCDB objects using a provided table
460481
DeploymentMode deploymentMode = DefaultsHelpers::deploymentMode();
461482
if (deploymentMode != DeploymentMode::OnlineDDS && deploymentMode != DeploymentMode::OnlineECS) {

0 commit comments

Comments
 (0)