Загадочные исключения при создании множества одновременных запросов от urllib.request к HTTPServer

Ответ 1

Вы используете значение по умолчанию listen(), которое, вероятно, является причиной многих из этих ошибок. Это не количество одновременных клиентов с уже установленным соединением, а количество клиентов, ожидающих очереди прослушивания до установления соединения. Измените свой класс сервера на:

class FancyHTTPServer(ThreadingMixIn, HTTPServer):
    def server_activate(self):
        self.socket.listen(128)

128 является разумным пределом. Возможно, вы захотите проверить socket.SOMAXCONN или вашу ОС somaxconn, если хотите увеличить ее дальше. Если у вас все еще есть случайные ошибки при большой нагрузке, вы должны проверить свои настройки ulimit и при необходимости увеличить.

Я сделал это с вашим примером, и я получил более 1000 потоков, работающих нормально, поэтому я думаю, что это должно решить вашу проблему.

Обновление

Если он улучшится, но он все еще терпит крах 200 одновременных клиентов, то я уверен, что основной проблемой является размер отставания. Имейте в виду, что ваша проблема заключается не в количестве одновременных клиентов, а в количестве одновременных запросов на соединение. Краткое объяснение того, что это значит, не заходя слишком глубоко в внутренние сети TCP.

s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
s.bind((HOST, PORT))
s.listen(BACKLOG)
while running:
    conn, addr = s.accept()
    do_something(conn, addr)

В этом примере сокет теперь принимает соединения на данном порту, а вызов s.accept() блокируется до тех пор, пока клиент не подключится. У вас может быть много клиентов, пытающихся подключиться одновременно, и в зависимости от вашего приложения вы не сможете вызвать s.accept() и отправить клиентское соединение так же быстро, как клиенты пытаются подключиться. Ожидающие клиенты поставлены в очередь, а максимальный размер этой очереди определяется значением BACKLOG. Если очередь заполнена, клиенты будут терпеть неудачу с ошибкой Connection Refused.

Threading не помогает, так как класс ThreadingMixIn выполняет вызов do_something(conn, addr) в отдельном потоке, поэтому сервер может вернуться к mainloop и вызову s.accept().

Вы можете попытаться увеличить отставание дальше, но будет точка, в которой это не поможет, потому что, если очередь становится слишком большой, некоторые клиенты будут таймаутом, прежде чем сервер выполнит вызов s.accept().

Итак, как я сказал выше, ваша проблема - это количество одновременных попыток подключения, а не количество одновременных клиентов. Возможно, 128 достаточно для вашего реального приложения, но вы получаете ошибку в своем тесте, потому что пытаетесь подключиться со всеми 200 потоками сразу и наводнять очередь.

Не беспокойтесь о ulimit, если вы не получите ошибку Too many open files, но если вы хотите увеличить отставание за пределами 128, выполните некоторые исследования socket.SOMAXCONN. Это хорошее начало: https://utcc.utoronto.ca/~cks/space/blog/python/AvoidSOMAXCONN

Ответ 2

Я бы сказал, что ваша проблема связана с некоторой блокировкой ввода-вывода, поскольку я успешно выполнил ваш код в NodeJs. Я также заметил, что и сервер, и клиент могут работать индивидуально.

Но можно увеличить количество запросов с несколькими изменениями:

• Определите количество параллельных подключений:

  http.server.HTTPServer.request_queue_size = 500

• Запустите сервер в другом процессе:

  server = multiprocessing.Process(target = RunHTTPServer) server.start()

• Используйте пул соединений на стороне клиента для выполнения запросов

• Используйте пул потоков на стороне сервера для обработки запросов

• Разрешить повторное использование соединения на стороне клиента, установив схему и используя заголовок "keep-alive"

Со всеми этими изменениями мне удалось запустить код с 500 потоками без каких-либо проблем. Поэтому, если вы хотите попробовать, вот полный код:

import random
from time import sleep, clock
from http.server import BaseHTTPRequestHandler, HTTPServer
from multiprocessing import Process
from multiprocessing.pool import ThreadPool
from socketserver import ThreadingMixIn
from concurrent.futures import ThreadPoolExecutor
from urllib3 import HTTPConnectionPool
from urllib.error import HTTPError


class HTTPServerThreaded(HTTPServer):
    request_queue_size = 500
    allow_reuse_address = True

    def serve_forever(self):
        executor = ThreadPoolExecutor(max_workers=self.request_queue_size)

        while True:
          try:
              request, client_address = self.get_request()
              executor.submit(ThreadingMixIn.process_request_thread, self, request, client_address)
          except OSError:
              break

        self.server_close()


class MyRequestHandler(BaseHTTPRequestHandler):
    default_request_version = 'HTTP/1.1'

    def do_GET(self):
        sleep(random.uniform(0, 1) / 100.0)

        data = b"abcdef"
        self.send_response(200)
        self.send_header("Content-type", 'text/html')
        self.send_header("Content-length", len(data))
        self.end_headers()
        self.wfile.write(data)

    def log_request(self, code=None, size=None):
        pass


def RunHTTPServer():
    server = HTTPServerThreaded(('127.0.0.1', 5674), MyRequestHandler)
    server.serve_forever()


client_headers = { 
    'User-Agent' : 'Mozilla/5.0 (Windows NT 6.1; Win64; x64)',
    'Content-Type': 'text/plain',
    'Connection': 'keep-alive'
}

client_pool = None

def request_is_ok(number):
    response = client_pool.request('GET', "/test" + str(number), headers=client_headers)
    return response.status == 200 and response.data == b"abcdef"


if __name__ == '__main__':

    # start the server in another process
    server = Process(target=RunHTTPServer)
    server.start()

    # start a connection pool for the clients
    client_pool = HTTPConnectionPool('127.0.0.1', 5674)

    # execute the requests
    with ThreadPool(500) as thread_pool:
        start = clock()

        for i in range(5):
            numbers = [random.randint(0, 99999) for j in range(20000)]
            for j, result in enumerate(thread_pool.imap(request_is_ok, numbers)):
                if j % 1000 == 0:
                    print(i, j, result)

        end = clock()
        print("execution time: %s" % (end-start,))

Обновление 1:

Увеличение request_queue_size просто дает вам больше места для хранения запросов, которые не могут быть выполнены в то время, чтобы их можно было выполнить позже. Таким образом, чем дольше очередь, тем выше дисперсия для времени отклика, что я считаю противоположной вашей цели здесь. Что касается ThreadingMixIn, он не идеален, так как он создает и уничтожает поток для каждого запроса, и это дорого. Лучшим решением для сокращения очереди ожидания является использование пула повторных потоков для обработки запросов.

Причиной запуска сервера в другом процессе является использование другого ЦП для сокращения времени выполнения.

Для клиентской стороны, использующей HTTPConnectionPool, был единственный способ сохранить постоянный поток запросов, поскольку у меня было какое-то странное поведение с urlopen при анализе соединений.

Ответ 3

В норме используется только столько потоков, сколько ядер, следовательно, потребность в потоке 8 (включая виртуальные ядра). Модель потоковой обработки проще всего работать, но это действительно мусорный способ сделать это. Лучшим способом обработки нескольких соединений является использование асинхронного подхода. Это сложнее, хотя.

С помощью метода потоков вы можете начать с изучения того, остается ли процесс открытым после выхода из программы. Это будет означать, что ваши потоки не закрываются и, очевидно, будут вызывать проблемы.

Попробуйте это...

class FancyHTTPServer(ThreadingMixIn, HTTPServer):
    daemon_threads = True

Это обеспечит правильное закрытие ваших потоков. Это может произойти автоматически в пуле потоков, но, вероятно, стоит попробовать.