title: Start asyncio event loop in separate thread and consume queue items
body: <p>I am writing a Python program that run tasks taken from a queue concurrently, to learn  <code>asyncio</code>.</p>
  <p>Items will be put onto a queue by interacting with a main thread (within REPL).
  Whenever a task is put onto the queue, it should be consumed and executed immediately.
  My approach is to kick off a separate thread and pass a queue to the event loop within that thread.</p>
  <p>The tasks are running but only sequentially and I am not clear on how to run the tasks concurrently. My attempt is as follows:</p>
  <pre><code>import asyncio
  import time
  import queue
  import threading
  
  def do_it(task_queue):
      '''Process tasks in the queue until the sentinel value is received'''
      _sentinel = 'STOP'
  
      def clock():
          return time.strftime(&quot;%X&quot;)
  
      async def process(name, total_time):
          status = f'{clock()} {name}_{total_time}:'
          print(status, 'START')
          current_time = time.time()
          end_time = current_time + total_time
          while current_time &lt; end_time:
              print(status, 'processing...')
              await asyncio.sleep(1)
              current_time = time.time()
          print(status, 'DONE.')
  
      async def main():
          while True:
              item = task_queue.get()
              if item == _sentinel:
                  break
              await asyncio.create_task(process(*item))
  
      print('event loop start')
      asyncio.run(main())
      print('event loop end')
  
  
  if __name__ == '__main__':
      tasks = queue.Queue()
      th = threading.Thread(target=do_it, args=(tasks,))
      th.start()
  
      tasks.put(('abc', 5))
      tasks.put(('def', 3))
  </code></pre>
  <p>Any advice pointing me in the direction of running these tasks concurrently would be greatly appreciated!<br />
  Thanks</p>
  <p><strong>UPDATE</strong><br />
  Thank you Frank Yellin and cynthi8! I have reformed main() according to your advice:</p>
  <ul>
  <li>removed <code>await</code> before <code>asyncio.create_task</code> - fixed concurrency</li>
  <li>added wait while loop so that main would not return prematurely</li>
  <li>used non-blocking mode of Queue.get()</li>
  </ul>
  <p>The program now works as expected </p>
  <p><strong>UPDATE 2</strong><br />
  user4815162342 has offered further improvements, I have annotated his suggestions below.</p>
  <pre><code>'''
  Starts auxiliary thread which establishes a queue and consumes tasks within a
  queue.
      
  Allow enqueueing of tasks from within __main__ and termination of aux thread
  '''
  import asyncio
  import time
  import threading
  import functools
  
  def do_it(started):
      '''Process tasks in the queue until the sentinel value is received'''
      _sentinel = 'STOP'
  
      def clock():
          return time.strftime(&quot;%X&quot;)
  
      async def process(name, total_time):
          print(f'{clock()} {name}_{total_time}:', 'Started.')
          current_time = time.time()
          end_time = current_time + total_time
          while current_time &lt; end_time:
              print(f'{clock()} {name}_{total_time}:', 'Processing...')
              await asyncio.sleep(1)
              current_time = time.time()
          print(f'{clock()} {name}_{total_time}:', 'Done.')
  
      async def main():
          # get_running_loop() get the running event loop in the current OS thread
          # out to __main__ thread
          started.loop = asyncio.get_running_loop()
          started.queue = task_queue = asyncio.Queue()
          started.set()
          while True:
              item = await task_queue.get()
              if item == _sentinel:
                  # task_done is used to tell join when the work in the queue is 
                  # actually finished. A queue length of zero does not mean work
                  # is complete.
                  task_queue.task_done()
                  break
              task = asyncio.create_task(process(*item))
              # Add a callback to be run when the Task is done.
              # Indicate that a formerly enqueued task is complete. Used by queue 
              # consumer threads. For each get() used to fetch a task, a 
              # subsequent call to task_done() tells the queue that the processing
              # on the task is complete.
              task.add_done_callback(lambda _: task_queue.task_done())            
  
          # keep loop going until all the work has completed
          # When the count of unfinished tasks drops to zero, join() unblocks.
          await task_queue.join()
  
      print('event loop start')
      asyncio.run(main())
      print('event loop end')
  
  if __name__ == '__main__':
      # started Event is used for communication with thread th
      started = threading.Event()
      th = threading.Thread(target=do_it, args=(started,))
      th.start()
      # started.wait() blocks until started.set(), ensuring that the tasks and
      # loop variables are available from the event loop thread
      started.wait()
      tasks, loop = started.queue, started.loop
  
      # call_soon schedules the callback callback to be called with args arguments
      # at the next iteration of the event loop.
      # call_soon_threadsafe is required to schedule callbacks from another thread 
      
      # put_nowait enqueues items in non-blocking fashion, == put(block=False)
      loop.call_soon_threadsafe(tasks.put_nowait, ('abc', 5))
      loop.call_soon_threadsafe(tasks.put_nowait, ('def', 3))
      loop.call_soon_threadsafe(tasks.put_nowait, 'STOP')
  </code></pre>