Merge pull request #36 from solidoss/relay_multicast

Relay multicast - preparing the stage + threadpool improvements
solidoss · Jun 29, 2024 · fc163bc · fc163bc
2 parents ac26b55 + 0b9d7b8
commit fc163bc
Show file tree

Hide file tree

Showing 106 changed files with 1,673 additions and 1,093 deletions.
diff --git a/CMakeLists.txt b/CMakeLists.txt
@@ -2,7 +2,7 @@ cmake_minimum_required(VERSION 3.13 FATAL_ERROR)
 #-----------------------------------------------------------------
 # The project
 #-----------------------------------------------------------------
-project (SolidFrame VERSION 11.1)
+project (SolidFrame VERSION 12.0)
 
 message("SolidFrame version: ${PROJECT_VERSION} - ${PROJECT_VERSION_MAJOR}.${PROJECT_VERSION_MINOR}")
 #-----------------------------------------------------------------

diff --git a/RELEASES.md b/RELEASES.md
@@ -1,5 +1,11 @@
 # SolidFrame Releases
 
+## Version 12.0
+* utility: support pthread_spin_lock when available
+* utility: ThreadPool improvements and fixes
+* frame: Reactor and aio::Reactor using the same new technique from ThreadPool for event passing
+* mprpc: preparing the stage for relay multicast support
+
 ## Version 11.1
 * mprpc: Split Connection in ClientConnection, ServerConnection, RelayConnection
 * mprpc: Some cleanup and some small improvements on the MessageReader and MessageWriter

diff --git a/cmake/check.config.cmake b/cmake/check.config.cmake
@@ -45,5 +45,9 @@ file (READ "${CMAKE_CURRENT_SOURCE_DIR}/cmake/check/epoll2.cpp" source_code)
 
 CHECK_CXX_SOURCE_RUNS("${source_code}" SOLID_USE_EPOLL2)
 
+file (READ "${CMAKE_CURRENT_SOURCE_DIR}/cmake/check/pthread_spinlock.cpp" source_code)
+
+CHECK_CXX_SOURCE_COMPILES("${source_code}" SOLID_USE_PTHREAD_SPINLOCK)
+
 #TODO:
 #set(SOLID_FRAME_AIO_REACTOR_USE_SPINLOCK TRUE)
diff --git a/cmake/check/pthread_spinlock.cpp b/cmake/check/pthread_spinlock.cpp
@@ -0,0 +1,8 @@
+#include <pthread.h>
+
+int main(){
+    pthread_spinlock_t spl;
+    auto rv = pthread_spin_init(&spl, PTHREAD_PROCESS_PRIVATE);
+    if(rv != 0) return -1;
+    return 0;
+}
diff --git a/examples/frame/aio_echo/example_echo_auto_client.cpp b/examples/frame/aio_echo/example_echo_auto_client.cpp
@@ -196,7 +196,7 @@ int main(int argc, char* argv[])
             1024 * 1024 * 64);
     }
 
-    CallPoolT            cwp{1, 100, 0, [](const size_t) {}, [](const size_t) {}};
+    CallPoolT            cwp{{1, 100, 0}, [](const size_t) {}, [](const size_t) {}};
     frame::aio::Resolver resolver([&cwp](std::function<void()>&& _fnc) { cwp.pushOne(std::move(_fnc)); });
 
     async_resolver(&resolver);

diff --git a/examples/frame/aio_echo/example_secure_echo_client.cpp b/examples/frame/aio_echo/example_secure_echo_client.cpp
@@ -174,7 +174,7 @@ int main(int argc, char* argv[])
         frame::ServiceT service(manager);
         frame::ActorIdT actuid;
 
-        CallPoolT            cwp{1, 100, 0, [](const size_t) {}, [](const size_t) {}};
+        CallPoolT            cwp{{1, 100, 0}, [](const size_t) {}, [](const size_t) {}};
         frame::aio::Resolver resolver([&cwp](std::function<void()>&& _fnc) { cwp.pushOne(std::move(_fnc)); });
         ErrorConditionT      err;
 

diff --git a/examples/frame/mprpc_echo/example_mprpc_echo.cpp b/examples/frame/mprpc_echo/example_mprpc_echo.cpp
@@ -173,7 +173,7 @@ int main(int argc, char* argv[])
         frame::Manager         m;
         frame::mprpc::ServiceT ipcsvc(m);
         ErrorConditionT        err;
-        CallPoolT              cwp{1, 100, 0, [](const size_t) {}, [](const size_t) {}};
+        CallPoolT              cwp{{1, 100, 0}, [](const size_t) {}, [](const size_t) {}};
         frame::aio::Resolver   resolver([&cwp](std::function<void()>&& _fnc) { cwp.pushOne(std::move(_fnc)); });
 
         if (!restart(ipcsvc, resolver, sch)) {
@@ -211,7 +211,7 @@ bool restart(
         reflection::v1::metadata::factory,
         [&](auto& _rmap) {
             _rmap.template registerMessage<FirstMessage>(1, "FirstMessage",
-                [](
+                           [](
                     frame::mprpc::ConnectionContext&             _rctx,
                     frame::mprpc::MessagePointerT<FirstMessage>& _rsend_msg,
                     frame::mprpc::MessagePointerT<FirstMessage>& _rrecv_msg,
@@ -347,7 +347,7 @@ void broadcast_message(frame::mprpc::Service& _rsvc, frame::mprpc::MessagePointe
     solid_log(generic_logger, Verbose, "done stop===============================");
 
     for (Params::StringVectorT::const_iterator it(params.connectstringvec.begin()); it != params.connectstringvec.end(); ++it) {
-        _rsvc.sendMessage(it->c_str(), _rmsgptr, {frame::mprpc::MessageFlagsE::AwaitResponse});
+        _rsvc.sendMessage({*it}, _rmsgptr, {frame::mprpc::MessageFlagsE::AwaitResponse});
     }
 }
 

diff --git a/examples/frame/relay_server/example_relay_server.cpp b/examples/frame/relay_server/example_relay_server.cpp
@@ -217,7 +217,7 @@ int main(int argc, char* argv[])
             3,
             1024 * 1024 * 64);
     }
-    CallPoolT            cwp{1, 100, 0, [](const size_t) {}, [](const size_t) {}};
+    CallPoolT            cwp{{1, 100, 0}, [](const size_t) {}, [](const size_t) {}};
     frame::aio::Resolver resolver([&cwp](std::function<void()>&& _fnc) { cwp.pushOne(std::move(_fnc)); });
 
     async_resolver(&resolver);

diff --git a/examples/frame/relay_server/example_relay_server_bi.cpp b/examples/frame/relay_server/example_relay_server_bi.cpp
@@ -180,7 +180,7 @@ int main(int argc, char* argv[])
             1024 * 1024 * 64);
     }
 
-    CallPoolT            cwp{1, 100, 0, [](const size_t) {}, [](const size_t) {}};
+    CallPoolT            cwp{{1, 100, 0}, [](const size_t) {}, [](const size_t) {}};
     frame::aio::Resolver resolver([&cwp](std::function<void()>&& _fnc) { cwp.pushOne(std::move(_fnc)); });
 
     async_resolver(&resolver);

diff --git a/examples/frame/relay_server/example_relay_server_bi_cp.cpp b/examples/frame/relay_server/example_relay_server_bi_cp.cpp
@@ -181,7 +181,7 @@ int main(int argc, char* argv[])
             3,
             1024 * 1024 * 64);
     }
-    CallPoolT            cwp{1, 100, 0, [](const size_t) {}, [](const size_t) {}};
+    CallPoolT            cwp{{1, 100, 0}, [](const size_t) {}, [](const size_t) {}};
     frame::aio::Resolver resolver([&cwp](std::function<void()>&& _fnc) { cwp.pushOne(std::move(_fnc)); });
 
     async_resolver(&resolver);

diff --git a/examples/frame/relay_server/example_relay_server_bi_ex.cpp b/examples/frame/relay_server/example_relay_server_bi_ex.cpp
@@ -382,7 +382,7 @@ int main(int argc, char* argv[])
 
     cout << "sizeof(Connection) = " << sizeof(Connection) << endl;
 
-    CallPoolT            cwp{1, 100, 0, [](const size_t) {}, [](const size_t) {}};
+    CallPoolT            cwp{{1, 100, 0}, [](const size_t) {}, [](const size_t) {}};
     frame::aio::Resolver resolver([&cwp](std::function<void()>&& _fnc) { cwp.pushOne(std::move(_fnc)); });
 
     async_resolver(&resolver);

diff --git a/examples/frame/relay_server/example_relay_server_bi_sh.cpp b/examples/frame/relay_server/example_relay_server_bi_sh.cpp
@@ -311,7 +311,7 @@ int main(int argc, char* argv[])
 
     cout << "sizeof(Connection) = " << sizeof(Connection) << endl;
 
-    CallPoolT            cwp{1, 100, 0, [](const size_t) {}, [](const size_t) {}};
+    CallPoolT            cwp{{1, 100, 0}, [](const size_t) {}, [](const size_t) {}};
     frame::aio::Resolver resolver([&cwp](std::function<void()>&& _fnc) { cwp.pushOne(std::move(_fnc)); });
 
     async_resolver(&resolver);

diff --git a/examples/utility/threadpool/example_file_open_pool.cpp b/examples/utility/threadpool/example_file_open_pool.cpp
@@ -108,7 +108,7 @@ int main(int argc, char* argv[])
     using ThreadPoolT = ThreadPool<FileDevice*, size_t>;
     Context     context;
     ThreadPoolT wp{
-        1, 100, 0, [](const size_t, Context&) {}, [](const size_t, Context&) {},
+        {1, 100, 0}, [](const size_t, Context&) {}, [](const size_t, Context&) {},
         [](FileDevice* _pfile, Context& _rctx) {
             int64_t sz = _pfile->size();
             int     toread;

diff --git a/examples/utility/threadpool/example_threadpool.cpp b/examples/utility/threadpool/example_threadpool.cpp
@@ -20,7 +20,7 @@ int main(int argc, char* argv[])
     solid::log_start(std::cerr, {".*:VIEW"});
 
     ThreadPool<int, size_t> wp{
-        1, 100, 0, [](const size_t) {}, [](const size_t) {},
+        {1, 100, 0}, [](const size_t) {}, [](const size_t) {},
         [](int _v) {
             solid_log(generic_logger, Info, "v = " << _v);
             std::this_thread::sleep_for(std::chrono::milliseconds(_v * 10));

diff --git a/solid/frame/aio/aioreactor.hpp b/solid/frame/aio/aioreactor.hpp
@@ -10,6 +10,8 @@
 
 #pragma once
 
+#include <tuple>
+
 #include "solid/frame/aio/aiocommon.hpp"
 #include "solid/frame/aio/aioreactorcontext.hpp"
 #include "solid/frame/common.hpp"
@@ -140,9 +142,10 @@ class Reactor : public frame::ReactorBase {
     Reactor(SchedulerBase& _rsched, StatisticT& _rstatistic, const size_t _schedidx, const size_t _wake_capacity);
     ~Reactor();
 
-    size_t pushWakeIndex() noexcept
+    std::tuple<frame::impl::AtomicIndexValueT, frame::impl::AtomicCounterValueT> pushWakeIndex() noexcept
     {
-        return push_wake_index_.fetch_add(1) % wake_capacity_;
+        const auto index = push_wake_index_.fetch_add(1);
+        return {index % wake_capacity_, frame::impl::computeCounter(index, wake_capacity_)};
     }
 
     template <typename Function>
@@ -374,7 +377,7 @@ class Reactor : public impl::Reactor {
 
     Reactor(SchedulerBase& _rsched, StatisticT& _rstatistic, const size_t _sched_idx, const size_t _wake_capacity)
         : impl::Reactor(_rsched, _rstatistic, _sched_idx, _wake_capacity)
-        , wake_arr_(new WakeStubT[_wake_capacity])
+        , wake_arr_(new WakeStubT[wake_capacity_])
     {
     }
 
@@ -385,10 +388,10 @@ class Reactor : public impl::Reactor {
             mutex().lock();
             const UniqueId uid = this->popUid(*_ract);
             mutex().unlock();
-            const auto index = pushWakeIndex();
-            auto&      rstub = wake_arr_[index];
+            const auto [index, count] = pushWakeIndex();
+            auto& rstub               = wake_arr_[index];
 
-            rstub.waitWhilePush(rstatistic_);
+            rstub.waitWhilePush(rstatistic_, count);
 
             rstub.reset(uid, _revent, std::move(_ract), &_rsvc);
 
@@ -414,10 +417,10 @@ class Reactor : public impl::Reactor {
             mutex().lock();
             const UniqueId uid = this->popUid(*_ract);
             mutex().unlock();
-            const auto index = pushWakeIndex();
-            auto&      rstub = wake_arr_[index];
+            const auto [index, count] = pushWakeIndex();
+            auto& rstub               = wake_arr_[index];
 
-            rstub.waitWhilePush(rstatistic_);
+            rstub.waitWhilePush(rstatistic_, count);
 
             rstub.reset(uid, std::move(_revent), std::move(_ract), &_rsvc);
 
@@ -442,10 +445,10 @@ class Reactor : public impl::Reactor {
     {
         bool notify = false;
         {
-            const auto index = pushWakeIndex();
-            auto&      rstub = wake_arr_[index];
+            const auto [index, count] = pushWakeIndex();
+            auto& rstub               = wake_arr_[index];
 
-            rstub.waitWhilePush(rstatistic_);
+            rstub.waitWhilePush(rstatistic_, count);
 
             rstub.reset(_ractuid, _revent);
 
@@ -468,10 +471,10 @@ class Reactor : public impl::Reactor {
     {
         bool notify = false;
         {
-            const auto index = pushWakeIndex();
-            auto&      rstub = wake_arr_[index];
+            const auto [index, count] = pushWakeIndex();
+            auto& rstub               = wake_arr_[index];
 
-            rstub.waitWhilePush(rstatistic_);
+            rstub.waitWhilePush(rstatistic_, count);
 
             rstub.reset(_ractuid, std::move(_revent));
 
@@ -533,7 +536,7 @@ class Reactor : public impl::Reactor {
         while (true) {
             const size_t index = pop_wake_index_ % wake_capacity_;
             auto&        rstub = wake_arr_[index];
-            if (rstub.isFilled()) {
+            if (rstub.isFilled(pop_wake_index_, wake_capacity_)) {
                 if (rstub.actor_ptr_) [[unlikely]] {
                     ++actor_count_;
                     rstatistic_.actorCount(actor_count_);

diff --git a/solid/frame/aio/src/aioreactor.cpp b/solid/frame/aio/src/aioreactor.cpp
@@ -343,7 +343,7 @@ Reactor::Reactor(
     SchedulerBase& _rsched, StatisticT& _rstatistic,
     const size_t _idx, const size_t _wake_capacity)
     : ReactorBase(_rsched, _idx)
-    , wake_capacity_(_wake_capacity)
+    , wake_capacity_(std::bit_ceil(_wake_capacity))
     , rstatistic_(_rstatistic)
 {
     solid_log(logger, Verbose, "");

diff --git a/solid/frame/aio/test/test_echo_tcp_stress.cpp b/solid/frame/aio/test/test_echo_tcp_stress.cpp
@@ -487,7 +487,7 @@ int test_echo_tcp_stress(int argc, char* argv[])
         frame::Manager       srv_mgr;
         SecureContextT       srv_secure_ctx{SecureContextT::create()};
         frame::ServiceT      srv_svc{srv_mgr};
-        CallPoolT            cwp{1, 100, 0, [](const size_t) {}, [](const size_t) {}};
+        CallPoolT            cwp{{1, 100, 0}, [](const size_t) {}, [](const size_t) {}};
         frame::aio::Resolver resolver([&cwp](std::function<void()>&& _fnc) { cwp.pushOne(std::move(_fnc)); });
 
         async_resolver(&resolver);
@@ -695,8 +695,7 @@ void Listener::onAccept(frame::aio::ReactorContext& _rctx, SocketDevice& _rsd)
             break;
         }
         --repeatcnt;
-    } while (repeatcnt != 0u && sock.accept(
-                 _rctx, [this](frame::aio::ReactorContext& _rctx, SocketDevice& _rsd) { onAccept(_rctx, _rsd); }, _rsd));
+    } while (repeatcnt != 0u && sock.accept(_rctx, [this](frame::aio::ReactorContext& _rctx, SocketDevice& _rsd) { onAccept(_rctx, _rsd); }, _rsd));
 
     if (repeatcnt == 0u) {
         sock.postAccept(
@@ -1004,8 +1003,7 @@ void Listener::onAccept(frame::aio::ReactorContext& _rctx, SocketDevice& _rsd)
             break;
         }
         --repeatcnt;
-    } while (repeatcnt != 0u && sock.accept(
-                 _rctx, [this](frame::aio::ReactorContext& _rctx, SocketDevice& _rsd) { onAccept(_rctx, _rsd); }, _rsd));
+    } while (repeatcnt != 0u && sock.accept(_rctx, [this](frame::aio::ReactorContext& _rctx, SocketDevice& _rsd) { onAccept(_rctx, _rsd); }, _rsd));
 
     if (repeatcnt == 0u) {
         sock.postAccept(

diff --git a/solid/frame/aio/test/test_event_stress_wp.cpp b/solid/frame/aio/test/test_event_stress_wp.cpp
@@ -224,11 +224,11 @@ int test_event_stress_wp(int argc, char* argv[])
             gctx.stopping_ = false;
 
             account_cp.start(
-                thread_count, account_count, 0, [](const size_t, AccountContext&) {}, [](const size_t, AccountContext&) {}, std::ref(acc_ctx));
+                {thread_count, account_count, 0}, [](const size_t, AccountContext&) {}, [](const size_t, AccountContext&) {}, std::ref(acc_ctx));
             connection_cp.start(
-                thread_count, account_count * account_connection_count, 0, [](const size_t, ConnectionContext&) {}, [](const size_t, ConnectionContext&) {}, std::ref(conn_ctx));
+                {thread_count, account_count * account_connection_count, 0}, [](const size_t, ConnectionContext&) {}, [](const size_t, ConnectionContext&) {}, std::ref(conn_ctx));
             device_cp.start(
-                thread_count, account_count * account_connection_count, 0, [](const size_t, DeviceContext&) {}, [](const size_t, DeviceContext&) {}, std::ref(dev_ctx));
+                {thread_count, account_count * account_connection_count, 0}, [](const size_t, DeviceContext&) {}, [](const size_t, DeviceContext&) {}, std::ref(dev_ctx));
 
             conn_ctx.conn_cnt_  = (account_connection_count * account_count);
             auto produce_lambda = [&]() {

diff --git a/solid/frame/aio/test/test_perf_threadpool_lockfree.cpp b/solid/frame/aio/test/test_perf_threadpool_lockfree.cpp
@@ -48,7 +48,7 @@ int test_perf_threadpool_lockfree(int argc, char* argv[])
     (void)context_count;
     auto lambda = [&]() {
         ThreadPoolT wp{
-            thread_count, 10000, 0, [](const size_t) {}, [](const size_t) {},
+            {thread_count, 10000, 0}, [](const size_t) {}, [](const size_t) {},
             [&](EventBase& _event) {
                 if (_event == generic_event<GenericEventE::Wake>) {
                     ++received_events;

diff --git a/solid/frame/aio/test/test_perf_threadpool_synch_context.cpp b/solid/frame/aio/test/test_perf_threadpool_synch_context.cpp
@@ -51,7 +51,7 @@ int test_perf_threadpool_synch_context(int argc, char* argv[])
     auto lambda = [&]() {
         auto        start = std::chrono::steady_clock::now();
         ThreadPoolT wp{
-            thread_count, 10000, 0, [](const size_t) {}, [](const size_t) {},
+            {thread_count, 10000, 0}, [](const size_t) {}, [](const size_t) {},
             [&](EventBase& _event) {
                 if (_event == generic_event<GenericEventE::Wake>) {
                     ++received_events;