Python - Timer with asyncio/coroutine

Question

I am trying to set a timer that will interrupt the running process and call a coroutine when it fires. However, I'm not sure what the right way to accomplish this is. I've found AbstractEventLoop.call_later, along with threading.Timer but neither of these seem to work (or I'm using them incorrectly). The code is pretty basic and looks something like this:

def set_timer( time ):
    self.timer = Timer( 10.0, timeout )
    self.timer.start()
    #v2
    #self.timer = get_event_loop()
    #self.timer.call_later( 10.0, timeout )
    return

async def timeout():
    await some_func()
    return

What is the correct way to set a non-blocking timer, that will call a callback function after some number of seconds? Being able to cancel the timer would be a bonus but is not a requirement. The major things I need are: non-blocking and successfully calling the co-routine. Right now it returns an error that the object can't be await'd (if I toss an await in) or that some_func was never await'd, and the expected output never happens.

Answer 1

The solution proposed by Mikhail has one drawback. Calling cancel() cancels both: the timer and the actual callback (if cancel() fired after timeout is passed, but actual job is still in progress). Canceling the job itself may be not the desired behavior.

An alternative approach is to use loop.call_later:

async def some_job():
    print('Job started')
    await asyncio.sleep(5)
    print('Job is done')

loop = asyncio.get_event_loop() # or asyncio.get_running_loop()

timeout = 5
timer = loop.call_later(timeout, lambda: asyncio.ensure_future(some_job()))

timer.cancel() # cancels the timer, but not the job, if it's already started

Answer 2

Thanks Mikhail Gerasimov for your answer, it was very useful. Here is an extension to Mikhail’s anwer. This is an interval timer with some twists. Perhaps it is useful for some users.

import asyncio


class Timer:
    def __init__(self, interval, first_immediately, timer_name, context, callback):
        self._interval = interval
        self._first_immediately = first_immediately
        self._name = timer_name
        self._context = context
        self._callback = callback
        self._is_first_call = True
        self._ok = True
        self._task = asyncio.ensure_future(self._job())
        print(timer_name + " init done")

    async def _job(self):
        try:
            while self._ok:
                if not self._is_first_call or not self._first_immediately:
                    await asyncio.sleep(self._interval)
                await self._callback(self._name, self._context, self)
                self._is_first_call = False
        except Exception as ex:
            print(ex)

    def cancel(self):
        self._ok = False
        self._task.cancel()


async def some_callback(timer_name, context, timer):
    context['count'] += 1
    print('callback: ' + timer_name + ", count: " + str(context['count']))

    if timer_name == 'Timer 2' and context['count'] == 3:
        timer.cancel()
        print(timer_name + ": goodbye and thanks for all the fish")


timer1 = Timer(interval=1, first_immediately=True, timer_name="Timer 1", context={'count': 0}, callback=some_callback)
timer2 = Timer(interval=5, first_immediately=False, timer_name="Timer 2", context={'count': 0}, callback=some_callback)

try:
    loop = asyncio.get_event_loop()
    loop.run_forever()
except KeyboardInterrupt:
    timer1.cancel()
    timer2.cancel()
    print("clean up done")

Answer 3

Creating Task using ensure_future is a common way to start some job executing without blocking your execution flow. You can also cancel tasks.

I wrote example implementation for you to have something to start from:

import asyncio


class Timer:
    def __init__(self, timeout, callback):
        self._timeout = timeout
        self._callback = callback
        self._task = asyncio.ensure_future(self._job())

    async def _job(self):
        await asyncio.sleep(self._timeout)
        await self._callback()

    def cancel(self):
        self._task.cancel()


async def timeout_callback():
    await asyncio.sleep(0.1)
    print('echo!')


async def main():
    print('\nfirst example:')
    timer = Timer(2, timeout_callback)  # set timer for two seconds
    await asyncio.sleep(2.5)  # wait to see timer works

    print('\nsecond example:')
    timer = Timer(2, timeout_callback)  # set timer for two seconds
    await asyncio.sleep(1)
    timer.cancel()  # cancel it
    await asyncio.sleep(1.5)  # and wait to see it won't call callback


loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
try:
    loop.run_until_complete(main())
finally:
    loop.run_until_complete(loop.shutdown_asyncgens())
    loop.close()

Output:

first example:
echo!

second example: