You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@tomcat.apache.org by re...@apache.org on 2019/11/21 16:30:06 UTC
[tomcat] branch 8.5.x updated: Port current version of the async API
This is an automated email from the ASF dual-hosted git repository.
remm pushed a commit to branch 8.5.x
in repository https://gitbox.apache.org/repos/asf/tomcat.git
The following commit(s) were added to refs/heads/8.5.x by this push:
new f627f8e Port current version of the async API
f627f8e is described below
commit f627f8e9ea2761960af8248b86d77515b4478f42
Author: remm <re...@apache.org>
AuthorDate: Thu Nov 21 17:29:27 2019 +0100
Port current version of the async API
Add async IO to APR and NIO.
Add vectoring to the NioChannel.
NIO2 code is pulled up in the superclass.
This reduces the differences between the connectors from 9.0 and 8.5.
I will continue backporting other now tested refactorings to reduce the
differences.
---
.../catalina/security/SecurityClassLoad.java | 6 +-
java/org/apache/tomcat/util/net/AprEndpoint.java | 126 +++++++-
java/org/apache/tomcat/util/net/Nio2Endpoint.java | 311 +++++--------------
java/org/apache/tomcat/util/net/NioChannel.java | 27 +-
java/org/apache/tomcat/util/net/NioEndpoint.java | 109 +++++++
.../apache/tomcat/util/net/SecureNioChannel.java | 145 +++++++++
.../apache/tomcat/util/net/SocketWrapperBase.java | 335 ++++++++++++++++++++-
webapps/docs/changelog.xml | 6 +
8 files changed, 820 insertions(+), 245 deletions(-)
diff --git a/java/org/apache/catalina/security/SecurityClassLoad.java b/java/org/apache/catalina/security/SecurityClassLoad.java
index b6e2be7..5afa0a0 100644
--- a/java/org/apache/catalina/security/SecurityClassLoad.java
+++ b/java/org/apache/catalina/security/SecurityClassLoad.java
@@ -183,12 +183,14 @@ public final class SecurityClassLoad {
loader.loadClass(basePackage + "util.net.NioBlockingSelector$BlockPoller$RunnableAdd");
loader.loadClass(basePackage + "util.net.NioBlockingSelector$BlockPoller$RunnableCancel");
loader.loadClass(basePackage + "util.net.NioBlockingSelector$BlockPoller$RunnableRemove");
- loader.loadClass(basePackage + "util.net.Nio2Endpoint$Nio2SocketWrapper$OperationState");
- loader.loadClass(basePackage + "util.net.Nio2Endpoint$Nio2SocketWrapper$VectoredIOCompletionHandler");
+ loader.loadClass(basePackage + "util.net.AprEndpoint$AprSocketWrapper$AprOperationState");
+ loader.loadClass(basePackage + "util.net.NioEndpoint$NioSocketWrapper$NioOperationState");
+ loader.loadClass(basePackage + "util.net.Nio2Endpoint$Nio2SocketWrapper$Nio2OperationState");
loader.loadClass(basePackage + "util.net.SocketWrapperBase$BlockingMode");
loader.loadClass(basePackage + "util.net.SocketWrapperBase$CompletionCheck");
loader.loadClass(basePackage + "util.net.SocketWrapperBase$CompletionHandlerCall");
loader.loadClass(basePackage + "util.net.SocketWrapperBase$CompletionState");
+ loader.loadClass(basePackage + "util.net.SocketWrapperBase$VectoredIOCompletionHandler");
// security
loader.loadClass(basePackage + "util.security.PrivilegedGetTccl");
loader.loadClass(basePackage + "util.security.PrivilegedSetTccl");
diff --git a/java/org/apache/tomcat/util/net/AprEndpoint.java b/java/org/apache/tomcat/util/net/AprEndpoint.java
index 460b075..d325b3c 100644
--- a/java/org/apache/tomcat/util/net/AprEndpoint.java
+++ b/java/org/apache/tomcat/util/net/AprEndpoint.java
@@ -21,6 +21,7 @@ import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.SocketTimeoutException;
import java.nio.ByteBuffer;
+import java.nio.channels.CompletionHandler;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.HashMap;
@@ -28,6 +29,8 @@ import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.RejectedExecutionException;
+import java.util.concurrent.Semaphore;
+import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantReadWriteLock.WriteLock;
@@ -112,6 +115,11 @@ public class AprEndpoint extends AbstractEndpoint<Long> implements SNICallBack {
// ------------------------------------------------------------ Constructor
public AprEndpoint() {
+ // Asynchronous IO has significantly lower performance with APR:
+ // - no IO vectoring
+ // - mandatory use of direct buffers forces output buffering
+ // - needs extra output flushes due to buffering
+ setUseAsyncIO(false);
// Need to override the default for maxConnections to align it with what
// was pollerSize (before the two were merged)
setMaxConnections(8 * 1024);
@@ -1178,7 +1186,7 @@ public class AprEndpoint extends AbstractEndpoint<Long> implements SNICallBack {
while (info != null) {
// Make sure we aren't trying add the socket as well as close it
addList.remove(info.socket);
- // Make sure the socket isn't in the poller before we close it
+ // Make sure the socket isn't in the poller before we close it
removeFromPoller(info.socket);
// Poller isn't running at this point so use destroySocket()
// directly
@@ -1189,7 +1197,7 @@ public class AprEndpoint extends AbstractEndpoint<Long> implements SNICallBack {
// Close all sockets in the add queue
info = addList.get();
while (info != null) {
- // Make sure the socket isn't in the poller before we close it
+ // Make sure the socket isn't in the poller before we close it
removeFromPoller(info.socket);
// Poller isn't running at this point so use destroySocket()
// directly
@@ -2049,7 +2057,7 @@ public class AprEndpoint extends AbstractEndpoint<Long> implements SNICallBack {
* This class is the equivalent of the Worker, but will simply use in an
* external Executor thread pool.
*/
- protected class SocketProcessor extends SocketProcessorBase<Long> {
+ protected class SocketProcessor extends SocketProcessorBase<Long> {
public SocketProcessor(SocketWrapperBase<Long> socketWrapper, SocketEvent event) {
super(socketWrapper, event);
@@ -2660,5 +2668,117 @@ public class AprEndpoint extends AbstractEndpoint<Long> implements SNICallBack {
}
}
}
+
+ @Override
+ protected <A> OperationState<A> newOperationState(boolean read,
+ ByteBuffer[] buffers, int offset, int length,
+ BlockingMode block, long timeout, TimeUnit unit, A attachment,
+ CompletionCheck check, CompletionHandler<Long, ? super A> handler,
+ Semaphore semaphore, VectoredIOCompletionHandler<A> completion) {
+ return new AprOperationState<>(read, buffers, offset, length, block,
+ timeout, unit, attachment, check, handler, semaphore, completion);
+ }
+
+ private class AprOperationState<A> extends OperationState<A> {
+ private volatile boolean inline = true;
+ private volatile long flushBytes = 0;
+ private AprOperationState(boolean read, ByteBuffer[] buffers, int offset, int length,
+ BlockingMode block, long timeout, TimeUnit unit, A attachment, CompletionCheck check,
+ CompletionHandler<Long, ? super A> handler, Semaphore semaphore,
+ VectoredIOCompletionHandler<A> completion) {
+ super(read, buffers, offset, length, block,
+ timeout, unit, attachment, check, handler, semaphore, completion);
+ }
+
+ @Override
+ protected boolean isInline() {
+ return inline;
+ }
+
+ @Override
+ public void run() {
+ // Perform the IO operation
+ // Called from the poller to continue the IO operation
+ long nBytes = 0;
+ if (getError() == null) {
+ try {
+ synchronized (this) {
+ if (!completionDone) {
+ // This filters out same notification until processing
+ // of the current one is done
+ if (log.isDebugEnabled()) {
+ log.debug("Skip concurrent " + (read ? "read" : "write") + " notification");
+ }
+ return;
+ }
+ // Find the buffer on which the operation will be performed (no vectoring with APR)
+ ByteBuffer buffer = null;
+ for (int i = 0; i < length; i++) {
+ if (buffers[i + offset].hasRemaining()) {
+ buffer = buffers[i + offset];
+ break;
+ }
+ }
+ if (buffer == null && flushBytes == 0) {
+ // Nothing to do
+ return;
+ }
+ if (read) {
+ nBytes = read(false, buffer);
+ } else {
+ if (!flush(block == BlockingMode.BLOCK)) {
+ if (flushBytes > 0) {
+ // Flushing was done, continue processing
+ nBytes = flushBytes;
+ flushBytes = 0;
+ } else {
+ @SuppressWarnings("null") // Not possible
+ int remaining = buffer.remaining();
+ write(block == BlockingMode.BLOCK, buffer);
+ nBytes = remaining - buffer.remaining();
+ if (nBytes > 0 && flush(block == BlockingMode.BLOCK)) {
+ // We have to flush and it's incomplete, save the bytes written until done
+ inline = false;
+ registerWriteInterest();
+ flushBytes = nBytes;
+ return;
+ }
+ }
+ } else {
+ // Continue flushing
+ inline = false;
+ registerWriteInterest();
+ return;
+ }
+ }
+ if (nBytes != 0) {
+ completionDone = false;
+ }
+ }
+ } catch (IOException e) {
+ setError(e);
+ }
+ }
+ if (nBytes > 0) {
+ // The bytes processed are only updated in the completion handler
+ completion.completed(Long.valueOf(nBytes), this);
+ } else if (nBytes < 0 || getError() != null) {
+ IOException error = getError();
+ if (error == null) {
+ error = new EOFException();
+ }
+ completion.failed(error, this);
+ } else {
+ // As soon as the operation uses the poller, it is no longer inline
+ inline = false;
+ if (read) {
+ registerReadInterest();
+ } else {
+ registerWriteInterest();
+ }
+ }
+ }
+ }
+
}
}
diff --git a/java/org/apache/tomcat/util/net/Nio2Endpoint.java b/java/org/apache/tomcat/util/net/Nio2Endpoint.java
index 030236e..ddf9d5e 100644
--- a/java/org/apache/tomcat/util/net/Nio2Endpoint.java
+++ b/java/org/apache/tomcat/util/net/Nio2Endpoint.java
@@ -29,10 +29,7 @@ import java.nio.channels.AsynchronousServerSocketChannel;
import java.nio.channels.AsynchronousSocketChannel;
import java.nio.channels.CompletionHandler;
import java.nio.channels.FileChannel;
-import java.nio.channels.InterruptedByTimeoutException;
import java.nio.channels.NetworkChannel;
-import java.nio.channels.ReadPendingException;
-import java.nio.channels.WritePendingException;
import java.nio.file.StandardOpenOption;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
@@ -933,254 +930,104 @@ public class Nio2Endpoint extends AbstractJsseEndpoint<Nio2Channel> {
return getEndpoint().getUseAsyncIO();
}
- /**
- * Internal state tracker for scatter/gather operations.
- */
- private static class OperationState<A> {
- private final boolean read;
- private final ByteBuffer[] buffers;
- private final int offset;
- private final int length;
- private final A attachment;
- private final long timeout;
- private final TimeUnit unit;
- private final BlockingMode block;
- private final CompletionCheck check;
- private final CompletionHandler<Long, ? super A> handler;
- private final Semaphore semaphore;
- private OperationState(boolean read, ByteBuffer[] buffers, int offset, int length,
- BlockingMode block, long timeout, TimeUnit unit, A attachment,
- CompletionCheck check, CompletionHandler<Long, ? super A> handler,
- Semaphore semaphore) {
- this.read = read;
- this.buffers = buffers;
- this.offset = offset;
- this.length = length;
- this.block = block;
- this.timeout = timeout;
- this.unit = unit;
- this.attachment = attachment;
- this.check = check;
- this.handler = handler;
- this.semaphore = semaphore;
- }
- private volatile long nBytes = 0;
- private volatile CompletionState state = CompletionState.PENDING;
+ @Override
+ public boolean needSemaphores() {
+ return true;
}
@Override
- public <A> CompletionState read(ByteBuffer[] dsts, int offset, int length,
- BlockingMode block, long timeout, TimeUnit unit, A attachment,
- CompletionCheck check, CompletionHandler<Long, ? super A> handler) {
- IOException ioe = getError();
- if (ioe != null) {
- handler.failed(ioe, attachment);
- return CompletionState.ERROR;
- }
- if (timeout == -1) {
- timeout = toTimeout(getReadTimeout());
- }
- // Disable any regular read notifications caused by registerReadInterest
- readNotify = true;
- if (block == BlockingMode.BLOCK || block == BlockingMode.SEMI_BLOCK) {
- try {
- if (!readPending.tryAcquire(timeout, unit)) {
- handler.failed(new SocketTimeoutException(), attachment);
- return CompletionState.ERROR;
- }
- } catch (InterruptedException e) {
- handler.failed(e, attachment);
- return CompletionState.ERROR;
- }
- } else {
- if (!readPending.tryAcquire()) {
- if (block == BlockingMode.NON_BLOCK) {
- return CompletionState.NOT_DONE;
- } else {
- handler.failed(new ReadPendingException(), attachment);
- return CompletionState.ERROR;
- }
- }
- }
- OperationState<A> state = new OperationState<>(true, dsts, offset, length, block,
- timeout, unit, attachment, check, handler, readPending);
- VectoredIOCompletionHandler<A> completion = new VectoredIOCompletionHandler<>();
- Nio2Endpoint.startInline();
- long nBytes = 0;
- if (!socketBufferHandler.isReadBufferEmpty()) {
- // There is still data inside the main read buffer, use it to fill out the destination buffers
- synchronized (readCompletionHandler) {
- // Note: It is not necessary to put this code in the completion handler
- socketBufferHandler.configureReadBufferForRead();
- for (int i = 0; i < length && !socketBufferHandler.isReadBufferEmpty(); i++) {
- nBytes += transfer(socketBufferHandler.getReadBuffer(), dsts[offset + i]);
- }
- }
- if (nBytes > 0) {
- completion.completed(Long.valueOf(nBytes), state);
- }
- }
- if (nBytes == 0) {
- getSocket().read(dsts, offset, length, timeout, unit, state, completion);
- }
- Nio2Endpoint.endInline();
- if (block == BlockingMode.BLOCK) {
- synchronized (state) {
- if (state.state == CompletionState.PENDING) {
- try {
- state.wait(unit.toMillis(timeout));
- if (state.state == CompletionState.PENDING) {
- return CompletionState.ERROR;
- }
- } catch (InterruptedException e) {
- handler.failed(new SocketTimeoutException(), attachment);
- return CompletionState.ERROR;
- }
- }
- }
- }
- return state.state;
+ public boolean hasPerOperationTimeout() {
+ return true;
}
@Override
- public <A> CompletionState write(ByteBuffer[] srcs, int offset, int length,
+ protected <A> OperationState<A> newOperationState(boolean read,
+ ByteBuffer[] buffers, int offset, int length,
BlockingMode block, long timeout, TimeUnit unit, A attachment,
- CompletionCheck check, CompletionHandler<Long, ? super A> handler) {
- IOException ioe = getError();
- if (ioe != null) {
- handler.failed(ioe, attachment);
- return CompletionState.ERROR;
- }
- if (timeout == -1) {
- timeout = toTimeout(getWriteTimeout());
- }
- // Disable any regular write notifications caused by registerWriteInterest
- writeNotify = true;
- if (block == BlockingMode.BLOCK || block == BlockingMode.SEMI_BLOCK) {
- try {
- if (!writePending.tryAcquire(timeout, unit)) {
- handler.failed(new SocketTimeoutException(), attachment);
- return CompletionState.ERROR;
- }
- } catch (InterruptedException e) {
- handler.failed(e, attachment);
- return CompletionState.ERROR;
- }
- } else {
- if (!writePending.tryAcquire()) {
- if (block == BlockingMode.NON_BLOCK) {
- return CompletionState.NOT_DONE;
- } else {
- handler.failed(new WritePendingException(), attachment);
- return CompletionState.ERROR;
- }
- }
+ CompletionCheck check, CompletionHandler<Long, ? super A> handler,
+ Semaphore semaphore, VectoredIOCompletionHandler<A> completion) {
+ return new Nio2OperationState<>(read, buffers, offset, length, block,
+ timeout, unit, attachment, check, handler, semaphore, completion);
+ }
+
+ private class Nio2OperationState<A> extends OperationState<A> {
+ private Nio2OperationState(boolean read, ByteBuffer[] buffers, int offset, int length,
+ BlockingMode block, long timeout, TimeUnit unit, A attachment,
+ CompletionCheck check, CompletionHandler<Long, ? super A> handler,
+ Semaphore semaphore, VectoredIOCompletionHandler<A> completion) {
+ super(read, buffers, offset, length, block,
+ timeout, unit, attachment, check, handler, semaphore, completion);
}
- if (!socketBufferHandler.isWriteBufferEmpty()) {
- // First flush the main buffer as needed
- try {
- doWrite(true);
- } catch (IOException e) {
- handler.failed(e, attachment);
- return CompletionState.ERROR;
- }
+
+ @Override
+ protected boolean isInline() {
+ return Nio2Endpoint.isInline();
}
- OperationState<A> state = new OperationState<>(false, srcs, offset, length, block,
- timeout, unit, attachment, check, handler, writePending);
- VectoredIOCompletionHandler<A> completion = new VectoredIOCompletionHandler<>();
- Nio2Endpoint.startInline();
- // It should be less necessary to check the buffer state as it is easy to flush before
- getSocket().write(srcs, offset, length, timeout, unit, state, completion);
- Nio2Endpoint.endInline();
- if (block == BlockingMode.BLOCK) {
- synchronized (state) {
- if (state.state == CompletionState.PENDING) {
- try {
- state.wait(unit.toMillis(timeout));
- if (state.state == CompletionState.PENDING) {
- return CompletionState.ERROR;
- }
- } catch (InterruptedException e) {
- handler.failed(new SocketTimeoutException(), attachment);
- return CompletionState.ERROR;
- }
- }
+
+ @Override
+ protected void start() {
+ if (read) {
+ // Disable any regular read notifications caused by registerReadInterest
+ readNotify = true;
+ } else {
+ // Disable any regular write notifications caused by registerWriteInterest
+ writeNotify = true;
}
+ Nio2Endpoint.startInline();
+ run();
+ Nio2Endpoint.endInline();
}
- return state.state;
- }
- private class VectoredIOCompletionHandler<A> implements CompletionHandler<Long, OperationState<A>> {
@Override
- public void completed(Long nBytes, OperationState<A> state) {
- if (nBytes.longValue() < 0) {
- failed(new EOFException(), state);
- } else {
- state.nBytes += nBytes.longValue();
- CompletionState currentState = Nio2Endpoint.isInline() ? CompletionState.INLINE : CompletionState.DONE;
- boolean complete = true;
- boolean completion = true;
- if (state.check != null) {
- CompletionHandlerCall call = state.check.callHandler(currentState, state.buffers, state.offset, state.length);
- if (call == CompletionHandlerCall.CONTINUE) {
- complete = false;
- } else if (call == CompletionHandlerCall.NONE) {
- completion = false;
- }
- }
- if (complete) {
- boolean notify = false;
- state.semaphore.release();
- if (state.block == BlockingMode.BLOCK && currentState != CompletionState.INLINE) {
- notify = true;
- } else {
- state.state = currentState;
- }
- if (completion && state.handler != null) {
- state.handler.completed(Long.valueOf(state.nBytes), state.attachment);
- }
- if (notify) {
- synchronized (state) {
- state.state = currentState;
- state.notify();
+ public void run() {
+ if (read) {
+ long nBytes = 0;
+ // If there is still data inside the main read buffer, it needs to be read first
+ if (!socketBufferHandler.isReadBufferEmpty()) {
+ synchronized (readCompletionHandler) {
+ socketBufferHandler.configureReadBufferForRead();
+ for (int i = 0; i < length && !socketBufferHandler.isReadBufferEmpty(); i++) {
+ nBytes += transfer(socketBufferHandler.getReadBuffer(), buffers[offset + i]);
}
}
- } else {
- if (state.read) {
- getSocket().read(state.buffers, state.offset, state.length,
- state.timeout, state.unit, state, this);
- } else {
- getSocket().write(state.buffers, state.offset, state.length,
- state.timeout, state.unit, state, this);
+ if (nBytes > 0) {
+ completion.completed(Long.valueOf(nBytes), this);
}
}
- }
- }
- @Override
- public void failed(Throwable exc, OperationState<A> state) {
- IOException ioe = null;
- if (exc instanceof InterruptedByTimeoutException) {
- ioe = new SocketTimeoutException();
- exc = ioe;
- } else if (exc instanceof IOException) {
- ioe = (IOException) exc;
- }
- setError(ioe);
- boolean notify = false;
- state.semaphore.release();
- if (state.block == BlockingMode.BLOCK) {
- notify = true;
+ if (nBytes == 0) {
+ getSocket().read(buffers, offset, length, timeout, unit, this, completion);
+ }
} else {
- state.state = Nio2Endpoint.isInline() ? CompletionState.ERROR : CompletionState.DONE;
- }
- if (state.handler != null) {
- state.handler.failed(exc, state.attachment);
- }
- if (notify) {
- synchronized (state) {
- state.state = Nio2Endpoint.isInline() ? CompletionState.ERROR : CompletionState.DONE;
- state.notify();
+ // If there is still data inside the main write buffer, it needs to be written first
+ if (!socketBufferHandler.isWriteBufferEmpty()) {
+ synchronized (writeCompletionHandler) {
+ socketBufferHandler.configureWriteBufferForRead();
+ final ByteBuffer[] array = nonBlockingWriteBuffer.toArray(socketBufferHandler.getWriteBuffer());
+ if (arrayHasData(array)) {
+ getSocket().write(array, 0, array.length, timeout, unit,
+ array, new CompletionHandler<Long, ByteBuffer[]>() {
+ @Override
+ public void completed(Long nBytes, ByteBuffer[] buffers) {
+ if (nBytes.longValue() < 0) {
+ failed(new EOFException(), null);
+ } else if (arrayHasData(buffers)) {
+ getSocket().write(array, 0, array.length, toTimeout(getWriteTimeout()),
+ TimeUnit.MILLISECONDS, array, this);
+ } else {
+ // Continue until everything is written
+ process();
+ }
+ }
+ @Override
+ public void failed(Throwable exc, ByteBuffer[] buffers) {
+ completion.failed(exc, Nio2OperationState.this);
+ }
+ });
+ return;
+ }
+ }
}
+ getSocket().write(buffers, offset, length, timeout, unit, this, completion);
}
}
}
diff --git a/java/org/apache/tomcat/util/net/NioChannel.java b/java/org/apache/tomcat/util/net/NioChannel.java
index 29e2c2a..adaa39f 100644
--- a/java/org/apache/tomcat/util/net/NioChannel.java
+++ b/java/org/apache/tomcat/util/net/NioChannel.java
@@ -19,6 +19,8 @@ package org.apache.tomcat.util.net;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.channels.ByteChannel;
+import java.nio.channels.GatheringByteChannel;
+import java.nio.channels.ScatteringByteChannel;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.SocketChannel;
@@ -33,7 +35,7 @@ import org.apache.tomcat.util.res.StringManager;
*
* @version 1.0
*/
-public class NioChannel implements ByteChannel {
+public class NioChannel implements ByteChannel, ScatteringByteChannel, GatheringByteChannel {
protected static final StringManager sm = StringManager.getManager(NioChannel.class);
@@ -134,6 +136,18 @@ public class NioChannel implements ByteChannel {
return sc.write(src);
}
+ @Override
+ public long write(ByteBuffer[] srcs) throws IOException {
+ return write(srcs, 0, srcs.length);
+ }
+
+ @Override
+ public long write(ByteBuffer[] srcs, int offset, int length)
+ throws IOException {
+ checkInterruptStatus();
+ return sc.write(srcs, offset, length);
+ }
+
/**
* Reads a sequence of bytes from this channel into the given buffer.
*
@@ -147,6 +161,17 @@ public class NioChannel implements ByteChannel {
return sc.read(dst);
}
+ @Override
+ public long read(ByteBuffer[] dsts) throws IOException {
+ return read(dsts, 0, dsts.length);
+ }
+
+ @Override
+ public long read(ByteBuffer[] dsts, int offset, int length)
+ throws IOException {
+ return sc.read(dsts, offset, length);
+ }
+
public Object getAttachment() {
Poller pol = getPoller();
Selector sel = pol!=null?pol.getSelector():null;
diff --git a/java/org/apache/tomcat/util/net/NioEndpoint.java b/java/org/apache/tomcat/util/net/NioEndpoint.java
index 3526687..da1df9d 100644
--- a/java/org/apache/tomcat/util/net/NioEndpoint.java
+++ b/java/org/apache/tomcat/util/net/NioEndpoint.java
@@ -27,6 +27,7 @@ import java.net.SocketTimeoutException;
import java.nio.ByteBuffer;
import java.nio.channels.CancelledKeyException;
import java.nio.channels.Channel;
+import java.nio.channels.CompletionHandler;
import java.nio.channels.FileChannel;
import java.nio.channels.NetworkChannel;
import java.nio.channels.SelectionKey;
@@ -37,6 +38,7 @@ import java.nio.channels.WritableByteChannel;
import java.util.ConcurrentModificationException;
import java.util.Iterator;
import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
@@ -1445,6 +1447,113 @@ public class NioEndpoint extends AbstractJsseEndpoint<NioChannel> {
public void setAppReadBufHandler(ApplicationBufferHandler handler) {
getSocket().setAppReadBufHandler(handler);
}
+
+ @Override
+ protected <A> OperationState<A> newOperationState(boolean read,
+ ByteBuffer[] buffers, int offset, int length,
+ BlockingMode block, long timeout, TimeUnit unit, A attachment,
+ CompletionCheck check, CompletionHandler<Long, ? super A> handler,
+ Semaphore semaphore, VectoredIOCompletionHandler<A> completion) {
+ return new NioOperationState<>(read, buffers, offset, length, block,
+ timeout, unit, attachment, check, handler, semaphore, completion);
+ }
+
+ private class NioOperationState<A> extends OperationState<A> {
+ private volatile boolean inline = true;
+ private NioOperationState(boolean read, ByteBuffer[] buffers, int offset, int length,
+ BlockingMode block, long timeout, TimeUnit unit, A attachment, CompletionCheck check,
+ CompletionHandler<Long, ? super A> handler, Semaphore semaphore,
+ VectoredIOCompletionHandler<A> completion) {
+ super(read, buffers, offset, length, block,
+ timeout, unit, attachment, check, handler, semaphore, completion);
+ }
+
+ @Override
+ protected boolean isInline() {
+ return inline;
+ }
+
+ @Override
+ public void run() {
+ // Perform the IO operation
+ // Called from the poller to continue the IO operation
+ long nBytes = 0;
+ if (getError() == null) {
+ try {
+ synchronized (this) {
+ if (!completionDone) {
+ // This filters out same notification until processing
+ // of the current one is done
+ if (log.isDebugEnabled()) {
+ log.debug("Skip concurrent " + (read ? "read" : "write") + " notification");
+ }
+ return;
+ }
+ if (read) {
+ // Read from main buffer first
+ if (!socketBufferHandler.isReadBufferEmpty()) {
+ // There is still data inside the main read buffer, it needs to be read first
+ socketBufferHandler.configureReadBufferForRead();
+ for (int i = 0; i < length && !socketBufferHandler.isReadBufferEmpty(); i++) {
+ nBytes += transfer(socketBufferHandler.getReadBuffer(), buffers[offset + i]);
+ }
+ }
+ if (nBytes == 0) {
+ nBytes = getSocket().read(buffers, offset, length);
+ updateLastRead();
+ }
+ } else {
+ boolean doWrite = true;
+ // Write from main buffer first
+ if (!socketBufferHandler.isWriteBufferEmpty()) {
+ // There is still data inside the main write buffer, it needs to be written first
+ socketBufferHandler.configureWriteBufferForRead();
+ do {
+ nBytes = getSocket().write(socketBufferHandler.getWriteBuffer());
+ } while (!socketBufferHandler.isWriteBufferEmpty() && nBytes > 0);
+ if (!socketBufferHandler.isWriteBufferEmpty()) {
+ doWrite = false;
+ }
+ // Preserve a negative value since it is an error
+ if (nBytes > 0) {
+ nBytes = 0;
+ }
+ }
+ if (doWrite) {
+ nBytes = getSocket().write(buffers, offset, length);
+ updateLastWrite();
+ }
+ }
+ if (nBytes != 0) {
+ completionDone = false;
+ }
+ }
+ } catch (IOException e) {
+ setError(e);
+ }
+ }
+ if (nBytes > 0) {
+ // The bytes processed are only updated in the completion handler
+ completion.completed(Long.valueOf(nBytes), this);
+ } else if (nBytes < 0 || getError() != null) {
+ IOException error = getError();
+ if (error == null) {
+ error = new EOFException();
+ }
+ completion.failed(error, this);
+ } else {
+ // As soon as the operation uses the poller, it is no longer inline
+ inline = false;
+ if (read) {
+ registerReadInterest();
+ } else {
+ registerWriteInterest();
+ }
+ }
+ }
+
+ }
+
}
diff --git a/java/org/apache/tomcat/util/net/SecureNioChannel.java b/java/org/apache/tomcat/util/net/SecureNioChannel.java
index d972492..483d076 100644
--- a/java/org/apache/tomcat/util/net/SecureNioChannel.java
+++ b/java/org/apache/tomcat/util/net/SecureNioChannel.java
@@ -651,6 +651,117 @@ public class SecureNioChannel extends NioChannel {
return read;
}
+ @Override
+ public long read(ByteBuffer[] dsts, int offset, int length)
+ throws IOException {
+ //are we in the middle of closing or closed?
+ if (closing || closed) {
+ return -1;
+ }
+ //did we finish our handshake?
+ if (!handshakeComplete) {
+ throw new IllegalStateException(sm.getString("channel.nio.ssl.incompleteHandshake"));
+ }
+
+ //read from the network
+ int netread = sc.read(netInBuffer);
+ //did we reach EOF? if so send EOF up one layer.
+ if (netread == -1) {
+ return -1;
+ }
+
+ //the data read
+ int read = 0;
+ //the SSL engine result
+ SSLEngineResult unwrap;
+ boolean processOverflow = false;
+ do {
+ boolean useOverflow = false;
+ if (processOverflow) {
+ useOverflow = true;
+ }
+ processOverflow = false;
+ //prepare the buffer
+ netInBuffer.flip();
+ //unwrap the data
+ unwrap = sslEngine.unwrap(netInBuffer, dsts, offset, length);
+ //compact the buffer
+ netInBuffer.compact();
+
+ if (unwrap.getStatus() == Status.OK || unwrap.getStatus() == Status.BUFFER_UNDERFLOW) {
+ //we did receive some data, add it to our total
+ read += unwrap.bytesProduced();
+ if (useOverflow) {
+ // Remove the data read into the overflow buffer
+ read -= getBufHandler().getReadBuffer().position();
+ }
+ //perform any tasks if needed
+ if (unwrap.getHandshakeStatus() == HandshakeStatus.NEED_TASK) {
+ tasks();
+ }
+ //if we need more network data, then bail out for now.
+ if (unwrap.getStatus() == Status.BUFFER_UNDERFLOW) {
+ break;
+ }
+ } else if (unwrap.getStatus() == Status.BUFFER_OVERFLOW) {
+ if (read > 0) {
+ // Buffer overflow can happen if we have read data. Return
+ // so the destination buffer can be emptied before another
+ // read is attempted
+ break;
+ } else {
+ ByteBuffer readBuffer = getBufHandler().getReadBuffer();
+ boolean found = false;
+ boolean resized = true;
+ for (int i = 0; i < length; i++) {
+ // The SSL session has increased the required buffer size
+ // since the buffer was created.
+ if (dsts[offset + i] == getBufHandler().getReadBuffer()) {
+ getBufHandler().expand(sslEngine.getSession().getApplicationBufferSize());
+ if (dsts[offset + i] == getBufHandler().getReadBuffer()) {
+ resized = false;
+ }
+ dsts[offset + i] = getBufHandler().getReadBuffer();
+ found = true;
+ } else if (getAppReadBufHandler() != null && dsts[offset + i] == getAppReadBufHandler().getByteBuffer()) {
+ getAppReadBufHandler().expand(sslEngine.getSession().getApplicationBufferSize());
+ if (dsts[offset + i] == getAppReadBufHandler().getByteBuffer()) {
+ resized = false;
+ }
+ dsts[offset + i] = getAppReadBufHandler().getByteBuffer();
+ found = true;
+ }
+ }
+ if (found) {
+ if (!resized) {
+ throw new IOException(sm.getString("channel.nio.ssl.unwrapFail", unwrap.getStatus()));
+ }
+ } else {
+ // Add the main read buffer in the destinations and try again
+ ByteBuffer[] dsts2 = new ByteBuffer[dsts.length + 1];
+ int dstOffset = 0;
+ for (int i = 0; i < dsts.length + 1; i++) {
+ if (i == offset + length) {
+ dsts2[i] = readBuffer;
+ dstOffset = -1;
+ } else {
+ dsts2[i] = dsts[i + dstOffset];
+ }
+ }
+ dsts = dsts2;
+ length++;
+ getBufHandler().configureReadBufferForWrite();
+ processOverflow = true;
+ }
+ }
+ } else {
+ // Something else went wrong
+ throw new IOException(sm.getString("channel.nio.ssl.unwrapFail", unwrap.getStatus()));
+ }
+ } while (netInBuffer.position() != 0 || processOverflow); //continue to unwrapping as long as the input buffer has stuff
+ return read;
+ }
+
/**
* Writes a sequence of bytes to this channel from the given buffer.
*
@@ -699,6 +810,40 @@ public class SecureNioChannel extends NioChannel {
}
@Override
+ public long write(ByteBuffer[] srcs, int offset, int length)
+ throws IOException {
+ checkInterruptStatus();
+ // Are we closing or closed?
+ if (closing || closed) {
+ throw new IOException(sm.getString("channel.nio.ssl.closing"));
+ }
+
+ if (!flush(netOutBuffer)) {
+ // We haven't emptied out the buffer yet
+ return 0;
+ }
+
+ // The data buffer is empty, we can reuse the entire buffer.
+ netOutBuffer.clear();
+
+ SSLEngineResult result = sslEngine.wrap(srcs, offset, length, netOutBuffer);
+ // The number of bytes written
+ int written = result.bytesConsumed();
+ netOutBuffer.flip();
+
+ if (result.getStatus() == Status.OK) {
+ if (result.getHandshakeStatus() == HandshakeStatus.NEED_TASK) tasks();
+ } else {
+ throw new IOException(sm.getString("channel.nio.ssl.wrapFail", result.getStatus()));
+ }
+
+ // Force a flush
+ flush(netOutBuffer);
+
+ return written;
+ }
+
+ @Override
public int getOutboundRemaining() {
return netOutBuffer.remaining();
}
diff --git a/java/org/apache/tomcat/util/net/SocketWrapperBase.java b/java/org/apache/tomcat/util/net/SocketWrapperBase.java
index 658d4e4..1470a6c 100644
--- a/java/org/apache/tomcat/util/net/SocketWrapperBase.java
+++ b/java/org/apache/tomcat/util/net/SocketWrapperBase.java
@@ -16,9 +16,17 @@
*/
package org.apache.tomcat.util.net;
+import java.io.EOFException;
import java.io.IOException;
+import java.net.SocketTimeoutException;
import java.nio.ByteBuffer;
import java.nio.channels.CompletionHandler;
+import java.nio.channels.InterruptedByTimeoutException;
+import java.nio.channels.ReadPendingException;
+import java.nio.channels.WritePendingException;
+import java.util.concurrent.Executor;
+import java.util.concurrent.RejectedExecutionException;
+import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantReadWriteLock;
@@ -26,6 +34,7 @@ import java.util.concurrent.locks.ReentrantReadWriteLock.WriteLock;
import org.apache.juli.logging.Log;
import org.apache.juli.logging.LogFactory;
+import org.apache.tomcat.util.ExceptionUtils;
import org.apache.tomcat.util.res.StringManager;
public abstract class SocketWrapperBase<E> {
@@ -46,6 +55,7 @@ public abstract class SocketWrapperBase<E> {
private volatile boolean upgraded = false;
private boolean secure = false;
private String negotiatedProtocol = null;
+
/*
* Following cached for speed / reduced GC
*/
@@ -91,12 +101,27 @@ public abstract class SocketWrapperBase<E> {
*/
protected final WriteBuffer nonBlockingWriteBuffer = new WriteBuffer(bufferedWriteSize);
+ /*
+ * Asynchronous operations.
+ */
+ protected final Semaphore readPending;
+ protected volatile OperationState<?> readOperation = null;
+ protected final Semaphore writePending;
+ protected volatile OperationState<?> writeOperation = null;
+
public SocketWrapperBase(E socket, AbstractEndpoint<E> endpoint) {
this.socket = socket;
this.endpoint = endpoint;
ReentrantReadWriteLock lock = new ReentrantReadWriteLock();
this.blockingStatusReadLock = lock.readLock();
this.blockingStatusWriteLock = lock.writeLock();
+ if (endpoint.getUseAsyncIO() || needSemaphores()) {
+ readPending = new Semaphore(1);
+ writePending = new Semaphore(1);
+ } else {
+ readPending = null;
+ writePending = null;
+ }
}
public E getSocket() {
@@ -107,6 +132,21 @@ public abstract class SocketWrapperBase<E> {
return endpoint;
}
+ /**
+ * Transfers processing to a container thread.
+ *
+ * @param runnable The actions to process on a container thread
+ *
+ * @throws RejectedExecutionException If the runnable cannot be executed
+ */
+ public void execute(Runnable runnable) {
+ Executor executor = endpoint.getExecutor();
+ if (!endpoint.isRunning() || executor == null) {
+ throw new RejectedExecutionException();
+ }
+ executor.execute(runnable);
+ }
+
public IOException getError() { return error; }
public void setError(IOException error) {
// Not perfectly thread-safe but good enough. Just needs to ensure that
@@ -935,12 +975,198 @@ public abstract class SocketWrapperBase<E> {
public static final CompletionCheck COMPLETE_READ = COMPLETE_WRITE;
/**
- * Allows using NIO2 style read/write only for connectors that can
- * efficiently support it.
+ * Internal state tracker for vectored operations.
+ */
+ protected abstract class OperationState<A> implements Runnable {
+ protected final boolean read;
+ protected final ByteBuffer[] buffers;
+ protected final int offset;
+ protected final int length;
+ protected final A attachment;
+ protected final long timeout;
+ protected final TimeUnit unit;
+ protected final BlockingMode block;
+ protected final CompletionCheck check;
+ protected final CompletionHandler<Long, ? super A> handler;
+ protected final Semaphore semaphore;
+ protected final VectoredIOCompletionHandler<A> completion;
+ protected OperationState(boolean read, ByteBuffer[] buffers, int offset, int length,
+ BlockingMode block, long timeout, TimeUnit unit, A attachment,
+ CompletionCheck check, CompletionHandler<Long, ? super A> handler,
+ Semaphore semaphore, VectoredIOCompletionHandler<A> completion) {
+ this.read = read;
+ this.buffers = buffers;
+ this.offset = offset;
+ this.length = length;
+ this.block = block;
+ this.timeout = timeout;
+ this.unit = unit;
+ this.attachment = attachment;
+ this.check = check;
+ this.handler = handler;
+ this.semaphore = semaphore;
+ this.completion = completion;
+ }
+ protected volatile long nBytes = 0;
+ protected volatile CompletionState state = CompletionState.PENDING;
+ protected boolean completionDone = true;
+
+ /**
+ * @return true if the operation is still inline, false if the operation
+ * is running on a thread that is not the original caller
+ */
+ protected abstract boolean isInline();
+
+ /**
+ * Process the operation using the connector executor.
+ * @return true if the operation was accepted, false if the executor
+ * rejected execution
+ */
+ protected boolean process() {
+ try {
+ getEndpoint().getExecutor().execute(this);
+ return true;
+ } catch (RejectedExecutionException ree) {
+ log.warn(sm.getString("endpoint.executor.fail", SocketWrapperBase.this) , ree);
+ } catch (Throwable t) {
+ ExceptionUtils.handleThrowable(t);
+ // This means we got an OOM or similar creating a thread, or that
+ // the pool and its queue are full
+ log.error(sm.getString("endpoint.process.fail"), t);
+ }
+ return false;
+ }
+
+ /**
+ * Start the operation, this will typically call run.
+ */
+ protected void start() {
+ run();
+ }
+
+ /**
+ * End the operation.
+ */
+ protected void end() {
+ }
+
+ }
+
+ /**
+ * Completion handler for vectored operations. This will check the completion of the operation,
+ * then either continue or call the user provided completion handler.
+ */
+ protected class VectoredIOCompletionHandler<A> implements CompletionHandler<Long, OperationState<A>> {
+ @Override
+ public void completed(Long nBytes, OperationState<A> state) {
+ if (nBytes.longValue() < 0) {
+ failed(new EOFException(), state);
+ } else {
+ state.nBytes += nBytes.longValue();
+ CompletionState currentState = state.isInline() ? CompletionState.INLINE : CompletionState.DONE;
+ boolean complete = true;
+ boolean completion = true;
+ if (state.check != null) {
+ CompletionHandlerCall call = state.check.callHandler(currentState, state.buffers, state.offset, state.length);
+ if (call == CompletionHandlerCall.CONTINUE) {
+ complete = false;
+ } else if (call == CompletionHandlerCall.NONE) {
+ completion = false;
+ }
+ }
+ if (complete) {
+ boolean notify = false;
+ state.semaphore.release();
+ if (state.read) {
+ readOperation = null;
+ } else {
+ writeOperation = null;
+ }
+ if (state.block == BlockingMode.BLOCK && currentState != CompletionState.INLINE) {
+ notify = true;
+ } else {
+ state.state = currentState;
+ }
+ state.end();
+ if (completion && state.handler != null) {
+ state.handler.completed(Long.valueOf(state.nBytes), state.attachment);
+ }
+ synchronized (state) {
+ state.completionDone = true;
+ if (notify) {
+ state.state = currentState;
+ state.notify();
+ }
+ }
+ } else {
+ synchronized (state) {
+ state.completionDone = true;
+ }
+ state.run();
+ }
+ }
+ }
+ @Override
+ public void failed(Throwable exc, OperationState<A> state) {
+ IOException ioe = null;
+ if (exc instanceof InterruptedByTimeoutException) {
+ ioe = new SocketTimeoutException();
+ exc = ioe;
+ } else if (exc instanceof IOException) {
+ ioe = (IOException) exc;
+ }
+ setError(ioe);
+ boolean notify = false;
+ state.semaphore.release();
+ if (state.read) {
+ readOperation = null;
+ } else {
+ writeOperation = null;
+ }
+ if (state.block == BlockingMode.BLOCK) {
+ notify = true;
+ } else {
+ state.state = state.isInline() ? CompletionState.ERROR : CompletionState.DONE;
+ }
+ state.end();
+ if (state.handler != null) {
+ state.handler.failed(exc, state.attachment);
+ }
+ synchronized (state) {
+ state.completionDone = true;
+ if (notify) {
+ state.state = state.isInline() ? CompletionState.ERROR : CompletionState.DONE;
+ state.notify();
+ }
+ }
+ }
+ }
+
+ /**
+ * Allows using NIO2 style read/write.
*
- * @return This default implementation always returns {@code false}
+ * @return {@code true} if the connector has the capability enabled
*/
public boolean hasAsyncIO() {
+ // The semaphores are only created if async IO is enabled
+ return (readPending != null);
+ }
+
+ /**
+ * Allows indicating if the connector needs semaphores.
+ *
+ * @return This default implementation always returns {@code false}
+ */
+ public boolean needSemaphores() {
+ return false;
+ }
+
+ /**
+ * Allows indicating if the connector supports per operation timeout.
+ *
+ * @return This default implementation always returns {@code false}
+ */
+ public boolean hasPerOperationTimeout() {
return false;
}
@@ -1066,10 +1292,10 @@ public abstract class SocketWrapperBase<E> {
* @param <A> The attachment type
* @return the completion state (done, done inline, or still pending)
*/
- public <A> CompletionState read(ByteBuffer[] dsts, int offset, int length,
+ public final <A> CompletionState read(ByteBuffer[] dsts, int offset, int length,
BlockingMode block, long timeout, TimeUnit unit, A attachment,
CompletionCheck check, CompletionHandler<Long, ? super A> handler) {
- throw new UnsupportedOperationException();
+ return vectoredOperation(true, dsts, offset, length, block, timeout, unit, attachment, check, handler);
}
/**
@@ -1149,13 +1375,108 @@ public abstract class SocketWrapperBase<E> {
* @param <A> The attachment type
* @return the completion state (done, done inline, or still pending)
*/
- public <A> CompletionState write(ByteBuffer[] srcs, int offset, int length,
+ public final <A> CompletionState write(ByteBuffer[] srcs, int offset, int length,
BlockingMode block, long timeout, TimeUnit unit, A attachment,
CompletionCheck check, CompletionHandler<Long, ? super A> handler) {
- throw new UnsupportedOperationException();
+ return vectoredOperation(false, srcs, offset, length, block, timeout, unit, attachment, check, handler);
}
+ /**
+ * Vectored operation. The completion handler will be called once
+ * the operation is complete or an error occurred. If a CompletionCheck
+ * object has been provided, the completion handler will only be
+ * called if the callHandler method returned true. If no
+ * CompletionCheck object has been provided, the default NIO2
+ * behavior is used: the completion handler will be called, even
+ * if the operation is incomplete, or if the operation completed inline.
+ *
+ * @param read true if the operation is a read, false if it is a write
+ * @param buffers buffers
+ * @param offset in the buffer array
+ * @param length in the buffer array
+ * @param block is the blocking mode that will be used for this operation
+ * @param timeout timeout duration for the write
+ * @param unit units for the timeout duration
+ * @param attachment an object to attach to the I/O operation that will be
+ * used when calling the completion handler
+ * @param check for the IO operation completion
+ * @param handler to call when the IO is complete
+ * @param <A> The attachment type
+ * @return the completion state (done, done inline, or still pending)
+ */
+ protected final <A> CompletionState vectoredOperation(boolean read,
+ ByteBuffer[] buffers, int offset, int length,
+ BlockingMode block, long timeout, TimeUnit unit, A attachment,
+ CompletionCheck check, CompletionHandler<Long, ? super A> handler) {
+ IOException ioe = getError();
+ if (ioe != null) {
+ handler.failed(ioe, attachment);
+ return CompletionState.ERROR;
+ }
+ if (timeout == -1) {
+ timeout = AbstractEndpoint.toTimeout(read ? getReadTimeout() : getWriteTimeout());
+ unit = TimeUnit.MILLISECONDS;
+ } else if (!hasPerOperationTimeout() && (unit.toMillis(timeout) != (read ? getReadTimeout() : getWriteTimeout()))) {
+ if (read) {
+ setReadTimeout(unit.toMillis(timeout));
+ } else {
+ setWriteTimeout(unit.toMillis(timeout));
+ }
+ }
+ if (block == BlockingMode.BLOCK || block == BlockingMode.SEMI_BLOCK) {
+ try {
+ if (read ? !readPending.tryAcquire(timeout, unit) : !writePending.tryAcquire(timeout, unit)) {
+ handler.failed(new SocketTimeoutException(), attachment);
+ return CompletionState.ERROR;
+ }
+ } catch (InterruptedException e) {
+ handler.failed(e, attachment);
+ return CompletionState.ERROR;
+ }
+ } else {
+ if (read ? !readPending.tryAcquire() : !writePending.tryAcquire()) {
+ if (block == BlockingMode.NON_BLOCK) {
+ return CompletionState.NOT_DONE;
+ } else {
+ handler.failed(read ? new ReadPendingException() : new WritePendingException(), attachment);
+ return CompletionState.ERROR;
+ }
+ }
+ }
+ VectoredIOCompletionHandler<A> completion = new VectoredIOCompletionHandler<>();
+ OperationState<A> state = newOperationState(read, buffers, offset, length, block, timeout, unit,
+ attachment, check, handler, read ? readPending : writePending, completion);
+ if (read) {
+ readOperation = state;
+ } else {
+ writeOperation = state;
+ }
+ state.start();
+ if (block == BlockingMode.BLOCK) {
+ synchronized (state) {
+ if (state.state == CompletionState.PENDING) {
+ try {
+ state.wait(unit.toMillis(timeout));
+ if (state.state == CompletionState.PENDING) {
+ return CompletionState.ERROR;
+ }
+ } catch (InterruptedException e) {
+ completion.failed(new SocketTimeoutException(), state);
+ return CompletionState.ERROR;
+ }
+ }
+ }
+ }
+ return state.state;
+ }
+
+ protected abstract <A> OperationState<A> newOperationState(boolean read,
+ ByteBuffer[] buffers, int offset, int length,
+ BlockingMode block, long timeout, TimeUnit unit, A attachment,
+ CompletionCheck check, CompletionHandler<Long, ? super A> handler,
+ Semaphore semaphore, VectoredIOCompletionHandler<A> completion);
+
// --------------------------------------------------------- Utility methods
protected static int transfer(byte[] from, int offset, int length, ByteBuffer to) {
diff --git a/webapps/docs/changelog.xml b/webapps/docs/changelog.xml
index 2381942..b3ec86e 100644
--- a/webapps/docs/changelog.xml
+++ b/webapps/docs/changelog.xml
@@ -61,6 +61,12 @@
Windows operating systems that required multiple smaller pollsets to be
used are no longer supported. (markt)
</scode>
+ <update>
+ Add vectoring for NIO in the base and SSL channels. (remm)
+ </update>
+ <add>
+ Add async API to the NIO and APR connector. (remm)
+ </add>
</changelog>
</subsection>
<subsection name="Web applications">
---------------------------------------------------------------------
To unsubscribe, e-mail: dev-unsubscribe@tomcat.apache.org
For additional commands, e-mail: dev-help@tomcat.apache.org