Hatena::ブログ(Diary)

CLOVER

2017-09-30

UndertowでHttpHandlerを実行するスレッドを切り替える

Undertow…というかXNIOでは、2種類のスレッドがあります。

IO Threadと、ブロッキングなタスクに使用することができるWorker Threadです。

Management of IO and Worker threads

XNIO workers

※XNIOのドキュメント、あんまり書かれてないですねぇ…

IO Threadは、ノンブロッキングなタスクを実行するためのスレッドで、ブロックするような処理をしてはいけません(他の接続への処理が止まる可能性があります)。
CPUコアあたり、2つのスレッドが適切なデフォルト値のようですが…?

Worker Threadは、ブロッキングなタスク用のスレッドプールで管理されます。一般に、CPUコア数あたり10スレッドで十分高いはず、
という感じのドキュメントになっています。

なお、Undertowのコードを見ると、それぞれのスレッドのデフォルト数はIO ThreadがCPUコア数 or 2の大きい方、
Worker ThreadがIO Thread×8です。
https://github.com/undertow-io/undertow/blob/1.4.20.Final/core/src/main/java/io/undertow/Undertow.java#L414-L415

で、ドキュメントにIO ThreadからWorker Threadに切り替える方法が書いてあります。

Request Lifecycle

Undertow Request Lifecyle

HttpServerExchange#exchangeを使いなさい、と。

今回、これを試してみましょう。

準備

まずは、Maven依存関係から。

        <dependency>
            <groupId>io.undertow</groupId>
            <artifactId>undertow-core</artifactId>
            <version>1.4.20.Final</version>
        </dependency>

簡単な起動クラスを作成します。
src/main/java/org/littlewings/undertow/App.java

package org.littlewings.undertow;

import io.undertow.Undertow;

public class App {
    public static void main(String... args) {
        Undertow undertow =
                Undertow
                        .builder()
                        .addHttpListener(8080, "localhost")
                        .setHandler(/** HttpHandlerを設定する **/)
                        .build();

        undertow.start();

        System.console().readLine("> Enter, stop.");

        undertow.stop();
    }
}

このBuilder#setHandlerに設定するHandlerを、ちょっとずつ変えながら見ていってみましょう。

                        .setHandler(/** HttpHandlerを設定する **/)

単純なサンプル

とりあえず、動かすための簡単なサンプルを書いてみます。
src/main/java/org/littlewings/undertow/SimpleHttpHandler.java

package org.littlewings.undertow;

import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;

import io.undertow.server.HttpHandler;
import io.undertow.server.HttpServerExchange;
import org.jboss.logging.Logger;

public class SimpleHttpHandler implements HttpHandler {
    Logger logger = Logger.getLogger(getClass());

    @Override
    public void handleRequest(HttpServerExchange exchange) throws Exception {
        logger.infof(
                "%s [%s/%s] access, is io-thread? %b, is blocking? %b",
                LocalDateTime.now().format(DateTimeFormatter.BASIC_ISO_DATE),
                Thread.currentThread().getName(),
                Thread.currentThread().getClass().getName(),
                exchange.isInIoThread(),
                exchange.isBlocking());

        exchange
                .getResponseSender()
                .send(
                        String.format(
                                "Response from: %s/%s, is io-thread? %b, is blocking? %b%n",
                                Thread.currentThread().getName(),
                                Thread.currentThread().getClass().getName(),
                                exchange.isInIoThread(),
                                exchange.isBlocking()
                        ));
    }
}

スレッド名やクラス、現在がIO Threadかどうか?ブロッキングしているかどうか?を返すようにしてみました。

これを設定して

                        .setHandler(new SimpleHttpHandler())

確認。

$ curl http://localhost:8080
Response from: XNIO-1 I/O-1/org.xnio.nio.WorkerThread, is io-thread? true, is blocking? false
$ curl http://localhost:8080
Response from: XNIO-1 I/O-2/org.xnio.nio.WorkerThread, is io-thread? true, is blocking? false
$ curl http://localhost:8080
Response from: XNIO-1 I/O-3/org.xnio.nio.WorkerThread, is io-thread? true, is blocking? false
$ curl http://localhost:8080
Response from: XNIO-1 I/O-4/org.xnio.nio.WorkerThread, is io-thread? true, is blocking? false

HttpServerExchange#isInIoThreadがtrueを返すので、IO Threadで動作しているようです。

とりあえず、わっとアクセスしてみて

$ for i in `seq 200`; do curl http://localhost:8080; done

スレッドダンプを見ると、XNIOのIO Threadしかいないようです。

$ jstack [PID] | grep XNIO
"XNIO-1 Accept" #20 prio=5 os_prio=0 tid=0x00007f567c53e000 nid=0x2d5b runnable [0x00007f56579f8000]
"XNIO-1 I/O-8" #19 prio=5 os_prio=0 tid=0x00007f567c53c800 nid=0x2d5a runnable [0x00007f5657af9000]
"XNIO-1 I/O-7" #18 prio=5 os_prio=0 tid=0x00007f567c53b000 nid=0x2d59 runnable [0x00007f5657bfa000]
"XNIO-1 I/O-6" #17 prio=5 os_prio=0 tid=0x00007f567c539000 nid=0x2d58 runnable [0x00007f5657cfb000]
"XNIO-1 I/O-5" #16 prio=5 os_prio=0 tid=0x00007f567c537800 nid=0x2d57 runnable [0x00007f5657dfc000]
"XNIO-1 I/O-4" #15 prio=5 os_prio=0 tid=0x00007f567c535800 nid=0x2d56 runnable [0x00007f5657efd000]
"XNIO-1 I/O-3" #14 prio=5 os_prio=0 tid=0x00007f567c534000 nid=0x2d55 runnable [0x00007f5657ffe000]
"XNIO-1 I/O-2" #13 prio=5 os_prio=0 tid=0x00007f567c532000 nid=0x2d54 runnable [0x00007f5664192000]
"XNIO-1 I/O-1" #12 prio=5 os_prio=0 tid=0x00007f567c530800 nid=0x2d53 runnable [0x00007f5664293000]

なるほど?

IO ThreadからWorker Threadに切り替える

では、IO ThreadからWorker Threadに切り替えてみましょう。
src/main/java/org/littlewings/undertow/DispatchWorkerHttpHandler.java

package org.littlewings.undertow;

import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;

import io.undertow.server.HttpHandler;
import io.undertow.server.HttpServerExchange;
import org.jboss.logging.Logger;

public class DispatchWorkerHttpHandler implements HttpHandler {
    Logger logger = Logger.getLogger(getClass());

    @Override
    public void handleRequest(HttpServerExchange exchange) throws Exception {
        logger.infof(
                "%s [%s/%s] access, is io-thread? %b, is blocking? %b",
                LocalDateTime.now().format(DateTimeFormatter.BASIC_ISO_DATE),
                Thread.currentThread().getName(),
                Thread.currentThread().getClass().getName(),
                exchange.isInIoThread(),
                exchange.isBlocking());

        if (exchange.isInIoThread()) {
            exchange.dispatch(this);
            logger.infof("return, dispatch to worker");
            return;
        }

        exchange
                .getResponseSender()
                .send(
                        String.format(
                                "Response from: %s/%s, is io-thread? %b, is blocking? %b%n",
                                Thread.currentThread().getName(),
                                Thread.currentThread().getClass().getName(),
                                exchange.isInIoThread(),
                                exchange.isBlocking()
                        ));
    }
}

ここでのポイントは、HttpServerExchange#isInIoThreadがtrueの場合、HttpServerExchange#dispatchでHttpHandlerを渡して折り返すことですね。

        if (exchange.isInIoThread()) {
            exchange.dispatch(this);
            logger.infof("return, dispatch to worker");
            return;
        }

渡しているのがthisなので、このHttpHandlerがWorker Threadで処理されるようになります。

こちらのHttpHandlerを使用するように、Appクラスを修正します。

                        .setHandler(new DispatchWorkerHttpHandler())

確認してみましょう。

$ curl http://localhost:8080
Response from: XNIO-1 task-5/java.lang.Thread, is io-thread? false, is blocking? false

この時、HttpHandler側にはhandleRequestの先頭のログが2回出力されることになります。

9 30, 2017 12:16:53 午後 org.littlewings.undertow.DispatchWorkerHttpHandler handleRequest
INFO: return, dispatch to worker
9 30, 2017 12:16:53 午後 org.littlewings.undertow.DispatchWorkerHttpHandler handleRequest
INFO: 20170930 [XNIO-1 task-5/java.lang.Thread] access, is io-thread? false, is blocking? false

HttpServerExchange#isInIoThreadがfalseになり、出てくるスレッド名も変わりましたね。

また、ざっとアクセスしてみます。

$ for i in `seq 200`; do curl http://localhost:8080; done

スレッドダンプを見てみると、大量のスレッドが出現します。

$ jstack [PID] | grep XNIO
"XNIO-1 task-64" #85 prio=5 os_prio=0 tid=0x00007f211c011800 nid=0x3287 waiting on condition [0x00007f216cdcc000]
"XNIO-1 task-63" #84 prio=5 os_prio=0 tid=0x00007f20f8048000 nid=0x3284 waiting on condition [0x00007f216cecd000]
"XNIO-1 task-62" #83 prio=5 os_prio=0 tid=0x00007f2104011000 nid=0x3281 waiting on condition [0x00007f216cfce000]
"XNIO-1 task-61" #82 prio=5 os_prio=0 tid=0x00007f2100011800 nid=0x327e waiting on condition [0x00007f216d0cf000]
"XNIO-1 task-60" #81 prio=5 os_prio=0 tid=0x00007f210c011000 nid=0x327b waiting on condition [0x00007f216d1d0000]
"XNIO-1 task-59" #80 prio=5 os_prio=0 tid=0x00007f2108011800 nid=0x3278 waiting on condition [0x00007f216d2d1000]
"XNIO-1 task-58" #79 prio=5 os_prio=0 tid=0x00007f2114011000 nid=0x3275 waiting on condition [0x00007f216d3d2000]
"XNIO-1 task-57" #78 prio=5 os_prio=0 tid=0x00007f211000f800 nid=0x3272 waiting on condition [0x00007f216d4d3000]
"XNIO-1 task-56" #77 prio=5 os_prio=0 tid=0x00007f211c00f800 nid=0x326f waiting on condition [0x00007f216d5d4000]
"XNIO-1 task-55" #76 prio=5 os_prio=0 tid=0x00007f20f8046800 nid=0x326c waiting on condition [0x00007f216d6d5000]
"XNIO-1 task-54" #75 prio=5 os_prio=0 tid=0x00007f210400f800 nid=0x3269 waiting on condition [0x00007f216d7d6000]
"XNIO-1 task-53" #74 prio=5 os_prio=0 tid=0x00007f210000f800 nid=0x3266 waiting on condition [0x00007f216d8d7000]
"XNIO-1 task-52" #73 prio=5 os_prio=0 tid=0x00007f210c00f800 nid=0x3263 waiting on condition [0x00007f216d9d8000]
"XNIO-1 task-51" #72 prio=5 os_prio=0 tid=0x00007f210800f800 nid=0x3260 waiting on condition [0x00007f216dad9000]
"XNIO-1 task-50" #71 prio=5 os_prio=0 tid=0x00007f211400f800 nid=0x325d waiting on condition [0x00007f216dbda000]
"XNIO-1 task-49" #70 prio=5 os_prio=0 tid=0x00007f211000e000 nid=0x325a waiting on condition [0x00007f216dcdb000]
"XNIO-1 task-48" #69 prio=5 os_prio=0 tid=0x00007f211c00e000 nid=0x3257 waiting on condition [0x00007f216dddc000]
"XNIO-1 task-47" #68 prio=5 os_prio=0 tid=0x00007f20f8044800 nid=0x3254 waiting on condition [0x00007f216dedd000]
"XNIO-1 task-46" #67 prio=5 os_prio=0 tid=0x00007f210400d800 nid=0x3251 waiting on condition [0x00007f216dfde000]
"XNIO-1 task-45" #66 prio=5 os_prio=0 tid=0x00007f210000e000 nid=0x324e waiting on condition [0x00007f216e0df000]
"XNIO-1 task-44" #65 prio=5 os_prio=0 tid=0x00007f210c00d800 nid=0x324b waiting on condition [0x00007f216e1e0000]
"XNIO-1 task-43" #64 prio=5 os_prio=0 tid=0x00007f210800e000 nid=0x3248 waiting on condition [0x00007f216e2e1000]
"XNIO-1 task-42" #63 prio=5 os_prio=0 tid=0x00007f211400d800 nid=0x3245 waiting on condition [0x00007f216e3e2000]
"XNIO-1 task-41" #62 prio=5 os_prio=0 tid=0x00007f211000c000 nid=0x3242 waiting on condition [0x00007f216e4e3000]
"XNIO-1 task-40" #61 prio=5 os_prio=0 tid=0x00007f211c00c000 nid=0x323f waiting on condition [0x00007f216e5e4000]
"XNIO-1 task-39" #60 prio=5 os_prio=0 tid=0x00007f20f803d800 nid=0x323c waiting on condition [0x00007f216e6e5000]
"XNIO-1 task-38" #59 prio=5 os_prio=0 tid=0x00007f210400c000 nid=0x3239 waiting on condition [0x00007f216e7e6000]
"XNIO-1 task-37" #58 prio=5 os_prio=0 tid=0x00007f210000c000 nid=0x3236 waiting on condition [0x00007f216e8e7000]
"XNIO-1 task-36" #57 prio=5 os_prio=0 tid=0x00007f210c00c000 nid=0x3233 waiting on condition [0x00007f216e9e8000]
"XNIO-1 task-35" #56 prio=5 os_prio=0 tid=0x00007f210800c000 nid=0x3230 waiting on condition [0x00007f216eae9000]
"XNIO-1 task-34" #55 prio=5 os_prio=0 tid=0x00007f211400c000 nid=0x322c waiting on condition [0x00007f216ebea000]
"XNIO-1 task-33" #54 prio=5 os_prio=0 tid=0x00007f211000a800 nid=0x3229 waiting on condition [0x00007f216eceb000]
"XNIO-1 task-32" #53 prio=5 os_prio=0 tid=0x00007f211c00a800 nid=0x3226 waiting on condition [0x00007f216edec000]
"XNIO-1 task-31" #52 prio=5 os_prio=0 tid=0x00007f20f803c000 nid=0x3223 waiting on condition [0x00007f216eeed000]
"XNIO-1 task-30" #51 prio=5 os_prio=0 tid=0x00007f210400a000 nid=0x3220 waiting on condition [0x00007f216efee000]
"XNIO-1 task-29" #50 prio=5 os_prio=0 tid=0x00007f210000a800 nid=0x321d waiting on condition [0x00007f216f0ef000]
"XNIO-1 task-28" #49 prio=5 os_prio=0 tid=0x00007f210c00a000 nid=0x321a waiting on condition [0x00007f216f1f0000]
"XNIO-1 task-27" #48 prio=5 os_prio=0 tid=0x00007f210800a800 nid=0x3217 waiting on condition [0x00007f216f2f1000]
"XNIO-1 task-26" #47 prio=5 os_prio=0 tid=0x00007f211400a000 nid=0x3214 waiting on condition [0x00007f216f3f2000]
"XNIO-1 task-25" #46 prio=5 os_prio=0 tid=0x00007f2110008800 nid=0x3211 waiting on condition [0x00007f216f4f3000]
"XNIO-1 task-24" #45 prio=5 os_prio=0 tid=0x00007f211c008800 nid=0x320e waiting on condition [0x00007f216f5f4000]
"XNIO-1 task-23" #44 prio=5 os_prio=0 tid=0x00007f20f803a000 nid=0x320b waiting on condition [0x00007f216f6f5000]
"XNIO-1 task-22" #43 prio=5 os_prio=0 tid=0x00007f2104008800 nid=0x3208 waiting on condition [0x00007f216f7f6000]
"XNIO-1 task-21" #42 prio=5 os_prio=0 tid=0x00007f2100008800 nid=0x3205 waiting on condition [0x00007f216f8f7000]
"XNIO-1 task-20" #41 prio=5 os_prio=0 tid=0x00007f210c008800 nid=0x3202 waiting on condition [0x00007f216f9f8000]
"XNIO-1 task-19" #40 prio=5 os_prio=0 tid=0x00007f2108008800 nid=0x31ff waiting on condition [0x00007f216faf9000]
"XNIO-1 task-18" #39 prio=5 os_prio=0 tid=0x00007f2114008800 nid=0x31fc waiting on condition [0x00007f216fbfa000]
"XNIO-1 task-17" #38 prio=5 os_prio=0 tid=0x00007f2110007000 nid=0x31f9 waiting on condition [0x00007f216fcfb000]
"XNIO-1 task-16" #37 prio=5 os_prio=0 tid=0x00007f211c007000 nid=0x31f6 waiting on condition [0x00007f216fdfc000]
"XNIO-1 task-15" #36 prio=5 os_prio=0 tid=0x00007f20f8038800 nid=0x31f3 waiting on condition [0x00007f216fefd000]
"XNIO-1 task-14" #35 prio=5 os_prio=0 tid=0x00007f2104007000 nid=0x31f0 waiting on condition [0x00007f216fffe000]
"XNIO-1 task-13" #34 prio=5 os_prio=0 tid=0x00007f2100007000 nid=0x31ed waiting on condition [0x00007f21741cb000]
"XNIO-1 task-12" #33 prio=5 os_prio=0 tid=0x00007f210c007000 nid=0x31ea waiting on condition [0x00007f21742cc000]
"XNIO-1 task-11" #32 prio=5 os_prio=0 tid=0x00007f2108007000 nid=0x31e7 waiting on condition [0x00007f21743cd000]
"XNIO-1 task-10" #31 prio=5 os_prio=0 tid=0x00007f2114007000 nid=0x31e4 waiting on condition [0x00007f21744ce000]
"XNIO-1 task-9" #30 prio=5 os_prio=0 tid=0x00007f2110005000 nid=0x31e1 waiting on condition [0x00007f21745cf000]
"XNIO-1 task-8" #29 prio=5 os_prio=0 tid=0x00007f211c005800 nid=0x31de waiting on condition [0x00007f21746d0000]
"XNIO-1 task-7" #28 prio=5 os_prio=0 tid=0x00007f20f8037000 nid=0x31db waiting on condition [0x00007f21747d1000]
"XNIO-1 task-6" #27 prio=5 os_prio=0 tid=0x00007f2104005000 nid=0x31d8 waiting on condition [0x00007f21748d2000]
"XNIO-1 task-5" #26 prio=5 os_prio=0 tid=0x00007f2100005000 nid=0x31d5 waiting on condition [0x00007f21749d3000]
"XNIO-1 task-4" #25 prio=5 os_prio=0 tid=0x00007f210c005000 nid=0x31d2 waiting on condition [0x00007f2174ad4000]
"XNIO-1 task-3" #24 prio=5 os_prio=0 tid=0x00007f2108005000 nid=0x31cf waiting on condition [0x00007f2174bd5000]
"XNIO-1 task-2" #23 prio=5 os_prio=0 tid=0x00007f2114005000 nid=0x31cc waiting on condition [0x00007f2174cd6000]
"XNIO-1 task-1" #22 prio=5 os_prio=0 tid=0x00007f20f8035000 nid=0x31c2 waiting on condition [0x00007f2174dd7000]
"XNIO-1 Accept" #20 prio=5 os_prio=0 tid=0x00007f219056e800 nid=0x31ba runnable [0x00007f21750d8000]
"XNIO-1 I/O-8" #19 prio=5 os_prio=0 tid=0x00007f219056c800 nid=0x31b9 runnable [0x00007f21751d9000]
"XNIO-1 I/O-7" #18 prio=5 os_prio=0 tid=0x00007f219056a800 nid=0x31b8 runnable [0x00007f21752da000]
"XNIO-1 I/O-6" #17 prio=5 os_prio=0 tid=0x00007f2190568800 nid=0x31b7 runnable [0x00007f21753db000]
"XNIO-1 I/O-5" #16 prio=5 os_prio=0 tid=0x00007f2190566800 nid=0x31b6 runnable [0x00007f21754dc000]
"XNIO-1 I/O-4" #15 prio=5 os_prio=0 tid=0x00007f2190564800 nid=0x31b5 runnable [0x00007f21755dd000]
"XNIO-1 I/O-3" #14 prio=5 os_prio=0 tid=0x00007f2190562800 nid=0x31b4 runnable [0x00007f21756de000]
"XNIO-1 I/O-2" #13 prio=5 os_prio=0 tid=0x00007f2190561000 nid=0x31b3 runnable [0x00007f21757df000]
"XNIO-1 I/O-1" #12 prio=5 os_prio=0 tid=0x00007f219055f800 nid=0x31b2 runnable [0x00007f21758e0000]

Worker Threadが使われるようになったようです。

同じスレッドにdispatchする

HttpServerExchange#dispatchには、いくつかオーバーロードされたバージョンがあります。
http://undertow.io/javadoc/1.4.x/io/undertow/server/HttpServerExchange.html

java.util.concurrent.Executorを渡せるdispatchメソッドもあり、Undertowの提供するSameThreadExecutorを使用すると、同じスレッドで処理を続けるように
dispatchすることができます。
src/main/java/org/littlewings/undertow/SameThreadHttpHandler.java

package org.littlewings.undertow;

import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;

import io.undertow.server.HttpHandler;
import io.undertow.server.HttpServerExchange;
import io.undertow.util.SameThreadExecutor;
import org.jboss.logging.Logger;

public class SameThreadHttpHandler implements HttpHandler {
    Logger logger = Logger.getLogger(getClass());

    @Override
    public void handleRequest(HttpServerExchange exchange) throws Exception {
        logger.infof(
                "%s [%s/%s] access, is io-thread? %b, is blocking? %b",
                LocalDateTime.now().format(DateTimeFormatter.BASIC_ISO_DATE),
                Thread.currentThread().getName(),
                Thread.currentThread().getClass().getName(),
                exchange.isInIoThread(),
                exchange.isBlocking());

        exchange.dispatch(SameThreadExecutor.INSTANCE, () ->
                exchange
                        .getResponseSender()
                        .send(
                                String.format(
                                        "Response from: %s/%s, is io-thread? %b, is blocking? %b%n",
                                        Thread.currentThread().getName(),
                                        Thread.currentThread().getClass().getName(),
                                        exchange.isInIoThread(),
                                        exchange.isBlocking()
                                ))
        );
    }
}

SameThreadExecutorの中身は、単純ですけれど。

IO Threadに切り替える

さらに、IO Threadに切り替えることもできます。
src/main/java/org/littlewings/undertow/ToIoThreadHttpHandler.java

package org.littlewings.undertow;

import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;

import io.undertow.server.HttpHandler;
import io.undertow.server.HttpServerExchange;
import io.undertow.util.SameThreadExecutor;
import org.jboss.logging.Logger;

public class ToIoThreadHttpHandler implements HttpHandler {
    Logger logger = Logger.getLogger(getClass());

    @Override
    public void handleRequest(HttpServerExchange exchange) throws Exception {
        logger.infof(
                "%s [%s/%s] access, is io-thread? %b, is blocking? %b",
                LocalDateTime.now().format(DateTimeFormatter.BASIC_ISO_DATE),
                Thread.currentThread().getName(),
                Thread.currentThread().getClass().getName(),
                exchange.isInIoThread(),
                exchange.isBlocking());

        if (!exchange.isInIoThread()) {
            exchange.dispatch(exchange.getIoThread());
            logger.infof("return, dispatch to io");
            return;
        }

        exchange.dispatch(SameThreadExecutor.INSTANCE, () ->
                exchange
                        .getResponseSender()
                        .send(
                                String.format(
                                        "Response from: %s/%s, is io-thread? %b, is blocking? %b%n",
                                        Thread.currentThread().getName(),
                                        Thread.currentThread().getClass().getName(),
                                        exchange.isInIoThread(),
                                        exchange.isBlocking()
                                )
                        ));
    }
}

ここでは、IO ThreadでなければIO Threadにdispatchするという内容になっています。

        if (!exchange.isInIoThread()) {
            exchange.dispatch(exchange.getIoThread());
            logger.infof("return, dispatch to io");
            return;
        }

この場合、dispatchの引数はRunnableとして扱われています。

極端なことをすると、先ほど作成したWorker Threadに切り替えるHttpHandlerから、さらにIO Threadに切り替えるみたいなことができます。
src/main/java/org/littlewings/undertow/DispatchWorkerHttpHandler.java

package org.littlewings.undertow;

import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;

import io.undertow.server.HttpHandler;
import io.undertow.server.HttpServerExchange;
import org.jboss.logging.Logger;

public class DispatchWorkerHttpHandler implements HttpHandler {
    Logger logger = Logger.getLogger(getClass());

    @Override
    public void handleRequest(HttpServerExchange exchange) throws Exception {
        logger.infof(
                "%s [%s/%s] access, is io-thread? %b, is blocking? %b",
                LocalDateTime.now().format(DateTimeFormatter.BASIC_ISO_DATE),
                Thread.currentThread().getName(),
                Thread.currentThread().getClass().getName(),
                exchange.isInIoThread(),
                exchange.isBlocking());

        if (exchange.isInIoThread()) {
            exchange.dispatch(this);
            logger.infof("return, dispatch to worker");
            return;
        }

        exchange.dispatch(exchange.getIoThread(), new ToIoThreadHttpHandler());
    }
}

この場合、HttpServerExchange#getIoThreadを使ってdispatchするところがポイントです。

        exchange.dispatch(exchange.getIoThread(), new ToIoThreadHttpHandler());

Blocking?

ところで、ここまでずっと

exchange.isBlocking()

とかでブロッキングかどうかを見ていたのですが、実はこれ、いつもfalseです。

これがtrueを返すようになるには、HttpServerExchange#startBlockingする必要があります。

Non-blocking IO

Blocking IO

HttpServerExchange#startBlockingを使うことで、ブロッキングIOのAPIを使うことができるようになります。

UndertowのServlet実装が、これを使っています。
https://github.com/undertow-io/undertow/blob/1.4.20.Final/servlet/src/main/java/io/undertow/servlet/handlers/ServletInitialHandler.java#L185

で、Worker Threadに切り替えます、と。
https://github.com/undertow-io/undertow/blob/1.4.20.Final/servlet/src/main/java/io/undertow/servlet/handlers/ServletInitialHandler.java#L193-L196

注意点としては、HttpServerExchange#startBlockingした後は、Blocking IOを使うことになるので、IO ThreadからRead/Writeは
禁止されます。
https://github.com/undertow-io/undertow/blob/1.4.20.Final/core/src/main/java/io/undertow/io/UndertowInputStream.java#L83-L85
https://github.com/undertow-io/undertow/blob/1.4.20.Final/core/src/main/java/io/undertow/io/UndertowOutputStream.java#L112-L114

このstartBlockingを使う前は、リクエスト/レスポンスへの入出力はAsyncなSender/Receiverが利用されています。
https://github.com/undertow-io/undertow/blob/1.4.20.Final/core/src/main/java/io/undertow/server/HttpServerExchange.java#L1306-L1332
https://github.com/undertow-io/undertow/blob/1.4.20.Final/core/src/main/java/io/undertow/io/AsyncSenderImpl.java
https://github.com/undertow-io/undertow/blob/1.4.20.Final/core/src/main/java/io/undertow/io/AsyncReceiverImpl.java

startBlockingすることで、BlockingなSender/Receiverが使われるようになります。
https://github.com/undertow-io/undertow/blob/1.4.20.Final/core/src/main/java/io/undertow/io/BlockingSenderImpl.java
https://github.com/undertow-io/undertow/blob/1.4.20.Final/core/src/main/java/io/undertow/io/BlockingReceiverImpl.java

startBlockingで、HttpServerExchange中にBlockingHttpExchangeへの参照が作られるようになりますよ、と。
https://github.com/undertow-io/undertow/blob/1.4.20.Final/core/src/main/java/io/undertow/server/HttpServerExchange.java#L1458-L1492

もうちょっとスレッドの管理を見てみよう

スレッドダンプを取った時に、こんな感じのスレッドがわらわらと現れました。

"XNIO-1 task-4" #25 prio=5 os_prio=0 tid=0x00007f210c005000 nid=0x31d2 waiting on condition [0x00007f2174ad4000]
"XNIO-1 task-3" #24 prio=5 os_prio=0 tid=0x00007f2108005000 nid=0x31cf waiting on condition [0x00007f2174bd5000]
"XNIO-1 task-2" #23 prio=5 os_prio=0 tid=0x00007f2114005000 nid=0x31cc waiting on condition [0x00007f2174cd6000]
"XNIO-1 task-1" #22 prio=5 os_prio=0 tid=0x00007f20f8035000 nid=0x31c2 waiting on condition [0x00007f2174dd7000]
"XNIO-1 Accept" #20 prio=5 os_prio=0 tid=0x00007f219056e800 nid=0x31ba runnable [0x00007f21750d8000]
"XNIO-1 I/O-8" #19 prio=5 os_prio=0 tid=0x00007f219056c800 nid=0x31b9 runnable [0x00007f21751d9000]
"XNIO-1 I/O-7" #18 prio=5 os_prio=0 tid=0x00007f219056a800 nid=0x31b8 runnable [0x00007f21752da000]
"XNIO-1 I/O-6" #17 prio=5 os_prio=0 tid=0x00007f2190568800 nid=0x31b7 runnable [0x00007f21753db000]

これらを、もう少し見てみたいと思います。

IO Threadも含めて、Workerという扱いになっているっぽい…?
https://github.com/undertow-io/undertow/blob/1.4.20.Final/core/src/main/java/io/undertow/Undertow.java#L119

スレッドの作成は、XNIO側になります。
https://github.com/xnio/xnio/blob/3.3.8.Final/api/src/main/java/org/xnio/Xnio.java#L476-L511
https://github.com/xnio/xnio/blob/3.3.8.Final/nio-impl/src/main/java/org/xnio/nio/NioXnio.java#L202

IO Threadと、Acceptを管理するスレッドが作られるのはNioXnioWorkerのようですね。
https://github.com/xnio/xnio/blob/3.3.8.Final/nio-impl/src/main/java/org/xnio/nio/NioXnioWorker.java#L96-L107

そのNioXnioWorkerの親クラスであるXniWorkerが、Worker Threadの管理をしています。
https://github.com/xnio/xnio/blob/3.3.8.Final/api/src/main/java/org/xnio/XnioWorker.java#L117-L124
https://github.com/xnio/xnio/blob/3.3.8.Final/api/src/main/java/org/xnio/XnioWorker.java#L954

HttpServerExchange#dispatchでWorker Threadに委譲されるのは、現在の接続からXniWorkerが取得でき、そこからexecuteすると
https://github.com/undertow-io/undertow/blob/1.4.20.Final/core/src/main/java/io/undertow/server/HttpServerExchange.java#L795

XniWorkerが管理しているTaskPoolに突っ込まれるからですね。
https://github.com/xnio/xnio/blob/3.3.8.Final/api/src/main/java/org/xnio/XnioWorker.java#L748-L750

なお、HttpServerExchange#getIoThreadを行うと、現在の接続に紐づいているIO Threadを直接取得できます。
https://github.com/undertow-io/undertow/blob/1.4.20.Final/core/src/main/java/io/undertow/server/HttpServerExchange.java#L1767-L1769

戻り値の型はXnioIoThreadですが、先ほどNioXnioWorkerで作成していたIO Threadは、WorkerThreadというクラスでした。
https://github.com/xnio/xnio/blob/3.3.8.Final/nio-impl/src/main/java/org/xnio/nio/WorkerThread.java#L79

これは、XnioIoThreadのサブクラスです。

Worker Threadはどうなのか?ふつうのThreadですね。
https://github.com/xnio/xnio/blob/3.3.8.Final/api/src/main/java/org/xnio/XnioWorker.java#L954

そういえば、curlの結果で戻ってきたスレッドのクラス名も、IO ThreadがWorkerThreadで

$ curl http://localhost:8080
Response from: XNIO-1 I/O-1/org.xnio.nio.WorkerThread, is io-thread? true, is blocking? false

Worker ThreadがThreadでしたね。

$ curl http://localhost:8080
Response from: XNIO-1 task-5/java.lang.Thread, is io-thread? false, is blocking? false

まとめ

ちょっとした理由からなのですが、UndertowでのIO Thread、Worker Threadの切り替えや、スレッドまわりをちょっと追ってみました。

前々から少しは気になっていた箇所なのですが、意外と面白かったです。

スパム対策のためのダミーです。もし見えても何も入力しないでください
ゲスト


画像認証

トラックバック - http://d.hatena.ne.jp/Kazuhira/20170930/1506743975