This is an automated email from the ASF dual-hosted git repository.
markt-asf pushed a commit to branch 11.0.x
in repository https://gitbox.apache.org/repos/asf/tomcat.git
The following commit(s) were added to refs/heads/11.0.x by this push:
new 7ab11d10de WebSocket write timeout improvements
7ab11d10de is described below
commit 7ab11d10de79a7a2226f41c8289871db69c6ca9c
Author: Mark Thomas <[email protected]>
AuthorDate: Thu Sep 3 10:51:02 2026 +0100
WebSocket write timeout improvements
---
.../server/WsRemoteEndpointImplServer.java | 123 ++++++++++++++-------
.../tomcat/websocket/server/WsWriteTimeout.java | 72 +++++++-----
webapps/docs/changelog.xml | 4 +
3 files changed, 127 insertions(+), 72 deletions(-)
diff --git
a/java/org/apache/tomcat/websocket/server/WsRemoteEndpointImplServer.java
b/java/org/apache/tomcat/websocket/server/WsRemoteEndpointImplServer.java
index 660085e5be..c0bd087140 100644
--- a/java/org/apache/tomcat/websocket/server/WsRemoteEndpointImplServer.java
+++ b/java/org/apache/tomcat/websocket/server/WsRemoteEndpointImplServer.java
@@ -53,6 +53,7 @@ public class WsRemoteEndpointImplServer extends
WsRemoteEndpointImplBase {
private final UpgradeInfo upgradeInfo;
private final WebConnection connection;
private final WsWriteTimeout wsWriteTimeout;
+ private final ReentrantLock writeCompletionLock = new ReentrantLock();
private volatile SendHandler handler = null;
private volatile ByteBuffer[] buffers = null;
@@ -163,12 +164,17 @@ public class WsRemoteEndpointImplServer extends
WsRemoteEndpointImplBase {
return;
}
} else {
- this.handler = handler;
- timeout = getSendTimeout();
- if (timeout > 0) {
- // Register with timeout thread
- timeoutExpiry = timeout + System.currentTimeMillis();
- wsWriteTimeout.register(this);
+ writeCompletionLock.lock();
+ try {
+ this.handler = handler;
+ timeout = getSendTimeout();
+ if (timeout > 0) {
+ // Register with timeout thread
+ timeoutExpiry = timeout + System.currentTimeMillis();
+ wsWriteTimeout.register(this);
+ }
+ } finally {
+ writeCompletionLock.unlock();
}
}
socketWrapper.write(block ? BlockingMode.BLOCK :
BlockingMode.SEMI_BLOCK, timeout, TimeUnit.MILLISECONDS,
@@ -183,7 +189,6 @@ public class WsRemoteEndpointImplServer extends
WsRemoteEndpointImplBase {
handler.onResult(new
SendResult(getSession()));
}
} else {
-
wsWriteTimeout.unregister(WsRemoteEndpointImplServer.this);
clearHandler(null, true);
}
}
@@ -194,7 +199,6 @@ public class WsRemoteEndpointImplServer extends
WsRemoteEndpointImplBase {
SendResult sr = new SendResult(getSession(),
exc);
handler.onResult(sr);
} else {
-
wsWriteTimeout.unregister(WsRemoteEndpointImplServer.this);
clearHandler(exc, true);
close();
}
@@ -202,8 +206,13 @@ public class WsRemoteEndpointImplServer extends
WsRemoteEndpointImplBase {
}, buffers);
} else {
if (blockingWriteTimeoutExpiry == -1) {
- this.handler = handler;
- this.buffers = buffers;
+ writeCompletionLock.lock();
+ try {
+ this.handler = handler;
+ this.buffers = buffers;
+ } finally {
+ writeCompletionLock.unlock();
+ }
// This is definitely the same thread that triggered the write
so a
// dispatch will be required.
onWritePossible(true);
@@ -274,25 +283,29 @@ public class WsRemoteEndpointImplServer extends
WsRemoteEndpointImplBase {
socketWrapper.flush(false);
complete = socketWrapper.isReadyForWrite();
if (complete) {
- wsWriteTimeout.unregister(this);
clearHandler(null, useDispatch);
}
break;
}
}
} catch (IOException | IllegalStateException e) {
- wsWriteTimeout.unregister(this);
clearHandler(e, useDispatch);
close();
}
if (!complete) {
- // Async write is in progress
- long timeout = getSendTimeout();
- if (timeout > 0) {
- // Register with timeout thread
- timeoutExpiry = timeout + System.currentTimeMillis();
- wsWriteTimeout.register(this);
+ writeCompletionLock.lock();
+ try {
+ // The write may have completed or timed out while obtaining
the lock
+ if (handler != null) {
+ long timeout = getSendTimeout();
+ if (timeout > 0) {
+ timeoutExpiry = timeout + System.currentTimeMillis();
+ wsWriteTimeout.register(this);
+ }
+ }
+ } finally {
+ writeCompletionLock.unlock();
}
}
}
@@ -334,13 +347,22 @@ public class WsRemoteEndpointImplServer extends
WsRemoteEndpointImplBase {
*/
/**
* Handles a write timeout event.
- * @param useDispatch whether to use a dispatch for callback
+ *
+ * @param useDispatch Whether to use a dispatch for callback
+ * @param now The time to which the timeout should be compared
*/
- protected void onTimeout(boolean useDispatch) {
- if (handler != null) {
- clearHandler(new SocketTimeoutException(), useDispatch);
+ protected void onTimeout(boolean useDispatch, long now) {
+ writeCompletionLock.lock();
+ try {
+ // Re-check timeout in case of concurrent completion and new write
+ if (handler != null && getTimeoutExpiry() < now) {
+ wsWriteTimeout.unregister(this);
+ clearHandlerInternal(new SocketTimeoutException(),
useDispatch);
+ close();
+ }
+ } finally {
+ writeCompletionLock.unlock();
}
- close();
}
@@ -357,6 +379,23 @@ public class WsRemoteEndpointImplServer extends
WsRemoteEndpointImplBase {
* the requirements of {@link
jakarta.websocket.RemoteEndpoint.Async}
*/
void clearHandler(Throwable t, boolean useDispatch) {
+ writeCompletionLock.lock();
+ try {
+ if (handler != null) {
+ /*
+ * Unregister before invoking the callback since the callback
may synchronously start and register the
+ * next write.
+ */
+ wsWriteTimeout.unregister(this);
+ clearHandlerInternal(t, useDispatch);
+ }
+ } finally {
+ writeCompletionLock.unlock();
+ }
+ }
+
+
+ private void clearHandlerInternal(Throwable t, boolean useDispatch) {
// Setting the result marks this (partial) message as
// complete which means the next one may be sent which
// could update the value of the handler. Therefore, keep a
@@ -365,27 +404,25 @@ public class WsRemoteEndpointImplServer extends
WsRemoteEndpointImplBase {
SendHandler sh = handler;
handler = null;
buffers = null;
- if (sh != null) {
- if (useDispatch) {
- OnResultRunnable r = new OnResultRunnable(getSession(), sh, t);
- try {
- socketWrapper.execute(r);
- } catch (RejectedExecutionException ree) {
- // Can't use the executor so call the runnable directly.
- // This may not be strictly specification compliant in all
- // cases but during shutdown only close messages are going
- // to be sent so there should not be the issue of nested
- // calls leading to stack overflow as described in bug
- // 55715. The issues with nested calls was the reason for
- // the separate thread requirement in the specification.
- r.run();
- }
+ if (useDispatch) {
+ OnResultRunnable r = new OnResultRunnable(getSession(), sh, t);
+ try {
+ socketWrapper.execute(r);
+ } catch (RejectedExecutionException ree) {
+ // Can't use the executor so call the runnable directly.
+ // This may not be strictly specification compliant in all
+ // cases but during shutdown only close messages are going
+ // to be sent so there should not be the issue of nested
+ // calls leading to stack overflow as described in bug
+ // 55715. The issues with nested calls was the reason for
+ // the separate thread requirement in the specification.
+ r.run();
+ }
+ } else {
+ if (t == null) {
+ sh.onResult(new SendResult(getSession()));
} else {
- if (t == null) {
- sh.onResult(new SendResult(getSession()));
- } else {
- sh.onResult(new SendResult(getSession(), t));
- }
+ sh.onResult(new SendResult(getSession(), t));
}
}
}
diff --git a/java/org/apache/tomcat/websocket/server/WsWriteTimeout.java
b/java/org/apache/tomcat/websocket/server/WsWriteTimeout.java
index 7877f6acb3..db6d9963a7 100644
--- a/java/org/apache/tomcat/websocket/server/WsWriteTimeout.java
+++ b/java/org/apache/tomcat/websocket/server/WsWriteTimeout.java
@@ -16,10 +16,9 @@
*/
package org.apache.tomcat.websocket.server;
-import java.util.Comparator;
+import java.util.HashSet;
import java.util.Set;
-import java.util.concurrent.ConcurrentSkipListSet;
-import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.locks.ReentrantLock;
import org.apache.tomcat.websocket.BackgroundProcess;
import org.apache.tomcat.websocket.BackgroundProcessManager;
@@ -37,13 +36,11 @@ public class WsWriteTimeout implements BackgroundProcess {
public WsWriteTimeout() {
}
- /**
- * Note: The comparator imposes orderings that are inconsistent with equals
- */
- private final Set<WsRemoteEndpointImplServer> endpoints =
- new
ConcurrentSkipListSet<>(Comparator.comparingLong(WsRemoteEndpointImplServer::getTimeoutExpiry));
- private final AtomicInteger count = new AtomicInteger(0);
- private int backgroundProcessCount = 0;
+ private final Set<WsRemoteEndpointImplServer> endpoints = new HashSet<>();
+ private final ReentrantLock backgroundProcessLock = new ReentrantLock();
+ private int count = 0;
+
+ private volatile int backgroundProcessCount = 0;
private volatile int processPeriod = 1;
@Override
@@ -54,17 +51,24 @@ public class WsWriteTimeout implements BackgroundProcess {
if (backgroundProcessCount >= processPeriod) {
backgroundProcessCount = 0;
+ Set<WsRemoteEndpointImplServer> endpointsForTimeoutCheck = new
HashSet<>();
+ backgroundProcessLock.lock();
+ try {
+ endpointsForTimeoutCheck.addAll(endpoints);
+ } finally {
+ backgroundProcessLock.unlock();
+ }
+
+ /*
+ * The timeout expiry is not fixed. A completed write or a new
write can change the expiry at any point.
+ * Since it is not possible to order the endpoints by expiry time
and process the endpoints in expiry time
+ * order, every endpoint is checked.
+ */
long now = System.currentTimeMillis();
- for (WsRemoteEndpointImplServer endpoint : endpoints) {
+ for (WsRemoteEndpointImplServer endpoint :
endpointsForTimeoutCheck) {
if (endpoint.getTimeoutExpiry() < now) {
- // Background thread, not the thread that triggered the
- // write so no need to use a dispatch
- endpoint.onTimeout(false);
- } else {
- // Endpoints are ordered by timeout expiry so if this point
- // is reached there is no need to check the remaining
- // endpoints
- break;
+ // Background thread, not the thread that triggered the
write, so no need to use a dispatch
+ endpoint.onTimeout(false, now);
}
}
}
@@ -92,12 +96,17 @@ public class WsWriteTimeout implements BackgroundProcess {
* @param endpoint the endpoint to register
*/
public void register(WsRemoteEndpointImplServer endpoint) {
- boolean result = endpoints.add(endpoint);
- if (result) {
- int newCount = count.incrementAndGet();
- if (newCount == 1) {
- BackgroundProcessManager.getInstance().register(this);
+ backgroundProcessLock.lock();
+ try {
+ boolean result = endpoints.add(endpoint);
+ if (result) {
+ if (count == 0) {
+ BackgroundProcessManager.getInstance().register(this);
+ }
+ count++;
}
+ } finally {
+ backgroundProcessLock.unlock();
}
}
@@ -108,12 +117,17 @@ public class WsWriteTimeout implements BackgroundProcess {
* @param endpoint the endpoint to unregister
*/
public void unregister(WsRemoteEndpointImplServer endpoint) {
- boolean result = endpoints.remove(endpoint);
- if (result) {
- int newCount = count.decrementAndGet();
- if (newCount == 0) {
- BackgroundProcessManager.getInstance().unregister(this);
+ backgroundProcessLock.lock();
+ try {
+ boolean result = endpoints.remove(endpoint);
+ if (result) {
+ count--;
+ if (count == 0) {
+ BackgroundProcessManager.getInstance().unregister(this);
+ }
}
+ } finally {
+ backgroundProcessLock.unlock();
}
}
}
diff --git a/webapps/docs/changelog.xml b/webapps/docs/changelog.xml
index eb2eb8aced..d75def4f3f 100644
--- a/webapps/docs/changelog.xml
+++ b/webapps/docs/changelog.xml
@@ -272,6 +272,10 @@
<fix>
Improve robustness of client handshakes. (remm)
</fix>
+ <fix>
+ Ensure that WebSocket non-blocking write timeouts are not lost if two
+ writes have the same timeout. (markt)
+ </fix>
</changelog>
</subsection>
<subsection name="Web applications">
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
