Async API simplifications -- no more fine-grained thread-safety

[Kenton Varda]

Hey all, particularly Jason,

I just simplified the async API quite a bit:

- Taking Jason's suggestion, an EventLoop now becomes the current loop for the thread immediately upon construction, so that you don't have to explicitly pass it around or use loop.there() instead of promise.then().  This makes main functions look a lot nicer -- call whatever you want, and use promise.wait() to wait for it.  It should also simplify usage from Python, where callbacks are awkward due to lack of multi-line lambdas.  It occurred to me at some point that Python scripts are far more likely to be clients than servers, and as such a Python Cap'n Proto program may in fact be fine with never using callbacks, but just always using .wait() to wait for each promise.  (Sorry for resisting this earlier, Jason -- I just had to think about it more to realize you were right.)

- Most of EventLoop's methods have been removed in favor of the now-equivalent Promise methods.

- It is no longer possible to wait on a Promise created in a different thread, or generally interact with the EventLoop of another thread.  This has allowed me to massively simplify a lot of code, getting rid of mutexes and atomic ops that were slowing things down even for single-threaded programs.  I will re-introduce a new way to communicate between threads in the future, but it will be more limited.

- On a similar note, I removed the thread-safety from reference counting and message building, because they also impose too much overhead for people who don't actually need them.  If you use an explicitly thread-safe Cap'n Proto adapter to bridge between threads (not yet written) then there's no need for these low-level details to be thread-safe, and you end up with much less synchronization overhead overall.

I still have some more tweaks to make, like making it possible to integrate with an existing event loop library not based on KJ.

-Kenton

[Naveen Michaud-Agrawal]

Hi Kenton,

Will this integrate with the new asyncio library in python 3.4?

Naveen

[Kenton Varda]

On Mon, Dec 2, 2013 at 9:37 AM, Naveen Michaud-Agrawal <naveen.mic...@gmail.com> wrote:

Hi Kenton,

Will this integrate with the new asyncio library in python 3.4?

Good question.

From a brief glance at PEP3156, the models look very similar, and I think it should be possible to have KJ-based code which is using the Python event loop.  Or, it will be, after some enhancements I intend to make to the KJ EventLoop API to make it easier to integrate with existing event loops.

Someone will actually have to do the work of integrating them, though.

-Kenton

[Jason Paryani]

 Awesome work Kenton. I'm glad I convinced you, since the python interface for clients does look a lot nicer now:

c = socket.create_connection(('localhost', 49999))

client = capnp.RpcClient(c)

cap = client.restore(test_capnp.TestSturdyRefObjectId.new_message(tag='testInterface'))
cap = cap.cast_as(test_capnp.TestInterface)

response = cap.foo(i=5).wait()

assert response.x == '125'
c.close()

I just have 2 questions for you:

1. I'm not really liking the names RpcServer/Client, since as you mentioned before, there will be other implementations of RpcSystem. Would TwoPartyVatNetwork{Server,Client} be better names for them?

2. I'm getting random errors during the Schema load in my tests. It's a transient bug when compiled with clang installed from Xcode 5, and is of the form:

E   KjException: /Users/jason/workspace/pycapnp/test/addressbook.capnp:0: bug in code: Internal compiler bug: Schema failed validation:

E   src/capnp/layout.c++:570: requirement not met: expected boundsCheck(segment, ptr, ptr + ref->structRef.wordSize()); Message contained out-of-bounds struct pointer.

E   stack: 0x10fe5740a 0x10fe5afaf 0x10fca21a6 0x10fd2cec1 0x10fd273f7 0x10fd37826 0x10fd36c00 0x10fd365c4 0x10fd384dd 0x10fe5828e 0x10fca5d4f 0x10fca60a5 0x10fca8dfa 0x10fca93fb

 0x10fcace20 0x10fbce857

E   stack: 0x10fe5758b 0x10fcaefb4 0x10fcaf1d7 0x10fca5f5c 0x10fca60a5 0x10fca8dfa 0x10fca93fb 0x10fcace20 0x10fbce857 0x10f2a7f6d 0x10fbcf3cd 0x10f2a7f6d 0x10fbc2e89 0x10f320656

 0x10f31e480 0x10f2c568b

Also a full output compiled with gcc-4.8 is available at https://travis-ci.org/jparyani/pycapnp/jobs/14841552

From a brief glance at PEP3156, the models look very similar, and I think it should be possible to have KJ-based code which is using the Python event loop.  Or, it will be, after some enhancements I intend to make to the KJ EventLoop API to make it easier to integrate with existing event loops.

Someone will actually have to do the work of integrating them, though.

-Kenton

I definitely want to allow integration of different EventLoops from the python wrappers, but for the initial release this won't be supported. Eventually, I'd like to support gevent and asyncio, and make it extensible enough that other Python event libraries could be wrapped from pure python.

 

[Kenton Varda]

On Mon, Dec 2, 2013 at 6:53 PM, Jason Paryani <jpar...@gmail.com> wrote:

I just have 2 questions for you:

1. I'm not really liking the names RpcServer/Client, since as you mentioned before, there will be other implementations of RpcSystem. Would TwoPartyVatNetwork{Server,Client} be better names for them?

Maybe just TwoPartyServer / TwoPartyClient?  Or TwoPartyRpcServer / TwoPartyRpcClient.

 

2. I'm getting random errors during the Schema load in my tests. It's a transient bug when compiled with clang installed from Xcode 5, and is of the form:

E   KjException: /Users/jason/workspace/pycapnp/test/addressbook.capnp:0: bug in code: Internal compiler bug: Schema failed validation:

E   src/capnp/layout.c++:570: requirement not met: expected boundsCheck(segment, ptr, ptr + ref->structRef.wordSize()); Message contained out-of-bounds struct pointer.

E   stack: 0x10fe5740a 0x10fe5afaf 0x10fca21a6 0x10fd2cec1 0x10fd273f7 0x10fd37826 0x10fd36c00 0x10fd365c4 0x10fd384dd 0x10fe5828e 0x10fca5d4f 0x10fca60a5 0x10fca8dfa 0x10fca93fb

 0x10fcace20 0x10fbce857

E   stack: 0x10fe5758b 0x10fcaefb4 0x10fcaf1d7 0x10fca5f5c 0x10fca60a5 0x10fca8dfa 0x10fca93fb 0x10fcace20 0x10fbce857 0x10f2a7f6d 0x10fbcf3cd 0x10f2a7f6d 0x10fbc2e89 0x10f320656

 0x10f31e480 0x10f2c568b

Weird.  Are you potentially calling SchemaParser from multiple threads?  It should still work but I moved the mutex way up the stack so I could easily have broken something.

[Jason Paryani]

I'm not sure if py.test uses multiple threads, I'll check later tonight. I'll also try to reproduce in a simple c++ example. 

—
Jason

[Jason Paryani]

I've attached some stack traces from the problem. I temporarily turned off exception handling on the python end, and ran a script that just calls capnp.load("test.capnp") a hundred times. This still doesn't reproduce the problem reliably, so I run it over and over again in a bash loop until one crashes and then inspect the core dump. I'm still trying to reproduce it in a simpler form, but it's definitely occurring in SchemaParser::parseDiskFile. Below are two different ways I've seen it crash.

#0  0x00007fff96623866 in __pthread_kill ()

#1  0x00007fff95b8f35c in pthread_kill ()

#2  0x00007fff9aa1fbba in abort ()

#3  0x00007fff9a901141 in abort_message ()

#4  0x00007fff9a926aa4 in default_terminate_handler ()

#5  0x00007fff9ae04322 in _objc_terminate ()

#6  0x00007fff9a9243e1 in std::__terminate ()

#7  0x00007fff9a923e6b in __cxa_throw ()

#8  0x000000010977f684 in kj::ExceptionCallback::RootExceptionCallback::onRecoverableException (this=0x1097b2828, exception=@0x7fff5744b690) at src/kj/exception.c++:298

#9  0x000000010977a409 in kj::throwRecoverableException (exception=@0x7fff5744b690) at src/kj/exception.c++:349

#10 0x0000000109781fc7 in kj::_::Debug::Fault::~Fault (this=0x7fff5744b948) at src/kj/debug.c++:201

#11 0x0000000109781f35 in kj::_::Debug::Fault::~Fault (this=0x7fff5744b948) at src/kj/debug.c++:197

#12 0x000000010924fe9d in capnp::_::WireHelpers::readListPointer (segment=0x7fcd60482ed0, ref=0x7fcd608c3428, refTarget=0x7fcd71a36360, defaultValue=0x0, expectedElementSize=capnp::_::INLINE_COMPOSITE, nestingLimit=2147483646) at src/capnp/layout.c++:1903

#13 0x000000010924bfe7 in capnp::_::WireHelpers::readListPointer (segment=0x7fcd60482ed0, ref=0x7fcd608c3428, defaultValue=0x0, expectedElementSize=capnp::_::INLINE_COMPOSITE, nestingLimit=2147483646) at src/capnp/layout.c++:1873

#14 0x0000000109246dd9 in capnp::_::PointerReader::getList (this=0x7fff5744bb30, expectedElementSize=capnp::_::INLINE_COMPOSITE, defaultValue=0x0) at src/capnp/layout.c++:2239

#15 0x0000000108c87683 in capnp::List<capnp::schema::Annotation, (capnp::Kind)3>::getFromPointer (reader=@0x7fff5744bb30, defaultValue=0x0) at list.h:297

#16 0x0000000108c875b6 in capnp::_::PointerHelpers<capnp::List<capnp::schema::Annotation, (capnp::Kind)3>, (capnp::Kind)6>::get (reader={segment = 0x7fcd60482ed0, pointer = 0x7fcd608c3428, nestingLimit = 2147483646}, defaultValue=0x0) at pointer-helpers.h:71

#17 0x0000000108c72b2c in capnp::schema::Node::Reader::getAnnotations (this=0x7fff5744c058) at schema.capnp.h:2888

#18 0x0000000108c6c156 in capnp::compiler::Compiler::Node::traverseNodeDependencies (this=0x7fcd608c6fe8, schemaNode=@0x7fff5744c058, eagerness=1073774593, seen=@0x7fff5744c278, finalLoader=@0x7fcd604b1980) at src/capnp/compiler/compiler.c++:775

#19 0x0000000108c6b9aa in capnp::compiler::Compiler::Node::traverse (this=0x7fcd608c6fe8, eagerness=1073774597, seen=@0x7fff5744c278, finalLoader=@0x7fcd604b1980) at src/capnp/compiler/compiler.c++:707

#20 0x0000000108c6bbdb in capnp::compiler::Compiler::Node::traverse (this=0x7fcd604ccf38, eagerness=1073774597, seen=@0x7fff5744c278, finalLoader=@0x7fcd604b1980) at src/capnp/compiler/compiler.c++:724

#21 0x0000000108c6eeba in capnp::compiler::Compiler::Impl::eagerlyCompile (this=0x7fcd60482e80, id=15350781828677780152, eagerness=1073774597, finalLoader=@0x7fcd604b1980) at src/capnp/compiler/compiler.c++:1070

#22 0x0000000108c6f4ee in capnp::compiler::Compiler::eagerlyCompile (this=0x7fcd604b18a0, id=15350781828677780152, eagerness=1073774597) at src/capnp/compiler/compiler.c++:1113

#23 0x0000000108c88b85 in capnp::SchemaParser::parseFile (this=0x7fcd604a8890, file=@0x7fff5744c6e8) at src/capnp/schema-parser.c++:182

#24 0x0000000108c88a2c in capnp::SchemaParser::parseDiskFile (this=0x7fcd604a8890, displayName={content = {<kj::DisallowConstCopyIfNotConst<const char>> = {<No data fields>}, ptr = 0x108a9cc84 "test.capnp", size_ = 11}}, diskPath={content = {<kj::DisallowConstCopyIfNotConst<const char>> = {<No data fields>}, ptr = 0x108a9cc84 "test.capnp", size_ = 11}}, importPath={<kj::DisallowConstCopyIfNotConst<const kj::StringPtr>> = {<No data fields>}, ptr = 0x7fcd604d1940, size_ = 0}) at src/capnp/schema-parser.c++:176

#25 0x0000000108af591f in __pyx_pw_5capnp_5capnp_12SchemaParser_5_parse_disk_file (__pyx_v_self=0x108992120, __pyx_args=<value temporarily unavailable, due to optimizations>, __pyx_kwds=0x0) at capnp/capnp.cpp:32795

#26 0x00000001087c3f6d in PyObject_Call ()

#27 0x0000000108af643d in __pyx_pw_5capnp_5capnp_12SchemaParser_7load (__pyx_v_self=0x108992120, __pyx_args=<value temporarily unavailable, due to optimizations>, __pyx_kwds=<value temporarily unavailable, due to optimizations>) at capnp/capnp.cpp:33818

#28 0x00000001087c3f6d in PyObject_Call ()

#29 0x0000000108ae9f79 in __pyx_pf_5capnp_5capnp_10load () at /Users/jason/workspace/pycapnp/capnp/capnp.cpp:38704

#30 0x0000000108ae9f79 in __pyx_pw_5capnp_5capnp_11load (__pyx_self=<value temporarily unavailable, due to optimizations>, __pyx_args=<value temporarily unavailable, due to optimizations>, __pyx_kwds=<value temporarily unavailable, due to optimizations>) at capnp/capnp.cpp:38633

#31 0x000000010883c656 in PyEval_EvalFrameEx ()

#32 0x000000010883a480 in PyEval_EvalCodeEx ()

#33 0x0000000108839e14 in PyEval_EvalCode ()

#34 0x00000001088593c4 in run_mod ()

#35 0x000000010885946b in PyRun_FileExFlags ()

#36 0x0000000108858fb9 in PyRun_SimpleFileExFlags ()

#37 0x0000000108869a67 in Py_Main ()

#38 0x00007fff97f915fd in start ()

#0  0x00007fff96623866 in __pthread_kill ()

#1  0x00007fff95b8f35c in pthread_kill ()

#2  0x00007fff9aa1fbba in abort ()

#3  0x00007fff9a901141 in abort_message ()

#4  0x00007fff9a926aa4 in default_terminate_handler ()

#5  0x00007fff9ae04322 in _objc_terminate ()

#6  0x00007fff9a9243e1 in std::__terminate ()

#7  0x00007fff9a923e6b in __cxa_throw ()

#8  0x0000000108682684 in kj::ExceptionCallback::RootExceptionCallback::onRecoverableException (this=0x1086b5828, exception=@0x7fff58548a30) at src/kj/exception.c++:298

#9  0x0000000107b91d29 in capnp::SchemaFile::DiskSchemaFile::reportError (this=0x7fc6e9d1bcb0, start={byte = 6808, line = 181, column = 7}, end={byte = 6817, line = 181, column = 16}, message={content = {<kj::DisallowConstCopyIfNotConst<const char>> = {<No data fields>}, ptr = 0x7fc6e9e96970 "Internal compiler bug: Schema failed validation:\nsrc/capnp/layout.c++:570: requirement not met: expected boundsCheck(segment, ptr, ptr + ref->structRef.wordSize()); Message contained out-of-bounds str"..., size_ = 412}}) at src/capnp/schema-parser.c++:445

#10 0x0000000107b9294b in capnp::SchemaParser::ModuleImpl::addError (this=0x7fc6e9d1c270, startByte=6808, endByte=6817, message={content = {<kj::DisallowConstCopyIfNotConst<const char>> = {<No data fields>}, ptr = 0x7fc6e9e96970 "Internal compiler bug: Schema failed validation:\nsrc/capnp/layout.c++:570: requirement not met: expected boundsCheck(segment, ptr, ptr + ref->structRef.wordSize()); Message contained out-of-bounds str"..., size_ = 412}}) at src/capnp/schema-parser.c++:123

#11 0x0000000107b6d746 in capnp::compiler::Compiler::Node::addError (this=0x7fc6ea801830, error={content = {<kj::DisallowConstCopyIfNotConst<const char>> = {<No data fields>}, ptr = 0x7fc6e9e96970 "Internal compiler bug: Schema failed validation:\nsrc/capnp/layout.c++:570: requirement not met: expected boundsCheck(segment, ptr, ptr + ref->structRef.wordSize()); Message contained out-of-bounds str"..., size_ = 412}}) at src/capnp/compiler/compiler.c++:826

#12 0x0000000107b6e6f1 in capnp::compiler::Compiler::Node::loadFinalSchema (this=0x7fc6ea801830, loader=@0x7fc6e9cb1980) at src/capnp/compiler/compiler.c++:683

#13 0x0000000107b6e8c4 in capnp::compiler::Compiler::Node::traverse (this=0x7fc6ea801830, eagerness=1073774597, seen=@0x7fff58549278, finalLoader=@0x7fc6e9cb1980) at src/capnp/compiler/compiler.c++:699

#14 0x0000000107b6ebdb in capnp::compiler::Compiler::Node::traverse (this=0x7fc6e9d011e8, eagerness=1073774597, seen=@0x7fff58549278, finalLoader=@0x7fc6e9cb1980) at src/capnp/compiler/compiler.c++:724

#15 0x0000000107b71eba in capnp::compiler::Compiler::Impl::eagerlyCompile (this=0x7fc6e9c82e80, id=15350781828677780152, eagerness=1073774597, finalLoader=@0x7fc6e9cb1980) at src/capnp/compiler/compiler.c++:1070

#16 0x0000000107b724ee in capnp::compiler::Compiler::eagerlyCompile (this=0x7fc6e9cb18a0, id=15350781828677780152, eagerness=1073774597) at src/capnp/compiler/compiler.c++:1113

#17 0x0000000107b8bb85 in capnp::SchemaParser::parseFile (this=0x7fc6e9c3aa50, file=@0x7fff585496e8) at src/capnp/schema-parser.c++:182

#18 0x0000000107b8ba2c in capnp::SchemaParser::parseDiskFile (this=0x7fc6e9c3aa50, displayName={content = {<kj::DisallowConstCopyIfNotConst<const char>> = {<No data fields>}, ptr = 0x10799fc84 "test.capnp", size_ = 11}}, diskPath={content = {<kj::DisallowConstCopyIfNotConst<const char>> = {<No data fields>}, ptr = 0x10799fc84 "test.capnp", size_ = 11}}, importPath={<kj::DisallowConstCopyIfNotConst<const kj::StringPtr>> = {<No data fields>}, ptr = 0x7fc6e9e7b6e0, size_ = 0}) at src/capnp/schema-parser.c++:176

#19 0x00000001079f891f in __pyx_pw_5capnp_5capnp_12SchemaParser_5_parse_disk_file (__pyx_v_self=0x107895120, __pyx_args=<value temporarily unavailable, due to optimizations>, __pyx_kwds=0x0) at capnp/capnp.cpp:32795

#20 0x00000001076c2f6d in PyObject_Call ()

#21 0x00000001079f943d in __pyx_pw_5capnp_5capnp_12SchemaParser_7load (__pyx_v_self=0x107895120, __pyx_args=<value temporarily unavailable, due to optimizations>, __pyx_kwds=<value temporarily unavailable, due to optimizations>) at capnp/capnp.cpp:33818

#22 0x00000001076c2f6d in PyObject_Call ()

#23 0x00000001079ecf79 in __pyx_pf_5capnp_5capnp_10load () at /Users/jason/workspace/pycapnp/capnp/capnp.cpp:38704

#24 0x00000001079ecf79 in __pyx_pw_5capnp_5capnp_11load (__pyx_self=<value temporarily unavailable, due to optimizations>, __pyx_args=<value temporarily unavailable, due to optimizations>, __pyx_kwds=<value temporarily unavailable, due to optimizations>) at capnp/capnp.cpp:38633

#25 0x000000010773b656 in PyEval_EvalFrameEx ()

#26 0x0000000107739480 in PyEval_EvalCodeEx ()

#27 0x0000000107738e14 in PyEval_EvalCode ()

#28 0x00000001077583c4 in run_mod ()

#29 0x000000010775846b in PyRun_FileExFlags ()

#30 0x0000000107757fb9 in PyRun_SimpleFileExFlags ()

#31 0x0000000107768a67 in Py_Main ()

#32 0x00007fff97f915fd in start ()

[Kenton Varda]

Have you tried running under valgrind?  I bet the problem involves reading already-freed memory, in which case valgrind should reliably catch it.

[Ed Hemphill]

[Waiting across Promises from a different thread ...]   I was wondering if there is any update on this in 4.1 or 5.0.

We are currently developing a multi-threaded database server, and are using KJ and capnp and love the Promise implementation, but we need Promises to work across threads.

If not... could I get a couple of suggestions on the best classes to extend in order to make this happen?

-Ed

[Kenton Varda]

Hi Ed,

I'd love to hear more about what you're working on!

Making it possible to communicate cross-thread will require extending the EventLoop implementation to include a second queue which is populated with cross-thread events in a thread-safe way.  Whenever the thread running the loop runs out of events in its main queue (or maybe just after every N events processed), it should check the inter-thread queue and transfer any events there over to the main queue.

Then we need a higher-level interface for using this queue.  I haven't totally figured out what it might look like yet.  Perhaps some sort of thread-safe promise whose destructor merely sends a message to the other thread notifying it to destroy its end, or maybe we need some other abstraction.

Obviously, this is all very low-level work in KJ itself, not something you can implement on top.  If you want a more stopgap solution, you can open some pipes/socketpairs and send messages through them.  You could even use Cap'n Proto RPC on top of those sock.  :)  In fact, this may end up being my recommended way to communicate between threads -- as RPCs.  Though, obviously for efficiency we'll want some sort of memory-passing RPC transport, rather than the socket-based one.

-Kenton

--

You received this message because you are subscribed to the Google Groups "Cap'n Proto" group.
To unsubscribe from this group and stop receiving emails from it, send an email to capnproto+...@googlegroups.com.
Visit this group at http://groups.google.com/group/capnproto.

[Ed Hemphill]

Kenton, thanks for the reply...

So far I am using a thread pool template I made using libkj promises along with libuv function calls. The target for the database is mostly v8/node.js users, to include our own deviceJS (http://devicejs.org/)

The main issue I am having is how to use the EzRpc system with a bunch of worker threads. The primary issue is that the threadpool uses it's EventPort, and the EzRpc server has it's own. ..and it looks like we can't have two EventLoops running in the same thread. So the issue is how do we get these fulfilled Promises done by the work pool to be the returned promises by the EzRPC calls we implement.

---

I made a little git project which shows the threadpool code in case you're curious... any tips / comments from the group appreciated. Here's the test file, calculating Fibonacci numbers: https://github.com/WigWagCo/threadpool-kj/blob/master/testPool.cxx

-Ed

[Kenton Varda]

Hi Ed,

EzRpc is designed for people with really simple situations.  It sets up the whole event framework in one particular way, just so that you don't have to do the work.  If you need to do anything custom, it's likely that you'll need to skip the EzRpc classes and go directly to dealing with RpcSystem, TwoPartyVatNetwork, etc.  See the comments in ez-rpc.h for more on this.

If you need to integrate with libuv, you definitely won't be able to use EzRpc.  But I have already written some code to do just this.  Check out my node.js Cap'n Proto wrapper:

https://github.com/kentonv/sandstorm/blob/master/src/sandstorm/v8capnp.c++

Digging a bit into your code, I see that you actually have your own PoolEventPort you've implemented which waits on a queue of notifications coming in from your thread pool.  This may be hard to use in a thread that also supports RPC because you need some way to wake up whenever there is activity on the RPC socket, in addition to waking up when a notification arrives from a worker thread.

What you may end up having to do instead is have each work thread, when it completes its task, write a byte to a pipe, and then you can wait for events on that pipe in the main thread by wrapping it in an AsyncIoStream.  Each time you read a byte from the pipe, you know that there is at least one notification on the worker thread queue, so you can receive it without blocking.

Alternatively, if you're integrating with libuv anyway, maybe it already has a more convenient primitive for cross-thread messaging.

-Kenton

[Ed Hemphill]

Kenton:

Thanks a bunch for the feedback... I will definitely be digging into sandstorm and the Cap'n Proto node.js wrapper as time allows.

In the mean time I am following your advice - and dropping EzRpc for now. Definitely looking forward to 5.0 and what may lie ahead with shared memory IPC.

-Ed

[Alex Elsayed]

For that in-memory transport, liburcu{,-cds} might be a useful tool -
especially wfcqueue or rculfqueue:

http://git.lttng.org/?p=userspace-rcu.git;a=blob;f=doc/cds-api.txt