asyncioを使ったwebsocketの処理

Question

以下のとおり、djangoのカスタムコマンドでasyncioを使ってwebsocketのデータを加工しています。
なお、元のソースコードはものすごく長いので、省略しています。

デバッグで追ったところ、method3のreturnまではいけたのですが、method2のreturn out_layerで止まってしまいます。
どなたかアドバイスをいただけないでしょうか？

（asyncioを使っているとエラーメッセージが表示できず、なぜ止まっているか判断が難しいです。）

実行したいことは、save_dataが終わった後に、method2と新たなメッセージの受信を同時に始めたいです。

class Command(BaseCommand):

    def handle(self, *args, **kwargs):

            """ループの開始"""
            loop = asyncio.get_event_loop()
            loop.create_task(self.stream())
            print('start receiving')
            try:
                loop.run_forever()
            except KeyboardInterrupt:
                exit()

    async def stream(self):
        uri = 'ws://localhost:1234/'

        async with websockets.connect(uri, ping_interval=None) as ws:
            while not ws.closed:
                response = await ws.recv()
                content = json.loads(response)
                task = loop.create_task(self.save_pushdata(content, sid))
                task.add_done_callback(self.finish)

    async def save_pushdata(self, content, sid):
        loop = asyncio.get_event_loop()
        newData = list(
            symbol=content['****'],
        )
        newData.save()
        task = loop.create_task(self.method2(content, sid))
        task.add_done_callback(self.finish)
        save_pushdata_sid = str(sid) + 'save_pushdata'
        return save_pushdata_sid

    async def method2(self, content):

        args1 = self.method4(content) 
        args2 = self.method5(content)
        self.method3(args1, args2)
        method2_sid = str(sid) + 'method2'
        return method2_sid

    def method3(self, content):
        print('123')
        return content

    def method4(self, content):
        newData = list(
            symbol=content['****'],
        )
        newData.save()
        return content['****']

    def method5(self, content):
        newData = list(
            symbol=content['****'],
        )
        newData.save()
        return content['****']

    def finish(self, text):
        logging.info('%s', text.result())

    def handle(self, *args, **kwargs):

            """ループの開始"""
            loop = asyncio.get_event_loop()
            loop.create_task(self.stream())
            print('start receiving')
            try:
                loop.run_forever()
            except KeyboardInterrupt:
                exit()

    async def stream(self):
        uri = 'ws://localhost:1234/'

        async with websockets.connect(uri, ping_interval=None) as ws:
            while not ws.closed:
                response = await ws.recv()
                content = json.loads(response)
                
                executor = concurrent.futures.ProcessPoolExecutor()
                queue = asyncio.Queue()

                for i in range(10):
                    queue.put_nowait(i)

                async def proc(q):
                    while not q.empty():
                        i = await q.get()
                        future = loop.run_in_executor(executor, save_pushdata(content, sid), i)
                        await future

                tasks = [proc(queue) for i in range(4)]  # 4 = number of cpu core
                return await asyncio.wait(tasks)

def save_pushdata(self, content, sid):
        loop = asyncio.get_event_loop()
        newData = list(
            symbol=content['****'],
        )
        newData.save()
        self.method2(content, sid)

Answer 1

asyncioを使っているとエラーメッセージが表示できず、なぜ止まっているか判断が難しいです。

asyncio自身はエラーメッセージに干渉しません。
django 側の設定や実行環境の問題ではないでしょうか？

djangoのカスタムコマンドについては知りませんが、コードには多数の問題があるので、
エラーメッセージを得られる環境設定を先に目指した方が良いと思います。

追記: カスタムコマンドではなく、djangoに依存しない通常のスクリプトとして
同様のコードを実装・実行しエラーメッセージは得られますか？

(モデル関連のコードは省き、asyncio/websockets の部分だけ動作確認する)

---

未定義のメソッド: save_data -> save_pushdata ?

    await self.save_pushdata(content)

---

組み込みのリストはキーワード引数やsave()メソッドを持ちません。

newData = list(
            symbol=content['****'],
        )
newData.save()

model の save() を呼び出したい？

---

async def method2(self, content):
    ...
    return out_layer

コルーチンは最低限一つ以上の await が必要です。
内容次第では、コルーチンにする必要があるかどうかから検討された方が良いと思います。

• method4, method5 が未定義
• method3 の引数が一致しない、等

Comments

[taro_yamada]

ご回答ありがとうございます。
とにかく元のソースコードが長く、djangoの別のクラスで様々なメソッドも定義してしまっていて、djangoでないスクリプトで実行するのが困難な状況に陥ってしまっています涙

＜エラーメッセージについて＞
djangoだけの場合エラーメッセージを出すので、asyncioの方が原因ではないかと思っています。
＜メソッドの定義エラー＞
ちなみに、メソッドの定義エラーは、省略したときのミスで、直します。
＜save()＞
djangoのORMのメソッドでDBにinsertしているので、一応問題なく動いています。こちらも省略したときの表現が悪かったです。
＜コルーチン化＞
問題は、コルーチンにする必要があるかどうかという点です。
目的として、メッセージを受信してからIO処理を含むいろいろなことをやっている間に次々とメッセージが溜まってtimeoutになってしまうという課題感から、並列処理を目指しています。
データ受信とその後の処理を切り分けて並列処理をするためにコルーチン化以外で適切な方法はありますでしょうか？
初心者で、どこにawaitを付けることができるかも今一つ理解していません。すみません。
ちなみに、method4とmethod5では、省略していますが、save()と同様にdjangoのORMを使ってDBへの書き込みをしています。

[teamikl]

エラーメッセージの表示に関しては、
asyncio単体で利用する場合も特に問題ないので、
django/asyncio で発生する問題なのかな？
どこかで例外が適切に伝搬されてないのだと思います。

djangoのORMに依存するモデル関連のコードは省き、
asyncio/websockets の部分だけで動作確認してみてください。

> どこにawaitを付けることができるか

これに関しては、method3～5 がもう少し具体的でないと、わかりませんが、
await を付けることができるのは、コールーチン、future, task 等
__await__ を実装するオブジェクトです。
通常は、何かの完了待ちをする処理を実装したオブジェクトです。

while not ws.closed:
　　...
　　await self.save_pushdata(content)

では、save_pushdata が完了するまで待ってから次の読み出しとなるので、
完了を待つ必要がないなら、ここは await 不要で task にするべきだと思います。
例えば、save_pushdata はコルーチンにはせずに、task を返す関数にする、等。

====

ちなみに、「並行」処理です。
await 迄処理 → asyncio のイベントループに戻る →他のコルーチンを処理
→ await の次から処理を再開～

と順番に実行されるので、
何処かで時間の掛かる処理があると全体の遅延に繋がります。

コードの見かけは複数の関数を同時に実行してるように見えますが、
内部ではawait の時点で切り替えて実行されます。
完了待ちが終了したもの次第再開される為、
どの関数が次に再開されるか決まってないという点が非同期です。
最初は「非同期」と「並列」「平行」の区別が解りにくいかもしれません。

[taro_yamada]

save_pushdata(content)をtaskを返すように変更したところ回り始めました。
ただ、contentに次の値が入ってきているのですが、なぜか2回目以降はtask = loop.create_task(self.save_pushdata(content))のループが始まりません。

[teamikl]

> save_pushdata(content)をtaskを返すように変更した

だと

＞ loop.create_task(self.save_pushdata(content))

は、task が２重になってませんか？
save_pushdata() が返すのは task オブジェクト
create_task() に必要なのはコルーチンです。(※ ここでは厳密に task != コルーチン)

恐らくエラーが出るはずだけど、
エラーメッセージが見れない原因を先に調べたほうが良さそうですね。

[teamikl]

django 3.1.4/asyncio 利用でエラーメッセージの表示確認出来ました。

例外がどこかで補足されてエラーが表示されてない、もしくは
コードが期待通りに実行されてない可能性もありますね。

----
他の問題点。
エラーメッセージ表示の問題が解消すれば簡単に修正できると思います。

> self.method3(args1, args2)
メソッドの定義と引き数が一致しない

> return out_layer
変数 out_layter は未定義

> task.add_done_callback(self.finish('test'))
コールバック登録になっていない。
finishを呼び出してその戻り値をcallbackに登録している。

[taro_yamada]

task.add_done_callback(self.finish)のfinishの引数を外したらなぜか動き始めました・・・

しかも、ほとんどすべて自分が望んでいた通りの動きをしています。
（DB処理等が終わる前に次のpushdataの受け取り）

かなり、申し訳ないソースコードでご対応いただいて本当にありがとうございました。

ちなみに、taskが二重のところですが、save_pushdataのコルーチンの中でmethod2のコルーチンがループしていて二重という意味でしょうか？
現状、問題なく動いていますが、やり方に問題がありますでしょうか？

[teamikl]

> 現状、問題なく動いていますが、やり方に問題がありますでしょうか？

どの様に修正されたのか把握してないので不明な部分です。

＞DB処理等が終わる前に次のpushdataの受け取り

を解決したのは、await self.save_pushdata() が処理をブロックしていた点ですね。

動いているが、冗長になってないか不安であれば、
質問のコードを修正して頂ければ確認できます。

想定した修正は、
- awaitself. save_pushdata(content) -> self.save_data(content)
- save_data 関数は async を外し、最後に return task

「２重に」というのは save_data 関数内で task を作るので、
その戻り値のtaskに再び loop.create_task を使う必要はないという点です。

[teamikl]

> task.add_done_callback(self.finish)のfinishの引数を外したらなぜか動き始めました・・・

future オブジェクトがコールバックの引数に渡されて
future.result() で結果を取り出しみたいな感じになってると思います。

add_done_callbackの時点で引数を渡したい場合は、
functools.partial も使えます
https://docs.python.org/3/library/asyncio-future.html#asyncio.Future.add_done_callback

とりあえずは、期待通りに動いたようなのは良かったですが、
エラーメッセージが表示されない点は気になりますね。

[taro_yamada]

最終のバージョンに直しました。
ついでに処理の管理をするためにsidを付けてみました。
ちなみに、save_data()はsave_pushdata()の誤りです。失礼しました。

さて、save_pushdata()もmethod2()もstrを戻していますが動いてしまっています・・・。add_done_callbackでfutureを戻しているからこれでOKということでしょうか？

>「２重に」というのは save_data 関数内で task を作るので、
>その戻り値のtaskに再び loop.create_task を使う必要はないという点です。
この文が少し難しくてわかりにくかったですが、
save_data内でmethod2のtaskを作る。
すなわち、save_dataの戻り値はsave_data内で宣言したtaskにするといけない
ということでいいですか？

ほんの少しずつですがasyncioの動きがわかってきたような気がします。とにかく難しいですね！

[taro_yamada]

ちなみに、エラーメッセージが表示されないのは、djangoでasyincioを使い始めたときからずっと表示されていないので、やはり、djangoとasyncioの関係性にあるのかもしれません。
かれこれ１か月くらいエラーメッセージなしにデバッグしています笑
改めて調べてみます。

[teamikl]

＞さて、save_pushdata()もmethod2()もstrを戻していますが動いてしまっています・・・。

エラーメッセージを得られないと、
編集してるコードと実行してるコードが違う可能性もあるので、
前提から見直した方が良さそうです。

＞新しい実行環境、新しいプロジェクトで djangoとasyncioを使った最小限のコードで試してみる、等。

私の視点からすると method3 の呼び出しも引数が違いエラーが出るはずなのに、
動いてしまうのも疑問なので、質問に掲載されたコードと実際のコードに大きな隔たりがありそうで、現状の提示されたコードでは正確な事が何も解りません。

２重に～というのは、
- stream 関数内で create_task
- save_pushdata 関数内でも create_task
という部分を指します。

[teamikl]

method3 の呼び出しに関しては

> TypeError: method3() takes 2 positional arguments but 3 were given
というエラーが出るはずです。

----
method2～method5 のコードに await を使う箇所がないのであれば、
非同期になってないので、タスクにする必要もないかもしれません。

コルーチン(method2) →create_task(save_pushdata内) → create_task(stream内) となっていますが、
タスク等の入れ物を余分に作ってるだけで、
実際のコードは同期的に実行されてます。
(await が時間の掛かる処理の待機から、他のコードの実行に処理を切り替えるタイミングなので、
await を使ってないコードでは、非同期の恩恵を付けられません）

追記: もしバックグラウンドで処理したいなら、
asyncioのタスクよりもconcurrent.future の方が適切です。
asyncioから使うなら、run_in_eecutor等。

DBへの保存部分もasyncioで処理したい場合は、
django の ORM が非同期処理に対応してるかどうかを調べると良いと思います。
(この辺りのdjangoの最近の情報には詳しくないので、言及のみ)

----
> かれこれ１か月くらいエラーメッセージなしにデバッグしています笑
> 改めて調べてみます。

ロギングの設定でエラー出力を無効にしてないか、等も調べてみてはどうでしょう。
https://docs.djangoproject.com/en/3.1/topics/logging/

一応、新規プロジェクトで何も特別な設定してない状態でも、
django command で asyncio を使ったコードでエラー表示は確認出来ました。

[taro_yamada]

methodの中で使っているのはkerasのpredict()とdjangoのORMとpandasとnumpyです。
どれもawaitを使えないので、concurrent.futureに切り替えたほうがいいのかもしれません。
ちなみに、これまでは、ORM→numpy、pandas→predict→ORMの処理をしている間にpushがtimeoutしてしまうといった状態だったのが、今回のawaitなしのasyncioで、logでsidの進行状況を見ると、save_dataやmethodのsidよりも先に次のデータ受信ができているように見えました。
これは、awaitで作ったsave_dataやmethodは、その下に非同期処理に対応したmethodがなくても一度上のネストに戻っていくということでしょうか？
もし、awaitの時点で順番を変えて受信だけ優先的にしてもらえらばいいという観点でいけば、最低限はクリアできていそうな気がしますが、concurrent.futureの方が並列処理の効率がいいということでよろしいでしょうか？
なお、method3の引数はソースを省略した際の転記ミスです。
色々と教えていただいてありがとうございます。助かっています。

[teamikl]

> concurrent.futureの方が並列処理の効率がいいということでよろしいでしょうか？

await のないコードの場合は、期待する並列処理にもなってません。
質問に提示されたコード内で非同期なのは await でデータ受信の箇所のみですが、
await のないコードを実行してる間は、コルーチンであったとしても
データ受信処理は止まってます。

現在のコードでは
1. データ受信 await ws.recv()
2. タスク生成
3. 2回目のデータ受信待機 await ws.recv()
4. タスク実行開始 (この間のデータ受信は中断されて出来ない)
なので、実質タスクを使わないで書いた場合と処理時間は変わらない
でコードのみ複雑になってる状態です。

優先的に処理という点は、タスクの実行時間が短い場合はそのように思うかもしれませんが、保証された動作ではありません。

コルーチンのコードは自動でバックグラウンド動作しているわけではなくて、
待ち時間が掛かる処理を await ～ に記述し、
await により適切に実行待ちの処理を切り替える必要があります。

恐らくここで望まれる動作は、データ受信後に即次のデータ受信を待機し、
バックグラウンドでDBへの保存だと思いますが、
await を使うコードがない場合は、
選択肢としては concurrent.future (asyncioから使う場合は run_in_executor) になると思います。

[taro_yamada]

即席でrun_in_executorのバージョンを作ってみました。
大体、こんなイメージでいいのでしょうか？

[teamikl]

＞大体、こんなイメージでいいのでしょうか？

恐らく期待する挙動になってないと思います。
（単体で実行可能なコードでないと、こちらでは動作確認できません）

エラーメッセージが読めれば気付ける事ばかりなので、
手探りでコードを書くよりも、先にそちらの問題を解決した方がよさそうです。

どうしても解決できない場合は、
django を使わずにコードを書いてテストして、
django command からはそのモジュールを呼び出すだけにするような工夫をして見て下さい。
(django に依存する部分を切り分けてモジュールを設計します）

----
＞loop.run_in_executor(executor, save_pushdata(content, sid), i)

このコードでは、save_pushdata を呼び出してから、
その戻り値を run_in_executor に渡してます。
run_in_executor で save_pushdata を実行したいはずなので、
関数と引数を別々にrun_in_executor に渡すようにしてください。

run_in_executor(executor, func(arg)) # 間違い
run_in_executor(executor, func, arg) # OK

https://docs.python.org/ja/3/library/asyncio-eventloop.html#asyncio.loop.run_in_executor

----
- queue や proc はwhileループの外側に出した方が良いです。
- asyncio.wait(tasks) は終了待機する為、次の await ws.recv() が実行されません。
タスク完了まで次のデータを読めないので、処理時間次第では一番最初にあった問題が再発すると思います。

[teamikl]

以前のコメントに誤解があるようなので、説明しておきます。

＞save_pushdata()もmethod2()もstrを戻していますが動いてしまっています・・・。

関数定義に async が付いてる場合 save_pushdata() の戻り値は
str ではなくコルーチン オブジェクトです。
create_task に str を渡してるわけではありません。

[taro_yamada]

ありがとうございました。
次のスレッドを立ててみたいと思います。
またよろしくお願いします。