初質問です pythonでのsocket通信でクライアントを同時に処理開始させたい

Question

前提

現在サーバー側Windows搭載のPC
クライアント側ラズパイ４が3台
このような構成になっています

実現したいこと

実現したことしては

1. サーバで値を入力する。
   2）3台のクライアントに同じ値を同時に送信
2. 3台のクライアントで処理を実施。
3. すべての結果を収集。
4. 1に戻る

最終的には、PC(サーバー)からの指令が来るたびにラズパイ(クライアント)が同時にプログラムを実行し
実行結果をサーバに返したいです

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

エラーメッセージ

現在のプログラムだと一度通信を切らないと、サーバー側の数字が更新できない
whileループの中に入れると、複数台同時に送信できない、あるいは
Bad adressといったエラーがでる

該当のソースコード

サーバー側

import socket
import threading
import time

IPADDR = ""
PORT =
sock_sv = socket.socket(socket.AF_INET)
sock_sv.bind((IPADDR, PORT))
sock_sv.listen(3)

def recv_client(sock, addr):
    try:      
        sock.send(signal.encode("utf-8"))
        data = sock.recv(4096)
        time.sleep(2)
        print(data)
        
    except ConnectionResetError:
        pass
    
while True:
    signal=input("signal=")

    sock_cl1, addr1 = sock_sv.accept()
    sock_cl2, addr2 = sock_sv.accept()
    sock_cl3, addr3 = sock_sv.accept()

    thread1 = threading.Thread(target=recv_client, args=(sock_cl1, addr1))
    thread2 = threading.Thread(target=recv_client, args=(sock_cl2, addr2))
    thread3 = threading.Thread(target=recv_client, args=(sock_cl3, addr3))
    # スレッド処理開始
    thread1.start()
    thread2.start()
    thread3.start()
    #スレッド終了待ち合わせ
    thread1.join()
    thread2.join()
    thread3.join()    
       

クライアント側

import socket
import queue
import threading
import numpy as np
import time

IPADDR = ""
PORT =
sock =socket.socket(socket.AF_INET)
sock.connect((IPADDR,PORT))
signal=0
q= queue.Queue()

def recv_data(sock):
    while True:
        try:
            signal_data = sock.recv(1024)
            print(signal_data.decode("utf-8"))
            q.put(signal_data)
        except ConnectionResetError:
            break
    sock.shutdown(socket.SHUT_RDWR)
    sock.close()
  
thread = threading.Thread(target=recv_data, args=(sock,))
thread.start()

while True:
    signal=q.get()
    signal=float(signal)
    try:
        if signal > 0:
            ser = serial.Serial("/dev/ttyS0", 38400,bytesize=7)
            msg =ser.readline()
            msg = msg.decode()
            tok = msg.split(',')
            S_data=tok[1]+","+tok[2]+","+tok[4]
            S_data=str(S_data)
　　　　S_data=S_data+"signal= "+str(signal)
            sock.send(S_data.encode("utf-8"))

        else:  
            data=("error")
            sock.send(data.encode("utf-8"))
　　except ConnectionResetError:
        　break 
        
sock.shutdown(socket.SHUT_RDWR)
sock.close()

試したこと

ループ外にinputを設けるとエラーは起きないが、数字が更新できない
thread.start()の上にinputを設けると個々に数字を送るので同時に実行できない

補足情報（FW/ツールのバージョンなど）

ここにより詳細な情報を記載してください。

Comments

[TakaiY]

実現したい処理の流れがよくわかりません。 特にサーバ(PC)側での値の入力。
1) サーバで値を入力する。
2) 3台のクライアントで処理を実施。
3) すべての結果を収集。
4) 1に戻る
でしょうか？ それとも入力はクライアントごとに行ないたいのでしょうか？

[Ayatakamaru]

処理の流れはTakaiYさんのおっしゃる通りです
入力は同じものを全クライアントを同時に送信したいです

[退会済みユーザー]

socket.ioのタグは外した方がいいです。あと初質問じゃないですよね？
前の同じような質問が削除申請中みたいです。

[退会済みユーザー]

クライアント3台から接続した後、例えばサーバー側から開始コマンドを決めて送ればいいかと思います。
クライアント側は送られたコマンドが開始コマンドなら処理を開始する、とかでしょうか・・・

サーバー側はacceptするスレッドを作成し、接続したソケットをリストか何かに並べておき、メインスレッドで開始の入力待ちをします。入力されたらスレッドを起動して接続したソケットの数分起動して、それぞれのスレッドで開始コマンドを送る、といった感じです。クライアントは今のままでいいでしょう。

[Ayatakamaru]

すみません，前回の質問は編集方法が分からなかったので，新しく上げなおしたので全く同じものです
実現したいのはおっしゃる通りの手法です．クライアントの数だけスレッドを用意するイメージですか？
あとメインループではなく，新しくスレッドで分けるってことですか？

[退会済みユーザー]

> クライアントの数だけスレッドを用意するイメージですか？
私が言ったのは(メインスレッドの他に)「クライアントの数＋１」ですね

> 新しくスレッドで分けるってことですか？
多分Yesです

inputより前にacceptすることで、TCPの3Way Handshakeを事前に済ませています。1パケット届けばクライアントは処理を開始できます。

[68user]

スレッドでもいいですが、
https://docs.python.org/ja/3/library/selectors.html
https://docs.python.org/ja/3/library/select.html
などを使うことで、1スレッドでソケットや標準入力を一度に扱う手もあります。

[Ayatakamaru]

>>dameo
acceptを別スレッドを移行せずに無事に同時に処理を開始することができましたが，
「inputより前にacceptすることで、TCPの3Way Handshakeを事前に済ませています。1パケット届けばクライアントは処理を開始できます。」
これをする利点がつかみ切れていません，知識不足で申し訳ないのですが，ご教授してもらえると助かります．
クライアント側からの送信を接続を切らずに止めたいのですが，上記のことで解決できますでしょうか？

[退会済みユーザー]

利点は「同時に送信」の精度を上げる目的です。必要な精度が分からなかったので。
「クライアント側からの送信を接続を切らずに止めたい」はちょっと意味が分かりません。

[Ayatakamaru]

現在稼働しているときは，
1)サーバで値を入力する。
2）3台のクライアントに同じ値を同時に送信
3) 3台のクライアントで処理を実施。
4) すべての結果を収集。
5) 1に戻る

1に戻れているのですが，クライアントから何度も結果が送られているので
サーバ側から信号があれば，クライアントから一度(あるいは一定時間)のみ返事が来るようにしたいです

プログラムを再起動とかは避けたいです

[退会済みユーザー]

すみません。プロトコルを決めれば済む話かと思いますが、それは別の質問ですね。
お手数ですが、別質問としてあげて他の方に聞いてください。

[TakaiY]

「クライアントから何度も結果が送られている」は、何が送られてきているんでしょうか？

[Ayatakamaru]

singal=1として入力するとsignal=1を利用した処理結果が送られ続けます

signalは新しく入力を要求していますが，その間も結果が流れているので，信号入力毎に一度の処理といった形を実現したいです．

[Ayatakamaru]

>>68user
ありがとうございます．スレッド処理よりも良い点が分かれば使ってみます

[退会済みユーザー]

二度目ですが、当初の質問の回答は得られているので、ベストアンサーを選び、別の質問にしてください。

[Ayatakamaru]

分かりました，ありがとうございます

Answer 1

何ができていて何ができていないのか、提示されているコードではわからないので、外しているかもしれません。

サーバ側のループ処理はこういう感じになるはずです。
入力してスレッドを起動して終了待ち、を繰り返す。

while true:
    # 入力を求める。
    signal=input("New signal>> ")
    # スレッドをクライアント毎に起動する。
    # thread生成は略
    thread1.start()
    thread2.start()
    thread3.start()
    # thread終了をまちあわせ
    thread1.join()
    thread2.join()
    thread3.join()
    # 必要であれば以下でthread結果の処理  

で、現状のコードでは受信処理が無限ループになっています。これではthreadが終らなくて次に進めません。また、この中でソケットを止めてしまっては通信が継続しないのでそれも削除します。
で、以下のような感じでしょうか。(試していないのでバグあるはず。)
単にデータを送って結果を受け取ったら終了する感じ。

def recv_client(sock, addr):
    try:
        sock.send(signal.encode("utf-8"))
        data = sock.recv(1024)
        print(data )            
    except ConnectionResetError:
        pass # 未検討

ちなみに、thread.Threadを使うとあまりスマートなコードになりません。(start()が並ぶような感じ)。
最近の、concurrent.futuerあたりを使うとスッキリ書けます。

Comments

[Ayatakamaru]

知識不足で本当にに申し訳ないのですが，
sock_c1, addr1 = sock_sv.accept()
thread1 = threading.Thread(target=recv_client, args=(sock_cl1, addr1))
sock_c2, addr2 = sock_sv.accept()
thread2 = threading.Thread(target=recv_client, args=(sock_cl2, addr2))
であっていますか？

主な変更点としては，クライアント毎にスレッドの用意ってことですか？
試してみたとことjoinで終了した後，signalの再入力は求められますが，クライアントから結果が送られ続けます
ここは止めれないでしょうか？
concurrent.futuerのアドバイスありがとうございます。あまりにも散らかってきたら勉強してみます

[TakaiY]

通信をどのように制御するかはシステム設計によるのですが、今の形だと、ループの中でなくて、事前に3台分の接続をしておいて、メッセージ(signalとdata)のみ送受信するかたちになると思います。

> クライアントから結果が送られ続けます
全部理解しながら見てはいませんが、サーバとクライアントのやりとりがうまくいっていないのであれば、まずは3台同時にとかでなく、1対1での通信ができるようにするのが先ではないでしょうか。

- ソケットの通信の開始はどの単位で実施するのか。
入力を受けてから socket 生成 -> signal送信 -> クライアントで処理 -> data返信 -> ソケット切断 (繰り返し)
socket 生成 -> 入力 -> signal送信 -> クライアントで処理 -> data返信 (繰り返し or 何らかの切っ掛けで停止) -> ソケット切断

[Ayatakamaru]

返事遅くなりました，申し訳ございません
やり取りはできているのですが，クライアントからのデータ送信がサーバ側で止めれないです

① socket 生成 -> signal送信 -> クライアントで処理 -> data返信 -> ソケット切断 (繰り返し)
この場合はサーバ側クライアント側を毎回プログラムを再起動して接続するってことですか？

②socket 生成 -> 入力 -> signal送信 -> クライアントで処理 -> data返信 (繰り返し or 何らかの切っ掛けで停止) -> ソケット切断
こちらが理想に近いです，接続を保ったまま送信の停止開始を操作したいです

[TakaiY]

「クライアントからのデータ送信がサーバ側で止めれない」とのことですが、クライアントから何が送られてくるのですか？現状のソースであれば、サーバ側が何も送信しなければ、 q.get() で 止ってくれると思いますけど。

[TakaiY]

「singal=1として入力するとsignal=1を利用した処理結果が送られ続けます」とのことですが、
- 質問を編集して、現状のソースコードを追加していただけますか。古いソースで話をしても意味がないので。
このとき、クライアント側も現状のものに更新してください。
- 送られ続けているのは、どうやって確認していますか？

[Ayatakamaru]

シグナルを入力後も再入力せずに一度入力したシグナルでの結果をかえし続けています

[TakaiY]

「かえし続けています」それはどうやって確認していますか？

[Ayatakamaru]

クライアントで受信したsignalをサーバに送り返しています

[Ayatakamaru]

ループのところを最新版にすると、返しつつづけることはなくなりましたが、新しくシグナルを入力してもクライアント側で受信できていません。エラーも出ずに止まっています