Trochę historii

Wiecie, że w Javie możliwe było programowanie współbieżne już w wersji 1.0? Oczywiście, przy użyciu klasy Thread. Możliwość ta pojawiła się w 1996 (obchodziłam w tym roku osiemnaste urodziny!), jednak dość szybko się okazało, że nie jest ono wcale proste… Większe i bardziej złożone aplikacje potrzebowały lepszych narzędzi do okiełznania wątków; w bibliotece standardowej Javiy brakowało konstrukcji niezbędnych do modelowania pewnych wzorców programowania współbieżnego, a sam kod był zwykle mało czytelny i bardzo podatny na błędy.

Tak więc osiem lat później - w 2004 roku - wydano wersję Java SE 5.0, w której znalazły się - przyćmione przez genericsy, enumy, varargi i adnotacje - nowe klasy w java.util.concurrent (concurrency utilities overview):

• Executor framework - który sprawił, że nie musimy już odpalać wątków ręcznie, dostajemy możliwość zakolejkowania tasków (Runnable) w schedulerze
• nowe, efektywne w programowaniu współbieżnym implementacje kolekcji Map, List i Queue (interfejsy Queue i BlockingQueue pojawiły się dopiero w tej wersji)
• narzędzia do synchronizacji wątków: muteksy, semafory, bariery, latche
• locki w java.util.concurrent.lock - “poprawiające” nieco synchronizację przy użyciu mechanizmu “synchronized {}” (dodano możliwosć zdefiniowania timeoutu przy ooczekiwaniu na lock, wielu zmiennych warunkowych per lock czy przerwania wątków czekających na złapanie locka).

Minęło kolejne dziesięć lat i w 2014 pojawiłą się Java 8. I znowu: lambdy przyćmiły nieco nowy mechanizm zarządzania współbieżnością: mechanizm fork-join, dzięki któremu zamiast jednej listy wątków oczekuących na wykonanie przez egzekutor, możliwe jest wykorzystanie wielu takich list, z których wątki mogą “podbierać” zadania (algorytm “work-stealing”). Pojawiło się też potężne rozszerzenie biblioteki standardowej o Completabe Future API (próbę stworzenia modelu programowania asymchronicznego w Javie, który jednak - wobec modelu async/await obecnego w wielu innych językach programowania - niewielu programistom przypadł do gustu.

A teraz mija kolejne 10 lat - i na scenę wkracza projekt Loom, który przynosi kolejny upgrade: wątki wirtualne (virtual threads), współbieżność strukturalną - a może raczej: ustrukturyzowaną? (structured concurrency) - oraz wartości kontekstowe (scoped values)

Przykłady i demo programów z wątkami wirualnymi: demo repo

Co to jest wątek?

Wątek w javie był do tej pory jedynie cienkim wraperem na wątek systemu operacyjnego (OS thread lub kernel thread). Taki wątek (zwany pthread od platform thread) jest dość drogim zasobem systemowym:

• wystartowanie wątku to czas ok 1 milisekundy
• potrzebuje alokacji pamięci (ok 2MB) na trzymanie stosu
• podlega schedulerowi w systemie operacyjnym, a przełączenie między wątkami jest ciężką operacją (patrz context switch).

Słabość modelu thread per request

Z tego też powodu “commodity hardware” jest dość mocno ograniczony jeśli chodzi o tworzenie wątków platformowych. Okazuje się, że w przypadku mocno obciążonych serwerów aplikacyjnych, które działają w modelu “odpalamy jeden wątek na obsługę jednego żądania” po osiągnieciu kilkuset tysięcy requestów po prostu brakuje pamięci na odpalenie kolejnego wątku i serwer umiera…Jeśli zaś mądrze wykorzystujemy pule wątków (na przykład ograniczając liczbę wątków) to i tak możemy mieć problem z wydajnością (np. przepustowością systemu), bo mimo wykorzystania wszystkich wątków w puli brakuje mocy przerobowych i kolejne połączenia/requesty nie mogą być obsłużone.

W świecie JVM powstało kilka prób radzenia sobie z tym problemem, na przykład: model programowania przy użyciu aktorów (Akka w Scali i Javie), Korutyny w Kotlinie, Vertx, a w samej Javie był jeszcze …

Niedoskonały CompletableFuture

Model programowana asynchronicznego w Javie (supplyAsync…, andThenApply… andThenApply…) pozostawia wciąż wiele do życzenia:

• jest bardzo mało czytelny
• trudno jest wnioskować na temat przebiegu sterowania
• nie wiadomo, w jakim wątku wykonuą się Runnable (debugowanie i logowanie jest bardzo trudne)
• trudno testować jednostkowo

Wątki wirtualne i wątki platformowe

Wątki platformowe

• mapują się 1 do 1 na wątki kernela, które są kolejkowane przez scheduler systemu operacyjnego
• posiadają duży stos i wymagają zasobów zarządzanych przez system operacyjny
• nadają się do wielu zadań, ale są drogim zasobem
• mają domyślnie nadane nazwy
• mogą być demonowe lub nie-demonowe; wątek w którym działa metoda main() to główny wątek nie-demonowy; JVM rozpoczyna sekwencję zamykania gdy wszystkie nie-demonowe wątki się zakończyły

Wątki wirtualne

• wątki wirtualne to wątki w trybie użytkownika (user-mode threads) kolejkowane przez JVM, a nie przez schedulera systemu operacyjnego
• mapują się w proporcji v:p (v > p) na wątki systemu operacyjnego, gdzie wiele v-wątków mapuje się na małą liczbę p-wątków
• p-wątki to tzw. carrier threads (“przewoźnicy” albo “kurierzy”) których nie tworzymy już bezpośrednio. To wątki w specjalnej puli fork-join zarządzanej przez JVM. Jeden wątek platformowy może wykonywać wiele różnych wątków wirtualnych na raz
• gdy v-wątek działający na pewnym p-wątku rozpoczyna wykonywanie blokującej operacji I/O, jego “struktura” (dane) może zostać przeniesiona na heap, a gdy kernel obsłuży operację I/O, JVM przywróci v-wątek z heapa z powrotem na - być może już inny - p-wątek.
• currentThread() zwróci informację o v-wątku (gdy odpalamy v-wątek, nie mamy dostępu do informacji o p-wątku)
• nie mają domyślnej nazwy; jeśli nie ustawimy jej, getName() zwróci pusty napis
• mają stały i niezmienialny priorytet
• są demonowe i nie blokują zamknięcia JVM-a

Tworzenie i uruchamianie wątków

Programista pisze kod “po staremu” - tworzy runnable i uruchamia je przy pomocy wątków , aby odpalić w nich długo trwające operacje. Robi to po to, aby nie blokować głównego wątku aplikacji. Może to zrobić korzystając z metod-fabryk wątków.

W Javie 21 dodano dwie metody, z których pierwsza tworzy bezpośrednio wątki platformowe, a druga - wątki wirtualne:

• Thead.ofPlatform() tworzy buildera/fabrykę do tworzenia wątków platformowych

1
2
3
4
5
6
7
8

 // Wystartuj wątek-demon i  wykonaj w nim runnable
  Thread thread = Thread.ofPlatform().daemon().start(runnable);

  // Utwórz niewystartowany wątek o nazwie "duke"; należy wywołać na nim metodę "start" aby został on zakolejkowany do wykonania
  Thread thread = Thread.ofPlatform().name("duke").unstarted(runnable);

  // Fabryka wątków tworząca wątki-demony o nazwach "worker-0", "worker-1", ...
  ThreadFactory factory = Thread.ofPlatform().daemon().name("worker-", 0).factory();

• Thread.ofVirtual() - tworzy buildera do tworzenia wątków wirtualnych lub do utworzenia fabryki wątków wirtalnych

1
2
3
4
5

  // Uruchom runnable na wątku wirtualnym
  Thread thread = Thread.ofVirtual().start(runnable);

  // Utwórz fabrykę wątków wirtualnych
  ThreadFactory factory = Thread.ofVirtual().factory();

Prosty testowy programik

Na podstawie: JosePaumard

Poniższy program MaxThreads.java uruchamia podaną z wiersza poleceń liczbę wątków wirtualnych bądź platformowych, każe każdemu z nich spać przez dwie sekundy, czeka, aż wątki się zakończą i wypisuje czas od ich uruchomienia do zakończenia.

W taki sposób uruchomię dziesięć wątków platformowych:

java --source 21 --enable-preview MaxThreads.java plat 10

A tak uruchomię 15 wątków wirtualnych:

java --source 21 --enable-preview MaxThreads.java virt 15

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58

import java.time.Duration;
import java.time.Instant;
import java.util.Set;
import java.util.Arrays;
import java.util.concurrent.ConcurrentHashMap;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
import java.util.stream.IntStream;

public class MaxThreads {

    // --enable-preview

    private static final void exitWithMessage() {
          System.out.println("Expected argument: virt or plat");
          System.exit(1);
    }
    public static void main(String[] args) throws InterruptedException {
        System.out.println(Arrays.toString(args));
        if (args.length != 2) {
          exitWithMessage();
        }

        var builder = switch(args[0]) {
          case "virt" -> Thread.ofVirtual();
          case "plat" -> Thread.ofPlatform();
          default -> null;
        };

        var count = Integer.parseInt(args[1]);
        
        if (builder == null) {
          exitWithMessage();
        }
        // virtual thread
        var threads =
              IntStream.range(0, count)
                    .mapToObj(index ->
                          builder
                                .name("platform-", index)
                                .unstarted(() -> {
                                    try {
                                        Thread.sleep(2_000);
                                    } catch (InterruptedException e) {
                                        throw new RuntimeException(e);
                                    }
                                }))
                    .toList();

        Instant begin = Instant.now();
        threads.forEach(Thread::start);
        for (Thread thread : threads) {
            thread.join();
        }
        Instant end = Instant.now();
        System.out.println("Duration = " + Duration.between(begin, end));
    }
}

Wątki platformowe - uruchomienie

Spróbuję uruchomić ten program w pętli, za każdym razem dziesięciokrotnie zwiększając liczbę wątków. Zaczynam od wątków platformowych, iteruję od dziesięciu do miliona.

JVM na mojej maszynie nie udźwignął diesięciu tysięcy wątków:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37

➜  jvm for i in {1..6}; do java --source 21  --enable-preview MaxThreads.java plat $(( 10 ** $i )); done > out.txt 2>&1
➜  jvm cat out.txt
───────┬──────────────────────────
       │ File: out.txt
───────┼─────────────────────────
   1   │ [plat, 10]
   2   │ Duration = PT2.00168193S
   3   │ [plat, 100]
   4   │ Duration = PT2.009612897S
   5   │ [plat, 1000]
   6   │ Duration = PT2.056957326S
   7   │ [plat, 10000]
   8   │ [1,604s][warning][os,thread] Failed to start thread "Unknown thread" - pthread_create failed (EAGAIN) for attributes: stacksize: 1024k, guardsize: 0k, detached.
   9   │ [1,604s][warning][os,thread] Failed to start the native thread for java.lang.Thread "platform-8979"
  10   │ Exception in thread "main" java.lang.OutOfMemoryError: unable to create native thread: possibly out of memory or process/resource limits reached
  11   │     at java.base/java.lang.Thread.start0(Native Method)
  12   │     at java.base/java.lang.Thread.start(Thread.java:1526)
  13   │     at java.base/java.lang.Iterable.forEach(Iterable.java:75)
  14   │     at MaxThreads.main(MaxThreads.java:51)
  15   │ [plat, 100000]
  16   │ [2,080s][warning][os,thread] Failed to start thread "Unknown thread" - pthread_create failed (EAGAIN) for attributes: stacksize: 1024k, guardsize: 0k, detached.
  17   │ [2,080s][warning][os,thread] Failed to start the native thread for java.lang.Thread "platform-8979"
  18   │ Exception in thread "main" java.lang.OutOfMemoryError: unable to create native thread: possibly out of memory or process/resource limits reached
  19   │     at java.base/java.lang.Thread.start0(Native Method)
  20   │     at java.base/java.lang.Thread.start(Thread.java:1526)
  21   │     at java.base/java.lang.Iterable.forEach(Iterable.java:75)
  22   │     at MaxThreads.main(MaxThreads.java:51)
  23   │ [plat, 1000000]
  24   │ [4,175s][warning][os,thread] Failed to start thread "Unknown thread" - pthread_create failed (EAGAIN) for attributes: stacksize: 1024k, guardsize: 0k, detached.
  25   │ [4,175s][warning][os,thread] Failed to start the native thread for java.lang.Thread "platform-8976"
  26   │ Exception in thread "main" java.lang.OutOfMemoryError: unable to create native thread: possibly out of memory or process/resource limits reached
  27   │     at java.base/java.lang.Thread.start0(Native Method)
  28   │     at java.base/java.lang.Thread.start(Thread.java:1526)
  29   │     at java.base/java.lang.Iterable.forEach(Iterable.java:75)
  30   │     at MaxThreads.main(MaxThreads.java:51)
───────┴────────────────────────
➜  jvm

Wątki wirtualne - uruchomienie

A jak sobie radzą wirtualne? Dają radę :)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19

➜  jvm for i in {1..6}; do java --source 21  --enable-preview MaxThreads.java virt $(( 10 ** $i )); done > out_virt.txt  2>&1
➜  jvm cat out_virt.txt
───────┬──────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────
       │ File: out_virt.txt
───────┼──────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────
   1   │ [virt, 10]
   2   │ Duration = PT2.005876979S
   3   │ [virt, 100]
   4   │ Duration = PT2.00849789S
   5   │ [virt, 1000]
   6   │ Duration = PT2.025809059S
   7   │ [virt, 10000]
   8   │ Duration = PT2.138172909S
   9   │ [virt, 100000]
  10   │ Duration = PT2.785108472S
  11   │ [virt, 1000000]
  12   │ Duration = PT11.038094874S
───────┴──────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────
➜  jvm

W środku

Dotychczas wątek mógł być przerwany (InterruptedException), mógł się zakończyć normalnie lub rzucić wyjątkiem. Teraz wątek moze się “zaparkować”, “zawiesić” - jak zwał, tak zwał - i po pewnym czasie, zwykle po wykonaiu przez kernel operacji I/O - zostanie on przywrócony do wykonania już na innym wątku platformowym. Jak to się dzieje?

Jeśli spojrzycie w kod klasy Thread, w sczególności na implementację metody sleep, zobaczycie klasę Continuation, która jest wewnętrznym “wraperem” wokół wątku platformowego. Jej metoda yialdContinuation jest odpowiedzialna za “oddanie sterowania” do wątku platformowego. Jeśli włączymy dostęp do wewnętrznego modułu (patrz wątek na liście openjdk): java --add-opens java.base/jdk.internal.vm=ALL-UNNAMED <your-main-class> wówczas możemy użyć ContinuationScope do tego, aby - będąc w tasku wewnątrz “Continuation” - zawiesić bieżący task, a reszta kodu nie będzie wykonana. Jeśli uruchimimy Continuation ponownie, zostanie ona wykonana do końca. Nie jest to publiczne API (i nie wiadomo, czy zostanie upublicznione), ale ciekawie jest zobaczyć bebechy.

Na przykład ten kod pokazuje, w jaki sposób można sterować wykonaniem wątków wirtualnych przy pomocy Continuation: G3_Continuation_Yield.

Structured concurency

Jeśli mamy już wątki wirtualne, możemy spróbować programowania asynchronicznego przy użyciu strukturalnej współbieżności - będziemy pisać kod imperatywny bez sięgania po CompletableFuture. StructuredTaskScope pozwala na utworzenie lekkigo “zakresu” (scope) wewnątrz metody, pozwalającego na

• utworzenie wielu v-wątków (fork())
• zdefiniowanie, kiedy scope ma zostać zamknięty (on success, on failure)
• uwidocznienie zależności między wątkami w thread dump-ach
• można skasować (cancel) wątek nadrzędny, a wszystkie podrzędne wątki również się skasują
• nie trzeba jawnie deklarować ExecutorSerwisu (globalnego z punktu widzenia lokalnej metody), którego nie wiadomo kiedy zamknąć

Na przykład:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13

    Callable<String> task1 = ...
    Callable<Integer> task1 = ...

    try (var scope = new StructuredTaskScope<Object>()) {

        Subtask<String> subtask1 = scope.fork(task1);
        Subtask<Integer> subtask2 = scope.fork(task2);

        scope.join();

        ... process results/exceptions ...

    } // close

Scoped value

Dotychczasowy ThreadLocal został - wewnątrz StructuredTaskScope - zatąpiony mechanizmem ScopedValue.

Podobnie jak ThreadLocal służy do “dzielenia się” wartoscią z wątkami-dziećmi bez konieczności przekazywania jej w parametrze. Jednak działanie ScopedValue jest takie, że jej wartość jes “związana” i widoczna jedynie w czasie trwania dynamcznego zakresu (“scope-u”) i nie jest dostępna poza nim. Może zostać ona związana z inną wartością (i wątki-dzieci będą widziały nową wartość).

Przykład użycia

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

  private static final ScopedValue<String> NAME = ScopedValue.newInstance();

    ScopedValue.runWhere(NAME, "duke", () -> {
        try (var scope = new StructuredTaskScope<String>()) {

            scope.fork(() -> childTask1());
            scope.fork(() -> childTask2());
            scope.fork(() -> childTask3());

            ...
         }
    });

O Scoped Values przeczytasz więcej we wpisie Jeśli nie ThreadLocal, to ScopedValue.

Przygotowania w ekosystemie

Co ciekawe, nawet Spring Boot przygotował się na to, aby uruchamiać na swoim wewnętrzym Tomkacie obsługę żądań na wątkach wirtualnych. Wystarczy zaimplementować i wystawić jako @Bean klasę TomcatProtocolHandlerCustomizer która definiuje callable służące do ustawienia odpowiedniego executora użuwanego w synchroniczncych żądaniach: retur handler -> handler.setExecutor(Executors.newVirtualThreadPoolPerTaskExecutor()). Po szczegóły i demo odsyłam do świetnej prezentacji JosePaumard: Virtual Threads and Structured Concurrency in Java 21 With Loom, z której zaczerpnęłam inspirację (i kod) do napisania tego wpisu.

Zbliża się wieczór, czas zająć się dziećmi i domem.

Happy coding!