I am having issues with multiprocessing.Manager().Queue()

In short: Some of the processes get clogged and stay clogged, whilst other processes have consistent run time, and are always consistent.

Below is a snippet of the relevant code, but providing full code is not possible as I am connecting to an API that requires usernames and passwords etc. My code basically creates 16 websocket connections for 16 different stocks for 16 live price feeds and plots them all on 1 figure with 16 subplots using matplotlib.animation.FuncAnimation(). Some of the stocks will never be delayed and will always plot when there is a websocket reponse, whilst some of the other stocks will always be delayed from when it got the websocket reponse and plotting. That is, it will ping the server but the process will get stuck in the queue.put() and queue.get(). (IT IS NOT AN ISSUE WITH PLOTTING).

I know it is a queue issue as I have printed the websocket response right after it gets the response from the API (before the put()) and printed the response after it’s through the pipe (after the get()) and it’s just a response that occurred a few minutes after it was called (response from 8:21am but it’s currently 8:52am). It is not an issue with the API because I can use asyncio and threading to print the API response and that works fine. And it is not a plotting issue as I have printed the response before any plotting and removed plotting functionality and the same issue persists.

If you look at the figure, you will see that some of the plots are behind others with their by looking at their price and the last update time. E.g. HK50 is behind US500, even though there will be an updated price at the same time as US500.

I don’t expect a clear solution, but I am new to multiprocessing so maybe there is a really simple mistake I have made.

The synopsis of my code is:

In __name__ == "__main__":

1. The rest api is called for basic information and create dictionaries to store the data.
2. manager = multiprocessing.Manager() is created and its Queue() method is used in a list by queues = [manager.Queue() for _ in range(len(epics))] (length is # of stocks).
3. The processes are spawned with a for loop multiprocessing.Process(target = get_live_prices, args=(queues[idx],...), for each stock and appended to a list called processes within the loop.
4. Loop on the processes lists and process.start() each process.
5. Plot specifications are created with a list of lines from axes, which is from plt.subplots().
6. Then animation plot is created which takes the queue list with anim = FuncAnimation(fig, update_plot, fargs=(queues,...)

In get_live_prices():

1. Take the queue’s as in input and the connection is created with try: inside a while loop so if the connection breaks, the error response will keep the loop alive and create a new connection.
2. Create a threading.Thread() proccess which calls the function keep_connection_alive since the connection will close (long story as to why, but the only way the connection won’t close for some of the ticker’s is that I must continuously ping it).
3. call the function getting_websocket_responses() which is a function that will be looping and getting responses from the connection forever

In getting_websocket_responses():

1. Loop forever collecting price data and timestamp from the websocket response until there is a bad response or closed connection. The latter which will trigger the except in get_live_prices(), which will exit the function.
2. The price and timestamp and stock name is place in the pipe with queue.put() (the queue.get() is in update_plot() function.

In update_plot():

1. The update_plot() takes the queues list as an input and this function is triggered by the `FuncAnimation()’
2. for loop on list of queues, subplots (axs) and lines.
3. Immediately check if the queue is not empty. If the queue is empty, that queue (therefore that stock) is skipped.
4. If the queue isn’t empty, get the API response from the pipe with queue.() and draw the updated plot with line.set_data().
5. For the list, set the flag did_we_update[i] = True
   (After the above for loop)
6. After the for loop, loop on did_we_update and update the last response times (what we see on the plot).
7. Return the lines and timers whhich

import http.client
import json
from websocket import create_connection
from websocket._exceptions import WebSocketConnectionClosedException
from datetime import datetime, timedelta,timezone
import credentials 
import multiprocessing 
from matplotlib.animation import FuncAnimation
import matplotlib.pyplot as plt
import numpy as np
import pytz
import time
from matplotlib.ticker import MaxNLocator
import matplotlib.dates as mdates
import threading


def update_plot(frame, queues, lines, axs, timer_texts, prev_time):
        
    did_we_update = [False] * len(queues)
    
    """
    The for-loop below loops for each plot. At each loop, the multiprocess queue is checked using the queue.empty() method to see if there was a tick from the API.
    If there is not a tick, the subplot will not be updated.
    
    """
    for i, (queue, line, ax)in enumerate(zip(queues, lines, axs)):
        
        if not queue.empty():
            
            new_data, epic, data_dict, pair = queue.get()
            old_xdata, old_ydata = line.get_data()
            
            if new_data[1] == None:
                continue
            else:
                new_price = new_data[1]
                perc_diff = 100 * (new_data[1] - data_dict[epic]['price_4PM_NY']) /  data_dict[epic]['price_4PM_NY']
            
          
            new_xdata = np.append(old_xdata,new_data[0])
            new_ydata = np.append(old_ydata,new_price)
         
            ax.set_xlim(new_xdata[0], new_data[0] + timedelta(minutes=30))
            city_tz= pytz.timezone('some_timezone')

            ax.xaxis.set_major_formatter(mdates.DateFormatter('%d %b %H:%M', tz=city_tz))
            ax.tick_params(axis='x', rotation=15, labelsize=8) 
            ax.xaxis.set_major_locator(MaxNLocator(6))
            
            if min(new_ydata) <= 0:
                ymin = min(new_ydata)*( 1 + (25/10000))

            else:
                ymin = min(new_ydata)*( 1 - (25/10000))
            
            if max(new_ydata) <= 0:
                ymax = max(new_ydata)*( 1 - (25/10000))
            
            else:
                ymax = max(new_ydata)*( 1 + (25/10000))
                
            ax.set_ylim(ymin, ymax)
            
            current_time = time.time() 
            time_diff = round(current_time - prev_time[i],3)
            prev_time[i] = current_time
            timer_texts[i].set_text(f'Last update: {str(time_diff)} (s)')
            
            line.set_data(new_xdata, new_ydata)
            did_we_update[i] = True
            label_offset = (ymax - ymin) * 0.02  # Adjust the offset as needed
            

            if hasattr(ax, 'perc_label'):
                ax.perc_label.remove()
            

            ax.perc_label = ax.text(new_xdata[-1], new_ydata[-1] + label_offset, f'{perc_diff:.2f}%', fontsize=12, verticalalignment='top', horizontalalignment='left')
            
              
        else:
            pass
        
    current_time = time.time()
    idx=0
    for update in did_we_update:
        if not update:
            timer_texts[idx].set_text(f'Last update: {str(round(current_time- prev_time[idx],3))} (s)')

        idx+=1
    return lines, timer_texts, prev_time


def keep_connection_alive(ws, epic):
    while True:
        time.sleep(30) 
        try:
            ws.ping("KeepAlive")
        except Exception as e:
            print(f"Error sending ping: {e}")
            return 

def getting_websocket_responses(queue, epic, cst, x_security_token, ws, data_dict):
    response_counter = 0
    pair = 0
    while True:
        
        response = json.loads(ws.recv())
        if response_counter >= 2:
        
            if 'payload' in response and 'bid' in response['payload']:
      
                bid_price = response['payload']['bid']
                ask_price = response['payload']['ofr']
                mid_price = (bid_price + ask_price) / 2
                
                timestamp = response['payload']['timestamp']
                time_UTC = datetime.utcfromtimestamp(timestamp/1000).replace(tzinfo=pytz.utc)
                
                city_tz = pytz.timezone('some city')
                tick = time_UTC.astimezone(city_tz)

                queue.put(([tick,mid_price],epic, data_dict, pair))

        response_counter += 1


def get_live_prices(queue, epic, cst, x_security_token, data_dict):
  
    looping = True
    while looping:

        try:
            ws = create_connection('wss://some_websocket_connection', timeout=999999)
    
            payload = {
                "epics": [epic]
            }
            req = {
                'destination': 'marketData.subscribe',
                'securityToken': x_security_token,
                'cst': cst,
                'correlationId': 1,
                'payload': payload
            }
            
            ws.send(json.dumps(req))
            
            threading.Thread(target=keep_connection_alive, args=(ws,epic,), daemon=True).start()
  
            getting_websocket_responses(queue, epic, cst, x_security_token, ws, data_dict)
            
        except WebSocketConnectionClosedException:
            
            print('websocket close: ', epic)
            
            pass
                        
           
if __name__ == '__main__':
    
    username, password, api_key, base_url = credentials.get_logins()
    cst, x_security_token = credentials.create_new_session(username, password, api_key)
    resolution = 'MINUTE'
    data_points = '1000'
    epics = get_epics(cst,x_security_token)
    epics = epics['Indices']

    historical_dict = {}
    for epic in epics[:]:
        snapshot_dict = get_UTC_prices(epic, cst, x_security_token)
        if snapshot_dict == None:
            tick=None
            idx = 0
            
            while tick==None:
                print(f'Failed to get tick data for {epic}, retrying...')
                snapshot_dict = get_UTC_prices(epic, x_security_token, cst)

                if idx==5:
                    print(f'Too many failures for {epic}, removing from chart')
                    epics.remove(epic)
                    break
                idx+=1
        else:  
            historical_dict[epic] = snapshot_dict
    
    data_dict = {}
    bid_price_5pm_NY = historical_dict
    
    for epic in epics:
        tick, price = get_snapshot(epic, cst, x_security_token) 
        bid_price_5pm_NY[epic]
        bid_perc_list = [ (price - bid_price_5pm_NY[epic]) / bid_price_5pm_NY[epic]]
        
        data = {'time' : [], 'offer' : [], 'perc_change': bid_perc_list, 'price_4PM_NY' : bid_price_5pm_NY[epic]}
        
        data_dict[epic] = data

    manager = multiprocessing.Manager()
    queues = [manager.Queue() for _ in range(len(epics))]
    processes = []
    
    idx=0
   
    for epic in epics:
        process = multiprocessing.Process(target = get_live_prices, args=(queues[idx], epic, cst, x_security_token, data_dict))
        processes.append(process)
        idx+=1
        
    for process in processes:
        process.start()
        
    n = len(epics)
    ncols = int(np.ceil(np.sqrt(n)))
    nrows = int(np.ceil(n / ncols))
    
    fig, axs = plt.subplots(nrows, ncols, constrained_layout=True)
    
    axs = axs.flatten()
    lines = [ax.plot(data_dict[epic]['time'], data_dict[epic]['offer'])[0] for ax,epic in zip(axs,epics)]
    timer_texts = [ax.text(0.02, 0.95, '', transform=ax.transAxes, verticalalignment='top', horizontalalignment='left', fontsize=8) for ax in axs]
    
    prev_time = [time.time()] * len(queues)

    for ax,epic in zip(axs,epics):
        ax.set_title(epic,fontsize=8)
        ax.set_ylabel('Change (%)',fontsize=8)

    anim = FuncAnimation(fig, update_plot, fargs=(queues, lines, axs, timer_texts, prev_time), blit=False, interval=1000, cache_frame_data=False)
    plt.show()
    
    for process in processes:
        process.terminate()
        
    try:
        for process in processes:
            process.join()

    except KeyboardInterrupt:
           print("Program interrupted")