title: Using Multiple Asyncio Queues Effectively
body: <p>I am currently building a project that requires multiple requests made to various endpoints. I am wrapping these requests in Aiohttp to allow for async.  </p>
  
  <p><strong>The problem:</strong><br>
  I have three Queues: <code>queue1</code>, <code>queue2</code>, and <code>queue3</code>. Additionally, I have three worker functions (<code>worker1</code>, <code>worker2</code>, <code>worker3</code>) that are associated with their respective Queue. The first queue is populated immediately with a list IDs that is known prior to running. When the request is finished and the data is committed to a database, it passes the ID to <code>queue2</code>. A <code>worker2</code> will take this ID and request more data. From this data it will begin to generate a list of IDs (different from the IDs in <code>queue1/queue2</code>. <code>worker2</code> will put the IDs in <code>queue3</code>. Finally <code>worker3</code> will grab this ID from <code>queue3</code> and request more data before committing to a database.</p>
  
  <p>The issue arises with the fact <code>queue.join()</code> is a blocking call. Each worker is tied to a separate Queue so the join for <code>queue1</code> will block until its finished. This is fine, but it also defeats the purpose of using async. Without using <code>join()</code> the program is unable to detect when the Queues are totally empty. The other issue is that there may be silent errors when one of the Queues is empty but there is still data that hasn't been added yet. </p>
  
  <p>The basic code outline is as follows:</p>
  
  <pre class="lang-py prettyprint-override"><code>queue1 = asyncio.Queue()
  queue2 = asyncio.Queue()
  queue3 = asyncio.Queue()
  
  async with aiohttp.ClientSession() as session:
      for i in range(3):
          tasks.append(asyncio.create_task(worker1(queue1)))
  
      for i in range(3):
          tasks.append(asyncio.create_task(worker2(queue2)))
  
      for i in range(10):
          tasks.append(asyncio.create_task(worker3(queue3)))
  
      for i in IDs:
         queue1.put_nowait(i)
  
      await asyncio.gather(*tasks)
  </code></pre>
  
  <p>The worker functions sit in an infinite loop waiting for items to enter the queue.</p>
  
  <p>When the data has all been processed there will be no exit and the program will hang. </p>
  
  <p>Is there a way to effectively manage the workers and end properly?</p>