compatibility issues with python 3.10, ThreadedApiManager.start_listener stuck in infinite loop

[MoeQatoum]

there is compatibility issues with python 3.10 connection closes after a few seconds for some reason. also looking at debug logs Kline stream did not connect.

EDIT: after further investigation , I found out the program is stuck in an infinite loop in start_listener ln 42 at threaded_stream.py. the exception is catching asyncio.TimeoutError but doing nothing.

this the function code:

    async def start_listener(self, socket, path: str, callback):
        async with socket as s:
            while self._socket_running[path]:
                try:
                    msg = await asyncio.wait_for(s.recv(), 3)
                except asyncio.TimeoutError:
                    ...
                    continue
                if not msg:
                    continue
                callback(msg)
        del self._socket_running[path]

It works perfectly with python 3.9, websokets 10.2, python-binance 1.0.15.

To Reproduce

#!/usr/bin/env python

from binance import ThreadedWebsocketManager

def main():
    def msg(msg):
        print(msg)

    streams = ["btcusdt@trade", "btcusdt@kline_1m"]
    wsm = ThreadedWebsocketManager()
    wsm.start()
    wsm.start_multiplex_socket(callback=msg, streams=streams)
    wsm.join()

if __name__ == "__main__":
    main()

Environment:

• Python version: 3.10
• websockets version: 10.2
• OS: Mac
• python-binance version: latest

DEBUG LOG

[DEBUG:252] websockets.client: = connection is CONNECTING
[DEBUG:111] websockets.client: > GET /stream?streams=btcusdt@trade HTTP/1.1
[DEBUG:113] websockets.client: > Host: stream.binance.com:9443
[DEBUG:113] websockets.client: > Upgrade: websocket
[DEBUG:113] websockets.client: > Connection: Upgrade
[DEBUG:113] websockets.client: > Sec-WebSocket-Key: q3TyFtRFJ6z9PwmsV0s44Q==
[DEBUG:113] websockets.client: > Sec-WebSocket-Version: 13
[DEBUG:113] websockets.client: > Sec-WebSocket-Extensions: permessage-deflate; client_max_window_bits
[DEBUG:113] websockets.client: > User-Agent: Python/3.10 websockets/10.3.dev7+g2867685
[DEBUG:144] websockets.client: < HTTP/1.1 101 Switching Protocols
[DEBUG:146] websockets.client: < Date: Fri, 01 Apr 2022 18:18:41 GMT
[DEBUG:146] websockets.client: < Connection: upgrade
[DEBUG:146] websockets.client: < Upgrade: websocket
[DEBUG:146] websockets.client: < Sec-WebSocket-Accept: PEYzEw2FvwMQN3apcGdjGD+ofFY=
[DEBUG:146] websockets.client: < Sec-WebSocket-Extensions: permessage-deflate; server_no_context_takeover; client_max_window_bits=15
[DEBUG:341] websockets.client: = connection is OPEN
[DEBUG:1150] websockets.client: < TEXT '{"stream":"btcusdt@trade","data":{"e":"trade","...53,"m":false,"M":true}}' [201 bytes]
[DEBUG:1150] websockets.client: < TEXT '{"stream":"btcusdt@trade","data":{"e":"trade","...01,"m":false,"M":true}}' [201 bytes]
[DEBUG:1150] websockets.client: < TEXT '{"stream":"btcusdt@trade","data":{"e":"trade","...730,"m":true,"M":true}}' [200 bytes]
[DEBUG:1150] websockets.client: < TEXT '{"stream":"btcusdt@trade","data":{"e":"trade","...822,"m":true,"M":true}}' [200 bytes]
[DEBUG:1150] websockets.client: < TEXT '{"stream":"btcusdt@trade","data":{"e":"trade","...03,"m":false,"M":true}}' [201 bytes]
[DEBUG:1150] websockets.client: < TEXT '{"stream":"btcusdt@trade","data":{"e":"trade","...43,"m":false,"M":true}}' [201 bytes]
[DEBUG:1150] websockets.client: < TEXT '{"stream":"btcusdt@trade","data":{"e":"trade","...197,"m":true,"M":true}}' [200 bytes]
[DEBUG:1150] websockets.client: < TEXT '{"stream":"btcusdt@trade","data":{"e":"trade","...264,"m":true,"M":true}}' [200 bytes]
[DEBUG:1150] websockets.client: < TEXT '{"stream":"btcusdt@trade","data":{"e":"trade","...264,"m":true,"M":true}}' [200 bytes]
[DEBUG:1150] websockets.client: < TEXT '{"stream":"btcusdt@trade","data":{"e":"trade","...264,"m":true,"M":true}}' [200 bytes]
[DEBUG:1150] websockets.client: < TEXT '{"stream":"btcusdt@trade","data":{"e":"trade","...267,"m":true,"M":true}}' [200 bytes]
[DEBUG:1150] websockets.client: < TEXT '{"stream":"btcusdt@trade","data":{"e":"trade","...267,"m":true,"M":true}}' [200 bytes]
[DEBUG:1150] websockets.client: < TEXT '{"stream":"btcusdt@trade","data":{"e":"trade","...267,"m":true,"M":true}}' [200 bytes]
[DEBUG:1150] websockets.client: < TEXT '{"stream":"btcusdt@trade","data":{"e":"trade","...267,"m":true,"M":true}}' [200 bytes]
[DEBUG:1150] websockets.client: < TEXT '{"stream":"btcusdt@trade","data":{"e":"trade","...311,"m":true,"M":true}}' [200 bytes]
[DEBUG:1150] websockets.client: < TEXT '{"stream":"btcusdt@trade","data":{"e":"trade","...321,"m":true,"M":true}}' [200 bytes]
[DEBUG:1150] websockets.client: < TEXT '{"stream":"btcusdt@trade","data":{"e":"trade","...321,"m":true,"M":true}}' [200 bytes]
[DEBUG:1150] websockets.client: < TEXT '{"stream":"btcusdt@trade","data":{"e":"trade","...336,"m":true,"M":true}}' [200 bytes]
[DEBUG:1243] websockets.client: % sending keepalive ping
[DEBUG:1156] websockets.client: > PING bf d7 f2 2b [binary, 4 bytes]
[DEBUG:1256] websockets.client: ! timed out waiting for keepalive pong
[DEBUG:1401] websockets.client: ! failing connection with code 1011
[DEBUG:1425] websockets.client: = connection is CLOSING
[DEBUG:1156] websockets.client: > CLOSE 1011 (unexpected error) keepalive ping timeout [24 bytes]
[DEBUG:1302] websockets.client: ! timed out waiting for TCP close
[DEBUG:1341] websockets.client: x closing TCP connection
[DEBUG:1486] websockets.client: = connection is CLOSED

[MoeQatoum]

working with this library is becoming exhausting and frustrating, its not well maintained you have to go through the library and debug its issues, this is distracting and a wast of time, I recommend migrating to unicorn-binance-websocket-api , its well maintained and very easy to use. or use binance-ru and port it to your python code. which is easier than looking at and debugging 1000s of code lines! I will keep this issue open for others who might be willing to fix this issue ...

[halfelf]

I can confirm this bug. Same problem with #1174 . Have found a quick fix there, but no promise since I don't use threaded socket.

[MoeQatoum]

hi @halfelf, thanks for your reply, your suggestion actually fixed the issue.

TO FIX THIS ISSUE: at threaded_stream.py modify self._loop as below

class ThreadedApiManager(threading.Thread):
    def __init__(
        self,
        api_key: Optional[str] = None,
        api_secret: Optional[str] = None,
        requests_params: Optional[Dict[str, str]] = None,
        tld: str = "com",
        testnet: bool = False,
    ):
        """Initialise the BinanceSocketManager"""
        super().__init__()
        #self._loop: asyncio.AbstractEventLoop = asyncio.new_event_loop()
        self._loop: asyncio.AbstractEventLoop = asyncio.get_event_loop()
        self._client: Optional[AsyncClient] = None
        self._running: bool = True
        self._socket_running: Dict[str, bool] = {}
        self._client_params = {
            "api_key": api_key,
            "api_secret": api_secret,
            "requests_params": requests_params,
            "tld": tld,
            "testnet": testnet,
        }

[Kncokoff]

tried @halfelf solution but when running two sockets got another problem

self.tdcm = ThreadedDepthCacheManager()
        self.twsm= ThreadedWebsocketManager()
        self.twsm.start()
        self.tdcm.start()
RuntimeError: This event loop is already running

any work around?

[dashko]

I can confirm the fix from @MoeQatoum here worked for me.