Mechanizm anulowania ForkJoinTask opiera się na flagi współpracy, a nie wymuszonym przerwaniu wątku. Oznacza to, że cancel() jedynie ustawia wewnętrzny zmienny status, który zadania muszą jawnie sprawdzać, aby dostrzec prośby o zakończenie. W rezultacie, ten projekt nie odblokowuje wątków oczekujących na monolityczne operacje WE/WY, takie jak odczyty FileChannel lub operacje InputStream gniazda. Te blokujące wywołania nie sprawdzają flagi anulowania i nie mogą być przerywane przez standardowe mechanizmy przerwania wątków.
Aby zapobiec wygłodzeniu puli, gdy pracownicy blokują, API ForkJoinPool.managedBlock pozwala deweloperom rejestrować instancję ForkJoinPool.ManagedBlocker. Ten blokujący sygnalizuje puli, aby uruchomić kompensacyjny wątek roboczy, utrzymując docelowy poziom równoległości mimo blokującej pracy. Metoda isReleasable blokady zapewnia możliwość sprawdzenia statusu anulowania lub programowego przerwania zablokowanej operacji. Umożliwia to puli łagodną degradację, a nie wyczerpywanie budżetu wątków w przypadku nieodpowiednich operacji WE/WY.
Napotkaliśmy to ograniczenie podczas budowania równoległego procesora logów, który używał Files.lines() w ramach niestandardowej RecursiveTask. Zadanie analizowało logi o skali terabajtów z urządzenia pamięci masowej zamontowanego w sieci. Kiedy użytkownicy żądali anulowania długotrwałych analizy, wątki ForkJoinPool utknęły w blokujących wywołaniach systemowych read() na minuty. Całkowicie ignorowały flagę anulowania, co uniemożliwiało rozpoczęcie nowych zadań i powodowało poważne wygłodzenie wątków.
Rozważaliśmy trzy różne podejścia dla rozwiązania zakleszczenia. Pierwsze podejście polegało na całkowitym porzuceniu ForkJoinPool i przejściu na cache'owany ThreadPoolExecutor. To oferowało prostsze semantyki przerwania i natychmiastową wymianę wątku, ale poświęcało efektywność kradzieży pracy, co było kluczowe dla naszych wymagających etapów analizy CPU.
Drugie podejście zakładało owinięcie każdego wywołania WE/WY w logikę Thread.interrupt() i przejście na przerywalne kanały, takie jak SocketChannel. Chociaż to wspierało natychmiastowe anulowanie, okazało się to inwazyjne i niekompatybilne z kodem starszych bibliotek, który polegał na standardowych blokujących strumieniach i parserach stron trzecich.
Trzecie podejście wykorzystało ForkJoinPool.managedBlock, implementując niestandardową ManagedBlocker, która owinęła pętlę odczytu pliku. Ta blokada sprawdzała okresowo isCancelled(), jednocześnie umożliwiając puli uruchamianie kompensacyjnych wątków poprzez protokół blokujący. Wybraliśmy trzecie rozwiązanie, ponieważ zachowało istniejącą architekturę równoległych strumieni, jednocześnie wyraźnie informując pulę o blokujących operacjach. Umożliwiło to zrównoważenie responsywności anulowania i przepustowości bez przepisania całej warstwy WE/WY.
Wynikiem był system, w którym prośby o anulowanie propagowały się w ciągu kilku sekund zamiast minut. Pula dynamicznie zwiększała się do pięćdziesięciu wątków podczas szczytów WE/WY bez ręcznej konfiguracji. Saturacja CPU pozostawała wysoka przez cały czas trwania obciążenia, a zakończenie zadania stało się niezawodne, nawet podczas dużego zatoru w sieci.
Jak ForkJoinPool wykrywa blokowanie wątków bez eksploracyjnych wywołań managedBlock, i jaki jest próg do uruchomienia wątków kompensacyjnych?
Pula wewnętrznie śledzi stany wątków roboczych za pomocą 64-bitowego pola ctl, które reprezentuje liczbę aktywnych wątków w porównaniu do liczby zaparkowanych. Liczy wątki jako "aktywne", gdy wykonują zadania, ale nie może rozróżnić pomiędzy intensywną pracą CPU a blokującym WE/WY bez wskazówek programistycznych. Gdy wątek blokuje na monitorze synchronizacji lub WE/WY bez użycia managedBlock, pula obserwuje tylko spadek liczby pracy, którą można ukraść, i dostępnych wątków. Może ostatecznie utknąć, gdy osiągnięty zostanie poziom równoległości, a żadne sygnały postępu nie docierają. Wątki kompensacyjne są uruchamiane niezawodnie tylko wtedy, gdy managedBlock jest wywoływane, lub kiedy wewnętrzne blokowanie JVM jest wykrywane za pomocą liczników Unsafe.park, ale domyślny próg jest nieprzejrzysty i nierzetelny dla niestandardowego kodu blokującego.
Dlaczego ForkJoinTask.join() nie zwraca natychmiastowo, gdy zadanie jest anulowane, i jak różni się to od Future.get() z limitem czasu?
join() wewnętrznie wywołuje doJoin(), które implementuje mechanizm "pomocy", gdzie wątek wywołujący wykonuje lub kradnie inną pracę, aż docelowe zadanie zostanie ukończone. Dzieje się tak bez względu na status anulowania, ponieważ anulowanie jedynie zapobiega nowemu rozdzielaniu subtasks i ustawia flagę zakończenia. Metoda nie sprawdza flagi anulowania przed oczekiwaniem, ani nie rzuca CancellationException przy wejściu. W przeciwieństwie do tego, Future.get() na ForkJoinTask (który implementuje Future) sprawdza status anulowania natychmiastowo i może rzucić CancellationException bez oczekiwania. Ta różnica jest kluczowa, ponieważ join() jest zaprojektowane dla współpracy w ramach puli, podczas gdy get() jest dla zewnętrznych klientów oczekujących standardowych semantyk Future.
Jaka jest interakcja między poziomem równoległości ForkJoinPool a Runtime.availableProcessors(), i dlaczego ustawienie równoległości wyższej niż dostępne procesory może poprawić przepustowość dla blokujących operacji?
Domyślna wspólna pula inicjalizuje się z availableProcessors() - 1, aby zarezerwować jeden rdzeń dla wątku aplikacji lub zbierania śmieci. Równoległość definiuje docelową liczbę aktywnych wątków, nie jest twardym maksymalnym; pula może tworzyć więcej wątków, jeśli managedBlock wskazuje na blokującą pracę, ale dąży do utrzymania tylko wielkości równoległości wątków naprawdę aktywnych. W przypadku operacji blokujących, ustawienie równoległości wyższej niż liczba rdzeni (np. 2x lub 3x rdzeni) pozwala harmonogramowi utrzymać CPU w ruchu, podczas gdy inne wątki czekają na WE/WY. Modeluje to ograniczenie "wątek-na-rdzeń", zapewniając istnienie zadań do uruchomienia dla każdego rdzenia mimo blokowania. Jednak wymaga to ostrożnego dopasowania, aby zapobiec nadmiernym narzutom kontekstu, gdy wskaźnik blokady jest błędnie oszacowany.