diff --git a/capnp/async_cpp.pxd b/capnp/async_cpp.pxd deleted file mode 100644 index 21d3a8b..0000000 --- a/capnp/async_cpp.pxd +++ /dev/null @@ -1,18 +0,0 @@ -# schema.capnp.cpp.pyx -# distutils: language = c++ -# distutils: extra_compile_args = --std=c++11 - -from cpython.ref cimport PyObject - -cdef extern from "kj/exception.h" namespace " ::kj": - cdef cppclass Exception: - pass - -cdef extern from "kj/async.h" namespace " ::kj": - cdef cppclass Promise[T]: - Promise() - Promise(Promise) - T wait() - -ctypedef Promise[PyObject *] PyPromise -ctypedef Promise[void] VoidPromise diff --git a/capnp/capnp.pyx b/capnp/capnp.pyx index f8f5d69..ec6e379 100644 --- a/capnp/capnp.pyx +++ b/capnp/capnp.pyx @@ -9,11 +9,10 @@ cimport cython cimport capnp_cpp as capnp cimport schema_cpp -from capnp_cpp cimport Schema as C_Schema, StructSchema as C_StructSchema, InterfaceSchema as C_InterfaceSchema, DynamicStruct as C_DynamicStruct, DynamicValue as C_DynamicValue, Type as C_Type, DynamicList as C_DynamicList, fixMaybe, getEnumString, SchemaParser as C_SchemaParser, ParsedSchema as C_ParsedSchema, VOID, ArrayPtr, StringPtr, String, StringTree, DynamicOrphan as C_DynamicOrphan, ObjectPointer as C_DynamicObject, DynamicCapability as C_DynamicCapability, new_client, new_server, server_to_client, Request, Response, RemotePromise, convert_to_pypromise, UnixEventLoop, PyPromise, VoidPromise, CallContext, PyRestorer, RpcSystem, makeRpcServer, makeRpcClient, makeRpcClientWithRestorer, restoreHelper, Capability as C_Capability, TwoPartyVatNetwork as C_TwoPartyVatNetwork, Side, AsyncIoStream, Own, makeTwoPartyVatNetwork, PromiseFulfillerPair as C_PromiseFulfillerPair, copyPromiseFulfillerPair, newPromiseAndFulfiller, reraise_kj_exception +from capnp_cpp cimport Schema as C_Schema, StructSchema as C_StructSchema, InterfaceSchema as C_InterfaceSchema, DynamicStruct as C_DynamicStruct, DynamicValue as C_DynamicValue, Type as C_Type, DynamicList as C_DynamicList, fixMaybe, getEnumString, SchemaParser as C_SchemaParser, ParsedSchema as C_ParsedSchema, VOID, ArrayPtr, StringPtr, String, StringTree, DynamicOrphan as C_DynamicOrphan, ObjectPointer as C_DynamicObject, DynamicCapability as C_DynamicCapability, new_client, new_server, server_to_client, Request, Response, RemotePromise, convert_to_pypromise, PyPromise, VoidPromise, CallContext, PyRestorer, RpcSystem, makeRpcServer, makeRpcClient, makeRpcClientWithRestorer, restoreHelper, Capability as C_Capability, TwoPartyVatNetwork as C_TwoPartyVatNetwork, Side, AsyncIoStream, Own, makeTwoPartyVatNetwork, PromiseFulfillerPair as C_PromiseFulfillerPair, copyPromiseFulfillerPair, newPromiseAndFulfiller, reraise_kj_exception from schema_cpp cimport Node as C_Node, EnumNode as C_EnumNode from cython.operator cimport dereference as deref -cimport async_cpp from cpython.ref cimport PyObject, Py_INCREF, Py_DECREF from cpython.exc cimport PyErr_Clear @@ -90,7 +89,7 @@ cdef public VoidPromise * call_server_method(PyObject * _server, char * _method_ func = getattr(server, method_name) # will raise if no function found params = context.params params_dict = {name : getattr(params, name) for name in params.schema.fieldnames} - params_dict['_results'] = context.results + params_dict['_context'] = context ret = func(**params_dict) if ret is not None: @@ -287,6 +286,7 @@ cdef extern from "" namespace "std": CallContext moveCallContext"std::move"(CallContext) Own[AsyncIoStream] moveOwnAsyncIOStream"std::move"(Own[AsyncIoStream]) capnp.Exception moveException"std::move"(capnp.Exception) + capnp.AsyncIoContext moveAsyncContext"std::move"(capnp.AsyncIoContext) cdef extern from "" namespace " ::capnp": StringTree printStructReader" ::capnp::prettyPrint"(C_DynamicStruct.Reader) @@ -1164,16 +1164,16 @@ cdef class _DynamicObjectBuilder: return _DynamicStructBuilder()._init(self.thisptr.getAs(s.thisptr), self._parent) cdef class _EventLoop: - cdef Own[capnp.AsyncIoProvider] thisptr + cdef capnp.AsyncIoContext * thisptr def __init__(self): self._init() cdef _init(self) except +reraise_kj_exception: - self.thisptr = capnp.setupIoEventLoop() + self.thisptr = new capnp.AsyncIoContext(moveAsyncContext(capnp.setupAsyncIo())) cdef Own[AsyncIoStream] wrapSocketFd(self, int fd): - return deref(self.thisptr).wrapSocketFd(fd) + return deref(self.thisptr.lowLevelProvider).wrapSocketFd(fd) # def __dealloc__(self): # self.remove() @@ -1236,6 +1236,18 @@ cdef class _CallContext: def __get__(self): return self._get_results() + cpdef release_params(self): + self.thisptr.releaseParams() + + cpdef allow_async_cancellation(self): + self.thisptr.allowAsyncCancellation() + + cpdef is_canceled(self): + return self.thisptr.isCanceled() + + cpdef tail_call(self, _Request tailRequest): + return _VoidPromise()._init(self.thisptr.tailCall(moveRequest(deref(tailRequest.thisptr_child)))) + cdef class _Promise: cdef PyPromise * thisptr cdef public bint is_consumed @@ -1255,7 +1267,7 @@ cdef class _Promise: if self.is_consumed: raise ValueError('Promise was already used in a consuming operation. You can no longer use this Promise object') - ret = self.thisptr.wait() # TODO: make sure refcount is fine here... + ret = self.thisptr.wait(C_DEFAULT_EVENT_LOOP.thisptr.waitScope) # TODO: make sure refcount is fine here... self.is_consumed = True return ret @@ -1288,7 +1300,7 @@ cdef class _VoidPromise: if self.is_consumed: raise ValueError('Promise was already used in a consuming operation. You can no longer use this Promise object') - self.thisptr.wait() + self.thisptr.wait(C_DEFAULT_EVENT_LOOP.thisptr.waitScope) self.is_consumed = True @@ -1322,7 +1334,7 @@ cdef class _RemotePromise: if self.is_consumed: raise ValueError('Promise was already used in a consuming operation. You can no longer use this Promise object') - ret = _Response()._init_child(self.thisptr.wait(), self._parent) + ret = _Response()._init_child(self.thisptr.wait(C_DEFAULT_EVENT_LOOP.thisptr.waitScope), self._parent) self.is_consumed = True return ret @@ -1435,9 +1447,7 @@ cdef class _DynamicCapabilityClient: return _find_field_order(params.struct) - cpdef _send_helper(self, name, firstSegmentWordSize, args, kwargs) except +reraise_kj_exception: - cdef Request * request = new Request(self.thisptr.newRequest(name, firstSegmentWordSize)) - + cdef _set_fields(self, Request * request, name, args, kwargs): if args is not None: arg_names = self._find_method_args(name) if len(args) > len(arg_names): @@ -1445,16 +1455,26 @@ cdef class _DynamicCapabilityClient: for arg_name, arg_val in zip(arg_names, args): _setDynamicFieldPtr(request, arg_name, arg_val, self) - for key, val in kwargs.items(): - _setDynamicFieldPtr(request, key, val, self) + if kwargs is not None: + for key, val in kwargs.items(): + _setDynamicFieldPtr(request, key, val, self) + + cpdef _send_helper(self, name, firstSegmentWordSize, args, kwargs) except +reraise_kj_exception: + cdef Request * request = new Request(self.thisptr.newRequest(name, firstSegmentWordSize)) + + self._set_fields(request, name, args, kwargs) return _RemotePromise()._init(request.send(), self) - cpdef _request_helper(self, name, firstSegmentWordSize=0) except +reraise_kj_exception: - return _Request()._init_child(self.thisptr.newRequest(name, firstSegmentWordSize), self) + cpdef _request_helper(self, name, firstSegmentWordSize, args, kwargs) except +reraise_kj_exception: + cdef _Request req = _Request()._init_child(self.thisptr.newRequest(name, firstSegmentWordSize), self) - def _request(self, name, firstSegmentWordSize=0): - return self._request_helper(name, firstSegmentWordSize) + self._set_fields(req.thisptr_child, name, args, kwargs) + + return req + + def _request(self, name, *args, firstSegmentWordSize=0, **kwargs): + return self._request_helper(name, firstSegmentWordSize, args, kwargs) def _send(self, name, *args, firstSegmentWordSize=0, **kwargs): return self._send_helper(name, firstSegmentWordSize, args, kwargs) diff --git a/capnp/capnp_cpp.pxd b/capnp/capnp_cpp.pxd index aba305f..f360599 100644 --- a/capnp/capnp_cpp.pxd +++ b/capnp/capnp_cpp.pxd @@ -2,13 +2,21 @@ # distutils: language = c++ # distutils: extra_compile_args = --std=c++11 from schema_cpp cimport Node, Data, StructNode, EnumNode, InterfaceNode, MessageBuilder, MessageReader -from async_cpp cimport PyPromise, VoidPromise, Promise from cpython.ref cimport PyObject from libc.stdint cimport * ctypedef unsigned int uint from libcpp cimport bool as cbool +cdef extern from "kj/async.h" namespace " ::kj": + cdef cppclass Promise[T]: + Promise() + Promise(Promise) + T wait(WaitScope) + +ctypedef Promise[PyObject *] PyPromise +ctypedef Promise[void] VoidPromise + cdef extern from "capabilityHelper.h": void reraise_kj_exception() @@ -59,11 +67,21 @@ cdef extern from "kj/array.h" namespace " ::kj": cdef extern from "kj/async-io.h" namespace " ::kj": cdef cppclass AsyncIoStream: pass - cdef cppclass AsyncIoProvider: + cdef cppclass LowLevelAsyncIoProvider: # Own[AsyncInputStream] wrapInputFd(int) # Own[AsyncOutputStream] wrapOutputFd(int) Own[AsyncIoStream] wrapSocketFd(int) - Own[AsyncIoProvider] setupIoEventLoop() + cdef cppclass AsyncIoProvider: + pass + cdef cppclass WaitScope: + pass + cdef cppclass AsyncIoContext: + AsyncIoContext(AsyncIoContext&) + Own[LowLevelAsyncIoProvider] lowLevelProvider + Own[AsyncIoProvider] provider + WaitScope waitScope + + AsyncIoContext setupAsyncIo() cdef extern from "capnp/schema.h" namespace " ::capnp": cdef cppclass Schema: @@ -358,15 +376,16 @@ cdef extern from "capnp/capability.h" namespace " ::capnp": cdef cppclass CallContext' ::capnp::CallContext< ::capnp::DynamicStruct, ::capnp::DynamicStruct>': CallContext(CallContext&) DynamicStruct.Reader getParams() except +reraise_kj_exception - void releaseParams() + void releaseParams() except +reraise_kj_exception DynamicStruct.Builder getResults(uint firstSegmentWordSize) DynamicStruct.Builder initResults(uint firstSegmentWordSize) void setResults(DynamicStruct.Reader value) # void adoptResults(Orphan&& value); # Orphanage getResultsOrphanage(uint firstSegmentWordSize = 0); - void allowAsyncCancellation(bint allow = true) - bint isCanceled() + VoidPromise tailCall(Request & tailRequest) + void allowAsyncCancellation() except +reraise_kj_exception + bint isCanceled() except +reraise_kj_exception cdef extern from "kj/async.h" namespace " ::kj": cdef cppclass EventLoop: @@ -386,8 +405,3 @@ cdef extern from "kj/async.h" namespace " ::kj": VoidPromise promise Own[PromiseFulfiller] fulfiller PromiseFulfillerPair newPromiseAndFulfiller" ::kj::newPromiseAndFulfiller"() - -cdef extern from "kj/async-unix.h" namespace " ::kj": - cdef cppclass UnixEventLoop(EventLoop): - pass - diff --git a/test/test_capability.capnp b/test/test_capability.capnp index f4ae9e1..b9c1076 100644 --- a/test/test_capability.capnp +++ b/test/test_capability.capnp @@ -54,4 +54,25 @@ struct TestSturdyRefObjectId { testExtends @1; testPipeline @2; } +} + +interface TestCallOrder { + getCallSequence @0 (expected: UInt32) -> (n: UInt32); + # First call returns 0, next returns 1, ... + # + # The input `expected` is ignored but useful for disambiguating debug logs. +} + +interface TestTailCallee { + struct TailResult { + i @0 :UInt32; + t @1 :Text; + c @2 :TestCallOrder; + } + + foo @0 (i :Int32, t :Text) -> TailResult; +} + +interface TestTailCaller { + foo @0 (i :Int32, callee :TestTailCallee) -> TestTailCallee.TailResult; } \ No newline at end of file diff --git a/test/test_capability.py b/test/test_capability.py index 75a468b..f87e301 100644 --- a/test/test_capability.py +++ b/test/test_capability.py @@ -22,8 +22,9 @@ class Server: return i.host + '_test' class PipelineServer: - def getCap(self, n, inCap, _results, **kwargs): + def getCap(self, n, inCap, _context, **kwargs): def _then(response): + _results = _context.results _results.s = response.x + '_foo' _results.outBox.cap = capability().TestInterface.new_server(Server(100)) @@ -148,8 +149,9 @@ def test_exception_client(capability): remote.wait() class BadPipelineServer: - def getCap(self, n, inCap, _results, **kwargs): + def getCap(self, n, inCap, _context, **kwargs): def _then(response): + _results = _context.results _results.s = response.x + '_foo' _results.outBox.cap = capability().TestInterface.new_server(Server(100)) def _error(error): @@ -190,3 +192,61 @@ def test_casting(capability): with pytest.raises(Exception): client.upcast(capability.TestPipeline) + +class TailCallOrder: + def __init__(self): + self.count = -1 + + def getCallSequence(self, expected, **kwargs): + self.count += 1 + return self.count + +class TailCaller: + def __init__(self): + self.count = 0 + + def foo(self, i, callee, _context, **kwargs): + self.count += 1 + + tail = callee.foo_request(i=i, t='from TailCaller') + return _context.tail_call(tail) + +class TailCallee: + def __init__(self): + self.count = 0 + + def foo(self, i, t, _context, **kwargs): + self.count += 1 + + results = _context.results + results.i = i + results.t = t + results.c = capability().TestCallOrder.new_server(TailCallOrder()) + +def test_tail_call(capability): + callee_server = TailCallee() + caller_server = TailCaller() + + callee = capability.TestTailCallee._new_client(callee_server) + caller = capability.TestTailCaller._new_client(caller_server) + + promise = caller.foo(i=456, callee=callee) + dependent_call1 = promise.c.getCallSequence() + + response = promise.wait() + + assert response.i == 456 + assert response.i == 456 + + dependent_call2 = response.c.getCallSequence() + dependent_call3 = response.c.getCallSequence() + + result = dependent_call1.wait() + assert result.n == 0 + result = dependent_call2.wait() + assert result.n == 1 + result = dependent_call3.wait() + assert result.n == 2 + + assert callee_server.count == 1 + assert caller_server.count == 1 \ No newline at end of file diff --git a/test/test_capability_context.py b/test/test_capability_context.py index 2cee72b..7c0c962 100644 --- a/test/test_capability_context.py +++ b/test/test_capability_context.py @@ -188,3 +188,61 @@ def test_casting_context(capability): with pytest.raises(Exception): client.upcast(capability.TestPipeline) + +class TailCallOrder: + def __init__(self): + self.count = -1 + + def getCallSequence_context(self, context): + self.count += 1 + context.results.n = self.count + +class TailCaller: + def __init__(self): + self.count = 0 + + def foo_context(self, context): + self.count += 1 + + tail = context.params.callee.foo_request(i=context.params.i, t='from TailCaller') + return context.tail_call(tail) + +class TailCallee: + def __init__(self): + self.count = 0 + + def foo_context(self, context): + self.count += 1 + + results = context.results + results.i = context.params.i + results.t = context.params.t + results.c = capability().TestCallOrder.new_server(TailCallOrder()) + +def test_tail_call(capability): + callee_server = TailCallee() + caller_server = TailCaller() + + callee = capability.TestTailCallee._new_client(callee_server) + caller = capability.TestTailCaller._new_client(caller_server) + + promise = caller.foo(i=456, callee=callee) + dependent_call1 = promise.c.getCallSequence() + + response = promise.wait() + + assert response.i == 456 + assert response.i == 456 + + dependent_call2 = response.c.getCallSequence() + dependent_call3 = response.c.getCallSequence() + + result = dependent_call1.wait() + assert result.n == 0 + result = dependent_call2.wait() + assert result.n == 1 + result = dependent_call3.wait() + assert result.n == 2 + + assert callee_server.count == 1 + assert caller_server.count == 1