Merge branch 'feat/support_event_bus' of code.uocat.com:tqcq/sled into feat/support_event_bus

.gitea/workflows/linux-aarch64-gcc.yml

@@ -27,7 +27,7 @@ concurrency:
 permissions:
   contents: read
 jobs:
-  linux-gcc:
+  linux-gcc-aarch64:
     runs-on: ubuntu-20.04
     steps:
       - uses: actions/checkout@v4
@@ -69,108 +69,10 @@ jobs:
       - name: build
         run: |
           mkdir build && cd build
-          cmake -DCMAKE_TOOLCHAIN_FILE=../toolchains/aarch64-linux-gnu.toolchain.cmake -DSLED_BUILD_TESTS=ON -DSLED_BUILD_BENCHMARK=ON
+          cmake -DCMAKE_TOOLCHAIN_FILE=../toolchains/aarch64-linux-gnu.toolchain.cmake -DSLED_BUILD_TESTS=ON -DSLED_BUILD_BENCHMARK=ON ..
           cmake --build . -j $(nproc)
       - name: test
         run: |
           export PATH=$GITHUB_WORKSPACE/qemu-install/bin:$PATH
           cd build
           TESTS_EXECUTABLE_LOADER=qemu-aarch64 TESTS_EXECUTABLE_LOADER_ARGUMENTS="-L;/usr/aarch64-linux-gnu" ctest --output-on-failure -j $(nproc)
-  linux-gcc-arm82:
-    runs-on: ubuntu-20.04
-    steps:
-      - uses: actions/checkout@v4
-      - name: cache-qemu
-        id: cache-qemu
-        uses: actions/cache@v4
-        with:
-          path: qemu-install
-          key: qemu-aarch64-install-20220502-ubuntu-2004-2
-      - name: install-qemu-build-deps
-        if: steps.cache-qemu.outputs.cache-hit != 'true'
-        run: |
-          sudo apt-get update -y
-          sudo apt-get install -y autoconf automake autotools-dev ninja-build
-      - name: checkout-qemu
-        if: steps.cache-qemu.outputs.cache-hit != 'true'
-        run: |
-          mkdir qemu && cd qemu
-          git init
-          git remote add origin https://gitlab.com/qemu-project/qemu.git
-          git fetch --depth=1 origin f5643914a9e8f79c606a76e6a9d7ea82a3fc3e65
-          git checkout f5643914a9e8f79c606a76e6a9d7ea82a3fc3e65
-        # uses: actions/checkout@v4
-        # with:
-        #   repository: qemu/qemu
-        #   path: qemu
-        #   ref: f5643914a9e8f79c606a76e6a9d7ea82a3fc3e65
-      - name: qemu
-        if: steps.cache-qemu.outputs.cache-hit != 'true'
-        run: |
-          cd qemu
-          ./configure --prefix=$GITHUB_WORKSPACE/qemu-install --target-list=aarch64-linux-user --disable-system
-          make -j$(nproc)
-          make install
-      - name: aarch64-gnu-toolchain
-        run: |
-          sudo apt-get update -y
-          sudo apt-get install -y cmake make g++-aarch64-linux-gnu
-      - name: build
-        run: |
-          mkdir build && cd build
-          cmake -DCMAKE_TOOLCHAIN_FILE=../toolchains/aarch64-linux-gnu.toolchain.cmake -DSLED_BUILD_TESTS=ON -DSLED_BUILD_BENCHMARK=ON
-          cmake --build . -j $(nproc)
-      - name: test
-        run: |
-          export PATH=$GITHUB_WORKSPACE/qemu-install/bin:$PATH
-          cd build
-          TESTS_EXECUTABLE_LOADER=qemu-aarch64 TESTS_EXECUTABLE_LOADER_ARGUMENTS="-L;/usr/aarch64-linux-gnu" ctest --output-on-failure -j $(nproc)
-  linux-gcc-arm86:
-    runs-on: ubuntu-22.04
-    steps:
-      - uses: actions/checkout@v4
-      - name: cache-qemu
-        id: cache-qemu
-        uses: actions/cache@v4
-        with:
-          path: qemu-install
-          key: qemu-aarch64-install-20230717
-      - name: install-qemu-build-deps
-        if: steps.cache-qemu.outputs.cache-hit != 'true'
-        run: |
-          sudo apt-get update -y
-          sudo apt-get install -y autoconf automake autotools-dev ninja-build
-      - name: checkout-qemu
-        if: steps.cache-qemu.outputs.cache-hit != 'true'
-        run: |
-          mkdir qemu && cd qemu
-          git init
-          git remote add origin https://gitlab.com/qemu-project/qemu.git
-          git fetch --depth=1 origin f5643914a9e8f79c606a76e6a9d7ea82a3fc3e65
-          git checkout f5643914a9e8f79c606a76e6a9d7ea82a3fc3e65
-        # uses: actions/checkout@v4
-        # with:
-        #   repository: qemu/qemu
-        #   path: qemu
-        #   ref: ed8ad9728a9c0eec34db9dff61dfa2f1dd625637
-      - name: qemu
-        if: steps.cache-qemu.outputs.cache-hit != 'true'
-        run: |
-          cd qemu
-          ./configure --prefix=$GITHUB_WORKSPACE/qemu-install --target-list=aarch64-linux-user --disable-system
-          make -j$(nproc)
-          make install
-      - name: aarch64-gnu-toolchain
-        run: |
-          sudo apt-get update -y
-          sudo apt-get install -y cmake make g++-aarch64-linux-gnu
-      - name: build
-        run: |
-          mkdir build && cd build
-          cmake -DCMAKE_TOOLCHAIN_FILE=../toolchains/aarch64-linux-gnu.toolchain.cmake -DSLED_BUILD_TESTS=ON -DSLED_BUILD_BENCHMARK=ON
-          cmake --build . -j $(nproc)
-      - name: test
-        run: |-
-          export PATH=$GITHUB_WORKSPACE/qemu-install/bin:$PATH
-          cd build
-          TESTS_EXECUTABLE_LOADER=qemu-aarch64 TESTS_EXECUTABLE_LOADER_ARGUMENTS="-L;/usr/aarch64-linux-gnu" ctest --output-on-failure -j $(nproc)

CMakeLists.txt

@@ -111,6 +111,7 @@ if(SLED_BUILD_BENCHMARK)
 
   add_executable(
     sled_benchmark
+    src/sled/event_bus/event_bus_bench.cc
     src/sled/random_bench.cc
     src/sled/strings/base64_bench.cc
     # src/sled/system/fiber/fiber_bench.cc
@@ -187,6 +188,7 @@ if(SLED_BUILD_TESTS)
                 src/sled/utility/move_on_copy_test.cc)
   sled_add_test(NAME sled_symbolize_test SRCS
                 src/sled/debugging/symbolize_test.cc NO_MAIN)
+  sled_add_test(NAME sled_sigslot_test SRCS src/sled/sigslot_test.cc)
 endif(SLED_BUILD_TESTS)
 
 if(SLED_BUILD_FUZZ)

src/sled/event_bus/event_bus.h

@@ -8,14 +8,14 @@
 namespace sled {
 
 class EventBus;
-class EventSubscriber;
+class Subscriber;
 
 namespace {
 
 template<typename Event>
 class EventRegistry {
 public:
-    using Dispatcher      = sigslot::signal1<Event, sigslot::single_threaded>;
+    using Dispatcher      = sigslot::signal1<Event>;
     using SubscriberTable = std::unordered_map<EventBus *, Dispatcher>;
 
     static EventRegistry &Instance()
@@ -60,6 +60,7 @@ public:
         if (iter == signals_.end()) { return; }
         auto &dispatcher = iter->second;
         dispatcher.disconnect(instance);
+        if (dispatcher.is_empty()) { signals_.erase(iter); }
     }
 
     bool IsEmpty(EventBus *bus) const
@@ -94,13 +95,13 @@ private:
 
 }// namespace
 
-class EventSubscriber : public sigslot::has_slots<> {
-public:
-    virtual ~EventSubscriber() {}
-};
-
 class EventBus {
 public:
+    class Subscriber : public sigslot::has_slots<> {
+    public:
+        virtual ~Subscriber() = default;
+    };
+
     EventBus() = default;
 
     ~EventBus()
@@ -119,7 +120,7 @@ public:
 
     // On<Event1> ([](const Event1 &){})
     template<typename Event, typename C>
-    typename std::enable_if<std::is_base_of<EventSubscriber, C>::value>::type
+    typename std::enable_if<std::is_base_of<Subscriber, C>::value>::type
     Subscribe(C *instance, void (C::*method)(Event))
     {
         {
@@ -131,7 +132,7 @@ public:
     }
 
     template<typename Event, typename C>
-    typename std::enable_if<std::is_base_of<EventSubscriber, C>::value>::type Unsubscribe(C *instance)
+    typename std::enable_if<std::is_base_of<Subscriber, C>::value>::type Unsubscribe(C *instance)
     {
         EventRegistry<Event>::Instance().Unsubscribe(this, instance);
         {

src/sled/event_bus/event_bus_bench.cc

@@ -0,0 +1,78 @@
+#include <sled/event_bus/event_bus.h>
+#include <sled/log/log.h>
+#include <sled/system/fiber/wait_group.h>
+#include <sled/system/thread_pool.h>
+
+struct Event {
+    std::shared_ptr<int> data = std::make_shared<int>(0);
+};
+
+struct AtomicEvent {
+    AtomicEvent(std::atomic<int> &v) : data(v) {}
+
+    std::atomic<int> &data;
+};
+
+struct Subscriber : public sled::EventBus::Subscriber {
+    void OnEvent(Event event) { (*event.data)++; }
+
+    void OnAtomicnEvent(AtomicEvent event) { event.data.fetch_add(1); }
+};
+
+void
+BMEventBusPost_1_to_1(picobench::state &s)
+{
+    sled::EventBus event_bus;
+    Subscriber subscriber;
+    event_bus.Subscribe(&subscriber, &Subscriber::OnEvent);
+    for (auto _ : s) { event_bus.Post(Event{}); }
+}
+
+void
+BMEventBusPost_1_to_1k(picobench::state &s)
+{
+    sled::EventBus event_bus;
+    std::vector<Subscriber> subscribers(1000);
+    for (auto &subscriber : subscribers) { event_bus.Subscribe(&subscriber, &Subscriber::OnEvent); }
+
+    for (auto _ : s) {
+        Event event;
+        event_bus.Post(event);
+        SLED_ASSERT(*event.data == 1000, "");
+    }
+}
+
+void
+BMEventBusPost_10_to_1k(picobench::state &s)
+{
+    constexpr int kPublishCount    = 10;
+    constexpr int kSubscriberCount = 1000;
+
+    sled::EventBus event_bus;
+    std::vector<Subscriber> subscribers(kSubscriberCount);
+    for (auto &subscriber : subscribers) { event_bus.Subscribe(&subscriber, &Subscriber::OnAtomicnEvent); }
+    sled::ThreadPool pool(kPublishCount);
+
+    for (auto _ : s) {
+        std::atomic<int> value(0);
+        AtomicEvent atomic_event(value);
+        sled::WaitGroup wg(kPublishCount);
+        for (int i = 0; i < kPublishCount; i++) {
+            pool.PostTask([atomic_event, wg, &event_bus]() {
+                event_bus.Post(atomic_event);
+                wg.Done();
+            });
+        }
+        wg.Wait();
+        SLED_ASSERT(value.load() == kPublishCount * kSubscriberCount,
+                    "{} != {}",
+                    value.load(),
+                    kPublishCount * kSubscriberCount);
+    }
+}
+
+PICOBENCH_SUITE("EventBus");
+
+PICOBENCH(BMEventBusPost_1_to_1);
+PICOBENCH(BMEventBusPost_1_to_1k);
+PICOBENCH(BMEventBusPost_10_to_1k);

src/sled/event_bus/event_bus_test.cc

@@ -1,6 +1,23 @@
 #include <sled/event_bus/event_bus.h>
+#include <sled/log/log.h>
+#include <sled/system/fiber/wait_group.h>
 #include <sled/system/thread_pool.h>
 
+using namespace fakeit;
+
+template<typename T, typename R, typename... Args>
+void *
+GetPtr(R (T::*p)(Args...))
+{
+    union {
+        R (T::*ptr)(Args...);
+        void *void_ptr;
+    } _;
+
+    _.ptr = p;
+    return _.void_ptr;
+}
+
 struct Event1 {
     int a;
 };
@@ -9,7 +26,7 @@ struct Event2 {
     std::string str;
 };
 
-struct Subscriber : public sled::EventSubscriber {
+struct Subscriber : public sled::EventBus::Subscriber {
     void OnEvent1(Event1 event) { a += event.a; }
 
     void OnEvent2(Event2 event) { str += event.str; }
@@ -62,4 +79,42 @@ TEST_SUITE("EventBus")
         CHECK_EQ(subscriber.a, 1);
         CHECK_EQ(subscriber.str, "1");
     }
+
+    TEST_CASE("thread_pool")
+    {
+        constexpr int kPublishCount    = 10;
+        constexpr int kSubscriberCount = 1000;
+
+        struct AtomicEvent {
+            std::atomic<int> &data;
+        };
+
+        struct AotmicEventSubscriber : public sled::EventBus::Subscriber {
+            virtual ~AotmicEventSubscriber() = default;
+
+            void OnEvent(AtomicEvent event) { event.data.fetch_add(1); }
+        };
+
+        std::atomic<int> value(0);
+        AtomicEvent atomic_event{value};
+
+        sled::WaitGroup wg(kPublishCount);
+        sled::ThreadPool pool(kPublishCount);
+        sled::EventBus bus;
+        std::vector<AotmicEventSubscriber> subscribers(kSubscriberCount);
+
+        for (auto &sub : subscribers) { bus.Subscribe(&sub, &AotmicEventSubscriber::OnEvent); }
+        std::atomic<int> invoke_count(0);
+        for (int i = 0; i < kPublishCount; i++) {
+            pool.PostTask([wg, atomic_event, &bus, &invoke_count]() {
+                bus.Post(atomic_event);
+                invoke_count.fetch_add(1);
+                wg.Done();
+            });
+        }
+        wg.Wait();
+
+        CHECK_EQ(invoke_count.load(), kPublishCount);
+        CHECK_EQ(value.load(), kPublishCount * kSubscriberCount);
+    }
 }

src/sled/random_bench.cc

@@ -1,5 +1,6 @@
 #include <sled/random.h>
 #include <sled/testing/benchmark.h>
+PICOBENCH_SUITE("Random");
 
 PICOBENCH([](picobench::state &s) {
     sled::Random rand(s.user_data());

src/sled/sigslot.h

@@ -104,7 +104,7 @@
 #include <set>
 
 // On our copy of sigslot.h, we set single threading as default.
-#define SIGSLOT_DEFAULT_MT_POLICY single_threaded
+#define SIGSLOT_DEFAULT_MT_POLICY multi_threaded_local
 
 #if defined(SIGSLOT_PURE_ISO) || (!defined(WEBRTC_WIN) && !defined(__GNUG__) && !defined(SIGSLOT_USE_POSIX_THREADS))
 #define _SIGSLOT_SINGLE_THREADED

src/sled/sigslot_test.cc

@@ -0,0 +1,16 @@
+#include <sled/sigslot.h>
+
+struct DeleteSelf : public sigslot::has_slots<> {
+    void DeleteThis() {
+        delete this;
+    }
+};
+
+TEST_SUITE("sigslot") {
+    TEST_CASE("delete this") {
+        DeleteSelf* d = new DeleteSelf();
+        sigslot::signal0<> sig;
+        sig.connect(d, &DeleteSelf::DeleteThis);
+        sig.emit();
+    }
+}

src/sled/strings/base64_bench.cc

@@ -38,6 +38,6 @@ Base64Decode(picobench::state &state)
         (void) sled::Base64::Decode(base64_input);
     }
 }
-
+PICOBENCH_SUITE("Base64");
 PICOBENCH(Base64Decode);
 PICOBENCH(Base64Encode);

src/sled/system/thread_pool_bench.cc

@@ -13,4 +13,5 @@ ThreadPoolBench(picobench::state &state)
 }
 
 // BENCHMARK(ThreadPoolBench)->RangeMultiplier(10)->Range(10, 10000);
+PICOBENCH_SUITE("TheadPool");
 PICOBENCH(ThreadPoolBench);

src/sled/system_time_bench.cc

@@ -7,4 +7,5 @@ SystemTimeNanos(picobench::state &state)
     for (auto _ : state) { (void) sled::SystemTimeNanos(); }
 }
 
+PICOBENCH_SUITE("SystemTime");
 PICOBENCH(SystemTimeNanos);

src/sled/uri_bench.cc

@@ -34,4 +34,5 @@ ParseURI(picobench::state &s)
     }
 }
 
+PICOBENCH_SUITE("URI");
 PICOBENCH(ParseURI);