0

マルチプロセスアプリケーション内部通信のため、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_()

1 件の回答 1

0

停止する原因は不明ですが、get_event_loopをget_running_loopとした方法で解決しました。

Pythonドキュメントのイベントループget_event_loopに関して、

「この関数の振る舞いは (特にイベントループポリシーをカスタマイズした場合) 複雑なため、コルーチンやコールバックでは get_event_loop() よりも get_running_loop() を使うほうが好ましいと考えられます。
また、低水準の関数を使って手作業でイベントループの管理をするかわりに、 asyncio.run() を使うことを検討してください。」

という記述を見つけ、これに対応するように変更しました。

以下のドキュメント参照。
イベントループ

com_server


class com_server_protcol(asyncio.Protocol):
    def __init__(self, recv_queue, send_queue):
        self.transport = None
        self.recv_queue = recv_queue
        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):
        time_out = 3.0
        interval = 0.03
        isLoop = True

        try:
            self.send_queue.put(data)
            timer = time.time()

            while (isLoop):
                time.sleep(interval)
                while (not self.recv_queue.empty()):
                    item = self.recv_queue.get()
                    self.transport.write(item)
                    self.transport.close()
                    isLoop = False
                    timer = None

                if timer is not None:
                    if time.time() - timer > time_out:
                        self.transport.close()
                        isLoop = False

        except Exception as e:
            print(f'com_server_protcol:{e}')

    def connection_lost(self, exc):
        self.transport.close()


async def com_server(recv_queue, send_queue):
    loop = asyncio.get_running_loop()

    server = await loop.create_server(
        lambda: com_server_protcol(recv_queue, send_queue),
        srv_addr, srv_port)

    async with server:
        await server.serve_forever()


def server_run(recv_queue, send_queue):
    asyncio.run(com_server(recv_queue, send_queue))

com_client

変更なしなので省略

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


if __name__ == '__main__':
    freeze_support()
    set_start_method('spawn')

    recv_queue, send_queue = MP_Queue(), MP_Queue()

    instProcess = Process(
        name='ComServer',
        target=com.server_run,
        args=[
            send_queue, recv_queue
        ]
    )
    instProcess.start()

    def client_thread():
        time.sleep(3)

        interval = 0.05
        const = 1
        counter = 0
        rcnt = 0

        while (True):
            time.sleep(interval)

            counter += 1
            if int(counter) == int(const/interval):
                counter = 0
                rcnt += 1
                try:
                    res = com.send_message(json.dumps({'GET': None}))
                except Exception as e:
                    print(f'recv{rcnt}:{e}')
                if rcnt % 20 == 0:
                    dt = datetime.datetime.now()
                    print(dt, rcnt)
        return

    client_thread = Thread(name='com_server_thread', target=client_thread)
    client_thread.daemon = True
    client_thread.start()

    while (True):
        time.sleep(0.05)
        while (not recv_queue.empty()):
            item = recv_queue.get()
            send_queue.put(item)

この質問に回答するには、ログインする必要があります。

求めていた回答ではありませんか? のタグが付いた他の質問を参照する。