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

Вы используете значение невыполненной работы по умолчанию 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.

Многопоточность не помогает, потому что класс ThreadingMixIn выполняет вызов do_something(conn, addr) в отдельном потоке, чтобы сервер мог вернуться к основному циклу и вызову s.accept().

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

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

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

Я бы сказал, что ваша проблема связана с некоторой блокировкой ввода-вывода, поскольку я успешно выполнил ваш код на 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 при анализе соединений.

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

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

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

class FancyHTTPServer(ThreadingMixIn, HTTPServer):
    daemon_threads = True

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