WebSocket text event execute in same thread as running binary event and destroy Threadlocal

[Horcrux7]

Jetty version: 9.4.31

Java version: 11.09 but also with 8.0.x

OS type/version: any

Description

If the WebSocket MessageHandler is registered as:

public void onOpen( Session session, EndpointConfig config ) {
  session.addMessageHandler( new MessageHandler.Whole<InputStream>() {
    public void onMessage( InputStream message ) {
    ...
    }
  }
}

then is can occur that inside message.read() in the same thread a text message will be processed. In this case two MessageHandler are share the same thread local variables. On security reasons our MessageHandlers destroy all ThreadLocal variables on finish for the current thread. The result for the first binary MessageHandler is that there is no current user scope after some message.read().

This has work in version 9.4.20. It is a regression in a micro version or fatal behavior change. We was not able to find a documentation that forbids Threadlocal with WebSocket events handling. We expected that an intermediate event will process in another thread or later.

A workaround is to switch to MessageHandler.Whole<byte[]>. Then there is no message.read() call anymore. But it sounds fewer effective.

[joakime]

Your description has not been a supported feature before or now.
The fact that it worked in 9.4.20 is likely a total accident.

How it works ...
A new thread is created for dispatching the initial message for InputStream, once you receive the onMessage(InputStream) you are in your own thread (a requirement of the javax.websocket spec) until you return from that dispatch/method call.
All updates to the InputStream from the underlying websocket connection is asynchronous (and can/will use multiple threads over the InputStream lifespan)
So, don't spawn a new thread or anything from the onMessage(InputStream) to handle that InputStream.

Other notes about your question/comment ...
Also, know that if MessageHandler.Whole<byte[]> is even an option for you, then you are likely not using the API most effectively.
The MessageHandler.Whole<InputStream> API is best used for very large (not able to fit into memory) messages and long term messages (think voip, steaming, gaming, etc), as that API is very heavy handed with its Thread requirements.
If you have many messages a minute then you are better off (performance wise, and scalability wise) to use the MessageHandler.Whole<byte[]> or MessageHandler.Whole<ByteBuffer> APIs instead.
You can often handle many multiples of more users by switching to whole message delivery vs streaming delivery.

[Horcrux7]

A new thread is created for dispatching the initial message for InputStream

Yes, this is the feature that I expected and that Jetty not implement. Every event should be processed in its own thread from a pool.

So, don't spawn a new thread or anything from the onMessage(InputStream) to handle that InputStream.

Sounds you misunderstand me. I does not switch the thread. We save the scope of the current event in ThreadLocal variables. The problem is that Jetty handle a second event in the same thread inside the message.read() call. This override the ThreadLocal variables of the first event.

Many APIs from Java use ThreadLocal for states and communication. For example the ContextClassLoader. Your implementation break all this java API.

[joakime]

A new thread is created for dispatching the initial message for InputStream

Yes, this is the feature that I expected and that Jetty not implement. Every event should be processed in its own thread from a pool.

This is standard, and has been this way since Jetty 9.1.0 (when javax.websocket was first supported)
That thread is created and handed off to your application via the onMessage(InputStream) call.
That's the thread you start your InputStream.read() logic on.
As soon as you return from the onMessage(InputStream) that thread is no longer used.
ThreadLocal works fine on that Thread, it has, and still does.

The next onMessage(InputStream) call (which starts a NEW message) is dispatched to the next available thread from the thread pool. (luck of the draw if it's the same thread or a prior one).

So, don't spawn a new thread or anything from the onMessage(InputStream) to handle that InputStream.

Sounds you misunderstand me. I does not switch the thread. We save the scope of the current event in ThreadLocal variables. The problem is that Jetty handle a second event in the same thread inside the message.read() call. This override the ThreadLocal variables of the first event.

The inputstream.read() is on your thread, not Jetty.
Jetty has it's own thread that wakes up on network read advice to read from the network and populate an internal buffer that your thread's inputstream.read() waits on. (two threads are present at this point, and your thread still hasn't changed or been modified).
The only case where this isn't true is if you start a thread in onMessage(InputStream) to read from that InputStream. (now you have 3 threads in play).

Many APIs from Java use ThreadLocal for states and communication. For example the ContextClassLoader. Your implementation break all this java API.

Most of these have been systematically removed over the years as they are hostile to async development efforts.
Java 16 has almost all of them removed from the classlib now as result of the project loom efforts.

The javax.websocket spec does have some threading statements, but they are limited to how the endpoint is called.

See https://github.com/eclipse-ee4j/websocket-api/blob/master/spec/src/main/asciidoc/WebSocket.adoc#51-threading-considerations

This means that the Endpoint is called in a way to guarantee no two container threads are calling the endpoint at the same time.
The thing is, onMessage(InputStream) is a call to the endpoint.
The updates for the buffers within the InputStream are not an "endpoint call" and are not subject to the threading considerations.

[Horcrux7]

From your link:

If the implementation decides to process an incoming message in parts, it must ensure that the corresponding onMessage() calls are called sequentially, and do not interleave either with parts of the same message or with other messages [WSC-5.1.5].

This is the problem. Jetty interleave one onMessage() with another onMessage(). The stacktrace look like:

at onMessage()
....
at read()
...
at onMessage()

[joakime]

What's read() to you?

There's no method called "read()" in the javax.websocket APIs.

Note that you are using MessageHandler.Whole<InputStream> so that means the "must ensure that the corresponding onMessage() calls are called sequentially" refers only to the start of a NEW message, the actually "endpoint call" is the onMessage(InputStream).

If the entire message fits in a handful of websocket frames (again, you are abusing MessageHandler.Whole<InputStream> if you are using it for small messages) then you are notified of each NEW message, and that entire message is considered delivered to your application once the last websocket frame is (a frame with fin==true) is received. We are free to start reading a new message at that point, as we don't deliver the parts of the message as a javax.websocket API call to your application, as that's not how MessageHandler.Whole<InputStream> works.

So you can get ...

1. network event.
2. read() from network. (a single buffer was read, it has 3 websocket frames)
3. Frame[1](BIN:fin=false) - dispatched on new thread (A) to your application as onMessage(InputStream) - we called the application here.
4. Frame[2](CONTINUATION:fin=true) - sent to the InputStream buffers. - there is no call to the application here.
5. Frame[3](BIN:fin=true) - dispatched on new thread (B) to your application as onMessage(InputStream) - we called the application here.

From the point of view of both javax.websocket and Jetty both of those messages were delivered on separate threads.
Since they were so small, there's even a window of chance that your thread dispatching might see those in the wrong order. (standard java JVM thread dispatching behavior here)
Again, if your messages are small, using MessageHandler.Whole<InputStream> is the wrong API to use.

[joakime]

If the implementation decides to process an incoming message in parts ...

Also, this part of the spec refers to use of the MessageHandler.Partial<> APIs.
Which are delivered sequentially, as parts of a message, even without using thread dispatch.
This text does not refer the MessageHandler.Whole<> API in combination to the various java.io streams InputStream or Reader (the "whole" in these APIs are the dispatch of a new thread to the application notifying it that a message has started)

You basically have this ...

API	Message Delivery Mode	Thread Mode
MessageHandler.Whole<String>	whole Text message (at fin=true)	no dispatch
MessageHandler.Whole<ByteBuffer>	whole Binary message (at fin=true)	no dispatch
MessageHandler.Whole<byte[]>	whole Binary message (at fin=true)	no dispatch
MessageHandler.Partial<String>	each Text frame and subsequent Continuation frames	no dispatch
MessageHandler.Partial<ByteBuffer>	each Binary frame and subsequent Continuation frames	no dispatch
MessageHandler.Whole<InputStream>	on new Text frame	dispatched on new Text frame
MessageHandler.Whole<Reader>	on new Binary frame	dispatched on new Binary frame

Notice that both of the Whole streaming API choices don't deliver on WHOLE message received, but only on the start of a new TEXT or BINARY websocket frame. (a frame that triggered this call to onMessage(InputStream) can even be 0 length of payload data).
Since it's new thread dispatch, it is essentially equivalent to ...

  new Thread(new Runnable() {
      public void run(){
          registeredEndpointMessageHandler.onMessage(inputStream);
      } } ).start();

... for each new message, imagine what happens if many new messages arrive close to each other (even in the same read network buffer)?
This is important to understand about MessageHandler.Whole<> with InputStream, the dispatch is the start of the message, not the end of the message. If you have many small messages this behavior means that you are subject to standard threading behaviors with new threads being created and started, which might take just long enough that the order your application sees it could be out of order.

[lachlan-roberts]

The changes made in #4486 seem relevant to this issue. Since 9.4.27 we use suspend and resume to control the incoming frames into the MessageInputStream. This means we should now only read the next frame when someone has read the data from the previous frame from the InputStream, this should also mean that the onMessage() calls are done sequentially in the correct order. If you return from the onMessage() method the InputStream will be closed and remaining data will be discarded until the start of the next message which will result in a new thread to call the new onMessage.

The problem is that Jetty handle a second event in the same thread inside the message.read() call. This override the ThreadLocal variables of the first event.

I think this is the core of the issue. I think the call to resume() inside MessageInputStream.read() is causing our dispatched onMessage() thread to be the one to start parsing again.

But I don't understand how this would override your ThreadLocal variables of the first onMessage() event because even if it called another onMessage() during parsing it would also dispatch a new thread for it.

[lachlan-roberts]

@Horcrux7 can you give a full stacktrace of where the second onMessage() is called after you call InputStream.read() inside the first onMessage()?

[Horcrux7]

Here is a stacktrace where:

• com.inet.http.websocket.WebSocketEndpoint$1 is a - MessageHandler.Whole<String>
• com.inet.http.websocket.WebSocketEndpoint$2 is a - MessageHandler.Whole<InputStream>

java.lang.Exception
	at com.inet.http.websocket.WebSocketEndpoint$1.onMessage(WebSocketEndpoint.java:170)
	at com.inet.http.websocket.WebSocketEndpoint$1.onMessage(WebSocketEndpoint.java:1)
	at org.eclipse.jetty.websocket.jsr356.messages.TextWholeMessage.messageComplete(TextWholeMessage.java:59)
	at org.eclipse.jetty.websocket.jsr356.endpoints.JsrEndpointEventDriver.onTextFrame(JsrEndpointEventDriver.java:227)
	at org.eclipse.jetty.websocket.common.events.AbstractEventDriver.incomingFrame(AbstractEventDriver.java:150)
	at org.eclipse.jetty.websocket.common.WebSocketSession.incomingFrame(WebSocketSession.java:326)
	at org.eclipse.jetty.websocket.common.extensions.ExtensionStack.incomingFrame(ExtensionStack.java:202)
	at org.eclipse.jetty.websocket.common.Parser.notifyFrame(Parser.java:225)
	at org.eclipse.jetty.websocket.common.Parser.parseSingleFrame(Parser.java:259)
	at org.eclipse.jetty.websocket.common.io.AbstractWebSocketConnection.onFillable(AbstractWebSocketConnection.java:459)
	at org.eclipse.jetty.websocket.common.io.AbstractWebSocketConnection.resume(AbstractWebSocketConnection.java:524)
	at org.eclipse.jetty.websocket.common.message.MessageInputStream.read(MessageInputStream.java:237)
	at java.io.InputStream.read(InputStream.java:179)
	at java.io.InputStream.read(InputStream.java:101)
	at com.inet.pdfc.server.persistence.impl.file.CoreComparePersistence.a(SourceFile:888)
	at com.inet.pdfc.server.persistence.impl.file.CoreComparePersistence.saveDocument(SourceFile:780)
	at com.inet.pdfc.rpc.websocket.a.a(SourceFile:146)
	at com.inet.pdfc.rpc.websocket.events.g.a(SourceFile:29)
	at com.inet.pdfc.rpc.websocket.events.g.handle(SourceFile:17)
	at com.inet.http.websocket.WebSocketEventHandler.callHandler(WebSocketEventHandler.java:348)
	at com.inet.http.websocket.WebSocketEventHandler.handleEvent(WebSocketEventHandler.java:326)
	at com.inet.http.websocket.WebSocketEndpoint$2.onMessage(WebSocketEndpoint.java:245)
	at com.inet.http.websocket.WebSocketEndpoint$2.onMessage(WebSocketEndpoint.java:1)
	at org.eclipse.jetty.websocket.jsr356.endpoints.JsrEndpointEventDriver.lambda$onBinaryFrame$0(JsrEndpointEventDriver.java:98)
	at org.eclipse.jetty.util.thread.QueuedThreadPool.runJob(QueuedThreadPool.java:806)
	at org.eclipse.jetty.util.thread.QueuedThreadPool$Runner.run(QueuedThreadPool.java:938)
	at java.lang.Thread.run(Thread.java:748)

Message count and size: We use a large count of small messages (some KB) and a few large messages (<100MB) for file uploads. It is a mix of small and large messages.

[lachlan-roberts]

This is fixed by PR #5377 and will be in the 9.4.33 release.

[lachlan-roberts]

Reopening to verify fix in Jetty 10 as most of changes couldn't be merged.

[github-actions]

This issue has been automatically marked as stale because it has been a
full year without activity. It will be closed if no further activity occurs.
Thank you for your contributions.

[lachlan-roberts]

This is not an issue in 10.0.x as the demand mechanism is used to control when frames are requested for the MessageInputStream.