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

.github/workflows/merge_build.yml

@@ -11,6 +11,7 @@ on:
 
 jobs:
   GccThreadSanitize:
+    if: ${{ false }}
     uses: ./.github/workflows/build_dependencies.yml
     with:
       platform: "ubuntu-24.04"

conanfile.py

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

include/sisl/async/coro.hpp

@@ -22,34 +22,91 @@
 
 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, exec::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&& task, std::chrono::milliseconds timeout) {
+    auto done = std::make_shared< std::promise< void > >();
+    auto fut = done->get_future();
+    start_coro([](std::decay_t< Task > t, std::shared_ptr< std::promise< void > > d) -> task< void > {
+        try {
+            co_await std::move(t);
+        } catch (...) {}
+        d->set_value();
+    }(std::forward< Task >(task), 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 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{}}));
+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 > task) {
+    start_coro(detach_wrapper< T >(std::move(task)));
+}
+
+// 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 > task, Fn fn) {
+    start_coro([](task< T > t, Fn f) -> task< void > { f(co_await std::move(t)); }(std::move(task), 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

src/fds/CMakeLists.txt

@@ -88,6 +88,13 @@ if(BUILD_TESTING)
     target_link_libraries(test_idreserver sisl_buffer GTest::gtest)
     add_test(NAME IdReserver COMMAND test_idreserver)
 
+    add_executable(test_bounded_mpmc_queue)
+    target_sources(test_bounded_mpmc_queue PRIVATE
+      tests/test_bounded_mpmc_queue.cpp
+      )
+    target_link_libraries(test_bounded_mpmc_queue sisl_buffer GTest::gtest)
+    add_test(NAME BoundedMPMCQueue COMMAND test_bounded_mpmc_queue)
+
 
     if (DEFINED MALLOC_IMPL)
       if (${MALLOC_IMPL} STREQUAL "jemalloc")

src/fds/tests/test_bounded_mpmc_queue.cpp

@@ -0,0 +1,149 @@
+/*********************************************************************************
+ * 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.
+ *
+ *********************************************************************************/
+#include <atomic>
+#include <cstdint>
+#include <thread>
+#include <vector>
+
+#include <sisl/logging/logging.h>
+#include <sisl/options/options.h>
+
+#include <gtest/gtest.h>
+
+#include <sisl/fds/bounded_mpmc_queue.hpp>
+
+using namespace sisl;
+
+SISL_OPTIONS_ENABLE(logging, test_bounded_mpmc_queue)
+SISL_OPTION_GROUP(test_bounded_mpmc_queue,
+                  (num_threads, "", "num_threads", "number of producer/consumer threads",
+                   ::cxxopts::value< uint32_t >()->default_value("4"), "number"),
+                  (num_entries, "", "num_entries", "number of entries per producer thread",
+                   ::cxxopts::value< uint32_t >()->default_value("5000"), "number"))
+
+TEST(BoundedMPMCQueueTest, WriteAndReadSingleValue) {
+    BoundedMPMCQueue< int > q{4};
+    EXPECT_EQ(q.sizeGuess(), 0u);
+
+    EXPECT_TRUE(q.write(42));
+    EXPECT_EQ(q.sizeGuess(), 1u);
+
+    int out{0};
+    EXPECT_TRUE(q.read(out));
+    EXPECT_EQ(out, 42);
+    EXPECT_EQ(q.sizeGuess(), 0u);
+}
+
+TEST(BoundedMPMCQueueTest, ReadFailsWhenEmpty) {
+    BoundedMPMCQueue< int > q{4};
+    int out{0};
+    EXPECT_FALSE(q.read(out));
+}
+
+TEST(BoundedMPMCQueueTest, WriteFailsWhenFull) {
+    constexpr size_t capacity{4};
+    BoundedMPMCQueue< int > q{capacity};
+
+    for (size_t i{0}; i < capacity; ++i) {
+        EXPECT_TRUE(q.write(static_cast< int >(i)));
+    }
+    EXPECT_EQ(q.sizeGuess(), capacity);
+    EXPECT_FALSE(q.write(999));
+    EXPECT_EQ(q.sizeGuess(), capacity);
+
+    int out{0};
+    EXPECT_TRUE(q.read(out));
+    EXPECT_EQ(out, 0);
+    EXPECT_TRUE(q.write(999));
+    EXPECT_EQ(q.sizeGuess(), capacity);
+}
+
+TEST(BoundedMPMCQueueTest, PreservesFifoOrder) {
+    BoundedMPMCQueue< int > q{8};
+    for (int i{0}; i < 8; ++i) {
+        EXPECT_TRUE(q.write(i));
+    }
+
+    for (int i{0}; i < 8; ++i) {
+        int out{-1};
+        EXPECT_TRUE(q.read(out));
+        EXPECT_EQ(out, i);
+    }
+}
+
+TEST(BoundedMPMCQueueTest, ConcurrentMultiProducerMultiConsumer) {
+    auto const num_threads = SISL_OPTIONS["num_threads"].as< uint32_t >();
+    auto const num_entries = SISL_OPTIONS["num_entries"].as< uint32_t >();
+    auto const total_entries = num_threads * num_entries;
+
+    BoundedMPMCQueue< uint64_t > q{16};
+    std::vector< std::atomic< bool > > received(total_entries);
+    for (auto& r : received) {
+        r.store(false);
+    }
+    std::atomic< uint32_t > consumed_count{0};
+
+    std::vector< std::thread > producers;
+    for (uint32_t t{0}; t < num_threads; ++t) {
+        producers.emplace_back([&q, t, num_entries]() {
+            for (uint32_t i{0}; i < num_entries; ++i) {
+                const uint64_t value{static_cast< uint64_t >(t) * num_entries + i};
+                while (!q.write(value)) {
+                    std::this_thread::yield();
+                }
+            }
+        });
+    }
+
+    std::vector< std::thread > consumers;
+    for (uint32_t t{0}; t < num_threads; ++t) {
+        consumers.emplace_back([&q, &received, &consumed_count, total_entries]() {
+            uint64_t value{0};
+            while (consumed_count.load(std::memory_order_relaxed) < total_entries) {
+                if (q.read(value)) {
+                    ASSERT_LT(value, total_entries);
+                    ASSERT_FALSE(received[value].exchange(true));
+                    consumed_count.fetch_add(1, std::memory_order_relaxed);
+                } else {
+                    std::this_thread::yield();
+                }
+            }
+        });
+    }
+
+    for (auto& thr : producers) {
+        thr.join();
+    }
+    for (auto& thr : consumers) {
+        thr.join();
+    }
+
+    EXPECT_EQ(consumed_count.load(), total_entries);
+    EXPECT_EQ(q.sizeGuess(), 0u);
+    for (const auto& r : received) {
+        EXPECT_TRUE(r.load());
+    }
+}
+
+int main(int argc, char* argv[]) {
+    int parsed_argc{argc};
+    ::testing::InitGoogleTest(&parsed_argc, argv);
+    SISL_OPTIONS_LOAD(parsed_argc, argv, logging, test_bounded_mpmc_queue);
+
+    sisl::logging::SetLogger("test_bounded_mpmc_queue");
+    spdlog::set_pattern("[%D %T%z] [%^%l%$] [%n] [%t] %v");
+
+    return RUN_ALL_TESTS();
+}