diff --git a/capnp/lib/capnp.pyx b/capnp/lib/capnp.pyx index 1b2c89d..014b1eb 100644 --- a/capnp/lib/capnp.pyx +++ b/capnp/lib/capnp.pyx @@ -10,8 +10,7 @@ cimport cython # noqa: E402 from capnp.helpers.helpers cimport init_capnp_api -from capnp.includes.capnp_cpp cimport AsyncIoStream, WaitScope, PyPromise, VoidPromise, EventPort, EventLoop, WaitScope, LowLevelAsyncIoProvider, AsyncIoProvider, newAsyncIoProvider, MonotonicClock, Timer, TimerImpl, systemPreciseMonotonicClock, MILLISECONDS, Canceler, PyAsyncIoStream, PromiseFulfiller, VoidPromiseFulfiller, makeException -from capnp.includes.capnp_cpp cimport AsyncIoStream, WaitScope, PyPromise, VoidPromise, EventPort, EventLoop, WaitScope, LowLevelAsyncIoProvider, AsyncIoProvider, newAsyncIoProvider, MonotonicClock, Timer, TimerImpl, systemPreciseMonotonicClock, MILLISECONDS, Canceler, PyAsyncIoStream, PromiseFulfiller, VoidPromiseFulfiller, tryReadMessage, writeMessage, makeException +from capnp.includes.capnp_cpp cimport AsyncIoStream, WaitScope, PyPromise, VoidPromise, EventPort, EventLoop, WaitScope, LowLevelAsyncIoProvider, AsyncIoProvider, newAsyncIoProvider, Canceler, PyAsyncIoStream, PromiseFulfiller, VoidPromiseFulfiller, tryReadMessage, writeMessage, makeException from capnp.includes.schema_cpp cimport (MessageReader,) from cpython cimport array, Py_buffer, PyObject_CheckBuffer, memoryview, buffer @@ -1817,28 +1816,19 @@ cdef class _DynamicObjectBuilder: cdef void kjloop_runnable_callback(void* data) with gil: cdef AsyncIoEventPort *port = data assert port.runHandle is not None - port.timerImpl.advanceTo(systemPreciseMonotonicClock().now()) port.kjLoop.run() -cdef void kjloop_advance_callback(void* data) with gil: - cdef AsyncIoEventPort *port = data - assert port.runHandle is not None - port.timerImpl.advanceTo(systemPreciseMonotonicClock().now()) - cdef cppclass AsyncIoEventPort(EventPort): EventLoop *kjLoop - TimerImpl *timerImpl; object asyncioLoop; object runHandle; __init__(object asyncioLoop): this.kjLoop = new EventLoop(deref(this)) - this.timerImpl = new TimerImpl(systemPreciseMonotonicClock().now()) this.runHandle = None this.asyncioLoop = asyncioLoop __dealloc__(): - del this.timerImpl del this.kjLoop cbool wait() except* with gil: @@ -1852,33 +1842,17 @@ cdef cppclass AsyncIoEventPort(EventPort): void setRunnable(cbool runnable) except* with gil: if runnable: - if this.runHandle is not None: - # If a timer was running, cancel it and schedule a run immediately - # The timer will be re-scheduled once the kj loop becomes un-runnable again. - this.runHandle.cancel() + assert this.runHandle is None us = this; this.runHandle = this.asyncioLoop.call_soon(lambda: kjloop_runnable_callback(us)) else: assert this.runHandle is not None this.runHandle.cancel() - this.scheduleAdvance() - - void scheduleAdvance() with gil: - cdef uint64_t nextEvent = this.timerImpl.timeoutToNextEvent( - systemPreciseMonotonicClock().now(), MILLISECONDS, -1).orDefault(-1) - if nextEvent == -1: this.runHandle = None - else: - seconds = nextEvent / 1000 - us = this; - this.runHandle = this.asyncioLoop.call_later(seconds, lambda: kjloop_advance_callback(us)) EventLoop *getKjLoop(): return this.kjLoop - Timer *getTimer(): - return this.timerImpl; - def _asyncio_close_patch(loop, oldclose, _EventLoop kjloop): # The purpose of patching the asyncio close() function is to set up the kj-loop to be closed as well. # We replace the event loop getter with a weakref, such that it can be destroyed when all other @@ -1893,7 +1867,6 @@ cdef class _EventLoop: cdef Own[LowLevelAsyncIoProvider] lowLevelProvider cdef Own[AsyncIoProvider] provider cdef WaitScope * waitScope - cdef Timer* timer cdef readonly in_asyncio_mode cdef AsyncIoEventPort *customPort @@ -1907,7 +1880,6 @@ cdef class _EventLoop: self.customPort = new AsyncIoEventPort(loop) kjLoop = self.customPort.getKjLoop() self.waitScope = new WaitScope(deref(kjLoop)) - self.timer = self.customPort.getTimer() loop.close = _partial(_asyncio_close_patch, loop, loop.close, self) self.in_asyncio_mode = True except RuntimeError: @@ -1915,7 +1887,6 @@ cdef class _EventLoop: self.lowLevelProvider = move(ptr.lowLevelProvider) self.provider = move(ptr.provider) self.waitScope = &ptr.waitScope - self.timer = &self.lowLevelProvider.get().getTimer() del ptr self.in_asyncio_mode = False @@ -1960,24 +1931,6 @@ cdef _EventLoop C_DEFAULT_EVENT_LOOP_GETTER(): return _C_DEFAULT_EVENT_LOOP_LOCAL.loop -cdef class _Timer: - cdef capnp.Timer * thisptr - - cdef _init(self, capnp.Timer * timer): - self.thisptr = timer - return self - - cpdef after_delay(self, time) except +reraise_kj_exception: - return _VoidPromise()._init(self.thisptr.afterDelay(capnp.Nanoseconds(time))) - - -def getTimer(): - """ - Get libcapnp event loop timer - """ - return _Timer()._init(C_DEFAULT_EVENT_LOOP_GETTER().timer) - - cpdef remove_event_loop(): '''Remove the event loop''' global _C_DEFAULT_EVENT_LOOP_LOCAL diff --git a/examples/async_ssl_server.py b/examples/async_ssl_server.py index fc3941a..3d84e2d 100755 --- a/examples/async_ssl_server.py +++ b/examples/async_ssl_server.py @@ -20,16 +20,13 @@ 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(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(1) def alive(self, **kwargs): return True diff --git a/examples/thread_server.py b/examples/thread_server.py index 25b2ae0..79b0dea 100755 --- a/examples/thread_server.py +++ b/examples/thread_server.py @@ -11,14 +11,12 @@ class ExampleImpl(thread_capnp.Example.Server): def subscribeStatus(self, subscriber, **kwargs): return ( - capnp.getTimer() - .after_delay(10**9) - .then(lambda: subscriber.status(True)) + subscriber.status(True) .then(lambda _: self.subscribeStatus(subscriber)) ) def longRunning(self, **kwargs): - return capnp.getTimer().after_delay(1 * 10**9) + return def parse_args(): diff --git a/test/test_capability.py b/test/test_capability.py index 01c6f25..19229f6 100644 --- a/test/test_capability.py +++ b/test/test_capability.py @@ -300,35 +300,6 @@ def test_cancel(): req.wait() -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(): client = capability.TestInterface._new_client(Server()) @@ -349,11 +320,6 @@ def test_then_args(): 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)