pw_transport#

Asynchronous datagram-oriented transport interface

Experimental C++

Note

pw_transport is still considered experimental. While the core interfaces are likely stable, some changes are still expected to be made around the edges. The documentation is also incomplete.

pw_transport defines interfaces for async-native, datagram-oriented network sockets built on pw_async2.

The core interfaces in pw_transport are Socket and ReliableSocket: abstract handles to some underlying transport, used for reading and writing. Any ReliableSocket can be implicitly converted to and used as a Socket. Sockets make use of pw_buf to enable zero-copy reads and writes directly into transport buffers.

Sockets are provisioned by one of two transport interfaces: a Connector / ReliableConnector which establishes them, or a Listener / ReliableListener which accepts them from a peer.

Usage#

Once a ReliableSocket is obtained from a ReliableConnector or ReliableListener, tasks read and write messages asynchronously. Writes use a two-phase reservation model: ReliableSocket::ReserveWrite returns a future that resolves to a WriteReservation once buffer space is available. If the underlying transport’s outgoing queue is full, the future pends, naturally applying backpressure to the sender.

Zero-length datagrams are supported: ReserveWrite(0) reserves a zero-byte write slot, and Read resolves to a non-null, empty pw::ConstBuf (such as pw::ConstBuf::ZeroLength()) when a zero-byte datagram is received. A null ConstBuf (response == nullptr) indicates that the socket has closed.

For example, consider a task which sends a message and waits for the peer’s response:

class RequestResponseTask : public pw::async2::Task {
 public:
  explicit RequestResponseTask(pw::transport::ReliableConnector& connector)
      : connector_(connector) {}

 private:
  enum class State {
    kConnecting,
    kWriting,
    kReading,
  };

  pw::async2::Poll<> DoPend(pw::async2::Context& cx) override {
    while (true) {
      switch (state_) {
        case State::kConnecting: {
          if (!connect_future_.is_pendable()) {
            connect_future_ = connector_.Connect();
          }
          PW_AWAIT(auto result, connect_future_, cx);
          if (!result.ok()) {
            return pw::async2::Ready();
          }
          socket_ = std::move(*result);
          state_ = State::kWriting;
          break;
        }

        case State::kWriting: {
          // Reserve space in an outgoing transport buffer.
          if (!reserve_future_.is_pendable()) {
            reserve_future_ = socket_.ReserveWrite(kPayload.size());
          }
          PW_AWAIT(auto reservation, reserve_future_, cx);
          if (!reservation.has_value()) {
            return pw::async2::Ready();
          }

          // Write the payload directly into the transport buffer without
          // intermediate storage or copying.
          SerializePayload(pw::ByteSpan(*reservation));
          if (!reservation->Commit(kPayload.size())) {
            return pw::async2::Ready();
          }
          state_ = State::kReading;
          break;
        }

        case State::kReading: {
          // Read the peer's response.
          if (!read_future_.is_pendable()) {
            read_future_ = socket_.Read();
          }
          PW_AWAIT(pw::ConstBuf response, read_future_, cx);
          if (response != nullptr) {
            ProcessResponse(response);
          }
          return pw::async2::Ready();
        }
      }
    }
  }

  pw::transport::ReliableConnector& connector_;
  pw::transport::ReliableSocket socket_;
  State state_ = State::kConnecting;
  pw::transport::ReliableConnector::ConnectFuture connect_future_;
  pw::transport::ReserveWriteFuture reserve_future_;
  pw::transport::ReadFuture read_future_;
};

pw::async2::Coro<pw::Status> SendAndReceive(
    pw::async2::CoroContext&, pw::transport::ReliableConnector& connector) {
  PW_CO_TRY_ASSIGN(pw::transport::ReliableSocket socket,
                   co_await connector.Connect());

  // Reserve space in an outgoing transport buffer.
  std::optional<pw::transport::WriteReservation> reservation =
      co_await socket.ReserveWrite(kPayload.size());
  if (!reservation.has_value()) {
    co_return pw::Status::Aborted();
  }

  // Write the payload directly into the transport buffer without
  // intermediate storage or copying.
  SerializePayload(pw::ByteSpan(*reservation));
  if (!reservation->Commit(kPayload.size())) {
    co_return pw::Status::Aborted();
  }

  // Read the peer's response.
  pw::ConstBuf response = co_await socket.Read();
  if (response == nullptr) {
    co_return pw::Status::Aborted();
  }

  ProcessResponse(response);
  co_return pw::OkStatus();
}

Addressing#

pw_transport does not define an addressing layer, as this is inherently platform and protocol dependent. Where required, addressing should be built into implementations of Listener / ReliableListener and Connector / ReliableConnector. For example:

class FramedTcpConnector : public pw::transport::ReliableConnector {
 public:
  FramedTcpConnector(std::string_view host, uint16_t port);

  pw::transport::ReliableConnector::ConnectFuture Connect() override;
};

Instances of these objects can then be configured for specific endpoints and passed around into APIs that consume pw_transport.

In-Memory Forwarding Sockets#

ForwardingDatagramSocketPair creates a connected pair of in-memory ReliableSocket endpoints (first() and second()) that forward datagrams to each other without copying. It is useful for unit testing and in-process communication between tasks. ForwardingDatagramSocketOptions configures the queue depth, which limits the write reservations and unread datagrams in each direction, and the maximum datagram size in each direction.

Implementing#

Authors writing an implementation for your own transport are required to implement three virtual interfaces:

Refer to the doxygen docs for details of each of these, with requirements and preconditions.