Attempt #1

tests/pipe_test.cc

@@ -2,11 +2,14 @@
 #include "tlb/pipe.h"
 
 #include "tlb/event_loop.h"
+#include "tlb/private/event_loop.h"
 
 #include <gtest/gtest.h>
 #include <string.h>
 
 #include "test_helpers.h"
+#include <array>
+#include <chrono>
 #include <unordered_set>
 
 namespace tlb_test {
@@ -260,5 +263,101 @@ TEST_P(PipeTest, ReadableWritable) {
 
 TLB_INSTANTIATE_TEST(PipeTest);
 
+class MultiPipeTest : public TlbTest {
+ public:
+  static constexpr size_t kPipeCount = 100;
+
+  void SetUp() override {
+    TlbTest::SetUp();
+
+    for (auto &pipe : pipes) {
+      ASSERT_EQ(0, tlb_pipe_open(&pipe));
+    }
+  }
+
+  void TearDown() override {
+    for (auto &pipe : pipes) {
+      tlb_pipe_close(&pipe);
+    }
+
+    TlbTest::TearDown();
+  }
+
+  void SubscribeRead(tlb_on_event *on_event, void *userdata) {
+    for (const auto &pipe : pipes) {
+      ASSERT_NE(nullptr, tlb_evl_add_fd(loop(), pipe.fd_read, TLB_EV_READ, on_event, userdata));
+    }
+  }
+
+  void SubscribeWrite(tlb_on_event *on_event, void *userdata) {
+    for (const auto &pipe : pipes) {
+      ASSERT_NE(nullptr, tlb_evl_add_fd(loop(), pipe.fd_write, TLB_EV_WRITE, on_event, userdata));
+    }
+  }
+
+  template <typename T>
+  T Read(tlb_handle handle) {
+    auto *sub = static_cast<tlb_subscription *>(handle);
+
+    T out;
+    static constexpr size_t size = sizeof(T);
+
+    tlb_pipe fake_pipe;
+    fake_pipe.fd_read = sub->ident.fd;
+    EXPECT_EQ(size, tlb_pipe_read_buf(&fake_pipe, &out, size));
+
+    Log() << "Reading from pipe " << sub->ident.fd << ": " << out << std::endl;
+    return out;
+  }
+
+  template <typename T>
+  void Write(const T &in) {
+    auto &pipe = pipes[pipe_index];
+    Log() << "Writing to pipe " << &pipe << ": " << in << std::endl;
+
+    static constexpr size_t size = sizeof(T);
+    EXPECT_EQ(size, tlb_pipe_write_buf(&pipe, &in, size));
+
+    pipe_index = (pipe_index + 6) % pipes.size();
+  }
+
+  std::array<tlb_pipe, kPipeCount> pipes;
+
+  size_t pipe_index = 0;
+};
+
+TEST_P(MultiPipeTest, Readible) {
+  static constexpr size_t kTargetReadCount = 5000;
+  struct TestState {
+    MultiPipeTest *test = nullptr;
+    std::atomic<size_t> read_count{0};
+  } state;
+  state.test = this;
+
+  SubscribeRead(
+      +[](tlb_handle handle, int events, void *userdata) {
+        TestState *state = static_cast<TestState *>(userdata);
+        const uint64_t value = state->test->Read<uint64_t>(handle);
+
+        const auto new_read_count = state->read_count.fetch_add(1) + 1;
+        if (new_read_count == kTargetReadCount) {
+          auto lock = state->test->lock();
+          state->test->notify();
+        } else {
+          state->test->Write(new_read_count);
+        }
+      },
+      &state);
+
+  /* Start by writing to a pipe per thread */
+  for (size_t i = 0; i < thread_count(); ++i) {
+    Write(s_test_value);
+  }
+  wait([&]() { return state.read_count >= kTargetReadCount && state.read_count <= kTargetReadCount + thread_count(); },
+       std::chrono::milliseconds(7500) / thread_count());
+}
+
+TLB_INSTANTIATE_TEST(MultiPipeTest);
+
 }  // namespace
 }  // namespace tlb_test

tests/test_helpers.cc

@@ -4,6 +4,7 @@
 #include "tlb/event_loop.h"
 #include "tlb/tlb.h"
 
+#include <chrono>
 #include <iostream>
 
 namespace tlb_test {
@@ -53,9 +54,9 @@ tlb_allocator *TlbTest::alloc() {
   return &alloc;
 }
 
-void TlbTest::wait(const std::function<bool()> &predicate) {
-  const auto timeout = std::chrono::milliseconds(50 * std::max<size_t>(thread_count(), 1));
-  const auto run_until = std::chrono::steady_clock::now() + timeout;
+void TlbTest::wait(const std::function<bool()> &predicate, std::chrono::milliseconds timeout) {
+  const auto total_timeout = timeout * std::max<size_t>(thread_count(), 1);
+  const auto run_until = std::chrono::steady_clock::now() + total_timeout;
   bool passed = false;
   // Fire the loop if necessary
   if (thread_count() == 0) {

tests/test_helpers.h

@@ -58,7 +58,7 @@ class TlbTest : public ::testing::TestWithParam<std::tuple<LoopMode, size_t>> {
     }
   }
 
-  void wait(const std::function<bool()> &predicate);
+  void wait(const std::function<bool()> &predicate, std::chrono::milliseconds timeout = std::chrono::milliseconds(50));
 
   void SetUp() override;
   void TearDown() override;