proxy, refactor: Change ConnectStream and ServeStream to accept stream objects

Instead of accepting raw file descriptor integers and wrapping them
internally, ConnectStream and ServeStream now accept
kj::Own<kj::AsyncIoStream> directly. This removes the assumption that
the transport is always a local unix fd, making the API easier to adapt
to other I/O types (e.g. Windows handles).

The Stream type alias (kj::Own<kj::AsyncIoStream>) is added as a
convenience, along with StreamSocketId() to extract the underlying fd
from a Stream when needed.

Callers are updated to wrap their fd with wrapSocketFd() before calling.

Author: ryanofsky

example/calculator.cpp

@@ -11,6 +11,7 @@
 #include <functional>
 #include <iostream>
 #include <kj/async.h>
+#include <kj/async-io.h>
 #include <kj/common.h>
 #include <kj/memory.h>
 #include <memory>
@@ -53,7 +54,8 @@ int main(int argc, char** argv)
     mp::SocketId socket{mp::StartSpawned(argv[1])};
     mp::EventLoop loop("mpcalculator", LogPrint);
     std::unique_ptr<Init> init = std::make_unique<InitImpl>();
-    mp::ServeStream<InitInterface>(loop, socket, *init);
+    mp::Stream stream{loop.m_io_context.lowLevelProvider->wrapSocketFd(socket, kj::LowLevelAsyncIoProvider::TAKE_OWNERSHIP)};
+    mp::ServeStream<InitInterface>(loop, kj::mv(stream), *init);
     loop.loop();
     return 0;
 }

example/example.cpp

@@ -11,7 +11,9 @@
 #include <future>
 #include <iostream>
 #include <kj/async.h>
+#include <kj/async-io.h>
 #include <kj/common.h>
+#include <kj/memory.h>
 #include <memory>
 #include <mp/proxy-io.h>
 #include <mp/util.h>
@@ -32,7 +34,7 @@ static auto Spawn(mp::EventLoop& loop, const std::string& process_argv0, const s
         path.append(new_exe_name);
         return {path.string(), std::move(info)};
     });
-    return std::make_tuple(mp::ConnectStream<InitInterface>(loop, socket), pid);
+    return std::make_tuple(mp::ConnectStream<InitInterface>(loop, loop.m_io_context.lowLevelProvider->wrapSocketFd(socket)), pid);
 }
 
 static void LogPrint(mp::LogMessage log_data)

example/printer.cpp

@@ -11,6 +11,7 @@
 #include <fstream>
 #include <iostream>
 #include <kj/async.h>
+#include <kj/async-io.h>
 #include <kj/common.h>
 #include <kj/memory.h>
 #include <memory>
@@ -47,7 +48,8 @@ int main(int argc, char** argv)
     mp::SocketId socket{mp::StartSpawned(argv[1])};
     mp::EventLoop loop("mpprinter", LogPrint);
     std::unique_ptr<Init> init = std::make_unique<InitImpl>();
-    mp::ServeStream<InitInterface>(loop, socket, *init);
+    mp::Stream stream{loop.m_io_context.lowLevelProvider->wrapSocketFd(socket, kj::LowLevelAsyncIoProvider::TAKE_OWNERSHIP)};
+    mp::ServeStream<InitInterface>(loop, std::move(stream), *init);
     loop.loop();
     return 0;
 }

include/mp/proxy-io.h

@@ -211,6 +211,12 @@ class Logger
 
 std::string LongThreadName(const char* exe_name);
 
+inline SocketId StreamSocketId(const Stream& stream)
+{
+    if (stream) KJ_IF_MAYBE(fd, stream->getFd()) return *fd;
+    throw std::logic_error("Stream socket unset");
+}
+
 //! Event loop implementation.
 //!
 //! Cap'n Proto threading model is very simple: all I/O operations are
@@ -795,17 +801,15 @@ kj::Promise<T> ProxyServer<Thread>::post(Fn&& fn)
     return ret;
 }
 
-//! Given stream file descriptor, make a new ProxyClient object to send requests
-//! over the stream. Also create a new Connection object embedded in the
-//! client that is freed when the client is closed.
+//! Given a stream, make a new ProxyClient object to send requests over it.
+//! Also create a new Connection object embedded in the client that is freed
+//! when the client is closed.
 template <typename InitInterface>
-std::unique_ptr<ProxyClient<InitInterface>> ConnectStream(EventLoop& loop, int fd)
+std::unique_ptr<ProxyClient<InitInterface>> ConnectStream(EventLoop& loop, kj::Own<kj::AsyncIoStream> stream)
 {
     typename InitInterface::Client init_client(nullptr);
     std::unique_ptr<Connection> connection;
     loop.sync([&] {
-        auto stream =
-            loop.m_io_context.lowLevelProvider->wrapSocketFd(fd, kj::LowLevelAsyncIoProvider::TAKE_OWNERSHIP);
         connection = std::make_unique<Connection>(loop, kj::mv(stream));
         init_client = connection->m_rpc_system->bootstrap(ServerVatId().vat_id).castAs<InitInterface>();
         Connection* connection_ptr = connection.get();
@@ -853,13 +857,12 @@ void _Listen(EventLoop& loop, kj::Own<kj::ConnectionReceiver>&& listener, InitIm
         }));
 }
 
-//! Given stream file descriptor and an init object, handle requests on the
-//! stream by calling methods on the Init object.
+//! Given a stream and an init object, handle requests on the stream by calling
+//! methods on the Init object.
 template <typename InitInterface, typename InitImpl>
-void ServeStream(EventLoop& loop, int fd, InitImpl& init)
+void ServeStream(EventLoop& loop, kj::Own<kj::AsyncIoStream> stream, InitImpl& init)
 {
-    _Serve<InitInterface>(
-        loop, loop.m_io_context.lowLevelProvider->wrapSocketFd(fd, kj::LowLevelAsyncIoProvider::TAKE_OWNERSHIP), init);
+    _Serve<InitInterface>(loop, kj::mv(stream), init);
 }
 
 //! Given listening socket file descriptor and an init object, handle incoming

include/mp/util.h

@@ -12,6 +12,8 @@
 #include <cstring>
 #include <exception>
 #include <functional>
+#include <kj/async-io.h>
+#include <kj/memory.h>
 #include <kj/string-tree.h>
 #include <mutex>
 #include <string>
@@ -250,6 +252,8 @@ std::string ThreadName(const char* exe_name);
 //! errors in python unit tests.
 std::string LogEscape(const kj::StringTree& string, size_t max_size);
 
+using Stream = kj::Own<kj::AsyncIoStream>;
+
 using ProcessId = int;
 using SocketId = int;
 constexpr SocketId SocketError{-1};