#include <cppcoro/read_only_file.hpp>
#include <cppcoro/task.hpp>

cppcoro::task<int> count_lines(std::string path)
{
  auto file = co_await cppcoro::read_only_file::open(path);

  int lineCount = 0;

  char buffer[1024];
  size_t bytesRead;
  std::uint64_t offset = 0;
  do
  {
    bytesRead = co_await file.read(offset, buffer, sizeof(buffer));
    lineCount += std::count(buffer, buffer + bytesRead, '\n');
    offset += bytesRead;
  } while (bytesRead > 0);

  co_return lineCount;
}

cppcoro::task<> usage_example()
{
  // Calling function creates a new task but doesn't start
  // executing the coroutine yet.
  cppcoro::task<int> countTask = count_lines("foo.txt");

  // ...

  // Coroutine is only started when we later co_await the task.
  int lineCount = co_await countTask;

  std::cout << "line count = " << lineCount << std::endl;
}

// <cppcoro/task.hpp>
namespace cppcoro
{
  template<typename T>
  class task
  {
  public:

    using promise_type = <unspecified>;
    using value_type = T;

    task() noexcept;

    task(task&& other) noexcept;
    task& operator=(task&& other);

    // task is a move-only type.
    task(const task& other) = delete;
    task& operator=(const task& other) = delete;

    // Query if the task result is ready.
    bool is_ready() const noexcept;

    // Wait for the task to complete and return the result or rethrow the
    // exception if the operation completed with an unhandled exception.
    //
    // If the task is not yet ready then the awaiting coroutine will be
    // suspended until the task completes. If the the task is_ready() then
    // this operation will return the result synchronously without suspending.
    Awaiter<T&> operator co_await() const & noexcept;
    Awaiter<T&&> operator co_await() const && noexcept;

    // Returns an awaitable that can be co_await'ed to suspend the current
    // coroutine until the task completes.
    //
    // The 'co_await t.when_ready()' expression differs from 'co_await t' in
    // that when_ready() only performs synchronization, it does not return
    // the result or rethrow the exception.
    //
    // This can be useful if you want to synchronize with the task without
    // the possibility of it throwing an exception.
    Awaitable<void> when_ready() const noexcept;
  };

  template<typename T>
  void swap(task<T>& a, task<T>& b);

  // Creates a task that yields the result of co_await'ing the specified awaitable.
  //
  // This can be used as a form of type-erasure of the concrete awaitable, allowing
  // different awaitables that return the same await-result type to be stored in
  // the same task<RESULT> type.
  template<
    typename AWAITABLE,
    typename RESULT = typename awaitable_traits<AWAITABLE>::await_result_t>
  task<RESULT> make_task(AWAITABLE awaitable);
}

namespace cppcoro
{
  template<typename T = void>
  class shared_task
  {
  public:

    using promise_type = <unspecified>;
    using value_type = T;

    shared_task() noexcept;
    shared_task(const shared_task& other) noexcept;
    shared_task(shared_task&& other) noexcept;
    shared_task& operator=(const shared_task& other) noexcept;
    shared_task& operator=(shared_task&& other) noexcept;

    void swap(shared_task& other) noexcept;

    // Query if the task has completed and the result is ready.
    bool is_ready() const noexcept;

    // Returns an operation that when awaited will suspend the
    // current coroutine until the task completes and the result
    // is available.
    //
    // The type of the result of the 'co_await someTask' expression
    // is an l-value reference to the task's result value (unless T
    // is void in which case the expression has type 'void').
    // If the task completed with an unhandled exception then the
    // exception will be rethrown by the co_await expression.
    Awaiter<T&> operator co_await() const noexcept;

    // Returns an operation that when awaited will suspend the
    // calling coroutine until the task completes and the result
    // is available.
    //
    // The result is not returned from the co_await expression.
    // This can be used to synchronize with the task without the
    // possibility of the co_await expression throwing an exception.
    Awaiter<void> when_ready() const noexcept;

  };

  template<typename T>
  bool operator==(const shared_task<T>& a, const shared_task<T>& b) noexcept;
  template<typename T>
  bool operator!=(const shared_task<T>& a, const shared_task<T>& b) noexcept;

  template<typename T>
  void swap(shared_task<T>& a, shared_task<T>& b) noexcept;

  // Wrap an awaitable value in a shared_task to allow multiple coroutines
  // to concurrently await the result.
  template<
    typename AWAITABLE,
    typename RESULT = typename awaitable_traits<AWAITABLE>::await_result_t>
  shared_task<RESULT> make_shared_task(AWAITABLE awaitable);
}

cppcoro::generator<const std::uint64_t> fibonacci()
{
  std::uint64_t a = 0, b = 1;
  while (true)
  {
    co_yield b;
    auto tmp = a;
    a = b;
    b += tmp;
  }
}

void usage()
{
  for (auto i : fibonacci())
  {
    if (i > 1'000'000) break;
    std::cout << i << std::endl;
  }
}

namespace cppcoro
{
    template<typename T>
    class generator
    {
    public:

        using promise_type = <unspecified>;

        class iterator
        {
        public:
            using iterator_category = std::input_iterator_tag;
            using value_type = std::remove_reference_t<T>;
            using reference = value_type&;
            using pointer = value_type*;
            using difference_type = std::size_t;

            iterator(const iterator& other) noexcept;
            iterator& operator=(const iterator& other) noexcept;

            // If the generator coroutine throws an unhandled exception before producing
            // the next element then the exception will propagate out of this call.
            iterator& operator++();

            reference operator*() const noexcept;
            pointer operator->() const noexcept;

            bool operator==(const iterator& other) const noexcept;
            bool operator!=(const iterator& other) const noexcept;
        };

        // Constructs to the empty sequence.
        generator() noexcept;

        generator(generator&& other) noexcept;
        generator& operator=(generator&& other) noexcept;

        generator(const generator& other) = delete;
        generator& operator=(const generator&) = delete;

        ~generator();

        // Starts executing the generator coroutine which runs until either a value is yielded
        // or the coroutine runs to completion or an unhandled exception propagates out of the
        // the coroutine.
        iterator begin();

        iterator end() noexcept;

        // Swap the contents of two generators.
        void swap(generator& other) noexcept;

    };

    template<typename T>
    void swap(generator<T>& a, generator<T>& b) noexcept;

    // Apply function, func, lazily to each element of the source generator
    // and yield a sequence of the results of calls to func().
    template<typename FUNC, typename T>
    generator<std::invoke_result_t<FUNC, T&>> fmap(FUNC func, generator<T> source);
}

// Lists the immediate contents of a directory.
cppcoro::generator<dir_entry> list_directory(std::filesystem::path path);

cppcoro::recursive_generator<dir_entry> list_directory_recursive(std::filesystem::path path)
{
  for (auto& entry : list_directory(path))
  {
    co_yield entry;
    if (entry.is_directory())
    {
      co_yield list_directory_recursive(entry.path());
    }
  }
}

cppcoro::async_generator<int> ticker(int count, threadpool& tp)
{
  for (int i = 0; i < count; ++i)
  {
    co_await tp.delay(std::chrono::seconds(1));
    co_yield i;
  }
}

cppcoro::task<> consumer(threadpool& tp)
{
  auto sequence = ticker(10, tp);
  for co_await(std::uint32_t i : sequence)
  {
    std::cout << "Tick " << i << std::endl;
  }
}

// <cppcoro/async_generator.hpp>
namespace cppcoro
{
  template<typename T>
  class async_generator
  {
  public:

    class iterator
    {
    public:
      using iterator_tag = std::forward_iterator_tag;
      using difference_type = std::size_t;
      using value_type = std::remove_reference_t<T>;
      using reference = value_type&;
      using pointer = value_type*;

      iterator(const iterator& other) noexcept;
      iterator& operator=(const iterator& other) noexcept;

      // Resumes the generator coroutine if suspended
      // Returns an operation object that must be awaited to wait
      // for the increment operation to complete.
      // If the coroutine runs to completion then the iterator
      // will subsequently become equal to the end() iterator.
      // If the coroutine completes with an unhandled exception then
      // that exception will be rethrown from the co_await expression.
      Awaitable<iterator&> operator++() noexcept;

      // Dereference the iterator.
      pointer operator->() const noexcept;
      reference operator*() const noexcept;

      bool operator==(const iterator& other) const noexcept;
      bool operator!=(const iterator& other) const noexcept;
    };

    // Construct to the empty sequence.
    async_generator() noexcept;
    async_generator(const async_generator&) = delete;
    async_generator(async_generator&& other) noexcept;
    ~async_generator();

    async_generator& operator=(const async_generator&) = delete;
    async_generator& operator=(async_generator&& other) noexcept;

    void swap(async_generator& other) noexcept;

    // Starts execution of the coroutine and returns an operation object
    // that must be awaited to wait for the first value to become available.
    // The result of co_await'ing the returned object is an iterator that
    // can be used to advance to subsequent elements of the sequence.
    //
    // This method is not valid to be called once the coroutine has
    // run to completion.
    Awaitable<iterator> begin() noexcept;
    iterator end() noexcept;

  };

  template<typename T>
  void swap(async_generator<T>& a, async_generator<T>& b);

  // Apply 'func' to each element of the source generator, yielding a sequence of
  // the results of calling 'func' on the source elements.
  template<typename FUNC, typename T>
  async_generator<std::invoke_result_t<FUNC, T&>> fmap(FUNC func, async_generator<T> source);
}

// <cppcoro/single_consumer_event.hpp>
namespace cppcoro
{
  class single_consumer_event
  {
  public:
    single_consumer_event(bool initiallySet = false) noexcept;
    bool is_set() const noexcept;
    void set();
    void reset() noexcept;
    Awaiter<void> operator co_await() const noexcept;
  };
}

#include <cppcoro/single_consumer_event.hpp>

cppcoro::single_consumer_event event;
std::string value;

cppcoro::task<> consumer()
{
  // Coroutine will suspend here until some thread calls event.set()
  // eg. inside the producer() function below.
  co_await event;

  std::cout << value << std::endl;
}

void producer()
{
  value = "foo";

  // This will resume the consumer() coroutine inside the call to set()
  // if it is currently suspended.
  event.set();
}

// <cppcoro/single_consumer_async_auto_reset_event.hpp>
namespace cppcoro
{
  class single_consumer_async_auto_reset_event
  {
  public:

    single_consumer_async_auto_reset_event(
      bool initiallySet = false) noexcept;

    // Change the event to the 'set' state. If a coroutine is awaiting the
    // event then the event is immediately transitioned back to the 'not set'
    // state and the coroutine is resumed.
    void set() noexcept;

    // Returns an Awaitable type that can be awaited to wait until
    // the event becomes 'set' via a call to the .set() method. If
    // the event is already in the 'set' state then the coroutine
    // continues without suspending.
    // The event is automatically reset back to the 'not set' state
    // before resuming the coroutine.
    Awaiter<void> operator co_await() const noexcept;

  };
}

std::atomic<int> value;
cppcoro::single_consumer_async_auto_reset_event valueDecreasedEvent;

cppcoro::task<> wait_until_value_is_below(int limit)
{
  while (value.load(std::memory_order_relaxed) >= limit)
  {
    // Wait until there has been some change that we're interested in.
    co_await valueDecreasedEvent;
  }
}

void change_value(int delta)
{
  value.fetch_add(delta, std::memory_order_relaxed);
  // Notify the waiter if there has been some change.
  if (delta < 0) valueDecreasedEvent.set();
}

// <cppcoro/async_mutex.hpp>
namespace cppcoro
{
  class async_mutex_lock;
  class async_mutex_lock_operation;
  class async_mutex_scoped_lock_operation;

  class async_mutex
  {
  public:
    async_mutex() noexcept;
    ~async_mutex();

    async_mutex(const async_mutex&) = delete;
    async_mutex& operator(const async_mutex&) = delete;

    bool try_lock() noexcept;
    async_mutex_lock_operation lock_async() noexcept;
    async_mutex_scoped_lock_operation scoped_lock_async() noexcept;
    void unlock();
  };

  class async_mutex_lock_operation
  {
  public:
    bool await_ready() const noexcept;
    bool await_suspend(std::experimental::coroutine_handle<> awaiter) noexcept;
    void await_resume() const noexcept;
  };

  class async_mutex_scoped_lock_operation
  {
  public:
    bool await_ready() const noexcept;
    bool await_suspend(std::experimental::coroutine_handle<> awaiter) noexcept;
    [[nodiscard]] async_mutex_lock await_resume() const noexcept;
  };

  class async_mutex_lock
  {
  public:
    // Takes ownership of the lock.
    async_mutex_lock(async_mutex& mutex, std::adopt_lock_t) noexcept;

    // Transfer ownership of the lock.
    async_mutex_lock(async_mutex_lock&& other) noexcept;

    async_mutex_lock(const async_mutex_lock&) = delete;
    async_mutex_lock& operator=(const async_mutex_lock&) = delete;

    // Releases the lock by calling unlock() on the mutex.
    ~async_mutex_lock();
  };
}

#include <cppcoro/async_mutex.hpp>
#include <cppcoro/task.hpp>
#include <set>
#include <string>

cppcoro::async_mutex mutex;
std::set<std::string> values;

cppcoro::task<> add_item(std::string value)
{
  cppcoro::async_mutex_lock lock = co_await mutex.scoped_lock_async();
  values.insert(std::move(value));
}

cppcoro::async_manual_reset_event event;
std::string value;

void producer()
{
  value = get_some_string_value();

  // Publish a value by setting the event.
  event.set();
}

// Can be called many times to create many tasks.
// All consumer tasks will wait until value has been published.
cppcoro::task<> consumer()
{
  // Wait until value has been published by awaiting event.
  co_await event;

  consume_value(value);
}

namespace cppcoro
{
  class async_manual_reset_event_operation;

  class async_manual_reset_event
  {
  public:
    async_manual_reset_event(bool initiallySet = false) noexcept;
    ~async_manual_reset_event();

    async_manual_reset_event(const async_manual_reset_event&) = delete;
    async_manual_reset_event(async_manual_reset_event&&) = delete;
    async_manual_reset_event& operator=(const async_manual_reset_event&) = delete;
    async_manual_reset_event& operator=(async_manual_reset_event&&) = delete;

    // Wait until the event becomes set.
    async_manual_reset_event_operation operator co_await() const noexcept;

    bool is_set() const noexcept;

    void set() noexcept;

    void reset() noexcept;

  };

  class async_manual_reset_event_operation
  {
  public:
    async_manual_reset_event_operation(async_manual_reset_event& event) noexcept;

    bool await_ready() const noexcept;
    bool await_suspend(std::experimental::coroutine_handle<> awaiter) noexcept;
    void await_resume() const noexcept;
  };
}

// <cppcoro/async_auto_reset_event.hpp>
namespace cppcoro
{
  class async_auto_reset_event_operation;

  class async_auto_reset_event
  {
  public:

    async_auto_reset_event(bool initiallySet = false) noexcept;

    ~async_auto_reset_event();

    async_auto_reset_event(const async_auto_reset_event&) = delete;
    async_auto_reset_event(async_auto_reset_event&&) = delete;
    async_auto_reset_event& operator=(const async_auto_reset_event&) = delete;
    async_auto_reset_event& operator=(async_auto_reset_event&&) = delete;

    // Wait for the event to enter the 'set' state.
    //
    // If the event is already 'set' then the event is set to the 'not set'
    // state and the awaiting coroutine continues without suspending.
    // Otherwise, the coroutine is suspended and later resumed when some
    // thread calls 'set()'.
    //
    // Note that the coroutine may be resumed inside a call to 'set()'
    // or inside another thread's call to 'operator co_await()'.
    async_auto_reset_event_operation operator co_await() const noexcept;

    // Set the state of the event to 'set'.
    //
    // If there are pending coroutines awaiting the event then one
    // pending coroutine is resumed and the state is immediately
    // set back to the 'not set' state.
    //
    // This operation is a no-op if the event was already 'set'.
    void set() noexcept;

    // Set the state of the event to 'not-set'.
    //
    // This is a no-op if the state was already 'not set'.
    void reset() noexcept;

  };

  class async_auto_reset_event_operation
  {
  public:
    explicit async_auto_reset_event_operation(async_auto_reset_event& event) noexcept;
    async_auto_reset_event_operation(const async_auto_reset_event_operation& other) noexcept;

    bool await_ready() const noexcept;
    bool await_suspend(std::experimental::coroutine_handle<> awaiter) noexcept;
    void await_resume() const noexcept;

  };
}

// <cppcoro/async_latch.hpp>
namespace cppcoro
{
  class async_latch
  {
  public:

    // Initialise the latch with the specified count.
    async_latch(std::ptrdiff_t initialCount) noexcept;

    // Query if the count has reached zero yet.
    bool is_ready() const noexcept;

    // Decrement the count by n.
    // This will resume any waiting coroutines if the count reaches zero
    // as a result of this call.
    // It is undefined behaviour to decrement the count below zero.
    void count_down(std::ptrdiff_t n = 1) noexcept;

    // Wait until the latch becomes ready.
    // If the latch count is not yet zero then the awaiting coroutine will
    // be suspended and later resumed by a call to count_down() that decrements
    // the count to zero. If the latch count was already zero then the coroutine
    // continues without suspending.
    Awaiter<void> operator co_await() const noexcept;

  };
}

namespace cppcoro
{
  template<typename SEQUENCE = std::size_t,
           typename TRAITS = sequence_traits<SEQUENCE>>
  class sequence_barrier
  {
  public:
    sequence_barrier(SEQUENCE initialSequence = TRAITS::initial_sequence) noexcept;
	~sequence_barrier();

	SEQUENCE last_published() const noexcept;

	// Wait until the specified targetSequence number has been published.
	//
	// If the operation does not complete synchronously then the awaiting
	// coroutine is resumed on the specified scheduler. Otherwise, the
	// coroutine continues without suspending.
	//
	// The co_await expression resumes with the updated last_published()
	// value, which is guaranteed to be at least 'targetSequence'.
	template<typename SCHEDULER>
	[[nodiscard]]
	Awaitable<SEQUENCE> wait_until_published(SEQUENCE targetSequence,
                                             SCHEDULER& scheduler) const noexcept;

    void publish(SEQUENCE sequence) noexcept;
  };
}

// <cppcoro/single_producer_sequencer.hpp>
namespace cppcoro
{
  template<
    typename SEQUENCE = std::size_t,
    typename TRAITS = sequence_traits<SEQUENCE>>
  class single_producer_sequencer
  {
  public:
    using size_type = typename sequence_range<SEQUENCE, TRAITS>::size_type;

    single_producer_sequencer(
      const sequence_barrier<SEQUENCE, TRAITS>& consumerBarrier,
      std::size_t bufferSize,
      SEQUENCE initialSequence = TRAITS::initial_sequence) noexcept;

    // Publisher API:

    template<typename SCHEDULER>
    [[nodiscard]]
    Awaitable<SEQUENCE> claim_one(SCHEDULER& scheduler) noexcept;

    template<typename SCHEDULER>
    [[nodiscard]]
    Awaitable<sequence_range<SEQUENCE>> claim_up_to(
      std::size_t count,
      SCHEDULER& scheduler) noexcept;

    void publish(SEQUENCE sequence) noexcept;

    // Consumer API:

    SEQUENCE last_published() const noexcept;

    template<typename SCHEDULER>
    [[nodiscard]]
    Awaitable<SEQUENCE> wait_until_published(
      SEQUENCE targetSequence,
      SCHEDULER& scheduler) const noexcept;

  };
}

using namespace cppcoro;
using namespace std::chrono;

struct message
{
  int id;
  steady_clock::time_point timestamp;
  float data;
};

constexpr size_t bufferSize = 16384; // Must be power-of-two
constexpr size_t indexMask = bufferSize - 1;
message buffer[bufferSize];

task<void> producer(
  io_service& ioSvc,
  single_producer_sequencer<size_t>& sequencer)
{
  auto start = steady_clock::now();
  for (int i = 0; i < 1'000'000; ++i)
  {
    // Wait until a slot is free in the buffer.
    size_t seq = co_await sequencer.claim_one(ioSvc);

    // Populate the message.
    auto& msg = buffer[seq & indexMask];
    msg.id = i;
    msg.timestamp = steady_clock::now();
    msg.data = 123;

    // Publish the message.
    sequencer.publish(seq);
  }

  // Publish a sentinel
  auto seq = co_await sequencer.claim_one(ioSvc);
  auto& msg = buffer[seq & indexMask];
  msg.id = -1;
  sequencer.publish(seq);
}

task<void> consumer(
  static_thread_pool& threadPool,
  const single_producer_sequencer<size_t>& sequencer,
  sequence_barrier<size_t>& consumerBarrier)
{
  size_t nextToRead = 0;
  while (true)
  {
    // Wait until the next message is available
    // There may be more than one available.
    const size_t available = co_await sequencer.wait_until_published(nextToRead, threadPool);
    do {
      auto& msg = buffer[nextToRead & indexMask];
      if (msg.id == -1)
      {
        consumerBarrier.publish(nextToRead);
        co_return;
      }

      processMessage(msg);
    } while (nextToRead++ != available);

    // Notify the producer that we've finished processing
    // up to 'nextToRead - 1'.
    consumerBarrier.publish(available);
  }
}

task<void> example(io_service& ioSvc, static_thread_pool& threadPool)
{
  sequence_barrier<size_t> barrier;
  single_producer_sequencer<size_t> sequencer{barrier, bufferSize};

  co_await when_all(
    producer(tp, sequencer),
    consumer(tp, sequencer, barrier));
}

// <cppcoro/multi_producer_sequencer.hpp>
namespace cppcoro
{
  template<typename SEQUENCE = std::size_t,
           typename TRAITS = sequence_traits<SEQUENCE>>
  class multi_producer_sequencer
  {
  public:
    multi_producer_sequencer(
      const sequence_barrier<SEQUENCE, TRAITS>& consumerBarrier,
      SEQUENCE initialSequence = TRAITS::initial_sequence);

    std::size_t buffer_size() const noexcept;

    // Consumer interface
    //
    // Each consumer keeps track of their own 'lastKnownPublished' value
    // and must pass this to the methods that query for an updated last-known
    // published sequence number.

    SEQUENCE last_published_after(SEQUENCE lastKnownPublished) const noexcept;

    template<typename SCHEDULER>
    Awaitable<SEQUENCE> wait_until_published(
      SEQUENCE targetSequence,
      SEQUENCE lastKnownPublished,
      SCHEDULER& scheduler) const noexcept;

    // Producer interface

    // Query whether any slots available for claiming (approx.)
    bool any_available() const noexcept;

    template<typename SCHEDULER>
    Awaitable<SEQUENCE> claim_one(SCHEDULER& scheduler) noexcept;

    template<typename SCHEDULER>
    Awaitable<sequence_range<SEQUENCE, TRAITS>> claim_up_to(
      std::size_t count,
      SCHEDULER& scheduler) noexcept;

    // Mark the specified sequence number as published.
    void publish(SEQUENCE sequence) noexcept;

    // Mark all sequence numbers in the specified range as published.
    void publish(const sequence_range<SEQUENCE, TRAITS>& range) noexcept;
  };
}

namespace cppcoro
{
  class cancellation_source
  {
  public:
    // Construct a new, independently cancellable cancellation source.
    cancellation_source();

    // Construct a new reference to the same cancellation state.
    cancellation_source(const cancellation_source& other) noexcept;
    cancellation_source(cancellation_source&& other) noexcept;

    ~cancellation_source();

    cancellation_source& operator=(const cancellation_source& other) noexcept;
    cancellation_source& operator=(cancellation_source&& other) noexcept;

    bool is_cancellation_requested() const noexcept;
    bool can_be_cancelled() const noexcept;
    void request_cancellation();

    cancellation_token token() const noexcept;
  };

  class cancellation_token
  {
  public:
    // Construct a token that can't be cancelled.
    cancellation_token() noexcept;

    cancellation_token(const cancellation_token& other) noexcept;
    cancellation_token(cancellation_token&& other) noexcept;

    ~cancellation_token();

    cancellation_token& operator=(const cancellation_token& other) noexcept;
    cancellation_token& operator=(cancellation_token&& other) noexcept;

    bool is_cancellation_requested() const noexcept;
    void throw_if_cancellation_requested() const;

    // Query if this token can ever have cancellation requested.
    // Code can use this to take a more efficient code-path in cases
    // that the operation does not need to handle cancellation.
    bool can_be_cancelled() const noexcept;
  };

  // RAII class for registering a callback to be executed if cancellation
  // is requested on a particular cancellation token.
  class cancellation_registration
  {
  public:

    // Register a callback to be executed if cancellation is requested.
    // Callback will be called with no arguments on the thread that calls
    // request_cancellation() if cancellation is not yet requested, or
    // called immediately if cancellation has already been requested.
    // Callback must not throw an unhandled exception when called.
    template<typename CALLBACK>
    cancellation_registration(cancellation_token token, CALLBACK&& callback);

    cancellation_registration(const cancellation_registration& other) = delete;

    ~cancellation_registration();
  };

  class operation_cancelled : public std::exception
  {
  public:
    operation_cancelled();
    const char* what() const override;
  };
}

cppcoro::task<> do_something_async(cppcoro::cancellation_token token)
{
  // Explicitly define cancellation points within the function
  // by calling throw_if_cancellation_requested().
  token.throw_if_cancellation_requested();

  co_await do_step_1();

  token.throw_if_cancellation_requested();

  do_step_2();

  // Alternatively, you can query if cancellation has been
  // requested to allow yourself to do some cleanup before
  // returning.
  if (token.is_cancellation_requested())
  {
    display_message_to_user("Cancelling operation...");
    do_cleanup();
    throw cppcoro::operation_cancelled{};
  }

  do_final_step();
}

// Say we already have a timer abstraction that supports being
// cancelled but it doesn't support cancellation_tokens natively.
// You can use a cancellation_registration to register a callback
// that calls the existing cancellation API. e.g.
cppcoro::task<> cancellable_timer_wait(cppcoro::cancellation_token token)
{
  auto timer = create_timer(10s);

  cppcoro::cancellation_registration registration(token, [&]
  {
    // Call existing timer cancellation API.
    timer.cancel();
  });

  co_await timer;
}

namespace cppcoro
{
  class static_thread_pool
  {
  public:
    // Initialise the thread-pool with a number of threads equal to
    // std::thread::hardware_concurrency().
    static_thread_pool();

    // Initialise the thread pool with the specified number of threads.
    explicit static_thread_pool(std::uint32_t threadCount);

    std::uint32_t thread_count() const noexcept;

    class schedule_operation
    {
    public:
      schedule_operation(static_thread_pool* tp) noexcept;

      bool await_ready() noexcept;
      bool await_suspend(std::experimental::coroutine_handle<> h) noexcept;
      bool await_resume() noexcept;

    private:
      // unspecified
    };

    // Return an operation that can be awaited by a coroutine.
    //
    //
    [[nodiscard]]
    schedule_operation schedule() noexcept;

  private:

    // Unspecified

  };
}

cppcoro::task<std::string> do_something_on_threadpool(cppcoro::static_thread_pool& tp)
{
  // First schedule the coroutine onto the threadpool.
  co_await tp.schedule();

  // When it resumes, this coroutine is now running on the threadpool.
  do_something();
}

cppcoro::task<double> dot_product(static_thread_pool& tp, double a[], double b[], size_t count)
{
  if (count > 1000)
  {
    // Subdivide the work recursively into two equal tasks
    // The first half is scheduled to the thread pool so it can run concurrently
    // with the second half which continues on this thread.
    size_t halfCount = count / 2;
    auto [first, second] = co_await when_all(
      schedule_on(tp, dot_product(tp, a, b, halfCount),
      dot_product(tp, a + halfCount, b + halfCount, count - halfCount));
    co_return first + second;
  }
  else
  {
    double sum = 0.0;
    for (size_t i = 0; i < count; ++i)
    {
      sum += a[i] * b[i];
    }
    co_return sum;
  }
}

namespace cppcoro
{
  class io_service
  {
  public:

    class schedule_operation;
    class timed_schedule_operation;

    io_service();
    io_service(std::uint32_t concurrencyHint);

    io_service(io_service&&) = delete;
    io_service(const io_service&) = delete;
    io_service& operator=(io_service&&) = delete;
    io_service& operator=(const io_service&) = delete;

    ~io_service();

    // Scheduler methods

    [[nodiscard]]
    schedule_operation schedule() noexcept;

    template<typename REP, typename RATIO>
    [[nodiscard]]
    timed_schedule_operation schedule_after(
      std::chrono::duration<REP, RATIO> delay,
      cppcoro::cancellation_token cancellationToken = {}) noexcept;

    // Event-loop methods
    //
    // I/O threads must call these to process I/O events and execute
    // scheduled coroutines.

    std::uint64_t process_events();
    std::uint64_t process_pending_events();
    std::uint64_t process_one_event();
    std::uint64_t process_one_pending_event();

    // Request that all threads processing events exit their event loops.
    void stop() noexcept;

    // Query if some thread has called stop()
    bool is_stop_requested() const noexcept;

    // Reset the event-loop after a call to stop() so that threads can
    // start processing events again.
    void reset();

    // Reference-counting methods for tracking outstanding references
    // to the io_service.
    //
    // The io_service::stop() method will be called when the last work
    // reference is decremented.
    //
    // Use the io_work_scope RAII class to manage calling these methods on
    // entry-to and exit-from a scope.
    void notify_work_started() noexcept;
    void notify_work_finished() noexcept;

  };

  class io_service::schedule_operation
  {
  public:
    schedule_operation(const schedule_operation&) noexcept;
    schedule_operation& operator=(const schedule_operation&) noexcept;

    bool await_ready() const noexcept;
    void await_suspend(std::experimental::coroutine_handle<> awaiter) noexcept;
    void await_resume() noexcept;
  };

  class io_service::timed_schedule_operation
  {
  public:
    timed_schedule_operation(timed_schedule_operation&&) noexcept;

    timed_schedule_operation(const timed_schedule_operation&) = delete;
    timed_schedule_operation& operator=(const timed_schedule_operation&) = delete;
    timed_schedule_operation& operator=(timed_schedule_operation&&) = delete;

    bool await_ready() const noexcept;
    void await_suspend(std::experimental::coroutine_handle<> awaiter);
    void await_resume();
  };

  class io_work_scope
  {
  public:

    io_work_scope(io_service& ioService) noexcept;

    io_work_scope(const io_work_scope& other) noexcept;
    io_work_scope(io_work_scope&& other) noexcept;

    ~io_work_scope();

    io_work_scope& operator=(const io_work_scope& other) noexcept;
    io_work_scope& operator=(io_work_scope&& other) noexcept;

    io_service& service() const noexcept;
  };

}

#include <cppcoro/task.hpp>
#include <cppcoro/task.hpp>
#include <cppcoro/io_service.hpp>
#include <cppcoro/read_only_file.hpp>

#include <experimental/filesystem>
#include <memory>
#include <algorithm>
#include <iostream>

namespace fs = std::experimental::filesystem;

cppcoro::task<std::uint64_t> count_lines(cppcoro::io_service& ioService, fs::path path)
{
  auto file = cppcoro::read_only_file::open(ioService, path);

  constexpr size_t bufferSize = 4096;
  auto buffer = std::make_unique<std::uint8_t[]>(bufferSize);

  std::uint64_t newlineCount = 0;

  for (std::uint64_t offset = 0, fileSize = file.size(); offset < fileSize;)
  {
    const auto bytesToRead = static_cast<size_t>(
      std::min<std::uint64_t>(bufferSize, fileSize - offset));

    const auto bytesRead = co_await file.read(offset, buffer.get(), bytesToRead);

    newlineCount += std::count(buffer.get(), buffer.get() + bytesRead, '\n');

    offset += bytesRead;
  }

  co_return newlineCount;
}

cppcoro::task<> run(cppcoro::io_service& ioService)
{
  cppcoro::io_work_scope ioScope(ioService);

  auto lineCount = co_await count_lines(ioService, fs::path{"foo.txt"});

  std::cout << "foo.txt has " << lineCount << " lines." << std::endl;;
}

cppcoro::task<> process_events(cppcoro::io_service& ioService)
{
  // Process events until the io_service is stopped.
  // ie. when the last io_work_scope goes out of scope.
  ioService.process_events();
  co_return;
}

int main()
{
  cppcoro::io_service ioService;

  cppcoro::sync_wait(cppcoro::when_all_ready(
    run(ioService),
    process_events(ioService)));

  return 0;
}

cppcoro::task<> do_something(cppcoro::io_service& ioService)
{
  // Coroutine starts execution on the thread of the task awaiter.

  // A coroutine can transfer execution to an I/O thread by awaiting the
  // result of io_service::schedule().
  co_await ioService.schedule();

  // At this point, the coroutine is now executing on an I/O thread
  // inside a call to one of the io_service event processing methods.

  // A coroutine can also perform a delayed-schedule that will suspend
  // the coroutine for a specified duration of time before scheduling
  // it for resumption on an I/O thread.
  co_await ioService.schedule_after(100ms);

  // At this point, the coroutine is executing on a potentially different I/O thread.
}

namespace cppcoro
{
  class file_read_operation;
  class file_write_operation;

  class file
  {
  public:

    virtual ~file();

    std::uint64_t size() const;

  protected:

    file(file&& other) noexcept;

  };

  class readable_file : public virtual file
  {
  public:

    [[nodiscard]]
    file_read_operation read(
      std::uint64_t offset,
      void* buffer,
      std::size_t byteCount,
      cancellation_token ct = {}) const noexcept;

  };

  class writable_file : public virtual file
  {
  public:

    void set_size(std::uint64_t fileSize);

    [[nodiscard]]
    file_write_operation write(
      std::uint64_t offset,
      const void* buffer,
      std::size_t byteCount,
      cancellation_token ct = {}) noexcept;

  };

  class file_read_operation
  {
  public:

    file_read_operation(file_read_operation&& other) noexcept;

    bool await_ready() const noexcept;
    bool await_suspend(std::experimental::coroutine_handle<> awaiter);
    std::size_t await_resume();

  };

  class file_write_operation
  {
  public:

    file_write_operation(file_write_operation&& other) noexcept;

    bool await_ready() const noexcept;
    bool await_suspend(std::experimental::coroutine_handle<> awaiter);
    std::size_t await_resume();

  };
}

namespace cppcoro
{
  class read_only_file : public readable_file
  {
  public:

    [[nodiscard]]
    static read_only_file open(
      io_service& ioService,
      const std::experimental::filesystem::path& path,
      file_share_mode shareMode = file_share_mode::read,
      file_buffering_mode bufferingMode = file_buffering_mode::default_);

  };

  class write_only_file : public writable_file
  {
  public:

    [[nodiscard]]
    static write_only_file open(
      io_service& ioService,
      const std::experimental::filesystem::path& path,
      file_open_mode openMode = file_open_mode::create_or_open,
      file_share_mode shareMode = file_share_mode::none,
      file_buffering_mode bufferingMode = file_buffering_mode::default_);

  };

  class read_write_file : public readable_file, public writable_file
  {
  public:

    [[nodiscard]]
    static read_write_file open(
      io_service& ioService,
      const std::experimental::filesystem::path& path,
      file_open_mode openMode = file_open_mode::create_or_open,
      file_share_mode shareMode = file_share_mode::none,
      file_buffering_mode bufferingMode = file_buffering_mode::default_);

  };
}

// <cppcoro/net/socket.hpp>
namespace cppcoro::net
{
  class socket
  {
  public:

    static socket create_tcpv4(ip_service& ioSvc);
    static socket create_tcpv6(ip_service& ioSvc);
    static socket create_updv4(ip_service& ioSvc);
    static socket create_udpv6(ip_service& ioSvc);

    socket(socket&& other) noexcept;

    ~socket();

    socket& operator=(socket&& other) noexcept;

    // Return the native socket handle for the socket
    <platform-specific> native_handle() noexcept;

    const ip_endpoint& local_endpoint() const noexcept;
    const ip_endpoint& remote_endpoint() const noexcept;

    void bind(const ip_endpoint& localEndPoint);

    void listen();

    [[nodiscard]]
    Awaitable<void> connect(const ip_endpoint& remoteEndPoint) noexcept;
    [[nodiscard]]
    Awaitable<void> connect(const ip_endpoint& remoteEndPoint,
                            cancellation_token ct) noexcept;

    [[nodiscard]]
    Awaitable<void> accept(socket& acceptingSocket) noexcept;
    [[nodiscard]]
    Awaitable<void> accept(socket& acceptingSocket,
                           cancellation_token ct) noexcept;

    [[nodiscard]]
    Awaitable<void> disconnect() noexcept;
    [[nodiscard]]
    Awaitable<void> disconnect(cancellation_token ct) noexcept;

    [[nodiscard]]
    Awaitable<std::size_t> send(const void* buffer, std::size_t size) noexcept;
    [[nodiscard]]
    Awaitable<std::size_t> send(const void* buffer,
                                std::size_t size,
                                cancellation_token ct) noexcept;

    [[nodiscard]]
    Awaitable<std::size_t> recv(void* buffer, std::size_t size) noexcept;
    [[nodiscard]]
    Awaitable<std::size_t> recv(void* buffer,
                                std::size_t size,
                                cancellation_token ct) noexcept;

    [[nodiscard]]
    socket_recv_from_operation recv_from(
        void* buffer,
        std::size_t size) noexcept;
    [[nodiscard]]
    socket_recv_from_operation_cancellable recv_from(
        void* buffer,
        std::size_t size,
        cancellation_token ct) noexcept;

    [[nodiscard]]
    socket_send_to_operation send_to(
        const ip_endpoint& destination,
        const void* buffer,
        std::size_t size) noexcept;
    [[nodiscard]]
    socket_send_to_operation_cancellable send_to(
        const ip_endpoint& destination,
        const void* buffer,
        std::size_t size,
        cancellation_token ct) noexcept;

    void close_send();
    void close_recv();

  };
}

#include <cppcoro/net/socket.hpp>
#include <cppcoro/io_service.hpp>
#include <cppcoro/cancellation_source.hpp>
#include <cppcoro/async_scope.hpp>
#include <cppcoro/on_scope_exit.hpp>

#include <memory>
#include <iostream>

cppcoro::task<void> handle_connection(socket s)
{
  try
  {
    const size_t bufferSize = 16384;
    auto buffer = std::make_unique<unsigned char[]>(bufferSize);
    size_t bytesRead;
    do {
      // Read some bytes
      bytesRead = co_await s.recv(buffer.get(), bufferSize);

      // Write some bytes
      size_t bytesWritten = 0;
      while (bytesWritten < bytesRead) {
        bytesWritten += co_await s.send(
          buffer.get() + bytesWritten,
          bytesRead - bytesWritten);
      }
    } while (bytesRead != 0);

    s.close_send();

    co_await s.disconnect();
  }
  catch (...)
  {
    std::cout << "connection failed" << std::
  }
}

cppcoro::task<void> echo_server(
  cppcoro::net::ipv4_endpoint endpoint,
  cppcoro::io_service& ioSvc,
  cancellation_token ct)
{
  cppcoro::async_scope scope;

  std::exception_ptr ex;
  try
  {
    auto listeningSocket = cppcoro::net::socket::create_tcpv4(ioSvc);
    listeningSocket.bind(endpoint);
    listeningSocket.listen();

    while (true) {
      auto connection = cppcoro::net::socket::create_tcpv4(ioSvc);
      co_await listeningSocket.accept(connection, ct);
      scope.spawn(handle_connection(std::move(connection)));
    }
  }
  catch (cppcoro::operation_cancelled)
  {
  }
  catch (...)
  {
    ex = std::current_exception();
  }

  // Wait until all handle_connection tasks have finished.
  co_await scope.join();

  if (ex) std::rethrow_exception(ex);
}

int main(int argc, const char* argv[])
{
    cppcoro::io_service ioSvc;

    if (argc != 2) return -1;

    auto endpoint = cppcoro::ipv4_endpoint::from_string(argv[1]);
    if (!endpoint) return -1;

    (void)cppcoro::sync_wait(cppcoro::when_all(
        [&]() -> task<>
        {
            // Shutdown the event loop once finished.
            auto stopOnExit = cppcoro::on_scope_exit([&] { ioSvc.stop(); });

            cppcoro::cancellation_source canceller;
            co_await cppcoro::when_all(
                [&]() -> task<>
                {
                    // Run for 30s then stop accepting new connections.
                    co_await ioSvc.schedule_after(std::chrono::seconds(30));
                    canceller.request_cancellation();
                }(),
                echo_server(*endpoint, ioSvc, canceller.token()));
        }(),
        [&]() -> task<>
        {
            ioSvc.process_events();
        }()));

    return 0;
}

namespace cppcoro::net
{
  class ipv4_address
  {
    using bytes_t = std::uint8_t[4];
  public:
    constexpr ipv4_address();
    explicit constexpr ipv4_address(std::uint32_t integer);
    explicit constexpr ipv4_address(const std::uint8_t(&bytes)[4]);
    explicit constexpr ipv4_address(std::uint8_t b0,
                                    std::uint8_t b1,
                                    std::uint8_t b2,
                                    std::uint8_t b3);

    constexpr const bytes_t& bytes() const;

    constexpr std::uint32_t to_integer() const;

    static constexpr ipv4_address loopback();

    constexpr bool is_loopback() const;
    constexpr bool is_private_network() const;

    constexpr bool operator==(ipv4_address other) const;
    constexpr bool operator!=(ipv4_address other) const;
    constexpr bool operator<(ipv4_address other) const;
    constexpr bool operator>(ipv4_address other) const;
    constexpr bool operator<=(ipv4_address other) const;
    constexpr bool operator>=(ipv4_address other) const;

    std::string to_string();

    static std::optional<ipv4_address> from_string(std::string_view string) noexcept;
  };

  class ipv6_address
  {
    using bytes_t = std::uint8_t[16];
  public:
    constexpr ipv6_address();

    explicit constexpr ipv6_address(
      std::uint64_t subnetPrefix,
      std::uint64_t interfaceIdentifier);

    constexpr ipv6_address(
      std::uint16_t part0,
      std::uint16_t part1,
      std::uint16_t part2,
      std::uint16_t part3,
      std::uint16_t part4,
      std::uint16_t part5,
      std::uint16_t part6,
      std::uint16_t part7);

    explicit constexpr ipv6_address(
        const std::uint16_t(&parts)[8]);

    explicit constexpr ipv6_address(
        const std::uint8_t(bytes)[16]);

    constexpr const bytes_t& bytes() const;

    constexpr std::uint64_t subnet_prefix() const;
    constexpr std::uint64_t interface_identifier() const;

    static constexpr ipv6_address unspecified();
    static constexpr ipv6_address loopback();

    static std::optional<ipv6_address> from_string(std::string_view string) noexcept;

    std::string to_string() const;

    constexpr bool operator==(const ipv6_address& other) const;
    constexpr bool operator!=(const ipv6_address& other) const;
    constexpr bool operator<(const ipv6_address& other) const;
    constexpr bool operator>(const ipv6_address& other) const;
    constexpr bool operator<=(const ipv6_address& other) const;
    constexpr bool operator>=(const ipv6_address& other) const;

  };

  class ip_address
  {
  public:

    // Constructs to IPv4 address 0.0.0.0
    ip_address() noexcept;

    ip_address(ipv4_address address) noexcept;
    ip_address(ipv6_address address) noexcept;

    bool is_ipv4() const noexcept;
    bool is_ipv6() const noexcept;

    const ipv4_address& to_ipv4() const;
    const ipv6_address& to_ipv6() const;

    const std::uint8_t* bytes() const noexcept;

    std::string to_string() const;

    static std::optional<ip_address> from_string(std::string_view string) noexcept;

    bool operator==(const ip_address& rhs) const noexcept;
    bool operator!=(const ip_address& rhs) const noexcept;

    //  ipv4_address sorts less than ipv6_address
    bool operator<(const ip_address& rhs) const noexcept;
    bool operator>(const ip_address& rhs) const noexcept;
    bool operator<=(const ip_address& rhs) const noexcept;
    bool operator>=(const ip_address& rhs) const noexcept;

  };
}

namespace cppcoro::net
{
  class ipv4_endpoint
  {
  public:
    ipv4_endpoint() noexcept;
    explicit ipv4_endpoint(ipv4_address address, std::uint16_t port = 0) noexcept;

    const ipv4_address& address() const noexcept;
    std::uint16_t port() const noexcept;

    std::string to_string() const;
    static std::optional<ipv4_endpoint> from_string(std::string_view string) noexcept;
  };

  bool operator==(const ipv4_endpoint& a, const ipv4_endpoint& b);
  bool operator!=(const ipv4_endpoint& a, const ipv4_endpoint& b);
  bool operator<(const ipv4_endpoint& a, const ipv4_endpoint& b);
  bool operator>(const ipv4_endpoint& a, const ipv4_endpoint& b);
  bool operator<=(const ipv4_endpoint& a, const ipv4_endpoint& b);
  bool operator>=(const ipv4_endpoint& a, const ipv4_endpoint& b);

  class ipv6_endpoint
  {
  public:
    ipv6_endpoint() noexcept;
    explicit ipv6_endpoint(ipv6_address address, std::uint16_t port = 0) noexcept;

    const ipv6_address& address() const noexcept;
    std::uint16_t port() const noexcept;

    std::string to_string() const;
    static std::optional<ipv6_endpoint> from_string(std::string_view string) noexcept;
  };

  bool operator==(const ipv6_endpoint& a, const ipv6_endpoint& b);
  bool operator!=(const ipv6_endpoint& a, const ipv6_endpoint& b);
  bool operator<(const ipv6_endpoint& a, const ipv6_endpoint& b);
  bool operator>(const ipv6_endpoint& a, const ipv6_endpoint& b);
  bool operator<=(const ipv6_endpoint& a, const ipv6_endpoint& b);
  bool operator>=(const ipv6_endpoint& a, const ipv6_endpoint& b);

  class ip_endpoint
  {
  public:
     // Constructs to IPv4 end-point 0.0.0.0:0
     ip_endpoint() noexcept;

     ip_endpoint(ipv4_endpoint endpoint) noexcept;
     ip_endpoint(ipv6_endpoint endpoint) noexcept;

     bool is_ipv4() const noexcept;
     bool is_ipv6() const noexcept;

     const ipv4_endpoint& to_ipv4() const;
     const ipv6_endpoint& to_ipv6() const;

     ip_address address() const noexcept;
     std::uint16_t port() const noexcept;

     std::string to_string() const;

     static std::optional<ip_endpoint> from_string(std::string_view string) noexcept;

     bool operator==(const ip_endpoint& rhs) const noexcept;
     bool operator!=(const ip_endpoint& rhs) const noexcept;

     //  ipv4_endpoint sorts less than ipv6_endpoint
     bool operator<(const ip_endpoint& rhs) const noexcept;
     bool operator>(const ip_endpoint& rhs) const noexcept;
     bool operator<=(const ip_endpoint& rhs) const noexcept;
     bool operator>=(const ip_endpoint& rhs) const noexcept;
  };
}

// <cppcoro/sync_wait.hpp>
namespace cppcoro
{
  template<typename AWAITABLE>
  auto sync_wait(AWAITABLE&& awaitable)
    -> typename awaitable_traits<AWAITABLE&&>::await_result_t;
}

void example_task()
{
  auto makeTask = []() -> task<std::string>
  {
    co_return "foo";
  };

  auto task = makeTask();

  // start the lazy task and wait until it completes
  sync_wait(task); // -> "foo"
  sync_wait(makeTask()); // -> "foo"
}

void example_shared_task()
{
  auto makeTask = []() -> shared_task<std::string>
  {
    co_return "foo";
  };

  auto task = makeTask();
  // start the shared task and wait until it completes
  sync_wait(task) == "foo";
  sync_wait(makeTask()) == "foo";
}

// <cppcoro/when_all_ready.hpp>
namespace cppcoro
{
  // Concurrently await multiple awaitables.
  //
  // Returns an awaitable object that, when co_await'ed, will co_await each of the input
  // awaitable objects and will resume the awaiting coroutine only when all of the
  // component co_await operations complete.
  //
  // Result of co_await'ing the returned awaitable is a std::tuple of detail::when_all_task<T>,
  // one for each input awaitable and where T is the result-type of the co_await expression
  // on the corresponding awaitable.
  //
  // AWAITABLES must be awaitable types and must be movable (if passed as rvalue) or copyable
  // (if passed as lvalue). The co_await expression will be executed on an rvalue of the
  // copied awaitable.
  template<typename... AWAITABLES>
  auto when_all_ready(AWAITABLES&&... awaitables)
    -> Awaitable<std::tuple<detail::when_all_task<typename awaitable_traits<AWAITABLES>::await_result_t>...>>;

  // Concurrently await each awaitable in a vector of input awaitables.
  template<
    typename AWAITABLE,
    typename RESULT = typename awaitable_traits<AWAITABLE>::await_result_t>
  auto when_all_ready(std::vector<AWAITABLE> awaitables)
    -> Awaitable<std::vector<detail::when_all_task<RESULT>>>;
}

task<std::string> get_record(int id);

task<> example1()
{
  // Run 3 get_record() operations concurrently and wait until they're all ready.
  // Returns a std::tuple of tasks that can be unpacked using structured bindings.
  auto [task1, task2, task3] = co_await when_all_ready(
    get_record(123),
    get_record(456),
    get_record(789));

  // Unpack the result of each task
  std::string& record1 = task1.result();
  std::string& record2 = task2.result();
  std::string& record3 = task3.result();

  // Use records....
}

task<> example2()
{
  // Create the input tasks. They don't start executing yet.
  std::vector<task<std::string>> tasks;
  for (int i = 0; i < 1000; ++i)
  {
    tasks.emplace_back(get_record(i));
  }

  // Execute all tasks concurrently.
  std::vector<detail::when_all_task<std::string>> resultTasks =
    co_await when_all_ready(std::move(tasks));

  // Unpack and handle each result individually once they're all complete.
  for (int i = 0; i < 1000; ++i)
  {
    try
    {
      std::string& record = tasks[i].result();
      std::cout << i << " = " << record << std::endl;
    }
    catch (const std::exception& ex)
    {
      std::cout << i << " : " << ex.what() << std::endl;
    }
  }
}

// <cppcoro/when_all.hpp>
namespace cppcoro
{
  // Variadic version.
  //
  // Note that if the result of `co_await awaitable` yields a void-type
  // for some awaitables then the corresponding component for that awaitable
  // in the tuple will be an empty struct of type detail::void_value.
  template<typename... AWAITABLES>
  auto when_all(AWAITABLES&&... awaitables)
    -> Awaitable<std::tuple<typename awaitable_traits<AWAITABLES>::await_result_t...>>;

  // Overload for vector<Awaitable<void>>.
  template<
    typename AWAITABLE,
    typename RESULT = typename awaitable_traits<AWAITABLE>::await_result_t,
    std::enable_if_t<std::is_void_v<RESULT>, int> = 0>
  auto when_all(std::vector<AWAITABLE> awaitables)
    -> Awaitable<void>;

  // Overload for vector<Awaitable<NonVoid>> that yield a value when awaited.
  template<
    typename AWAITABLE,
    typename RESULT = typename awaitable_traits<AWAITABLE>::await_result_t,
    std::enable_if_t<!std::is_void_v<RESULT>, int> = 0>
  auto when_all(std::vector<AWAITABLE> awaitables)
    -> Awaitable<std::vector<std::conditional_t<
         std::is_lvalue_reference_v<RESULT>,
         std::reference_wrapper<std::remove_reference_t<RESULT>>,
         std::remove_reference_t<RESULT>>>>;
}

task<A> get_a();
task<B> get_b();

task<> example1()
{
  // Run get_a() and get_b() concurrently.
  // Task yields a std::tuple<A, B> which can be unpacked using structured bindings.
  auto [a, b] = co_await when_all(get_a(), get_b());

  // use a, b
}

task<std::string> get_record(int id);

task<> example2()
{
  std::vector<task<std::string>> tasks;
  for (int i = 0; i < 1000; ++i)
  {
    tasks.emplace_back(get_record(i));
  }

  // Concurrently execute all get_record() tasks.
  // If any of them fail with an exception then the exception will propagate
  // out of the co_await expression once they have all completed.
  std::vector<std::string> records = co_await when_all(std::move(tasks));

  // Process results
  for (int i = 0; i < 1000; ++i)
  {
    std::cout << i << " = " << records[i] << std::endl;
  }
}

// Given a function you want to apply that converts
// a value of type A to value of type B.
B a_to_b(A value);

// And a task that yields a value of type A
cppcoro::task<A> get_an_a();

// We can apply the function to the result of the task using fmap()
// and obtain a new task yielding the result.
cppcoro::task<B> bTask = fmap(a_to_b, get_an_a());

// An alternative syntax is to use the pipe notation.
cppcoro::task<B> bTask = get_an_a() | cppcoro::fmap(a_to_b);

// <cppcoro/fmap.hpp>
namespace cppcoro
{
  template<typename FUNC>
  struct fmap_transform
  {
    fmap_transform(FUNC&& func) noexcept(std::is_nothrow_move_constructible_v<FUNC>);
    FUNC func;
  };

  // Type-deducing constructor for fmap_transform object that can be used
  // in conjunction with operator|.
  template<typename FUNC>
  fmap_transform<FUNC> fmap(FUNC&& func);

  // operator| overloads for providing pipe-based syntactic sugar for fmap()
  // such that the expression:
  //   <value-expr> | cppcoro::fmap(<func-expr>)
  // is equivalent to:
  //   fmap(<func-expr>, <value-expr>)

  template<typename T, typename FUNC>
  decltype(auto) operator|(T&& value, fmap_transform<FUNC>&& transform);

  template<typename T, typename FUNC>
  decltype(auto) operator|(T&& value, fmap_transform<FUNC>& transform);

  template<typename T, typename FUNC>
  decltype(auto) operator|(T&& value, const fmap_transform<FUNC>& transform);

  // Generic overload for all awaitable types.
  //
  // Returns an awaitable that when co_awaited, co_awaits the specified awaitable
  // and applies the specified func to the result of the 'co_await awaitable'
  // expression as if by 'std::invoke(func, co_await awaitable)'.
  //
  // If the type of 'co_await awaitable' expression is 'void' then co_awaiting the
  // returned awaitable is equivalent to 'co_await awaitable, func()'.
  template<
    typename FUNC,
    typename AWAITABLE,
    std::enable_if_t<is_awaitable_v<AWAITABLE>, int> = 0>
  auto fmap(FUNC&& func, AWAITABLE&& awaitable)
    -> Awaitable<std::invoke_result_t<FUNC, typename awaitable_traits<AWAITABLE>::await_result_t>>;
}

task<record> load_record(int id);

ui_thread_scheduler uiThreadScheduler;

task<> example()
{
  // This will start load_record() on the current thread.
  // Then when load_record() completes (probably on an I/O thread)
  // it will reschedule execution onto thread pool and call to_json
  // Once to_json completes it will transfer execution onto the
  // ui thread before resuming this coroutine and returning the json text.
  task<std::string> jsonTask =
    load_record(123)
    | cppcoro::resume_on(threadpool::default())
    | cppcoro::fmap(to_json)
    | cppcoro::resume_on(uiThreadScheduler);

  // At this point, all we've done is create a pipeline of tasks.
  // The tasks haven't started executing yet.

  // Await the result. Starts the pipeline of tasks.
  std::string jsonText = co_await jsonTask;

  // Guaranteed to be executing on ui thread here.

  someUiControl.set_text(jsonText);
}

// <cppcoro/resume_on.hpp>
namespace cppcoro
{
  template<typename SCHEDULER, typename AWAITABLE>
  auto resume_on(SCHEDULER& scheduler, AWAITABLE awaitable)
    -> Awaitable<typename awaitable_traits<AWAITABLE>::await_traits_t>;

  template<typename SCHEDULER, typename T>
  async_generator<T> resume_on(SCHEDULER& scheduler, async_generator<T> source);

  template<typename SCHEDULER>
  struct resume_on_transform
  {
    explicit resume_on_transform(SCHEDULER& scheduler) noexcept;
    SCHEDULER& scheduler;
  };

  // Construct a transform/operation that can be applied to a source object
  // using "pipe" notation (ie. operator|).
  template<typename SCHEDULER>
  resume_on_transform<SCHEDULER> resume_on(SCHEDULER& scheduler) noexcept;

  // Equivalent to 'resume_on(transform.scheduler, std::forward<T>(value))'
  template<typename T, typename SCHEDULER>
  decltype(auto) operator|(T&& value, resume_on_transform<SCHEDULER> transform)
  {
    return resume_on(transform.scheduler, std::forward<T>(value));
  }
}

task<int> get_value();
io_service ioSvc;

task<> example()
{
  // Starts executing get_value() on the current thread.
  int a = co_await get_value();

  // Starts executing get_value() on a thread associated with ioSvc.
  int b = co_await schedule_on(ioSvc, get_value());
}

// <cppcoro/schedule_on.hpp>
namespace cppcoro
{
  // Return a task that yields the same result as 't' but that
  // ensures that 't' is co_await'ed on a thread associated with
  // the specified scheduler. Resulting task will complete on
  // whatever thread 't' would normally complete on.
  template<typename SCHEDULER, typename AWAITABLE>
  auto schedule_on(SCHEDULER& scheduler, AWAITABLE awaitable)
    -> Awaitable<typename awaitable_traits<AWAITABLE>::await_result_t>;

  // Return a generator that yields the same sequence of results as
  // 'source' but that ensures that execution of the coroutine starts
  // execution on a thread associated with 'scheduler' and resumes
  // after a 'co_yield' on a thread associated with 'scheduler'.
  template<typename SCHEDULER, typename T>
  async_generator<T> schedule_on(SCHEDULER& scheduler, async_generator<T> source);

  template<typename SCHEDULER>
  struct schedule_on_transform
  {
    explicit schedule_on_transform(SCHEDULER& scheduler) noexcept;
    SCHEDULER& scheduler;
  };

  template<typename SCHEDULER>
  schedule_on_transform<SCHEDULER> schedule_on(SCHEDULER& scheduler) noexcept;

  template<typename T, typename SCHEDULER>
  decltype(auto) operator|(T&& value, schedule_on_transform<SCHEDULER> transform);
}

// <cppcoro/awaitable_traits.hpp>
namespace cppcoro
{
  template<typename T>
  struct awaitable_traits
  {
    // The type that results from applying `operator co_await()` to a value
    // of type T, if T supports an `operator co_await()`, otherwise is type `T&&`.
    typename awaiter_t = <unspecified>;

    // The type of the result of co_await'ing a value of type T.
    typename await_result_t = <unspecified>;
  };
}

// <cppcoro/is_awaitable.hpp>
namespace cppcoro
{
  template<typename T>
  struct is_awaitable : std::bool_constant<...>
  {};

  template<typename T>
  constexpr bool is_awaitable_v = is_awaitable<T>::value;
}

concept Scheduler
{
  Awaitable<void> schedule();
}

cppcoro::task<> f(Scheduler& scheduler)
{
  // Execution of the coroutine is initially on the caller's execution context.

  // Suspends execution of the coroutine and schedules it for resumption on
  // the scheduler's execution context.
  co_await scheduler.schedule();

  // At this point the coroutine is now executing on the scheduler's
  // execution context.
}

concept DelayedScheduler : Scheduler
{
  template<typename REP, typename RATIO>
  Awaitable<void> schedule_after(std::chrono::duration<REP, RATIO> delay);

  template<typename REP, typename RATIO>
  Awaitable<void> schedule_after(
    std::chrono::duration<REP, RATIO> delay,
    cppcoro::cancellation_token cancellationToken);
}