asyncioでタスクをスケジュールしたりキャンセルするには

Question

私はクライアント/サーバアプリケーションを作成しています。接続されている間、クライアントはサーバに「ハートビート」信号、例えば毎秒を送信する。 サーバ側では、タスク（またはコルーチンなど）を非同期に実行するためのメカニズムが必要です。さらに、 "ハートビート"信号の送信を停止したときに、クライアントからタスクを取り消したい。

つまり、サーバーがタスクを開始するときには、タイムアウトの種類があるか、ttl（例：3秒）です。サーバーは "ハートビート"信号を受信すると、タスクが完了するかクライアントが切断されるまで（信号の送信を停止する）、さらに3秒間タイマーをリセットします。

ここではexampleからasyncioチュートリアルpymotw.comのタスクをキャンセルしています。しかしここでは、event_loopが開始される前にタスクがキャンセルされています。これは私にとっては適切ではありません。

import asyncio 

async def task_func(): 
    print('in task_func') 
    return 'the result' 


event_loop = asyncio.get_event_loop() 
try: 
    print('creating task') 
    task = event_loop.create_task(task_func()) 

    print('canceling task') 
    task.cancel() 

    print('entering event loop') 
    event_loop.run_until_complete(task) 
    print('task: {!r}'.format(task)) 
except asyncio.CancelledError: 
    print('caught error from cancelled task') 
else: 
    print('task result: {!r}'.format(task.result())) 
finally: 
    event_loop.close() 

Answer 1

あなたはensure_future()メソッドを介してタスクを実行するためにasyncioTaskラッパーを使用することができます。

ensure_futureは、コルーチンを自動的にTaskラッパーにラップしてイベントループにアタッチします。また、Taskラッパーは、awaitからawaitまでのコルーチン 'クランクオーバー'（またはコルーチンが終了するまで）を保証します。

つまり、通常のコルーチンをensure_futureに渡し、結果のTaskオブジェクトを変数に割り当てます。停止する必要がある場合は、Task.cancel()に電話をかけることができます。

import asyncio 

async def task_func(): 
    print('in task_func') 
    # if the task needs to run for a while you'll need an await statement 
    # to provide a pause point so that other coroutines can run in the mean time 
    await some_db_or_long_running_background_coroutine() 
    # or if this is a once-off thing, then return the result, 
    # but then you don't really need a Task wrapper... 
    # return 'the result' 

async def my_app(): 
    my_task = None 
    while True: 
     await asyncio.sleep(0) 

     # listen for trigger/heartbeat 
     if heartbeat and not my_task: 
      my_task = asyncio.ensure_future(task_func()) 

     # also listen for termination of hearbeat/connection 
     elif not heartbeat and my_task: 
      if not my_task.cancelled(): 
       my_task.cancel() 
      else: 
       my_task = None 

run_app = asyncio.ensure_future(my_app()) 
event_loop = asyncio.get_event_loop() 
event_loop.run_forever() 

タスクは、メインフローを中断することなくバックグラウンドで作業を続ける必要がある長期実行タスク用です。あなたが必要とするのは、すぐに一度だけの方法であれば、代わりに関数を直接呼び出すことです。