Проблемы при запуске http-сервера с отложенными потоками при высокой нагрузке в скрученном ⇐ Python

[Anonymous]

У меня возникли некоторые проблемы, связанные с искривленной платформой HTTP. В частности, я пытаюсь реализовать использование threads.deferToThread с обратными вызовами, чтобы обеспечить лучший параллелизм, который работает нормально, пока у вас не будет более высокой нагрузки трафика, пытающегося получить доступ к серверу одновременно. В этом случае обратные вызовы часто вообще не выполняются, и в транспорт ничего не записывается.
Короче говоря, при каждом запросе сервер будет обрабатывать пакеты, отправленные клиент внутри отложенного потока. Затем сервер записывает пакеты ответа в очередь, которые обратный вызов on_request_done затем удаляет из очереди и возвращает в качестве ответа. Как упоминалось ранее, это прекрасно работает с 1 или 2 клиентами.
Вот упрощенная версия кода, с которым я работаю:

Код: Выделить всё

class HttpPlayer(Player):
def __init__(self, address: str, port: int) -> None:
super().__init__(address, port)
self.queue = Queue()
self.token = ""

def enqueue(self, data: bytes):
self.queue.put(data)

def dequeue(self, max: int = 4096) -> bytes:
data = b""

while not self.queue.empty():
data += self.queue.get()

return data

class HttpBanchoProtocol(Resource):
isLeaf = True

def __init__(self) -> None:
self.player: HttpPlayer | None = None
self.children = {}

def handle_login_request(self, request: Request) -> bytes:
username, password, client = (
request.content.read().decode().splitlines()
)

deferred = threads.deferToThread(
self.player.login_received,
username,
password,
client
)

deferred.addCallbacks(
lambda _: self.on_request_done(request),
lambda f: self.on_request_error(request, f)
)

return NOT_DONE_YET

def handle_request(self, request: Request) -> bytes:
deferred = threads.deferToThread(
self.process_packets,
request.content.read()
)

deferred.addCallbacks(
lambda _: self.on_request_done(request),
lambda f: self.on_request_error(request, f)
)

return NOT_DONE_YET

def login_received(self, username: str, password: str, client: str) -> None:
# Processing login here
...

def process_packets(self, request: bytes):
# Processing packets here
...

def on_request_done(self, request: Request) -> None:
if request._disconnected:
self.player.logger.warning('Client disconnected before response')
return

if request.finished:
self.player.logger.warning('Request finished before response')
return

request.write(self.player.dequeue())
request.finish()

def on_request_error(
self,
request: Request,
failiure: Failure
) -> None:
request.setResponseCode(500)
self.player.send_error()
self.player.logger.error(
f'Failed to process request: {failiure.getErrorMessage()}',
exc_info=failiure.value
)
self.on_login_done(request)

def render_POST(self, request: Request) -> bytes:
request.setResponseCode(200)

if not (token := request.getHeader('token')):
return self.handle_login_request(request)

if not (player := app.session.players.by_token(token)):
request.setResponseCode(401)
return b""

self.player = player
return self.handle_request(request)

Вот код, над которым я работаю, чтобы при необходимости получить дополнительную информацию: https://github.com/osuTitanic/anchor/bl ... ttp.py#L99
п>

Подробнее здесь: https://stackoverflow.com/questions/787 ... in-twisted