diff --git a/Framework/Core/CMakeLists.txt b/Framework/Core/CMakeLists.txt index a958d9fa57fca..8cbbf639e300a 100644 --- a/Framework/Core/CMakeLists.txt +++ b/Framework/Core/CMakeLists.txt @@ -76,6 +76,7 @@ o2_add_library(Framework src/ConfigurationOptionsRetriever.cxx src/FreePortFinder.cxx src/GraphvizHelpers.cxx + src/MermaidHelpers.cxx src/HTTPParser.cxx src/InputRecord.cxx src/InputRouteHelpers.cxx @@ -208,6 +209,7 @@ foreach(t InputSpec Kernels LogParsingHelpers + Mermaid OptionsHelpers OverrideLabels PtrHelpers diff --git a/Framework/Core/src/MermaidHelpers.cxx b/Framework/Core/src/MermaidHelpers.cxx new file mode 100644 index 0000000000000..2acef99180386 --- /dev/null +++ b/Framework/Core/src/MermaidHelpers.cxx @@ -0,0 +1,51 @@ +// Copyright 2019-2020 CERN and copyright holders of ALICE O2. +// See https://alice-o2.web.cern.ch/copyright for details of the copyright holders. +// All rights not expressly granted are reserved. +// +// This software is distributed under the terms of the GNU General Public +// License v3 (GPL Version 3), copied verbatim in the file "COPYING". +// +// In applying this license CERN does not waive the privileges and immunities +// granted to it by virtue of its status as an Intergovernmental Organization +// or submit itself to any jurisdiction. + +#include "MermaidHelpers.h" +#include +#include +#include + +namespace o2 +{ +namespace framework +{ + +namespace +{ +std::string quote(std::string const& s) { return R"(")" + s + R"(")"; } +} // namespace + +/// Helper to dump a set of devices as a mermaid file +void MermaidHelpers::dumpDeviceSpec2Mermaid(std::ostream& out, const std::vector& specs) +{ + out << "flowchart TD\n"; + std::map outputChannel2Device; + std::map outputChannel2Port; + + for (auto& spec : specs) { + auto id = spec.id; + out << " " << id << "\n"; + for (auto&& output : spec.outputChannels) { + outputChannel2Device.insert(std::make_pair(output.name, id)); + outputChannel2Port.insert(std::make_pair(output.name, output.port)); + } + } + for (auto& spec : specs) { + for (auto& input : spec.inputChannels) { + auto outputName = input.name; + out << " " << outputChannel2Device[outputName] << "-- " << input.port << ":" << outputName << " -->" << spec.id << "\n"; + } + } +} + +} // namespace framework +} // namespace o2 diff --git a/Framework/Core/src/MermaidHelpers.h b/Framework/Core/src/MermaidHelpers.h new file mode 100644 index 0000000000000..08e7ff249a4e9 --- /dev/null +++ b/Framework/Core/src/MermaidHelpers.h @@ -0,0 +1,32 @@ +// Copyright 2019-2020 CERN and copyright holders of ALICE O2. +// See https://alice-o2.web.cern.ch/copyright for details of the copyright holders. +// All rights not expressly granted are reserved. +// +// This software is distributed under the terms of the GNU General Public +// License v3 (GPL Version 3), copied verbatim in the file "COPYING". +// +// In applying this license CERN does not waive the privileges and immunities +// granted to it by virtue of its status as an Intergovernmental Organization +// or submit itself to any jurisdiction. +#ifndef FRAMEWORK_MERMAIDHELPERS_H +#define FRAMEWORK_MERMAIDHELPERS_H + +#include "Framework/WorkflowSpec.h" +#include "Framework/DeviceSpec.h" +#include +#include + +namespace o2 +{ +namespace framework +{ + +struct MermaidHelpers { + using Devices = std::vector; + static void dumpDataProcessorSpec2Mermaid(std::ostream&, const WorkflowSpec& specs); + static void dumpDeviceSpec2Mermaid(std::ostream&, const Devices& specs); +}; + +} // namespace framework +} // namespace o2 +#endif // FRAMEWORK_MERMAIDHELPERS_H diff --git a/Framework/Core/src/runDataProcessing.cxx b/Framework/Core/src/runDataProcessing.cxx index 67dbd66615f81..05b5bc4c397fd 100644 --- a/Framework/Core/src/runDataProcessing.cxx +++ b/Framework/Core/src/runDataProcessing.cxx @@ -64,6 +64,7 @@ #include "O2ControlHelpers.h" #include "DeviceSpecHelpers.h" #include "GraphvizHelpers.h" +#include "MermaidHelpers.h" #include "PropertyTreeHelpers.h" #include "SimpleResourceManager.h" #include "WorkflowSerializationHelpers.h" @@ -2172,7 +2173,23 @@ void initialiseDriverControl(bpo::variables_map const& varmap, control.state = DriverControlState::PLAY; } - if (varmap["graphviz"].as()) { + if (varmap["mermaid"].as()) { + // Dump a mermaid representation of what I will do. + control.callbacks = {[](WorkflowSpec const&, + DeviceSpecs const& specs, + DeviceExecutions const&, + DataProcessorInfos&, + CommandInfo const&) { + MermaidHelpers::dumpDeviceSpec2Mermaid(std::cout, specs); + }}; + control.forcedTransitions = { + DriverState::EXIT, // + DriverState::PERFORM_CALLBACKS, // + DriverState::MERGE_CONFIGS, // + DriverState::IMPORT_CURRENT_WORKFLOW, // + DriverState::MATERIALISE_WORKFLOW // + }; + } else if (varmap["graphviz"].as()) { // Dump a graphviz representation of what I will do. control.callbacks = {[](WorkflowSpec const&, DeviceSpecs const& specs, @@ -2387,7 +2404,8 @@ int doMain(int argc, char** argv, o2::framework::WorkflowSpec const& workflow, "what to do when a device has an error: quit, wait") // // ("min-failure-level", bpo::value(&minFailureLevel)->default_value(LogParsingHelpers::LogLevel::Fatal), // // "minimum message level which will be considered as fatal and exit with 1") // // - ("graphviz,g", bpo::value()->zero_tokens()->default_value(false), "produce graph output") // // + ("graphviz,g", bpo::value()->zero_tokens()->default_value(false), "produce graphviz output") // // + ("mermaid", bpo::value()->zero_tokens()->default_value(false), "produce graph output in mermaid format") // // ("timeout,t", bpo::value()->default_value(0), "forced exit timeout (in seconds)") // // ("dds,D", bpo::value()->zero_tokens()->default_value(false), "create DDS configuration") // // ("dds-workflow-suffix,D", bpo::value()->default_value(""), "suffix for DDS names") // // @@ -2601,12 +2619,16 @@ int doMain(int argc, char** argv, o2::framework::WorkflowSpec const& workflow, conflicting_options(varmap, "dds", "dump-workflow"); conflicting_options(varmap, "dds", "run"); conflicting_options(varmap, "dds", "graphviz"); + conflicting_options(varmap, "dds", "mermaid"); conflicting_options(varmap, "o2-control", "dump-workflow"); conflicting_options(varmap, "o2-control", "run"); conflicting_options(varmap, "o2-control", "graphviz"); + conflicting_options(varmap, "o2-control", "mermaid"); conflicting_options(varmap, "run", "dump-workflow"); conflicting_options(varmap, "run", "graphviz"); + conflicting_options(varmap, "run", "mermaid"); conflicting_options(varmap, "dump-workflow", "graphviz"); + conflicting_options(varmap, "dump-workflow", "mermaid"); conflicting_options(varmap, "no-batch", "batch"); if (varmap.count("help")) { diff --git a/Framework/Core/test/test_Mermaid.cxx b/Framework/Core/test/test_Mermaid.cxx new file mode 100644 index 0000000000000..c692b0be76359 --- /dev/null +++ b/Framework/Core/test/test_Mermaid.cxx @@ -0,0 +1,146 @@ +// Copyright 2019-2020 CERN and copyright holders of ALICE O2. +// See https://alice-o2.web.cern.ch/copyright for details of the copyright holders. +// All rights not expressly granted are reserved. +// +// This software is distributed under the terms of the GNU General Public +// License v3 (GPL Version 3), copied verbatim in the file "COPYING". +// +// In applying this license CERN does not waive the privileges and immunities +// granted to it by virtue of its status as an Intergovernmental Organization +// or submit itself to any jurisdiction. +#define BOOST_TEST_MODULE Test Framework MermaidHelpers +#define BOOST_TEST_MAIN +#define BOOST_TEST_DYN_LINK + +#include "Mocking.h" +#include "../src/ComputingResourceHelpers.h" +#include "../src/DeviceSpecHelpers.h" +#include "../src/MermaidHelpers.h" +#include "../src/SimpleResourceManager.h" +#include "Framework/DeviceSpec.h" +#include "Framework/WorkflowSpec.h" +#include "Headers/DataHeader.h" + +#include +#include +#include + +using namespace o2::framework; + +// because comparing the whole thing is a pain. +void lineByLineComparison(const std::string& as, const std::string& bs) +{ + std::istringstream a(as); + std::istringstream b(bs); + + char bufferA[1024]; + char bufferB[1024]; + while (a.good() && b.good()) { + a.getline(bufferA, 1024); + b.getline(bufferB, 1024); + BOOST_CHECK_EQUAL(std::string(bufferA), std::string(bufferB)); + } + BOOST_CHECK(a.eof()); + BOOST_CHECK(b.eof()); +} + +// This is how you can define your processing in a declarative way +WorkflowSpec defineDataProcessing() +{ + return {{"A", Inputs{}, + Outputs{OutputSpec{"TST", "A1"}, + OutputSpec{"TST", "A2"}}}, + {"B", + {InputSpec{"x", "TST", "A1"}}, + Outputs{OutputSpec{"TST", "B1"}}}, + {"C", Inputs{InputSpec{"x", "TST", "A2"}}, + Outputs{OutputSpec{"TST", "C1"}}}, + {"D", + Inputs{InputSpec{"i1", "TST", "B1"}, + InputSpec{"i2", "TST", "C1"}}, + Outputs{}}}; +} + +WorkflowSpec defineDataProcessing2() +{ + return { + {"A", + {}, + { + OutputSpec{"TST", "A"}, + }}, + timePipeline({"B", + {InputSpec{"a", "TST", "A"}}, + {OutputSpec{"TST", "B"}}}, + 3), + timePipeline({"C", + {InputSpec{"b", "TST", "B"}}, + {OutputSpec{"TST", "C"}}}, + 2), + }; +} + +BOOST_AUTO_TEST_CASE(TestMermaid) +{ + auto workflow = defineDataProcessing(); + std::ostringstream str; + std::vector devices; + for (auto& device : devices) { + BOOST_CHECK(device.id != ""); + } + auto configContext = makeEmptyConfigContext(); + auto channelPolicies = ChannelConfigurationPolicy::createDefaultPolicies(*configContext); + auto completionPolicies = CompletionPolicy::createDefaultPolicies(); + auto callbacksPolicies = CallbacksPolicy::createDefaultPolicies(); + std::vector resources = {ComputingResourceHelpers::getLocalhostResource()}; + SimpleResourceManager rm(resources); + DeviceSpecHelpers::dataProcessorSpecs2DeviceSpecs(workflow, channelPolicies, completionPolicies, callbacksPolicies, devices, rm, "workflow-id", *configContext); + str.str(""); + MermaidHelpers::dumpDeviceSpec2Mermaid(str, devices); + lineByLineComparison(str.str(), R"EXPECTED(flowchart TD + A + B + C + D + A-- 22000:from_A_to_B -->B + A-- 22001:from_A_to_C -->C + B-- 22002:from_B_to_D -->D + C-- 22003:from_C_to_D -->D +)EXPECTED"); +} + +BOOST_AUTO_TEST_CASE(TestMermaidWithPipeline) +{ + auto workflow = defineDataProcessing2(); + std::ostringstream str; + std::vector devices; + for (auto& device : devices) { + BOOST_CHECK(device.id != ""); + } + auto configContext = makeEmptyConfigContext(); + auto channelPolicies = ChannelConfigurationPolicy::createDefaultPolicies(*configContext); + auto completionPolicies = CompletionPolicy::createDefaultPolicies(); + auto callbacksPolicies = CallbacksPolicy::createDefaultPolicies(); + std::vector resources = {ComputingResourceHelpers::getLocalhostResource()}; + SimpleResourceManager rm(resources); + DeviceSpecHelpers::dataProcessorSpecs2DeviceSpecs(workflow, channelPolicies, completionPolicies, callbacksPolicies, devices, rm, "workflow-id", *configContext); + str.str(""); + MermaidHelpers::dumpDeviceSpec2Mermaid(str, devices); + lineByLineComparison(str.str(), R"EXPECTED(flowchart TD + A + B_t0 + B_t1 + B_t2 + C_t0 + C_t1 + A-- 22000:from_A_to_B_t0 -->B_t0 + A-- 22001:from_A_to_B_t1 -->B_t1 + A-- 22002:from_A_to_B_t2 -->B_t2 + B_t0-- 22003:from_B_t0_to_C_t0 -->C_t0 + B_t1-- 22005:from_B_t1_to_C_t0 -->C_t0 + B_t2-- 22007:from_B_t2_to_C_t0 -->C_t0 + B_t0-- 22004:from_B_t0_to_C_t1 -->C_t1 + B_t1-- 22006:from_B_t1_to_C_t1 -->C_t1 + B_t2-- 22008:from_B_t2_to_C_t1 -->C_t1 +)EXPECTED"); +}