evl: make edge triggering optional

ColdenCullen · Sep 29, 2023 · 622093b · 622093b
1 parent cac568c
commit 622093b
Show file tree

Hide file tree

Showing 5 changed files with 75 additions and 22 deletions.
diff --git a/include/tlb/event_loop.h b/include/tlb/event_loop.h
@@ -26,7 +26,8 @@ struct tlb_event_loop *tlb_evl_new(struct tlb_allocator *alloc);
 void tlb_evl_destroy(struct tlb_event_loop *loop);
 
 /** Subscribe a file descriptor to the event loop. */
-tlb_handle tlb_evl_add_fd(struct tlb_event_loop *loop, int fd, int events, tlb_on_event *on_event, void *userdata);
+tlb_handle tlb_evl_add_fd(struct tlb_event_loop *loop, int fd, int events, bool edge_trigger, tlb_on_event *on_event,
+                          void *userdata);
 
 /** Add a timer to fire in timeout milliseconds */
 tlb_handle tlb_evl_add_timer(struct tlb_event_loop *loop, int timeout, tlb_on_event *trigger, void *userdata);

diff --git a/source/event_loop.c b/source/event_loop.c
@@ -1,5 +1,7 @@
 #include "tlb/private/event_loop.h"
 
+#include "tlb/event_loop.h"
+
 #include <errno.h>
 
 /**********************************************************************************************************************
@@ -40,11 +42,14 @@ static struct tlb_subscription *s_sub_new(struct tlb_event_loop *loop, tlb_on_ev
  * File descriptor                                                                                                    *
  **********************************************************************************************************************/
 
-tlb_handle tlb_evl_add_fd(struct tlb_event_loop *loop, int fd, int events, tlb_on_event *on_event, void *userdata) {
+tlb_handle tlb_evl_add_fd(struct tlb_event_loop *loop, int fd, int events, bool edge_trigger, tlb_on_event *on_event,
+                          void *userdata) {
   struct tlb_subscription *sub = TLB_CHECK(NULL !=, s_sub_new(loop, on_event, userdata, "fd"));
   sub->ident.fd = fd;
   sub->events = events;
-  sub->sub_mode |= TLB_SUB_EDGE;
+  if (edge_trigger) {
+    sub->sub_mode |= TLB_SUB_EDGE;
+  }
 
   tlb_evl_impl_fd_init(sub);
   TLB_CHECK_GOTO(0 ==, tlb_evl_impl_subscribe(loop, sub), sub_failed);
@@ -76,18 +81,12 @@ tlb_handle tlb_evl_add_timer(struct tlb_event_loop *loop, int timeout, tlb_on_ev
  **********************************************************************************************************************/
 
 tlb_handle tlb_evl_add_evl(struct tlb_event_loop *loop, struct tlb_event_loop *sub_loop) {
-  struct tlb_subscription *sub = TLB_CHECK(NULL !=, s_sub_new(loop, tlb_evl_sub_loop_on_event, sub_loop, "sub-loop"));
-  sub->ident.fd = sub_loop->fd;
-  sub->events = TLB_EV_READ;
-  sub->sub_mode |= 0;
-
-  tlb_evl_impl_fd_init(sub);
-  TLB_CHECK_GOTO(0 ==, tlb_evl_impl_subscribe(loop, sub), sub_failed);
+  struct tlb_subscription *sub =
+      tlb_evl_add_fd(loop, sub_loop->fd, TLB_EV_READ, false, tlb_evl_sub_loop_on_event, sub_loop);
+  if (sub) {
+    sub->name = "sub-loop";
+  }
   return sub;
-
-sub_failed:
-  tlb_free(loop->alloc, sub);
-  return NULL;
 }
 
 void tlb_evl_sub_loop_on_event(tlb_handle subscription, int events, void *userdata) {

diff --git a/source/linux/event_loop_epoll.c b/source/linux/event_loop_epoll.c
@@ -92,7 +92,7 @@ void tlb_evl_cleanup(struct tlb_event_loop *loop) {
  **********************************************************************************************************************/
 
 void tlb_evl_impl_fd_init(struct tlb_subscription *sub) {
-  TLB_ASSERT(sub->sub_mode);
+  (void)sub;
 }
 
 /**********************************************************************************************************************

diff --git a/source/tlb.c b/source/tlb.c
@@ -12,6 +12,11 @@ enum {
   TLB_MAX_THREADS = 128,
 };
 
+struct tlb_thread {
+  thrd_t handle;
+  struct tlb_event_loop loop;
+};
+
 struct tlb {
   struct tlb_allocator *alloc;
   struct tlb_options options;
@@ -27,7 +32,7 @@ struct tlb {
   mtx_t mtx;
   cnd_t cnd;
 
-  thrd_t threads[];
+  struct tlb_thread threads[];
 };
 
 static _Thread_local bool s_should_stop;
@@ -93,7 +98,7 @@ int tlb_start(struct tlb *tlb) {
 
   const size_t target_threads = tlb->options.max_thread_count;
   for (size_t ii = 0; ii < target_threads; ++ii) {
-    thrd_create(&tlb->threads[ii], s_thread_start, tlb);
+    thrd_create(&tlb->threads[ii].handle, s_thread_start, tlb);
     cnd_wait(&tlb->cnd, &tlb->mtx);
   }
 
@@ -127,7 +132,7 @@ int tlb_stop(struct tlb *tlb) {
   for (size_t ii = 0; ii < active_threads; ++ii) {
     /* Join the threads */
     int result = 0;
-    thrd_join(tlb->threads[ii], &result);
+    thrd_join(tlb->threads[ii].handle, &result);
     TLB_ASSERT(thrd_success == result);
   }
 

diff --git a/tests/pipe_test.cc b/tests/pipe_test.cc
@@ -2,6 +2,7 @@
 #include "tlb/pipe.h"
 
 #include "tlb/event_loop.h"
+#include "tlb/private/event_loop.h"
 
 #include <gtest/gtest.h>
 #include <string.h>
@@ -32,14 +33,14 @@ class PipeTest : public TlbTest {
     TlbTest::TearDown();
   }
 
-  tlb_handle SubscribeRead(tlb_on_event *on_event, void *userdata) {
-    const tlb_handle handle = tlb_evl_add_fd(loop(), pipe.fd_read, TLB_EV_READ, on_event, userdata);
+  tlb_handle SubscribeRead(tlb_on_event *on_event, void *userdata, bool edge_trigger = true) {
+    const tlb_handle handle = tlb_evl_add_fd(loop(), pipe.fd_read, TLB_EV_READ, edge_trigger, on_event, userdata);
     open_subscriptions.emplace(handle);
     return handle;
   }
 
-  tlb_handle SubscribeWrite(tlb_on_event *on_event, void *userdata) {
-    const tlb_handle handle = tlb_evl_add_fd(loop(), pipe.fd_write, TLB_EV_WRITE, on_event, userdata);
+  tlb_handle SubscribeWrite(tlb_on_event *on_event, void *userdata, bool edge_trigger = true) {
+    const tlb_handle handle = tlb_evl_add_fd(loop(), pipe.fd_write, TLB_EV_WRITE, edge_trigger, on_event, userdata);
     open_subscriptions.emplace(handle);
     return handle;
   }
@@ -258,6 +259,53 @@ TEST_P(PipeTest, ReadableWritable) {
   wait([&]() { return state.read_count == 1; });
 }
 
+TEST_P(PipeTest, Testbed) {
+  struct TestState {
+    PipeTest *test = nullptr;
+    std::atomic<int> read_count{};
+    std::atomic<int> second_loop_read_count{};
+  } state;
+  state.test = this;
+
+  tlb_handle read_sub = SubscribeRead(
+      +[](tlb_handle handle, int events, void *userdata) {
+        TestState *state = static_cast<TestState *>(userdata);
+
+        Log() << "Read received on MAIN loop";
+
+        state->test->Unsubscribe(handle);
+
+        auto lock = state->test->lock();
+        state->read_count++;
+        state->test->notify();
+      },
+      &state);
+  ASSERT_NE(nullptr, read_sub);
+
+  struct tlb_event_loop *second_loop = tlb_evl_new(alloc());
+  const tlb_handle handle = tlb_evl_add_fd(
+      second_loop, pipe.fd_read, TLB_EV_READ, false,
+      +[](tlb_handle handle, int events, void *userdata) {
+        TestState *state = static_cast<TestState *>(userdata);
+
+        Log() << "Read received on SECOND loop";
+
+        state->second_loop_read_count++;
+      },
+      &state);
+
+  Write(s_test_value);
+
+  wait([&]() { return state.read_count == 1; });
+
+  Log() << "Handling events on second loop";
+  tlb_evl_handle_events(second_loop, 100, 0);
+  ASSERT_EQ(1, state.second_loop_read_count);
+
+  tlb_evl_remove(second_loop, handle);
+  tlb_evl_destroy(second_loop);
+}
+
 TLB_INSTANTIATE_TEST(PipeTest);
 
 }  // namespace