Reconnection with disptacher, issue with close about websocket-client

[bubbleboy14]

@devanshubisht interesting. As a test, could you put this line near the top of your code:

rel.initialize(["poll", "select"])

On FreeBSD, kqueue seems to work fine. However, I guess your system might be a little different. LMK how that test goes!

[bubbleboy14]

@devanshubisht rel can be used instead of threads to run multiple WebSocketApp instances.

I wasn't able to reproduce your issue - calling ws.close() inside the on_message() callback works fine on my end.

If you provide a concise example that demonstrates your issue, I can take a look.

[devanshubisht]

` def trade_monitor(self):
print("Delta Hedge Bot : Running <> ***")
recon_attempt = [False]
random = [0]
def on_message(ws, message):
ws.close()

    def on_error(ws, error):
        print("error")
        print(error)
        ws.close()
            
    def on_close(ws, close_status_code, close_msg):
        print("### closed ###")
        
        # self.telegram_alert("Option's websocket Connection Closed")

    def on_open(ws):
        # Send the subscription message when the connection is opened

        message = {
            "type": "subscribe",
            "channels": ["user_trade"],
            "pairs": self.instrument_list,  # param is pairs instead of currencies
            "categories": ["option"],
            "interval": "raw",
            "token": self.token,
        }
        ws.send(json.dumps(message))  # Send the subscription message

    websocket.enableTrace(True)
    socket = self.ws_host_url
    ws = websocket.WebSocketApp(socket,
                                on_open=on_open,  # Specify the on_open callback
                                on_message=on_message,
                                on_error=on_error, 
                                on_close = on_close,
                                )
    ws.run_forever(dispatcher=rel,reconnect=5 )
    rel.dispatch()`

This is the code adn this is the error.

I have been working with Binance API and BIT API. I have also realised that when I attempt to reconnect using websocket-client, it attempts to reconnect with the same token which was established before. This prevents reconnections from happening again. Hence, I need to teardown and create a different object.

This is how I am actually doing it now:
while True: # BIT.telegram_alert("Option's websocket Reconnecting...") BIT = TradeMonitor(instrument_list=instrument_list, api_key=api_key, secret_key=secret_key, ws_host_url=ws_host_url, test_mode=test_mode) BIT.trade_monitor()
and my ws.run_forever() has no recon inside.

Have you ever encountered or faced such an issue before?

[waza-ari]

Thanks so much! I had an issue with rel based reconnects on MacOS, adding this code worked fine. For reference, my code was:

import rel
import websocket

class WebsocketClient:
    def __init__(self):
        scheme = "wss" if settings.BACKEND_SECURE else "ws"
        self.ws = websocket.WebSocketApp(
            f"{scheme}://{settings.BACKEND_BASE_URL}/ws/storage", on_message=self.on_message, on_close=self.on_close
        )
        self.ws.run_forever(dispatcher=rel, reconnect=5)
        rel.signal(2, rel.abort)
        rel.dispatch()

    @staticmethod
    def on_message(wsapp: websocket.WebSocketApp, message: str):
        # Do something with message
        pass

    def on_close(self, ws: websocket.WebSocketApp, close_status_code: int, close_msg: str):
        rel.abort()
        self.ws.run_forever(dispatcher=rel, reconnect=5)
        rel.signal(2, rel.abort)
        rel.dispatch()


if __name__ == "__main__":
    WebsocketClient()

When the connection to the server was lost (e.g. because the server was shutting down) and it got re-established, the following Exception was thrown:

Websocket connected
I/O operation on closed kqueue object - reconnect
Calling custom dispatcher reconnect [19 frames in stack]
error from callback <bound method WebsocketClient.on_close of <__main__.WebsocketClient object at 0x100723110>>: I/O operation on closed kqueue object
Traceback (most recent call last):
  File "/project/websocket_listener.py", line 46, in <module>
    WebsocketClient()
  File "/project/websocket_listener.py", line 19, in __init__
    rel.dispatch()
  File "/venv/lib/python3.12/site-packages/rel/rel.py", line 234, in dispatch
    registrar.dispatch()
  File "/venv/lib/python3.12/site-packages/rel/registrar.py", line 129, in dispatch
    if not self.loop():
           ^^^^^^^^^^^
  File "/venv/lib/python3.12/site-packages/rel/registrar.py", line 138, in loop
    e = self.check_events()
        ^^^^^^^^^^^^^^^^^^^
  File "/venv/lib/python3.12/site-packages/rel/registrar.py", line 226, in check_events
    elist = self.kq.control(None, 1000, LISTEN_KQUEUE)
            ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
ValueError: I/O operation on closed kqueue object

Adding the proposed line rel.initialize(["poll", "select"]) directly below the import statement fixed the issue and reconnect works as expected. Can you give some insight on what is happening behind the scenes? I was unable to find a lot of references to that error message

[bubbleboy14]

Thanks for the bug report @waza-ari , much appreciated. The problem appears to be with rel's usage of kqueue. Would you be willing to help me debug this issue?

Here's the simplest possible test case for reconnect. For the server, I'm using dez, which can be installed thusly:

pip install dez

Then to run the server I type:

dez_test wsecho

Here's (roughly) the simplest possible client ("client.py") for this:

import rel, websocket

domain = input("domain? [default: localhost] ") or "localhost"

websocket.enableTrace(True)
websocket.WebSocketApp("ws://%s:8888"%(domain,)).run_forever(dispatcher=rel, reconnect=1)
rel.dispatch()

Which I run in a different terminal with:

python client.py

And then back in the server ("dez_test wsecho") terminal I repeatedly stop (with Ctr-C) and restart the server, and watch the logs to confirm that reconnects are working.

I tested this in a FreeBSD virtual machine (because Linux doesn't have kqueue but FreeBSD does), and it seems to work fine. Does it work fine for you? If not, I wonder whether the OSX implementation of kqueue is different from FreeBSD's in some way. But if it does work for you, that would suggest that some other factor (reads/writes?) may be at play.

Anyway, thanks for debugging this with me, LMK how it goes!

[waza-ari]

Good morning,

thanks for your answer and your detailed explanation! Happy to help. Unfortunately, the first issue is already around the test server, for some reason it doesn't work:

> dez_test wsecho                        
loading: dez.samples.wsecho
error: invalid module specified
type "dez_test -h" for help

I did inspect the wheel and it does contain the wsecho sample file. Otherwise I've been unable to find any documentation, so not sure what else to do. Instead I used my own server (Uvicorn based) for testing, and simply killed the process instead of shutting it down to avoid graceful shutdown behaviour.

domain? [default: localhost] 
--- request header ---
GET /ws/storage HTTP/1.1
Upgrade: websocket
Host: localhost:8000
Origin: http://localhost:8000
Sec-WebSocket-Key: aN+lFqRipNmWjfvHN5x3RQ==
Sec-WebSocket-Version: 13
Connection: Upgrade


-----------------------
--- response header ---
HTTP/1.1 101 Switching Protocols
Upgrade: websocket
Connection: Upgrade
Sec-WebSocket-Accept: kqln5At5glzDNke68RLa2KW1E6A=
date: Fri, 17 May 2024 05:11:36 GMT
server: uvicorn
-----------------------
Websocket connected
++Rcv raw: b'\x89\x04\xce\x86\xb5\xe9'
++Rcv decoded: fin=1 opcode=9 data=b'\xce\x86\xb5\xe9'
[... a few messages removed before I figured out there was a second Uvicorn thread spawned... [
Connection to remote host was lost. - reconnect
Calling custom dispatcher reconnect [9 frames in stack]
[Errno 61] Connection refused - reconnect
Calling custom dispatcher reconnect [8 frames in stack]
[Errno 61] Connection refused - reconnect
Calling custom dispatcher reconnect [8 frames in stack]
[Errno 61] Connection refused - reconnect
Calling custom dispatcher reconnect [8 frames in stack]
[Errno 61] Connection refused - reconnect
Calling custom dispatcher reconnect [8 frames in stack]
[Errno 61] Connection refused - reconnect
Calling custom dispatcher reconnect [8 frames in stack]
[Errno 61] Connection refused - reconnect
Calling custom dispatcher reconnect [8 frames in stack]
[Errno 61] Connection refused - reconnect
Calling custom dispatcher reconnect [8 frames in stack]

[... server restarted now ... ]

--- request header ---
GET /ws/storage HTTP/1.1
Upgrade: websocket
Host: localhost:8000
Origin: http://localhost:8000
Sec-WebSocket-Key: KBjyyTeBYzJwjVuzx7lonw==
Sec-WebSocket-Version: 13
Connection: Upgrade


-----------------------
--- response header ---
HTTP/1.1 101 Switching Protocols
Upgrade: websocket
Connection: Upgrade
Sec-WebSocket-Accept: 4rM0+2bzh4ecGXQ3nnPpmR/6g5U=
date: Fri, 17 May 2024 05:15:15 GMT
server: uvicorn
-----------------------
Websocket connected
++Rcv raw: b'\x89\x04\x91\x0e9\x10'
++Rcv decoded: fin=1 opcode=9 data=b'\x91\x0e9\x10'
++Sent raw: b'\x8a\x84\xf4K\xc5\xcfeE\xfc\xdf'
++Sent decoded: fin=1 opcode=10 data=b'\x91\x0e9\x10'

[ ... try again ... ]

Connection to remote host was lost. - reconnect
Calling custom dispatcher reconnect [9 frames in stack]
[Errno 61] Connection refused - reconnect
Calling custom dispatcher reconnect [8 frames in stack]
[Errno 61] Connection refused - reconnect
Calling custom dispatcher reconnect [8 frames in stack]
[Errno 61] Connection refused - reconnect
Calling custom dispatcher reconnect [8 frames in stack]
[Errno 61] Connection refused - reconnect
Calling custom dispatcher reconnect [8 frames in stack]
[Errno 61] Connection refused - reconnect
Calling custom dispatcher reconnect [8 frames in stack]
[Errno 61] Connection refused - reconnect
Calling custom dispatcher reconnect [8 frames in stack]
--- request header ---
GET /ws/storage HTTP/1.1
Upgrade: websocket
Host: localhost:8000
Origin: http://localhost:8000
Sec-WebSocket-Key: CiLif5gmRNekEc4mdn9D2Q==
Sec-WebSocket-Version: 13
Connection: Upgrade


-----------------------
--- response header ---
HTTP/1.1 101 Switching Protocols
Upgrade: websocket
Connection: Upgrade
Sec-WebSocket-Accept: E8RvTOqvLC6V8Pg82+kVg1f76jM=
date: Fri, 17 May 2024 05:16:03 GMT
server: uvicorn
-----------------------
Websocket connected
++Rcv raw: b'\x88\x02\x03\xf4'
++Rcv decoded: fin=1 opcode=8 data=b'\x03\xf4'
++Sent raw: b'\x88\x82;B\x8c\x148\xaa'
++Sent decoded: fin=1 opcode=8 data=b'\x03\xe8'

With the basic test it does seem to work fine. The main difference I can come up with so far is that the error occurred when reconnecting in the on_close() callback, as I'd like the client to attempt reconnecting even if the server properly shut down the connection, for example due to an update or when developing locally.

[bubbleboy14]

Thanks for testing @waza-ari! Hm, that's weird about dez_test wsecho, thanks for the heads up.

If the issue is happening in the on_close() callback, I'm guessing it has to do with the three rel calls (abort/signal/dispatch) in that function. This pattern smells off to me - I'd recommend removing all three calls.

Does the kqueue issue still happen without those three rel calls?

Regarding the on_close() callback, generally, I've been working on a branch (PR: #983) that improves a few things including reconnect behavior.

Previously, I was also manually reconnecting in an on_close() callback. However, with the improvements in that branch, my on_close() callback never gets called - it just reconnects instead.

I'm wondering - does this branch improve reconnect behavior (or anything else) with your use case? Or degrade performance in any way?

Thanks for testing this stuff with me!