Skip to content

Commit bb18dfd

Browse files
aalkinaalkin
committed
DPL Analysis: use shm metadata in CCDB tables instead of binary view
1 parent f2a59e7 commit bb18dfd

8 files changed

Lines changed: 354 additions & 117 deletions

File tree

‎Framework/CCDBSupport/src/AnalysisCCDBHelpers.cxx‎

Lines changed: 60 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -11,6 +11,7 @@
1111

1212
#include "AnalysisCCDBHelpers.h"
1313
#include "CCDBFetcherHelper.h"
14+
#include "Framework/ArrowTypes.h"
1415
#include "Framework/DataProcessingStats.h"
1516
#include "Framework/DeviceSpec.h"
1617
#include "Framework/TimingInfo.h"
@@ -22,6 +23,7 @@
2223
#include "Framework/DanglingEdgesContext.h"
2324
#include "Framework/ConfigContext.h"
2425
#include "Framework/ConfigParamsHelper.h"
26+
#include <fairmq/Version.h>
2527
#include <arrow/array/builder_binary.h>
2628
#include <arrow/type.h>
2729
#include <arrow/type_fwd.h>
@@ -109,20 +111,49 @@ AlgorithmSpec AnalysisCCDBHelpers::fetchFromCCDB(ConfigContext const& /*ctx*/)
109111
auto it = ccdbUrls.find(m.name);
110112
fieldMetadata->Append("url", it != ccdbUrls.end() ? it->second : m.defaultValue.asString());
111113
auto columnName = m.name.substr(strlen("ccdb:"));
114+
#if (FAIRMQ_VERSION_DEC >= 111000)
115+
fields.emplace_back(std::make_shared<arrow::Field>(columnName, soa::asArrowDataType<int64_t[3]>(), false, fieldMetadata));
116+
#else
112117
fields.emplace_back(std::make_shared<arrow::Field>(columnName, arrow::binary_view(), false, fieldMetadata));
118+
#endif
113119
}
114120
schemas.emplace_back(std::make_shared<arrow::Schema>(fields, schemaMetadata));
115121
}
116122

123+
#if (FAIRMQ_VERSION_DEC >= 111000)
124+
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
128+
allbuilders.resize([&schemas]() { size_t size = 0; for (auto& schema : schemas) { size += schema->num_fields(); }; return size; }());
129+
auto* pool = arrow::default_memory_pool();
130+
131+
int idx = 0;
132+
int sidx = 0;
133+
for (auto const& schema : schemas) {
134+
for (auto const& _ : schema->fields()) {
135+
#if (FAIRMQ_VERSION_DEC >= 111000)
136+
auto value_builder = std::make_shared<arrow::Int64Builder>();
137+
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
141+
++idx;
142+
}
143+
++sidx;
144+
}
145+
117146
std::shared_ptr<CCDBFetcherHelper> helper = std::make_shared<CCDBFetcherHelper>();
118147
CCDBFetcherHelper::initialiseHelper(*helper, options);
119148
std::unordered_map<std::string, int> bindings;
120149
fillValidRoutes(*helper, spec.outputs, bindings);
121150

122-
return adaptStateless([schemas, bindings, helper](InputRecord& inputs, DataTakingContext& dtc, DataAllocator& allocator, TimingInfo& timingInfo, DataProcessingStats& stats) {
151+
return adaptStateless([schemas, bindings, helper, allbuilders](InputRecord& inputs, DataTakingContext& dtc, DataAllocator& allocator, TimingInfo& timingInfo, DataProcessingStats& stats) {
123152
O2_SIGNPOST_ID_GENERATE(sid, ccdb);
124153
O2_SIGNPOST_START(ccdb, sid, "fetchFromAnalysisCCDB", "Fetching CCDB objects for analysis%" PRIu64, (uint64_t)timingInfo.timeslice);
125-
for (auto& schema : schemas) {
154+
std::ranges::for_each(allbuilders, [](auto& builder) { builder.second->Reset(); });
155+
for (auto i = 0U; i < schemas.size(); ++i) {
156+
auto& schema = schemas[i];
126157
std::vector<CCDBFetcherHelper::FetchOp> ops;
127158
auto inputBinding = *schema->metadata()->Get("sourceTable");
128159
auto inputMatcher = DataSpecUtils::fromString(*schema->metadata()->Get("sourceMatcher"));
@@ -134,6 +165,7 @@ AlgorithmSpec AnalysisCCDBHelpers::fetchFromCCDB(ConfigContext const& /*ctx*/)
134165
auto table = inputs.get<TableConsumer>(inputMatcher)->asArrowTable();
135166
// FIXME: make the fTimestamp column configurable.
136167
auto timestampColumn = table->GetColumnByName("fTimestamp");
168+
auto reserveSize = timestampColumn->length();
137169
O2_SIGNPOST_EVENT_EMIT_INFO(ccdb, sid, "fetchFromAnalysisCCDB",
138170
"There are %zu bindings available", bindings.size());
139171
for (auto& binding : bindings) {
@@ -143,9 +175,16 @@ AlgorithmSpec AnalysisCCDBHelpers::fetchFromCCDB(ConfigContext const& /*ctx*/)
143175
}
144176
int outputRouteIndex = bindings.at(outRouteDesc);
145177
auto& spec = helper->routes[outputRouteIndex].matcher;
146-
std::vector<std::shared_ptr<arrow::BinaryViewBuilder>> builders;
147-
for (auto const& _ : schema->fields()) {
148-
builders.emplace_back(std::make_shared<arrow::BinaryViewBuilder>());
178+
auto builders = allbuilders | std::views::filter([&i](auto const& builder) { return builder.first == i; });
179+
unsigned int numBuilders = std::ranges::count_if(allbuilders, [&i](auto const& builder) { return builder.first == i; });
180+
arrow::Status status;
181+
std::ranges::for_each(builders, [&status, &reserveSize](auto& builder) {
182+
if (reserveSize > builder.second->capacity()) {
183+
status &= builder.second->Reserve(reserveSize - builder.second->capacity());
184+
}
185+
});
186+
if (!status.ok()) {
187+
throw framework::runtime_error_f("Failed to reserve arrays: ", status.ToString().c_str());
149188
}
150189

151190
for (auto ci = 0; ci < timestampColumn->num_chunks(); ++ci) {
@@ -171,15 +210,25 @@ AlgorithmSpec AnalysisCCDBHelpers::fetchFromCCDB(ConfigContext const& /*ctx*/)
171210
O2_SIGNPOST_START(ccdb, sid, "handlingResponses",
172211
"Got %zu responses from server.",
173212
responses.size());
174-
if (builders.size() != responses.size()) {
175-
LOGP(fatal, "Not enough responses (expected {}, found {})", builders.size(), responses.size());
213+
if (numBuilders != responses.size()) {
214+
LOGP(fatal, "Not enough responses (expected {}, found {})", numBuilders, responses.size());
176215
}
177216
arrow::Status result;
178-
for (size_t bi = 0; bi < responses.size(); bi++) {
179-
auto& builder = builders[bi];
217+
218+
int bi = 0;
219+
for (auto& builder : builders) {
180220
auto& response = responses[bi];
221+
#if (FAIRMQ_VERSION_DEC >= 111000)
222+
result &= builder.second->Append();
223+
auto* value_builder = dynamic_cast<arrow::Int64Builder*>(builder.second->value_builder());
224+
result &= value_builder->Append(response.id.handle);
225+
result &= value_builder->Append(response.id.segment);
226+
result &= value_builder->Append(response.size);
227+
#else
181228
char const* address = reinterpret_cast<char const*>(response.id.value);
182-
result &= builder->Append(std::string_view(address, response.size));
229+
result &= builder.second->Append(std::string_view(address, response.size));
230+
#endif
231+
++bi;
183232
}
184233
if (!result.ok()) {
185234
LOGP(fatal, "Error adding results from CCDB");
@@ -188,9 +237,7 @@ AlgorithmSpec AnalysisCCDBHelpers::fetchFromCCDB(ConfigContext const& /*ctx*/)
188237
}
189238
}
190239
arrow::ArrayVector arrays;
191-
for (auto& builder : builders) {
192-
arrays.push_back(*builder->Finish());
193-
}
240+
std::ranges::for_each(builders, [&arrays](auto& builder) { arrays.push_back(*builder.second->Finish()); });
194241
auto outTable = arrow::Table::Make(schema, arrays);
195242
auto concrete = DataSpecUtils::asConcreteDataMatcher(spec);
196243
allocator.adopt(Output{concrete.origin, concrete.description, concrete.subSpec}, outTable);

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

Lines changed: 91 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -28,6 +28,7 @@
2828
#include "Framework/ArrowTableSlicingCache.h" // IWYU pragma: export
2929
#include "Framework/SliceCache.h" // IWYU pragma: export
3030
#include "Framework/VariantHelpers.h" // IWYU pragma: export
31+
#include <fairmq/Version.h>
3132
#include <arrow/array/array_binary.h>
3233
#include <arrow/table.h> // IWYU pragma: export
3334
#include <arrow/array.h> // IWYU pragma: export
@@ -40,9 +41,17 @@
4041
#include <cstring>
4142
#include <gsl/span> // IWYU pragma: export
4243

44+
namespace fair::mq::shmem
45+
{
46+
struct MetaHeader;
47+
}
48+
4349
namespace o2::framework
4450
{
4551
using ListVector = std::vector<std::vector<int64_t>>;
52+
#if (FAIRMQ_VERSION_DEC >= 111000)
53+
using PointerReconstructor = std::function<std::byte*(fair::mq::shmem::MetaHeader&&)>;
54+
#endif
4655

4756
std::string cutString(std::string&& str);
4857
std::string strToUpper(std::string&& str);
@@ -1085,6 +1094,9 @@ concept can_bind = requires(T&& t) {
10851094
template <typename... C>
10861095
concept has_index = (is_indexing_column<C> || ...);
10871096

1097+
template <typename C>
1098+
concept needs_ptr_rec = C::needs_ptr_rec;
1099+
10881100
template <typename D, typename O, typename IP, typename... C>
10891101
struct TableIterator : IP, C... {
10901102
public:
@@ -1252,6 +1264,21 @@ struct TableIterator : IP, C... {
12521264
{
12531265
doSetCurrentInternal(internal_index_columns_t{}, table);
12541266
}
1267+
#if (FAIRMQ_VERSION_DEC >= 111000)
1268+
void setPointerReconstructor(framework::PointerReconstructor const& pointerReconstructor)
1269+
{
1270+
[&pointerReconstructor, this]<typename... Cs>(framework::pack<Cs...>) {
1271+
([&pointerReconstructor, this]<typename CC>() {
1272+
if constexpr (needs_ptr_rec<CC>) {
1273+
if (pointerReconstructor) {
1274+
CC::ptrRec = &pointerReconstructor;
1275+
}
1276+
}
1277+
}.template operator()<Cs>(),
1278+
...);
1279+
}(all_columns{});
1280+
}
1281+
#endif
12551282

12561283
private:
12571284
/// Helper to move at the end of columns which actually have an iterator.
@@ -2289,7 +2316,12 @@ class Table
22892316
{
22902317
return self_t{mTable->Slice(0, 0), 0};
22912318
}
2292-
2319+
#if (FAIRMQ_VERSION_DEC >= 111000)
2320+
void setPointerReconstructor(framework::PointerReconstructor const& pointerReconstructor)
2321+
{
2322+
mBegin.setPointerReconstructor(pointerReconstructor);
2323+
}
2324+
#endif
22932325
private:
22942326
template <typename T>
22952327
arrow::ChunkedArray* lookupColumn()
@@ -2468,6 +2500,57 @@ consteval static std::string_view namespace_prefix()
24682500
}; \
24692501
[[maybe_unused]] static constexpr o2::framework::expressions::BindingNode _Getter_ { _Label_, _Name_::hash, o2::framework::expressions::selectArrowType<_Type_>() }
24702502

2503+
#if (FAIRMQ_VERSION_DEC >= 111000)
2504+
#define DECLARE_SOA_CCDB_COLUMN_FULL(_Name_, _Label_, _Getter_, _ConcreteType_, _CCDBQuery_) \
2505+
struct _Name_ : o2::soa::Column<int64_t[3], _Name_> { \
2506+
static constexpr const char* mLabel = _Label_; \
2507+
static constexpr const char* query = _CCDBQuery_; \
2508+
static constexpr const uint32_t hash = crc32(namespace_prefix<_Name_>(), std::string_view{#_Getter_}); \
2509+
static constexpr bool needs_ptr_rec = true; \
2510+
std::function<std::byte*(fair::mq::shmem::MetaHeader&&)> const* ptrRec = nullptr; \
2511+
using base = o2::soa::Column<int64_t[3], _Name_>; \
2512+
using type = int64_t[3]; \
2513+
using column_t = _Name_; \
2514+
_Name_(arrow::ChunkedArray const* column) \
2515+
: o2::soa::Column<int64_t[3], _Name_>(o2::soa::ColumnIterator<int64_t[3]>(column)) \
2516+
{ \
2517+
} \
2518+
\
2519+
_Name_() = default; \
2520+
_Name_(_Name_ const& other) = default; \
2521+
_Name_& operator=(_Name_ const& other) = default; \
2522+
\
2523+
decltype(auto) _Getter_() const \
2524+
{ \
2525+
auto& [handle, segment, size] = *mColumnIterator; \
2526+
auto span = std::span<std::byte>{(*ptrRec)(fair::mq::shmem::MetaHeader{ \
2527+
static_cast<size_t>(size), \
2528+
0, handle, 0, 0, \
2529+
static_cast<uint16_t>(segment), true}), \
2530+
static_cast<size_t>(size)}; \
2531+
if constexpr (std::same_as<_ConcreteType_, std::span<std::byte>>) { \
2532+
return span; \
2533+
} else { \
2534+
static std::byte* payload = nullptr; \
2535+
static _ConcreteType_* deserialised = nullptr; \
2536+
static TClass* c = TClass::GetClass(#_ConcreteType_); \
2537+
if (payload != (std::byte*)span.data()) { \
2538+
payload = (std::byte*)span.data(); \
2539+
delete deserialised; \
2540+
TBufferFile f(TBufferFile::EMode::kRead, span.size(), (char*)span.data(), kFALSE); \
2541+
deserialised = (_ConcreteType_*)soa::extractCCDBPayload((char*)payload, span.size(), c, "ccdb_object"); \
2542+
} \
2543+
return *deserialised; \
2544+
} \
2545+
} \
2546+
\
2547+
decltype(auto) \
2548+
get() const \
2549+
{ \
2550+
return _Getter_(); \
2551+
} \
2552+
};
2553+
#else
24712554
#define DECLARE_SOA_CCDB_COLUMN_FULL(_Name_, _Label_, _Getter_, _ConcreteType_, _CCDBQuery_) \
24722555
struct _Name_ : o2::soa::Column<std::span<std::byte>, _Name_> { \
24732556
static constexpr const char* mLabel = _Label_; \
@@ -2510,6 +2593,7 @@ consteval static std::string_view namespace_prefix()
25102593
return _Getter_(); \
25112594
} \
25122595
};
2596+
#endif
25132597

25142598
#define DECLARE_SOA_CCDB_COLUMN(_Name_, _Getter_, _ConcreteType_, _CCDBQuery_) \
25152599
DECLARE_SOA_CCDB_COLUMN_FULL(_Name_, "f" #_Name_, _Getter_, _ConcreteType_, _CCDBQuery_)
@@ -3853,7 +3937,12 @@ class FilteredBase : public T
38533937
{
38543938
return mCached;
38553939
}
3856-
3940+
#if (FAIRMQ_VERSION_DEC >= 111000)
3941+
void setPointerReconstructor(framework::PointerReconstructor const& pointerReconstructor)
3942+
{
3943+
mFilteredBegin.setPointerReconstructor(pointerReconstructor);
3944+
}
3945+
#endif
38573946
private:
38583947
void resetRanges()
38593948
{

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

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -26,6 +26,11 @@
2626
#include "SimulationDataFormat/MCGenProperties.h"
2727
#include "Framework/PID.h"
2828

29+
#include <fairmq/Version.h>
30+
#if (FAIRMQ_VERSION_DEC >= 111000)
31+
#include <fairmq/shmem/Common.h>
32+
#endif
33+
2934
namespace o2
3035
{
3136
namespace aod

0 commit comments

Comments
 (0)