multiprocessing error : EOFError: Ran out of input

[Zourvan]

Hi
I have a program written to cash full candles to my database I want use multiprocessing to speed my process isolate from other, but when I ran my code show me this error:

OS is Windows 7
CPU : core i7

Traceback (most recent call last):
File "d:\MD\Project\Xtrader\Trace\xTM\test.py", line 178, in
daq(exchangeName = exchangeName,Channels = Channels , wsBundelSize = wsBundelSize)
File "d:\MD\Project\Xtrader\Trace\xTM\test.py", line 148, in daq
p1.start()
File "C:\Program Files\Python\Python310\lib\multiprocessing\process.py", line 121, in start
self._popen = self._Popen(self)
File "C:\Program Files\Python\Python310\lib\multiprocessing\context.py", line 224, in _Popen
return _default_context.get_context().Process._Popen(process_obj)
File "C:\Program Files\Python\Python310\lib\multiprocessing\context.py", line 327, in _Popen
return Popen(process_obj)
File "C:\Program Files\Python\Python310\lib\multiprocessing\popen_spawn_win32.py", line 93, in init
reduction.dump(process_obj, to_child)
File "C:\Program Files\Python\Python310\lib\multiprocessing\reduction.py", line 60, in dump
ForkingPickler(file, protocol).dump(obj)
TypeError: cannot pickle '_thread.lock' object
Traceback (most recent call last):
File "", line 1, in
File "C:\Program Files\Python\Python310\lib\multiprocessing\spawn.py", line 116, in spawn_main
exitcode = _main(fd, parent_sentinel)
File "C:\Program Files\Python\Python310\lib\multiprocessing\spawn.py", line 126, in _main
self = reduction.pickle.load(from_parent)
EOFError: Ran out of input

and here is my code:

class exDAQ(mngdb):
    def __init__(self,exchangeName,name="DAQ") -> None:
        super().__init__(exchangeName)
        self.exchangeName = exchangeName
        self.name = name
        self.db = mngdb(self.exchangeName)
        self.ubwa = BinanceWebSocketApiManager(exchange="binance.com", output_default="dict",high_performance=True)
        self.db.log("info_"+self.name,"class streated")

    def config(self,markets,channel):
        self.ubwa.create_stream(channel, markets, stream_label="dict", output="dict")            
        self.db.log("info_"+self.name,"Config Done!!")
        

    def run(self):
        self.db.log("info_"+self.name,"Ran!!")
        while True:
            print(self.ubwa.stream_is_crashing())
            if self.ubwa.is_manager_stopping():
                exit(0)
            osdfsb= self.ubwa.pop_stream_data_from_stream_buffer()
            if osdfsb is False:
                time.sleep(0.01)
            else:
                if osdfsb is not None:
                    try :
                        if osdfsb['data']['k']['x']:
                            candle = osdfsb['data']['k']
                            if candle['x']:
                                self.db.insert_one("spot_" + osdfsb['data']['s'] + "_" + candle['i']
                                , dict(ThreadName = self.name , timestamp = candle['t'] , DateTime = datetime.utcfromtimestamp(int(candle['t'] / 1000)) 
                                , InsertTime = datetime.utcnow(), open=candle['o'], close=candle['c'], low=candle['l']
                                , high=candle['h'], volume=candle['v']))
                            # print(f"{osdfsb}")
                    except:
                        pass

def daq(exchangeName = 'binance',Channels =['kline_1m', 'kline_3m' , 'kline_5m' , 'kline_15m'] , wsBundelSize = 100):
    print("*"*3, f" Loop is start in {datetime.utcnow()}", "*"*3)
    db = mngdb(exchangeName)
    loading = load_address(exchangeName)
    value = loading.load_markets()

    markets = []
    symbols = []
    for i in range(0,len(value)):
        index = value[i].find('/')
        symbol = value[i][0:index] + value[i][index+1:]
        symbols.append(dict(id=i, symbol= symbol.lower(),raw_symbol=value[i],insert_time=datetime.utcnow()))
        markets.append(symbol.lower())
    markets.sort()

    db.insert_many(db.symbolNAmeDBT, symbols)
    section = int(len(symbols) / wsBundelSize)

    markets_section_list = []
    for i in range(0, section):
        markets_section_list.append(markets[i*wsBundelSize:(i+1)*wsBundelSize])
    markets_section_list.append(markets[section*wsBundelSize:len(symbols)])
    print(f"-> Number of streams are {len(markets_section_list)*len(Channels)}")
    
    db.log('info', f'creating a list of section for websocket thread pool at: {datetime.utcnow()}')
   
    exDAQ_= list()
    for i in range(0,len(markets_section_list)*len(Channels)):
        exDAQ_.append(exDAQ(exchangeName,f"DAQ_{i}"))
    
    for i in range(0,len(markets_section_list)):
        for j in range(0,len(Channels)):
            print(f'Stream  #{i*(j+1):3d} is configed!!', end= "  ")
            exDAQ_[i*(j+1)].config(markets_section_list[i],Channels[j])
   # just for test
    p1 = Process(target=exDAQ_[0].run)
    p1.start()


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

    exchangeName = 'binance_ubw'
    Channels =['kline_1m', 'kline_3m' , 'kline_5m' , 'kline_15m']
    wsBundelSize = 500
    logger.addHandler(MongoHandler(host='localhost',database_name=exchangeName,collection="_log_"+exchangeName))
    daq(exchangeName = exchangeName,Channels = Channels , wsBundelSize = wsBundelSize)

[oliver-zehentleitner]

hi!

self.ubwa.stream_is_crashing() needs a stream_id like self.ubwa.stream_is_crashing(stream_id).

I have quickly read your code but dont get behind it, but here is an issue about multiprocessing in general: #126