python 3.7 websocket使用時におけるマルチプロセス処理のエラー

Question

websocketを使用しながらマルチプロセスでサブプロセスを動作させておきメモリ空間を共有させたいと思っているのですが、サブプロセスのルーチンが早々に停止してしまいます。
マルチプロセスだけなら走るのですがwebsocketを組み合わせるとうまく動かず困っています。

エラーメッセージ

該当のソースコード

# websocket
class WebsocketAPI(object):
    def __init__():

    def run():

    def on_message():

    def on_close():

    def on_open():

# サブプロセス
def test_func(d):
    print("in test_func started")
    d[1] = 200
    print(d[1])
    d["test"] = "test1"
    print(d["test"])
    while True:
        print(d)
        sleep(0.5)


if __name__ == '__main__':

    while True:
        try:
            with Manager() as manager:
                d = manager.dict()
                test_process = Process(target=test_func, args=(d))
                print("start")
                test_process.start()
                print("finished")
                break
        except ccxt.BaseError as e:
            print(" error in SubProcess : test_func()",e)
            print("recall")
            sleep(1)

    #API endpoint
    url, channel, json_rpc = make_connection()
    #ctrl + cで終了
    while True:
        try:
            json_rpc.run()
            break
        except ccxt.BaseError as e:
            print(" error in _MAIN_ json_rpc.run()", e)
            url, channel, json_rpc = make_connection()

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

(base) C:\test>python websocket.py
start
finished
called ＿init＿
websocket.WebSocketApp open 正常終了
json_rpc :  <__main__.RealtimeAPI object at 0x0000023E460B6D68>
--- request header ---

-----------------------
--- response header ---

-----------------------
called on_open
connected streaming server
send: b'\x81\xd1\x8b\xd1t\xbf\xf0\xf3\x19\xda\xff\xb9\x1b\xdb\xa9\xebT\x9d\xf8\xa4\x16\xcc\xe8\xa3\x1d\xdd\xee\xf3X\x9f\xa9\xa1\x15\xcd\xea\xbc\x07\x9d\xb1\xf1\x0f\x9d\xe8\xb9\x15\xd1\xe5\xb4\x18\x9d\xb1\xf1V\xd3\xe2\xb6\x1c\xcb\xe5\xb8\x1a\xd8\xd4\xb4\x0c\xda\xe8\xa4\x00\xd6\xe4\xbf\x07\xe0\xcd\x89+\xfd\xdf\x92+\xf5\xdb\x88V\xc2\xf6'

～～～～～～～～～～～～
～～ websocket 処理～～
～～～～～～～～～～～～


～～～～～～～～～～～サブプロセスエラー～～～～～～～～～～～～～～～～
Process Process-2:
Traceback (most recent call last):
  File "C:\Anaconda3_2\lib\multiprocessing\process.py", line 297, in _bootstrap
    self.run()
  File "C:\Anaconda3_2\lib\multiprocessing\process.py", line 99, in run
    self._target(*self._args, **self._kwargs)
TypeError: test_func() missing 1 required positional argument: 'd'

～～～～～～～～～～～～～～～～～～～～～～～～～～～～～～～～～～～～

～～～～～～～～～～～～
～～ websocket 処理～～
～～～～～～～～～～～～

サブプロセスについては最初のプリント文すら表示されません。
何が原因かわかれば教えてください。よろしくお願いします。

追記

初期エラーが改善され、次のエラーが生じたため編集コードとエラーを抜粋して追記します。
またサブプロセス１行目のプリント文までは出力されるようになりました。
引き続きよろしくお願いします。

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

～～～～～～～～～～～サブプロセスエラー～～～～～～～～～～～～～～～～
in test_func started
Process Process-2:
Traceback (most recent call last):
  File "C:\Anaconda3_2\lib\multiprocessing\managers.py", line 788, in _callmethod
    conn = self._tls.connection
AttributeError: 'ForkAwareLocal' object has no attribute 'connection'

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
  File "C:\Anaconda3_2\lib\multiprocessing\process.py", line 297, in _bootstrap
    self.run()
  File "C:\Anaconda3_2\lib\multiprocessing\process.py", line 99, in run
    self._target(*self._args, **self._kwargs)
  File "C:\test\websocket.py", line 563, in test_func
    d[1] = 200
  File "<string>", line 2, in __setitem__
  File "C:\Anaconda3_2\lib\multiprocessing\managers.py", line 792, in _callmethod
    self._connect()
  File "C:\Anaconda3_2\lib\multiprocessing\managers.py", line 779, in _connect
    conn = self._Client(self._token.address, authkey=self._authkey)
  File "C:\Anaconda3_2\lib\multiprocessing\connection.py", line 490, in Client
    c = PipeClient(address)
  File "C:\Anaconda3_2\lib\multiprocessing\connection.py", line 691, in PipeClient
    _winapi.WaitNamedPipe(address, 1000)
FileNotFoundError: [WinError 2] 指定されたファイルが見つかりません。

～～～～～～～～～～～～～～～～～～～～～～～～～～～～～～～～～～～～

該当のソースコード

# サブプロセス
# 変数を２つに修正
def test_func(d,l):
    print("in test_func")
    d[1] = 200
    print(d[1])
    d["test"] = "test1"
    print(d["test"])
    while True:
        print(d)
        print(l)
        sleep(0.5)

if __name__ == '__main__':

    while True:
        try:
            with Manager() as manager:
                d = manager.dict()
                ### 変数 l を追加 ###
                l = manager.list(range(5))
                test_process = Process(target=test_func, args=(d,l))
                print("start")
                test_process.start()
                print("finished")
                break
        except ccxt.BaseError as e:
            print(" error in SubProcess : test_func()",e)
            print("recall")
            sleep(1)

    #API endpoint
    url, channel, json_rpc = make_connection()
    #ctrl + cで終了
    while True:
        try:
            json_rpc.run()
            break
        except ccxt.BaseError as e:
            print(" error in _MAIN_ json_rpc.run()", e)
            url, channel, json_rpc = make_connection()

追記２

※抜けていたプログラムコードを追記します。

該当のソースコード２

# API 接続
def make_connection():
    #API endpoint
    url = 'URL'
    channel = '希望するAPI情報の種類(str)'
    json_rpc = WebsocketAPI(url=url, channel=channel)
    print("json_rpc : ", json_rpc)
    return url, channel, json_rpc

Answer 1

                test_process = Process(target=test_func, args=(d))

は

                test_process = Process(target=test_func, args=(d,))

にしないとtupleとみなされません。

他にもまずいことがあるかもしれませんが、ぱっと見て気づいた点だけ。

---

以下解決方法の要約。詳細はコメント欄参照。

AttributeError: 'ForkAwareLocal' object has no attribute 'connection'の直接の原因は、Managerのインスタンス生成をwith文で行っており、ブロックを抜けたあとにも使用しようとしていたため。withのブロックを抜けるときに閉じられてしまうので、その後にManagerの機能を使おうとしてもエラーになる。

Comments

[wiz_planet]

ありがとうございます。
リストを追加して変数を二つにして対応してみました。

[wiz_planet]

そして先ほどのエラーは解消され次は別のエラーが出ました。。

[hayataka2049]

修正後のコードとエラーを質問文に追記してみてください。

[wiz_planet]

追記してみました。。
ファイルを指定した覚えもありませんし、このエラーがどういうことなのかよくわかりませんでした。

[hayataka2049]

前の内容は残した上で「追記」にしてください。どういう流れなのかわけわからなくなるので。

[hayataka2049]

メインプロセスが死ぬと問題になるのではないでしょうか。
https://stackoverflow.com/questions/25455462/share-list-between-process-in-python-server
子が終わるまで待つ必要があるかと。joinを呼ぶか、単にsleepでもさせるか・・・

[wiz_planet]

すみません、内容を修正しました。。

[wiz_planet]

メインが死ぬとはどういうことでしょうか？
websocketのことでしょうか？

[wiz_planet]

メインはjson_rpc.run()で回り続けているんじゃないでしょうか？
サブはループで回し続けるのでjoinで待つと json_rpc.run()までたどり着けなくなります。

[wiz_planet]

while True:
try:
with Manager() as manager:
d = manager.dict()
l = manager.list(range(5))
test_process = Process(target=test_func, args=(d,l))
print("start")
test_process.start()
test_process.join()
print("finished")
break

[hayataka2049]

＞メインはjson_rpc.run()で回り続けているんじゃないでしょうか？
と言われても、私にこれらが何者なのかわかる情報は与えられていませんので・・・

[wiz_planet]

上記の箇所にjoinを挿入してみたらサブは問題なく回りました。
これをメインと同時にやりたいです。。

[hayataka2049]

下のwhileを抜けたところでjoinしてみては？

[wiz_planet]

すみません、情報不足でしたね。。
json_rpc.run()を回し続けつつ、サブも回し続けたいです。
今はサブが回り続けるようになりました。

[wiz_planet]

このプログラムでは下のwhileは抜けない仕様になっています。
このwhileのjson_rpc.run()で取得し続けるAPI情報を加工したデータをプリントし続けるプログラムとなっています。

[hayataka2049]

・make_connectionという関数の仕様
・変数json_rpcとして受け取っている返り値のクラスの仕様
の2点は示していただけませんか？　それがないと適切に助言することは難しいです。

[wiz_planet]

だいぶ端折って書いたので抜けてるところがあるのですが、
json_rpc.run()はclass WebsocketAPI(object)のon_message()関数を実行し続ける仕組みになってます。

[wiz_planet]

追記しますのでちょっと待ってください。

[wiz_planet]

make_connection()追記しておきました。

[hayataka2049]

了解です。要するにWebsocketAPIのインスタンスを返すだけということですね。

[wiz_planet]

APIの返り値のクラスは list になります。

[hayataka2049]

はい、url, channel, json_rpcを返すということですね。

見落としていたのですが（すみません）、withでManagerを作るとwithを抜けた瞬間にクローズされますので、こちらが原因かもしれません。使わない方法で書き直してみてください（manager = Manager()に置き換え）

[wiz_planet]

そのようだと思います。
ただwebsocketは受け取り方がややこしいため、まだあまり深く理解はできていないところです。。

[wiz_planet]

いけました！！
いやはや、お付き合いいただき本当にありがとうございます。

[wiz_planet]

このような書き方があることすら知りませんでした。。
withだとクローズされるんですね。
助かりました。ありがとうございます。

[hayataka2049]

「開始処理と終了処理が必要なリソースなどを特定のブロック内でだけ扱う」のがwith文の機能ですね。抜けたあとはもう使い物にならない訳です。
https://www.python-izm.com/advanced/with/

# 違うこと疑ってたせいもあるけど気づくのに時間がかかったし、ちょっと体調が悪いですね。遅くなってすみませんでした。

[wiz_planet]

いえ、こんなに一発で解決できるとは全く思っておりませんでしたので自分でも驚きました！
リンクまで貼っていただきありがとうございます。
またコツコツやって行こうと思います。。