How to properly delete a stream completely from the manager?

[Scoooooba]

Hi Guys,

given the following simple code:

from unicorn_binance_websocket_api.manager import BinanceWebSocketApiManager
websocket_manager = BinanceWebSocketApiManager()
stream_id = websocket_manager.create_stream('depth', 'BNBBUSD')

When it is executed, entries for this particular stream_id are made at least in the following dictionaries (maybe in more, these were the ones I've quickly been able to find):

websocket_manager.specific_process_stream_data
websocket_manager.stream_threading_lock
websocket_manager.stream_list
websocket_manager.event_loops
websocket_manager.socket_is_ready
websocket_manager.stream_threads
websocket_manager.stream_threading_lock
websocket_manager.websocket_list

After stopping the stream with the websocket_manager.stop_stream(stream_id) command, the entries for this stream_id in the different dictionaries are still existing. There is a function websocket_manager.delete_stream_from_stream_list(stream_id) to remove the stream_id entry from the websocket_manager.stream_list dicionary, but as far as I know there are no functions to delete the stream_id from all the other dictionaries.

Wouldn't it make sense to implement (a) function(s) to remove a stream_id completely from the websocket_manager, so the manager can be properly cleaned up? Otherwise these dictionaries will grow a lot, if several streams will be created and stopped over some time.

Best regards
Sebastian

[oliver-zehentleitner]

I think a parameter like ubwa.stop_stream(stream_id, clean_up=True) could make sense...

Activating this parameter could call a new function like ubwa.remove_all_entries_of_stream_id(stream_id) which removes everything related to that stream_id, but only if the stream is not running anymore.

[Scoooooba]

Sounds good!

I tried to execute the websocket_manager.delete_stream_from_stream_list(stream_id) command right after the websocket_manager.stop_stream(stream_id) command. This resulted in an exception - I guess because the stream was still running. So maybe a function like ubwa.remove_all_entries_of_stream_id(stream_id) should be executed in a new thread (or in one of the existing ubwa threads that are always running in background) cleaning up as soon as the stream has stopped.

Maybe for this purpose there could be a new list like ubwa.streams_to_clean that is checked regularly by an always running thread.

[oliver-zehentleitner]

Combine it with ubwa.wait_till_stream_has_stopped(stream_id) - does it avoid the exception you got?

[oliver-zehentleitner]

there is already a thread which could do that job:

unicorn-binance-websocket-api/unicorn_binance_websocket_api/manager.py

Line 557 in 0dfeed6

def _frequent_checks(self):

[Scoooooba]

Combine it with ubwa.wait_till_stream_has_stopped(stream_id) - does it avoid the exception you got?

No. This code ...

from unicorn_binance_websocket_api.manager import BinanceWebSocketApiManager

import time

def callback_func(stream_data):
    print(stream_data)


ubwa = BinanceWebSocketApiManager()

stream_id = ubwa.create_stream('depth20@1000ms', 'BNBBUSD', output='dict', process_stream_data=callback_func)


time.sleep(5)

stopped = ubwa.stop_stream(stream_id)
ubwa.wait_till_stream_has_stopped(stream_id)
deleted = ubwa.delete_stream_from_stream_list(stream_id)

... results in the following (repeated) exception:

Exception in thread _restart_stream_thread: stream_id=d78a8feed045-6eb0-b798-619d-47e34970, time=1654809827.4938142: Traceback (most recent call last): File "C:\Users\Sebi\AppData\Local\Programs\Python\Python39\lib\threading.py", line 973, in _bootstrap_inner self.run() File "C:\Users\Sebi\AppData\Local\Programs\Python\Python39\lib\threading.py", line 910, in run self._target(*self._args, **self._kwargs) File "A:\CryptoTrading\Tradetool\.Tradetool_1_virtualenv\lib\site-packages\unicorn_binance_websocket_api\manager.py", line 814, in _restart_stream_thread str(self.stream_list[stream_id]['channels']) + KeyError: 'd78a8feed045-6eb0-b798-619d-47e34970'

[oliver-zehentleitner]

Fixed in 7b0dbea

[oliver-zehentleitner]

It seems wait_till_stream_has_stopped() is buggy or at least not working the way we would expect it.

That way it works: https://github.com/LUCIT-Systems-and-Development/unicorn-binance-websocket-api/blob/master/example_remove_streams_from_manager.py

So we have to improve wait_till_stream_has_stopped() (I will do that this weekend) In the meantime you can proceed by working with stream signals.

[Scoooooba]

Perfect, thank you!

[Scoooooba]

Hey Oliver,
have you been able to solve the issue with wait_till_stream_has_stopped()?

[oliver-zehentleitner]

Its already solved! Did not mention here, sorry

[oliver-zehentleitner]

No, i will update the full suite asap...

[oliver-zehentleitner]

In the next release (>1.42.0) this will work:

from unicorn_binance_websocket_api.manager import BinanceWebSocketApiManager
import logging
import os
import time


logging.getLogger("unicorn_binance_websocket_api")
logging.basicConfig(level=logging.DEBUG,
                    filename=os.path.basename(__file__) + '.log',
                    format="{asctime} [{levelname:8}] {process} {thread} {module}: {message}",
                    style="{")


def callback_func(stream_data):
    print(stream_data)


ubwa = BinanceWebSocketApiManager(debug=True)
ubwa.print_summary()
stream_id = ubwa.create_stream('depth20@1000ms', 'BNBBUSD', output='dict', process_stream_data=callback_func)
time.sleep(5)
stopped = ubwa.stop_stream(stream_id)
ubwa.wait_till_stream_has_stopped(stream_id)
ubwa.print_summary()
deleted = ubwa.delete_stream_from_stream_list(stream_id)
ubwa.stop_manager_with_all_streams()

[oliver-zehentleitner]

Disccusion about the implementation of deleting old streams: #307