Sisl::Async::Result, when_quorum and sync_get

CHANGELOG.md

@@ -4,6 +4,14 @@ All notable changes to this project will be documented in this file.
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
+## [14.7.0] - 2026-06-04
+- Added new generic sisl::result generic type
+- Added new generic when_quorum async reactor
+
+## [14.6.1]
+
+## [14.6.0]
+
 ## [14.5.0] - 2026-06-04
 - Add deregister_generic_rpc()
 

conanfile.py

@@ -10,7 +10,7 @@
 
 class SISLConan(ConanFile):
     name = "sisl"
-    version = "14.6.1"
+    version = "14.7.0"
 
     homepage = "https://github.com/eBay/sisl"
     description = "Library for fast data structures, utilities"

include/sisl/async/coro.hpp

@@ -0,0 +1,55 @@
+#pragma once
+
+// The two bridges between synchronous, non-coroutine code and the stdexec / sisl::async::task world:
+//   - sync_get(task) : block the calling thread until the task completes and return its value (void for
+//                      task<void>). The task is fulfilled by other threads; sync_wait drains a run_loop here.
+//                      Safe only OFF an event-loop / reactor thread (a test main, a synchronous control path);
+//                      blocking a reactor with it would park that loop.
+//   - detach(task)   : fire-and-forget a coroutine from a non-coroutine context (a void callback, a timer).
+//                      exec::task is lazy, so an un-awaited task never runs -- this starts it.
+//
+// Requires stdexec on the include path (same opt-in as <sisl/async/task.hpp>).
+
+#include <exception>
+#include <tuple>
+#include <utility>
+
+#include <exec/inline_scheduler.hpp>
+#include <stdexec/execution.hpp>
+
+#include <sisl/async/task.hpp>
+#include <sisl/logging/logging.h>
+
+namespace sisl::async {
+
+// Block the calling thread until the task completes and return its value (void for task<void>). Do NOT call on
+// an event-loop / reactor thread.
+template < typename Task >
+inline auto sync_get(Task&& task) {
+    auto result = stdexec::sync_wait(std::forward< Task >(task)).value();
+    if constexpr (std::tuple_size_v< decltype(result) > == 0) {
+        return;
+    } else {
+        return std::get< 0 >(std::move(result));
+    }
+}
+
+// Fire-and-forget a coroutine whose result we don't need. The task is taken by value (copied into the
+// self-owning wrapper frame); the wrapper swallows exceptions so a throwing body can't reach start_detached's
+// receiver (which would std::terminate) -- tasks normally complete errors-as-values, so this is a backstop.
+// write_env injects an inline scheduler so the sticky-affinity exec::task can start without an enclosing
+// scheduler (it resumes inline on whatever thread completes its awaited work) -- the same idiom as when_all.
+template < typename T >
+inline void detach(task< T > t) {
+    auto wrapper = [](task< T > inner) -> task< void > {
+        try {
+            co_await std::move(inner);
+        } catch (const std::exception& e) { LOGERROR("Detached task threw, swallowing: {}", e.what()); } catch (...) {
+            LOGERROR("Detached task threw an unknown exception, swallowing");
+        }
+    }(std::move(t));
+    stdexec::start_detached(
+        stdexec::write_env(std::move(wrapper), stdexec::prop{stdexec::get_scheduler, exec::inline_scheduler{}}));
+}
+
+} // namespace sisl::async

include/sisl/async/result.hpp

@@ -0,0 +1,28 @@
+#pragma once
+
+// The asynchronous form of sisl::result: a coroutine you co_await that resolves to a sisl::result<T>. It layers
+// the sisl::async::task carrier (exec::task / stdexec) over the pure-std sisl::result, so it lives beside
+// task.hpp and inherits the SAME opt-in stdexec dependency -- consumers without stdexec use <sisl/result.hpp>
+// directly. Because sisl::async::result<T> IS sisl::async::task<sisl::result<T>>, it composes with
+// stdexec::when_all / when_any and co_awaits from any task, which is what lets higher layers gather async
+// operations on one coroutine model.
+//
+// It sits in sisl::async (beside task) because that is the coroutine-carried world; the plain sisl::result is
+// top-level vocabulary. Splitting them by namespace mirrors the header split. A consumer that wants a flat
+// `async_result` name exposes its own alias: `template < class T > using async_result = sisl::async::result< T >;`.
+
+#include <variant>
+
+#include <sisl/async/task.hpp>
+#include <sisl/result.hpp>
+
+namespace sisl::async {
+
+// Operational result of an asynchronous call: a coroutine that resolves to a sisl::result<T>. `sisl::result` is
+// qualified deliberately -- unqualified `result` here would name THIS alias and self-refer.
+template < typename T >
+using result = task< sisl::result< T > >;
+
+using status = result< std::monostate >;
+
+} // namespace sisl::async

include/sisl/async/when_quorum.hpp

@@ -0,0 +1,120 @@
+#pragma once
+
+// when_quorum: fan out over a runtime vector of tasks and resume as soon as `quorum` of them SUCCEED (complete
+// with a value), leaving the stragglers running detached. The k-of-n sibling of sisl::async::when_all (same
+// start_detached + value_awaitable latch), differing only in when the latch fires. The canonical use is
+// quorum-acknowledged replication -- acknowledge at a majority instead of waiting for the slowest peer -- but
+// the combinator is domain-agnostic.
+//
+// THE PAYLOAD CONTRACT. Children keep running after we resume, so a child must not reference a caller-owned
+// buffer past its first suspension point -- by then the caller may have recycled it. Note the asymmetry that
+// makes early return sound: the caller's buffer must survive every child's *send/first-suspension*, but we only
+// wait for a quorum of *completions*. Every child runs synchronously to its first suspension inside the fan-out
+// loop below, so all first suspensions are reached before we ever await the latch.
+//
+// THE RESULTS RACE, AND WHY IT IS NOT ONE. The latch fires on exactly two triggers:
+//
+//     (a) `quorum` children have succeeded  -- stragglers may still be writing into `results`
+//     (b) every child has finished          -- nothing is writing into `results`
+//
+// So `results` is safe to read IF AND ONLY IF fewer than `quorum` successes came back: (a) cannot have fired,
+// therefore (b) did. Do not read `results` without checking `acks < quorum` first.
+
+#include <atomic>
+#include <cstddef>
+#include <functional>
+#include <memory>
+#include <utility>
+#include <variant>
+#include <vector>
+
+#include <exec/inline_scheduler.hpp>
+#include <stdexec/execution.hpp>
+
+#include <sisl/async/task.hpp>
+#include <sisl/async/value_awaitable.hpp>
+
+namespace sisl::async {
+
+namespace detail {
+
+// Fires once, on whichever trigger lands first. `fired_` guards value_awaitable's exactly-once contract.
+struct quorum_latch {
+    std::atomic< std::size_t > remaining;
+    std::atomic< std::size_t > acks{0};
+    std::atomic< bool > fired{false};
+    std::size_t const quorum;
+    value_awaitable< std::monostate > done;
+
+    quorum_latch(std::size_t n, std::size_t q) noexcept : remaining{n}, quorum{q} {}
+
+    void arrive(bool acked) noexcept {
+        // Order matters: publish the ack BEFORE the completion, so a waiter woken by the last child's
+        // count_down observes every ack that preceded it.
+        if (acked && (acks.fetch_add(1, std::memory_order_acq_rel) + 1 == quorum)) fire();
+        if (remaining.fetch_sub(1, std::memory_order_acq_rel) == 1) fire();
+    }
+    void fire() noexcept {
+        if (!fired.exchange(true, std::memory_order_acq_rel)) done.complete({});
+    }
+};
+
+// Called once per child as it completes, on whatever thread completed it (stragglers included), with the
+// child's fan-out index and whether it succeeded. Keyed on the local `acked` flag, NOT on results[index]: a
+// throwing child leaves results[index] default-constructed, which for e.g. expected<void> reads has_value() ==
+// true, a phantom success the hook must never publish.
+using completion_hook = std::function< void(std::size_t index, bool acked) >;
+
+// Awaits one child, records its result, and arrives at the latch. Swallows exceptions (a throwing child must
+// not strand the latch), leaving that slot default-constructed and counted as a non-success.
+template < typename T >
+task< void > quorum_run_one(task< T > child, std::shared_ptr< std::vector< T > > results, std::size_t index,
+                            std::shared_ptr< quorum_latch > latch, completion_hook on_each) {
+    bool acked = false;
+    try {
+        auto r = co_await std::move(child);
+        acked = r.has_value();
+        (*results)[index] = std::move(r);
+    } catch (...) {
+        // leave results[index] default-constructed; counted as a non-success
+    }
+    // Publish evidence BEFORE arriving: on the quorum-th success, arrive() fires the latch and the awaiter
+    // resumes, so a hook recording here is visible to that awaiter before it observes the result.
+    if (on_each) on_each(index, acked);
+    latch->arrive(acked);
+}
+
+} // namespace detail
+
+template < typename T >
+struct quorum_result {
+    std::size_t acks{0};
+    // Readable ONLY when acks < quorum (see the header comment): then every child has finished.
+    std::shared_ptr< std::vector< T > > results;
+};
+
+// Start every task concurrently; resume once `quorum` of them have completed with a value, or once all have
+// finished, whichever comes first. Stragglers keep running detached and keep `results` alive. `on_each`, if set,
+// fires once per child as it completes (see completion_hook); default {} leaves behavior unchanged for callers
+// that do not observe per-child outcomes.
+template < typename T >
+task< quorum_result< T > > when_quorum(std::vector< task< T > > tasks, std::size_t quorum,
+                                       detail::completion_hook on_each = {}) {
+    auto const n = tasks.size();
+    auto results = std::make_shared< std::vector< T > >(n);
+    if (n == 0) co_return quorum_result< T >{0, std::move(results)};
+
+    auto latch = std::make_shared< detail::quorum_latch >(n, quorum);
+    for (std::size_t i = 0; i < n; ++i) {
+        // Each child runs synchronously to its first suspension right here, which is where it consumes the
+        // caller's payload. exec::task has sticky scheduler affinity, so inject inline_scheduler to let the
+        // detached child resume on whatever thread completes it (identical to sisl::async::when_all).
+        stdexec::start_detached(
+            stdexec::write_env(detail::quorum_run_one< T >(std::move(tasks[i]), results, i, latch, on_each),
+                               stdexec::prop{stdexec::get_scheduler, exec::inline_scheduler{}}));
+    }
+    co_await latch->done;
+    co_return quorum_result< T >{latch->acks.load(std::memory_order_acquire), std::move(results)};
+}
+
+} // namespace sisl::async

include/sisl/result.hpp

@@ -0,0 +1,38 @@
+#pragma once
+
+// The stack-wide operational-result vocabulary. Every operation reports success-or-failure the SAME way: it
+// returns a value or a std::error_condition. sisl::result<T> is that value-or-condition. The specific reason
+// rides in the std::error_condition, whose category-registered domain enum -- homestore's ReplServiceError,
+// iomgr's codes, CRAFT's craft_error, ... -- the caller branches on:
+//
+//     auto r = repl_dev->create(...);
+//     if (!r) {
+//         if (r.error() == ReplServiceError::NOT_LEADER) { /* redirect */ }
+//         else { LOGERROR("create failed: {}", r.error().message()); }
+//     }
+//
+// This header stays pure-std: it names NO domain, so any layer's error enum flows through result<T> without
+// sisl knowing it. That type erasure is exactly what lets a leaf library (a wire codec, a client) speak the
+// same result vocabulary as homestore without depending on it. The asynchronous form (sisl::async_result<T>,
+// a co_await-able task) layers stdexec on top in <sisl/async/result.hpp>; keep this one dependency-free so
+// non-async and stdexec-free consumers can still use it.
+//
+// Reserve exceptions for precondition violations that signal a caller bug, never for operational failures
+// that are part of the contract.
+
+#include <expected>
+#include <system_error>
+#include <variant>
+
+namespace sisl {
+
+// Operational result of a synchronous call: a value, or a std::error_condition describing the failure.
+template < typename T >
+using result = std::expected< T, std::error_condition >;
+
+// A result that carries no value on success (just success/failure).
+using status = result< std::monostate >;
+
+inline status ok() noexcept { return status{std::monostate{}}; }
+
+} // namespace sisl

src/async/tests/test_manual_scheduler.cpp

@@ -17,8 +17,8 @@ namespace {
 
 TEST(manual_scheduler, step_below_deadline_does_not_fire) {
     sisl::async::manual_scheduler sched;
-    bool                          fired  = false;
-    auto                          runner = [&]() -> exec::task< void > {
+    bool fired = false;
+    auto runner = [&]() -> exec::task< void > {
         co_await sched.schedule_at(50'000'000);
         fired = true;
     };
@@ -34,8 +34,8 @@ TEST(manual_scheduler, step_below_deadline_does_not_fire) {
 
 TEST(manual_scheduler, step_past_deadline_fires_in_one_drain) {
     sisl::async::manual_scheduler sched;
-    bool                          fired  = false;
-    auto                          runner = [&]() -> exec::task< void > {
+    bool fired = false;
+    auto runner = [&]() -> exec::task< void > {
         co_await sched.schedule_at(50'000'000);
         fired = true;
     };
@@ -50,7 +50,7 @@ TEST(manual_scheduler, step_past_deadline_fires_in_one_drain) {
 
 TEST(manual_scheduler, multiple_timers_fire_in_deadline_order) {
     sisl::async::manual_scheduler sched;
-    std::vector< int >            order;
+    std::vector< int > order;
 
     auto schedule_one = [&](int label, uint64_t wall_ns) -> exec::task< void > {
         co_await sched.schedule_at(wall_ns);
@@ -82,7 +82,7 @@ TEST(manual_scheduler, deadline_already_past_fires_immediately) {
     sisl::async::manual_scheduler sched;
     sched.step(std::chrono::milliseconds{100}); // now=100ms
 
-    bool fired  = false;
+    bool fired = false;
     auto runner = [&]() -> exec::task< void > {
         // Schedule a deadline that is already in the past.
         co_await sched.schedule_at(50'000'000);

src/fds/tests/test_atomic_status_counter.cpp

@@ -29,7 +29,7 @@ SISL_OPTIONS_ENABLE(logging)
 namespace {
 class AtomicStatusCounterTest : public testing::Test {
 public:
-    AtomicStatusCounterTest() : testing::Test(){};
+    AtomicStatusCounterTest() : testing::Test() {};
     AtomicStatusCounterTest(const AtomicStatusCounterTest&) = delete;
     AtomicStatusCounterTest(AtomicStatusCounterTest&&) noexcept = delete;
     AtomicStatusCounterTest& operator=(const AtomicStatusCounterTest&) = delete;

src/file_watcher/file_watcher_test.cpp

@@ -47,7 +47,7 @@ void monitor_file_changes(FileWatcherTest::FileChangeParams& file_change_params,
         file_change_params.file_str, listener,
         [&file_change_params, listener](const std::string filepath, const bool deleted) {
             EXPECT_EQ(file_change_params.file_str, filepath);
-	    LOGWARN("CB called with deleted = {}", deleted);
+            LOGWARN("CB called with deleted = {}", deleted);
             {
                 std::lock_guard< std::mutex > lg(file_change_params.file_change_lock);
                 file_change_params.is_deleted = deleted;

src/grpc/rpc_server.cpp

@@ -35,8 +35,8 @@ GrpcServer::GrpcServer(const std::string& listen_addr, uint32_t threads, const s
 GrpcServer::GrpcServer(const std::string& listen_addr, uint32_t threads, const std::string& ssl_key,
                        const std::string& ssl_cert, const std::shared_ptr< sisl::GrpcTokenVerifier >& auth_mgr) :
         GrpcServer::GrpcServer(listen_addr, threads, 0, 0, ssl_key, ssl_cert, auth_mgr) {}
-GrpcServer::GrpcServer(const std::string& listen_addr, uint32_t threads, int max_receive_msg_size, int max_send_msg_size,
-                       const std::string& ssl_key, const std::string& ssl_cert,
+GrpcServer::GrpcServer(const std::string& listen_addr, uint32_t threads, int max_receive_msg_size,
+                       int max_send_msg_size, const std::string& ssl_key, const std::string& ssl_cert,
                        const std::shared_ptr< sisl::GrpcTokenVerifier >& auth_mgr) :
         m_num_threads{threads}, m_auth_mgr{auth_mgr} {
     if (listen_addr.empty() || threads == 0) { throw std::invalid_argument("Invalid parameter to start grpc server"); }
@@ -164,7 +164,9 @@ void GrpcServer::shutdown() {
 
 bool GrpcServer::is_auth_enabled() const { return m_auth_mgr != nullptr; }
 
-grpc::Status GrpcServer::auth_verify(grpc::ServerContext const* srv_ctx) const { return m_auth_mgr->verify_ctx(srv_ctx); }
+grpc::Status GrpcServer::auth_verify(grpc::ServerContext const* srv_ctx) const {
+    return m_auth_mgr->verify_ctx(srv_ctx);
+}
 
 bool GrpcServer::run_generic_handler_cb(const std::string& rpc_name, boost::intrusive_ptr< GenericRpcData >& rpc_data) {
     generic_rpc_handler_cb_t cb;

src/logging/backtrace.h

@@ -52,7 +52,7 @@
 #include <execinfo.h>
 #endif
 
-#if defined(__linux__) 
+#if defined(__linux__)
 #include <linux/limits.h>
 #endif
 
@@ -68,7 +68,6 @@ constexpr size_t address_length{16};
 constexpr uint64_t pipe_timeout_ms{15000}; // 15 seconds.  Addr2line can be extremely slow the first time
 } // namespace backtrace_detail
 
-
 [[maybe_unused]] static size_t stack_backtrace_impl(void** const stack_ptr, const size_t stack_ptr_capacity) {
     return ::backtrace(stack_ptr, static_cast< int >(stack_ptr_capacity));
 }
@@ -94,20 +93,19 @@ constexpr uint64_t pipe_timeout_ms{15000}; // 15 seconds.  Addr2line can be extr
                                                const bool trim_internal) {
 #if defined(__linux__)
     std::unique_ptr< FILE, std::function< void(FILE* const) > > stack_file{std::tmpfile(), [](FILE* const fp) {
-                                                                               if (fp)
-                                                                                   std::fclose(fp);
+                                                                               if (fp) std::fclose(fp);
                                                                            }};
-    if (!stack_file)
-        return 0;
+    if (!stack_file) return 0;
 
     ::backtrace_symbols_fd(stack_ptr, static_cast< int >(stack_size), ::fileno(stack_file.get()));
 
     const size_t len{
         stack_interpret_linux_file(stack_ptr, stack_file.get(), stack_size, output_buf, output_buflen, trim_internal)};
 #else
     const std::unique_ptr< char*, std::function< void(char** const) > > stack_msg{
-        ::backtrace_symbols(stack_ptr, static_cast< int >(stack_size)),
-        [](char** const ptr) { if (ptr) std::free(static_cast< void* >(ptr)); }};
+        ::backtrace_symbols(stack_ptr, static_cast< int >(stack_size)), [](char** const ptr) {
+            if (ptr) std::free(static_cast< void* >(ptr));
+        }};
 #if defined(__APPLE__)
     const size_t len{
         stack_interpret_apple(stack_ptr, stack_msg.get(), stack_size, output_buf, output_buflen, trim_internal)};