Pythonでwebsocketのから逐次送ったデータをマルチプロセスで処理したいです

Question

実現したいこと

① websocketを使ってデータをリアルタイムで取得
② サブプロセスをマルチプロセスで実行
③ ①で取得したデータをpipeで②のサブプロセスに送って処理する

前提

PythonでBybitのAPIを叩いています。websocketを用いてデータを逐次取得しながら、サブプロセスで分析するようなコードを書こうとしています。
構成としては、websocketを動かすプロセスはデータを取得することだけをさせて、サブプロセスでは一定時間ごとにそのデータを加工して分析するというように、作業を二つのプロセスに分けようとしています。
とりあえず、全体の動きを見るために、websocketを動かすメインプロセスからpipeで 送られてきたデータを、サブプロセスでプリントすると言うだけのプログラムを書きました。
ここでは、サブプロセスでのデータのプリントを実現したいです。

プラットフォームはmacOSです。
本番環境はLinax（ubuntu）です。mac上でうまくいけば、Linax上でも大丈夫かなと考えていますが、安易でしょうか？

発生している問題・エラーメッセージ

特にエラーは発生していませんが、別プロセスからのデータの表示も全くされません。ps でプロセスIDを確認すると、プログラムが停止しているわけではなさそうでした。

該当のソースコード

import json
import websocket
from pprint import pprint
import multiprocessing as mp

class MyWS():
    def __init__(self):
        self.endpoint_public = 'wss://stream.bybit.com/v5/public/linear'
        self.resv_conn, self.send_conn = mp.Pipe()
        self.connect(self.endpoint_public)
        self.child = mp.Process(target=self.create_df, args=(self.resv_conn, ))
        self.child.start()
        self.child.join()

    def connect(self, endpoint):
        self.ws = websocket.WebSocketApp(url = endpoint,
                                    on_open = self.on_open,                                         
                                    on_message = self.on_message,
                                    on_error = self.on_error,
                                    on_close = self.on_close,)
        self.ws.run_forever()

    def on_message(self, ws, message):
        try:
            message = json.loads(message)
            if message.get('data'):
                for d in message.get('data'):
                    self.send_conn.send(d)  ### ここでデータをサブプロセスに送る。
            # pprint(message) ### A ###
        except Exception as e:
            self.on_error(ws, e)

    def on_open(self, ws):
        self.ws.send(json.dumps({"op":"subscribe", "args":["publicTrade.BTCUSDT"]}))
    
    def exit(self):
        self.ws.close()

    def on_error(self, ws, error):
        print(error)
        self.exit()

    def on_close(self, ws, close_status_code, close_msg):
        print("### closed ###")

    # データ処理用のサブプロセス
    def create_df(self, connection):
        while True:
            received_data = connection.recv()
            if received_data:
                print(received_data)


if __name__ == '__main__':  
    myws = MyWS()

試したこと

関数on_messageの中の「### A ###」の箇所を有効にすると、データは取得できています。
ですので、pipeでうまく送信できていないか、そもそもサブプロセスが動いていないのか、そのような問題だと思います。しかし、問題の切り分けができません。どなたか、websocketとマルチプロセスに詳しい方、教えていただけないでしょうか。

追記
コンストラクタ__init__()のコードの順序を変えました。

    def __init__(self):
        self.endpoint_public = 'wss://stream.bybit.com/v5/public/linear'
        self.resv_conn, self.send_conn = mp.Pipe()
        # 以下のようにコードの順序を変更
        self.child = mp.Process(target=self.create_df, args=(self.resv_conn, ))
        self.child.start()
        self.child.join()
        self.connect(self.endpoint_public)

Comments

[TakaiY]

プラットフォームはWindowsですか？ Linuxですか？ multiprocessの挙動が違うのでこの情報は重要です。

create_df(self, connection) メソッドのprintに flush=trueを指定してみるとどうなりますか？
また、whileループの前に 何かprintしてみたらそれは表示されますか？
https://docs.python.org/ja/3.7/library/functions.html#print

回答はここではなく、質問を編集して追記するといいでしょう。

[fu_3823]

コメントありがとうございます。
本文の方も加筆いたしましたが、こちらにも転記いたします。

プラットフォームはmacOSです。
本番環境はLinax（ubuntu）ですが、mac上でうまくいけば、Linax上でも大丈夫かなと考えていますが、安易でしょうか？

[fu_3823]

printに flush=trueを指定　については後ほど試して返信いたします。

[bsdfan]

self.connect(...) の中で、self.ws.run_forever() しているので、connect() からはもどってこないのではないでしょうか。だとすると、それより後ろのコードは実行されないのでサブプロセスは作られていないと思います。

[TakaiY]

ああ、たしかにそうですね。
子プロセス起動してから、connectですかね。

[fu_3823]

ありがとうございます。 コンストラクタのコードの順序をbsdfanのご指摘とおり変えました。
質問欄に追記した通りです。
また、TakaiY さんのおっしゃる通り、whileループの前に print("test") を置いたところ、コンストラクタのコードの順序変更後はサブプロセスが作成されたようで、うまく"test" と表示されました。
しかし、メインプロセスのwebsocket からのデータは表示されません。pipe経由で送れていないか、受け取れていないかどちらかのようです。
あと、printの引数、flush=True では目に見える変化はありませんでした。

[TakaiY]

join()は、生成した子プロセスの終了を待つ命令なので、joinまで入れてしまってはだめですね。

[fu_3823]

何度もご意見ありがとうございます。joinのドキュメントを再読して、もう一度改善にトライしてみます。

[fu_3823]

コードの順序とjoinの使い方を見直すことで解決しました。
ありがとうございました。
マルチプロセスの書き方のセオリーのようなものを知らないので、下手なロジックだったと思いますが、ご丁寧にありがとうございまいた。
BAしたいですが、コメント欄からの助けていただいた場合は自己解決で良いのでしょうか。内容を回答欄にコピーしていただいたら、BAにさせていただきたいです。

[TakaiY]

キーとなるコメントは bsdfanさんのものなので僕の出る幕ではないのですが、夕方くらいまで待って応答なければ、簡単な修正ポイントのまとめを書いて自己解決にしてしまっていいと思います。

[fu_3823]

ありがとうございます。
おっしゃる通りにさせて頂きたいと思います。

[bsdfan]

やったことをまとめて、自己解決にしていただくほうが、いい回答になると思いますので、お願いします。

[fu_3823]

わかりました。おふたりともありがとうございました。

Answer 1

コンストラクタを以下のように書き換えることで、うまく動作するようになりました。

    def __init__(self):
        self.endpoint_public = 'wss://stream.bybit.com/v5/public/linear'
        self.resv_conn, self.send_conn = mp.Pipe()
# 削除 -    self.connect(self.endpoint_public)
        self.child = mp.Process(target=self.create_df, args=(self.resv_conn, ))
        self.child.start()
# 削除 -   self.child.join()
# 追加 +   self.connect(self.endpoint_public)

問題点は、.run_forever()が実行されるタイミングがサブプロセスの作成・実行後にすることと、join()が不要だったことです。
サブプロセスは終了させず、メインプロセスからのデータを受け取り続けるように動かすことを意図していたので、サブロセスの終了を待つjoin()は不要でした。