Python: asyncioを用いて，複数クライアントからUDPデータを同時に受け取りたい

Question

前提

python3.9を用いて，UDP/TCPによるソケット通信を行っています．サーバは，複数のクライアントからUDPでデータを受け取り，そのデータを加工し，TCPでクライアントに返送します．

実現したいこと

複数のクライアントからのUDPのデータを一台のサーバで並列(非同期)で捌きたいと考えています．
その際に，asyncioでタスクを複数作り，同時にUDPデータをrecvfrom()したいと考えています

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

1つ目のクライアントからのデータは正常に受け取れますが，2つめのクライアントからのデータは受け取れず，recvfromがそのままタイムアウトしてしまいます．

ちなみに，asyncioは並列処理ではなく，シングルスレッドで動作していることは理解していますが，全く知見はありません．

該当のソースコード

10つのクライアントからUDPを受信しようと考えているため，def main()内のfor処理を10回繰り返し，10個のタスクを生成しています．

※主にソケット生成/受信/送信部分を見てもらえればよいと思います．

import socket
import asyncio


M_Group = "239.1.2.3"
LOCALHOST = socket.gethostbyname(socket.gethostname())
PORT = 1234


FILE_NAME = 'send.txt'

# UDPソケットの設定
udp_sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
udp_sock.setsockopt(socket.IPPROTO_IP, socket.IP_MULTICAST_IF, socket.inet_aton(LOCALHOST))
udp_sock.setsockopt(
    socket.IPPROTO_IP, socket.IP_ADD_MEMBERSHIP,
    socket.inet_aton(M_Group) + socket.inet_aton(LOCALHOST)
)
# 自分の送ったマルチキャストパケットを自身で受け取らない
udp_sock.setsockopt(socket.IPPROTO_IP, socket.IP_MULTICAST_LOOP, 0)
udp_sock.bind(('', PORT))
udp_sock.settimeout(5)

# udp_sock.connect((M_Group, PORT)) 

# TCPソケットの設定
tcp_sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)


file_data = ""
len_data = 0

with open(FILE_NAME, 'r') as f:
    file_data = f.read()
    len_data = len(file_data)


async def recv_loss_packet():    

    print("time to wait for response...")
    try:
        res_data, (src_ip, src_port) = udp_sock.recvfrom(1024)

        print(f'src_ip: {src_ip}, src_port: {src_port}')
        
        
        # packet loss seq split
        recv_seq_ary = res_data.decode().split(",")
        recv_seq_ary.pop(-1)

        recv_seq_ary = list(map(int, recv_seq_ary))
        print(recv_seq_ary)

        # packet loss data resend
        tcp_sock.connect((src_ip, src_port))
        
        for i in recv_seq_ary:
            chunk = file_data[i*10:i*10+10]
            chunk = str(i) + ":" + chunk
            print(chunk)
            
            tcp_sock.send(chunk.encode())
            
    except:
        print(f"Timeout!")


async def main():
    count = 0

    # データを指定バイトごとに分割
    chunks = [file_data[i:i+10] for i in range(0, len_data, 10)]

    for chunk in chunks:
        count+=1
        chunk = str(count) + ":" + chunk
        print(chunk)
   
        udp_sock.sendto(chunk.encode(), (M_Group, PORT))
        time.sleep(0.05)

    for i in range(10):
        asyncio.create_task(recv_loss_packet())

    await asyncio.sleep(10)

asyncio.run(main())

試したこと

"発生している問題"でも述べましたが，2つのクライアントで試してみたところ，1つめのクライアントからは受け取れて，2つめのクライアントからは，データを受け取れませんでした．

並列処理ではないため，2つのクライアントからの送信を，十分な間隔(2秒)を空けて送信しましたが結果は変わりませんでした．

Comments

[8524ba23]

ブロッキングソケットを利用しているのではないか？など詳細が不明なので
ソケット生成なども含めたコード部分、できれば現象が再現する最小限のコード全体を提示すると回答得られやすくなるかと思います。

[bsdfan]

udpで受けているのに、tcpで返そうとしていませんか？
そこで止まっていないでしょうか。

[Yochyee]

すいません．udpで受け取ったデータを加工をしてTCPで返信しています．
止まっているのは，udpのrecvfromの部分です．

[matukeso]

udp_sockをnonblockにした上、recv_loss_packetはasyncioのeventloopのadd_readerからのコールバックで呼ばれる必要があるでしょう。

[Yochyee]

@matukeso
どこでnonblockされているでしょうか？settimeoutの部分でしょうか？
settimeoutでコントロールが戻ってくるのに，わざわざadd_readerのコールバックを呼ぶ必要がないということですか？

[matukeso]

nonblockにしてないから、すべき、と言いました。settimeoutつけても、recfronはtimeoutになるまでpythonに制御が戻らないから意味がなくて、、add_readerで読めるタイミングを通知してもらう必要があるんでしょ。

[Yochyee]

@matukeso
ご提言感謝します！参考にさせていただきます。

Answer 1

udpのsocketをnon blockingにし，add_readerでイベントを登録することで実装できました．
ちなみに私の環境であるwindowsでは，asyncio.get_event_loop()が使えないようなので代替記法で書いています．

本質問では，UDPで受信した後，TCPで返送するプログラムでしたがTCP部分が正常に動かなかったため．TCPの部分は削除しました．

M_Group = "239.1.2.3"
LOCALHOST = socket.gethostbyname(socket.gethostname())
PORT = 1234

# UDPソケットの設定
udp_sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
udp_sock.setsockopt(socket.IPPROTO_IP, socket.IP_MULTICAST_IF, socket.inet_aton(LOCALHOST))
udp_sock.setsockopt(
    socket.IPPROTO_IP, socket.IP_ADD_MEMBERSHIP,
    socket.inet_aton(M_Group) + socket.inet_aton(LOCALHOST)
)
# 自分の送ったマルチキャストパケットを自身で受け取らない
udp_sock.setsockopt(socket.IPPROTO_IP, socket.IP_MULTICAST_LOOP, 0)
udp_sock.bind(('', PORT))

# ノンブロッキングに指定
udp_sock.setblocking(False)


FILE_NAME = 'send.txt'
file_data = ""
len_data = 0

with open(FILE_NAME, 'r') as f:
    file_data = f.read()
    len_data = len(file_data)


def recv_loss_packet():    

    print("time to wait for response...")
    try:
        res_data, (src_ip, src_port) = udp_sock.recvfrom(1024)

        print(f'src_ip: {src_ip}, src_port: {src_port}')
            
    except:
        print(f"Timeout!")


count = 0

# データを指定バイトごとに分割
chunks = [file_data[i:i+10] for i in range(0, len_data, 10)]

for chunk in chunks:
    count+=1
    chunk = str(count) + ":" + chunk
    print(chunk)
   
    udp_sock.sendto(chunk.encode(), (M_Group, PORT))
    time.sleep(0.05)


# イベントループの取得 & セット
# 以下3行は，windows以外なら，asyncio.get_event_loop() のみで良い
selector = selectors.SelectSelector()
loop = asyncio.SelectorEventLoop(selector)
asyncio.set_event_loop(loop)

# udp_sockのファイルディスクリプタが空いたら，紐づけられた関数を実行
loop.add_reader(udp_sock.fileno(), recv_loss_packet)

# ループを回す
try:
   loop.run_forever()
finally:
　loop.close()