diff --git a/runtime-light/components/confdata/bindings/bindings.cpp b/runtime-light/components/confdata/bindings/bindings.cpp index 5d3ae4366a..9ab3cb6c46 100644 --- a/runtime-light/components/confdata/bindings/bindings.cpp +++ b/runtime-light/components/confdata/bindings/bindings.cpp @@ -7,6 +7,7 @@ #include "runtime-common/core/runtime-core.h" #include "runtime-light/allocator/allocator-state.h" +#include "runtime-light/components/confdata/state/component-state.h" #include "runtime-light/components/confdata/state/instance-state.h" #include "runtime-light/coroutine/coroutine-state.h" #include "runtime-light/coroutine/io-scheduler.h" @@ -18,11 +19,11 @@ namespace kphp::coro { auto instance_state::get() noexcept -> instance_state& { - return InstanceState::get().coroutine_instance_state; + return InstanceState::get().m_coroutine_instance_state; } auto io_scheduler::get() noexcept -> io_scheduler& { - return InstanceState::get().io_scheduler; + return InstanceState::get().m_io_scheduler; } } // namespace kphp::coro @@ -30,17 +31,16 @@ auto io_scheduler::get() noexcept -> io_scheduler& { namespace kphp::log { auto contextual_tags::try_get() noexcept -> std::optional> { - if (k2::instance_state() != nullptr) [[likely]] { - return InstanceState::get().instance_tags; - } return std::nullopt; } } // namespace kphp::log auto AllocatorState::get() noexcept -> const AllocatorState& { - if (k2::instance_state() != nullptr) [[likely]] { - return InstanceState::get().instance_allocator_state; + if (const auto* instance_state_ptr{k2::instance_state()}; instance_state_ptr != nullptr) [[likely]] { + return instance_state_ptr->m_allocator_state; + } else if (const auto* component_state_ptr{k2::component_state()}; component_state_ptr != nullptr) { + return component_state_ptr->m_allocator_state; } kphp::log::error("can't find allocator state"); } diff --git a/runtime-light/components/confdata/confdata-component.cpp b/runtime-light/components/confdata/confdata-component.cpp index d846ce020c..4f7761cd05 100644 --- a/runtime-light/components/confdata/confdata-component.cpp +++ b/runtime-light/components/confdata/confdata-component.cpp @@ -59,7 +59,22 @@ VISIBILITY_DEFAULT void k2_init_instance() { } VISIBILITY_DEFAULT k2::PollStatus k2_warmup() { - return k2::PollStatus::PollFinishedOk; + k2::details::image_state_ptr = k2_image_state(); + k2::details::component_state_ptr = k2_component_state(); + k2::details::instance_state_ptr = k2_instance_state(); + + auto& instance{InstanceState::get()}; + if (instance.m_warmup_status == InstanceState::warmup_status::done) { + return k2::PollStatus::PollFinishedOk; + } + + // the initial sync is performed by the service loop; pump the scheduler and observe the status it sets + const auto poll_status{kphp::coro::io_scheduler::get().process_events()}; + if (instance.m_warmup_status == InstanceState::warmup_status::done) { + return k2::PollStatus::PollFinishedOk; + } + // PollFinishedOk while the sync is still incomplete means the scheduler has drained unexpectedly + return poll_status == k2::PollStatus::PollFinishedOk ? k2::PollStatus::PollFinishedError : poll_status; } VISIBILITY_DEFAULT k2::PollStatus k2_poll() { diff --git a/runtime-light/components/confdata/confdata-proxy/sync-functions.h b/runtime-light/components/confdata/confdata-proxy/sync-functions.h new file mode 100644 index 0000000000..d6e0d49bb7 --- /dev/null +++ b/runtime-light/components/confdata/confdata-proxy/sync-functions.h @@ -0,0 +1,153 @@ +// Compiler for PHP (aka KPHP) +// Copyright (c) 2026 LLC «V Kontakte» +// Distributed under the GPL v3 License, see LICENSE.notice.txt + +#pragma once + +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include "common/wrappers/overloaded.h" +#include "runtime-common/core/allocator/script-allocator.h" +#include "runtime-common/core/std/containers.h" +#include "runtime-light/components/confdata/confdata-proxy/tl.h" +#include "runtime-light/coroutine/io-scheduler.h" +#include "runtime-light/coroutine/task.h" +#include "runtime-light/k2-platform/k2-api.h" +#include "runtime-light/stdlib/diagnostics/logs.h" +#include "runtime-light/stdlib/rpc/rpc-query.h" +#include "runtime-light/tl/tl-core.h" +#include "runtime-light/tl/tl-types.h" + +namespace kphp::confdata { + +struct pagination { + kphp::stl::string m_page; + int64_t m_offset{}; + bool m_has_synced{}; +}; + +enum class subscribe_error : uint8_t { transport, old_offset, malformed_response, not_synced }; + +namespace details { + +// Performs a single confdata.subscribe round-trip. +// On success, invokes `event_handler(events)` once with the batch of received events and updates `to` pagination. +// The batch is a view into the response buffer and is only valid for the duration of the call; empty batches are not delivered. +// An empty event value means that the key has been deleted. +template> event_handler_type> +auto subscribe(std::string_view confdata_proxy_actor, kphp::confdata::pagination& to, + const event_handler_type& event_handler) noexcept -> kphp::coro::task> { + // subscribe is a longpoll method, so the timeout must cover the time confdata-proxy may hold the request open + static constexpr auto SUBSCRIBE_TIMEOUT{std::chrono::milliseconds{45'000}}; + + const tl::RpcDestActorFlags request{.inner = {.actor_id = {}, + .flags = {.value = tl::rpcInvokeReqExtra::CUSTOM_TIMEOUT_MS_FLAG}, + .extra = {.opt_custom_timeout_ms = tl::i32{.value = SUBSCRIBE_TIMEOUT.count()}}, + .query = tl::confdata::Subscribe{ + .fields_mask = {}, + .access_token = {}, + .page = {.value = to.m_page}, + .offset = {.value = to.m_offset}, + .has_synced = {.value = to.m_has_synced}, + .prefixes = {.value = {{/* a single empty prefix subscribes to all keys */}}}, + + }}}; + tl::storer tls{request.footprint()}; + request.store(tls); + + // client-side timeout must outlive the server-side longpoll (SUBSCRIBE_TIMEOUT); 10x is a safe margin + auto expected_query{kphp::rpc::query::send(confdata_proxy_actor, SUBSCRIBE_TIMEOUT * 10, tls.view(), k2::RpcKind::TL_RPC)}; + if (!expected_query) [[unlikely]] { + kphp::log::warning("confdata: failed to send subscribe request: {}", expected_query.error()); + co_return std::unexpected{kphp::confdata::subscribe_error::transport}; + } + + kphp::stl::vector response_buffer{}; + auto expected_response{co_await kphp::rpc::query::response(std::move(*expected_query), [&response_buffer](size_t size) noexcept -> std::span { + response_buffer.resize(size); + return {response_buffer.data(), response_buffer.size()}; + })}; + if (!expected_response) [[unlikely]] { + kphp::log::warning("confdata: failed to fetch subscribe response: {}", expected_response.error()); + co_return std::unexpected{kphp::confdata::subscribe_error::transport}; + } + + tl::fetcher tlf{*expected_response}; + tl::confdata::SubscribeResponse response{}; + if (!response.fetch(tlf)) [[unlikely]] { + kphp::log::warning("confdata: failed to parse subscribe response"); + co_return std::unexpected{kphp::confdata::subscribe_error::malformed_response}; + } + + co_return std::visit( + overloaded{ + [&event_handler, &to](const tl::confdata::subscribeResponseOk& response) noexcept -> std::expected { + if (const auto& events{response.events}; events.size() != 0) { + std::invoke(event_handler, std::span{events.value}); + } + + to.m_page = response.new_page.value; + to.m_offset = response.new_offset.value; + to.m_has_synced = response.new_has_synced.value; + return {}; + }, + [](const tl::confdata::subscribeResponseOldOffsetError& /* unused */) noexcept -> std::expected { + return std::unexpected{kphp::confdata::subscribe_error::old_offset}; + }, + }, + response.value); +} + +} // namespace details + +// Paginates through a consistent snapshot of all subscribed keys until it has been fully synced. +// Returns the final pagination that should be passed to `update`. +// +// `event_handler` is invoked once per round-trip with a batch of events; the batch is only valid +// for the duration of the call and must be copied if it needs to be retained. +template> event_handler_type> +auto sync(std::string_view confdata_proxy_actor, + event_handler_type event_handler) noexcept -> kphp::coro::task> { + kphp::confdata::pagination p{}; + for (; !p.m_has_synced;) { + if (auto expected{co_await details::subscribe(confdata_proxy_actor, p, event_handler)}; !expected) [[unlikely]] { + co_return std::unexpected{expected.error()}; + } + } + co_return std::move(p); +} + +// Longpoll loop: invokes `event_handler` for each event as it arrives, throttled to at most one batch per second: +// events that arrive between round-trips are buffered by the proxy and coalesced into the next batch. Returns only on error; +// `subscribe_error::old_offset` means that the local version is too old and a clean `sync` is required. +// `from` must be a synced pagination, typically the one returned by `sync`; `subscribe_error::not_synced` is returned otherwise. +// +// `event_handler` is invoked once per round-trip with a batch of events; the batch is only valid +// for the duration of the call and must be copied if it needs to be retained. +template> event_handler_type> +auto update(std::string_view confdata_proxy_actor, kphp::confdata::pagination& from, + event_handler_type event_handler) noexcept -> kphp::coro::task> { + // limits the update rate to at most one batch per interval + static constexpr auto UPDATE_INTERVAL{std::chrono::seconds{1}}; + + if (!from.m_has_synced) [[unlikely]] { + co_return std::unexpected{kphp::confdata::subscribe_error::not_synced}; + } + + for (;;) { + if (auto expected{co_await details::subscribe(confdata_proxy_actor, from, event_handler)}; !expected) [[unlikely]] { + co_return std::unexpected{expected.error()}; + } + co_await kphp::coro::io_scheduler::get().schedule(UPDATE_INTERVAL); + } +} + +} // namespace kphp::confdata diff --git a/runtime-light/components/confdata/confdata-proxy/tl.h b/runtime-light/components/confdata/confdata-proxy/tl.h new file mode 100644 index 0000000000..dfc8fa0082 --- /dev/null +++ b/runtime-light/components/confdata/confdata-proxy/tl.h @@ -0,0 +1,136 @@ +// Compiler for PHP (aka KPHP) +// Copyright (c) 2026 LLC «V Kontakte» +// Distributed under the GPL v3 License, see LICENSE.notice.txt + +#pragma once + +#include +#include +#include +#include +#include + +#include "runtime-light/tl/tl-core.h" +#include "runtime-light/tl/tl-types.h" + +namespace tl::confdata { + +struct keyValuePair final { + tl::string key{}; + tl::string value{}; + tl::Bool is_php_serialized{}; + tl::Bool is_json_serialized{}; + + bool fetch(tl::fetcher& tlf) noexcept { + return key.fetch(tlf) && value.fetch(tlf) && is_php_serialized.fetch(tlf) && is_json_serialized.fetch(tlf); + } + + constexpr size_t footprint() const noexcept { + return key.footprint() + value.footprint() + is_php_serialized.footprint() + is_json_serialized.footprint(); + } +}; + +class KeyValuePair final { + static constexpr tl::magic MAGIC{.value = 0xff1c'b454}; + +public: + tl::confdata::keyValuePair inner{}; + + bool fetch(tl::fetcher& tlf) noexcept { + tl::magic magic{}; + return magic.fetch(tlf) && magic.expect(MAGIC) && inner.fetch(tlf); + } + + constexpr size_t footprint() const noexcept { + return MAGIC.footprint() + inner.footprint(); + } +}; + +struct subscribeResponseOk final { + tl::vector events{}; + tl::string new_page{}; + tl::i64 new_offset{}; + tl::Bool new_has_synced{}; + + bool fetch(tl::fetcher& tlf) noexcept { + return events.fetch(tlf) && new_page.fetch(tlf) && new_offset.fetch(tlf) && new_has_synced.fetch(tlf); + } + + constexpr size_t footprint() const noexcept { + return events.footprint() + new_page.footprint() + new_offset.footprint() + new_has_synced.footprint(); + } +}; + +struct subscribeResponseOldOffsetError final { + bool fetch(tl::fetcher& /*unused*/) noexcept { + return true; + } + + constexpr size_t footprint() const noexcept { + return 0; + } +}; + +class SubscribeResponse final { + static constexpr tl::magic SUBSCRIBE_RESPONSE_OK_MAGIC{.value = 0x2709'63e8}; + static constexpr tl::magic SUBSCRIBE_RESPONSE_OLD_OFFSET_ERROR_MAGIC{.value = 0x11eb'eb02}; + +public: + std::variant value; + + bool fetch(tl::fetcher& tlf) noexcept { + tl::magic magic{}; + if (!magic.fetch(tlf)) { + return false; + } + + if (tl::confdata::subscribeResponseOk response_ok{}; magic.expect(SUBSCRIBE_RESPONSE_OK_MAGIC) && response_ok.fetch(tlf)) { + value.emplace(std::move(response_ok)); + return true; + } + if (tl::confdata::subscribeResponseOldOffsetError old_offset_error{}; + magic.expect(SUBSCRIBE_RESPONSE_OLD_OFFSET_ERROR_MAGIC) && old_offset_error.fetch(tlf)) { + value.emplace(old_offset_error); + return true; + } + return false; + } + + constexpr size_t footprint() const noexcept { + return std::visit( + [](const auto& value) noexcept { + using value_t = std::remove_cvref_t; + if constexpr (std::same_as) { + return SUBSCRIBE_RESPONSE_OK_MAGIC.footprint() + value.footprint(); + } else if constexpr (std::same_as) { + return SUBSCRIBE_RESPONSE_OLD_OFFSET_ERROR_MAGIC.footprint() + value.footprint(); + } else { + static_assert(false, "non-exhaustive visitor!"); + } + }, + value); + } +}; + +class Subscribe final { + static constexpr tl::magic MAGIC{.value = 0xfebd'1230}; + +public: + tl::mask fields_mask{}; + tl::string access_token{}; + tl::string page{}; + tl::i64 offset{}; + tl::Bool has_synced{}; + tl::vector prefixes{}; + + void store(tl::storer& tls) const noexcept { + MAGIC.store(tls), fields_mask.store(tls), access_token.store(tls), page.store(tls), offset.store(tls), has_synced.store(tls), prefixes.store(tls); + } + + constexpr size_t footprint() const noexcept { + return MAGIC.footprint() + fields_mask.footprint() + access_token.footprint() + page.footprint() + offset.footprint() + has_synced.footprint() + + prefixes.footprint(); + } +}; + +} // namespace tl::confdata diff --git a/runtime-light/components/confdata/confdata.cmake b/runtime-light/components/confdata/confdata.cmake index a6f2478d31..ecff9f47b1 100644 --- a/runtime-light/components/confdata/confdata.cmake +++ b/runtime-light/components/confdata/confdata.cmake @@ -3,8 +3,13 @@ set(K2_CONFDATA_COMPONENT_SRC ${RUNTIME_LIGHT_DIR}/components/confdata/confdata-component.cpp ${RUNTIME_LIGHT_DIR}/components/confdata/bindings/bindings.cpp + ${RUNTIME_LIGHT_DIR}/components/confdata/state/component-state.cpp ${RUNTIME_LIGHT_DIR}/components/confdata/state/instance-state.cpp) +set(K2_CONFDATA_TL_SRC + ${RUNTIME_LIGHT_DIR}/tl/tl-types.cpp + ${RUNTIME_LIGHT_DIR}/tl/tl-functions.cpp) + set(K2_CONFDATA_ALLOCATOR_SRC ${RUNTIME_LIGHT_DIR}/allocator/runtime-light-allocator.cpp ${RUNTIME_LIGHT_DIR}/memory-resource-impl/monotonic-light-buffer-resource.cpp) @@ -21,6 +26,7 @@ set(K2_CONFDATA_MEMORY_RESOURCE_SRC set(K2_CONFDATA_SRC ${K2_CONFDATA_COMPONENT_SRC} + ${K2_CONFDATA_TL_SRC} ${K2_CONFDATA_ALLOCATOR_SRC} ${K2_CONFDATA_DIAGNOSTICS_SRC} ${K2_CONFDATA_MEMORY_RESOURCE_SRC} diff --git a/runtime-light/components/confdata/state/component-state.cpp b/runtime-light/components/confdata/state/component-state.cpp new file mode 100644 index 0000000000..335a56a1eb --- /dev/null +++ b/runtime-light/components/confdata/state/component-state.cpp @@ -0,0 +1,28 @@ +// Compiler for PHP (aka KPHP) +// Copyright (c) 2026 LLC «V Kontakte» +// Distributed under the GPL v3 License, see LICENSE.notice.txt + +#include "runtime-light/components/confdata/state/component-state.h" + +#include + +#include "runtime-light/k2-platform/k2-api.h" +#include "runtime-light/stdlib/diagnostics/logs.h" + +auto ComponentState::parse_confdata_proxy_actor_name_arg(std::string_view value_view) noexcept -> void { + m_confdata_proxy_actor_name = value_view; +} + +auto ComponentState::parse_args() noexcept -> void { + for (auto i = 0; i < m_argc; ++i) { + const auto [arg_key, arg_value]{k2::arg_fetch(i)}; + const std::string_view key_view{arg_key.get(), std::strlen(arg_key.get())}; + const std::string_view value_view{arg_value.get(), std::strlen(arg_value.get())}; + + if (key_view == CONFDATA_PROXY_ACTOR_NAME_ARG) { + parse_confdata_proxy_actor_name_arg(value_view); + } else { + kphp::log::error("unexpected argument: {}", key_view); + } + } +} diff --git a/runtime-light/components/confdata/state/component-state.h b/runtime-light/components/confdata/state/component-state.h index c5e3059aa6..9e39724264 100644 --- a/runtime-light/components/confdata/state/component-state.h +++ b/runtime-light/components/confdata/state/component-state.h @@ -5,16 +5,43 @@ #pragma once #include +#include #include "common/mixin/not_copyable.h" +#include "runtime-common/core/allocator/script-allocator.h" +#include "runtime-common/core/std/containers.h" +#include "runtime-light/allocator/allocator-state.h" #include "runtime-light/k2-platform/k2-api.h" +#include "runtime-light/stdlib/diagnostics/logs.h" struct ComponentState final : private vk::not_copyable { - ComponentState() noexcept = default; + AllocatorState m_allocator_state{INIT_COMPONENT_ALLOCATOR_SIZE, DEFAULT_MIN_EXTRA_MEMORY_POOL_SIZE, 0}; + kphp::stl::string m_confdata_proxy_actor_name; + +private: + const uint32_t m_argc{k2::args_count()}; + +public: + ComponentState() noexcept; static auto get() noexcept -> const ComponentState&; static auto get_mutable() noexcept -> ComponentState&; + +private: + auto parse_confdata_proxy_actor_name_arg(std::string_view) noexcept -> void; + auto parse_args() noexcept -> void; + + static constexpr std::string_view CONFDATA_PROXY_ACTOR_NAME_ARG{"confdata-proxy-actor-name"}; + static constexpr auto INIT_COMPONENT_ALLOCATOR_SIZE{static_cast(1024U * 1024U)}; // 1MiB }; +inline ComponentState::ComponentState() noexcept { + parse_args(); + + if (m_confdata_proxy_actor_name.empty()) { + kphp::log::error("{} argument is required", CONFDATA_PROXY_ACTOR_NAME_ARG); + } +} + inline auto ComponentState::get() noexcept -> const ComponentState& { return *k2::component_state(); } diff --git a/runtime-light/components/confdata/state/instance-state.cpp b/runtime-light/components/confdata/state/instance-state.cpp index 1aaa3c42f4..01510f1bb1 100644 --- a/runtime-light/components/confdata/state/instance-state.cpp +++ b/runtime-light/components/confdata/state/instance-state.cpp @@ -4,36 +4,98 @@ #include "runtime-light/components/confdata/state/instance-state.h" +#include #include +#include #include #include +#include #include +#include "runtime-light/components/confdata/confdata-proxy/sync-functions.h" +#include "runtime-light/components/confdata/confdata-proxy/tl.h" +#include "runtime-light/components/confdata/state/component-state.h" #include "runtime-light/coroutine/task.h" +#include "runtime-light/coroutine/when-all.h" #include "runtime-light/stdlib/diagnostics/logs.h" #include "runtime-light/streams/stream.h" +namespace { + +auto sync_handler(std::span events) noexcept -> void { + kphp::log::info("got {} events on sync", events.size()); +} + +auto update_handler(std::span events) noexcept -> void { + kphp::log::info("got {} events on update", events.size()); +} + +} // namespace + auto InstanceState::init() noexcept -> void { auto main_task{run()}; // initialize async stack auto& main_task_async_stack_frame{main_task.get_handle().promise().get_async_stack_frame()}; - main_task_async_stack_frame.async_stack_root = std::addressof(coroutine_instance_state.coroutine_stack_root); - coroutine_instance_state.coroutine_stack_root.top_async_stack_frame = std::addressof(main_task_async_stack_frame); + main_task_async_stack_frame.async_stack_root = std::addressof(m_coroutine_instance_state.coroutine_stack_root); + m_coroutine_instance_state.coroutine_stack_root.top_async_stack_frame = std::addressof(main_task_async_stack_frame); // spawn main task onto the scheduler - kphp::log::assertion(io_scheduler.spawn(std::move(main_task))); + kphp::log::assertion(m_io_scheduler.spawn(std::move(main_task))); } auto InstanceState::run() noexcept -> kphp::coro::task<> { - auto opt_stream{co_await kphp::component::stream::accept()}; - if (!opt_stream.has_value()) [[unlikely]] { - kphp::log::warning("failed to accept a stream"); - co_return; + co_await kphp::coro::when_all(service_loop(), accept_loop()); // both never return + kphp::log::assertion(false); +} + +auto InstanceState::accept_loop() noexcept -> kphp::coro::task<> { + for (;;) { + auto opt_stream{co_await kphp::component::stream::accept()}; + if (!opt_stream.has_value()) [[unlikely]] { + kphp::log::warning("failed to accept a stream"); + continue; + } + auto request_stream{std::move(*opt_stream)}; + kphp::log::info("accepted a stream: descriptor -> {}", request_stream.descriptor()); + + // dummy implementation: drain the request and close + if (auto expected{co_await request_stream.read_all([](std::span) noexcept {})}; !expected) [[unlikely]] { + kphp::log::warning("failed to read a request: error -> {}", expected.error()); + } } - auto request_stream{std::move(*opt_stream)}; - kphp::log::info("accepted a stream: descriptor -> {}", request_stream.descriptor()); +} + +auto InstanceState::service_loop() noexcept -> kphp::coro::task<> { + static constexpr auto CONFDATA_RETRY_INTERVAL{std::chrono::seconds{1}}; + const std::string_view confdata_proxy_actor{ComponentState::get().m_confdata_proxy_actor_name}; + + for (;;) { + if (!m_pagination.m_has_synced) { + auto sync{co_await kphp::confdata::sync(confdata_proxy_actor, sync_handler)}; + if (!sync) [[unlikely]] { + kphp::log::warning("confdata sync failed: error -> {}, retrying", std::to_underlying(std::move(sync).error())); + co_await m_io_scheduler.schedule(CONFDATA_RETRY_INTERVAL); + continue; + } + m_pagination = *std::move(sync); + m_warmup_status = InstanceState::warmup_status::done; + } - // dummy implementation: drain the request and close - if (auto expected{co_await request_stream.read_all([](std::span) noexcept {})}; !expected) [[unlikely]] { - kphp::log::warning("failed to read a request: error -> {}", expected.error()); + auto update{co_await kphp::confdata::update(confdata_proxy_actor, m_pagination, update_handler)}; + // update returns only on error; m_pagination was advanced in place up to the last applied batch + kphp::log::assertion(!update.has_value()); + switch (update.error()) { + case kphp::confdata::subscribe_error::old_offset: + case kphp::confdata::subscribe_error::not_synced: + // local version is too old: clean re-sync required + kphp::log::warning("confdata update failed: error -> {}, resyncing", std::to_underlying(std::move(update).error())); + m_pagination = {}; + break; + case kphp::confdata::subscribe_error::transport: + case kphp::confdata::subscribe_error::malformed_response: + // pagination is still valid; the longpoll resumes from the last applied position + kphp::log::warning("confdata update failed: error -> {}, retrying", std::to_underlying(std::move(update).error())); + break; + } + co_await m_io_scheduler.schedule(CONFDATA_RETRY_INTERVAL); } } diff --git a/runtime-light/components/confdata/state/instance-state.h b/runtime-light/components/confdata/state/instance-state.h index 4a7243baea..8bd4c66434 100644 --- a/runtime-light/components/confdata/state/instance-state.h +++ b/runtime-light/components/confdata/state/instance-state.h @@ -5,9 +5,11 @@ #pragma once #include +#include #include "common/mixin/not_copyable.h" #include "runtime-light/allocator/allocator-state.h" +#include "runtime-light/components/confdata/confdata-proxy/sync-functions.h" #include "runtime-light/coroutine/coroutine-state.h" #include "runtime-light/coroutine/io-scheduler.h" #include "runtime-light/coroutine/task.h" @@ -15,12 +17,17 @@ #include "runtime-light/stdlib/diagnostics/contextual-tags.h" struct InstanceState final : vk::not_copyable { - AllocatorState instance_allocator_state{INIT_INSTANCE_ALLOCATOR_SIZE, DEFAULT_MIN_EXTRA_MEMORY_POOL_SIZE, 0}; + enum class warmup_status : uint8_t { pending, done }; - kphp::log::contextual_tags instance_tags; + AllocatorState m_allocator_state{INIT_INSTANCE_ALLOCATOR_SIZE, DEFAULT_MIN_EXTRA_MEMORY_POOL_SIZE, 0}; - kphp::coro::instance_state coroutine_instance_state; - kphp::coro::io_scheduler io_scheduler{coroutine_instance_state}; + warmup_status m_warmup_status{warmup_status::pending}; + kphp::confdata::pagination m_pagination{}; + + kphp::log::contextual_tags m_instance_tags; + + kphp::coro::instance_state m_coroutine_instance_state; + kphp::coro::io_scheduler m_io_scheduler{m_coroutine_instance_state}; InstanceState() noexcept = default; static auto get() noexcept -> InstanceState&; @@ -31,6 +38,8 @@ struct InstanceState final : vk::not_copyable { static constexpr auto INIT_INSTANCE_ALLOCATOR_SIZE = static_cast(16U * 1024U * 1024U); // 16MiB auto run() noexcept -> kphp::coro::task<>; + auto accept_loop() noexcept -> kphp::coro::task<>; + auto service_loop() noexcept -> kphp::coro::task<>; }; inline auto InstanceState::get() noexcept -> InstanceState& { diff --git a/runtime-light/tl/tl-types.h b/runtime-light/tl/tl-types.h index 63b4d42a29..7fa507edf6 100644 --- a/runtime-light/tl/tl-types.h +++ b/runtime-light/tl/tl-types.h @@ -44,6 +44,10 @@ struct magic final { return expected == value; } + bool expect(tl::magic expected) const noexcept { + return expect(expected.value); + } + constexpr size_t footprint() const noexcept { return sizeof(underlying_type); } @@ -1283,7 +1287,7 @@ class traceContext final { } // namespace tracing -class rpcInvokeReqExtra final { +struct rpcInvokeReqExtra final { static constexpr uint32_t RETURN_BINLOG_POS_FLAG = vk::tl::common::rpc_invoke_req_extra_flags::return_binlog_pos; static constexpr uint32_t RETURN_BINLOG_TIME_FLAG = vk::tl::common::rpc_invoke_req_extra_flags::return_binlog_time; static constexpr uint32_t RETURN_PID_FLAG = vk::tl::common::rpc_invoke_req_extra_flags::return_pid; @@ -1304,7 +1308,6 @@ class rpcInvokeReqExtra final { static constexpr uint32_t TRACE_CONTEXT_FLAG = vk::tl::common::rpc_invoke_req_extra_flags::trace_context; static constexpr uint32_t EXECUTION_CONTEXT_FLAG = vk::tl::common::rpc_invoke_req_extra_flags::execution_context; -public: bool return_binlog_pos{}; bool return_binlog_time{}; bool return_pid{};