diff --git a/org.restlet/src/main/java/org/restlet/engine/netty/ChannelOutputStream.java b/org.restlet/src/main/java/org/restlet/engine/netty/ChannelOutputStream.java index 3945ee057a..b7e79884b3 100644 --- a/org.restlet/src/main/java/org/restlet/engine/netty/ChannelOutputStream.java +++ b/org.restlet/src/main/java/org/restlet/engine/netty/ChannelOutputStream.java @@ -12,7 +12,12 @@ import io.netty.channel.Channel; import io.netty.channel.ChannelFuture; import java.io.IOException; +import java.io.InterruptedIOException; import java.io.OutputStream; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicLong; +import java.util.concurrent.locks.Condition; +import java.util.concurrent.locks.ReentrantLock; import java.util.function.Function; /** @@ -42,6 +47,16 @@ final class ChannelOutputStream extends OutputStream { private boolean closed; + /** Bytes written to the channel whose write has not completed yet. */ + private final AtomicLong inFlightBytes = new AtomicLong(); + + private final ReentrantLock inFlightLock = new ReentrantLock(); + + private final Condition inFlightChanged = inFlightLock.newCondition(); + + /** The cause of the first failed write, if any. */ + private volatile Throwable writeFailure; + /** * Constructor. * @@ -75,8 +90,8 @@ public void write(int b) throws IOException { @Override public void write(byte[] b, int off, int len) throws IOException { - ensureOpen(); while (len > 0) { + ensureOpen(); ensureBuffer(); int n = Math.min(len, buffer.writableBytes()); buffer.writeBytes(b, off, n); @@ -131,19 +146,88 @@ private void ensureOpen() throws IOException { if (closed) { throw new IOException("Stream closed"); } + checkWriteFailure(); if (!channel.isActive()) { throw new IOException("Connection closed by peer"); } } + private void checkWriteFailure() throws IOException { + Throwable cause = writeFailure; + if (cause != null) { + throw new IOException("Unable to write to the connection", cause); + } + } + private void send(boolean flush) throws IOException { Writability.await(channel, writeTimeoutMs); + awaitInFlightBytes(); ByteBuf chunk = buffer; buffer = null; + int size = chunk.readableBytes(); + inFlightBytes.addAndGet(size); Object msg = contentFactory.apply(chunk); ChannelFuture future = flush ? channel.writeAndFlush(msg) : channel.write(msg); + future.addListener(f -> onWriteDone(size, f.cause())); if (future.isDone() && !future.isSuccess()) { throw new IOException("Unable to write to the connection", future.cause()); } } + + /** Called when a write completes, possibly on the event loop. */ + private void onWriteDone(int size, Throwable cause) { + inFlightLock.lock(); + try { + inFlightBytes.addAndGet(-size); + if (cause != null && writeFailure == null) { + writeFailure = cause; + } + inFlightChanged.signalAll(); + } finally { + inFlightLock.unlock(); + } + } + + /** + * Waits until the bytes written but not yet sent fall below the channel high water mark. Writes + * made outside the event loop only update the channel writability once the event loop has + * processed them, so a fast producer could otherwise queue unbounded data, and miss the failure + * of writes to a connection closed in the meantime. + */ + private void awaitInFlightBytes() throws IOException { + int limit = Math.max(bufferSize, channel.config().getWriteBufferHighWaterMark()); + if (inFlightBytes.get() <= limit || channel.eventLoop().inEventLoop()) { + checkWriteFailure(); + return; + } + + channel.flush(); + long remaining = + (writeTimeoutMs > 0) + ? TimeUnit.MILLISECONDS.toNanos(writeTimeoutMs) + : Long.MAX_VALUE; + inFlightLock.lock(); + try { + while (inFlightBytes.get() > limit && writeFailure == null) { + if (!channel.isActive()) { + throw new IOException("Connection closed by peer"); + } + if (remaining <= 0) { + throw new InterruptedIOException("Write timed out"); + } + // Wake up periodically in case the channel closes without failing a write. + long slice = Math.min(remaining, TimeUnit.MILLISECONDS.toNanos(100)); + long left = inFlightChanged.awaitNanos(slice); + if (writeTimeoutMs > 0) { + remaining -= (slice - Math.max(0, left)); + } + } + } catch (InterruptedException _) { + Thread.currentThread().interrupt(); + throw new InterruptedIOException("Interrupted while writing"); + } finally { + inFlightLock.unlock(); + } + checkWriteFailure(); + } } diff --git a/org.restlet/src/main/java/org/restlet/engine/netty/NettyServerCall.java b/org.restlet/src/main/java/org/restlet/engine/netty/NettyServerCall.java index 43d7e14c3a..96793418b0 100644 --- a/org.restlet/src/main/java/org/restlet/engine/netty/NettyServerCall.java +++ b/org.restlet/src/main/java/org/restlet/engine/netty/NettyServerCall.java @@ -608,6 +608,27 @@ private void writeHead(boolean emptyBody) { responseHeadWritten = true; int code = statusCode; + HttpResponse head = newResponseHead(code); + boolean bodyAllowed = + code != 204 && code != 304 && code >= 200 && !"HEAD".equals(getMethod()); + boolean hasTrailers = bodyAllowed && !responseTrailers.isEmpty(); + // Streams can carry trailers added after the head was written. + responseTrailersEnabled = hasTrailers || (kind.isStream() && bodyAllowed); + + if (kind.isStream()) { + // Connection-specific headers are forbidden in HTTP/2 and HTTP/3. + head.headers().remove(HttpHeaderNames.CONNECTION); + head.headers().remove(HttpHeaderNames.TRANSFER_ENCODING); + head.headers().remove("Keep-Alive"); + } else { + setHttp1Framing(head, emptyBody, hasTrailers, bodyAllowed); + } + + channel.write(head); + } + + /** Creates the response head with the status and the response headers. */ + private HttpResponse newResponseHead(int code) { String reason = reasonPhrase; HttpResponseStatus status = (reason != null && !reason.isEmpty()) @@ -624,45 +645,27 @@ private void writeHead(boolean emptyBody) { if (altSvc != null && !head.headers().contains("Alt-Svc")) { head.headers().set(kind.isStream() ? "alt-svc" : "Alt-Svc", altSvc); } - - boolean hasTrailers = !responseTrailers.isEmpty(); - boolean hasLength = head.headers().contains(HttpHeaderNames.CONTENT_LENGTH); - boolean noBodyStatus = code == 204 || code == 304 || code < 200; - - if (noBodyStatus || "HEAD".equals(getMethod())) { - hasTrailers = false; - } - responseTrailersEnabled = - hasTrailers || (kind.isStream() && !noBodyStatus && !"HEAD".equals(getMethod())); - - if (kind.isStream()) { - // Connection-specific headers are forbidden in HTTP/2 and HTTP/3. - head.headers().remove(HttpHeaderNames.CONNECTION); - head.headers().remove(HttpHeaderNames.TRANSFER_ENCODING); - head.headers().remove("Keep-Alive"); - } else { - if (emptyBody) { - if (hasTrailers) { - head.headers().remove(HttpHeaderNames.CONTENT_LENGTH); - HttpUtil.setTransferEncodingChunked(head, true); - } else if (!hasLength && !noBodyStatus && !"HEAD".equals(getMethod())) { - head.headers().set(HttpHeaderNames.CONTENT_LENGTH, 0); - } - } else if (hasTrailers) { - head.headers().remove(HttpHeaderNames.CONTENT_LENGTH); - HttpUtil.setTransferEncodingChunked(head, true); - } else if (!hasLength) { - head.headers().set(HttpHeaderNames.TRANSFER_ENCODING, HttpHeaderValues.CHUNKED); - } - if (hasTrailers) { - setTrailerNames(head.headers(), responseTrailers); - } - if (!HttpUtil.isKeepAlive(request)) { - head.headers().set(HttpHeaderNames.CONNECTION, HttpHeaderValues.CLOSE); + return head; + } + + /** Sets the HTTP/1.1 message framing and connection headers of the response head. */ + private void setHttp1Framing( + HttpResponse head, boolean emptyBody, boolean hasTrailers, boolean bodyAllowed) { + HttpHeaders headers = head.headers(); + if (hasTrailers) { + headers.remove(HttpHeaderNames.CONTENT_LENGTH); + HttpUtil.setTransferEncodingChunked(head, true); + setTrailerNames(headers, responseTrailers); + } else if (!headers.contains(HttpHeaderNames.CONTENT_LENGTH)) { + if (!emptyBody) { + headers.set(HttpHeaderNames.TRANSFER_ENCODING, HttpHeaderValues.CHUNKED); + } else if (bodyAllowed) { + headers.set(HttpHeaderNames.CONTENT_LENGTH, 0); } } - - channel.write(head); + if (!HttpUtil.isKeepAlive(request)) { + headers.set(HttpHeaderNames.CONNECTION, HttpHeaderValues.CLOSE); + } } /** Completes the exchange: ends the response and frees the connection for the next request. */ diff --git a/org.restlet/src/main/java/org/restlet/engine/netty/NettyServerHandler.java b/org.restlet/src/main/java/org/restlet/engine/netty/NettyServerHandler.java index d6c65a2310..e6c7223b69 100644 --- a/org.restlet/src/main/java/org/restlet/engine/netty/NettyServerHandler.java +++ b/org.restlet/src/main/java/org/restlet/engine/netty/NettyServerHandler.java @@ -115,6 +115,11 @@ private void onRequest(ChannelHandlerContext ctx, HttpRequest request) { ChannelInputStream body = new ChannelInputStream(ctx.channel(), helper.getConnectorIdleTimeout()); currentBody = body; + if (kind == StreamKind.HTTP1 && !hasBody(request)) { + // End the body now: the call could otherwise complete before the last (empty) content + // is delivered, and close a connection that the response announced as kept alive. + body.end(); + } NettyServerCall call = new NettyServerCall(helper, ctx.channel(), kind, request, body, confidential); @@ -190,6 +195,12 @@ private void onContent(ChannelHandlerContext ctx, HttpContent content) { // Otherwise the body stream requests the next read once the consumer catches up. } + /** Indicates whether an HTTP/1.1 request has a body, according to its framing headers. */ + private static boolean hasBody(HttpRequest request) { + return HttpUtil.isTransferEncodingChunked(request) + || HttpUtil.getContentLength(request, 0L) != 0; + } + private void runCall(NettyServerCall call) { try { helper.handle(call);