diff --git a/.github/workflows/wheels.yml b/.github/workflows/wheels.yml index 76d1fd4..57fb23d 100644 --- a/.github/workflows/wheels.yml +++ b/.github/workflows/wheels.yml @@ -60,7 +60,7 @@ jobs: # TODO: Disable building PyPy wheels. If the build system gets modernized, this should be # auto-detected based on the Cython dependency. CIBW_SKIP: pp* - CIBW_TEST_REQUIRES: pytest + CIBW_TEST_REQUIRES: pytest pytest-asyncio CIBW_TEST_COMMAND: pytest {project} # Only needed to make the macosx arm64 build work CMAKE_OSX_ARCHITECTURES: "${{ matrix.arch == 'arm64' && 'arm64' || '' }}" diff --git a/benchmark/bin/run_all.py b/benchmark/bin/run_all.py index c3458d9..36a7930 100755 --- a/benchmark/bin/run_all.py +++ b/benchmark/bin/run_all.py @@ -10,28 +10,58 @@ import time _this_dir = os.path.dirname(__file__) + def parse_args(): parser = argparse.ArgumentParser() - parser.add_argument('-l', "--langs", help="Add languages to test, ie: -l pyproto -l pyproto_cpp", action='append', default=['pycapnp']) - parser.add_argument("-r", "--reuse", help="If this flag is passed, re-use tests will be run", action='store_true') - parser.add_argument("-c", "--compression", help="If this flag is passed, compression tests will be run", action='store_true') - parser.add_argument("-i", "--scale_iters", help="Scaling factor to multiply the default iters by", type=float, default=1.0) + parser.add_argument( + "-l", + "--langs", + help="Add languages to test, ie: -l pyproto -l pyproto_cpp", + action="append", + default=["pycapnp"], + ) + parser.add_argument( + "-r", + "--reuse", + help="If this flag is passed, re-use tests will be run", + action="store_true", + ) + parser.add_argument( + "-c", + "--compression", + help="If this flag is passed, compression tests will be run", + action="store_true", + ) + parser.add_argument( + "-i", + "--scale_iters", + help="Scaling factor to multiply the default iters by", + type=float, + default=1.0, + ) return parser.parse_args() + def run_one(prefix, name, mode, iters, faster, compression): res_type = prefix - reuse = 'no-reuse' + reuse = "no-reuse" if faster: - reuse = 'reuse' - res_type += '_reuse' - if compression != 'none': - res_type += '_' + compression + reuse = "reuse" + res_type += "_reuse" + if compression != "none": + res_type += "_" + compression - command = [os.path.join(_this_dir, prefix+"-"+name), mode, reuse, compression, str(iters)] + command = [ + os.path.join(_this_dir, prefix + "-" + name), + mode, + reuse, + compression, + str(iters), + ] start = time.time() - print('running: ' + ' '.join(command), file=sys.stderr) + print("running: " + " ".join(command), file=sys.stderr) p = Popen(command, stdout=PIPE, stderr=PIPE) res = p.wait() end = time.time() @@ -39,14 +69,19 @@ def run_one(prefix, name, mode, iters, faster, compression): data = {} if p.returncode != 0: - sys.stderr.write(' '.join(command) + ' failed to run with errors: \n' + p.stderr.read().decode(sys.stdout.encoding) + '\n') + sys.stderr.write( + " ".join(command) + + " failed to run with errors: \n" + + p.stderr.read().decode(sys.stdout.encoding) + + "\n" + ) sys.stderr.flush() - data['type'] = res_type - data['mode'] = mode - data['name'] = name - data['iters'] = iters - data['time'] = end - start + data["type"] = res_type + data["mode"] = mode + data["name"] = name + data["iters"] = iters + data["time"] = end - start return data @@ -55,28 +90,40 @@ def run_each(name, langs, reuse, compression, iters): ret = [] for lang_name in langs: - ret.append(run_one(lang_name, name, 'object', iters, False, 'none')) - ret.append(run_one(lang_name, name, 'bytes', iters, False, 'none')) + ret.append(run_one(lang_name, name, "object", iters, False, "none")) + ret.append(run_one(lang_name, name, "bytes", iters, False, "none")) if reuse: - ret.append(run_one(lang_name, name, 'object', iters, True, 'none')) - ret.append(run_one(lang_name, name, 'bytes', iters, True, 'none')) + ret.append(run_one(lang_name, name, "object", iters, True, "none")) + ret.append(run_one(lang_name, name, "bytes", iters, True, "none")) if compression: - ret.append(run_one(lang_name, name, 'bytes', iters, True, 'packed')) + ret.append(run_one(lang_name, name, "bytes", iters, True, "packed")) if compression: - ret.append(run_one(lang_name, name, 'bytes', iters, False, 'packed')) + ret.append(run_one(lang_name, name, "bytes", iters, False, "packed")) return ret + def main(): args = parse_args() - os.environ['PATH'] += ':.' + os.environ["PATH"] += ":." data = [] - data += run_each('carsales', args.langs, args.reuse, args.compression, int(2000 * args.scale_iters)) - data += run_each('catrank', args.langs, args.reuse, args.compression, int(100 * args.scale_iters)) - data += run_each('eval', args.langs, args.reuse, args.compression, int(10000 * args.scale_iters)) - json.dump(data, sys.stdout, sort_keys=True, indent=4, separators=(',', ': ')) + data += run_each( + "carsales", + args.langs, + args.reuse, + args.compression, + int(2000 * args.scale_iters), + ) + data += run_each( + "catrank", args.langs, args.reuse, args.compression, int(100 * args.scale_iters) + ) + data += run_each( + "eval", args.langs, args.reuse, args.compression, int(10000 * args.scale_iters) + ) + json.dump(data, sys.stdout, sort_keys=True, indent=4, separators=(",", ": ")) -if __name__ == '__main__': + +if __name__ == "__main__": main() diff --git a/benchmark/bin/runner.py b/benchmark/bin/runner.py index 1627bd8..d4654b8 100755 --- a/benchmark/bin/runner.py +++ b/benchmark/bin/runner.py @@ -8,46 +8,83 @@ from timeit import default_timer import random _this_dir = os.path.dirname(__file__) -sys.path.append(os.path.join(_this_dir, '..')) +sys.path.append(os.path.join(_this_dir, "..")) from common import do_benchmark + def parse_args_simple(): parser = argparse.ArgumentParser() - parser.add_argument("mode", help="Mode to use for serialization, ie. object or bytes") + parser.add_argument( + "mode", help="Mode to use for serialization, ie. object or bytes" + ) parser.add_argument("reuse", help="Currently ignored") parser.add_argument("compression", help="Valid values are none or packed") parser.add_argument("iters", help="Number of iterations to run for", type=int) - parser.add_argument("-I", "--includes", help="Directories to add to PYTHONPATH", default='/usr/local/include') + parser.add_argument( + "-I", + "--includes", + help="Directories to add to PYTHONPATH", + default="/usr/local/include", + ) return parser.parse_args() + def parse_args(): parser = argparse.ArgumentParser() - parser.add_argument("name", help="Name of the benchmark to run, eg. carsales", nargs='?', default='carsales') - parser.add_argument("-c", "--compression", help="Specify the compression type", default=None) - parser.add_argument("-s", "--suffix", help="Choose the protocol type.", default='pycapnp') - parser.add_argument("-m", "--mode", help="Specify the mode", default='object') - parser.add_argument("-i", "--iters", help="Specify the number of iterations manually. By default, it will be looked up in preset table", default=10, type=int) - parser.add_argument("-r", "--reuse", help="If this flag is passed, objects will be re-used", action='store_true') - parser.add_argument("-I", "--includes", help="Directories to add to PYTHONPATH", default='/usr/local/include') + parser.add_argument( + "name", + help="Name of the benchmark to run, eg. carsales", + nargs="?", + default="carsales", + ) + parser.add_argument( + "-c", "--compression", help="Specify the compression type", default=None + ) + parser.add_argument( + "-s", "--suffix", help="Choose the protocol type.", default="pycapnp" + ) + parser.add_argument("-m", "--mode", help="Specify the mode", default="object") + parser.add_argument( + "-i", + "--iters", + help="Specify the number of iterations manually. By default, it will be looked up in preset table", + default=10, + type=int, + ) + parser.add_argument( + "-r", + "--reuse", + help="If this flag is passed, objects will be re-used", + action="store_true", + ) + parser.add_argument( + "-I", + "--includes", + help="Directories to add to PYTHONPATH", + default="/usr/local/include", + ) return parser.parse_args() + def run_test(name, mode, reuse, compression, iters, suffix, includes): tic = default_timer() name = name sys.path.append(includes) - module = import_module(name + '_' + suffix) + module = import_module(name + "_" + suffix) benchmark = module.Benchmark(compression=compression) do_benchmark(mode=mode, benchmark=benchmark, iters=iters, reuse=reuse) toc = default_timer() return toc - tic + def main(): args = parse_args() run_test(**vars(args)) -if __name__ == '__main__': - main() \ No newline at end of file + +if __name__ == "__main__": + main() diff --git a/capnp/__init__.py b/capnp/__init__.py index 0482b03..aba313f 100644 --- a/capnp/__init__.py +++ b/capnp/__init__.py @@ -52,7 +52,6 @@ from .lib.capnp import ( _StructModule, _write_message_to_fd, _write_packed_message_to_fd, - _Promise as Promise, _AsyncIoStream as AsyncIoStream, _init_capnp_api, ) diff --git a/capnp/helpers/asyncHelper.h b/capnp/helpers/asyncHelper.h deleted file mode 100644 index f939b3f..0000000 --- a/capnp/helpers/asyncHelper.h +++ /dev/null @@ -1,13 +0,0 @@ -#pragma once - -#include "kj/async.h" -#include "capabilityHelper.h" - -void waitNeverDone(kj::WaitScope & scope) { - kj::NEVER_DONE.wait(scope); -} - -capnp::Response< ::capnp::DynamicStruct> * waitRemote(kj::Own> promise, - kj::WaitScope & scope) { - return new capnp::Response< ::capnp::DynamicStruct>(promise->wait(scope)); -} diff --git a/capnp/helpers/capabilityHelper.cpp b/capnp/helpers/capabilityHelper.cpp index 353f4b5..fda7560 100644 --- a/capnp/helpers/capabilityHelper.cpp +++ b/capnp/helpers/capabilityHelper.cpp @@ -1,8 +1,8 @@ #include "capnp/helpers/capabilityHelper.h" #include "capnp/lib/capnp_api.h" -::kj::Promise> convert_to_pypromise(kj::Own> promise) { - return promise->then([](capnp::Response&& response) { +::kj::Promise> convert_to_pypromise(capnp::RemotePromise promise) { + return promise.then([](capnp::Response&& response) { return stealPyRef(wrap_dynamic_struct_reader(response)); } ); } @@ -58,75 +58,26 @@ void check_py_error() { } } -inline kj::Promise> maybeUnwrapPromise(PyObject * result) { - check_py_error(); - auto promise = extract_promise(result); - Py_DECREF(result); - return kj::mv(*promise); -} - kj::Promise> wrapPyFunc(kj::Own func, kj::Own arg) { GILAcquire gil; - // Creates an owned reference, which will be destroyed in maybeUnwrapPromise PyObject * result = PyObject_CallFunctionObjArgs(func->obj, arg->obj, NULL); - return maybeUnwrapPromise(result); + check_py_error(); + return stealPyRef(result); } -kj::Promise> wrapPyFuncNoArg(kj::Own func) { - GILAcquire gil; - // Creates an owned reference, which will be destroyed in maybeUnwrapPromise - PyObject * result = PyObject_CallFunctionObjArgs(func->obj, NULL); - return maybeUnwrapPromise(result); -} - -kj::Promise> wrapRemoteCall(kj::Own func, capnp::Response & arg) { - GILAcquire gil; - // Creates an owned reference, which will be destroyed in maybeUnwrapPromise - PyObject * ret = wrap_remote_call(func->obj, arg); - return maybeUnwrapPromise(ret); -} - -::kj::Promise> then(kj::Own>> promise, +::kj::Promise> then(kj::Promise> promise, kj::Own func, kj::Own error_func) { if(error_func->obj == Py_None) - return promise->then(kj::mvCapture(func, [](auto func, kj::Own arg) { + return promise.then(kj::mvCapture(func, [](auto func, kj::Own arg) { return wrapPyFunc(kj::mv(func), kj::mv(arg)); } )); else - return promise->then + return promise.then (kj::mvCapture(func, [](auto func, kj::Own arg) { return wrapPyFunc(kj::mv(func), kj::mv(arg)); }), kj::mvCapture(error_func, [](auto error_func, kj::Exception arg) { return wrapPyFunc(kj::mv(error_func), stealPyRef(wrap_kj_exception(arg))); } )); } -::kj::Promise> then(kj::Own<::capnp::RemotePromise<::capnp::DynamicStruct>> promise, - kj::Own func, kj::Own error_func) { - if(error_func->obj == Py_None) - return promise->then(kj::mvCapture(func, [](auto func, capnp::Response&& arg) { - return wrapRemoteCall(kj::mv(func), arg); } )); - else - return promise->then - (kj::mvCapture(func, [](auto func, capnp::Response&& arg) { - return wrapRemoteCall(kj::mv(func), arg); }), - kj::mvCapture(error_func, [](auto error_func, kj::Exception arg) { - return wrapPyFunc(kj::mv(error_func), stealPyRef(wrap_kj_exception(arg))); } )); -} - -::kj::Promise> then(kj::Own> promise, - kj::Own func, kj::Own error_func) { - if(error_func->obj == Py_None) - return promise->then(kj::mvCapture(func, [](auto func) { return wrapPyFuncNoArg(kj::mv(func)); } )); - else - return promise->then(kj::mvCapture(func, [](auto func) { return wrapPyFuncNoArg(kj::mv(func)); }), - kj::mvCapture(error_func, [](auto error_func, kj::Exception arg) { - return wrapPyFunc(kj::mv(error_func), stealPyRef(wrap_kj_exception(arg))); } )); -} - -::kj::Promise> then(kj::Promise> > && promise) { - return promise.then([](kj::Array>&& arg) { - return stealPyRef(convert_array_pyobject(arg)); } ); -} - kj::Promise PythonInterfaceDynamicImpl::call(capnp::InterfaceSchema::Method method, capnp::CallContext< capnp::DynamicStruct, capnp::DynamicStruct> context) { auto methodName = method.getProto().getName(); diff --git a/capnp/helpers/capabilityHelper.h b/capnp/helpers/capabilityHelper.h index 7c81f50..d1d270a 100644 --- a/capnp/helpers/capabilityHelper.h +++ b/capnp/helpers/capabilityHelper.h @@ -54,37 +54,21 @@ inline kj::Own stealPyRef(PyObject* o) { return ret; } -::kj::Promise> convert_to_pypromise(kj::Own> promise); +::kj::Promise> convert_to_pypromise(capnp::RemotePromise promise); -inline ::kj::Promise> convert_to_pypromise(kj::Own> promise) { - return promise->then([]() { +inline ::kj::Promise> convert_to_pypromise(kj::Promise promise) { + return promise.then([]() { GILAcquire gil; return kj::heap(Py_None); }); } -template -::kj::Promise convert_to_voidpromise(kj::Own> promise) { - return promise->then([](T) { } ); -} - void reraise_kj_exception(); void check_py_error(); -inline kj::Promise> wrapSizePromise(kj::Promise promise) { - return promise.then([](size_t response) { return stealPyRef(PyLong_FromSize_t(response)); } ); -} - -::kj::Promise> then(kj::Own>> promise, +::kj::Promise> then(kj::Promise> promise, kj::Own func, kj::Own error_func); -::kj::Promise> then(kj::Own<::capnp::RemotePromise< ::capnp::DynamicStruct>> promise, - kj::Own func, kj::Own error_func); - -::kj::Promise> then(kj::Own> promise, - kj::Ownfunc, kj::Own error_func); - -::kj::Promise> then(kj::Promise> > && promise); class PythonInterfaceDynamicImpl final: public capnp::DynamicCapability::Server { public: @@ -105,17 +89,6 @@ public: capnp::CallContext< capnp::DynamicStruct, capnp::DynamicStruct> context); }; -inline capnp::DynamicCapability::Client new_client(capnp::InterfaceSchema & schema, PyObject * server) { - return capnp::DynamicCapability::Client(kj::heap(schema, server)); -} -inline capnp::DynamicValue::Reader new_server(capnp::InterfaceSchema & schema, PyObject * server) { - return capnp::DynamicValue::Reader(kj::heap(schema, server)); -} - -inline capnp::Capability::Client server_to_client(capnp::InterfaceSchema & schema, PyObject * server) { - return kj::heap(schema, server); -} - class PyAsyncIoStream: public kj::AsyncIoStream { public: kj::Own protocol; diff --git a/capnp/helpers/helpers.pxd b/capnp/helpers/helpers.pxd index 5682d12..64ddcd0 100644 --- a/capnp/helpers/helpers.pxd +++ b/capnp/helpers/helpers.pxd @@ -1,9 +1,7 @@ from capnp.includes.capnp_cpp cimport ( - Maybe, ReaderOptions, DynamicStruct, Request, Response, Promise, PyPromise, VoidPromise, PyPromiseArray, - RemotePromise, DynamicCapability, InterfaceSchema, EnumSchema, StructSchema, DynamicValue, - Capability, RpcSystem, MessageBuilder, MessageReader, TwoPartyVatNetwork, AnyPointer, - DynamicStruct_Builder, WaitScope, AsyncIoContext, StringPtr, TaskSet, Timer, - LowLevelAsyncIoProvider, AsyncIoProvider, Own, PyRefCounter + Maybe, PyPromise, VoidPromise, RemotePromise, + DynamicCapability, InterfaceSchema, EnumSchema, StructSchema, DynamicValue, Capability, + RpcSystem, MessageBuilder, Own, PyRefCounter ) from capnp.includes.schema_cpp cimport ByteArray @@ -12,37 +10,23 @@ from non_circular cimport reraise_kj_exception from cpython.ref cimport PyObject -from libcpp cimport bool - cdef extern from "capnp/helpers/fixMaybe.h": EnumSchema.Enumerant fixMaybe(Maybe[EnumSchema.Enumerant]) except +reraise_kj_exception StructSchema.Field fixMaybe(Maybe[StructSchema.Field]) except +reraise_kj_exception cdef extern from "capnp/helpers/capabilityHelper.h": - # PyPromise evalLater(EventLoop &, PyObject * func) - # PyPromise there(EventLoop & loop, PyPromise & promise, PyObject * func, PyObject * error_func) - PyPromise then(Own[PyPromise] promise, Own[PyRefCounter] func, Own[PyRefCounter] error_func) - PyPromise then(Own[RemotePromise] promise, Own[PyRefCounter] func, Own[PyRefCounter] error_func) - PyPromise then(Own[VoidPromise] promise, Own[PyRefCounter] func, Own[PyRefCounter] error_func) - PyPromise then(PyPromiseArray & promise) + PyPromise then(PyPromise promise, Own[PyRefCounter] func, Own[PyRefCounter] error_func) DynamicCapability.Client new_client(InterfaceSchema&, PyObject *) DynamicValue.Reader new_server(InterfaceSchema&, PyObject *) Capability.Client server_to_client(InterfaceSchema&, PyObject *) - PyPromise convert_to_pypromise(Own[RemotePromise]) - PyPromise convert_to_pypromise(Own[VoidPromise]) - VoidPromise convert_to_voidpromise(Own[PyPromise]) - PyPromise wrapSizePromise(Promise[size_t]) + PyPromise convert_to_pypromise(RemotePromise) + PyPromise convert_to_pypromise(VoidPromise) VoidPromise taskToPromise(Own[PyRefCounter] coroutine, PyObject* callback) void init_capnp_api() cdef extern from "capnp/helpers/rpcHelper.h": Capability.Client bootstrapHelper(RpcSystem&) Capability.Client bootstrapHelperServer(RpcSystem&) - PyPromise connectServer(TaskSet &, Capability.Client, AsyncIoProvider *, StringPtr, ReaderOptions &) cdef extern from "capnp/helpers/serialize.h": ByteArray messageToPackedBytes(MessageBuilder &, size_t wordCount) - -cdef extern from "capnp/helpers/asyncHelper.h": - void waitNeverDone(WaitScope&) except +reraise_kj_exception nogil - Response * waitRemote(Own[RemotePromise], WaitScope&) except +reraise_kj_exception nogil diff --git a/capnp/helpers/non_circular.pxd b/capnp/helpers/non_circular.pxd index a6246b3..164a569 100644 --- a/capnp/helpers/non_circular.pxd +++ b/capnp/helpers/non_circular.pxd @@ -1,16 +1,8 @@ from cpython.ref cimport PyObject from libcpp cimport bool -cdef extern from "capnp/helpers/capabilityHelper.h": - cppclass PythonInterfaceDynamicImpl: - PythonInterfaceDynamicImpl(PyObject *) - cdef extern from "capnp/helpers/capabilityHelper.h": void reraise_kj_exception() cdef cppclass PyRefCounter: PyRefCounter(PyObject *) PyObject * obj - -cdef extern from "capnp/helpers/rpcHelper.h": - cdef cppclass ErrorHandler: - pass diff --git a/capnp/helpers/rpcHelper.h b/capnp/helpers/rpcHelper.h index a0c68e4..0681039 100644 --- a/capnp/helpers/rpcHelper.h +++ b/capnp/helpers/rpcHelper.h @@ -19,52 +19,3 @@ capnp::Capability::Client bootstrapHelperServer(capnp::RpcSystem stream; - capnp::TwoPartyVatNetwork network; - capnp::RpcSystem rpcSystem; - - ServerContext(kj::Own&& stream, capnp::Capability::Client client, capnp::ReaderOptions & opts) - : stream(kj::mv(stream)), - network(*this->stream, capnp::rpc::twoparty::Side::SERVER, opts), - rpcSystem(makeRpcServer(network, client)) {} -}; - -void acceptLoop(kj::TaskSet & tasks, capnp::Capability::Client client, kj::Own&& listener, capnp::ReaderOptions & opts) { - auto ptr = listener.get(); - tasks.add(ptr->accept().then(kj::mvCapture(kj::mv(listener), - [&, client, opts](kj::Own&& listener, - kj::Own&& connection) mutable { - acceptLoop(tasks, client, kj::mv(listener), opts); - - auto server = kj::heap(kj::mv(connection), client, opts); - - // Arrange to destroy the server context when all references are gone, or when the - // EzRpcServer is destroyed (which will destroy the TaskSet). - tasks.add(server->network.onDisconnect().attach(kj::mv(server))); - }))); -} - -kj::Promise> connectServer(kj::TaskSet & tasks, capnp::Capability::Client client, kj::AsyncIoProvider * provider, kj::StringPtr bindAddress, capnp::ReaderOptions & opts) { - auto paf = kj::newPromiseAndFulfiller(); - auto portPromise = paf.promise.fork(); - - tasks.add(provider->getNetwork().parseAddress(bindAddress) - .then(kj::mvCapture(paf.fulfiller, - [&, client, opts](kj::Own>&& portFulfiller, - kj::Own&& addr) mutable { - auto listener = addr->listen(); - portFulfiller->fulfill(listener->getPort()); - acceptLoop(tasks, client, kj::mv(listener), opts); - }))); - - return portPromise.addBranch().then([&](unsigned int port) { - return stealPyRef(PyLong_FromUnsignedLong(port)); }); -} diff --git a/capnp/includes/capnp_cpp.pxd b/capnp/includes/capnp_cpp.pxd index 46dfb02..8428987 100644 --- a/capnp/includes/capnp_cpp.pxd +++ b/capnp/includes/capnp_cpp.pxd @@ -5,7 +5,7 @@ cdef extern from "capnp/helpers/checkCompiler.h": from libcpp cimport bool from capnp.helpers.non_circular cimport ( - PythonInterfaceDynamicImpl, reraise_kj_exception, PyRefCounter, ErrorHandler, + reraise_kj_exception, PyRefCounter, ) from capnp.includes.schema_cpp cimport ( Node, Data, StructNode, EnumNode, InterfaceNode, MessageBuilder, MessageReader, ReaderOptions, @@ -52,10 +52,6 @@ cdef extern from "kj/memory.h" namespace " ::kj": T& operator*() T* get() Own[T] heap[T](...) - Own[TwoPartyVatNetwork] makeTwoPartyVatNetwork" ::kj::heap< ::capnp::TwoPartyVatNetwork>"( - AsyncIoStream& stream, Side, ReaderOptions) - Own[PromiseFulfillerPair] copyPromiseFulfillerPair" ::kj::heap< ::kj::PromiseFulfillerPair >"( - PromiseFulfillerPair&) cdef extern from "kj/async.h" namespace " ::kj": cdef cppclass Promise[T] nogil: @@ -110,72 +106,12 @@ cdef extern from "kj/array.h" namespace " ::kj": T& add(T&) Array[T] finish() - ArrayBuilder[PyPromise] heapArrayBuilderPyPromise"::kj::heapArrayBuilder< ::kj::Promise> >"(size_t) nogil - - ctypedef Array[Own[PyRefCounter]] PyArray' ::kj::Array>' - -ctypedef Promise[PyArray] PyPromiseArray - -cdef extern from "kj/time.h" namespace " ::kj": - cdef cppclass Duration nogil: - Duration operator*(int64_t) - Duration NANOSECONDS - Duration MICROSECONDS - Duration MILLISECONDS - Duration SECONDS - Duration MINUTES - Duration HOURS - Duration DAYS - cdef cppclass TimePoint: - TimePoint(Duration) - cdef cppclass MonotonicClock nogil: - MonotonicClock(MonotonicClock&) - TimePoint now() - MonotonicClock systemPreciseMonotonicClock() - -cdef extern from "kj/timer.h" namespace " ::kj": - cdef cppclass Timer nogil: - # int64_t now() - # VoidPromise atTime(TimePoint time) - VoidPromise afterDelay(Duration delay) - cdef cppclass TimerImpl(Timer) nogil: - TimerImpl(TimePoint startTime) - Maybe[TimePoint] nextEvent() - Maybe[uint64_t] timeoutToNextEvent(TimePoint start, Duration unit, uint64_t max) - void advanceTo(TimePoint newTime) - -cdef inline Duration Nanoseconds(int64_t nanos): - return NANOSECONDS * nanos cdef extern from "kj/async-io.h" namespace " ::kj": cdef cppclass AsyncIoStream nogil: Promise[size_t] read(void*, size_t, size_t) Promise[void] write(const void*, size_t) - cdef cppclass LowLevelAsyncIoProvider: - # Own[AsyncInputStream] wrapInputFd(int) - # Own[AsyncOutputStream] wrapOutputFd(int) - Own[AsyncIoStream] wrapSocketFd(int) - Timer& getTimer() except +reraise_kj_exception - - cdef cppclass AsyncIoProvider nogil: - TwoWayPipe newTwoWayPipe() - - cdef cppclass AsyncIoContext nogil: - AsyncIoContext(AsyncIoContext&) - Own[LowLevelAsyncIoProvider] lowLevelProvider - Own[AsyncIoProvider] provider - WaitScope waitScope - - cdef cppclass TaskSet nogil: - TaskSet(ErrorHandler &) - - cdef cppclass TwoWayPipe nogil: - Own[AsyncIoStream] ends[2] - - AsyncIoContext setupAsyncIo() nogil - Own[AsyncIoProvider] newAsyncIoProvider(LowLevelAsyncIoProvider& lowLevel); - cdef extern from "capnp/schema.capnp.h" namespace " ::capnp": enum TypeWhich" ::capnp::schema::Type::Which": TypeWhichVOID " ::capnp::schema::Type::Which::VOID" @@ -360,6 +296,7 @@ cdef extern from "capnp/dynamic.h" namespace " ::capnp": cppclass Client nogil: Client() Client(Client&) + Client(Own[PythonInterfaceDynamicImpl]) Client upcast(InterfaceSchema requestedSchema) DynamicCapability.Client castAs"castAs< ::capnp::DynamicCapability>"(InterfaceSchema) InterfaceSchema getSchema() @@ -390,7 +327,7 @@ cdef extern from "capnp/rpc-twoparty.h" namespace " ::capnp": TwoPartyVatNetwork(EventLoop &, AsyncIoStream& stream, Side, ReaderOptions) VoidPromise onDisconnect() VoidPromise onDrained() - RpcSystem makeRpcServerBootstrap"makeRpcServer"(TwoPartyVatNetwork&, Capability.Client) nogil + RpcSystem makeRpcServer(TwoPartyVatNetwork&, Capability.Client) nogil RpcSystem makeRpcClient(TwoPartyVatNetwork&) nogil cdef extern from "capnp/dynamic.h" namespace " ::capnp": @@ -474,7 +411,7 @@ cdef extern from "capnp/dynamic.h" namespace " ::capnp": Reader(DynamicEnum value) Reader(DynamicStruct.Reader& value) Reader(DynamicCapability.Client& value) - Reader(PythonInterfaceDynamicImpl& value) + Reader(Own[PythonInterfaceDynamicImpl] value) Reader(AnyPointer.Reader& value) Type getType() int64_t asInt"as"() @@ -553,19 +490,16 @@ cdef extern from "kj/async.h" namespace " ::kj": cdef cppclass VoidPromiseFulfiller"::kj::PromiseFulfiller" nogil: void fulfill() void reject(Exception&& exception) - cdef cppclass PromiseFulfillerPair" ::kj::PromiseFulfillerPair" nogil: - VoidPromise promise - Own[VoidPromiseFulfiller] fulfiller - PromiseFulfillerPair newPromiseAndFulfiller" ::kj::newPromiseAndFulfiller"() nogil - PyPromiseArray joinPromises(Array[PyPromise]) nogil cdef extern from "capnp/helpers/capabilityHelper.h": cdef cppclass PyAsyncIoStream(AsyncIoStream): - PyAsyncIoStream(PyObject* thisptr) + PyAsyncIoStream(Own[PyRefCounter] thisptr) void rejectDisconnected[T](PromiseFulfiller[T]& fulfiller, StringPtr message) void rejectVoidDisconnected(VoidPromiseFulfiller& fulfiller, StringPtr message) Exception makeException(StringPtr message) PyPromise tryReadMessage(AsyncIoStream& stream, ReaderOptions opts) + cppclass PythonInterfaceDynamicImpl: + PythonInterfaceDynamicImpl(InterfaceSchema&, PyObject *) cdef extern from "capnp/serialize-async.h" namespace " ::capnp": VoidPromise writeMessage(AsyncIoStream& output, MessageBuilder& builder) diff --git a/capnp/lib/capnp.pxd b/capnp/lib/capnp.pxd index 2e319d3..aed34d1 100644 --- a/capnp/lib/capnp.pxd +++ b/capnp/lib/capnp.pxd @@ -9,10 +9,9 @@ from capnp.includes.capnp_cpp cimport ( 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, Promise, - CallContext, RpcSystem, makeRpcServerBootstrap, makeRpcClient, Capability as C_Capability, - TwoPartyVatNetwork as C_TwoPartyVatNetwork, Side, AsyncIoStream, Own, makeTwoPartyVatNetwork, - PromiseFulfillerPair as C_PromiseFulfillerPair, copyPromiseFulfillerPair, newPromiseAndFulfiller, - PyArray, DynamicStruct_Builder, TwoWayPipe, PyRefCounter, PyAsyncIoStream + CallContext, RpcSystem, makeRpcServer, makeRpcClient, Capability as C_Capability, + TwoPartyVatNetwork as C_TwoPartyVatNetwork, Side, AsyncIoStream, Own, + DynamicStruct_Builder, PyRefCounter, PyAsyncIoStream ) from capnp.includes.schema_cpp cimport Node as C_Node, EnumNode as C_EnumNode from capnp.includes.types cimport * @@ -161,7 +160,6 @@ cdef _setDynamicFieldStatic(DynamicStruct_Builder thisptr, field, value, parent) cdef api object wrap_dynamic_struct_reader(Response & r) with gil cdef api Promise[void] * call_server_method( object server, char * _method_name, CallContext & _context) except * with gil -cdef api convert_array_pyobject(PyArray & arr) with gil cdef api object wrap_kj_exception(capnp.Exception & exception) with gil cdef api object wrap_kj_exception_for_reraise(capnp.Exception & exception) with gil cdef api object get_exception_info(object exc_type, object exc_obj, object exc_tb) with gil diff --git a/capnp/lib/capnp.pyx b/capnp/lib/capnp.pyx index 1b2c89d..9016f82 100644 --- a/capnp/lib/capnp.pyx +++ b/capnp/lib/capnp.pyx @@ -10,8 +10,7 @@ cimport cython # noqa: E402 from capnp.helpers.helpers cimport init_capnp_api -from capnp.includes.capnp_cpp cimport AsyncIoStream, WaitScope, PyPromise, VoidPromise, EventPort, EventLoop, WaitScope, LowLevelAsyncIoProvider, AsyncIoProvider, newAsyncIoProvider, MonotonicClock, Timer, TimerImpl, systemPreciseMonotonicClock, MILLISECONDS, Canceler, PyAsyncIoStream, PromiseFulfiller, VoidPromiseFulfiller, makeException -from capnp.includes.capnp_cpp cimport AsyncIoStream, WaitScope, PyPromise, VoidPromise, EventPort, EventLoop, WaitScope, LowLevelAsyncIoProvider, AsyncIoProvider, newAsyncIoProvider, MonotonicClock, Timer, TimerImpl, systemPreciseMonotonicClock, MILLISECONDS, Canceler, PyAsyncIoStream, PromiseFulfiller, VoidPromiseFulfiller, tryReadMessage, writeMessage, makeException +from capnp.includes.capnp_cpp cimport AsyncIoStream, WaitScope, PyPromise, VoidPromise, EventPort, EventLoop, Canceler, PyAsyncIoStream, PromiseFulfiller, VoidPromiseFulfiller, tryReadMessage, writeMessage, makeException, PythonInterfaceDynamicImpl from capnp.includes.schema_cpp cimport (MessageReader,) from cpython cimport array, Py_buffer, PyObject_CheckBuffer, memoryview, buffer @@ -37,6 +36,7 @@ import threading as _threading import traceback as _traceback import warnings as _warnings import weakref as _weakref +import traceback as _traceback from types import ModuleType as _ModuleType from operator import attrgetter as _attrgetter @@ -60,12 +60,7 @@ def deregister_all_types(): # By making it public, we'll be able to call it from capabilityHelper.h cdef api object wrap_dynamic_struct_reader(Response & r) with gil: - return _Response()._init_childptr(new Response(moveResponse(r)), None) - - -cdef api object wrap_remote_call(object func, Response & r): - response = _Response()._init_childptr(new Response(moveResponse(r)), None) - return func(response) + return _Response()._init_childptr(new Response(move(r)), None) cdef _find_field_order(struct_node): return [f.name for f in sorted(struct_node.fields, key=_attrgetter('codeOrder'))] @@ -85,7 +80,8 @@ def void_task_done_callback(method_name, _VoidPromiseFulfiller fulfiller, task): exc = task.exception() if exc is not None: - fulfiller.fulfiller.reject(makeException(capnp.StringPtr(str(exc)))) + fulfiller.fulfiller.reject(makeException(capnp.StringPtr(''.join( + _traceback.format_exception(type(exc), exc, exc.__traceback__))))) return res = task.result() @@ -124,27 +120,16 @@ cdef api VoidPromise * call_server_method(object server, func = getattr(server, method_name+'_context', None) if func is not None: ret = func(context) - 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(move((<_Promise>ret).thisptr))) - elif asyncio.iscoroutine(ret): - task = asyncio.create_task(ret) - callback = _partial(void_task_done_callback, method_name) - return new VoidPromise(helpers.taskToPromise( - capnp.heap[PyRefCounter](task), - callback)) - else: - try: - warning_msg = ( - "Server function ({}) returned a value that was not a Promise: return = {}" - .format(method_name, str(ret))) - except Exception: - 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 asyncio.iscoroutine(ret): + task = asyncio.create_task(ret) + callback = _partial(void_task_done_callback, method_name) + return new VoidPromise(helpers.taskToPromise( + capnp.heap[PyRefCounter](task), + callback)) + else: + raise ValueError( + "Server function ({}) is not a coroutine" + .format(method_name, str(ret))) else: func = getattr(server, method_name) # will raise if no function found params = context.params @@ -152,40 +137,18 @@ cdef api VoidPromise * call_server_method(object server, params_dict['_context'] = context ret = func(**params_dict) - 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(move((<_Promise>ret).thisptr))) - elif asyncio.iscoroutine(ret): - async def finalize(): - fill_context(method_name, context, await ret) - task = asyncio.create_task(finalize()) - callback = _partial(void_task_done_callback, method_name) - return new VoidPromise(helpers.taskToPromise( - capnp.heap[PyRefCounter](task), - callback)) - else: - fill_context(method_name, context, ret) - - return NULL - - -cdef api object convert_array_pyobject(PyArray & arr) with gil: - return [arr[i].get().obj for i in range(arr.size())] - - -cdef api Own[Promise[Own[PyRefCounter]]] extract_promise(object obj): - if type(obj) is _Promise: - return move((<_Promise>obj).thisptr) - elif type(obj) is _RemotePromise: - parent = (<_RemotePromise>obj)._parent - # We don't need parent anymore. Setting to none allows quicker garbage collection - (<_RemotePromise>obj)._parent = None - return capnp.heap[PyPromise](helpers.convert_to_pypromise(move((<_RemotePromise>obj).thisptr)) - .attach(capnp.heap[PyRefCounter](parent))) - else: - return capnp.heap[PyPromise](capnp.heap[PyRefCounter](obj)) + if asyncio.iscoroutine(ret): + async def finalize(): + fill_context(method_name, context, await ret) + task = asyncio.create_task(finalize()) + callback = _partial(void_task_done_callback, method_name) + return new VoidPromise(helpers.taskToPromise( + capnp.heap[PyRefCounter](task), + callback)) + else: + raise ValueError( + "Server function ({}) is not a coroutine" + .format(method_name, str(ret))) cdef extern from "" namespace " ::kj": @@ -218,7 +181,7 @@ cdef class _KjExceptionWrapper: cdef capnp.Exception * thisptr cdef _init(self, capnp.Exception & other): - self.thisptr = new capnp.Exception(moveException(other)) + self.thisptr = new capnp.Exception(move(other)) return self def __dealloc__(self): @@ -339,13 +302,6 @@ ctypedef fused _DynamicSetterClasses: DynamicStruct_Builder -ctypedef fused PromiseTypes: - _Promise - _RemotePromise - _VoidPromise - # PromiseFulfillerPair - - cdef extern from "Python.h": cdef int PyObject_GetBuffer(object, Py_buffer *view, int flags) cdef void PyBuffer_Release(Py_buffer *view) @@ -361,20 +317,6 @@ cdef extern from "capnp/list.h" namespace " ::capnp": uint size() -cdef extern from "" namespace "std": - C_DynamicStruct.Pipeline moveStructPipeline"std::move"(C_DynamicStruct.Pipeline) - C_DynamicOrphan moveOrphan"std::move"(C_DynamicOrphan) - Request moveRequest"std::move"(Request) - Response moveResponse"std::move"(Response) - PyPromise movePromise"std::move"(PyPromise) - VoidPromise moveVoidPromise"std::move"(VoidPromise) - RemotePromise moveRemotePromise"std::move"(RemotePromise) - CallContext moveCallContext"std::move"(CallContext) - Own[AsyncIoStream] moveOwnAsyncIOStream"std::move"(Own[AsyncIoStream]) - capnp.Exception moveException"std::move"(capnp.Exception) - capnp.AsyncIoContext moveAsyncContext"std::move"(capnp.AsyncIoContext) - - cdef extern from "" namespace " ::capnp": StringTree printStructReader" ::capnp::prettyPrint"(C_DynamicStruct.Reader) except +reraise_kj_exception StringTree printStructBuilder" ::capnp::prettyPrint"(DynamicStruct_Builder) except +reraise_kj_exception @@ -760,7 +702,7 @@ cdef C_DynamicValue.Reader _extract_dynamic_client(_DynamicCapabilityClient valu cdef C_DynamicValue.Reader _extract_dynamic_server(object value): cdef _InterfaceSchema schema = value.schema - return helpers.new_server(schema.thisptr, value) + return C_DynamicValue.Reader(capnp.heap[PythonInterfaceDynamicImpl](schema.thisptr, value)) cdef C_DynamicValue.Reader _extract_dynamic_enum(_DynamicEnum value): @@ -843,7 +785,7 @@ cdef _setDynamicField(_DynamicSetterClasses thisptr, field, value, parent): thisptr.set(field, _extract_dynamic_struct_reader(value)) elif value_type is _DynamicCapabilityClient: thisptr.set(field, _extract_dynamic_client(value)) - elif value_type is _DynamicCapabilityServer or isinstance(value, _DynamicCapabilityServer): + elif isinstance(value, _DynamicCapabilityServer): thisptr.set(field, _extract_dynamic_server(value)) elif value_type is _DynamicEnum: thisptr.set(field, _extract_dynamic_enum(value)) @@ -892,7 +834,7 @@ cdef _setDynamicFieldWithField(DynamicStruct_Builder thisptr, _StructSchemaField thisptr.setByField(field.thisptr, _extract_dynamic_struct_reader(value)) elif value_type is _DynamicCapabilityClient: thisptr.setByField(field.thisptr, _extract_dynamic_client(value)) - elif value_type is _DynamicCapabilityServer or isinstance(value, _DynamicCapabilityServer): + elif isinstance(value, _DynamicCapabilityServer): thisptr.setByField(field.thisptr, _extract_dynamic_server(value)) elif value_type is _DynamicEnum: thisptr.setByField(field.thisptr, _extract_dynamic_enum(value)) @@ -941,7 +883,7 @@ cdef _setDynamicFieldStatic(DynamicStruct_Builder thisptr, field, value, parent) thisptr.set(field, _extract_dynamic_struct_reader(value)) elif value_type is _DynamicCapabilityClient: thisptr.set(field, _extract_dynamic_client(value)) - elif value_type is _DynamicCapabilityServer or isinstance(value, _DynamicCapabilityServer): + elif isinstance(value, _DynamicCapabilityServer): thisptr.set(field, _extract_dynamic_server(value)) elif value_type is _DynamicEnum: thisptr.set(field, _extract_dynamic_enum(value)) @@ -1323,7 +1265,8 @@ cdef class _DynamicStructBuilder: :Raises: :exc:`KjException` if this isn't the message's root struct. """ self._check_write() - await _VoidPromise()._init(writeMessage(deref(stream.thisptr.get()), deref((<_MessageBuilder>self._parent).thisptr))) + await _voidpromise_to_asyncio( + writeMessage(deref(stream.thisptr), deref((<_MessageBuilder>self._parent).thisptr))) self._is_written = True def write_packed(self, file): @@ -1691,12 +1634,12 @@ cdef class _DynamicStructPipeline: cdef class _DynamicOrphan: cdef _init(self, C_DynamicOrphan other, object parent): - self.thisptr = moveOrphan(other) + self.thisptr = move(other) self._parent = parent return self cdef C_DynamicOrphan move(self): - return moveOrphan(self.thisptr) + return move(self.thisptr) cpdef get(self): """Returns a python object corresponding to the DynamicValue owned by this orphan @@ -1817,28 +1760,19 @@ cdef class _DynamicObjectBuilder: cdef void kjloop_runnable_callback(void* data) with gil: cdef AsyncIoEventPort *port = data assert port.runHandle is not None - port.timerImpl.advanceTo(systemPreciseMonotonicClock().now()) port.kjLoop.run() -cdef void kjloop_advance_callback(void* data) with gil: - cdef AsyncIoEventPort *port = data - assert port.runHandle is not None - port.timerImpl.advanceTo(systemPreciseMonotonicClock().now()) - cdef cppclass AsyncIoEventPort(EventPort): EventLoop *kjLoop - TimerImpl *timerImpl; object asyncioLoop; object runHandle; __init__(object asyncioLoop): this.kjLoop = new EventLoop(deref(this)) - this.timerImpl = new TimerImpl(systemPreciseMonotonicClock().now()) this.runHandle = None this.asyncioLoop = asyncioLoop __dealloc__(): - del this.timerImpl del this.kjLoop cbool wait() except* with gil: @@ -1852,88 +1786,44 @@ cdef cppclass AsyncIoEventPort(EventPort): void setRunnable(cbool runnable) except* with gil: if runnable: - if this.runHandle is not None: - # If a timer was running, cancel it and schedule a run immediately - # The timer will be re-scheduled once the kj loop becomes un-runnable again. - this.runHandle.cancel() + assert this.runHandle is None us = this; this.runHandle = this.asyncioLoop.call_soon(lambda: kjloop_runnable_callback(us)) else: assert this.runHandle is not None this.runHandle.cancel() - this.scheduleAdvance() - - void scheduleAdvance() with gil: - cdef uint64_t nextEvent = this.timerImpl.timeoutToNextEvent( - systemPreciseMonotonicClock().now(), MILLISECONDS, -1).orDefault(-1) - if nextEvent == -1: this.runHandle = None - else: - seconds = nextEvent / 1000 - us = this; - this.runHandle = this.asyncioLoop.call_later(seconds, lambda: kjloop_advance_callback(us)) EventLoop *getKjLoop(): return this.kjLoop - Timer *getTimer(): - return this.timerImpl; - def _asyncio_close_patch(loop, oldclose, _EventLoop kjloop): # The purpose of patching the asyncio close() function is to set up the kj-loop to be closed as well. # We replace the event loop getter with a weakref, such that it can be destroyed when all other # references to it are gone. Then, if a new asyncio loop ever gets started, a new kj-loop can also be # started. _C_DEFAULT_EVENT_LOOP_LOCAL.loop = _weakref.ref(kjloop) - loop.close = oldclose() + loop.close = oldclose return oldclose() cdef class _EventLoop: cdef object __weakref__ # Needed to make this class weak-referenceable - cdef Own[LowLevelAsyncIoProvider] lowLevelProvider - cdef Own[AsyncIoProvider] provider - cdef WaitScope * waitScope - cdef Timer* timer - cdef readonly in_asyncio_mode - - cdef AsyncIoEventPort *customPort + cdef WaitScope* waitScope + cdef AsyncIoEventPort* customPort def __init__(self): self._init() cdef _init(self) except +reraise_kj_exception: - try: - loop = asyncio.get_running_loop() - self.customPort = new AsyncIoEventPort(loop) - kjLoop = self.customPort.getKjLoop() - self.waitScope = new WaitScope(deref(kjLoop)) - self.timer = self.customPort.getTimer() - loop.close = _partial(_asyncio_close_patch, loop, loop.close, self) - self.in_asyncio_mode = True - except RuntimeError: - ptr = new capnp.AsyncIoContext(capnp.setupAsyncIo()) - self.lowLevelProvider = move(ptr.lowLevelProvider) - self.provider = move(ptr.provider) - self.waitScope = &ptr.waitScope - self.timer = &self.lowLevelProvider.get().getTimer() - del ptr - self.in_asyncio_mode = False + loop = asyncio.get_running_loop() + self.customPort = new AsyncIoEventPort(loop) + kjLoop = self.customPort.getKjLoop() + self.waitScope = new WaitScope(deref(kjLoop)) + loop.close = _partial(_asyncio_close_patch, loop, loop.close, self) def __dealloc__(self): - if not self.customPort == NULL: - # If we have a custom port, the waitscope is not owned by provider, we have to delete it manually - del self.waitScope - del self.customPort - - cdef TwoWayPipe makeTwoWayPipe(self): - if self.in_asyncio_mode: - raise RuntimeError("Cannot call makeTwoWayPipe in asyncio mode") - return deref(self.provider).newTwoWayPipe() - - cdef Own[AsyncIoStream] wrapSocketFd(self, int fd): - if self.in_asyncio_mode: - raise RuntimeError("Cannot call wrapSocketFd in asyncio mode") - return deref(self.lowLevelProvider).wrapSocketFd(fd) + del self.waitScope + del self.customPort _C_DEFAULT_EVENT_LOOP_LOCAL = _threading.local() @@ -1960,57 +1850,11 @@ cdef _EventLoop C_DEFAULT_EVENT_LOOP_GETTER(): return _C_DEFAULT_EVENT_LOOP_LOCAL.loop -cdef class _Timer: - cdef capnp.Timer * thisptr - - cdef _init(self, capnp.Timer * timer): - self.thisptr = timer - return self - - cpdef after_delay(self, time) except +reraise_kj_exception: - return _VoidPromise()._init(self.thisptr.afterDelay(capnp.Nanoseconds(time))) - - -def getTimer(): - """ - Get libcapnp event loop timer - """ - return _Timer()._init(C_DEFAULT_EVENT_LOOP_GETTER().timer) - - -cpdef remove_event_loop(): - '''Remove the event loop''' - global _C_DEFAULT_EVENT_LOOP_LOCAL - - loop = getattr(_C_DEFAULT_EVENT_LOOP_LOCAL, 'loop', None) - if loop is not None: - loop._remove() - del _C_DEFAULT_EVENT_LOOP_LOCAL.loop - - -def wait_forever(): - """ - Use libcapnp event loop to poll/wait forever - """ - cdef _EventLoop loop = C_DEFAULT_EVENT_LOOP_GETTER() - with nogil: - helpers.waitNeverDone(deref(loop.waitScope)) - - -def poll_once(): - """ - Poll libcapnp event loop once - """ - cdef _EventLoop loop = C_DEFAULT_EVENT_LOOP_GETTER() - with nogil: - loop.waitScope.poll() - - cdef class _CallContext: cdef CallContext * thisptr cdef _init(self, CallContext other): - self.thisptr = new CallContext(moveCallContext(other)) + self.thisptr = new CallContext(move(other)) return self def __dealloc__(self): @@ -2034,146 +1878,51 @@ cdef class _CallContext: self.thisptr.allowCancellation() cpdef tail_call(self, _Request tailRequest): - promise = _VoidPromise()._init(self.thisptr.tailCall(moveRequest(deref(tailRequest.thisptr_child)))) - return promise + return _voidpromise_to_asyncio(self.thisptr.tailCall(move(deref(tailRequest.thisptr_child)))) -cdef void _promise_check_consumed(PromiseTypes promise) except*: - if promise.thisptr.get() == NULL: - raise KjException( - "Promise was already used in a consuming operation. You can no longer use this Promise object") - -cdef _promise_then(PromiseTypes self, func, error_func, num_args, attach=None) except +reraise_kj_exception: - _promise_check_consumed(self) - - argspec = None - try: - argspec = _inspect.getfullargspec(func) - except (TypeError, ValueError): - pass - if argspec: - args_length = len(argspec.args) if argspec.args else 0 - defaults_length = len(argspec.defaults) if argspec.defaults else 0 - if args_length - defaults_length != num_args: - raise KjException(f'Function passed to `then` call must take exactly {num_args} arguments') - - return _Promise()._init( - helpers.then(move(self.thisptr), capnp.heap[PyRefCounter](func), - capnp.heap[PyRefCounter](error_func)) - .attach(capnp.heap[PyRefCounter]( attach))) - -cdef _promise_to_asyncio(PromiseTypes promise): - _promise_check_consumed(promise) +cdef _promise_to_asyncio(PyPromise promise): fut = asyncio.get_running_loop().create_future() + def success(res): return fut.set_result(res) if not fut.cancelled() else None + def exception(err): return fut.set_exception(err) if not fut.cancelled() else None + def done(fut): return fut.kjpromise.cancel() if fut.cancelled() else None # Attach the promise to the future, so that it doesn't get destroyed - fut.kjpromise = promise.then( - lambda res: fut.set_result(res) if not fut.cancelled() else None, - lambda err: fut.set_exception(err) if not fut.cancelled() else None) - del promise - fut.add_done_callback( - lambda fut: fut.kjpromise.cancel() if fut.cancelled() else None) + fut.kjpromise = _Promise()._init(helpers.then( + move(promise), + capnp.heap[PyRefCounter](success), + capnp.heap[PyRefCounter](exception))) + fut.add_done_callback(done) return fut +cdef _voidpromise_to_asyncio(VoidPromise promise): + return _promise_to_asyncio(helpers.convert_to_pypromise(move(promise))) + cdef class _Promise: cdef Own[PyPromise] thisptr - def __init__(self, obj=None): - if obj is not None: - self.thisptr = capnp.heap[PyPromise](capnp.heap[PyRefCounter](obj)) - cdef _init(self, PyPromise other): - self.thisptr = capnp.heap[PyPromise](movePromise(other)) + self.thisptr = capnp.heap[PyPromise](move(other)) return self - cpdef wait(self) except +reraise_kj_exception: - _promise_check_consumed(self) - cdef Own[PyPromise] prom = move(self.thisptr) # Explicit move to not leave thisptr dangling - cdef Own[PyRefCounter] ret - cdef _EventLoop loop = C_DEFAULT_EVENT_LOOP_GETTER() - with nogil: - ret = move(prom.get().wait(deref(loop.waitScope))) - return ret.get().obj - - async def a_wait(self): - """ - Asyncio version of wait(). - Required when using asyncio for socket communication. - - Will still work with non-asyncio socket communication, but requires async handling of the function call. - """ - return await _promise_to_asyncio(self) - - def __await__(self): - return _promise_to_asyncio(self).__await__() - - cpdef then(self, func, error_func=None) except +reraise_kj_exception: - return _promise_then(self, func, error_func, 1) - cpdef cancel(self) except +reraise_kj_exception: self.thisptr = Own[PyPromise]() -cdef class _VoidPromise: - cdef Own[VoidPromise] thisptr - - - cdef _init(self, VoidPromise other): - self.thisptr = capnp.heap[VoidPromise](moveVoidPromise(other)) - return self - - cpdef wait(self) except +reraise_kj_exception: - _promise_check_consumed(self) - cdef Own[VoidPromise] prom = move(self.thisptr) # Explicit move to not leave thisptr dangling - cdef _EventLoop loop = C_DEFAULT_EVENT_LOOP_GETTER() - with nogil: - prom.get().wait(deref(loop.waitScope)) - - async def a_wait(self): - """ - Asyncio version of wait(). - Required when using asyncio for socket communication. - - Will still work with non-asyncio socket communication, but requires async handling of the function call. - """ - # TODO: Is keeping a separate _VoidPromise class really worth it? Does it make things faster? - return await _promise_to_asyncio[_Promise](self.as_pypromise()) - - def __await__(self): - return _promise_to_asyncio[_Promise](self.as_pypromise()).__await__() - - cpdef as_pypromise(self) except +reraise_kj_exception: - _promise_check_consumed(self) - return _Promise()._init(helpers.convert_to_pypromise(move(self.thisptr))) - - cpdef then(self, func, error_func=None) except +reraise_kj_exception: - return _promise_then(self, func, error_func, 0) - - cpdef cancel(self) except +reraise_kj_exception: - self.thisptr = Own[VoidPromise]() - - - cdef class _RemotePromise: cdef object _parent - """A pointer to a parent object that needs to be kept alive for this promise to function. - Note that _Promise and _VoidPromise don't have such pointer. The reason is that in _RemotePromise - the parent pointer needs to be passed around through _RemotePromise._get. If an object needs to - be kept alive in _Promise or _VoidPromise, it can be attached to the underlying C++ promise.""" + """A pointer to a parent object that needs to be kept alive for this promise to function.""" cdef Own[RemotePromise] thisptr cdef _init(self, RemotePromise other, object parent=None): - self.thisptr = capnp.heap[RemotePromise](moveRemotePromise(other)) + self.thisptr = capnp.heap[RemotePromise](move(other)) self._parent = parent return self - cpdef wait(self) except +reraise_kj_exception: - """Wait on the promise. This will block until the promise has completed.""" - _promise_check_consumed(self) - cdef _EventLoop loop = C_DEFAULT_EVENT_LOOP_GETTER() - with nogil: - response = helpers.waitRemote(move(self.thisptr), deref(loop.waitScope)) - return _Response()._init_childptr(response, None) + cdef void _check_consumed(self) except*: + if self.thisptr.get() == NULL: + raise KjException( + "Promise was already used in a consuming operation. You can no longer use this Promise object") async def a_wait(self): """ @@ -2182,20 +1931,17 @@ cdef class _RemotePromise: Will still work with non-asyncio socket communication, but requires async handling of the function call. """ - return await _promise_to_asyncio(self) + self._check_consumed() + cdef Own[RemotePromise] thisptr = move(self.thisptr) + return await _promise_to_asyncio(helpers.convert_to_pypromise(move(deref(thisptr)))) def __await__(self): - return _promise_to_asyncio(self).__await__() - - cpdef as_pypromise(self) except +reraise_kj_exception: - _promise_check_consumed(self) - parent = self._parent - self._parent = None # We don't need parent anymore. Setting to none allows quicker garbage collection - return _Promise()._init(helpers.convert_to_pypromise(move(self.thisptr)) - .attach(capnp.heap[PyRefCounter](parent))) + self._check_consumed() + cdef Own[RemotePromise] thisptr = move(self.thisptr) + return _promise_to_asyncio(helpers.convert_to_pypromise(move(deref(thisptr)))).__await__() cpdef _get(self, field) except +reraise_kj_exception: - _promise_check_consumed(self) + self._check_consumed() cdef int type = (self.thisptr.get().get(field)).getType() if type == capnp.TYPE_CAPABILITY: return _DynamicCapabilityClient()._init( @@ -2218,7 +1964,7 @@ cdef class _RemotePromise: property schema: """A property that returns the _StructSchema object matching this reader""" def __get__(self): - _promise_check_consumed(self) + self._check_consumed() return _StructSchema()._init_child(self.thisptr.get().getSchema()) def __dir__(self): @@ -2227,43 +1973,17 @@ cdef class _RemotePromise: def to_dict(self, verbose=False, ordered=False): return _to_dict(self, verbose, ordered) - cpdef then(self, func, error_func=None) except +reraise_kj_exception: - parent = self._parent - self._parent = None # We don't need parent anymore. Setting to none allows quicker garbage collection - return _promise_then(self, func, error_func, 1, attach=parent) - cpdef cancel(self) except +reraise_kj_exception: self.thisptr = Own[RemotePromise]() self._parent = None # We don't need parent anymore. Setting to none allows quicker garbage collection -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>promise - elif promise_type is _RemotePromise or promise_type is _VoidPromise: - pyPromise = <_Promise>promise.as_pypromise() - new_promises_append(pyPromise) - else: - raise KjException( - "One of the promises passed to `join_promises` had a non promise value of: {}".format(promise)) - heap.add(movePromise(deref(pyPromise.thisptr))) - - return _Promise()._init(helpers.then(capnp.joinPromises(heap.finish()))) - - cdef class _Request(_DynamicStructBuilder): cdef Request * thisptr_child cdef public bint is_consumed cdef _init_child(self, Request other, parent): - self.thisptr_child = new Request(moveRequest(other)) + self.thisptr_child = new Request(move(other)) self._init(deref(self.thisptr_child), parent) self.is_consumed = False return self @@ -2282,7 +2002,7 @@ cdef class _Response(_DynamicStructReader): cdef Response * thisptr_child cdef _init_child(self, Response other, parent): - self.thisptr_child = new Response(moveResponse(other)) + self.thisptr_child = new Response(move(other)) self._init(deref(self.thisptr_child), parent) return self @@ -2295,30 +2015,17 @@ cdef class _Response(_DynamicStructReader): return self cdef class _DynamicCapabilityServer: - cdef public _InterfaceSchema schema - cdef public object server - - def __init__(self, schema, server): - cdef _InterfaceSchema s - if hasattr(schema, 'schema'): - s = schema.schema - else: - s = schema - - self.schema = s - self.server = server - - def __getattr__(self, field): - try: - return getattr(self.server, field) - except KjException as e: - raise e._to_python(), None, _sys.exc_info()[2] - + pass cdef class _DynamicCapabilityClient: cdef C_DynamicCapability.Client thisptr cdef public object _parent, _cached_schema + def __dealloc__(self): + # Needed to make Python 3.7 happy, which seems to have trouble deallocating stack objects + # appropriately + self.thisptr = C_DynamicCapability.Client() + cdef _init(self, C_DynamicCapability.Client other, object parent): self.thisptr = other self._parent = parent @@ -2331,7 +2038,8 @@ cdef class _DynamicCapabilityClient: else: s = schema - self.thisptr = helpers.new_client(s.thisptr, server) + self.thisptr = C_DynamicCapability.Client( + capnp.heap[PythonInterfaceDynamicImpl](s.thisptr, server)) self._parent = server return self @@ -2456,251 +2164,97 @@ cdef class _TwoPartyVatNetwork: cdef _init(self, _AsyncIoStream stream, Side side, schema_cpp.ReaderOptions opts): self.stream = stream - self.thisptr = makeTwoPartyVatNetwork(deref(stream.thisptr), side, opts) - return self - - cdef _init_pipe(self, _TwoWayPipe pipe, Side side, schema_cpp.ReaderOptions opts): - self.thisptr = makeTwoPartyVatNetwork(deref(pipe._pipe.ends[0]), side, opts) + self.thisptr = capnp.heap[C_TwoPartyVatNetwork](deref(stream.thisptr), side, opts) return self cpdef on_disconnect(self) except +reraise_kj_exception: - return _VoidPromise()._init(deref(self.thisptr).onDisconnect()) + return _voidpromise_to_asyncio(deref(self.thisptr).onDisconnect()) cdef class TwoPartyClient: """ TwoPartyClient for RPC Communication - Can be initialized using a socket wrapper (libcapnp) or using asyncio controlled sockets. - - :param socket: Can be a string defining a socket (e.g. localhost:12345) or a file descriptor for a socket. - Passes socket directly to libcapnp. Do not use with asyncio. + :param socket: AsyncIoStream :param traversal_limit_in_words: Pointer derefence limit (see https://capnproto.org/cxx.html). :param nesting_limit: Recursive limit when reading types (see https://capnproto.org/cxx.html). """ - cdef RpcSystem * thisptr - cdef public _TwoPartyVatNetwork _network - cdef public _TwoWayPipe _pipe + cdef Own[RpcSystem] thisptr + cdef _TwoPartyVatNetwork _network + + def __dealloc__(self): + # Needed to make Python 3.7 happy, which seems to have trouble deallocating stack objects + # appropriately + self.thisptr = Own[RpcSystem]() def __init__(self, socket=None, traversal_limit_in_words=None, nesting_limit=None): - if isinstance(socket, basestring): - if C_DEFAULT_EVENT_LOOP_GETTER().in_asyncio_mode: - raise RuntimeError("Pycapnp is in asyncio mode. Pass a AsyncIoStream") - socket = self._connect(socket) cdef schema_cpp.ReaderOptions opts = make_reader_opts(traversal_limit_in_words, nesting_limit) - if socket is None: - # Initialize TwoWayPipe, to use pipe() acquire other end of the pipe using read() and write() methods - self._pipe = _TwoWayPipe() - self._network = _TwoPartyVatNetwork()._init_pipe(self._pipe, capnp.CLIENT, opts) - elif isinstance(socket, _AsyncIoStream): + if isinstance(socket, _AsyncIoStream): self._network = _TwoPartyVatNetwork()._init(socket, capnp.CLIENT, opts) - elif isinstance(socket, _socket.socket): - stream = _FdAsyncIoStream(socket) - self._network = _TwoPartyVatNetwork()._init(stream, capnp.CLIENT, opts) else: - raise ValueError(f"Argument socket should be a string, socket, AsyncIoStream or None, was {type(socket)}") + raise ValueError(f"Argument socket should be a AsyncIoStream, was {type(socket)}") - self.thisptr = new RpcSystem(makeRpcClient(deref(self._network.thisptr))) - - async def read(self, bufsize): - """ - libcapnp reader (asyncio sockets only) - - :param bufsize: Buffer size to read from the libcapnp library - """ - - cdef array.array read_buffer = array.array('b', []) - array.resize(read_buffer, bufsize) - read_size_actual = await _Promise()._init( - helpers.wrapSizePromise( - self._pipe._pipe.ends[1].get().read(read_buffer.data.as_voidptr, 1, bufsize))) - array.resize(read_buffer, read_size_actual) - return read_buffer - - async def write(self, data): - """ - libcapnp writer (asyncio sockets only) - - :param data: Buffer to write to the libcapnp library - """ - cdef array.array write_buffer = array.array('b', data) - await _VoidPromise()._init( - deref(self._pipe._pipe.ends[1]).write( - write_buffer.data.as_voidptr, - len(data) - )) - - def __dealloc__(self): - if not self.thisptr == NULL: - del self.thisptr - - def _connect(self, host_string): - if host_string.startswith('unix:'): - path = host_string[5:] - sock = _socket.socket(_socket.AF_UNIX, _socket.SOCK_STREAM) - sock.connect(path) - else: - host, port = host_string.split(':') - sock = _socket.create_connection((host, port)) - # Set TCP_NODELAY on socket to disable Nagle's algorithm. This is not - # neccessary, but it speeds things up. - sock.setsockopt(_socket.IPPROTO_TCP, _socket.TCP_NODELAY, 1) - return sock + self.thisptr = capnp.heap[RpcSystem](makeRpcClient(deref(self._network.thisptr))) cpdef bootstrap(self) except +reraise_kj_exception: return _CapabilityClient()._init(helpers.bootstrapHelper(deref(self.thisptr)), self) cpdef on_disconnect(self) except +reraise_kj_exception: - return _VoidPromise()._init(deref(self._network.thisptr).onDisconnect()) + return self._network.on_disconnect() cdef class TwoPartyServer: """ TwoPartyServer for RPC Communication - Can be initialized using a socket wrapper (libcapnp) or using asyncio controlled sockets. - - :param socket: Can be a string defining a socket (e.g. localhost:12345, also supports *:12345) - or a file descriptor for a socket. - Passes socket directly to libcapnp. Do not use with asyncio. + :param socket: AsyncIoStream :param bootstrap: Class object defining the implementation of the Cap'n'proto interface. :param traversal_limit_in_words: Pointer derefence limit (see https://capnproto.org/cxx.html). :param nesting_limit: Recursive limit when reading types (see https://capnproto.org/cxx.html). """ - cdef RpcSystem * thisptr - cdef public _TwoPartyVatNetwork _network - cdef public _TwoWayPipe _pipe - cdef object _port - cdef public object port_promise, _bootstrap - cdef capnp.TaskSet * _task_set - cdef capnp.ErrorHandler _error_handler + cdef Own[RpcSystem] thisptr + cdef _TwoPartyVatNetwork _network + + def __dealloc__(self): + # Needed to make Python 3.7 happy, which seems to have trouble deallocating stack objects + # appropriately + self.thisptr = Own[RpcSystem]() def __init__(self, socket=None, bootstrap=None, traversal_limit_in_words=None, nesting_limit=None): if not bootstrap: raise KjException("You must provide a bootstrap interface to a server constructor.") - cdef _InterfaceSchema schema - self._bootstrap = None - - if isinstance(socket, basestring): - if C_DEFAULT_EVENT_LOOP_GETTER().in_asyncio_mode: - raise RuntimeError("Pycapnp is in asyncio mode. Please start an asyncio server using" - "TwoPartyClient.create_server and pass any resulting connection to this class.") - self._connect(socket, bootstrap, traversal_limit_in_words, nesting_limit) - return - opts = make_reader_opts(traversal_limit_in_words, nesting_limit) if isinstance(socket, _AsyncIoStream): self._network = _TwoPartyVatNetwork()._init(socket, capnp.SERVER, opts) - elif isinstance(socket, _socket.socket): - if C_DEFAULT_EVENT_LOOP_GETTER().in_asyncio_mode: - raise RuntimeError("Pycapnp is in asyncio mode. Please pass an AsyncIoStream instance.") - stream = _FdAsyncIoStream(socket) - self._network = _TwoPartyVatNetwork()._init(stream, capnp.SERVER, opts) - elif socket is None: - # Initialize TwoWayPipe, to use pipe() acquire other end of the pipe using read() and write() methods - self._pipe = _TwoWayPipe() - self._network = _TwoPartyVatNetwork()._init_pipe( - self._pipe, capnp.SERVER, opts) else: - raise KjException("Unexpected typ for socket in TwoPartyServer") + raise ValueError(f"Argument socket should be a AsyncIoStream, was {type(socket)}") - self._port = 0 - - if bootstrap: - self._bootstrap = bootstrap - schema = bootstrap.schema - self.thisptr = new RpcSystem(makeRpcServerBootstrap( - deref(self._network.thisptr), helpers.server_to_client(schema.thisptr, bootstrap))) - - async def read(self, bufsize): - """ - libcapnp reader (asyncio sockets only) - - :param bufsize: Buffer size to read from the libcapnp library - """ - cdef array.array read_buffer = array.array('b', []) - array.resize(read_buffer, bufsize) - read_size_actual = await _Promise()._init( - helpers.wrapSizePromise( - self._pipe._pipe.ends[1].get().read(read_buffer.data.as_voidptr, 1, bufsize))) - array.resize(read_buffer, read_size_actual) - return read_buffer - - async def write(self, data): - """ - libcapnp writer (asyncio sockets only) - - :param data: Buffer to write to the libcapnp library - """ - cdef array.array write_buffer = array.array('b', data) - await _VoidPromise()._init( - deref(self._pipe._pipe.ends[1]).write( - write_buffer.data.as_voidptr, - len(data) - )) - - cpdef _connect(self, host_string, bootstrap, traversal_limit_in_words, nesting_limit): - cdef schema_cpp.ReaderOptions opts = make_reader_opts(traversal_limit_in_words, nesting_limit) - cdef _InterfaceSchema schema - cdef _EventLoop loop = C_DEFAULT_EVENT_LOOP_GETTER() - cdef capnp.StringPtr temp_string = capnp.StringPtr(host_string, len(host_string)) - self._task_set = new capnp.TaskSet(self._error_handler) - - self._bootstrap = bootstrap - schema = bootstrap.schema - self.port_promise = _Promise()._init( - helpers.connectServer( - deref(self._task_set), - helpers.server_to_client(schema.thisptr, bootstrap), - loop.provider.get(), temp_string, opts)) - - def __dealloc__(self): - del self.thisptr - del self._task_set - - cpdef on_disconnect(self) except +reraise_kj_exception: - if self._task_set != NULL: - raise KjException("Currently, you can only call on_disconnect on a server without an internal socket") - return _VoidPromise()._init(deref(self._network.thisptr).onDisconnect()) - - def poll_once(self): - """ - Poll libcapnp library one cycle. - """ - return poll_once() - - async def poll_forever(self): - """Deprecated. Do not use. - Poll libcapnp library forever (asyncio) - """ - raise KjException("This functionality has been removed. If you wish to wait forever, use \n" + - "'await asyncio._get_running_loop().create_future()'") - - cpdef run_forever(self): - if self.port_promise is None: - raise KjException("You must pass a string as the socket parameter in __init__ to use this function") - - wait_forever() + cdef _InterfaceSchema schema = bootstrap.schema + self.thisptr = capnp.heap[RpcSystem](makeRpcServer( + deref(self._network.thisptr), + C_DynamicCapability.Client(capnp.heap[PythonInterfaceDynamicImpl]( + schema.thisptr, bootstrap)))) cpdef bootstrap(self) except +reraise_kj_exception: return _CapabilityClient()._init(helpers.bootstrapHelperServer(deref(self.thisptr)), self) - property port: - def __get__(self): - if self._port is None: - self._port = self.port_promise.wait() - return self._port - else: - return self._port + cpdef on_disconnect(self) except +reraise_kj_exception: + return _voidpromise_to_asyncio(deref(self._network.thisptr).onDisconnect() + .attach(capnp.heap[PyRefCounter](self))) cdef class _AsyncIoStream: cdef Own[AsyncIoStream] thisptr cdef _EventLoop _event_loop # We hold a pointer to the event loop here, to ensure it remains alive + def __dealloc__(self): + # Needed to make Python 3.7 happy, which seems to have trouble deallocating stack objects + # appropriately + self.thisptr = Own[AsyncIoStream]() + @staticmethod async def create_connection(host = None, port = None, **kwargs): """Create a TCP connection. @@ -2789,7 +2343,7 @@ cdef class _PyAsyncIoStreamProtocol(DummyBaseClass, asyncio.BufferedProtocol): cdef cbool read_eof # TODO: Temporary. This is an overflow buffer, which is needed for two blatant violations of the protocol. - # The first violation is int the SSL transport implementation. + # The first violation is in the the SSL transport implementation. # See https://github.com/python/cpython/issues/89322, fixed in Python 3.11. This bug causes the # SSL transport to force data upon us even when we've asked it to pause sending us data. Therefore, # we have to store the data in a overflow buffer. @@ -2969,53 +2523,6 @@ cdef api void _asyncio_stream_close(object thisptr) except*: if self.transport is not None and hasattr(self.transport, "close"): self.transport.close() -cdef class _TwoWayPipe: - cdef _EventLoop _event_loop - cdef TwoWayPipe _pipe - - def __init__(self): - self._init() - - cpdef _init(self) except +reraise_kj_exception: - self._event_loop = C_DEFAULT_EVENT_LOOP_GETTER() - # Create two way pipe using AsyncIoContext - self._pipe = self._event_loop.makeTwoWayPipe() - - -cdef class _FdAsyncIoStream(_AsyncIoStream): - """Wraps a socket for usage with pycapnp. - Note that this class does not own the socket. Instead, it receives the fileno from a python object, - which will continue to own it. This object is kept alive as long as this class is alive. Ultimately, - the python object is responsible for closing.""" - cdef object _socket - - def __init__(self, object socket): - self._socket = socket - self._init(socket.fileno()) - - cdef _init(self, int fd) except +reraise_kj_exception: - self._event_loop = C_DEFAULT_EVENT_LOOP_GETTER() - self.thisptr = self._event_loop.wrapSocketFd(fd) - - def __dealloc__(self): - # The AsyncIoStream must be destroyed before self._socket is removed, to ensure the socket is still - # open when the destructor is called. Therefore, we do this manually to have control over the ordering - self.thisptr = Own[AsyncIoStream]() - - -cdef class PromiseFulfillerPair: - cdef Own[C_PromiseFulfillerPair] thisptr - cdef public bint is_consumed - cdef public _VoidPromise promise - - def __init__(self): - self.thisptr = copyPromiseFulfillerPair(newPromiseAndFulfiller()) - self.is_consumed = False - self.promise = _VoidPromise()._init(moveVoidPromise(deref(self.thisptr).promise)) - - cpdef fulfill(self): - deref(deref(self.thisptr).fulfiller).fulfill() - cdef class _Schema: cdef _init(self, C_Schema other): @@ -3511,7 +3018,7 @@ class _StructModule(object): :rtype: :class:`_DynamicStructReader`""" cdef schema_cpp.ReaderOptions opts = make_reader_opts(traversal_limit_in_words, nesting_limit) - reader = await _Promise()._init(tryReadMessage(deref(stream.thisptr.get()), opts)) + reader = await _promise_to_asyncio(tryReadMessage(deref(stream.thisptr), opts)) if reader is None: return return reader.get_root(self.schema) @@ -3703,10 +3210,6 @@ class _InterfaceModule(object): C_DEFAULT_EVENT_LOOP_GETTER() # Make sure that the event loop has been initialized return _DynamicCapabilityClient()._init_vals(self.schema, server) - def _new_server(self, server): - C_DEFAULT_EVENT_LOOP_GETTER() # Make sure that the event loop has been initialized - return _DynamicCapabilityServer(self.schema, server) - class _EnumModule(object): def __init__(self, schema, name): diff --git a/docs/capnp.rst b/docs/capnp.rst index d895ef8..7bf8938 100644 --- a/docs/capnp.rst +++ b/docs/capnp.rst @@ -27,7 +27,6 @@ Promise may be one of: * :meth:`capnp.lib.capnp._Promise` * :meth:`capnp.lib.capnp._RemotePromise` * :meth:`capnp.lib.capnp._VoidPromise` -* :meth:`PromiseFulfillerPair` .. autoclass:: capnp.lib.capnp._Promise :members: @@ -44,11 +43,6 @@ Promise may be one of: :undoc-members: :inherited-members: -.. autoclass:: PromiseFulfillerPair - :members: - :undoc-members: - :inherited-members: - Communication ############# diff --git a/examples/async_calculator_client.py b/examples/async_calculator_client.py index 41ba8f8..c3e3e52 100755 --- a/examples/async_calculator_client.py +++ b/examples/async_calculator_client.py @@ -13,7 +13,7 @@ class PowerFunction(calculator_capnp.Calculator.Function.Server): we're implementing this on the client side and will pass a reference to the server. The server will then be able to make calls back to the client.""" - def call(self, params, **kwargs): + async def call(self, params, **kwargs): """Note the **kwargs. This is very necessary to include, since protocols can add parameters over time. Also, by default, a _context variable is passed to all server methods, but you can also return @@ -25,17 +25,14 @@ class PowerFunction(calculator_capnp.Calculator.Function.Server): def parse_args(): parser = argparse.ArgumentParser( - usage="Connects to the Calculator server \ -at the given address and does some RPCs" + 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() -async def main(host): - host, port = parse_args().host.split(":") - connection = await capnp.AsyncIoStream.create_connection(host=host, port=port) +async def main(connection): client = capnp.TwoPartyClient(connection) # Bootstrap the Calculator interface @@ -303,5 +300,10 @@ async def main(host): print("PASS") +async def cmd_main(host): + host, port = host.split(":") + await main(await capnp.AsyncIoStream.create_connection(host=host, port=port)) + + if __name__ == "__main__": - asyncio.run(main(parse_args().host)) + asyncio.run(cmd_main(parse_args().host)) diff --git a/examples/async_calculator_server.py b/examples/async_calculator_server.py index 3e00a00..f8302c3 100755 --- a/examples/async_calculator_server.py +++ b/examples/async_calculator_server.py @@ -48,7 +48,7 @@ class ValueImpl(calculator_capnp.Calculator.Value.Server): def __init__(self, value): self.value = value - def read(self, **kwargs): + async def read(self, **kwargs): return self.value @@ -79,7 +79,7 @@ class OperatorImpl(calculator_capnp.Calculator.Function.Server): def __init__(self, op): self.op = op - def call(self, params, **kwargs): + async def call(self, params, **kwargs): assert len(params) == 2 op = self.op @@ -102,22 +102,20 @@ class CalculatorImpl(calculator_capnp.Calculator.Server): async def evaluate(self, expression, _context, **kwargs): return ValueImpl(await evaluate_impl(expression)) - def defFunction(self, paramCount, body, _context, **kwargs): + async def defFunction(self, paramCount, body, _context, **kwargs): return FunctionImpl(paramCount, body) - def getOperator(self, op, **kwargs): + async def getOperator(self, op, **kwargs): return OperatorImpl(op) async def new_connection(stream): - server = capnp.TwoPartyServer(stream, bootstrap=CalculatorImpl()) - await server.on_disconnect() + await capnp.TwoPartyServer(stream, bootstrap=CalculatorImpl()).on_disconnect() def parse_args(): parser = argparse.ArgumentParser( - usage="""Runs the server bound to the\ - given address/port ADDRESS. """ + usage="""Runs the server bound to the given address/port ADDRESS. """ ) parser.add_argument("address", help="ADDRESS:PORT") diff --git a/examples/async_client.py b/examples/async_client.py index a97e212..47492d3 100755 --- a/examples/async_client.py +++ b/examples/async_client.py @@ -10,8 +10,7 @@ import thread_capnp def parse_args(): parser = argparse.ArgumentParser( - usage="Connects to the Example thread server \ -at the given address and does some RPCs" + usage="Connects to the Example thread server at the given address and does some RPCs" ) parser.add_argument("host", help="HOST:PORT") @@ -21,7 +20,7 @@ at the given address and does some RPCs" class StatusSubscriber(thread_capnp.Example.StatusSubscriber.Server): """An implementation of the StatusSubscriber interface""" - def status(self, value, **kwargs): + async def status(self, value, **kwargs): print("status: {}".format(time.time())) diff --git a/examples/async_reconnecting_ssl_client.py b/examples/async_reconnecting_ssl_client.py index c5f40d1..3d3acf7 100755 --- a/examples/async_reconnecting_ssl_client.py +++ b/examples/async_reconnecting_ssl_client.py @@ -16,8 +16,7 @@ this_dir = os.path.dirname(os.path.abspath(__file__)) def parse_args(): parser = argparse.ArgumentParser( - usage="Connects to the Example thread server \ -at the given address and does some RPCs" + usage="Connects to the Example thread server at the given address and does some RPCs" ) parser.add_argument("host", help="HOST:PORT") diff --git a/examples/async_server.py b/examples/async_server.py index 00e85f6..5d5b63d 100755 --- a/examples/async_server.py +++ b/examples/async_server.py @@ -16,23 +16,21 @@ class ExampleImpl(thread_capnp.Example.Server): "Implementation of the Example threading Cap'n Proto interface." async def subscribeStatus(self, subscriber, **kwargs): - await asyncio.sleep(1) + await asyncio.sleep(0.1) await subscriber.status(True) await self.subscribeStatus(subscriber) async def longRunning(self, **kwargs): - await asyncio.sleep(1) + await asyncio.sleep(0.1) async def new_connection(stream): - server = capnp.TwoPartyServer(stream, bootstrap=ExampleImpl()) - await server.on_disconnect() + await capnp.TwoPartyServer(stream, bootstrap=ExampleImpl()).on_disconnect() def parse_args(): parser = argparse.ArgumentParser( - usage="""Runs the server bound to the\ - given address/port ADDRESS. """ + usage="""Runs the server bound to the given address/port ADDRESS. """ ) parser.add_argument("address", help="ADDRESS:PORT") diff --git a/examples/async_socket_message_client.py b/examples/async_socket_message_client.py index ef08f33..bb8397e 100644 --- a/examples/async_socket_message_client.py +++ b/examples/async_socket_message_client.py @@ -9,8 +9,7 @@ import addressbook_capnp def parse_args(): parser = argparse.ArgumentParser( - usage="Connects to the Example thread server \ -at the given address and does some RPCs" + usage="Connects to the Example thread server at the given address and does some RPCs" ) parser.add_argument("host", help="HOST:PORT") diff --git a/examples/async_socket_message_server.py b/examples/async_socket_message_server.py index e261d7e..8896ff0 100644 --- a/examples/async_socket_message_server.py +++ b/examples/async_socket_message_server.py @@ -43,8 +43,7 @@ async def new_connection(stream): def parse_args(): parser = argparse.ArgumentParser( - usage="""Runs the server bound to the\ -given address/port ADDRESS. """ + usage="""Runs the server bound to the given address/port ADDRESS. """ ) parser.add_argument("address", help="ADDRESS:PORT") diff --git a/examples/async_ssl_calculator_client.py b/examples/async_ssl_calculator_client.py index 342701a..63e3d8d 100755 --- a/examples/async_ssl_calculator_client.py +++ b/examples/async_ssl_calculator_client.py @@ -19,7 +19,7 @@ class PowerFunction(calculator_capnp.Calculator.Function.Server): we're implementing this on the client side and will pass a reference to the server. The server will then be able to make calls back to the client.""" - def call(self, params, **kwargs): + async def call(self, params, **kwargs): """Note the **kwargs. This is very necessary to include, since protocols can add parameters over time. Also, by default, a _context variable is passed to all server methods, but you can also return @@ -31,8 +31,7 @@ class PowerFunction(calculator_capnp.Calculator.Function.Server): def parse_args(): parser = argparse.ArgumentParser( - usage="Connects to the Calculator server \ -at the given address and does some RPCs" + usage="Connects to the Calculator server at the given address and does some RPCs" ) parser.add_argument("host", help="HOST:PORT") diff --git a/examples/async_ssl_calculator_server.py b/examples/async_ssl_calculator_server.py index 4404889..8657d72 100755 --- a/examples/async_ssl_calculator_server.py +++ b/examples/async_ssl_calculator_server.py @@ -17,15 +17,7 @@ logger.setLevel(logging.DEBUG) this_dir = os.path.dirname(os.path.abspath(__file__)) -def read_value(value): - """Helper function to asynchronously call read() on a Calculator::Value and - return a promise for the result. (In the future, the generated code might - include something like this automatically.)""" - - return value.read().then(lambda result: result.value) - - -def evaluate_impl(expression, params=None): +async def evaluate_impl(expression, params=None): """Implementation of CalculatorImpl::evaluate(), also shared by FunctionImpl::call(). In the latter case, `params` are the parameter values passed to the function; in the former case, `params` is just an @@ -34,26 +26,23 @@ def evaluate_impl(expression, params=None): which = expression.which() if which == "literal": - return capnp.Promise(expression.literal) + return expression.literal elif which == "previousResult": - return read_value(expression.previousResult) + return (await expression.previousResult.read()).value elif which == "parameter": assert expression.parameter < len(params) - return capnp.Promise(params[expression.parameter]) + return params[expression.parameter] elif which == "call": call = expression.call func = call.function # Evaluate each parameter. paramPromises = [evaluate_impl(param, params) for param in call.params] + vals = await asyncio.gather(*paramPromises) - joinedParams = capnp.join_promises(paramPromises) # When the parameters are complete, call the function. - ret = joinedParams.then(lambda vals: func.call(vals)).then( - lambda result: result.value - ) - - return ret + result = await func.call(vals) + return result.value else: raise ValueError("Unknown expression type: " + which) @@ -64,7 +53,7 @@ class ValueImpl(calculator_capnp.Calculator.Value.Server): def __init__(self, value): self.value = value - def read(self, **kwargs): + async def read(self, **kwargs): return self.value @@ -77,17 +66,14 @@ class FunctionImpl(calculator_capnp.Calculator.Function.Server): self.paramCount = paramCount self.body = body.as_builder() - def call(self, params, _context, **kwargs): + async def call(self, params, _context, **kwargs): """Note that we're returning a Promise object here, and bypassing the helper functionality that normally sets the results struct from the returned object. Instead, we set _context.results directly inside of another promise""" assert len(params) == self.paramCount - # using setattr because '=' is not allowed inside of lambdas - return evaluate_impl(self.body, params).then( - lambda value: setattr(_context.results, "value", value) - ) + return await evaluate_impl(self.body, params) class OperatorImpl(calculator_capnp.Calculator.Function.Server): @@ -98,7 +84,7 @@ class OperatorImpl(calculator_capnp.Calculator.Function.Server): def __init__(self, op): self.op = op - def call(self, params, **kwargs): + async def call(self, params, **kwargs): assert len(params) == 2 op = self.op @@ -118,22 +104,19 @@ class OperatorImpl(calculator_capnp.Calculator.Function.Server): class CalculatorImpl(calculator_capnp.Calculator.Server): "Implementation of the Calculator Cap'n Proto interface." - def evaluate(self, expression, _context, **kwargs): - return evaluate_impl(expression).then( - lambda value: setattr(_context.results, "value", ValueImpl(value)) - ) + async def evaluate(self, expression, _context, **kwargs): + return ValueImpl(await evaluate_impl(expression)) - def defFunction(self, paramCount, body, _context, **kwargs): + async def defFunction(self, paramCount, body, _context, **kwargs): return FunctionImpl(paramCount, body) - def getOperator(self, op, **kwargs): + async def getOperator(self, op, **kwargs): return OperatorImpl(op) def parse_args(): parser = argparse.ArgumentParser( - usage="""Runs the server bound to the\ -given address/port ADDRESS. """ + usage="""Runs the server bound to the given address/port ADDRESS. """ ) parser.add_argument("address", help="ADDRESS:PORT") @@ -142,8 +125,7 @@ given address/port ADDRESS. """ async def new_connection(stream): - server = capnp.TwoPartyServer(stream, bootstrap=CalculatorImpl()) - await server.on_disconnect() + await capnp.TwoPartyServer(stream, bootstrap=CalculatorImpl()).on_disconnect() async def main(): diff --git a/examples/async_ssl_server.py b/examples/async_ssl_server.py index fc3941a..9d629a5 100755 --- a/examples/async_ssl_server.py +++ b/examples/async_ssl_server.py @@ -20,24 +20,20 @@ this_dir = os.path.dirname(os.path.abspath(__file__)) class ExampleImpl(thread_capnp.Example.Server): "Implementation of the Example threading Cap'n Proto interface." - def subscribeStatus(self, subscriber, **kwargs): - return ( - capnp.getTimer() - .after_delay(10**9) - .then(lambda: subscriber.status(True)) - .then(lambda _: self.subscribeStatus(subscriber)) - ) + async def subscribeStatus(self, subscriber, **kwargs): + await asyncio.sleep(0.1) + await subscriber.status(True) + await self.subscribeStatus(subscriber) - def longRunning(self, **kwargs): - return capnp.getTimer().after_delay(1 * 10**9) + async def longRunning(self, **kwargs): + await asyncio.sleep(0.1) - def alive(self, **kwargs): + async def alive(self, **kwargs): return True async def new_connection(stream): - server = capnp.TwoPartyServer(stream, bootstrap=ExampleImpl()) - await server.on_disconnect() + await capnp.TwoPartyServer(stream, bootstrap=ExampleImpl()).on_disconnect() def parse_args(): diff --git a/examples/calculator_client.py b/examples/calculator_client.py deleted file mode 100755 index 9cb9fa4..0000000 --- a/examples/calculator_client.py +++ /dev/null @@ -1,304 +0,0 @@ -#!/usr/bin/env python3 - -import argparse -import capnp - -import calculator_capnp - - -class PowerFunction(calculator_capnp.Calculator.Function.Server): - - """An implementation of the Function interface wrapping pow(). Note that - we're implementing this on the client side and will pass a reference to - the server. The server will then be able to make calls back to the client.""" - - def call(self, params, **kwargs): - """Note the **kwargs. This is very necessary to include, since - protocols can add parameters over time. Also, by default, a _context - variable is passed to all server methods, but you can also return - results directly as python objects, and they'll be added to the - results struct in the correct order""" - - return pow(params[0], params[1]) - - -def parse_args(): - 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() - - -def main(host): - client = capnp.TwoPartyClient(host) - - # Bootstrap the server capability and cast it to the Calculator interface - calculator = client.bootstrap().cast_as(calculator_capnp.Calculator) - - """Make a request that just evaluates the literal value 123. - - What's interesting here is that evaluate() returns a "Value", which is - another interface and therefore points back to an object living on the - server. We then have to call read() on that object to read it. - However, even though we are making two RPC's, this block executes in - *one* network round trip because of promise pipelining: we do not wait - for the first call to complete before we send the second call to the - server.""" - - print("Evaluating a literal... ", end="") - - # Make the request. Note we are using the shorter function form (instead - # of evaluate_request), and we are passing a dictionary that represents a - # struct and its member to evaluate - eval_promise = calculator.evaluate({"literal": 123}) - - # This is equivalent to: - """ - request = calculator.evaluate_request() - request.expression.literal = 123 - - # Send it, which returns a promise for the result (without blocking). - eval_promise = request.send() - """ - - # Using the promise, create a pipelined request to call read() on the - # returned object. Note that here we are using the shortened method call - # syntax read(), which is mostly just sugar for read_request().send() - read_promise = eval_promise.value.read() - - # Now that we've sent all the requests, wait for the response. Until this - # point, we haven't waited at all! - response = read_promise.wait() - assert response.value == 123 - - print("PASS") - - """Make a request to evaluate 123 + 45 - 67. - - The Calculator interface requires that we first call getOperator() to - get the addition and subtraction functions, then call evaluate() to use - them. But, once again, we can get both functions, call evaluate(), and - then read() the result -- four RPCs -- in the time of *one* network - round trip, because of promise pipelining.""" - - print("Using add and subtract... ", end="") - - # Get the "add" function from the server. - add = calculator.getOperator(op="add").func - # Get the "subtract" function from the server. - subtract = calculator.getOperator(op="subtract").func - - # Build the request to evaluate 123 + 45 - 67. Note the form is 'evaluate' - # + '_request', where 'evaluate' is the name of the method we want to call - request = calculator.evaluate_request() - subtract_call = request.expression.init("call") - subtract_call.function = subtract - subtract_params = subtract_call.init("params", 2) - subtract_params[1].literal = 67.0 - - add_call = subtract_params[0].init("call") - add_call.function = add - add_params = add_call.init("params", 2) - add_params[0].literal = 123 - add_params[1].literal = 45 - - # Send the evaluate() request, read() the result, and wait for read() to finish. - eval_promise = request.send() - read_promise = eval_promise.value.read() - - response = read_promise.wait() - assert response.value == 101 - - print("PASS") - - """ - Note: a one liner version of building the previous request (I highly - recommend not doing it this way for such a complicated structure, but I - just wanted to demonstrate it is possible to set all of the fields with a - dictionary): - - eval_promise = calculator.evaluate( -{'call': {'function': subtract, - 'params': [{'call': {'function': add, - 'params': [{'literal': 123}, - {'literal': 45}]}}, - {'literal': 67.0}]}}) - """ - - """Make a request to evaluate 4 * 6, then use the result in two more - requests that add 3 and 5. - - Since evaluate() returns its result wrapped in a `Value`, we can pass - that `Value` back to the server in subsequent requests before the first - `evaluate()` has actually returned. Thus, this example again does only - one network round trip.""" - - print("Pipelining eval() calls... ", end="") - - # Get the "add" function from the server. - add = calculator.getOperator(op="add").func - # Get the "multiply" function from the server. - multiply = calculator.getOperator(op="multiply").func - - # Build the request to evaluate 4 * 6 - request = calculator.evaluate_request() - - multiply_call = request.expression.init("call") - multiply_call.function = multiply - multiply_params = multiply_call.init("params", 2) - multiply_params[0].literal = 4 - multiply_params[1].literal = 6 - - multiply_result = request.send().value - - # Use the result in two calls that add 3 and add 5. - - add_3_request = calculator.evaluate_request() - add_3_call = add_3_request.expression.init("call") - add_3_call.function = add - add_3_params = add_3_call.init("params", 2) - add_3_params[0].previousResult = multiply_result - add_3_params[1].literal = 3 - add_3_promise = add_3_request.send().value.read() - - add_5_request = calculator.evaluate_request() - add_5_call = add_5_request.expression.init("call") - add_5_call.function = add - add_5_params = add_5_call.init("params", 2) - add_5_params[0].previousResult = multiply_result - add_5_params[1].literal = 5 - add_5_promise = add_5_request.send().value.read() - - # Now wait for the results. - assert add_3_promise.wait().value == 27 - assert add_5_promise.wait().value == 29 - - print("PASS") - - """Our calculator interface supports defining functions. Here we use it - to define two functions and then make calls to them as follows: - - f(x, y) = x * 100 + y - g(x) = f(x, x + 1) * 2; - f(12, 34) - g(21) - - Once again, the whole thing takes only one network round trip.""" - - print("Defining functions... ", end="") - - # Get the "add" function from the server. - add = calculator.getOperator(op="add").func - # Get the "multiply" function from the server. - multiply = calculator.getOperator(op="multiply").func - - # Define f. - request = calculator.defFunction_request() - request.paramCount = 2 - - # Build the function body. - add_call = request.body.init("call") - add_call.function = add - add_params = add_call.init("params", 2) - add_params[1].parameter = 1 # y - - multiply_call = add_params[0].init("call") - multiply_call.function = multiply - multiply_params = multiply_call.init("params", 2) - multiply_params[0].parameter = 0 # x - multiply_params[1].literal = 100 - - f = request.send().func - - # Define g. - request = calculator.defFunction_request() - request.paramCount = 1 - - # Build the function body. - multiply_call = request.body.init("call") - multiply_call.function = multiply - multiply_params = multiply_call.init("params", 2) - multiply_params[1].literal = 2 - - f_call = multiply_params[0].init("call") - f_call.function = f - f_params = f_call.init("params", 2) - f_params[0].parameter = 0 - - add_call = f_params[1].init("call") - add_call.function = add - add_params = add_call.init("params", 2) - add_params[0].parameter = 0 - add_params[1].literal = 1 - - g = request.send().func - - # OK, we've defined all our functions. Now create our eval requests. - - # f(12, 34) - f_eval_request = calculator.evaluate_request() - f_call = f_eval_request.expression.init("call") - f_call.function = f - f_params = f_call.init("params", 2) - f_params[0].literal = 12 - f_params[1].literal = 34 - f_eval_promise = f_eval_request.send().value.read() - - # g(21) - g_eval_request = calculator.evaluate_request() - g_call = g_eval_request.expression.init("call") - g_call.function = g - g_call.init("params", 1)[0].literal = 21 - g_eval_promise = g_eval_request.send().value.read() - - # Wait for the results. - assert f_eval_promise.wait().value == 1234 - assert g_eval_promise.wait().value == 4244 - - print("PASS") - - """Make a request that will call back to a function defined locally. - - Specifically, we will compute 2^(4 + 5). However, exponent is not - defined by the Calculator server. So, we'll implement the Function - interface locally and pass it to the server for it to use when - evaluating the expression. - - This example requires two network round trips to complete, because the - server calls back to the client once before finishing. In this - particular case, this could potentially be optimized by using a tail - call on the server side -- see CallContext::tailCall(). However, to - keep the example simpler, we haven't implemented this optimization in - the sample server.""" - - print("Using a callback... ", end="") - - # Get the "add" function from the server. - add = calculator.getOperator(op="add").func - - # Build the eval request for 2^(4+5). - request = calculator.evaluate_request() - - pow_call = request.expression.init("call") - pow_call.function = PowerFunction() - pow_params = pow_call.init("params", 2) - pow_params[0].literal = 2 - - add_call = pow_params[1].init("call") - add_call.function = add - add_params = add_call.init("params", 2) - add_params[0].literal = 4 - add_params[1].literal = 5 - - # Send the request and wait. - response = request.send().value.read().wait() - assert response.value == 512 - - print("PASS") - - -if __name__ == "__main__": - main(parse_args().host) diff --git a/examples/calculator_server.py b/examples/calculator_server.py deleted file mode 100755 index 8464439..0000000 --- a/examples/calculator_server.py +++ /dev/null @@ -1,145 +0,0 @@ -#!/usr/bin/env python3 - -import argparse -import capnp -import time - -import calculator_capnp - - -def read_value(value): - """Helper function to asynchronously call read() on a Calculator::Value and - return a promise for the result. (In the future, the generated code might - include something like this automatically.)""" - - return value.read().then(lambda result: result.value) - - -def evaluate_impl(expression, params=None): - """Implementation of CalculatorImpl::evaluate(), also shared by - FunctionImpl::call(). In the latter case, `params` are the parameter - values passed to the function; in the former case, `params` is just an - empty list.""" - - which = expression.which() - - if which == "literal": - return capnp.Promise(expression.literal) - elif which == "previousResult": - return read_value(expression.previousResult) - elif which == "parameter": - assert expression.parameter < len(params) - return capnp.Promise(params[expression.parameter]) - elif which == "call": - call = expression.call - func = call.function - - # Evaluate each parameter. - paramPromises = [evaluate_impl(param, params) for param in call.params] - - joinedParams = capnp.join_promises(paramPromises) - # When the parameters are complete, call the function. - ret = joinedParams.then(lambda vals: func.call(vals)).then( - lambda result: result.value - ) - - return ret - else: - raise ValueError("Unknown expression type: " + which) - - -class ValueImpl(calculator_capnp.Calculator.Value.Server): - "Simple implementation of the Calculator.Value Cap'n Proto interface." - - def __init__(self, value): - self.value = value - - def read(self, **kwargs): - return self.value - - -class FunctionImpl(calculator_capnp.Calculator.Function.Server): - - """Implementation of the Calculator.Function Cap'n Proto interface, where the - function is defined by a Calculator.Expression.""" - - def __init__(self, paramCount, body): - self.paramCount = paramCount - self.body = body.as_builder() - - def call(self, params, _context, **kwargs): - """Note that we're returning a Promise object here, and bypassing the - helper functionality that normally sets the results struct from the - returned object. Instead, we set _context.results directly inside of - another promise""" - - assert len(params) == self.paramCount - # using setattr because '=' is not allowed inside of lambdas - return evaluate_impl(self.body, params).then( - lambda value: setattr(_context.results, "value", value) - ) - - -class OperatorImpl(calculator_capnp.Calculator.Function.Server): - - """Implementation of the Calculator.Function Cap'n Proto interface, wrapping - basic binary arithmetic operators.""" - - def __init__(self, op): - self.op = op - - def call(self, params, **kwargs): - assert len(params) == 2 - - op = self.op - - if op == "add": - return params[0] + params[1] - elif op == "subtract": - return params[0] - params[1] - elif op == "multiply": - return params[0] * params[1] - elif op == "divide": - return params[0] / params[1] - else: - raise ValueError("Unknown operator") - - -class CalculatorImpl(calculator_capnp.Calculator.Server): - "Implementation of the Calculator Cap'n Proto interface." - - def evaluate(self, expression, _context, **kwargs): - return evaluate_impl(expression).then( - lambda value: setattr(_context.results, "value", ValueImpl(value)) - ) - - def defFunction(self, paramCount, body, _context, **kwargs): - return FunctionImpl(paramCount, body) - - def getOperator(self, op, **kwargs): - return OperatorImpl(op) - - -def parse_args(): - parser = argparse.ArgumentParser( - usage="""Runs the server bound to the\ -given address/port ADDRESS may be '*' to bind to all local addresses.\ -:PORT may be omitted to choose a port automatically. """ - ) - - parser.add_argument("address", help="ADDRESS[:PORT]") - - return parser.parse_args() - - -def main(): - address = parse_args().address - - server = capnp.TwoPartyServer(address, bootstrap=CalculatorImpl()) - while True: - server.poll_once() - time.sleep(0.001) - - -if __name__ == "__main__": - main() diff --git a/examples/thread_client.py b/examples/thread_client.py deleted file mode 100755 index 317bc62..0000000 --- a/examples/thread_client.py +++ /dev/null @@ -1,56 +0,0 @@ -#!/usr/bin/env python3 - -import argparse -import threading -import time -import capnp - -import thread_capnp - - -def parse_args(): - parser = argparse.ArgumentParser( - usage="Connects to the Example thread server \ -at the given address and does some RPCs" - ) - parser.add_argument("host", help="HOST:PORT") - - return parser.parse_args() - - -class StatusSubscriber(thread_capnp.Example.StatusSubscriber.Server): - - """An implementation of the StatusSubscriber interface""" - - def status(self, value, **kwargs): - print("status: {}".format(time.time())) - - -def start_status_thread(host): - client = capnp.TwoPartyClient(host) - cap = client.bootstrap().cast_as(thread_capnp.Example) - - subscriber = StatusSubscriber() - promise = cap.subscribeStatus(subscriber) - promise.wait() - - -def main(host): - client = capnp.TwoPartyClient(host) - cap = client.bootstrap().cast_as(thread_capnp.Example) - - status_thread = threading.Thread(target=start_status_thread, args=(host,)) - status_thread.daemon = True - status_thread.start() - - print("main: {}".format(time.time())) - cap.longRunning().wait() - print("main: {}".format(time.time())) - cap.longRunning().wait() - print("main: {}".format(time.time())) - cap.longRunning().wait() - print("main: {}".format(time.time())) - - -if __name__ == "__main__": - main(parse_args().host) diff --git a/examples/thread_server.py b/examples/thread_server.py deleted file mode 100755 index 25b2ae0..0000000 --- a/examples/thread_server.py +++ /dev/null @@ -1,44 +0,0 @@ -#!/usr/bin/env python3 - -import argparse -import capnp - -import thread_capnp - - -class ExampleImpl(thread_capnp.Example.Server): - "Implementation of the Example threading Cap'n Proto interface." - - def subscribeStatus(self, subscriber, **kwargs): - return ( - capnp.getTimer() - .after_delay(10**9) - .then(lambda: subscriber.status(True)) - .then(lambda _: self.subscribeStatus(subscriber)) - ) - - def longRunning(self, **kwargs): - return capnp.getTimer().after_delay(1 * 10**9) - - -def parse_args(): - parser = argparse.ArgumentParser( - usage="""Runs the server bound to the\ -given address/port ADDRESS may be '*' to bind to all local addresses.\ -:PORT may be omitted to choose a port automatically. """ - ) - - parser.add_argument("address", help="ADDRESS[:PORT]") - - return parser.parse_args() - - -def main(): - address = parse_args().address - - server = capnp.TwoPartyServer(address, bootstrap=ExampleImpl()) - server.run_forever() - - -if __name__ == "__main__": - main() diff --git a/pyproject.toml b/pyproject.toml index 801edb2..fd9aa9d 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,2 +1,5 @@ [build-system] requires = ["setuptools", "wheel", "pkgconfig", "cython"] + +[tool.pytest.ini_options] +asyncio_mode = "auto" \ No newline at end of file diff --git a/test/test_capability.py b/test/test_capability.py index 01c6f25..158ec7e 100644 --- a/test/test_capability.py +++ b/test/test_capability.py @@ -1,5 +1,4 @@ import pytest -import time import capnp import test_capability_capnp as capability @@ -9,37 +8,35 @@ class Server(capability.TestInterface.Server): def __init__(self, val=1): self.val = val - def foo(self, i, j, **kwargs): + async def foo(self, i, j, **kwargs): extra = 0 if j: extra = 1 return str(i * 5 + extra + self.val) - def buz(self, i, **kwargs): + async def buz(self, i, **kwargs): return i.host + "_test" - def bam(self, i, **kwargs): + async def bam(self, i, **kwargs): return str(i) + "_test", i class PipelineServer(capability.TestPipeline.Server): - def getCap(self, n, inCap, _context, **kwargs): - def _then(response): - _results = _context.results - _results.s = response.x + "_foo" - _results.outBox.cap = Server(100) - - return inCap.foo(i=n).then(_then) + async def getCap(self, n, inCap, _context, **kwargs): + response = await inCap.foo(i=n) + _results = _context.results + _results.s = response.x + "_foo" + _results.outBox.cap = Server(100) -def test_client(): +async def test_client(): client = capability.TestInterface._new_client(Server()) req = client._request("foo") req.i = 5 remote = req.send() - response = remote.wait() + response = await remote assert response.x == "26" @@ -47,7 +44,7 @@ def test_client(): req.i = 5 remote = req.send() - response = remote.wait() + response = await remote assert response.x == "26" @@ -65,46 +62,46 @@ def test_client(): req.baz = 1 -def test_simple_client(): +async def test_simple_client(): client = capability.TestInterface._new_client(Server()) remote = client._send("foo", i=5) - response = remote.wait() + response = await remote assert response.x == "26" remote = client.foo(i=5) - response = remote.wait() + response = await remote assert response.x == "26" remote = client.foo(i=5, j=True) - response = remote.wait() + response = await remote assert response.x == "27" remote = client.foo(5) - response = remote.wait() + response = await remote assert response.x == "26" remote = client.foo(5, True) - response = remote.wait() + response = await remote assert response.x == "27" remote = client.foo(5, j=True) - response = remote.wait() + response = await remote assert response.x == "27" remote = client.buz(capability.TestSturdyRefHostId.new_message(host="localhost")) - response = remote.wait() + response = await remote assert response.x == "localhost_test" remote = client.bam(i=5) - response = remote.wait() + response = await remote assert response.x == "5_test" assert response.i == 5 @@ -125,7 +122,7 @@ def test_simple_client(): remote = client.foo(baz=5) -def test_pipeline(): +async def test_pipeline(): client = capability.TestPipeline._new_client(PipelineServer()) foo_client = capability.TestInterface._new_client(Server()) @@ -134,10 +131,10 @@ def test_pipeline(): outCap = remote.outBox.cap pipelinePromise = outCap.foo(i=10) - response = pipelinePromise.wait() + response = await pipelinePromise assert response.x == "150" - response = remote.wait() + response = await remote assert response.s == "26_foo" @@ -145,47 +142,42 @@ class BadServer(capability.TestInterface.Server): def __init__(self, val=1): self.val = val - def foo(self, i, j, **kwargs): + async def foo(self, i, j, **kwargs): extra = 0 if j: extra = 1 return str(i * 5 + extra + self.val), 10 # returning too many args -def test_exception_client(): +async def test_exception_client(): client = capability.TestInterface._new_client(BadServer()) remote = client._send("foo", i=5) with pytest.raises(capnp.KjException): - remote.wait() + await remote class BadPipelineServer(capability.TestPipeline.Server): - def getCap(self, n, inCap, _context, **kwargs): - def _then(response): - _results = _context.results - _results.s = response.x + "_foo" - _results.outBox.cap = Server(100) - - def _error(error): + async def getCap(self, n, inCap, _context, **kwargs): + try: + await inCap.foo(i=n) + except capnp.KjException: raise Exception("test was a success") - return inCap.foo(i=n).then(_then, _error) - -def test_exception_chain(): +async def test_exception_chain(): client = capability.TestPipeline._new_client(BadPipelineServer()) foo_client = capability.TestInterface._new_client(BadServer()) remote = client.getCap(n=5, inCap=foo_client) try: - remote.wait() + await remote except Exception as e: assert "test was a success" in str(e) -def test_pipeline_exception(): +async def test_pipeline_exception(): client = capability.TestPipeline._new_client(BadPipelineServer()) foo_client = capability.TestInterface._new_client(BadServer()) @@ -195,13 +187,13 @@ def test_pipeline_exception(): pipelinePromise = outCap.foo(i=10) with pytest.raises(Exception): - pipelinePromise.wait() + await pipelinePromise with pytest.raises(Exception): - remote.wait() + await remote -def test_casting(): +async def test_casting(): client = capability.TestExtends._new_client(Server()) client2 = client.upcast(capability.TestInterface) _ = client2.cast_as(capability.TestInterface) @@ -214,7 +206,7 @@ class TailCallOrder(capability.TestCallOrder.Server): def __init__(self): self.count = -1 - def getCallSequence(self, expected, **kwargs): + async def getCallSequence(self, expected, **kwargs): self.count += 1 return self.count @@ -223,18 +215,18 @@ class TailCaller(capability.TestTailCaller.Server): def __init__(self): self.count = 0 - def foo(self, i, callee, _context, **kwargs): + async def foo(self, i, callee, _context, **kwargs): self.count += 1 tail = callee.foo_request(i=i, t="from TailCaller") - return _context.tail_call(tail) + return await _context.tail_call(tail) class TailCallee(capability.TestTailCallee.Server): def __init__(self): self.count = 0 - def foo(self, i, t, _context, **kwargs): + async def foo(self, i, t, _context, **kwargs): self.count += 1 results = _context.results @@ -243,7 +235,7 @@ class TailCallee(capability.TestTailCallee.Server): results.c = TailCallOrder() -def test_tail_call(): +async def test_tail_call(): callee_server = TailCallee() caller_server = TailCaller() @@ -253,7 +245,7 @@ def test_tail_call(): promise = caller.foo(i=456, callee=callee) dependent_call1 = promise.c.getCallSequence() - response = promise.wait() + response = await promise assert response.i == 456 assert response.i == 456 @@ -261,18 +253,18 @@ def test_tail_call(): dependent_call2 = response.c.getCallSequence() dependent_call3 = response.c.getCallSequence() - result = dependent_call1.wait() + result = await dependent_call1 assert result.n == 0 - result = dependent_call2.wait() + result = await dependent_call2 assert result.n == 1 - result = dependent_call3.wait() + result = await dependent_call3 assert result.n == 2 assert callee_server.count == 1 assert caller_server.count == 1 -def test_cancel(): +async def test_cancel(): client = capability.TestInterface._new_client(Server()) req = client._request("foo") @@ -282,167 +274,100 @@ def test_cancel(): remote.cancel() with pytest.raises(Exception): - remote.wait() + await remote req = client.foo(5) - trans = req.then(lambda x: 5) + await req req.cancel() # Cancel a promise that was already consumed - assert trans.wait() == 5 req = client.foo(5) req.cancel() with pytest.raises(Exception): - trans = req.then(lambda x: 5) + await req req = client.foo(5) - assert req.wait().x == "26" + assert (await req).x == "26" with pytest.raises(Exception): - req.wait() + await req -def test_timer(): - global test_timer_var - test_timer_var = False - - def set_timer_var(): - global test_timer_var - test_timer_var = True - - capnp.getTimer().after_delay(1).then(set_timer_var).wait() - - assert test_timer_var is True - - test_timer_var = False - promise = ( - capnp.Promise(0) - .then(lambda x: time.sleep(0.1)) - .then(lambda x: time.sleep(0.1)) - .then(lambda x: set_timer_var()) - ) - - canceller = capnp.getTimer().after_delay(1).then(lambda: promise.cancel()) - - joined = capnp.join_promises([canceller, promise]) - joined.wait() - - # faling for now, not sure why... - # assert test_timer_var is False - - -def test_double_send(): +async def test_double_send(): client = capability.TestInterface._new_client(Server()) req = client._request("foo") req.i = 5 - req.send() + await req.send() with pytest.raises(Exception): - req.send() - - -def test_then_args(): - capnp.Promise(0).then(lambda x: 1) - - with pytest.raises(Exception): - capnp.Promise(0).then(lambda: 1) - - with pytest.raises(Exception): - capnp.Promise(0).then(lambda x, y: 1) - - capnp.getTimer().after_delay(1).then(lambda: 1) # after_delay is a VoidPromise - - with pytest.raises(Exception): - capnp.getTimer().after_delay(1).then(lambda x: 1) - - client = capability.TestInterface._new_client(Server()) - - client.foo(i=5).then(lambda x: 1) - - with pytest.raises(Exception): - client.foo(i=5).then(lambda: 1) - - with pytest.raises(Exception): - client.foo(i=5).then(lambda x, y: 1) + await req.send() class PromiseJoinServer(capability.TestPipeline.Server): - def getCap(self, n, inCap, _context, **kwargs): - def _then(response): - _results = _context.results - _results.s = response.x + "_bar" - _results.outBox.cap = inCap - - return ( - inCap.foo(i=n) - .then( - lambda res: capnp.Promise(int(res.x)) - ) # Make sure that Promise is flattened - .then( - lambda x: inCap.foo(i=x + 1) - ) # Make sure that RemotePromise is flattened - .then(_then) - ) + async def getCap(self, n, inCap, _context, **kwargs): + res = await inCap.foo(i=n) + response = await inCap.foo(i=int(res.x) + 1) + _results = _context.results + _results.s = response.x + "_bar" + _results.outBox.cap = inCap -def test_promise_joining(): +async def test_promise_joining(): client = capability.TestPipeline._new_client(PromiseJoinServer()) foo_client = capability.TestInterface._new_client(Server()) remote = client.getCap(n=5, inCap=foo_client) - assert remote.wait().s == "136_bar" + assert (await remote).s == "136_bar" class ExtendsServer(Server): - def qux(self, **kwargs): + async def qux(self, **kwargs): pass -def test_inheritance(): +async def test_inheritance(): client = capability.TestExtends._new_client(ExtendsServer()) - client.qux().wait() + await client.qux() remote = client.foo(i=5) - response = remote.wait() + response = await remote assert response.x == "26" class PassedCapTest(capability.TestPassedCap.Server): - def foo(self, cap, _context, **kwargs): - def set_result(res): - _context.results.x = res.x - - return cap.foo(5).then(set_result) + async def foo(self, cap, _context, **kwargs): + res = await cap.foo(5) + _context.results.x = res.x -def test_null_cap(): +async def test_null_cap(): client = capability.TestPassedCap._new_client(PassedCapTest()) - assert client.foo(Server()).wait().x == "26" + assert (await client.foo(Server())).x == "26" with pytest.raises(capnp.KjException): - client.foo().wait() + await client.foo() class StructArgTest(capability.TestStructArg.Server): - def bar(self, a, b, **kwargs): + async def bar(self, a, b, **kwargs): return a + str(b) -def test_struct_args(): +async def test_struct_args(): client = capability.TestStructArg._new_client(StructArgTest()) - assert client.bar(a="test", b=1).wait().c == "test1" + assert (await client.bar(a="test", b=1)).c == "test1" with pytest.raises(capnp.KjException): - assert client.bar("test", 1).wait().c == "test1" + assert (await client.bar("test", 1)).c == "test1" class GenericTest(capability.TestGeneric.Server): - def foo(self, a, **kwargs): + async def foo(self, a, **kwargs): return a.as_text() + "test" -def test_generic(): +async def test_generic(): client = capability.TestGeneric._new_client(GenericTest()) obj = capnp._MallocMessageBuilder().get_root_as_any() obj.set_as_text("anypointer_") - assert client.foo(obj).wait().b == "anypointer_test" + assert (await client.foo(obj)).b == "anypointer_test" diff --git a/test/test_capability_context.py b/test/test_capability_context.py index e669191..70c03c5 100644 --- a/test/test_capability_context.py +++ b/test/test_capability_context.py @@ -1,52 +1,38 @@ -import os import pytest import capnp - -this_dir = os.path.dirname(__file__) - -# flake8: noqa: E501 +import test_capability_capnp as capability -@pytest.fixture -def capability(): - capnp.cleanup_global_schema_parser() - return capnp.load(os.path.join(this_dir, "test_capability.capnp")) - - -class Server: +class Server(capability.TestInterface.Server): def __init__(self, val=1): self.val = val - def foo_context(self, context): + async def foo_context(self, context): extra = 0 if context.params.j: extra = 1 context.results.x = str(context.params.i * 5 + extra + self.val) - def buz_context(self, context): + async def buz_context(self, context): context.results.x = context.params.i.host + "_test" -class PipelineServer: - def getCap_context(self, context): - def _then(response): - context.results.s = response.x + "_foo" - context.results.outBox.cap = capability().TestInterface._new_server( - Server(100) - ) - - return context.params.inCap.foo(i=context.params.n).then(_then) +class PipelineServer(capability.TestPipeline.Server): + async def getCap_context(self, context): + response = await context.params.inCap.foo(i=context.params.n) + context.results.s = response.x + "_foo" + context.results.outBox.cap = Server(100) -def test_client_context(capability): +async def test_client_context(): client = capability.TestInterface._new_client(Server()) req = client._request("foo") req.i = 5 remote = req.send() - response = remote.wait() + response = await remote assert response.x == "26" @@ -54,7 +40,7 @@ def test_client_context(capability): req.i = 5 remote = req.send() - response = remote.wait() + response = await remote assert response.x == "26" @@ -72,41 +58,41 @@ def test_client_context(capability): req.baz = 1 -def test_simple_client_context(capability): +async def test_simple_client_context(): client = capability.TestInterface._new_client(Server()) remote = client._send("foo", i=5) - response = remote.wait() + response = await remote assert response.x == "26" remote = client.foo(i=5) - response = remote.wait() + response = await remote assert response.x == "26" remote = client.foo(i=5, j=True) - response = remote.wait() + response = await remote assert response.x == "27" remote = client.foo(5) - response = remote.wait() + response = await remote assert response.x == "26" remote = client.foo(5, True) - response = remote.wait() + response = await remote assert response.x == "27" remote = client.foo(5, j=True) - response = remote.wait() + response = await remote assert response.x == "27" remote = client.buz(capability.TestSturdyRefHostId.new_message(host="localhost")) - response = remote.wait() + response = await remote assert response.x == "localhost_test" @@ -126,15 +112,7 @@ def test_simple_client_context(capability): remote = client.foo(baz=5) -@pytest.mark.xfail -def test_pipeline_context(capability): - """ - E capnp.lib.capnp.KjException: capnp/lib/capnp.pyx:61: failed: :Fixture "capability" called directly. Fixtures are not meant to be called directly, - E but are created automatically when test functions request them as parameters. - E See https://docs.pytest.org/en/latest/fixture.html for more information about fixtures, and - E https://docs.pytest.org/en/latest/deprecations.html#calling-fixtures-directly about how to update your code. - E stack: 7f87c1ac6e40 7f87c17c3250 7f87c17be260 7f87c17c49f0 7f87c17c0f50 7f87c17c5540 7f87c17d7bf0 7f87c1acb768 7f87c1aaf185 7f87c1aaf2dc 7f87c1a6da1d 7f87c3895459 7f87c3895713 7f87c38c72eb 7f87c3901409 7f87c38b5767 7f87c38b6e7e 7f87c38fe48d 7f87c38b5767 7f87c38b6e7e 7f87c38fe48d 7f87c38b5767 7f87c38b67d2 7f87c38c71cf 7f87c38fdb77 7f87c38b5767 7f87c38b67d2 7f87c38c71cf 7f87c3901409 7f87c38b6632 7f87c38c71cf 7f87c3901409 - """ +async def test_pipeline_context(): client = capability.TestPipeline._new_client(PipelineServer()) foo_client = capability.TestInterface._new_client(Server()) @@ -143,57 +121,51 @@ def test_pipeline_context(capability): outCap = remote.outBox.cap pipelinePromise = outCap.foo(i=10) - response = pipelinePromise.wait() + response = await pipelinePromise assert response.x == "150" - response = remote.wait() + response = await remote assert response.s == "26_foo" -class BadServer: +class BadServer(capability.TestInterface.Server): def __init__(self, val=1): self.val = val - def foo_context(self, context): + async def foo_context(self, context): context.results.x = str(context.params.i * 5 + self.val) context.results.x2 = 5 # raises exception -def test_exception_client_context(capability): +async def test_exception_client_context(): client = capability.TestInterface._new_client(BadServer()) remote = client._send("foo", i=5) with pytest.raises(capnp.KjException): - remote.wait() + await remote -class BadPipelineServer: - def getCap_context(self, context): - def _then(response): - context.results.s = response.x + "_foo" - context.results.outBox.cap = capability().TestInterface._new_server( - Server(100) - ) - - def _error(error): +class BadPipelineServer(capability.TestPipeline.Server): + async def getCap_context(self, context): + try: + await context.params.inCap.foo(i=context.params.n) + except capnp.KjException: raise Exception("test was a success") - return context.params.inCap.foo(i=context.params.n).then(_then, _error) - -def test_exception_chain_context(capability): +async def test_exception_chain_context(): client = capability.TestPipeline._new_client(BadPipelineServer()) foo_client = capability.TestInterface._new_client(BadServer()) remote = client.getCap(n=5, inCap=foo_client) try: - remote.wait() + await remote except Exception as e: assert "test was a success" in str(e) -def test_pipeline_exception_context(capability): +async def test_pipeline_exception_context(): client = capability.TestPipeline._new_client(BadPipelineServer()) foo_client = capability.TestInterface._new_client(BadServer()) @@ -203,13 +175,13 @@ def test_pipeline_exception_context(capability): pipelinePromise = outCap.foo(i=10) with pytest.raises(Exception): - pipelinePromise.wait() + await pipelinePromise with pytest.raises(Exception): - remote.wait() + await remote -def test_casting_context(capability): +async def test_casting_context(): client = capability.TestExtends._new_client(Server()) client2 = client.upcast(capability.TestInterface) _ = client2.cast_as(capability.TestInterface) @@ -218,50 +190,42 @@ def test_casting_context(capability): client.upcast(capability.TestPipeline) -class TailCallOrder: +class TailCallOrder(capability.TestCallOrder.Server): def __init__(self): self.count = -1 - def getCallSequence_context(self, context): + async def getCallSequence_context(self, context): self.count += 1 context.results.n = self.count -class TailCaller: +class TailCaller(capability.TestTailCaller.Server): def __init__(self): self.count = 0 - def foo_context(self, context): + async def foo_context(self, context): self.count += 1 tail = context.params.callee.foo_request( i=context.params.i, t="from TailCaller" ) - return context.tail_call(tail) + await context.tail_call(tail) -class TailCallee: +class TailCallee(capability.TestTailCallee.Server): def __init__(self): self.count = 0 - def foo_context(self, context): + async def foo_context(self, context): self.count += 1 results = context.results results.i = context.params.i results.t = context.params.t - results.c = capability().TestCallOrder._new_server(TailCallOrder()) + results.c = TailCallOrder() -@pytest.mark.xfail -def test_tail_call(capability): - """ - E capnp.lib.capnp.KjException: capnp/lib/capnp.pyx:75: failed: :Fixture "capability" called directly. Fixtures are not meant to be called directly, - E but are created automatically when test functions request them as parameters. - E See https://docs.pytest.org/en/latest/fixture.html for more information about fixtures, and - E https://docs.pytest.org/en/latest/deprecations.html#calling-fixtures-directly about how to update your code. - E stack: 7f87c17c5540 7f87c17c51b0 7f87c17c5540 7f87c17d7bf0 7f87c1acb768 7f87c1aaf185 7f87c1aaf2dc 7f87c1a6da1d 7f87c3895459 7f87c3895713 7f87c38c72eb 7f87c3901409 7f87c38b5767 7f87c38b6e7e 7f87c38fe48d 7f87c38b5767 7f87c38b6e7e 7f87c38fe48d 7f87c38b5767 7f87c38b67d2 7f87c38c71cf 7f87c38fdb77 7f87c38b5767 7f87c38b67d2 7f87c38c71cf 7f87c3901409 7f87c38b6632 7f87c38c71cf 7f87c3901409 7f87c38b5767 7f87c38b6e7e 7f87c388ace7 - """ +async def test_tail_call(): callee_server = TailCallee() caller_server = TailCaller() @@ -271,7 +235,7 @@ def test_tail_call(capability): promise = caller.foo(i=456, callee=callee) dependent_call1 = promise.c.getCallSequence() - response = promise.wait() + response = await promise assert response.i == 456 assert response.i == 456 @@ -279,11 +243,11 @@ def test_tail_call(capability): dependent_call2 = response.c.getCallSequence() dependent_call3 = response.c.getCallSequence() - result = dependent_call1.wait() + result = await dependent_call1 assert result.n == 0 - result = dependent_call2.wait() + result = await dependent_call2 assert result.n == 1 - result = dependent_call3.wait() + result = await dependent_call3 assert result.n == 2 assert callee_server.count == 1 diff --git a/test/test_capability_old.py b/test/test_capability_old.py deleted file mode 100644 index c99e493..0000000 --- a/test/test_capability_old.py +++ /dev/null @@ -1,287 +0,0 @@ -import os -import pytest - -import capnp - -this_dir = os.path.dirname(__file__) - -# flake8: noqa: E501 - - -@pytest.fixture -def capability(): - return capnp.load(os.path.join(this_dir, "test_capability.capnp")) - - -class Server: - def __init__(self, val=1): - self.val = val - - def foo(self, i, j, **kwargs): - extra = 0 - if j: - extra = 1 - return str(i * 5 + extra + self.val) - - def buz(self, i, **kwargs): - return i.host + "_test" - - -class PipelineServer: - def getCap(self, n, inCap, _context, **kwargs): - def _then(response): - _results = _context.results - _results.s = response.x + "_foo" - _results.outBox.cap = capability().TestInterface._new_server(Server(100)) - - return inCap.foo(i=n).then(_then) - - -def test_client(capability): - client = capability.TestInterface._new_client(Server()) - - req = client._request("foo") - req.i = 5 - - remote = req.send() - response = remote.wait() - - assert response.x == "26" - - req = client.foo_request() - req.i = 5 - - remote = req.send() - response = remote.wait() - - assert response.x == "26" - - with pytest.raises(AttributeError): - client.foo2_request() - - req = client.foo_request() - - with pytest.raises(Exception): - req.i = "foo" - - req = client.foo_request() - - with pytest.raises(AttributeError): - req.baz = 1 - - -def test_simple_client(capability): - client = capability.TestInterface._new_client(Server()) - - remote = client._send("foo", i=5) - response = remote.wait() - - assert response.x == "26" - - remote = client.foo(i=5) - response = remote.wait() - - assert response.x == "26" - - remote = client.foo(i=5, j=True) - response = remote.wait() - - assert response.x == "27" - - remote = client.foo(5) - response = remote.wait() - - assert response.x == "26" - - remote = client.foo(5, True) - response = remote.wait() - - assert response.x == "27" - - remote = client.foo(5, j=True) - response = remote.wait() - - assert response.x == "27" - - remote = client.buz(capability.TestSturdyRefHostId.new_message(host="localhost")) - response = remote.wait() - - assert response.x == "localhost_test" - - with pytest.raises(Exception): - remote = client.foo(5, 10) - - with pytest.raises(Exception): - remote = client.foo(5, True, 100) - - with pytest.raises(Exception): - remote = client.foo(i="foo") - - with pytest.raises(AttributeError): - remote = client.foo2(i=5) - - with pytest.raises(Exception): - remote = client.foo(baz=5) - - -@pytest.mark.xfail -def test_pipeline(capability): - """ - E capnp.lib.capnp.KjException: capnp/lib/capnp.pyx:61: failed: :Fixture "capability" called directly. Fixtures are not meant to be called directly, - E but are created automatically when test functions request them as parameters. - E See https://docs.pytest.org/en/latest/fixture.html for more information about fixtures, and - E https://docs.pytest.org/en/latest/deprecations.html#calling-fixtures-directly about how to update your code. - E stack: 7f680f7fce40 7f680f4f9250 7f680f4f4260 7f680f4fa9f0 7f680f4f6f50 7f680f4fb540 7f680f50dbf0 7f680f801768 7f680f7e5185 7f680f7e52dc 7f680f7a3a1d 7f68115cb459 7f68115cb713 7f68115fd2eb 7f6811637409 7f68115eb767 7f68115ece7e 7f681163448d 7f68115eb767 7f68115ece7e 7f681163448d 7f68115eb767 7f68115ec7d2 7f68115fd1cf 7f6811633b77 7f68115eb767 7f68115ec7d2 7f68115fd1cf 7f6811637409 7f68115ec632 7f68115fd1cf 7f6811637409 - """ - client = capability.TestPipeline._new_client(PipelineServer()) - foo_client = capability.TestInterface._new_client(Server()) - - remote = client.getCap(n=5, inCap=foo_client) - - outCap = remote.outBox.cap - pipelinePromise = outCap.foo(i=10) - - response = pipelinePromise.wait() - assert response.x == "150" - - response = remote.wait() - assert response.s == "26_foo" - - -class BadServer: - def __init__(self, val=1): - self.val = val - - def foo(self, i, j, **kwargs): - extra = 0 - if j: - extra = 1 - return str(i * 5 + extra + self.val), 10 # returning too many args - - -def test_exception_client(capability): - client = capability.TestInterface._new_client(BadServer()) - - remote = client._send("foo", i=5) - with pytest.raises(capnp.KjException): - remote.wait() - - -class BadPipelineServer: - def getCap(self, n, inCap, _context, **kwargs): - def _then(response): - _results = _context.results - _results.s = response.x + "_foo" - _results.outBox.cap = capability().TestInterface._new_server(Server(100)) - - def _error(error): - raise Exception("test was a success") - - return inCap.foo(i=n).then(_then, _error) - - -def test_exception_chain(capability): - client = capability.TestPipeline._new_client(BadPipelineServer()) - foo_client = capability.TestInterface._new_client(BadServer()) - - remote = client.getCap(n=5, inCap=foo_client) - - try: - remote.wait() - except Exception as e: - assert "test was a success" in str(e) - - -def test_pipeline_exception(capability): - client = capability.TestPipeline._new_client(BadPipelineServer()) - foo_client = capability.TestInterface._new_client(BadServer()) - - remote = client.getCap(n=5, inCap=foo_client) - - outCap = remote.outBox.cap - pipelinePromise = outCap.foo(i=10) - - with pytest.raises(Exception): - pipelinePromise.wait() - - with pytest.raises(Exception): - remote.wait() - - -def test_casting(capability): - client = capability.TestExtends._new_client(Server()) - client2 = client.upcast(capability.TestInterface) - _ = client2.cast_as(capability.TestInterface) - - with pytest.raises(Exception): - client.upcast(capability.TestPipeline) - - -class TailCallOrder: - def __init__(self): - self.count = -1 - - def getCallSequence(self, expected, **kwargs): - self.count += 1 - return self.count - - -class TailCaller: - def __init__(self): - self.count = 0 - - def foo(self, i, callee, _context, **kwargs): - self.count += 1 - - tail = callee.foo_request(i=i, t="from TailCaller") - return _context.tail_call(tail) - - -class TailCallee: - def __init__(self): - self.count = 0 - - def foo(self, i, t, _context, **kwargs): - self.count += 1 - - results = _context.results - results.i = i - results.t = t - results.c = capability().TestCallOrder._new_server(TailCallOrder()) - - -@pytest.mark.xfail -def test_tail_call(capability): - """ - E capnp.lib.capnp.KjException: capnp/lib/capnp.pyx:104: failed: :Fixture "capability" called directly. Fixtures are not meant to be called directly, - E but are created automatically when test functions request them as parameters. - E See https://docs.pytest.org/en/latest/fixture.html for more information about fixtures, and - E https://docs.pytest.org/en/latest/deprecations.html#calling-fixtures-directly about how to update your code. - E stack: 7f680f4fb540 7f680f4fb1b0 7f680f4fb540 7f680f50dbf0 7f680f801768 7f680f7e5185 7f680f7e52dc 7f680f7a3a1d 7f68115cb459 7f68115cb713 7f68115fd2eb 7f6811637409 7f68115eb767 7f68115ece7e 7f681163448d 7f68115eb767 7f68115ece7e 7f681163448d 7f68115eb767 7f68115ec7d2 7f68115fd1cf 7f6811633b77 7f68115eb767 7f68115ec7d2 7f68115fd1cf 7f6811637409 7f68115ec632 7f68115fd1cf 7f6811637409 7f68115eb767 7f68115ece7e 7f68115c0ce7 - """ - callee_server = TailCallee() - caller_server = TailCaller() - - callee = capability.TestTailCallee._new_client(callee_server) - caller = capability.TestTailCaller._new_client(caller_server) - - promise = caller.foo(i=456, callee=callee) - dependent_call1 = promise.c.getCallSequence() - - response = promise.wait() - - assert response.i == 456 - assert response.i == 456 - - dependent_call2 = response.c.getCallSequence() - dependent_call3 = response.c.getCallSequence() - - result = dependent_call1.wait() - assert result.n == 0 - result = dependent_call2.wait() - assert result.n == 1 - result = dependent_call3.wait() - assert result.n == 2 - - assert callee_server.count == 1 - assert caller_server.count == 1 diff --git a/test/test_examples.py b/test/test_examples.py index 40d496b..387eec9 100644 --- a/test/test_examples.py +++ b/test/test_examples.py @@ -117,23 +117,13 @@ def run_subprocesses( serverp.kill() -def test_async_calculator_example(cleanup): - address = "{}:36432".format(hostname) +def test_async_calculator_example(unused_tcp_port, cleanup): + address = "{}:{}".format(hostname, unused_tcp_port) server = "async_calculator_server.py" client = "async_calculator_client.py" run_subprocesses(address, server, client) -@pytest.mark.xfail( - reason="Some versions of python don't like to share ports, don't worry if this fails" -) -def test_thread_example(cleanup): - address = "{}:36433".format(hostname) - server = "thread_server.py" - client = "thread_client.py" - run_subprocesses(address, server, client, wildcard_server=True) - - def test_addressbook_example(cleanup): proc = subprocess.Popen( [sys.executable, os.path.join(examples_dir, "addressbook.py")] @@ -142,57 +132,36 @@ def test_addressbook_example(cleanup): assert ret == 0 -@pytest.mark.skipif( - sys.platform == "win32", - reason=""" -Asyncio bug with libcapnp timer, likely due to asyncio starving some event loop. -See https://github.com/capnproto/pycapnp/issues/196 -""", -) -def test_async_example(cleanup): - address = "{}:36434".format(hostname) +def test_async_example(unused_tcp_port, cleanup): + address = "{}:{}".format(hostname, unused_tcp_port) server = "async_server.py" client = "async_client.py" run_subprocesses(address, server, client) -@pytest.mark.skipif( - sys.platform == "win32", - reason=""" -Asyncio bug with libcapnp timer, likely due to asyncio starving some event loop. -See https://github.com/capnproto/pycapnp/issues/196 -""", -) -def test_ssl_async_example(cleanup): - address = "{}:36435".format(hostname) +def test_ssl_async_example(unused_tcp_port, cleanup): + address = "{}:{}".format(hostname, unused_tcp_port) server = "async_ssl_server.py" client = "async_ssl_client.py" run_subprocesses(address, server, client, ipv4_force=False) -@pytest.mark.skipif( - sys.platform == "win32", - reason=""" -Asyncio bug with libcapnp timer, likely due to asyncio starving some event loop. -See https://github.com/capnproto/pycapnp/issues/196 -""", -) -def test_ssl_reconnecting_async_example(cleanup): - address = "{}:36436".format(hostname) +def test_ssl_reconnecting_async_example(unused_tcp_port, cleanup): + address = "{}:{}".format(hostname, unused_tcp_port) server = "async_ssl_server.py" client = "async_reconnecting_ssl_client.py" run_subprocesses(address, server, client, ipv4_force=False) -def test_async_ssl_calculator_example(cleanup): - address = "{}:36437".format(hostname) +def test_async_ssl_calculator_example(unused_tcp_port, cleanup): + address = "{}:{}".format(hostname, unused_tcp_port) server = "async_ssl_calculator_server.py" client = "async_ssl_calculator_client.py" run_subprocesses(address, server, client, ipv4_force=False) -def test_async_socket_message_example(cleanup): - address = "{}:36438".format(hostname) +def test_async_socket_message_example(unused_tcp_port, cleanup): + address = "{}:{}".format(hostname, unused_tcp_port) server = "async_socket_message_server.py" client = "async_socket_message_client.py" run_subprocesses(address, server, client) diff --git a/test/test_response.py b/test/test_response.py index 0d49e2a..d1f1c49 100644 --- a/test/test_response.py +++ b/test/test_response.py @@ -5,7 +5,7 @@ class FooServer(test_response_capnp.Foo.Server): def __init__(self, val=1): self.val = val - def foo(self, **kwargs): + async def foo(self, **kwargs): return 1 @@ -13,27 +13,27 @@ class BazServer(test_response_capnp.Baz.Server): def __init__(self, val=1): self.val = val - def grault(self, **kwargs): + async def grault(self, **kwargs): return {"foo": FooServer()} -def test_response_reference(): +async def test_response_reference(): baz = test_response_capnp.Baz._new_client(BazServer()) - bar = baz.grault().wait().bar + bar = (await baz.grault()).bar foo = bar.foo # This used to cause an exception about invalid pointers because the response got garbage collected - assert foo.foo().wait().val == 1 + assert (await foo.foo()).val == 1 -def test_response_reference2(): +async def test_response_reference2(): baz = test_response_capnp.Baz._new_client(BazServer()) - bar = baz.grault().wait().bar + bar = (await baz.grault()).bar # This always worked since it saved the intermediate response object - response = baz.grault().wait() + response = await baz.grault() bar = response.bar foo = bar.foo - assert foo.foo().wait().val == 1 + assert (await foo.foo()).val == 1 diff --git a/test/test_rpc.py b/test/test_rpc.py index fa5daa7..c6ae055 100644 --- a/test/test_rpc.py +++ b/test/test_rpc.py @@ -13,12 +13,14 @@ class Server(test_capability_capnp.TestInterface.Server): def __init__(self, val=100): self.val = val - def foo(self, i, j, **kwargs): + async def foo(self, i, j, **kwargs): return str(i * 5 + self.val) -def test_simple_rpc_with_options(): +async def test_simple_rpc_with_options(): read, write = socket.socketpair() + read = await capnp.AsyncIoStream.create_connection(sock=read) + write = await capnp.AsyncIoStream.create_connection(sock=write) _ = capnp.TwoPartyServer(write, bootstrap=Server()) # This traversal limit is too low to receive the response in, so we expect @@ -32,8 +34,10 @@ def test_simple_rpc_with_options(): _ = remote.wait() -def test_simple_rpc_bootstrap(): +async def test_simple_rpc_bootstrap(): read, write = socket.socketpair() + read = await capnp.AsyncIoStream.create_connection(sock=read) + write = await capnp.AsyncIoStream.create_connection(sock=write) _ = capnp.TwoPartyServer(write, bootstrap=Server(100)) client = capnp.TwoPartyClient(read) @@ -42,6 +46,6 @@ def test_simple_rpc_bootstrap(): cap = cap.cast_as(test_capability_capnp.TestInterface) remote = cap.foo(i=5) - response = remote.wait() + response = await remote assert response.x == "125" diff --git a/test/test_rpc_calculator.py b/test/test_rpc_calculator.py index ca5fd16..19317c9 100644 --- a/test/test_rpc_calculator.py +++ b/test/test_rpc_calculator.py @@ -1,6 +1,5 @@ import gc import os -import pytest import socket import sys # add examples dir to sys.path @@ -9,57 +8,20 @@ import capnp examples_dir = os.path.join(os.path.dirname(__file__), "..", "examples") sys.path.append(examples_dir) -import calculator_client # noqa: E402 -import calculator_server # noqa: E402 - -# Uses run_subprocesses function -import test_examples # noqa: E402 - -processes = [] +import async_calculator_client # noqa: E402 +import async_calculator_server # noqa: E402 -@pytest.fixture -def cleanup(): - yield - for p in processes: - p.kill() - - -def test_calculator(): +async def test_calculator(): read, write = socket.socketpair() + read = await capnp.AsyncIoStream.create_connection(sock=read) + write = await capnp.AsyncIoStream.create_connection(sock=write) - _ = capnp.TwoPartyServer(write, bootstrap=calculator_server.CalculatorImpl()) - calculator_client.main(read) + _ = capnp.TwoPartyServer(write, bootstrap=async_calculator_server.CalculatorImpl()) + await async_calculator_client.main(read) -@pytest.mark.xfail( - reason="Some versions of python don't like to share ports, don't worry if this fails" -) -def test_calculator_tcp(cleanup): - address = "localhost:36431" - test_examples.run_subprocesses( - address, "calculator_server.py", "calculator_client.py", wildcard_server=True - ) - - -@pytest.mark.xfail( - reason="Some versions of python don't like to share ports, don't worry if this fails" -) -@pytest.mark.skipif(os.name == "nt", reason="socket.AF_UNIX not supported on Windows") -def test_calculator_unix(cleanup): - path = "/tmp/pycapnp-test" - try: - os.unlink(path) - except OSError: - pass - - address = "unix:" + path - test_examples.run_subprocesses( - address, "calculator_server.py", "calculator_client.py" - ) - - -def test_calculator_gc(): +async def test_calculator_gc(): def new_evaluate_impl(old_evaluate_impl): def call(*args, **kwargs): gc.collect() @@ -68,12 +30,14 @@ def test_calculator_gc(): return call read, write = socket.socketpair() + read = await capnp.AsyncIoStream.create_connection(sock=read) + write = await capnp.AsyncIoStream.create_connection(sock=write) # inject a gc.collect to the beginning of every evaluate_impl call - evaluate_impl_orig = calculator_server.evaluate_impl - calculator_server.evaluate_impl = new_evaluate_impl(evaluate_impl_orig) + evaluate_impl_orig = async_calculator_server.evaluate_impl + async_calculator_server.evaluate_impl = new_evaluate_impl(evaluate_impl_orig) - _ = capnp.TwoPartyServer(write, bootstrap=calculator_server.CalculatorImpl()) - calculator_client.main(read) + _ = capnp.TwoPartyServer(write, bootstrap=async_calculator_server.CalculatorImpl()) + await async_calculator_client.main(read) - calculator_server.evaluate_impl = evaluate_impl_orig + async_calculator_server.evaluate_impl = evaluate_impl_orig diff --git a/test/test_serialization.py b/test/test_serialization.py index bc3a428..bcbb910 100644 --- a/test/test_serialization.py +++ b/test/test_serialization.py @@ -81,9 +81,6 @@ def test_roundtrip_bytes_mmap(all_types): test_regression.check_all_types(msg) -@pytest.mark.skipif( - sys.version_info[0] < 3, reason="memoryview is a builtin on Python 3" -) def test_roundtrip_bytes_buffer(all_types): msg = all_types.TestAllTypes.new_message() test_regression.init_all_types(msg) diff --git a/test/test_threads.py b/test/test_threads.py deleted file mode 100644 index 4756294..0000000 --- a/test/test_threads.py +++ /dev/null @@ -1,55 +0,0 @@ -""" -thread test -""" - -import platform -import socket -import threading - -import pytest - -import capnp - -import test_capability_capnp - - -class Server(test_capability_capnp.TestInterface.Server): - """ - Server - """ - - def __init__(self, val=100): - self.val = val - - def foo(self, i, j, **kwargs): - """ - foo - """ - return str(i * 5 + self.val) - - -@pytest.mark.skipif( - platform.python_implementation() == "PyPy", - reason="pycapnp's GIL handling isn't working properly at the moment for PyPy", -) -def test_using_threads(): - """ - Thread test - """ - read, write = socket.socketpair() - - def run_server(): - _ = capnp.TwoPartyServer(write, bootstrap=Server()) - capnp.wait_forever() - - server_thread = threading.Thread(target=run_server) - server_thread.daemon = True - server_thread.start() - - client = capnp.TwoPartyClient(read) - cap = client.bootstrap().cast_as(test_capability_capnp.TestInterface) - - remote = cap.foo(i=5) - response = remote.wait() - - assert response.x == "125"