Tajemnicze wyjątki podczas robienia wielu równoczesnych żądań od adresu urllib.request do HTTPServer

Question

Próbuję wykonać this Matasano crypto challenge, który polega na wykonaniu ataku synchronizującego na serwer ze sztucznie spowolnioną funkcją porównywania ciągów. Mówi się, aby użyć "wybranej przez siebie struktury sieci", ale nie miałem ochoty instalować ram internetowych, więc zdecydowałem się użyć modułu HTTPServer class wbudowanego w moduł http.server.

Wymyśliłem coś, co działało, ale działo się bardzo wolno, więc starałem się przyspieszyć za pomocą (słabo udokumentowanej) puli wątków wbudowanej w multiprocessing.dummy. Było znacznie szybciej, ale zauważyłem coś dziwnego: jeśli wykonam 8 lub mniej żądań jednocześnie, to działa dobrze. Jeśli mam więcej, to działa przez chwilę i daje mi błędy w pozornie przypadkowych czasach. Błędy wydają się być niespójne i nie zawsze takie same, ale zwykle mają w sobie Connection refused, invalid argument, OSError: [Errno 22] Invalid argument, urllib.error.URLError: <urlopen error [Errno 22] Invalid argument>, BrokenPipeError: [Errno 32] Broken pipe lub urllib.error.URLError: <urlopen error [Errno 61] Connection refused>.

Czy istnieje ograniczenie liczby połączeń obsługiwanych przez serwer? Nie sądzę, że liczba wątków jako takich jest problemem, ponieważ napisałem prostą funkcję, która spowodowała spowolnienie porównywania ciągów bez uruchamiania serwera WWW i wywołała go z 500 równoczesnymi wątkami i działała dobrze. Nie sądzę, że problem polega na zwykłym wysyłaniu żądań z wielu wątków, ponieważ zrobiłem roboty, które używały ponad 100 wątków (wszystkie zgłaszały jednoczesne żądania do tej samej witryny) i działały dobrze. Wygląda na to, że być może HTTPServer nie jest przeznaczony do hostowania witryn produkcyjnych, które generują duży ruch, ale jestem zaskoczony, że tak łatwo jest go zawiesić.

Próbowałem stopniowo usuwać rzeczy z mojego kodu, który wyglądał niezwiązany z problemem, jak zwykle robię, kiedy diagnozuję takie tajemnicze błędy, ale to nie było bardzo pomocne w tym przypadku. Wyglądało na to, że usuwając pozornie niepowiązany kod, liczba połączeń, które serwer mógł obsłużyć, stopniowo wzrastała, ale nie było wyraźnej przyczyny awarii.

Czy ktoś wie, jak zwiększyć liczbę żądań, które mogę wykonać naraz, a przynajmniej dlaczego tak się dzieje?

Mój kod jest skomplikowany, ale wpadłem na to prosty program, który pokazuje problem:

#!/usr/bin/env python3 

import os 
import random 

from http.server import BaseHTTPRequestHandler, HTTPServer 
from multiprocessing.dummy import Pool as ThreadPool 
from socketserver import ForkingMixIn, ThreadingMixIn 
from threading import Thread 
from time import sleep 
from urllib.error import HTTPError 
from urllib.request import urlopen 


class FancyHTTPServer(ThreadingMixIn, HTTPServer): 
    pass 


class MyRequestHandler(BaseHTTPRequestHandler): 
    def do_GET(self): 
     sleep(random.uniform(0, 2)) 
     self.send_response(200) 
     self.end_headers() 
     self.wfile.write(b"foo") 

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

def request_is_ok(number): 
    try: 
     urlopen("http://localhost:31415/test" + str(number)) 
    except HTTPError: 
     return False 
    else: 
     return True 


server = FancyHTTPServer(("localhost", 31415), MyRequestHandler) 
try: 
    Thread(target=server.serve_forever).start() 
    with ThreadPool(200) as pool: 
     for i in range(10): 
      numbers = [random.randint(0, 99999) for j in range(20000)] 
      for j, result in enumerate(pool.imap(request_is_ok, numbers)): 
       if j % 20 == 0: 
        print(i, j) 
finally: 
    server.shutdown() 
    server.server_close() 
    print("done testing server") 

Z jakiegoś powodu, program powyżej działa poprawnie, chyba że ma ponad 100 wątków lub tak, ale moja prawdziwy kod wyzwania może obsłużyć tylko 8 wątków. Jeśli uruchomię go z 9, zwykle otrzymuję błędy połączenia, a przy 10 zawsze otrzymuję błędy połączenia. Próbowałem używać concurrent.futures.ThreadPoolExecutor, i multiprocessing.pool zamiast z multiprocessing.dummy.pool i żadne z nich nie pomogło. Próbowałem używać zwykłego obiektu HTTPServer (bez ThreadingMixIn) i to tylko sprawiało, że rzeczy działały bardzo wolno i nie rozwiązały problemu. Próbowałem użyć ForkingMixIn i to też nie naprawiło.

Co mam z tym zrobić? Używam Python 3.5.1 na MacBooka Pro z końca 2013 roku z systemem OS X 10.11.3.

EDIT: Próbowałem kilka rzeczy, w tym uruchomienie serwera w procesie zamiast wątku, jako prosty HTTPServer, z ForkingMixIn iz ThreadingMixIn. Żadne z nich nie pomogło.

EDIT: Ten problem jest dziwniejszy niż myślałem.Próbowałem utworzyć jeden skrypt z serwerem, a drugi z wieloma wątkami wysyłającymi żądania i uruchamiając je w różnych zakładkach w moim terminalu. Proces z serwerem działał dobrze, ale ten, który zgłasza żądania, zawiesił się. Wyjątkami były mieszanką ConnectionResetError: [Errno 54] Connection reset by peer, urllib.error.URLError: <urlopen error [Errno 54] Connection reset by peer>, OSError: [Errno 41] Protocol wrong type for socket, urllib.error.URLError: <urlopen error [Errno 41] Protocol wrong type for socket>, urllib.error.URLError: <urlopen error [Errno 22] Invalid argument>.

Próbowałem go z fałszywym serwerem, jak ten powyżej, a jeśli ograniczyłem liczbę współbieżnych żądań do 5 lub mniej, to działało dobrze, ale z 6 żądaniami, proces klienta się zawiesił. Wystąpiły błędy na serwerze, ale nadal działało. Klient zawiesił się niezależnie od tego, czy używałem wątków lub procesów do wysyłania żądań. Następnie spróbowałem umieścić spowolnioną funkcję na serwerze i było w stanie obsłużyć 60 równoczesnych żądań, ale rozbił się z 70. Wydaje się, że może to być sprzeczne z dowodami, że problem dotyczy serwera.

EDIT: Próbowałem większość rzeczy, opisanych za pomocą requests zamiast urllib.request i pobiegł do podobnych problemów.

EDYCJA: Używam teraz systemu OS X 10.11.4 i uruchamiam te same problemy.

Answer 1

Używasz wartość domyślna listen() zaległości, co jest prawdopodobnie przyczyną wielu tych błędów. Nie jest to liczba równoczesnych klientów z nawiązanym połączeniem, ale liczba klientów oczekujących w kolejce odsłuchu przed nawiązaniem połączenia. Zmień klasę serwera na:

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

128 to rozsądny limit. Możesz chcieć sprawdzić socket.SOMAXCONN lub twój system operacyjny somaxconn, jeśli chcesz go jeszcze bardziej zwiększyć. Jeśli nadal masz losowe błędy przy dużym obciążeniu, powinieneś sprawdzić ustawienia ulimit i zwiększyć w razie potrzeby.

Zrobiłem to z przykładu i mam ponad 1000 wątków uruchomionych w porządku, więc myślę, że powinno rozwiązać problem.

---

Aktualizacja

Jeśli to poprawić, ale to jeszcze upaść 200 jednoczesnych klientów, to jestem całkiem pewny, że głównym problemem była wielkość zaległości. Należy pamiętać, że problem nie dotyczy liczby współbieżnych klientów, ale liczby jednoczesnych żądań połączeń. Krótkie wyjaśnienie, co to oznacza, bez zbytniego zagłębiania się w elementy 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) 

W tym przykładzie, gniazdo jest teraz przyjmowanie połączeń na danym porcie, a połączenie s.accept() zablokuje aż klient łączy. Można mieć wielu klientów próbujących połączyć się jednocześnie, a w zależności od aplikacji może nie być możliwe wywołanie s.accept() i wywołanie połączenia klienta tak szybko, jak klienci próbują się połączyć. Oczekujący klienci są ustawieni w kolejce, a maksymalny rozmiar tej kolejki jest określony przez wartość BACKLOG. Jeśli kolejka jest pełna, klienty zostaną odrzucone z błędem połączenia odrzuconego.

Nawlekanie nie pomaga, ponieważ klasa ThreadingMixIn wykonuje wywołanie do_something(conn, addr) w osobnym wątku, dzięki czemu serwer może wrócić do wywołania i wywołania s.accept().

Możesz spróbować zwiększyć zaległości dalej, ale będzie punkt, w którym to nie pomoże, ponieważ jeśli kolejka będzie zbyt duża, niektórzy klienci przekroczą limit czasu, zanim serwer wykona połączenie s.accept().

Tak jak powiedziałem powyżej, Twoim problemem jest liczba jednoczesnych prób połączenia, a nie liczba jednoczesnych klientów. Być może 128 wystarczy do rzeczywistej aplikacji, ale dostajesz błąd w teście, ponieważ próbujesz połączyć się ze wszystkimi 200 wątkami naraz i zalanie kolejki.

nie martw się o ulimit chyba że pojawi się błąd Too many open files, ale jeśli chcesz zwiększyć zaległości ponad 128, zrobić rozeznanie na socket.SOMAXCONN. To jest dobry początek: https://utcc.utoronto.ca/~cks/space/blog/python/AvoidSOMAXCONN

Answer 2

Normą jest użycie tylko tylu nici co rdzeni, stąd potrzeba 8 wątków (w tym wirtualne rdzenie). Model z wątkami jest najłatwiejszy do wykonania, ale naprawdę jest to bzdurny sposób na zrobienie tego. Lepszym sposobem na obsługę wielu połączeń jest użycie podejścia asynchronicznego. Jest to jednak trudniejsze.

Metodą threading można rozpocząć od sprawdzenia, czy proces jest otwarty po zamknięciu programu. Oznaczałoby to, że twoje wątki nie zamykają się i oczywiście powodują problemy.

Spróbuj tego ...

class FancyHTTPServer(ThreadingMixIn, HTTPServer): 
    daemon_threads = True 

To zapewni, że nici zamyka się prawidłowo. Może się zdarzyć automatycznie w puli wątków, ale i tak warto spróbować.

Answer 3

Powiedziałbym, że twój problem jest związany z blokowaniem IO, ponieważ pomyślnie wykonałem twój kod na NodeJs. Zauważyłem również, że zarówno serwer, jak i klient mają problemy z pracą indywidualnie.

Ale możliwe jest zwiększenie liczby wniosków z kilkoma modyfikacjami:

• Określić liczbę jednoczesnych połączeń:

  http.server.HTTPServer.request_queue_size = 500

• Uruchom serwer w innym procesie:

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

• Użyj puli połączeń po stronie klienta, aby wykonać żądania

• pomocą puli wątków po stronie serwera do obsługi żądań

• Zezwalaj na ponowne wykorzystanie Przyłącze po stronie klienta poprzez ustawienie schematu i za pomocą „keep-alive” nagłówek

z tych wszystkich modyfikacji, udało mi się uruchomić kod z 500 wątków bez problemu. Więc jeśli chcesz spróbować, oto kompletny kod:

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

Aktualizacja 1:

Zwiększenie request_queue_size tylko daje więcej miejsca do przechowywania wniosków, które nie mogą być wykonywane na czas, aby mogły zostać wykonane później. Im dłuższa kolejka, tym większa dyspersja dla czasu odpowiedzi, co jest moim zdaniem przeciwne do celu tutaj. Jeśli chodzi o ThreadingMixIn, nie jest idealny, ponieważ tworzy i niszczy wątek dla każdego żądania i jest drogi. Lepszym wyborem do zmniejszenia kolejki oczekujących jest użycie puli wielokrotnego użytku do obsługi żądań.

Powodem uruchomienia serwera w innym procesie jest skorzystanie z innego procesora w celu skrócenia czasu wykonania.

Dla klientów korzystających z HTTPConnectionPool był jedyny sposób, w jaki znalazłem stały przepływ żądań, ponieważ miałem dziwne zachowanie z urlopen podczas analizy połączeń.