Javaの仮想スレッドはネットワークI/Oとどう連携するのか：Carrier、Continuation、Poller

·(編集済み)· / , , , , · 閲覧

この記事はに最終更新されました。一部の内容が古くなっている可能性があります。ご質問があれば著者にお問い合わせください。

AI翻訳簡体中文日本語

仮想スレッドが次のコードを実行するとき、表面上は通常のスレッドと区別がつかない。

JAVA

int count = socket.getInputStream().read(buffer);
process(buffer, count);

Socketに一時的にデータがない場合、現在の仮想スレッドは中断される。その仮想スレッドを運んでいたCarrier Threadは、ほかの仮想スレッドを実行できる。ネットワークデータが届くと、元の仮想スレッドが再びスケジュールされ、read()が戻ってprocess()の実行が続く。

ここには몇 가지 질문이 생긴다。

Carrier Threadはコードを順番に実行するだけだ。現在のメソッドがまだ終わっていないのに、どうやって別のタスクを実行するのか？
仮想スレッドを中断した後はどこに保存されるのか？
ネットワークイベントが到着すると、Pollerはどうやって対応する仮想スレッドを見つけるのか？
スケジュールキューには、仮想スレッド、コールバック関数、完全な呼び出しスタックのどれが保存されるのか？
ネットワークデータは、元のbufferにどう戻ってくるのか？
Continuation.yield()、Thread.yield()、LockSupport.park()はそれぞれ何を担当するのか？

本文ではJDK 21の実装を基準に、Socketの読み取りが1回行われるまでの呼び出し過程を追いながら、これらの問題を解析する。JDK 21の仮想スレッドはM:Nスケジューリングを使う。JDKは大量の仮想スレッドを少数のプラットフォームスレッドへスケジュールして実行する。デフォルトのスケジューラは独立したForkJoinPoolだ。（OpenJDK）

Carrier Threadにタスク切り替え命令はない

まず仮想スレッドを取り除き、通常のワーカースレッドだけを見てみよう。

CodeBlock Loading...

このスレッドは常に順番に実行する。キューからRunnableを取り出してrun()を呼び、そのメソッドが戻るまで待ってから次のループへ進む。

JAVA

Runnable taskA = queue.take();
taskA.run();
Runnable taskB = queue.take();
taskB.run();

Carrier Threadも同じように動く。taskA.run()がまだ戻っていない最中に、taskB.run()へ唐突に移動することはない。

仮想スレッドが切り替えられるのは、taskA.run()がアプリ側の呼び出しを終える前に一時的に戻れるからだ。戻る前に、JVMは仮想スレッドの呼び出しスタックを保存する。後からこのタスクを実行するときは、JVMが元の呼び出しスタックを復元して、続きのコードを実行する。

したがって、Carrier Thread上で実際に起きるのは次の処理になる。

JAVA

virtualThreadAResumeTask.run(); // 保存调用栈后暂时返回
virtualThreadBResumeTask.run(); // Carrier 回到调度循环后执行

Carrier Threadは引き続き通常のメソッド呼び出し規則に従う。特別な仕組みはContinuationが持つ。未完了の呼び出しスタックを中断して再開できる。

スケジュールキューに保存されるもの

JDK 21のVirtualThread内部にはいくつかの重要なフィールドがある。診断情報、スレッドコンテナー、中断処理などを除くと、次のように整理できる。

CodeBlock Loading...

各フィールドの意味は次の通りだ。

schedulerは仮想スレッドのスケジューラを指す。
contは中断・再開できる実行状態を保持する。
runContinuationはスケジューラへ投入するタスクである。
stateは、仮想スレッドが実行中、中断中、または実行可能な状態にあることを記録する。
parkPermitはunparkの許可を1つ保持する。
carrierThreadは、現在その仮想スレッドを運んでいるプラットフォームスレッドを指す。

仮想スレッドを構築するとContinuationが生成され、現在の仮想スレッドに束縛されたメソッド参照が作られる。

JAVA

this.cont = new VThreadContinuation(this, task);
this.runContinuation = this::runContinuation;

そのメソッド参照は、次のような通常のRunnableへ展開できる。

CodeBlock Loading...

スケジュールキューの中核に保存されるのは、このような再開タスクだ。タスクはVirtualThreadへの参照を保持し、VirtualThreadはさらにContinuationを保持する。

CodeBlock Loading...

VirtualThreadはForkJoinPool内の特定のワークキューを直接操作しない。次の呼び出しを行うだけだ。

JAVA

scheduler.execute(runContinuation);

デフォルトのスケジューラは通常のRunnableを受け取るとForkJoinTaskへラップし、内部キューに入れる。OpenJDKのコメントによれば、スケジューラのWorkerから投入された場合はタスクがローカルキューに入り、ほかのスレッドから投入された場合は外部提出キューに入る。（GitHub）

仮想スレッドが最初に実行されるとき

仮想スレッドを起動すると、内部で状態がNEWからSTARTEDへ変わり、runContinuationが投入される。

CodeBlock Loading...

Carrier ThreadのいずれかがForkJoinPoolのキューからタスクを取り出し、最終的に次を呼び出す。

JAVA

virtualThread.runContinuation();

runContinuation()の主な構造は次の通りだ。

CodeBlock Loading...

mount()は現在のプラットフォームスレッドをCarrierとして記録し、JVMが保持する現在のスレッドの同一性を変更する。

JAVA

private void mount() {
    Thread carrier = Thread.currentCarrierThread();
    setCarrierThread(carrier);
    carrier.setCurrentThread(this);
}

そのため、仮想スレッド内の業務コードで

JAVA

Thread.currentThread()

を呼び出すと、得られるのはVirtualThreadオブジェクトであり、基盤のCarrier Threadではない。

続いて実行するのは

JAVA

cont.run();

だ。最初の呼び出しでは、Continuationはユーザータスクの入口から実行を始める。OpenJDKのrunContinuation()ではmount()の後にcont.run()を呼び出し、finallyでunmount()と後続の状態処理を行っている。（GitHub）

Continuation.yield()がCarrierを解放する方法

業務コードの呼び出し関係が次のだと仮定する。

CodeBlock Loading...

仮想スレッドがネットワーク待ちの位置まで実行すると、Carrier Thread上の論理呼び出しスタックは次のようになる。

CodeBlock Loading...

Continuation.yield()を実行すると、JVMは現在のContinuationに属するスタックフレームを凍結する。ローカル変数、オブジェクト参照、呼び出し関係、プログラムの実行位置がすべて保持される。

たとえば、handleRequest()に対応するスタックフレームには、次のような値が存在するかもしれない。

TEXT

socket  -> Socket@100
user    -> User@200
buffer  -> byte[]@300
count   -> 尚未赋值

このデータは仮想スレッドの実行状態に属する。現在のCarrier Threadのプラットフォームスタックには依存しない。

凍結に成功すると、制御フローはContinuationから抜け、cont.run()の呼び出し元へ戻る。この時点で、仮想スレッドの深い呼び出しスタックはCarrier Threadからすでに外れている。Carrier Threadにはスケジューラ関連のスタックフレームだけが残る。

TEXT

ForkJoinPool.runWorker()
VirtualThread.runContinuation()

cont.run()が戻ると、runContinuation()はfinallyブロックに入り、unmount()を実行する。

JAVA

private void unmount() {
    Thread carrier = this.carrierThread;
    carrier.setCurrentThread(carrier);
    setCarrierThread(null);
}

runContinuation()が戻った後、Carrier ThreadはForkJoinPoolのワークループへ戻り、キューから別のタスクを取る。

後続のあるCarrier Threadが再びcont.run()を呼び出すと、JVMは以前凍結したスタックフレームを復元する。以前中断したContinuation.yield()から戻り始め、呼び出しチェーンが順に再開する。

TEXT

Continuation.yield() 返回
VirtualThread.park() 返回
LockSupport.park() 返回
Poller.poll() 返回
NioSocketImpl.implRead() 继续

handleRequest()を再呼び出したり、メソッドの先頭から実行し直したりすることはない。再開地点は、以前中断した位置にある。

park()がすぐには実行キューへ戻らない理由

Continuation.yield()は実行の中断だけを担当し、仮想スレッドをいつ再開するかは決めない。再開の扱いは、Continuation.yield()を呼び出した側によって決まる。

仮想スレッドがThread.yield()を呼び出した場合、現在のスレッドは実行可能なままCarrierを一時的に譲るだけだ。実行スタックを凍結した後、afterYield()が直ちにrunContinuationを再投入する。

TEXT

RUNNING
  -> YIELDING
  -> YIELDED
  -> 重新提交
  -> RUNNING

LockSupport.park()は、待機条件がまだ満たされていないことを表す。たとえば、ネットワークデータの到着、ロックの解放、キュー要素の到着を待つ場合だ。

実行スタックが凍結された直後に仮想スレッドを実行キューへ戻すと、毎回再開して条件を確認して再び中断するため、CPUを空回りさせてしまう。

TEXT

RUNNING
  -> PARKING
  -> PARKED

他のスレッドがunpark()を呼んで初めて、仮想スレッドは実行可能状態に戻る。

TEXT

PARKED
  -> UNPARKED
  -> 重新提交
  -> RUNNING

OpenJDKのafterYield()は現在状態を確認する。PARKINGはPARKEDへ遷移し、デフォルトでは再投入しない。YIELDINGはYIELDEDへ遷移して、直ちに投入する。（GitHub）

対応するコードは次のように簡略化できる。

CodeBlock Loading...

Socketの読み取りがPollerへ入る仕組み

ここからネットワークI/O経路へ入る。

JAVA

int count = socket.getInputStream().read(buffer);

JDK 21のLinux上の実装を例にすると、JDKはまずSocketの読み取りを試す。カーネルの受信バッファにすでにデータがあれば、読み取りはそのまま完了し、仮想スレッドを中断する必要はない。

現在データがない場合、非ブロッキング読み取りはEAGAIN、または同等の状態を返す。JDKは続くファイル記述子をPollerへ登録し、現在の仮想スレッドを中断する。JEP 444ではこの処理が次のように説明されている。JDK内のブロッキングネットワーク操作がすぐには完了しない場合、仮想スレッドはアンマウントされる。I/Oが完了すると、仮想スレッドはスケジューラへ再提出される。（OpenJDK）

簡略化した読み取りロジックは次の通りだ。

CodeBlock Loading...

Poller.poll(fd)の内部は、次のように抽象化できる。

JAVA

void poll(int fd) {
    Thread thread = Thread.currentThread();
    waiters.put(fd, thread);
    registerWithEpoll(fd);
    LockSupport.park();
}

ここでは仮想スレッドを実行しているため、Thread.currentThread()は対応するVirtualThreadを返す。

現在、2つの対応付けが存在する。

Linuxカーネルのepoll記録は次の通りだ。

TEXT

监听 fd 37 的可读事件

JDKのPoller記録は次の通りだ。

TEXT

fd 37 -> VirtualThread@500

OSはファイル記述子だけを知っており、Javaの仮想スレッドは知らない。Poller内のマッピングが、カーネルイベントをJavaのスレッドオブジェクトへ結び直す。JDK 21のPollerは、ネットワークイベント通知と仮想スレッドの覚醒との間をつなぐ層だ。（GitHub）

登録後、現在の仮想スレッドは次を実行する。

JAVA

LockSupport.park();

park()はVirtualThread.park()へ入り、PARKING状態を設定してContinuation.yield()を呼ぶ。実行スタックが凍結されると、仮想スレッドはPARKEDになり、Carrier Threadが解放される。

待機中も、仮想スレッドはCarrier Threadを占有せず、実行キューにも入っていない。Pollerが待機関係を保持し、Continuationが実行状態を保持する。

CodeBlock Loading...

ネットワークが準備できた後の再スケジュール

ネットワークデータが届くと、Linuxのプロトコルスタックはバイト列をSocketのカーネル受信バッファへ入れ、fdを可読としてマークする。

Linux上のPollerスレッドは通常epoll_wait()でブロックしている。イベントが返ると、準備できたファイル記述子が得られる。

JAVA

int[] readyFds = epollWait();
for (int fd : readyFds) {
    Thread thread = waiters.remove(fd);
    if (thread != null) {
        LockSupport.unpark(thread);
    }
}

PollerはユーザーのhandleRequest()を実行せず、Poller Threadで元のInputStream.read()を呼ぶこともない。fdを待っている仮想スレッドを見つけ、unpark()を呼ぶだけだ。

VirtualThread.unpark()の主要なロジックは次の通りだ。

CodeBlock Loading...

submitRunContinuation()は最終的に次を実行する。

JAVA

scheduler.execute(runContinuation);

ここで仮想スレッドをスケジューラへ戻す。OpenJDKのunpark()はまずparkPermitを設定し、次に状態CASによって中断中の仮想スレッドをUNPARKEDへ移し、その後に再開タスクを投入する。（GitHub）

unpark()を呼ぶのは通常Poller Threadだ。Poller ThreadはデフォルトスケジューラのWorkerではないため、このタスクは通常ForkJoinPoolの外部提出経路を通る。

後続のCarrier Threadがタスクを取り出し、再び次を呼び出す。

JAVA

virtualThread.runContinuation();

状態はUNPARKEDからRUNNINGへ変わり、仮想スレッドは新しいCarrier Threadへマウントされる。そのうえで次を実行する。

JAVA

cont.run();

以前凍結された呼び出しスタックが復元され、LockSupport.park()が戻り、Socketの読み取り処理が続く。

スケジュールキューにネットワークデータは入らない

スケジュールキューにネットワークデータが入ることはない。次のような物体も存在しない。

JAVA

new ResumeTask(virtualThread, networkData);

Pollerの責務は「fdが今読み取り可能かもしれない」と通知することであり、スケジューラは仮想スレッドの続行を安排する。ネットワークバイトは、そのままSocketのカーネル受信バッファに残る。

仮想スレッドの再開後、読み取りループはもう一度OSを呼び出す。

CodeBlock Loading...

最初のnonBlockingRead()がEAGAINを返すと、仮想スレッドは待ち状態に入る。

ネットワークが準備できた後、Pollerはunpark()を呼び、仮想スレッドが再開する。続いてもう一度次を実行する。

JAVA

nonBlockingRead(fd, buffer);

今度はカーネル受信バッファにデータがある。システムコールはバイト列を元のbufferへコピーし、バイト数を返す。

元のbufferへ引き続きアクセスできるのは、Continuationが保存した呼び出しスタックがまだbufferを保持しているからだ。

TEXT

buffer -> byte[]@300

データの流れは次の通りだ。

CodeBlock Loading...

スケジュールキューが渡すのは実行資格だけだ。Pollerが渡すのは準備完了通知だけだ。ネットワークデータはカーネルのSocketバッファが保持する。

parkPermitが失われた通知を防ぐ仕組み

仮想スレッドが中断を準備している間に、ネットワークイベントが先に到達することがある。

次の時系列を考える。

TEXT

虚拟线程发现当前没有数据
Poller 注册 fd
网络数据立即到达
Poller 调用 unpark()
虚拟线程随后才执行 park()

unpark()がすでにPARKED状態のスレッドだけを起こせるなら、この通知は失われる。そのあと仮想スレッドがpark()に入っても、永久に再開されない可能性がある。

parkPermitは容量1の許可のように機能する。

JAVA

private volatile boolean parkPermit;

unpark()はまず次を設定する。

JAVA

parkPermit = true;

park()は開始時にまず許可を消費しようとする。

JAVA

if (getAndSetParkPermit(false)) {
    return;
}

ネットワークイベントがすでに到着していれば、park()はそのまま戻り、呼び出しスタックを凍結しない。

unpark()を連続して複数回呼んでも、保存される許可は1つだけだ。

JAVA

unpark(thread);
unpark(thread);
unpark(thread);

最終的な状態は次の通りになる。

TEXT

parkPermit = true

次のpark()が許可を消費すると、falseに戻る。

JUCの同期機構が通常条件のチェックをループで行うのも、この理由だ。

JAVA

while (!conditionSatisfied()) {
    LockSupport.park();
}

park()が戻ることは、スレッドが実行機会を得たことだけを意味している。業務条件は依然として再確認が必要だ。

PARKINGによる重複スケジュールの防止

呼び出しスタックの凍結には一段の実行過程を要するため、RUNNINGとPARKEDの2つの状態だけでは足りない。

仮想スレッドはまず次を実行する。

JAVA

state = PARKING;
Continuation.yield();

PARKINGを設定してから、Continuationの凍結が完了するまでの間に、Pollerがすでにunpark()を呼ぶ可能性がある。

この時点では元のCarrier Threadが現在の仮想スレッドを実行している，也可能だ。直ちに再開タスクをスケジューラへ投入すると、別のCarrier Threadが同じ仮想スレッドを同時に取り出してしまう。

そのため、PARKING状態ではunpark()はparkPermitを設定するだけで、タスクを投入しない。

Continuationの凍結後、元のCarrierはafterYield()を実行する。

JAVA

state = PARKED;
if (parkPermit
        && compareAndSetState(PARKED, UNPARKED)) {
    submitRunContinuation();
}

これにより，提前到着した覚醒通知を残しつつ、同じ仮想スレッドを2本のCarrier Threadが同時に実行することも防げる。

また、runContinuation()は再開前にも状態へCASを行う。万一スケジュールキューに同じ仮想スレッドを指す2つの再開タスクが入っても、RUNNINGへ状態を変更 successfully にできるのは1本のCarrier Threadだけだ。

3つのスレッドの連携

完全な実装にはForkJoinPool、Continuation、Poller、オペレーティングシステムが関わるが、実行過程は3つの通常のスレッドループに分解できる。

Carrier Threadのループは次の通りだ。

JAVA

while (true) {
    Runnable task = scheduler.takeTask();
    task.run();
}

Poller Threadのループは次の通りだ。

CodeBlock Loading...

仮想スレッド内で行 되는Socketの読み取りループは次の通りだ。

CodeBlock Loading...

3つの実行フローはいずれも順番に処理を進める。次のオブジェクトを通じて彼此がつながっている。

CodeBlock Loading...

Carrier Threadは再開タスクを実行し、Pollerはfdから待機中のスレッドを探し、仮想スレッドはscheduler経由で自分自身を再提出する。Continuationは中断位置とローカルオブジェクト参照を保持する。

完全な呼び出し経路

全体を接続すると、次の主要経路が得られる。

CodeBlock Loading...

この経路を理解すると、仮想スレッドのスケジューリングはもう謎ではない。Carrier Threadは常に通常のタスクループを実行する。Continuationは、呼び出しスタックを保持したまま現在のタスクを一時的に戻せるようにする。Pollerはネットワークfdと待機中の仮想スレッドを結び付ける。unpark()はイベント到着後に再開タスクを再提出する。

JDK 21とJDK 24のPinningの違い

JDK 21では、仮想スレッドがsynchronized Monitorを保持したまま一部のブロッキング操作へ入ると、Continuationをアンマウントできないことがある。仮想スレッドとCarrier Threadが一緒にブロックし、この状況をPinningと呼ぶ。

そのため、JDK 21の段階では、长时间のsynchronizedクリティカルセクション内でネットワークI/Oを行わないようにすることがよく推奨される。必要ならReentrantLockを使う。

JDK 24ではJEP 491が提供され、JVM Monitorと仮想スレッドの連携方法が変更された。仮想スレッドがsynchronizedメソッドまたはコードブロック内でブロックしても、大半の場合はアンマウントしてCarrier Threadを解放できるようになった。JDK 24以降は、synchronizedとjava.util.concurrent.locksのどちらを選ぶかを、Monitor Pinningを避けるためだけでなく、中断可能な取得、公平性、タイムアウト、条件変数などの意味に基づいて判断する重要性が増した。（OpenJDK）

ロックを保持したまま長時間ネットワークI/Oを行う場合は、依然として谨慎が必要だ。Carrierを解放できても、ロック自体は保持されたままだ。同じクリティカルセクションに入りたい他のスレッドは待つことになる。

参照ソース

JEP 444：Virtual Threads。（OpenJDK）
JDK 21 java.lang.VirtualThread。（GitHub）
JDK 21 jdk.internal.vm.Continuation。（GitHub）
JDK 21 sun.nio.ch.Poller。（GitHub）
JEP 491：Synchronize Virtual Threads without Pinning。（OpenJDK）

仮想スレッドが次のコードを実行するとき、表面上は通常のスレッドと区別がつかない。

JAVA

int count = socket.getInputStream().read(buffer);
process(buffer, count);

ここには몇 가지 질문이 생긴다。

Carrier Threadはコードを順番に実行するだけだ。現在のメソッドがまだ終わっていないのに、どうやって別のタスクを実行するのか？
仮想スレッドを中断した後はどこに保存されるのか？
ネットワークイベントが到着すると、Pollerはどうやって対応する仮想スレッドを見つけるのか？
スケジュールキューには、仮想スレッド、コールバック関数、完全な呼び出しスタックのどれが保存されるのか？
ネットワークデータは、元のbufferにどう戻ってくるのか？
Continuation.yield()、Thread.yield()、LockSupport.park()はそれぞれ何を担当するのか？

Carrier Threadにタスク切り替え命令はない

まず仮想スレッドを取り除き、通常のワーカースレッドだけを見てみよう。

JAVA

final class Worker extends Thread {
    private final BlockingQueue<Runnable> queue;
    Worker(BlockingQueue<Runnable> queue) {
        this.queue = queue;
    }
    @Override
    public void run() {
        while (!isInterrupted()) {
            try {
                Runnable task = queue.take();
                task.run();
            } catch (InterruptedException e) {
                interrupt();
            }
        }
    }
}

CodeBlock Loading...

このスレッドは常に順番に実行する。キューからRunnableを取り出してrun()を呼び、そのメソッドが戻るまで待ってから次のループへ進む。

JAVA

Runnable taskA = queue.take();
taskA.run();
Runnable taskB = queue.take();
taskB.run();

Carrier Threadも同じように動く。taskA.run()がまだ戻っていない最中に、taskB.run()へ唐突に移動することはない。

したがって、Carrier Thread上で実際に起きるのは次の処理になる。

JAVA

virtualThreadAResumeTask.run(); // 保存调用栈后暂时返回
virtualThreadBResumeTask.run(); // Carrier 回到调度循环后执行

Carrier Threadは引き続き通常のメソッド呼び出し規則に従う。特別な仕組みはContinuationが持つ。未完了の呼び出しスタックを中断して再開できる。

スケジュールキューに保存されるもの

JDK 21のVirtualThread内部にはいくつかの重要なフィールドがある。診断情報、スレッドコンテナー、中断処理などを除くと、次のように整理できる。

JAVA

final class VirtualThread extends Thread {
    private final Executor scheduler;
    private final Continuation cont;
    private final Runnable runContinuation;
    private volatile int state;
    private volatile boolean parkPermit;
    private volatile Thread carrierThread;
}

CodeBlock Loading...

各フィールドの意味は次の通りだ。

schedulerは仮想スレッドのスケジューラを指す。
contは中断・再開できる実行状態を保持する。
runContinuationはスケジューラへ投入するタスクである。
stateは、仮想スレッドが実行中、中断中、または実行可能な状態にあることを記録する。
parkPermitはunparkの許可を1つ保持する。
carrierThreadは、現在その仮想スレッドを運んでいるプラットフォームスレッドを指す。

仮想スレッドを構築するとContinuationが生成され、現在の仮想スレッドに束縛されたメソッド参照が作られる。

JAVA

this.cont = new VThreadContinuation(this, task);
this.runContinuation = this::runContinuation;

そのメソッド参照は、次のような通常のRunnableへ展開できる。

JAVA

final class ResumeTask implements Runnable {
    private final VirtualThread virtualThread;
    ResumeTask(VirtualThread virtualThread) {
        this.virtualThread = virtualThread;
    }
    @Override
    public void run() {
        virtualThread.runContinuation();
    }
}

CodeBlock Loading...

TEXT

ForkJoinPool.WorkQueue
    └── ForkJoinTask
          └── runContinuation
                └── VirtualThread
                      └── Continuation

CodeBlock Loading...

VirtualThreadはForkJoinPool内の特定のワークキューを直接操作しない。次の呼び出しを行うだけだ。

JAVA

scheduler.execute(runContinuation);

仮想スレッドが最初に実行されるとき

仮想スレッドを起動すると、内部で状態がNEWからSTARTEDへ変わり、runContinuationが投入される。

JAVA

void start() {
    if (!compareAndSetState(NEW, STARTED)) {
        throw new IllegalThreadStateException();
    }
    submitRunContinuation();
}
private void submitRunContinuation() {
    scheduler.execute(runContinuation);
}

CodeBlock Loading...

Carrier ThreadのいずれかがForkJoinPoolのキューからタスクを取り出し、最終的に次を呼び出す。

JAVA

virtualThread.runContinuation();

runContinuation()の主な構造は次の通りだ。

JAVA

private void runContinuation() {
    int initialState = state();
    if (initialState != STARTED
            && initialState != UNPARKED
            && initialState != YIELDED) {
        return;
    }
    if (!compareAndSetState(initialState, RUNNING)) {
        return;
    }
    mount();
    try {
        cont.run();
    } finally {
        unmount();
        if (cont.isDone()) {
            afterDone();
        } else {
            afterYield();
        }
    }
}

CodeBlock Loading...

mount()は現在のプラットフォームスレッドをCarrierとして記録し、JVMが保持する現在のスレッドの同一性を変更する。

JAVA

private void mount() {
    Thread carrier = Thread.currentCarrierThread();
    setCarrierThread(carrier);
    carrier.setCurrentThread(this);
}

そのため、仮想スレッド内の業務コードで

JAVA

Thread.currentThread()

を呼び出すと、得られるのはVirtualThreadオブジェクトであり、基盤のCarrier Threadではない。

続いて実行するのは

JAVA

cont.run();

Continuation.yield()がCarrierを解放する方法

業務コードの呼び出し関係が次のだと仮定する。

JAVA

void handleRequest(Socket socket) throws IOException {
    User user = loadUser();
    byte[] buffer = new byte[1024];
    int count = socket.getInputStream().read(buffer);
    process(user, buffer, count);
}

CodeBlock Loading...

仮想スレッドがネットワーク待ちの位置まで実行すると、Carrier Thread上の論理呼び出しスタックは次のようになる。

TEXT

ForkJoinPool.runWorker()
VirtualThread.runContinuation()
Continuation.run()
handleRequest()
InputStream.read()
NioSocketImpl.implRead()
Poller.poll()
LockSupport.park()
VirtualThread.park()
Continuation.yield()

CodeBlock Loading...

たとえば、handleRequest()に対応するスタックフレームには、次のような値が存在するかもしれない。

TEXT

socket  -> Socket@100
user    -> User@200
buffer  -> byte[]@300
count   -> 尚未赋值

このデータは仮想スレッドの実行状態に属する。現在のCarrier Threadのプラットフォームスタックには依存しない。

TEXT

ForkJoinPool.runWorker()
VirtualThread.runContinuation()

cont.run()が戻ると、runContinuation()はfinallyブロックに入り、unmount()を実行する。

JAVA

private void unmount() {
    Thread carrier = this.carrierThread;
    carrier.setCurrentThread(carrier);
    setCarrierThread(null);
}

runContinuation()が戻った後、Carrier ThreadはForkJoinPoolのワークループへ戻り、キューから別のタスクを取る。

TEXT

Continuation.yield() 返回
VirtualThread.park() 返回
LockSupport.park() 返回
Poller.poll() 返回
NioSocketImpl.implRead() 继续

handleRequest()を再呼び出したり、メソッドの先頭から実行し直したりすることはない。再開地点は、以前中断した位置にある。

park()がすぐには実行キューへ戻らない理由

TEXT

RUNNING
  -> YIELDING
  -> YIELDED
  -> 重新提交
  -> RUNNING

実行スタックが凍結された直後に仮想スレッドを実行キューへ戻すと、毎回再開して条件を確認して再び中断するため、CPUを空回りさせてしまう。

TEXT

RUNNING
  -> PARKING
  -> PARKED

他のスレッドがunpark()を呼んで初めて、仮想スレッドは実行可能状態に戻る。

TEXT

PARKED
  -> UNPARKED
  -> 重新提交
  -> RUNNING

OpenJDKのafterYield()は現在状態を確認する。PARKINGはPARKEDへ遷移し、デフォルトでは再投入しない。YIELDINGはYIELDEDへ遷移して、直ちに投入する。（GitHub）

対応するコードは次のように簡略化できる。

JAVA

private void afterYield() {
    int s = state();
    if (s == PARKING) {
        setState(PARKED);
        if (parkPermit
                && compareAndSetState(PARKED, UNPARKED)) {
            submitRunContinuation();
        }
        return;
    }
    if (s == YIELDING) {
        setState(YIELDED);
        submitRunContinuation();
    }
}

CodeBlock Loading...

Socketの読み取りがPollerへ入る仕組み

ここからネットワークI/O経路へ入る。

JAVA

int count = socket.getInputStream().read(buffer);

簡略化した読み取りロジックは次の通りだ。

JAVA

int read(int fd, byte[] buffer) throws IOException {
    while (true) {
        int result = nonBlockingRead(fd, buffer);
        if (result >= 0) {
            return result;
        }
        Poller.poll(fd);
    }
}

CodeBlock Loading...

Poller.poll(fd)の内部は、次のように抽象化できる。

JAVA

void poll(int fd) {
    Thread thread = Thread.currentThread();
    waiters.put(fd, thread);
    registerWithEpoll(fd);
    LockSupport.park();
}

ここでは仮想スレッドを実行しているため、Thread.currentThread()は対応するVirtualThreadを返す。

現在、2つの対応付けが存在する。

Linuxカーネルのepoll記録は次の通りだ。

TEXT

监听 fd 37 的可读事件

JDKのPoller記録は次の通りだ。

TEXT

fd 37 -> VirtualThread@500

登録後、現在の仮想スレッドは次を実行する。

JAVA

LockSupport.park();

待機中も、仮想スレッドはCarrier Threadを占有せず、実行キューにも入っていない。Pollerが待機関係を保持し、Continuationが実行状態を保持する。

TEXT

Linux epoll
    └── fd 37
Poller
    └── fd 37 -> VirtualThread@500
VirtualThread@500
    ├── state = PARKED
    ├── carrierThread = null
    └── Continuation
          └── 保存 read() 及其上层调用栈
ForkJoinPool
    └── 当前没有该虚拟线程的恢复任务

CodeBlock Loading...

ネットワークが準備できた後の再スケジュール

ネットワークデータが届くと、Linuxのプロトコルスタックはバイト列をSocketのカーネル受信バッファへ入れ、fdを可読としてマークする。

Linux上のPollerスレッドは通常epoll_wait()でブロックしている。イベントが返ると、準備できたファイル記述子が得られる。

JAVA

int[] readyFds = epollWait();
for (int fd : readyFds) {
    Thread thread = waiters.remove(fd);
    if (thread != null) {
        LockSupport.unpark(thread);
    }
}

VirtualThread.unpark()の主要なロジックは次の通りだ。

JAVA

void unpark() {
    Thread currentThread = Thread.currentThread();
    if (!getAndSetParkPermit(true)
            && currentThread != this) {
        int s = state();
        if (s == PARKED
                && compareAndSetState(PARKED, UNPARKED)) {
            submitRunContinuation();
        }
    }
}

CodeBlock Loading...

submitRunContinuation()は最終的に次を実行する。

JAVA

scheduler.execute(runContinuation);

unpark()を呼ぶのは通常Poller Threadだ。Poller ThreadはデフォルトスケジューラのWorkerではないため、このタスクは通常ForkJoinPoolの外部提出経路を通る。

後続のCarrier Threadがタスクを取り出し、再び次を呼び出す。

JAVA

virtualThread.runContinuation();

状態はUNPARKEDからRUNNINGへ変わり、仮想スレッドは新しいCarrier Threadへマウントされる。そのうえで次を実行する。

JAVA

cont.run();

以前凍結された呼び出しスタックが復元され、LockSupport.park()が戻り、Socketの読み取り処理が続く。

スケジュールキューにネットワークデータは入らない

スケジュールキューにネットワークデータが入ることはない。次のような物体も存在しない。

JAVA

new ResumeTask(virtualThread, networkData);

仮想スレッドの再開後、読み取りループはもう一度OSを呼び出す。

JAVA

int read(int fd, byte[] buffer) throws IOException {
    while (true) {
        int result = nonBlockingRead(fd, buffer);
        if (result >= 0) {
            return result;
        }
        Poller.poll(fd);
        // park 返回后，再次进入循环
    }
}

CodeBlock Loading...

最初のnonBlockingRead()がEAGAINを返すと、仮想スレッドは待ち状態に入る。

ネットワークが準備できた後、Pollerはunpark()を呼び、仮想スレッドが再開する。続いてもう一度次を実行する。

JAVA

nonBlockingRead(fd, buffer);

今度はカーネル受信バッファにデータがある。システムコールはバイト列を元のbufferへコピーし、バイト数を返す。

元のbufferへ引き続きアクセスできるのは、Continuationが保存した呼び出しスタックがまだbufferを保持しているからだ。

TEXT

buffer -> byte[]@300

データの流れは次の通りだ。

TEXT

网卡
  -> Linux 网络协议栈
  -> Socket 内核接收缓冲区
  -> 虚拟线程被唤醒
  -> 恢复 read() 调用栈
  -> 再次执行 read()
  -> 数据复制到原 buffer

CodeBlock Loading...

スケジュールキューが渡すのは実行資格だけだ。Pollerが渡すのは準備完了通知だけだ。ネットワークデータはカーネルのSocketバッファが保持する。

parkPermitが失われた通知を防ぐ仕組み

仮想スレッドが中断を準備している間に、ネットワークイベントが先に到達することがある。

次の時系列を考える。

TEXT

虚拟线程发现当前没有数据
Poller 注册 fd
网络数据立即到达
Poller 调用 unpark()
虚拟线程随后才执行 park()

parkPermitは容量1の許可のように機能する。

JAVA

private volatile boolean parkPermit;

unpark()はまず次を設定する。

JAVA

parkPermit = true;

park()は開始時にまず許可を消費しようとする。

JAVA

if (getAndSetParkPermit(false)) {
    return;
}

ネットワークイベントがすでに到着していれば、park()はそのまま戻り、呼び出しスタックを凍結しない。

unpark()を連続して複数回呼んでも、保存される許可は1つだけだ。

JAVA

unpark(thread);
unpark(thread);
unpark(thread);

最終的な状態は次の通りになる。

TEXT

parkPermit = true

次のpark()が許可を消費すると、falseに戻る。

JUCの同期機構が通常条件のチェックをループで行うのも、この理由だ。

JAVA

while (!conditionSatisfied()) {
    LockSupport.park();
}

park()が戻ることは、スレッドが実行機会を得たことだけを意味している。業務条件は依然として再確認が必要だ。

PARKINGによる重複スケジュールの防止

呼び出しスタックの凍結には一段の実行過程を要するため、RUNNINGとPARKEDの2つの状態だけでは足りない。

仮想スレッドはまず次を実行する。

JAVA

state = PARKING;
Continuation.yield();

PARKINGを設定してから、Continuationの凍結が完了するまでの間に、Pollerがすでにunpark()を呼ぶ可能性がある。

そのため、PARKING状態ではunpark()はparkPermitを設定するだけで、タスクを投入しない。

Continuationの凍結後、元のCarrierはafterYield()を実行する。

JAVA

state = PARKED;
if (parkPermit
        && compareAndSetState(PARKED, UNPARKED)) {
    submitRunContinuation();
}

これにより，提前到着した覚醒通知を残しつつ、同じ仮想スレッドを2本のCarrier Threadが同時に実行することも防げる。

3つのスレッドの連携

完全な実装にはForkJoinPool、Continuation、Poller、オペレーティングシステムが関わるが、実行過程は3つの通常のスレッドループに分解できる。

Carrier Threadのループは次の通りだ。

JAVA

while (true) {
    Runnable task = scheduler.takeTask();
    task.run();
}

Poller Threadのループは次の通りだ。

JAVA

while (true) {
    int[] readyFds = epollWait();
    for (int fd : readyFds) {
        VirtualThread thread = waiters.remove(fd);
        LockSupport.unpark(thread);
    }
}

CodeBlock Loading...

仮想スレッド内で行 되는Socketの読み取りループは次の通りだ。

JAVA

while (true) {
    int count = tryRead(fd, buffer);
    if (count >= 0) {
        return count;
    }
    registerPoller(fd, Thread.currentThread());
    LockSupport.park();
}

CodeBlock Loading...

3つの実行フローはいずれも順番に処理を進める。次のオブジェクトを通じて彼此がつながっている。

TEXT

ForkJoinPool WorkQueue
    保存可运行的 runContinuation
VirtualThread
    持有 scheduler、Continuation 和线程状态
Poller
    保存 fd 到 VirtualThread 的等待关系
Socket 内核接收缓冲区
    保存网络数据

CodeBlock Loading...

完全な呼び出し経路

全体を接続すると、次の主要経路が得られる。

TEXT

Thread.startVirtualThread
  -> VirtualThread.start
  -> submitRunContinuation
  -> ForkJoinPool 工作队列
  -> Carrier 取出任务
  -> VirtualThread.runContinuation
  -> mount
  -> Continuation.run
  -> 用户代码
  -> Socket.read
  -> 非阻塞 read 返回 EAGAIN
  -> Poller 注册 fd
  -> LockSupport.park
  -> VirtualThread.park
  -> Continuation.yield
  -> 调用栈被冻结
  -> cont.run 暂时返回
  -> unmount
  -> VirtualThread 进入 PARKED
  -> Carrier 返回调度循环
网络数据到达
  -> Socket 内核接收缓冲区
  -> epoll_wait 返回 fd
  -> Poller 找到 VirtualThread
  -> LockSupport.unpark
  -> VirtualThread.unpark
  -> submitRunContinuation
  -> ForkJoinPool 工作队列
  -> Carrier 取出恢复任务
  -> VirtualThread.runContinuation
  -> mount
  -> Continuation.run
  -> 恢复被冻结的调用栈
  -> LockSupport.park 返回
  -> 再次执行 read
  -> 数据复制到原 buffer
  -> 用户代码继续

CodeBlock Loading...

JDK 21とJDK 24のPinningの違い

参照ソース

JEP 444：Virtual Threads。（OpenJDK）
JDK 21 java.lang.VirtualThread。（GitHub）
JDK 21 jdk.internal.vm.Continuation。（GitHub）
JDK 21 sun.nio.ch.Poller。（GitHub）
JEP 491：Synchronize Virtual Threads without Pinning。（OpenJDK）