TwoWayPipe and basic asyncio support

Note: I've tried not to break any behaviour of the previously working APIs

Python API Changes / Additions
- capnp/lib/capnp.pyx
  * class _RemotePromise
    + [Added] cpdef _wait(self)
      = Exception raising code that used to be inside of wait(self)
    + [Modified] def wait(self)
      = Same functionality as before
    + [Added] async def a_wait(self)
      = Cannot use await as that's a reserved keyword
      = Uses pollRemote and asyncio.sleep(0) to make call asynchronous
  * class _TwoPartyVatNetwork
    + [Added] cdef _init_pipe(self, _TwoWayPipe pipe, Side side,
    schema_cpp.ReaderOptions opts)
      = Instanciates a TwoPartyVatNetwork using a TwoWayPipe (instead of
      using a file handle or connection as before)
  * class TwoPartyClient
    + [Modified] def __init__(self, socket=None, restorer=None,
    traversal_limit_in_words=None, nesting_limit=None)
      = Changes the socket parameter to be optional
      = If socket is not specified, default to using a TwoWayPipe
    + [Added] async def read(self, bufsize)
      = awaitable function that blocks until data has been read
      = bufsize defines the maximum amount of data to be read back
        (e.g. 4096 bytes)
      = Reads data from TwoWayPipe
    + [Added] def write(self, data)
      = Write data to TwoWayPipe
      = Not awaitable as the write interface of the TwoWayPipe doesn't
      have poll functionality
  * class TwoPartyServer
    + [Modified] def __init__(self, socket=None, restorer=None,
    server_socket=None, bootstrap=None, traversal_limit_in_words=None,
    nesting_limit=None)
      = Changes the socket parameter to be optional
      = If socket is not specified, default to using a TwoWayPipe
      = Simplified code by removing an else (self._connect)
    + [Added] async def read(self, bufsize)
      = awaitable function that blocks until data has been read
      = bufsize defines the maximum amount of data to be read back
        (e.g. 4096 bytes)
      = Reads data from TwoWayPipe
    + [Added] def write(self, data)
      = Write data to TwoWayPipe
      = Not awaitable as the write interface of the TwoWayPipe doesn't
      have poll functionality
    + [Added] async def poll_forever(self)
      = asyncio equivalent of run_forever()
  * class _TwoWayPipe
    + Wrapper class for TwoWayPipe

Other Additions
- capnp/helpers/asyncHelper.h
  * pollWaitScope
    + Pumps the kj event handler
    + Used for the TwoWayServer
  * pollRemote
    + Polls a remote promise
    + i.e. a capnp RPC call
- capnp/helpers/asyncIoHelper.h
  * AsyncIoStreamReadHelper
    + I wasn't able to figure out Promise[size_t] using Cython so this was
    the next best thing I could think of doing
    + Was needed to handle read polling from a read promise
      = Polling is used for asyncio as kj waits need a wrapper to be
      compatible
- capnp/lib/capnp.pyx
  * makeTwoWayPipe
    + Wrapper for kj newTwoWayPipe function
  * poll_once
    + Single pump of the kj event handler (used with pollWaitScope)

TwoWayClient Usage - TwoWayPipe
- See examples/async_client.py

TwoWayServer Usage - TwoWayPipe
- See examples/async_server.py

capnp/helpers/asyncIoHelper.h

Misc Changes
- Fixed thread_server.py and thread_client.py to use bootstrap instead
of ez_restore
- async_client.py and async_server.py examples
  * Uses the same thread.capnp as thread_client.py and thread_server.py
  * They are compatible, so you can mix and match client and server for
  compatibility testing
  * async_client.py and async_server.py require <address>:<port>
  formatting (unlike autodetection from thread_client.py and
  thread_server.py)

capnp/helpers/asyncHelper.h

@@ -39,6 +39,11 @@ void waitNeverDone(kj::WaitScope & scope) {
   kj::NEVER_DONE.wait(scope);
 }
 
+void pollWaitScope(kj::WaitScope & scope) {
+  GILRelease gil;
+  scope.poll();
+}
+
 kj::Timer * getTimer(kj::AsyncIoContext * context) {
   return &context->lowLevelProvider->getTimer();
 }
@@ -57,3 +62,8 @@ capnp::Response< ::capnp::DynamicStruct> * waitRemote(capnp::RemotePromise< ::ca
   GILRelease gil;
   return new capnp::Response< ::capnp::DynamicStruct>(promise->wait(scope));
 }
+
+bool pollRemote(capnp::RemotePromise< ::capnp::DynamicStruct> * promise, kj::WaitScope & scope) {
+  GILRelease gil;
+  return promise->poll(scope);
+}

capnp/helpers/asyncIoHelper.h

@@ -0,0 +1,53 @@
+#pragma once
+
+#include "kj/async.h"
+#include "kj/async-io.h"
+
+class AsyncIoStreamReadHelper {
+public:
+  AsyncIoStreamReadHelper(kj::AsyncIoStream * _stream, kj::WaitScope * _scope, size_t bufsize) {
+    io_stream = _stream;
+    wait_scope = _scope;
+    ready = false;
+    buffer_read_size = 0;
+    buffer = new unsigned char[bufsize];
+    promise = io_stream->read(buffer, 1, bufsize);
+  }
+
+  ~AsyncIoStreamReadHelper() {
+    delete[] buffer;
+  }
+
+  bool poll() {
+    bool result = promise.poll(*wait_scope);
+    if (result) {
+      ready = true;
+      buffer_read_size = promise.wait(*wait_scope);
+    }
+    return result;
+  }
+
+  size_t read_size() {
+    if (!ready) {
+      return 0;
+    }
+    return buffer_read_size;
+  }
+
+  void * read_buffer() {
+    if (!ready) {
+      return 0;
+    }
+    return buffer;
+  }
+
+private:
+  kj::AsyncIoStream * io_stream;
+  kj::WaitScope * wait_scope;
+  kj::Promise<size_t> promise = nullptr;
+
+  unsigned char *buffer;
+  size_t buffer_read_size;
+
+  bool ready;
+};

capnp/helpers/helpers.pxd

@@ -2,10 +2,12 @@ from capnp.includes.capnp_cpp cimport Maybe, DynamicStruct, Request, Response, P
 
 from capnp.includes.schema_cpp cimport ByteArray
 
-from non_circular cimport reraise_kj_exception
+from non_circular cimport reraise_kj_exception, AsyncIoStreamReadHelper
 
 from cpython.ref cimport PyObject
 
+from libcpp cimport bool
+
 cdef extern from "capnp/helpers/fixMaybe.h":
     EnumSchema.Enumerant fixMaybe(Maybe[EnumSchema.Enumerant]) except +reraise_kj_exception
     StructSchema.Field fixMaybe(Maybe[StructSchema.Field]) except +reraise_kj_exception
@@ -41,7 +43,9 @@ cdef extern from "capnp/helpers/serialize.h":
 
 cdef extern from "capnp/helpers/asyncHelper.h":
     void waitNeverDone(WaitScope&)
+    void pollWaitScope(WaitScope&)
     Response * waitRemote(RemotePromise *, WaitScope&)
+    bool pollRemote(RemotePromise *, WaitScope&)
     PyObject * waitPyPromise(PyPromise *, WaitScope&)
     void waitVoidPromise(VoidPromise *, WaitScope&)
     Timer * getTimer(AsyncIoContext *) except +reraise_kj_exception

capnp/helpers/non_circular.pxd

@@ -1,4 +1,6 @@
 from cpython.ref cimport PyObject
+from capnp.includes.capnp_cpp cimport AsyncIoStream, WaitScope
+from libcpp cimport bool
 
 cdef extern from "capnp/helpers/capabilityHelper.h":
     cppclass PythonInterfaceDynamicImpl:
@@ -18,3 +20,10 @@ cdef extern from "capnp/helpers/rpcHelper.h":
 cdef extern from "capnp/helpers/asyncHelper.h":
     cdef cppclass PyEventPort:
         PyEventPort(PyObject *)
+
+cdef extern from "capnp/helpers/asyncIoHelper.h":
+    cdef cppclass AsyncIoStreamReadHelper:
+        AsyncIoStreamReadHelper(AsyncIoStream *, WaitScope *, size_t)
+        bool poll()
+        size_t read_size()
+        void* read_buffer()

capnp/includes/capnp_cpp.pxd

@@ -4,6 +4,7 @@
 cdef extern from "capnp/helpers/checkCompiler.h":
     pass
 
+from libcpp cimport bool
 from schema_cpp cimport Node, Data, StructNode, EnumNode, InterfaceNode, MessageBuilder, MessageReader, ReaderOptions
 from capnp.helpers.non_circular cimport PythonInterfaceDynamicImpl, reraise_kj_exception, PyRefCounter, PyRestorer, PyEventPort, ErrorHandler
 from capnp.includes.types cimport *
@@ -45,6 +46,7 @@ cdef extern from "kj/exception.h" namespace " ::kj":
 cdef extern from "kj/memory.h" namespace " ::kj":
     cdef cppclass Own[T]:
         T& operator*()
+        T* get()
     Own[TwoPartyVatNetwork] makeTwoPartyVatNetwork" ::kj::heap< ::capnp::TwoPartyVatNetwork>"(AsyncIoStream& stream, Side, ReaderOptions)
     Own[PromiseFulfillerPair] copyPromiseFulfillerPair" ::kj::heap< ::kj::PromiseFulfillerPair<void> >"(PromiseFulfillerPair&)
     Own[PyRefCounter] makePyRefCounter" ::kj::heap< PyRefCounter >"(PyObject *)
@@ -55,6 +57,7 @@ cdef extern from "kj/async.h" namespace " ::kj":
         Promise(Promise)
         Promise(T)
         T wait(WaitScope)
+        bool poll(WaitScope)
         # ForkedPromise<T> fork()
         # Promise<T> exclusiveJoin(Promise<T>&& other)
         # Promise[T] eagerlyEvaluate()
@@ -121,16 +124,21 @@ cdef inline Duration Nanoseconds(int64_t nanos):
 
 cdef extern from "kj/async-io.h" namespace " ::kj":
     cdef cppclass AsyncIoStream:
-        pass
+        Promise[size_t] read(void*, size_t, size_t)
+        Promise[void] write(const void*, size_t)
+
     cdef cppclass LowLevelAsyncIoProvider:
         # Own[AsyncInputStream] wrapInputFd(int)
         # Own[AsyncOutputStream] wrapOutputFd(int)
         Own[AsyncIoStream] wrapSocketFd(int)
         Timer& getTimer() except +reraise_kj_exception
+
     cdef cppclass AsyncIoProvider:
-        pass
+        TwoWayPipe newTwoWayPipe()
+
     cdef cppclass WaitScope:
         pass
+
     cdef cppclass AsyncIoContext:
         AsyncIoContext(AsyncIoContext&)
         Own[LowLevelAsyncIoProvider] lowLevelProvider
@@ -140,6 +148,9 @@ cdef extern from "kj/async-io.h" namespace " ::kj":
     cdef cppclass TaskSet:
         TaskSet(ErrorHandler &)
 
+    cdef cppclass TwoWayPipe:
+        Own[AsyncIoStream] ends[2]
+
     AsyncIoContext setupAsyncIo()
 
 cdef extern from "capnp/schema.capnp.h" namespace " ::capnp":

capnp/lib/capnp.pxd

@@ -2,7 +2,7 @@
 
 from capnp.includes cimport capnp_cpp as capnp
 from capnp.includes cimport schema_cpp
-from capnp.includes.capnp_cpp cimport Schema as C_Schema, StructSchema as C_StructSchema, InterfaceSchema as C_InterfaceSchema, EnumSchema as C_EnumSchema, ListSchema as C_ListSchema, DynamicStruct as C_DynamicStruct, DynamicValue as C_DynamicValue, Type as C_Type, DynamicList as C_DynamicList, SchemaParser as C_SchemaParser, ParsedSchema as C_ParsedSchema, VOID, ArrayPtr, StringPtr, String, StringTree, DynamicOrphan as C_DynamicOrphan, AnyPointer as C_DynamicObject, DynamicCapability as C_DynamicCapability, Request, Response, RemotePromise, PyPromise, VoidPromise, CallContext, PyRestorer, RpcSystem, makeRpcServer, makeRpcServerBootstrap, makeRpcClient, Capability as C_Capability, TwoPartyVatNetwork as C_TwoPartyVatNetwork, Side, AsyncIoStream, Own, makeTwoPartyVatNetwork, PromiseFulfillerPair as C_PromiseFulfillerPair, copyPromiseFulfillerPair, newPromiseAndFulfiller, PyArray, DynamicStruct_Builder
+from capnp.includes.capnp_cpp cimport Schema as C_Schema, StructSchema as C_StructSchema, InterfaceSchema as C_InterfaceSchema, EnumSchema as C_EnumSchema, ListSchema as C_ListSchema, DynamicStruct as C_DynamicStruct, DynamicValue as C_DynamicValue, Type as C_Type, DynamicList as C_DynamicList, SchemaParser as C_SchemaParser, ParsedSchema as C_ParsedSchema, VOID, ArrayPtr, StringPtr, String, StringTree, DynamicOrphan as C_DynamicOrphan, AnyPointer as C_DynamicObject, DynamicCapability as C_DynamicCapability, Request, Response, RemotePromise, PyPromise, VoidPromise, CallContext, PyRestorer, RpcSystem, makeRpcServer, makeRpcServerBootstrap, makeRpcClient, Capability as C_Capability, TwoPartyVatNetwork as C_TwoPartyVatNetwork, Side, AsyncIoStream, Own, makeTwoPartyVatNetwork, PromiseFulfillerPair as C_PromiseFulfillerPair, copyPromiseFulfillerPair, newPromiseAndFulfiller, PyArray, DynamicStruct_Builder, TwoWayPipe
 from capnp.includes.schema_cpp cimport Node as C_Node, EnumNode as C_EnumNode
 from capnp.includes.types cimport *
 from capnp.helpers.non_circular cimport reraise_kj_exception

capnp/lib/capnp.pyx

@@ -11,6 +11,8 @@
 cimport cython
 
 from capnp.helpers.helpers cimport makeRpcClientWithRestorer
+from capnp.helpers.helpers cimport AsyncIoStreamReadHelper
+from capnp.includes.capnp_cpp cimport AsyncIoStream, WaitScope
 
 from libc.stdlib cimport malloc, free
 from libc.string cimport memcpy
@@ -33,6 +35,7 @@ import socket as _socket
 import random as _random
 import collections as _collections
 import array
+import asyncio
 
 _CAPNP_VERSION_MAJOR = capnp.CAPNP_VERSION_MAJOR
 _CAPNP_VERSION_MINOR = capnp.CAPNP_VERSION_MINOR
@@ -1655,6 +1658,9 @@ cdef class _EventLoop:
         del self.thisptr
         self.thisptr = NULL
 
+    cdef TwoWayPipe makeTwoWayPipe(self):
+        return deref(deref(self.thisptr).provider).newTwoWayPipe()
+
     cdef Own[AsyncIoStream] wrapSocketFd(self, int fd):
         return deref(deref(self.thisptr).lowLevelProvider).wrapSocketFd(fd)
 
@@ -1738,6 +1744,10 @@ def wait_forever():
     cdef _EventLoop loop = C_DEFAULT_EVENT_LOOP_GETTER()
     helpers.waitNeverDone(deref(loop.thisptr).waitScope)
 
+def poll_once():
+    cdef _EventLoop loop = C_DEFAULT_EVENT_LOOP_GETTER()
+    helpers.pollWaitScope(deref(loop.thisptr).waitScope)
+
 cdef class _CallContext:
     cdef CallContext * thisptr
 
@@ -1929,11 +1939,24 @@ cdef class _RemotePromise:
     def __dealloc__(self):
         del self.thisptr
 
-    cpdef wait(self) except +reraise_kj_exception:
+    cpdef _wait(self) except +reraise_kj_exception:
+        return _Response()._init_childptr(helpers.waitRemote(self.thisptr, deref(self._event_loop.thisptr).waitScope), self._parent)
+
+    def wait(self):
         if self.is_consumed:
             raise KjException('Promise was already used in a consuming operation. You can no longer use this Promise object')
 
-        ret = _Response()._init_childptr(helpers.waitRemote(self.thisptr, deref(self._event_loop.thisptr).waitScope), self._parent)
+        ret = self._wait()
+        self.is_consumed = True
+        return ret
+
+    async def a_wait(self):
+        if self.is_consumed:
+            raise KjException('Promise was already used in a consuming operation. You can no longer use this Promise object')
+
+        while not helpers.pollRemote(self.thisptr, deref(self._event_loop.thisptr).waitScope):
+            await asyncio.sleep(0)
+        ret = self._wait()
         self.is_consumed = True
 
         return ret
@@ -2234,6 +2257,10 @@ cdef class _TwoPartyVatNetwork:
         self.thisptr = makeTwoPartyVatNetwork(deref(stream.thisptr), side, opts)
         return self
 
+    cdef _init_pipe(self, _TwoWayPipe pipe, Side side, schema_cpp.ReaderOptions opts):
+        self.thisptr = makeTwoPartyVatNetwork(deref(pipe._pipe.ends[0]), side, opts)
+        return self
+
     cpdef on_disconnect(self) except +reraise_kj_exception:
         return _VoidPromise()._init(deref(self.thisptr).onDisconnect(), self)
 
@@ -2255,16 +2282,23 @@ cdef class TwoPartyClient:
     cdef public object _orig_stream
     cdef public _Restorer _restorer
     cdef public _AsyncIoStream _stream
+    cdef public _TwoWayPipe _pipe
 
-    def __init__(self, socket, restorer=None, traversal_limit_in_words=None, nesting_limit=None):
+    def __init__(self, socket=None, restorer=None, traversal_limit_in_words=None, nesting_limit=None):
         if isinstance(socket, basestring):
             socket = self._connect(socket)
 
         cdef schema_cpp.ReaderOptions opts = make_reader_opts(traversal_limit_in_words, nesting_limit)
 
         self._orig_stream = socket
-        self._stream = _FdAsyncIoStream(socket.fileno())
-        self._network = _TwoPartyVatNetwork()._init(self._stream, capnp.CLIENT, opts)
+        if self._orig_stream:
+            self._stream = _FdAsyncIoStream(socket.fileno())
+            self._network = _TwoPartyVatNetwork()._init(self._stream, capnp.CLIENT, opts)
+        else:
+            # Initialize TwoWayPipe, to use pipe() acquire other end of the pipe using read() and write() methods
+            self._pipe = _TwoWayPipe()
+            self._network = _TwoPartyVatNetwork()._init_pipe(self._pipe, capnp.CLIENT, opts)
+
         if restorer is None:
             self.thisptr = new RpcSystem(makeRpcClient(deref(self._network.thisptr)))
             self._restorer = None
@@ -2274,10 +2308,35 @@ cdef class TwoPartyClient:
             self.thisptr = new RpcSystem(makeRpcClientWithRestorer(deref(self._network.thisptr), deref(self._restorer.thisptr)))
 
         Py_INCREF(self._restorer)
-        Py_INCREF(self._orig_stream)
-        Py_INCREF(self._stream)
+        if self._orig_stream:
+            Py_INCREF(self._orig_stream)
+            Py_INCREF(self._stream)
+        else:
+            Py_INCREF(self._pipe)
         Py_INCREF(self._network) # TODO:MEMORY: attach this to onDrained, also figure out what's leaking
 
+    async def read(self, bufsize):
+        cdef AsyncIoStreamReadHelper *reader = new AsyncIoStreamReadHelper(
+            self._pipe._pipe.ends[1].get(),
+            &self._pipe._event_loop.thisptr.waitScope,
+            bufsize
+        )
+        while not reader.poll():
+            await asyncio.sleep(0)
+
+        cdef array.array read_buffer = array.array('b', [])
+        array.resize(read_buffer, reader.read_size())
+        memcpy(read_buffer.data.as_voidptr, reader.read_buffer(), reader.read_size())
+        del reader
+        return read_buffer
+
+    def write(self, data):
+        cdef array.array write_buffer = array.array('b', data)
+        deref(self._pipe._pipe.ends[1]).write(
+            write_buffer.data.as_voidptr,
+            len(data)
+        ).wait(self._pipe._event_loop.thisptr.waitScope)
+
     def __dealloc__(self):
         del self.thisptr
 
@@ -2349,12 +2408,13 @@ cdef class TwoPartyServer:
     cdef public object _orig_stream, _server_socket, _disconnect_promise
     cdef public _Restorer _restorer
     cdef public _AsyncIoStream _stream
+    cdef public _TwoWayPipe _pipe
     cdef object _port
     cdef public object port_promise, _bootstrap
     cdef capnp.TaskSet * _task_set
     cdef capnp.ErrorHandler _error_handler
 
-    def __init__(self, socket, restorer=None, server_socket=None, bootstrap=None,
+    def __init__(self, socket=None, restorer=None, server_socket=None, bootstrap=None,
                  traversal_limit_in_words=None, nesting_limit=None):
         if not restorer and not bootstrap:
             raise KjException("You must provide either a bootstrap interface or a restorer (deperecated) to a server constructor.")
@@ -2366,28 +2426,58 @@ cdef class TwoPartyServer:
 
         if isinstance(socket, basestring):
             self._connect(socket, restorer, bootstrap)
-        else:
-            self._orig_stream = socket
+            return
+
+        self._orig_stream = socket
+        if self._orig_stream:
             self._stream = _FdAsyncIoStream(socket.fileno())
-            self._server_socket = server_socket
-            self._port = 0
             self._network = _TwoPartyVatNetwork()._init(self._stream, capnp.SERVER, opts)
+        else:
+            # Initialize TwoWayPipe, to use pipe() acquire other end of the pipe using read() and write() methods
+            self._pipe = _TwoWayPipe()
+            self._network = _TwoPartyVatNetwork()._init_pipe(self._pipe, capnp.SERVER, opts)
 
-            if bootstrap:
-                self._bootstrap = bootstrap
-                schema = bootstrap.schema
-                self.thisptr = new RpcSystem(makeRpcServerBootstrap(deref(self._network.thisptr), helpers.server_to_client(schema.thisptr, <PyObject *>bootstrap)))
-            elif restorer:
-                _warnings.warn('Restorers are deprecated. Please use the new bootstrap methods.', UserWarning)
-                self._restorer = _convert_restorer(restorer)
-                self.thisptr = new RpcSystem(makeRpcServer(deref(self._network.thisptr), deref(self._restorer.thisptr)))
+        self._server_socket = server_socket
+        self._port = 0
 
-            Py_INCREF(self._orig_stream)
-            Py_INCREF(self._stream)
-            Py_INCREF(self._restorer)
-            Py_INCREF(self._bootstrap)
-            Py_INCREF(self._network)
-            self._disconnect_promise = self.on_disconnect().then(self._decref)
+        if bootstrap:
+            self._bootstrap = bootstrap
+            schema = bootstrap.schema
+            self.thisptr = new RpcSystem(makeRpcServerBootstrap(deref(self._network.thisptr), helpers.server_to_client(schema.thisptr, <PyObject *>bootstrap)))
+        elif restorer:
+            _warnings.warn('Restorers are deprecated. Please use the new bootstrap methods.', UserWarning)
+            self._restorer = _convert_restorer(restorer)
+            self.thisptr = new RpcSystem(makeRpcServer(deref(self._network.thisptr), deref(self._restorer.thisptr)))
+
+        Py_INCREF(self._restorer)
+        Py_INCREF(self._orig_stream)
+        Py_INCREF(self._stream)
+        Py_INCREF(self._pipe)
+        Py_INCREF(self._bootstrap)
+        Py_INCREF(self._network)
+        self._disconnect_promise = self.on_disconnect().then(self._decref)
+
+    async def read(self, bufsize):
+        cdef AsyncIoStreamReadHelper *reader = new AsyncIoStreamReadHelper(
+            self._pipe._pipe.ends[1].get(),
+            &self._pipe._event_loop.thisptr.waitScope,
+            bufsize
+        )
+        while not reader.poll():
+            await asyncio.sleep(0)
+
+        cdef array.array read_buffer = array.array('b', [])
+        array.resize(read_buffer, reader.read_size())
+        memcpy(read_buffer.data.as_voidptr, reader.read_buffer(), reader.read_size())
+        del reader
+        return read_buffer
+
+    async def write(self, data):
+        cdef array.array write_buffer = array.array('b', data)
+        deref(self._pipe._pipe.ends[1]).write(
+            write_buffer.data.as_voidptr,
+            len(data)
+        ).wait(self._pipe._event_loop.thisptr.waitScope)
 
     cpdef _connect(self, host_string, restorer, bootstrap):
         cdef _InterfaceSchema schema
@@ -2406,6 +2496,7 @@ cdef class TwoPartyServer:
     def _decref(self):
         Py_DECREF(self._bootstrap)
         Py_DECREF(self._restorer)
+        Py_INCREF(self._pipe)
         Py_DECREF(self._orig_stream)
         Py_DECREF(self._stream)
         Py_DECREF(self._network)
@@ -2417,6 +2508,11 @@ cdef class TwoPartyServer:
     cpdef on_disconnect(self) except +reraise_kj_exception:
         return _VoidPromise()._init(deref(self._network.thisptr).onDisconnect())
 
+    async def poll_forever(self):
+        while True:
+            poll_once()
+            await asyncio.sleep(0)
+
     cpdef run_forever(self):
         if self.port_promise is None:
             raise KjException("You must pass a string as the socket parameter in __init__ to use this function")
@@ -2437,6 +2533,18 @@ cdef class TwoPartyServer:
 cdef class _AsyncIoStream:
     cdef Own[AsyncIoStream] thisptr
 
+cdef class _TwoWayPipe:
+    cdef _EventLoop _event_loop
+    cdef TwoWayPipe _pipe
+
+    def __init__(self):
+        self._init()
+
+    cpdef _init(self) except +reraise_kj_exception:
+        self._event_loop = C_DEFAULT_EVENT_LOOP_GETTER()
+        # Create two way pipe using AsyncIoContext
+        self._pipe = self._event_loop.makeTwoWayPipe()
+
 cdef class _FdAsyncIoStream(_AsyncIoStream):
     cdef _EventLoop _event_loop