diff --git a/PendingReleaseNotes b/PendingReleaseNotes index c1289d2fd10f..0a8ae908214a 100644 --- a/PendingReleaseNotes +++ b/PendingReleaseNotes @@ -1,3 +1,113 @@ +* librados/neorados: The C++ APIs for executing Ceph Class (CLS) methods + have undergone a breaking change to enforce compile-time type safety, replacing + legacy string-based parameters with strongly-typed ClsMethod structs. C++ + developers must update their method calls to use these new structs, which ensure + read/write semantics are correctly applied. + +* RGW: S3 ListObjects and ListObjectVersions now support the + ``x-amz-optional-object-attributes: RestoreStatus`` request header to include + restore status in listing responses. Restore status is stored in the bucket + index, so only objects written or restored after this upgrade will populate + the field. Existing objects are unaffected. +* RGW: New S3 Control APIs to apply PublicAccessBlock configuration to User Accounts. +* ceph-volume: Raw BlueStore OSD preparation now pre-formats NVMe devices and + skips the slower BlueStore discard phase,reducing mkfs time on + very large namespaces. +* RGW: Bucket Logging suppports creating log buckets in EC pools. + Implicit logging object commits are now performed asynchronously. +* RGW: radosgw-admin bucket list now supports pagination for versioned buckets by using + both --marker and --object-version options together (e.g., ``--marker=obj1 --object-version=abc123``). + This enables proper pagination through versioned bucket listings without duplicates or + infinite loops. For non-versioned buckets, use only --marker as before (backward compatible). +* RGW: OpenSSL engine support is deprecated in favor of provider support. + - Removed the `openssl_engine_opts` configuration option. OpenSSL engine configuration in string format is no longer supported. + - Added the `openssl_conf` configuration option for loading specified providers as default providers. + Configuration file syntax follows the OpenSSL standard (see https://github.com/openssl/openssl/blob/master/doc/man5/config.pod). + If the default provider is required when also using custom providers, + it must be explicitly loaded in the configuration file or code (see https://github.com/openssl/openssl/blob/master/README-PROVIDERS.md). +* RGW: Fixed bucket notification events so the 'x_amz_request_id' in NotificationEvent now matches the 'x_amz_request_id' returned by the corresponding S3 operation. + +* DASHBOARD: Introduces a new landing page - "Overview". This revamps UX and adds more information in the landing page - overall cluster health, health checks, resilency panel (showing PG status, active/clean percent), total and used raw capacity, alerts, capacity breakdown by object, file and block and performance charts - throughput, latency and IOPS. This renames the landing page from "Dashboard" to "Overview" +* DASHBOARD: Removed the older landing page which was deprecated in Quincy. + Admins can no longer enable the older, deprecated landing page layout by + adjusting FEATURE_TOGGLE_DASHBOARD. +* DASHBOARD: RGW Service form updated to take input regarding QAT compression. + - QAT compression is an optional field which can be set to 'Hardware' or 'Software' by selecting options from provided dropdwon. If 'None' is selected, compression is removed altogether. +* Monitoring: Added new Prometheus alerts for certificate management: + - CephCertificateError: Fires when a Ceph certificate has expired (critical severity). + - CephCertificateWarning: Fires when a Ceph certificate is about to expire (warning severity). +* CephFS: The `peer_add` command is deprecated in favor of the `peer_bootstrap` command. +* RGW: For `radosgw-admin account rm`, the `--purge-data` option previously applied only + to buckets and objects, but now deletes account users, roles, groups, and oidc-providers + too instead of failing with ENOTEMPTY. +* RADOS: When objects are read during deep scrubs, the data is read in strides, + and the scrubbing process is delayed between each read in order to avoid monopolizing + the I/O capacity of the OSD. + The default stride size (``osd_deep_scrub_stride``) was 512 KBytes, and is now 4 MBytes. +* RADOS: When an OSD is overloaded with queued snap-trim operations, no + regular (non-urgent) scrubs will be scheduled on that OSD. This is + determined by comparing the total snap-trim queue lengths for all PGs + for which the OSD is a primary against ``osd_scrub_queued_snaptrims_limit``, + which defaults to 500. + This restriction does not apply to operator-initiated scrubs, nor to repair scrubs. + It can be disabled by setting ``osd_scrub_queued_snaptrims_limit`` to 0. +* RGW: Add SSE-KMS secrets cache + +* RADOS: Stretch mode can now be entered even if the two dividing buckets differ + in weight by a small fraction (default 0.1). This is tunable via + `mon_stretch_max_bucket_weight_delta`. + +* CephFS: The offline CephFS tools (cephfs-data-scan, cephfs-journal-tool, + and cephfs-table-tool) now include progress tracking with ETA (Estimated Time of + Arrival) for long-running operations. Progress updates are displayed automatically + at regular intervals, showing completion percentage, processed items, and time + estimates. This feature is enabled by default for relevant commands including + scan_extents, scan_inodes, and other state-changing operations. + Related Tracker: https://tracker.ceph.com/issues/63191 +* RBD: Fixed incorrect behavior of the "start-time" argument for mirror + snapshot and trash purge schedules, where it previously offset the schedule + anchor instead of defining it. The argument now requires an ISO 8601 + date-time. The `schedule ls` output displays the start time in UTC, including + the date and time in the format "%Y-%m-%d %H:%M:00". The `schedule status` + output now displays the next schedule time in UTC. +* CephFS Mirroring: Now utilizes a multi-threaded architecture to improve synchronization + performance. The workload is split into two distinct thread pools: a crawler thread pool, which + manages snapshot crawl and a data synchronization thread pool, which handles concurrent file + transfers. Users can fine-tune these operations using configuration parameters: + cephfs_mirror_max_concurrent_directory_syncs (controlling the number of concurrent snapshots being crawled) + and cephfs_mirror_max_datasync_threads (controlling the total threads available for data sync). + For more information, see https://tracker.ceph.com/issues/73452 +* CephFS Mirroring: Improved incremental synchronization behavior in CephFS mirroring. Previously, + block-level delta synchronization was used for all files regardless of size. With this change, + blockdiff is applied only to files larger than a configurable threshold, while smaller files are + synchronized using full copy, as blockdiff is not efficient for small files. The threshold is + controlled by the new configuration option cephfs_mirror_blockdiff_min_file_size (default: 16_M). + For more information, see https://tracker.ceph.com/issues/73452 +* CephFS Mirroring: Improved mirror daemon status reporting. The command + ``ceph fs snapshot mirror daemon status`` now shows the remote cluster's + monitor addresses and cluster ID for each configured peer, making it easier + to verify peer connectivity and troubleshoot mirroring issues. +* RBD: Mirror snapshot creation and trash purge schedules are now automatically + staggered when no explicit "start-time" is specified. This reduces scheduling + spikes and distributes work more evenly over time. +* RBD: Introduced a new ``RBD_LOCK_MODE_EXCLUSIVE_TRANSIENT`` policy for + ``rbd_lock_acquire()``. This is a low-level interface intended to allow + a peer to grab exclusive lock manually for short periods of time with other + peers pausing their activity and waiting for the lock to be released rather + than instantly aborting I/O and returning an error. It's possible to switch + from ``RBD_LOCK_MODE_EXCLUSIVE`` to ``RBD_LOCK_MODE_EXCLUSIVE_TRANSIENT`` + policy and vice versa even if the lock is already held. +* OSD: A health warning is reported when BlueFS usage exceeds the + configured ratio of the main OSD data device size. This warning is + informational and can be muted with: + ``ceph health mute BLUESTORE_BLUEFS_OVERSIZED`` +* MGR: The Manager now automatically increases ``mgr_stats_period`` when its + message queue is congested, reducing daemon reporting frequency to prevent + overload. The period recovers automatically once the queue clears. This + behavior is controlled by the new ``mgr_stats_period_autotune`` (default: + ``true``) and ``mgr_stats_period_autotune_queue_threshold`` (default: ``100``) + config options. + >=20.0.0 * RGW: Lua scripts will not run against health checks. diff --git a/doc/radosgw/config-ref.rst b/doc/radosgw/config-ref.rst index b4c4829ae9d3..c403ba320427 100644 --- a/doc/radosgw/config-ref.rst +++ b/doc/radosgw/config-ref.rst @@ -271,6 +271,16 @@ SSE-S3 Settings .. confval:: rgw_crypt_sse_s3_vault_ssl_clientcert .. confval:: rgw_crypt_sse_s3_vault_ssl_clientkey +KMS Secrets Cache Settings +========================== + +.. confval:: rgw_crypt_s3_kms_cache_enabled +.. confval:: rgw_crypt_s3_kms_cache_max_size +.. confval:: rgw_crypt_s3_kms_cache_positive_ttl +.. confval:: rgw_crypt_s3_kms_cache_transient_error_ttl +.. confval:: rgw_crypt_s3_kms_cache_negative_ttl + +.. _Encryption: ../encryption QoS settings ------------ diff --git a/doc/radosgw/encryption.rst b/doc/radosgw/encryption.rst index e30fe1468725..dc55a5ec0b68 100644 --- a/doc/radosgw/encryption.rst +++ b/doc/radosgw/encryption.rst @@ -81,7 +81,53 @@ The configuration expects a base64-encoded 256 bit key. For example:: file is not a secure method for storing encryption keys. Keys that are accidentally exposed in this way should be considered compromised. - +Caching +======= + +The caching feature for Key Management Service (KMS) secrets greatly +improves the performance of server-side encryption and lessens the +load on the KMS. + +Secrets are stored using the `Linux Kernel Key Retention Service`_ in +the RGW processes' process keyring. This is subject to a global quota +and must be set in accordance with the configured cache size. +Depending on whether RGW runs as root, these quotas can be managed by adjusting: +- ``/proc/sys/kernel/keys/root_maxkeys`` and ``/proc/sys/kernel/keys/root_maxbytes`` +- ``/proc/sys/kernel/keys/maxkeys`` and ``/proc/sys/kernel/keys/maxbytes`` + +Exceeding a quota will disable the cache, fail the request with an +internal error, and log a failure message. + +Three different Cache Time-to-Live (TTL) values can be set: +- **Positive TTL**: How long a successfully retrieved secret remains + in the cache. +- **Negative TTL**: How long to remember that a key does not exist, + preventing unnecessary requests to the KMS. +- **Transient Error TTL**: How long to cache failures due to temporary + issues like KMS timeouts. + +Metrics +--------- + +The cache exports metrics under the ``kms-cache`` collection. +- ``hit``: Hit counter +- ``miss``: Miss counter +- ``expired``: Number of TTL expired entries +- ``size``: Current cache size +- ``capacity``: Cache maximum size +- ``clear``: Number of cache clears. Resets ``size``, ``hit``, + ``miss``, ``expired`` + +In addition the ``rgw`` collection has: +- ``kms_fetch_lat``: Average KMS fetch latency. Also includes a + successful request counter. Each event results in a positive cache + entry. +- ``kms_error_transient``: Transient KMS fetch error counter. Each + event results in a transient error cache entry. +- ``kms_error_permanent``: Permanent KMS fetch error counter. Each + event results in a negative cache cache entry. + +.. _Linux Kernel Key Retention Service: https://www.kernel.org/doc/html/latest/security/keys/core.html .. _Amazon SSE-C: https://docs.aws.amazon.com/AmazonS3/latest/dev/ServerSideEncryptionCustomerKeys.html .. _Amazon SSE-KMS: http://docs.aws.amazon.com/AmazonS3/latest/dev/UsingKMSEncryption.html .. _Amazon SSE-S3: https://docs.aws.amazon.com/AmazonS3/latest/userguide/UsingServerSideEncryption.html diff --git a/release-management/backports/019eb08c-159f-7a78-b689-5d37a7d343d0.md b/release-management/backports/019eb08c-159f-7a78-b689-5d37a7d343d0.md new file mode 100644 index 000000000000..03299d70ba38 --- /dev/null +++ b/release-management/backports/019eb08c-159f-7a78-b689-5d37a7d343d0.md @@ -0,0 +1,49 @@ +--- +id: 019eb08c-159f-7a78-b689-5d37a7d343d0 +title: KMS Cache +component: rgw +category: feature +provenance: + type: upstream_pr + upstream_prs: + - https://github.com/ceph/ceph/pull/61256 + upstream_pr_state: merged-main + upstream_commits: + - 56253c4ae02ecffcf878f5046839991f3e7b0d0e + upstream_trackers: + - tracker.ceph.com/issues/68524 +justification: + type: customer + refs: + - cobaltcore-dev/cloud-storage#463 + - https://github.com/cobaltcore-dev/cloud-storage/issues/125 +risk: + blast: data-loss + conflict: substantive + coverage: partial +--- + +## Justification (public) + +Adds a secure in-memory cache for SSE-KMS encryption keys retrieved from Barbican (or other KMS backends), eliminating redundant round-trips on repeated access to the same encrypted objects. Uses a new SIEVE-based eviction cache (WebCache) and stores decrypted secrets in the Linux kernel keyring. +Please refer to the Ceph documentation on how to configure this feature. + +## Justification (internal) + +Customer roadmap feature. + +## Risk notes + +- TTLs mean that a deleted or rotated Barbican key won't be immediately reflected. +- New async primitives (async::call_once) and the WebCache are significant new library code shipping for the first time. +- Cache size tuning and system wide keyring quota adjustment needed for production workloads (refer to the docs). + +## Behavior change + +Repeated SSE-KMS access no longer hits the KMS backend on every +request; decrypted secrets are cached in-memory and in the Linux +kernel keyring. KMS-cache is ON out of the box. + +## Upgrade notes + +Feature flag: rgw_crypt_s3_kms_cache_enabled. Set false to disable caching entirely and revert to per-request KMS lookups. diff --git a/src/common/CMakeLists.txt b/src/common/CMakeLists.txt index 3da83cd9ed1c..641bebe7ca46 100644 --- a/src/common/CMakeLists.txt +++ b/src/common/CMakeLists.txt @@ -246,4 +246,9 @@ if(HAVE_KEYUTILS) set(parse_secret_srcs secret.c) add_library(parse_secret_objs OBJECT ${parse_secret_srcs}) + + set(keyring_srcs + keyring.cc) + add_library(keyring STATIC ${keyring_srcs}) + target_link_libraries(keyring keyutils::keyutils) endif() diff --git a/src/common/async/call_once.h b/src/common/async/call_once.h new file mode 100644 index 000000000000..8c48a2c3c6a8 --- /dev/null +++ b/src/common/async/call_once.h @@ -0,0 +1,113 @@ +// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*- +// vim: ts=8 sw=2 smarttab ft=cpp + +/* + * Ceph - scalable distributed file system + * + * Copyright contributors to the Ceph project + * + * This is free software; you can redistribute it and/or + * modify it under the terms of the GNU Lesser General Public + * License version 2.1, as published by the Free Software + * Foundation. See file COPYING. + * + */ + +#pragma once + +#include +#include +#include +#include + +#include "include/function2.hpp" + +#include "yield_context.h" +#include "detail/call_once.h" + +namespace ceph::async { + +/// Stores the result of call_once(), if any. +/// +/// The cached Result type must be semiregular (copyable and default- +/// constructible). +template +class once_result { + public: + /// Call f() once and cache its return value. + /// + /// If a call has already completed, return the cached result. If a call + /// is already in progress, wait for its completion and return the result. + /// If a yield context is provided in y, its coroutine will be suspended + /// while waiting. f() may also suspend/resume the same coroutine. + /// + /// f() must return a Result when called with no arguments. If it throws, + /// that exception is rethrown by all calls. + /// + /// This function is thread-safe. + template + friend Result call_once(once_result& self, optional_yield y, Callable&& f) + requires std::convertible_to, Result> + { + auto lock = std::unique_lock{self.mutex}; + return std::visit( + fu2::overload( + [&] (const Init&) { return self.do_init(lock, f); }, + [&] (Wait& w) { return self.do_wait(lock, y, w); }, + [] (const Complete& c) { return do_complete(c); }), + self.state); + } + + private: + std::mutex mutex; + + // state machine: [Init] -> [Wait] -> [Complete] + using Init = std::monostate; + using Wait = detail::call_once::WaitState; + struct Complete { + Result result; + std::exception_ptr eptr; + }; + std::variant state; + + Result do_init(std::unique_lock& lock, auto&& f) + { + // transition to Wait state + auto& wait = state.template emplace(); + lock.unlock(); + + Complete complete; + try { + complete.result = f(); + } catch (const std::exception&) { + complete.eptr = std::current_exception(); + } + + lock.lock(); + // wake any waiters + wait.wake_all(); + + // transition to Complete state and return the result + return do_complete(state.template emplace(std::move(complete))); + } + + Result do_wait(std::unique_lock& lock, + optional_yield y, Wait& wait) + { + // wait for the response to an outstanding request + wait.wait(lock, y); + + // state must be Complete after wakeup + return do_complete(std::get(state)); + } + + static Result do_complete(const Complete& complete) + { + if (complete.eptr) { // rethrow cached exception + std::rethrow_exception(complete.eptr); + } + return complete.result; // return cached result + } +}; + +} // namespace ceph::async diff --git a/src/common/async/detail/call_once.h b/src/common/async/detail/call_once.h new file mode 100644 index 000000000000..c7d676457934 --- /dev/null +++ b/src/common/async/detail/call_once.h @@ -0,0 +1,129 @@ +// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*- +// vim: ts=8 sw=2 smarttab ft=cpp + +/* + * Ceph - scalable distributed file system + * + * Copyright contributors to the Ceph project + * + * This is free software; you can redistribute it and/or + * modify it under the terms of the GNU Lesser General Public + * License version 2.1, as published by the Free Software + * Foundation. See file COPYING. + * + */ + +#pragma once + +#include +#include + +#include + +#include "common/async/service.h" +#include "common/async/yield_context.h" +#include "common/async/yield_waiter.h" + +namespace ceph::async::detail::call_once { + +class AsyncWaiter : public boost::intrusive::list_base_hook<> { + yield_waiter impl; + public: + void wait(std::unique_lock& lock, + boost::asio::yield_context yield) + { + impl.async_wait(lock, yield); + } + void wake() + { + // async_wait() may complete inline via boost::asio::dispatch(), but would + // immediately try to reacquire the lock we're already holding. defer the + // completion until after we've returned and dropped our lock + boost::asio::defer(impl.get_executor(), [this] { + impl.complete(boost::system::error_code{}); + }); + } + void destroy() + { + impl.shutdown(); + } +}; + +class SyncWaiter : public boost::intrusive::list_base_hook<> { + std::condition_variable cond; + bool done = false; + public: + void wait(std::unique_lock& lock) + { + cond.wait(lock, [this] { return done; }); + } + void wake() + { + done = true; + cond.notify_one(); + } +}; + +struct WaitState : service_list_base_hook { + service* svc = nullptr; + boost::intrusive::list sync_waiters; + boost::intrusive::list async_waiters; + + ~WaitState() + { + if (svc) { + svc->remove(*this); + } + } + + void wait(std::unique_lock& lock, optional_yield y) + { + if (y) { + auto yield = y.get_yield_context(); + if (!svc) { + // on first async wait, register for service_shutdown() notifications + svc = &boost::asio::use_service>( + boost::asio::query(yield.get_executor(), + boost::asio::execution::context)); + svc->add(*this); + } + + AsyncWaiter waiter; + async_waiters.push_back(waiter); + + waiter.wait(lock, yield); + } else { + SyncWaiter waiter; + sync_waiters.push_back(waiter); + + waiter.wait(lock); + } + } + + void wake_all() + { + while (!sync_waiters.empty()) { + auto& waiter = sync_waiters.front(); + sync_waiters.pop_front(); + waiter.wake(); + } + while (!async_waiters.empty()) { + auto& waiter = async_waiters.front(); + async_waiters.pop_front(); + waiter.wake(); + } + } + + // on shutdown, destroy async waiters that didn't complete. this breaks any + // ownership cycles between once_result and the completion handlers + void service_shutdown() + { + while (!async_waiters.empty()) { + auto& waiter = async_waiters.front(); + async_waiters.pop_front(); + waiter.destroy(); + } + } +}; + +} // namespace ceph::async::detail::call_once diff --git a/src/common/async/service.h b/src/common/async/service.h new file mode 100644 index 000000000000..f611e7c81298 --- /dev/null +++ b/src/common/async/service.h @@ -0,0 +1,75 @@ +// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*- +// vim: ts=8 sw=2 smarttab +/* + * Ceph - scalable distributed file system + * + * Copyright contributors to the Ceph project + * + * This is free software; you can redistribute it and/or + * modify it under the terms of the GNU Lesser General Public + * License version 2.1, as published by the Free Software + * Foundation. See file COPYING. + * + */ + +#pragma once + +#include +#include +#include + +namespace ceph::async { + +struct service_tag {}; +using service_list_base_hook = boost::intrusive::list_base_hook< + boost::intrusive::tag>; + +/// Service for two-phase execution_context shutdown, which breaks ownership +/// cycles between completion handlers and their io objects. Tracks objects +/// which may have outstanding completion handlers, and calls their member +/// function service_shutdown() when the execution_context is shutting down. +/// This member function should destroy any memory associated with its +/// outstanding completion handlers. +/// +/// Requirements for IoObject: +/// * Inherits publicly from service_list_base_hook +/// * Has public member function service_shutdown() +/// * Calls add(*this) on construction and remove(*this) on destruction. +template +class service : public boost::asio::execution_context::service { + using base_hook = boost::intrusive::base_hook; + boost::intrusive::list entries; + std::mutex mutex; + + /// Called by the execution_context on shutdown + void shutdown() override { + while (!entries.empty()) { + auto& entry = entries.front(); + entries.pop_front(); + entry.service_shutdown(); + } + } + public: + using key_type = service; + static inline boost::asio::execution_context::id id; + + explicit service(boost::asio::execution_context& ctx) + : boost::asio::execution_context::service(ctx) {} + + /// Register an io object for notification of service_shutdown() + void add(IoObject& entry) { + auto lock = std::scoped_lock{mutex}; + entries.push_back(entry); + } + /// Unregister an object + void remove(IoObject& entry) { + auto lock = std::scoped_lock{mutex}; + if (entries.empty()) { + // already shut down + } else { + entries.erase(entries.iterator_to(entry)); + } + } +}; + +} // namespace ceph::async diff --git a/src/common/async/yield_waiter.h b/src/common/async/yield_waiter.h index 9c14d9bafe4b..667dbbbb2521 100644 --- a/src/common/async/yield_waiter.h +++ b/src/common/async/yield_waiter.h @@ -16,15 +16,40 @@ #pragma once #include +#include #include #include #include +#include #include #include #include namespace ceph::async { +namespace detail { + +// handler wrapper that reacquires a lock immediately before completion +template +struct lock_handler { + Handler handler; + BasicLockable* lock = nullptr; + + template + void operator()(Args&& ...args) { + if (lock) { + lock->lock(); + } + std::move(handler)(std::forward(args)...); + } +}; + +// deduction guide required by windows? +template +lock_handler(Handler&&, BasicLockable*) -> lock_handler; + +} // namespace detail + /// Captures a yield_context handler for deferred completion or cancellation. template class yield_waiter { @@ -41,6 +66,18 @@ class yield_waiter { /// Returns true if there's a handler awaiting completion. operator bool() const { return state.has_value(); } + /// Returns false if there's a handler awaiting completion. + bool empty() const { return !state.has_value(); } + + /// Returns the captured completion handler's executor. + /// Precondition: !empty() + using executor_type = boost::asio::any_io_executor; + executor_type get_executor() const + { + return boost::asio::get_associated_executor( + state->handler, state->work.get_executor()); + } + /// Suspends the given yield_context until the captured handler is invoked /// via complete() or cancel(). template @@ -52,7 +89,25 @@ class yield_waiter { if (slot.is_connected()) { slot.template emplace(this); } - state.emplace(std::move(h)); + constexpr std::unique_lock* lock = nullptr; + state.emplace(std::move(h), lock); + }, token); + } + + /// Suspends the given yield_context until the captured handler is invoked + /// via complete() or cancel(). The given lock is released immediately before + /// it suspends and reacquired immediately after it resumes. + template + auto async_wait(std::unique_lock& lock, CompletionToken&& token) + { + return boost::asio::async_initiate( + [this, &lock] (handler_type h) { + auto slot = get_associated_cancellation_slot(h); + if (slot.is_connected()) { + slot.template emplace(this); + } + state.emplace(std::move(h), &lock); + lock.unlock(); // unlock before suspend }, token); } @@ -61,8 +116,10 @@ class yield_waiter { { auto s = std::move(*state); state.reset(); - auto h = boost::asio::append(std::move(s.handler), ec, std::move(value)); - boost::asio::dispatch(std::move(h)); + boost::asio::dispatch( + boost::asio::append( + detail::lock_handler{std::move(s.handler), s.lock}, + ec, std::move(value))); } /// Destroy the completion handler. @@ -80,10 +137,10 @@ class yield_waiter { struct handler_state { handler_type handler; work_guard work; + std::unique_lock* lock = nullptr; - explicit handler_state(handler_type&& h) - : handler(std::move(h)), - work(make_work_guard(handler)) + handler_state(handler_type&& h, std::unique_lock* lock) + : handler(std::move(h)), work(make_work_guard(handler)), lock(lock) {} }; std::optional state; @@ -123,6 +180,18 @@ class yield_waiter { /// Returns true if there's a handler awaiting completion. operator bool() const { return state.has_value(); } + /// Returns false if there's a handler awaiting completion. + bool empty() const { return !state.has_value(); } + + /// Returns the captured completion handler's executor. + /// Precondition: !empty() + using executor_type = boost::asio::any_io_executor; + executor_type get_executor() const + { + return boost::asio::get_associated_executor( + state->handler, state->work.get_executor()); + } + /// Suspends the given yield_context until the captured handler is invoked /// via complete() or cancel(). template @@ -134,7 +203,25 @@ class yield_waiter { if (slot.is_connected()) { slot.template emplace(this); } - state.emplace(std::move(h)); + constexpr std::unique_lock* lock = nullptr; + state.emplace(std::move(h), lock); + }, token); + } + + /// Suspends the given yield_context until the captured handler is invoked + /// via complete() or cancel(). The given lock is released immediately before + /// it suspends and reacquired immediately after it resumes. + template + auto async_wait(std::unique_lock& lock, CompletionToken&& token) + { + return boost::asio::async_initiate( + [this, &lock] (handler_type h) { + auto slot = get_associated_cancellation_slot(h); + if (slot.is_connected()) { + slot.template emplace(this); + } + state.emplace(std::move(h), &lock); + lock.unlock(); // unlock before suspend }, token); } @@ -143,7 +230,9 @@ class yield_waiter { { auto s = std::move(*state); state.reset(); - boost::asio::dispatch(boost::asio::append(std::move(s.handler), ec)); + boost::asio::dispatch( + boost::asio::append( + detail::lock_handler{std::move(s.handler), s.lock}, ec)); } /// Destroy the completion handler. @@ -161,10 +250,10 @@ class yield_waiter { struct handler_state { handler_type handler; work_guard work; + std::unique_lock* lock = nullptr; - explicit handler_state(handler_type&& h) - : handler(std::move(h)), - work(make_work_guard(handler)) + handler_state(handler_type&& h, std::unique_lock* lock) + : handler(std::move(h)), work(make_work_guard(handler)), lock(lock) {} }; std::optional state; @@ -189,3 +278,23 @@ class yield_waiter { }; } // namespace ceph::async + +namespace boost::asio { + +// forward the handler's associated executor, allocator, cancellation slot, etc +template