How to detect if solara server has been aborted?

[imrenor]

Hi Guys,
So I'm trying to build a component that uses binance-connector to fetch data from a stream.
I'm using the use_effect hook to create the client after component render and close it after component destroy.

import time
from binance.websocket.spot.websocket_stream import SpotWebsocketStreamClient
import solara

def stopBinanceClient(client):
    print("Closing client", client)
    # unsubscribe 1m kline
    client.kline(symbol="xrpfdusd", interval="1m", action=SpotWebsocketStreamClient.ACTION_UNSUBSCRIBE)
    client.stop()    

def initBinanceClient(output):
    rClient = solara.use_reactive(0)
    def effectHandler():
        def onMessage(_, message):
            output.set(message)
            print("Message RECV")
        def onClose(_):
            print("Client closed", client)
        client = SpotWebsocketStreamClient(on_message=onMessage, on_close=onClose, timeout=15)
        print("Created client", client)
        rClient.set(client)
        def cleanup():
            stopBinanceClient(client)
        # subscribe to 1m kline
        client.kline(symbol="xrpfdusd", interval="1m")
        return cleanup
    solara.use_effect(effectHandler, dependencies=[])
    return rClient

@solara.component
def BinanceConnectorWs():
    output = solara.use_reactive("Waiting for response...")
    client = initBinanceClient(output=output)
    solara.Button(label="Stop Task", on_click=lambda:stopBinanceClient(client.get()))
    solara.Text(text=output.get())
    
@solara.component
def Page():
    BinanceConnectorWs()

It looks great when I run it then close the page in the browser. The component gets destroyed and the client is closed properly.
I can abort the solara server with ctrl-c in this case and it stops just fine:

(.venv) d:\python\home\git\python-crypto>solara run examples\task_test_binance_connector_api.ipynb
Solara server is starting at http://localhost:8765
Created client <binance.websocket.spot.websocket_stream.SpotWebsocketStreamClient object at 0x000002589A415400>
Message RECV
...
Message RECV
Closing client <binance.websocket.spot.websocket_stream.SpotWebsocketStreamClient object at 0x000002589A415400>
Message RECV
Message RECV
CLOSE frame received, closing websocket connection
Client closed <binance.websocket.spot.websocket_stream.SpotWebsocketStreamClient object at 0x000002589A415400>

Aborted.

(.venv) d:\python\home\git\python-crypto>

However when I abort solara server before the page is closed it gets stuck, in that case I have to find python process in task manager and kill it:

(.venv) d:\python\home\git\python-crypto>solara run examples\task_test_binance_connector_api.ipynb
Solara server is starting at http://localhost:8765
Created client <binance.websocket.spot.websocket_stream.SpotWebsocketStreamClient object at 0x0000018E99415400>
Message RECV
...
Message RECV

Aborted.
Message RECV
Message RECV
...
This goes on forever until I kill the python process...

I have been playing around with solara.use_task() before and examined this exact same behavior. If I have a function/coro that runs an infinite loop there is no way to detect that the server had been aborted. I tried checking task.is_current, task.cancelled, asyncio.get_running_loop, all of them show that the task and the loop is still fine and running. No signs of any cancellation.
I tried adding signal handler for SIGINT/SIGKILL using asyncio.get_event_loop.add_signal_handler() which is not implemented on Windows. Then I tried adding handler using signal.signal() which says signal handlers can only be added on main thread (which I thought I was on when Page() was executed, but it turns out to be not the case.

I would also expect my component to get destroyed (and use_effect cleanup to run) when server is aborted, but that's not the case either.

So, is there a way to listen for server shutdown and cleanup my open connections?

Any help is appreciated! Thanks a lot in advance!

[imrenor]

One more thing I discovered earlier. If I use solara.use_thread() to run my function threaded, then it gets terminated on server abort. So that could be a workaround to have this whole stuff wrapped in use_thread.
Still in that case when the thread is terminated there is no sign of it in the code. No exception and thread event is not set.
The only thing I'm winning with the thread version is solara server can be aborted with no issues. But my code still has no idea about the abort, therefor I can't close my open connection properly.

[maartenbreddels]

That is because we set daemon=True

solara/solara/hooks/use_thread.py

Line 120 in cbe9a7e

thread = threading.Thread(target=runner, daemon=True)

Without this set, a process does not quit until the last thread is stopped.

[imrenor]

Some further trial and error...

Try registering a function with atexit module

def exitFunc():
    print("Exit handler")

atexit.register(exitFunc)

The function only gets called when solara server can abort normally (when the component is already destroyed and binance client is closed). Does not help in my case, but at least, good to know.

Studying the threads and trying to run code on main thread
I went further in experimenting. I know I could try reading the solara docs to look for some in depth info on how the server itself works or look into the sources, but I'm a little too lazy to do that. And I'm only a hobby python developer, my main area is Java, so most of the time I'm just guessing around and using trial and error :)

So next experiment is:

print(threading.current_thread(), threading.main_thread())
result:
<WorkerThread(AnyIO worker thread, started 996)> <_MainThread(MainThread, started 8156)>

Hm, AnyIO. So I just need to find a way to run code synchronized from an AnyIO WorkerThread on the main thread. Should be easy, lol. And it turns out it actually is freakin' easy!
I've added this extra piece of code to the end of my effectHandler function (so that it only runs once, after my component is rendered):

        # Trying to run code on the main thread from the AnyIO worker thread
        print(threading.current_thread(), threading.main_thread())
        def syncedFunc():
            print(threading.current_thread(), threading.main_thread())
        from_thread.run_sync(syncedFunc)
        return cleanup

Result:

$ solara run examples/task_test_binance_connector_api.py
Solara server is starting at http://localhost:8765
Created client <binance.websocket.spot.websocket_stream.SpotWebsocketStreamClient object at 0x00000184CCEB1FD0>
<WorkerThread(AnyIO worker thread, started 3560)> <_MainThread(MainThread, started 10108)>
<_MainThread(MainThread, started 10108)> <_MainThread(MainThread, started 10108)>

Next up...
Now that I can run code on main thread, try signal.signal() again to register callbacks on SIGINT and SIGKILL and see if they work...
I'll be back...

[imrenor]

This is where I finally landed

It works now, but I'm not sure if this is the best way. I think it would be good to be able to register custom callbacks to Uvicorn's SIGINT handler. Maybe it is possible, I'll have to look into Uvicorn for that.
This solution is ok for having a single component, but not scalable for having multiple components. Worst case scenario I'll need to change the implementation to install a custom SIGINT handler that allows registering of callbacks and calls them one by one. Once all callbacks are done, it calls the original Uvicorn SIGINT handler.

Full code:

import threading
from anyio import from_thread
import signal
from binance.websocket.spot.websocket_stream import SpotWebsocketStreamClient
import solara

def stopBinanceClient(client):
    print(threading.current_thread())
    print("Closing client", client)
    # unsubscribe 1m kline
    client.kline(symbol="xrpfdusd", interval="1m", action=SpotWebsocketStreamClient.ACTION_UNSUBSCRIBE)
    client.stop()    

def initBinanceClient(output):
    rClient = solara.use_reactive(0)
    def effectHandler():
        def onMessage(_, message):
            output.set(message)
            print("Message RECV")
        def onClose(_):
            print("Client closed", client)
        client = SpotWebsocketStreamClient(on_message=onMessage, on_close=onClose, timeout=15)
        print("Created client", client)
        rClient.set(client)
        def cleanup():
            stopBinanceClient(client)
        # subscribe to 1m kline
        client.kline(symbol="xrpfdusd", interval="1m")
        # We are in an AnyIO WorkerThread
        print(threading.current_thread(), threading.main_thread())
        # Install custom SIGINT handler to close binance connection
        def syncedFunc():
            # We are on main thread yeah, let's install our custom SIGINT handler
            print(threading.current_thread(), threading.main_thread())
            # This is cool but turns out we are replacing Uvicorn's own handler that would take care of server shutdown.
            # oldHandler is: <bound method Server.handle_exit of <uvicorn.server.Server object at 0x...>>
            # If we don't store and call old handler, then server will still not shut down
            # Store server's SIGINT handler in oldHandler
            oldHandler = signal.getsignal(signal.SIGINT) 
            def signalHandler(sig, frame):
                print("Handling signal SIGINT")
                # We are on main thread here and binance client close hangs on main thread.
                # Start a new thread for the stop function to work.
                thread = threading.Thread(target = stopBinanceClient, args = (client, ))
                thread.start()
                # We don't have to wait for the thread to finish.
                # Server will wait for it to complete, that's where this whole issue started :D
                # Simply call Uvicorn's SIGINT handler to shut the server down
                if oldHandler:
                    print("Calling old handler:", oldHandler)
                    oldHandler(sig, frame)
            signal.signal(signal.SIGINT, signalHandler)
            print("Replaced old handler:", oldHandler)
        # Solara's server runs components on AnyIO WorkerThreads. Run our signal handler installer function on main thread.
        from_thread.run_sync(syncedFunc)
        return cleanup
    solara.use_effect(effectHandler, dependencies=[])
    return rClient

@solara.component
def BinanceConnectorWs():
    output = solara.use_reactive("Waiting for response...")
    client = initBinanceClient(output=output)
    solara.Button(label="Stop Task", on_click=lambda:stopBinanceClient(client.get()))
    solara.Text(text=output.get())
        
@solara.component
def Page():
    BinanceConnectorWs()

Result:

(.venv) d:\python\home\git\python-crypto>solara run examples\task_test_binance_connector_api.py
Solara server is starting at http://localhost:8765
Created client <binance.websocket.spot.websocket_stream.SpotWebsocketStreamClient object at 0x000002970E641FD0>
<WorkerThread(AnyIO worker thread, started 3000)> <_MainThread(MainThread, started 7120)>
<_MainThread(MainThread, started 7120)> <_MainThread(MainThread, started 7120)>
Replaced old handler: <bound method Server.handle_exit of <uvicorn.server.Server object at 0x000002970D5397F0>>
Message RECV
...
Message RECV
Handling signal SIGINT
<Thread(Thread-14 (stopBinanceClient), started 8012)>
Closing client <binance.websocket.spot.websocket_stream.SpotWebsocketStreamClient object at 0x000002970E641FD0>
Calling old handler: <bound method Server.handle_exit of <uvicorn.server.Server object at 0x000002970D5397F0>>

Aborted.
Message RECV
CLOSE frame received, closing websocket connection
Client closed <binance.websocket.spot.websocket_stream.SpotWebsocketStreamClient object at 0x000002970E641FD0>

(.venv) d:\python\home\git\python-crypto>

[imrenor]

And now that I know what I am looking for I found this:
Graceful shutdown of uvicorn starlette app with websockets

It all makes sense now and I like the solution in the above answer a lot more than what I came up with :)

[maartenbreddels]

The problem happens because they start a thread:

• https://github.com/binance/binance-connector-python/blob/5d76597b8b79d23e589706bf96f0c837da991ad4/binance/websocket/websocket_client.py#L29-L30
• https://github.com/binance/binance-connector-python/blob/5d76597b8b79d23e589706bf96f0c837da991ad4/binance/websocket/websocket_client.py#L59-L60
• https://github.com/binance/binance-connector-python/blob/5d76597b8b79d23e589706bf96f0c837da991ad4/binance/websocket/binance_socket_manager.py#L29-L30

If you managed to override/monkeypatch to set the daemon flag before the thread starts, you will not have this problem.

However, we can see if we can more gracefully shut down. I'll explore that a bit based on your info above.

[maartenbreddels]

Should be fixed by #930

[imrenor]

Thanks a lot for your replies!
If I get this correct, the behavior after the fix is that the kernel will be forced to stop and the threads killed. Similar to what happens with use_thread on shutdown.
Or will the components get correctly destroyed on shutdown and my use_effect cleanup will run?

[maartenbreddels]

Or will the components get correctly destroyed on shutdown and my use_effect cleanup will run?

This one. Threads cannot be stopped.
We might do a release tomorrow, or very soon, I hope we can get this in.

[imrenor]

Sounds great! Thanks a lot for such a great support!
I really enjoy exploring Solara, nice work!