マルチプロセスアプリケーション内部通信のため、asyncioを用いたcom server/clientを作成しています。
ソースコードでは、com_serverは受け取ったデータをそのまま返送するだけです。
試行回数(app.pyのrcnt値)200~2000と幅があるですが、com serverがinvalid_stateの例外エラーを送出して停止してしまいます。
com_serverのstart_serverメソッド内のexcept 例外送出部分において、「invalid state」が送出されますが、それ以外の情報がなく、どの部分でinvalid stateなのかが判断がつきません。
さらに、例外発生時のfinally節において self.event_loop.close()
の部分で処理が戻ってこなくなります。
self.event_loop.close()
部分から処理を戻す方法があれば、com_serverを再起動できるのですが。
なぜ停止してしまうのか、ご教示をお願いします。
環境
OS:Windows10 Pro(64bit)
Python 3.9
IDE:VSCode
※2台で実施しましたが、結果は同じでした。
追加情報
TCPコネクションリソースを疑っていましたが、ループ試行回数に大きなばらつきがあり、おそらくリソース問題ではないだろうと考えています。
確認方法
1)管理ツール:リソースモニターを起動し、TCP接続を確認。
2)Powershell において、2秒ごとに以下のコマンドを実行・表示
(& netstat -n -b | Select-String "127.0.0.1:5638").Length
この結果により、上限100-104で安定します。
ソースコード
com_server
import asyncio
from queue import Queue
from threading import Thread
import time
srv_addr = '127.0.0.1'
srv_port = 5638
class com_server_protcol(asyncio.Protocol):
def __init__(self, send_queue):
self.transport = None
self.send_queue = send_queue
super().__init__()
return
def connection_made(self, transport):
try:
self.transport = transport
client_address, client_port = self.transport.get_extra_info(
'peername')
except Exception as e:
print(f'com_server_protcol:{e}')
# ホストマシン以外の接続は拒否
if client_address != srv_addr:
self.transport.close()
def data_received(self, data):
self.send_queue.put(data)
def connection_lost(self, exc):
self.transport.close()
class com_server():
def com_thread(self):
timer = None
time_out = 3.0
try:
while (self.is_ThrLoop):
time.sleep(0.03)
while (not self.con_recv_queue.empty()):
item = self.con_recv_queue.get()
self.instProt.transport.write(item)
self.instProt.transport.close()
except Exception as e:
print(f'com_thread:{e}')
return
def start_server(self):
self.Thr = Thread(name='com_server_thread', target=self.com_thread)
self.Thr.daemon = True
self.Thr.start()
self.event_loop = asyncio.get_event_loop()
self.instProt = com_server_protcol(self.con_recv_queue)
factory = self.event_loop.create_server(
lambda: self.instProt, srv_addr, srv_port)
self.server = self.event_loop.run_until_complete(factory)
try:
self.event_loop.run_forever()
except Exception as e:
print(f'com_server:{e}')
finally:
self.is_ThrLoop = False
self.Thr.join()
self.server.close()
self.event_loop.run_until_complete(self.server.wait_closed())
self.event_loop.stop()
self.event_loop.close()
return
def run(self, recv_queue, send_queue):
self.is_ThrLoop = True
self.recv_queue, self.send_queue = recv_queue, send_queue
self.con_recv_queue = Queue()
self.start_server()
return
com_client
import asyncio
srv_addr = '127.0.0.1'
srv_port = 5638
async def client_message(message):
reader, writer = await asyncio.open_connection(srv_addr, srv_port)
writer.write(message.encode())
await writer.drain()
data = await reader.read()
writer.close()
await writer.wait_closed()
return data
def send_message(message):
try:
recv_data = asyncio.run(client_message(message))
except Exception as e:
print('client_message:', e)
return recv_data
app
from multiprocessing import freeze_support, set_start_method, Process
from multiprocessing import Queue as MP_Queue
import json
import os
import sys
import time
import com_server as com # nopep8
from com_client import * # nopep8
class AppServer():
def run(self):
self.work_dict = dict()
self.com_send_queue, self.com_recv_queue = MP_Queue(), MP_Queue()
instCom = com.com_server()
instComProcess = Process(
name='ComServer',
target=instCom.run,
args=(self.com_send_queue, self.com_recv_queue)
)
instComProcess.start()
while (True):
time.sleep(0.05)
if __name__ == '__main__':
freeze_support()
set_start_method('spawn')
instServer = AppServer()
instProcess = Process(
name='ComServer',
target=instServer.run,
)
instProcess.start()
time.sleep(3)
counter = 0
interval = 0.05
const = 1
rcnt = 0
def _session_test_():
global rcnt
global counter
counter += 1
if int(counter) == int(const/0.05):
counter = 0
rcnt += 1
try:
res = send_message(json.dumps({'GET': None}))
except Exception as e:
ptint(f'recv{rcnt}:{e}')
print(rcnt)
return
while (True):
time.sleep(interval)
_session_test_()