Skip to content

DPL Analysis: allow finalising callback also for CCDB columns #15732

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

Open
ktf wants to merge 1 commit into
base: dev
Choose a base branch
from
+71 −48
Open

DPL Analysis: allow finalising callback also for CCDB columns #15732

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
119 changes: 71 additions & 48 deletions Framework/Core/include/Framework/ASoA.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2370,59 +2370,82 @@ consteval static std::string_view namespace_prefix()
}; \
[[maybe_unused]] static constexpr o2::framework::expressions::BindingNode _Getter_ { _Label_, _Name_::hash, o2::framework::expressions::selectArrowType<_Type_>() }

#define DECLARE_SOA_CCDB_COLUMN_FULL(_Name_, _Label_, _Getter_, _ConcreteType_, _CCDBQuery_) \
struct _Name_ : o2::soa::Column<int64_t[3], _Name_> { \
static constexpr const char* mLabel = _Label_; \
static constexpr const char* query = _CCDBQuery_; \
static constexpr const uint32_t hash = crc32(namespace_prefix<_Name_>(), std::string_view{#_Getter_}); \
static constexpr bool needs_ptr_rec = true; \
std::function<std::byte*(fair::mq::shmem::MetaHeader&&)> const* ptrRec = nullptr; \
using base = o2::soa::Column<int64_t[3], _Name_>; \
using type = int64_t[3]; \
using column_t = _Name_; \
_Name_(arrow::ChunkedArray const* column) \
: o2::soa::Column<int64_t[3], _Name_>(o2::soa::ColumnIterator<int64_t[3]>(column)) \
{ \
} \
\
_Name_() = default; \
_Name_(_Name_ const& other) = default; \
_Name_& operator=(_Name_ const& other) = default; \
\
decltype(auto) _Getter_() const \
{ \
auto& [handle, segment, size] = *mColumnIterator; \
auto span = std::span<std::byte>{(*ptrRec)(fair::mq::shmem::MetaHeader{ \
static_cast<size_t>(size), \
0, handle, 0, 0, \
static_cast<uint16_t>(segment), true}), \
static_cast<size_t>(size)}; \
if constexpr (std::same_as<_ConcreteType_, std::span<std::byte>>) { \
return span; \
} else { \
static std::byte* payload = nullptr; \
static _ConcreteType_* deserialised = nullptr; \
static TClass* c = TClass::GetClass(#_ConcreteType_); \
if (payload != (std::byte*)span.data()) { \
payload = (std::byte*)span.data(); \
delete deserialised; \
TBufferFile f(TBufferFile::EMode::kRead, span.size(), (char*)span.data(), kFALSE); \
deserialised = (_ConcreteType_*)soa::extractCCDBPayload((char*)payload, span.size(), c, "ccdb_object"); \
} \
return *deserialised; \
} \
} \
\
decltype(auto) \
get() const \
{ \
return _Getter_(); \
} \
#define DECLARE_SOA_CCDB_COLUMN_FULL_FINALISED(_Name_, _Label_, _Getter_, _ConcreteType_, _CCDBQuery_, ...) \
struct _Name_ : o2::soa::Column<int64_t[3], _Name_> { \
static constexpr const char* mLabel = _Label_; \
static constexpr const char* query = _CCDBQuery_; \
static constexpr const uint32_t hash = crc32(namespace_prefix<_Name_>(), std::string_view{#_Getter_}); \
static constexpr bool needs_ptr_rec = true; \
/* Post-deserialisation fixup for objects which are not usable straight out of the ROOT */ \
/* streamer, e.g. FlatObjects whose internal pointers must be rectified first. Runs on */ \
/* the receiving device, once per (re)deserialisation, before the object is ever handed */ \
/* out. Returns the object to cache: a finaliser returning a different instance owns */ \
/* disposing of the one it was given. */ \
using finaliser_t = _ConcreteType_* (*)(_ConcreteType_*); \
static constexpr finaliser_t finalise = __VA_ARGS__; \
std::function<std::byte*(fair::mq::shmem::MetaHeader&&)> const* ptrRec = nullptr; \
using base = o2::soa::Column<int64_t[3], _Name_>; \
using type = int64_t[3]; \
using column_t = _Name_; \
_Name_(arrow::ChunkedArray const* column) \
: o2::soa::Column<int64_t[3], _Name_>(o2::soa::ColumnIterator<int64_t[3]>(column)) \
{ \
} \
\
_Name_() = default; \
_Name_(_Name_ const& other) = default; \
_Name_& operator=(_Name_ const& other) = default; \
\
decltype(auto) _Getter_() const \
{ \
auto& [handle, segment, size] = *mColumnIterator; \
auto span = std::span<std::byte>{(*ptrRec)(fair::mq::shmem::MetaHeader{ \
static_cast<size_t>(size), \
0, handle, 0, 0, \
static_cast<uint16_t>(segment), true}), \
static_cast<size_t>(size)}; \
if constexpr (std::same_as<_ConcreteType_, std::span<std::byte>>) { \
return span; \
} else { \
static std::byte* payload = nullptr; \
static _ConcreteType_* deserialised = nullptr; \
static TClass* c = TClass::GetClass(#_ConcreteType_); \
if (payload != (std::byte*)span.data()) { \
payload = (std::byte*)span.data(); \
delete deserialised; \
TBufferFile f(TBufferFile::EMode::kRead, span.size(), (char*)span.data(), kFALSE); \
auto* streamed = (_ConcreteType_*)soa::extractCCDBPayload((char*)payload, span.size(), c, "ccdb_object"); \
if (!streamed) { \
LOGP(fatal, \
"Could not deserialise a {} from the CCDB payload for {} ({} bytes). Check the configured " \
"path (option \"ccdb:{}\") and that the object exists for this timestamp.", \
#_ConcreteType_, _CCDBQuery_, span.size(), _Label_); \
} \
deserialised = finalise(streamed); \
} \
return *deserialised; \
} \
} \
\
decltype(auto) \
get() const \
{ \
return _Getter_(); \
} \
};

#define DECLARE_SOA_CCDB_COLUMN_FULL(_Name_, _Label_, _Getter_, _ConcreteType_, _CCDBQuery_) \
DECLARE_SOA_CCDB_COLUMN_FULL_FINALISED(_Name_, _Label_, _Getter_, _ConcreteType_, _CCDBQuery_, \
[](_ConcreteType_* ccdbObject) { return ccdbObject; })

#define DECLARE_SOA_CCDB_COLUMN(_Name_, _Getter_, _ConcreteType_, _CCDBQuery_) \
DECLARE_SOA_CCDB_COLUMN_FULL(_Name_, "f" #_Name_, _Getter_, _ConcreteType_, _CCDBQuery_)

/* As DECLARE_SOA_CCDB_COLUMN, plus a finalisation function run on the freshly deserialised
object. The finaliser must be the last argument so that commas in a lambda body are absorbed. */
#define DECLARE_SOA_CCDB_COLUMN_FINALISED(_Name_, _Getter_, _ConcreteType_, _CCDBQuery_, ...) \
DECLARE_SOA_CCDB_COLUMN_FULL_FINALISED(_Name_, "f" #_Name_, _Getter_, _ConcreteType_, _CCDBQuery_, __VA_ARGS__)

#define DECLARE_SOA_COLUMN(_Name_, _Getter_, _Type_) \
DECLARE_SOA_COLUMN_FULL(_Name_, _Getter_, _Type_, "f" #_Name_)

Expand Down