title: python 3 asyncio and MotorClient: how to use motor with multithreading and multiple event loops
body: <p>I am back with a question about asyncio. I find it very useful (especially due to the GIL with threads) and I am trying to boost performances of some pieces of code.</p>
  
  <p>My application is doing the following:</p>
  
  <ul>
  <li>1 Background daemon thread "A" receives events from connected clients and reacts by populating a SetQueue (that simply is an event queue that removes duplicate ids) and by doing some insertions in a DB. I get this daemon from another module (basically I control a callback from when an event is received). In my sample code below I substituted this with a thread I generate and that very simply just populates the queue with 20 items and mimics DB inserts before exiting.</li>
  <li><p>1 Background daemon thread "B" is launched (loop_start) and he just loops over running until completion a coroutine that:</p>
  
  <ul>
  <li>Fetches all the items in the queue (if not empty, otherwise it release the control for x seconds and then the coroutine is re-launched)</li>
  <li><p>For each id in the queue it launches a chained coroutine that:</p>
  
  <ul>
  <li><p>Creates and waits for a task that just fetches all relevant information for that id from the DB. I am using MotorClient that supports asyncio to do await in the task itself.</p></li>
  <li><p>Uses an Pool of Processes executor to launch a process per id that uses the DB data to do some CPU intensive processing.</p></li>
  </ul></li>
  </ul></li>
  <li><p>The main thread just initializes the db_client and takes loop_start and stop commands.</p></li>
  </ul>
  
  <p>That is basically it.</p>
  
  <p>Now I am trying to boost performance as much as possible.</p>
  
  <p>My current issue is in using <code>motor.motor_asyncio.AsyncioMotorClient()</code> in this way:</p>
  
  <ol>
  <li>It gets initialized in the main thread and there I want to create indexes</li>
  <li>Thread "A" needs to perform DB insertions</li>
  <li>Thread "B" needs to perform DB finds/reads</li>
  </ol>
  
  <p>How can I do this? Motor states that it is meant for a single thread application where you use obviously a single event loop.
  Here I found myself forced to have two events loops, one in thread "A" and one in thread "B". This is not optimal, but I didn't manage to use a single event loop with call_soon_threadsafe while keeping the same behavior...and I think performance wise I am still gaining much with two events loop that release control over the gil bound cpu core.</p>
  
  <p>Should I use three different AsyncioMotorClient instances (one per thread) and use them as stated above? I failed with different errors while trying.</p>
  
  <p>Here is my sample code that doesn't include just the the MotorClient initialization in Asynchro's <code>__init__</code></p>
  
  <pre><code>import threading
  import asyncio
  import concurrent.futures
  import functools
  import os
  import time
  import logging
  from random import randint
  from queue import Queue
  
  
  
  
  
  # create logger
  logger = logging.getLogger(__name__)
  logger.setLevel(logging.DEBUG)
  # create file handler which logs even debug messages
  fh = logging.FileHandler('{}.log'.format(__name__))
  fh.setLevel(logging.DEBUG)
  # create console handler with a higher log level
  ch = logging.StreamHandler()
  ch.setLevel(logging.DEBUG)
  # create formatter and add it to the handlers
  formatter = logging.Formatter('%(asctime)s - %(name)s - %(processName)s - %(threadName)s - %(levelname)s - %(message)s')
  fh.setFormatter(formatter)
  ch.setFormatter(formatter)
  # add the handlers to the logger
  logger.addHandler(fh)
  logger.addHandler(ch)
  
  
  class SetQueue(Queue):
      """Queue that avoids duplicate entries while keeping an order."""
      def _init(self, maxsize):
          self.maxsize = maxsize
          self.queue = set()
  
      def _put(self, item):
          if type(item) is not int:
              raise TypeError
          self.queue.add(item)
  
      def _get(self):
          # Get always all items in a thread-safe manner
          ret = self.queue.copy()
          self.queue.clear()
          return ret
  
  
  class Asynchro:
      def __init__(self, event_queue):
          self.__daemon = None
          self.__daemon_terminate = False
          self.__queue = event_queue
  
      def fake_populate(self, size):
          t = threading.Thread(target=self.worker, args=(size,))
          t.daemon = True
          t.start()
  
      def worker(self, size):
          run = True
          populate_event_loop = asyncio.new_event_loop()
          asyncio.set_event_loop(populate_event_loop)
          cors = [self.worker_cor(i, populate_event_loop) for i in range(size)]
          done, pending = populate_event_loop.run_until_complete(asyncio.wait(cors))
          logger.debug('Finished to populate event queue with result done={}, pending={}.'.format(done, pending))
          while run:
              # Keep it alive to simulate something still alive (minor traffic)
              time.sleep(5)
              rand = randint(100, 200)
              populate_event_loop.run_until_complete(self.worker_cor(rand, populate_event_loop))
              if self.__daemon_terminate:
                  logger.debug('Closed the populate_event_loop.')
                  populate_event_loop.close()
                  run = False
  
      async def worker_cor(self, i, loop):
          time.sleep(0.5)
          self.__queue.put(i)
          logger.debug('Wrote {} in the event queue that has now size {}.'.format(i, self.__queue.qsize()))
          # Launch fake DB Insertions
          #db_task = loop.create_task(self.fake_db_insert(i))
          db_data = await self.fake_db_insert(i)
          logger.info('Finished to populate with id {}'.format(i))
          return db_data
  
      @staticmethod
      async def fake_db_insert(item):
          # Fake some DB insert
          logger.debug('Starting fake db insertion with id {}'.format(item))
          st = randint(1, 101) / 100
          await asyncio.sleep(st)
          logger.debug('Finished db insertion with id {}, sleep {}'.format(item, st))
          return item
  
      def loop_start(self):
          logger.info('Starting the loop.')
          if self.__daemon is not None:
              raise Exception
          self.__daemon_terminate = False
          self.__daemon = threading.Thread(target=self.__daemon_main)
          self.__daemon.daemon = True
          self.__daemon.start()
  
      def loop_stop(self):
          logger.info('Stopping the loop.')
          if self.__daemon is None:
              raise Exception
          self.__daemon_terminate = True
          if threading.current_thread() != self.__daemon:
              self.__daemon.join()
              self.__daemon = None
              logger.debug('Stopped the loop and closed the event_loop.')
  
      def __daemon_main(self):
          logger.info('Background daemon started (inside __daemon_main).')
          event_loop = asyncio.new_event_loop()
          asyncio.set_event_loop(event_loop)
          run, rc = True, 0
          while run:
              logger.info('Inside \"while run\".')
              event_loop.run_until_complete(self.__cor_main())
              if self.__daemon_terminate:
                  event_loop.close()
                  run = False
                  rc = 1
          return rc
  
      async def __cor_main(self):
          # If nothing in the queue release control for a bit
          if self.__queue.qsize() == 0:
              logger.info('Event queue is empty, going to sleep (inside __cor_main).')
              await asyncio.sleep(10)
              return
          # Extract all items from event queue
          items = self.__queue.get()
          # Run asynchronously DB extraction and processing on the ids (using pool of processes)
          with concurrent.futures.ProcessPoolExecutor(max_workers=8) as executor:
              cors = [self.__cor_process(item, executor) for item in items]
              logger.debug('Launching {} coroutines to elaborate queue items (inside __cor_main).'.format(len(items)))
              done, pending = await asyncio.wait(cors)
              logger.debug('Finished to execute __cor_main with result {}, pending {}'
                           .format([t.result() for t in done], pending))
  
      async def __cor_process(self, item, executor):
          # Extract corresponding DB data
          event_loop = asyncio.get_event_loop()
          db_task = event_loop.create_task(self.fake_db_access(item))
          db_data = await db_task
          # Heavy processing of data done in different processes
          logger.debug('Launching processes to elaborate db_data.')
          res = await event_loop.run_in_executor(executor, functools.partial(self.fake_processing, db_data, None))
          return res
  
      @staticmethod
      async def fake_db_access(item):
          # Fake some db access
          logger.debug('Starting fake db access with id {}'.format(item))
          st = randint(1, 301) / 100
          await asyncio.sleep(st)
          logger.debug('Finished db access with id {}, sleep {}'.format(item, st))
          return item
  
      @staticmethod
      def fake_processing(db_data, _):
          # fake some CPU processing
          logger.debug('Starting fake processing with data {}'.format(db_data))
          st = randint(1, 101) / 10
          time.sleep(st)
          logger.debug('Finished fake processing with data {}, sleep {}, process id {}'.format(db_data, st, os.getpid()))
          return db_data
  
  
  def main():
      # Event queue
      queue = SetQueue()
      return Asynchro(event_queue=queue)
  
  
  if __name__ == '__main__':
      a = main()
      a.fake_populate(20)
      time.sleep(5)
      a.loop_start()
      time.sleep(20)
      a.loop_stop()
  </code></pre>