Hi; how can i interrupt an async function during its allotted runtime?

Question

We are coding behaviors for a robot in python async; we would like the robot to have multiple states, within each of which it will exhibit multiple behaviors, some random, some in order. These different state will be triggered by an input from a camera.

One such example is:

async def state1():
    while True:
        await behaviourA()
        await asyncio.sleep(3)
        await behaviourB()
        if finalstate != 1
            break

Lets assume behaviourA and B are both non-loop functions that run for 5 seconds.

We have a state function, simplified as such:

async def states():               # conditions functions
    while True:
        global finalstate           # this is the variable that the camera outputs that decides the state
        
        async def choice():
            while i < 10:
                print('choosing',finalstate)
                if finalstate == 1:
                    print('chose 1')
                    await state1()
                
                elif finalstate == 2:
                    print('chose 2')
                    await state2()
                    
                elif finalstate == 3:
                    print('chose 3')
                    await state3()
                    
                else:
                    await asyncio.sleep(0.5)
        
        tasks = [
        asyncio.ensure_future(decision()),            # this function modifies the finalstate variable
        asyncio.ensure_future(choice())
        ]
        
        await asyncio.wait(tasks)

The problem is that, for example in state1, during behaviourA(), the program is unresponsive to changes in finalstate. we would like the robot to change state immediately, or as quickly as possible, in reaction to finalstate; yet at the moment it seems to be either stuck on behaviourA indefinitely, or fails to run the choice subfunction until behaviourA completes.

I understand that it is probably a problem with using 'await state' in choice, but I'm not sure what we can do. We have tried instead using create_task and calling the state, but that doesn't seem to work either.

What can we do to remedy this?

It depends on how you want to handle the interrupt. You could try to catch `KeyboardInterrupt` and look for `CTRL-C` (for example). In your try/catch block, you can quit the various tasks in `tasks` by `task.cancel()` them. See https://stackoverflow.com/a/30766124/1141805 for more. — davidlowryduda, Apr 24 '23 at 15:45

score 0 · Answer 1 · answered Apr 24 '23 at 15:53

Your question is not perfectly clear, however...
Maybe you can try to modify the state1() function that is blocking the event loop, preventing the program from being responsive to changes in finalstate. The state1() method can be changed into smaller, non-blocking tasks. For instance:

async def state1():
    while True:
        # run coroutines concurrently as asyncio Tasks, instead of being awaited directly
        task1 = asyncio.create_task(behaviourA())
        await asyncio.sleep(3)
        task2 = asyncio.create_task(behaviourB())
        # wait for both tasks to complete before moving on to the next iteration
        await asyncio.gather(task1, task2)
        if finalstate != 1:
        #if finalstate changes before the tasks complete, break and exit
            break

You can apply the same approach to state2() and state3().

Hi; how can i interrupt an async function during its allotted runtime?

1 Answers1