Project
Loading...
Searching...
No Matches
WorkflowSerializationHelpers.cxx
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#include "WorkflowSerializationHelpers.h"
13#include "Framework/WorkflowSpec.h"
14#include "Framework/DataDescriptorQueryBuilder.h"
15#include "Framework/DataSpecUtils.h"
16#include "Framework/VariantHelpers.h"
17#include "Framework/VariantJSONHelpers.h"
18#include "Framework/DataDescriptorMatcher.h"
19#include "Framework/DataMatcherWalker.h"
20#include "Framework/Logger.h"
21#include "Framework/Signpost.h"
22
23#include <rapidjson/reader.h>
24#include <rapidjson/prettywriter.h>
25#include <rapidjson/istreamwrapper.h>
26#include <rapidjson/ostreamwrapper.h>
27#include <iostream>
28#include <algorithm>
29#include <memory>
30
31O2_DECLARE_DYNAMIC_LOG(workflow_importer);
32O2_DECLARE_DYNAMIC_LOG(post_workflow_importer);
33
34namespace o2::framework
35{
36
37using namespace rapidjson;
38using namespace o2::framework::data_matcher;
39
40struct WorkflowImporter : public rapidjson::BaseReaderHandler<rapidjson::UTF8<>, WorkflowImporter> {
101
102 friend std::ostream& operator<<(std::ostream& s, State state)
103 {
104 switch (state) {
105 case State::IN_START:
106 s << "IN_START";
107 break;
108 case State::IN_EXECUTION:
109 s << "IN_EXECUTION";
110 break;
111 case State::IN_WORKFLOW:
112 s << "IN_WORKFLOW";
113 break;
114 case State::IN_COMMAND:
115 s << "IN_COMMAND";
116 break;
117 case State::IN_DATAPROCESSORS:
118 s << "IN_DATAPROCESSORS";
119 break;
120 case State::IN_DATAPROCESSOR:
121 s << "IN_DATAPROCESSOR";
122 break;
123 case State::IN_DATAPROCESSOR_NAME:
124 s << "IN_DATAPROCESSOR_NAME";
125 break;
126 case State::IN_DATAPROCESSOR_RANK:
127 s << "IN_DATAPROCESSOR_RANK";
128 break;
129 case State::IN_DATAPROCESSOR_N_SLOTS:
130 s << "IN_DATAPROCESSOR_N_SLOTS";
131 break;
132 case State::IN_DATAPROCESSOR_TIMESLICE_ID:
133 s << "IN_DATAPROCESSOR_TIMESLICE_ID";
134 break;
135 case State::IN_DATAPROCESSOR_MAX_TIMESLICES:
136 s << "IN_DATAPROCESSOR_MAX_TIMESLICES";
137 break;
138 case State::IN_INPUTS:
139 s << "IN_INPUTS";
140 break;
141 case State::IN_OUTPUTS:
142 s << "IN_OUTPUTS";
143 break;
144 case State::IN_OPTIONS:
145 s << "IN_OPTIONS";
146 break;
147 case State::IN_LABELS:
148 s << "IN_LABELS";
149 break;
150 case State::IN_METADATA:
151 s << "IN_METADATA";
152 break;
153 case State::IN_WORKFLOW_OPTIONS:
154 s << "IN_WORKFLOW_OPTIONS";
155 break;
156 case State::IN_INPUT:
157 s << "IN_INPUT";
158 break;
159 case State::IN_INPUT_BINDING:
160 s << "IN_INPUT_BINDING";
161 break;
162 case State::IN_INPUT_ORIGIN:
163 s << "IN_INPUT_ORIGIN";
164 break;
165 case State::IN_INPUT_DESCRIPTION:
166 s << "IN_INPUT_DESCRIPTION";
167 break;
168 case State::IN_INPUT_SUBSPEC:
169 s << "IN_INPUT_SUBSPEC";
170 break;
171 case State::IN_INPUT_ORIGIN_REF:
172 s << "IN_INPUT_ORIGIN_REF";
173 break;
174 case State::IN_INPUT_DESCRIPTION_REF:
175 s << "IN_INPUT_DESCRIPTION_REF";
176 break;
177 case State::IN_INPUT_SUBSPEC_REF:
178 s << "IN_INPUT_SUBSPEC_REF";
179 break;
180 case State::IN_INPUT_MATCHER:
181 s << "IN_INPUT_MATCHER";
182 break;
183 case State::IN_INPUT_MATCHER_OPERATION:
184 s << "IN_INPUT_MATCHER_OPERATION";
185 break;
186 case State::IN_INPUT_LEFT_MATCHER:
187 s << "IN_INPUT_LEFT_MATCHER";
188 break;
189 case State::IN_INPUT_RIGHT_MATCHER:
190 s << "IN_INPUT_RIGHT_MATCHER";
191 break;
192 case State::IN_INPUT_LIFETIME:
193 s << "IN_INPUT_LIFETIME";
194 break;
195 case State::IN_INPUT_STARTTIME:
196 s << "IN_INPUT_STARTTIME";
197 break;
198 case State::IN_INPUT_OPTIONS:
199 s << "IN_INPUT_OPTIONS";
200 break;
201 case State::IN_OUTPUT:
202 s << "IN_OUTPUT";
203 break;
204 case State::IN_OUTPUT_BINDING:
205 s << "IN_OUTPUT_BINDING";
206 break;
207 case State::IN_OUTPUT_ORIGIN:
208 s << "IN_OUTPUT_ORIGIN";
209 break;
210 case State::IN_OUTPUT_DESCRIPTION:
211 s << "IN_OUTPUT_DESCRIPTION";
212 break;
213 case State::IN_OUTPUT_SUBSPEC:
214 s << "IN_OUTPUT_SUBSPEC";
215 break;
216 case State::IN_OUTPUT_LIFETIME:
217 s << "IN_OUTPUT_LIFETIME";
218 break;
219 case State::IN_OUTPUT_OPTIONS:
220 s << "IN_OUTPUT_OPTIONS";
221 break;
222 case State::IN_OPTION:
223 s << "IN_OPTION";
224 break;
225 case State::IN_OPTION_NAME:
226 s << "IN_OPTION_NAME";
227 break;
228 case State::IN_OPTION_TYPE:
229 s << "IN_OPTION_TYPE";
230 break;
231 case State::IN_OPTION_DEFAULT:
232 s << "IN_OPTION_DEFAULT";
233 break;
234 case State::IN_OPTION_HELP:
235 s << "IN_OPTION_HELP";
236 break;
237 case State::IN_OPTION_KIND:
238 s << "IN_OPTION_KIND";
239 break;
240 case State::IN_LABEL:
241 s << "IN_LABEL";
242 break;
243 case State::IN_METADATUM:
244 s << "IN_METADATUM";
245 break;
246 case State::IN_METADATUM_KEY:
247 s << "IN_METADATUM_KEY";
248 break;
249 case State::IN_METADATUM_VALUE:
250 s << "IN_METADATUM_VALUE";
251 break;
252 case State::IN_ERROR:
253 s << "IN_ERROR";
254 break;
255 case State::IN_DATAPROCESSOR_INFOS:
256 s << "IN_DATAPROCESSOR_INFOS";
257 break;
258 case State::IN_DATAPROCESSOR_INFO:
259 s << "IN_DATAPROCESSOR_INFO";
260 break;
261 case State::IN_DATAPROCESSOR_INFO_NAME:
262 s << "IN_DATAPROCESSOR_INFO_NAME";
263 break;
264 case State::IN_DATAPROCESSOR_INFO_EXECUTABLE:
265 s << "IN_DATAPROCESSOR_INFO_EXECUTABLE";
266 break;
267 case State::IN_DATAPROCESSOR_INFO_ARGS:
268 s << "IN_DATAPROCESSOR_INFO_ARGS";
269 break;
270 case State::IN_DATAPROCESSOR_INFO_ARG:
271 s << "IN_DATAPROCESSOR_INFO_ARG";
272 break;
273 case State::IN_DATAPROCESSOR_INFO_CHANNELS:
274 s << "IN_DATAPROCESSOR_INFO_CHANNELS";
275 break;
276 case State::IN_DATAPROCESSOR_INFO_CHANNEL:
277 s << "IN_DATAPROCESSOR_INFO_CHANNEL";
278 break;
279 }
280 return s;
281 }
282
283 WorkflowImporter(std::vector<DataProcessorSpec>& o,
284 std::vector<DataProcessorInfo>& m,
285 CommandInfo& c)
286 : states{},
287 dataProcessors{o},
288 metadata{m},
289 command{c}
290 {
291 push(State::IN_START);
292 }
293
294 bool StartObject()
295 {
296 enter("START_OBJECT");
297 if (in(State::IN_START)) {
298 push(State::IN_EXECUTION);
299 } else if (in(State::IN_DATAPROCESSORS)) {
300 push(State::IN_DATAPROCESSOR);
301 dataProcessors.push_back(DataProcessorSpec{});
302 } else if (in(State::IN_DATAPROCESSOR)) {
303 dataProcessors.push_back(DataProcessorSpec{});
304 } else if (in(State::IN_INPUTS)) {
305 push(State::IN_INPUT);
306 inputMatcherNodes.clear();
307 } else if (in(State::IN_INPUT_MATCHER)) {
308 // start a new embedded matcher
309 } else if (in(State::IN_INPUT_LEFT_MATCHER)) {
310 // this is a matcher leaf, i.e. last matcher of a branch
311 // will be merged into the parent matcher
312 } else if (in(State::IN_INPUT_RIGHT_MATCHER)) {
313 // this is a matcher leaf, i.e. last matcher of a branch
314 // will be merged into the parent matcher
315 } else if (in(State::IN_OUTPUTS)) {
316 push(State::IN_OUTPUT);
317 outputHasSubSpec = false;
318 } else if (in(State::IN_OPTIONS)) {
319 push(State::IN_OPTION);
320 } else if (in(State::IN_INPUT_OPTIONS)) {
321 push(State::IN_OPTION);
322 } else if (in(State::IN_OUTPUT_OPTIONS)) {
323 push(State::IN_OPTION);
324 } else if (in(State::IN_WORKFLOW_OPTIONS)) {
325 push(State::IN_OPTION);
326 } else if (in(State::IN_DATAPROCESSOR_INFOS)) {
327 push(State::IN_DATAPROCESSOR_INFO);
328 metadata.push_back(DataProcessorInfo{});
329 } else if (in(State::IN_DATAPROCESSOR_INFO)) {
330 metadata.push_back(DataProcessorInfo{});
331 } else if (in(State::IN_METADATA)) {
332 push(State::IN_METADATUM);
333 metadatumKey.clear();
334 metadatumValue.clear();
335 } else if (in(State::IN_COMMAND)) {
336 command = CommandInfo{};
337 }
338 return true;
339 }
340
341 bool EndObject(SizeType memberCount)
342 {
343 enter("END_OBJECT");
344 if (in(State::IN_INPUT)) {
345 auto buildMatcher = [](auto& nodes) -> std::unique_ptr<DataDescriptorMatcher> {
346 auto lastMatcher =
347 std::make_unique<DataDescriptorMatcher>(DataDescriptorMatcher::Op::Just,
348 StartTimeValueMatcher(ContextRef{ContextPos::STARTTIME_POS}));
349 for (size_t ni = 0, ne = nodes.size(); ni < ne; ++ni) {
350 auto& node = nodes[nodes.size() - 1 - ni];
351 auto tmp = std::make_unique<DataDescriptorMatcher>(DataDescriptorMatcher::Op::And,
352 std::move(node),
353 std::move(lastMatcher));
354 assert(lastMatcher.get() == nullptr);
355 lastMatcher = std::move(tmp);
356 }
357 return lastMatcher;
358 };
359
360 std::unique_ptr<DataDescriptorMatcher> matcher;
361 if (auto* pval = std::get_if<std::unique_ptr<DataDescriptorMatcher>>(&inputMatcherNodes[0])) {
362 assert(inputMatcherNodes.size() == 1);
363 matcher = std::move(*pval);
364 } else {
365 matcher = buildMatcher(inputMatcherNodes);
366 }
367 auto concrete = DataSpecUtils::optionalConcreteDataMatcherFrom(*matcher);
368 if (concrete.has_value()) {
369 // the matcher is fully qualified with unique parameters so we add ConcreteDataMatcher
370 dataProcessors.back().inputs.push_back(InputSpec({binding}, (*concrete).origin, (*concrete).description, (*concrete).subSpec, lifetime, inputOptions));
371 } else {
372 dataProcessors.back().inputs.push_back(InputSpec({binding}, std::move(*matcher), lifetime, inputOptions));
373 }
374 inputMatcherNodes.clear();
375 inputOptions.clear();
376
377 } else if (in(State::IN_INPUT_MATCHER) && inputMatcherNodes.size() > 1) {
378 data_matcher::Node child = std::move(inputMatcherNodes.back());
379 inputMatcherNodes.pop_back();
380 auto* matcher = std::get_if<std::unique_ptr<DataDescriptorMatcher>>(&child);
381 assert(matcher != nullptr);
382 auto* parent = std::get_if<std::unique_ptr<DataDescriptorMatcher>>(&inputMatcherNodes.back());
383 assert(parent != nullptr);
384 std::unique_ptr<DataDescriptorMatcher> node;
385 auto mergeDown = [&node, &parent, &child]() -> bool {
386 // FIXME: do we need a dedicated default state, or can we simply use ConstantValueMatcher
387 if (auto* pval1 = std::get_if<ConstantValueMatcher>(&((*parent)->getLeft()))) {
388 if (*pval1 == ConstantValueMatcher{false}) {
389 node = std::make_unique<DataDescriptorMatcher>((*parent)->getOp(),
390 std::move(child),
391 std::move((*parent)->getRight()));
392 return true;
393 }
394 }
395 if (auto* pval2 = std::get_if<ConstantValueMatcher>(&((*parent)->getRight()))) {
396 if (*pval2 == ConstantValueMatcher{false}) {
397 node = std::make_unique<DataDescriptorMatcher>((*parent)->getOp(),
398 std::move((*parent)->getLeft()),
399 std::move(child));
400 return true;
401 }
402 }
403 return false;
404 };
405 if (!mergeDown()) {
406 states.push_back(State::IN_ERROR);
407 }
408 inputMatcherNodes.pop_back();
409 inputMatcherNodes.push_back(std::move(node));
410 } else if (in(State::IN_INPUT_LEFT_MATCHER)) {
411 assert(inputMatcherNodes.size() >= 2);
412 size_t nMatchers = inputMatcherNodes.size();
413 auto* parent = std::get_if<std::unique_ptr<DataDescriptorMatcher>>(&inputMatcherNodes[nMatchers - 2]);
414 assert(parent != nullptr);
415 auto node = std::make_unique<DataDescriptorMatcher>((*parent)->getOp(),
416 std::move(inputMatcherNodes[nMatchers - 1]),
417 std::move((*parent)->getRight()));
418 inputMatcherNodes.pop_back();
419 inputMatcherNodes.pop_back();
420 inputMatcherNodes.push_back(std::move(node));
421 } else if (in(State::IN_INPUT_RIGHT_MATCHER)) {
422 data_matcher::Node child = std::move(inputMatcherNodes.back());
423 inputMatcherNodes.pop_back();
424 auto* parent = std::get_if<std::unique_ptr<DataDescriptorMatcher>>(&inputMatcherNodes.back());
425 assert(parent != nullptr);
426 auto node = std::make_unique<DataDescriptorMatcher>((*parent)->getOp(),
427 std::move((*parent)->getLeft()),
428 std::move(child));
429 inputMatcherNodes.pop_back();
430 inputMatcherNodes.push_back(std::move(node));
431 } else if (in(State::IN_OUTPUT)) {
432 if (outputHasSubSpec) {
433 dataProcessors.back().outputs.push_back(OutputSpec({binding}, origin, description, subspec, lifetime, outputOptions));
434 } else {
435 dataProcessors.back().outputs.push_back(OutputSpec({binding}, {origin, description}, lifetime, outputOptions));
436 }
437 outputOptions.clear();
438 outputHasSubSpec = false;
439 } else if (in(State::IN_OPTION)) {
440 std::unique_ptr<ConfigParamSpec> opt{nullptr};
441
442 using HelpString = ConfigParamSpec::HelpString;
443 std::stringstream is;
444 is.str(optionDefault);
445 switch (optionType) {
446 case VariantType::String:
447 opt = std::make_unique<ConfigParamSpec>(optionName, optionType, optionDefault.c_str(), HelpString{optionHelp}, optionKind);
448 break;
449 case VariantType::Int:
450 opt = std::make_unique<ConfigParamSpec>(optionName, optionType, std::stoi(optionDefault, nullptr), HelpString{optionHelp}, optionKind);
451 break;
452 case VariantType::Int8:
453 opt = std::make_unique<ConfigParamSpec>(optionName, optionType, static_cast<int8_t>(std::stoi(optionDefault, nullptr)), HelpString{optionHelp}, optionKind);
454 break;
455 case VariantType::Int16:
456 opt = std::make_unique<ConfigParamSpec>(optionName, optionType, static_cast<int16_t>(std::stoi(optionDefault, nullptr)), HelpString{optionHelp}, optionKind);
457 break;
458 case VariantType::UInt8:
459 opt = std::make_unique<ConfigParamSpec>(optionName, optionType, static_cast<uint8_t>(std::stoi(optionDefault, nullptr)), HelpString{optionHelp}, optionKind);
460 break;
461 case VariantType::UInt16:
462 opt = std::make_unique<ConfigParamSpec>(optionName, optionType, static_cast<uint16_t>(std::stoi(optionDefault, nullptr)), HelpString{optionHelp}, optionKind);
463 break;
464 case VariantType::UInt32:
465 opt = std::make_unique<ConfigParamSpec>(optionName, optionType, static_cast<uint32_t>(std::stoul(optionDefault, nullptr)), HelpString{optionHelp}, optionKind);
466 break;
467 case VariantType::UInt64:
468 opt = std::make_unique<ConfigParamSpec>(optionName, optionType, std::stoul(optionDefault, nullptr), HelpString{optionHelp}, optionKind);
469 break;
470 case VariantType::Int64:
471 opt = std::make_unique<ConfigParamSpec>(optionName, optionType, std::stol(optionDefault, nullptr), HelpString{optionHelp}, optionKind);
472 break;
473 case VariantType::Float:
474 opt = std::make_unique<ConfigParamSpec>(optionName, optionType, std::stof(optionDefault, nullptr), HelpString{optionHelp}, optionKind);
475 break;
476 case VariantType::Double:
477 opt = std::make_unique<ConfigParamSpec>(optionName, optionType, std::stod(optionDefault, nullptr), HelpString{optionHelp}, optionKind);
478 break;
479 case VariantType::Bool:
480 opt = std::make_unique<ConfigParamSpec>(optionName, optionType, (bool)std::stoi(optionDefault, nullptr), HelpString{optionHelp}, optionKind);
481 break;
482 case VariantType::ArrayInt:
483 opt = std::make_unique<ConfigParamSpec>(optionName, optionType, VariantJSONHelpers::read<VariantType::ArrayInt>(is), HelpString{optionHelp}, optionKind);
484 break;
485 case VariantType::ArrayFloat:
486 opt = std::make_unique<ConfigParamSpec>(optionName, optionType, VariantJSONHelpers::read<VariantType::ArrayFloat>(is), HelpString{optionHelp}, optionKind);
487 break;
488 case VariantType::ArrayDouble:
489 opt = std::make_unique<ConfigParamSpec>(optionName, optionType, VariantJSONHelpers::read<VariantType::ArrayDouble>(is), HelpString{optionHelp}, optionKind);
490 break;
491 // case VariantType::ArrayBool:
492 // opt = std::make_unique<ConfigParamSpec>(optionName, optionType, VariantJSONHelpers::read<VariantType::ArrayBool>(is), HelpString{optionHelp}, optionKind);
493 // break;
494 case VariantType::ArrayString:
495 opt = std::make_unique<ConfigParamSpec>(optionName, optionType, VariantJSONHelpers::read<VariantType::ArrayString>(is), HelpString{optionHelp}, optionKind);
496 break;
497 case VariantType::Array2DInt:
498 opt = std::make_unique<ConfigParamSpec>(optionName, optionType, VariantJSONHelpers::read<VariantType::Array2DInt>(is), HelpString{optionHelp}, optionKind);
499 break;
500 case VariantType::Array2DFloat:
501 opt = std::make_unique<ConfigParamSpec>(optionName, optionType, VariantJSONHelpers::read<VariantType::Array2DFloat>(is), HelpString{optionHelp}, optionKind);
502 break;
503 case VariantType::Array2DDouble:
504 opt = std::make_unique<ConfigParamSpec>(optionName, optionType, VariantJSONHelpers::read<VariantType::Array2DDouble>(is), HelpString{optionHelp}, optionKind);
505 break;
506 case VariantType::LabeledArrayInt:
507 opt = std::make_unique<ConfigParamSpec>(optionName, optionType, VariantJSONHelpers::read<VariantType::LabeledArrayInt>(is), HelpString{optionHelp}, optionKind);
508 break;
509 case VariantType::LabeledArrayFloat:
510 opt = std::make_unique<ConfigParamSpec>(optionName, optionType, VariantJSONHelpers::read<VariantType::LabeledArrayFloat>(is), HelpString{optionHelp}, optionKind);
511 break;
512 case VariantType::LabeledArrayDouble:
513 opt = std::make_unique<ConfigParamSpec>(optionName, optionType, VariantJSONHelpers::read<VariantType::LabeledArrayDouble>(is), HelpString{optionHelp}, optionKind);
514 break;
515 case VariantType::LabeledArrayString:
516 opt = std::make_unique<ConfigParamSpec>(optionName, optionType, VariantJSONHelpers::read<VariantType::LabeledArrayString>(is), HelpString{optionHelp}, optionKind);
517 break;
518 case VariantType::Dict:
519 opt = std::make_unique<ConfigParamSpec>(optionName, optionType, emptyDict(), HelpString{optionHelp}, optionKind);
520 break;
521 default:
522 opt = std::make_unique<ConfigParamSpec>(optionName, optionType, optionDefault, HelpString{optionHelp}, optionKind);
523 }
524 // Depending on the previous state, push options to the right place.
525 if (previousIs(State::IN_OPTIONS)) {
526 dataProcessors.back().options.push_back(*opt);
527 } else if (previousIs(State::IN_WORKFLOW_OPTIONS)) {
528 metadata.back().workflowOptions.push_back(*opt);
529 } else if (previousIs(State::IN_INPUT_OPTIONS)) {
530 inputOptions.push_back(*opt);
531 } else if (previousIs(State::IN_OUTPUT_OPTIONS)) {
532 outputOptions.push_back(*opt);
533 } else {
534 assert(false);
535 }
536 } else if (in(State::IN_METADATUM)) {
537 dataProcessors.back().metadata.push_back({metadatumKey, metadatumValue});
538 }
539 pop();
540 return true;
541 }
542
543 bool StartArray()
544 {
545 enter("START_ARRAY");
546 if (in(State::IN_WORKFLOW)) {
547 push(State::IN_DATAPROCESSORS);
548 } else if (in(State::IN_INPUTS)) {
549 push(State::IN_INPUT);
550 } else if (in(State::IN_INPUT_OPTIONS)) {
551 push(State::IN_OPTION);
552 } else if (in(State::IN_OUTPUT_OPTIONS)) {
553 push(State::IN_OPTION);
554 } else if (in(State::IN_OUTPUTS)) {
555 push(State::IN_OUTPUT);
556 outputHasSubSpec = false;
557 } else if (in(State::IN_OPTIONS)) {
558 push(State::IN_OPTION);
559 } else if (in(State::IN_WORKFLOW_OPTIONS)) {
560 push(State::IN_OPTION);
561 } else if (in(State::IN_LABELS)) {
562 push(State::IN_LABEL);
563 } else if (in(State::IN_METADATA)) {
564 push(State::IN_METADATUM);
565 } else if (in(State::IN_DATAPROCESSOR_INFOS)) {
566 push(State::IN_DATAPROCESSOR_INFO);
567 } else if (in(State::IN_DATAPROCESSOR_INFO_ARGS)) {
568 push(State::IN_DATAPROCESSOR_INFO_ARG);
569 } else if (in(State::IN_DATAPROCESSOR_INFO_CHANNELS)) {
570 push(State::IN_DATAPROCESSOR_INFO_CHANNEL);
571 }
572 return true;
573 }
574
575 bool EndArray(SizeType count)
576 {
577 enter("END_ARRAY");
578 // Handle the case in which inputs / options / outputs are
579 // empty.
580 if (in(State::IN_INPUT) || in(State::IN_OUTPUT) || in(State::IN_OPTION) || in(State::IN_LABEL) || in(State::IN_METADATUM) ||
581 in(State::IN_DATAPROCESSOR_INFO) || in(State::IN_DATAPROCESSOR_INFO_ARG) || in(State::IN_DATAPROCESSOR_INFO_CHANNEL) ||
582 in(State::IN_DATAPROCESSORS)) {
583 pop();
584 }
585 pop();
586 return true;
587 }
588
589 bool Key(const Ch* str, SizeType length, bool copy)
590 {
591 enter("KEY");
592 enter(str);
593 if (in(State::IN_INPUT) && strncmp(str, "binding", length) == 0) {
594 push(State::IN_INPUT_BINDING);
595 } else if (in(State::IN_INPUT) && strncmp(str, "origin", length) == 0) {
596 push(State::IN_INPUT_ORIGIN);
597 } else if (in(State::IN_INPUT) && strncmp(str, "description", length) == 0) {
598 push(State::IN_INPUT_DESCRIPTION);
599 } else if (in(State::IN_INPUT) && strncmp(str, "subspec", length) == 0) {
600 push(State::IN_INPUT_SUBSPEC);
601 } else if (in(State::IN_INPUT) && strncmp(str, "originRef", length) == 0) {
602 push(State::IN_INPUT_ORIGIN_REF);
603 } else if (in(State::IN_INPUT) && strncmp(str, "descriptionRef", length) == 0) {
604 push(State::IN_INPUT_DESCRIPTION_REF);
605 } else if (in(State::IN_INPUT) && strncmp(str, "subspecRef", length) == 0) {
606 push(State::IN_INPUT_SUBSPEC_REF);
607 } else if (in(State::IN_INPUT) && strncmp(str, "matcher", length) == 0) {
608 // the outermost matcher is starting here
609 // we create a placeholder which is being updated later
610 inputMatcherNodes.push_back(std::make_unique<DataDescriptorMatcher>(DataDescriptorMatcher::Op::And, ConstantValueMatcher{false}));
611 push(State::IN_INPUT_MATCHER);
612 } else if (in(State::IN_INPUT_MATCHER) && strncmp(str, "matcher", length) == 0) {
613 // recursive matchers
614 inputMatcherNodes.push_back(std::make_unique<DataDescriptorMatcher>(DataDescriptorMatcher::Op::And, ConstantValueMatcher{false}));
615 push(State::IN_INPUT_MATCHER);
616 } else if (in(State::IN_INPUT_MATCHER) && strncmp(str, "operation", length) == 0) {
617 push(State::IN_INPUT_MATCHER_OPERATION);
618 } else if (in(State::IN_INPUT_MATCHER) && strncmp(str, "left", length) == 0) {
619 push(State::IN_INPUT_LEFT_MATCHER);
620 } else if (in(State::IN_INPUT_MATCHER) && strncmp(str, "right", length) == 0) {
621 push(State::IN_INPUT_RIGHT_MATCHER);
622 } else if (in(State::IN_INPUT_LEFT_MATCHER) && strncmp(str, "origin", length) == 0) {
623 push(State::IN_INPUT_ORIGIN);
624 } else if (in(State::IN_INPUT_RIGHT_MATCHER) && strncmp(str, "origin", length) == 0) {
625 push(State::IN_INPUT_ORIGIN);
626 } else if (in(State::IN_INPUT_LEFT_MATCHER) && strncmp(str, "description", length) == 0) {
627 push(State::IN_INPUT_DESCRIPTION);
628 } else if (in(State::IN_INPUT_RIGHT_MATCHER) && strncmp(str, "description", length) == 0) {
629 push(State::IN_INPUT_DESCRIPTION);
630 } else if (in(State::IN_INPUT_LEFT_MATCHER) && strncmp(str, "subspec", length) == 0) {
631 push(State::IN_INPUT_SUBSPEC);
632 } else if (in(State::IN_INPUT_RIGHT_MATCHER) && strncmp(str, "subspec", length) == 0) {
633 push(State::IN_INPUT_SUBSPEC);
634 } else if (in(State::IN_INPUT_LEFT_MATCHER) && strncmp(str, "originRef", length) == 0) {
635 push(State::IN_INPUT_ORIGIN_REF);
636 } else if (in(State::IN_INPUT_RIGHT_MATCHER) && strncmp(str, "originRef", length) == 0) {
637 push(State::IN_INPUT_ORIGIN_REF);
638 } else if (in(State::IN_INPUT_LEFT_MATCHER) && strncmp(str, "descriptionRef", length) == 0) {
639 push(State::IN_INPUT_DESCRIPTION_REF);
640 } else if (in(State::IN_INPUT_RIGHT_MATCHER) && strncmp(str, "descriptionRef", length) == 0) {
641 push(State::IN_INPUT_DESCRIPTION_REF);
642 } else if (in(State::IN_INPUT_LEFT_MATCHER) && strncmp(str, "subspecRef", length) == 0) {
643 push(State::IN_INPUT_SUBSPEC_REF);
644 } else if (in(State::IN_INPUT_RIGHT_MATCHER) && strncmp(str, "subspecRef", length) == 0) {
645 push(State::IN_INPUT_SUBSPEC_REF);
646 } else if (in(State::IN_INPUT_LEFT_MATCHER) && strncmp(str, "starttime", length) == 0) {
647 push(State::IN_INPUT_STARTTIME);
648 } else if (in(State::IN_INPUT_RIGHT_MATCHER) && strncmp(str, "starttime", length) == 0) {
649 push(State::IN_INPUT_STARTTIME);
650 } else if (in(State::IN_INPUT) && strncmp(str, "lifetime", length) == 0) {
651 push(State::IN_INPUT_LIFETIME);
652 } else if (in(State::IN_INPUT) && strncmp(str, "starttime", length) == 0) {
653 push(State::IN_INPUT_STARTTIME);
654 } else if (in(State::IN_INPUT) && strncmp(str, "metadata", length) == 0) {
655 push(State::IN_INPUT_OPTIONS);
656 } else if (in(State::IN_OUTPUT) && strncmp(str, "binding", length) == 0) {
657 push(State::IN_OUTPUT_BINDING);
658 } else if (in(State::IN_OUTPUT) && strncmp(str, "origin", length) == 0) {
659 push(State::IN_OUTPUT_ORIGIN);
660 } else if (in(State::IN_OUTPUT) && strncmp(str, "description", length) == 0) {
661 push(State::IN_OUTPUT_DESCRIPTION);
662 } else if (in(State::IN_OUTPUT) && strncmp(str, "subspec", length) == 0) {
663 push(State::IN_OUTPUT_SUBSPEC);
664 outputHasSubSpec = true;
665 } else if (in(State::IN_OUTPUT) && strncmp(str, "lifetime", length) == 0) {
666 push(State::IN_OUTPUT_LIFETIME);
667 } else if (in(State::IN_OUTPUT) && strncmp(str, "metadata", length) == 0) {
668 push(State::IN_OUTPUT_OPTIONS);
669 } else if (in(State::IN_DATAPROCESSOR) && strncmp(str, "name", length) == 0) {
670 push(State::IN_DATAPROCESSOR_NAME);
671 } else if (in(State::IN_DATAPROCESSOR) && strncmp(str, "ranks", length) == 0) {
672 push(State::IN_DATAPROCESSOR_RANK);
673 } else if (in(State::IN_DATAPROCESSOR) && strncmp(str, "nSlots", length) == 0) {
674 push(State::IN_DATAPROCESSOR_N_SLOTS);
675 } else if (in(State::IN_DATAPROCESSOR) && strncmp(str, "inputTimeSliceId", length) == 0) {
676 push(State::IN_DATAPROCESSOR_TIMESLICE_ID);
677 } else if (in(State::IN_DATAPROCESSOR) && strncmp(str, "maxInputTimeslices", length) == 0) {
678 push(State::IN_DATAPROCESSOR_MAX_TIMESLICES);
679 } else if (in(State::IN_DATAPROCESSOR) && strncmp(str, "inputs", length) == 0) {
680 push(State::IN_INPUTS);
681 } else if (in(State::IN_DATAPROCESSOR) && strncmp(str, "outputs", length) == 0) {
682 push(State::IN_OUTPUTS);
683 } else if (in(State::IN_DATAPROCESSOR) && strncmp(str, "options", length) == 0) {
684 push(State::IN_OPTIONS);
685 } else if (in(State::IN_DATAPROCESSOR) && strncmp(str, "labels", length) == 0) {
686 push(State::IN_LABELS);
687 } else if (in(State::IN_DATAPROCESSOR) && strncmp(str, "metadata", length) == 0) {
688 push(State::IN_METADATA);
689 } else if (in(State::IN_METADATUM) && strncmp(str, "key", length) == 0) {
690 push(State::IN_METADATUM_KEY);
691 } else if (in(State::IN_METADATUM) && strncmp(str, "value", length) == 0) {
692 push(State::IN_METADATUM_VALUE);
693 } else if (in(State::IN_EXECUTION) && strncmp(str, "workflow", length) == 0) {
694 push(State::IN_WORKFLOW);
695 } else if (in(State::IN_EXECUTION) && strncmp(str, "metadata", length) == 0) {
696 push(State::IN_DATAPROCESSOR_INFOS);
697 } else if (in(State::IN_OPTION) && strncmp(str, "name", length) == 0) {
698 push(State::IN_OPTION_NAME);
699 } else if (in(State::IN_OPTION) && strncmp(str, "type", length) == 0) {
700 push(State::IN_OPTION_TYPE);
701 } else if (in(State::IN_OPTION) && strncmp(str, "defaultValue", length) == 0) {
702 push(State::IN_OPTION_DEFAULT);
703 } else if (in(State::IN_OPTION) && strncmp(str, "help", length) == 0) {
704 push(State::IN_OPTION_HELP);
705 } else if (in(State::IN_OPTION) && strncmp(str, "kind", length) == 0) {
706 push(State::IN_OPTION_KIND);
707 } else if (in(State::IN_DATAPROCESSOR_INFO) && strncmp(str, "name", length) == 0) {
708 push(State::IN_DATAPROCESSOR_INFO_NAME);
709 } else if (in(State::IN_DATAPROCESSOR_INFO) && strncmp(str, "executable", length) == 0) {
710 push(State::IN_DATAPROCESSOR_INFO_EXECUTABLE);
711 } else if (in(State::IN_DATAPROCESSOR_INFO) && strncmp(str, "cmdLineArgs", length) == 0) {
712 push(State::IN_DATAPROCESSOR_INFO_ARGS);
713 } else if (in(State::IN_DATAPROCESSOR_INFO) && strncmp(str, "workflowOptions", length) == 0) {
714 push(State::IN_WORKFLOW_OPTIONS);
715 } else if (in(State::IN_DATAPROCESSOR_INFO) && strncmp(str, "channels", length) == 0) {
716 push(State::IN_DATAPROCESSOR_INFO_CHANNELS);
717 } else if (in(State::IN_EXECUTION) && strncmp(str, "command", length) == 0) {
718 push(State::IN_COMMAND);
719 }
720 return true;
721 }
722
723 bool String(const Ch* str, SizeType length, bool copy)
724 {
725 enter("STRING");
726 enter(str);
727 auto s = std::string(str, length);
728 if (in(State::IN_DATAPROCESSOR_NAME)) {
729 assert(dataProcessors.size());
730 dataProcessors.back().name = s;
731 } else if (in(State::IN_DATAPROCESSOR_INFO_NAME)) {
732 assert(metadata.size());
733 metadata.back().name = s;
734 } else if (in(State::IN_DATAPROCESSOR_INFO_EXECUTABLE)) {
735 assert(metadata.size());
736 metadata.back().executable = s;
737 } else if (in(State::IN_INPUT_BINDING)) {
738 binding = s;
739 } else if (in(State::IN_INPUT_ORIGIN)) {
740 origin.runtimeInit(s.c_str(), std::min(s.size(), 4UL));
741 std::string v(s.c_str(), std::min(s.size(), 4UL));
742 inputMatcherNodes.push_back(OriginValueMatcher{v});
743 } else if (in(State::IN_INPUT_DESCRIPTION)) {
744 description.runtimeInit(s.c_str(), std::min(s.size(), 16UL));
745 std::string v(s.c_str(), std::min(s.size(), 16UL));
746 inputMatcherNodes.push_back(DescriptionValueMatcher{v});
747 } else if (in(State::IN_INPUT_STARTTIME)) {
748 // we add StartTimeValueMatcher with ContextRef for starttime, no matter what
749 // has been in the configuration.
750 inputMatcherNodes.push_back(StartTimeValueMatcher(ContextRef{ContextPos::STARTTIME_POS}));
751 } else if (in(State::IN_INPUT_MATCHER_OPERATION)) {
752 // FIXME: need to implement operator>> to read the op parameter
753 DataDescriptorMatcher::Op op = DataDescriptorMatcher::Op::And;
754 if (s == "and") {
755 op = DataDescriptorMatcher::Op::And;
756 } else if (s == "or") {
757 op = DataDescriptorMatcher::Op::Or;
758 } else if (s == "xor") {
759 op = DataDescriptorMatcher::Op::Xor;
760 } else if (s == "just") {
761 op = DataDescriptorMatcher::Op::Just;
762 } else if (s == "not") {
763 op = DataDescriptorMatcher::Op::Not;
764 }
765 // FIXME: we could drop the placeholder which has been added when entering
766 // the states which can read key 'operation', but then we need to make sure
767 // that this key is always present
768 auto node = std::make_unique<DataDescriptorMatcher>(op, ConstantValueMatcher{false});
769 inputMatcherNodes.pop_back();
770 inputMatcherNodes.push_back(std::move(node));
771 } else if (in(State::IN_OUTPUT_BINDING)) {
772 binding = s;
773 } else if (in(State::IN_OUTPUT_ORIGIN)) {
774 origin.runtimeInit(s.c_str(), std::min(s.size(), 4UL));
775 } else if (in(State::IN_OUTPUT_DESCRIPTION)) {
776 description.runtimeInit(s.c_str(), std::min(s.size(), 16UL));
777 } else if (in(State::IN_OPTION_NAME)) {
778 optionName = s;
779 } else if (in(State::IN_OPTION_TYPE)) {
780 optionType = (VariantType)std::stoi(s, nullptr);
781 } else if (in(State::IN_OPTION_KIND)) {
782 optionKind = (ConfigParamKind)std::stoi(s, nullptr);
783 } else if (in(State::IN_OPTION_DEFAULT)) {
784 optionDefault = s;
785 } else if (in(State::IN_OPTION_HELP)) {
786 optionHelp = s;
787 } else if (in(State::IN_LABEL)) {
788 dataProcessors.back().labels.push_back({s});
789 // This is in an array, so we do not actually want to
790 // exit from the state.
791 push(State::IN_LABEL);
792 } else if (in(State::IN_METADATUM_KEY)) {
793 metadatumKey = s;
794 } else if (in(State::IN_METADATUM_VALUE)) {
795 metadatumValue = s;
796 } else if (in(State::IN_DATAPROCESSOR_INFO_ARG)) {
797 metadata.back().cmdLineArgs.push_back(s);
798 // This is in an array, so we do not actually want to
799 // exit from the state.
800 push(State::IN_DATAPROCESSOR_INFO_ARG);
801 } else if (in(State::IN_DATAPROCESSOR_INFO_CHANNEL)) {
802 metadata.back().channels.push_back(s);
803 // This is in an array, so we do not actually want to
804 // exit from the state.
805 push(State::IN_DATAPROCESSOR_INFO_CHANNEL);
806 } else if (in(State::IN_COMMAND)) {
807 command.merge({s});
808 } else {
809 std::stringstream errstr;
810 errstr << "No string handling for argument '" << std::string(str, length) << "' in state " << states.back() << std::endl;
811 throw std::runtime_error(errstr.str());
812 }
813 pop();
814 return true;
815 }
816
817 bool Uint(unsigned i)
818 {
819 O2_SIGNPOST_EVENT_EMIT(workflow_importer, _o2_signpost_id_t{(int64_t)states.size()}, "import", "Uint(%d)", i);
820 if (in(State::IN_INPUT_SUBSPEC)) {
821 subspec = i;
822 inputMatcherNodes.push_back(SubSpecificationTypeValueMatcher{i});
823 } else if (in(State::IN_INPUT_ORIGIN_REF)) {
824 ref = i;
825 inputMatcherNodes.push_back(OriginValueMatcher{ContextRef{i}});
826 } else if (in(State::IN_INPUT_DESCRIPTION_REF)) {
827 ref = i;
828 inputMatcherNodes.push_back(DescriptionValueMatcher{ContextRef{i}});
829 } else if (in(State::IN_INPUT_SUBSPEC_REF)) {
830 ref = i;
831 inputMatcherNodes.push_back(SubSpecificationTypeValueMatcher{ContextRef{i}});
832 } else if (in(State::IN_OUTPUT_SUBSPEC)) {
833 subspec = i;
834 } else if (in(State::IN_INPUT_LIFETIME)) {
835 lifetime = (Lifetime)i;
836 } else if (in(State::IN_OUTPUT_LIFETIME)) {
837 lifetime = (Lifetime)i;
838 } else if (in(State::IN_DATAPROCESSOR_RANK)) {
839 dataProcessors.back().rank = i;
840 } else if (in(State::IN_DATAPROCESSOR_N_SLOTS)) {
841 dataProcessors.back().nSlots = i;
842 } else if (in(State::IN_DATAPROCESSOR_TIMESLICE_ID)) {
843 dataProcessors.back().inputTimeSliceId = i;
844 } else if (in(State::IN_DATAPROCESSOR_MAX_TIMESLICES)) {
845 dataProcessors.back().maxInputTimeslices = i;
846 }
847 pop();
848 return true;
849 }
850
851 bool Int(int i)
852 {
853 O2_SIGNPOST_EVENT_EMIT(workflow_importer, _o2_signpost_id_t{(int64_t)states.size()}, "import", "Int(%d)", i);
854 return true;
855 }
856 bool Uint64(uint64_t u)
857 {
858 O2_SIGNPOST_EVENT_EMIT(workflow_importer, _o2_signpost_id_t{(int64_t)states.size()}, "import", "Uint64(%" PRIu64 ")", u);
859 return true;
860 }
861 bool Double(double d)
862 {
863 O2_SIGNPOST_EVENT_EMIT(workflow_importer, _o2_signpost_id_t{(int64_t)states.size()}, "import", "Double(%f)", d);
864 return true;
865 }
866
867 void enter(char const* what)
868 {
869 O2_SIGNPOST_EVENT_EMIT(workflow_importer, _o2_signpost_id_t{(int64_t)states.size()}, "import", "ENTER: %s", what);
870 }
871
872 void push(State state)
873 {
874 debug.str("");
875 debug << state;
876 states.push_back(state);
877 O2_SIGNPOST_START(workflow_importer, _o2_signpost_id_t{(int64_t)states.size()}, "import", "PUSH: %s", debug.str().c_str());
878 }
879
880 State pop()
881 {
882 if (states.empty()) {
883 states.push_back(State::IN_ERROR);
884 return State::IN_ERROR;
885 }
886 auto result = states.back();
887 states.pop_back();
888 debug.str("");
889 debug << result;
890 if (!states.empty()) {
891 debug << " now in " << states.back();
892 }
893 O2_SIGNPOST_END(workflow_importer, _o2_signpost_id_t{(int64_t)states.size()+1}, "import", "POP: %s", debug.str().c_str());
894 return result;
895 }
896 bool in(State o)
897 {
898 return states.back() == o;
899 }
900
901 bool previousIs(State o)
902 {
903 assert(states.size() > 1);
904 return states[states.size() - 2] == o;
905 }
906
907 std::ostringstream debug;
908 std::vector<State> states;
909 std::string spec;
910 std::vector<DataProcessorSpec>& dataProcessors;
911 std::vector<DataProcessorInfo>& metadata;
912 CommandInfo& command;
913 std::vector<ConfigParamSpec> inputOptions;
914 std::vector<ConfigParamSpec> outputOptions;
915 std::string binding;
916 header::DataOrigin origin;
917 header::DataDescription description;
918 size_t subspec;
919 size_t ref;
920 Lifetime lifetime;
921 std::string metadatumKey;
922 std::string metadatumValue;
923 std::string optionName;
924 VariantType optionType;
925 std::string optionDefault;
926 std::string optionHelp;
927 ConfigParamKind optionKind;
928 bool outputHasSubSpec;
929 std::vector<data_matcher::Node> inputMatcherNodes;
930};
931
932bool WorkflowSerializationHelpers::import(std::istream& s,
933 std::vector<DataProcessorSpec>& workflow,
934 std::vector<DataProcessorInfo>& metadata,
935 CommandInfo& command)
936{
937 // Skip any line which does not start with '{'
938 // If we do not find a starting {, we simply assume that no workflow
939 // was actually passed on the PIPE.
940 // FIXME: not particularly resilient, but works for now.
941 // FIXME: this will fail if { is found at char 1024.
942 char buf[1024];
943 bool hasFatalImportError = false;
944 while (s.peek() != '{') {
945 if (s.eof()) {
946 return !hasFatalImportError;
947 }
948 if (s.fail() || s.bad()) {
949 throw std::runtime_error("Malformatted input workflow");
950 }
951 s.getline(buf, 1024, '\n');
952 // FairLogger messages (starting with [) simply get forwarded.
953 // Other messages we consider them as ERRORs since they
954 // were printed out without FairLogger.
955 if (buf[0] == '[') {
956 if (strncmp(buf, "[ERROR] invalid workflow in", strlen("[ERROR] invalid workflow in")) == 0 ||
957 strncmp(buf, "[ERROR] error while setting up workflow", strlen("[ERROR] error while setting up workflow")) == 0 ||
958 strncmp(buf, "[ERROR] error parsing options of", strlen("[ERROR] error parsing options of")) == 0) {
959 hasFatalImportError = true;
960 }
961 std::cout << buf << std::endl;
962 } else {
963 LOG(error) << buf;
964 }
965 }
966 if (hasFatalImportError) {
967 return false;
968 }
969 rapidjson::Reader reader;
970 s.tie(nullptr);
971 rapidjson::IStreamWrapper isw(s);
972 WorkflowImporter importer{workflow, metadata, command};
973 bool ok = reader.Parse(isw, importer);
974 if (ok == false) {
975 if (s.eof()) {
976 throw std::runtime_error("Error while parsing serialised workflow");
977 }
978 // clean up possible leftovers at the end of the input stream, e.g. [DEBUG] message from destructors
979 O2_SIGNPOST_ID_GENERATE(sid, post_workflow_importer);
980 while (true) {
981 s.getline(buf, 1024, '\n');
982 if (s.eof()) {
983 break;
984 }
985 O2_SIGNPOST_EVENT_EMIT(post_workflow_importer, sid, "post import", "Following leftover line found in input stream after parsing workflow: %{public}s", buf);
986 }
987 }
988 return true;
989}
990
991void WorkflowSerializationHelpers::dump(std::ostream& out,
992 std::vector<DataProcessorSpec> const& workflow,
993 std::vector<DataProcessorInfo> const& metadata,
994 CommandInfo const& commandInfo)
995{
996 rapidjson::OStreamWrapper osw(out);
997 rapidjson::PrettyWriter<rapidjson::OStreamWrapper> w(osw);
998
999 // handlers for serialization of InputSpec matchers
1000 auto edgeWalker = overloaded{
1001 [&w](EdgeActions::EnterNode action) {
1002 w.Key("matcher");
1003 w.StartObject();
1004 w.Key("operation");
1005 std::stringstream ss;
1006 ss << action.node->getOp();
1007 w.String(ss.str().c_str());
1008 if (action.node->getOp() == DataDescriptorMatcher::Op::Just ||
1009 action.node->getOp() == DataDescriptorMatcher::Op::Not) {
1010 return ChildAction::VisitLeft;
1011 }
1012 return ChildAction::VisitBoth;
1013 },
1014 [&w](EdgeActions::EnterLeft) {
1015 w.Key("left");
1016 w.StartObject();
1017 },
1018 [&w](EdgeActions::ExitLeft) {
1019 w.EndObject();
1020 },
1021 [&w](EdgeActions::EnterRight) {
1022 w.Key("right");
1023 w.StartObject();
1024 },
1025 [&w](EdgeActions::ExitRight) {
1026 w.EndObject();
1027 },
1028 [&w](EdgeActions::ExitNode) {
1029 w.EndObject();
1030 },
1031 [&w](auto) {}};
1032 auto leafWalker = overloaded{
1033 [&w](OriginValueMatcher const& origin) {
1034 origin.visit(overloaded{
1035 [&w](ContextRef const& ref) {
1036 w.Key("originRef");
1037 w.Uint64(ref.index);
1038 },
1039 [&w](auto const& value) {
1040 w.Key("origin");
1041 std::stringstream ss;
1042 ss << value;
1043 w.String(ss.str().c_str());
1044 }});
1045 },
1046 [&w](DescriptionValueMatcher const& description) {
1047 description.visit(overloaded{
1048 [&w](ContextRef const& ref) {
1049 w.Key("descriptionRef");
1050 w.Uint64(ref.index);
1051 },
1052 [&w](auto const& value) {
1053 w.Key("description");
1054 std::stringstream ss;
1055 ss << value;
1056 w.String(ss.str().c_str());
1057 }});
1058 },
1059 [&w](SubSpecificationTypeValueMatcher const& subspec) {
1060 subspec.visit(overloaded{
1061 [&w](ContextRef const& ref) {
1062 w.Key("subspecRef");
1063 w.Uint64(ref.index);
1064 },
1065 [&w](auto const& value) {
1066 w.Key("subspec");
1067 std::stringstream ss;
1068 ss << value;
1069 w.Uint64(std::stoul(ss.str()));
1070 }});
1071 },
1072 [&w](StartTimeValueMatcher const& startTime) {
1073 w.Key("starttime");
1074 std::stringstream ss;
1075 ss << startTime;
1076 w.String(ss.str().c_str());
1077 },
1078 [&w](ConstantValueMatcher const& constant) {},
1079 [&w](auto t) {}};
1080
1081 w.StartObject();
1082 w.Key("workflow");
1083 w.StartArray();
1084
1085 for (auto& processor : workflow) {
1086 if (processor.name.rfind("internal-dpl", 0) == 0) {
1087 continue;
1088 }
1089 w.StartObject();
1090 w.Key("name");
1091 w.String(processor.name.c_str());
1092
1093 w.Key("inputs");
1094 w.StartArray();
1095 for (auto const& input : processor.inputs) {
1099 w.StartObject();
1100 w.Key("binding");
1101 w.String(input.binding.c_str());
1102 if (auto const* concrete = std::get_if<ConcreteDataMatcher>(&input.matcher)) {
1103 w.Key("origin");
1104 w.String(concrete->origin.str, strnlen(concrete->origin.str, 4));
1105 w.Key("description");
1106 w.String(concrete->description.str, strnlen(concrete->description.str, 16));
1107 w.Key("subspec");
1108 w.Uint64(concrete->subSpec);
1109 // auto tmp = DataSpecUtils::dataDescriptorMatcherFrom(*concrete);
1110 // DataMatcherWalker::walk(tmp,
1111 // edgeWalker,
1112 // leafWalker);
1113 } else if (auto const* matcher = std::get_if<DataDescriptorMatcher>(&input.matcher)) {
1114 DataMatcherWalker::walk(*matcher,
1115 edgeWalker,
1116 leafWalker);
1117 }
1118 w.Key("lifetime");
1119 w.Uint((int)input.lifetime);
1120 if (input.metadata.empty() == false) {
1121 w.Key("metadata");
1122 w.StartArray();
1123 for (auto& metadata : input.metadata) {
1124 w.StartObject();
1125 w.Key("name");
1126 w.String(metadata.name.c_str());
1127 auto s = std::to_string(int(metadata.type));
1128 w.Key("type");
1129 w.String(s.c_str());
1130 std::ostringstream oss;
1131 oss << metadata.defaultValue;
1132 w.Key("defaultValue");
1133 w.String(oss.str().c_str());
1134 w.Key("help");
1135 w.String(metadata.help.c_str());
1136 w.EndObject();
1137 }
1138 w.EndArray();
1139 }
1140 w.EndObject();
1141 }
1142 w.EndArray();
1143
1144 w.Key("outputs");
1145 w.StartArray();
1146 for (auto& output : processor.outputs) {
1147 w.StartObject();
1148 w.Key("binding");
1149 if (output.binding.value.empty()) {
1150 auto autogenerated = DataSpecUtils::describe(output);
1151 w.String(autogenerated.c_str());
1152 } else {
1153 w.String(output.binding.value.c_str());
1154 }
1155 ConcreteDataTypeMatcher dataType = DataSpecUtils::asConcreteDataTypeMatcher(output);
1156 w.Key("origin");
1157 w.String(dataType.origin.str, strnlen(dataType.origin.str, 4));
1158 w.Key("description");
1159 w.String(dataType.description.str, strnlen(dataType.description.str, 16));
1160 // FIXME: this will have to change once we introduce wildcards for
1161 // OutputSpec
1162 auto subSpec = DataSpecUtils::getOptionalSubSpec(output);
1163 if (subSpec.has_value()) {
1164 w.Key("subspec");
1165 w.Uint64(*subSpec);
1166 }
1167 w.Key("lifetime");
1168 w.Uint((int)output.lifetime);
1169 if (output.metadata.empty() == false) {
1170 w.Key("metadata");
1171 w.StartArray();
1172 for (auto& metadata : output.metadata) {
1173 w.StartObject();
1174 w.Key("name");
1175 w.String(metadata.name.c_str());
1176 auto s = std::to_string(int(metadata.type));
1177 w.Key("type");
1178 w.String(s.c_str());
1179 std::ostringstream oss;
1180 oss << metadata.defaultValue;
1181 w.Key("defaultValue");
1182 w.String(oss.str().c_str());
1183 w.Key("help");
1184 w.String(metadata.help.c_str());
1185 w.EndObject();
1186 }
1187 w.EndArray();
1188 }
1189 w.EndObject();
1190 }
1191 w.EndArray();
1192
1193 w.Key("options");
1194 w.StartArray();
1195 for (auto& option : processor.options) {
1196 if (option.name == "start-value-enumeration" || option.name == "end-value-enumeration" || option.name == "step-value-enumeration" || option.name == "orbit-offset-enumeration" || option.name == "orbit-multiplier-enumeration") {
1197 continue;
1198 }
1199 w.StartObject();
1200 w.Key("name");
1201 w.String(option.name.c_str());
1202 auto s = std::to_string(int(option.type));
1203 w.Key("type");
1204 w.String(s.c_str());
1205 std::ostringstream oss;
1206 switch (option.type) {
1207 case VariantType::ArrayInt:
1208 case VariantType::ArrayFloat:
1209 case VariantType::ArrayDouble:
1210 case VariantType::ArrayBool:
1211 case VariantType::ArrayString:
1212 case VariantType::Array2DInt:
1213 case VariantType::Array2DFloat:
1214 case VariantType::Array2DDouble:
1215 case VariantType::LabeledArrayInt:
1216 case VariantType::LabeledArrayFloat:
1217 case VariantType::LabeledArrayDouble:
1218 case VariantType::LabeledArrayString:
1219 case VariantType::Dict:
1220 VariantJSONHelpers::write(oss, option.defaultValue);
1221 break;
1222 default:
1223 oss << option.defaultValue;
1224 break;
1225 }
1226 w.Key("defaultValue");
1227 w.String(oss.str().c_str());
1228 w.Key("help");
1229 w.String(option.help.c_str());
1230 w.Key("kind");
1231 w.String(std::to_string((int)option.kind).c_str());
1232 w.EndObject();
1233 }
1234 w.EndArray();
1235 w.Key("labels");
1236 w.StartArray();
1237 for (auto& label : processor.labels) {
1238 w.String(label.value.c_str());
1239 }
1240 w.EndArray();
1241 w.Key("metadata");
1242 w.StartArray();
1243 for (auto& metadatum : processor.metadata) {
1244 w.StartObject();
1245 w.Key("key");
1246 w.String(metadatum.key.c_str());
1247 w.Key("value");
1248 w.String(metadatum.value.c_str());
1249 w.EndObject();
1250 }
1251 w.EndArray();
1252 w.Key("rank");
1253 w.Int(processor.rank);
1254 w.Key("nSlots");
1255 w.Int(processor.nSlots);
1256 w.Key("inputTimeSliceId");
1257 w.Int(processor.inputTimeSliceId);
1258 w.Key("maxInputTimeslices");
1259 w.Int(processor.maxInputTimeslices);
1260
1261 w.EndObject();
1262 }
1263 w.EndArray();
1264
1265 w.Key("metadata");
1266 w.StartArray();
1267 for (auto& info : metadata) {
1268 w.StartObject();
1269 w.Key("name");
1270 w.String(info.name.c_str());
1271 w.Key("executable");
1272 w.String(info.executable.c_str());
1273 w.Key("cmdLineArgs");
1274 w.StartArray();
1275 for (auto& arg : info.cmdLineArgs) {
1276 w.String(arg.c_str());
1277 }
1278 w.EndArray();
1279 w.Key("workflowOptions");
1280 w.StartArray();
1281 for (auto& option : info.workflowOptions) {
1282 w.StartObject();
1283 w.Key("name");
1284 w.String(option.name.c_str());
1285 auto s = std::to_string(int(option.type));
1286 w.Key("type");
1287 w.String(s.c_str());
1288 std::ostringstream oss;
1289 oss << option.defaultValue;
1290 w.Key("defaultValue");
1291 w.String(oss.str().c_str());
1292 w.Key("help");
1293 w.String(option.help.c_str());
1294 w.EndObject();
1295 }
1296 w.EndArray();
1297 w.Key("channels");
1298 w.StartArray();
1299 for (auto& channel : info.channels) {
1300 w.String(channel.c_str());
1301 }
1302 w.EndArray();
1303 w.EndObject();
1304 }
1305 w.EndArray();
1306
1307 w.Key("command");
1308 w.String(commandInfo.command.c_str());
1309
1310 w.EndObject();
1311}
1312
1313} // namespace o2::framework
origin
header::DataOrigin origin
Definition AODReaderHelpers.cxx:34
description
header::DataDescription description
Definition AODReaderHelpers.cxx:35
state
benchmark::State & state
Definition BenchCathodeSegmentation.cxx:58
node
std::unique_ptr< expressions::Node > node
Definition ExpressionJSONHelpers.cxx:88
i
int32_t i
Definition GPUCommonAlgorithm.h:451
op
uint32_t op
Definition GPUReconstruction.cxx:56
output
void output(const std::map< std::string, ChannelStat > &channels)
Definition rawdump.cxx:197
c
uint32_t c
Definition RawData.h:2
O2_DECLARE_DYNAMIC_LOG
#define O2_DECLARE_DYNAMIC_LOG(name)
Definition Signpost.h:490
O2_SIGNPOST_END
#define O2_SIGNPOST_END(log, id, name, format,...)
Definition Signpost.h:609
O2_SIGNPOST_ID_GENERATE
#define O2_SIGNPOST_ID_GENERATE(name, log)
Definition Signpost.h:507
O2_SIGNPOST_EVENT_EMIT
#define O2_SIGNPOST_EVENT_EMIT(log, id, name, format,...)
Definition Signpost.h:523
O2_SIGNPOST_START
#define O2_SIGNPOST_START(log, id, name, format,...)
Definition Signpost.h:603
o2::framework::data_matcher::ConstantValueMatcher
Definition DataDescriptorMatcher.h:216
o2::framework::data_matcher::DataDescriptorMatcher::Op
Op
Definition DataDescriptorMatcher.h:263
o2::framework::data_matcher::DescriptionValueMatcher
Something which can be matched against a header::DataDescription.
Definition DataDescriptorMatcher.h:165
o2::framework::data_matcher::OriginValueMatcher
Something which can be matched against a header::DataOrigin.
Definition DataDescriptorMatcher.h:153
o2::framework::data_matcher::StartTimeValueMatcher
Matcher on actual time, as reported in the DataProcessingHeader.
Definition DataDescriptorMatcher.h:193
o2::framework::data_matcher::SubSpecificationTypeValueMatcher
Something which can be matched against a header::SubSpecificationType.
Definition DataDescriptorMatcher.h:177
m
const GLfloat * m
Definition glcorearb.h:4066
count
GLint GLsizei count
Definition glcorearb.h:399
result
GLuint64EXT * result
Definition glcorearb.h:5662
v
const GLdouble * v
Definition glcorearb.h:832
value
GLsizei const GLfloat * value
Definition glcorearb.h:819
length
GLuint GLsizei GLsizei * length
Definition glcorearb.h:790
label
GLuint GLsizei const GLchar * label
Definition glcorearb.h:2519
ref
GLint ref
Definition glcorearb.h:291
buf
GLenum GLuint GLenum GLsizei const GLchar * buf
Definition glcorearb.h:2514
w
GLubyte GLubyte GLubyte GLubyte w
Definition glcorearb.h:852
states
GLuint * states
Definition glcorearb.h:4932
o2::framework::data_matcher
Definition DataDescriptorMatcher.h:34
o2::framework::data_matcher::Node
std::variant< OriginValueMatcher, DescriptionValueMatcher, SubSpecificationTypeValueMatcher, std::unique_ptr< DataDescriptorMatcher >, ConstantValueMatcher, StartTimeValueMatcher > Node
Definition DataDescriptorMatcher.h:251
o2::framework
Defining ITS Vertex explicitly as messageable.
Definition Cartesian.h:288
o2::framework::Lifetime
Lifetime
Possible Lifetime of objects being exchanged by the DPL.
Definition Lifetime.h:18
o2::framework::emptyDict
Variant emptyDict()
Definition Variant.h:428
std::to_string
std::string to_string(gsl::span< T, Size > span)
Definition common.h:52
_o2_signpost_id_t
Definition Signpost.h:171
o2::framework::CommandInfo::merge
void merge(CommandInfo const &other)
Definition CommandInfo.cxx:41
o2::framework::DataSpecUtils::describe
static std::string describe(InputSpec const &spec)
Definition DataSpecUtils.cxx:72
o2::framework::DataSpecUtils::asConcreteDataTypeMatcher
static ConcreteDataTypeMatcher asConcreteDataTypeMatcher(OutputSpec const &spec)
Definition DataSpecUtils.cxx:429
o2::framework::DataSpecUtils::getOptionalSubSpec
static std::optional< header::DataHeader::SubSpecificationType > getOptionalSubSpec(OutputSpec const &spec)
Get the subspec, if available.
Definition DataSpecUtils.cxx:745
o2::framework::DataSpecUtils::optionalConcreteDataMatcherFrom
static std::optional< framework::ConcreteDataMatcher > optionalConcreteDataMatcherFrom(data_matcher::DataDescriptorMatcher const &matcher)
Definition DataSpecUtils.cxx:569
o2::framework::VariantJSONHelpers::write
static void write(std::ostream &o, Variant const &v)
Definition VariantJSONHelpers.cxx:418
o2::framework::WorkflowImporter::metadata
std::vector< DataProcessorInfo > & metadata
Definition WorkflowSerializationHelpers.cxx:911
o2::framework::WorkflowImporter::operator<<
friend std::ostream & operator<<(std::ostream &s, State state)
Definition WorkflowSerializationHelpers.cxx:102
o2::framework::WorkflowImporter::outputOptions
std::vector< ConfigParamSpec > outputOptions
Definition WorkflowSerializationHelpers.cxx:914
o2::framework::WorkflowImporter::Key
bool Key(const Ch *str, SizeType length, bool copy)
Definition WorkflowSerializationHelpers.cxx:589
o2::framework::WorkflowImporter::dataProcessors
std::vector< DataProcessorSpec > & dataProcessors
Definition WorkflowSerializationHelpers.cxx:910
o2::framework::WorkflowImporter::inputOptions
std::vector< ConfigParamSpec > inputOptions
Definition WorkflowSerializationHelpers.cxx:913
o2::framework::WorkflowImporter::String
bool String(const Ch *str, SizeType length, bool copy)
Definition WorkflowSerializationHelpers.cxx:723
o2::framework::WorkflowImporter::WorkflowImporter
WorkflowImporter(std::vector< DataProcessorSpec > &o, std::vector< DataProcessorInfo > &m, CommandInfo &c)
Definition WorkflowSerializationHelpers.cxx:283
o2::framework::WorkflowImporter::inputMatcherNodes
std::vector< data_matcher::Node > inputMatcherNodes
Definition WorkflowSerializationHelpers.cxx:929
o2::framework::WorkflowSerializationHelpers::dump
static void dump(std::ostream &o, std::vector< DataProcessorSpec > const &workflow, std::vector< DataProcessorInfo > const &metadata, CommandInfo const &commandInfo)
Definition WorkflowSerializationHelpers.cxx:991
o2::framework::WorkflowSerializationHelpers::import
static bool import(std::istream &s, std::vector< DataProcessorSpec > &workflow, std::vector< DataProcessorInfo > &metadata, CommandInfo &command)
Definition WorkflowSerializationHelpers.cxx:932
o2::framework::data_matcher::ContextRef
A typesafe reference to an element of the context.
Definition DataDescriptorMatcher.h:41
o2::framework::data_matcher::DataMatcherWalker::walk
static void walk(DataDescriptorMatcher const &top, EDGEWALKER edgeWalker, LEAFWALKER leafWalker)
Definition DataMatcherWalker.h:55
o2::framework::data_matcher::EdgeActions::EnterLeft
Definition DataMatcherWalker.h:27
o2::framework::data_matcher::EdgeActions::EnterNode
Definition DataMatcherWalker.h:21
o2::framework::data_matcher::EdgeActions::EnterRight
Definition DataMatcherWalker.h:31
o2::framework::data_matcher::EdgeActions::ExitLeft
Definition DataMatcherWalker.h:29
o2::framework::data_matcher::EdgeActions::ExitNode
Definition DataMatcherWalker.h:24
o2::framework::data_matcher::EdgeActions::ExitRight
Definition DataMatcherWalker.h:33
o2::framework::overloaded
From https://en.cppreference.com/w/cpp/utility/variant/visit.
Definition VariantHelpers.h:20
o2::header::Descriptor::runtimeInit
void runtimeInit(const char *string, short length=-1)
Definition DataHeader.h:261
LOG
LOG(info)<< "Compressed in "<< sw.CpuTime()<< " s"
str
const std::string str
Definition test_ransEncodeDecode.cxx:37