Thread Wars: Project Loom Strikes Back

thread wars: project loom strikes back

jep-425 is here!

файберы корутины continuations легковесные потоки user-level потоки

structured concurrency JEP 428 файберы легковесные потоки user-level потоки корутины
continuations

questions

questions • Java повторяет за Котлином? • Котлин теперь перейдет
на Loom? • Почему так долго делали? Вон в Котлине давно уже…

• А вообще уже работает? Пора переходить? • У меня
теперь все ускорится? • У меня теперь все замедлится? questions • Java повторяет за Котлином? • Котлин теперь перейдет на Loom? • Почему так долго делали? Вон в Котлине давно уже…

agenda 1. Разбираемся с проблемой и терминологией 2. Сравниваем Loom
c другими реализациями 3. Loom: смотрим на подводные камни

a long time ago... Хотим писать код, который будет параллельно
исполняться

Хотим писать код последовательный: int v = foo(); v +=
2; bar(v); the problem

Хотим писать код параллельный: Thread t = new Thread(() ->
{ int v = foo(); v += 2; bar(v); }); t.start(); ... t.join(); the problem

А как реализовать java.lang.Thread? the problem

А как реализовать java.lang.Thread? Очевидное решение – встать на плечи
гигантов! java.lang.Thread <=> OS-thread (1:1 схема, один к одному) the problem

java.lang.Thread <=> OS-thread + в JVM ничего делать не нужно
+ scheduler, над которым работают 50 лет the problem

java.lang.Thread <=> OS-thread + в JVM ничего делать не нужно
+ scheduler, над которым работают 50 лет − ограничение по количеству − долгое создание − ограниченный стек the problem

final int count = 100_000; final Thread[] threads = new
Thread[count]; for (int i = 0; i < count; i++) { System.out.println("Starting thread #" + i); final int n = i; threads[i] = new Thread(() -> { Thread.sleep(Duration.ofSeconds(10)); System.out.println("Thread #" + n + " finished"); }); threads[i].start(); } for (int i = 0; i < count; i++) { threads[i].join(); } the problem

Thread[count]; for (int i = 0; i < count; i++) { System.out.println("Starting thread #" + i); final int n = i; threads[i] = new Thread(() -> { Thread.sleep(Duration.ofSeconds(10)); System.out.println("Thread #" + n + " finished"); }); threads[i].start(); } for (int i = 0; i < count; i++) { threads[i].join(); } Starting thread #10775 Starting thread #10776 Starting thread #10777 Starting thread #10778 Starting thread #10779 Starting thread #10780 [1.232s][warning][os,thread] Failed to start thread "Unknown thread" – pthread_create failed (EAGAIN) for attributes: stacksize: 1024k, guardsize: 0k, detached. [1.232s][warning][os,thread] Failed to start the native thread for java.lang.Thread "Thread-10780" the problem

java.lang.Thread <=> OS-thread Следствие реализации: o Треды – дорогие! Лишний
раз их создавать не стоит (а хочется) o Значит треды будем пулить the problem

j.l.Thread j.l.Thread j.l.Thread j.l.Thread Thread pool OS-thread OS-thread OS-thread OS-thread
Tasks thread pool to the rescue

Вместо тредов таски в тредпуле: + Больше не боимся ограничений
OS! thread pool to the rescue

Thread[count]; for (int i = 0; i < count; i++) { System.out.println("Starting thread #" + i); final int n = i; threads[i] = new Thread(() -> { Thread.sleep(Duration.ofSeconds(10)); System.out.println("Thread #" + n + " finished"); }); threads[i].start(); } for (int i = 0; i < count; i++) { threads[i].join(); } thread pool to the rescue

ExecutorService es = Executors.newFixedThreadPool(4); for (int i = 0; i
< 100_000; i++) { System.out.println("Submitting task #" + i); es.execute(() -> { Thread.sleep(Duration.ofSeconds(10)); }); } thread pool to the rescue

OS! + Специализированные scheduler-ы thread pool to the rescue

OS! + Специализированные scheduler-ы − Долгий (бесконечный) таск забирает OS-thread thread pool to the rescue

j.l.Thread j.l.Thread j.l.Thread j.l.Thread OS-thread OS-thread OS-thread OS-thread Tasks Долгий
таск Thread pool thread pool to the rescue

Блокирующие вызовы: o Ожидание ответа по сети o Ожидание окончания
файловой операции o Ожидание монитора o Ожидание по таймауту (Thread.sleep) o … thread pool to the rescue

таск Thread pool socketChannel.read(buffer); thread pool to the rescue

таск Thread pool Блокирующий вызов Блокирующий вызов Блокирующий вызов thread pool to the rescue

OS! + Специализированные scheduler-ы − Долгий (бесконечный) таск забирает OS-thread − Блокирующие вызовы, как крайний случай thread pool to the rescue

to block or not to block Блокирующие вызовы: o Ожидание
ответа по сети o Ожидание окончания файловой операции o Ожидание монитора o Ожидание по таймауту (Thread.sleep) o …

файловой операции o Ожидание монитора o Ожидание по таймауту (Thread.sleep) o … Ожидание реакции OS. to block or not to block

файловой операции o Ожидание монитора o Ожидание по таймауту (Thread.sleep) o … Ожидание реакции OS. Но можно и не ждать! to block or not to block

OS предаставляет асинхронный API (EPOLL) Вместо занимающего тред ожидания ответа
подписываемся на событие. Когда оно случится, вызывается колбек. to block or not to block

Через EPOLL реализованы асинхронные API для работы с сетью и
файловой системой в managed языках. AsynchronousSocketChannel AsynchronousFileChannel ... to block or not to block

Через EPOLL реализованы асинхронные API для работы с сетью и
файловой системой в managed языках. <A> void read(ByteBuffer dst, long position, A attachment, CompletionHandler<Integer,? super A> handler); callback to block or not to block

j.l.Thread j.l.Thread j.l.Thread j.l.Thread OS-thread OS-thread OS-thread OS-thread Tasks Thread
pool socketChannel.read(buffer); thread pool to the rescue

pool asyncSocket.read(buffer, ..., callback); thread pool to the rescue

pool callback thread pool to the rescue

Проблема подхода с callback – уродование кода и сложность отладки
thread pool to the rescue

int v = foo(); validate(v); int k = bar(v); validate(k); baz(k); thread pool to the rescue

main callback int v = foo(); validate(v); int k = bar(v); validate(k); baz(k); foo(v -> { validate(v); bar(v, k -> { validate(k); baz(k); }); }); callback thread pool to the rescue

Альтернатива колбекам - Future Future<Integer> read(ByteBuffer dst, long position); to
block or not to block

Альтернатива колбекам - Future Future<Integer> read(ByteBuffer dst, long position); o
Вызов read неблокирующий, сразу возвращает результат o Вызов Future.get – блокирующий to block or not to block

Альтернатива колбекам - Future Future<Integer> read(ByteBuffer dst, long position); o
Вызов read неблокирующий, сразу возвращает результат o Вызов Future.get – блокирующий o Само по себе это не поможет, но помогут комбинаторы o Вместо вызова get описываем следующие шаги to block or not to block

main callback int v = foo(); validate(v); int k = bar(v); validate(k); baz(k); foo(v -> { validate(v); bar(v, k -> { validate(k); baz(k); }); }); callback to block or not to block

С комбинаторами – лучше. Callback hell больше нет, но код
все еще другой и непривычный. CompletableFuture.supplyAsync(this::foo). thenApply(v -> { validate(v); return v; }). thenApplyAsync(this::bar). thenApply(k -> { validate(k); return k; }). thenAcceptAsync(this::baz); to block or not to block

Итого: первый путь решения проблемы тредов – изменение исходного кода
to block or not to block

o Используем асинхронные API o Декомпозируем код с помощью Future и комбинаторов o Избегаем кода с блокировками на мониторах to block or not to block

o Используем асинхронные API o Декомпозируем код с помощью Future и комбинаторов o Избегаем кода с блокировками на мониторах После этого начинает работать подход с тредпулом. to block or not to block

таск Thread pool Блокирующий вызов to block or not to block

Основанные на ThreadPool и изменении подхода к написанию кода: +
Реализация на уровне библиотеки ± Сильное изменение кода

Основанные на ThreadPool и изменении подхода к написанию кода: +
Реализация на уровне библиотеки ± Сильное изменение кода – Хрупкое решение (в случае Java)

Аналогичный подход в функциональных языках: o Scala: ZIO, Cats Effect
o Pure functional

o Pure functional Реактивные фреймворки

o Pure functional Реактивные фреймворки: youtube.com/watch?v=YwG04UZP2a0

больше Влияние на код и язык меньше

больше Влияние на код и язык меньше ThreadPool + Callbacks/Future/
Combinators

больше меньше ThreadPool + Callbacks/Future/ Combinators CompletableFuture ReactiveFrameworks Влияние на
код и язык

thread wars episode i

to hide a thread pool И все же писать и
отлаживать императивный код хочется

И все же писать и отлаживать императивный код хочется А
что если преобразовывать код для работы с тредпулом… автоматически? to hide a thread pool

И все же писать и отлаживать императивный код хочется А
что если преобразовывать код для работы с тредпулом… автоматически? Так делают в C#/JS/С++/Rust/Kotlin to hide a thread pool

youtube.com/watch?v=rB5Q3y73FTo to hide a thread pool

main callback int v = foo(); validate(v); int k =
bar(v); validate(k); baz(k); foo(v -> { validate(v); bar(v, k -> { validate(k); baz(k); }); }); callback to hide a thread pool

fun test(input: Int): Int { val seed = Random.nextInt() delay(100)
print("$input -> $seed") delay(100) return input + seed } to hide a thread pool

public suspend fun delay(timeMillis: Long) to hide a thread pool
fun test(input: Int): Int { val seed = Random.nextInt() delay(100) print("$input -> $seed") delay(100) return input + seed }

public suspend fun delay(timeMillis: Long) to hide a thread pool
suspend fun test(input: Int): Int { val seed = Random.nextInt() delay(100) print("$input -> $seed") delay(100) return input + seed }

fun main() = runBlocking { for (i in 1..100_000) {
launch { val result = test(i) println("Coroutine $i returned $result") } } } suspend fun test(input: Int): Int { ... } to hide a thread pool

suspend fun test(input: Int): Int { val seed = Random.nextInt()
delay(100) print("$input -> $seed") delay(100) return input + seed } А как это работает то? to hide a thread pool

fun main() = runBlocking { for (i in 1..100_000) {
launch { val result = test(i) println("Coroutine $i returned $result") } } } suspend fun test(input: Int): Int { ... } Запуск таска на тредпуле to hide a thread pool

delay(100) print("$input -> $seed") delay(100) return input + seed } to hide a thread pool

delay(100) print("$input -> $seed") delay(100) return input + seed } callback to hide a thread pool

delay(100) print("$input -> $seed") delay(100) return input + seed } Object test(int input, Continuation<?> cont) { ... } to hide a thread pool

delay(100) print("$input -> $seed") delay(100) return input + seed } Object test(int input, Continuation<?> cont) { ... switch (cont.label) { } ... } 0 to hide a thread pool

delay(100) print("$input -> $seed") delay(100) return input + seed } Object test(int input, Continuation<?> cont) { ... switch (cont.label) { case 0: cont.input = input; cont.seed = Random.nextInt(); cont.label = 1; delay(100, cont); break; ... } ... } 1 0 to hide a thread pool

delay(100) print("$input -> $seed") delay(100) return input + seed } Object test(int input, Continuation<?> cont) { ... switch (cont.label) { ... case 1: print("$cont.input -> $cont.seed") cont.label = 2; delay(100, cont); break; ... } ... } 2 1 0 to hide a thread pool

delay(100) print("$input -> $seed") delay(100) return input + seed } Object test(int input, Continuation<?> cont) { ... switch (cont.label) { ... case 2: cont.caller.resumeWith( cont.input + cont.seed); } ... } 2 1 0 to hide a thread pool

delay(100) print("$input -> $seed") delay(100) return input + seed } to hide a thread pool

delay(100) print("$input -> $seed") delay(100) return input + seed } callback callback to hide a thread pool

Suspend функции: o Преобразуются компилятором в switch и работу с
Continuation to hide a thread pool

Continuation o Изначально появляются в библиотеке или как обертки над async API (см. suspendCoroutine) to hide a thread pool

Continuation o Изначально появляются в библиотеке или как обертки над async API (см. suspendCoroutine) o Caller suspend функции тоже suspend to hide a thread pool

to hide a thread pool

А чем за это все заплатим? to hide a thread
pool

fun test(): Int { val r = foo() validate(r) val
l = bar(r + 42) validate(l) return baz(r + l) } to hide a thread pool

l = bar(r + 42) validate(l) return baz(r + l) } fun validate() { ... } fun foo() { validate(...) ... } fun bar(i: Int) { foo() ... } fun baz(i: Int) { validate(i) ... } to hide a thread pool

l = bar(r + 42) validate(l) return baz(r + l) } fun validate() { ... } @Benchmark fun baseline() = runBlocking { test() } fun foo() { validate(...) ... } fun bar(i: Int) { foo() ... } fun baz(i: Int) { validate(i) ... } to hide a thread pool

l = bar(r + 42) validate(l) return baz(r + l) } fun validate() { ... } @Benchmark fun baseline() = runBlocking { test() } fun foo() { validate(...) ... } fun bar(i: Int) { foo() ... } fun baz(i: Int) { validate(i) ... } Benchmark Score Error Units baseline 0,337 ± 0,027 us/op to hide a thread pool

l = bar(r + 42) validate(l) return baz(r + l) } suspend fun validate() {…} @Benchmark fun baseline() = runBlocking { test() } fun foo() { validate(...) ... } fun bar(i: Int) { foo() ... } fun baz(i: Int) { validate(i) ... } Benchmark Score Error Units baseline 0,337 ± 0,027 us/op to hide a thread pool

suspend fun test(): Int { val r = foo() validate(r)
val l = bar(r + 42) validate(l) return baz(r + l) } suspend fun validate() {…} @Benchmark fun testSuspend() = runBlocking { test() } suspend fun foo() { validate(...) ... } suspend fun bar(i: Int) { foo() ... } suspend fun baz(i: Int) { validate(i) ... } Benchmark Score Error Units baseline 0,337 ± 0,027 us/op to hide a thread pool

suspend fun test(): Int { val r = foo() validate(r)
val l = bar(r + 42) validate(l) return baz(r + l) } suspend fun validate() {…} @Benchmark fun testSuspend() = runBlocking { test() } suspend fun foo() { validate(...) ... } suspend fun bar(i: Int) { foo() ... } suspend fun baz(i: Int) { validate(i) ... } Benchmark Score Error Units baseline 0,337 ± 0,027 us/op testSuspend 3,843 ± 0,278 us/op to hide a thread pool

Benchmark Score Error Units baseline 0,337 ± 0,027 us/op testSuspend
3,843 ± 0,278 us/op fun test(): Int { val r = foo() validate(r) val l = bar(r + 42) validate(l) return baz(r + l) } fun validate() { ... } @Benchmark fun baseline() = runBlocking { test() } fun foo() { validate(...) ... } fun bar(i: Int) { foo() ... } fun baz(i: Int) { validate(i) ... } to hide a thread pool

3,843 ± 0,278 us/op fun test(): Int { val r = foo() validate(r) val l = bar(r + 42) validate(l) return baz(r + l) } fun validate() { ... } @Benchmark fun testNoInline() = runBlocking { test() } fun foo() { validate(...) ... } fun bar(i: Int) { foo() ... } fun baz(i: Int) { validate(i) ... } @CompilerControl(CompilerControl.Mode.DONT_INLINE) to hide a thread pool

3,843 ± 0,278 us/op testNoInline 2,677 ± 0,075 us/op fun test(): Int { val r = foo() validate(r) val l = bar(r + 42) validate(l) return baz(r + l) } fun validate() { ... } @Benchmark fun testNoInline() = runBlocking { test() } fun foo() { validate(...) ... } fun bar(i: Int) { foo() ... } fun baz(i: Int) { validate(i) ... } @CompilerControl(CompilerControl.Mode.DONT_INLINE) to hide a thread pool

А чем за это все заплатим? o Производительностью преобразованного кода
to hide a thread pool

А чем за это все заплатим? o Производительностью преобразованного кода
o Памятью (из-за создания Continuation-ов) to hide a thread pool

Автоматическое преобразование кода в CPS o Используется в C#, JS,
C++, Kotlin

C++, Kotlin o Реализация на уровне компилятора и библиотеки + Минимальное изменение исходного кода − Издержки по производительности и по памяти

C++, Kotlin o Реализация на уровне компилятора и библиотеки + Минимальное изменение исходного кода − Издержки по производительности и по памяти ± Раскраска кода (suspend функциями) – Все еще хрупкое решение!

код и язык

код и язык Автоматическое преобразование кода в CPS a.k.a stackless корутины

код и язык Автоматическое преобразование кода в CPS a.k.a stackless корутины С#/JS/C++ Kotlin

Thread[count]; for (int i = 0; i < count; i++) { System.out.println("Starting thread #" + i); final int n = i; threads[i] = new Thread(() -> { Thread.sleep(Duration.ofSeconds(10)); System.out.println("Thread #" + n + " finished"); }); threads[i].start(); } for (int i = 0; i < count; i++) { threads[i].join(); } loom

Thread[count]; for (int i = 0; i < count; i++) { System.out.println("Starting thread #" + i); final int n = i; threads[i] = new Thread(() -> { Thread.sleep(Duration.ofSeconds(10)); System.out.println("Thread #" + n + " finished"); }); threads[i].start(); } for (int i = 0; i < count; i++) { threads[i].join(); } Starting thread #10775 Starting thread #10776 Starting thread #10777 Starting thread #10778 Starting thread #10779 Starting thread #10780 [1.232s][warning][os,thread] Failed to start thread "Unknown thread" – pthread_create failed (EAGAIN) for attributes: stacksize: 1024k, guardsize: 0k, detached. [1.232s][warning][os,thread] Failed to start the native thread for java.lang.Thread "Thread-10780" loom

Thread[count]; for (int i = 0; i < count; i++) { System.out.println("Starting thread #" + i); final int n = i; threads[i] = new Thread(() -> { Thread.sleep(Duration.ofSeconds(10)); System.out.println("Thread #" + n + " finished"); }); threads[i].start(); } for (int i = 0; i < count; i++) { threads[i].join(); } loom

Thread[count]; for (int i = 0; i < count; i++) { System.out.println("Starting thread #" + i); final int n = i; threads[i] = Thread.ofVirtual().start(() -> { Thread.sleep(Duration.ofSeconds(10)); System.out.println("Thread #" + n + " finished"); }); } for (int i = 0; i < count; i++) { threads[i].join(); } loom

Thread[count]; for (int i = 0; i < count; i++) { System.out.println("Starting thread #" + i); final int n = i; threads[i] = Thread.ofVirtual().start(() -> { Thread.sleep(Duration.ofSeconds(10)); System.out.println("Thread #" + n + " finished"); }); } for (int i = 0; i < count; i++) { threads[i].join(); } $ javac --release 20 --enable-preview Test.java Note: Test.java uses preview features of Java SE 20. $ java --enable-preview Test loom

Thread[count]; for (int i = 0; i < count; i++) { System.out.println("Starting thread #" + i); final int n = i; threads[i] = Thread.ofVirtual().start(() -> { Thread.sleep(Duration.ofSeconds(10)); System.out.println("Thread #" + n + " finished"); }); } for (int i = 0; i < count; i++) { threads[i].join(); } $ javac --release 20 --enable-preview Test.java Note: Test.java uses preview features of Java SE 20. $ java --enable-preview Test Thread #99986 finished Thread #99982 finished Thread #99990 finished Thread #99993 finished Thread #99996 finished Thread #99994 finished Thread #99997 finished Thread #99999 finished Thread #99988 finished loom

try (var executor = Executors.newVirtualThreadPerTaskExecutor()) { IntStream.range(0, 100_000).forEach(i -> {
executor.submit(() -> { Thread.sleep(Duration.ofSeconds(1)); System.out.println("worker " + i + " finished"); return i; }); }); } loom

executor.submit(() -> { Thread.sleep(Duration.ofSeconds(1)); System.out.println("worker " + i + " finished"); return i; }); }); } worker #99989 finished worker #99982 finished worker #99991 finished worker #99994 finished worker #99993 finished worker #99999 finished worker #99995 finished worker #99992 finished worker #99998 finished loom

loom: under the hood

Важно! Виртуальные потоки – это инкубатор фича. К релизу многое
может поменяться миллион раз!

V N M P P P P виртуальные потоки OS-потоки
V V V V V V V V V V V V V o N:M threading loom: under the hood

V V V V V V V V V V V V V o N:M threading o Виртуальный поток: o исполняется (на OS- потоке) o ожидает loom: under the hood

V V N M P P P P виртуальные потоки
OS-потоки V V V V V V V V V V V V o N:M threading o Виртуальный поток: o исполняется (на OS- потоке) o ожидает o Снятие с OS-потока кооперативное loom: under the hood

V V N M o N:M threading o Виртуальный поток:
o исполняется (на OS- потоке) o ожидает o Снятие с OS-потока кооперативное o Планировщик решает, кто следующий P P P P виртуальные потоки OS-потоки V V V V V V V V V V V V loom: under the hood

Когда виртуальный поток снимается с OS потока? loom: under the
hood

Когда виртуальный поток снимается с OS потока? o Thread.yield() public
static void yield() { if (currentThread() instanceof VirtualThread vthread) { vthread.tryYield(); } else { yield0(); } } loom: under the hood

Когда виртуальный поток снимается с OS потока? o Thread.yield(), Thread.sleep(…)
loom: under the hood

o LockSupport.park(…) loom: under the hood

o LockSupport.park(…) public static void park(Object blocker) { Thread t = Thread.currentThread(); setBlocker(t, blocker); try { if (t.isVirtual()) { VirtualThreads.park(); } else { U.park(false, 0L); } } finally { setBlocker(t, null); } } loom: under the hood

o LockSupport.park(…) o ReentrantLock, BlockingQueue, Mutex loom: under the hood

o LockSupport.park(…) o ReentrantLock, BlockingQueue, Mutex o java.net/java.nio.channels loom: under the hood

o LockSupport.park(…) o ReentrantLock, BlockingQueue, Mutex o java.net/java.nio.channels "Блокирующие" чтение и запись по сети вместо блокировки OS-потока снимают виртуальный поток loom: under the hood

public int read(ByteBuffer buf) throws IOException { ... configureSocketNonBlockingIfVirtualThread(); n
= IOUtil.read(fd, buf, -1, nd); if (blocking) { while (IOStatus.okayToRetry(n) && isOpen()) { park(Net.POLLIN); n = IOUtil.read(fd, buf, -1, nd); } } ... sun.nio.ch.SocketChannelImpl.java loom: under the hood

А что, если поток заблокируется на IO? loom: under the
hood

V А что, если поток заблокируется на IO? V N
M P P P P V V V V V V V V V V V V FileInputStream.read() loom: under the hood

M P P P P V V V V V V V V V V V V FileInputStream.read() public int read() throws IOException { long comp = Blocker.begin(); try { return read0(); } finally { Blocker.end(comp); } } loom: under the hood

M P P P P V V V V V V V V V V V V FileInputStream.read() public int read() throws IOException { long comp = Blocker.begin(); try { return read0(); } finally { Blocker.end(comp); } } P loom: under the hood

M P P P V V V V V V V V V V V V public int read() throws IOException { long comp = Blocker.begin(); try { return read0(); } finally { Blocker.end(comp); } } P loom: under the hood

При блокирующих операциях: o Либо виртуальные потоки снимаются с OS-потоков
o Либо временно добавляются новые OS-потоки loom: under the hood

При блокирующих операциях: o Либо виртуальные потоки снимаются с OS-потоков
o Либо временно добавляются новые OS-потоки Но есть два исключения! loom: under the hood

Но есть два исключения! o synchronized. Внутри виртуальные потоки не
снимаются loom: under the hood

executor.submit(() -> { Thread.sleep(Duration.ofSeconds(1)); System.out.println("worker " + i + " finished"); return i; }); }); } $ time java --enable-preview Test >log real 0m2.526s user 0m6.683s sys 0m0.865s loom: under the hood

Object[] monitors = new Object[100_000]; ... try (var executor =
Executors.newVirtualThreadPerTaskExecutor()) { IntStream.range(0, 100_000).forEach(i -> { executor.submit(() -> { synchronized (monitors[i]) { Thread.sleep(Duration.ofSeconds(1)); System.out.println("worker " + i + " finished"); return i; } }); }); } loom: under the hood

Executors.newVirtualThreadPerTaskExecutor()) { IntStream.range(0, 100_000).forEach(i -> { executor.submit(() -> { synchronized (monitors[i]) { Thread.sleep(Duration.ofSeconds(1)); System.out.println("worker " + i + " finished"); return i; } }); }); } $ time java --enable-preview Test >log real 208m25.003s user 0m20.695s sys 0m13.830s loom: under the hood

Executors.newVirtualThreadPerTaskExecutor()) { IntStream.range(0, 100_000).forEach(i -> { executor.submit(() -> { Thread.sleep(Duration.ofSeconds(1)); synchronized (monitors[i]) { System.out.println("worker " + i + " finished"); return i; } }); }); } loom: under the hood

Executors.newVirtualThreadPerTaskExecutor()) { IntStream.range(0, 100_000).forEach(i -> { executor.submit(() -> { Thread.sleep(Duration.ofSeconds(1)); synchronized (monitors[i]) { System.out.println("worker " + i + " finished"); return i; } }); }); } $ time java --enable-preview Test >log real 0m2.532s user 0m5.044s sys 0m0.601s loom: under the hood

снимаются o Причина: сложность реализации o Диагностика: -Djdk.tracePinnedThreads=full loom: under the hood

Executors.newVirtualThreadPerTaskExecutor()) { IntStream.range(0, 100_000).forEach(i -> { executor.submit(() -> { synchronized(monitors[i]) { Thread.sleep(Duration.ofSeconds(1)); System.out.println("worker " + i + " finished"); return i; } }); }); } $ java -Djdk.tracePinnedThreads=full --enable-preview Test loom: under the hood

Executors.newVirtualThreadPerTaskExecutor()) { IntStream.range(0, 100_000).forEach(i -> { executor.submit(() -> { synchronized(monitors[i]) { Thread.sleep(Duration.ofSeconds(1)); System.out.println("worker " + i + " finished"); return i; } }); }); } $ java -Djdk.tracePinnedThreads=full --enable-preview Test Thread[#22,ForkJoinPool-1-worker-1,5,CarrierThreads] java.base/java.lang.VirtualThread$VThreadContinuation.onPinned(VirtualThread.java:180) java.base/jdk.internal.vm.Continuation.onPinned0(Continuation.java:398) java.base/jdk.internal.vm.Continuation.yield0(Continuation.java:390) java.base/jdk.internal.vm.Continuation.yield(Continuation.java:357) java.base/java.lang.VirtualThread.yieldContinuation(VirtualThread.java:370) java.base/java.lang.VirtualThread.parkNanos(VirtualThread.java:532) java.base/java.lang.VirtualThread.doSleepNanos(VirtualThread.java:713) java.base/java.lang.VirtualThread.sleepNanos(VirtualThread.java:686) java.base/java.lang.Thread.sleep(Thread.java:538) Test.lambda$main$0(Test.java:78) <== monitors:1 ... loom: under the hood

снимаются o Причина: сложность реализации o Диагностика: -Djdk.tracePinnedThreads=full o Лечение: замена на ReentrantLock loom: under the hood

снимаются … o Если хоть один фрейм нативный Java => C++ => JNI => Java => … => Thread.yield() loom: under the hood

снимаются … o Если хоть один фрейм нативный o Диагностика: -Djdk.tracePinnedThreads=full o Лечение: нет, избегайте сквозных нативов loom: under the hood

А что, если в виртуальном потоке исполняется просто долгий CPU
intensive код? loom: under the hood

intensive код? for (int i = 0; i < 100_000; i++) { for (int j = 0; j < 100_000; j++) { while (z > 0) { // CPU intensive делишки } } } loom: under the hood

intensive код? V V N M P P P P V V V V V V V V V V V V FileInputStream.read() P CPU intensive loom: under the hood

executor.submit(() -> { Thread.sleep(Duration.ofSeconds(1)); System.out.println("worker " + i + " finished"); return i; }); }); } loom: under the hood

executor.submit(() -> { if (i < 8) { long result = getNthPrimeNumber(1_000_000 + i); System.out.println("Heavy worker " + i + " finished"); return (int) result; } else { Thread.sleep(Duration.ofSeconds(1)); System.out.println("worker " + i + " finished"); return i; }); }); } loom: under the hood

executor.submit(() -> { if (i < 8) { long result = getNthPrimeNumber(1_000_000 + i); System.out.println("Heavy worker " + i + " finished"); return (int) result; } else { Thread.sleep(Duration.ofSeconds(1)); System.out.println("worker " + i + " finished"); return i; }); }); } Heavy worker 6 finished Heavy worker 4 finished Heavy worker 0 finished Heavy worker 1 finished Heavy worker 5 finished Heavy worker 3 finished Heavy worker 2 finished Heavy worker 7 finished worker 14 finished worker 15 finished worker 51 finished worker 53 finished worker 55 finished ... loom: under the hood

intensive код? loom: under the hood

intensive код? o Это может быть проблемой (несколько CPU intensive потоков забьют весь пул) loom: under the hood

intensive код? o Это может быть проблемой (несколько CPU intensive потоков забьют весь пул) o Лечение: • Ручной Thread.yield() • Использование старых тредов loom: under the hood

Практические советы: 1. Избегайте synchronized, заменяйте на ReentrantLock 2. Избегайте
сквозных нативов 3. Избегайте CPU-intensive кода, вставляйте ручной yield -Djdk.tracePinnedThreads для обнаружения проблем 1 и 2. loom: under the hood

сквозных нативов 3. Избегайте CPU-intensive кода, вставляйте ручной yield -Djdk.tracePinnedThreads для обнаружения проблем 1 и 2. Пункты 1 и 3 можно реализовать на стороне рантайма! loom: under the hood

сквозных нативов 3. Избегайте CPU-intensive кода, вставляйте ручной yield -Djdk.tracePinnedThreads для обнаружения проблем 1 и 2. Пункты 1 и 3 можно реализовать на стороне рантайма! loom: under the hood they are billions

Но почему миллион виртуальных потоков создать можно, а миллион OS-тредов
– нет? better threads

Но почему миллион виртуальных потоков создать можно, а миллион OS-тредов
– нет? Стеки! better threads

better threads Стек Классические стеки: o 1MB зарезервировано 1MB

Стек Классические стеки: o 1MB зарезервировано o Большая часть страниц
не подключена o Страницы по 4KB или больше 4KB better threads

не подключена o Страницы по 4KB или больше o Подключаем лениво main() main better threads

не подключена o Страницы по 4KB или больше o Подключаем лениво main() main foo() foo better threads

не подключена o Страницы по 4KB или больше o Подключаем лениво main() main foo() bar() foo bar better threads

не подключена o Страницы по 4KB или больше o Подключаем лениво main() main foo() bar() foo bar baz() baz better threads

не подключена o Страницы по 4KB или больше o Подключаем лениво o В конце guard page main() main foo() bar() foo bar baz() baz better threads

Почему не сопоставить такой стек каждому виртуальному потоку? main() foo()
bar() baz() better threads Стек main foo bar baz

bar() baz() o Дорогое создание better threads Стек main foo bar baz

bar() baz() o Дорогое создание o Потребление памяти (страницы бывают и по 64к) better threads Стек main foo bar baz

bar() baz() o Дорогое создание o Потребление памяти (страницы бывают и по 64к) o Адресное пространство конечно better threads Стек main foo bar baz

o Дорогое создание o Потребление памяти (страницы бывают и по
64к) o Адресное пространство конечно o Ограничения OS (vm.max_map_count) Почему не сопоставить такой стек каждому виртуальному потоку? main() foo() bar() baz() better threads Стек main foo bar baz

final class VirtualThread extends Thread { ... // scheduler and
continuation private final Executor scheduler; private final Continuation cont; private final Runnable runContinuation; ... } better threads

continuation private final Executor scheduler; private final Continuation cont; private final Runnable runContinuation; ... } public class Continuation { ... private StackChunk tail; ... } better threads

continuation private final Executor scheduler; private final Continuation cont; private final Runnable runContinuation; ... } public class Continuation { ... private StackChunk tail; ... } public final class StackChunk { public static void init() {} private StackChunk parent; private int size; // in words private int sp; // in words private int argsize; // bottom stack-passed arguments, in words // The stack itself is appended here by the VM, as well as some injected fields } better threads

OS-потоки V V V V V V V V better threads

OS-потоки V V V V V V V V o Классические стеки у потоков носителей better threads

N M P P P P виртуальные потоки OS-потоки o
Классические стеки у потоков носителей o Собственные маленькие стеки в хипе у виртуальных потоков V V V V V V V V V V better threads

OS-потоки V V V V V V V V o Классические стеки у потоков носителей o Собственные маленькие стеки в хипе у виртуальных потоков o Копирование стека при возобновлении better threads

OS-потоки V V V V V V V V o Классические стеки у потоков носителей o Собственные маленькие стеки в хипе у виртуальных потоков o Копирование стека при возобновлении o Копирование обратно в хип при приостановке better threads

Теперь понятно, почему не работают нативы! Там ведь могли быть
указатели на стек! struct test a; struct test* pa = &a; better threads

Теперь понятно, почему не работают нативы! Там ведь могли быть
указатели на стек! struct test a; struct test* pa = &a; Копирование стека сделает такие указатели невалидными. better threads

Получается, что context-switch – работает за O(N) от размера стека?!
better threads

run(depth) run(depth - 1) ... run(0) for (; ;) {
Thread.yield() } better threads

Thread.yield() } Стек run run run … … run Копируется каждый раз при context-switch? better threads

import jdk.internal.vm.Continuation; import jdk.internal.vm.ContinuationScope; static class YieldAtLevel implements Runnable {
private void run(int depth) { if (depth > 0) { run(depth - 1); } else if (depth == 0) { for (;;) { Continuation.yield(SCOPE); } } } ... } better threads

import jdk.internal.vm.Continuation; import jdk.internal.vm.ContinuationScope; static class YieldAtLevel implements Runnable {
private void run(int depth) { if (depth > 0) { run(depth - 1); } else if (depth == 0) { for (;;) { Continuation.yield(SCOPE); } } } ... } @Setup(Level.Trial) public void setup() { cont = YieldAtLevel.cont(stackDepth); cont.run(); } @Benchmark public void yieldAtTheBottom() { cont.run(); } better threads

Benchmark (stackDepth) Score Error Units yieldAtTheBottom 5 115.436 ± 0.674
ns/op yieldAtTheBottom 10 114.703 ± 2.482 ns/op yieldAtTheBottom 20 116.356 ± 1.025 ns/op yieldAtTheBottom 100 115.887 ± 1.295 ns/op yieldAtTheBottom 200 115.369 ± 0.520 ns/op yieldAtTheBottom 500 113.776 ± 0.384 ns/op run(depth) run(depth - 1) ... run(0) for (; ;) { Thread.yield() } better threads

Да, но это оптимизируется ленивым копированием: o Копируется N фреймов
o Затем return барьер P V Получается, что context-switch – работает за O(N) от размера стека?! foo bar baz better threads

o Затем return барьер P V Получается, что context-switch – работает за O(N) от размера стека?! foo bar better threads

o Затем return барьер Но есть и цена! P V Получается, что context-switch – работает за O(N) от размера стека?! foo better threads

Thread.yield() } better threads

run(depth) run(depth - 1) ... run(0) Thread.yield() return; better threads

run(depth) run(depth - 1) ... run(0) Thread.yield() return; run(depth) run(depth
- 1) ... run(0) return; baseline better threads

0 5000 10000 15000 20000 25000 30000 5 100 200
300 400 500 No yielding Yield at the bottom and return ns/op stack depth

Выводы по стекам: o Context-switch не бесплатный, но оптимизации работают
better threads

Выводы по стекам: o Context-switch не бесплатный, но оптимизации работают
o Контрпример: обратный ход глубокой рекурсии better threads

mighty scheduler

OS-потоки V V V V V V V V V V V V Как понять, кто следующий исполняется? mighty scheduler

V V V V V V V V V V V V Как понять, кто следующий исполняется? o Очередь? mighty scheduler

V V V V V V V V V V V Как понять, кто следующий исполняется? o Очередь? V mighty scheduler

V V V V V V V V V V V Как понять, кто следующий исполняется? o Очередь? o Много очередей! V mighty scheduler

V V V V V V V V V V V Как понять, кто следующий исполняется? o Очередь? o Много очередей! o Work stealing V mighty scheduler

V V V V V V V V V V V Как понять, кто следующий исполняется? o Очередь? o Много очередей! o Work stealing Такая функциональность есть в ForkJoinPool! V mighty scheduler

В качестве планировщика используется ForkJoinPool o Написан Дагом Ли! o
Мощно оптимизирован youtube.com/watch?v=t0dGLFtRR9c mighty scheduler

В качестве планировщика используется ForkJoinPool o Написан Дагом Ли! o
Мощно оптимизирован Но так ли он подходит именно для виртуальных потоков? mighty scheduler

Раньше в Котлине тоже использовали FJP, как дефолтный планировщик. mighty
scheduler

Раньше в Котлине тоже использовали FJP, как дефолтный планировщик. А
теперь – нет. Почему? mighty scheduler

PingPongActorBenchmark.kt mighty scheduler

Benchmark (dispatcher) Score Error Units PingPongActorBenchmark.coresCountPingPongs scheduler 58,211 ± 2,074
ms/op PingPongActorBenchmark.coresCountPingPongs fjp 58,506 ± 10,942 ms/op PingPongActorBenchmark.coresCountPingPongs ftp_1 548,916 ± 37,671 ms/op ------------------------------------------------------------------------------- PingPongActorBenchmark.singlePingPong scheduler 23,855 ± 1,436 ms/op PingPongActorBenchmark.singlePingPong fjp 81,871 ± 1,816 ms/op PingPongActorBenchmark.singlePingPong ftp_1 29,627 ± 0,290 ms/op PingPongActorBenchmark.kt mighty scheduler

PingPongActorBenchmark.kt Benchmark (dispatcher) Score Error Units PingPongActorBenchmark.coresCountPingPongs scheduler 58,211 ±
2,074 ms/op PingPongActorBenchmark.coresCountPingPongs fjp 58,506 ± 10,942 ms/op PingPongActorBenchmark.coresCountPingPongs ftp_1 548,916 ± 37,671 ms/op ------------------------------------------------------------------------------- PingPongActorBenchmark.singlePingPong scheduler 23,855 ± 1,436 ms/op PingPongActorBenchmark.singlePingPong fjp 81,871 ± 1,816 ms/op PingPongActorBenchmark.singlePingPong ftp_1 29,627 ± 0,290 ms/op mighty scheduler

Эмпирические наблюдения: o Зачастую приостановленная корутина скоро снова будет готова
работать mighty scheduler

Эмпирические наблюдения: o Зачастую приостановленная корутина скоро снова будет готова
работать o Хорошо бы, чтобы она осталась на том же треде и на том же CPU mighty scheduler

ForkJoinPool: o Написан Дагом Ли! o Мощно оптимизирован mighty scheduler

ForkJoinPool: o Написан Дагом Ли! o Мощно оптимизирован o Нужна
более тонкая настройка очередности o Слишком агрессивно ворует mighty scheduler

ForkJoinPool: o Написан Дагом Ли! o Мощно оптимизирован o Нужна
более тонкая настройка очередности o Слишком агрессивно ворует У большинства JVM-based решений свой планировщик, который эти проблемы старается решать. mighty scheduler

Нужна возможность писать и использовать с виртуальными потоками свои планировщики!
mighty scheduler

... Thus, full support for NUMA requires the ability to
create several different pools of carrier threads with the ability to configure them. This is not possible in the current version of the API. ... Vladimir Ogorodnikov [email protected] Нужна возможность писать и использовать с виртуальными потоками свои планировщики! mighty scheduler

... Thus, full support for NUMA requires the ability to
create several different pools of carrier threads with the ability to configure them. This is not possible in the current version of the API. ... Vladimir Ogorodnikov [email protected] Yes, the plan is to ultimately allow custom schedulers, but we want to first let the ecosystem learn about virtual threads and their uses with the default scheduler. ... Ron Pressler Нужна возможность писать и использовать с виртуальными потоками свои планировщики! mighty scheduler

Нужна возможность писать и использовать с виртуальными потоками свои планировщики!
... Thus, full support for NUMA requires the ability to create several different pools of carrier threads with the ability to configure them. This is not possible in the current version of the API. ... Vladimir Ogorodnikov [email protected] Yes, the plan is to ultimately allow custom schedulers, but we want to first let the ecosystem learn about virtual threads and their uses with the default scheduler. ... Ron Pressler mighty scheduler

is loom a hero?

Из хорошего: + Низкоуровневые и оптимизированные примитивы: Continuation/Stack chunk +
Переработана стандартная библиотека под виртуальные потоки is loom a hero?

Из хорошего: + Низкоуровневые и оптимизированные примитивы: Continuation/Stack chunk +
Переработана стандартная библиотека под виртуальные потоки + Влияние на код минимальное! Никакого coloring is loom a hero?

Что нужно доделать: ? Synchronized ? Принудительное переключение потоков ?
Humongous stack chunks (SOE при стеке >256КБ) ? Возможность делать свои планировщики is loom a hero?

Из плохого: − Реализация сложна (кишки JVM + переделки в
стандартной библиотеке) − Сквозные нативы пи́нают виртуальный тред − Обратный ход рекурсии может быть дорог − Context-switch не бесплатен is loom a hero?

Вывод: когда (если) допилят, будет хорошо! is loom a hero?

Вывод: когда (если) допилят, будет хорошо! Пробуйте и пишите фидбек
is loom a hero?

код и язык Автоматическое преобразование кода в CPS a.k.a stackless корутины С#/JS/C++ Kotlin

код и язык Автоматическое преобразование кода в CPS a.k.a stackless корутины С#/JS/C++ Kotlin Виртуальные потоки в Loom a.k.a stackful корутины

код и язык Автоматическое преобразование кода в CPS a.k.a stackless корутины С#/JS/C++ Kotlin Виртуальные потоки в Loom a.k.a stackful корутины Loom идеальный

код и язык Автоматическое преобразование кода в CPS a.k.a stackless корутины С#/JS/C++ Kotlin Виртуальные потоки в Loom a.k.a stackful корутины Loom идеальный Loom сейчас

brave new world func worker(id int) { fmt.Printf("Worker %d starting\n",
id) time.Sleep(time.Second) fmt.Printf("Worker %d done\n", id) } func main() { for i := 1; i <= 10; i++ { go worker(i) } }

func main() { var wg sync.WaitGroup for i := 1;
i <= 100000; i++ { wg.Add(1) i := i go func() { defer wg.Done() worker(i) }() } wg.Wait() } brave new world

… Worker 99833 done Worker 76674 done Worker 99825 done
Worker 99821 done Worker 99799 done Worker 67867 done Worker 99815 done Worker 99804 done Worker 99816 done Worker 99863 done Worker 87131 done Worker 84239 done Worker 99808 done func main() { var wg sync.WaitGroup for i := 1; i <= 100000; i++ { wg.Add(1) i := i go func() { defer wg.Done() worker(i) }() } wg.Wait() } brave new world

Горутины в Go: o Любую функцию можно запустить параллельно brave
new world

Горутины в Go: o Любую функцию можно запустить параллельно o
Специальных модификаторов/раскраски – нет o Ограничений на код – нет brave new world

Горутины в Go: o Любую функцию можно запустить параллельно o
Специальных модификаторов/раскраски – нет o Ограничений на код – нет o Миллион горутин создается и работает brave new world

youtube.com/watch?v=-K11rY57K7k brave new world

Реализация горутин в Go: o N:M:P схема brave new world

Реализация горутин в Go: o N:M:P схема o Бесконечные стеки
o Context-switch без копирования стеков brave new world

o Context-switch без копирования стеков + Бесплатный! За O(1) + Нативы работают − Платим скоростью каждой функции brave new world

o Context-switch без копирования стеков o Preemption! o Честность планирования горутин brave new world

код и язык Автоматическое преобразование кода в CPS a.k.a stackless корутины С#/JS/C++ Kotlin Виртуальные потоки в Loom a.k.a stackful корутины Loom идеальный Loom сейчас

код и язык Автоматическое преобразование кода в CPS a.k.a stackless корутины С#/JS/C++ Kotlin Виртуальные потоки в Loom или горутины в Go a.k.a stackful корутины Go прямо сейчас Loom сейчас

o Решений проблемы легковесных потоков существует множество o Loom –
лишь один из путей, дающий свободу выбора

лишь один из путей, дающий свободу выбора o Реализация иногда сильно влияет на язык

лишь один из путей, дающий свободу выбора o Реализация иногда сильно влияет на язык. Но это не значит, что с этим нужно мириться

o Андрею Паньгину o Севе Толстопятову o Ринчину Запанову o
Кате Углянской

@dbg_nsk @ugliansky t.me/jugnsk Бенчмарки и примеры здесь: github.com/ugliansky/jpoint2022 Пробуйте Loom
в Java 19 или сейчас

Benchmark Score Error Units baseline 0,337 ± 0,027 us/op suspendUsual
3,843 ± 0,278 us/op noInline 2,677 ± 0,075 us/op @CompilerControl(CompilerControl.Mode.DONT_INLINE) suspend fun test(): Int { val r = foo() validate(r) val l = bar(r + 42) validate(l) return baz(r + l) } suspend fun validate() {…} @Benchmark fun suspendNoInline() = runBlocking { test() } suspend fun foo() { validate(...) ... } suspend fun bar(i: Int) { foo() ... } suspend fun baz(i: Int) { validate(i) ... } to hide a thread pool

Benchmark Score Error Units baseline 0,337 ± 0,027 us/op suspendUsual
3,843 ± 0,278 us/op noInline 2,677 ± 0,075 us/op suspendNoInline 10,460 ± 0,110 us/op @CompilerControl(CompilerControl.Mode.DONT_INLINE) suspend fun test(): Int { val r = foo() validate(r) val l = bar(r + 42) validate(l) return baz(r + l) } suspend fun validate() {…} @Benchmark fun suspendNoInline() = runBlocking { test() } suspend fun foo() { validate(...) ... } suspend fun bar(i: Int) { foo() ... } suspend fun baz(i: Int) { validate(i) ... } to hide a thread pool

to hide a thread pool Benchmark Score Error Units baseline
0,337 ± 0,027 us/op suspendUsual 3,843 ± 0,278 us/op noInline 2,677 ± 0,075 us/op suspendNoInline 10,460 ± 0,110 us/op testRunBlocking 0,191 ± 0,011 us/op fun test(): Int { val r = foo() validate(r) val l = bar(r + 42) validate(l) return baz(r + l) } fun validate() {…} @Benchmark fun testRunBlocking() = runBlocking { // test() } fun foo() { validate(...) ... } fun bar(i: Int) { foo() ... } fun baz(i: Int) { validate(i) ... }

better threads Benchmark (stackDepth) Score Error Units noYielding 5 1007.704
± 41.553 ns/op noYielding 50 1401.873 ± 36.681 ns/op noYielding 100 2034.516 ± 56.821 ns/op noYielding 200 3429.187 ± 73.425 ns/op noYielding 300 4754.425 ± 139.880 ns/op noYielding 400 6043.952 ± 231.421 ns/op ---------------------------------------------------- yieldAndReturn 5 1051.371 ± 26.917 ns/op yieldAndReturn 50 1880.173 ± 32.059 ns/op yieldAndReturn 100 4423.114 ± 70.889 ns/op yieldAndReturn 200 9993.032 ± 73.666 ns/op yieldAndReturn 300 15378.223 ± 346.750 ns/op yieldAndReturn 400 20933.469 ± 120.113 ns/op Понятно, что baseline быстрее, но какой из этого можно сделать вывод?

И еще про стеки: better threads

И еще про стеки: o Стеки в хипе => размечаются
сборщиком мусора better threads

сборщиком мусора o VirtualThread → Continuation → StackChunk → локалы better threads

сборщиком мусора o VirtualThread → Continuation → StackChunk → локалы Из-за недостаточной поддержки stack chunks в G1 могут случаться SOE на стеке размером 256KB better threads

0 5000 10000 15000 20000 25000 30000 5 100 200
300 400 500 Invoke => yield Resume => return ns/op stack depth

Thread Wars: Project Loom Strikes Back

Thread Wars: Project Loom Strikes Back

More Decks by Ivan Ugliansky

Other Decks in Programming

Featured

Transcript