Flask-socketio: Where to put the code to stop a background thread?

[johschmitz]

I have been unsuccessfully searching for quite a long time now to find the recommended way of stopping a background thread that has been started with something like:

socketio = SocketIO(app, async_mode='gevent')
background_thread = socketio.start_background_task(target=bg_task.watcher)

Can somebody tell me where to put a piece of code like:

bg_task.stop()
background_thread.join()

[miguelgrinberg]

You don't normally stop a thread like that, you have to signal to the thread so that it stops gracefully. For example you can use an Event object for that. Have a look at this question on Stack Overflow: https://stackoverflow.com/questions/323972/is-there-any-way-to-kill-a-thread.

[johschmitz]

My bg_task.stop() is supposed to gracefully stop the thread by flipping a Boolean and allowing it to return from the loop, similar to what you suggest.
My question though is where can I call it?
I was trying something like this:

def sigint_handler(signal, frame):
    bg_task.stop()
    background_thread.join()

signal.signal(signal.SIGINT, sigint_handler)

Or something like this:

import atexit
atexit.register(bg_task.stop())

What is the recommended way of stopping such a thread during shutdown of gunicorn? It doesn't even let me define my own handler for SIGINT...

[johschmitz]

I found that my background thread only works with async_mode='threading' and not with 'gevent' or 'eventlet'. I probably need to work on a minimal example to properly discuss my whole problem with you.

[johschmitz]

I refactored my code and figured out that the important thing missing for me was a socketio.sleep(0) or a socketio.sleep(0.01) which works slightly better (not exactly sure why). This needs to be in the background thread loop. I guess I need to study some async/await/asyncio internals to gain a better understanding of the sleep calls.

[miguelgrinberg]

@johschmitz the sleep call in async applications allows the CPU to be shared among all the running tasks. If your function does not perform any I/O then calling sleep is the only way to release the CPU and trigger a context switch. Without the sleep that one task will continue to hold on to the CPU and all other running tasks will starve.

[johschmitz]

How could I forget the Python GIL issues .. now everything becomes clear. Always good to refresh one's knowledge.
For other people that will stumble over this Python GIL related issue in the future, (re)consider this minimal example: https://stackoverflow.com/questions/92928/time-sleep-sleeps-thread-or-process

@miguelgrinberg, before closing this ticket, since you are the expert: is there any way to circumvent the GIL issue in Flask with Python multiprocessing or is it eventually unavoidable to use something like Celery when my app performance becomes CPU limited?

[miguelgrinberg]

@johschmitz sorry, but you are wrong, this has nothing to do with the GIL. The Python GIL affects multithreaded applications. This is an async application, so it uses a single thread, and thus is immune to GIL effects since there is no thread concurrency.

This is a specific characteristic of async applications, which have cooperative context switches. If a task does not willingly release the CPU, then none of the other concurrent tasks get to run.

If your CPU intensive tasks cannot be sprinkled with sleep calls to allow other tasks to run, then you'll have to consider moving the CPU work to another process, either via Celery or something else.

[johschmitz]

Might be that I misunderstood it all but I was under the impression that start_background_task() starts a new thread, especially since the documentation says:
This function returns an object compatible with the Thread class in the Python standard library. The start() method on this object is already called by this function.
So is it that the background thread/task is "simulated" with asyncio or what is the meaning of "compatible with"? In that case it should maybe be documented more clearly that it is not a real python thread.
Sorry for the confusion. I promise to also have a look into asyncio to close my knowledge gap.

[miguelgrinberg]

There is no simulation, each async framework implements concurrency in its own way, and the start_background_task uses whatever the async framework uses. In the case of asyncio, tasks are started as coroutines. In the case of eventlet and gevent they are greenlets. If you need to work with these elements at a low level I suggest you familiarize yourself with how async applications work, everything will start to make more sense when you do.

[VladimirVLF]

Hi, I have somewhat related question.
I have a function running in background thread that sleeps most of the times using socketio.sleep(some_time). Basically it is a function doing some periodic work.
Now, when I shutdown/restart the server (uwsgi) it hangs for that some_time and in the worst case just kills the thread.

Is there a possibility to interrupt long socketio.sleep()? Basically I am looking for non-blocking Event.wait() functionality.

[miguelgrinberg]

@VladimirVLF Both Eventlet and Gevent have green versions of Event.

[VladimirVLF]

Now used Gevent's Event to wait for stop signal but hit another wall, which is graceful shutdown from the server. Are there any common recipes to implement that in long-waiting workers?

I use uwsgi.atexit, which apparently doesn't work, especially in async mode (https://github.com/unbit/uwsgi/issues...)

from gevent.event import Event
import uwsgi

stop_event = Event()

def cleanup():
    stop_event.set()

uwsgi.atexit = cleanup

def thread_function():
    while not stop_event.wait(600):
        # Some thread work
        pass
    # Some thread clean-ups
    pass

[miguelgrinberg]

@VladimirVLF Not sure I understand. Are you saying that in addition to your background thread, now you also want to stop the web server?

Web servers are not designed to stop in general, but some provide mechanisms to do so. Many servers respond to the SIGINT signal, the same signal that is issued when you press Ctrl-C on a server that is running as a foreground process.

[VladimirVLF]

@miguelgrinberg I don't want to stop the server from my application, just to gracefully handle the shutdown coming from the server. Basically, when I need to update my application I ask uwsgi to reload and it hangs as I have "infinite" threads running there. I thought there is some common way to do it, apparently not.

[miguelgrinberg]

Sending SIGINT to the process is the most generic solution. Have you tried it? If you have other threads, you can add a custom SIGINT handler where you signal and join all those threads.

[VladimirVLF]

@miguelgrinberg I tried the code below, hoping to catch one of the three signals, but no luck. This is due to uwsgi specific way of reloading: I use --touch-reload option and on reload uwsgi gives workers 60 seconds to finish, but does not really send any signal to them. I haven't figured out how to gracefully stop my threads yet.

Update: I use uwsgi with 1 worker with 1 thread - should be no issues from this point

Update2: Seems like after 60 sec to finish, uwsgi sends SIGKILL to workers, so there is no way to gracefully finish from inside a worker...

from gevent.event import Event
import signal

stop_event = Event()

def cleanup(*args):
    stop_event.set()

for signal_no in (signal.SIGINT, signal.SIGTERM, signal.SIGHUP):
    signal.signal(signal_no, cleanup)

def thread_function():
    while not stop_event.wait(600):
        # Some thread work
        pass
    # Some thread clean-ups
    pass