Skip to content

[k2] instance_cache_store: freshness optimization #1695

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
denisichh wants to merge 1 commit into
base: master
Choose a base branch
from
+102 −43
Open

[k2] instance_cache_store: freshness optimization #1695

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
34 changes: 22 additions & 12 deletions runtime-light/k2-platform/k2-api.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -81,6 +81,8 @@ using PollStatus = PollStatus;

using ImageInfo = ImageInfo;

using SharedMemoryInfo = SharedMemoryInfo;

using RpcKind = RpcKind;

using ControlFlags = ControlFlags;
Expand Down Expand Up @@ -129,54 +131,62 @@ inline void free_checked(void* ptr, size_t size, size_t align) noexcept {
k2_free_checked(ptr, size, align);
}

inline std::expected<void*, int32_t> alloc_shared_memory(size_t size, size_t align = k2::details::DEFAULT_MEMORY_ALIGN) noexcept {
inline std::expected<void*, int32_t> shared_memory_alloc(size_t size, size_t align = k2::details::DEFAULT_MEMORY_ALIGN) noexcept {
void* pointer{nullptr};
if (auto error_code{k2_alloc_shared_memory(size, align, std::addressof(pointer))}; error_code != k2::errno_ok) [[unlikely]] {
if (auto error_code{k2_shared_memory_alloc(size, align, std::addressof(pointer))}; error_code != k2::errno_ok) [[unlikely]] {
return std::unexpected{error_code};
}
return {pointer};
}

inline std::expected<void, int32_t> publish_shared_memory(std::string_view name, const void* memory, uint64_t ttl_ms, bool as_mut,
inline std::expected<void, int32_t> shared_memory_publish(std::string_view name, const void* memory, uint64_t ttl_ms, bool as_mut,
bool ignore_if_exist) noexcept {
if (auto error_code{k2_publish_shared_memory(name.data(), name.length(), memory, ttl_ms, as_mut, ignore_if_exist)}; error_code != k2::errno_ok) [[unlikely]] {
if (auto error_code{k2_shared_memory_publish(name.data(), name.length(), memory, ttl_ms, as_mut, ignore_if_exist)}; error_code != k2::errno_ok) [[unlikely]] {
return std::unexpected{error_code};
}
return {};
}

inline std::expected<std::span<const std::byte>, int32_t> get_shared_memory(std::string_view name) noexcept {
inline std::expected<std::span<const std::byte>, int32_t> shared_memory_get(std::string_view name) noexcept {
const void* pointer{nullptr};
size_t size{};
if (auto error_code{k2_get_shared_memory(name.data(), name.length(), std::addressof(pointer), std::addressof(size))}; error_code != k2::errno_ok)
if (auto error_code{k2_shared_memory_get(name.data(), name.length(), std::addressof(pointer), std::addressof(size))}; error_code != k2::errno_ok)
[[unlikely]] {
return std::unexpected{error_code};
}
return {std::span{static_cast<const std::byte*>(pointer), size}};
}

inline std::expected<void, int32_t> republish_shared_memory(std::string_view name, uint64_t ttl) noexcept {
if (auto error_code{k2_republish_shared_memory(name.data(), name.length(), ttl)}; error_code != k2::errno_ok) [[unlikely]] {
inline std::expected<void, int32_t> shared_memory_republish(std::string_view name, uint64_t ttl) noexcept {
if (auto error_code{k2_shared_memory_republish(name.data(), name.length(), ttl)}; error_code != k2::errno_ok) [[unlikely]] {
return std::unexpected{error_code};
}
return {};
}

inline std::expected<void, int32_t> seek_ttl_to_shared_memory(std::string_view name, uint8_t percentile, uint64_t remaining_lifetime_limit) noexcept {
if (auto error_code{k2_seek_ttl_to_shared_memory(name.data(), name.length(), percentile, remaining_lifetime_limit)}; error_code != k2::errno_ok)
inline std::expected<void, int32_t> shared_memory_seek_ttl_to(std::string_view name, uint8_t percentile, uint64_t remaining_lifetime_limit) noexcept {
if (auto error_code{k2_shared_memory_seek_ttl_to(name.data(), name.length(), percentile, remaining_lifetime_limit)}; error_code != k2::errno_ok)
[[unlikely]] {
return std::unexpected{error_code};
}
return {};
}

inline std::expected<void, int32_t> release_shared_memory(const void* ptr) noexcept {
if (auto error_code{k2_release_shared_memory(ptr)}; error_code != k2::errno_ok) [[unlikely]] {
inline std::expected<void, int32_t> shared_memory_release(const void* ptr) noexcept {
if (auto error_code{k2_shared_memory_release(ptr)}; error_code != k2::errno_ok) [[unlikely]] {
return std::unexpected{error_code};
}
return {};
}

inline std::expected<const k2::SharedMemoryInfo, int32_t> shared_memory_info(std::string_view name) noexcept {
k2::SharedMemoryInfo info{};
if (auto error_code{k2_shared_memory_info(name.data(), name.length(), std::addressof(info))}; error_code != k2::errno_ok) [[unlikely]] {
return std::unexpected{error_code};
}
return info;
}

[[noreturn]] inline void exit(int32_t exit_code) noexcept {
k2_exit(exit_code);
}
Expand Down
82 changes: 58 additions & 24 deletions runtime-light/k2-platform/k2-header.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -28,7 +28,7 @@
#include <time.h>
#endif

#define K2_PLATFORM_HEADER_H_VERSION 16
#define K2_PLATFORM_HEADER_H_VERSION 17

// Always check that enum value is a valid value!

Expand Down Expand Up @@ -124,6 +124,18 @@ struct ImageInfo {
const struct KeyValuePair* extra_info;
};

// `stored_at` (the moment of the last publish/republish) is always treated as time origin 0,
// so both `now` and `expires_at` are nanosecond offsets from it, not wall-clock timestamps.
struct SharedMemoryInfo {
// Nanoseconds elapsed since `stored_at`.
uint64_t now;
// Nanoseconds from `stored_at` to the expiration moment. `0` means the memory is immortal
// (it was published/republished with a zero TTL) and never expires.
uint64_t expires_at;
// `true` if the memory was last published by this instance.
bool is_own;
};

/**
* Symbols provided by component image
* Every image should provide these symbols
Expand Down Expand Up @@ -199,21 +211,21 @@ void k2_free_checked(void* ptr, size_t size, size_t align);

/**
* Shared memory provides a mechanism for instances to share data.
* To use it, first allocate memory with `k2_alloc_shared_memory`, then publish
* it with a unique name using `k2_publish_shared_memory`. Other instances can
* then retrieve the memory by name with `k2_get_shared_memory`.
* To use it, first allocate memory with `k2_shared_memory_alloc`, then publish
* it with a unique name using `k2_shared_memory_publish`. Other instances can
* then retrieve the memory by name with `k2_shared_memory_get`.
*
* Lifecycle:
* - TTL defines when memory becomes eligible for reclamation
* - Memory is reclaimed lazily after TTL expires AND no references remain
*
* Reference counting:
* - Calling `k2_publish_shared_memory` sets the reference count to one
* - Calling `k2_get_shared_memory` increments the reference count by one if
* - Calling `k2_shared_memory_publish` sets the reference count to one
* - Calling `k2_shared_memory_get` increments the reference count by one if
* this instance does not already hold a reference; repeated calls for the
* same underlying pointer while already held do not increment it further
* - Calling `k2_release_shared_memory` clears the held reference, if `pointer`
* is currently held; unpublished memory from `k2_alloc_shared_memory` is
* - Calling `k2_shared_memory_release` clears the held reference, if `pointer`
* is currently held; unpublished memory from `k2_shared_memory_alloc` is
* just freed
*/

Expand All @@ -235,33 +247,33 @@ void k2_free_checked(void* ptr, size_t size, size_t align);
* `ENOMEM` => Not enough memory to allocate the requested region.
* `ENOSYS` => Shared memory subsystem is unavailable on this host.
*/
int32_t k2_alloc_shared_memory(size_t size, size_t align, void** pointer);
int32_t k2_shared_memory_alloc(size_t size, size_t align, void** pointer);

/**
* Publishes shared memory with a name and TTL, making it discoverable by other instances.
*
* @param `name` Name to associate with the memory region. Must be unique.
* Should be valid UTF-8 and not contain null bytes.
* @param `name_len` Length of the name in bytes. Must be greater than 0.
* @param `memory` Pointer to memory previously allocated via `k2_alloc_shared_memory`.
* @param `memory` Pointer to memory previously allocated via `k2_shared_memory_alloc`.
* @param `ttl` Time-to-live in milliseconds. Memory becomes eligible for
* reclamation after this duration, but only when the reference
* count reaches zero. Zero TTL means infinite life.
* @param `as_mut` Controls whether write protection is disabled. If true, the memory is made writable.
* @param `ignore_if_exist` If true, allows re-publishing memory previously allocated via
* `k2_alloc_shared_memory` even if a live published memory with the same name already exists.
* `k2_shared_memory_alloc` even if a live published memory with the same name already exists.
*
* @return `0` on success. libc-like `errno` on error.
*
* Possible `errno`:
* `EINVAL` => `name` is NULL, `name_len` is 0, `memory` is NULL, `name` is
* not valid UTF-8, or `ttl` is too small (non-zero and less than
* 100 ms).
* `ENOENT` => `memory` was not allocated by `k2_alloc_shared_memory`.
* `ENOENT` => `memory` was not allocated by `k2_shared_memory_alloc`.
* `EEXIST` => Memory with this name already exists.
* `ENOSYS` => Shared memory subsystem is unavailable on this host.
*/
int32_t k2_publish_shared_memory(const char* name, size_t name_len, const void* memory, uint64_t ttl, bool as_mut, bool ignore_if_exist);
int32_t k2_shared_memory_publish(const char* name, size_t name_len, const void* memory, uint64_t ttl, bool as_mut, bool ignore_if_exist);

/**
* Retrieves shared memory by name and, unless already held by this
Expand All @@ -278,10 +290,11 @@ int32_t k2_publish_shared_memory(const char* name, size_t name_len, const void*
* Possible `errno`:
* `EINVAL` => `name` is NULL, `name_len` is 0, `pointer` is NULL, or `name`
* is not valid UTF-8.
* `ENOENT` => No memory found with the given name (or TTL expired and memory was freed).
* `ENOENT` => No memory found with the given name (memory was freed).
* `EKEYEXPIRED` => TTL for this name has already expired.
* `ENOSYS` => Shared memory subsystem is unavailable on this host.
*/
int32_t k2_get_shared_memory(const char* name, size_t name_len, const void** pointer, size_t* size);
int32_t k2_shared_memory_get(const char* name, size_t name_len, const void** pointer, size_t* size);

/**
* Republishes shared memory, resetting its TTL and restarting the age counter from this call.
Expand All @@ -299,7 +312,7 @@ int32_t k2_get_shared_memory(const char* name, size_t name_len, const void** poi
* `ENOENT` => No memory found with the given name (or TTL expired and memory was freed).
* `ENOSYS` => Shared memory subsystem is unavailable on this host.
*/
int32_t k2_republish_shared_memory(const char* name, size_t name_len, uint64_t ttl);
int32_t k2_shared_memory_republish(const char* name, size_t name_len, uint64_t ttl);

/**
* Fast-forwards the expiration of published shared memory ahead of its TTL, by name.
Expand Down Expand Up @@ -334,21 +347,21 @@ int32_t k2_republish_shared_memory(const char* name, size_t name_len, uint64_t t
* `ENOENT` => No memory found with the given name (or TTL expired and memory was freed).
* `ENOSYS` => Shared memory subsystem is unavailable on this host.
*/
int32_t k2_seek_ttl_to_shared_memory(const char* name, size_t name_len, uint8_t percentile, uint64_t remaining_lifetime_limit);
int32_t k2_shared_memory_seek_ttl_to(const char* name, size_t name_len, uint8_t percentile, uint64_t remaining_lifetime_limit);

/**
* Frees shared memory allocated via `k2_alloc_shared_memory` that was never
* Frees shared memory allocated via `k2_shared_memory_alloc` that was never
* published, or releases the reference to shared memory held by this
* instance, acquired via `k2_publish_shared_memory` or
* `k2_get_shared_memory` (see the reference counting notes above).
* instance, acquired via `k2_shared_memory_publish` or
* `k2_shared_memory_get` (see the reference counting notes above).
*
* Releasing a reference does not free the underlying memory immediately: the
* memory is only physically reclaimed once its TTL has expired AND the
* reference count has reached zero (see the lifecycle notes above).
*
* @param `pointer` Pointer to shared memory, as returned by
* `k2_alloc_shared_memory` or `k2_get_shared_memory`, or
* passed as the `memory` argument to `k2_publish_shared_memory`.
* `k2_shared_memory_alloc` or `k2_shared_memory_get`, or
* passed as the `memory` argument to `k2_shared_memory_publish`.
*
* @return `0` on success: either the unpublished allocation was freed, or the
* reference held by this instance was released. libc-like `errno`
Expand All @@ -357,10 +370,31 @@ int32_t k2_seek_ttl_to_shared_memory(const char* name, size_t name_len, uint8_t
* Possible `errno`:
* `EINVAL` => `pointer` is NULL.
* `ENOENT` => `pointer` does not correspond to memory allocated by this
* instance via `k2_alloc_shared_memory`, nor to a reference
* instance via `k2_shared_memory_alloc`, nor to a reference
* currently held by this instance.
*/
int32_t k2_release_shared_memory(const void* pointer);
int32_t k2_shared_memory_release(const void* pointer);

/**
* Retrieves timing and ownership info about published shared memory, by name, without
* taking a reference to it.
*
* @param `name` Name of the shared memory to query.
* @param `name_len` Length of the name in bytes. Must be greater than 0.
* @param `info` Out parameter, filled with the current info on success. `stored_at` (the
* moment of the last publish/republish) is always treated as time origin 0, so
* `info->now` and `info->expires_at` are nanosecond offsets from it, not
* wall-clock timestamps. `info->expires_at == 0` means the memory is immortal
* (published/republished with a zero TTL) and never expires.
*
* @return `0` on success. libc-like `errno` on error.
*
* Possible `errno`:
* `EINVAL` => `name` is NULL, `name_len` is 0, `info` is NULL, or `name` is not valid UTF-8.
* `ENOENT` => No memory found with the given name (memory was freed).
* `EKEYEXPIRED` => TTL for this name has already expired.
*/
int32_t k2_shared_memory_info(const char* name, size_t name_len, struct SharedMemoryInfo* info);

/**
* Immediately abort component execution.
Expand Down
29 changes: 22 additions & 7 deletions runtime-light/stdlib/instance-cache/instance-cache-functions.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -25,6 +25,12 @@
// shared memory layout: class_name_hash(u64) | class_instance shell | inner data
template<typename InstanceType>
bool f$instance_cache_store(const string& key, class_instance<InstanceType> instance, int64_t ttl_sec = 0) noexcept {
static constexpr double FRESHNESS_ELEMENT_RATIO{0.2};
static constexpr auto freshness_ratio{[](uint64_t now, uint64_t expires_at) noexcept {
static constexpr double IMMORTAL_RATIO{0.5};
return expires_at == 0 ? IMMORTAL_RATIO : static_cast<double>(now) / static_cast<double>(expires_at);
}};

if (key.empty()) [[unlikely]] {
kphp::log::warning("instance_cache_store. empty key is not supported");
return false;
Expand All @@ -43,6 +49,15 @@ bool f$instance_cache_store(const string& key, class_instance<InstanceType> inst
ttl_sec = 0;
}

std::string_view name{std::string_view{key.c_str(), key.size()}};

if (auto shared_memory_info_result{k2::shared_memory_info(name)}; shared_memory_info_result.has_value()) {
const k2::SharedMemoryInfo info{*shared_memory_info_result};
if (!info.is_own && freshness_ratio(info.now, info.expires_at) < FRESHNESS_ELEMENT_RATIO) {
return false;
}
}

kphp::visitors::instance_deep_estimate_size_visitor estimate_size_visitor{};
if (!estimate_size_visitor.process_instance(instance)) [[unlikely]] {
kphp::log::warning("instance_cache_store. failed to estimate instance size: key -> {}", key.c_str());
Expand All @@ -52,7 +67,7 @@ bool f$instance_cache_store(const string& key, class_instance<InstanceType> inst
constexpr size_t instance_size{sizeof(class_instance<InstanceType>)};
constexpr size_t hash_size{sizeof(uint64_t)};

auto alloc_result{k2::alloc_shared_memory(hash_size + instance_size + estimated_size)};
auto alloc_result{k2::shared_memory_alloc(hash_size + instance_size + estimated_size)};
if (!alloc_result.has_value()) [[unlikely]] {
kphp::log::warning("instance_cache_store. failed to allocate shared memory: error -> {}, key -> {}", alloc_result.error(), key.c_str());
return false;
Expand All @@ -67,19 +82,19 @@ bool f$instance_cache_store(const string& key, class_instance<InstanceType> inst
kphp::visitors::instance_deep_copy_visitor copy_visitor{std::span{std::next(mem, hash_size + instance_size), estimated_size},
ExtraRefCnt::for_instance_cache};
if (!copy_visitor.process_instance(instance)) [[unlikely]] {
kphp::log::assertion(k2::release_shared_memory(mem).has_value());
kphp::log::assertion(k2::shared_memory_release(mem).has_value());
kphp::log::warning("instance_cache_store. failed to deep-copy instance into shared memory: estimated size -> {}, key -> {}", estimated_size, key.c_str());
return false;
}
std::construct_at(reinterpret_cast<class_instance<InstanceType>*>(std::next(mem, hash_size)), std::move(instance));

// the platform expects ttl in milliseconds, while the PHP API accepts seconds
if (auto publish_result{k2::publish_shared_memory(std::string_view{key.c_str(), key.size()}, mem, ttl_sec * 1000, false, true)}; publish_result.has_value()) {
if (auto publish_result{k2::shared_memory_publish(name, mem, ttl_sec * 1000, false, true)}; publish_result.has_value()) {
InstanceCacheInstanceState::get().request_cache.insert_or_assign(key, std::span{mem, hash_size + instance_size + estimated_size});
return true;
} else {
// publish is expected to always succeed here (ignore_if_exist=true, valid key/memory), so this should never actually happen.
kphp::log::assertion(k2::release_shared_memory(mem).has_value());
kphp::log::assertion(k2::shared_memory_release(mem).has_value());
kphp::log::warning("instance_cache_store. failed to publish shared memory: error -> {}, key -> {}", publish_result.error(), key.c_str());
return false;
}
Expand Down Expand Up @@ -119,7 +134,7 @@ ClassInstanceType f$instance_cache_fetch(const string& class_name, const string&
return materialize(it->second);
}

auto get_result{k2::get_shared_memory(std::string_view{key.c_str(), key.size()})};
auto get_result{k2::shared_memory_get(std::string_view{key.c_str(), key.size()})};
if (!get_result.has_value()) {
return {};
}
Expand All @@ -142,7 +157,7 @@ inline bool f$instance_cache_update_ttl(const string& key, int64_t ttl_sec = 0)
ttl_sec = 0;
}
// the platform expects ttl in milliseconds, while the PHP API accepts seconds
return k2::republish_shared_memory(std::string_view{key.c_str(), key.size()}, ttl_sec * 1000).has_value();
return k2::shared_memory_republish(std::string_view{key.c_str(), key.size()}, ttl_sec * 1000).has_value();
}

inline bool f$instance_cache_delete(const string& key) noexcept {
Expand All @@ -154,7 +169,7 @@ inline bool f$instance_cache_delete(const string& key) noexcept {
return false;
}
InstanceCacheInstanceState::get().request_cache.erase(key);
return k2::seek_ttl_to_shared_memory(std::string_view{key.c_str(), key.size()}, EARLY_EXPIRATION_ELEMENT_PERCENTILE,
return k2::shared_memory_seek_ttl_to(std::string_view{key.c_str(), key.size()}, EARLY_EXPIRATION_ELEMENT_PERCENTILE,
EXPIRED_ELEMENT_REMAINING_LIFETIME_LIMIT_MS)
.has_value();
}
Loading