Skip to content

[k2] add tl::RpcDestActorFlags type #1676

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
apolyakov merged 1 commit into from
Aug 13, 2026
Merged

[k2] add tl::RpcDestActorFlags type #1676

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
74 changes: 74 additions & 0 deletions runtime-light/tl/tl-types.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -193,6 +193,80 @@ bool rpcInvokeReqExtra::fetch(tl::fetcher& tlf, const tl::mask& flags) noexcept
return ok;
}

void rpcInvokeReqExtra::store(tl::storer& tls, const tl::mask& flags) const noexcept { // NOLINT(readability-function-cognitive-complexity)
if (static_cast<bool>(flags.value & WAIT_BINLOG_POS_FLAG)) {
opt_wait_binlog_pos.has_value() ? (*opt_wait_binlog_pos).store(tls) : tl::i64{}.store(tls);
}
if (static_cast<bool>(flags.value & STRING_FORWARD_KEYS_FLAG)) {
opt_string_forward_keys.has_value() ? (*opt_string_forward_keys).store(tls) : tl::vector<tl::string>{}.store(tls);
}
if (static_cast<bool>(flags.value & INT_FORWARD_KEYS_FLAG)) {
opt_int_forward_keys.has_value() ? (*opt_int_forward_keys).store(tls) : tl::vector<tl::i64>{}.store(tls);
}
if (static_cast<bool>(flags.value & STRING_FORWARD_FLAG)) {
opt_string_forward.has_value() ? (*opt_string_forward).store(tls) : tl::string{}.store(tls);
}
if (static_cast<bool>(flags.value & INT_FORWARD_FLAG)) {
opt_int_forward.has_value() ? (*opt_int_forward).store(tls) : tl::i64{}.store(tls);
}
if (static_cast<bool>(flags.value & CUSTOM_TIMEOUT_MS_FLAG)) {
opt_custom_timeout_ms.has_value() ? (*opt_custom_timeout_ms).store(tls) : tl::i32{}.store(tls);
}
if (static_cast<bool>(flags.value & SUPPORTED_COMPRESSION_VERSION_FLAG)) {
opt_supported_compression_version.has_value() ? (*opt_supported_compression_version).store(tls) : tl::i32{}.store(tls);
}
if (static_cast<bool>(flags.value & RANDOM_DELAY_FLAG)) {
opt_random_delay.has_value() ? (*opt_random_delay).store(tls) : tl::f64{}.store(tls);
}
if (static_cast<bool>(flags.value & PERSISTENT_QUERY_FLAG)) {
opt_persistent_query.has_value() ? (*opt_persistent_query).store(tls) : tl::exactlyOnce::PersistentRequest{}.store(tls);
}
if (static_cast<bool>(flags.value & TRACE_CONTEXT_FLAG)) {
opt_trace_context.has_value() ? (*opt_trace_context).store(tls) : tl::tracing::traceContext{}.store(tls);
}
if (static_cast<bool>(flags.value & EXECUTION_CONTEXT_FLAG)) {
opt_execution_context.has_value() ? (*opt_execution_context).store(tls) : tl::string{}.store(tls);
}
}

size_t rpcInvokeReqExtra::footprint(const tl::mask& flags) const noexcept { // NOLINT(readability-function-cognitive-complexity)
size_t footprint{};
if (static_cast<bool>(flags.value & WAIT_BINLOG_POS_FLAG)) {
footprint += opt_wait_binlog_pos.has_value() ? (*opt_wait_binlog_pos).footprint() : tl::i64{}.footprint();
}
if (static_cast<bool>(flags.value & STRING_FORWARD_KEYS_FLAG)) {
footprint += opt_string_forward_keys.has_value() ? (*opt_string_forward_keys).footprint() : tl::vector<tl::string>{}.footprint();
}
if (static_cast<bool>(flags.value & INT_FORWARD_KEYS_FLAG)) {
footprint += opt_int_forward_keys.has_value() ? (*opt_int_forward_keys).footprint() : tl::vector<tl::i64>{}.footprint();
}
if (static_cast<bool>(flags.value & STRING_FORWARD_FLAG)) {
footprint += opt_string_forward.has_value() ? (*opt_string_forward).footprint() : tl::string{}.footprint();
}
if (static_cast<bool>(flags.value & INT_FORWARD_FLAG)) {
footprint += opt_int_forward.has_value() ? (*opt_int_forward).footprint() : tl::i64{}.footprint();
}
if (static_cast<bool>(flags.value & CUSTOM_TIMEOUT_MS_FLAG)) {
footprint += opt_custom_timeout_ms.has_value() ? (*opt_custom_timeout_ms).footprint() : tl::i32{}.footprint();
}
if (static_cast<bool>(flags.value & SUPPORTED_COMPRESSION_VERSION_FLAG)) {
footprint += opt_supported_compression_version.has_value() ? (*opt_supported_compression_version).footprint() : tl::i32{}.footprint();
}
if (static_cast<bool>(flags.value & RANDOM_DELAY_FLAG)) {
footprint += opt_random_delay.has_value() ? (*opt_random_delay).footprint() : tl::f64{}.footprint();
}
if (static_cast<bool>(flags.value & PERSISTENT_QUERY_FLAG)) {
footprint += opt_persistent_query.has_value() ? (*opt_persistent_query).footprint() : tl::exactlyOnce::PersistentRequest{}.footprint();
}
if (static_cast<bool>(flags.value & TRACE_CONTEXT_FLAG)) {
footprint += opt_trace_context.has_value() ? (*opt_trace_context).footprint() : tl::tracing::traceContext{}.footprint();
}
if (static_cast<bool>(flags.value & EXECUTION_CONTEXT_FLAG)) {
footprint += opt_execution_context.has_value() ? (*opt_execution_context).footprint() : tl::string{}.footprint();
}
return footprint;
}

tl::mask rpcInvokeReqExtra::get_flags() const noexcept {
tl::mask flags{.value = static_cast<tl::mask::underlying_type>(return_binlog_pos)};

Expand Down
110 changes: 110 additions & 0 deletions runtime-light/tl/tl-types.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1090,6 +1090,14 @@ struct uuid final {
bool fetch(tl::fetcher& tlf) noexcept {
return lo.fetch(tlf) && hi.fetch(tlf);
}

void store(tl::storer& tls) const noexcept {
lo.store(tls), hi.store(tls);
}

constexpr size_t footprint() const noexcept {
return lo.footprint() + hi.footprint();
}
};

struct prepareRequest final {
Expand All @@ -1098,6 +1106,14 @@ struct prepareRequest final {
bool fetch(tl::fetcher& tlf) noexcept {
return persistent_query_uuid.fetch(tlf);
}

void store(tl::storer& tls) const noexcept {
persistent_query_uuid.store(tls);
}

constexpr size_t footprint() const noexcept {
return persistent_query_uuid.footprint();
}
};

struct commitRequest final {
Expand All @@ -1107,6 +1123,14 @@ struct commitRequest final {
bool fetch(tl::fetcher& tlf) noexcept {
return persistent_query_uuid.fetch(tlf) && persistent_slot_uuid.fetch(tlf);
}

void store(tl::storer& tls) const noexcept {
persistent_query_uuid.store(tls), persistent_slot_uuid.store(tls);
}

constexpr size_t footprint() const noexcept {
return persistent_query_uuid.footprint() + persistent_slot_uuid.footprint();
}
};

class PersistentRequest final {
Expand All @@ -1131,6 +1155,24 @@ class PersistentRequest final {
}
return false;
}

void store(tl::storer& tls) const noexcept {
std::visit(
[&tls](const auto& request) noexcept {
using request_t = std::remove_cvref_t<decltype(request)>;
if constexpr (std::same_as<request_t, tl::exactlyOnce::prepareRequest>) {
tl::magic{.value = PREPARE_REQUEST_MAGIC}.store(tls);
} else if constexpr (std::same_as<request_t, tl::exactlyOnce::commitRequest>) {
tl::magic{.value = COMMIT_REQUEST_MAGIC}.store(tls);
}
request.store(tls);
},
request);
}

constexpr size_t footprint() const noexcept {
return std::visit([](const auto& request) noexcept { return tl::magic{}.footprint() + request.footprint(); }, request);
}
};
} // namespace exactlyOnce

Expand All @@ -1143,6 +1185,14 @@ struct traceID final {
bool fetch(tl::fetcher& tlf) noexcept {
return lo.fetch(tlf) && hi.fetch(tlf);
}

void store(tl::storer& tls) const noexcept {
lo.store(tls), hi.store(tls);
}

constexpr size_t footprint() const noexcept {
return lo.footprint() + hi.footprint();
}
};

class traceContext final {
Expand Down Expand Up @@ -1195,6 +1245,28 @@ class traceContext final {
return ok;
}

void store(tl::storer& tls) const noexcept {
get_flags().store(tls);
trace_id.store(tls);
if (opt_parent_id.has_value()) {
opt_parent_id->store(tls);
}
if (opt_source_id.has_value()) {
opt_source_id->store(tls);
}
}

constexpr size_t footprint() const noexcept {
size_t footprint{tl::mask{}.footprint() + trace_id.footprint()};
if (opt_parent_id.has_value()) {
footprint += opt_parent_id->footprint();
}
if (opt_source_id.has_value()) {
footprint += opt_source_id->footprint();
}
return footprint;
}

tl::mask get_flags() const noexcept {
tl::mask flags{.value = static_cast<tl::mask::underlying_type>(reserved_status_0)};
flags.value |= static_cast<tl::mask::underlying_type>(reserved_status_1) << 1;
Expand Down Expand Up @@ -1254,6 +1326,8 @@ class rpcInvokeReqExtra final {
bool return_view_number{};

bool fetch(tl::fetcher& tlf, const tl::mask& flags) noexcept;
void store(tl::storer& tls, const tl::mask& flags) const noexcept;
size_t footprint(const tl::mask& flags) const noexcept;

tl::mask get_flags() const noexcept;
};
Expand Down Expand Up @@ -1333,6 +1407,42 @@ struct rpcReqError final {
}
};

template<std::default_initializable T>
struct rpcDestActorFlags final {
tl::i64 actor_id{};
tl::mask flags{};
tl::rpcInvokeReqExtra extra{};
T query{};

void store(tl::storer& tls) const noexcept
requires tl::serializable<T>
{
actor_id.store(tls), flags.store(tls), extra.store(tls, flags), query.store(tls);
}

constexpr size_t footprint() const noexcept
requires tl::footprintable<T>
{
return actor_id.footprint() + flags.footprint() + extra.footprint(flags) + query.footprint();
}
};

template<typename T>
class RpcDestActorFlags final {
static constexpr tl::magic MAGIC{.value = TL_RPC_DEST_ACTOR_FLAGS};

public:
tl::rpcDestActorFlags<T> inner{};

void store(tl::storer& tls) const noexcept {
MAGIC.store(tls), inner.store(tls);
}

constexpr size_t footprint() const noexcept {
return MAGIC.footprint() + inner.footprint();
}
};

struct k2RpcResponseError final {
tl::i32 error_code{};
tl::string error{};
Expand Down
Loading