From ebbd83a425fe8f55137ddd1840044f2f0355ec08 Mon Sep 17 00:00:00 2001 From: Jason Paryani Date: Wed, 11 Dec 2013 13:50:50 -0800 Subject: [PATCH] Fix as_pypromise not returning correctly --- capnp/lib/capnp.pyx | 53 ++++++++++++++++++++++++++++++++++++--------- test/test_rpc.py | 2 +- 2 files changed, 44 insertions(+), 11 deletions(-) diff --git a/capnp/lib/capnp.pyx b/capnp/lib/capnp.pyx index aca7b7b..a1687c7 100644 --- a/capnp/lib/capnp.pyx +++ b/capnp/lib/capnp.pyx @@ -22,6 +22,7 @@ from functools import partial as _partial import warnings as _warnings import inspect as _inspect from operator import attrgetter as _attrgetter +import threading as _threading # By making it public, we'll be able to call it from capabilityHelper.h cdef public object wrap_dynamic_struct_reader(Response & r): @@ -1192,7 +1193,7 @@ cdef class _EventLoop: def __dealloc__(self): self._remove() - cdef _remove(self) except +reraise_kj_exception: + cpdef _remove(self) except +reraise_kj_exception: del self.thisptr self.thisptr = NULL @@ -1201,12 +1202,44 @@ cdef class _EventLoop: raise ValueError('Event loop has already been destroyed') return deref(self.thisptr.lowLevelProvider).wrapSocketFd(fd) -cdef _EventLoop C_DEFAULT_EVENT_LOOP = _EventLoop() +C_DEFAULT_EVENT_LOOP = _EventLoop() -cpdef reset_event_loop(): - global C_DEFAULT_EVENT_LOOP - C_DEFAULT_EVENT_LOOP._remove() - C_DEFAULT_EVENT_LOOP = _EventLoop() +_C_DEFAULT_EVENT_LOOP_LOCAL = None + +cdef _EventLoop C_DEFAULT_EVENT_LOOP_GETTER(): + 'Optimization for not having to deal with threadlocal event loops unless we need to' + if C_DEFAULT_EVENT_LOOP is not None: + print C_DEFAULT_EVENT_LOOP + return <_EventLoop>C_DEFAULT_EVENT_LOOP + elif _C_DEFAULT_EVENT_LOOP_LOCAL is not None: + loop = getattr(_C_DEFAULT_EVENT_LOOP_LOCAL, 'loop', None) + if loop is not None: + return <_EventLoop>_C_DEFAULT_EVENT_LOOP_LOCAL.loop + + raise RuntimeError("You don't have any EventLoops running. Please make sure to add ") + +# cpdef remove_event_loop(): +# 'Remove the global event loop' +# global C_DEFAULT_EVENT_LOOP +# C_DEFAULT_EVENT_LOOP._remove() +# C_DEFAULT_EVENT_LOOP = None + +# cpdef create_event_loop(is_thread_local=True): +# '''Create a new global event loop. This will not remove the previous +# EventLoop for you, so make sure to do that first''' +# global C_DEFAULT_EVENT_LOOP +# global _C_DEFAULT_EVENT_LOOP_LOCAL +# if is_thread_local: +# if _C_DEFAULT_EVENT_LOOP_LOCAL is None: +# _C_DEFAULT_EVENT_LOOP_LOCAL = _threading.local() +# _C_DEFAULT_EVENT_LOOP_LOCAL.loop = _EventLoop() +# else: +# C_DEFAULT_EVENT_LOOP = _EventLoop() + +# cpdef reset_event_loop(): +# global C_DEFAULT_EVENT_LOOP +# C_DEFAULT_EVENT_LOOP._remove() +# C_DEFAULT_EVENT_LOOP = _EventLoop() cdef class _CallContext: cdef CallContext * thisptr @@ -1267,7 +1300,7 @@ cdef class Promise: if self.is_consumed: raise ValueError('Promise was already used in a consuming operation. You can no longer use this Promise object') - ret = self.thisptr.wait(C_DEFAULT_EVENT_LOOP.thisptr.waitScope) # TODO: make sure refcount is fine here... + ret = self.thisptr.wait(C_DEFAULT_EVENT_LOOP_GETTER().thisptr.waitScope) # TODO: make sure refcount is fine here... self.is_consumed = True return ret @@ -1300,7 +1333,7 @@ cdef class _VoidPromise: if self.is_consumed: raise ValueError('Promise was already used in a consuming operation. You can no longer use this Promise object') - self.thisptr.wait(C_DEFAULT_EVENT_LOOP.thisptr.waitScope) + self.thisptr.wait(C_DEFAULT_EVENT_LOOP_GETTER().thisptr.waitScope) self.is_consumed = True @@ -1339,7 +1372,7 @@ cdef class _RemotePromise: if self.is_consumed: raise ValueError('Promise was already used in a consuming operation. You can no longer use this Promise object') - ret = _Response()._init_child(self.thisptr.wait(C_DEFAULT_EVENT_LOOP.thisptr.waitScope), self._parent) + ret = _Response()._init_child(self.thisptr.wait(C_DEFAULT_EVENT_LOOP_GETTER().thisptr.waitScope), self._parent) self.is_consumed = True return ret @@ -1695,7 +1728,7 @@ cdef class _FdAsyncIoStream: self._init(fd) cdef _init(self, int fd) except +reraise_kj_exception: - self.thisptr = C_DEFAULT_EVENT_LOOP.wrapSocketFd(fd) + self.thisptr = C_DEFAULT_EVENT_LOOP_GETTER().wrapSocketFd(fd) cdef class PromiseFulfillerPair: cdef Own[C_PromiseFulfillerPair] thisptr diff --git a/test/test_rpc.py b/test/test_rpc.py index 03ab995..900587d 100644 --- a/test/test_rpc.py +++ b/test/test_rpc.py @@ -23,7 +23,7 @@ class SimpleRestorer(test_capability_capnp.TestSturdyRefObjectId.Restorer): def restore(self, ref_id): assert ref_id.tag == 'testInterface' return Server(100) - + def test_simple_rpc(): read, write = socket.socketpair(socket.AF_UNIX)