co::Coroutine -> toolbelt::Poller

CMakeLists.txt

@@ -35,19 +35,6 @@ FetchContent_Declare(
 FetchContent_MakeAvailable(abseil)
 # Abseil provides targets like absl::base, absl::strings, etc.
 
-# --- External Dependency: co ---
-# cpp_toolbelt depends on co, so we need to fetch and make it available.
-FetchContent_Declare(
-    co
-    GIT_REPOSITORY https://github.com/dallison/co.git
-    GIT_TAG cf1252b2f5952d7cba83b67dd69288971c0a2b57
-    # Pass architecture settings to co's CMake build
-    CMAKE_ARGS
-        CMAKE_OSX_ARCHITECTURES="${CMAKE_OSX_ARCHITECTURES}"
-)
-FetchContent_MakeAvailable(co)
-# co provides the 'co' target.
-
 # --- toolbelt Library Target ---
 # Corresponds to //toolbelt:toolbelt in Bazel
 # Source files are omitted here for manual addition by the user.
@@ -67,6 +54,8 @@ add_library(toolbelt STATIC
     toolbelt/payload_buffer.h
     toolbelt/pipe.cc
     toolbelt/pipe.h
+    toolbelt/poller.cc
+    toolbelt/poller.h
     toolbelt/sockets.cc
     toolbelt/sockets.h
     toolbelt/table.cc
@@ -88,7 +77,6 @@ target_link_libraries(toolbelt PUBLIC
     absl::log # Added as it's used by toolbelt/logging
     absl::time # Added as it's used by toolbelt/time
     absl::synchronization # Added as it's used by toolbelt/utilities/thread_pool
-    co # Link against the co library
 )
 
 # Note: If there are tests in cpp_toolbelt, they would need a separate target

MODULE.bazel

@@ -15,4 +15,9 @@ bazel_dep(name = "rules_cc", version = "0.2.17")
 #   module_name = "coroutines",
 #   path = "../co",
 # )
+git_override(
+    module_name = "coroutines",
+    commit = "88c4013c209bc0815cf7a8b20579b4f353c1b1be",
+    remote = "https://github.com/mikael-s-persson/co.git",
+)
 

toolbelt/BUILD.bazel

@@ -11,6 +11,7 @@ cc_library(
         "logging.cc",
         "payload_buffer.cc",
         "pipe.cc",
+        "poller.cc",
         "sockets.cc",
         "table.cc",
         "triggerfd.cc",
@@ -26,6 +27,7 @@ cc_library(
         "mutex.h",
         "payload_buffer.h",
         "pipe.h",
+        "poller.h",
         "sockets.h",
         "table.h",
         "triggerfd.h",
@@ -37,7 +39,6 @@ cc_library(
         "@abseil-cpp//absl/strings",
         "@abseil-cpp//absl/strings:str_format",
         "@abseil-cpp//absl/types:span",
-        "@coroutines//:co",
     ],
 )
 
@@ -91,6 +92,7 @@ cc_test(
         "@abseil-cpp//absl/status:status_matchers",
         "@abseil-cpp//absl/status:statusor",
         "@googletest//:gtest_main",
+        "@coroutines//:co",
     ],
 )
 
@@ -105,6 +107,7 @@ cc_test(
         "@abseil-cpp//absl/status:status_matchers",
         "@abseil-cpp//absl/status:statusor",
         "@googletest//:gtest_main",
+        "@coroutines//:co",
     ],
 )
 

toolbelt/fd.cc

@@ -16,7 +16,7 @@ void CloseAllFds(std::function<bool(int)> predicate) {
 }
 
 absl::StatusOr<ssize_t> FileDescriptor::Read(void *buffer, size_t length,
-                                             const co::Coroutine *c) {
+                                             const Poller *c) {
   char *buf = reinterpret_cast<char *>(buffer);
   size_t total = 0;
   while (total < length) {
@@ -58,7 +58,7 @@ absl::StatusOr<ssize_t> FileDescriptor::Read(void *buffer, size_t length,
 }
 
 absl::StatusOr<ssize_t> FileDescriptor::Write(const void *buffer, size_t length,
-                                              const co::Coroutine *c) {
+                                              const Poller *c) {
   const char *buf = reinterpret_cast<const char *>(buffer);
 
   size_t total = 0;

toolbelt/fd.h

@@ -18,7 +18,7 @@
 #include <sys/resource.h>
 #include <sys/stat.h>
 #include <unistd.h>
-#include "co/coroutine.h"
+#include "toolbelt/poller.h"
 
 namespace toolbelt {
 
@@ -188,9 +188,9 @@ class FileDescriptor {
     return absl::OkStatus();
   }
 
-  absl::StatusOr<ssize_t> Read(void* buffer, size_t length, const co::Coroutine* c = nullptr);
+  absl::StatusOr<ssize_t> Read(void* buffer, size_t length, const Poller* c = nullptr);
   absl::StatusOr<ssize_t> Write(const void* buffer, size_t length,
-                                      const co::Coroutine* c = nullptr);
+                                      const Poller* c = nullptr);
 private:
   // Reference counted OS fd, shared among all FileDescriptors with the
   // same OS fd, provided you don't create two FileDescriptors with the

toolbelt/pipe.cc

@@ -83,7 +83,7 @@ absl::Status Pipe::SetPipeSize(size_t size) {
 }
 
 absl::StatusOr<ssize_t> Pipe::Read(char *buffer, size_t length,
-                                   const co::Coroutine *c) {
+                                   const Poller *c) {
   size_t total = 0;
   ScopedRead sc(*this, c);
 
@@ -123,7 +123,7 @@ absl::StatusOr<ssize_t> Pipe::Read(char *buffer, size_t length,
 }
 
 absl::StatusOr<ssize_t> Pipe::Write(const char *buffer, size_t length,
-                                    const co::Coroutine *c) {
+                                    const Poller *c) {
   size_t total = 0;
   ScopedWrite sc(*this, c);
 

toolbelt/pipe.h

@@ -3,7 +3,7 @@
 #include "absl/status/status.h"
 #include "absl/status/statusor.h"
 #include "absl/strings/str_format.h"
-#include "co/coroutine.h"
+#include "toolbelt/poller.h"
 #include "toolbelt/fd.h"
 
 #include <chrono>
@@ -66,9 +66,9 @@ class Pipe {
   absl::Status SetPipeSize(size_t size);
 
   virtual absl::StatusOr<ssize_t> Read(char *buffer, size_t length,
-                                       const co::Coroutine *c = nullptr);
+                                       const Poller *c = nullptr);
   virtual absl::StatusOr<ssize_t> Write(const char *buffer, size_t length,
-                                        const co::Coroutine *c = nullptr);
+                                        const Poller *c = nullptr);
 
 protected:
   // RAII classes for keeping coroutines from interleaving reads or writes on a
@@ -78,15 +78,12 @@ class Pipe {
   //
   // Same applies to non-coroutine use except we block with a sleep.
   struct ScopedRead {
-    ScopedRead(Pipe &p, const co::Coroutine *c) : pipe(p) {
+    ScopedRead(Pipe &p, const Poller *c) : pipe(p) {
       while (pipe.read_in_progress_) {
         if (c) {
           c->Yield();
         } else {
-          if (!pipe.read_.IsNonBlocking()) {
-            break;
-          }
-          std::this_thread::sleep_for(std::chrono::microseconds(10));
+          PosixPoller{/*yield_sleep_ns=*/10000ULL}.Yield();
         }
       }
       pipe.read_in_progress_ = true;
@@ -97,15 +94,12 @@ class Pipe {
   };
 
   struct ScopedWrite {
-    ScopedWrite(Pipe &p, const co::Coroutine *c) : pipe(p) {
+    ScopedWrite(Pipe &p, const Poller *c) : pipe(p) {
       while (pipe.write_in_progress_) {
         if (c) {
           c->Yield();
         } else {
-          if (!pipe.write_.IsNonBlocking()) {
-            break;
-          }
-          std::this_thread::sleep_for(std::chrono::microseconds(10));
+          PosixPoller{/*yield_sleep_ns=*/10000ULL}.Yield();
         }
       }
       pipe.write_in_progress_ = true;
@@ -151,15 +145,15 @@ template <typename T> class SharedPtrPipe : public Pipe {
 
   // You can't use raw buffers with shared ptr pipes.
   absl::StatusOr<ssize_t> Read(char *, size_t ,
-                               const co::Coroutine * = nullptr) override {
+                               const Poller * = nullptr) override {
     return absl::InternalError("Not supported on SharedPtrPipe");
   }
   absl::StatusOr<ssize_t> Write(const char *, size_t ,
-                                const co::Coroutine *c = nullptr) override {
+                                const Poller *c = nullptr) override {
     return absl::InternalError("Not supported on SharedPtrPipe");
   }
 
-  absl::StatusOr<std::shared_ptr<T>> Read(const co::Coroutine *c = nullptr) {
+  absl::StatusOr<std::shared_ptr<T>> Read(const Poller *c = nullptr) {
     char buffer[sizeof(std::shared_ptr<T>)];
     size_t length = sizeof(buffer);
     size_t total = 0;
@@ -205,7 +199,7 @@ template <typename T> class SharedPtrPipe : public Pipe {
   }
 
   // This makes the pipe an owner of the pointer.
-  absl::Status Write(std::shared_ptr<T> p, const co::Coroutine *c = nullptr) {
+  absl::Status Write(std::shared_ptr<T> p, const Poller *c = nullptr) {
     // On entry, ref count for p = N
     char buffer[sizeof(std::shared_ptr<T>)];