Get rid of capnp timer functionality.
The asyncio timer should now be used
This commit is contained in:
@@ -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 = <AsyncIoEventPort*>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 = <AsyncIoEventPort*>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 = <void*>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 = <double>nextEvent / 1000
|
||||
us = <void*>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
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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():
|
||||
|
||||
@@ -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)
|
||||
|
||||
Reference in New Issue
Block a user