91230


マルチプロセスアプリケーション内部通信のため、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コネクションリソースを疑っていましたが、ループ試行回数に大きなばらつきがあり、おそらくリソース問題ではないだろうと考えています。

## 確認方法

１）管理ツール：リソースモニターを起動し、TCP接続を確認。  
２）Powershell において、2秒ごとに以下のコマンドを実行・表示

[code]

    (& netstat -n -b | Select-String "127.0.0.1:5638").Length
    
[/code]

この結果により、上限100-104で安定します。

# ソースコード

## com_server

[code]

    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
    
[/code]

## com_client

[code]

    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
    
[/code]

## app

[code]

    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_()
    
[/code]


CC BY-SA 4.0
2022-09-12T05:55:48.350
91049
2022-09-22T04:47:09.133
2022-09-13T02:56:33.973
32891
32891
question
python python3 非同期
asyncioを用いたサーバーがinvalid_stateで停止してしまう


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

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

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

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

以下のドキュメント参照。  
[イベントループ](https://docs.python.org/ja/3/library/asyncio-eventloop.html)

## com_server

[code]

    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))
    
    
[/code]

## com_client

変更なしなので省略

## app

[code]

    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)
    
    
[/code]


CC BY-SA 4.0
2022-09-22T04:47:09.133
91230
2022-09-22T04:47:09.133
32891
91049
answer