新しいリクエストが同じセッションの同じユーザーから来たときに以前のリクエストをキャンセルする方法

Question

aiohttpを使用して残りのAPIを構築しています。私たちのアプリは、ユーザーが応答時間よりも頻繁にリクエストを送信するように設計されています（計算時間のため）。ユーザーにとっては、最新のリクエストの結果のみが重要です。古くなったリクエストの計算を停止することは可能ですか？

は、あなたは非常に非HTTP-のようなものを構築しているあなたに

Answer 1

ありがとうございます。 HTTP要求には数ミリ秒以上の時間がかかりません.HTTP要求は相互に依存してはいけません。時間がかかりすぎる計算を実行する必要がある場合は、アーキテクチャ/モデル/キャッシングなどを変更して速度を上げたり、HTTPインタフェースで制御できる長時間実行ジョブとして明示的に処理してください。つまり、「ジョブ」はHTTP経由で照会できる「物理リソース」です。あなたはPOSTリクエストを使用して、リソースを作成します。

GET /tasks/42 

{"status": "pending"} 

、最終的な結果を得る：

POST /tasks 
Content-Type: application/json 

{"some": "parameters", "go": "here"} 

{"resource": "/tasks/42"} 

その後、タスクのステータスを照会することができます

GET /tasks/42 

あなたは古いものに取って代わる新しいタスクを投稿するとき

{"status": "done", "results": [...]} 

、あなたのバックエンドは、それが適当と考えるどのような方法で、古いタスクをキャンセルすることができます。リソースは「取り消された」またはそれに類するステータスを返します。クライアントは、新しいタスクを開始した後、単に古いリソースを再度照会しません。

クライアントが1秒に1回リソースを照会しても、サーバー上のリソースは少なくなります（1つの接続が10秒間開いているのに対して、10の接続は同じ時間枠内で200ms開いています）。インテリジェントなキャッシングを適用します。これは、HTTPフロントエンドとは独立してタスクバックエンドをスケーリングすることができ、HTTPフロントエンドを複数のサーバーやロードバランサに簡単に拡張できるので、スケーラビリティも大幅に向上します。

Answer 2

私は@ Drizzt1991から解決策を投稿します：

ちょっとそこ、アルテムを。あなたがそこに持っている非常に奇妙な要件。 1つのクライアントがキープアライブのソケットを使用している場合、最初のクライアントに応答する前に次の要求を見ることは実際には不可能であることを理解してください。これはHTTPの動作方法で、別のリクエストを送信する前に結果を期待します。 したがって、クライアントが2つの別々のソケットで動作する場合にのみ動作しますが、同じクライアントからの2つのソケットが同じマシン上にルーティングされることを再度宣言する必要があります。練習では、これはフェイルオーバーやもので実際にはうまく動作しません。基本的にはステートフルAPIになります。 すべてのライブラリでキャンセルをサポートしているわけではありません。従来のリレーショナルDBでは、結果は無視されますが、依然として保留中のクエリが処理されます。複雑なものをグラフのトラバースとしてやっていて、たくさんのステップがあると、それはキャンセルできます。

しかし、あなたが主張する場合、そのクライアントは、ソケットのプールを使用し、それらが同一のマシンにルーティングされ、要求がキャンセルの恩恵を受け、このような何かが、トリックを行う必要があります。

import asyncio 
import random 

from aiohttp import web 


def get_session_id(request): 
    # I don't know how you do session management, so left it out 
    return "" 


async def handle(request): 
    session_id = get_session_id(request) 
    request['tr_id'] = tr_id = int(random.random() * 1000000) 

    running_tasks = request.app['running_tasks'] 
    if session_id in running_tasks and not running_tasks[session_id].done(): 
     running_tasks[session_id].cancel() 
     del running_tasks[session_id] 

    current_task = asyncio.ensure_future(_handle_impl(request)) 
    running_tasks[session_id] = current_task 

    try: 
     resp = await current_task 
    except asyncio.CancelledError: 
     print("Cancelled request", tr_id) 
     resp = web.Response(text="Cancelled {}".format(tr_id)) 
    finally: 
     if running_tasks[session_id] is current_task: 
      del running_tasks[session_id] 
    return resp 


async def _handle_impl(request): 
    tr_id = request['tr_id'] 
    print("Start request", tr_id) 
    await asyncio.sleep(10) 
    print("Finished request", tr_id) 
    return web.Response(text="Finished {}".format(tr_id)) 


app = web.Application() 
app.router.add_get('/', handle) 
app.router.add_get('/{name}', handle) 

app['running_tasks'] = {} 

web.run_app(app, host="127.0.0.1", port=8080)