channel.h

// Copyright 2025 The Pigweed Authors
//
// Licensed under the Apache License, Version 2.0 (the "License"); you may not
// use this file except in compliance with the License. You may obtain a copy of
// the License at
//
//     https://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
// WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
// License for the specific language governing permissions and limitations under
// the License.
#pragma once
 
#include <mutex>
 
#include "pw_allocator/allocator.h"
#include "pw_async2/callback_task.h"
#include "pw_async2/dispatcher.h"
#include "pw_async2/future.h"
#include "pw_containers/deque.h"
#include "pw_numeric/checked_arithmetic.h"
#include "pw_result/result.h"
#include "pw_sync/interrupt_spin_lock.h"
#include "pw_sync/lock_annotations.h"
#include "pw_sync/timed_thread_notification.h"
 
namespace pw::async2 {
 
template <typename T>
class Receiver;
 
template <typename T>
class ReceiveFuture;
 
template <typename T>
class Sender;
 
template <typename T>
class SendFuture;
 
template <typename T>
class ReserveSendFuture;
 
template <typename T>
class SendReservation;
 
template <typename T, uint16_t kCapacity>
class ChannelStorage;
 
namespace internal {
 
template <typename T>
class ChannelHandle;
 
template <typename T>
class Channel {
 public:
  static Channel* Allocated(Allocator& alloc, uint16_t capacity) {
    FixedDeque<T> deque = FixedDeque<T>::TryAllocate(alloc, capacity);
    if (deque.capacity() == 0) {
      return nullptr;
    }
    return alloc.New<Channel<T>>(std::move(deque));
  }
 
  ~Channel() { PW_ASSERT(ref_count_ == 0); }
 
  [[nodiscard]] bool closed() const PW_LOCKS_EXCLUDED(lock_) {
    std::lock_guard lock(lock_);
    return closed_;
  }
 
 protected:
  explicit Channel(FixedDeque<T>&& deque) : deque_(std::move(deque)) {}
 
  template <size_t kAlignment, size_t kCapacity>
  explicit Channel(containers::Storage<kAlignment, kCapacity>& storage)
      : deque_(storage) {}
 
  uint16_t ref_count() const PW_LOCKS_EXCLUDED(lock_) {
    std::lock_guard lock(lock_);
    return ref_count_;
  }
 
 private:
  friend Allocator;
  friend ChannelHandle<T>;
  friend SendFuture<T>;
  friend ReserveSendFuture<T>;
  friend ReceiveFuture<T>;
  friend Sender<T>;
  friend SendReservation<T>;
  friend Receiver<T>;
 
  void Destroy() PW_LOCKS_EXCLUDED(lock_) {
    Deallocator* deallocator = nullptr;
    {
      std::lock_guard lock(lock_);
      deallocator = deque_.deallocator();
    }
 
    if (deallocator != nullptr) {
      std::destroy_at(this);
      deallocator->Deallocate(this);
    }
  }
 
  void Close() PW_LOCKS_EXCLUDED(lock_) {
    std::lock_guard lock(lock_);
    CloseLocked();
  }
 
  void CloseLocked() PW_EXCLUSIVE_LOCKS_REQUIRED(lock_) {
    closed_ = true;
    while (auto send_future = send_futures_.Pop()) {
      send_future->get().Wake();
    }
    while (auto reserve_send_future = reserve_send_futures_.Pop()) {
      reserve_send_future->get().Wake();
    }
    while (auto receive_future = receive_futures_.Pop()) {
      receive_future->get().Wake();
    }
  }
 
  Sender<T> CreateSender() PW_LOCKS_EXCLUDED(lock_) {
    if (closed()) {
      return Sender<T>(nullptr);
    }
    return Sender<T>(this);
  }
 
  Receiver<T> CreateReceiver() PW_LOCKS_EXCLUDED(lock_) {
    if (closed()) {
      return Receiver<T>(nullptr);
    }
    return Receiver<T>(this);
  }
 
  void PushAndWake(T&& value) PW_EXCLUSIVE_LOCKS_REQUIRED(lock_) {
    deque_.push_back(std::move(value));
    WakeOneReceiver();
  }
 
  void PushAndWake(const T& value) PW_EXCLUSIVE_LOCKS_REQUIRED(lock_) {
    deque_.push_back(value);
    WakeOneReceiver();
  }
 
  template <typename... Args>
  void EmplaceAndWake(Args&&... args) PW_EXCLUSIVE_LOCKS_REQUIRED(lock_) {
    deque_.emplace_back(std::forward<Args>(args)...);
    WakeOneReceiver();
  }
 
  void WakeOneReceiver() PW_EXCLUSIVE_LOCKS_REQUIRED(lock_) {
    if (auto future = receive_futures_.Pop()) {
      future->get().Wake();
    }
  }
 
  T PopAndWake() PW_EXCLUSIVE_LOCKS_REQUIRED(lock_) {
    PW_ASSERT(!deque_.empty());
 
    T value = std::move(deque_.front());
    deque_.pop_front();
 
    WakeOneSender();
    return value;
  }
 
  void WakeOneSender() PW_EXCLUSIVE_LOCKS_REQUIRED(lock_) {
    // TODO: b/456507134 - Store both future types in the same list.
    if (prioritize_reserve_) {
      if (auto reserve_future = reserve_send_futures_.Pop()) {
        reserve_future->get().Wake();
      } else if (auto send_future = send_futures_.Pop()) {
        send_future->get().Wake();
      }
    } else {
      if (auto send_future = send_futures_.Pop()) {
        send_future->get().Wake();
      } else if (auto reserve_future = reserve_send_futures_.Pop()) {
        reserve_future->get().Wake();
      }
    }
 
    prioritize_reserve_ = !prioritize_reserve_;
  }
 
  bool full() PW_LOCKS_EXCLUDED(lock_) { return remaining_capacity() == 0; }
  bool full_locked() const PW_EXCLUSIVE_LOCKS_REQUIRED(lock_) {
    return remaining_capacity_locked() == 0;
  }
 
  uint16_t remaining_capacity() PW_LOCKS_EXCLUDED(lock_) {
    std::lock_guard lock(lock_);
    return remaining_capacity_locked();
  }
  uint16_t remaining_capacity_locked() const
      PW_EXCLUSIVE_LOCKS_REQUIRED(lock_) {
    return deque_.capacity() - deque_.size() - reservations_;
  }
 
  uint16_t capacity() const PW_NO_LOCK_SAFETY_ANALYSIS {
    // SAFETY: The capacity of `deque_` cannot change.
    return deque_.capacity();
  }
 
  bool empty() PW_LOCKS_EXCLUDED(lock_) {
    std::lock_guard lock(lock_);
    return deque_.empty();
  }
 
  void Push(const T& value) PW_EXCLUSIVE_LOCKS_REQUIRED(lock_) {
    PW_ASSERT(!closed_);
    PushAndWake(value);
  }
 
  void Push(T&& value) PW_EXCLUSIVE_LOCKS_REQUIRED(lock_) {
    PW_ASSERT(!closed_);
    PushAndWake(std::move(value));
  }
 
  bool TryPush(const T& value) PW_LOCKS_EXCLUDED(lock_) {
    std::lock_guard lock(lock_);
    if (closed_ || remaining_capacity_locked() == 0) {
      return false;
    }
    PushAndWake(value);
    return true;
  }
 
  bool TryPush(T&& value) PW_LOCKS_EXCLUDED(lock_) {
    std::lock_guard lock(lock_);
    if (closed_ || remaining_capacity_locked() == 0) {
      return false;
    }
    PushAndWake(std::move(value));
    return true;
  }
 
  std::optional<T> TryPop() PW_LOCKS_EXCLUDED(lock_) {
    std::lock_guard lock(lock_);
    if (deque_.empty()) {
      return std::nullopt;
    }
    return PopAndWake();
  }
 
  bool Reserve() PW_LOCKS_EXCLUDED(lock_) {
    std::lock_guard lock(lock_);
    if (closed_ || remaining_capacity_locked() == 0) {
      return false;
    }
    reservations_++;
    return true;
  }
 
  void DropReservation() PW_LOCKS_EXCLUDED(lock_) {
    std::lock_guard lock(lock_);
    PW_ASSERT(reservations_ > 0);
    reservations_--;
    if (!closed_) {
      WakeOneSender();
    }
  }
 
  template <typename... Args>
  void CommitReservation(Args&&... args) PW_LOCKS_EXCLUDED(lock_) {
    std::lock_guard lock(lock_);
    PW_ASSERT(reservations_ > 0);
    reservations_--;
    if (!closed_) {
      EmplaceAndWake(std::forward<Args>(args)...);
    }
  }
 
  void add_receiver() PW_LOCKS_EXCLUDED(lock_) {
    std::lock_guard lock(lock_);
    add_object(receiver_count_);
  }
 
  void add_sender() PW_LOCKS_EXCLUDED(lock_) {
    std::lock_guard lock(lock_);
    add_object(sender_count_);
  }
 
  void add_handle() PW_LOCKS_EXCLUDED(lock_) {
    std::lock_guard lock(lock_);
    add_object(handle_count_);
  }
 
  void add_object(uint8_t& counter) PW_EXCLUSIVE_LOCKS_REQUIRED(lock_) {
    if (!closed_) {
      PW_ASSERT(CheckedAdd(counter, 1, counter));
    }
    PW_ASSERT(CheckedAdd(ref_count_, 1, ref_count_));
  }
 
  void remove_sender() PW_LOCKS_EXCLUDED(lock_) {
    std::lock_guard lock(lock_);
    remove_object(sender_count_);
  }
 
  void remove_receiver() PW_LOCKS_EXCLUDED(lock_) {
    std::lock_guard lock(lock_);
    remove_object(receiver_count_);
  }
 
  void remove_handle() PW_LOCKS_EXCLUDED(lock_) {
    std::lock_guard lock(lock_);
    remove_object(handle_count_);
  }
 
  void remove_object(uint8_t& counter) PW_UNLOCK_FUNCTION(lock_) {
    if (!closed_) {
      PW_ASSERT(counter > 0);
      counter--;
      if (should_close()) {
        CloseLocked();
      }
    }
    bool destroy = decrement_ref_locked();
    lock_.unlock();
 
    if (destroy) {
      Destroy();
    }
  }
 
  void add_ref() PW_LOCKS_EXCLUDED(lock_) {
    std::lock_guard lock(lock_);
    ref_count_++;
  }
 
  void remove_ref() PW_LOCKS_EXCLUDED(lock_) {
    bool destroy;
    {
      std::lock_guard lock(lock_);
      destroy = decrement_ref_locked();
    }
    if (destroy) {
      Destroy();
    }
  }
 
  bool decrement_ref_locked() PW_EXCLUSIVE_LOCKS_REQUIRED(lock_) {
    ref_count_--;
    return ref_count_ == 0;
  }
 
  bool should_close() const PW_EXCLUSIVE_LOCKS_REQUIRED(lock_) {
    if (handle_count_ > 0) {
      return false;
    }
    return sender_count_ == 0 || receiver_count_ == 0;
  }
 
  // ListFutureProvider is internally synchronized.
  ListFutureProvider<SendFuture<T>> send_futures_;
  ListFutureProvider<ReserveSendFuture<T>> reserve_send_futures_;
  ListFutureProvider<ReceiveFuture<T>> receive_futures_;
 
  mutable sync::InterruptSpinLock lock_;
  FixedDeque<T> deque_ PW_GUARDED_BY(lock_);
  uint16_t reservations_ PW_GUARDED_BY(lock_) = 0;
  bool closed_ PW_GUARDED_BY(lock_) = false;
  bool prioritize_reserve_ PW_GUARDED_BY(lock_) = true;
 
  // Channels are reference counted in two ways:
  //
  // - Senders and receivers are tracked independently. Once either reaches
  //   zero, the channel is closed, but not destroyed. No new values can be
  //   sent, but any buffered values can still be read.
  //
  // - Overall object reference count, including senders, receivers, futures,
  //   and channel handles. Once this reaches zero, the channel is destroyed.
  //
  uint8_t sender_count_ PW_GUARDED_BY(lock_) = 0;
  uint8_t receiver_count_ PW_GUARDED_BY(lock_) = 0;
  uint8_t handle_count_ PW_GUARDED_BY(lock_) = 0;
  uint16_t ref_count_ PW_GUARDED_BY(lock_) = 0;
};
 
template <typename T>
class ChannelHandle {
 public:
  constexpr ChannelHandle() : channel_(nullptr) {}
 
  ChannelHandle(const ChannelHandle& other) : channel_(other.channel_) {
    if (channel_ != nullptr) {
      channel_->add_handle();
    }
  }
 
  ChannelHandle& operator=(const ChannelHandle& other) {
    if (channel_ != nullptr) {
      channel_->remove_handle();
    }
    channel_ = other.channel_;
    if (channel_ != nullptr) {
      channel_->add_handle();
    }
    return *this;
  }
 
  ChannelHandle(ChannelHandle&& other) noexcept
      : channel_(std::exchange(other.channel_, nullptr)) {}
 
  ChannelHandle& operator=(ChannelHandle&& other) noexcept {
    if (this == &other) {
      return *this;
    }
    if (channel_ != nullptr) {
      channel_->remove_handle();
    }
    channel_ = std::exchange(other.channel_, nullptr);
    return *this;
  }
 
  ~ChannelHandle() { Release(); }
 
  [[nodiscard]] bool is_open() const {
    return channel_ != nullptr && !channel_->closed();
  }
 
  Sender<T> CreateSender() {
    PW_ASSERT(channel_ != nullptr);
    return channel_->CreateSender();
  }
 
  Receiver<T> CreateReceiver() {
    PW_ASSERT(channel_ != nullptr);
    return channel_->CreateReceiver();
  }
 
  void Close() {
    if (channel_ != nullptr) {
      channel_->Close();
    }
  }
 
  void Release() {
    if (channel_ != nullptr) {
      channel_->remove_handle();
      channel_ = nullptr;
    }
  }
 
 protected:
  explicit ChannelHandle(internal::Channel<T>* channel) : channel_(channel) {
    if (channel_ != nullptr) {
      channel_->add_handle();
    }
  }
 
 private:
  internal::Channel<T>* channel_;
};
 
}  // namespace internal
 
template <typename T>
class MpmcChannelHandle : private internal::ChannelHandle<T> {
 public:
  constexpr MpmcChannelHandle() = default;
 
  using internal::ChannelHandle<T>::is_open;
  using internal::ChannelHandle<T>::Close;
  using internal::ChannelHandle<T>::CreateReceiver;
  using internal::ChannelHandle<T>::CreateSender;
  using internal::ChannelHandle<T>::Release;
 
 private:
  explicit MpmcChannelHandle(internal::Channel<T>* channel)
      : internal::ChannelHandle<T>(channel) {}
 
  template <typename U>
  friend std::optional<MpmcChannelHandle<U>> CreateMpmcChannel(Allocator&,
                                                               uint16_t);
 
  template <typename U, uint16_t kCapacity>
  friend MpmcChannelHandle<U> CreateMpmcChannel(
      ChannelStorage<U, kCapacity>& storage);
};
 
template <typename T>
class MpscChannelHandle : private internal::ChannelHandle<T> {
 public:
  constexpr MpscChannelHandle() = default;
 
  using internal::ChannelHandle<T>::is_open;
  using internal::ChannelHandle<T>::Close;
  using internal::ChannelHandle<T>::CreateSender;
  using internal::ChannelHandle<T>::Release;
 
 private:
  explicit MpscChannelHandle(internal::Channel<T>* channel)
      : internal::ChannelHandle<T>(channel) {}
 
  template <typename U>
  friend std::optional<std::tuple<MpscChannelHandle<U>, Receiver<U>>>
  CreateMpscChannel(Allocator&, uint16_t);
 
  template <typename U, uint16_t kCapacity>
  friend std::tuple<MpscChannelHandle<U>, Receiver<U>> CreateMpscChannel(
      ChannelStorage<U, kCapacity>& storage);
};
 
template <typename T>
class SpmcChannelHandle : private internal::ChannelHandle<T> {
 public:
  constexpr SpmcChannelHandle() = default;
 
  using internal::ChannelHandle<T>::is_open;
  using internal::ChannelHandle<T>::Close;
  using internal::ChannelHandle<T>::CreateReceiver;
  using internal::ChannelHandle<T>::Release;
 
 private:
  explicit SpmcChannelHandle(internal::Channel<T>* channel)
      : internal::ChannelHandle<T>(channel) {}
 
  template <typename U>
  friend std::optional<std::tuple<SpmcChannelHandle<U>, Sender<U>>>
  CreateSpmcChannel(Allocator&, uint16_t);
 
  template <typename U, uint16_t kCapacity>
  friend std::tuple<SpmcChannelHandle<U>, Sender<U>> CreateSpmcChannel(
      ChannelStorage<U, kCapacity>& storage);
};
 
template <typename T>
class SpscChannelHandle : private internal::ChannelHandle<T> {
 public:
  constexpr SpscChannelHandle() = default;
 
  using internal::ChannelHandle<T>::is_open;
  using internal::ChannelHandle<T>::Close;
  using internal::ChannelHandle<T>::Release;
 
 private:
  explicit SpscChannelHandle(internal::Channel<T>* channel)
      : internal::ChannelHandle<T>(channel) {}
 
  template <typename U>
  friend std::optional<std::tuple<SpscChannelHandle<U>, Sender<U>, Receiver<U>>>
  CreateSpscChannel(Allocator&, uint16_t);
 
  template <typename U, uint16_t kCapacity>
  friend std::tuple<SpscChannelHandle<U>, Sender<U>, Receiver<U>>
  CreateSpscChannel(ChannelStorage<U, kCapacity>& storage);
};
 
template <typename T, uint16_t kCapacity>
class ChannelStorage : private containers::internal::ArrayStorage<T, kCapacity>,
                       public internal::Channel<T> {
 public:
  ChannelStorage() : internal::Channel<T>(this->storage_array) {}
 
  [[nodiscard]] bool active() const { return this->ref_count() != 0; }
 
 private:
  using internal::Channel<T>::Allocated;
};
 
template <typename T>
class [[nodiscard]] ReceiveFuture
    : public ListableFutureWithWaker<ReceiveFuture<T>, std::optional<T>> {
 public:
  ReceiveFuture(ReceiveFuture&& other)
      : Base(Base::kMovedFrom),
        channel_(std::exchange(other.channel_, nullptr)) {
    Base::MoveFrom(other);
  }
 
  ReceiveFuture& operator=(ReceiveFuture&& other) {
    if (this == &other) {
      return *this;
    }
    if (channel_ != nullptr) {
      channel_->remove_ref();
    }
    channel_ = std::exchange(other.channel_, nullptr);
    Base::MoveFrom(other);
    return *this;
  }
 
  ~ReceiveFuture() { reset(); }
 
 private:
  using Base = ListableFutureWithWaker<ReceiveFuture<T>, std::optional<T>>;
  friend Base;
  friend internal::Channel<T>;
  friend Receiver<T>;
 
  static constexpr const char kWaitReason[] = "Receiver::Receive";
 
  explicit ReceiveFuture(internal::Channel<T>& channel)
      : Base(channel.receive_futures_), channel_(&channel) {
    channel_->add_ref();
  }
 
  ReceiveFuture() : Base(Base::kReadyForCompletion), channel_(nullptr) {}
 
  Poll<std::optional<T>> DoPend(Context&) {
    if (channel_ == nullptr) {
      return Ready<std::optional<T>>(std::nullopt);
    }
 
    std::optional<T> value = channel_->TryPop();
    if (!value.has_value()) {
      if (channel_->closed()) {
        reset();
        return Ready<std::optional<T>>(std::nullopt);
      }
      return Pending();
    }
 
    reset();
    return Ready(std::move(value));
  }
 
  void reset() {
    if (channel_ != nullptr) {
      channel_->remove_ref();
      channel_ = nullptr;
    }
  }
 
  using Base::Wake;
 
  internal::Channel<T>* channel_;
};
 
template <typename T>
class Receiver {
 public:
  constexpr Receiver() : channel_(nullptr) {}
 
  Receiver(const Receiver& other) = delete;
  Receiver& operator=(const Receiver& other) = delete;
 
  Receiver(Receiver&& other) noexcept
      : channel_(std::exchange(other.channel_, nullptr)) {}
 
  Receiver& operator=(Receiver&& other) noexcept {
    if (this == &other) {
      return *this;
    }
    if (channel_ != nullptr) {
      channel_->remove_receiver();
    }
    channel_ = std::exchange(other.channel_, nullptr);
    return *this;
  }
 
  ~Receiver() {
    if (channel_ != nullptr) {
      channel_->remove_receiver();
    }
  }
 
  Result<T> BlockingReceive(
      Dispatcher& dispatcher,
      chrono::SystemClock::duration timeout = kWaitForever) {
    if (channel_ == nullptr) {
      return Status::FailedPrecondition();
    }
 
    Result<T> available = TryReceive();
    if (available.ok()) {
      return available;
    }
 
    std::optional<T> result;
    sync::TimedThreadNotification notification;
 
    FutureCallbackTask task(ReceiveFuture<T>(*channel_),
                            [&result, &notification](std::optional<T> value) {
                              result = std::move(value);
                              notification.release();
                            });
    dispatcher.Post(task);
 
    if (timeout == kWaitForever) {
      notification.acquire();
      if (!result.has_value()) {
        return Status::FailedPrecondition();
      }
      return Result<T>(std::move(result.value()));
    }
 
    if (!notification.try_acquire_for(timeout)) {
      task.Deregister();
      return Status::DeadlineExceeded();
    }
 
    if (!result.has_value()) {
      return Status::FailedPrecondition();
    }
    return Result<T>(std::move(result.value()));
  }
 
  ReceiveFuture<T> Receive() {
    if (channel_ == nullptr) {
      return ReceiveFuture<T>();
    }
    return ReceiveFuture<T>(*channel_);
  }
 
  Result<T> TryReceive() {
    if (channel_ == nullptr) {
      return Status::FailedPrecondition();
    }
    std::optional<T> value = channel_->TryPop();
    if (value.has_value()) {
      return std::move(*value);
    }
    if (channel_->closed()) {
      return Status::FailedPrecondition();
    }
    return Status::Unavailable();
  }
 
  void Disconnect() {
    if (channel_ != nullptr) {
      channel_->remove_receiver();
      channel_ = nullptr;
    }
  }
 
  [[nodiscard]] bool is_open() const {
    return channel_ != nullptr && !channel_->closed();
  }
 
 private:
  static constexpr chrono::SystemClock::duration kWaitForever =
      chrono::SystemClock::duration::max();
 
  template <typename U>
  friend class internal::Channel;
 
  template <typename U>
  friend std::optional<std::tuple<MpscChannelHandle<U>, Receiver<U>>>
  CreateMpscChannel(Allocator&, uint16_t);
 
  template <typename U, uint16_t kCapacity>
  friend std::tuple<MpscChannelHandle<U>, Receiver<U>> CreateMpscChannel(
      ChannelStorage<U, kCapacity>& storage);
 
  template <typename U>
  friend std::optional<std::tuple<SpscChannelHandle<U>, Sender<U>, Receiver<U>>>
  CreateSpscChannel(Allocator&, uint16_t);
 
  template <typename U, uint16_t kCapacity>
  friend std::tuple<SpscChannelHandle<U>, Sender<U>, Receiver<U>>
  CreateSpscChannel(ChannelStorage<U, kCapacity>& storage);
 
  explicit Receiver(internal::Channel<T>* channel) : channel_(channel) {
    if (channel_ != nullptr) {
      channel_->add_receiver();
    }
  }
 
  internal::Channel<T>* channel_;
};
 
template <typename T>
class [[nodiscard]] SendFuture
    : public ListableFutureWithWaker<SendFuture<T>, bool> {
 public:
  SendFuture(SendFuture&& other)
      : Base(Base::kMovedFrom),
        channel_(std::exchange(other.channel_, nullptr)),
        value_(std::move(other.value_)) {
    Base::MoveFrom(other);
  }
 
  SendFuture& operator=(SendFuture&& other) {
    if (this == &other) {
      return *this;
    }
    if (channel_ != nullptr) {
      channel_->remove_ref();
    }
    channel_ = std::exchange(other.channel_, nullptr);
    value_ = std::move(other.value_);
    Base::MoveFrom(other);
    return *this;
  }
 
  ~SendFuture() { reset(); }
 
 private:
  using Base = ListableFutureWithWaker<SendFuture<T>, bool>;
  friend Base;
  friend internal::Channel<T>;
  friend Sender<T>;
 
  static constexpr const char kWaitReason[] = "Sender::Send";
 
  SendFuture(internal::Channel<T>& channel, const T& value)
      : Base(channel.send_futures_), channel_(&channel), value_(value) {
    channel_->add_ref();
  }
 
  SendFuture(internal::Channel<T>& channel, T&& value)
      : Base(channel.send_futures_),
        channel_(&channel),
        value_(std::move(value)) {
    channel_->add_ref();
  }
 
  enum ClosedState { kClosed };
 
  SendFuture(ClosedState, const T& value)
      : Base(Base::kReadyForCompletion), channel_(nullptr), value_(value) {}
 
  SendFuture(ClosedState, T&& value)
      : Base(Base::kReadyForCompletion),
        channel_(nullptr),
        value_(std::move(value)) {}
 
  Poll<bool> DoPend(async2::Context&) {
    if (channel_ == nullptr || channel_->closed()) {
      reset();
      return Ready(false);
    }
 
    {
      std::lock_guard lock(channel_->lock_);
      if (channel_->full_locked()) {
        return Pending();
      }
 
      channel_->Push(std::move(value_));
    }
 
    reset();
    return Ready(true);
  }
 
  void reset() {
    if (channel_ != nullptr) {
      channel_->remove_ref();
      channel_ = nullptr;
    }
  }
 
  using Base::Wake;
 
  internal::Channel<T>* channel_;
  T value_;
};
 
template <typename T>
class SendReservation {
 public:
  SendReservation(const SendReservation& other) = delete;
  SendReservation& operator=(const SendReservation& other) = delete;
 
  SendReservation(SendReservation&& other)
      : channel_(std::exchange(other.channel_, nullptr)) {}
 
  SendReservation& operator=(SendReservation&& other) {
    if (this == &other) {
      return *this;
    }
    Cancel();
    channel_ = std::exchange(other.channel_, nullptr);
    return *this;
  }
 
  ~SendReservation() { Cancel(); }
 
  template <typename... Args>
  void Commit(Args&&... args) {
    PW_ASSERT(channel_ != nullptr);
    channel_->CommitReservation(std::forward<Args>(args)...);
    channel_->remove_ref();
    channel_ = nullptr;
  }
 
  void Cancel() {
    if (channel_ != nullptr) {
      channel_->DropReservation();
      channel_->remove_ref();
      channel_ = nullptr;
    }
  }
 
 private:
  friend class ReserveSendFuture<T>;
  friend class Sender<T>;
 
  explicit SendReservation(internal::Channel<T>& channel) : channel_(&channel) {
    channel_->add_ref();
  }
 
  internal::Channel<T>* channel_;
};
 
template <typename T>
class [[nodiscard]] ReserveSendFuture
    : public ListableFutureWithWaker<ReserveSendFuture<T>,
                                     std::optional<SendReservation<T>>> {
 public:
  ReserveSendFuture(ReserveSendFuture&& other)
      : Base(Base::kMovedFrom),
        channel_(std::exchange(other.channel_, nullptr)) {
    Base::MoveFrom(other);
  }
 
  ReserveSendFuture& operator=(ReserveSendFuture&& other) {
    if (channel_ != nullptr) {
      channel_->remove_ref();
    }
    channel_ = std::exchange(other.channel_, nullptr);
    Base::MoveFrom(other);
    return *this;
  }
 
  ~ReserveSendFuture() { reset(); }
 
 private:
  using Base = ListableFutureWithWaker<ReserveSendFuture<T>,
                                       std::optional<SendReservation<T>>>;
  friend Base;
  friend internal::Channel<T>;
  friend Sender<T>;
 
  static constexpr const char kWaitReason[] = "Sender::ReserveSend";
 
  explicit ReserveSendFuture(internal::Channel<T>* channel)
      : Base(channel->reserve_send_futures_), channel_(channel) {
    channel_->add_ref();
  }
 
  enum ClosedState { kClosed };
 
  explicit ReserveSendFuture(ClosedState)
      : Base(Base::kReadyForCompletion), channel_(nullptr) {}
 
  Poll<std::optional<SendReservation<T>>> DoPend(async2::Context&) {
    if (channel_ == nullptr || channel_->closed()) {
      reset();
      return Ready<std::optional<SendReservation<T>>>(std::nullopt);
    }
 
    if (!channel_->Reserve()) {
      return Pending();
    }
 
    SendReservation<T> reservation(*channel_);
    reset();
    return reservation;
  }
 
  void reset() {
    if (channel_ != nullptr) {
      channel_->remove_ref();
      channel_ = nullptr;
    }
  }
 
  using Base::Wake;
 
  internal::Channel<T>* channel_;
};
 
template <typename T>
class Sender {
 public:
  constexpr Sender() : channel_(nullptr) {}
 
  Sender(const Sender& other) = delete;
  Sender& operator=(const Sender& other) = delete;
 
  Sender(Sender&& other) noexcept
      : channel_(std::exchange(other.channel_, nullptr)) {}
 
  Sender& operator=(Sender&& other) noexcept {
    if (this == &other) {
      return *this;
    }
    if (channel_ != nullptr) {
      channel_->remove_sender();
    }
    channel_ = std::exchange(other.channel_, nullptr);
    return *this;
  }
 
  ~Sender() {
    if (channel_ != nullptr) {
      channel_->remove_sender();
    }
  }
 
  SendFuture<T> Send(const T& value) {
    if (channel_ == nullptr) {
      return SendFuture<T>(SendFuture<T>::kClosed, value);
    }
    return SendFuture<T>(*channel_, value);
  }
 
  SendFuture<T> Send(T&& value) {
    if (channel_ == nullptr) {
      return SendFuture<T>(SendFuture<T>::kClosed, std::move(value));
    }
    return SendFuture<T>(*channel_, std::move(value));
  }
 
  ReserveSendFuture<T> ReserveSend() {
    if (channel_ == nullptr) {
      return ReserveSendFuture<T>(ReserveSendFuture<T>::kClosed);
    }
    return ReserveSendFuture<T>(channel_);
  }
 
  std::optional<SendReservation<T>> TryReserveSend() {
    if (channel_ == nullptr) {
      return std::nullopt;
    }
    if (!channel_->Reserve()) {
      return std::nullopt;
    }
    return SendReservation<T>(*channel_);
  }
 
  bool TrySend(const T& value) {
    if (channel_ == nullptr) {
      return false;
    }
    return channel_->TryPush(value);
  }
 
  bool TrySend(T&& value) {
    if (channel_ == nullptr) {
      return false;
    }
    return channel_->TryPush(std::move(value));
  }
 
  Status BlockingSend(Dispatcher& dispatcher,
                      const T& value,
                      chrono::SystemClock::duration timeout = kWaitForever) {
    if (channel_ == nullptr) {
      return Status::FailedPrecondition();
    }
    return BlockingSend(dispatcher, SendFuture<T>(*channel_, value), timeout);
  }
 
  Status BlockingSend(Dispatcher& dispatcher,
                      T&& value,
                      chrono::SystemClock::duration timeout = kWaitForever) {
    if (channel_ == nullptr) {
      return Status::FailedPrecondition();
    }
    return BlockingSend(
        dispatcher, SendFuture<T>(*channel_, std::move(value)), timeout);
  }
 
  void Disconnect() {
    if (channel_ != nullptr) {
      channel_->remove_sender();
      channel_ = nullptr;
    }
  }
 
  uint16_t remaining_capacity() const {
    return channel_ != nullptr ? channel_->remaining_capacity() : 0;
  }
 
  uint16_t capacity() const {
    return channel_ != nullptr ? channel_->capacity() : 0;
  }
 
  [[nodiscard]] bool is_open() const {
    return channel_ != nullptr && !channel_->closed();
  }
 
 private:
  static constexpr chrono::SystemClock::duration kWaitForever =
      chrono::SystemClock::duration::max();
 
  template <typename U>
  friend class internal::Channel;
 
  template <typename U>
  friend std::optional<std::tuple<SpmcChannelHandle<U>, Sender<U>>>
  CreateSpmcChannel(Allocator&, uint16_t);
 
  template <typename U, uint16_t kCapacity>
  friend std::tuple<SpmcChannelHandle<U>, Sender<U>> CreateSpmcChannel(
      ChannelStorage<U, kCapacity>& storage);
 
  template <typename U>
  friend std::optional<std::tuple<SpscChannelHandle<U>, Sender<U>, Receiver<U>>>
  CreateSpscChannel(Allocator&, uint16_t);
 
  template <typename U, uint16_t kCapacity>
  friend std::tuple<SpscChannelHandle<U>, Sender<U>, Receiver<U>>
  CreateSpscChannel(ChannelStorage<U, kCapacity>& storage);
 
  explicit Sender(internal::Channel<T>* channel) : channel_(channel) {
    if (channel_ != nullptr) {
      channel_->add_sender();
    }
  }
 
  Status BlockingSend(Dispatcher& dispatcher,
                      SendFuture<T>&& future,
                      chrono::SystemClock::duration timeout) {
    Status status;
    sync::TimedThreadNotification notification;
 
    FutureCallbackTask task(
        std::move(future), [&status, &notification](bool result) {
          status = result ? OkStatus() : Status::FailedPrecondition();
          notification.release();
        });
    dispatcher.Post(task);
 
    if (timeout == kWaitForever) {
      notification.acquire();
      return status;
    }
 
    if (!notification.try_acquire_for(timeout)) {
      task.Deregister();
      return Status::DeadlineExceeded();
    }
    return status;
  }
 
  internal::Channel<T>* channel_;
};
 
template <typename T>
std::optional<MpmcChannelHandle<T>> CreateMpmcChannel(Allocator& alloc,
                                                      uint16_t capacity) {
  auto channel = internal::Channel<T>::Allocated(alloc, capacity);
  if (channel == nullptr) {
    return std::nullopt;
  }
  return MpmcChannelHandle<T>(channel);
}
 
template <typename T, uint16_t kCapacity>
MpmcChannelHandle<T> CreateMpmcChannel(ChannelStorage<T, kCapacity>& storage) {
  PW_ASSERT(!storage.active());
  return MpmcChannelHandle<T>(&storage);
}
 
template <typename T>
std::optional<std::tuple<MpscChannelHandle<T>, Receiver<T>>> CreateMpscChannel(
    Allocator& alloc, uint16_t capacity) {
  auto channel = internal::Channel<T>::Allocated(alloc, capacity);
  if (channel == nullptr) {
    return std::nullopt;
  }
  return std::make_tuple(MpscChannelHandle<T>(channel), Receiver<T>(channel));
}
 
template <typename T, uint16_t kCapacity>
std::tuple<MpscChannelHandle<T>, Receiver<T>> CreateMpscChannel(
    ChannelStorage<T, kCapacity>& storage) {
  PW_ASSERT(!storage.active());
  return std::make_tuple(MpscChannelHandle<T>(&storage), Receiver<T>(&storage));
}
 
template <typename T>
std::optional<std::tuple<SpmcChannelHandle<T>, Sender<T>>> CreateSpmcChannel(
    Allocator& alloc, uint16_t capacity) {
  auto channel = internal::Channel<T>::Allocated(alloc, capacity);
  if (channel == nullptr) {
    return std::nullopt;
  }
  return std::make_tuple(SpmcChannelHandle<T>(channel), Sender<T>(channel));
}
 
template <typename T, uint16_t kCapacity>
std::tuple<SpmcChannelHandle<T>, Sender<T>> CreateSpmcChannel(
    ChannelStorage<T, kCapacity>& storage) {
  PW_ASSERT(!storage.active());
  return std::make_tuple(SpmcChannelHandle<T>(&storage), Sender<T>(&storage));
}
 
template <typename T>
std::optional<std::tuple<SpscChannelHandle<T>, Sender<T>, Receiver<T>>>
CreateSpscChannel(Allocator& alloc, uint16_t capacity) {
  auto channel = internal::Channel<T>::Allocated(alloc, capacity);
  if (channel == nullptr) {
    return std::nullopt;
  }
  return std::make_tuple(
      SpscChannelHandle<T>(channel), Sender<T>(channel), Receiver<T>(channel));
}
 
template <typename T, uint16_t kCapacity>
std::tuple<SpscChannelHandle<T>, Sender<T>, Receiver<T>> CreateSpscChannel(
    ChannelStorage<T, kCapacity>& storage) {
  PW_ASSERT(!storage.active());
  return std::make_tuple(SpscChannelHandle<T>(&storage),
                         Sender<T>(&storage),
                         Receiver<T>(&storage));
}
 
}  // namespace pw::async2