asyncio に関する質問に戻ってきました。私はそれが非常に便利だと思います (特にスレッドを使用した GIL のおかげで)。いくつかのコードのパフォーマンスを向上させようとしています。
私のアプリケーションは次のことを行っています:
- 1 バックグラウンド デーモン スレッド「A」は、接続されたクライアントからイベントを受信し、SetQueue (重複 ID を削除する単純なイベント キュー) に入力し、DB に挿入することによって反応します。このデーモンは別のモジュールから取得します (基本的には、イベントが受信されたときからのコールバックを制御します)。以下のサンプル コードでは、これを私が生成したスレッドに置き換えました。これは、キューに 20 個のアイテムを入力し、終了する前に DB 挿入を模倣するだけです。
1 バックグラウンド デーモン スレッド "B" が起動され (loop_start)、次のコルーチンの実行が完了するまでループします。
- キュー内のすべてのアイテムを取得します (空でない場合、x 秒間コントロールを解放してから、コルーチンを再起動します)
キュー内の ID ごとに、次のような連鎖コルーチンを起動します。
DB からその ID に関連するすべての情報を取得するだけのタスクを作成して待機します。asyncio をサポートする MotorClient を使用して、タスク自体で await を実行しています。
Pool of Processes executor を使用して、ID ごとにプロセスを起動し、DB データを使用して CPU を集中的に処理します。
メイン スレッドは db_client を初期化し、loop_start コマンドと stop コマンドを受け取るだけです。
それは基本的にそれです。
今、私は可能な限りパフォーマンスを向上させようとしています。
私の現在の問題はmotor.motor_asyncio.AsyncioMotorClient()、このように使用することです:
- メインスレッドで初期化され、そこでインデックスを作成したい
- スレッド「A」は DB 挿入を実行する必要があります
- スレッド「B」は DB の検索/読み取りを実行する必要があります
これどうやってするの？Motor は、明らかに単一のイベントループを使用する単一スレッドアプリケーション向けであると述べています。ここで、スレッド「A」とスレッド「B」に 1 つずつ、合計 2 つのイベント ループを作成する必要があることに気付きました。これは最適ではありませんが、同じ動作を維持しながら call_soon_threadsafe で単一のイベントループを使用することはできませんでした...そして、ギルバウンドのCPUコアの制御を解放する2つのイベントループでパフォーマンスが向上していると思います.
3 つの異なる AsyncioMotorClient インスタンス (スレッドごとに 1 つ) を使用し、上記のように使用する必要がありますか? 試行中にさまざまなエラーで失敗しました。
これは、Asynchro の MotorClient 初期化だけを含まない私のサンプル コードです。__init__
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()