SDSTOR-25631: move some common headers from homestore into sisl

conanfile.py

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

include/sisl/async/coro.hpp

@@ -10,46 +10,107 @@
 //
 // Requires stdexec on the include path (same opt-in as <sisl/async/task.hpp>).
 
+#include <chrono>
 #include <exception>
+#include <future>
 #include <tuple>
 #include <utility>
 
-#include <exec/inline_scheduler.hpp>
 #include <stdexec/execution.hpp>
 
+#include <sisl/async/shared_awaitable.hpp>
 #include <sisl/async/task.hpp>
+#include <sisl/async/value_awaitable.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.
+// Return a task<T> that awaits a heap-held completion. Used by the "do a synchronous side effect, then return a
+// task that awaits the eventual completion" pattern (e.g. CP trigger / open_log_store: the switchover or map
+// insert must run at call time, only the wait is deferred). The shared_ptr is copied into the task frame so the
+// completion outlives the await.
+template < typename T >
+inline task< T > await_shared(std::shared_ptr< shared_awaitable< T > > aw) {
+    co_return co_await *aw;
+}
+template < typename T >
+inline task< T > await_value(std::shared_ptr< value_awaitable< T > > aw) {
+    co_return co_await *aw;
+}
+
+// Await a value_awaitable held by reference (e.g. a long-lived member of a heap object that outlives the await).
+// The reference is bound into the returned task's frame; the awaitable must stay alive until this task completes.
+template < typename T >
+inline task< T > await_value_ref(value_awaitable< T >& aw) {
+    co_return co_await aw;
+}
+
+// write_env injects an inline scheduler so the sticky-affinity exec::task can be started without an enclosing
+// scheduler context (it resumes inline on whatever thread completes its awaited sender); start_detached owns the
+// operation-state on the heap and frees it on completion. Same pattern as iomgr's io_launch.hpp and sisl's
+// when_all.
+template < typename Task >
+inline void start_coro(Task&& t) {
+    stdexec::start_detached(stdexec::write_env(std::forward< Task >(t),
+                                               stdexec::prop{stdexec::get_scheduler, stdexec::inline_scheduler{}}));
+}
+
+// Block the calling thread until the task completes and return its value. For the infrequent control-plane and
+// shutdown paths that are synchronous today (e.g. a forced CP flush awaited before proceeding). The task is
+// fulfilled by other threads, so this drains its run_loop here without self-deadlocking a data path.
 template < typename Task >
 inline auto sync_get(Task&& task) {
     auto result = stdexec::sync_wait(std::forward< Task >(task)).value();
+    // result is a tuple of the task's completion values; for task<void> it is empty (nothing to return).
     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
+// Block the calling thread until the task completes OR `timeout` elapses; returns true iff it completed in time.
+// The task is started detached and signals a std::promise on completion, which we time-wait on via its future.
+// On timeout the detached task remains pending -- it must keep alive whatever it awaits (e.g. by holding a strong
+// ref into its frame); we simply stop waiting and leave it to complete (or leak) later. Used by the data-receive
+// timeout path, which then inspects per-item readiness and remediates the stragglers.
+template < typename Task >
+inline bool sync_wait_for(Task&& t, std::chrono::milliseconds timeout) {
+    auto done = std::make_shared< std::promise< void > >();
+    auto fut = done->get_future();
+    start_coro([](std::decay_t< Task > inner, std::shared_ptr< std::promise< void > > d) -> task< void > {
+        try {
+            co_await std::move(inner);
+        } catch (...) {}
+        d->set_value();
+    }(std::forward< Task >(t), std::move(done)));
+    return fut.wait_for(timeout) == std::future_status::ready;
+}
+
+// Take the task BY VALUE so it is copied into the self-owning coroutine frame; a captured-by-reference task
+// would dangle once start_detached returns. Swallows exceptions so a throwing body cannot 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 task< void > detach_wrapper(task< T > t) {
+    try {
+        co_await std::move(t);
+    } catch (const std::exception& e) { LOGERROR("Detached task threw, swallowing: {}", e.what()); } catch (...) {
+        LOGERROR("Detached task threw an unknown exception, swallowing");
+    }
+}
+
+// Fire-and-forget a task whose result is not needed (e.g. a non-forced CP trigger). Starts it detached.
 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{}}));
+    start_coro(detach_wrapper< T >(std::move(t)));
+}
+
+// Fire-and-forget a task but invoke fn(result) when it completes (the non-blocking ".thenValue(cb)" shape).
+// fn runs on whatever thread completes the task. Both task and fn are copied into the self-owning frame.
+template < typename T, typename Fn >
+inline void detach_then(task< T > t, Fn fn) {
+    start_coro(
+        [](task< T > inner, Fn f) -> task< void > { f(co_await std::move(inner)); }(std::move(t), std::move(fn)));
 }
 
 } // namespace sisl::async

include/sisl/fds/bounded_mpmc_queue.hpp

@@ -0,0 +1,64 @@
+/*********************************************************************************
+ * Modifications Copyright 2017-2019 eBay Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *    https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software distributed
+ * under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR
+ * CONDITIONS OF ANY KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations under the License.
+ *
+ *********************************************************************************/
+#pragma once
+
+#include <atomic>
+#include <cstddef>
+
+#include <boost/lockfree/queue.hpp>
+
+namespace sisl {
+
+// Folly-free replacement for folly::MPMCQueue: a capacity-bounded boost::lockfree MPMC queue plus an
+// approximate size counter (folly exposed this as sizeGuess()). The queue is bounded -- write() returns false
+// when full -- which blkalloc relies on (the slab cache spills to the next level on a full level, and bounds
+// total cached free-blocks). We use the default (pointer-freelist) boost::lockfree::queue pre-reserved to
+// `capacity` and push only via bounded_push() (never grows past the reserved nodes): this gives bounded
+// behavior WITHOUT boost::lockfree::fixed_sized<true>'s hard 65535-element cap (16-bit freelist indices),
+// which the blkalloc free-block / slab-cache capacities exceed. boost::lockfree::queue requires a
+// trivially-copyable element type (blk_num_t, blk_cache_entry both satisfy this). The size counter is the
+// central accounting a bounded+queryable queue inherently needs; boost::lockfree itself keeps no size.
+template < typename T >
+class BoundedMPMCQueue {
+public:
+    explicit BoundedMPMCQueue(const size_t capacity) : m_q{capacity} {}
+
+    // Non-blocking enqueue; returns false if the queue is full. (folly::MPMCQueue::write)
+    bool write(const T& value) {
+        if (m_q.bounded_push(value)) {
+            m_size.fetch_add(1, std::memory_order_relaxed);
+            return true;
+        }
+        return false;
+    }
+
+    // Non-blocking dequeue; returns false if the queue is empty. (folly::MPMCQueue::read)
+    bool read(T& out_value) {
+        if (m_q.pop(out_value)) {
+            m_size.fetch_sub(1, std::memory_order_relaxed);
+            return true;
+        }
+        return false;
+    }
+
+    // Approximate number of elements (racy under concurrency, like folly's). (folly::MPMCQueue::sizeGuess)
+    size_t sizeGuess() const { return m_size.load(std::memory_order_relaxed); }
+
+private:
+    boost::lockfree::queue< T > m_q;
+    std::atomic< size_t > m_size{0};
+};
+
+} // namespace sisl

include/sisl/fds/lru_map.hpp

@@ -0,0 +1,73 @@
+/*********************************************************************************
+ * Modifications Copyright 2017-2019 eBay Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *    https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software distributed
+ * under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR
+ * CONDITIONS OF ANY KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations under the License.
+ *
+ *********************************************************************************/
+#pragma once
+
+#include <cstddef>
+#include <list>
+#include <unordered_map>
+#include <utility>
+
+namespace sisl {
+
+// Tiny bounded cache used by simple in-memory LRU scenarios.
+//
+// The implementation keeps the most recently touched item at the front of `order_` and
+// uses `index_` to map a key to the corresponding list iterator. `std::size_t` is needed
+// for the cache capacity, while `std::pair` and `std::move` are required to represent each
+// cached element and move replacement values into the list without copying.
+template < typename Key, typename Value >
+class LruMap {
+public:
+    using value_type = std::pair< Key, Value >;
+    using list_type = std::list< value_type >;
+    using iterator = typename list_type::iterator;
+    using const_iterator = typename list_type::const_iterator;
+
+    explicit LruMap(std::size_t capacity) : capacity_{capacity} {}
+
+    // Inserts or updates a key. Updating an existing key refreshes the value and moves it to
+    // the front to preserve the recency ordering. When the cache exceeds its capacity, the
+    // least recently used item is evicted from both `order_` and `index_`.
+    void set(Key const& key, Value value) {
+        if (auto it = index_.find(key); it != index_.end()) {
+            it->second->second = std::move(value);
+            order_.splice(order_.begin(), order_, it->second);
+            return;
+        }
+        order_.emplace_front(key, std::move(value));
+        index_[order_.front().first] = order_.begin();
+        while (order_.size() > capacity_) {
+            index_.erase(order_.back().first);
+            order_.pop_back();
+        }
+    }
+
+    // Missing keys return a default-constructed Value (shared_ptr -> nullptr).
+    Value get(Key const& key) const {
+        auto it = index_.find(key);
+        if (it == index_.end()) { return Value{}; }
+        return it->second->second;
+    }
+
+    const_iterator begin() const { return order_.begin(); }
+    const_iterator end() const { return order_.end(); }
+
+private:
+    std::size_t capacity_;
+    list_type order_;
+    std::unordered_map< Key, iterator > index_;
+};
+
+} // namespace sisl