Skip to content

[EMCAL-507] Remove custom end-of-stream handling via EMCALBlockHeader #7761

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
shahor02 merged 1 commit into from
Dec 1, 2021
Merged

[EMCAL-507] Remove custom end-of-stream handling via EMCALBlockHeader #7761

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
9 changes: 9 additions & 0 deletions Detectors/EMCAL/workflow/include/EMCALWorkflow/RawToCellConverterSpec.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -103,6 +103,15 @@ class RawToCellConverterSpec : public framework::Task
int mHWAddressHG; ///< HW address of HG (for monitoring)
};
bool isLostTimeframe(framework::ProcessingContext& ctx) const;

/// \brief Send data to output channels
/// \param cells Container with output cells for timeframe
/// \param triggers Container with trigger records for timeframe
/// \param decodingErrors Container with decoding errors for timeframe
///
/// Send data to all output channels for the given subspecification. The subspecification
/// is determined on the fly in the run method and therefore used as parameter. Consumers
/// must use wildcard subspecification via ConcreteDataTypeMatcher.
void sendData(framework::ProcessingContext& ctx, const std::vector<o2::emcal::Cell>& cells, const std::vector<o2::emcal::TriggerRecord>& triggers, const std::vector<ErrorTypeFEE>& decodingErrors) const;

header::DataHeader::SubSpecificationType mSubspecification = 0; ///< Subspecification for output channels
Expand Down
9 changes: 0 additions & 9 deletions Detectors/EMCAL/workflow/src/AnalysisClusterSpec.cxx
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -14,7 +14,6 @@

#include "DataFormatsEMCAL/Digit.h"
#include "DataFormatsEMCAL/Cluster.h"
#include "DataFormatsEMCAL/EMCALBlockHeader.h"
#include "DataFormatsEMCAL/TriggerRecord.h"
#include "EMCALWorkflow/AnalysisClusterSpec.h"
#include "Framework/ControlService.h"
Expand Down Expand Up @@ -73,14 +72,6 @@ void AnalysisClusterSpec<InputType>::run(framework::ProcessingContext& ctx)
TrigName = "cellstrgr";
}

auto dataref = ctx.inputs().get(inputname.c_str());
auto const* emcheader = o2::framework::DataRefUtils::getHeader<o2::emcal::EMCALBlockHeader*>(dataref);
if (!emcheader->mHasPayload) {
LOG(debug) << "[EMCALClusterizer - run] No more cells/digits" << std::endl;
ctx.services().get<o2::framework::ControlService>().readyToQuit(framework::QuitRequest::Me);
return;
}

auto Inputs = ctx.inputs().get<gsl::span<InputType>>(inputname.c_str());
LOG(debug) << "[EMCALClusterizer - run] Received " << Inputs.size() << " Cells/digits, running clusterizer ...";

Expand Down
1 change: 0 additions & 1 deletion Detectors/EMCAL/workflow/src/CellConverterSpec.cxx
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -12,7 +12,6 @@
#include "FairLogger.h"

#include "DataFormatsEMCAL/Digit.h"
#include "DataFormatsEMCAL/EMCALBlockHeader.h"
#include "EMCALWorkflow/CellConverterSpec.h"
#include "Framework/ControlService.h"
#include "Framework/ConfigParamRegistry.h"
Expand Down
1 change: 0 additions & 1 deletion Detectors/EMCAL/workflow/src/ClusterizerSpec.cxx
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -14,7 +14,6 @@

#include "DataFormatsEMCAL/Digit.h"
#include "DataFormatsEMCAL/Cluster.h"
#include "DataFormatsEMCAL/EMCALBlockHeader.h"
#include "DataFormatsEMCAL/TriggerRecord.h"
#include "EMCALWorkflow/ClusterizerSpec.h"
#include "Framework/ControlService.h"
Expand Down
1 change: 0 additions & 1 deletion Detectors/EMCAL/workflow/src/DigitsPrinterSpec.cxx
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,7 +18,6 @@

#include "Framework/ControlService.h"
#include "Framework/DataRefUtils.h"
#include "DataFormatsEMCAL/EMCALBlockHeader.h"
#include "DataFormatsEMCAL/Cell.h"
#include "DataFormatsEMCAL/Digit.h"
#include "DataFormatsEMCAL/TriggerRecord.h"
Expand Down
4 changes: 1 addition & 3 deletions Detectors/EMCAL/workflow/src/PublisherSpec.cxx
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -10,7 +10,6 @@
// or submit itself to any jurisdiction.

#include "DetectorsCommonDataFormats/NameConf.h"
#include "DataFormatsEMCAL/EMCALBlockHeader.h"
#include "EMCALWorkflow/PublisherSpec.h"
#include "Framework/ConfigParamRegistry.h"
#include "Framework/ControlService.h"
Expand Down Expand Up @@ -67,9 +66,8 @@ o2::framework::DataProcessorSpec createPublisherSpec(PublisherConf const& config
}

auto publish = [&processAttributes, &pc, propagateMC]() {
o2::emcal::EMCALBlockHeader emcheader(true);
if (processAttributes->reader->next()) {
(*processAttributes->reader)(pc, emcheader);
(*processAttributes->reader)(pc);
} else {
processAttributes->reader.reset();
return false;
Expand Down
5 changes: 2 additions & 3 deletions Detectors/EMCAL/workflow/src/RawToCellConverterSpec.cxx
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,7 +21,6 @@
#include "Framework/InputRecordWalker.h"
#include "Framework/Logger.h"
#include "Framework/WorkflowSpec.h"
#include "DataFormatsEMCAL/EMCALBlockHeader.h"
#include "DataFormatsEMCAL/Constants.h"
#include "DataFormatsEMCAL/TriggerRecord.h"
#include "DataFormatsEMCAL/ErrorTypeFEE.h"
Expand Down Expand Up @@ -499,12 +498,12 @@ void RawToCellConverterSpec::run(framework::ProcessingContext& ctx)
ncellsEvent++;
mOutputCells.push_back(cell.mCellData);
}
LOG(debug) << "Accepted " << ncellsEvent << " cells";
LOG(debug) << "Next event [Orbit " << bc.orbit << ", BC (" << bc.bc << "]: Accepted " << ncellsEvent << " cells";
}
mOutputTriggerRecords.emplace_back(bc, triggerBuffer[bc], eventstart, ncellsEvent);
}

LOG(debug) << "[EMCALRawToCellConverter - run] Writing " << mOutputCells.size() << " cells ...";
LOG(info) << "[EMCALRawToCellConverter - run] Writing " << mOutputCells.size() << " cells from " << mOutputTriggerRecords.size() << " events ...";
sendData(ctx, mOutputCells, mOutputTriggerRecords, mOutputDecoderErrors);
}

Expand Down
46 changes: 10 additions & 36 deletions Detectors/EMCAL/workflow/src/RecoWorkflow.cxx
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,7 +17,6 @@

#include "Algorithm/RangeTokenizer.h"
#include "DPLUtils/MakeRootTreeWriterSpec.h"
#include "DataFormatsEMCAL/EMCALBlockHeader.h"
#include "DataFormatsEMCAL/Cell.h"
#include "DataFormatsEMCAL/Digit.h"
#include "DataFormatsEMCAL/Cluster.h"
Expand Down Expand Up @@ -187,75 +186,50 @@ o2::framework::WorkflowSpec getWorkflow(bool propagateMC,
specs.emplace_back(o2::emcal::reco_workflow::getAnalysisClusterSpec(inputType == InputType::Digits));
}

// check if the process is ready to quit
// this is decided upon the meta information in the EMCAL block header, the operation is set
// value kNoPayload in case of no data or no operation
// see also PublisherSpec.cxx
// in this workflow, the EOD is sent after the last real data, and all inputs will receive EOD,
// so it is enough to check on the first occurence
// FIXME: this will be changed once DPL can propagate control events like EOD
auto checkReady = [](o2::framework::DataRef const& ref) {
auto const* emcalheader = o2::framework::DataRefUtils::getHeader<o2::emcal::EMCALBlockHeader*>(ref);
// sector number -1 indicates end-of-data
if (emcalheader != nullptr) {
// indicate normal processing if not ready and skip if ready
if (!emcalheader->mHasPayload) {
return std::make_tuple(o2::framework::MakeRootTreeWriterSpec::TerminationCondition::Action::SkipProcessing, true);
}
}
return std::make_tuple(o2::framework::MakeRootTreeWriterSpec::TerminationCondition::Action::DoProcessing, false);
};

auto makeWriterSpec = [propagateMC, checkReady](const char* processName, const char* defaultFileName, const char* defaultTreeName,
auto&& databranch, auto&& triggerbranch, auto&& mcbranch) {
auto makeWriterSpec = [propagateMC](const char* processName, const char* defaultFileName, const char* defaultTreeName,
auto&& databranch, auto&& triggerbranch, auto&& mcbranch) {
// depending on the MC propagation flag, the RootTreeWriter spec is created with two
// or one branch definition
if (propagateMC) {
return std::move(o2::framework::MakeRootTreeWriterSpec(processName, defaultFileName, defaultTreeName,
o2::framework::MakeRootTreeWriterSpec::TerminationCondition{checkReady},
std::move(databranch),
std::move(triggerbranch),
std::move(mcbranch)));
}
return std::move(o2::framework::MakeRootTreeWriterSpec(processName, defaultFileName, defaultTreeName,
o2::framework::MakeRootTreeWriterSpec::TerminationCondition{checkReady},
std::move(databranch),
std::move(triggerbranch)));
};

// TODO: Write comment in push comment @matthiasrichter
auto makeWriterSpec_Cluster = [checkReady](const char* processName, const char* defaultFileName, const char* defaultTreeName,
auto&& clusterbranch, auto&& digitindicesbranch, auto&& clustertriggerbranch, auto&& indicestriggerbranch) {
auto makeWriterSpec_Cluster = [](const char* processName, const char* defaultFileName, const char* defaultTreeName,
auto&& clusterbranch, auto&& digitindicesbranch, auto&& clustertriggerbranch, auto&& indicestriggerbranch) {
// RootTreeWriter spec is created with one branch definition
return std::move(o2::framework::MakeRootTreeWriterSpec(processName, defaultFileName, defaultTreeName,
o2::framework::MakeRootTreeWriterSpec::TerminationCondition{checkReady},
std::move(clusterbranch),
std::move(digitindicesbranch),
std::move(clustertriggerbranch),
std::move(indicestriggerbranch)));
};

auto makeWriterSpec_AnalysisCluster = [checkReady](const char* processName, const char* defaultFileName, const char* defaultTreeName,
auto&& analysisclusterbranch) {
auto makeWriterSpec_AnalysisCluster = [](const char* processName, const char* defaultFileName, const char* defaultTreeName,
auto&& analysisclusterbranch) {
// RootTreeWriter spec is created with one branch definition
return std::move(o2::framework::MakeRootTreeWriterSpec(processName, defaultFileName, defaultTreeName,
o2::framework::MakeRootTreeWriterSpec::TerminationCondition{checkReady},
std::move(analysisclusterbranch)));
};

auto makeWriterSpec_CellsTR = [disableDecodingErrors, checkReady](const char* processName, const char* defaultFileName, const char* defaultTreeName,
auto&& CellsBranch, auto&& TriggerRecordBranch, auto&& DecoderErrorsBranch) {
auto makeWriterSpec_CellsTR = [disableDecodingErrors](const char* processName, const char* defaultFileName, const char* defaultTreeName,
auto&& CellsBranch, auto&& TriggerRecordBranch, auto&& DecoderErrorsBranch) {
return std::move(o2::framework::MakeRootTreeWriterSpec(processName, defaultFileName, defaultTreeName,
o2::framework::MakeRootTreeWriterSpec::TerminationCondition{checkReady},
std::move(CellsBranch),
std::move(TriggerRecordBranch),
std::move(DecoderErrorsBranch)));
};

auto makeWriterSpec_CellsTR_noerrors = [checkReady](const char* processName, const char* defaultFileName, const char* defaultTreeName,
auto&& CellsBranch, auto&& TriggerRecordBranch) {
auto makeWriterSpec_CellsTR_noerrors = [](const char* processName, const char* defaultFileName, const char* defaultTreeName,
auto&& CellsBranch, auto&& TriggerRecordBranch) {
return std::move(o2::framework::MakeRootTreeWriterSpec(processName, defaultFileName, defaultTreeName,
o2::framework::MakeRootTreeWriterSpec::TerminationCondition{checkReady},
std::move(CellsBranch),
std::move(TriggerRecordBranch)));
};
Expand Down