Async IO interfaces for low-level networking and similar.

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

+// Copyright (c) 2013, Kenton Varda <temporal@gmail.com>
+// All rights reserved.
+//
+// Redistribution and use in source and binary forms, with or without
+// modification, are permitted provided that the following conditions are met:
+//
+// 1. Redistributions of source code must retain the above copyright notice, this
+//    list of conditions and the following disclaimer.
+// 2. Redistributions in binary form must reproduce the above copyright notice,
+//    this list of conditions and the following disclaimer in the documentation
+//    and/or other materials provided with the distribution.
+//
+// THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND
+// ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
+// WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
+// DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE LIABLE FOR
+// ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES
+// (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
+// LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND
+// ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+// (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
+// SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+
+#include "async-io.h"
+#include "async-unix.h"
+#include "debug.h"
+#include <gtest/gtest.h>
+
+namespace kj {
+namespace {
+
+class DummyErrorHandler: public TaskSet::ErrorHandler {
+public:
+  void taskFailed(kj::Exception&& exception) override {
+    kj::throwRecoverableException(kj::mv(exception));
+  }
+};
+
+TEST(AsyncIo, SimpleNetwork) {
+  UnixEventLoop loop;
+  DummyErrorHandler dummyHandler;
+  TaskSet tasks(loop, dummyHandler);
+  auto& network = getOperatingSystemSingleton().getNetwork();
+
+  Own<ConnectionReceiver> listener;
+  Own<AsyncIoStream> server;
+  Own<AsyncIoStream> client;
+
+  char receiveBuffer[4];
+
+  auto port = newPromiseAndFulfiller<uint>();
+
+  tasks.add(loop.evalLater([&]() {
+    return port.promise
+        .then([&](uint portnum) {
+          return network.parseRemoteAddress("127.0.0.1", portnum);
+        }).then([&](Own<RemoteAddress>&& result) {
+          return result->connect();
+        }).then([&](Own<AsyncIoStream>&& result) {
+          client = kj::mv(result);
+          return client->write("foo", 3);
+        });
+  }));
+
+  kj::String result = loop.wait(loop.evalLater([&]() {
+    return network.parseLocalAddress("*")
+        .then([&](Own<LocalAddress>&& result) {
+          listener = result->listen();
+          port.fulfiller->fulfill(listener->getPort());
+          return listener->accept();
+        }).then([&](Own<AsyncIoStream>&& result) {
+          server = kj::mv(result);
+          return server->tryRead(receiveBuffer, 3, 4);
+        }).then([&](size_t n) {
+          EXPECT_EQ(3u, n);
+          return heapString(receiveBuffer, n);
+        });
+  }));
+
+  EXPECT_EQ("foo", result);
+}
+
+String tryParseLocal(EventLoop& loop, Network& network, StringPtr text, uint portHint = 0) {
+  return loop.wait(loop.evalLater([&]() {
+    return network.parseLocalAddress(text, portHint);
+  }))->toString();
+}
+
+String tryParseRemote(EventLoop& loop, Network& network, StringPtr text, uint portHint = 0) {
+  return loop.wait(loop.evalLater([&]() {
+    return network.parseRemoteAddress(text, portHint);
+  }))->toString();
+}
+
+TEST(AsyncIo, AddressParsing) {
+  UnixEventLoop loop;
+  auto& network = getOperatingSystemSingleton().getNetwork();
+
+  EXPECT_EQ("*:0", tryParseLocal(loop, network, "*"));
+  EXPECT_EQ("*:123", tryParseLocal(loop, network, "123"));
+  EXPECT_EQ("*:123", tryParseLocal(loop, network, ":123"));
+  EXPECT_EQ("[::]:123", tryParseLocal(loop, network, "0::0", 123));
+  EXPECT_EQ("0.0.0.0:0", tryParseLocal(loop, network, "0.0.0.0"));
+  EXPECT_EQ("1.2.3.4:5678", tryParseRemote(loop, network, "1.2.3.4", 5678));
+  EXPECT_EQ("[12ab:cd::34]:321", tryParseRemote(loop, network, "[12ab:cd:0::0:34]:321", 432));
+
+  EXPECT_EQ("unix:foo/bar/baz", tryParseLocal(loop, network, "unix:foo/bar/baz"));
+  EXPECT_EQ("unix:foo/bar/baz", tryParseRemote(loop, network, "unix:foo/bar/baz"));
+}
+
+}  // namespace
+}  // namespace kj

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

+// Copyright (c) 2013, Kenton Varda <temporal@gmail.com>
+// All rights reserved.
+//
+// Redistribution and use in source and binary forms, with or without
+// modification, are permitted provided that the following conditions are met:
+//
+// 1. Redistributions of source code must retain the above copyright notice, this
+//    list of conditions and the following disclaimer.
+// 2. Redistributions in binary form must reproduce the above copyright notice,
+//    this list of conditions and the following disclaimer in the documentation
+//    and/or other materials provided with the distribution.
+//
+// THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND
+// ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
+// WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
+// DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE LIABLE FOR
+// ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES
+// (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
+// LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND
+// ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+// (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
+// SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+
+#ifndef KJ_ASYNC_IO_H_
+#define KJ_ASYNC_IO_H_
+
+#include "async.h"
+
+namespace kj {
+
+class AsyncInputStream {
+public:
+  virtual Promise<size_t> read(void* buffer, size_t minBytes, size_t maxBytes) = 0;
+  virtual Promise<size_t> tryRead(void* buffer, size_t minBytes, size_t maxBytes) = 0;
+
+  Promise<size_t> read(void* buffer, size_t bytes);
+};
+
+class AsyncOutputStream {
+public:
+  virtual Promise<void> write(const void* buffer, size_t size) = 0;
+  virtual Promise<void> write(ArrayPtr<const ArrayPtr<const byte>> pieces) = 0;
+};
+
+class AsyncIoStream: public AsyncInputStream, public AsyncOutputStream {
+public:
+};
+
+class ConnectionReceiver {
+public:
+  virtual Promise<Own<AsyncIoStream>> accept() = 0;
+
+  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.
+};
+
+class RemoteAddress {
+  // Represents a remote address to which the application can connect.
+
+public:
+  virtual Promise<Own<AsyncIoStream>> connect() = 0;
+
+  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
+  // implementation.  This should be called only to display the address to human users, who will
+  // hopefully know what they are able to do with it.
+};
+
+class LocalAddress {
+  // Represents a local address on which the application can potentially accept connections.
+
+public:
+  virtual Own<ConnectionReceiver> listen() = 0;
+
+  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
+  // implementation.  This should be called only to display the address to human users, who will
+  // hopefully know what they are able to do with it.
+};
+
+class Network {
+  // Factory for LocalAddress and RemoteAddress instances, representing the network services
+  // offered by the operating system.
+  //
+  // This interface typically represents broad authority, and well-designed code should limit its
+  // use to high-level startup code and user interaction.  Low-level APIs should accept
+  // LocalAddress and/or RemoteAddress instances directly and work from there, if at all possible.
+
+public:
+  virtual Promise<Own<LocalAddress>> parseLocalAddress(StringPtr addr, uint portHint = 0) = 0;
+  virtual Promise<Own<RemoteAddress>> parseRemoteAddress(StringPtr addr, uint portHint = 0) = 0;
+  // Construct a local or remote address from a user-provided string.  The format of the address
+  // strings is not specified at the API level, and application code should make no assumptions
+  // about them.  These strings should always be provided by humans, and said humans will know
+  // what format to use in their particular context.
+  //
+  // `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.
+  //
+  // In practice, a local address is usually just a port number (or even an empty string, if a
+  // reasonable `portHint` is provided), whereas a remote address usually requires a hostname.
+
+  virtual Own<LocalAddress> getLocalSockaddr(const void* sockaddr, uint len) = 0;
+  virtual Own<RemoteAddress> getRemoteSockaddr(const void* sockaddr, uint len) = 0;
+  // Construct a local or remote address from a legacy struct sockaddr.
+};
+
+class OperatingSystem {
+  // Interface representing the I/O facilities offered to a process by the operating system.  This
+  // interface usually should be used only in the highest levels of the application, in order to
+  // set up the right connections to pass down to lower levels that do the actual work.
+
+public:
+  virtual AsyncIoStream& getStandardIo() = 0;
+  virtual AsyncOutputStream& getStandardError() = 0;
+
+  virtual Network& getNetwork() = 0;
+
+  // TODO(someday):  Filesystem.  Should it even be async?
+//  virtual Directory& getCurrentDir() = 0;
+//  virtual Directory& getRootDir() = 0;
+};
+
+OperatingSystem& getOperatingSystemSingleton();
+// Get the EVIL singleton instance of OperatingSystem representing the real kernel.
+//
+// DO NOT USE THIS except at the highest levels of your code, ideally in the main() function.  If
+// you call this from low-level code, then you are preventing higher-level code from injecting an
+// alternative implementation.  Instead, if your code needs to use OS functionality, it should ask
+// for an OperatingSystem as a parameter.  See:
+//     http://www.object-oriented-security.org/lets-argue/singletons
+//
+// If you use KJ_MAIN, you never have to call this at all, because your main function will receive
+// an OperatingSystem as part of the process context.
+
+}  // namespace kj
+
+#endif  // KJ_ASYNC_IO_H_

c++/src/kj/common-test.c++

@@ -52,6 +52,7 @@ TEST(Common, Maybe) {
     } else {
       ADD_FAILURE();
     }
+    EXPECT_EQ(123, m.orDefault(456));
   }
 
   {
@@ -66,6 +67,7 @@ TEST(Common, Maybe) {
       ADD_FAILURE();
       EXPECT_EQ(0, *v);  // avoid unused warning
     }
+    EXPECT_EQ(456, m.orDefault(456));
   }
 
   int i = 234;
@@ -83,6 +85,7 @@ TEST(Common, Maybe) {
     } else {
       ADD_FAILURE();
     }
+    EXPECT_EQ(234, m.orDefault(456));
   }
 
   {
@@ -97,6 +100,7 @@ TEST(Common, Maybe) {
       ADD_FAILURE();
       EXPECT_EQ(0, *v);  // avoid unused warning
     }
+    EXPECT_EQ(456, m.orDefault(456));
   }
 
   {
@@ -113,6 +117,7 @@ TEST(Common, Maybe) {
     } else {
       ADD_FAILURE();
     }
+    EXPECT_EQ(234, m.orDefault(456));
   }
 
   {
@@ -127,6 +132,7 @@ TEST(Common, Maybe) {
       ADD_FAILURE();
       EXPECT_EQ(0, *v);  // avoid unused warning
     }
+    EXPECT_EQ(456, m.orDefault(456));
   }
 
   {

c++/src/kj/common.h

@@ -730,6 +730,21 @@ public:
   inline bool operator==(decltype(nullptr)) const { return ptr == nullptr; }
   inline bool operator!=(decltype(nullptr)) const { return ptr != nullptr; }
 
+  T& orDefault(T& defaultValue) {
+    if (ptr == nullptr) {
+      return defaultValue;
+    } else {
+      return *ptr;
+    }
+  }
+  const T& orDefault(const T& defaultValue) const {
+    if (ptr == nullptr) {
+      return defaultValue;
+    } else {
+      return *ptr;
+    }
+  }
+
   template <typename Func>
   auto map(Func&& f) -> Maybe<decltype(f(instance<T&>()))> {
     if (ptr == nullptr) {
@@ -787,6 +802,21 @@ public:
   inline bool operator==(decltype(nullptr)) const { return ptr == nullptr; }
   inline bool operator!=(decltype(nullptr)) const { return ptr != nullptr; }
 
+  T& orDefault(T& defaultValue) {
+    if (ptr == nullptr) {
+      return defaultValue;
+    } else {
+      return *ptr;
+    }
+  }
+  const T& orDefault(const T& defaultValue) const {
+    if (ptr == nullptr) {
+      return defaultValue;
+    } else {
+      return *ptr;
+    }
+  }
+
   template <typename Func>
   auto map(Func&& f) -> Maybe<decltype(f(instance<T&>()))> {
     if (ptr == nullptr) {

c++/src/kj/debug.c++

@@ -222,9 +222,14 @@ String Debug::makeContextDescriptionInternal(const char* macroArgs, ArrayPtr<Str
   return makeDescription(LOG, nullptr, 0, macroArgs, argValues);
 }
 
-int Debug::getOsErrorNumber() {
+int Debug::getOsErrorNumber(bool nonblocking) {
   int result = errno;
-  return result == EINTR ? -1 : result;
+
+  // On many systems, EAGAIN and EWOULDBLOCK have the same value, but this is not strictly required
+  // by POSIX, so we need to check both.
+  return result == EINTR ? -1
+       : nonblocking && (result == EAGAIN || result == EWOULDBLOCK) ? 0
+       : result;
 }
 
 Debug::Context::Context(): logged(false) {}

c++/src/kj/debug.h

@@ -79,6 +79,11 @@
 //
 //   `KJ_SYSCALL` can be followed by a recovery block, just like `KJ_ASSERT`.
 //
+// * `KJ_NONBLOCKING_SYSCALL(code, ...)`:  Like KJ_SYSCALL, but will not throw an exception on
+//   EAGAIN/EWOULDBLOCK.  The calling code should check the syscall's return value to see if it
+//   indicates an error; in this case, it can assume the error was EAGAIN because any other error
+//   would have caused an exception to be thrown.
+//
 // * `KJ_CONTEXT(...)`:  Notes additional contextual information relevant to any exceptions thrown
 //   from within the current scope.  That is, until control exits the block in which KJ_CONTEXT()
 //   is used, if any exception is generated, it will contain the given information in its context
@@ -129,7 +134,13 @@ namespace kj {
 #define KJ_FAIL_REQUIRE(...) _kJ_FAIL_FAULT(PRECONDITION, ##__VA_ARGS__)
 
 #define KJ_SYSCALL(call, ...) \
-  if (auto _kjSyscallResult = ::kj::_::Debug::syscall([&](){return (call);})) {} else \
+  if (auto _kjSyscallResult = ::kj::_::Debug::syscall([&](){return (call);}, false)) {} else \
+    for (::kj::_::Debug::Fault f( \
+             __FILE__, __LINE__, ::kj::Exception::Nature::OS_ERROR, \
+             _kjSyscallResult.getErrorNumber(), #call, #__VA_ARGS__, ##__VA_ARGS__);; f.fatal())
+
+#define KJ_NONBLOCKING_SYSCALL(call, ...) \
+  if (auto _kjSyscallResult = ::kj::_::Debug::syscall([&](){return (call);}, true)) {} else \
     for (::kj::_::Debug::Fault f( \
              __FILE__, __LINE__, ::kj::Exception::Nature::OS_ERROR, \
              _kjSyscallResult.getErrorNumber(), #call, #__VA_ARGS__, ##__VA_ARGS__);; f.fatal())
@@ -227,7 +238,7 @@ public:
   };
 
   template <typename Call>
-  static SyscallResult syscall(Call&& call);
+  static SyscallResult syscall(Call&& call, bool nonblocking);
 
   class Context: public ExceptionCallback {
   public:
@@ -280,7 +291,7 @@ private:
                           ArrayPtr<String> argValues);
   static String makeContextDescriptionInternal(const char* macroArgs, ArrayPtr<String> argValues);
 
-  static int getOsErrorNumber();
+  static int getOsErrorNumber(bool nonblocking);
   // Get the error code of the last error (e.g. from errno).  Returns -1 on EINTR.
 };
 
@@ -303,10 +314,12 @@ Debug::Fault::Fault(const char* file, int line, Exception::Nature nature, int er
 }
 
 template <typename Call>
-Debug::SyscallResult Debug::syscall(Call&& call) {
+Debug::SyscallResult Debug::syscall(Call&& call, bool nonblocking) {
   while (call() < 0) {
-    int errorNum = getOsErrorNumber();
-    // getOsErrorNumber() returns -1 to indicate EINTR
+    int errorNum = getOsErrorNumber(nonblocking);
+    // getOsErrorNumber() returns -1 to indicate EINTR.
+    // Also, if nonblocking is true, then it returns 0 on EAGAIN, which will then be treated as a
+    // non-error.
     if (errorNum != -1) {
       return SyscallResult(errorNum);
     }

c++/src/kj/memory.h

@@ -212,6 +212,21 @@ public:
   inline bool operator==(decltype(nullptr)) const { return ptr == nullptr; }
   inline bool operator!=(decltype(nullptr)) const { return ptr != nullptr; }
 
+  Own<T>& orDefault(Own<T>& defaultValue) {
+    if (ptr == nullptr) {
+      return defaultValue;
+    } else {
+      return ptr;
+    }
+  }
+  const Own<T>& orDefault(const Own<T>& defaultValue) const {
+    if (ptr == nullptr) {
+      return defaultValue;
+    } else {
+      return ptr;
+    }
+  }
+
   template <typename Func>
   auto map(Func&& f) -> Maybe<decltype(f(instance<T&>()))> {
     if (ptr == nullptr) {