2

環境

  • Python 3.12.4

やりたいこと

以下のlib.pyには、外部に公開しているsay_after関数と、それを呼び出しているdo_say_after関数が定義されています。

lib.py

import time


def say_after(delay, what):
    """外部に公開している関数"""
    time.sleep(delay)
    print(what)


def do_say_after():
    """外部に公開している関数"""
    print(f"started at {time.strftime('%X')}")
    say_after(1, "hello")
    say_after(2, "world")
    print(f"finished at {time.strftime('%X')}")


if __name__ == "__main__":
    do_say_after()
$ python lib.py
started at 15:12:13
hello
world
finished at 15:12:16

do_say_afterの処理は同期的なので、実行時間は約3秒(=1秒+2秒)です。

今回、do_say_afterの処理をasyncioを使って非同期的な処理に変更したいです。
say_after関数とdo_say_after関数はすでに外部に公開しているので、asyncを付けてコルーチンには変更できません。
もしコルーチンに変更してよいならば、以下のようなコードにしたいです。

import time
import asyncio

async def say_after(delay, what):
    """外部に公開している関数"""
    await asyncio.sleep(delay)
    print(what)

async def do_say_after():
    async with asyncio.TaskGroup() as tg:
        task1 = tg.create_task(say_after(1, "hello"))
        task2 = tg.create_task(say_after(2, "world"))
        print(f"started at {time.strftime('%X')}")

    print(f"finished at {time.strftime('%X')}")

if __name__ == "__main__":
    asyncio.run(do_say_after())

質問

say_after関数とdo_say_after関数のインターフェイスを変えずに、do_say_after関数内でsay_after関数を並行で呼び出すには、どのような処理にすればよいでしょうか?

試したこと

asyncio.get_event_loop()でイベントループを取得してrun_in_executorで実行する

以下の記事を参考にして、loop.run_in_executorを利用してみました。

https://qiita.com/nilwurtz/items/85b81febefd3f3f2899a

new_lib1.py

import time
import asyncio

def say_after(delay, what):
    """外部に公開している関数"""
    time.sleep(delay)
    print(what)


def do_say_after():
    """外部に公開している関数"""
    loop = asyncio.get_event_loop()
    
    # asyncio.gatherを使って並行実行
    task1 = loop.run_in_executor(None, say_after, 1, "hello")
    task2 = loop.run_in_executor(None, say_after, 2, "world")
    
    print(f"started at {time.strftime('%X')}")
    
    gather = asyncio.gather(task1, task2)
    result = loop.run_until_complete(gather)
    print(f"finished at {time.strftime('%X')}")

if __name__ == "__main__":
    do_say_after()

$ python new_lib1.py
/home/yuji/tmp/20240809/new_lib1.py:12: DeprecationWarning: There is no current event loop
  loop = asyncio.get_event_loop()
started at 15:22:54
hello
world
finished at 15:22:56

実行時間は約2秒なので、say_after関数を並行で実行できました。
ただし、asyncio.get_event_loop()で、There is no current event loopという警告が発生しました。

asyncio.get_running_loop()でイベントループを取得してrun_in_executorで実行する

公式ドキュメントによると、get_event_loop()よりもget_running_loop()が好ましいようなので、get_running_loop()に置き換えて実行しました。

この関数の振る舞いは (特にイベントループポリシーをカスタマイズした場合) 複雑なため、コルーチンやコールバックでは get_event_loop() よりも get_running_loop() を使うほうが好ましいと考えられます。

https://docs.python.org/ja/3.12/library/asyncio-eventloop.html#asyncio.get_event_loop

new_lib2.py

import time
import asyncio

def say_after(delay, what):
    """外部に公開している関数"""
    time.sleep(delay)
    print(what)


def do_say_after():
    """外部に公開している関数"""
    loop = asyncio.get_running_loop()
    
    # asyncio.gatherを使って並行実行
    task1 = loop.run_in_executor(None, say_after, 1, "hello")
    task2 = loop.run_in_executor(None, say_after, 2, "world")
    
    print(f"started at {time.strftime('%X')}")
    
    gather = asyncio.gather(task1, task2)
    result = loop.run_until_complete(gather)
    print(f"finished at {time.strftime('%X')}")

if __name__ == "__main__":
    do_say_after()
$ python new_lib2.py
Traceback (most recent call last):
  File "/home/yuji/tmp/20240809/new_lib2.py", line 25, in <module>
    do_say_after()
  File "/home/yuji/tmp/20240809/new_lib2.py", line 12, in do_say_after
    loop = asyncio.get_running_loop()
           ^^^^^^^^^^^^^^^^^^^^^^^^^^
RuntimeError: no running event loop

no running event loopというRuntimeErrorが発生しました。

asyncio.new_event_loop()でイベントループを取得してrun_in_executorで実行する

ChatGPTに質問して、asyncio.new_event_loopを使う方法を試しました。

new_lib3.py

import time
import asyncio


def say_after(delay, what):
    """外部に公開している関数"""
    time.sleep(delay)
    print(what)


def do_say_after():
    """外部に公開している関数"""
    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop)

    # asyncio.gatherを使って並行実行
    task1 = loop.run_in_executor(None, say_after, 1, "hello")
    task2 = loop.run_in_executor(None, say_after, 2, "world")

    print(f"started at {time.strftime('%X')}")

    gather = asyncio.gather(task1, task2)
    result = loop.run_until_complete(gather)
    print(f"finished at {time.strftime('%X')}")
    loop.close()


if __name__ == "__main__":
    do_say_after()
$ python new_lib3.py 
started at 21:28:31
hello
world
finished at 21:28:33

実行時間は約2秒なので、say_after関数を並行で実行できました。

分かっていないこと

  • new_lib1.pyasyncio.get_event_loopを呼び出すと、There is no current event loopという警告が発生する理由。また、new_lib2.pyasyncio.get_running_loopを呼び出すと、no running event loopというエラーが発生する理由。
    • ソースコードを見ると、get_event_loopget_running_loopは内部で_get_running_loopを呼び出していた。
  • new_lib3.pyasyncio.new_event_loopを使うことによる問題はあるかどうか。
    • 公式サイトには、以下のように記載さていたので、イベントループを参照するコードは何か間違っている気がする。

    アプリケーション開発者は通常 asyncio.run() のような高水準の ayncio 関数だけを利用し、ループオブジェクトを参照したり、ループオブジェクトのメソッドを呼び出したりすることはほとんどありません。

  • 外部に公開する関数内で、イベントループを扱ってよいかどうか。たとえば、do_say_afterをコルーチンで呼び出したら、Cannot run the event loop while another loop is runningが発生した(以下のnew_lib4.py参照)

new_lib4.py

import time
import asyncio


def say_after(delay, what):
    """外部に公開している関数"""
    time.sleep(delay)
    print(what)


def do_say_after():
    """外部に公開している関数"""
    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop)

    # asyncio.gatherを使って並行実行
    task1 = loop.run_in_executor(None, say_after, 1, "hello")
    task2 = loop.run_in_executor(None, say_after, 2, "world")

    print(f"started at {time.strftime('%X')}")

    gather = asyncio.gather(task1, task2)
    result = loop.run_until_complete(gather)
    print(f"finished at {time.strftime('%X')}")
    loop.close()

async def main():
    do_say_after()
    
if __name__ == "__main__":
    asyncio.run(main())
$ python new_lib4.py 
started at 22:20:36
Traceback (most recent call last):
  File "/home/yuji/tmp/20240809/new_lib4.py", line 31, in <module>
    asyncio.run(main())
  File "/home/yuji/.pyenv/versions/3.12.4/lib/python3.12/asyncio/runners.py", line 194, in run
    return runner.run(main)
           ^^^^^^^^^^^^^^^^
  File "/home/yuji/.pyenv/versions/3.12.4/lib/python3.12/asyncio/runners.py", line 118, in run
    return self._loop.run_until_complete(task)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/yuji/.pyenv/versions/3.12.4/lib/python3.12/asyncio/base_events.py", line 687, in run_until_complete
    return future.result()
           ^^^^^^^^^^^^^^^
  File "/home/yuji/tmp/20240809/new_lib4.py", line 28, in main
    do_say_after()
  File "/home/yuji/tmp/20240809/new_lib4.py", line 23, in do_say_after
    result = loop.run_until_complete(gather)
             ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/yuji/.pyenv/versions/3.12.4/lib/python3.12/asyncio/base_events.py", line 663, in run_until_complete
    self._check_running()
  File "/home/yuji/.pyenv/versions/3.12.4/lib/python3.12/asyncio/base_events.py", line 624, in _check_running
    raise RuntimeError(
RuntimeError: Cannot run the event loop while another loop is running
hello
world

改めて質問

say_after関数とdo_say_after関数のインターフェイスを変えずに、do_say_after関数内でsay_after関数を並行で呼び出すには、new_lib3.pyのようなコードでよいでしょうか?
イベントループの取得にはasyncio.new_event_loopを使っていますが、この方法で正しいのかが分かっていません。

追記

new_lib11.py

oriri さんの回答で教えていただいたコードからawait asyncio.sleepを削除して、実行しました。

import time
import asyncio

def say_after(delay, what):
    """外部に公開している関数"""
    time.sleep(delay)
    print(what)

def do_say_after():
    """外部に公開している関数"""

    async def f1():
        say_after(1, "hello")

    async def f2():
        say_after(2, "world")

    async def main():
        async with asyncio.TaskGroup() as tg:
            task1 = tg.create_task(
                f1())

            task2 = tg.create_task(
                f2())

            print(f"started at {time.strftime('%X')}")

        # The await is implicit when the context manager exits.

        print(f"finished at {time.strftime('%X')}")

    asyncio.run(main())

if __name__ == "__main__":
    do_say_after()

実行時間は約3秒でした。

$ python new_lib11.py
started at 14:59:36
hello
world
finished at 14:59:39
3

2 件の回答 2

1
+200

定義など

コメントで記したように, say_after()を共通で呼び出す, 何らかの処理を行う関数 (たぶん演算処理)と定義します。
非同期 I/O タスクの中では少し毛色の違う処理といえる
(逆にsay_after()内部に I/O が存在するのなら, 非同期タスクを再設計・再配置したほうがよいでしょう)

また, ルールが不明瞭で, なにが是で なにが非なのか, 第三者である回答者には判断しづらく
とりあえずの回答ということになります


並行で呼び出す

say_after関数とdo_say_after関数のインターフェイスを変えずに、do_say_after関数内でsay_after関数を並行で呼び出すには

import time
import asyncio

def say_after(delay, what):
    """外部に公開している関数"""
    time.sleep(delay)
    print(what)

def do_say_after():
    """外部に公開している関数"""

    async def f1():
        await asyncio.sleep(1)   # 何らかの前処理, 非同期 I/O処理を予定
        say_after(1, "hello")

    async def f2():
        await asyncio.sleep(.5)   # 何らかの前処理, 非同期 I/O処理を予定
        say_after(2, "world")

    async def main():
        async with asyncio.TaskGroup() as tg:
            task1 = tg.create_task(
                f1())

            task2 = tg.create_task(
                f2())

            print(f"started at {time.strftime('%X')}")

        # The await is implicit when the context manager exits.

        print(f"finished at {time.strftime('%X')}")

    asyncio.run(main())

if __name__ == "__main__":
    do_say_after()
    # time.sleep(1)
    # do_say_after()
  • do_say_after() 関数内部に組み込んでいる f1(), f2(), main() は, 外部に設置してもそのまま動くはず。
    ただし, 外部に出す => 公開 API となる => 公開 APIでは async が外される, というルールが存在するのなら, 内部関数のままがよいでしょう

  • 前処理は, asyncio.sleep() で表している, await を多用するであろう処理

  • say_after() 共通処理では I/Oを行わないので, 走り出したらノンストップで最後まで走り切ることになる

  • do_say_after() を二度呼び出しても二度目も動作するのは確認済み

処理結果については, new_lib1.pynew_lib4.py での動作より遅くなります
なぜならば, 速度的な同等のコードは以下のようになるため
(run_in_executorに Noneを指定すると, ThreadPoolExecutorが利用されることになっているはず)

def do_say_after():
    """外部に公開している関数"""
    def f1():
        ### async await
        say_after(1, "hello")
    def f2():
        ### async await
        say_after(2, "world")

    def main():
        with ThreadPoolExecutor(max_workers=4) as e:
            e.submit(f1)
            e.submit(f2)

            print(f"started at {time.strftime('%X')}")

        # The await is implicit when the context manager exits.

        print(f"finished at {time.strftime('%X')}")

    main()

つまり「並行」ではなく「並列」だから高速ということ

基本的に, 非同期 I/O 処理は高速で動かすためのものではなく, 効率よくタスクを動かすためのものです
(ただし複数スレッドを用意し, それぞれにイベントループ準備し, それぞれで非同期 I/Oタスクを動かす荒業もあるようですが)


run_in_executorについて

run_in_executorメソッドは
それこそ requests.get()に代表されるような,「ブロッキングを行う可能性の(ある|高い) レガシーコード」を, 非同期 I/O の中に組み込む機能(メソッド)といえます

参考: 低水準の API インデックス

  • await loop.run_in_executor()
    CPUバウンドなブロッキング関数、またはその他のブロッキング関数を concurrent.futures 実行オブジェクト (executor) 上で実行します。

(そうでない関数を,run_in_executor で利用するのは, 個人的にはおすすめできないと考えます)

requests に関して

  • 案外高機能 (セッションがあったり, リトライ回数指定できたり)
  • 内部では urllib3 を利用しているらしい
  • asyncio には対応予定はないと公言していたはず
  • asyncio 対応の requestsもどきが幾つか存在するが, 機能豊富なものはなかったように思えた

詳しくは(興味のある人は), 各自で調べてみてください


(new)

コメントに対する応答

say_after関数の中ではsleepしているだけなので、スレッドでもasyncioでも結果は同じだと感がていました。
コードを一部修正して、new_lib11.pyとして実行しました。結果は3秒でした。実行結果が2秒でないのはなぜでしょうか?

ThreadPoolExecutor で「並列」に動かせば, 1s と 2s では, 2秒あればどちらも完了しているはずです。
非同期 I/O 処理は「並列」ではないので, 順次実行することになり, 完了時間は合計時間の 3秒です

この辺りは, 非同期 I/O 処理の基礎的なことであり解説しだすと巨大になりがち
元の質問と外れてきていることもあり, さらに詳しい回答が必要であれば, 別の質問にするのがよいでしょう

say_after()はWebAPIに定期的にアクセスして状態を確認する処理を想定していました。したがって、say_after()にI/Oは存在します。

WebAPI がどのようになっているか不明なので適当ですが
私ならば(可能ならば) 以下のようにするかもしれません

async def f1():
    say_after(1, "hello")   # WebAPI 送信 
    await WebAPI_reply()   # WebAPIリプライ待ち

async def f2():
    say_after(2, "world")   # WebAPI 送信 
    await WebAPI_reply()   # WebAPIリプライ待ち
9
  • run_in_executor の説明で適切に記してあるページを見つけたので, 「参考」として付記しました
    – oriri
    Commented 8月22日 4:50
  • >なぜならば, 速度的な同等のコードは以下のようになるため new_lib1.py 〜 new_lib4.py のコードど同等という意味で合っていますか? ThreadPoolExecutorを使っていることは理解できましたが、ThreadPoolExecutorを使っているから高速なのが分かりませんでした。 say_after関数の中ではsleepしているだけなので、スレッドでもasyncioでも結果は同じだと感がていました。
    – yuji38kwmt
    Commented 8月25日 6:07
  • >(逆にsay_after()内部に I/O が存在するのなら, 非同期タスクを再設計・再配置したほうがよいでしょう) 回答が遅くなってすみません。say_after()はWebAPIに定期的にアクセスして状態を確認する処理を想定していました。したがって、say_after()にI/Oは存在します。
    – yuji38kwmt
    Commented 8月25日 6:08
  • 関数の内部でもf1f2のようにコルーチンを定義できるのですね(当然といえば当然)。知りませんでした。。。
    – yuji38kwmt
    Commented 8月25日 6:11
  • oririさんの回答で教えてもらったコードを一部修正して、new_lib11.pyとして実行しました。結果は3秒でした。実行結果が2秒でないのはなぜでしょうか?
    – yuji38kwmt
    Commented 8月25日 6:11
1

以下のような方法ではどうでしょうか。

import time
from concurrent.futures import ThreadPoolExecutor


def say_after(delay, what):
    """外部に公開している関数"""
    time.sleep(delay)
    print(what)


def do_say_after():
    """外部に公開している関数"""
    first_list = [1, 2]  # say_after関数に渡す第1引数のデータをリストで用意
    second_list = ['hello', 'world']  # say_after関数に渡す第2引数のデータをリストで用意
    assert len(first_list) == len(second_list), 'Invalid parameter.'  # リストの長さが等しいことを確認

    with ThreadPoolExecutor(max_workers=len(first_list)) as executor:
        # _ = list(executor.map(say_after, first_list, second_list))  # listですべての並列処理が完了するまで待つ
        executor.map(say_after, first_list, second_list)  # 2024/08/21訂正 すべての実行が完了してから、withブロックを抜けるようです。

if __name__ == "__main__":
    do_say_after()

この質問に回答するには、ログインする必要があります。

求めていた回答ではありませんか? のタグが付いた他の質問を参照する。