Update for API changes and add tail_call functionality

capnp/async_cpp.pxd

@@ -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

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 "<utility>" 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 "<capnp/pretty-print.h>" 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 = <object>self.thisptr.wait() # TODO: make sure refcount is fine here...
+        ret = <object>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 _set_fields(self, Request * request, name, args, kwargs):
-        cdef Request * request = new Request(self.thisptr.newRequest(name, firstSegmentWordSize))
-
         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)
 
+        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:
+    cpdef _request_helper(self, name, firstSegmentWordSize, args, kwargs) except +reraise_kj_exception:
-        return _Request()._init_child(self.thisptr.newRequest(name, firstSegmentWordSize), self)
+        cdef _Request req = _Request()._init_child(self.thisptr.newRequest(name, firstSegmentWordSize), self)
 
-    def _request(self, name, firstSegmentWordSize=0):
+        self._set_fields(req.thisptr_child, name, args, kwargs)
-        return self._request_helper(name, firstSegmentWordSize)
+
+        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)

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<Results>&& value);
         # Orphanage getResultsOrphanage(uint firstSegmentWordSize = 0);
-        void allowAsyncCancellation(bint allow = true)
+        VoidPromise tailCall(Request & tailRequest)
-        bint isCanceled()
+        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<void>"()
-
-cdef extern from "kj/async-unix.h" namespace " ::kj":
-    cdef cppclass UnixEventLoop(EventLoop):
-        pass
-

test/test_capability.capnp

@@ -55,3 +55,24 @@ struct TestSturdyRefObjectId {
     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;
+}

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

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