Add interfaces for UDP networking, with support for receiving ancillary data.

c++/src/kj/async-io-test.c++

@@ -221,5 +221,151 @@ TEST(AsyncIo, Timeouts) {
   EXPECT_EQ(123, promise2.wait(ioContext.waitScope));
 }
 
+TEST(AsyncIo, Udp) {
+  auto ioContext = setupAsyncIo();
+
+  auto addr = ioContext.provider->getNetwork().parseAddress("127.0.0.1").wait(ioContext.waitScope);
+
+  auto port1 = addr->bindDatagramPort();
+  auto port2 = addr->bindDatagramPort();
+
+  auto addr1 = ioContext.provider->getNetwork().parseAddress("127.0.0.1", port1->getPort())
+      .wait(ioContext.waitScope);
+  auto addr2 = ioContext.provider->getNetwork().parseAddress("127.0.0.1", port2->getPort())
+      .wait(ioContext.waitScope);
+
+  Own<NetworkAddress> receivedAddr;
+
+  {
+    // Send a message and receive it.
+    EXPECT_EQ(3, port1->send("foo", 3, *addr2).wait(ioContext.waitScope));
+    auto receiver = port2->makeReceiver();
+
+    receiver->receive().wait(ioContext.waitScope);
+    {
+      auto content = receiver->getContent();
+      EXPECT_EQ("foo", kj::heapString(content.value.asChars()));
+      EXPECT_FALSE(content.isTruncated);
+    }
+    receivedAddr = receiver->getSource().clone();
+    EXPECT_EQ(addr1->toString(), receivedAddr->toString());
+    {
+      auto ancillary = receiver->getAncillary();
+      EXPECT_EQ(0, ancillary.value.size());
+      EXPECT_FALSE(ancillary.isTruncated);
+    }
+
+    // Receive a second message with the same receiver.
+    {
+      auto promise = receiver->receive();  // This time, start receiving before sending
+      EXPECT_EQ(6, port1->send("barbaz", 6, *addr2).wait(ioContext.waitScope));
+      promise.wait(ioContext.waitScope);
+      auto content = receiver->getContent();
+      EXPECT_EQ("barbaz", kj::heapString(content.value.asChars()));
+      EXPECT_FALSE(content.isTruncated);
+    }
+  }
+
+  DatagramReceiver::Capacity capacity;
+  capacity.content = 8;
+  capacity.ancillary = 1024;
+
+  {
+    // Send a reply that will be truncated.
+    EXPECT_EQ(16, port2->send("0123456789abcdef", 16, *receivedAddr).wait(ioContext.waitScope));
+    auto recv1 = port1->makeReceiver(capacity);
+
+    recv1->receive().wait(ioContext.waitScope);
+    {
+      auto content = recv1->getContent();
+      EXPECT_EQ("01234567", kj::heapString(content.value.asChars()));
+      EXPECT_TRUE(content.isTruncated);
+    }
+    EXPECT_EQ(addr2->toString(), recv1->getSource().toString());
+    {
+      auto ancillary = recv1->getAncillary();
+      EXPECT_EQ(0, ancillary.value.size());
+      EXPECT_FALSE(ancillary.isTruncated);
+    }
+
+#ifdef IP_PKTINFO
+    // Set IP_PKTINFO header and try to receive it.
+    int one = 1;
+    port1->setsockopt(IPPROTO_IP, IP_PKTINFO, &one, sizeof(one));
+
+    EXPECT_EQ(3, port2->send("foo", 3, *addr1).wait(ioContext.waitScope));
+
+    recv1->receive().wait(ioContext.waitScope);
+    {
+      auto content = recv1->getContent();
+      EXPECT_EQ("foo", kj::heapString(content.value.asChars()));
+      EXPECT_FALSE(content.isTruncated);
+    }
+    EXPECT_EQ(addr2->toString(), recv1->getSource().toString());
+    {
+      auto ancillary = recv1->getAncillary();
+      EXPECT_FALSE(ancillary.isTruncated);
+      ASSERT_EQ(1, ancillary.value.size());
+
+      auto message = ancillary.value[0];
+      EXPECT_EQ(IPPROTO_IP, message.getLevel());
+      EXPECT_EQ(IP_PKTINFO, message.getType());
+      EXPECT_EQ(sizeof(struct in_pktinfo), message.asArray<byte>().size());
+      auto& pktinfo = KJ_ASSERT_NONNULL(message.as<struct in_pktinfo>());
+      EXPECT_EQ(htonl(0x7F000001), pktinfo.ipi_addr.s_addr);  // 127.0.0.1
+    }
+
+    // See what happens if there's not quite enough space for in_pktinfo.
+    capacity.ancillary = CMSG_SPACE(sizeof(struct in_pktinfo)) - 8;
+    recv1 = port1->makeReceiver(capacity);
+
+    EXPECT_EQ(3, port2->send("bar", 3, *addr1).wait(ioContext.waitScope));
+
+    recv1->receive().wait(ioContext.waitScope);
+    {
+      auto content = recv1->getContent();
+      EXPECT_EQ("bar", kj::heapString(content.value.asChars()));
+      EXPECT_FALSE(content.isTruncated);
+    }
+    EXPECT_EQ(addr2->toString(), recv1->getSource().toString());
+    {
+      auto ancillary = recv1->getAncillary();
+      EXPECT_TRUE(ancillary.isTruncated);
+
+      // We might get a message, but it will be truncated.
+      if (ancillary.value.size() != 0) {
+        EXPECT_EQ(1, ancillary.value.size());
+
+        auto message = ancillary.value[0];
+        EXPECT_EQ(IPPROTO_IP, message.getLevel());
+        EXPECT_EQ(IP_PKTINFO, message.getType());
+
+        EXPECT_TRUE(message.as<struct in_pktinfo>() == nullptr);
+        EXPECT_LT(message.asArray<byte>().size(), sizeof(struct in_pktinfo));
+      }
+    }
+
+    // See what happens if there's not enough space even for the cmsghdr.
+    capacity.ancillary = CMSG_SPACE(0) - 8;
+    recv1 = port1->makeReceiver(capacity);
+
+    EXPECT_EQ(3, port2->send("baz", 3, *addr1).wait(ioContext.waitScope));
+
+    recv1->receive().wait(ioContext.waitScope);
+    {
+      auto content = recv1->getContent();
+      EXPECT_EQ("baz", kj::heapString(content.value.asChars()));
+      EXPECT_FALSE(content.isTruncated);
+    }
+    EXPECT_EQ(addr2->toString(), recv1->getSource().toString());
+    {
+      auto ancillary = recv1->getAncillary();
+      EXPECT_TRUE(ancillary.isTruncated);
+      EXPECT_EQ(0, ancillary.value.size());
+    }
+#endif
+  }
+}
+
 }  // namespace
 }  // namespace kj

c++/src/kj/async-io.c++

@@ -39,6 +39,11 @@
 #include <netdb.h>
 #include <set>
 #include <poll.h>
+#include <limits.h>
+
+#if !defined(IOV_MAX) && defined(UIO_MAXIOV)
+#define IOV_MAX UIO_MAXIOV
+#endif
 
 namespace kj {
 
@@ -175,6 +180,16 @@ public:
     KJ_SYSCALL(shutdown(fd, SHUT_WR));
   }
 
+  void getsockopt(int level, int option, void* value, uint* length) override {
+    socklen_t socklen = *length;
+    KJ_SYSCALL(::getsockopt(fd, level, option, value, &socklen));
+    *length = socklen;
+  }
+
+  void setsockopt(int level, int option, const void* value, uint length) override {
+    KJ_SYSCALL(::setsockopt(fd, level, option, value, length));
+  }
+
   Promise<void> waitConnected() {
     // Wait until initial connection has completed. This actually just waits until it is writable.
 
@@ -269,14 +284,19 @@ private:
 
   Promise<void> writeInternal(ArrayPtr<const byte> firstPiece,
                               ArrayPtr<const ArrayPtr<const byte>> morePieces) {
-    KJ_STACK_ARRAY(struct iovec, iov, 1 + morePieces.size(), 16, 128);
+    // If there are more than IOV_MAX pieces, we'll only write the first IOV_MAX for now, and
+    // then we'll loop later.
+    KJ_STACK_ARRAY(struct iovec, iov, kj::min(1 + morePieces.size(), IOV_MAX), 16, 128);
+    size_t iovTotal = 0;
 
     // writev() interface is not const-correct.  :(
     iov[0].iov_base = const_cast<byte*>(firstPiece.begin());
     iov[0].iov_len = firstPiece.size();
-    for (uint i = 0; i < morePieces.size(); i++) {
-      iov[i + 1].iov_base = const_cast<byte*>(morePieces[i].begin());
-      iov[i + 1].iov_len = morePieces[i].size();
+    iovTotal += iov[0].iov_len;
+    for (uint i = 1; i < iov.size(); i++) {
+      iov[i].iov_base = const_cast<byte*>(morePieces[i - 1].begin());
+      iov[i].iov_len = morePieces[i - 1].size();
+      iovTotal += iov[i].iov_len;
     }
 
     ssize_t writeResult;
@@ -302,6 +322,13 @@ private:
       if (n < firstPiece.size()) {
         // Only part of the first piece was consumed.  Wait for buffer space and then write again.
         firstPiece = firstPiece.slice(n, firstPiece.size());
+        iovTotal -= n;
+
+        if (iovTotal == 0) {
+          // Oops, what actually happened is that we hit the IOV_MAX limit. Don't wait.
+          return writeInternal(firstPiece, morePieces);
+        }
+
         return observer.whenBecomesWritable().then([=]() {
           return writeInternal(firstPiece, morePieces);
         });
@@ -312,6 +339,7 @@ private:
       } else {
         // First piece was fully consumed, so move on to the next piece.
         n -= firstPiece.size();
+        iovTotal -= firstPiece.size();
         firstPiece = morePieces[0];
         morePieces = morePieces.slice(1, morePieces.size());
       }
@@ -340,6 +368,9 @@ public:
     return memcmp(&addr.generic, &other.addr.generic, addrlen) < 0;
   }
 
+  const struct sockaddr* getRaw() const { return &addr.generic; }
+  socklen_t getRawSize() const { return addrlen; }
+
   int socket(int type) const {
     bool isStream = type == SOCK_STREAM;
 
@@ -782,7 +813,50 @@ public:
     return SocketAddress::getLocalAddress(fd).getPort();
   }
 
+  void getsockopt(int level, int option, void* value, uint* length) override {
+    socklen_t socklen = *length;
+    KJ_SYSCALL(::getsockopt(fd, level, option, value, &socklen));
+    *length = socklen;
+  }
+  void setsockopt(int level, int option, const void* value, uint length) override {
+    KJ_SYSCALL(::setsockopt(fd, level, option, value, length));
+  }
+
+public:
+  UnixEventPort& eventPort;
+  UnixEventPort::FdObserver observer;
+};
+
+class DatagramPortImpl final: public DatagramPort, public OwnedFileDescriptor {
+public:
+  DatagramPortImpl(LowLevelAsyncIoProvider& lowLevel, UnixEventPort& eventPort, int fd, uint flags)
+      : OwnedFileDescriptor(fd, flags), lowLevel(lowLevel), eventPort(eventPort),
+        observer(eventPort, fd, UnixEventPort::FdObserver::OBSERVE_READ |
+                                UnixEventPort::FdObserver::OBSERVE_WRITE) {}
+
+  Promise<size_t> send(const void* buffer, size_t size, NetworkAddress& destination) override;
+  Promise<size_t> send(
+      ArrayPtr<const ArrayPtr<const byte>> pieces, NetworkAddress& destination) override;
+
+  class ReceiverImpl;
+
+  Own<DatagramReceiver> makeReceiver(DatagramReceiver::Capacity capacity) override;
+
+  uint getPort() override {
+    return SocketAddress::getLocalAddress(fd).getPort();
+  }
+
+  void getsockopt(int level, int option, void* value, uint* length) override {
+    socklen_t socklen = *length;
+    KJ_SYSCALL(::getsockopt(fd, level, option, value, &socklen));
+    *length = socklen;
+  }
+  void setsockopt(int level, int option, const void* value, uint length) override {
+    KJ_SYSCALL(::setsockopt(fd, level, option, value, length));
+  }
+
 public:
+  LowLevelAsyncIoProvider& lowLevel;
   UnixEventPort& eventPort;
   UnixEventPort::FdObserver observer;
 };
@@ -838,6 +912,9 @@ public:
   Own<ConnectionReceiver> wrapListenSocketFd(int fd, uint flags = 0) override {
     return heap<FdConnectionReceiver>(eventPort, fd, flags);
   }
+  Own<DatagramPort> wrapDatagramSocketFd(int fd, uint flags = 0) override {
+    return heap<DatagramPortImpl>(*this, eventPort, fd, flags);
+  }
 
   Timer& getTimer() override { return timer; }
 
@@ -887,13 +964,46 @@ public:
     return lowLevel.wrapListenSocketFd(fd, NEW_FD_FLAGS);
   }
 
+  Own<DatagramPort> bindDatagramPort() override {
+    if (addrs.size() > 1) {
+      KJ_LOG(WARNING, "Bind address resolved to multiple addresses.  Only the first address will "
+          "be used.  If this is incorrect, specify the address numerically.  This may be fixed "
+          "in the future.", addrs[0].toString());
+    }
+
+    int fd = addrs[0].socket(SOCK_DGRAM);
+
+    {
+      KJ_ON_SCOPE_FAILURE(close(fd));
+
+      // We always enable SO_REUSEADDR because having to take your server down for five minutes
+      // before it can restart really sucks.
+      int optval = 1;
+      KJ_SYSCALL(setsockopt(fd, SOL_SOCKET, SO_REUSEADDR, &optval, sizeof(optval)));
+
+      addrs[0].bind(fd);
+    }
+
+    return lowLevel.wrapDatagramSocketFd(fd, NEW_FD_FLAGS);
+  }
+
+  Own<NetworkAddress> clone() override {
+    return kj::heap<NetworkAddressImpl>(lowLevel, kj::heapArray(addrs.asPtr()));
+  }
+
   String toString() override {
     return strArray(KJ_MAP(addr, addrs) { return addr.toString(); }, ",");
   }
 
+  const SocketAddress& chooseOneAddress() {
+    KJ_REQUIRE(addrs.size() > 0, "No addresses available.");
+    return addrs[counter++ % addrs.size()];
+  }
+
 private:
   LowLevelAsyncIoProvider& lowLevel;
   Array<SocketAddress> addrs;
+  uint counter = 0;
 
   Promise<Own<AsyncIoStream>> connectImpl(uint index) {
     KJ_ASSERT(index < addrs.size());
@@ -957,6 +1067,189 @@ private:
 
 // =======================================================================================
 
+Promise<size_t> DatagramPortImpl::send(
+    const void* buffer, size_t size, NetworkAddress& destination) {
+  auto& addr = downcast<NetworkAddressImpl>(destination).chooseOneAddress();
+
+  ssize_t n;
+  KJ_NONBLOCKING_SYSCALL(n = sendto(fd, buffer, size, 0, addr.getRaw(), addr.getRawSize()));
+  if (n < 0) {
+    // Write buffer full.
+    return observer.whenBecomesWritable().then([this, buffer, size, &destination]() {
+      return send(buffer, size, destination);
+    });
+  } else {
+    // If less than the whole message was sent, then it got truncated, and there's nothing we can
+    // do about it.
+    return n;
+  }
+}
+
+Promise<size_t> DatagramPortImpl::send(
+    ArrayPtr<const ArrayPtr<const byte>> pieces, NetworkAddress& destination) {
+  struct msghdr msg;
+  memset(&msg, 0, sizeof(msg));
+
+  auto& addr = downcast<NetworkAddressImpl>(destination).chooseOneAddress();
+  msg.msg_name = const_cast<void*>(implicitCast<const void*>(addr.getRaw()));
+  msg.msg_namelen = addr.getRawSize();
+
+  KJ_STACK_ARRAY(struct iovec, iov, kj::min(pieces.size(), IOV_MAX), 16, 64);
+
+  for (size_t i: kj::indices(pieces)) {
+    iov[i].iov_base = const_cast<void*>(implicitCast<const void*>(pieces[i].begin()));
+    iov[i].iov_len = pieces[i].size();
+  }
+
+  Array<byte> extra;
+  if (pieces.size() > IOV_MAX) {
+    // Too many pieces, but we can't use multiple syscalls because they'd send separate
+    // datagrams. We'll have to copy the trailing pieces into a temporary array.
+    //
+    // TODO(perf): On Linux we could use multiple syscalls via MSG_MORE.
+    size_t extraSize = 0;
+    for (size_t i = IOV_MAX - 1; i < pieces.size(); i++) {
+      extraSize += pieces[i].size();
+    }
+    extra = kj::heapArray<byte>(extraSize);
+    extraSize = 0;
+    for (size_t i = IOV_MAX - 1; i < pieces.size(); i++) {
+      memcpy(extra.begin() + extraSize, pieces[i].begin(), pieces[i].size());
+      extraSize += pieces[i].size();
+    }
+    iov[IOV_MAX - 1].iov_base = extra.begin();
+    iov[IOV_MAX - 1].iov_len = extra.size();
+  }
+
+  msg.msg_iov = iov.begin();
+  msg.msg_iovlen = iov.size();
+
+  ssize_t n;
+  KJ_NONBLOCKING_SYSCALL(n = sendmsg(fd, &msg, 0));
+  if (n < 0) {
+    // Write buffer full.
+    return observer.whenBecomesWritable().then([this, pieces, &destination]() {
+      return send(pieces, destination);
+    });
+  } else {
+    // If less than the whole message was sent, then it was truncated, and there's nothing we can
+    // do about that now.
+    return n;
+  }
+}
+
+class DatagramPortImpl::ReceiverImpl final: public DatagramReceiver {
+public:
+  explicit ReceiverImpl(DatagramPortImpl& port, Capacity capacity)
+      : port(port),
+        contentBuffer(heapArray<byte>(capacity.content)),
+        ancillaryBuffer(capacity.ancillary > 0 ? heapArray<byte>(capacity.ancillary)
+                                               : Array<byte>(nullptr)) {}
+
+  Promise<void> receive() override {
+    struct msghdr msg;
+    memset(&msg, 0, sizeof(msg));
+
+    struct sockaddr_storage addr;
+    memset(&addr, 0, sizeof(addr));
+    msg.msg_name = &addr;
+    msg.msg_namelen = sizeof(addr);
+
+    struct iovec iov;
+    iov.iov_base = contentBuffer.begin();
+    iov.iov_len = contentBuffer.size();
+    msg.msg_iov = &iov;
+    msg.msg_iovlen = 1;
+    msg.msg_control = ancillaryBuffer.begin();
+    msg.msg_controllen = ancillaryBuffer.size();
+
+    ssize_t n;
+    KJ_NONBLOCKING_SYSCALL(n = recvmsg(port.fd, &msg, 0));
+
+    if (n < 0) {
+      // No data available. Wait.
+      return port.observer.whenBecomesReadable().then([this]() {
+        return receive();
+      });
+    } else {
+      receivedSize = n;
+      contentTruncated = msg.msg_flags & MSG_TRUNC;
+
+      source.emplace(port.lowLevel, msg.msg_name, msg.msg_namelen);
+
+      ancillaryList.resize(0);
+      ancillaryTruncated = msg.msg_flags & MSG_CTRUNC;
+
+      for (struct cmsghdr* cmsg = CMSG_FIRSTHDR(&msg); cmsg != nullptr;
+           cmsg = CMSG_NXTHDR(&msg, cmsg)) {
+        // On some platforms (OSX), a cmsghdr's length may cross the end of the ancillary buffer
+        // when truncated. On other platforms (Linux) the length in cmsghdr will itself be
+        // truncated to fit within the buffer.
+
+        const byte* pos = reinterpret_cast<const byte*>(cmsg);
+        size_t available = ancillaryBuffer.end() - pos;
+        if (available < CMSG_SPACE(0)) {
+          // The buffer ends in the middle of the header. We can't use this message.
+          // (On Linux, this never happens, because the message is not included if there isn't
+          // space for a header. I'm not sure how other systems behave, though, so let's be safe.)
+          break;
+        }
+
+        // OK, we know the cmsghdr is valid, at least.
+
+        // Find the start of the message payload.
+        const byte* begin = CMSG_DATA(cmsg);
+
+        // Cap the message length to the available space.
+        const byte* end = pos + kj::min(available, cmsg->cmsg_len);
+
+        ancillaryList.add(AncillaryMessage(
+            cmsg->cmsg_level, cmsg->cmsg_type, arrayPtr(begin, end)));
+      }
+
+      return READY_NOW;
+    }
+  }
+
+  MaybeTruncated<ArrayPtr<const byte>> getContent() override {
+    return { contentBuffer.slice(0, receivedSize), contentTruncated };
+  }
+
+  MaybeTruncated<ArrayPtr<const AncillaryMessage>> getAncillary() override {
+    return { ancillaryList.asPtr(), ancillaryTruncated };
+  }
+
+  NetworkAddress& getSource() override {
+    return KJ_REQUIRE_NONNULL(source, "Haven't sent a message yet.").abstract;
+  }
+
+private:
+  DatagramPortImpl& port;
+  Array<byte> contentBuffer;
+  Array<byte> ancillaryBuffer;
+  Vector<AncillaryMessage> ancillaryList;
+  size_t receivedSize = 0;
+  bool contentTruncated = false;
+  bool ancillaryTruncated = false;
+
+  struct StoredAddress {
+    StoredAddress(LowLevelAsyncIoProvider& lowLevel, const void* sockaddr, uint length)
+        : raw(sockaddr, length),
+          abstract(lowLevel, Array<SocketAddress>(&raw, 1, NullArrayDisposer::instance)) {}
+
+    SocketAddress raw;
+    NetworkAddressImpl abstract;
+  };
+
+  kj::Maybe<StoredAddress> source;
+};
+
+Own<DatagramReceiver> DatagramPortImpl::makeReceiver(DatagramReceiver::Capacity capacity) {
+  return kj::heap<ReceiverImpl>(*this, capacity);
+}
+
+// =======================================================================================
+
 class AsyncIoProviderImpl final: public AsyncIoProvider {
 public:
   AsyncIoProviderImpl(LowLevelAsyncIoProvider& lowLevel)
@@ -1030,6 +1323,31 @@ Promise<void> AsyncInputStream::read(void* buffer, size_t bytes) {
   return read(buffer, bytes, bytes).then([](size_t) {});
 }
 
+void AsyncIoStream::getsockopt(int level, int option, void* value, uint* length) {
+  KJ_UNIMPLEMENTED("Not a socket.");
+}
+void AsyncIoStream::setsockopt(int level, int option, const void* value, uint length) {
+  KJ_UNIMPLEMENTED("Not a socket.");
+}
+void ConnectionReceiver::getsockopt(int level, int option, void* value, uint* length) {
+  KJ_UNIMPLEMENTED("Not a socket.");
+}
+void ConnectionReceiver::setsockopt(int level, int option, const void* value, uint length) {
+  KJ_UNIMPLEMENTED("Not a socket.");
+}
+void DatagramPort::getsockopt(int level, int option, void* value, uint* length) {
+  KJ_UNIMPLEMENTED("Not a socket.");
+}
+void DatagramPort::setsockopt(int level, int option, const void* value, uint length) {
+  KJ_UNIMPLEMENTED("Not a socket.");
+}
+Own<DatagramPort> NetworkAddress::bindDatagramPort() {
+  KJ_UNIMPLEMENTED("Datagram sockets not implemented.");
+}
+Own<DatagramPort> LowLevelAsyncIoProvider::wrapDatagramSocketFd(int fd, uint flags) {
+  KJ_UNIMPLEMENTED("Datagram sockets not implemented.");
+}
+
 Own<AsyncIoProvider> newAsyncIoProvider(LowLevelAsyncIoProvider& lowLevel) {
   return kj::heap<AsyncIoProviderImpl>(lowLevel);
 }

c++/src/kj/async-io.h

@@ -34,6 +34,10 @@
 namespace kj {
 
 class UnixEventPort;
+class NetworkAddress;
+
+// =======================================================================================
+// Streaming I/O
 
 class AsyncInputStream {
   // Asynchronous equivalent of InputStream (from io.h).
@@ -59,6 +63,26 @@ class AsyncIoStream: public AsyncInputStream, public AsyncOutputStream {
 public:
   virtual void shutdownWrite() = 0;
   // Cleanly shut down just the write end of the stream, while keeping the read end open.
+
+  virtual void getsockopt(int level, int option, void* value, uint* length);
+  virtual void setsockopt(int level, int option, const void* value, uint length);
+  // Corresponds to getsockopt() and setsockopt() syscalls. Will throw an "unimplemented" exception
+  // if the stream is not a socket or the option is not appropriate for the socket type. The
+  // default implementations always throw "unimplemented".
+};
+
+struct OneWayPipe {
+  // A data pipe with an input end and an output end.  (Typically backed by pipe() system call.)
+
+  Own<AsyncInputStream> in;
+  Own<AsyncOutputStream> out;
+};
+
+struct TwoWayPipe {
+  // A data pipe that supports sending in both directions.  Each end's output sends data to the
+  // other end's input.  (Typically backed by socketpair() system call.)
+
+  Own<AsyncIoStream> ends[2];
 };
 
 class ConnectionReceiver {
@@ -70,9 +94,120 @@ public:
 
   virtual uint getPort() = 0;
   // Gets the port number, if applicable (i.e. if listening on IP).  This is useful if you didn't
-  // specify a port when constructing the LocalAddress -- one will have been assigned automatically.
+  // specify a port when constructing the NetworkAddress -- one will have been assigned
+  // automatically.
+
+  virtual void getsockopt(int level, int option, void* value, uint* length);
+  virtual void setsockopt(int level, int option, const void* value, uint length);
+  // Same as the methods of AsyncIoStream.
+};
+
+// =======================================================================================
+// Datagram I/O
+
+class AncillaryMessage {
+  // Represents an ancillary message (aka control message) received using the recvmsg() system
+  // call (or equivalent). Most apps will not use this.
+
+public:
+  inline AncillaryMessage(int level, int type, ArrayPtr<const byte> data);
+  AncillaryMessage() = default;
+
+  inline int getLevel() const;
+  // Originating protocol / socket level.
+
+  inline int getType() const;
+  // Protocol-specific message type.
+
+  template <typename T>
+  inline Maybe<const T&> as();
+  // Interpret the ancillary message as the given struct type. Most ancillary messages are some
+  // sort of struct, so this is a convenient way to access it. Returns nullptr if the message
+  // is smaller than the struct -- this can happen if the message was truncated due to
+  // insufficient ancillary buffer space.
+
+  template <typename T>
+  inline ArrayPtr<const T> asArray();
+  // Interpret the ancillary message as an array of items. If the message size does not evenly
+  // divide into elements of type T, the remainder is discarded -- this can happen if the message
+  // was truncated due to insufficient ancillary buffer space.
+
+private:
+  int level;
+  int type;
+  ArrayPtr<const byte> data;
+  // Message data. In most cases you should use `as()` or `asArray()`.
+};
+
+class DatagramReceiver {
+  // Class encapsulating the recvmsg() system call. You must specify the DatagramReceiver's
+  // capacity in advance; if a received packet is larger than the capacity, it will be truncated.
+
+public:
+  virtual Promise<void> receive() = 0;
+  // Receive a new message, overwriting this object's content.
+  //
+  // receive() may reuse the same buffers for content and ancillary data with each call.
+
+  template <typename T>
+  struct MaybeTruncated {
+    T value;
+
+    bool isTruncated;
+    // True if the Receiver's capacity was insufficient to receive the value and therefore the
+    // value is truncated.
+  };
+
+  virtual MaybeTruncated<ArrayPtr<const byte>> getContent() = 0;
+  // Get the content of the datagram.
+
+  virtual MaybeTruncated<ArrayPtr<const AncillaryMessage>> getAncillary() = 0;
+  // Ancilarry messages received with the datagram. See the recvmsg() system call and the cmsghdr
+  // struct. Most apps don't need this.
+  //
+  // If the returned value is truncated, then the last message in the array may itself be
+  // truncated, meaning its as<T>() method will return nullptr or its asArray<T>() method will
+  // return fewer elements than expected. Truncation can also mean that additional messages were
+  // available but discarded.
+
+  virtual NetworkAddress& getSource() = 0;
+  // Get the datagram sender's address.
+
+  struct Capacity {
+    size_t content = 8192;
+    // How much space to allocate for the datagram content. If a datagram is received that is
+    // larger than this, it will be truncated, with no way to recover the tail.
+
+    size_t ancillary = 0;
+    // How much space to allocate for ancillary messages. As with content, if the ancillary data
+    // is larger than this, it will be truncated.
+  };
+};
+
+class DatagramPort {
+public:
+  virtual Promise<size_t> send(const void* buffer, size_t size, NetworkAddress& destination) = 0;
+  virtual Promise<size_t> send(ArrayPtr<const ArrayPtr<const byte>> pieces,
+                               NetworkAddress& destination) = 0;
+
+  virtual Own<DatagramReceiver> makeReceiver(
+      DatagramReceiver::Capacity capacity = DatagramReceiver::Capacity()) = 0;
+  // Create a new `Receiver` that can be used to receive datagrams. `capacity` specifies how much
+  // space to allocate for the received message. The `DatagramPort` must outlive the `Receiver`.
+
+  virtual uint getPort() = 0;
+  // Gets the port number, if applicable (i.e. if listening on IP).  This is useful if you didn't
+  // specify a port when constructing the NetworkAddress -- one will have been assigned
+  // automatically.
+
+  virtual void getsockopt(int level, int option, void* value, uint* length);
+  virtual void setsockopt(int level, int option, const void* value, uint length);
+  // Same as the methods of AsyncIoStream.
 };
 
+// =======================================================================================
+// Networks
+
 class NetworkAddress {
   // Represents a remote address to which the application can connect.
 
@@ -87,6 +222,14 @@ public:
   //
   // The address must be local.
 
+  virtual Own<DatagramPort> bindDatagramPort();
+  // Open this address as a datagram (e.g. UDP) port.
+  //
+  // The address must be local.
+
+  virtual Own<NetworkAddress> clone() = 0;
+  // Returns an equivalent copy of this NetworkAddress.
+
   virtual String toString() = 0;
   // Produce a human-readable string which hopefully can be passed to Network::parseRemoteAddress()
   // to reproduce this address, although whether or not that works of course depends on the Network
@@ -123,26 +266,15 @@ public:
   // `portHint`, if provided, specifies the "standard" IP port number for the application-level
   // service in play.  If the address turns out to be an IP address (v4 or v6), and it lacks a
   // port number, this port will be used.  If `addr` lacks a port number *and* `portHint` is
-  // omitted, then the returned address will only support listen() (not connect()), and a port
-  // will be chosen when listen() is called.
+  // omitted, then the returned address will only support listen() and bindDatagramPort()
+  // (not connect()), and an unused port will be chosen each time one of those methods is called.
 
   virtual Own<NetworkAddress> getSockaddr(const void* sockaddr, uint len) = 0;
   // Construct a network address from a legacy struct sockaddr.
 };
 
-struct OneWayPipe {
-  // A data pipe with an input end and an output end.  (Typically backed by pipe() system call.)
-
-  Own<AsyncInputStream> in;
-  Own<AsyncOutputStream> out;
-};
-
-struct TwoWayPipe {
-  // A data pipe that supports sending in both directions.  Each end's output sends data to the
-  // other end's input.  (Typically backed by socketpair() system call.)
-
-  Own<AsyncIoStream> ends[2];
-};
+// =======================================================================================
+// I/O Provider
 
 class AsyncIoProvider {
   // Class which constructs asynchronous wrappers around the operating system's I/O facilities.
@@ -284,6 +416,8 @@ public:
   //
   // `flags` is a bitwise-OR of the values of the `Flags` enum.
 
+  virtual Own<DatagramPort> wrapDatagramSocketFd(int fd, uint flags = 0);
+
   virtual Timer& getTimer() = 0;
   // Returns a `Timer` based on real time.  Time does not pass while event handlers are running --
   // it only updates when the event loop polls for system events.  This means that calling `now()`
@@ -328,6 +462,30 @@ AsyncIoContext setupAsyncIo();
 //       return 0;
 //     }
 
+// =======================================================================================
+// inline implementation details
+
+inline AncillaryMessage::AncillaryMessage(
+    int level, int type, ArrayPtr<const byte> data)
+    : level(level), type(type), data(data) {}
+
+inline int AncillaryMessage::getLevel() const { return level; }
+inline int AncillaryMessage::getType() const { return type; }
+
+template <typename T>
+inline Maybe<const T&> AncillaryMessage::as() {
+  if (data.size() >= sizeof(T)) {
+    return *reinterpret_cast<const T*>(data.begin());
+  } else {
+    return nullptr;
+  }
+}
+
+template <typename T>
+inline ArrayPtr<const T> AncillaryMessage::asArray() {
+  return arrayPtr(reinterpret_cast<const T*>(data.begin()), data.size() / sizeof(T));
+}
+
 }  // namespace kj
 
 #endif  // KJ_ASYNC_IO_H_

c++/src/kj/common.h

@@ -797,6 +797,16 @@ public:
   inline operator T*() { return isSet ? &value : nullptr; }
   inline operator const T*() const { return isSet ? &value : nullptr; }
 
+  template <typename... Params>
+  inline void emplace(Params&&... params) {
+    if (isSet) {
+      isSet = false;
+      dtor(value);
+    }
+    ctor(value, kj::fwd<Params>(params)...);
+    isSet = true;
+  }
+
 private:  // internal interface used by friends only
   inline NullableValue() noexcept: isSet(false) {}
   inline NullableValue(T&& t) noexcept(noexcept(T(instance<T&&>())))
@@ -959,6 +969,14 @@ public:
 
   Maybe(decltype(nullptr)) noexcept: ptr(nullptr) {}
 
+  template <typename... Params>
+  inline void emplace(Params&&... params) {
+    // Replace this Maybe's content with a new value constructed by passing the given parametrs to
+    // T's constructor. This can be used to initialize a Maybe without copying or even moving a T.
+
+    ptr.emplace(kj::fwd<Params>(params)...);
+  }
+
   inline Maybe& operator=(Maybe&& other) { ptr = kj::mv(other.ptr); return *this; }
   inline Maybe& operator=(Maybe& other) { ptr = other.ptr; return *this; }
   inline Maybe& operator=(const Maybe& other) { ptr = other.ptr; return *this; }

super-test.sh

@@ -19,7 +19,7 @@ while [ $# -gt 0 ]; do
     caffeinate )
       # Re-run preventing sleep.
       shift
-      exec caffeinate $0 $@
+      exec caffeinate -ims $0 $@
       ;;
     tmpdir )
       # Clone to a temp directory.