Python 非同期通信でタスクオブジェクトが返ってくる際の対処法

Question

pythonの非同期通信に関して勉強し始めました。試しに色々書いてみているのですが、つまづいてしまったので質問させて頂きます。

async def main():
        var = 'おはよう'

        while True:

            task1 = asyncio.create_task(coro_1())
            var = task1
            
            task2 = asyncio.create_task(coro_2(var))
            await task2
            
            task3 = asyncio.create_task(coro_3())
            await task3
            

async def coro_1():
    await asyncio.sleep(3)
    var = 'こんばんは'
    return var

    
async def coro_2(var):
    await asyncio.sleep(1)
    pybotters.print(var)
    

async def coro_3():
    await asyncio.sleep(1)
    pybotters.print('こんにちは')
    
            

try:
    asyncio.run(main())
except KeyboardInterrupt:    
    pass

上記のコードで、coro_1関数でプログラム開始から3秒後にvarに'こんばんは'を代入し、printすること想定して書いてみました。(おはよう→こんにちは→こんにちは→こんばんは)のようにprintされると考えていました。

ですが、task1ではawaitをしていないためか、varにはタスクオブジェクトが代入されてしまいます。

awaitをするとプログラム開始から３秒後に'こんばんは'がvarに代入され、'おはよう'がprintされません。(こんばんは→こんにちは→こんばんは→....)となってしまいます。

私の想定通りにprintするにはどうすれば良いのでしょうか。

Answer 1

Python　非同期通信でタスクオブジェクトが返ってくる際の対処法

タスク内で return される値を扱うには、
task.add_done_callback メソッドで、タスク完了時に呼び出す関数を登録します。

import asyncio
from types import SimpleNamespace


async def main():
        shared = SimpleNamespace(
            var="おはよう",
        )

        task1 = asyncio.create_task(coro_1())
        
        def done(task):
            shared.var = task.result()

        task1.add_done_callback(done)

        while True:
            task2 = asyncio.create_task(coro_2(shared))
            
            await task2

            task3 = asyncio.create_task(coro_3())
            
            await task3    


async def coro_1():
    await asyncio.sleep(3)
    return 'こんばんは'

    
async def coro_2(shared):
    await asyncio.sleep(1)
    print(shared.var)
    

async def coro_3():
    await asyncio.sleep(1)
    print('こんにちは')


try:
    asyncio.run(main())
except KeyboardInterrupt:    
    pass

質問文: (おはよう→こんにちは→おはよう→こんにちは→こんばんは)のようにprintされると考えていました。
コメント: 私が想定していたのはおはよう→こんにちは→こんばんは→こんにちは→こんばんは→...というようにvarにはawaitの処理が終わるまでは「おはよう」、終わってからは「こんばんは」が代入されるというものでした。

想定する挙動がどちらか解りませんが、共通する問題点として

• coro_2 に渡す引数、コルーチン生成の時点でのvarの値が渡されます。

　coro_1 の結果を反映することができません。→ 対策: 辞書に入れて、引数には辞書を渡す等

• await task では、処理を中断し、右辺に指定された コルーチンの完了を待ちます 。

　質問に掲載のコードでは task2 と task3が完了する迄、次のループに移りません。

main側のループで実行したい場合は、メインのループの末尾で
await asyncio.sleep(適当な数値) としておけば、タスクの完了は待たずに次のループを実行できます。
(await task ではタスク完了を待ってしまう。await なしではコルーチンが実行されない。）
但し、ループ内で使う変数の上書き、デストラクタ等の注意が必要になったりする為、幾つか注意が必要。

main関数側のループで扱うのではなく、各コルーチン内で独立したループを持つ方法をお勧めします。
「〇〇秒後に実行」であれば、簡易的にはloop.call_later を使う方法もあります。

Comments

[paopao7]

ご回答ありがとうございます！なるほど、非同期にも様々なメソッドがあるんですね。
やっぱりコルーチン側でループさせるべきなんですね。色んなサイトを見てもmain側でループさせているのは少なかったです...笑
初歩的な質問で申し訳ないんですが、コルーチン側でループさせるとすると、処理の順番が自分の想定とずれてしまうんじゃないかと思ってしまうんですが、どうなんでしょうか？

[paopao7]

それと、main側でループさせるとするなら「await asyncio.sleep(適当な数値) としておけば、タスクの完了は待たずに次のループを実行できます。」とのことですが、例えばawait task2(何かしら時間のかかる処理だとして)がmainループにあるとすると、次のループに移ってから、task2の処理はまた初めから為されるということでしょうか？

[teamikl]

処理の順番は、想定されている順序がどちらか解らないので正解が解りませんが、
実行順序が重要になる場合は「非同期」ではなく「同期」コードとします。

現状のコードでは・・・
await task2 であ task2の完了を待ってから task3 を作る「同期」コードとなっているので、
分離する必要が無ければ一つのループに統合できます。

2つのコルーチンで実装する場合の例
- 処理が終わった後(3秒後) に変数を「こんばんは」に変更
　※ ３秒後というのが、何らかの時間の掛かる処理の場合は、別スレッドを使ったりもう少し複雑になります。
-「こんにちは」と変数を１秒毎に交互に表示するループ

>例えばawait task2(何かしら時間のかかる処理だとして)がmainループにあるとすると、
>次のループに移ってから、task2の処理はまた初めから為されるということでしょうか？

いいえ、この場合想定する前提が違い、
await task2 とはせずに、代わりに毎ループ await asyncio.sleep(n) とします。

task2を新たにループ毎に作るかどうかは、想定する挙動次第ですが、
task2の処理が勝手に巻き戻り最初から実行されることは、ありません。

Answer 2

var = task1 で おはよう が入っている変数が上書きされてしまうので、おはよう がprintされることはありませんね。

(おはよう→こんにちは→おはよう→こんにちは→こんばんは)のようにprintされると考えていました。

おはよう→こんにちは→こんばんは→おはよう→こんにちは→こんばんは の順ではないですか？
それならこれでどうでしょうか？

import asyncio


async def main():
    var = 'おはよう'

    while True:
        task1 = asyncio.create_task(coro_1())
        task2 = asyncio.create_task(coro_2(var))
        task3 = asyncio.create_task(coro_3())
        print(await task1)
        await task2
        await task3


async def coro_1():
    await asyncio.sleep(3)
    return  'こんばんは'


async def coro_2(var):
    await asyncio.sleep(1)
    print(var)


async def coro_3():
    await asyncio.sleep(1)
    print('こんにちは')


try:
    asyncio.run(main())
except KeyboardInterrupt:
    pass

こうすれば最初の1回だけ 'おはよう' が表示されます。

import asyncio


async def main():
    var = 'おはよう'

    while True:
        task1 = asyncio.create_task(coro_1())
        task2 = asyncio.create_task(coro_2(var))
        task3 = asyncio.create_task(coro_3())
        var = await task1
        await task2
        await task3


async def coro_1():
    await asyncio.sleep(3)
    return 'こんばんは'


async def coro_2(var):
    await asyncio.sleep(1)
    print(var)


async def coro_3():
    await asyncio.sleep(1)
    print('こんにちは')


try:
    asyncio.run(main())
except KeyboardInterrupt:
    pass

Taskが終わっていたらawaitするようにしたら望みの動作になりますか？

import asyncio


async def main():
    var = 'おはよう'
    task1 = task2 = task3 = None

    while True:
        if task1 is None:
            task1 = asyncio.create_task(coro_1())
        if task2 is None:
            task2 = asyncio.create_task(coro_2(var))
        if task3 is None:
            task3 = asyncio.create_task(coro_3())
        if task1.done():
            var = await task1
            task1 = None
        if task2.done():
            await task2
            task2 = None
        if task3.done():
            await task3
            task3 = None
        await asyncio.sleep(0.1)


async def coro_1():
    await asyncio.sleep(3)
    return 'こんばんは'


async def coro_2(var):
    await asyncio.sleep(1)
    print(var)


async def coro_3():
    await asyncio.sleep(1)
    print('こんにちは')


try:
    asyncio.run(main())
except KeyboardInterrupt:
    pass

少し簡略化

import asyncio


async def main():
    var = 'おはよう'
    task1 = None

    while True:
        if task1 is None:
            task1 = asyncio.create_task(coro_1())
        task2 = asyncio.create_task(coro_2(var))
        task3 = asyncio.create_task(coro_3())
        if task1.done():
            var = await task1
            task1 = None
        await task2
        await task3


async def coro_1():
    await asyncio.sleep(3)
    return 'こんばんは'


async def coro_2(var):
    await asyncio.sleep(1)
    print(var)


async def coro_3():
    await asyncio.sleep(1)
    print('こんにちは')


try:
    asyncio.run(main())
except KeyboardInterrupt:
    pass

Comments

[paopao7]

私が想定していたのはおはよう→こんにちは→こんばんは→こんにちは→こんばんは→...というようにvarにはawaitの処理が終わるまでは「おはよう」、終わってからは「こんばんは」が代入されるというものでした。
coro_1関数内で、await asyncio.sleep(3)としているため、varにはその時点では'こんばんは'は代入されず、awaitの処理が終わったら代入されると考えていたのですが、代入されてしまうんですね。
最終的にやりたいことは、awaitの処理が含まれるコルーチン関数は、awaitの処理が終わるまでは実行されず、他のコルーチン関数をmainのループで実行し続ける、というようなことなんですが、その場合はどうすればいいんでしょうか...？

[shiracamus]

await することでひとつの Task が動きます。
そのTaskの中で await すると他の Task が動きます。
await しない限り、他の Task は動けないです。

mainのループを実行し続けるには、作成した Task を await しつつ、ループ1回ごとに await して Task に実行権を譲渡しないといけません。
Task が終わっていたら await するようにすることで望みの動作になりますでしょうか？
回答にコードを追記しておきました。

[paopao7]

なるほど、工夫が必要なんですね。ありがとうございます！