Skip to content

Commit e9ccaaf

Browse files
aalkinaalkin
committed
DPL Analysis: remove compatibility with older FairMQ
1 parent 1183f52 commit e9ccaaf

7 files changed

Lines changed: 9 additions & 110 deletions

File tree

‎Framework/CCDBSupport/src/AnalysisCCDBHelpers.cxx‎

Lines changed: 0 additions & 17 deletions
Original file line numberDiff line numberDiff line change
@@ -111,33 +111,21 @@ AlgorithmSpec AnalysisCCDBHelpers::fetchFromCCDB(ConfigContext const& /*ctx*/)
111111
auto it = ccdbUrls.find(m.name);
112112
fieldMetadata->Append("url", it != ccdbUrls.end() ? it->second : m.defaultValue.asString());
113113
auto columnName = m.name.substr(strlen("ccdb:"));
114-
#if (FAIRMQ_VERSION_DEC >= 111000)
115114
fields.emplace_back(std::make_shared<arrow::Field>(columnName, soa::asArrowDataType<int64_t[3]>(), false, fieldMetadata));
116-
#else
117-
fields.emplace_back(std::make_shared<arrow::Field>(columnName, arrow::binary_view(), false, fieldMetadata));
118-
#endif
119115
}
120116
schemas.emplace_back(std::make_shared<arrow::Schema>(fields, schemaMetadata));
121117
}
122118

123-
#if (FAIRMQ_VERSION_DEC >= 111000)
124119
std::vector<std::pair<uint32_t, std::shared_ptr<arrow::FixedSizeListBuilder>>> allbuilders;
125-
#else
126-
std::vector<std::pair<uint32_t, std::shared_ptr<arrow::BinaryViewBuilder>>> allbuilders;
127-
#endif
128120
allbuilders.resize([&schemas]() { size_t size = 0; for (auto& schema : schemas) { size += schema->num_fields(); }; return size; }());
129121
auto* pool = arrow::default_memory_pool();
130122

131123
int idx = 0;
132124
int sidx = 0;
133125
for (auto const& schema : schemas) {
134126
for (auto const& _ : schema->fields()) {
135-
#if (FAIRMQ_VERSION_DEC >= 111000)
136127
auto value_builder = std::make_shared<arrow::Int64Builder>();
137128
allbuilders[idx] = std::make_pair(sidx, std::make_shared<arrow::FixedSizeListBuilder>(pool, std::move(value_builder), 3));
138-
#else
139-
allbuilders[idx] = std::make_pair(sidx, std::make_shared<arrow::BinaryViewBuilder>());
140-
#endif
141129
++idx;
142130
}
143131
++sidx;
@@ -227,16 +215,11 @@ AlgorithmSpec AnalysisCCDBHelpers::fetchFromCCDB(ConfigContext const& /*ctx*/)
227215
lastId.value = response.id.value;
228216
allocator.adoptFromCache(output, response.id, header::gSerializationMethodCCDB);
229217
}
230-
#if (FAIRMQ_VERSION_DEC >= 111000)
231218
result &= builder.second->Append();
232219
auto* value_builder = dynamic_cast<arrow::Int64Builder*>(builder.second->value_builder());
233220
result &= value_builder->Append(response.id.handle);
234221
result &= value_builder->Append(response.id.segment);
235222
result &= value_builder->Append(response.size);
236-
#else
237-
char const* address = reinterpret_cast<char const*>(response.id.value);
238-
result &= builder.second->Append(std::string_view(address, response.size));
239-
#endif
240223
++bi;
241224
}
242225
if (!result.ok()) {

‎Framework/Core/include/Framework/ASoA.h‎

Lines changed: 5 additions & 53 deletions
Original file line numberDiff line numberDiff line change
@@ -49,9 +49,7 @@ struct MetaHeader;
4949
namespace o2::framework
5050
{
5151
using ListVector = std::vector<std::vector<int64_t>>;
52-
#if (FAIRMQ_VERSION_DEC >= 111000)
5352
using PointerReconstructor = std::function<std::byte*(fair::mq::shmem::MetaHeader&&)>;
54-
#endif
5553

5654
std::string cutString(std::string&& str);
5755
std::string strToUpper(std::string&& str);
@@ -1196,7 +1194,7 @@ struct TableIterator : IP, C... {
11961194
{
11971195
doSetCurrentInternal(internal_index_columns_t{}, table);
11981196
}
1199-
#if (FAIRMQ_VERSION_DEC >= 111000)
1197+
12001198
void setPointerReconstructor(framework::PointerReconstructor const& pointerReconstructor)
12011199
{
12021200
[&pointerReconstructor, this]<typename... Cs>(framework::pack<Cs...>) {
@@ -1210,7 +1208,6 @@ struct TableIterator : IP, C... {
12101208
...);
12111209
}(all_columns{});
12121210
}
1213-
#endif
12141211

12151212
private:
12161213
/// Helper to move at the end of columns which actually have an iterator.
@@ -2185,12 +2182,12 @@ class Table
21852182
{
21862183
return self_t{mArrowTableRef.makeEmpty()};
21872184
}
2188-
#if (FAIRMQ_VERSION_DEC >= 111000)
2185+
21892186
void setPointerReconstructor(framework::PointerReconstructor const& pointerReconstructor)
21902187
{
21912188
mBegin.setPointerReconstructor(pointerReconstructor);
21922189
}
2193-
#endif
2190+
21942191
private:
21952192
template <typename T>
21962193
arrow::ChunkedArray* lookupColumn()
@@ -2372,7 +2369,6 @@ consteval static std::string_view namespace_prefix()
23722369
}; \
23732370
[[maybe_unused]] static constexpr o2::framework::expressions::BindingNode _Getter_ { _Label_, _Name_::hash, o2::framework::expressions::selectArrowType<_Type_>() }
23742371

2375-
#if (FAIRMQ_VERSION_DEC >= 111000)
23762372
#define DECLARE_SOA_CCDB_COLUMN_FULL(_Name_, _Label_, _Getter_, _ConcreteType_, _CCDBQuery_) \
23772373
struct _Name_ : o2::soa::Column<int64_t[3], _Name_> { \
23782374
static constexpr const char* mLabel = _Label_; \
@@ -2422,50 +2418,6 @@ consteval static std::string_view namespace_prefix()
24222418
return _Getter_(); \
24232419
} \
24242420
};
2425-
#else
2426-
#define DECLARE_SOA_CCDB_COLUMN_FULL(_Name_, _Label_, _Getter_, _ConcreteType_, _CCDBQuery_) \
2427-
struct _Name_ : o2::soa::Column<std::span<std::byte>, _Name_> { \
2428-
static constexpr const char* mLabel = _Label_; \
2429-
static constexpr const char* query = _CCDBQuery_; \
2430-
static constexpr const uint32_t hash = crc32(namespace_prefix<_Name_>(), std::string_view{#_Getter_}); \
2431-
using base = o2::soa::Column<std::span<std::byte>, _Name_>; \
2432-
using type = std::span<std::byte>; \
2433-
using column_t = _Name_; \
2434-
_Name_(arrow::ChunkedArray const* column) \
2435-
: o2::soa::Column<std::span<std::byte>, _Name_>(o2::soa::ColumnIterator<std::span<std::byte>>(column)) \
2436-
{ \
2437-
} \
2438-
\
2439-
_Name_() = default; \
2440-
_Name_(_Name_ const& other) = default; \
2441-
_Name_& operator=(_Name_ const& other) = default; \
2442-
\
2443-
decltype(auto) _Getter_() const \
2444-
{ \
2445-
if constexpr (std::same_as<_ConcreteType_, std::span<std::byte>>) { \
2446-
return *mColumnIterator; \
2447-
} else { \
2448-
static std::byte* payload = nullptr; \
2449-
static _ConcreteType_* deserialised = nullptr; \
2450-
static TClass* c = TClass::GetClass(#_ConcreteType_); \
2451-
auto span = *mColumnIterator; \
2452-
if (payload != (std::byte*)span.data()) { \
2453-
payload = (std::byte*)span.data(); \
2454-
delete deserialised; \
2455-
TBufferFile f(TBufferFile::EMode::kRead, span.size(), (char*)span.data(), kFALSE); \
2456-
deserialised = (_ConcreteType_*)soa::extractCCDBPayload((char*)payload, span.size(), c, "ccdb_object"); \
2457-
} \
2458-
return *deserialised; \
2459-
} \
2460-
} \
2461-
\
2462-
decltype(auto) \
2463-
get() const \
2464-
{ \
2465-
return _Getter_(); \
2466-
} \
2467-
};
2468-
#endif
24692421

24702422
#define DECLARE_SOA_CCDB_COLUMN(_Name_, _Getter_, _ConcreteType_, _CCDBQuery_) \
24712423
DECLARE_SOA_CCDB_COLUMN_FULL(_Name_, "f" #_Name_, _Getter_, _ConcreteType_, _CCDBQuery_)
@@ -3778,12 +3730,12 @@ class FilteredBase : public T
37783730
{
37793731
return mCached;
37803732
}
3781-
#if (FAIRMQ_VERSION_DEC >= 111000)
3733+
37823734
void setPointerReconstructor(framework::PointerReconstructor const& pointerReconstructor)
37833735
{
37843736
mFilteredBegin.setPointerReconstructor(pointerReconstructor);
37853737
}
3786-
#endif
3738+
37873739
private:
37883740
void resetRanges()
37893741
{

‎Framework/Core/include/Framework/AnalysisDataModel.h‎

Lines changed: 0 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -27,9 +27,7 @@
2727
#include "Framework/PID.h"
2828

2929
#include <fairmq/Version.h>
30-
#if (FAIRMQ_VERSION_DEC >= 111000)
3130
#include <fairmq/shmem/Common.h>
32-
#endif
3331

3432
namespace o2
3533
{

‎Framework/Core/include/Framework/AnalysisTask.h‎

Lines changed: 3 additions & 26 deletions
Original file line numberDiff line numberDiff line change
@@ -304,19 +304,15 @@ struct AnalysisDataProcessorBuilder {
304304
}
305305

306306
template <typename Task, is_table_iterator_or_enumeration Grouping, std::ranges::input_range R, soa::is_table... Associated>
307-
#if (FAIRMQ_VERSION_DEC >= 111000)
308307
static void invokeProcess(Task& task, InputRecord& inputs, R matchers, PointerReconstructor const& pointerReconstructor, void (Task::*processingFunction)(Grouping, Associated...), std::vector<ExpressionInfo>& infos, ArrowTableSlicingCache& slices, header::DataOrigin newOrigin = header::DataOrigin{"AOD"})
309-
#else
310-
static void invokeProcess(Task& task, InputRecord& inputs, R matchers, void (Task::*processingFunction)(Grouping, Associated...), std::vector<ExpressionInfo>& infos, ArrowTableSlicingCache& slices, header::DataOrigin newOrigin = header::DataOrigin{"AOD"})
311-
#endif
312308
{
313309
using G = std::decay_t<Grouping>;
314310
auto groupingTable = AnalysisDataProcessorBuilder::bindGroupingTable(inputs, matchers, processingFunction, infos);
315-
#if (FAIRMQ_VERSION_DEC >= 111000)
311+
316312
if constexpr (!is_enumeration<G>) {
317313
groupingTable.setPointerReconstructor(pointerReconstructor);
318314
}
319-
#endif
315+
320316
constexpr const int numElements = homogeneous_apply_refs_size<false, std::decay_t<Task>>();
321317

322318
// set filtered tables for partitions with grouping
@@ -357,12 +353,10 @@ struct AnalysisDataProcessorBuilder {
357353
...);
358354
},
359355
associatedTables);
360-
#if (FAIRMQ_VERSION_DEC >= 111000)
361356
std::apply([&pointerReconstructor](auto&... table) {
362357
(table.setPointerReconstructor(pointerReconstructor), ...);
363358
},
364359
associatedTables);
365-
#endif
366360

367361
auto binder = [&task, &groupingTable, &associatedTables](auto& x) mutable {
368362
x.bindExternalIndices(&groupingTable, &std::get<std::decay_t<Associated>>(associatedTables)...);
@@ -393,12 +387,11 @@ struct AnalysisDataProcessorBuilder {
393387
auto slicer = GroupSlicer(groupingTable, associatedTables, slices, newOrigin);
394388
for (auto& slice : slicer) {
395389
auto associatedSlices = slice.associatedTables();
396-
#if (FAIRMQ_VERSION_DEC >= 111000)
397390
std::apply([&pointerReconstructor](auto&... table) {
398391
(table.setPointerReconstructor(pointerReconstructor), ...);
399392
},
400393
associatedSlices);
401-
#endif
394+
402395
overwriteInternalIndices(associatedSlices, associatedTables);
403396
std::apply(
404397
[&binder](auto&... x) mutable {
@@ -667,7 +660,6 @@ DataProcessorSpec adaptAnalysisTask(ConfigContext const& ctx, Args&&... args)
667660
ic.services().get<ArrowTableSlicingCacheDef>().setCaches(std::move(bindingsKeys));
668661
ic.services().get<ArrowTableSlicingCacheDef>().setCachesUnsorted(std::move(bindingsKeysUnsorted));
669662
ic.services().get<ArrowTableSlicingCacheDef>().setOrigin(newOrigin);
670-
#if (FAIRMQ_VERSION_DEC >= 111000)
671663
PointerReconstructor pointerReconstructor(nullptr);
672664
bool hasCCDBTables = !ic.services().get<DanglingEdgesContext>().requestedTIMs.empty();
673665

@@ -677,9 +669,6 @@ DataProcessorSpec adaptAnalysisTask(ConfigContext const& ctx, Args&&... args)
677669
auto& spec = pc.services().get<DanglingEdgesContext>().requestedTIMs.front();
678670
pointerReconstructor = proxy.getShmPointerReconstructor(spec, 0);
679671
}
680-
#else
681-
return [task, expressionInfos, inputInfos, newOrigin](ProcessingContext& pc) mutable {
682-
#endif
683672
// load the ccdb object from their cache
684673
homogeneous_apply_refs_sized<numElements>([&pc](auto& element) { return analysis_task_parsers::newDataframeCondition(pc.inputs(), element); }, *task.get());
685674
// reset partitions once per dataframe
@@ -704,28 +693,16 @@ DataProcessorSpec adaptAnalysisTask(ConfigContext const& ctx, Args&&... args)
704693
if constexpr (requires { &T::process; }) {
705694
auto loc = std::ranges::find_if(inputInfos, [](auto const& info) { return info.hash == o2::framework::TypeIdHelpers::uniqueId<decltype(&T::process)>(); });
706695
auto matchers = loc == inputInfos.end() ? std::vector<std::pair<int, ConcreteDataMatcher>>{} : loc->matchers;
707-
#if (FAIRMQ_VERSION_DEC >= 111000)
708696
AnalysisDataProcessorBuilder::invokeProcess(*(task.get()), pc.inputs(), matchers, pointerReconstructor, &T::process, expressionInfos, slices, newOrigin);
709-
#else
710-
AnalysisDataProcessorBuilder::invokeProcess(*(task.get()), pc.inputs(), matchers, &T::process, expressionInfos, slices, newOrigin);
711-
#endif
712697
}
713698
// execute optional process()
714699
homogeneous_apply_refs_sized<numElements>(
715-
#if (FAIRMQ_VERSION_DEC >= 111000)
716700
[&pc, &expressionInfos, &task, &slices, &inputInfos, &newOrigin, &pointerReconstructor](auto& x) {
717-
#else
718-
[&pc, &expressionInfos, &task, &slices, &inputInfos, &newOrigin](auto& x) {
719-
#endif
720701
if constexpr (is_process_configurable<decltype(x)>) {
721702
if (x.value == true) {
722703
auto loc = std::ranges::find_if(inputInfos, [](auto const& info) { return info.hash == o2::framework::TypeIdHelpers::uniqueId<decltype(x.process)>(); });
723704
auto matchers = loc == inputInfos.end() ? std::vector<std::pair<int, ConcreteDataMatcher>>{} : loc->matchers;
724-
#if (FAIRMQ_VERSION_DEC >= 111000)
725705
AnalysisDataProcessorBuilder::invokeProcess(*task.get(), pc.inputs(), matchers, pointerReconstructor, x.process, expressionInfos, slices, newOrigin);
726-
#else
727-
AnalysisDataProcessorBuilder::invokeProcess(*task.get(), pc.inputs(), matchers, x.process, expressionInfos, slices, newOrigin);
728-
#endif
729706
return true;
730707
}
731708
return false;

‎Framework/Core/include/Framework/DataAllocator.h‎

Lines changed: 0 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -43,9 +43,7 @@
4343
// Do not change this for a full inclusion of fair::mq::Device.
4444
#include <fairmq/FwdDecls.h>
4545
#include <fairmq/Version.h>
46-
#if (FAIRMQ_VERSION_DEC >= 111000)
4746
#include <fairmq/shmem/Message.h>
48-
#endif
4947

5048
namespace arrow
5149
{
@@ -635,11 +633,9 @@ DataAllocator::CacheId DataAllocator::adoptContainer(const Output& spec, Contain
635633
// clone is indistinguishable from the original, we can keep sending
636634
// the original.
637635
cacheId.value = context.addToCache(payloadMessage);
638-
#if (FAIRMQ_VERSION_DEC >= 111000)
639636
auto meta = dynamic_cast<fair::mq::shmem::Message*>(payloadMessage.get())->GetMeta();
640637
cacheId.handle = meta.fHandle;
641638
cacheId.segment = meta.fSegmentId;
642-
#endif
643639
}
644640

645641
context.add<MessageContext::TrivialObject>(std::move(headerMessage), std::move(payloadMessage), routeIndex);

‎Framework/Core/include/Framework/FairMQDeviceProxy.h‎

Lines changed: 1 addition & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -21,9 +21,7 @@
2121
#include "Framework/ForwardRoute.h"
2222
#include <fairmq/FwdDecls.h>
2323
#include <fairmq/Version.h>
24-
#if (FAIRMQ_VERSION_DEC >= 111000)
2524
#include <fairmq/shmem/Common.h>
26-
#endif
2725
#include <vector>
2826

2927
namespace o2::header
@@ -33,9 +31,8 @@ struct DataHeader;
3331

3432
namespace o2::framework
3533
{
36-
#if (FAIRMQ_VERSION_DEC >= 111000)
3734
using PointerReconstructor = std::function<std::byte*(fair::mq::shmem::MetaHeader&&)>;
38-
#endif
35+
3936
/// Helper class to hide fair::mq::Device headers in the DataAllocator header.
4037
/// This is done because fair::mq::Device brings in a bunch of boost.mpl /
4138
/// boost.fusion stuff, slowing down compilation times enourmously.
@@ -65,10 +62,8 @@ class FairMQDeviceProxy
6562
[[nodiscard]] ChannelIndex getForwardChannelIndexByName(std::string const& channelName) const;
6663
/// Retrieve the channel index from a given OutputSpec and the associated timeslice
6764
[[nodiscard]] ChannelIndex getOutputChannelIndex(OutputSpec const& spec, size_t timeslice) const;
68-
#if (FAIRMQ_VERSION_DEC >= 111000)
6965
/// Retrieve the pointer-reconstruction function for the shm manager for a given input spec
7066
[[nodiscard]] PointerReconstructor getShmPointerReconstructor(InputSpec const& spec, size_t timeslice);
71-
#endif
7267
/// Retrieve the channel index from a given OutputSpec and the associated timeslice
7368
void getMatchingForwardChannelIndexes(std::vector<ChannelIndex>& result, header::DataHeader const& header, size_t timeslice) const;
7469
/// ChannelIndex from a RouteIndex

‎Framework/Core/src/FairMQDeviceProxy.cxx‎

Lines changed: 0 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -365,7 +365,6 @@ void FairMQDeviceProxy::bind(std::vector<OutputRoute> const& outputs, std::vecto
365365
mStateChangeCallback = newStatePending;
366366
}
367367

368-
#if (FAIRMQ_VERSION_DEC >= 111000)
369368
PointerReconstructor FairMQDeviceProxy::getShmPointerReconstructor(InputSpec const& spec, size_t timeslice)
370369
{
371370
assert(mInputRoutes.size() == mInputs.size());
@@ -384,5 +383,4 @@ PointerReconstructor FairMQDeviceProxy::getShmPointerReconstructor(InputSpec con
384383
}
385384
return {};
386385
}
387-
#endif
388386
} // namespace o2::framework

0 commit comments

Comments
 (0)