diff --git a/CMakeLists.txt b/CMakeLists.txt index a8dce7d..dcbed63 100644 --- a/CMakeLists.txt +++ b/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/async_waiter.cc + toolbelt/async_waiter.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 diff --git a/MODULE.bazel b/MODULE.bazel index 7117fe6..abbb0dc 100644 --- a/MODULE.bazel +++ b/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 = "21e28c2c6b28ea1f4cf5f9948a9c6dd379b5ec7e", + remote = "https://github.com/mikael-s-persson/co.git", +) diff --git a/MODULE.bazel.lock b/MODULE.bazel.lock index f8a48ff..fabf7a4 100644 --- a/MODULE.bazel.lock +++ b/MODULE.bazel.lock @@ -19,8 +19,8 @@ "https://bcr.bazel.build/modules/apple_support/1.21.0/MODULE.bazel": "ac1824ed5edf17dee2fdd4927ada30c9f8c3b520be1b5fd02a5da15bc10bff3e", "https://bcr.bazel.build/modules/apple_support/1.21.1/MODULE.bazel": "5809fa3efab15d1f3c3c635af6974044bac8a4919c62238cce06acee8a8c11f1", "https://bcr.bazel.build/modules/apple_support/1.22.1/MODULE.bazel": "90bd1a660590f3ceffbdf524e37483094b29352d85317060b2327fff8f3f4458", - "https://bcr.bazel.build/modules/apple_support/1.24.2/MODULE.bazel": "0e62471818affb9f0b26f128831d5c40b074d32e6dda5a0d3852847215a41ca4", - "https://bcr.bazel.build/modules/apple_support/1.24.2/source.json": "2c22c9827093250406c5568da6c54e6fdf0ef06238def3d99c71b12feb057a8d", + "https://bcr.bazel.build/modules/apple_support/2.5.2/MODULE.bazel": "bb1b7081e390a8d3885f642cb0073dfcf8e7a36525e079fa8f5835af7079c0fe", + "https://bcr.bazel.build/modules/apple_support/2.5.2/source.json": "dd51ee925a550e92b55eb41c4c809a967da311a2d2520241252a238a48b39c4c", "https://bcr.bazel.build/modules/bazel_features/1.1.1/MODULE.bazel": "27b8c79ef57efe08efccbd9dd6ef70d61b4798320b8d3c134fd571f78963dbcd", "https://bcr.bazel.build/modules/bazel_features/1.10.0/MODULE.bazel": "f75e8807570484a99be90abcd52b5e1f390362c258bcb73106f4544957a48101", "https://bcr.bazel.build/modules/bazel_features/1.11.0/MODULE.bazel": "f9382337dd5a474c3b7d334c2f83e50b6eaedc284253334cf823044a26de03e8", @@ -56,8 +56,6 @@ "https://bcr.bazel.build/modules/bazel_skylib/1.9.0/source.json": "7ad77c1e8c1b84222d9b3f3cae016a76639435744c19330b0b37c0a3c9da7dc0", "https://bcr.bazel.build/modules/buildozer/8.5.1/MODULE.bazel": "a35d9561b3fc5b18797c330793e99e3b834a473d5fbd3d7d7634aafc9bdb6f8f", "https://bcr.bazel.build/modules/buildozer/8.5.1/source.json": "e3386e6ff4529f2442800dee47ad28d3e6487f36a1f75ae39ae56c70f0cd2fbd", - "https://bcr.bazel.build/modules/coroutines/3.3.2/MODULE.bazel": "ad1395ae9758ee5d113acab755b4d972c368c4dab66513bfe31423136f3ee608", - "https://bcr.bazel.build/modules/coroutines/3.3.2/source.json": "fb47a4c3e13d730a1a58ed6a34add8f6aaeef655a47d0f179427f71bf58ba150", "https://bcr.bazel.build/modules/google_benchmark/1.8.2/MODULE.bazel": "a70cf1bba851000ba93b58ae2f6d76490a9feb74192e57ab8e8ff13c34ec50cb", "https://bcr.bazel.build/modules/googletest/1.11.0/MODULE.bazel": "3a83f095183f66345ca86aa13c58b59f9f94a2f81999c093d4eeaa2d262d12f4", "https://bcr.bazel.build/modules/googletest/1.14.0.bcr.1/MODULE.bazel": "22c31a561553727960057361aa33bf20fb2e98584bc4fec007906e27053f80c6", @@ -123,7 +121,7 @@ "https://bcr.bazel.build/modules/rules_cc/0.1.5/MODULE.bazel": "88dfc9361e8b5ae1008ac38f7cdfd45ad738e4fa676a3ad67d19204f045a1fd8", "https://bcr.bazel.build/modules/rules_cc/0.2.0/MODULE.bazel": "b5c17f90458caae90d2ccd114c81970062946f49f355610ed89bebf954f5783c", "https://bcr.bazel.build/modules/rules_cc/0.2.13/MODULE.bazel": "eecdd666eda6be16a8d9dc15e44b5c75133405e820f620a234acc4b1fdc5aa37", - "https://bcr.bazel.build/modules/rules_cc/0.2.16/MODULE.bazel": "9242fa89f950c6ef7702801ab53922e99c69b02310c39fb6e62b2bd30df2a1d4", + "https://bcr.bazel.build/modules/rules_cc/0.2.15/MODULE.bazel": "6a0a4a75a57aa6dc888300d848053a58c6b12a29f89d4304e1c41448514ec6e8", "https://bcr.bazel.build/modules/rules_cc/0.2.17/MODULE.bazel": "1849602c86cb60da8613d2de887f9566a6d354a6df6d7009f9d04a14402f9a84", "https://bcr.bazel.build/modules/rules_cc/0.2.17/source.json": "3832f45d145354049137c0090df04629d9c2b5493dc5c2bf46f1834040133a07", "https://bcr.bazel.build/modules/rules_cc/0.2.8/MODULE.bazel": "f1df20f0bf22c28192a794f29b501ee2018fa37a3862a1a2132ae2940a23a642", @@ -205,7 +203,7 @@ "moduleExtensions": { "@@rules_kotlin+//src/main/starlark/core/repositories:bzlmod_setup.bzl%rules_kotlin_extensions": { "general": { - "bzlTransitiveDigest": "+Kp6j204mBZ3mxlIDDR0gBoP45BZ4jYRhRAcB8sU0qc=", + "bzlTransitiveDigest": "WDuVCwqV73DvcHB5Of+02lhUZgPxr9MqtpvMRyUi0eQ=", "usagesDigest": "QI2z8ZUR+mqtbwsf2fLqYdJAkPOHdOV+tF2yVAUgRzw=", "recordedInputs": [ "REPO_MAPPING:rules_kotlin+,bazel_tools bazel_tools" @@ -262,7 +260,7 @@ }, "@@rules_python+//python/extensions:config.bzl%config": { "general": { - "bzlTransitiveDigest": "dzD8Q2YmrP3fz8saWLHPmlwPLO91ImtTmP/c9JKTStM=", + "bzlTransitiveDigest": "L+2ZCpb7GIJqvWaXpzkfk/80d916sIFVmSPzrSm6of8=", "usagesDigest": "ZVSXMAGpD+xzVNPuvF1IoLBkty7TROO0+akMapt1pAg=", "recordedInputs": [ "REPO_MAPPING:rules_python+,bazel_tools bazel_tools", @@ -430,7 +428,7 @@ }, "@@rules_python+//python/uv:uv.bzl%uv": { "general": { - "bzlTransitiveDigest": "ijW9KS7qsIY+yBVvJ+Nr1mzwQox09j13DnE3iIwaeTM=", + "bzlTransitiveDigest": "IgkgxNrtlAec7OnaFrWoCC5NQhHcY/aIz4jsqcNMm8Y=", "usagesDigest": "H8dQoNZcoqP+Mu0tHZTi4KHATzvNkM5ePuEqoQdklIU=", "recordedInputs": [ "REPO_MAPPING:rules_python+,bazel_tools bazel_tools", diff --git a/toolbelt/BUILD.bazel b/toolbelt/BUILD.bazel index 4fad7ab..0c9da20 100644 --- a/toolbelt/BUILD.bazel +++ b/toolbelt/BUILD.bazel @@ -5,6 +5,7 @@ load("@rules_cc//cc:defs.bzl", "cc_library", "cc_test") cc_library( name = "toolbelt", srcs = [ + "async_waiter.cc", "color.cc", "fd.cc", "hexdump.cc", @@ -17,6 +18,7 @@ cc_library( "stacktrace.cc", ], hdrs = [ + "async_waiter.h", "bitset.h", "clock.h", "color.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", ], ) diff --git a/toolbelt/async_waiter.cc b/toolbelt/async_waiter.cc new file mode 100644 index 0000000..e58d305 --- /dev/null +++ b/toolbelt/async_waiter.cc @@ -0,0 +1,60 @@ +// Copyright 2026 Mikael Persson +// All Rights Reserved +// See LICENSE file for licensing information. +#include "toolbelt/async_waiter.h" + +#include +#include +#include + +namespace toolbelt { + +PosixPoller::PosixPoller(uint64_t yield_sleep_ns) : AsyncWaiter(), yield_sleep_ns_(yield_sleep_ns) {} + +void PosixPoller::Yield() const { + if (yield_sleep_ns_ == 0) { + std::this_thread::yield(); + } else { + std::this_thread::sleep_for(std::chrono::nanoseconds(yield_sleep_ns_)); + } +} + +void PosixPoller::Nanosleep(uint64_t ns) const { + std::this_thread::sleep_for(std::chrono::nanoseconds(ns)); +} + +void PosixPoller::AddToUserWaitFds(int fd, uint32_t event_mask) const { + user_fds_.push_back({.fd = fd, .events = short(event_mask)}); +} + +int PosixPoller::PollImpl(std::vector &fds, int timeout_ms) const { + int ret = ::poll(user_fds_.data(), user_fds_.size(), timeout_ms); + if (ret < 0) { + return -2; + } + if (ret == 0) { + return -1; + } + for (auto &pfd : fds) { + if (pfd.revents & (pfd.events | POLLOUT)) { + return pfd.fd; + } + } + return -1; +} + +int PosixPoller::WaitOnUserWaitFds(uint64_t timeout_ns) const { + int timeout_ms = -1; + if (timeout_ns != 0) { + timeout_ms = std::max(1, static_cast(timeout_ns / 1000000ULL)); + } + int result = PollImpl(user_fds_, timeout_ms); + user_fds_.clear(); + return result; +} + +int PosixPoller::PollWithMutableFds(std::vector &fds) const { + return PollImpl(fds, 0); +} + +} // namespace toolbelt diff --git a/toolbelt/async_waiter.h b/toolbelt/async_waiter.h new file mode 100644 index 0000000..ed2f21e --- /dev/null +++ b/toolbelt/async_waiter.h @@ -0,0 +1,278 @@ +// Copyright 2026 Mikael Persson +// All Rights Reserved +// See LICENSE file for licensing information. + +#pragma once + +#include + +#include +#include +#include + +namespace toolbelt { + +namespace toolbelt_details { + +template +inline uint64_t ChronoToNanoseconds(std::chrono::duration duration) { + using FloatingNanoseconds = std::chrono::duration; + const long double count = FloatingNanoseconds(duration).count(); + if (!(count > 0)) { + return 0; + } + const long double maximum = + static_cast(std::numeric_limits::max()); + if (count >= maximum) { + return std::numeric_limits::max(); + } + return static_cast(count); +} + +} // namespace toolbelt_details + +// This is to provide the epoll equivalent of waiting for a set +// of pollfds +struct WaitFd { + WaitFd(int f, uint32_t e) : fd(f), events(e) {} + int fd; + uint32_t events; +}; + +// This is a AsyncWaiter. It's an abstract interface to a poll/wait/yield +// mechanism, such as a vanilla POSIX implementation or coroutines. +class AsyncWaiter { +public: + AsyncWaiter() = default; + AsyncWaiter(const AsyncWaiter&) = delete; + AsyncWaiter(AsyncWaiter&&) = delete; + AsyncWaiter& operator=(const AsyncWaiter&) = delete; + AsyncWaiter& operator=(AsyncWaiter&&) = delete; + virtual ~AsyncWaiter() = default; + + // Yield cpu time (to another thread, process, coroutine, etc.). + virtual void Yield() const = 0; + + // For all Poll functions, there is no timeout because those are immediate + // polls that just check current readiness status without blocking. + // Returns -1 for no fd ready, fd if one is ready. + // Returns -2 for an error during poll. + + // Poll a set of file descriptors with a single common event mask. + int Poll(const std::vector &fds, short event_mask = POLLIN) const { + std::vector pfds; + pfds.reserve(fds.size() + 1); + for (auto &fd : fds) { + pfds.push_back({.fd = fd, .events = short(event_mask), .revents = 0}); + } + return PollWithMutableFds(pfds); + } + + // Poll a set of file descriptors with their own event masks. + // Note that the pollfd's are immutable, therefore, this function may copy + // the vector for a typical POSIX poll call that fills in `revents`, and + // upon return, the `revents` are not set. Use the mutable version below. + int Poll(const std::vector &fds) const { + std::vector pfds = fds; + return PollWithMutableFds(pfds); + } + + // Poll a set of file descriptors with their own event masks. + // Upon return, the `revents` of the ready FDs will be set. + // Note that additional FDs might be appended to this set of pollfds for + // internal purposes. + int Poll(std::vector &fds) const { + return PollWithMutableFds(fds); + } + + // For all Wait functions, the timeout is optional and if greater than zero + // specifies a nanosecond timeout. If the timeout occurs before the fd (or + // one of the fds) becomes ready, Wait will return -1. If an fd is ready, Wait + // will return the fd that terminated the wait. + // If poll returned an error, Wait will return -2. + + // Wait for a file descriptor to become ready. Returns the fd if it + // was triggered or -1 for timeout. + int Wait(int fd, uint32_t event_mask = POLLIN, uint64_t timeout_ns = 0) const { + AddToUserWaitFds(fd, event_mask); + return WaitOnUserWaitFds(timeout_ns); + } + + // Wait for a set of fds, all with the same event mask. + int Wait(const std::vector &fds, uint32_t event_mask = POLLIN, + uint64_t timeout_ns = 0) const { + for (auto &fd : fds) { + AddToUserWaitFds(fd, event_mask); + } + return WaitOnUserWaitFds(timeout_ns); + } + + // Wait for a WaitFd. Returns the fd if it was triggered or -1 for timeout. + int Wait(WaitFd fd, uint64_t timeout_ns = 0) const { + AddToUserWaitFds(fd.fd, fd.events); + return WaitOnUserWaitFds(timeout_ns); + } + + // Wait for a set of WaitFds. Each needs to specify an fd and an event. + // Returns the fd that was triggered, or -1 for a timeout. + int Wait(const std::vector &fds, uint64_t timeout_ns = 0) const { + for (auto &fd : fds) { + AddToUserWaitFds(fd.fd, fd.events); + } + return WaitOnUserWaitFds(timeout_ns); + } + + // Wait for a pollfd. Returns the fd if it was triggered or -1 for timeout. + int Wait(struct pollfd fd, uint64_t timeout_ns = 0) const { + AddToUserWaitFds(fd.fd, fd.events); + return WaitOnUserWaitFds(timeout_ns); + } + + // Wait for a set of pollfds. Each needs to specify an fd and an event. + // Returns the fd that was triggered, or -1 for a timeout. + int Wait(const std::vector &fds, + uint64_t timeout_ns = 0) const { + for (auto &fd : fds) { + AddToUserWaitFds(fd.fd, fd.events); + } + return WaitOnUserWaitFds(timeout_ns); + } + + // For all PollAndWait functions, they combine an immediate poll to check for + // readiness and if not, enter a wait (timed-out poll). This can be more efficient + // in some implementation. If the timeout occurs before the fd (or + // one of the fds) becomes ready, Wait will return -1. If an fd is ready, Wait + // will return the fd that terminated the wait. + // If poll returned an error, Wait will return -2. + + // Poll first and if the fd is not ready, wait for it. + int PollAndWait(int fd, uint32_t event_mask = POLLIN, + uint64_t timeout_ns = 0) const { + int n = Poll({fd}, event_mask); + if (n != -1) { + return n; + } + return Wait(fd, event_mask, timeout_ns); + } + + // Wait for a set of fds, all with the same event mask. + int PollAndWait(const std::vector &fds, uint32_t event_mask = POLLIN, + uint64_t timeout_ns = 0) const { + int n = Poll(fds, event_mask); + if (n != -1) { + return n; + } + return Wait(fds, event_mask, timeout_ns); + } + + int PollAndWait(WaitFd fd, uint64_t timeout_ns = 0) const { + int n = Poll({(struct pollfd){.fd = fd.fd, .events = short(fd.events)}}); + if (n != -1) { + return fd.fd; + } + return Wait(fd, timeout_ns); + } + + int PollAndWait(const std::vector &fds, + uint64_t timeout_ns = 0) const { + std::vector pfds; + pfds.reserve(fds.size()); + for (auto &fd : fds) { + pfds.push_back({.fd = fd.fd, .events = short(fd.events), .revents = 0}); + } + int n = Poll(pfds); + if (n != -1) { + return n; + } + for (auto &fd : fds) { + AddToUserWaitFds(fd.fd, fd.events); + } + return WaitOnUserWaitFds(timeout_ns); + } + + // Wait for a pollfd. Returns the fd if it was triggered or -1 for timeout. + int PollAndWait(struct pollfd fd, uint64_t timeout_ns = 0) const { + int n = Poll({fd}); + if (n != -1) { + return fd.fd; + } + return Wait(fd, timeout_ns); + } + + // Wait for a set of pollfds. Each needs to specify an fd and an event. + // Returns the fd that was triggered, or -1 for a timeout. + int PollAndWait(const std::vector &fds, + uint64_t timeout_ns = 0) const { + int n = Poll(fds); + if (n != -1) { + return n; + } + return Wait(fds, timeout_ns); + } + + // Templated waits with chrono timeouts. + template + int Wait(const T &fd, uint32_t events, + std::chrono::duration duration) const { + return Wait( + fd, events, toolbelt_details::ChronoToNanoseconds(duration)); + } + + template + int Wait(const T &fd, std::chrono::duration duration) const { + return Wait( + fd, POLLIN, toolbelt_details::ChronoToNanoseconds(duration)); + } + + template + int PollAndWait(const T &fd, uint32_t events, + std::chrono::duration duration) const { + return PollAndWait( + fd, events, toolbelt_details::ChronoToNanoseconds(duration)); + } + + template + int PollAndWait(const T &fd, + std::chrono::duration duration) const { + return PollAndWait( + fd, POLLIN, toolbelt_details::ChronoToNanoseconds(duration)); + } + + // Sleeping functions. + virtual void Nanosleep(uint64_t ns) const = 0; + void Millisleep(time_t msecs) const { + Nanosleep(static_cast(msecs) * 1000000LL); + } + void Sleep(time_t secs) const { + Nanosleep(static_cast(secs) * 1000000000LL); + } + + template + void Sleep(std::chrono::duration duration) const { + Nanosleep(toolbelt_details::ChronoToNanoseconds(duration)); + } + +protected: + virtual void AddToUserWaitFds(int fd, uint32_t event_mask) const = 0; + virtual int WaitOnUserWaitFds(uint64_t timeout_ns) const = 0; + virtual int PollWithMutableFds(std::vector &fds) const = 0; +}; + +class PosixPoller : public AsyncWaiter { +public: + explicit PosixPoller(uint64_t yield_sleep_ns = 0); + + void Yield() const override; + void Nanosleep(uint64_t ns) const override; +protected: + void AddToUserWaitFds(int fd, uint32_t event_mask) const override; + int WaitOnUserWaitFds(uint64_t timeout_ns) const override; + int PollWithMutableFds(std::vector &fds) const override; +private: + uint64_t yield_sleep_ns_ = 0; + mutable std::vector user_fds_; + + int PollImpl(std::vector &fds, int timeout_ms) const; +}; + +} // namespace toolbelt diff --git a/toolbelt/fd.cc b/toolbelt/fd.cc index a160646..412c2d7 100644 --- a/toolbelt/fd.cc +++ b/toolbelt/fd.cc @@ -16,7 +16,7 @@ void CloseAllFds(std::function predicate) { } absl::StatusOr FileDescriptor::Read(void *buffer, size_t length, - const co::Coroutine *c) { + const AsyncWaiter *c) { char *buf = reinterpret_cast(buffer); size_t total = 0; while (total < length) { @@ -58,7 +58,7 @@ absl::StatusOr FileDescriptor::Read(void *buffer, size_t length, } absl::StatusOr FileDescriptor::Write(const void *buffer, size_t length, - const co::Coroutine *c) { + const AsyncWaiter *c) { const char *buf = reinterpret_cast(buffer); size_t total = 0; diff --git a/toolbelt/fd.h b/toolbelt/fd.h index 5760b56..4b63c37 100644 --- a/toolbelt/fd.h +++ b/toolbelt/fd.h @@ -18,7 +18,7 @@ #include #include #include -#include "co/coroutine.h" +#include "toolbelt/async_waiter.h" namespace toolbelt { @@ -188,9 +188,9 @@ class FileDescriptor { return absl::OkStatus(); } - absl::StatusOr Read(void* buffer, size_t length, const co::Coroutine* c = nullptr); + absl::StatusOr Read(void* buffer, size_t length, const AsyncWaiter* c = nullptr); absl::StatusOr Write(const void* buffer, size_t length, - const co::Coroutine* c = nullptr); + const AsyncWaiter* 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 diff --git a/toolbelt/payload_buffer.h b/toolbelt/payload_buffer.h index 3feca78..42bf568 100644 --- a/toolbelt/payload_buffer.h +++ b/toolbelt/payload_buffer.h @@ -531,7 +531,6 @@ inline void PayloadBuffer::VectorPush(PayloadBuffer **self, VectorHeader *hdr, // by the block size (in bytes). BufferOffset hdr_offset = (*self)->ToOffset(hdr); - uint32_t total_size = hdr->num_elements * sizeof(T); if (hdr->data == 0) { // The vector is empty, allocate it with a default size of 2. void *vecp = Allocate(self, 2 * sizeof(T), true, enable_small_block); diff --git a/toolbelt/pipe.cc b/toolbelt/pipe.cc index 0cb8312..919c690 100644 --- a/toolbelt/pipe.cc +++ b/toolbelt/pipe.cc @@ -83,7 +83,7 @@ absl::Status Pipe::SetPipeSize(size_t size) { } absl::StatusOr Pipe::Read(char *buffer, size_t length, - const co::Coroutine *c) { + const AsyncWaiter *c) { size_t total = 0; ScopedRead sc(*this, c); @@ -123,7 +123,7 @@ absl::StatusOr Pipe::Read(char *buffer, size_t length, } absl::StatusOr Pipe::Write(const char *buffer, size_t length, - const co::Coroutine *c) { + const AsyncWaiter *c) { size_t total = 0; ScopedWrite sc(*this, c); diff --git a/toolbelt/pipe.h b/toolbelt/pipe.h index cfe4749..fd0d27c 100644 --- a/toolbelt/pipe.h +++ b/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/async_waiter.h" #include "toolbelt/fd.h" #include @@ -66,9 +66,9 @@ class Pipe { absl::Status SetPipeSize(size_t size); virtual absl::StatusOr Read(char *buffer, size_t length, - const co::Coroutine *c = nullptr); + const AsyncWaiter *c = nullptr); virtual absl::StatusOr Write(const char *buffer, size_t length, - const co::Coroutine *c = nullptr); + const AsyncWaiter *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 AsyncWaiter *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 AsyncWaiter *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 class SharedPtrPipe : public Pipe { // You can't use raw buffers with shared ptr pipes. absl::StatusOr Read(char *, size_t , - const co::Coroutine * = nullptr) override { + const AsyncWaiter * = nullptr) override { return absl::InternalError("Not supported on SharedPtrPipe"); } absl::StatusOr Write(const char *, size_t , - const co::Coroutine *c = nullptr) override { + const AsyncWaiter *c = nullptr) override { return absl::InternalError("Not supported on SharedPtrPipe"); } - absl::StatusOr> Read(const co::Coroutine *c = nullptr) { + absl::StatusOr> Read(const AsyncWaiter *c = nullptr) { char buffer[sizeof(std::shared_ptr)]; size_t length = sizeof(buffer); size_t total = 0; @@ -205,7 +199,7 @@ template class SharedPtrPipe : public Pipe { } // This makes the pipe an owner of the pointer. - absl::Status Write(std::shared_ptr p, const co::Coroutine *c = nullptr) { + absl::Status Write(std::shared_ptr p, const AsyncWaiter *c = nullptr) { // On entry, ref count for p = N char buffer[sizeof(std::shared_ptr)]; diff --git a/toolbelt/sockets.cc b/toolbelt/sockets.cc index 965a0af..345af6c 100644 --- a/toolbelt/sockets.cc +++ b/toolbelt/sockets.cc @@ -2,7 +2,7 @@ // All Rights Reserved // See LICENSE file for licensing information. -#include "sockets.h" +#include "toolbelt/sockets.h" #include #include @@ -20,7 +20,7 @@ #include #include "absl/strings/str_format.h" -#include "hexdump.h" +#include "toolbelt/async_waiter.h" namespace toolbelt { @@ -157,7 +157,7 @@ std::string VirtualAddress::ToString() const { return absl::StrFormat("%d:%d", addr_.svm_cid, addr_.svm_port); } -static ssize_t ReceiveFully(const co::Coroutine *c, int fd, size_t length, +static ssize_t ReceiveFully(const AsyncWaiter *c, int fd, size_t length, char *buffer, size_t buflen) { int offset = 0; size_t remaining = length; @@ -193,7 +193,7 @@ static ssize_t ReceiveFully(const co::Coroutine *c, int fd, size_t length, return length; } -static ssize_t SendFully(const co::Coroutine *c, int fd, const char *buffer, +static ssize_t SendFully(const AsyncWaiter *c, int fd, const char *buffer, size_t length, bool blocking) { size_t remaining = length; size_t offset = 0; @@ -242,7 +242,7 @@ static ssize_t SendFully(const co::Coroutine *c, int fd, const char *buffer, } absl::StatusOr Socket::Receive(char *buffer, size_t buflen, - const co::Coroutine *c) { + const AsyncWaiter *c) { if (!Connected()) { return absl::InternalError("Socket is not connected"); } @@ -256,7 +256,7 @@ absl::StatusOr Socket::Receive(char *buffer, size_t buflen, } absl::StatusOr Socket::Send(const char *buffer, size_t length, - const co::Coroutine *c) { + const AsyncWaiter *c) { if (!Connected()) { return absl::InternalError("Socket is not connected"); } @@ -270,7 +270,7 @@ absl::StatusOr Socket::Send(const char *buffer, size_t length, } absl::StatusOr Socket::ReceiveMessage(char *buffer, size_t buflen, - const co::Coroutine *c) { + const AsyncWaiter *c) { if (!Connected()) { return absl::InternalError("Socket is not connected"); } @@ -303,7 +303,7 @@ absl::StatusOr Socket::ReceiveMessage(char *buffer, size_t buflen, } absl::StatusOr> -Socket::ReceiveVariableLengthMessage(const co::Coroutine *c) { +Socket::ReceiveVariableLengthMessage(const AsyncWaiter *c) { if (!Connected()) { return absl::InternalError("Socket is not connected"); } @@ -338,7 +338,7 @@ Socket::ReceiveVariableLengthMessage(const co::Coroutine *c) { } absl::StatusOr Socket::SendMessage(char *buffer, size_t length, - const co::Coroutine *c) { + const AsyncWaiter *c) { if (!Connected()) { return absl::InternalError("Socket is not connected"); } @@ -406,7 +406,7 @@ absl::Status UnixSocket::Bind(const std::string &pathname, bool listen) { return absl::OkStatus(); } -absl::StatusOr UnixSocket::Accept(const co::Coroutine *c) const { +absl::StatusOr UnixSocket::Accept(const AsyncWaiter *c) const { if (!fd_.Valid()) { return absl::InternalError("UnixSocket is not valid"); } @@ -457,7 +457,7 @@ absl::Status UnixSocket::Connect(const std::string &pathname) { } absl::Status UnixSocket::SendFds(const std::vector &fds, - const co::Coroutine *c) { + const AsyncWaiter *c) { if (!Connected()) { return absl::InternalError("Socket is not connected"); } @@ -522,7 +522,7 @@ absl::Status UnixSocket::SendFds(const std::vector &fds, } absl::Status UnixSocket::ReceiveFds(std::vector &fds, - const co::Coroutine *c) { + const AsyncWaiter *c) { if (!Connected()) { return absl::InternalError("Socket is not connected"); } @@ -734,7 +734,7 @@ absl::Status TCPSocket::Bind(const InetAddress &addr, bool listen) { return absl::OkStatus(); } -absl::StatusOr TCPSocket::Accept(const co::Coroutine *c) const { +absl::StatusOr TCPSocket::Accept(const AsyncWaiter *c) const { if (!fd_.Valid()) { return absl::InternalError("Socket is not valid"); } @@ -883,7 +883,7 @@ absl::Status UDPSocket::SetMulticastLoop() { } absl::Status UDPSocket::SendTo(const InetAddress &addr, const void *buffer, - size_t length, const co::Coroutine *c) { + size_t length, const AsyncWaiter *c) { if (c != nullptr) { int fd = c->Wait(fd_.Fd(), POLLOUT); if (fd != fd_.Fd()) { @@ -902,7 +902,7 @@ absl::Status UDPSocket::SendTo(const InetAddress &addr, const void *buffer, } absl::StatusOr UDPSocket::Receive(void *buffer, size_t buflen, - const co::Coroutine *c) { + const AsyncWaiter *c) { if (c != nullptr) { int fd = c->Wait(fd_.Fd(), POLLIN); if (fd != fd_.Fd()) { @@ -918,7 +918,7 @@ absl::StatusOr UDPSocket::Receive(void *buffer, size_t buflen, } absl::StatusOr UDPSocket::ReceiveFrom(InetAddress &sender, void *buffer, size_t buflen, - const co::Coroutine *c) { + const AsyncWaiter *c) { if (c != nullptr) { int fd = c->Wait(fd_.Fd(), POLLIN); if (fd != fd_.Fd()) { @@ -978,7 +978,7 @@ absl::Status VirtualStreamSocket::Bind(const VirtualAddress &addr, } absl::StatusOr -VirtualStreamSocket::Accept(const co::Coroutine *c) const { +VirtualStreamSocket::Accept(const AsyncWaiter *c) const { if (!fd_.Valid()) { return absl::InternalError("Socket is not valid"); } diff --git a/toolbelt/sockets.h b/toolbelt/sockets.h index a159894..9d7a5a8 100644 --- a/toolbelt/sockets.h +++ b/toolbelt/sockets.h @@ -6,8 +6,8 @@ #define __TOOLBELT_SOCKETS_H #include "absl/status/status.h" #include "absl/status/statusor.h" -#include "co/coroutine.h" -#include "fd.h" +#include "toolbelt/async_waiter.h" +#include "toolbelt/fd.h" #include #include #include @@ -354,17 +354,17 @@ class Socket { // Send and receive raw buffers. absl::StatusOr Receive(char *buffer, size_t buflen, - const co::Coroutine *c = nullptr); + const AsyncWaiter *c = nullptr); absl::StatusOr Send(const char *buffer, size_t length, - const co::Coroutine *c = nullptr); + const AsyncWaiter *c = nullptr); // Send and receive length-delimited message. The length is a 4-byte // network byte order (big endian) int as the first 4 bytes and // contains the length of the message. absl::StatusOr ReceiveMessage(char *buffer, size_t buflen, - const co::Coroutine *c = nullptr); + const AsyncWaiter *c = nullptr); absl::StatusOr> - ReceiveVariableLengthMessage(const co::Coroutine *c = nullptr); + ReceiveVariableLengthMessage(const AsyncWaiter *c = nullptr); // For SendMessage, the buffer pointer must be 4 bytes beyond // the actual buffer start, which must be length+4 bytes @@ -372,7 +372,7 @@ class Socket { // at buffer-4. This is to allow us to do a single send // to the socket rather than splitting it into 2. absl::StatusOr SendMessage(char *buffer, size_t length, - const co::Coroutine *c = nullptr); + const AsyncWaiter *c = nullptr); absl::Status SetNonBlocking() { if (absl::Status s = fd_.SetNonBlocking(); !s.ok()) { @@ -405,12 +405,12 @@ class UnixSocket : public Socket { absl::Status Bind(const std::string &pathname, bool listen); absl::Status Connect(const std::string &pathname); - absl::StatusOr Accept(const co::Coroutine *c = nullptr) const; + absl::StatusOr Accept(const AsyncWaiter *c = nullptr) const; absl::Status SendFds(const std::vector &fds, - const co::Coroutine *c = nullptr); + const AsyncWaiter *c = nullptr); absl::Status ReceiveFds(std::vector &fds, - const co::Coroutine *c = nullptr); + const AsyncWaiter *c = nullptr); std::string BoundAddress() const { return bound_address_; } absl::StatusOr GetPeerName() const; @@ -455,12 +455,12 @@ class UDPSocket : public NetworkSocket { // NOTE: Read and Write may or may not work on UDP sockets. Use SendTo and // Receive for datagrams. absl::Status SendTo(const InetAddress &addr, const void *buffer, - size_t length, const co::Coroutine *c = nullptr); + size_t length, const AsyncWaiter *c = nullptr); absl::StatusOr Receive(void *buffer, size_t buflen, - const co::Coroutine *c = nullptr); + const AsyncWaiter *c = nullptr); absl::StatusOr ReceiveFrom(InetAddress &sender, void *buffer, size_t buflen, - const co::Coroutine *c = nullptr); + const AsyncWaiter *c = nullptr); absl::Status SetBroadcast(); absl::Status SetMulticastLoop(); }; @@ -474,7 +474,7 @@ class TCPSocket : public NetworkSocket { absl::Status Bind(const InetAddress &addr, bool listen); - absl::StatusOr Accept(const co::Coroutine *c = nullptr) const; + absl::StatusOr Accept(const AsyncWaiter *c = nullptr) const; absl::StatusOr LocalAddress(int port) const; @@ -491,7 +491,7 @@ class VirtualStreamSocket : public Socket { absl::Status Bind(const VirtualAddress &addr, bool listen); - absl::StatusOr Accept(const co::Coroutine *c = nullptr) const; + absl::StatusOr Accept(const AsyncWaiter *c = nullptr) const; absl::StatusOr LocalAddress(uint32_t port) const; const VirtualAddress &BoundAddress() const { return bound_address_; } @@ -552,7 +552,7 @@ class StreamSocket { return absl::Status(absl::StatusCode::kInternal, "Invalid socket address"); } - absl::StatusOr Accept(const co::Coroutine *c = nullptr) const { + absl::StatusOr Accept(const AsyncWaiter *c = nullptr) const { return std::visit( EyeOfNewt{ [&](const TCPSocket &s) mutable -> absl::StatusOr { @@ -618,7 +618,7 @@ class StreamSocket { // Send and receive raw buffers. absl::StatusOr Receive(char *buffer, size_t buflen, - const co::Coroutine *c = nullptr) { + const AsyncWaiter *c = nullptr) { return std::visit( EyeOfNewt{[&](TCPSocket &s) { return s.Receive(buffer, buflen, c); }, [&](VirtualStreamSocket &s) { @@ -629,7 +629,7 @@ class StreamSocket { } absl::StatusOr Send(const char *buffer, size_t length, - const co::Coroutine *c = nullptr) { + const AsyncWaiter *c = nullptr) { return std::visit( EyeOfNewt{ [&](TCPSocket &s) { return s.Send(buffer, length, c); }, @@ -642,7 +642,7 @@ class StreamSocket { // network byte order (big endian) int as the first 4 bytes and // contains the length of the message. absl::StatusOr ReceiveMessage(char *buffer, size_t buflen, - const co::Coroutine *c = nullptr) { + const AsyncWaiter *c = nullptr) { return std::visit( EyeOfNewt{ [&](TCPSocket &s) { return s.ReceiveMessage(buffer, buflen, c); }, @@ -654,7 +654,7 @@ class StreamSocket { } absl::StatusOr> - ReceiveVariableLengthMessage(const co::Coroutine *c = nullptr) { + ReceiveVariableLengthMessage(const AsyncWaiter *c = nullptr) { return std::visit( EyeOfNewt{ [&](TCPSocket &s) { return s.ReceiveVariableLengthMessage(c); }, @@ -671,7 +671,7 @@ class StreamSocket { // at buffer-4. This is to allow us to do a single send // to the socket rather than splitting it into 2. absl::StatusOr SendMessage(char *buffer, size_t length, - const co::Coroutine *c = nullptr) { + const AsyncWaiter *c = nullptr) { return std::visit( EyeOfNewt{ [&](TCPSocket &s) { return s.SendMessage(buffer, length, c); }, diff --git a/toolbelt/sockets_test.cc b/toolbelt/sockets_test.cc index f766c8c..2a0ef9b 100644 --- a/toolbelt/sockets_test.cc +++ b/toolbelt/sockets_test.cc @@ -7,6 +7,7 @@ #include #include "absl/status/status_matchers.h" #include "toolbelt/hexdump.h" +#include "co/coroutine.h" #define VAR(a) a##__COUNTER__ #define EVAL_AND_ASSERT_OK(expr) EVAL_AND_ASSERT_OK2(VAR(r_), expr) @@ -109,7 +110,7 @@ TEST(SocketsTest, UnixSocket) { absl::Status s2 = socket.ReceiveFds(fds, c); ASSERT_TRUE(s2.ok()); - ASSERT_EQ(3, fds.size()); + ASSERT_EQ(std::size_t{3}, fds.size()); }); co::Coroutine outgoing(scheduler, [&socket_name](co::Coroutine* c) { @@ -198,7 +199,7 @@ TEST(SocketsTest, UnixSocketShortFdCountRead) { std::vector fds; absl::Status s2 = socket.ReceiveFds(fds, c); ASSERT_TRUE(s2.ok()); - ASSERT_EQ(1, fds.size()); + ASSERT_EQ(std::size_t{1}, fds.size()); }); co::Coroutine outgoing(scheduler, [&socket_name](co::Coroutine* c) { @@ -264,7 +265,7 @@ TEST(SocketsTest, TCPSocket) { absl::StatusOr> b = socket.ReceiveVariableLengthMessage(c); ASSERT_TRUE(b.ok()); auto buf = b.value(); - ASSERT_EQ(12, buf.size()); // "hello world\0" + ASSERT_EQ(std::size_t{12}, buf.size()); // "hello world\0" ASSERT_EQ("hello world", std::string(buf.data(), 11)); }); @@ -310,7 +311,7 @@ TEST(SocketsTest, BigTCPSocketNonblocking) { std::cerr << "Mismatch at " << i << ": " << buf[i] << " != " << 'a' + (i % 26) << "\n"; } - ASSERT_EQ('a' + ((i + 4) % 26), buf[i]); + ASSERT_EQ(static_cast('a' + ((i + 4) % 26)), buf[i]); } }); @@ -326,7 +327,7 @@ TEST(SocketsTest, BigTCPSocketNonblocking) { absl::StatusOr nsent = socket.SendMessage(buffer.data() + 4, buffer.size() - 4, c); ASSERT_TRUE(nsent.ok()); - ASSERT_EQ(buffer.size(), nsent.value()); + ASSERT_EQ(buffer.size(), static_cast(nsent.value())); }); scheduler.Run(); @@ -358,7 +359,7 @@ TEST(SocketsTest, BigTCPSocketBlocking) { std::cerr << "Mismatch at " << i << ": " << buf[i] << " != " << 'a' + (i % 26) << "\n"; } - ASSERT_EQ('a' + ((i + 4) % 26), buf[i]); + ASSERT_EQ(static_cast('a' + ((i + 4) % 26)), buf[i]); } }); @@ -373,7 +374,7 @@ TEST(SocketsTest, BigTCPSocketBlocking) { absl::StatusOr nsent = socket.SendMessage(buffer.data() + 4, buffer.size() - 4, c); ASSERT_TRUE(nsent.ok()); - ASSERT_EQ(buffer.size(), nsent.value()); + ASSERT_EQ(buffer.size(), static_cast(nsent.value())); }); std::thread sender([&sendScheduler]() { sendScheduler.Run(); }); @@ -609,7 +610,7 @@ TEST(SocketsTest, UDPSocket_SendAndReceiveUnicast) { ASSERT_TRUE(sender.SendTo(sendto_address, TEST_DATA.data(), TEST_DATA.size()).ok()); std::vector Receive_buffer(TEST_DATA.size()); - ASSERT_EQ(*Receiver.Receive(Receive_buffer.data(), Receive_buffer.size()), TEST_DATA.size()); + ASSERT_EQ(static_cast(*Receiver.Receive(Receive_buffer.data(), Receive_buffer.size())), TEST_DATA.size()); ASSERT_EQ(std::string_view(Receive_buffer.data(), Receive_buffer.size()), TEST_DATA); } @@ -632,7 +633,7 @@ TEST(SocketsTest, UDPSocket_SendAndReceiveBroadcast) { ASSERT_TRUE(sender.SendTo(sendto_address, TEST_DATA.data(), TEST_DATA.size()).ok()); std::vector Receive_buffer(TEST_DATA.size()); - ASSERT_EQ(*Receiver.Receive(Receive_buffer.data(), Receive_buffer.size()), TEST_DATA.size()); + ASSERT_EQ(static_cast(*Receiver.Receive(Receive_buffer.data(), Receive_buffer.size())), TEST_DATA.size()); ASSERT_EQ(std::string_view(Receive_buffer.data(), Receive_buffer.size()), TEST_DATA); } @@ -663,7 +664,7 @@ TEST(SocketsTest, UDPSocket_SendAndReceiveMulticast) { while (absl::Now() < timeout) { auto status_or_len = Receiver.Receive(Receive_buffer.data(), Receive_buffer.size()); if (status_or_len.ok()) { - ASSERT_EQ(*status_or_len, TEST_DATA.size()); + ASSERT_EQ(static_cast(*status_or_len), TEST_DATA.size()); ASSERT_EQ(std::string_view(Receive_buffer.data(), Receive_buffer.size()), TEST_DATA); break; }