python - Почему TCP-клиент замедляется, а затем выдает OSError: [Errno 99] Невозможно назначить запрошенный адрес

Question

Я пишу сервер для симулятора событий и использую asyncio TCP-сервер для этой цели.

#server.py
import asyncio
import itertools
import json


class Server:
    def __init__(self, loop=None):
        self.loop = loop
        self.pq = asyncio.PriorityQueue()
        self.counter = itertools.count()

    async def __call__(self, reader, writer):
        event = await reader.read(100)
        message = json.loads(event.decode())
        self.pq.put_nowait([next(self.counter), message])
        while self.pq.qsize():
            t = await self.pq.get()
            send_data = json.dumps(t).encode("utf-8")
            writer.write(send_data)
            await writer.drain()

loop = asyncio.get_event_loop()
s = Server(loop)
coro = asyncio.start_server(s, '127.0.0.1', 5000, loop=loop)
server = loop.run_until_complete(coro)

# Serve requests until Ctrl+C is pressed
print('Serving on {}'.format(server.sockets[0].getsockname()))
try:
    loop.run_forever()
except KeyboardInterrupt:
    pass

# Close the server
server.close()
loop.run_until_complete(server.wait_closed())
loop.close()

Я хочу, чтобы клиент быстро отправлял закодированные события json на этот сервер.

#client.py
import socket
import json
import datetime

host = "127.0.0.1"
port = 5000

N = 10000
start = datetime.datetime.utcnow()


for i in range(1, N + 1):
    s = socket.create_connection((host, port))
    send_message = {"id": i, "value": i * 3}
    send_json = json.dumps(send_message)
    send_data = send_json.encode("utf-8")
    s.sendall(send_data)
    receive_data = s.recv(1024)
    receive_json = receive_data.decode("utf-8")
    _ = json.loads(receive_json)
    s.close()

stop = datetime.datetime.utcnow()
print("Tasks per second: {}".format(N / (stop - start).total_seconds()))

Проблема Клиентская программа имеет разную производительность и генерирует ошибки, несмотря на то, что другие пользовательские программы не открыты.

Часто, но не всегда, первый запуск client.py выполняется со скоростью около 3000 задач в секунду. Время от времени первый запуск выполняется медленнее (~ 500-600 задач в секунду).

Как только производительность падает до 500-600 задач в секунду, дальнейшие запуски никогда не восстанавливаются до 3000 задач в секунду.

В конце концов, выполнение client.py вызывает следующее исключение:

Traceback (most recent call last):
  File "aioclient.py", line 12, in <module>
    s = socket.create_connection((host, port))
  File "/home/randm/Libraries/anaconda3/lib/python3.6/socket.py", line 724, in create_connection
    raise err
  File "/home/randm/Libraries/anaconda3/lib/python3.6/socket.py", line 713, in create_connection
    sock.connect(sa)
OSError: [Errno 99] Cannot assign requested address

Вопрос

Как мне переписать client.py (или server.py), чтобы избежать этого?

Прочитав https://docs.python.org/3/howto/sockets.html, возможно несколько замечаний:

1. Сообщения будут переменной длины.
2. Я согласен с разделением или отправкой длины сообщения вместе с сообщениями, а не с отключением соединения, которое, по-видимому, является основным источником проблемы здесь.
3. Я бы предпочел не добавлять протокол уровня приложения, такой как HTTP, поскольку я знаю, что сообщения всегда будут в кодировке UTF8 в формате JSON.

Answer 1

Я думаю, что использование разделителя сообщений (я выбрал b'\x1e') позволяет мне иметь 1 соединение для всего набора сообщений вместо того, чтобы устанавливать новое соединение для каждого сообщения. Метод StreamReader.readuntil отлично работает в этом случае.

# server.py
import asyncio
import itertools
import json


PQ = asyncio.PriorityQueue()
COUNTER = itertools.count()


async def handle_data_provider(reader, writer):
    try:
        while True:
            data = await reader.readuntil(b'\x1e')
            message = json.loads(data[:-1].decode())
            n = next(COUNTER)
            if n % 10000 == 0:
                print(n, PQ.qsize())
            PQ.put_nowait([n, message])
    except asyncio.streams.IncompleteReadError:
        pass

loop = asyncio.get_event_loop()
coro = asyncio.start_server(handle_data_provider, '127.0.0.1', 5000, loop=loop)
server = loop.run_until_complete(coro)

# Serve requests until Ctrl+C is pressed
print('Serving on {}'.format(server.sockets[0].getsockname()))
try:
    loop.run_forever()
except KeyboardInterrupt:
    pass

# Close the server
server.close()
loop.run_until_complete(server.wait_closed())
loop.close()

И клиент ...

# client.py
import socket
import json
import datetime

host = "127.0.0.1"
port = 5000

N = 10000
start = datetime.datetime.utcnow()

s = socket.create_connection((host, port))

for i in range(1, N + 1):
    message = {"id": i, "value": i * 3}
    json_message = json.dumps(message)
    data = json_message.encode("utf-8") + b'\x1e'
    s.sendall(data)

s.close()

stop = datetime.datetime.utcnow()
print("Tasks per second: {}".format(N / (stop - start).total_seconds()))