From e92f7b56c63252c6514f4b5a71b13b272ca189c1 Mon Sep 17 00:00:00 2001 From: Jason Paryani Date: Tue, 10 Dec 2013 22:57:21 -0800 Subject: [PATCH] Various fixups to the RPC api. * change how restorer works * fix join_promises * add incref's all around to make sure we aren't freeing objects early * make it so we return PyPromises everywhere and make chains collapsible --- capnp/helpers/capabilityHelper.h | 48 +- capnp/helpers/helpers.pxd | 7 +- capnp/helpers/rpcHelper.h | 8 +- capnp/includes/capnp_cpp.pxd | 22 +- capnp/lib/capnp.pxd | 2 +- capnp/lib/capnp.pyx | 202 +++-- examples/c++.capnp | 27 - examples/calculator-client.py | 5 +- examples/rpc.capnp | 1251 ------------------------------ test/test_rpc.py | 55 +- 10 files changed, 267 insertions(+), 1360 deletions(-) delete mode 100644 examples/c++.capnp delete mode 100644 examples/rpc.capnp diff --git a/capnp/helpers/capabilityHelper.h b/capnp/helpers/capabilityHelper.h index 26708f0..417fa9d 100644 --- a/capnp/helpers/capabilityHelper.h +++ b/capnp/helpers/capabilityHelper.h @@ -6,12 +6,14 @@ #include extern "C" { - void wrap_remote_call(PyObject * func, capnp::Response &); + PyObject * wrap_remote_call(PyObject * func, capnp::Response &); PyObject * wrap_dynamic_struct_reader(capnp::DynamicStruct::Reader &); ::kj::Promise * call_server_method(PyObject * py_server, char * name, capnp::CallContext< capnp::DynamicStruct, capnp::DynamicStruct> & context); PyObject * wrap_kj_exception(kj::Exception &); PyObject * wrap_kj_exception_for_reraise(kj::Exception &); PyObject * get_exception_info(PyObject *, PyObject *, PyObject *); + PyObject * convert_array_pyobject(kj::Array&); + ::kj::Promise * extract_promise(PyObject *); } void reraise_kj_exception() { @@ -39,6 +41,8 @@ void check_py_error() { if(err) { PyObject * ptype, *pvalue, *ptraceback; PyErr_Fetch(&ptype, &pvalue, &ptraceback); + if(ptype == NULL || pvalue == NULL || ptraceback == NULL) + throw kj::Exception(kj::Exception::Nature::OTHER, kj::Exception::Durability::PERMANENT, kj::heapString("capabilityHelper.h"), 44, kj::heapString("Unknown error occurred")); PyObject * info = get_exception_info(ptype, pvalue, ptraceback); @@ -62,26 +66,39 @@ void check_py_error() { } // TODO: need to decref error_func as well on successful run -PyObject * wrapPyFunc(PyObject * func, PyObject * arg) { +kj::Promise wrapPyFunc(PyObject * func, PyObject * arg) { PyObject * result = PyObject_CallFunctionObjArgs(func, arg, NULL); Py_DECREF(func); check_py_error(); + + auto promise = extract_promise(result); + if(promise != NULL) + return kj::mv(*promise); return result; } -PyObject * wrapPyFuncNoArg(PyObject * func) { +kj::Promise wrapPyFuncNoArg(PyObject * func) { PyObject * result = PyObject_CallFunctionObjArgs(func, NULL); Py_DECREF(func); check_py_error(); + + auto promise = extract_promise(result); + if(promise != NULL) + return kj::mv(*promise); return result; } -void wrapRemoteCall(PyObject * func, capnp::Response & arg) { - wrap_remote_call(func, arg); +kj::Promise wrapRemoteCall(PyObject * func, capnp::Response & arg) { + PyObject * ret = wrap_remote_call(func, arg); check_py_error(); + + auto promise = extract_promise(ret); + if(promise != NULL) + return kj::mv(*promise); + return ret; } ::kj::Promise then(kj::Promise & promise, PyObject * func, PyObject * error_func) { @@ -92,12 +109,12 @@ void wrapRemoteCall(PyObject * func, capnp::Response & arg , [error_func](kj::Exception arg) { return wrapPyFunc(error_func, wrap_kj_exception(arg)); } ); } -::kj::Promise then(::capnp::RemotePromise< ::capnp::DynamicStruct> & promise, PyObject * func, PyObject * error_func) { +::kj::Promise then(::capnp::RemotePromise< ::capnp::DynamicStruct> & promise, PyObject * func, PyObject * error_func) { if(error_func == Py_None) - return promise.then([func](capnp::Response&& arg) { wrapRemoteCall(func, arg); } ); + return promise.then([func](capnp::Response&& arg) { return wrapRemoteCall(func, arg); } ); else - return promise.then([func](capnp::Response&& arg) { wrapRemoteCall(func, arg); } - , [error_func](kj::Exception arg) { wrapPyFunc(error_func, wrap_kj_exception(arg)); } ); + return promise.then([func](capnp::Response&& arg) { return wrapRemoteCall(func, arg); } + , [error_func](kj::Exception arg) { return wrapPyFunc(error_func, wrap_kj_exception(arg)); } ); } ::kj::Promise then(kj::Promise & promise, PyObject * func, PyObject * error_func) { @@ -108,6 +125,10 @@ void wrapRemoteCall(PyObject * func, capnp::Response & arg , [error_func](kj::Exception arg) { return wrapPyFunc(error_func, wrap_kj_exception(arg)); } ); } +::kj::Promise then(kj::Promise > && promise) { + return promise.then([](kj::Array&& arg) { return convert_array_pyobject(arg); } ); +} + class PythonInterfaceDynamicImpl final: public capnp::DynamicCapability::Server { public: PyObject * py_server; @@ -152,3 +173,12 @@ capnp::Capability::Client server_to_client(capnp::InterfaceSchema & schema, PyOb ::kj::Promise convert_to_pypromise(capnp::RemotePromise & promise) { return promise.then([](capnp::Response&& response) { return wrap_dynamic_struct_reader(response); } ); } + +::kj::Promise convert_to_pypromise(kj::Promise & promise) { + return promise.then([]() { Py_RETURN_NONE;} ); +} + +template +::kj::Promise convert_to_voidpromise(kj::Promise & promise) { + return promise.then([](T) { } ); +} diff --git a/capnp/helpers/helpers.pxd b/capnp/helpers/helpers.pxd index 8b25aa6..833f65f 100644 --- a/capnp/helpers/helpers.pxd +++ b/capnp/helpers/helpers.pxd @@ -1,4 +1,4 @@ -from .capnp.includes.capnp_cpp cimport Maybe, DynamicStruct, Request, PyPromise, VoidPromise, RemotePromise, DynamicCapability, InterfaceSchema, EnumSchema, StructSchema, DynamicValue, Capability, RpcSystem, MessageBuilder, MessageReader, TwoPartyVatNetwork, PyRestorer, AnyPointer +from .capnp.includes.capnp_cpp cimport Maybe, DynamicStruct, Request, PyPromise, VoidPromise, PyPromiseArray, RemotePromise, DynamicCapability, InterfaceSchema, EnumSchema, StructSchema, DynamicValue, Capability, RpcSystem, MessageBuilder, MessageReader, TwoPartyVatNetwork, PyRestorer, AnyPointer from non_circular cimport reraise_kj_exception @@ -14,12 +14,15 @@ cdef extern from "../helpers/capabilityHelper.h": # PyPromise evalLater(EventLoop &, PyObject * func) # PyPromise there(EventLoop & loop, PyPromise & promise, PyObject * func, PyObject * error_func) PyPromise then(PyPromise & promise, PyObject * func, PyObject * error_func) - VoidPromise then(RemotePromise & promise, PyObject * func, PyObject * error_func) + PyPromise then(RemotePromise & promise, PyObject * func, PyObject * error_func) PyPromise then(VoidPromise & promise, PyObject * func, PyObject * error_func) + PyPromise then(PyPromiseArray & promise) DynamicCapability.Client new_client(InterfaceSchema&, PyObject *) DynamicValue.Reader new_server(InterfaceSchema&, PyObject *) Capability.Client server_to_client(InterfaceSchema&, PyObject *) PyPromise convert_to_pypromise(RemotePromise&) + PyPromise convert_to_pypromise(VoidPromise&) + VoidPromise convert_to_voidpromise(PyPromise&) cdef extern from "../helpers/rpcHelper.h": Capability.Client restoreHelper(RpcSystem&, MessageBuilder&) diff --git a/capnp/helpers/rpcHelper.h b/capnp/helpers/rpcHelper.h index d40af54..b453067 100644 --- a/capnp/helpers/rpcHelper.h +++ b/capnp/helpers/rpcHelper.h @@ -6,12 +6,12 @@ #include "capabilityHelper.h" extern "C" { - capnp::Capability::Client * call_py_restorer(PyObject *, capnp::DynamicStruct::Reader &); + capnp::Capability::Client * call_py_restorer(PyObject *, capnp::AnyPointer::Reader &); } class PyRestorer final: public capnp::SturdyRefRestorer { public: - PyRestorer(PyObject * _py_restorer, capnp::StructSchema& _schema): py_restorer(_py_restorer), schema(_schema) { + PyRestorer(PyObject * _py_restorer): py_restorer(_py_restorer) { // We don't need to incref/decref, since this C++ class will be owned by the Python wrapper class, and we'll make sure the python class doesn't refcount to 0 elsewhere. // Py_INCREF(py_restorer); } @@ -21,8 +21,7 @@ public: // } capnp::Capability::Client restore(capnp::AnyPointer::Reader objectId) override { - auto reader = objectId.getAs(schema); - capnp::Capability::Client * ret = call_py_restorer(py_restorer, reader); + capnp::Capability::Client * ret = call_py_restorer(py_restorer, objectId); check_py_error(); capnp::Capability::Client stack_ret(*ret); delete ret; @@ -32,7 +31,6 @@ public: private: PyObject * py_restorer; - capnp::StructSchema schema; }; capnp::Capability::Client restoreHelper(capnp::RpcSystem& client, capnp::MessageBuilder & objectId) { capnp::MallocMessageBuilder hostIdMessage(8); diff --git a/capnp/includes/capnp_cpp.pxd b/capnp/includes/capnp_cpp.pxd index 6b24193..e543969 100644 --- a/capnp/includes/capnp_cpp.pxd +++ b/capnp/includes/capnp_cpp.pxd @@ -12,6 +12,7 @@ cdef extern from "kj/async.h" namespace " ::kj": cdef cppclass Promise[T]: Promise() Promise(Promise) + Promise(T) T wait(WaitScope) ctypedef Promise[PyObject *] PyPromise @@ -60,6 +61,19 @@ cdef extern from "kj/array.h" namespace " ::kj": cdef cppclass Array[T]: T* begin() size_t size() + T& operator[](size_t index) + cdef cppclass ArrayBuilder[T]: + T* begin() + size_t size() + T& operator[](size_t index) + T& add(T&) + Array[T] finish() + + ArrayBuilder[PyPromise] heapArrayBuilderPyPromise"::kj::heapArrayBuilder< ::kj::Promise >"(size_t) + + ctypedef Array[PyObject *] PyArray' ::kj::Array' + +ctypedef Promise[PyArray] PyPromiseArray cdef extern from "kj/async-io.h" namespace " ::kj": cdef cppclass AsyncIoStream: @@ -217,7 +231,7 @@ cdef extern from "capnp/capability.h" namespace " ::capnp": cdef extern from "../helpers/rpcHelper.h": cdef cppclass PyRestorer: - PyRestorer(PyObject *, StructSchema&) + PyRestorer(PyObject *) cdef extern from "capnp/rpc-twoparty.h" namespace " ::capnp": cdef cppclass RpcSystem" ::capnp::RpcSystem": @@ -249,10 +263,13 @@ cdef extern from "capnp/any.h" namespace " ::capnp": cdef cppclass AnyPointer: cppclass Reader: DynamicStruct.Reader getAs"getAs< ::capnp::DynamicStruct>"(StructSchema) + String getAsText"getAs< ::capnp::Text>"() cppclass Builder: Builder(Builder) DynamicStruct.Builder getAs"getAs< ::capnp::DynamicStruct>"(StructSchema) - void setAsText"setAs< ::capnp::Text>"(char*) + String getAsText"getAs< ::capnp::Text>"() + void setAsStruct"setAs< ::capnp::DynamicStruct>"(DynamicStruct.Reader&) except +reraise_kj_exception + void setAsText"setAs< ::capnp::Text>"(char*) except +reraise_kj_exception cdef extern from "capnp/dynamic.h" namespace " ::capnp": cdef cppclass DynamicEnum: @@ -381,3 +398,4 @@ cdef extern from "kj/async.h" namespace " ::kj": VoidPromise promise Own[PromiseFulfiller] fulfiller PromiseFulfillerPair newPromiseAndFulfiller" ::kj::newPromiseAndFulfiller"() + PyPromiseArray joinPromises(Array[PyPromise]) diff --git a/capnp/lib/capnp.pxd b/capnp/lib/capnp.pxd index 9868714..fd76d2e 100644 --- a/capnp/lib/capnp.pxd +++ b/capnp/lib/capnp.pxd @@ -1,6 +1,6 @@ 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, 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, makeRpcClient, Capability as C_Capability, TwoPartyVatNetwork as C_TwoPartyVatNetwork, Side, AsyncIoStream, Own, makeTwoPartyVatNetwork, PromiseFulfillerPair as C_PromiseFulfillerPair, copyPromiseFulfillerPair, newPromiseAndFulfiller +from .capnp.includes.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, 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, makeRpcClient, Capability as C_Capability, TwoPartyVatNetwork as C_TwoPartyVatNetwork, Side, AsyncIoStream, Own, makeTwoPartyVatNetwork, PromiseFulfillerPair as C_PromiseFulfillerPair, copyPromiseFulfillerPair, newPromiseAndFulfiller, PyArray 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 36e71b1..90626d2 100644 --- a/capnp/lib/capnp.pyx +++ b/capnp/lib/capnp.pyx @@ -17,6 +17,7 @@ from types import ModuleType as _ModuleType import os as _os import sys as _sys import imp as _imp +import traceback as _traceback from functools import partial as _partial import warnings as _warnings import inspect as _inspect @@ -26,12 +27,14 @@ from operator import attrgetter as _attrgetter cdef public object wrap_dynamic_struct_reader(C_DynamicStruct.Reader & reader): return _DynamicStructReader()._init(reader, None) -cdef public void wrap_remote_call(PyObject * func, Response & r) except *: +cdef public PyObject * wrap_remote_call(PyObject * func, Response & r) except *: response = _Response()._init_childptr(new Response(moveResponse(r)), None) func_obj = func # TODO: decref func? - func_obj(response) + ret = func_obj(response) + Py_INCREF(ret) + return ret cdef _find_field_order(struct_node): return [f.name for f in sorted(struct_node.fields, key=_attrgetter('codeOrder'))] @@ -44,24 +47,29 @@ cdef public VoidPromise * call_server_method(PyObject * _server, char * _method_ func = getattr(server, method_name+'_context', None) if func is not None: ret = func(context) + Py_INCREF(ret) #TODO: stop leaking this if ret is not None: if type(ret) is _VoidPromise: return new VoidPromise(moveVoidPromise(deref((<_VoidPromise>ret).thisptr))) + elif type(ret) is Promise: + return new VoidPromise(helpers.convert_to_voidpromise(deref((ret).thisptr))) else: try: - warning_msg = 'Server function (%s) returned a value that was not a VoidPromise: return = %s' % (method_name, str(ret)) + warning_msg = 'Server function (%s) returned a value that was not a Promise: return = %s' % (method_name, str(ret)) except: - warning_msg = 'Server function (%s) returned a value that was not a VoidPromise' % (method_name) + warning_msg = 'Server function (%s) returned a value that was not a Promise' % (method_name) _warnings.warn_explicit(warning_msg, UserWarning, _inspect.getsourcefile(func), _inspect.getsourcelines(func)[1]) if ret is not None: - if type(ret) is _VoidPromise: - return new VoidPromise(moveVoidPromise(deref((<_VoidPromise>ret).thisptr))) + if type(ret) is Promise: + return new VoidPromise(helpers.convert_to_voidpromise(deref((ret).thisptr))) + elif type(ret) is Promise: + return new VoidPromise(helpers.convert_to_voidpromise(deref((ret).thisptr))) else: try: - warning_msg = 'Server function (%s) returned a value that was not a VoidPromise: return = %s' % (method_name, str(ret)) + warning_msg = 'Server function (%s) returned a value that was not a Promise: return = %s' % (method_name, str(ret)) except: - warning_msg = 'Server function (%s) returned a value that was not a VoidPromise' % (method_name) + warning_msg = 'Server function (%s) returned a value that was not a Promise' % (method_name) _warnings.warn_explicit(warning_msg, UserWarning, _inspect.getsourcefile(func), _inspect.getsourcelines(func)[1]) else: func = getattr(server, method_name) # will raise if no function found @@ -69,10 +77,13 @@ cdef public VoidPromise * call_server_method(PyObject * _server, char * _method_ params_dict = {name : getattr(params, name) for name in params.schema.fieldnames} params_dict['_context'] = context ret = func(**params_dict) + Py_INCREF(ret) #TODO: stop leaking this if ret is not None: if type(ret) is _VoidPromise: return new VoidPromise(moveVoidPromise(deref((<_VoidPromise>ret).thisptr))) + elif type(ret) is Promise: + return new VoidPromise(helpers.convert_to_voidpromise(deref((ret).thisptr))) if not isinstance(ret, tuple): ret = (ret,) names = _find_field_order(context.results.schema.node.struct) @@ -85,16 +96,29 @@ cdef public VoidPromise * call_server_method(PyObject * _server, char * _method_ return NULL -cdef public C_Capability.Client * call_py_restorer(PyObject * _restorer, C_DynamicStruct.Reader & _reader) except *: +cdef public C_Capability.Client * call_py_restorer(PyObject * _restorer, C_DynamicObject.Reader & _reader) except *: restorer = _restorer - reader = _DynamicStructReader()._init(_reader, None) + reader = _DynamicObjectReader()._init(_reader, None) - ret = restorer.restore(reader) + ret = restorer._restore(reader) cdef _DynamicCapabilityServer server = ret cdef _InterfaceSchema schema = ret.schema return new C_Capability.Client(helpers.server_to_client(schema.thisptr, server)) + +cdef public convert_array_pyobject(PyArray & arr): + return [arr[i] for i in range(arr.size())] + +cdef public PyPromise * extract_promise(object obj): + if type(obj) is Promise: + promise = obj + promise.is_consumed = True + Py_INCREF(promise) # TODO: fix leak + return promise.thisptr + + return NULL + cdef extern from "" namespace " ::kj": String strStructReader" ::kj::str"(C_DynamicStruct.Reader) String strStructBuilder" ::kj::str"(C_DynamicStruct.Builder) @@ -234,8 +258,8 @@ ctypedef fused _DynamicSetterClasses: C_DynamicStruct.Builder Request -ctypedef fused _PromiseTypes: - _Promise +ctypedef fused PromiseTypes: + Promise _RemotePromise _VoidPromise PromiseFulfillerPair @@ -637,7 +661,7 @@ cdef _setDynamicField(_DynamicSetterClasses thisptr, field, value, parent): elif value_type is _DynamicCapabilityServer or isinstance(value, _DynamicCapabilityServer): thisptr.set(field, _extract_dynamic_server(value)) else: - raise ValueError("Non primitive type") + raise ValueError("Tried to set field: '{}' with a value of: '{}' which is an unsupported type: '{}'".format(field, str(value), str(type(value)))) cdef _setDynamicFieldPtr(_DynamicSetterClasses * thisptr, field, value, parent): cdef C_DynamicValue.Reader temp @@ -672,7 +696,7 @@ cdef _setDynamicFieldPtr(_DynamicSetterClasses * thisptr, field, value, parent): elif value_type is _DynamicCapabilityClient: thisptr.set(field, _extract_dynamic_client(value)) else: - raise ValueError("Non primitive type") + raise ValueError("Tried to set field: '{}' with a value of: '{}' which is an unsupported type: '{}'".format(field, str(value), str(type(value)))) cdef _to_dict(msg, bint verbose): msg_type = type(msg) @@ -1108,7 +1132,7 @@ cdef class _DynamicObjectReader: self._parent = parent return self - cpdef as_struct(self, schema): + cpdef as_struct(self, schema) except +reraise_kj_exception: cdef _StructSchema s if hasattr(schema, 'schema'): s = schema.schema @@ -1117,6 +1141,9 @@ cdef class _DynamicObjectReader: return _DynamicStructReader()._init(self.thisptr.getAs(s.thisptr), self._parent) + cpdef as_text(self) except +reraise_kj_exception: + return (self.thisptr.getAsText().cStr())[:] + cdef class _DynamicObjectBuilder: cdef C_DynamicObject.Builder * thisptr cdef public object _parent @@ -1129,7 +1156,7 @@ cdef class _DynamicObjectBuilder: def __dealloc__(self): del self.thisptr - cpdef as_struct(self, schema): + cpdef as_struct(self, schema) except +reraise_kj_exception: cdef _StructSchema s if hasattr(schema, 'schema'): s = schema.schema @@ -1141,6 +1168,9 @@ cdef class _DynamicObjectBuilder: cpdef set_as_text(self, text): self.thisptr.setAsText(text) + cpdef as_text(self) except +reraise_kj_exception: + return (self.thisptr.getAsText().cStr())[:] + cdef class _EventLoop: cdef capnp.AsyncIoContext * thisptr @@ -1164,6 +1194,11 @@ cdef class _EventLoop: cdef _EventLoop C_DEFAULT_EVENT_LOOP = _EventLoop() +cpdef reset_event_loop(): + global C_DEFAULT_EVENT_LOOP + C_DEFAULT_EVENT_LOOP._remove() + C_DEFAULT_EVENT_LOOP = _EventLoop() + cdef class _CallContext: cdef CallContext * thisptr @@ -1192,18 +1227,28 @@ cdef class _CallContext: self.thisptr.allowCancellation() cpdef tail_call(self, _Request tailRequest): - return _VoidPromise()._init(self.thisptr.tailCall(moveRequest(deref(tailRequest.thisptr_child)))) + promise = _VoidPromise()._init(self.thisptr.tailCall(moveRequest(deref(tailRequest.thisptr_child)))) + promise.is_consumed = True + return promise -cdef class _Promise: +cdef class Promise: cdef PyPromise * thisptr cdef public bint is_consumed + cdef public object _parent, _obj - def __init__(self): - self.is_consumed = True + def __init__(self, obj=None): + if obj is None: + self.is_consumed = True + else: + self.is_consumed = False + self._obj = obj + Py_INCREF(obj) # TODO: fix this + self.thisptr = new PyPromise(obj) - cdef _init(self, PyPromise other): + cdef _init(self, PyPromise other, parent=None): self.is_consumed = False self.thisptr = new PyPromise(movePromise(other)) + self._parent = parent return self def __dealloc__(self): @@ -1225,7 +1270,7 @@ cdef class _Promise: Py_INCREF(func) Py_INCREF(error_func) - return _Promise()._init(helpers.then(deref(self.thisptr), func, error_func)) + return Promise()._init(helpers.then(deref(self.thisptr), func, error_func), self) cdef class _VoidPromise: cdef VoidPromise * thisptr @@ -1257,7 +1302,12 @@ cdef class _VoidPromise: Py_INCREF(func) Py_INCREF(error_func) - return _Promise()._init(helpers.then(deref(self.thisptr), func, error_func)) + return Promise()._init(helpers.then(deref(self.thisptr), func, error_func), self) + + cpdef as_pypromise(self) except +reraise_kj_exception: + if self.is_consumed: + raise RuntimeError('Promise was already used in a consuming operation. You can no longer use this Promise object') + Promise()._init(helpers.convert_to_pypromise(deref(self.thisptr)), self) cdef class _RemotePromise: cdef RemotePromise * thisptr @@ -1273,8 +1323,8 @@ cdef class _RemotePromise: self._parent = parent return self - def __dealloc__(self): - del self.thisptr + # def __dealloc__(self): + # del self.thisptr cpdef wait(self) except +reraise_kj_exception: if self.is_consumed: @@ -1286,7 +1336,9 @@ cdef class _RemotePromise: return ret cpdef as_pypromise(self) except +reraise_kj_exception: - _Promise()._init(helpers.convert_to_pypromise(deref(self.thisptr))) + if self.is_consumed: + raise RuntimeError('Promise was already used in a consuming operation. You can no longer use this Promise object') + Promise()._init(helpers.convert_to_pypromise(deref(self.thisptr)), self) cpdef then(self, func, error_func=None) except +reraise_kj_exception: if self.is_consumed: @@ -1295,7 +1347,7 @@ cdef class _RemotePromise: Py_INCREF(func) Py_INCREF(error_func) - return _VoidPromise()._init(helpers.then(deref(self.thisptr), func, error_func)) + return Promise()._init(helpers.then(deref(self.thisptr), func, error_func), self) cpdef _get(self, field) except +reraise_kj_exception: cdef int type = (self.thisptr.get(field)).getType() @@ -1322,6 +1374,26 @@ cdef class _RemotePromise: def to_dict(self, verbose=False): return _to_dict(self, verbose) +cpdef join_promises(promises) except +reraise_kj_exception: + heap = capnp.heapArrayBuilderPyPromise(len(promises)) + + new_promises = [] + new_promises_append = new_promises.append + + for promise in promises: + promise_type = type(promise) + if promise_type is Promise: + pyPromise = promise + elif promise_type is _RemotePromise or promise_type is _VoidPromise: + pyPromise = promise.as_pypromise() + new_promises_append(pyPromise) + else: + raise ValueError('One of the promises passed to `join_promises` had a non promise value of: ' + str(promise)) + heap.add(movePromise(deref(pyPromise.thisptr))) + pyPromise.is_consumed = True + + return Promise()._init(helpers.then(capnp.joinPromises(heap.finish()))) + cdef class _Request(_DynamicStructBuilder): cdef Request * thisptr_child @@ -1330,6 +1402,7 @@ cdef class _Request(_DynamicStructBuilder): self._init(deref(self.thisptr_child), parent) return self + #TODO: dealloc cpdef send(self): return _RemotePromise()._init(self.thisptr_child.send(), self._parent) @@ -1341,6 +1414,7 @@ cdef class _Response(_DynamicStructReader): self._init(deref(self.thisptr_child), parent) return self + #TODO: dealloc cdef _init_childptr(self, Response * other, parent): self.thisptr_child = other self._init(deref(self.thisptr_child), parent) @@ -1483,26 +1557,20 @@ cdef class _CapabilityClient: s = schema return _DynamicCapabilityClient()._init(self.thisptr.castAs(s.thisptr), self._parent) -cdef class Restorer: +cdef class _Restorer: cdef PyRestorer * thisptr - cdef C_StructSchema schema - cdef public object restore - def __init__(self, schema, restore_func): - cdef _StructSchema s - if hasattr(schema, 'schema'): - s = schema.schema - else: - s = schema - - self.schema = s.thisptr - self.restore = restore_func - self.thisptr = new PyRestorer(self, self.schema) + def __init__(self, restore): + self.thisptr = new PyRestorer(self) + self.restore = restore def __dealloc__(self): del self.thisptr + def _restore(self, obj): + return self.restore(obj) + cdef class _TwoPartyVatNetwork: cdef Own[C_TwoPartyVatNetwork] thisptr @@ -1510,21 +1578,34 @@ cdef class _TwoPartyVatNetwork: self.thisptr = makeTwoPartyVatNetwork(stream, side) return self -cdef class RpcClient: +cdef _Restorer _convert_restorer(restorer): + if isinstance(restorer, _RestorerImpl): + return _Restorer(restorer._restore) + elif type(restorer) is _Restorer: + return restorer + elif hasattr(restorer, 'restore'): + return _Restorer(restorer.restore) + elif callable(restorer): + return _Restorer(restorer) + else: + raise ValueError("Restorer object ({}) isn't able to be used as a restore".format(str(restorer))) + +cdef class TwoPartyClient: cdef RpcSystem * thisptr cdef public _TwoPartyVatNetwork network - cdef public object restorer, _stream + cdef public object _stream + cdef public _Restorer restorer cdef public _FdAsyncIoStream stream - def __init__(self, stream, Restorer restorer=None): + def __init__(self, stream, restorer=None): self._stream = stream self.stream = _FdAsyncIoStream(stream.fileno()) self.network = _TwoPartyVatNetwork()._init(deref(self.stream.thisptr), capnp.CLIENT) if restorer is None: self.thisptr = new RpcSystem(makeRpcClient(deref(self.network.thisptr))) else: - self.restorer = restorer - self.thisptr = new RpcSystem(makeRpcClientWithRestorer(deref(self.network.thisptr), deref(restorer.thisptr))) + self.restorer = _convert_restorer(restorer) + self.thisptr = new RpcSystem(makeRpcClientWithRestorer(deref(self.network.thisptr), deref(self.restorer.thisptr))) def __dealloc__(self): del self.thisptr @@ -1568,18 +1649,23 @@ cdef class RpcClient: return self.restore(ref.objectId) -cdef class RpcServer: +cdef class TwoPartyServer: cdef RpcSystem * thisptr cdef public _TwoPartyVatNetwork network - cdef public object restorer, _stream + cdef public object _stream + cdef public _Restorer restorer cdef public _FdAsyncIoStream stream - def __init__(self, stream, Restorer restorer): + def __init__(self, stream, restorer): self._stream = stream self.stream = _FdAsyncIoStream(stream.fileno()) - self.restorer = restorer + Py_INCREF(self._stream) + Py_INCREF(self.stream) # TODO: attach this to onDrained, also figure out what's leaking + self.restorer = _convert_restorer(restorer) self.network = _TwoPartyVatNetwork()._init(deref(self.stream.thisptr), capnp.SERVER) - self.thisptr = new RpcSystem(makeRpcServer(deref(self.network.thisptr), deref(restorer.thisptr))) + self.thisptr = new RpcSystem(makeRpcServer(deref(self.network.thisptr), deref(self.restorer.thisptr))) + Py_INCREF(self.restorer) # TODO: attach this to onDrained, also figure out what's leaking + Py_INCREF(self.network) # TODO: attach this to onDrained, also figure out what's leaking def __dealloc__(self): del self.thisptr @@ -1751,9 +1837,17 @@ cdef _new_message(self, kwargs): _from_dict(msg, kwargs) return msg +class _RestorerImpl(object): + pass + class _StructModule(object): - def __init__(self, schema): + def __init__(self, schema, name): + def blank_init(server_self): + pass + def _restore(self, obj): + return self.restore(obj.as_struct(self.schema)) self.schema = schema + self.Restorer = type(name + '.Restorer', (_RestorerImpl,), {'schema':schema, '_restore':_restore}) def read(self, file, traversal_limit_in_words = None, nesting_limit = None): """Returns a Reader for the unpacked object read from file. @@ -1876,7 +1970,7 @@ class _InterfaceModule(object): def server_init(server_self): pass self.schema = schema - self.Server = type(name, (_DynamicCapabilityServer,), {'__init__': server_init, 'schema':schema}) + self.Server = type(name + '.Server', (_DynamicCapabilityServer,), {'__init__': server_init, 'schema':schema}) def _new_client(self, server): return _DynamicCapabilityClient()._init_vals(self.schema, server) @@ -1955,7 +2049,7 @@ cdef class SchemaParser: schema = nodeSchema.get_nested(node.name) proto = schema.get_proto() if proto.isStruct: - local_module = _StructModule(schema.as_struct()) + local_module = _StructModule(schema.as_struct(), node.name) class Reader(_DynamicStructReader): """An abstract base class. Readers are 'instances' of this class.""" __metaclass__ = _StructABCMeta diff --git a/examples/c++.capnp b/examples/c++.capnp deleted file mode 100644 index 7f306a7..0000000 --- a/examples/c++.capnp +++ /dev/null @@ -1,27 +0,0 @@ -# Copyright (c) 2013, Kenton Varda -# All rights reserved. -# -# Redistribution and use in source and binary forms, with or without -# modification, are permitted provided that the following conditions are met: -# -# 1. Redistributions of source code must retain the above copyright notice, this -# list of conditions and the following disclaimer. -# 2. Redistributions in binary form must reproduce the above copyright notice, -# this list of conditions and the following disclaimer in the documentation -# and/or other materials provided with the distribution. -# -# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND -# ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED -# WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE -# DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE LIABLE FOR -# ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES -# (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; -# LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND -# ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT -# (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS -# SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. - -@0xbdf87d7bb8304e81; -$namespace("capnp::annotations"); - -annotation namespace(file): Text; diff --git a/examples/calculator-client.py b/examples/calculator-client.py index 3c6f7bc..5488e9d 100755 --- a/examples/calculator-client.py +++ b/examples/calculator-client.py @@ -6,7 +6,6 @@ import socket import capnp import calculator_capnp -import rpc_capnp class PowerFunction(calculator_capnp.Calculator.Function.Server): '''An implementation of the Function interface wrapping pow(). Note that @@ -18,7 +17,7 @@ class PowerFunction(calculator_capnp.Calculator.Function.Server): return pow(params[0], params[1]) def parse_args(): - parser = argparse.ArgumentParser('Connects to the Calculator server at the given address and does some RPCs') + parser = argparse.ArgumentParser(usage='Connects to the Calculator server at the given address and does some RPCs') parser.add_argument("host", help="HOST:PORT") return parser.parse_args() @@ -27,7 +26,7 @@ def main(): host, port = parse_args().host.split(':') sock = socket.create_connection((host, port)) - client = capnp.RpcClient(sock) + client = capnp.TwoPartyClient(sock) # Pass "calculator" to ez_restore (there's also a `restore` function that takes a struct or AnyPointer as an argument), and then cast the returned capability to it's proper type. This casting is due to capabilities not having a reference to their schema calculator = client.ez_restore('calculator').cast_as(calculator_capnp.Calculator) diff --git a/examples/rpc.capnp b/examples/rpc.capnp deleted file mode 100644 index ed99964..0000000 --- a/examples/rpc.capnp +++ /dev/null @@ -1,1251 +0,0 @@ -# Copyright (c) 2013, Kenton Varda -# All rights reserved. -# -# Redistribution and use in source and binary forms, with or without -# modification, are permitted provided that the following conditions are met: -# -# 1. Redistributions of source code must retain the above copyright notice, this -# list of conditions and the following disclaimer. -# 2. Redistributions in binary form must reproduce the above copyright notice, -# this list of conditions and the following disclaimer in the documentation -# and/or other materials provided with the distribution. -# -# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND -# ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED -# WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE -# DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE LIABLE FOR -# ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES -# (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; -# LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND -# ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT -# (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS -# SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. - -@0xb312981b2552a250; -# Recall that Cap'n Proto RPC allows messages to contain references to remote objects that -# implement interfaces. These references are called "capabilities", because they both designate -# the remote object to use and confer permission to use it. -# -# Recall also that Cap'n Proto RPC has the feature that when a method call itself returns a -# capability, the caller can begin calling methods on that capability _before the first call has -# returned_. The caller essentially sends a message saying "Hey server, as soon as you finish -# that previous call, do this with the result!". Cap'n Proto's RPC protocol makes this possible. -# As a result, it is more complicated than most. -# -# Cap'n Proto RPC is based heavily on CapTP: -# http://www.erights.org/elib/distrib/captp/index.html -# -# Cap'n Proto RPC takes place between "vats". A vat hosts some set of capabilities and talks to -# other vats through direct bilateral connections. Typically, there is a 1:1 correspondence -# between vats and processes (in the unix sense of the word), although this is not strictly always -# true (one process could run multiple vats, or a distributed vat might live across many processes). -# -# Cap'n Proto does not distinguish between "clients" and "servers" -- this is up to the application. -# Either end of any connection can potentially hold capabilities pointing to the other end, and -# can call methods on those capabilities. In the doc comments below, we use the words "sender" -# and "receiver". These refer to the sender and receiver of an instance of the struct or field -# being documented. Sometimes we refer to a "third-party" which is neither the sender nor the -# receiver. -# -# It is generally up to the vat network implementation to securely verify that connections are made -# to the intended vat as well as to encrypt transmitted data for privacy and integrity. See the -# `VatNetwork` example interface near the end of this file. -# -# Once a connection is formed, nothing interesting can happen until one side sends a Restore -# message to obtain a persistent capability. -# -# Unless otherwise specified, messages must be delivered to the receiving application in the same -# order in which they were initiated by the sending application, just like in E: -# http://erights.org/elib/concurrency/partial-order.html -# -# Since the full protocol is complicated, we define multiple levels of support which an -# implementation may target. For typical applications, level 1 support will be sufficient. -# Comments in this file indicate which level requires the corresponding feature to be -# implemented. -# -# * **Level 0:** The implementation does not support object references. `Restore` is supported -# only for looking up singleton objects which exist for the lifetime of the server, and only -# these singleton objects can receive calls. At this level, the implementation does not support -# object-oriented protocols and is similar in complexity to JSON-RPC or Protobuf "generic -# services". This level should be considered only a temporary stepping-stone toward level 1 as -# the lack of object references drastically changes how protocols are designed. Applications -# _should not_ attempt to design their protocols around the limitations of level 0 -# implementations. -# -# * **Level 1:** The implementation supports simple bilateral interaction with object references -# and promise pipelining, but interactions between three or more parties are supported only via -# proxying of objects. E.g. if Alice wants to send Bob a capability pointing to Carol, Alice -# must host a local proxy of Carol and send Bob a reference to that; Bob cannot form a direct -# connection to Carol. Level 1 implementations do not support "join" or "eq" across capabilities -# received from different vats, although they should be supported on capabilities received from -# the same vat. `Restore` is supported only for looking up singleton objects as in level 0. -# -# * **Level 2:** The implementation supports saving, restoring, and deleting persistent -# capabilities. -# -# * **Level 3:** The implementation supports three-way interactions but does not implement "Join" -# operations. The implementation can be used effectively on networks that do not require joins, -# or to implement objects that never need to be joined. -# -# * **Level 4:** The entire protocol is implemented, including joins. -# -# Note that an implementation must also support specific networks (transports), as described in -# the "Network-specific Parameters" section below. An implementation might have different levels -# depending on the network used. -# -# New implementations of Cap'n Proto should start out targeting the simplistic two-party network -# type as defined in `rpc-twoparty.capnp`. With this network type, level 3 is irrelevant and -# levels 2 and 4 are much easier than usual to implement. When such an implementation is paired -# with a container proxy, the contained app effectively gets to make full use of the proxy's -# network at level 4. And since Cap'n Proto IPC is extremely fast, it may never make sense to -# bother implementing any other vat network protocol -- just use the correct container type and get -# it for free. - -using Cxx = import "c++.capnp"; -$Cxx.namespace("capnp::rpc"); - -# ======================================================================================== -# The Four Tables -# -# Cap'n Proto RPC connections are stateful (although an application built on Cap'n Proto could -# export a stateless interface). As in CapTP, for each open connection, a vat maintains four state -# tables: questions, answers, imports, and exports. See the diagram at: -# http://www.erights.org/elib/distrib/captp/4tables.html -# -# The question table corresponds to the other end's answer table, and the imports table corresponds -# to the other end's exports table. -# -# The entries in each table are identified by ID numbers (defined below as 32-bit integers). These -# numbers are always specific to the connection; a newly-established connection starts with no -# valid IDs. Since low-numbered IDs will pack better, it is suggested that IDs be assigned like -# Unix file descriptors -- prefer the lowest-number ID that is currently available. -# -# IDs in the questions/answers tables are chosen by the questioner and generally represent method -# calls that are in progress. -# -# IDs in the imports/exports tables are chosen by the exporter and generally represent objects on -# which methods may be called. Exports may be "settled", meaning the exported object is an actual -# object living in the exporter's vat, or they may be "promises", meaning the exported object is -# the as-yet-unknown result of an ongoing operation and will eventually be resolved to some other -# object once that operation completes. Calls made to a promise will be forwarded to the eventual -# target once it is known. The eventual replacement object does *not* take the same ID as the -# promise, as it may turn out to be an object that is already exported (so already has an ID) or -# may even live in a completely different vat (and so won't get an ID on the same export table -# at all). -# -# IDs can be reused over time. To make this safe, we carefully define the lifetime of IDs. Since -# messages using the ID could be traveling in both directions simultaneously, we must define the -# end of life of each ID _in each direction_. The ID is only safe to reuse once it has been -# released by both sides. -# -# When a Cap'n Proto connection is lost, everything on the four tables is lost. All questions are -# canceled and throw exceptions. All imports become broken (all methods throw exceptions). All -# exports and answers are implicitly released. The only things not lost are persistent -# capabilities (`SturdyRef`s). The application must plan for this and should respond by -# establishing a new connection and restoring from these persistent capabilities. - -using QuestionId = UInt32; -# **(level 0)** -# -# Identifies a question in the questions/answers table. The questioner (caller) chooses an ID -# when making a call. The ID remains valid in caller -> callee messages until a Finish -# message is sent, and remains valid in callee -> caller messages until a Return message is sent. - -using ExportId = UInt32; -# **(level 1)** -# -# Identifies an exported capability or promise in the exports/imports table. The exporter chooses -# an ID before sending a capability over the wire. If the capability is already in the table, the -# exporter should reuse the same ID. If the ID is a promise (as opposed to a settled capability), -# this must be indicated at the time the ID is introduced; in this case, the importer shall expect -# a later Resolve message which replaces the promise. -# -# ExportIds are subject to reference counting. When an `ExportId` is sent embedded in an -# CapDescriptor, the export's reference count is incremented. The reference count is -# later decremented by a `Release` message. Since the `Release` message can specify an arbitrary -# number by which to reduce the reference count, the importer should usually batch reference -# decrements and only send a `Release` when it believes the reference count has hit zero. Of -# course, it is possible that a new reference to the released object is in-flight at the time -# that the `Release` message is sent, so it is necessary for the exporter to keep track of the -# reference count on its end as well to avoid race conditions. -# -# When an `ExportId` is received as part of a exporter -> importer message but not embedded in a -# question or answer, its reference count must be incremented unless otherwise specified. -# -# An `ExportId` remains valid in importer -> exporter messages until its reference count reaches -# zero and a `Release` message has been sent to release it. -# -# When a connection is lost, all exports are implicitly released. It is not possible to restore -# a connection state (or, restoration should be implemented at the transport layer without the RPC -# layer knowing that anything happened). - -# ======================================================================================== -# Messages - -struct Message { - # An RPC connection is a bi-directional stream of Messages. - - union { - unimplemented @0 :Message; - # When a peer receives a message of a type it doesn't recognize or doesn't support, it - # must immediately echo the message back to the sender in `unimplemented`. The sender is - # then able to examine the message and decide how to deal with it being unimplemented. - # - # For example, say `resolve` is received by a level 0 implementation (because a previous call - # or return happened to contain a promise). The receiver will echo it back as `unimplemented`. - # The sender can then simply release the cap to which the promise had resolved, thus avoiding - # a leak. - # - # For any message type that introduces a question, if the message comes back unimplemented, - # the sender may simply treat it as if the question failed with an exception. - # - # In cases where there is no sensible way to react to an `unimplemented` message (without - # resource leaks or other serious problems), the connection may need to be aborted. This is - # a gray area; different implementations may take different approaches. - - abort @1 :Exception; - # Sent when a connection is being aborted due to an unrecoverable error. This could be e.g. - # because the sender received an invalid or nonsensical message (`isCallersFault` is true) or - # because the sender had an internal error (`isCallersFault` is false). The sender will shut - # down the outgoing half of the connection after `abort` and will completely close the - # connection shortly thereafter (it's up to the sender how much of a time buffer they want to - # offer for the client to receive the `abort` before the connection is reset). - - # Level 0 features ----------------------------------------------- - - call @2 :Call; # Begin a method call. - return @3 :Return; # Complete a method call. - finish @4 :Finish; # Release a returned answer / cancel a call. - - # Level 1 features ----------------------------------------------- - - resolve @5 :Resolve; # Resolve a previously-sent promise. - release @6 :Release; # Release a capability so that the remote object can be deallocated. - disembargo @13 :Disembargo; # Lift an embargo used to enforce E-order over promise resolution. - - # Level 2 features ----------------------------------------------- - - save @7 :Save; # Save a capability persistently. - restore @8 :Restore; # Restore a persistent capability from a previous connection. - delete @9 :Delete; # Delete a persistent capability. - - # Level 3 features ----------------------------------------------- - - provide @10 :Provide; # Provide a capability to a third party. - accept @11 :Accept; # Accept a capability provided by a third party. - - # Level 4 features ----------------------------------------------- - - join @12 :Join; # Directly connect to the common root of two or more proxied caps. - } -} - -# Level 0 message types ---------------------------------------------- - -struct Call { - # **(level 0)** - # - # Message type initiating a method call on a capability. - - questionId @0 :QuestionId; - # A number, chosen by the caller, which identifies this call in future messages. This number - # must be different from all other calls originating from the same end of the connection (but - # may overlap with question IDs originating from the opposite end). A fine strategy is to use - # sequential question IDs, but the recipient should not assume this. - # - # A question ID can be reused once both: - # - A matching Return has been received from the callee. - # - A matching Finish has been sent from the caller. - - target @1 :MessageTarget; - # The object that should receive this call. - - interfaceId @2 :UInt64; - # The type ID of the interface being called. Each capability may implement multiple interfaces. - - methodId @3 :UInt16; - # The ordinal number of the method to call within the requested interface. - - allowThirdPartyTailCall @8 :Bool = false; - # Indicates whether or not the receiver is allowed to send a `Return` containing - # `acceptFromThirdParty`. Level 3 implementations should set this true. Otherwise, the callee - # will have to proxy the return in the case of a tail call to a third-party vat. - - params @4 :Payload; - # The call parameters. `params.content` is a struct whose fields correspond to the parameters of - # the method. - - sendResultsTo :union { - # Where should the return message be sent? - - caller @5 :Void; - # Send the return message back to the caller (the usual). - - yourself @6 :Void; - # **(level 1)** - # - # Don't actually return the results to the sender. Instead, hold on to them and await - # instructions from the sender regarding what to do with them. In particular, the sender - # may subsequently send a `Return` for some other call (which the receiver had previously made - # to the sender) with `takeFromOtherAnswer` set. The results from this call are then used - # as the results of the other call. - # - # When `yourself` is used, the receiver must still send a `Return` for the call, but sets the - # field `resultsSentElsewhere` in that `Return` rather than including the results. - # - # This feature can be used to implement tail calls in which a call from Vat A to Vat B ends up - # returning the result of a call from Vat B back to Vat A. - # - # In particular, the most common use case for this feature is when Vat A makes a call to a - # promise in Vat B, and then that promise ends up resolving to a capability back in Vat A. - # Vat B must forward all the queued calls on that promise back to Vat A, but can set `yourself` - # in the calls so that the results need not pass back through Vat B. - # - # For example: - # - Alice, in Vat A, call foo() on Bob in Vat B. - # - Alice makes a pipelined call bar() on the promise returned by foo(). - # - Later on, Bob resolves the promise from foo() to point at Carol, who lives in Vat A (next - # to Alice). - # - Vat B dutifully forwards the bar() call to Carol. Let us call this forwarded call bar'(). - # Notice that bar() and bar'() are travelling in opposite directions on the same network - # link. - # - The `Call` for bar'() has `sendResultsTo` set to `yourself`, with the value being the - # question ID originally assigned to the bar() call. - # - Vat A receives bar'() and delivers it to Carol. - # - When bar'() returns, Vat A immediately takes the results and returns them from bar(). - # - Meanwhile, Vat A sends a `Return` for bar'() to Vat B, with `resultsSentElsewhere` set in - # place of results. - # - Vat A sends a `Finish` for that call to Vat B. - # - Vat B receives the `Return` for bar'() and sends a `Return` for bar(), with - # `receivedFromYourself` set in place of the results. - # - Vat B receives the `Finish` for bar() and sends a `Finish` to bar'(). - - thirdParty @7 :RecipientId; - # **(level 3)** - # - # The call's result should be returned to a different vat. The receiver (the callee) expects - # to receive an `Accept` message from the indicated vat, and should return the call's result - # to it, rather than to the sender of the `Call`. - # - # This operates much like `yourself`, above, except that Carol is in a separate Vat C. `Call` - # messages are sent from Vat A -> Vat B and Vat B -> Vat C. A `Return` message is sent from - # Vat B -> Vat A that contains `acceptFromThirdParty` in place of results. When Vat A sends - # an `Accept` to Vat C, it receives back a `Return` containing the call's actual result. Vat C - # also sends a `Return` to Vat B with `resultsSentElsewhere`. - } -} - -struct Return { - # **(level 0)** - # - # Message type sent from callee to caller indicating that the call has completed. - - questionId @0 :QuestionId; - # Question ID which is being answered, as specified in the corresponding Call. - - releaseParamCaps @1 :Bool = true; - # If true, all capabilities that were in the params should be considered released. The sender - # must not send separate `Release` messages for them. Level 0 implementations in particular - # should always set this true. This defaults true because if level 0 implementations forgot to - # set it they'd never notice (just silently leak caps), but if level >=1 implementations forget - # set it false they'll quickly get errors. - - union { - results @2 :Payload; - # The result. - # - # For regular method calls, `results.content` points to the result struct. - # - # For a `Return` in response to an `Accept`, `results` contains a single capability (rather - # than a struct), and `results.content` is just a capability pointer with index 0. A `Finish` - # is still required in this case. - - exception @3 :Exception; - # Indicates that the call failed and explains why. - - canceled @4 :Void; - # Indicates that the call was canceled due to the caller sending a Finish message - # before the call had completed. - - resultsSentElsewhere @5 :Void; - # This is set when returning from a `Call` which had `sendResultsTo` set to something other - # than `caller`. - - takeFromOtherAnswer @6 :QuestionId; - # The sender has also sent (before this message) a `Call` with the given question ID and with - # `sendResultsTo.yourself` set, and the results of that other call should be used as the - # results here. - - acceptFromThirdParty @7 :ThirdPartyCapId; - # **(level 3)** - # - # The caller should contact a third-party vat to pick up the results. An `Accept` message - # sent to the vat will return the result. This pairs with `Call.sendResultsTo.thirdParty`. - } -} - -struct Finish { - # **(level 0)** - # - # Message type sent from the caller to the callee to indicate: - # 1) The questionId will no longer be used in any messages sent by the callee (no further - # pipelined requests). - # 2) Any capabilities in the results other than the ones listed below should be implicitly - # released. - # 3) If the call has not returned yet, the caller no longer cares about the result. If nothing - # else cares about the result either (e.g. there are to other outstanding calls pipelined on - # the result of this one) then the callee may wish to immediately cancel the operation and - # send back a Return message with "canceled" set. However, implementations are not requried - # to support premature cancellation -- instead, the implementation may wait until the call - # actually completes and send a normal `Return` message. - # - # TODO(someday): Should we separate (1) and (2)? It would be possible and useful to notify the - # server that it doesn't need to keep around the response to service pipeline requests even - # though the caller still wants to receive it / hasn't yet finished processing it. It could - # also be useful to notify the server that it need not marshal the results because the caller - # doesn't want them anyway, even if the caller is still sending pipelined calls, although this - # seems less useful (just saving some bytes on the wire). - - questionId @0 :QuestionId; - # ID of the call whose result is to be released. - - releaseResultCaps @1 :Bool = true; - # If true, all capabilities that were in the results should be considered released. The sender - # must not send separate `Release` messages for them. Level 0 implementations in particular - # should always set this true. This defaults true because if level 0 implementations forgot to - # set it they'd never notice (just silently leak caps), but if level >=1 implementations forget - # set it false they'll quickly get errors. -} - -# Level 1 message types ---------------------------------------------- - -struct Resolve { - # **(level 1)** - # - # Message type sent to indicate that a previously-sent promise has now been resolved to some other - # object (possibly another promise) -- or broken, or canceled. - # - # Keep in mind that it's possible for a `Resolve` to be sent to a level 0 implementation that - # doesn't implement it. For example, a method call or return might contain a capability in the - # payload. Normally this is fine even if the receiver is level 0, because they will implicitly - # release all such capabilities on return / finish. But if the cap happens to be a promise, then - # a follow-up `Resolve` will be sent regardless of this release. The level 0 receiver will reply - # with an `unimplemented` message. The sender (of the `Resolve`) can respond to this as if the - # receiver had immediately released any capability to which the promise resolved. - - promiseId @0 :ExportId; - # The ID of the promise to be resolved. - # - # Unlike all other instances of `ExportId` sent from the exporter, the `Resolve` message does - # _not_ increase the reference count of `promiseId`. - # - # When an export ID sent over the wire (e.g. in a `CapDescriptor`) is indicated to be a promise, - # this indicates that the sender will follow up at some point with a `Resolve` message. If the - # same `promiseId` is sent again before `Resolve`, still only one `Resolve` is sent. If the - # same ID is sent again later _after_ a `Resolve`, it can only be because the export's - # reference count hit zero in the meantime and the ID was re-assigned to a new export, therefore - # this later promise does _not_ correspond to the earlier `Resolve`. - # - # If a promise ID's reference count reaches zero before a `Resolve` is sent, the `Resolve` - # message may or may not still be sent (in particular, the `Resolve` may have already been - # in-flight when `Release` was sent). Thus a `Resolve` may be received for a promise of which - # the receiver has no knowledge, because it already released it earlier. In this case, the - # receiver should immediately release the capability to which the promise resolved, if - # applicable. - - union { - cap @1 :CapDescriptor; - # The object to which the promise resolved. - # - # The sender promises that from this point forth, until `promiseId` is released, it shall - # simply forward all messages to the capability designated by `cap`. This is true even if - # `cap` itself happens to desigate another promise, and that other promise later resolves -- - # messages sent to `promiseId` shall still go to that other promise, not to its resolution. - # This is important in the case that the receiver of the `Resolve` ends up sending a - # `Disembargo` message towards `promiseId` in order to control message ordering -- that - # `Disembargo` really needs to reflect back to exactly the object designated by `cap` even - # if that object is itself a promise. - - exception @2 :Exception; - # Indicates that the promise was broken. - } -} - -struct Release { - # **(level 1)** - # - # Message type sent to indicate that the sender is done with the given capability and the receiver - # can free resources allocated to it. - - id @0 :ExportId; - # What to release. - - referenceCount @1 :UInt32; - # The amount by which to decrement the reference count. The export is only actually released - # when the reference count reaches zero. -} - -struct Disembargo { - # **(level 1)** - # - # Message sent to indicate that an embargo on a recently-resolved promise may now be lifted. - # - # Embargos are used to enforce E-order in the presence of promise resolution. That is, if an - # application makes two calls foo() and bar() on the same capability reference, in that order, - # the calls should be delivered in the order in which they were made. But if foo() is called - # on a promise, and that promise happens to resolve before bar() is called, then the two calls - # may travel different paths over the network, and thus could arrive in the wrong order. In - # this case, the call to `bar()` must be embargoed, and a `Disembargo` message must be sent along - # the same path as `foo()` to ensure that the `Disembargo` arrives after `foo()`. Once the - # `Disembargo` arrives, `bar()` can then be delivered. - # - # There are two particular cases where embargos are important. Consider object Alice, in Vat A, - # who holds a promise P, pointing towards Vat B, which eventually resolves to Carol. The two - # cases are: - # - Carol lives in Vat A, i.e. next to Alice. In this case, Vat A needs to send a `Disembargo` - # message that echos through Vat B and back, to ensure that all pipelined calls on the promise - # have been delivered. - # - Carol lives in a different Vat C. When the promise resolves, a three-party handoff occurs - # (see `Provide` and `Accept`, which constitute level 3 of the protocol). In this case, we - # piggyback on the state that has already been set up to handle the handoff: the `Accept` - # message (from Vat A to Vat C) is embargoed, as are all pipelined messages sent to it, while - # a `Disembargo` message is sent from Vat A through Vat B to Vat C. See `Accept.embargo` for - # an example. - # - # Note that in the case where Carol actually lives in Vat B (i.e., the same vat that the promise - # already pointed at), no embargo is needed, because the pipelined calls are delivered over the - # same path as the later direct calls. - # - # An alternative strategy for enforcing E-order over promise resolution could be for Vat A to - # implement the embargo internally. When Vat A is notified of promise resolution, it could - # send a dummy no-op call to promise P and wait for it to complete. Until that call completes, - # all calls to the capability are queued locally. This strategy works, but is pessimistic: - # in the three-party case, it requires an A -> B -> C -> B -> A round trip before calls can start - # being delivered directly to from Vat A to Vat C. The `Disembargo` message allows latency to be - # reduced. (In the two-party loopback case, the `Disembargo` message is just a more explicit way - # of accomplishing the same thing as a no-op call, but isn't any faster.) - - target @0 :MessageTarget; - # What is to be disembargoed. - - using EmbargoId = UInt32; - # Used in `senderLoopback` and `receiverLoopback`, below. - - context :union { - senderLoopback @1 :EmbargoId; - # The sender is requesting a disembargo on a promise which is known to resolve back to a - # capability hoste by the sender. As soon as the receiver has echoed back all pipelined calls - # on this promise, it will deliver the Disembargo back to the sender with `receiverLoopback` - # set to the same value as `senderLoopback`. This value is chosen by the sender, and since - # it is also consumed be the sender, the sender can use whatever strategy it wants to make sure - # the value is unambiguous. - # - # The receiver must verify that the target capability actually resolves back to the sender's - # vat. Otherwise, the sender has committed a protocol error and should be disconnected. - - receiverLoopback @2 :EmbargoId; - # The receiver previously sent a `senderLoopback` Disembargo towards a promise resolving to - # this capability, and that Disembargo is now being echoed back. - - accept @3 :Void; - # **(level 3)** - # - # The sender is requesting a disembargo on a promise which is known to resolve to a third-party - # capability which the sender is currently in the process of accepting (using `Accept`). - # The receiver of this `Disembargo` has an outstanding `Provide` on said capability. The - # receiver should now send a `Disembargo` with `provide` set to the question ID of that - # `Provide` message. - # - # See `Accept.embargo` for an example. - - provide @4 :QuestionId; - # **(level 3)** - # - # The sender is requesting a disembargo on a capability currently being provided to a third - # party. The question ID identifies the `Provide` message previously sent by the sender to - # this capability. On receipt, the receiver (the capability host) shall release the embargo - # on the `Accept` message that it has received from the third party. See `Accept.embargo` for - # an example. - } -} - -# Level 2 message types ---------------------------------------------- - -struct Save { - # **(level 2)** - # - # Message type sent to save a capability persistently so that it can be restored by a future - # connection. Not all capabilities can be saved -- application interfaces should define which - # capabilities support this and which do not. - - questionId @0 :QuestionId; - # A new question ID identifying this request, which will eventually receive a Return - # message whose `results` is a SturdyRef. - - target @1 :MessageTarget; - # What is to be saved. -} - -struct Restore { - # **(mostly level 2)** - # - # Message type sent to restore a persistent capability obtained during a previous connection, or - # through other means. - # - # Level 0/1 implementations need to implement a limited version of `Restore` only for the purpose - # of bootstrapping a new connection (otherwise, there would be no objects to which to address - # methods). These levels may simply implement public singleton services that exist for the - # lifetime of the host process and probably have non-secret names. A level 0 receiver of - # `Restore` should never actually send a `Return` message, but should simply expect `Call` - # messages addressed to the `PromisedAnswer` corresponding to the `Restore`. A level 0 sender - # of `Restore` can ignore the corresponding `Return` and just keep addressing the - # `PromisedAnswer`. - - questionId @0 :QuestionId; - # A new question ID identifying this request, which will eventually receive a Return message - # containing the restored capability. - - objectId @1 :SturdyRefObjectId; - # Designates the capability to restore. -} - -struct Delete { - # **(level 2)** - # - # Message type sent to delete a previously-saved persistent capability. In other words, this - # means "this ref will no longer be used in the future", so that the host can potentially - # garbage collect resources associated with it. Note that if any ExportId still refers to a - # capability restored from this ref, that export should still remain valid until released. - # - # Different applications may define different policies regarding saved capability lifetimes that - # may or may not rely on `Delete`. For the purpose of implementation freedom, a receiver is - # allowed to silently ignore a delete request for a reference it doesn't recognize. This way, - # a persistent capability could be given an expiration time, after which the capability is - # automatically deleted, and any future `Delete` message is ignored. - # - # A client must send no more than one `Delete` message for any given `Save`, so that a host - # can potentially implement reference counting. However, hosts should be wary of reference - # counting across multiple clients, as a malicious client could of course send multiple - # `Delete`s. - - questionId @0 :QuestionId; - # A new question ID identifying this request, which will eventually receive a Return message - # with an empty (null) result. - - objectId @1 :SturdyRefObjectId; - # Designates the capability to delete. -} - -# Level 3 message types ---------------------------------------------- - -struct Provide { - # **(level 3)** - # - # Message type sent to indicate that the sender wishes to make a particular capability implemented - # by the receiver available to a third party for direct access (without the need for the third - # party to proxy through the sender). - # - # (In CapTP, `Provide` and `Accept` are methods of the global `NonceLocator` object exported by - # every vat. In Cap'n Proto, we bake this into the core protocol.) - - questionId @0 :QuestionId; - # Question ID to be held open until the recipient has received the capability. A result will - # be returned once the third party has successfully received the capability. The sender must - # at some point send a `Finish` message as with any other call, and such a message can be - # used to cancel the whole operation. - - target @1 :MessageTarget; - # What is to be provided to the third party. - - recipient @2 :RecipientId; - # Identity of the third party which is expected to pick up the capability. -} - -struct Accept { - # **(level 3)** - # - # Message type sent to pick up a capability hosted by the receiving vat and provided by a third - # party. The third party previously designated the capability using `Provide`. - # - # This message is also used to pick up a redirected return -- see `Return.redirect`. - - questionId @0 :QuestionId; - # A new question ID identifying this accept message, which will eventually receive a Return - # message containing the provided capability (or the call result in the case of a redirected - # return). - - provision @1 :ProvisionId; - # Identifies the provided object to be picked up. - - embargo @2 :Bool; - # If true, this accept shall be temporarily embargoed. The resulting `Return` will not be sent, - # and any pipelined calls will not be delivered, until the embargo is released. The receiver - # (the capability host) will expect the provider (the vat that sent the `Provide` message) to - # eventually send a `Disembargo` message with the field `context.provide` set to the question ID - # of the original `Provide` message. At that point, the embargo is released and the queued - # messages are delivered. - # - # For example: - # - Alice, in Vat A, holds a promise P, which currently points toward Vat B. - # - Alice calls foo() on P. The `Call` message is sent to Vat B. - # - The promise P in Vat B ends up resolving to Carol, in Vat C. - # - Vat B sends a `Provide` message to Vat C, identifying Vat A as the recipient. - # - Vat B sends a `Resolve` message to Vat A, indicating that the promise has resolved to a - # `ThirdPartyCapId` identifying Carol in Vat C. - # - Vat A sends an `Accept` message to Vat C to pick up the capability. Since Vat A knows that - # it has an outstanding call to the promise, it sets `embargo` to `true` in the `Accept` - # message. - # - Vat A sends a `Disembargo` message to Vat B on promise P, with `context.accept` set. - # - Alice makes a call bar() to promise P, which is now pointing towards Vat C. Alice doesn't - # know anything about the mechanics of promise resolution happening under the hood, but she - # expects that bar() will be delivered after foo() because that is the order in which she - # initiated the calls. - # - Vat A sends the bar() call to Vat C, as a pipelined call on the result of the `Accept` (which - # hasn't returned yet, due to the embargo). Since calls to the newly-accepted capability - # are embargoed, Vat C does not deliver the call yet. - # - At some point, Vat B forwards the foo() call from the beginning of this example on to Vat C. - # - Vat B forwards the `Disembargo` from Vat A on to vat C. It sets `context.provide` to the - # question ID of the `Provide` message it had sent previously. - # - Vat C receives foo() before `ReleaseEmbargo`, thus allowing it to correctly deliver foo() - # before delivering bar(). - # - Vat C receives `ReleaseEmbargo` from Vat B. It can now send a `Return` for the `Accept` from - # Vat A, as well as deliver bar(). -} - -# Level 4 message types ---------------------------------------------- - -struct Join { - # **(level 4)** - # - # Message type sent to implement E.join(), which, given a number of capabilities which are - # expected to be equivalent, finds the underlying object upon which they all agree and forms a - # direct connection to it, skipping any proxies which may have been constructed by other vats - # while transmitting the capability. See: - # http://erights.org/elib/equality/index.html - # - # Note that this should only serve to bypass fully-transparent proxies -- proxies that were - # created merely for convenience, without any intention of hiding the underlying object. - # - # For example, say Bob holds two capabilities hosted by Alice and Carol, but he expects that both - # are simply proxies for a capability hosted elsewhere. He then issues a join request, which - # operates as follows: - # - Bob issues Join requests on both Alice and Carol. Each request contains a different piece - # of the JoinKey. - # - Alice is proxying a capability hosted by Dana, so forwards the request to Dana's cap. - # - Dana receives the first request and sees that the JoinKeyPart is one of two. She notes that - # she doesn't have the other part yet, so she records the request and responds with a - # JoinResult. - # - Alice relays the JoinAswer back to Bob. - # - Carol is also proxying a capability from Dana, and so forwards her Join request to Dana as - # well. - # - Dana receives Carol's request and notes that she now has both parts of a JoinKey. She - # combines them in order to form information needed to form a secure connection to Bob. She - # also responds with another JoinResult. - # - Bob receives the responses from Alice and Carol. He uses the returned JoinResults to - # determine how to connect to Dana and attempts to form the connection. Since Bob and Dana now - # agree on a secret key which neither Alice nor Carol ever saw, this connection can be made - # securely even if Alice or Carol is conspiring against the other. (If Alice and Carol are - # conspiring _together_, they can obviously reproduce the key, but this doesn't matter because - # the whole point of the join is to verify that Alice and Carol agree on what capability they - # are proxying.) - # - # If the two capabilities aren't actually proxies of the same object, then the join requests - # will come back with conflicting `hostId`s and the join will fail before attempting to form any - # connection. - - questionId @0 :QuestionId; - # Question ID used to respond to this Join. (Note that this ID only identifies one part of the - # request for one hop; each part has a different ID and relayed copies of the request have - # (probably) different IDs still.) - # - # The receiver will reply with a `Return` whose `results` is a JoinResult. This `JoinResult` - # is relayed from the joined object's host, possibly with transformation applied as needed - # by the network. - # - # Like any return, the result must be released using a `Finish`. However, this release - # should not occur until the joiner has either successfully connected to the joined object. - # Vats relaying a `Join` message similarly must not release the result they receive until the - # return they relayed back towards the joiner has itself been released. This allows the - # joined object's host to detect when the Join operation is canceled before completing -- if - # it receives a `Finish` for one of the join results before the joiner successfully - # connects. It can then free any resources it had allocated as part of the join. - - capId @1 :ExportId; - # The capability to join. - - keyPart @2 :JoinKeyPart; - # A part of the join key. These combine to form the complete join key which is used to establish - # a direct connection. - - # TODO(before implementing): Change this so that multiple parts can be sent in a single Join - # message, so that if multiple join parts are going to cross the same connection they can be sent - # together, so that the receive can potentially optimize its handling of them. In the case where - # all parts are bundled together, should the recipient be expected to simply return a cap, so - # that the caller can immediately start pipelining to it? -} - -# ======================================================================================== -# Common structures used in messages - -struct MessageTarget { - # The target of a `Call` or other messages that target a capability. - - union { - exportedCap @0 :ExportId; - # This message is to a capability or promise previously exported by the receiver. - - promisedAnswer @1 :PromisedAnswer; - # This message is to a capability that is expected to be returned by another call that has not - # yet been completed. - # - # At level 0, this is supported only for addressing the result of a previous `Restore`, so that - # initial startup doesn't require a round trip. - } -} - -struct Payload { - # Represents some data structure that might contain capabilities. - - content @0 :AnyPointer; - # Some Cap'n Proto data structure. Capability pointers embedded in this structure index into - # `capTable`. - - capTable @1 :List(CapDescriptor); - # Descriptors corresponding to the cap pointers in `content`. -} - -struct CapDescriptor { - # **(level 1)** - # - # When an application-defined type contains an interface pointer, that pointer's encoding is the - # same as a struct pointer except that the bottom two bits are 1's instead of 0's. The pointer - # actually points to an instance of `CapDescriptor`. The runtime API should not reveal the - # CapDescriptor directly to the application, but should instead wrap it in some kind of callable - # object with methods corresponding to the interface that the capability implements. - # - # Keep in mind that `ExportIds` in a `CapDescriptor` are subject to reference counting. See the - # description of `ExportId`. - - union { - none @0 :Void; - # There is no capability here. This `CapDescriptor` should not appear in the payload content. - # A `none` CapDescriptor can be generated when an application inserts a capability into a - # message and then later changes its mind and removes it -- rewriting all of the other - # capability pointers may be hard, so instead a tombstone is left, similar to the way a removed - # struct or list instance is zeroed out of the message but the space is not reclaimed. - # Hopefully this is unusual. - - senderHosted @1 :ExportId; - # A capability newly exported by the sender. This is the ID of the new capability in the - # sender's export table (receiver's import table). - - senderPromise @2 :ExportId; - # A promise which the sender will resolve later. The sender will send exactly one Resolve - # message at a future point in time to replace this promise. Note that even if the same - # `senderPromise` is received multiple times, only one `Resolve` is sent to cover all of - # them. If `senderPromise` is released before the `Resolve` is sent, the sender (of this - # `CapDescriptor`) may choose not to send the `Resolve` at all. - - receiverHosted @3 :ExportId; - # A capability (or promise) previously exported by the receiver. - - receiverAnswer @4 :PromisedAnswer; - # A capability expected to be returned in the results of a currently-outstanding call posed - # by the sender. - - thirdPartyHosted @5 :ThirdPartyCapDescriptor; - # **(level 3)** - # - # A capability that lives in neither the sender's nor the receiver's vat. The sender needs - # to form a direct connection to a third party to pick up the capability. - # - # Level 1 and 2 implementations that receive a `thirdPartyHosted` may simply send calls to its - # `vine` instead. - } -} - -struct PromisedAnswer { - # **(mostly level 1)** - # - # Specifies how to derive a promise from an unanswered question, by specifying the path of fields - # to follow from the root of the eventual result struct to get to the desired capability. Used - # to address method calls to a not-yet-returned capability or to pass such a capability as an - # input to some other method call. - # - # Level 0 implementations must support `PromisedAnswer` only for the case where the answer is - # to a `Restore` message. In this case, `path` is always empty since `Restore` always returns - # a raw capability. - - questionId @0 :QuestionId; - # ID of the question (in the sender's question table / receiver's answer table) whose answer is - # expected to contain the capability. - - transform @1 :List(Op); - # Operations / transformations to apply to the result in order to get the capability actually - # being addressed. E.g. if the result is a struct and you want to call a method on a capability - # pointed to by a field of the struct, you need a `getPointerField` op. - - struct Op { - union { - noop @0 :Void; - # Does nothing. This member is mostly defined so that we can make `Op` a union even - # though (as of this writing) only one real operation is defined. - - getPointerField @1 :UInt16; - # Get a pointer field within a struct. The number is an index into the pointer section, NOT - # a field ordinal, so that the receiver does not need to understand the schema. - - # TODO(someday): We could add: - # - For lists, the ability to address every member of the list, or a slice of the list, the - # result of which would be another list. This is useful for implementing the equivalent of - # a SQL table join (not to be confused with the `Join` message type). - # - Maybe some ability to test a union. - # - Probably not a good idea: the ability to specify an arbitrary script to run on the - # result. We could define a little stack-based language where `PathPart` specifies one - # "instruction" or transformation to apply. Although this is not a good idea - # (over-engineered), any narrower additions to `PathPart` should be designed as if this - # were the eventual goal. - } - } -} - -struct SturdyRef { - # **(level 2)** - # - # A combination of a SturdyRefObjectId and SturdyRefHostId. This is what a client of the ref - # would typically save in its own storage. This type is also the result of a `Save` message. - - hostId @0 :SturdyRefHostId; - # Describes how to connect to and authenticate a vat that hosts this SturdyRef (and can therefore - # accept a `Restore` message for it). - - objectId @1 :SturdyRefObjectId; - # The opaque ref in the scope of the host vat, to be sent in the `Restore` message. -} - -struct ThirdPartyCapDescriptor { - # **(level 3)** - # - # Identifies a capability in a third-party vat which the sender wants the receiver to pick up. - - id @0 :ThirdPartyCapId; - # Identifies the third-party host and the specific capability to accept from it. - - vineId @1 :ExportId; - # A proxy for the third-party object exported by the sender. In CapTP terminology this is called - # a "vine", because it is an indirect reference to the third-party object that snakes through the - # sender vat. This serves two purposes: - # - # * Level 1 and 2 implementations that don't understand how to connect to a third party may - # simply send calls to the vine. Such calls will be forwarded to the third-party by the - # sender. - # - # * Level 3 implementations must release the vine once they have successfully picked up the - # object from the third party. This ensures that the capability is not released by the sender - # prematurely. - # - # The sender will close the `Provide` request that it has sent to the third party as soon as - # it receives either a `Call` or a `Release` message directed at the vine. -} - -struct Exception { - # **(level 0)** - # - # Describes an arbitrary error that prevented an operation (e.g. a call) from completing. - - reason @0 :Text; - # Human-readable failure description. - - isCallersFault @1 :Bool; - # In the best estimate of the error source, is it the caller's fault that this error occurred - # (like HTTP 400), or is it the callee's fault (like HTTP 500)? Or, put another way, if an - # automated bug report were to be generated for this error, should it be initially filed on the - # caller's code or the callee's? This is a guess. Generally guesses should err towards blaming - # the callee -- at the very least, the callee should be on the hook for improving their error - # handling to be more confident in assigning blame. - - durability @2 :Durability; - # In the best estimate of the error source, is this error likely to repeat if the same call is - # executed again? Callers might use this to decide when to retry a request. - - enum Durability { - permanent @0; # Retrying the exact same operation will fail in the same way. - temporary @1; # Retrying the exact same operation might succeed. - overloaded @2; # The error may be due to the system being overloaded. Retrying may work - # later on, but for now the caller should not retry right away as this will - # likely exacerbate the problem. - } -} - -# ======================================================================================== -# Network-specific Parameters -# -# Some parts of the Cap'n Proto RPC protocol are not specified here because different vat networks -# may wish to use different approaches to solving them. For example, on the public internet, you -# may want to authenticate vats using public-key cryptography, but on a local intranet with trusted -# infrastructure, you may be happy to authenticate based on network address only, or some other -# lightweight mechanism. -# -# To accommodate this, we specify several "parameter" types. Each type is defined here as an -# alias for `AnyPointer`, but a specific network will want to define a specific set of types to use. -# All vats in a vat network must agree on these parameters in order to be able to communicate. -# Inter-network communication can be accomplished through "gateways" that perform translation -# between the primitives used on each network; these gateways may need to be deeply stateful, -# depending on the translations they perform. -# -# For interaction over the global internet between parties with no other prior arrangement, a -# particular set of bindings for these types is defined elsewhere. (TODO(someday): Specify where -# these common definitions live.) -# -# Another common network type is the two-party network, in which one of the parties typically -# interacts with the outside world entirely through the other party. In such a connection between -# Alice and Bob, all objects that exist on Bob's other networks appear to Alice as if they were -# hosted by Bob himself, and similarly all objects on Alice's network (if she even has one) appear -# to Bob as if they were hosted by Alice. This network type is interesting because from the point -# of view of a simple application that communicates with only one other party via the two-party -# protocol, there are no three-party interactions at all, and joins are unusually simple to -# implement, so implementing at level 4 is barely more complicated than implementing at level 1. -# Moreover, if you pair an app implementing the two-party network with a container that implements -# some other network, the app can then participate on the container's network just as if it -# implemented that network directly. The types used by the two-party network are defined in -# `rpc-twoparty.capnp`. -# -# The things which we need to parameterize are: -# - How to store capabilities long-term without holding a connection open (mostly level 2). -# - How to authenticate vats in three-party introductions (level 3). -# - How to implement `Join` (level 4). -# -# Persistent references -# --------------------- -# -# **(mostly level 2)** -# -# We want to allow some capabilities to be stored long-term, even if a connection is lost and later -# recreated. ExportId is a short-term identifier that is specific to a connection, so it doesn't -# help here. We need a way to specify long-term identifiers, as well as a strategy for -# reconnecting to a referenced capability later. -# -# Three-party interactions -# ------------------------ -# -# **(level 3)** -# -# In cases where more than two vats are interacting, we have situations where VatA holds a -# capability hosted by VatB and wants to send that capability to VatC. This can be accomplished -# by VatA proxying requests on the new capability, but doing so has two big problems: -# - It's inefficient, requiring an extra network hop. -# - If VatC receives another capability to the same object from VatD, it is difficult for VatC to -# detect that the two capabilities are really the same and to implement the E "join" operation, -# which is necessary for certain four-or-more-party interactions, such as the escrow pattern. -# See: http://www.erights.org/elib/equality/grant-matcher/index.html -# -# Instead, we want a way for VatC to form a direct, authenticated connection to VatB. -# -# Join -# ---- -# -# **(level 4)** -# -# The `Join` message type and corresponding operation arranges for a direct connection to be formed -# between the joiner and the host of the joined object, and this connection must be authenticated. -# Thus, the details are network-dependent. - -using SturdyRefHostId = AnyPointer; -# **(level 2)** -# -# Identifies the host of a persistent capability which can be restored using a `Restore` message. -# That is, this identifies where the `Restore` message should be sent, but does not provide any -# part of the `Restore` message's content. `SturdyRefHostId` is usually paired with a -# `SturdyRefObjectId`, often in the form of a `SturdyRef`. -# -# `SturdyRefHostId` could be as simple as a network address and public key fingerprint. Or, it -# might be more complicated or abstract. For example, on some kinds of networks, `SturdyRefHostId` -# might be an abstract service name without any information on where that service is physically -# located; the network itself might provide a separate service for mapping such names to locations. -# It could even be the case that a particular service name maps to a group of vats, where any vat -# in the group is able to restore the ref. Such an approach would make `SturdyRefHostId`s more -# robust against changes in network topology. - -using SturdyRefObjectId = AnyPointer; -# **(mostly level 2)** -# -# A SturdyRefObjectId identifies a persistent object which may be restored later, within the scope -# of some host. The contents of a SturdyRefObjectId are entirely determined by the vat that hosts -# it. In fact, different vats on the same network may actually use different definitions for -# SturdyRefObjectId, so SturdyRefObjectId is not actually parameterized per-network but rather -# per-vat. A SturdyRefObjectId is typically paired with a `SturdyRefHostId` (in a -# `SturdyRef`) which describes how to find a vat capable of restoring the ref. - -using ProvisionId = AnyPointer; -# **(level 3)** -# -# The information which must be sent in an `Accept` message to identify the object being accepted. -# -# In a network where each vat has a public/private key pair, this could simply be the public key -# fingerprint of the provider vat along with the questionId used in the `Provide` message sent from -# that provider. - -using RecipientId = AnyPointer; -# **(level 3)** -# -# The information which must be sent in a `Provide` message to identify the recipient of the -# capability. -# -# In a network where each vat has a public/private key pair, this could simply be the public key -# fingerprint of the recipient. (CapTP also calls for a nonce to identify the object. In our -# case, the `Provide` message's `questionId` can serve as the nonce.) - -using ThirdPartyCapId = AnyPointer; -# **(level 3)** -# -# The information needed to connect to a third party and accept a capability from it. -# -# In a network where each vat has a public/private key pair, this could be a combination of the -# third party's public key fingerprint, hints on how to connect to the third party (e.g. an IP -# address), and the question ID used in the corresponding `Provide` mesasge sent to that third party -# (used to identify which capability to pick up). - -using JoinKeyPart = AnyPointer; -# **(level 4)** -# -# A piece of a secret key. One piece is sent along each path that is expected to lead to the same -# place. Once the pieces are combined, a direct connection may be formed between the sender and -# the receiver, bypassing any men-in-the-middle along the paths. See the `Join` message type. -# -# The motivation for Joins is discussed under "Supporting Equality" in the "Unibus" protocol -# sketch: http://www.erights.org/elib/distrib/captp/unibus.html -# -# In a network where each vat has a public/private key pair and each vat forms no more than one -# connection to each other vat, Joins will rarely -- perhaps never -- be needed, as objects never -# need to be transparently proxied and references to the same object sent over the same connection -# have the same export ID. Thus, a successful join requires only checking that the two objects -# come from the same connection and have the same ID, and then completes immediately. -# -# However, in networks where two vats may form more than one connection between each other, or -# where proxying of objects occurs, joins are necessary. -# -# Typically, each JoinKeyPart would include a fixed-length data value such that all value parts -# XOR'd together forms a shared secret which can be used to form an encrypted connection between -# the joiner and the joined object's host. Each JoinKeyPart should also include an indication of -# how many parts to expect and a hash of the shared secret (used to match up parts). - -using JoinResult = AnyPointer; -# **(level 4)** -# -# Information returned as the result to a `Join` message, needed by the joiner in order to form a -# direct connection to a joined object. This might simply be the address of the joined object's -# host vat, since the `JoinKey` has already been communicated so the two vats already have a shared -# secret to use to authenticate each other. -# -# The `JoinResult` should also contain information that can be used to detect when the Join -# requests ended up reaching different objects, so that this situation can be detected easily. -# This could be a simple matter of including a sequence number -- if the joiner receives two -# `JoinResult`s with sequence number 0, then they must have come from different objects and the -# whole join is a failure. - -# ======================================================================================== -# Network interface sketch -# -# The interfaces below are meant to be pseudo-code to illustrate how the details of a particular -# vat network might be abstracted away. They are written like Cap'n Proto interfaces, but in -# practice you'd probably define these interfaces manually in the target programming language. A -# Cap'n Proto RPC implementation should be able to use these interfaces without knowing the -# definitions of the various network-specific parameters defined above. - -# interface VatNetwork { -# # Represents a vat network, with the ability to connect to particular vats and receive -# # connections from vats. -# # -# # Note that methods returning a `Connection` may return a pre-existing `Connection`, and the -# # caller is expected to find and share state with existing users of the connection. -# -# # Level 0 features ----------------------------------------------- -# -# connectToRefHost(hostId :SturdyRefHostId) :Connection; -# # Connect to the given SturdyRef host. The transport should return a promise which does not -# # resolve until authentication has completed, but allows messages to be pipelined in before -# # that; the transport either queues these messages until authenticated, or sends them encrypted -# # such that only the authentic vat would be able to decrypt them. The latter approach avoids a -# # round trip for authentication. -# # -# # Once connected, the caller should start by sending a `Restore` message. -# -# acceptConnectionAsRefHost() :Connection; -# # Wait for the next incoming connection and return it. Only connections formed by -# # connectToHostOf() are returned by this method. -# # -# # Once connected, the first received message will usually be a `Restore`. -# -# # Level 4 features ----------------------------------------------- -# -# newJoiner(count :UInt32) :NewJoinerResponse; -# # Prepare a new Join operation, which will eventually lead to forming a new direct connection -# # to the host of the joined capability. `count` is the number of capabilities to join. -# -# struct NewJoinerResponse { -# joinKeyParts :List(JoinKeyPart); -# # Key parts to send in Join messages to each capability. -# -# joiner :Joiner; -# # Used to establish the final connection. -# } -# -# interface Joiner { -# addJoinResult(result :JoinResult) :Void; -# # Add a JoinResult received in response to one of the `Join` messages. All `JoinResult`s -# # returned from all paths must be added before trying to connect. -# -# connect() :ConnectionAndProvisionId; -# # Try to form a connection to the joined capability's host, verifying that it has received -# # all of the JoinKeyParts. Once the connection is formed, the caller should send an `Accept` -# # message on it with the specified `ProvisionId` in order to receive the final capability. -# } -# -# acceptConnectionFromJoiner(parts :List(JoinKeyPart), paths :List(VatPath)) -# :ConnectionAndProvisionId; -# # Called on a joined capability's host to receive the connection from the joiner, once all -# # key parts have arrived. The caller should expect to receive an `Accept` message over the -# # connection with the given ProvisionId. -# } -# -# interface Connection { -# # Level 0 features ----------------------------------------------- -# -# send(message :Message) :Void; -# # Send the message. Returns successfully when the message (and all preceding messages) has -# # been acknowledged by the recipient. -# -# receive() :Message; -# # Receive the next message, and acknowledges receipt to the sender. Messages are received in -# # the order in which they are sent. -# -# # Level 3 features ----------------------------------------------- -# -# introduceTo(recipient :Connection) :IntroductionInfo; -# # Call before starting a three-way introduction, assuming a `Provide` message is to be sent on -# # this connection and a `ThirdPartyCapId` is to be sent to `recipient`. -# -# struct IntroductionInfo { -# sendToRecipient :ThirdPartyCapId; -# sendToTarget :RecipientId; -# } -# -# connectToIntroduced(capId :ThirdPartyCapId) :ConnectionAndProvisionId; -# # Given a ThirdPartyCapId received over this connection, connect to the third party. The -# # caller should then send an `Accept` message over the new connection. -# -# acceptIntroducedConnection(recipientId :RecipientId) :Connection; -# # Given a RecipientId received in a `Provide` message on this `Connection`, wait for the -# # recipient to connect, and return the connection formed. Usually, the first message received -# # on the new connection will be an `Accept` message. -# } -# -# struct ConnectionAndProvisionId { -# # **(level 3)** -# -# connection :Connection; -# # Connection on which to issue `Accept` message. -# -# provision :ProvisionId; -# # `ProvisionId` to send in the `Accept` message. -# } diff --git a/test/test_rpc.py b/test/test_rpc.py index fc58843..03ab995 100644 --- a/test/test_rpc.py +++ b/test/test_rpc.py @@ -12,17 +12,60 @@ class Server(test_capability_capnp.TestInterface.Server): def foo(self, i, j, **kwargs): return str(i * 5 + self.val) -def test_simple_rpc(): - def _restore(ref_id): +class TypelessRestorer: + def restore(self, ref_id): return Server(100) +def restore_func(ref_id): + return Server(100) + +class SimpleRestorer(test_capability_capnp.TestSturdyRefObjectId.Restorer): + def restore(self, ref_id): + assert ref_id.tag == 'testInterface' + return Server(100) + +def test_simple_rpc(): + read, write = socket.socketpair(socket.AF_UNIX) - restorer = capnp.Restorer(test_capability_capnp.TestSturdyRefObjectId, _restore) - server = capnp.RpcServer(write, restorer) - client = capnp.RpcClient(read) + restorer = SimpleRestorer() + server = capnp.TwoPartyServer(write, restorer) + client = capnp.TwoPartyClient(read) - ref = test_capability_capnp.TestSturdyRefObjectId.new_message() + ref = test_capability_capnp.TestSturdyRefObjectId.new_message(tag='testInterface') + cap = client.restore(ref) + cap = cap.cast_as(test_capability_capnp.TestInterface) + + remote = cap.foo(i=5) + response = remote.wait() + + assert response.x == '125' + +def test_simple_rpc_typeless_restorer(): + + read, write = socket.socketpair(socket.AF_UNIX) + + restorer = TypelessRestorer() + server = capnp.TwoPartyServer(write, restorer) + client = capnp.TwoPartyClient(read) + + ref = test_capability_capnp.TestSturdyRefObjectId.new_message(tag='testInterface') + cap = client.restore(ref) + cap = cap.cast_as(test_capability_capnp.TestInterface) + + remote = cap.foo(i=5) + response = remote.wait() + + assert response.x == '125' + +def test_simple_rpc_restore_func(): + + read, write = socket.socketpair(socket.AF_UNIX) + + server = capnp.TwoPartyServer(write, restore_func) + client = capnp.TwoPartyClient(read) + + ref = test_capability_capnp.TestSturdyRefObjectId.new_message(tag='testInterface') cap = client.restore(ref) cap = cap.cast_as(test_capability_capnp.TestInterface)