Cryptofeed multiprocessing not pushing to queue

fvim 0
Viewed 37

I am attempting to use cryptofeed to get the feed of multiple tokens on multiple processes. Also to retreive these feeds and push them to a database using multiple worker processes. My attempted code below works for the feed processes but not the queue.get() in the worker process(es).


queue = multiprocessing.Queue()

async def ticker_function(data, time):
    # Send data to queue
    queue.put(data, block=False)

def new_feed(coin):
    f = FeedHandler()
    f.add_feed(cryptofeed.exchanges.Binance(
            channels=[TICKER], 
            symbols=[f"{coin}-USDT"],
            callbacks={TICKER: ticker_function}
        ))
    f.run()

def worker(queue):
    print("Starting worker")
    while True:
        if not queue.empty():
            data = queue.get(block=False) # THIS IS WHERE IT SAYS QUEUE EMPTY
            json_payload = {
            "measurement": data.symbol,
            "time": int(time.time()*1000000000),
            "fields": {
                'bid': float(data.bid),
                'ask': float(data.ask),
                }
            }
            # Push to influxdb
            write_api.write(bucket='test_bucket', org='mydb', record=json_payload)
        else:
            time.sleep(0.01)

if __name__ == '__main__':
    


    data_process = multiprocessing.Process(target=worker, args=(queue,))
    data_process.start()

    feed_process = multiprocessing.Process(target=new_feed, args=('BTC',))
    feed_process.start()


    # To stop the main process from ending, adding a join (won't ever trigger)
    feed_process.join()

Exception thrown (which I except and pass):

Exception has occurred: Empty
exception: no description
  File "/Users/user/sandbox/live-pnl/main.py", line 90, in worker
    data = queue.get(block=False)
  File "<string>", line 1, in <module>

What am I doing wrong?

EDIT: Trace for another error after changing code

Traceback (most recent call last):
  File "<string>", line 1, in <module>
  File "/opt/homebrew/Cellar/python@3.10/3.10.6_2/Frameworks/Python.framework/Versions/3.10/lib/python3.10/multiprocessing/spawn.py", line 116, in spawn_main
    exitcode = _main(fd, parent_sentinel)
  File "/opt/homebrew/Cellar/python@3.10/3.10.6_2/Frameworks/Python.framework/Versions/3.10/lib/python3.10/multiprocessing/spawn.py", line 126, in _main
    self = reduction.pickle.load(from_parent)
  File "/opt/homebrew/Cellar/python@3.10/3.10.6_2/Frameworks/Python.framework/Versions/3.10/lib/python3.10/multiprocessing/synchronize.py", line 110, in __setstate__
    self._semlock = _multiprocessing.SemLock._rebuild(*state)
FileNotFoundError: [Errno 2] No such file or directory
python-multiprocessing multiprocessing python
1 Answers

Following the docs, if the queue is empty, and you you request an item with queue.get(block=True), the exception queue.Empty will be raised.

Instead, you probably want block=False, to wait for a value till it is available in the queue.

Or, first check if the queue is empty, and then request an item:

import time
def worker(queue):
    print("Starting worker")
    while True:
        if not queue.empty()
            data = queue.get()
            json_payload = {
            "measurement": data.symbol,
            "time": int(time.time()*1000000000),
            "fields": {
                'bid': float(data.bid),
                'ask': float(data.ask),
                }
            }
            # Push to influxdb
            write_api.write(bucket='test_bucket', org='mydb', record=json_payload)
        else:
            time.sleep(0.01) # sanity delay
rinkert0
Related