Skip to content

Commit 5f086d1

Browse files
ktfktf
committed
DPL: stop at first workflow in the pipe which fails
This will also forward error messages from one workflow to the other, so that we can properly visualise what actually failed.
1 parent 1f66135 commit 5f086d1

5 files changed

Lines changed: 57 additions & 32 deletions

File tree

‎Framework/Core/include/Framework/runDataProcessing.h‎

Lines changed: 7 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -120,9 +120,9 @@ int doMain(int argc, char** argv, o2::framework::WorkflowSpec const& specs,
120120
std::vector<o2::framework::ConfigParamSpec> const& workflowOptions,
121121
o2::framework::ConfigContext& configContext);
122122

123-
void doBoostException(boost::exception& e);
124-
void doDPLException(o2::framework::RuntimeErrorRef& ref);
125-
void doUnknownException(std::string const& s);
123+
void doBoostException(boost::exception& e, const char*);
124+
void doDPLException(o2::framework::RuntimeErrorRef& ref, char const*);
125+
void doUnknownException(std::string const& s, char const*);
126126
void doDefaultWorkflowTerminationHook();
127127

128128
template <typename T>
@@ -185,13 +185,13 @@ int main(int argc, char** argv)
185185
channelPolicies.insert(std::end(channelPolicies), std::begin(defaultChannelPolicies), std::end(defaultChannelPolicies));
186186
result = doMain(argc, argv, specs, channelPolicies, completionPolicies, dispatchPolicies, resourcePolicies, workflowOptions, configContext);
187187
} catch (boost::exception& e) {
188-
doBoostException(e);
188+
doBoostException(e, argv[0]);
189189
} catch (std::exception const& error) {
190-
doUnknownException(error.what());
190+
doUnknownException(error.what(), argv[0]);
191191
} catch (o2::framework::RuntimeErrorRef& ref) {
192-
doDPLException(ref);
192+
doDPLException(ref, argv[0]);
193193
} catch (...) {
194-
doUnknownException("");
194+
doUnknownException("", argv[0]);
195195
}
196196

197197
char* idstring = nullptr;

‎Framework/Core/src/WorkflowSerializationHelpers.cxx‎

Lines changed: 20 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -656,7 +656,7 @@ struct WorkflowImporter : public rapidjson::BaseReaderHandler<rapidjson::UTF8<>,
656656
bool inputHasDescription;
657657
};
658658

659-
void WorkflowSerializationHelpers::import(std::istream& s,
659+
bool WorkflowSerializationHelpers::import(std::istream& s,
660660
std::vector<DataProcessorSpec>& workflow,
661661
std::vector<DataProcessorInfo>& metadata,
662662
CommandInfo& command)
@@ -667,14 +667,31 @@ void WorkflowSerializationHelpers::import(std::istream& s,
667667
// FIXME: not particularly resilient, but works for now.
668668
// FIXME: this will fail if { is found at char 1024.
669669
char buf[1024];
670+
bool hasFatalImportError = false;
670671
while (s.peek() != '{') {
671672
if (s.eof()) {
672-
return;
673+
return !hasFatalImportError;
673674
}
674675
if (s.fail() || s.bad()) {
675676
throw std::runtime_error("Malformatted input workflow");
676677
}
677678
s.getline(buf, 1024, '\n');
679+
// FairLogger messages (starting with [) simply get forwarded.
680+
// Other messages we consider them as ERRORs since they
681+
// were printed out without FairLogger.
682+
if (buf[0] == '[') {
683+
if (strncmp(buf, "[ERROR] invalid workflow in", strlen("[ERROR] invalid workflow in")) == 0 ||
684+
strncmp(buf, "[ERROR] error while setting up workflow", strlen("[ERROR] error while setting up workflow")) == 0 ||
685+
strncmp(buf, "[ERROR] error parsing options of", strlen("[ERROR] error parsing options of")) == 0) {
686+
hasFatalImportError = true;
687+
}
688+
std::cout << buf << std::endl;
689+
} else {
690+
LOG(ERROR) << buf;
691+
}
692+
}
693+
if (hasFatalImportError) {
694+
return false;
678695
}
679696
rapidjson::Reader reader;
680697
rapidjson::IStreamWrapper isw(s);
@@ -683,6 +700,7 @@ void WorkflowSerializationHelpers::import(std::istream& s,
683700
if (ok == false) {
684701
throw std::runtime_error("Error while parsing serialised workflow");
685702
}
703+
return true;
686704
}
687705

688706
void WorkflowSerializationHelpers::dump(std::ostream& out,

‎Framework/Core/src/WorkflowSerializationHelpers.h‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -21,7 +21,8 @@ namespace o2::framework
2121
{
2222

2323
struct WorkflowSerializationHelpers {
24-
static void import(std::istream& s,
24+
///@return false if the previous workflow failed to generate a valid config, true otherwise
25+
static bool import(std::istream& s,
2526
std::vector<DataProcessorSpec>& workflow,
2627
std::vector<DataProcessorInfo>& metadata,
2728
CommandInfo& command);

‎Framework/Core/src/runDataProcessing.cxx‎

Lines changed: 26 additions & 20 deletions
Original file line numberDiff line numberDiff line change
@@ -900,7 +900,6 @@ void spawnDevice(DeviceRef ref,
900900
gDeviceMetricsInfos.emplace_back(DeviceMetricsInfo{});
901901
}
902902

903-
904903
struct LogProcessingState {
905904
bool didProcessLog = false;
906905
bool didProcessControl = true;
@@ -1044,27 +1043,31 @@ bool processSigChild(DeviceInfos& infos)
10441043
return hasError;
10451044
}
10461045

1047-
void doDPLException(RuntimeErrorRef& e)
1046+
void doDPLException(RuntimeErrorRef& e, char const* processName)
10481047
{
10491048
auto& err = o2::framework::error_from_ref(e);
10501049
if (err.maxBacktrace != 0) {
1051-
LOG(ERROR) << "Unhandled o2::framework::runtime_error reached the top of main, device shutting down."
1052-
<< "\n Reason: " << err.what
1053-
<< "\n Backtrace follow: \n";
1050+
LOGP(ERROR,
1051+
"Unhandled o2::framework::runtime_error reached the top of main of {}, device shutting down."
1052+
"\n Reason: "
1053+
"\n Backtrace follow: \n",
1054+
processName, err.what);
10541055
backtrace_symbols_fd(err.backtrace, err.maxBacktrace, STDERR_FILENO);
10551056
} else {
1056-
LOG(ERROR) << "Unhandled o2::framework::runtime_error reached the top of main, device shutting down."
1057-
<< "\n Reason: " << err.what
1058-
<< "\n Recompile with DPL_ENABLE_BACKTRACE=1 to get more information.";
1057+
LOGP(ERROR,
1058+
"Unhandled o2::framework::runtime_error reached the top of main of {}, device shutting down."
1059+
"\n Reason: "
1060+
"\n Recompile with DPL_ENABLE_BACKTRACE=1 to get more information.",
1061+
processName, err.what);
10591062
}
10601063
}
10611064

1062-
void doUnknownException(std::string const& s)
1065+
void doUnknownException(std::string const& s, char const* processName)
10631066
{
10641067
if (s.empty()) {
1065-
LOG(ERROR) << "Unknown error while setting up workflow.";
1068+
LOGP(ERROR, "unknown error while setting up workflow in {}.", processName);
10661069
} else {
1067-
LOG(ERROR) << "error while setting up workflow: " << s;
1070+
LOGP(ERROR, "error while setting up workflow in {}: {}", processName, s);
10681071
}
10691072
}
10701073

@@ -1483,17 +1486,17 @@ int runStateMachine(DataProcessorSpecs const& workflow,
14831486

14841487
// This should expand nodes so that we can build a consistent DAG.
14851488
} catch (std::runtime_error& e) {
1486-
std::cerr << "Invalid workflow: " << e.what() << std::endl;
1489+
LOGP(ERROR, "invalid workflow in {}: {}", driverInfo.argv[0], e.what());
14871490
return 1;
14881491
} catch (o2::framework::RuntimeErrorRef ref) {
14891492
auto& err = o2::framework::error_from_ref(ref);
14901493
#ifdef DPL_ENABLE_BACKTRACE
14911494
backtrace_symbols_fd(err.backtrace, err.maxBacktrace, STDERR_FILENO);
14921495
#endif
1493-
std::cerr << "Invalid workflow: " << err.what << std::endl;
1496+
LOGP(ERROR, "invalid workflow in {}: {}", driverInfo.argv[0], err.what);
14941497
return 1;
14951498
} catch (...) {
1496-
std::cerr << "Unknown error while materialising workflow";
1499+
LOGP(ERROR, "invalid workflow in {}: Unknown error while materialising workflow", driverInfo.argv[0]);
14971500
return 1;
14981501
}
14991502
break;
@@ -1580,7 +1583,7 @@ int runStateMachine(DataProcessorSpecs const& workflow,
15801583
driverInfo.uniqueWorkflowId);
15811584
} catch (o2::framework::RuntimeErrorRef& ref) {
15821585
auto& err = o2::framework::error_from_ref(ref);
1583-
LOG(ERROR) << "Unable to merge configurations: " << err.what;
1586+
LOGP(ERROR, "unable to merge configurations in {}: {}", driverInfo.argv[0], err.what);
15841587
#ifdef DPL_ENABLE_BACKTRACE
15851588
std::cerr << "\nStacktrace follows:\n\n";
15861589
backtrace_symbols_fd(err.backtrace, err.maxBacktrace, STDERR_FILENO);
@@ -2211,7 +2214,10 @@ int doMain(int argc, char** argv, o2::framework::WorkflowSpec const& workflow,
22112214

22122215
if (isatty(STDIN_FILENO) == false && isInputConfig()) {
22132216
std::vector<DataProcessorSpec> importedWorkflow;
2214-
WorkflowSerializationHelpers::import(std::cin, importedWorkflow, dataProcessorInfos, commandInfo);
2217+
bool previousWorked = WorkflowSerializationHelpers::import(std::cin, importedWorkflow, dataProcessorInfos, commandInfo);
2218+
if (previousWorked == false) {
2219+
exit(1);
2220+
}
22152221

22162222
size_t workflowHashB = 0;
22172223
for (auto& dp : importedWorkflow) {
@@ -2341,7 +2347,7 @@ int doMain(int argc, char** argv, o2::framework::WorkflowSpec const& workflow,
23412347
.run(),
23422348
varmap);
23432349
} catch (std::exception const& e) {
2344-
std::cerr << "Error: " << e.what() << std::endl;
2350+
LOGP(ERROR, "error parsing options of {}: {}", argv[0], e.what());
23452351
exit(1);
23462352
}
23472353
conflicting_options(varmap, "dds", "o2-control");
@@ -2416,8 +2422,8 @@ int doMain(int argc, char** argv, o2::framework::WorkflowSpec const& workflow,
24162422
frameworkId);
24172423
}
24182424

2419-
void doBoostException(boost::exception& e)
2425+
void doBoostException(boost::exception& e, char const* processName)
24202426
{
2421-
LOG(ERROR) << "error while setting up workflow: \n"
2422-
<< boost::current_exception_diagnostic_information(true);
2427+
LOGP(ERROR, "error while setting up workflow in {}: {}",
2428+
processName, boost::current_exception_diagnostic_information(true));
24232429
}

‎Framework/Core/test/benchmark_WorkflowHelpers.cxx‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -57,7 +57,7 @@ static void BM_CreateGraphOverhead(benchmark::State& state)
5757
std::vector<LogicalForwardInfo> availableForwardsInfo;
5858

5959
if (WorkflowHelpers::verifyWorkflow(workflow) != WorkflowParsingState::Valid) {
60-
throw std::runtime_error("Invalid workflow");
60+
throw std::runtime_error("invalid workflow");
6161
};
6262
auto context = makeEmptyConfigContext();
6363
WorkflowHelpers::injectServiceDevices(workflow, *context);
@@ -95,7 +95,7 @@ static void BM_CreateGraphReverseOverhead(benchmark::State& state)
9595
std::vector<LogicalForwardInfo> availableForwardsInfo;
9696

9797
if (WorkflowHelpers::verifyWorkflow(workflow) != WorkflowParsingState::Valid) {
98-
throw std::runtime_error("Invalid workflow");
98+
throw std::runtime_error("invalid workflow");
9999
};
100100
auto context = makeEmptyConfigContext();
101101
WorkflowHelpers::injectServiceDevices(workflow, *context);

0 commit comments

Comments
 (0)