Improve scheduling performance

Author: dermesser

test/channel_test.cc

@@ -103,6 +103,35 @@ TEST(ChannelTest, DISABLED_basicWriteReadBench) {
   run_loop(setup);
 }
 
+TEST(ChannelTest, DISABLED_generatorWriteReadBench) {
+  static constexpr int N_iter = 1000000;
+
+  auto reader = [](Channel<int> &chan) -> Promise<void> {
+    MultiPromise<int> gen = chan.getAll();
+    for (int i = 1; i < N_iter; ++i) {
+      EXPECT_EQ(i, co_await gen);
+    }
+  };
+
+  auto writer = [](Channel<int> &chan) -> Promise<void> {
+    for (int i = 1; i < N_iter; ++i) {
+      co_await chan.put(i);
+    }
+  };
+
+  auto setup = [&](const Loop &) -> Promise<void> {
+    Channel<int> chan{2};
+
+    Promise<void> readerCoroutine = reader(chan);
+    Promise<void> writerCoroutine = writer(chan);
+
+    co_await readerCoroutine;
+    co_await writerCoroutine;
+  };
+
+  run_loop(setup);
+}
+
 TEST(ChannelTest, blockingRead) {
 
   auto drain = [](Channel<int> &chan) -> Promise<void> {

uvco/bounded_queue.h

@@ -43,7 +43,7 @@ template <typename T> class BoundedQueue {
   }
   /// Pop an item from the queue.
   T get() {
-    if (empty()) {
+    if (empty()) [[unlikely]] {
       throw UvcoException(UV_EAGAIN, "queue is empty");
     }
     T element = std::move(queue_.at(tail_++));

uvco/channel.h

@@ -144,7 +144,7 @@ template <typename T> class Channel {
       if (!waiters_.hasSpace()) {
         throw UvcoException(
             UV_EBUSY,
-            "only one coroutine can wait for reading/writing a channel");
+            "too many coroutines waiting for reading/writing a channel");
       }
       waiters_.put(handle);
       return true;

uvco/loop/scheduler.cc

@@ -1,5 +1,6 @@
 // uvco (c) 2023 Lewin Bormann. See LICENSE for specific terms.
 
+#include <array>
 #include <fmt/core.h>
 #include <uv.h>
 
@@ -14,37 +15,6 @@ namespace uvco {
 
 namespace {
 
-using BloomFilter = std::size_t;
-
-size_t splitmix64(size_t x) {
-  size_t z = (x + 0x9e3779b97f4a7c15);
-  z = (z ^ (z >> 30)) * 0xbf58476d1ce4e5b9;
-  z = (z ^ (z >> 27)) * 0x94d049bb133111eb;
-  return z ^ (z >> 31);
-}
-
-bool haveSeenOrAdd(BloomFilter &filter, std::coroutine_handle<> handle) {
-  static_assert(sizeof(BloomFilter) == 8, "BloomFilter is not 64 bits");
-  const size_t hash = splitmix64((size_t)handle.address());
-  const unsigned index1 = hash % 64;
-  const unsigned index2 = (hash >> 6U) % 64;
-  const unsigned index3 = (hash >> 12U) % 64;
-  const unsigned index4 = (hash >> 18U) % 64;
-  const unsigned index5 = (hash >> 24U) % 64;
-
-  // More than the first 32 bits appear to not gain much.
-  const size_t bloomIndex = (1U << index1) | (1U << index2) | (1U << index3) |
-                            (1U << index4) | (1U << index5);
-
-  if ((filter & bloomIndex) == bloomIndex) {
-    // Potentially a false positive.
-    return true;
-  }
-  // Definitely not seen before.
-  filter |= bloomIndex;
-  return false;
-}
-
 unsigned findFirstIndexOf(std::span<const std::coroutine_handle<>> handles,
                           std::coroutine_handle<> handle) {
   return std::ranges::find_if(
@@ -61,39 +31,35 @@ void Scheduler::runAll() {
   unsigned turns = 0;
 
   while (!resumableActive_.empty() && turns < maxTurnsBeforeReturning) {
-    BloomFilter seenHandles = 0;
     resumableRunning_.swap(resumableActive_);
     for (unsigned i = 0; i < resumableRunning_.size(); ++i) {
       auto &coro = resumableRunning_[i];
-
       // Defend against resuming the same coroutine twice in the same loop pass.
       // This happens when SelectSet selects two coroutines which return at the
       // same time. Resuming the same handle twice is not good, very bad, and
       // will usually at least cause a heap use-after-free.
 
-      // Explicitly written in an explicit way :)
-      if (!haveSeenOrAdd(seenHandles, coro)) [[likely]] {
-        coro.resume();
-      } else if (findFirstIndexOf(resumableRunning_, coro) == i) {
+      // Check if this coroutine handle has already been resumed. This has
+      // quadratic complexity, but appears to be faster than e.g. a Bloom
+      // filter, because it takes fewer calculations and is a nice linear search
+      // over a usually short vector.
+      if (findFirstIndexOf(resumableRunning_, coro) == i) {
         // This is only true if the coroutine is a false positive in the bloom
         // filter, and has not been run before. The linear search is slow (but
         // not too slow), and only happens in the case of a false positive.
         coro.resume();
-      } else {
-        // This is most likely a SelectSet being awaited, with two coroutines
-        // being ready at the same time.
       }
     }
     resumableRunning_.clear();
     ++turns;
   }
 }
 
-void Scheduler::close() { BOOST_ASSERT(resumableActive_.empty()); }
+void Scheduler::close() { BOOST_ASSERT(!resumableActive_.empty()); }
 
 void Scheduler::enqueue(std::coroutine_handle<> handle) {
   // Use of moved-out Scheduler?
-  BOOST_ASSERT(resumableActive_.capacity() != 0);
+  BOOST_ASSERT(resumableActive_.capacity() > 0);
 
   if (run_mode_ == RunMode::Immediate) {
     handle.resume();
@@ -108,9 +74,8 @@ void Scheduler::setUpLoop(uv_loop_t *loop) { uv_loop_set_data(loop, this); }
 Scheduler::~Scheduler() = default;
 
 Scheduler::Scheduler(RunMode mode) : run_mode_{mode} {
-  static constexpr size_t resumableBufferSize = 16;
-  resumableActive_.reserve(resumableBufferSize);
-  resumableRunning_.reserve(resumableBufferSize);
+  resumableActive_.reserve(16);
+  resumableRunning_.reserve(16);
 }
 
 } // namespace uvco

uvco/loop/scheduler.h

@@ -2,6 +2,7 @@
 
 #pragma once
 
+#include <array>
 #include <boost/assert.hpp>
 #include <uv.h>
 
@@ -62,9 +63,9 @@ class Scheduler {
   explicit Scheduler(RunMode mode = RunMode::Deferred);
 
   Scheduler(const Scheduler &) = delete;
-  Scheduler(Scheduler &&) = default;
+  Scheduler(Scheduler &&) = delete;
   Scheduler &operator=(const Scheduler &) = delete;
-  Scheduler &operator=(Scheduler &&) = default;
+  Scheduler &operator=(Scheduler &&) = delete;
   ~Scheduler();
 
   /// Set up scheduler with event loop. This is required for all uvco
@@ -88,10 +89,10 @@ class Scheduler {
   [[nodiscard]] bool empty() const { return resumableActive_.empty(); }
 
 private:
-  // Vectors of coroutine handles to be resumed.
-  std::vector<std::coroutine_handle<>> resumableActive_ = {};
-  // Vectors of coroutines currently being resumed (while in runAll()).
-  std::vector<std::coroutine_handle<>> resumableRunning_ = {};
+  // Vector of coroutine handles to be resumed.
+  std::vector<std::coroutine_handle<>> resumableActive_;
+  // Vector of coroutines currently being resumed (while in runAll()).
+  std::vector<std::coroutine_handle<>> resumableRunning_;
 
   RunMode run_mode_;
 };

uvco/promise/multipromise.h

@@ -62,8 +62,7 @@ template <typename T> class MultiPromiseCore : public PromiseCore<T> {
   /// next value.
   ///
   /// Called by a `MultiPromiseAwaiter_` when the `MultiPromise` is
-  /// `co_await`ed, so that the generator coroutine can continue yielding the
-  /// next value.
+  /// `co_await`ed, after a value has been successfully awaited.
   void resumeGenerator() {
     BOOST_ASSERT(PromiseCore<T>::state_ != PromiseState::finished);
     if (generatorHandle_) {
@@ -100,15 +99,17 @@ template <typename T> class MultiPromiseCore : public PromiseCore<T> {
     }
   }
 
-  /// Mark generator as finished, yielding no more values.
-  ///
-  /// Called when the generator coroutine returns or throws.
+  /// Mark generator as finished, yielding no more values. Called from within
+  /// the MultiPromise upon return_value and unhandled_exception. From hereon
+  /// awaiting the generator will either rethrow the thrown exception, or yield
+  /// nullopt.
   void terminated() { terminated_ = true; }
 
   /// Check if the generator has been cancelled or has returned.
   [[nodiscard]] bool isTerminated() const { return terminated_; }
 
 private:
+  /// Coroutine handle referring to the suspended generator.
   std::optional<std::coroutine_handle<>> generatorHandle_;
   /// Set to true once the generator coroutine has returned or has been
   /// cancelled.
@@ -161,7 +162,8 @@ template <typename T> class MultiPromise {
     }
   }
 
-  /// Obtain the next value yielded by a generator coroutine.
+  /// Obtain the next value yielded by a generator coroutine. This is less
+  /// efficient than awaiting the multipromise directly.
   Promise<std::optional<T>> next() { co_return (co_await *this); }
 
   /// Return an awaiter for this MultiPromise, which resumes the waiting
@@ -174,7 +176,8 @@ template <typename T> class MultiPromise {
     return MultiPromiseAwaiter_{core_};
   }
 
-  /// Returns true if a value is available.
+  /// Returns true if a value is available, or the generator has returned or
+  /// thrown.
   bool ready() { return core_->slot.has_value() || core_->isTerminated(); }
 
   /// Immediately cancel the suspended generator coroutine. This will drop all
@@ -230,21 +233,19 @@ template <typename T> class MultiPromise {
       if (!core_->slot) {
         // Terminated by co_return
         return std::nullopt;
-      } else {
-        switch (core_->slot->index()) {
-        case 0: {
-          std::optional<T> result = std::move(std::get<0>(core_->slot.value()));
-          core_->slot.reset();
-          return std::move(result);
-        }
-        case 1:
-          // Terminated by exception
-          BOOST_ASSERT(core_->isTerminated());
-          std::rethrow_exception(std::get<1>(core_->slot.value()));
-        default:
-          throw UvcoException(
-              "MultiPromiseAwaiter_::await_resume: invalid slot");
-        }
+      }
+      switch (core_->slot->index()) {
+      [[likely]] case 0: {
+        std::optional<T> result = std::move(std::get<0>(core_->slot.value()));
+        core_->slot.reset();
+        return std::move(result);
+      }
+      case 1:
+        // Terminated by exception
+        BOOST_ASSERT(core_->isTerminated());
+        std::rethrow_exception(std::get<1>(core_->slot.value()));
+      default:
+        throw UvcoException("MultiPromiseAwaiter_::await_resume: invalid slot");
       }
     }
 

uvco/promise/promise.h

@@ -151,9 +151,8 @@ template <typename T> class Promise {
       default:
         throw UvcoException("PromiseAwaiter_::await_resume: invalid slot");
       }
-    } else {
-      throw UvcoException("unwrap called on unfulfilled promise");
     }
+    throw UvcoException("unwrap called on unfulfilled promise");
   }
 
 protected:
@@ -203,10 +202,8 @@ template <typename T> class Promise {
         default:
           throw UvcoException("PromiseAwaiter_::await_resume: invalid slot");
         }
-      } else {
-        throw UvcoException(
-            "await_resume called on unfulfilled promise (bug?)");
       }
+      throw UvcoException("await_resume called on unfulfilled promise (bug?)");
     }
 
     PromiseCore_ &core_;