4:01 p.m.
Author: remy.maucherat(a)jboss.com
Date: 2013-09-27 10:01:51 -0400 (Fri, 27 Sep 2013)
New Revision: 2265
Modified:
branches/7.4.x/src/main/java/org/apache/coyote/http11/Http11AprProcessor.java
branches/7.4.x/src/main/java/org/apache/coyote/http11/InternalAprOutputBuffer.java
branches/7.4.x/src/main/java/org/apache/tomcat/websocket/FutureToSendHandler.java
branches/7.4.x/src/main/java/org/apache/tomcat/websocket/WsSession.java
branches/7.4.x/src/main/java/org/apache/tomcat/websocket/server/WsRemoteEndpointImplServer.java
Log:
- Add flushing at the end of a message write (Tomcat uses unbuffered stream and direct
socket writes)
- Try using autoblocking for the blocking get() on the future
Modified: branches/7.4.x/src/main/java/org/apache/coyote/http11/Http11AprProcessor.java
===================================================================
---
branches/7.4.x/src/main/java/org/apache/coyote/http11/Http11AprProcessor.java 2013-09-24
17:35:22 UTC (rev 2264)
+++
branches/7.4.x/src/main/java/org/apache/coyote/http11/Http11AprProcessor.java 2013-09-27
14:01:51 UTC (rev 2265)
@@ -96,6 +96,8 @@
initializeFilters();
+ Http11AbstractProcessor.containerThread.set(Boolean.FALSE);
+
// Cause loading of HexUtils
int foo = HexUtils.DEC[0];
@@ -109,12 +111,6 @@
/**
- * Thread local marker.
- */
- public static ThreadLocal<Boolean> containerThread = new
ThreadLocal<Boolean>();
-
-
- /**
* Associated adapter.
*/
protected Adapter adapter = null;
@@ -762,7 +758,7 @@
// Set error flag right away
error = true;
}
- containerThread.set(Boolean.TRUE);
+ Http11AbstractProcessor.containerThread.set(Boolean.TRUE);
rp.setStage(org.apache.coyote.Constants.STAGE_SERVICE);
error = !adapter.event(request, response, status);
} catch (InterruptedIOException e) {
Modified:
branches/7.4.x/src/main/java/org/apache/coyote/http11/InternalAprOutputBuffer.java
===================================================================
---
branches/7.4.x/src/main/java/org/apache/coyote/http11/InternalAprOutputBuffer.java 2013-09-24
17:35:22 UTC (rev 2264)
+++
branches/7.4.x/src/main/java/org/apache/coyote/http11/InternalAprOutputBuffer.java 2013-09-27
14:01:51 UTC (rev 2265)
@@ -538,7 +538,7 @@
// If non blocking (event) and there are leftover bytes,
// and lastWrite was 0 -> error
- if (leftover.getLength() > 0 &&
!(Http11AprProcessor.containerThread.get() == Boolean.TRUE)) {
+ if (leftover.getLength() > 0 &&
!(Http11AbstractProcessor.containerThread.get() == Boolean.TRUE)) {
throw new IOException(MESSAGES.invalidBacklog());
}
@@ -789,7 +789,7 @@
// - If the call is asynchronous, throw an exception
// - If the call is synchronous, make regular blocking writes to flush the data
if (leftover.getLength() > 0) {
- if (Http11AprProcessor.containerThread.get() == Boolean.TRUE) {
+ if (Http11AbstractProcessor.containerThread.get() == Boolean.TRUE) {
Socket.timeoutSet(socket, endpoint.getSoTimeout() * 1000);
// Send leftover bytes
res = Socket.send(socket, leftover.getBuffer(), leftover.getOffset(),
leftover.getEnd());
@@ -820,7 +820,7 @@
}
}
if (pos < end) {
- if (response.getFlushLeftovers() &&
(Http11AprProcessor.containerThread.get() == Boolean.TRUE)) {
+ if (response.getFlushLeftovers() &&
(Http11AbstractProcessor.containerThread.get() == Boolean.TRUE)) {
// Switch to blocking mode and write the data
Socket.timeoutSet(socket, endpoint.getSoTimeout() * 1000);
res = Socket.sendbb(socket, 0, end);
Modified:
branches/7.4.x/src/main/java/org/apache/tomcat/websocket/FutureToSendHandler.java
===================================================================
---
branches/7.4.x/src/main/java/org/apache/tomcat/websocket/FutureToSendHandler.java 2013-09-24
17:35:22 UTC (rev 2264)
+++
branches/7.4.x/src/main/java/org/apache/tomcat/websocket/FutureToSendHandler.java 2013-09-27
14:01:51 UTC (rev 2265)
@@ -16,6 +16,7 @@
*/
package org.apache.tomcat.websocket;
+import java.io.IOException;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
@@ -25,6 +26,8 @@
import javax.websocket.SendHandler;
import javax.websocket.SendResult;
+import org.apache.coyote.http11.Http11AbstractProcessor;
+
/**
* Converts a Future to a SendHandler.
*/
@@ -71,15 +74,26 @@
@Override
public Void get() throws InterruptedException,
ExecutionException {
- try {
- wsSession.registerFuture(this);
- latch.await();
- } finally {
- wsSession.unregisterFuture(this);
+ // If inside a container thread, must use an autoblocking flush as the write
+ // event will never come to the Servlet layer until the container thread returns
+ if (Http11AbstractProcessor.containerThread.get() == Boolean.TRUE) {
+ // FIXME: this uses the IO timeout rather than no timeout as per the API
contract
+ try {
+ wsSession.forceFlush();
+ } catch (IOException e) {
+ throw new ExecutionException(e);
+ }
+ } else {
+ try {
+ wsSession.registerFuture(this);
+ latch.await();
+ } finally {
+ wsSession.unregisterFuture(this);
+ }
+ if (result.getException() != null) {
+ throw new ExecutionException(result.getException());
+ }
}
- if (result.getException() != null) {
- throw new ExecutionException(result.getException());
- }
return null;
}
@@ -87,20 +101,31 @@
public Void get(long timeout, TimeUnit unit)
throws InterruptedException, ExecutionException,
TimeoutException {
- boolean retval = false;
- try {
- wsSession.registerFuture(this);
- retval = latch.await(timeout, unit);
- } finally {
- wsSession.unregisterFuture(this);
+ // If inside a container thread, must use an autoblocking flush as the write
+ // event will never come to the Servlet layer until the container thread returns
+ if (Http11AbstractProcessor.containerThread.get() == Boolean.TRUE) {
+ // FIXME: this uses the IO timeout rather than the timeout specified by the
user
+ try {
+ wsSession.forceFlush();
+ } catch (IOException e) {
+ throw new ExecutionException(e);
+ }
+ } else {
+ boolean retval = false;
+ try {
+ wsSession.registerFuture(this);
+ retval = latch.await(timeout, unit);
+ } finally {
+ wsSession.unregisterFuture(this);
+ }
+ if (retval == false) {
+ throw new TimeoutException();
+ }
+ if (result.getException() != null) {
+ throw new ExecutionException(result.getException());
+ }
}
- if (retval == false) {
- throw new TimeoutException();
- }
- if (result.getException() != null) {
- throw new ExecutionException(result.getException());
- }
return null;
}
}
Modified: branches/7.4.x/src/main/java/org/apache/tomcat/websocket/WsSession.java
===================================================================
--- branches/7.4.x/src/main/java/org/apache/tomcat/websocket/WsSession.java 2013-09-24
17:35:22 UTC (rev 2264)
+++ branches/7.4.x/src/main/java/org/apache/tomcat/websocket/WsSession.java 2013-09-27
14:01:51 UTC (rev 2265)
@@ -436,6 +436,12 @@
}
}
+ /**
+ * Force an autoblocking flush.
+ */
+ public void forceFlush() throws IOException {
+ wsRemoteEndpoint.flushBatch();
+ }
private void fireEndpointOnClose(CloseReason closeReason) {
Modified:
branches/7.4.x/src/main/java/org/apache/tomcat/websocket/server/WsRemoteEndpointImplServer.java
===================================================================
---
branches/7.4.x/src/main/java/org/apache/tomcat/websocket/server/WsRemoteEndpointImplServer.java 2013-09-24
17:35:22 UTC (rev 2264)
+++
branches/7.4.x/src/main/java/org/apache/tomcat/websocket/server/WsRemoteEndpointImplServer.java 2013-09-27
14:01:51 UTC (rev 2265)
@@ -82,9 +82,8 @@
}
if (complete) {
wsWriteTimeout.unregister(this);
- // ADD: Explicit flush
+ // Explicit flush for compatibility with buffered streams
sos.flush();
- // END ADD
if (close) {
close();
}