asyncio - how to stop background task cleanly

[Frank Millman]

Hi all

It is easy enough to set up a task to run in the background every 10 seconds
using asyncio -

async def background_task():
while True:
await perform_task()
await asyncio.sleep(10)

asyncio.ensure_future(background_task())

When shutting the main program down, I want to stop the task, but I cannot
figure out how to stop it cleanly - i.e. wait until it has finished the
current task and possibly performed some cleanup, before continuing.

async def background_task():
await perform_setup()
while condition:
await perform_task()
await asyncio.sleep(10)
await perform_cleanup()

Previously I would run the task in another thread, then set a flag to tell
it to stop, then join() the thread which would block until the task had
finished. I used threading.Event as the flag, which allows it to 'sleep'
using wait() with a timeout value, but reacts instantly when set() is
called, so it was ideal.

Is there a way to achieve this using asyncio?

Thanks

Frank Millman

[Marko Rauhamaa]

"Frank Millman" <fr...@chagford.com>:

> When shutting the main program down, I want to stop the task, but I
> cannot figure out how to stop it cleanly - i.e. wait until it has
> finished the current task and possibly performed some cleanup, before
> continuing.

Here (and really, only here) is where asyncio shows its superiority over
threads: you can multiplex.

You should

await asyncio.wait(..., return_when=asyncio.FIRST_COMPLETED)

to deal with multiple alternative stimuli.

In fact, since there is always a minimum of two alternative stimuli to
await, you should only ever await asyncio.wait().

And, while viable, that's what makes every asyncio program ugly as hell.


Marko

[Frank Millman]

"Marko Rauhamaa" wrote in message news:87lh6ys...@elektro.pacujo.net...

>
> "Frank Millman" <fr...@chagford.com>:
>
> > When shutting the main program down, I want to stop the task, but I
> > cannot figure out how to stop it cleanly - i.e. wait until it has
> > finished the current task and possibly performed some cleanup, before
> > continuing.
>
> Here (and really, only here) is where asyncio shows its superiority over
> threads: you can multiplex.
>
> You should
>
> await asyncio.wait(..., return_when=asyncio.FIRST_COMPLETED)
>
> to deal with multiple alternative stimuli.
>

Thanks, Marko, that works very well.

It took me a while to get it working, because I initiate shutting down the
program from another thread. Eventually I figured out that I could put all
my event loop shutdown procedures into a coroutine, and then call
asyncio.run_coroutine_threadsafe() from the main thread.

Now I just have one problem left. I will keep experimenting, but if someone
gives me a hint in the meantime it will be appreciated.

I run my background task like this -

stop_task = False

async def background_task():
while not stop_task:
await perform_task()
await asyncio.sleep(10)

I stop the task by setting stop_task to True. It works, but it waits for the
10-second sleep to expire before it is actioned.

With threading, I could set up a threading.Event(), call
evt.wait(timeout=10) to run the loop, and evt.set() to stop it. It stopped
instantly.

Is there an equivalent in asyncio?

Thanks

Frank

[Marko Rauhamaa]

"Frank Millman" <fr...@chagford.com>:

> "Marko Rauhamaa" wrote in message news:87lh6ys...@elektro.pacujo.net...

>> You should
>>
>> await asyncio.wait(..., return_when=asyncio.FIRST_COMPLETED)
>>
>> to deal with multiple alternative stimuli.
>>
>
> Thanks, Marko, that works very well.
>

> [...]

>
> Now I just have one problem left. I will keep experimenting, but if
> someone gives me a hint in the meantime it will be appreciated.
>
> I run my background task like this -
>
> stop_task = False
>
> async def background_task():
> while not stop_task:
> await perform_task()
> await asyncio.sleep(10)

You should set up a separate asyncio.Event or asyncio.Queue to send
"out-of-band" signals to your background_task:

async def background_task(cancel_event):
async def doze_off():
await asyncio.sleep(10)

while True:
await asyncio.wait(
perform_task, cancel_event.wait,
return_when=asyncio.FIRST_COMPETED)
if cancel_event_is_set()
break
await asyncio.wait(
doze_off, cancel_event.wait,
return_when=asyncio.FIRST_COMPETED)
if cancel_event_is_set()
break

That should be the idea, anyway. I didn't try it out.

> I stop the task by setting stop_task to True. It works, but it waits for
> the 10-second sleep to expire before it is actioned.
>
> With threading, I could set up a threading.Event(), call
> evt.wait(timeout=10) to run the loop, and evt.set() to stop it. It
> stopped instantly.
>
> Is there an equivalent in asyncio?

Yes, you could simplify the above thusly:

async def background_task(cancel_event):
while True:
await asyncio.wait(
perform_task, cancel_event.wait,
return_when=asyncio.FIRST_COMPETED)
if cancel_event_is_set()
break
await asyncio.wait(
cancel_event.wait, timeout=10,
return_when=asyncio.FIRST_COMPETED)
if cancel_event_is_set()
break


Marko

[Marko Rauhamaa]

Marko Rauhamaa <ma...@pacujo.net>:

> async def background_task(cancel_event):
> while True:
> await asyncio.wait(
> perform_task, cancel_event.wait,
> return_when=asyncio.FIRST_COMPETED)
> if cancel_event_is_set()
> break
> await asyncio.wait(
> cancel_event.wait, timeout=10,
> return_when=asyncio.FIRST_COMPETED)
> if cancel_event_is_set()
> break

[Typo: cancel_event_is_set() ==> cancel_event.is_set().]

Actually, cancellation is specially supported in asyncio (<URL:
https://docs.python.org/3/library/asyncio-task.html#asyncio.Task.cancel>)
so this should do:

async def background_task():
while True:
await perform_task()
await asyncio.sleep(10)

Marko

[Frank Millman]

"Marko Rauhamaa" wrote in message news:8737t5s...@elektro.pacujo.net...

> >
> Actually, cancellation is specially supported in asyncio (<URL:
> https://docs.python.org/3/library/asyncio-task.html#asyncio.Task.cancel>)
> so this should do:
>
> async def background_task():
> while True:
> await perform_task()
> await asyncio.sleep(10)
>

That's exactly what I needed - thanks, Marko

async def background_task()
try:

while True:
await perform_task()
await asyncio.sleep(10)

except asyncio.CancelledError:
await perform_cleanup()

At startup -

task = asyncio.ensure_future(background_task())

At shutdown -

task.cancel()
await asyncio.wait([task])

Works perfectly - thanks again.

Frank

[Frank Millman]

"Frank Millman" wrote in message news:n96kjr$mvl$1...@ger.gmane.org...

>
> "Marko Rauhamaa" wrote in message

> news:8737t5s...@elektro.pacujo.net...

>
> > Actually, cancellation is specially supported in asyncio (<URL:
> > https://docs.python.org/3/library/asyncio-task.html#asyncio.Task.cancel>)
> > so this should do:
> >
> > async def background_task():
> > while True:
> > await perform_task()
> > await asyncio.sleep(10)
> >
>

> That's exactly what I needed - thanks, Marko
>
> async def background_task()
> try:

> while True:
> await perform_task()
> await asyncio.sleep(10)

> except asyncio.CancelledError:
> await perform_cleanup()
>
> At startup -
>
> task = asyncio.ensure_future(background_task())
>
> At shutdown -
>
> task.cancel()
> await asyncio.wait([task])
>
> Works perfectly - thanks again.
>

Alas, I spoke too soon.

I tried to simulate what would happen if the background task was busy with a
task when it was cancelled -

async def background_task()
try:
while True:

print('start')
time.sleep(2)
print('done')
await asyncio.sleep(10)
except asyncio.CancelledError:
print('cleanup')
print('DONE')

If I cancel after a pair of 'start/done' appear, the background task is in
the 'asyncio.sleep' stage. The words 'cleanup' and 'DONE' appear instantly,
and the program halts.

If I cancel after 'start', but before 'done', the background task is
executing a task. There is a delay of up to 2 seconds, then the words
'done', 'cleanup', and 'DONE' appear, but the program hangs. If I press
Ctrl+C, I get a traceback from the threading module -

line 1288, in _shutdown
t.join()
line 1054, in join
self._wait_for_tstate_lock()
line 1070, in _wait_for_tstate_lock
KeyboardInterrupt

So it is waiting for join() to complete. I will continue investigating, but
will report it here to see if anyone can come up with an
explanation/solution.

Thanks

Frank

[Marko Rauhamaa]

"Frank Millman" <fr...@chagford.com>:

> Alas, I spoke too soon.
>

> [...]

>
> If I press Ctrl+C, I get a traceback from the threading module -
>
> line 1288, in _shutdown
> t.join()
> line 1054, in join
> self._wait_for_tstate_lock()
> line 1070, in _wait_for_tstate_lock
> KeyboardInterrupt
>
> So it is waiting for join() to complete. I will continue
> investigating, but will report it here to see if anyone can come up
> with an explanation/solution.

I can't see your complete program, but here's mine, and it seems to be
working:

========================================================================
#!/usr/bin/env python3

import asyncio, time

def main():
loop = asyncio.get_event_loop()
try:
task = asyncio.async(background_task())
loop.run_until_complete(asyncio.wait([ task, terminator(task) ]))
finally:
loop.close()

@asyncio.coroutine
def terminator(task):
yield from asyncio.sleep(5)
task.cancel()
yield from asyncio.wait([ task ])

@asyncio.coroutine

def background_task():
try:
while True:
print('start')
time.sleep(2)
print('done')

yield from asyncio.sleep(10)

except asyncio.CancelledError:
print('cleanup')
print('DONE')

if __name__ == '__main__':
main()
========================================================================

(My Python is slightly older, so replace

yield from ==> await
@asyncio.coroutine ==> async
asyncio.async ==> asyncio.ensure_future)


Marko

[Frank Millman]

"Marko Rauhamaa" wrote in message news:87r3gpq...@elektro.pacujo.net...

>
> I can't see your complete program, but here's mine, and it seems to be
> working
>

Thanks, Marko, I really appreciate your assistance.

I wanted to show you my complete program, but as it is quite long I
distilled it down to its essence, and lo and behold it works!

So now I just have to go through my program and find where it differs - I
must have a bug somewhere.

For the record, here is my stripped down version.

The main difference from yours is that I want to run a genuine 'loop
forever', and only shut it down on receipt of some external signal.

I have never been able to get Ctrl+C to work properly on Windows, so I use a
separate thread that simply waits for Enter.

You will see that if you press Enter after 'done' appears, the program
closes instantly, but if you press it in between 'start' and 'done', it
waits for the task to complete before it closes.

Frank

=============================================================

import asyncio, time
import threading

def main():
loop = asyncio.get_event_loop()

task = asyncio.async(background_task())
threading.Thread(target=stop, args=(loop, task)).start()
loop.run_forever()

@asyncio.coroutine
def background_task():
try:
while True:
print('start')
time.sleep(2)
print('done')

yield from asyncio.sleep(5)

except asyncio.CancelledError:
print('cleanup')
print('DONE')

@asyncio.coroutine
def shutdown(loop, task):
task.cancel()
yield from asyncio.wait([task], loop=loop)
loop.stop()

def stop(loop, task):
input('Press <enter> to stop\n')
asyncio.run_coroutine_threadsafe(shutdown(loop, task), loop)

[Marko Rauhamaa]

"Frank Millman" <fr...@chagford.com>:

> I have never been able to get Ctrl+C to work properly on Windows, so I
> use a separate thread that simply waits for Enter.

Now you are leaving my realm of expertise, as I haven't programmed for
windows since Windows 1.0. Mixing threads, asyncio, input() and Windows
seems like begging for trouble, though.


Marko

[Frank Millman]

"Marko Rauhamaa" wrote in message news:871t8or...@elektro.pacujo.net...

Well, it is not such a big deal :-) The program I posted works
cross-platform, so it will run on your linux box.

No matter, you have been an enormous help, and I am very grateful.

I found my bug, and my program is now running sweetly.

Frank