質問編集履歴

threadingでP2P通信を実現しようとして失敗した際のコードを追記

2020/01/03 07:45

投稿

attache-case

スコア6

test CHANGED Viewed

File without changes

test CHANGED Viewed

@@ -180,35 +180,7 @@
 Send: '15: Hello, World!\n'
-Send: '16: Hello, World!\n'
-Send: '17: Hello, World!\n'
-Send: '18: Hello, World!\n'
-Send: '19: Hello, World!\n'
-Send: '6: Hello, World!\n'
+...（略）...
-Send: '2: Hello, World!\n'
-Send: '4: Hello, World!\n'
-Send: '7: Hello, World!\n'
-Send: '1: Hello, World!\n'
-Send: '8: Hello, World!\n'
-Send: '9: Hello, World!\n'
-Send: '10: Hello, World!\n'
-Send: '5: Hello, World!\n'
-Send: '11: Hello, World!\n'
-Send: '12: Hello, World!\n'
 Received: '13: Hello, World!\n'
@@ -230,65 +202,7 @@
 Close the connection
-Received: '16: Hello, World!\n'
-Close the connection
+...（略）...
-Received: '17: Hello, World!\n'
-Close the connection
-Received: '18: Hello, World!\n'
-Close the connection
-Received: '19: Hello, World!\n'
-Close the connection
-Received: '6: Hello, World!\n'
-Close the connection
-Received: '2: Hello, World!\n'
-Close the connection
-Received: '4: Hello, World!\n'
-Close the connection
-Received: '7: Hello, World!\n'
-Close the connection
-Received: '1: Hello, World!\n'
-Close the connection
-Received: '8: Hello, World!\n'
-Close the connection
-Received: '9: Hello, World!\n'
-Close the connection
-Received: '10: Hello, World!\n'
-Close the connection
-Received: '5: Hello, World!\n'
-Close the connection
-Received: '11: Hello, World!\n'
-Close the connection
-Received: '12: Hello, World!\n'
-Close the connection
 □（待ち受けが続いている）
@@ -323,3 +237,387 @@
 Python 3.7.4
+追記[2020/01/03 16:40]：
+Threadingで1台vs1台のP2P通信の導通確認をするコードとエラーを補足として載せます
+実際には10台以上でこのようなコードを実行し，一定の失敗回数以内で一定数以上のノードに通信できる（send_status='A'）＆一定時間以内に一定数以上のノードから通信があった（receive_status='A'）という結果をハンドリングして，この後に本格的なP2P通信を開始したいというコードになります．
+Threadingとasyncioに関わらず，クライアントとサーバーの機能が共存するようなコードを書くヒントを頂きたく思います．
+```py
+# -*- coding: utf-8 -*-
+import asyncio
+import socket
+import threading
+from time import time
+import queue
+send_queue = queue.Queue()
+receive_set = set()
+send_status = None
+receive_status = None
+sendable_ips = []
+receivable_ips = []
+def try_init(my_ip, ips, dest_port=50008):
+    """
+    他のノードに自身の情報を伝え導通確認をする
+    """
+    global send_status
+    send_status = 'A'
+    fail_count = 0
+    n_nodes = len(ips) - 1
+    for addr in list(set(ips) - {my_ip}):
+        send_queue.put(addr)
+    print('[initial send_queue]')
+    print(list(set(ips) - {my_ip}))
+    while True:
+        next_queue = queue.Queue()
+        while send_queue.empty() is False:
+            addr = send_queue.get()
+            try:
+                with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
+                    s.settimeout(3)
+                    s.setblocking(False)
+                    print('trying to connect: ' + addr + ', ' + str(dest_port))
+                    s.connect((addr, dest_port))
+                    print('connected to: ' + addr + ', ' + str(dest_port))
+                    # サーバにメッセージを送る
+                    s.sendall(b'test msg')
+                    # サーバからの文字列を取得する。
+                    data = s.recv(4096)
+                    if data:
+                        sendable_ips.append(addr)
+            except (ConnectionRefusedError, TimeoutError, socket.timeout) as e:
+                print(e)
+                next_queue.put(addr)
+                fail_count += 1
+            except:
+                send_status = 'Z'
+                raise
+        if next_queue.empty():
+            break
+        elif fail_count < n_nodes//3:
+            while next_queue.empty() is False:
+                send_queue.put(next_queue.get())
+            continue
+        elif next_queue.qsize() < n_nodes//10:
+            send_status = 'B'
+            break
+        else:
+            send_status = 'C'
+            break
+    print('send_status: ' + send_status)
+    return
+def listen_init(my_ip, ips, listen_ip='0.0.0.0', listen_port=50008):
+    """
+    他のノードから通信が来るのを待ち、受信する
+    """
+    global receive_status
+    n_nodes = len(ips) - 1
+    receive_status = 'A'
+    t_listen_start_sec = time()
+    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
+        # IPアドレスとポートを指定してbindする
+        # FWやセキュリティポリシーで解放されているIP/Portにするべきである
+        s.bind((listen_ip, listen_port))
+        # 接続待ち受け 20秒何も来なかったら抜ける
+        s.settimeout(20)
+        s.listen(10)
+        s.setblocking(False)
+        # connectionするまで待つ
+        while True:
+            try:
+                # 接続
+                print('listening at: ' + listen_ip + ', ' + str(listen_port))
+                conn, addr = s.accept()
+                print('got connection from: ' + addr[0] + ', ' + str(addr[1]))
+                data = conn.recv(4096)
+                if data:
+                    conn.sendall(b'OK: Received your info.')
+                    receive_set.add(addr[0]) # 実際は相手にPublic IPv4 Addressをメッセージに入れてもらっている
+                else:
+                    conn.sendall(b'NG: Nothing received.')
+            except socket.timeout:
+                print('listen timeout')
+                pass
+            except:
+                receive_status = 'Z'
+                raise
+            if len(receive_set) == n_nodes:
+                break
+            else:
+                t_listen_current_sec = time()
+                t_listen_elapsed_sec = t_listen_current_sec - t_listen_start_sec
+                if t_listen_elapsed_sec < 100:
+                    continue
+                elif len(receive_set) > 9*n_nodes//10:
+                    receive_status = 'B'
+                    break
+                else:
+                    receive_status = 'C'
+                    break
+    for addr in receive_set:
+        receivable_ips.append(addr)
+    print('receive_status: ' + receive_status)
+    return
+def p2p_setup_main():
+    my_ip = 'x.x.x.x'
+    ips = ['x.x.x.x', 'y.y.y.y']
+    try:
+        t_try_init = threading.Thread(target=try_init, args=(my_ip, ips))
+        t_listen_init = threading.Thread(target=listen_init, args=(my_ip, ips))
+        t_try_init.start()
+        t_listen_init.start()
+        t_try_init.join()
+        t_listen_init.join()
+    except:
+        raise
+    return
+if __name__ == '__main__':
+    p2p_setup_main()
+```
+出力（エラー）
+```
+(python3) [ec2-user@ip-x-x-x-x ~]$ /home/ec2-user/venv/python3/bin/python /home/ec2-user/p2p_test/threading_p2p.py
+[initial send_queue]
+['y.y.y.y']
+trying to connect: y.y.y.y, 50008
+Exception in thread Thread-1:
+Traceback (most recent call last):
+  File "/usr/lib64/python3.7/threading.py", line 926, in _bootstrap_inner
+    self.run()
+  File "/usr/lib64/python3.7/threading.py", line 870, in run
+    self._target(*self._args, **self._kwargs)
+  File "/home/ec2-user/p2p_test/threading_p2p.py", line 43, in try_init
+    s.connect((addr, dest_port))
+BlockingIOError: [Errno 115] Operation now in progress
+listening at: 0.0.0.0, 50008
+Exception in thread Thread-2:
+Traceback (most recent call last):
+  File "/usr/lib64/python3.7/threading.py", line 926, in _bootstrap_inner
+    self.run()
+  File "/usr/lib64/python3.7/threading.py", line 870, in run
+    self._target(*self._args, **self._kwargs)
+  File "/home/ec2-user/p2p_test/threading_p2p.py", line 101, in listen_init
+    conn, addr = s.accept()
+  File "/usr/lib64/python3.7/socket.py", line 212, in accept
+    fd, addr = self._accept()
+BlockingIOError: [Errno 11] Resource temporarily unavailable
+```