combinators.h

Go to the documentation of this file.

// uvco (c) 2025 Lewin Bormann. See LICENSE for specific terms.
 
#pragma once
 
#include "uvco/promise/multipromise.h"
#include "uvco/promise/promise.h"
#include "uvco/promise/select.h"
#include <coroutine>
#include <cstddef>
#include <deque>
#include <exception>
#include <functional>
#include <memory>
#include <tuple>
#include <variant>
#include <vector>
 
namespace uvco {


Promise<void> yield();

MultiPromise<unsigned> yield(unsigned count);

template <typename... PromiseTypes>
Promise<std::vector<std::variant<PromiseTypes...>>>
waitAny(Promise<PromiseTypes> &...promises) {
  using S = SelectSet<PromiseTypes...>;
  using V = std::variant<PromiseTypes...>;
  S selectSet{promises...};
  auto readyPromises = co_await selectSet;
  std::vector<V> results;
  for (auto &promise : readyPromises) {
    results.emplace_back(std::visit([](auto *p) -> V { return p->unwrap(); },
                                    std::move(promise)));
  }
  co_return results;
}

template <typename... PromiseTypes>
Promise<std::vector<std::variant<PromiseTypes...>>>
race(Promise<PromiseTypes>... promises) {
  co_return (co_await waitAny(promises...));
}

template <typename... PromiseTypes>
Promise<void> raceIgnore(Promise<PromiseTypes>... promises) {
  using S = SelectSet<PromiseTypes...>;
  S selectSet{promises...};
  co_await selectSet;
}
 
namespace detail {
 
template <typename T> struct ReplaceVoid {
  using type = T;
};
 
struct Void {};
 
template <> struct ReplaceVoid<void> {
  using type = Void;
};
 
template <typename PromiseType>
Promise<typename ReplaceVoid<PromiseType>::type>
awaitAndReplaceVoid(Promise<PromiseType> &promise) {
  co_return (co_await promise);
}
 
template <> inline Promise<Void> awaitAndReplaceVoid(Promise<void> &promise) {
  co_await promise;
  co_return {};
}
 
} // namespace detail

template <typename... PromiseTypes>
Promise<std::tuple<typename detail::ReplaceVoid<PromiseTypes>::type...>>
waitAll(Promise<PromiseTypes>... promises) {
  co_return std::make_tuple(co_await detail::awaitAndReplaceVoid(promises)...);
}

class WaitPoint {
  struct WaitPointAwaiter_;
 
public:
  WaitPoint() = default;
  WaitPoint(const WaitPoint &) = delete;
  WaitPoint(WaitPoint &&) noexcept = default;
  WaitPoint &operator=(const WaitPoint &) = delete;
  WaitPoint &operator=(WaitPoint &&) noexcept = default;
  ~WaitPoint() = default;
 
  [[nodiscard]] size_t waiting() const { return waiters_.size(); }
  Promise<void> wait();
  void releaseOne();
  void releaseAll();
 
private:
  void enqueue(std::coroutine_handle<> handle);
  std::deque<std::coroutine_handle<>> waiters_;
};

class TaskSet {
public:
  TaskSet() = default;
  TaskSet(const TaskSet &) = delete;
  TaskSet(TaskSet &&) noexcept = default;
  TaskSet &operator=(const TaskSet &) = delete;
  TaskSet &operator=(TaskSet &&) noexcept = default;
  virtual ~TaskSet() = default;
 
  using Id = size_t;
  using ErrorCallback = std::function<void(Id, std::exception_ptr)>;

  static std::unique_ptr<TaskSet> create();

  virtual Id add(Promise<void> task) = 0;

  virtual bool empty() = 0;

  virtual Promise<void> onEmpty() = 0;

  virtual void setOnError(ErrorCallback callback) = 0;
};

 
} // namespace uvco