Dart: obsłużyć przychodzące żądania HTTP równolegle

Question

Próbuję napisać serwer HTTP w Dart, który może obsługiwać wiele żądań równolegle. Nie udało mi się jak dotąd osiągnąć części "równoległej".

Oto, co starałem się najpierw:

import 'dart:io'; 

main() { 
    HttpServer.bind(InternetAddress.ANY_IP_V4, 8080).then((HttpServer server) { 
    server.listen((HttpRequest request) { 
     Stopwatch stopwatch = new Stopwatch(); 
     stopwatch.start(); 
     while (stopwatch.elapsedMilliseconds < 1000) { /* do nothing */ } 
     request.response.statusCode = HttpStatus.OK; 
     request.response.write(stopwatch.elapsedMilliseconds.toString()); 
     request.response.close().catchError(print); 
    }); 
    }); 
} 

na każde żądanie robi zajęty pracą na jedną sekundę, a następnie uzupełnia. Poprosiłem o obsługę żądań w ten sposób, aby jego czas był przewidywalny, dzięki czemu mogłem z łatwością zobaczyć efekt żądania w menedżerze zadań Windows (rdzeń procesora przeskakuje do 100% użycia).

mogę powiedzieć, to nie obsługuje żądania równolegle, ponieważ:

1. Gdybym załadować kilka zakładek przeglądarki, aby http://example:8080/ a następnie odśwież je wszystkie, obciążenie jednej wypustki po drugim w kolejności, w przybliżeniu 1 sekunda między każdym.

2. Jeśli korzystam z narzędzia do sprawdzania obciążenia wrk z tymi ustawieniami ... wrk -d 10 -c 8 -t 8 http://example:8080/ ... wypełnia on od 5 do 8 żądań w ciągu 10 sekund, które mu dałem. Gdyby serwer wykorzystywał wszystkie moje 8 rdzeni, spodziewałbym się, że liczba bliższa 80 próśb.

3. Kiedy otwieram menedżera zadań Windows podczas testu wrk, zauważam, że tylko jeden z moich rdzeni jest prawie w 100% użyty, a reszta jest prawie bezczynna.

Tak więc próbowałem przy użyciu izolatów, chcąc ręcznie tarło nowy izolować/wątku dla każdego żądania:

import 'dart:io'; 
import 'dart:isolate'; 

main() { 
    HttpServer.bind(InternetAddress.ANY_IP_V4, 8080).then((HttpServer server) { 
    server.listen((HttpRequest request) { 
     spawnFunction(handleRequest).send(request); 
    }); 
    }); 
} 

handleRequest() { 
    port.receive((HttpRequest request, SendPort sender) { 
    Stopwatch stopwatch = new Stopwatch(); 
    stopwatch.start(); 
    while (stopwatch.elapsedMilliseconds < 1000) { /* do nothing */ } 
    request.response.statusCode = HttpStatus.OK; 
    request.response.write(stopwatch.elapsedMilliseconds.toString()); 
    request.response.close().catchError(print); 
    }); 
} 

To nie działa w ogóle. Nie podoba mi się, że próbuję wysłać HttpRequest jako wiadomość do izolatu. Tutaj jest błąd:

#0  _SendPortImpl._sendInternal (dart:isolate-patch/isolate_patch.dart:122:3) 
#1  _SendPortImpl._sendNow (dart:isolate-patch/isolate_patch.dart:95:18) 
#2  _SendPortImpl.send (dart:isolate-patch/isolate_patch.dart:91:18) 
#3  main.<anonymous closure>.<anonymous closure> (file:///C:/Development/dartbenchmark/simple2.dart:7:40) 
#4  _StreamSubscriptionImpl._sendData (dart:async/stream_impl.dart:475:12) 
#5  _StreamImpl._sendData.<anonymous closure> (dart:async/stream_impl.dart:251:29) 
#6  _SingleStreamImpl._forEachSubscriber (dart:async/stream_impl.dart:335:11) 
#7  _StreamImpl._sendData (dart:async/stream_impl.dart:249:23) 
#8  _StreamImpl._add (dart:async/stream_impl.dart:51:16) 
#9  StreamController.add (dart:async/stream_controller.dart:10:35) 
#10  _HttpServer._handleRequest (http_impl.dart:1261:20) 
#11  _HttpConnection._HttpConnection.<anonymous closure> (http_impl.dart:1188:33) 
#12  _StreamSubscriptionImpl._sendData (dart:async/stream_impl.dart:475:12) 
#13  _StreamImpl._sendData.<anonymous closure> (dart:async/stream_impl.dart:251:29) 
#14  _SingleStreamImpl._forEachSubscriber (dart:async/stream_impl.dart:335:11) 
#15  _StreamImpl._sendData (dart:async/stream_impl.dart:249:23) 
#16  _StreamImpl._add (dart:async/stream_impl.dart:51:16) 
#17  StreamController.add (dart:async/stream_controller.dart:10:35) 
#18  _HttpParser._doParse (http_parser.dart:415:26) 
#19  _HttpParser._parse (http_parser.dart:161:15) 
#20  _HttpParser._onData._onData (http_parser.dart:509:11) 
#21  _StreamSubscriptionImpl._sendData (dart:async/stream_impl.dart:475:12) 
#22  _StreamImpl._sendData.<anonymous closure> (dart:async/stream_impl.dart:251:29) 
#23  _SingleStreamImpl._forEachSubscriber (dart:async/stream_impl.dart:335:11) 
#24  _StreamImpl._sendData (dart:async/stream_impl.dart:249:23) 
#25  _StreamImpl._add (dart:async/stream_impl.dart:51:16) 
#26  StreamController.add (dart:async/stream_controller.dart:10:35) 
#27  _Socket._onData._onData (dart:io-patch/socket_patch.dart:726:42) 
#28  _StreamSubscriptionImpl._sendData (dart:async/stream_impl.dart:475:12) 
#29  _StreamImpl._sendData.<anonymous closure> (dart:async/stream_impl.dart:251:29) 
#30  _SingleStreamImpl._forEachSubscriber (dart:async/stream_impl.dart:335:11) 
#31  _StreamImpl._sendData (dart:async/stream_impl.dart:249:23) 
#32  _StreamImpl._add (dart:async/stream_impl.dart:51:16) 
#33  StreamController.add (dart:async/stream_controller.dart:10:35) 
#34  _RawSocket._RawSocket.<anonymous closure> (dart:io-patch/socket_patch.dart:452:52) 
#35  _NativeSocket.multiplex (dart:io-patch/socket_patch.dart:253:18) 
#36  _NativeSocket.connectToEventHandler.<anonymous closure> (dart:io-patch/socket_patch.dart:338:54) 
#37  _ReceivePortImpl._handleMessage (dart:isolate-patch/isolate_patch.dart:81:92) 

Unhandled exception: 
Illegal argument(s): Illegal argument in isolate message : (object is a closure) 
#0  _throwDelayed.<anonymous closure> (dart:async/stream_impl.dart:22:5) 
#1  _asyncRunCallback._asyncRunCallback (dart:async/event_loop.dart:15:17) 
#2  _asyncRunCallback._asyncRunCallback (dart:async/event_loop.dart:25:9) 
#3  Timer.run.<anonymous closure> (dart:async/timer.dart:17:21) 
#4  Timer.run.<anonymous closure> (dart:async/timer.dart:25:13) 
#5  Timer.Timer.<anonymous closure> (dart:async-patch/timer_patch.dart:9:15) 
#6  _Timer._createTimerHandler._handleTimeout (timer_impl.dart:99:28) 
#7  _Timer._createTimerHandler._handleTimeout (timer_impl.dart:107:7) 
#8  _Timer._createTimerHandler.<anonymous closure> (timer_impl.dart:115:23) 
#9  _ReceivePortImpl._handleMessage (dart:isolate-patch/isolate_patch.dart:81:92) 

wersje używane: wersja

• Dart Editor wersja 0.5.9_r22879
• Dart SDK 0.5.9.0_r22879

Czy to możliwe, aby obsłużyć te wnioski równolegle, z wszystkimi dostępnymi rdzeniami mojej maszyny, używając Dart'a?

Answer 1

Napisałem bibliotekę o nazwie dart-isoserver, aby zrobić to jakiś czas temu. Jest teraz nieco spróchniały, ale widać podejście.

https://code.google.com/p/dart-isoserver/

Co zrobiłem było proxy HttpRequest i HttpResponse poprzez porty izolować, ponieważ nie można wysłać je bezpośrednio. Udało się, chociaż było kilka zastrzeżeń:

1. We/wy na żądanie i odpowiedź przeszły przez główny izolator, więc część nie była równoległa. Inne prace wykonane w izolatorze pracującym nie blokowały jednak głównego izolatora. Tak naprawdę powinno się zdarzyć, że połączenie między gniazdami powinno być transferowalne między izolatami.
2. Wyjątki w izolacie spowodowałyby zmniejszenie całego serwera.spawnFunction() ma teraz parametr przechwytywania wyjątków wyjątku, więc jest to w pewnym stopniu poprawne, ale spawnUri() tego nie robi. dart-isoserver używał spawnUri() do zaimplementowania ładowania na gorąco, więc musiałoby zostać usunięte.
3. Isolates są nieco powolne do uruchomienia, a prawdopodobnie nie chcesz, aby jeden na połączenie dla tysięcy równoczesnych przypadków użycia połączenia, które są docelowe nginx i node.js. Pula izolatów z kolejkami pracy prawdopodobnie działałaby lepiej, ale eliminowałoby to fajną funkcję, którą można użyć blokowania operacji we/wy w module roboczym.

Uwaga dotycząca pierwszego przykładu kodu. To zdecydowanie nie będzie działać równolegle, jak zauważyłeś, ponieważ Dart jest jednowątkowe. Żaden kod darta w tym samym izolacie nigdy nie jest uruchamiany jednocześnie.

Answer 2

Nawet przy obecnych ograniczeniach HttpServer możliwe jest wykorzystanie wielu rdzeni poprzez uruchamianie wielu procesów serwera za odwrotnym serwerem proxy, takim jak Apache lub Nginx. Z poziomu Dart możesz także rozwidlić procesy potomne, aby podzielić zadania wymagające dużej mocy obliczeniowej.

Dobrym miejscem na rozpoczęcie byłoby przeczytanie o skalowaniu node.js, ponieważ używa ono również architektury pojedynczego wątku na proces.

Edycja: Odpowiedź jest już nieaktualna, możliwe jest teraz udostępnianie żądań między izolatami, dzięki czemu proces Dart może wykorzystywać wiele rdzeni.

Zobacz dokumenty dla ServerSocket.bind(shared).

"Opcjonalny udostępniony argument określa, czy możliwe jest dodatkowe powiązanie z tym samym adresem, portem i kombinacją vOnly tylko z tego samego procesu dartowania. Jeśli współdzielenie ma wartość true, a dodatkowe powiązania są wykonywane, wówczas połączenia przychodzące będą rozdzielone między tymi zestawami. ServerSockets. Jednym ze sposobów użycia tego jest posiadanie liczby izolatów, pomiędzy którymi rozprowadzane są połączenia przychodzące. "