为什么 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以每秒约 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，也许有一些注意事项：

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()))