Project
Loading...
Searching...
No Matches
AnalysisManagers.h
Go to the documentation of this file.
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 FRAMEWORK_ANALYSISMANAGERS_H
13#define FRAMEWORK_ANALYSISMANAGERS_H
17#include "Framework/ASoA.h"
25#include "Framework/Condition.h"
30
31namespace o2::framework
32{
33
34namespace
35{
36template <typename O>
37static inline auto extractOriginal(ProcessingContext& pc)
38{
39 return pc.inputs().get<TableConsumer>(aod::MetadataTrait<O>::metadata::tableLabel())->asArrowTable();
40}
41
42template <typename... Os>
43static inline std::vector<std::shared_ptr<arrow::Table>> extractOriginals(framework::pack<Os...>, ProcessingContext& pc)
44{
45 return {extractOriginal<Os>(pc)...};
46}
47
48template <size_t N, std::array<soa::TableRef, N> refs>
49static inline auto extractOriginals(ProcessingContext& pc)
50{
51 return [&]<size_t... Is>(std::index_sequence<Is...>) -> std::vector<std::shared_ptr<arrow::Table>> {
52 return {pc.inputs().get<TableConsumer>(o2::aod::label<refs[Is]>())->asArrowTable()...};
53 }(std::make_index_sequence<refs.size()>());
54}
55} // namespace
56
57namespace analysis_task_parsers
58{
59
61template <typename O>
62bool appendOption(std::vector<ConfigParamSpec>&, O&)
63{
64 return false;
65}
66
67template <is_configurable O>
68bool appendOption(std::vector<ConfigParamSpec>& options, O& option)
69{
70 return ConfigurableHelpers::appendOption(options, option);
71}
72
73template <is_configurable_group O>
74bool appendOption(std::vector<ConfigParamSpec>& options, O& optionGroup)
75{
76 if constexpr (requires { optionGroup.prefix; }) {
77 homogeneous_apply_refs<true>([prefix = optionGroup.prefix]<typename C>(C& option) { // apend group prefix if set
78 if constexpr (requires { option.name; }) {
79 option.name.insert(0, 1, '.');
80 option.name.insert(0, prefix);
81 }
82 return true;
83 },
84 optionGroup);
85 }
86 homogeneous_apply_refs<true>([&options](auto& option) { return appendOption(options, option); }, optionGroup);
87 return true;
88}
89
90template <typename O>
92{
93 return false;
94}
95
96template <is_configurable O>
97bool prepareOption(InitContext& context, O& configurable)
98{
99 if constexpr (variant_trait_v<typename O::type> != VariantType::Unknown) {
100 configurable.value = context.options().get<typename O::type>(configurable.name.c_str());
101 } else {
102 auto pt = context.options().get<boost::property_tree::ptree>(configurable.name.c_str());
103 configurable.value = RootConfigParamHelpers::as<typename O::type>(pt);
104 }
105 return true;
106}
107
108template <is_configurable_group O>
109bool prepareOption(InitContext& context, O& configurableGroup)
110{
111 homogeneous_apply_refs<true>([&context](auto&& configurable) { return prepareOption(context, configurable); }, configurableGroup);
112 return true;
113}
114
116template <typename C>
117bool appendCondition(std::vector<InputSpec>&, C&)
118{
119 return false;
120}
121
122template <is_condition C>
123bool appendCondition(std::vector<InputSpec>& inputs, C& condition)
124{
125 inputs.emplace_back(InputSpec{condition.path, "AODC", runtime_hash(condition.path.c_str()), Lifetime::Condition, ccdbParamSpec(condition.path)});
126 return true;
127}
128
129template <is_condition_group C>
130bool appendCondition(std::vector<InputSpec>& inputs, C& conditionGroup)
131{
132 homogeneous_apply_refs<true>([&inputs](auto& condition) { return appendCondition(inputs, condition); }, conditionGroup);
133 return true;
134}
135
137template <typename T>
138bool requestInputs(std::vector<InputSpec>&, T const&)
139{
140 return false;
141}
142
143template <is_spawns T>
144const char* controlOption()
145{
146 return "control:spawn";
147}
148
149template <is_builds T>
150const char* controlOption()
151{
152 return "control:build";
153}
154
155template <is_defines T>
156const char* controlOption()
157{
158 return "control:define";
159}
160
161template <typename T>
162 requires(is_spawns<T> || is_builds<T> || is_defines<T>)
163bool requestInputs(std::vector<InputSpec>& inputs, T const& entity)
164{
165 auto base_specs = entity.base_specs();
166 for (auto base_spec : base_specs) {
167 base_spec.metadata.push_back(ConfigParamSpec{std::string{controlOption<T>()}, VariantType::Bool, true, {"\"\""}});
168 DataSpecUtils::updateInputList(inputs, std::forward<InputSpec>(base_spec));
169 }
170 return true;
171}
172
173template <typename C>
175{
176 return false;
177}
178
179template <is_condition C>
180bool newDataframeCondition(InputRecord& record, C& condition)
181{
182 condition.instance = (typename C::type*)record.get<typename C::type*>(condition.path).get();
183 return true;
184}
185
186template <is_condition_group C>
187bool newDataframeCondition(InputRecord& record, C& conditionGroup)
188{
189 homogeneous_apply_refs<true>([&record](auto&& condition) { return newDataframeCondition(record, condition); }, conditionGroup);
190 return true;
191}
192
194template <typename T>
195bool appendOutput(std::vector<OutputSpec>&, T&, uint32_t)
196{
197 return false;
198}
199
200template <is_produces T>
201bool appendOutput(std::vector<OutputSpec>& outputs, T&, uint32_t)
202{
204 return true;
205}
206
207template <is_produces_group T>
208bool appendOutput(std::vector<OutputSpec>& outputs, T& producesGroup, uint32_t hash)
209{
210 homogeneous_apply_refs<true>([&outputs, hash](auto& produces) { return appendOutput(outputs, produces, hash); }, producesGroup);
211 return true;
212}
213
214template <is_histogram_registry T>
215bool appendOutput(std::vector<OutputSpec>& outputs, T& hr, uint32_t hash)
216{
217 hr.setHash(hash);
218 outputs.emplace_back(hr.spec());
219 return true;
220}
221
222template <is_outputobj T>
223bool appendOutput(std::vector<OutputSpec>& outputs, T& obj, uint32_t hash)
224{
225 obj.setHash(hash);
226 outputs.emplace_back(obj.spec());
227 return true;
228}
229
230template <typename T>
231 requires(is_spawns<T> || is_builds<T> || is_defines<T>)
232bool appendOutput(std::vector<OutputSpec>& outputs, T& entity, uint32_t)
233{
234 outputs.emplace_back(entity.spec());
235 return true;
236}
237
238template <typename T>
240{
241 return false;
242}
243
244template <is_histogram_registry T>
245bool postRunOutput(EndOfStreamContext& context, T& hr)
246{
247 auto& deviceSpec = context.services().get<o2::framework::DeviceSpec const>();
248 context.outputs().snapshot(hr.ref(deviceSpec.inputTimesliceId, deviceSpec.maxInputTimeslices), *(hr.getListOfHistograms()));
249 hr.clean();
250 return true;
251}
252
253template <is_outputobj T>
254bool postRunOutput(EndOfStreamContext& context, T& obj)
255{
256 auto& deviceSpec = context.services().get<o2::framework::DeviceSpec const>();
257 context.outputs().snapshot(obj.ref(deviceSpec.inputTimesliceId, deviceSpec.maxInputTimeslices), *obj);
258 return true;
259}
260
261template <typename T>
263{
264 return false;
265}
266
267template <is_produces T>
268bool prepareOutput(ProcessingContext& context, T& produces)
269{
270 produces.resetCursor(std::move(context.outputs().make<TableBuilder>(OutputForTable<typename T::persistent_table_t>::ref())));
271 return true;
272}
273
274template <is_produces_group T>
275bool prepareOutput(ProcessingContext& context, T& producesGroup)
276{
277 homogeneous_apply_refs<true>([&context](auto& produces) { return prepareOutput(context, produces); }, producesGroup);
278 return true;
279}
280
281template <is_spawns T>
282bool prepareOutput(ProcessingContext& context, T& spawns)
283{
284 using metadata = o2::aod::MetadataTrait<o2::aod::Hash<T::spawnable_t::ref.desc_hash>>::metadata;
285 auto originalTable = soa::ArrowHelpers::joinTables(extractOriginals<metadata::sources.size(), metadata::sources>(context), std::span{metadata::base_table_t::originalLabels});
286 if (originalTable->schema()->fields().empty() == true) {
287 using base_table_t = typename T::base_table_t::table_t;
288 originalTable = makeEmptyTable<base_table_t>(o2::aod::label<metadata::extension_table_t::ref>());
289 }
290 using D = o2::aod::Hash<metadata::extension_table_t::ref.desc_hash>;
291
292 spawns.extension = std::make_shared<typename T::extension_t>(o2::framework::spawner<D>(originalTable,
293 o2::aod::label<metadata::extension_table_t::ref>(),
294 spawns.projectors.data(),
295 spawns.projector,
296 spawns.schema));
297 spawns.table = std::make_shared<typename T::spawnable_t::table_t>(soa::ArrowHelpers::joinTables({spawns.extension->asArrowTable(), originalTable}, std::span{T::spawnable_t::table_t::originalLabels}));
298 return true;
299}
300
301template <is_builds T>
302bool prepareOutput(ProcessingContext& context, T& builds)
303{
304 using metadata = o2::aod::MetadataTrait<o2::aod::Hash<T::buildable_t::ref.desc_hash>>::metadata;
305 return builds.template build<typename T::buildable_t::indexing_t>(builds.pack(), extractOriginals<metadata::sources.size(), metadata::sources>(context));
306}
307
308template <is_defines T>
309bool prepareOutput(ProcessingContext& context, T& defines)
310 requires(T::delayed == false)
311{
312 using metadata = o2::aod::MetadataTrait<o2::aod::Hash<T::spawnable_t::ref.desc_hash>>::metadata;
313 auto originalTable = soa::ArrowHelpers::joinTables(extractOriginals<metadata::sources.size(), metadata::sources>(context), std::span{metadata::base_table_t::originalLabels});
314 if (originalTable->schema()->fields().empty() == true) {
315 using base_table_t = typename T::base_table_t::table_t;
316 originalTable = makeEmptyTable<base_table_t>(o2::aod::label<metadata::extension_table_t::ref>());
317 }
318 if (defines.inputSchema == nullptr) {
319 defines.inputSchema = originalTable->schema();
320 }
321 using D = o2::aod::Hash<metadata::extension_table_t::ref.desc_hash>;
322
323 defines.extension = std::make_shared<typename T::extension_t>(o2::framework::spawner<D>(originalTable,
324 o2::aod::label<metadata::extension_table_t::ref>(),
325 defines.projectors.data(),
326 defines.projector,
327 defines.schema));
328 defines.table = std::make_shared<typename T::spawnable_t::table_t>(soa::ArrowHelpers::joinTables({defines.extension->asArrowTable(), originalTable}, std::span{T::spawnable_t::table_t::originalLabels}));
329 return true;
330}
331
332template <typename T>
334{
335 return false;
336}
337
338template <is_defines T>
339 requires(T::delayed == true)
340bool prepareDelayedOutput(ProcessingContext& context, T& defines)
341{
342 if (defines.needRecompilation) {
343 defines.recompile();
344 }
345 using metadata = o2::aod::MetadataTrait<o2::aod::Hash<T::spawnable_t::ref.desc_hash>>::metadata;
346 auto originalTable = soa::ArrowHelpers::joinTables(extractOriginals<metadata::sources.size(), metadata::sources>(context), std::span{metadata::base_table_t::originalLabels});
347 if (originalTable->schema()->fields().empty() == true) {
348 using base_table_t = typename T::base_table_t::table_t;
349 originalTable = makeEmptyTable<base_table_t>(o2::aod::label<metadata::extension_table_t::ref>());
350 }
351 if (defines.inputSchema == nullptr) {
352 defines.inputSchema = originalTable->schema();
353 }
354 using D = o2::aod::Hash<metadata::extension_table_t::ref.desc_hash>;
355
356 defines.extension = std::make_shared<typename T::extension_t>(o2::framework::spawner<D>(originalTable,
357 o2::aod::label<metadata::extension_table_t::ref>(),
358 defines.projectors.data(),
359 defines.projector,
360 defines.schema));
361 defines.table = std::make_shared<typename T::spawnable_t::table_t>(soa::ArrowHelpers::joinTables({defines.extension->asArrowTable(), originalTable}, std::span{T::spawnable_t::table_t::originalLabels}));
362 return true;
363}
364
365template <typename T>
367{
368 return false;
369}
370
371template <is_produces T>
372bool finalizeOutput(ProcessingContext&, T& produces)
373{
374 produces.setLabel(o2::aod::label<T::persistent_table_t::ref>());
375 produces.release();
376 return true;
377}
378
379template <is_produces_group T>
380bool finalizeOutput(ProcessingContext& context, T& producesGroup)
381{
382 homogeneous_apply_refs<true>([&context](auto& produces) { return finalizeOutput(context, produces); }, producesGroup);
383 return true;
384}
385
386template <is_spawns T>
387bool finalizeOutput(ProcessingContext& context, T& spawns)
388{
389 context.outputs().adopt(spawns.output(), spawns.asArrowTable());
390 return true;
391}
392
393template <is_builds T>
394bool finalizeOutput(ProcessingContext& context, T& builds)
395{
396 context.outputs().adopt(builds.output(), builds.asArrowTable());
397 return true;
398}
399
400template <is_defines T>
401bool finalizeOutput(ProcessingContext& context, T& defines)
402{
403 context.outputs().adopt(defines.output(), defines.asArrowTable());
404 return true;
405}
406
408template <typename T>
409bool addService(std::vector<ServiceSpec>&, T&)
410{
411 return false;
412}
413
414template <is_service T>
415bool addService(std::vector<ServiceSpec>& specs, T&)
416{
418 auto p = typename T::service_t{};
419 auto loadableServices = PluginManager::parsePluginSpecString(p.loadSpec.c_str());
420 PluginManager::loadFromPlugin<ServiceSpec, ServicePlugin>(loadableServices, specs);
421 }
422 return true;
423}
424
425template <typename T>
427{
428 return false;
429}
430
431template <is_service T>
432bool prepareService(InitContext& context, T& service)
433{
434 using S = typename T::service_t;
435 if constexpr (requires { &S::instance; }) {
436 service.service = &(S::instance()); // Sigh...
437 return true;
438 } else {
439 service.service = &(context.services().get<S>());
440 return true;
441 }
442 return false;
443}
444
445template <typename T>
447{
448 return false;
449}
450
451template <is_service T>
452bool postRunService(EndOfStreamContext&, T& service)
453{
454 // FIXME: for the moment we only need endOfStream to be
455 // stateless. In the future we might want to pass it EndOfStreamContext
456 if constexpr (requires { &T::service_t::endOfStream; }) {
457 service.service->endOfStream();
458 return true;
459 }
460 return false;
461}
462
464template <typename T>
466{
467 return false;
468}
469
470template <expressions::is_filter T>
471bool updatePlaceholders(InitContext& context, T& filter)
472{
474 return true;
475}
476
477template <is_partition T>
478bool updatePlaceholders(InitContext& context, T& partition)
479{
480 partition.updatePlaceholders(context);
481 return true;
482}
483
484template <typename T>
485bool createExpressionTrees(std::vector<ExpressionInfo>&, T&)
486{
487 return false;
488}
489
490template <expressions::is_filter T>
491bool createExpressionTrees(std::vector<ExpressionInfo>& expressionInfos, T& filter)
492{
494 return true;
495}
496
497template <typename T>
499{
500 return false;
501}
502
503template <is_partition T>
504bool newDataframePartition(T& partition)
505{
506 partition.dataframeChanged = true;
507 return true;
508}
509
510template <typename P, typename... T>
511void setPartition(P&, T&...)
512{
513}
514
515template <is_partition P, typename... T>
516void setPartition(P& partition, T&... tables)
517{
518 ([&]() { if constexpr (std::same_as<typename P::content_t, T>) {partition.bindTable(tables);} }(), ...);
519}
520
521template <typename P, typename T>
523{
524}
525
526template <is_partition P, typename T>
527void bindInternalIndicesPartition(P& partition, T* table)
528{
529 if constexpr (o2::soa::is_binding_compatible_v<typename P::content_t, std::decay_t<T>>()) {
530 partition.bindInternalIndicesTo(table);
531 }
532}
533
534template <typename P, typename... T>
536{
537}
538
539template <is_partition P, typename... T>
540void bindExternalIndicesPartition(P& partition, T*... tables)
541{
542 partition.bindExternalIndices(tables...);
543}
544
546template <typename T>
548{
549 return false;
550}
551
552template <typename T>
554{
555 return false;
556}
557
558template <is_slice_cache T>
559bool initializeCache(ProcessingContext& context, T& cache)
560{
561 if (cache.ptr == nullptr) {
562 cache.ptr = &context.services().get<ArrowTableSlicingCache>();
563 }
564 return true;
565}
566
568template <typename C, typename TG, typename... Ts>
569void setGroupedCombination(C&, TG&, Ts&...)
570{
571}
572
573template <is_combinations_generator C, typename TG, typename... Ts>
574static void setGroupedCombination(C& comb, TG& grouping, std::tuple<Ts...>& associated)
575{
576 if constexpr (std::same_as<typename C::g_t, std::decay_t<TG>>) {
577 comb.setTables(grouping, associated);
578 }
579}
580
582template <typename T>
583 requires(!is_preslice<T>)
585{
586 return false;
587}
588
589template <is_preslice T>
590 requires std::same_as<typename T::policy_t, framework::PreslicePolicySorted>
591bool registerCache(T& preslice, Cache& bsks, Cache&)
592{
593 if constexpr (T::optional) {
594 if (preslice.binding == "[MISSING]") {
595 return true;
596 }
597 }
598 auto locate = std::find_if(bsks.begin(), bsks.end(), [&](auto const& entry) { return (entry.binding == preslice.bindingKey.binding) && (entry.key == preslice.bindingKey.key); });
599 if (locate == bsks.end()) {
600 bsks.emplace_back(preslice.getBindingKey());
601 } else if (locate->enabled == false) {
602 locate->enabled = true;
603 }
604 return true;
605}
606
607template <is_preslice T>
608 requires std::same_as<typename T::policy_t, framework::PreslicePolicyGeneral>
609bool registerCache(T& preslice, Cache&, Cache& bsksU)
610{
611 if constexpr (T::optional) {
612 if (preslice.binding == "[MISSING]") {
613 return true;
614 }
615 }
616 auto locate = std::find_if(bsksU.begin(), bsksU.end(), [&](auto const& entry) { return (entry.binding == preslice.bindingKey.binding) && (entry.key == preslice.bindingKey.key); });
617 if (locate == bsksU.end()) {
618 bsksU.emplace_back(preslice.getBindingKey());
619 } else if (locate->enabled == false) {
620 locate->enabled = true;
621 }
622 return true;
623}
624
625template <typename T>
626 requires(!is_preslice<T>)
628{
629 return false;
630}
631
632template <is_preslice T>
633static bool updateSliceInfo(T& preslice, ArrowTableSlicingCache& cache)
634 requires std::same_as<typename T::policy_t, framework::PreslicePolicySorted>
635{
636 if constexpr (T::optional) {
637 if (preslice.binding == "[MISSING]") {
638 return true;
639 }
640 }
641 preslice.updateSliceInfo(cache.getCacheFor(preslice.getBindingKey()));
642 return true;
643}
644
645template <is_preslice T>
646static bool updateSliceInfo(T& preslice, ArrowTableSlicingCache& cache)
647 requires std::same_as<typename T::policy_t, framework::PreslicePolicyGeneral>
648{
649 if constexpr (T::optional) {
650 if (preslice.binding == "[MISSING]") {
651 return true;
652 }
653 }
654 preslice.updateSliceInfo(cache.getCacheUnsortedFor(preslice.getBindingKey()));
655 return true;
656}
657
659template <typename T>
660static bool setProcessSwitch(std::pair<std::string, bool>, T&)
661{
662 return false;
663}
664
665template <is_process_configurable T>
666static bool setProcessSwitch(std::pair<std::string, bool> setting, T& pc)
667{
668 if (pc.name == setting.first) {
669 pc.value = setting.second;
670 return true;
671 }
672 return false;
673}
674
675} // namespace analysis_task_parsers
676} // namespace o2::framework
677
678#endif // ANALYSISMANAGERS_H
constexpr uint32_t runtime_hash(char const *str)
void snapshot(const Output &spec, T const &object)
decltype(auto) make(const Output &spec, Args... args)
ServiceRegistryRef services()
Definition InitContext.h:34
ConfigParamRegistry const & options()
Definition InitContext.h:33
The input API of the Data Processing Layer This class holds the inputs which are valid for processing...
decltype(auto) get(R binding, int part=0) const
DataAllocator & outputs()
The data allocator is used to allocate memory for the output data.
ServiceRegistryRef services()
The services registry associated with this processing context.
GLuint entry
Definition glcorearb.h:5735
GLsizeiptr size
Definition glcorearb.h:659
GLint GLint GLint GLint GLint GLint GLint GLbitfield GLenum filter
Definition glcorearb.h:1308
bool prepareService(InitContext &, T &)
bool requestInputs(std::vector< InputSpec > &, T const &)
Table auto-creation handling.
void setGroupedCombination(C &, TG &, Ts &...)
Combinations handling.
bool initializeCache(ProcessingContext &, T &)
bool preInitializeCache(InitContext &, T &)
Cache handling.
bool prepareDelayedOutput(ProcessingContext &, T &)
bool prepareOption(InitContext &, O &)
bool appendOutput(std::vector< OutputSpec > &, T &, uint32_t)
Outputs handling.
bool finalizeOutput(ProcessingContext &, T &)
bool postRunOutput(EndOfStreamContext &, T &)
bool createExpressionTrees(std::vector< ExpressionInfo > &, T &)
bool prepareOutput(ProcessingContext &, T &)
bool appendOption(std::vector< ConfigParamSpec > &, O &)
Options handling.
bool registerCache(T &, Cache &, Cache &)
Preslice handling.
bool updateSliceInfo(T &, ArrowTableSlicingCache &)
bool appendCondition(std::vector< InputSpec > &, C &)
Conditions handling.
bool postRunService(EndOfStreamContext &, T &)
bool addService(std::vector< ServiceSpec > &, T &)
Service handling.
bool updatePlaceholders(InitContext &, T &)
Filter handling.
bool newDataframeCondition(InputRecord &, C &)
void updateExpressionInfos(expressions::Filter const &filter, std::vector< ExpressionInfo > &eInfos)
Function for attaching gandiva filters to to compatible task inputs.
void updatePlaceholders(Filter &filter, InitContext &context)
Update placeholder nodes from context.
Defining PrimaryVertex explicitly as messageable.
Definition TFIDInfo.h:20
std::vector< Entry > Cache
std::vector< ConfigParamSpec > ccdbParamSpec(std::string const &path, int runDependent, std::vector< CCDBMetadata > metadata={}, int qrate=0)
static bool appendOption(std::vector< ConfigParamSpec > &options, Configurable< T, K, IP > &what)
static void updateInputList(std::vector< InputSpec > &list, InputSpec &&input)
Updates list of InputSpecs by merging metadata.
static std::vector< LoadablePlugin > parsePluginSpecString(char const *str)
Parse a comma separated list of <library>:<plugin-name> plugin declarations.
static std::shared_ptr< arrow::Table > joinTables(std::vector< std::shared_ptr< arrow::Table > > &&tables, std::span< const char *const > labels)
Definition ASoA.cxx:67