Stop a thread as soon as possible without having to add "if not running: break" before each line of code?

Question

^{Context: I've aleady read Is there any way to kill a Thread? and its answers, but here I'm focusing on a possible solution with a function (inside a wrapper?) that would automatically check a flag running before actually running each line of code of the thread target function. Also I have already read Python, Stop a Thread but it does not solve the problem since the flag checking "granularity" is only once every while loop, whereas I'd like it to be at each line of code.}

With the following code (simplified for this question):

import threading, time

def character_loop():
    while running:
        print('character moving left')
        time.sleep(1)
        print('character moving middle')
        time.sleep(1)
        print('character moving right')
        time.sleep(1)

running = True
threading.Thread(target=character_loop).start()
time.sleep(4.5)
running = False     # how to stop *any* movement here?
time.sleep(10)

it is in fact doing two full loops, i.e. left/mid/right/left/mid/right, whereas I would like to stop any movement after 4.5 seconds.

How to have left/mid/right/left/<stop here!> instead?

In other words, how to be able to stop a thread as soon as possible? (possibly after each line of code running in the thread target function)

This works:

def character_loop():
    while running:
        print('character moving left')
        if not running: 
            break
        time.sleep(1)
        if not running: 
            break
        print('character moving middle')
        if not running: 
            break
        time.sleep(1)
        if not running: 
            break
        print('character moving right')
        if not running: 
            break
        time.sleep(1)

but having to add

if not running: 
    break

after each line of code in the main loop is really annoying.

I'm sure there's a general threading / scheduling solution for this: how to be able to stop a thread function as soon as possible, ideally after each line of code, and not only when the function comes back to while running:?

PS: in my real code, it's not just print('character moving left') but another function that takes a few milliseconds to seconds to complete depending on the action.

This is or should be a part of Python threading FAQ - there is no such way even though it would be immensely useful. There is no way to signal a thread and tell it to exit or do something else. You can signal processes as you probably know - but it is not suitable for everything. — Hannu, Jun 01 '20 at 10:40
@Hannu We could probably create a wrapper function for the thread function `character_loop`, that automatically checks if `running` is `True` before running each line of code inside this function, it would be some sort of "signalling". I wonder if this could be done easily without dirty hacks? — Basj, Jun 01 '20 at 11:07
You still have to check each time, but loop over the items to clean it up? `items = ["left", "mid", "right"] for item in items: print(item) if not running: break time.sleep(1) ` — Kassym Dorsel, Jun 01 '20 at 12:19
@KassymDorsel In my code it's not as simple as left, mid, right, here it was just an example. The best would be to have a checking done *automatically* after each line of ceode. — Basj, Jun 01 '20 at 12:23
I agree, but this would be a workaround. You could turn each one into its own function and have `items = [func1, func2, func3]` and call them in the loop. This would be a, hopefully, easy wrapper — Kassym Dorsel, Jun 01 '20 at 12:27
@KassymDorsel Why not in general, byt in my real code, this would be even more unnatural in the code than doing a check `if not running: return` before every line. — Basj, Jun 01 '20 at 12:30
Can you turn it into asyncio? A task object has cancel signal that raises an error. https://docs.python.org/3/library/asyncio-task.html#asyncio.Task.cancel — Kassym Dorsel, Jun 01 '20 at 12:36
@KassymDorsel I'm not experienced enough with asyncio. Could you please post an answer showing how to do my original code with asyncio? — Basj, Jun 01 '20 at 12:53

score 1 · Answer 1 · answered Jun 01 '20 at 13:39

Switch from using threads to asyncio which supports immediate cancellation of tasks. However this could prove to be difficult depending on how the code is structured. You will need an asyncio loop in the main tread to be started. Currently this is done with the asyncio.run() and the main(), but will be different in your implementation.

import asyncio


async def runner():
    try:
        while True:
            print("left")
            await asyncio.sleep(1)
            print("mid")
            await asyncio.sleep(1)
            print("right")
            await asyncio.sleep(1)
    except asyncio.CancelledError:
        print("Task was canceled")


async def main():
    # In Python 3.7+
    task = asyncio.create_task(runner())
    # This works in all Python versions but is less readable
    # task = asyncio.ensure_future(runner())
    await asyncio.sleep(4.5)
    task.cancel()
    await asyncio.sleep(5)


if __name__ == "__main__":
    asyncio.run(main())

Stop a thread as soon as possible without having to add "if not running: break" before each line of code?

1 Answers1