回答編集履歴

コードレイアウト調整、不要な処理を消去。内容に変更はありません。

2024/12/11 23:43

投稿

teamikl

スコア8760

test CHANGED Viewed

@@ -49,7 +49,6 @@
 追記: 動作サンプルコード
 ```python
 import time
 import asyncio
 from concurrent.futures import ThreadPoolExecutor
@@ -60,7 +59,6 @@
 TASK_LIMIT = 5
 QUEUE_SIZE = 10
 INTERVAL = 10
 DATA_LIST = [
     {"key": "a", "value": 1},
     {"key": "b", "value": 2},
@@ -106,11 +104,9 @@
             await semaphore.acquire() # 複数実行中のタスク完了を待つ
             future = loop.run_in_executor(executor, heavy_task, item)
             future.add_done_callback(done)
-            await asyncio.sleep(0)
 def debug(func):
     # タスクの実行タイム計測
-    import time
     from datetime import datetime
     from functools import wraps

コード修正: 変数名修正、キューサイズ等の調整用変数をファイル冒頭へ移動

2024/12/11 23:39

投稿

teamikl

スコア8760

test CHANGED Viewed

@@ -49,6 +49,7 @@
 追記: 動作サンプルコード
 ```python
 import time
 import asyncio
 from concurrent.futures import ThreadPoolExecutor
@@ -56,13 +57,17 @@
 logger = logging.getLogger(__name__)
+TASK_LIMIT = 5
+QUEUE_SIZE = 10
+INTERVAL = 10
 DATA_LIST = [
     {"key": "a", "value": 1},
     {"key": "b", "value": 2},
     {"key": "c", "value": 3},
 ]
-def producer(data=DATA_LIST, interval=10):
+def producer(data=DATA_LIST, interval=INTERVAL):
     # sync generator
     while True:
         yield from data
@@ -85,7 +90,7 @@
 async def consumer_task(queue):
     logger.info("consumer start")
-    semapho = asyncio.Semaphore(5) # 同時実行数
+    semaphore = asyncio.Semaphore(TASK_LIMIT) # 同時実行数
     loop = asyncio.get_event_loop()
     def done(future):
@@ -93,12 +98,12 @@
         result = future.result()
         logger.debug(f"task done {result=}")
         queue.task_done()
-        semapho.release()
+        semaphore.release()
     with ThreadPoolExecutor(thread_name_prefix="worker") as executor:
         async for item in consumer(queue):
             logger.debug(f"C: {item=}")
-            await semapho.acquire() # 複数実行中のタスク完了を待つ
+            await semaphore.acquire() # 複数実行中のタスク完了を待つ
             future = loop.run_in_executor(executor, heavy_task, item)
             future.add_done_callback(done)
             await asyncio.sleep(0)
@@ -131,7 +136,7 @@
 async def main():
     # producer->consumer へ値を渡す、非同期キュー
-    queue = asyncio.Queue(maxsize=10)
+    queue = asyncio.Queue(maxsize=QUEUE_SIZE)
     async with asyncio.TaskGroup() as tg:
         task1 = tg.create_task(producer_task(queue, producer()))
@@ -148,5 +153,4 @@
         pass
     finally:
         logger.info("Done.")
 ```

文章・用語の校正

2024/12/11 19:42

投稿

teamikl

スコア8760

test CHANGED Viewed

@@ -17,10 +17,10 @@
 別スレッドでの実行は to_thread, run_in_executor 等。
-追記2: 要点が埋もれてしまったので、同期ジェネレータから非同期読み出し。for文は使わずに
+追記2: 要点が埋もれてしまったので、ブロッキング処理を含むジェネレータを別スレッドで実行し非同期読み出し。for文は使わずに
 ```python
-# 同期ジェネレーターからの非同期読み出し （別スレッドでジェネレーター内部の処理は進む)
+# ジェネレーターからの非同期読み出し （別スレッドでジェネレーター内部の処理は進む ⇒ asyncio のイベントループはブロックされない)
 value = await asyncio.to_thread(next, gen)
 ```

追記: 解決策の要点だった、同期ジェネレーターから値を非同期に読み出す方法を回答内に追記

2024/12/11 19:39

投稿

teamikl

スコア8760

test CHANGED Viewed

@@ -17,6 +17,14 @@
 別スレッドでの実行は to_thread, run_in_executor 等。
+追記2: 要点が埋もれてしまったので、同期ジェネレータから非同期読み出し。for文は使わずに
+```python
+# 同期ジェネレーターからの非同期読み出し （別スレッドでジェネレーター内部の処理は進む)
+value = await asyncio.to_thread(next, gen)
+```
+----
 解決策B
 タスク生成スケジュールの見直し、
 現状のコードではタスクの内容次第では、処理速度が追いつかないことがあるかもしれません。

回答例の実装サンプルコードを追記

2024/12/11 18:52

投稿

teamikl

スコア8760

test CHANGED Viewed

@@ -9,7 +9,7 @@
 解決策A;
 同期 <=> 非同期のやりとりとして、まず思いつくのは、
 janus というライブラリの同期<=>非同期キューを用いると
-例えば別スレッドで無限ループを処理し、非同期側で値を受け取ることが可能です。
+例えば別スレッドで無限ループを処理し、非同期側で値を受け取ることが可能です。(追記: 標準ライブラリの asyncio.Queue でも工夫すれば可能)
 実質、同期ジェネレーターで生成する値を、
 コルーチン側で非同期ジェネレーターとして扱う事が可能になります。
@@ -37,3 +37,108 @@
 何か重たい処理として仮置きしてるのかもしれませんが、
 コルーチン内での重たい処理は別スレッドで行いましょう。
 実際のコード次第では問題になる場合があります。
+追記: 動作サンプルコード
+```python
+import time
+import asyncio
+from concurrent.futures import ThreadPoolExecutor
+import logging
+logger = logging.getLogger(__name__)
+DATA_LIST = [
+    {"key": "a", "value": 1},
+    {"key": "b", "value": 2},
+    {"key": "c", "value": 3},
+]
+def producer(data=DATA_LIST, interval=10):
+    # sync generator
+    while True:
+        yield from data
+        logger.info(f"waiting {interval} sec")
+        time.sleep(interval)
+async def producer_task(queue, sync_gen):
+    # 非同期: ジェネレーターから値を取り出す
+    logger.info("producer start")
+    while True:
+        item = await asyncio.to_thread(next, sync_gen)
+        logger.debug(f"P: {item=}")
+        await queue.put(item) # キューの空を待つ (maxsize)
+async def consumer(queue):
+    # 非同期 generator: async for で扱う為の wrapper
+    while True:
+        item = await queue.get() # キューに値が入るのを待つ
+        yield item
+async def consumer_task(queue):
+    logger.info("consumer start")
+    semapho = asyncio.Semaphore(5) # 同時実行数
+    loop = asyncio.get_event_loop()
+    def done(future):
+        # 同期タスクの実行完了時
+        result = future.result()
+        logger.debug(f"task done {result=}")
+        queue.task_done()
+        semapho.release()
+    with ThreadPoolExecutor(thread_name_prefix="worker") as executor:
+        async for item in consumer(queue):
+            logger.debug(f"C: {item=}")
+            await semapho.acquire() # 複数実行中のタスク完了を待つ
+            future = loop.run_in_executor(executor, heavy_task, item)
+            future.add_done_callback(done)
+            await asyncio.sleep(0)
+def debug(func):
+    # タスクの実行タイム計測
+    import time
+    from datetime import datetime
+    from functools import wraps
+    @wraps(func)
+    def _func(*args, **kw):
+        result = None
+        start_time = datetime.now()
+        logger.debug(f"start {func.__name__}")
+        try:
+            result = func(*args, **kw)
+        except Exception as exc:
+            logger.error(f"error ", exc_info=exc)
+        end_time = datetime.now()
+        logger.debug(f"{result=} (time: {end_time - start_time})")
+        return result
+    return _func
+@debug
+def heavy_task(arg):
+    # 時間のかかるタスク。run_in_executor で実行する
+    time.sleep(5)
+    return arg
+async def main():
+    # producer->consumer へ値を渡す、非同期キュー
+    queue = asyncio.Queue(maxsize=10)
+    async with asyncio.TaskGroup() as tg:
+        task1 = tg.create_task(producer_task(queue, producer()))
+        task2 = tg.create_task(consumer_task(queue))
+if __name__ == '__main__':
+    logging.basicConfig(
+        format="[%(asctime)s][%(threadName)-10s][%(levelname)-8s] %(message)s",
+        level=logging.DEBUG,
+    )
+    try:
+        asyncio.run(main())
+    except (KeyboardInterrupt, SystemExit):
+        pass
+    finally:
+        logger.info("Done.")
+```

タスクの数量制限の解決策の追記

2024/12/11 16:48

投稿

teamikl

スコア8760

test CHANGED Viewed

@@ -27,6 +27,7 @@
 next() でジェネレーターから値を順次呼び出すことは可能です。
 （time.sleep(10) は不要になると思うので見直し候補）
+追記: Semaphore 等で数量制限
 ----
 ```python