Problem canceling tasks

[fmafma]

Hi!

In the follwoing code:

class Pipo:
    def __init__(self):
        loop = asyncio.get_event_loop()
        self._task1 = loop.create_task(self._loop1())
        self._task2 = loop.create_task(self._loop2())

    async def _loop1(self):
        print("entering loop1")
        while True:
            try:
                await asyncio.sleep(0.1)

            except asyncio.CancelledError:
                break

            except:
                print("Pipo._loop1(): unhandled exception")

        print("Pipo._loop1(): exiting")

    async def _loop2(self):
        print("entering loop2")
        while True:
            try:
                await asyncio.sleep(0)

            except asyncio.CancelledError:
                break

            except:
                print("Pipo._loop1(): unhandled exception")

        print("Pipo._loop2(): exiting")

    def stop(self):
        print("Pipo.stop()")

        self._task1.cancel()
        self._task2.cancel()


class Blah:
    def __init__(self):
        self._pipo = Pipo()

    def stop(self):
        print("Blah.stop()")
        self._pipo.stop()
        print("Blah.stop(): all tasks stopped")


async def dummy(stack):
    while True:
        await asyncio.sleep(0)


blah = Blah()
try:
    asyncio.run(dummy(blah))

except KeyboardInterrupt:
    print("execution canceled (SIGTERM)")

finally:
    blah.stop()
    asyncio.new_event_loop()

I sometimes get an error when I hit Ctrl-C (but not always):

$ micropython cancel.py
entering loop1
entering loop2
^Cexecution canceled (SIGTERM)
Blah.stop()
Pipo.stop()
Traceback (most recent call last):
  File "cancel.py", line 71, in <module>
  File "cancel.py", line 54, in stop
  File "cancel.py", line 45, in stop
RuntimeError: can't cancel self

Any idea what's going on?

[fmafma]

In fact, my code is more like this one, where I get the error every time:

# -*- coding: utf-8 -*-

import time
import asyncio

import socket

async def recv(s):
    while True:
        try:
            s.recvfrom(1024)
        except OSError:
            await asyncio.sleep(0)


class Pipo:
    def __init__(self):
        loop = asyncio.get_event_loop()
        self._task1 = loop.create_task(self._loop1())
        self._task2 = loop.create_task(self._loop2())

    async def _loop1(self):
        print("entering loop1")
        while True:
            try:
                await asyncio.sleep(0.1)

            except asyncio.CancelledError:
                break

        print("Pipo._loop1(): exiting")

    async def _loop2(self):
        print("entering loop2")

        s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
        while True:
            try:
                await recv(s)

            except asyncio.CancelledError:
                break

        print("Pipo._loop2(): exiting")

    def stop(self):
        print("Pipo.stop()")

        self._task1.cancel()
        self._task2.cancel()


class Blah:
    def __init__(self):
        self._pipo = Pipo()

    def stop(self):
        print("Blah.stop()")
        self._pipo.stop()
        print("Blah.stop(): all tasks stopped")


async def dummy():
    while True:
        await asyncio.sleep(0)


blah = Blah()
try:
    asyncio.run(dummy())

except KeyboardInterrupt:
    print("execution canceled (SIGTERM)")

finally:
    blah.stop()
    asyncio.new_event_loop()

[peterhinch]

The program structure needs significant revision. Firstly the run statement

asyncio.run(dummy())

is intended to start asyncio. When the run() statement terminates, asyncio is stopped. However by instantiating Blah you have already created tasks. All tasks should be started - directly or indirectly - by run().

The finally clause runs after run() has stopped and asyncio has terminated. Yet blah.stop() is issuing asyncio commands. This cannot end well :)

I suggest you look at the tutorial to learn how to structure asyncio programs.

The following works correctly:

# -*- coding: utf-8 -*-

import time
import asyncio

import socket

async def recv(s):
    while True:
        try:
            s.recvfrom(1024)
        except OSError:
            await asyncio.sleep(0)


class Pipo:
    def __init__(self):
        loop = asyncio.get_event_loop()
        self._task1 = loop.create_task(self._loop1())
        self._task2 = loop.create_task(self._loop2())

    async def _loop1(self):
        print("entering loop1")
        while True:
            try:
                await asyncio.sleep(0.1)

            except asyncio.CancelledError:
                break

        print("Pipo._loop1(): exiting")

    async def _loop2(self):
        print("entering loop2")

        s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
        while True:
            try:
                await recv(s)

            except asyncio.CancelledError:
                break

        print("Pipo._loop2(): exiting")

    def stop(self):
        print("Pipo.stop()")

        self._task1.cancel()
        self._task2.cancel()


class Blah:
    def __init__(self):
        self._pipo = Pipo()

    def stop(self):
        print("Blah.stop()")
        self._pipo.stop()
        print("Blah.stop(): all tasks stopped")


async def dummy():
    blah = Blah()
    try:
        while True:
            await asyncio.sleep(0)
    finally:
        blah.stop()
        await asyncio.sleep(0)  # Allow cancellation to take effect
        raise


try:
    asyncio.run(dummy())
except KeyboardInterrupt:
    print("execution canceled (SIGTERM)")
finally:
    asyncio.new_event_loop()

[fmafma]

I did read your great documentation, but it seems I missed an important point :o/
Anyway, thank you for your answer, Peter!

[fmafma]

Mmm, I just tested your code, and it does not work as expected, neither on the unix port nor on the rp2 port; when hitting Ctr-C, I only get the message "execution canceled (SIGTERM)", but none of the stop() routines are executed!

On CPython, however, it works, but I have to hit Ctrl-C twice...

[fmafma]

@peterhinch which version of unix port are you running? I'm using:
MicroPython v1.22.1+ds-1+b1 on 2024-03-24; linux [GCC 13.2.0] version...
And on my RPi Pico W:
MicroPython v1.23.0 on 2024-06-02; Raspberry Pi Pico W with RP2040

[peterhinch]

Apologies. My restructured program could be interrupted without error, but was not calling blah.stop().

On further thought and testing, I think what you are trying to do is impossible. The following is my opinion - I'll do some further thinking/testing.

The essence of the problem is a need to cancel tasks after a ctrl-c has been received. The KeyboardInterrupt interrupts the scheduler itself, so tasks which rely on the scheduler will never again get execution. It is therefore impossible to cancel them because they have effectively ceased to exist. Indeed it's not possible to execute any code that relies on the existence of the event loop.

Given that asyncio.run() has terminated it would be possible to issue a new asyncio.run() command, but this would be an entirely new instance with no "knowledge" of previous tasks.

I'd welcome any comments from others with experience of asyncio.

[fmafma]

Any idea why its works as expected on CPython but not on Micropython?

[peterhinch]

I don't know the answer to that, but I think it's worth bearing in mind that the focus of CPython is on desktop applications which may routinely be stopped with ctrl-c. In microcontroller (firmware) applications ctrl-c is typically an emergency exit when debugging code. Many firmware applications never exit; those that do usually use another mechanism such as polling a Pin or receiving a message on some interface.

You might want to raise an issue, but if you do, I suggest you write a minimal reproducer.

[beetlegigg]

Just a comment as I tried the initial code posted (I was intending to copy @peterhinch example but I mistakenly copied the code of the first post), and it ran and cancelled the tasks. It did this several times without a problem until, as @fmafma indicates it stops with the traceback as was shown. Trying again a short time later and it worked and cancelled the tasks again, until it did not again :-)

I was giving this a go to improve my understanding of asyncio. Reading the rational from Peter it all makes sense that if Ctrl C cancels the asyncio event loop then the tasks wont run and there's no point in cancelling them, but it does seem strange that using the original example code posted, sometimes the code will run to completion, and will cancel the tasks without any Traceback being issued. Maybe something to do with the current sequence that the program is executing when Ctrl C triggers the Terminate signal? But I guess that at the end of the day, what I've learn from this is that a Ctrl C will, in most cases, immediately stop the asynio event loop from running.

[peterhinch]

This type of non-deterministic behaviour is common in asynchronous code.

Maybe something to do with the current sequence that the program is executing when Ctrl C triggers the Terminate signal?

Very likely.

[kapetan]

I encountered the same issue trying to do some cleanup after a keyboard interrupt. Sometimes the following code would raise RuntimeError: can't cancel self.

import asyncio
import sys

loop = asyncio.get_event_loop()
task = loop.create_task(work())

try:
    loop.run_until_complete(task)
finally:
    if not task.done():
        task.cancel() # This raises RuntimeError

        try:
            loop.run_until_complete(task)
        except BaseException as e:
            sys.print_exception(e)

    asyncio.new_event_loop()

I managed to get more predictable behaviour by setting asyncio.core.cur_task to None.

import asyncio
import asyncio.core

loop = asyncio.get_event_loop()
task = loop.create_task(work())

try:
    loop.run_until_complete(task)
finally:
    if not task.done():
        asyncio.core.cur_task = None # Reset current task
        task.cancel()

        try:
            loop.run_until_complete(task)
        except BaseException as e:
            sys.print_exception(e)

    asyncio.new_event_loop()

This is still a hack, the KeyboardInterrupt may leave the program in an invalid state.

[peterhinch]

A KeyboardInterrupt is necessarily drastic because its function is to provide a guaranteed means of breaking into running code. In the case of asyncio programs there is a high probability that the scheduler itself will be interrupted at some arbitrary point in its execution. The status of the event loop, the task queue and other elements of state will be indeterminate. I don't think there is any way to resume a task or to do anything with asyncio after a KeyboardInterrupt.

The purpose of trapping the interrupt at the outermost scope is to perform synchronous operations like closing files and sockets. Creating a new event loop effectively clears asyncio state to enable another asyncio program to be started at the REPL.

[sosi-deadeye]

Just clear all tasks in asyncio.core._task_queue. This

import asyncio


async def worker(name):
    await asyncio.sleep(10)
    print(f"Worker: '{name}' done")


def sync_cancel_tasks():
    """
    which is better?
    
    creating a new queue and let the garbage collector do it's work?
    asyncio.core._task_queue = asyncio.TaskQueue()
    
    or popping all tasks and do not assign a new TaskQueue
    """
    try:
        while asyncio.core._task_queue.pop():
            pass
    except IndexError:
        pass


async def main():
    tasks = [asyncio.create_task(worker(f"Worker {n}")) for n in range(1, 5)]
    print("submit now a CTRL+C to stop the event loop") 

    # will block until all tasks are finished
    # if CTRL+C is submitted, the event loop stops
    # the exception does not occour here, it occours at asyncio.run
    await asyncio.gather(*tasks)


async def main2():
    await asyncio.sleep(11)
    print("Main 2 done")


try:
    asyncio.run(main())
except KeyboardInterrupt:
    # comment this out and the tasks will run
    # after main2() is executed long enough by the event loop
    sync_cancel_tasks()
    
    # IndexError: empty heap
    # print(asyncio.core._task_queue.pop())
    ...


print("Running now main2. If sync_calcel_tasks were not called, the old tasks from create_taks are executed also.")
asyncio.run(main2())

Output:

MPY: soft reboot
submit now a CTRL+C to stop the event loop
Running now main2. If sync_calcel_tasks were not called, the old tasks from create_taks are executed also
Main 2 done
MicroPython v1.24.0-preview.363.g17d823445 on 2024-09-26; Generic ESP32S3 module with Octal-SPIRAM with ESP32S3
Type "help()" for more information.

Output, if sync_cancel_taks is commented out:

MPY: soft reboot
submit now a CTRL+C to stop the event loop
Running now main2. If sync_calcel_tasks were not called, the old tasks from create_taks are executed also.
Worker: 'Worker 1' done
Worker: 'Worker 2' done
Worker: 'Worker 3' done
Worker: 'Worker 4' done
Main 2 done
MicroPython v1.24.0-preview.363.g17d823445 on 2024-09-26; Generic ESP32S3 module with Octal-SPIRAM with ESP32S3
Type "help()" for more information.

[kapetan]

I don't think this has the desired effect I'm after. I want to inject a CancelledError into the tasks so the except/finally statements get executed.