network.hpp

#pragma once
 
#include <coroutine>
#include <cstddef>
#include <deque>
#include <optional>
#include <string>
#include <utility>
 
#include "ravel/clock.hpp"
#include "ravel/rng.hpp"
#include "ravel/scheduler.hpp"
#include "ravel/trace.hpp"
 
namespace ravel {
 
using ChannelId = std::size_t;
using Message = std::string;
 
struct FaultSpec {
  double loss_probability = 0.0;  
  VirtualClock::Tick latency_min = 0;
  VirtualClock::Tick latency_max = 0;  
  bool allow_reorder = false;
};
 
class Channel {
 public:
  Channel(ChannelId id, std::string from, std::string to, FaultSpec fault,
          Scheduler& scheduler, VirtualRng& rng, Trace& trace);
 
  Channel(const Channel&) = delete;
  Channel& operator=(const Channel&) = delete;
 
  const std::string& from() const noexcept { return from_; }
  const std::string& to() const noexcept { return to_; }
  const FaultSpec& fault() const noexcept { return fault_; }
 
  void send(Message message);
 
  class ReceiveAwaiter {
   public:
    bool await_ready() const noexcept { return false; }
    void await_suspend(std::coroutine_handle<>) const { channel_.wait_for_message(); }
    Message await_resume() const { return channel_.take_message(); }
 
   private:
    friend class Channel;
    explicit ReceiveAwaiter(Channel& channel) noexcept : channel_(channel) {}
    Channel& channel_;
  };
 
  [[nodiscard]] ReceiveAwaiter receive() noexcept { return ReceiveAwaiter(*this); }
 
  class TimedReceiveAwaiter {
   public:
    bool await_ready() const noexcept { return false; }
    void await_suspend(std::coroutine_handle<>) const { channel_.wait_for_message(timeout_); }
    std::optional<Message> await_resume() const { return channel_.take_message_if_any(); }
 
   private:
    friend class Channel;
    TimedReceiveAwaiter(Channel& channel, VirtualClock::Tick timeout) noexcept
        : channel_(channel), timeout_(timeout) {}
    Channel& channel_;
    VirtualClock::Tick timeout_;
  };
 
  [[nodiscard]] TimedReceiveAwaiter receive_within(VirtualClock::Tick timeout) noexcept {
    return TimedReceiveAwaiter(*this, timeout);
  }
 
  void clear_inbox() { inbox_.clear(); }
 private:
  void deliver(Message message);
  void wait_for_message(std::optional<VirtualClock::Tick> timeout = std::nullopt);
  Message take_message();
  std::optional<Message> take_message_if_any();
  void record(TraceEventKind kind) { trace_.record({scheduler_.now(), id_, kind}); }
 
  ChannelId id_;
  std::string from_;
  std::string to_;
  FaultSpec fault_;
  Scheduler& scheduler_;
  VirtualRng& rng_;
  Trace& trace_;
 
  std::deque<Message> inbox_;                
 
  struct Waiter {
    TaskId task;
    std::uint64_t id;  
  };
  std::optional<Waiter> waiting_receiver_;   
  std::uint64_t waits_started_ = 0;
  VirtualClock::Tick last_delivery_at_ = 0;  
};
 
}  // namespace ravel