python - 为什么TCP客户端变慢然后产生OSError : [Errno 99] Cannot assign requested address

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 的第一次运行每秒运行大约 3,000 个任务。有时，第一次运行会变慢(每秒约 500-600 个任务)。

一旦性能下降到每秒 500-600 个任务，进一步的运行永远不会恢复到每秒 3,000 个任务。

最终，运行 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 ，也许有一些注意事项:

消息将是可变长度的。

我可以通过消息来定界或发送消息长度，而不是关闭连接，这似乎是问题的根本来源。

我不想在这里添加像 HTTP 这样的应用层协议(protocol)，因为我知道消息将始终是 UTF8 编码的 JSON。

Answer 1

我认为使用消息分隔符(我选择了 b'\x1e' )允许我为整个消息集建立一个连接，而不是为每条消息建立一个新连接。 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()))