Pythonでasyncioを使ってP2P通信をTCPで実現したい（お互いのIPアドレスは既知）

Question

前提・実現したいこと

• 決められた台数（2台〜）のAWS ec2インスタンス同士がP2PでTCP通信できるようにしたいです
  ※送受信する情報はPythonで処理するため，通信部分もPythonで実現しようと考えています
  ※お互いのIPアドレス・待ち受けポートは既知とします
• 各インスタンスは特定のポート（例えば50008）で他のインスタンスからの通信を待ち受けていると同時に，一定の周期で他のインスタンスの特定のポートへ通信をします
  ※待ち受けていたポートにメッセージが来た→メッセージを処理して返信用のメッセージを返す
  ※相手にメッセージを送る→相手がメッセージを処理して返信用のメッセージを返す→帰ってきたメッセージをもとに変数の増減などの処理を行う
• サーバーとしての受信待ちとクライアントとしての送信待ちが共存して走る形になるため，ブロッキングが起きないようasyncioを使いたいと考えています
• サーバー的な役割を果たすループでは一定の条件（例えば自分以外の全員から1回以上通信されるなど）を満たしたら止めたい
• クライアント的な役割を果たすループでは送信しようとして一定時間応答がなければキャンセルする処理をしたい

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

実現したいことの前段階として
1台 vs 1台で20個のメッセージを相手に送信しつつ，
相手からのメッセージを待ち受けるコードを書いたものの，
①待ち受けしている関数で相手から届いた情報がPrintされない
②待ち受けしている関数の止め方が分からない
（loop.stop()のような記述をどこに埋め込めば良いか分からない）

該当のソースコード

https://docs.python.org/ja/3/library/asyncio-stream.html#tcp-echo-client-using-streams
https://docs.python.org/ja/3/library/asyncio-stream.html#tcp-echo-server-using-streams
などを参考に作成

#encoding: utf-8
import asyncio

ip_to = 'x.x.x.x' # Public IPv4 Address

async def tcp_echo_client(message):
    reader, writer = await asyncio.wait_for(asyncio.open_connection(
        ip_to, 50008), timeout=5)

    print(f'Send: {message!r}')
    writer.write(message.encode())

    data = await reader.read(100)
    print(f'Received: {data.decode()!r}')

    print('Close the connection')
    writer.close()

async def handle_echo(reader, writer):
    data = await reader.read(100)
    message = data.decode()
    addr = writer.get_extra_info('peername')

    print(f"Received {message!r} from {addr!r}")

    print(f"Send: {message!r}")
    writer.write(data)
    await writer.drain()

    print("Close the connection")
    writer.close()

async def main():
    server = await asyncio.start_server(
        handle_echo, '0.0.0.0', 50008,
        reuse_address=True, reuse_port=True)

    addr = server.sockets[0].getsockname()
    print(f'Serving on {addr}')

    async with server:
        await server.serve_forever()

async def multiple_request(loop):
    tasks = []
    tasks.append(main())
    for i in range(20):
        tasks.append(tcp_echo_client(f'{i}: Hello, World!\n'))
    done, pending = await asyncio.wait(tasks)

loop = asyncio.get_event_loop()
loop.run_until_complete(multiple_request(loop))
loop.close()

2台のインスタンスで同じコードを実行したところ，出力は以下のようになりました

(python3) [ec2-user@ip-x-x-x-x ~]$ python "/home/ec2-user/p2p_test/p2p.py"
Serving on ('0.0.0.0', 50008)
Send: '13: Hello, World!\n'
Send: '0: Hello, World!\n'
Send: '3: Hello, World!\n'
Send: '14: Hello, World!\n'
Send: '15: Hello, World!\n'
...（略）...
Received: '13: Hello, World!\n'
Close the connection
Received: '0: Hello, World!\n'
Close the connection
Received: '3: Hello, World!\n'
Close the connection
Received: '14: Hello, World!\n'
Close the connection
Received: '15: Hello, World!\n'
Close the connection
...（略）...
□（待ち受けが続いている）

handle_echo側の関数によって以下のような出力が20個printされるはずですがprintされません

Received '0: Hello,World!\n' from ('x.x.x.x', 61827)
Send: '0: Hello,World!\n'

試したこと

asyncioを使う前はThreadingでサーバー的な関数（'0.0.0.0', 50008をbind）とクライアント的な関数を並行して実行させていましたが，サーバーの待ち受け部分conn, addr = s.accept()とクライアントの送信待ちs.connect((addr, dest_port))にブロッキングされていたようで，メッセージが相手に届きませんでした．s.setblocking(False)を設定すると，IOErrorとしてポート50008にバインドできないという旨のエラーが出たため，方針を切り替えてasyncioを使ってみようとしたところ，非同期処理でServerとClientを共存させているコード例が見当たらず詰まってしまいました．
初心者ながら，参考になる知識やコードの修正をいただけますと大変ありがたいです．

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

Python 3.7.4

追記[2020/01/03 16:40]：

Threadingで1台vs1台のP2P通信の導通確認をするコードとエラーを補足として載せます
実際には10台以上でこのようなコードを実行し，一定の失敗回数以内で一定数以上のノードに通信できる（send_status='A'）＆一定時間以内に一定数以上のノードから通信があった（receive_status='A'）という結果をハンドリングして，この後に本格的なP2P通信を開始したいというコードになります．
Threadingとasyncioに関わらず，クライアントとサーバーの機能が共存するようなコードを書くヒントを頂きたく思います．

# -*- 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