understand Undertow threads with a real life example

[Nishant Kumar]

Hi,

I am using Undertow 2.2.10.Final on Linux machine. I have written the complete code but am not sure the way I am moving between IO and Worker threads. I couldn't find out any similar code or doc over the internet which can help me. Based on my understanding of Undertow, I am posting here the sample code along with the proper comment. 

I will appreciate it if someone can review it and confirm if it is the correct way to do it. I also get the below error sometimes-

java.lang.IllegalStateException: UT000146: HttpServerExchange cannot have both async IO resumed and dispatch() called in the same cycle

Sample code snippet (also attached full sample code):

public void handleRequest(HttpServerExchange exchange) throws Exception {
// read the complete request in IO thread and then pass it to worker thread
exchange.getRequestReceiver().receiveFullBytes(
(httpServerExchange, reqBytes) -> {
// IO Thread
httpServerExchange.dispatch(() -> {
// Worker Thread
new TestProcessor().process(httpServerExchange, reqBytes);
});
},
(httpServerExchange, exception) -> {
// IO Thread
httpServerExchange.dispatch(() -> {
// Worker Thread
//optional error handler
// e.g. Remote peer closed connection before all data could be read
log.error("Exception: {}", exception.getMessage());
new TestProcessor().sendErrorResponse(httpServerExchange);
});
});
}

private class TestProcessor {

public void process(HttpServerExchange httpServerExchange, byte[] reqBytes) {
try {
// In worker thread

// parse POST request body

// do some more work

// http sync call over network

// do some more work

if (some condition){

// async http call
ayncHttpClient.send(body, new Callback() {
@Override
public void onSuccess(Key[] keys, Record[] records) {
// moving from NIO Event Loop to undertow worker thread
httpServerExchange.dispatch(() -> {
// In worker thread

//parse response of callback

someWork(httpServerExchange);

httpServerExchange.endExchange();
});
}

@Override
public void onFailure(Exception exception) {
// moving from NIO Event Loop to undertow worker thread
httpServerExchange.dispatch(() -> {
sendErrorResponse(httpServerExchange);
});
}
});
}else{
someWork(httpServerExchange);

httpServerExchange.endExchange();
}
} catch (Exception e) {
sendErrorResponse(httpServerExchange);
}
}

public void someWork(HttpServerExchange httpServerExchange) {
// do some work
// some more work
// send response to client
httpServerExchange.setStatusCode(200);
httpServerExchange.getResponseHeaders().put(Headers.CONTENT_TYPE, "application/json");
httpServerExchange.getResponseSender().send(body);
}

public void sendErrorResponse(HttpServerExchange httpServerExchange) {
// send error msg to client
httpServerExchange.setStatusCode(204);
httpServerExchange.getResponseHeaders().put(Headers.CONTENT_TYPE, "text/plain");
httpServerExchange.getResponseSender().send("");

httpServerExchange.endExchange();
}
}

[Nishant Kumar]

do I need to call httpServerExchange.endExchange()?

I think it's not required but got confused by the statement  (https://undertow.io/undertow-docs/undertow-docs-2.1.0/index.html#request-lifecycle)- 

" The root handler can return after HttpServerExchange.dispatch() has been called, or after async IO has been started

In this case the dispatched task will be submitted to the dispatch executor, or if async IO has been started on either the request or response channels then this will be started. In this case the exchange will not be finished, it is up to your async task to finish the exchange when it is done processing. "

[Ruslan Ibragimov]

Can you post minimal working example in GitHub/Gitlab/etc? It will help to review your code

> do I need to call httpServerExchange.endExchange()?

Depends on use-case. It's required if you dispatch a request from an IO thread. Then undertow will not end exchange upon returning from root handler.

Example of place where endExchange is required:

class TestHandler : HttpHandler {
    override fun handleRequest(exchange: HttpServerExchange) {
        exchange.dispatch(SameThreadExecutor.INSTANCE, Runnable {
            println("This exchange will timeout")
            // Client Error: java.net.SocketTimeoutException: Read timed out
        })
    }
}

class TestHandlerWithEnd : HttpHandler {
    override fun handleRequest(exchange: HttpServerExchange) {
        exchange.dispatch(SameThreadExecutor.INSTANCE, Runnable {
            println("This exchange will ended immediately")
            exchange.endExchange()
        })
    }
}

--

You received this message because you are subscribed to the Google Groups "Undertow Dev" group.

To unsubscribe from this group and stop receiving emails from it, send an email to undertow-dev...@googlegroups.com.

To view this discussion on the web visit https://groups.google.com/d/msgid/undertow-dev/d3ffccbd-8f30-400b-8869-a2cbea9e02a8n%40googlegroups.com.

For more options, visit https://groups.google.com/d/optout.

--

Best regards,

Ruslan Ibragimov

https://ibragimov.by/

[Nishant Kumar]

I have added my sample code in github.
https://github.com/onlynishant/poc

[Nishant Kumar]

what's the difference (performance-wise or best practice) between -

exchange.dispatch(SameThreadExecutor.INSTANCE, Runnable {
println("This exchange will timeout")

// Client Error: java.net.SocketTimeoutException: Read timed out
})

Vs

exchange.dispatch(() -> {

println("This exchange will timeout");
// Client Error: java.net.SocketTimeoutException: Read timed out

});

[Ruslan Ibragimov]

In the first case, dispatch happens to the same thread (so no context-switch overhead), it's good if operation inside asynchronous as well (like calling async http client, or using something like Rx/Coroutines).

In the second case dispatch happens to Undertow Worker Pool, so it's good for running blocking code (like blocking IO, heavy computations, JDBC, etc)

On Fri, Sep 10, 2021, at 08:01, Nishant Kumar wrote:

what's the difference between -

exchange.dispatch(SameThreadExecutor.INSTANCE, Runnable {
println("This exchange will timeout")

// Client Error: java.net.SocketTimeoutException: Read timed out
})

Vs

exchange.dispatch(() -> {

println("This exchange will timeout");
// Client Error: java.net.SocketTimeoutException: Read timed out

});

To view this discussion on the web visit https://groups.google.com/d/msgid/undertow-dev/3d4391ac-b7cc-4327-944c-1b190369b787n%40googlegroups.com.

For more options, visit https://groups.google.com/d/optout.

[Ruslan Ibragimov]

I would say that code looks mostly fine, but if you want to achieve better performance you need to do single dispatch on the same thread before you do any work in handler https://github.com/onlynishant/poc/blob/544ca385d2b34af982414b28afe550469631eea8/undertow/src/main/java/UndertowTest.java#L79 and replace syncHttp call with async, for example with org.apache.httpcomponents.client5:httpclient5

To view this discussion on the web visit https://groups.google.com/d/msgid/undertow-dev/2955a788-94eb-4b53-bb2d-09c7e0b2094an%40googlegroups.com.

For more options, visit https://groups.google.com/d/optout.

[Nishant Kumar]

Thanks for the reply and code review. Can you please also comment about the exception i have mentioned. Below is the stack trace. We get it often in production server -

2021-09-28 05:00:44.358 [ERROR] [XNIO-1 I/O-31] io.undertow.server.Connectors:executeRootHandler(388): UT005071: Undertow request failed HttpServerExchange{ POST /}

java.lang.IllegalStateException: UT000146: HttpServerExchange cannot have both async IO resumed and dispatch() called in the same cycle

        at io.undertow.server.HttpServerExchange$ReadDispatchChannel.resumeReads(HttpServerExchange.java:2229) ~[undertow-core-2.2.10.Final.jar:2.2.10.Final]

        at io.undertow.io.AsyncReceiverImpl.receiveFullBytes(AsyncReceiverImpl.java:457) ~[undertow-core-2.2.10.Final.jar:2.2.10.Final]

        at io.platform.handler.RequestHandler.handleRequest(RequestHandler.java:212) ~[platform-java-1.0-SNAPSHOT.jar:?]

        at io.undertow.server.RoutingHandler.handleRequest(RoutingHandler.java:93) ~[undertow-core-2.2.10.Final.jar:2.2.10.Final]

        at io.undertow.server.protocol.http2.Http2UpgradeHandler.handleRequest(Http2UpgradeHandler.java:102) ~[undertow-core-2.2.10.Final.jar:2.2.10.Final]

        at io.undertow.server.Connectors.executeRootHandler(Connectors.java:387) [undertow-core-2.2.10.Final.jar:2.2.10.Final]

        at io.undertow.server.protocol.http.HttpReadListener.handleEventWithNoRunningRequest(HttpReadListener.java:256) [undertow-core-2.2.10.Final.jar:2.2.10.Final]

        at io.undertow.server.protocol.http.HttpReadListener.handleEvent(HttpReadListener.java:136) [undertow-core-2.2.10.Final.jar:2.2.10.Final]

        at io.undertow.server.protocol.http.HttpReadListener.handleEvent(HttpReadListener.java:59) [undertow-core-2.2.10.Final.jar:2.2.10.Final]

        at org.xnio.ChannelListeners.invokeChannelListener(ChannelListeners.java:92) [xnio-api-3.8.4.Final.jar:3.8.4.Final]

        at org.xnio.conduits.ReadReadyHandler$ChannelListenerHandler.readReady(ReadReadyHandler.java:66) [xnio-api-3.8.4.Final.jar:3.8.4.Final]

        at org.xnio.nio.NioSocketConduit.handleReady(NioSocketConduit.java:89) [xnio-nio-3.8.4.Final.jar:3.8.4.Final]

        at org.xnio.nio.WorkerThread.run(WorkerThread.java:591) [xnio-nio-3.8.4.Final.jar:3.8.4.Final]

This is the code snippet of RequestHandler.Java

@Override

public void handleRequest(HttpServerExchange serverContext) throws Exception {

this.serverContext = serverContext;

// line #212

serverContext.getRequestReceiver().receiveFullBytes(

        (httpServerExchange, reqBytes) -> {

          // IO Thread

          serverContext.dispatch(() -> {

            // Worker Thread

            new RequestProcessor().process(httpServerExchange, reqBytes, Metrics.INSTANCE);

          });

        },

        (httpServerExchange, exception) -> {

          //optional error handler

          // e.g. Remote peer closed connection before all data could be read

          alertLog.error("Exception: {}", exception.getMessage());

          new RequestProcessor().sendErrorResponse(httpServerExchange);

        });

}

[Ruslan Ibragimov]

It's valid exception, but problem (probably) that code dispatching request somewhere after receiveFullBytes, like:

handle(exchange)

  exchange.receiveFullBytes {}

  // execution continued, since receiveFullBytes is async, callback-based API

  exchange.dispatch() // <- illegal

So check logic of application, code shouldn't dispatch same exchange while reading from request channel

To view this discussion on the web visit https://groups.google.com/d/msgid/undertow-dev/44f7dcf0-9932-47f8-96da-efc3243af7b1n%40googlegroups.com.

For more options, visit https://groups.google.com/d/optout.

[Nishant Kumar]

I am calling dispatch() in the callback of  receiveFullBytes(). shouldn't the callback be called after reading the full request msg from the request channel?

serverContext.getRequestReceiver().receiveFullBytes(

        (httpServerExchange, reqBytes) -> {

          // IO Thread

          serverContext.dispatch(() -> {

            // Worker Thread

            new RequestProcessor().process(httpServerExchange, reqBytes, Metrics.INSTANCE);

          });

        },

        (httpServerExchange, exception) -> {

          //optional error handler

          // e.g. Remote peer closed connection before all data could be read

          alertLog.error("Exception: {}", exception.getMessage());

          new RequestProcessor().sendErrorResponse(httpServerExchange);

        });

}

If it's not correct, can you please provide the right way of using it? I just want to read the POST request body & header and start processing it after that. 

[Ruslan Ibragimov]

As I try to point previously, interesting part is code that happens after in a function where receiveFullBytes is called.

To view this discussion on the web visit https://groups.google.com/d/msgid/undertow-dev/3993440e-c5fc-4112-9847-9baedb33e818n%40googlegroups.com.

For more options, visit https://groups.google.com/d/optout.