title: Multiprocessing Producer Consumer with Python 3.x on Linux
body: <p>The following code below is simulating a producer and consumer model, which will collect data from a forex broker FXCM and write to a database.
  Each producer process will have a session based connection with the broker.</p>
  
  <p>Both producer and consumer will run indefinitely until a 'poison pill' is placed into the queue, which happens at the close of business (Friday 22:00). I have omitted this part of the code as it is irrelevant to the problem. All of the examples I can find seem to spawn a process, do some work for a short period and then <code>join()</code> back to the parent process. Like this one <a href="https://gist.github.com/brantfaircloth/1255715" rel="nofollow noreferrer" title="one">here</a></p>
  
  <p>As mentioned above the producer will run indefinitely, and the reason for this is because it takes about 3 seconds to log in and create a session with the broker. </p>
  
  <p>When running the code below you will see the Queue backlog, although this seems to be even worst when running the real code.</p>
  
  <p>Not sure if it is relevant but the session is created using the <a href="https://github.com/neka-nat/python-forexconnect" rel="nofollow noreferrer" title="python-forexconnect">python-forexconnect</a> API, which is written in C++ and uses Boost.</p>
  
  <p>The problem is the consumer is taking too long to <code>get()</code> items from the queue, and I am wondering if this model is the correct way to approach this type of development.</p>
  
  <p>Thank-you</p>
  
  <p>Sample code</p>
  
  <pre><code>from multiprocessing import Process, Queue, cpu_count
  from datetime import datetime, timedelta
  
  import numpy as np
  import time
  
  def dummy_data(dtto):
      dates = np.array(
            [dtto - timedelta(days=i) for i in range(300)])
      price_data = np.random.rand(len(dates),5)
      return np.concatenate(  
        (np.vstack(dates),price_data), axis=1)
  
  def get_bars(q2, ms, symbol, dtfm, dtto, time_frame):
      stop_date = dtfm
      while dtto &gt; stop_date:
          data = dummy_data(dtto)
          dtfm = data[-1,0]
          dtto = data[0,0]
          q2.put((symbol, dtfm, dtto))
          # Switch to date
          dtto = dtfm
  
  def producer(q1,q2):  
      # client = fx.Client(....)
      client = 'broker session'
      while True:
          job = q1.get()
          if job == None:
              break
          sym, dtfm, dtto, tf = job
          # Get price data from broker
          get_bars(q2, client, sym, dtfm, dtto, tf)
      q2.put(None)
  
  def consumer(q2):
      while True:
          bars = q2.get()
          if bars == None:
              break
          print(q2.qsize(), bars[0], bars[1], bars[2]) # write to db
  
  q1, q2 = Queue(), Queue()
  # instruments = client.get_offers()
  # instruments = ['GBP/USD', 'EUR,USD',...]
  instruments = range(63) # 62 dummy instruments
  # Places jobs into the queue for each symbol
  for symbol in instruments:
      q1.put((symbol,
          datetime(2000,1,14,22,0),
          datetime(2018,1,14,22,0),
          'D1'))
  # Setup producers and consumers
  pp, cp = range(6), range(2)
  pro = [Process(target=producer, args=(q1, q2,)) for i in pp]
  con = [Process(target=consumer, args=(q2,)) for i in cp]
  for p in pro: p.start()
  for p in con: p.start()
  # This is just here to stop this script and does not
  # exist in the real version
  for i in pp: q1.put(None)
  for p in pro: p.join()
  for p in con: p.join()
  print('stopped')
  </code></pre>