From 8915ef79f14d72171bec7dc8dbbc9a8ab6b402b3 Mon Sep 17 00:00:00 2001 From: Jacob Alexander Date: Mon, 16 Sep 2019 21:34:29 -0700 Subject: [PATCH] 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
: formatting (unlike autodetection from thread_client.py and thread_server.py) --- capnp/helpers/asyncHelper.h | 10 +++ capnp/helpers/asyncIoHelper.h | 53 +++++++++++ capnp/helpers/helpers.pxd | 6 +- capnp/helpers/non_circular.pxd | 9 ++ capnp/includes/capnp_cpp.pxd | 15 +++- capnp/lib/capnp.pxd | 2 +- capnp/lib/capnp.pyx | 160 +++++++++++++++++++++++++++------ examples/async_client.py | 92 +++++++++++++++++++ examples/async_server.py | 87 ++++++++++++++++++ examples/thread_client.py | 4 +- examples/thread_server.py | 10 +-- 11 files changed, 407 insertions(+), 41 deletions(-) create mode 100644 capnp/helpers/asyncIoHelper.h create mode 100755 examples/async_client.py create mode 100755 examples/async_server.py diff --git a/capnp/helpers/asyncHelper.h b/capnp/helpers/asyncHelper.h index 8b4e1f1..4d74e43 100644 --- a/capnp/helpers/asyncHelper.h +++ b/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); +} diff --git a/capnp/helpers/asyncIoHelper.h b/capnp/helpers/asyncIoHelper.h new file mode 100644 index 0000000..c848ebb --- /dev/null +++ b/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 promise = nullptr; + + unsigned char *buffer; + size_t buffer_read_size; + + bool ready; +}; diff --git a/capnp/helpers/helpers.pxd b/capnp/helpers/helpers.pxd index 0d9884e..6e506f8 100644 --- a/capnp/helpers/helpers.pxd +++ b/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 diff --git a/capnp/helpers/non_circular.pxd b/capnp/helpers/non_circular.pxd index 58771dc..b0a9ffd 100644 --- a/capnp/helpers/non_circular.pxd +++ b/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() diff --git a/capnp/includes/capnp_cpp.pxd b/capnp/includes/capnp_cpp.pxd index f43956e..450a05e 100644 --- a/capnp/includes/capnp_cpp.pxd +++ b/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 >"(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 fork() # Promise exclusiveJoin(Promise&& 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": diff --git a/capnp/lib/capnp.pxd b/capnp/lib/capnp.pxd index e8dffa5..4d9c7e1 100644 --- a/capnp/lib/capnp.pxd +++ b/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 diff --git a/capnp/lib/capnp.pyx b/capnp/lib/capnp.pyx index d11c1e0..e5d420e 100644 --- a/capnp/lib/capnp.pyx +++ b/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, 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, 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 diff --git a/examples/async_client.py b/examples/async_client.py new file mode 100755 index 0000000..df70f27 --- /dev/null +++ b/examples/async_client.py @@ -0,0 +1,92 @@ +#!/usr/bin/env python + +from __future__ import print_function + +import asyncio +import argparse +import threading +import time +import capnp +import socket + +import thread_capnp + +capnp.remove_event_loop() +capnp.create_event_loop(threaded=True) + + +def parse_args(): + parser = argparse.ArgumentParser(usage='Connects to the Example thread server \ +at the given address and does some RPCs') + parser.add_argument("host", help="HOST:PORT") + + return parser.parse_args() + + +class StatusSubscriber(thread_capnp.Example.StatusSubscriber.Server): + + '''An implementation of the StatusSubscriber interface''' + + def status(self, value, **kwargs): + print('status: {}'.format(time.time())) + + +async def myreader(client, reader): + while True: + data = await reader.read(4096) + client.write(data) + + +async def mywriter(client, writer): + while True: + data = await client.read(4096) + writer.write(data.tobytes()) + await writer.drain() + + +async def background(cap): + subscriber = StatusSubscriber() + promise = cap.subscribeStatus(subscriber) + await promise.a_wait() + + +async def main(host): + host = host.split(':') + addr = host[0] + port = host[1] + # Handle both IPv4 and IPv6 cases + try: + print("Try IPv4") + reader, writer = await asyncio.open_connection( + addr, port, + ) + except: + print("Try IPv6") + reader, writer = await asyncio.open_connection( + addr, port, + family=socket.AF_INET6 + ) + + # Start TwoPartyClient using TwoWayPipe (takes no arguments in this mode) + client = capnp.TwoPartyClient() + cap = client.bootstrap().cast_as(thread_capnp.Example) + + # Assemble reader and writer tasks, run in the background + coroutines = [myreader(client, reader), mywriter(client, writer)] + asyncio.gather(*coroutines, return_exceptions=True) + + # Start background task for subscriber + tasks = [background(cap)] + asyncio.gather(*tasks, return_exceptions=True) + + # Run blocking tasks + print('main: {}'.format(time.time())) + await cap.longRunning().a_wait() + print('main: {}'.format(time.time())) + await cap.longRunning().a_wait() + print('main: {}'.format(time.time())) + await cap.longRunning().a_wait() + print('main: {}'.format(time.time())) + +if __name__ == '__main__': + asyncio.run(main(parse_args().host)) diff --git a/examples/async_server.py b/examples/async_server.py new file mode 100755 index 0000000..1c6d81e --- /dev/null +++ b/examples/async_server.py @@ -0,0 +1,87 @@ +#!/usr/bin/env python + +from __future__ import print_function + +import argparse +import capnp + +import thread_capnp +import asyncio +import socket + + +class ExampleImpl(thread_capnp.Example.Server): + + "Implementation of the Example threading Cap'n Proto interface." + + def subscribeStatus(self, subscriber, **kwargs): + return capnp.getTimer().after_delay(10**9) \ + .then(lambda: subscriber.status(True)) \ + .then(lambda _: self.subscribeStatus(subscriber)) + + def longRunning(self, **kwargs): + return capnp.getTimer().after_delay(3 * 10**9) + + +async def myreader(server, reader): + while True: + data = await reader.read(4096) + # Close connection if 0 bytes read + if len(data) == 0: + server.close() + await server.write(data) + + +async def mywriter(server, writer): + while True: + data = await server.read(4096) + writer.write(data.tobytes()) + await writer.drain() + + +async def myserver(reader, writer): + # Start TwoPartyServer using TwoWayPipe (only requires bootstrap) + server = capnp.TwoPartyServer(bootstrap=ExampleImpl()) + + # Assemble reader and writer tasks, run in the background + coroutines = [myreader(server, reader), mywriter(server, writer)] + asyncio.gather(*coroutines, return_exceptions=True) + + await server.poll_forever() + + +def parse_args(): + parser = argparse.ArgumentParser(usage='''Runs the server bound to the\ +given address/port ADDRESS. ''') + + parser.add_argument("address", help="ADDRESS:PORT") + + return parser.parse_args() + + +async def main(): + address = parse_args().address + host = address.split(':') + addr = host[0] + port = host[1] + + # Handle both IPv4 and IPv6 cases + try: + print("Try IPv4") + server = await asyncio.start_server( + myserver, + addr, port, + ) + except: + print("Try IPv6") + server = await asyncio.start_server( + myserver, + addr, port, + family=socket.AF_INET6 + ) + + async with server: + await server.serve_forever() + +if __name__ == '__main__': + asyncio.run(main()) diff --git a/examples/thread_client.py b/examples/thread_client.py index 7201dd0..459550b 100755 --- a/examples/thread_client.py +++ b/examples/thread_client.py @@ -31,7 +31,7 @@ class StatusSubscriber(thread_capnp.Example.StatusSubscriber.Server): def start_status_thread(host): client = capnp.TwoPartyClient(host) - cap = client.ez_restore('example').cast_as(thread_capnp.Example) + cap = client.bootstrap().cast_as(thread_capnp.Example) subscriber = StatusSubscriber() promise = cap.subscribeStatus(subscriber) @@ -40,7 +40,7 @@ def start_status_thread(host): def main(host): client = capnp.TwoPartyClient(host) - cap = client.ez_restore('example').cast_as(thread_capnp.Example) + cap = client.bootstrap().cast_as(thread_capnp.Example) status_thread = threading.Thread(target=start_status_thread, args=(host,)) status_thread.daemon = True diff --git a/examples/thread_server.py b/examples/thread_server.py index 04b1e2d..a61a767 100755 --- a/examples/thread_server.py +++ b/examples/thread_server.py @@ -31,18 +31,10 @@ given address/port ADDRESS may be '*' to bind to all local addresses.\ return parser.parse_args() -impl = ExampleImpl() - - -def restore(ref): - assert ref.as_text() == 'example' - return impl - - def main(): address = parse_args().address - server = capnp.TwoPartyServer(address, restore) + server = capnp.TwoPartyServer(address, bootstrap=ExampleImpl()) server.run_forever() if __name__ == '__main__':