IQ.Lvbs/pycapnp
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request #315 from LasseBlaauwbroek/cull-sync

Browse Source 
Remove the synchronous RPC mode
This commit is contained in:
Jacob Alexander
2023-06-12 00:31:16 -07:00
committed by GitHub
parent 8feb377217 1e94f2e321
commit b439993b1f
38 changed files with 506 additions and 2250 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
.github/workflows/wheels.yml vendored
View File

@@ -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' || '' }}"

105
benchmark/bin/run_all.py
View File

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

61
benchmark/bin/runner.py
View File

@@ -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__':
if __name__ == "__main__":
main()

1
capnp/__init__.py
View File

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

13
capnp/helpers/asyncHelper.h
View File

@@ -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<capnp::RemotePromise<::capnp::DynamicStruct>> promise,
kj::WaitScope & scope) {
return new capnp::Response< ::capnp::DynamicStruct>(promise->wait(scope));
}

63
capnp/helpers/capabilityHelper.cpp
View File

@@ -1,8 +1,8 @@
#include "capnp/helpers/capabilityHelper.h"
#include "capnp/lib/capnp_api.h"
::kj::Promise<kj::Own<PyRefCounter>> convert_to_pypromise(kj::Own<capnp::RemotePromise<capnp::DynamicStruct>> promise) {
return promise->then([](capnp::Response<capnp::DynamicStruct>&& response) {
::kj::Promise<kj::Own<PyRefCounter>> convert_to_pypromise(capnp::RemotePromise<capnp::DynamicStruct> promise) {
return promise.then([](capnp::Response<capnp::DynamicStruct>&& response) {
return stealPyRef(wrap_dynamic_struct_reader(response)); } );
}
@@ -58,75 +58,26 @@ void check_py_error() {
}
}
inline kj::Promise<kj::Own<PyRefCounter>> maybeUnwrapPromise(PyObject * result) {
check_py_error();
auto promise = extract_promise(result);
Py_DECREF(result);
return kj::mv(*promise);
}
kj::Promise<kj::Own<PyRefCounter>> wrapPyFunc(kj::Own<PyRefCounter> func, kj::Own<PyRefCounter> 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<kj::Own<PyRefCounter>> wrapPyFuncNoArg(kj::Own<PyRefCounter> 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<kj::Own<PyRefCounter>> wrapRemoteCall(kj::Own<PyRefCounter> func, capnp::Response<capnp::DynamicStruct> & 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<kj::Own<PyRefCounter>> then(kj::Own<kj::Promise<kj::Own<PyRefCounter>>> promise,
::kj::Promise<kj::Own<PyRefCounter>> then(kj::Promise<kj::Own<PyRefCounter>> promise,
kj::Own<PyRefCounter> func, kj::Own<PyRefCounter> error_func) {
if(error_func->obj == Py_None)
return promise->then(kj::mvCapture(func, [](auto func, kj::Own<PyRefCounter> arg) {
return promise.then(kj::mvCapture(func, [](auto func, kj::Own<PyRefCounter> arg) {
return wrapPyFunc(kj::mv(func), kj::mv(arg)); } ));
else
return promise->then
return promise.then
(kj::mvCapture(func, [](auto func, kj::Own<PyRefCounter> 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<kj::Own<PyRefCounter>> then(kj::Own<::capnp::RemotePromise<::capnp::DynamicStruct>> promise,
kj::Own<PyRefCounter> func, kj::Own<PyRefCounter> error_func) {
if(error_func->obj == Py_None)
return promise->then(kj::mvCapture(func, [](auto func, capnp::Response<capnp::DynamicStruct>&& arg) {
return wrapRemoteCall(kj::mv(func), arg); } ));
else
return promise->then
(kj::mvCapture(func, [](auto func, capnp::Response<capnp::DynamicStruct>&& 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<kj::Own<PyRefCounter>> then(kj::Own<kj::Promise<void>> promise,
kj::Own<PyRefCounter> func, kj::Own<PyRefCounter> 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<kj::Own<PyRefCounter>> then(kj::Promise<kj::Array<kj::Own<PyRefCounter>> > && promise) {
return promise.then([](kj::Array<kj::Own<PyRefCounter>>&& arg) {
return stealPyRef(convert_array_pyobject(arg)); } );
}
kj::Promise<void> PythonInterfaceDynamicImpl::call(capnp::InterfaceSchema::Method method,
capnp::CallContext< capnp::DynamicStruct, capnp::DynamicStruct> context) {
auto methodName = method.getProto().getName();

35
capnp/helpers/capabilityHelper.h
View File

@@ -54,37 +54,21 @@ inline kj::Own<PyRefCounter> stealPyRef(PyObject* o) {
return ret;
}
::kj::Promise<kj::Own<PyRefCounter>> convert_to_pypromise(kj::Own<capnp::RemotePromise<capnp::DynamicStruct>> promise);
::kj::Promise<kj::Own<PyRefCounter>> convert_to_pypromise(capnp::RemotePromise<capnp::DynamicStruct> promise);
inline ::kj::Promise<kj::Own<PyRefCounter>> convert_to_pypromise(kj::Own<kj::Promise<void>> promise) {
return promise->then([]() {
inline ::kj::Promise<kj::Own<PyRefCounter>> convert_to_pypromise(kj::Promise<void> promise) {
return promise.then([]() {
GILAcquire gil;
return kj::heap<PyRefCounter>(Py_None);
});
}
template<class T>
::kj::Promise<void> convert_to_voidpromise(kj::Own<kj::Promise<T>> promise) {
return promise->then([](T) { } );
}
void reraise_kj_exception();
void check_py_error();
inline kj::Promise<kj::Own<PyRefCounter>> wrapSizePromise(kj::Promise<size_t> promise) {
return promise.then([](size_t response) { return stealPyRef(PyLong_FromSize_t(response)); } );
}
::kj::Promise<kj::Own<PyRefCounter>> then(kj::Own<kj::Promise<kj::Own<PyRefCounter>>> promise,
::kj::Promise<kj::Own<PyRefCounter>> then(kj::Promise<kj::Own<PyRefCounter>> promise,
kj::Own<PyRefCounter> func, kj::Own<PyRefCounter> error_func);
::kj::Promise<kj::Own<PyRefCounter>> then(kj::Own<::capnp::RemotePromise< ::capnp::DynamicStruct>> promise,
kj::Own<PyRefCounter> func, kj::Own<PyRefCounter> error_func);
::kj::Promise<kj::Own<PyRefCounter>> then(kj::Own<kj::Promise<void>> promise,
kj::Own<PyRefCounter>func, kj::Own<PyRefCounter> error_func);
::kj::Promise<kj::Own<PyRefCounter>> then(kj::Promise<kj::Array<kj::Own<PyRefCounter>> > && 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<PythonInterfaceDynamicImpl>(schema, server));
}
inline capnp::DynamicValue::Reader new_server(capnp::InterfaceSchema & schema, PyObject * server) {
return capnp::DynamicValue::Reader(kj::heap<PythonInterfaceDynamicImpl>(schema, server));
}
inline capnp::Capability::Client server_to_client(capnp::InterfaceSchema & schema, PyObject * server) {
return kj::heap<PythonInterfaceDynamicImpl>(schema, server);
}
class PyAsyncIoStream: public kj::AsyncIoStream {
public:
kj::Own<PyRefCounter> protocol;

28
capnp/helpers/helpers.pxd
View File

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

8
capnp/helpers/non_circular.pxd
View File

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

49
capnp/helpers/rpcHelper.h
View File

@@ -19,52 +19,3 @@ capnp::Capability::Client bootstrapHelperServer(capnp::RpcSystem<capnp::rpc::two
hostId.setSide(capnp::rpc::twoparty::Side::CLIENT);
return client.bootstrap(hostId);
}
class ErrorHandler : public kj::TaskSet::ErrorHandler {
void taskFailed(kj::Exception&& exception) override {
kj::throwFatalException(kj::mv(exception));
}
};
struct ServerContext {
kj::Own<kj::AsyncIoStream> stream;
capnp::TwoPartyVatNetwork network;
capnp::RpcSystem<capnp::rpc::twoparty::SturdyRefHostId> rpcSystem;
ServerContext(kj::Own<kj::AsyncIoStream>&& 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<kj::ConnectionReceiver>&& listener, capnp::ReaderOptions & opts) {
auto ptr = listener.get();
tasks.add(ptr->accept().then(kj::mvCapture(kj::mv(listener),
[&, client, opts](kj::Own<kj::ConnectionReceiver>&& listener,
kj::Own<kj::AsyncIoStream>&& connection) mutable {
acceptLoop(tasks, client, kj::mv(listener), opts);
auto server = kj::heap<ServerContext>(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<kj::Own<PyRefCounter>> connectServer(kj::TaskSet & tasks, capnp::Capability::Client client, kj::AsyncIoProvider * provider, kj::StringPtr bindAddress, capnp::ReaderOptions & opts) {
auto paf = kj::newPromiseAndFulfiller<unsigned int>();
auto portPromise = paf.promise.fork();
tasks.add(provider->getNetwork().parseAddress(bindAddress)
.then(kj::mvCapture(paf.fulfiller,
[&, client, opts](kj::Own<kj::PromiseFulfiller<unsigned int>>&& portFulfiller,
kj::Own<kj::NetworkAddress>&& 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)); });
}

80
capnp/includes/capnp_cpp.pxd
View File

@@ -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<void> >"(
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<kj::Own<PyRefCounter>> >"(size_t) nogil
ctypedef Array[Own[PyRefCounter]] PyArray' ::kj::Array<kj::Own<PyRefCounter>>'
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<int64_t>"()
@@ -553,19 +490,16 @@ cdef extern from "kj/async.h" namespace " ::kj":
cdef cppclass VoidPromiseFulfiller"::kj::PromiseFulfiller<void>" nogil:
void fulfill()
void reject(Exception&& exception)
cdef cppclass PromiseFulfillerPair" ::kj::PromiseFulfillerPair<void>" nogil:
VoidPromise promise
Own[VoidPromiseFulfiller] fulfiller
PromiseFulfillerPair newPromiseAndFulfiller" ::kj::newPromiseAndFulfiller<void>"() 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)

8
capnp/lib/capnp.pxd
View File

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

701
capnp/lib/capnp.pyx
View File

File diff suppressed because it is too large Load Diff

6
docs/capnp.rst
View File

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

16
examples/async_calculator_client.py
View File

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

14
examples/async_calculator_server.py
View File

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

5
examples/async_client.py
View File

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

3
examples/async_reconnecting_ssl_client.py
View File

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

10
examples/async_server.py
View File

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

3
examples/async_socket_message_client.py
View File

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

3
examples/async_socket_message_server.py
View File

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

5
examples/async_ssl_calculator_client.py
View File

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

52
examples/async_ssl_calculator_server.py
View File

@@ -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():

20
examples/async_ssl_server.py
View File

@@ -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():

304
examples/calculator_client.py
View File

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

145
examples/calculator_server.py
View File

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

56
examples/thread_client.py
View File

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

44
examples/thread_server.py
View File

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

3
pyproject.toml
View File

@@ -1,2 +1,5 @@
[build-system]
requires = ["setuptools", "wheel", "pkgconfig", "cython"]
[tool.pytest.ini_options]
asyncio_mode = "auto"

219
test/test_capability.py
View File

@@ -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):
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)
return inCap.foo(i=n).then(_then)
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):
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
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)
)
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):
async def foo(self, cap, _context, **kwargs):
res = await cap.foo(5)
_context.results.x = res.x
return cap.foo(5).then(set_result)
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"

136
test/test_capability_context.py
View File

@@ -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):
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 = capability().TestInterface._new_server(
Server(100)
)
return context.params.inCap.foo(i=context.params.n).then(_then)
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: <class '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: <class '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

287
test/test_capability_old.py
View File

@@ -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: <class '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: <class '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

55
test/test_examples.py
View File

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

18
test/test_response.py
View File

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

12
test/test_rpc.py
View File

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

66
test/test_rpc_calculator.py
View File

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

3
test/test_serialization.py
View File

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

55
test/test_threads.py
View File

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