title: Asyncio worker that handles N jobs at a time?
body: <p>I'm trying to make an <code>asyncio</code> worker class that will consume jobs from a job queue and process up to <code>N</code> jobs in parallel. Some jobs may queue additional jobs. When the job queue is empty and the worker finishes all of its current jobs, it should end.</p>
  
  <p>I'm still struggling with <code>asyncio</code> conceptually. Here is one of my attempts, where <code>N=3</code>:</p>
  
  <pre><code>import asyncio, logging, random
  
  async def do_work(id_):
      await asyncio.sleep(random.random())
      return id_
  
  class JobQueue:
      ''' Maintains a list of all pendings jobs. '''
      def __init__(self):
          self._queue = asyncio.Queue()
          self._max_id = 10
          for id_ in range(self._max_id):
              self._queue.put_nowait(id_ + 1)
  
      def add_job(self):
          self._max_id += 1
          self._queue.put_nowait(self._max_id)
  
      async def get_job(self):
          return await self._queue.get()
  
      def has_jobs(self):
          return self._queue.qsize() &gt; 0
  
  class JobWorker:
      ''' Processes up to 3 jobs at a time in parallel. '''
      def __init__(self, job_queue):
          self._current_jobs = set()
          self._job_queue = job_queue
          self._semaphore = asyncio.Semaphore(3)
  
      async def run(self):
          while self._job_queue.has_jobs() or len(self._current_jobs) &gt; 0:
              print('Acquiring semaphore...')
              await self._semaphore.acquire()
              print('Getting a job...')
              job_id = await self._job_queue.get_job()
              print('Scheduling job {}'.format(job_id))
              self._current_jobs.add(job_id)
              task = asyncio.Task(do_work(job_id))
              task.add_done_callback(self.task_finished)
  
      def task_finished(self, task):
          job_id = task.result()
          print('Finished job {} / released semaphore'.format(job_id))
          self._current_jobs.remove(job_id)
          self._semaphore.release()
          if random.random() &lt; 0.2:
              print('Queuing a new job')
              self._job_queue.add_job()
  
  loop = asyncio.get_event_loop()
  jw = JobWorker(JobQueue())
  print('Starting event loop')
  loop.run_until_complete(jw.run())
  print('Event loop ended')
  loop.close()
  </code></pre>
  
  <p>An excerpt of the output:</p>
  
  <pre><code>Starting event loop
  Acquiring semaphore...
  Getting a job...
  Scheduling job 1
  Acquiring semaphore...
  Getting a job...
  Scheduling job 2
  Acquiring semaphore...
  Getting a job...
  Scheduling job 3
  Acquiring semaphore...
  Finished job 2 / released semaphore
  Getting a job...
  Scheduling job 4
  ...snip...
  Acquiring semaphore...
  Finished job 11 / released semaphore
  Getting a job...
  Finished job 12 / released semaphore
  Finished job 13 / released semaphore
  </code></pre>
  
  <p>It appears to correctly process all jobs while processing no more than 3 jobs at any one time. However, the program hangs after the last job is finished. As indicated by the output, it appears to be hanging at <code>job_id = await self._job_queue.get_job()</code>. Once the job queue is empty, this coroutine will never resume, and the check to see if the job queue is empty (at the top of the loop) isn't reached again.</p>
  
  <p>I've tried working around this in a number of ways but conceptually something just don't quite fit. My current WIP is passing some futures between the queue and the worker and then using some combination of <code>asyncio.wait(...)</code> on all of them, but it's getting ugly and I'm wondering if there is an elegant solution that I'm overlooking.</p>