Eclipse
/
jetty.project
mirror of https://github.com/jetty/jetty.project.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Experiment with IteratingCallback (#12040) 

The previous semantic of `onCompleteFailure` has been renamed to `onFailure(Throwable)`, which is called immediately (but serialized) on either an abort or a failure.   A new `onCompleteFailure(Throwable)` method has been added that is called only after a `failed(throwable)` or a `abort(Throwable)` followed by `succeeded()` or `failed(Throwable)``

No usage has yet been made of the new `onCompleteFailure`, but the ICB implementation has been completely replaced by the one developed in #11876

Signed-off-by: Simone Bordet <simone.bordet@gmail.com>
Signed-off-by: Ludovic Orban <lorban@bitronix.be>
Co-authored-by: Simone Bordet <simone.bordet@gmail.com>
Co-authored-by: Ludovic Orban <lorban@bitronix.be>
This commit is contained in:
Greg Wilkins 2024-08-26 02:18:57 +02:00 committed by GitHub
parent b9bcb58a6d
commit 7d7eeb3b1e
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
55 changed files with 1592 additions and 554 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

31
documentation/jetty/modules/code/examples/src/main/java/org/eclipse/jetty/docs/programming/ContentDocs.java
View File

@ -329,11 +329,11 @@ public class ContentDocs
// Read a chunk. // Read a chunk.
chunk = source.read(); chunk = source.read();
// No chunk, demand to be called back when there will be more chunks. // If no chunk, schedule a demand callback when there are more chunks.
if (chunk == null) if (chunk == null)
{ {
source.demand(this::iterate); source.demand(this::succeeded);
return Action.IDLE; return Action.SCHEDULED;
} }
// The read failed, re-throw the failure // The read failed, re-throw the failure
@ -341,7 +341,7 @@ public class ContentDocs
if (Content.Chunk.isFailure(chunk)) if (Content.Chunk.isFailure(chunk))
throw chunk.getFailure(); throw chunk.getFailure();
// Copy the chunk. // Copy the chunk by scheduling an asynchronous write.
sink.write(chunk.isLast(), chunk.getByteBuffer(), this); sink.write(chunk.isLast(), chunk.getByteBuffer(), this);
return Action.SCHEDULED; return Action.SCHEDULED;
} }
@ -349,8 +349,9 @@ public class ContentDocs
@Override @Override
protected void onSuccess() protected void onSuccess()
{ {
// After every successful write, release the chunk. // After every successful write, release the chunk
chunk.release(); // and reset to the next chunk
chunk = Content.Chunk.releaseAndNext(chunk);
} }
@Override @Override
@ -360,15 +361,21 @@ public class ContentDocs
callback.succeeded(); callback.succeeded();
} }
@Override
protected void onFailure(Throwable cause)
{
// The copy is failed, fail the callback.
// This method is invoked before a write() has completed, so
// the chunk is not released here, but in onCompleteFailure().
callback.failed(cause);
}
@Override @Override
protected void onCompleteFailure(Throwable failure) protected void onCompleteFailure(Throwable failure)
{ {
// In case of a failure, either on the // In case of a failure, this method is invoked when the write()
// read or on the write, release the chunk. // is completed, and it is now possible to release the chunk.
chunk.release(); chunk = Content.Chunk.releaseAndNext(chunk);
// The copy is failed, fail the callback.
callback.failed(failure);
} }
@Override @Override

36
documentation/jetty/modules/code/examples/src/main/java/org/eclipse/jetty/docs/programming/SelectorManagerDocs.java
View File

@ -24,9 +24,12 @@ import java.util.concurrent.atomic.AtomicInteger;
import org.eclipse.jetty.client.HttpClient; import org.eclipse.jetty.client.HttpClient;
import org.eclipse.jetty.io.AbstractConnection; import org.eclipse.jetty.io.AbstractConnection;
import org.eclipse.jetty.io.ByteBufferPool;
import org.eclipse.jetty.io.Connection; import org.eclipse.jetty.io.Connection;
import org.eclipse.jetty.io.ConnectionStatistics; import org.eclipse.jetty.io.ConnectionStatistics;
import org.eclipse.jetty.io.EndPoint; import org.eclipse.jetty.io.EndPoint;
import org.eclipse.jetty.io.Retainable;
import org.eclipse.jetty.io.RetainableByteBuffer;
import org.eclipse.jetty.io.SelectorManager; import org.eclipse.jetty.io.SelectorManager;
import org.eclipse.jetty.server.Server; import org.eclipse.jetty.server.Server;
import org.eclipse.jetty.server.ServerConnector; import org.eclipse.jetty.server.ServerConnector;
@ -225,11 +228,13 @@ public class SelectorManagerDocs
// tag::echo-correct[] // tag::echo-correct[]
class EchoConnection extends AbstractConnection class EchoConnection extends AbstractConnection
{ {
private final ByteBufferPool.Sized pool;
private final IteratingCallback callback = new EchoIteratingCallback(); private final IteratingCallback callback = new EchoIteratingCallback();
public EchoConnection(EndPoint endp, Executor executor) public EchoConnection(EndPoint endp, ByteBufferPool.Sized pool, Executor executor)
{ {
super(endp, executor); super(endp, executor);
this.pool = pool;
} }
@Override @Override
@ -250,20 +255,20 @@ public class SelectorManagerDocs
class EchoIteratingCallback extends IteratingCallback class EchoIteratingCallback extends IteratingCallback
{ {
private ByteBuffer buffer; private RetainableByteBuffer buffer;
@Override @Override
protected Action process() throws Throwable protected Action process() throws Throwable
{ {
// Obtain a buffer if we don't already have one. // Obtain a buffer if we don't already have one.
if (buffer == null) if (buffer == null)
buffer = BufferUtil.allocate(1024); buffer = pool.acquire();
int filled = getEndPoint().fill(buffer); int filled = getEndPoint().fill(buffer.getByteBuffer());
if (filled > 0) if (filled > 0)
{ {
// We have filled some bytes, echo them back. // We have filled some bytes, echo them back.
getEndPoint().write(this, buffer); getEndPoint().write(this, buffer.getByteBuffer());
// Signal that the iteration should resume // Signal that the iteration should resume
// when the write() operation is completed. // when the write() operation is completed.
@ -273,14 +278,15 @@ public class SelectorManagerDocs
{ {
// We don't need the buffer anymore, so // We don't need the buffer anymore, so
// don't keep it around while we are idle. // don't keep it around while we are idle.
buffer = null; buffer = Retainable.release(buffer);
// No more bytes to read, declare // No more bytes to read, declare
// again interest for fill events. // again interest for fill events.
fillInterested(); fillInterested(this);
// Signal that the iteration is now IDLE. // Signal that the iteration is now SCHEDULED
return Action.IDLE; // for a fillable callback.
return Action.SCHEDULED;
} }
else else
{ {
@ -291,17 +297,11 @@ public class SelectorManagerDocs
} }
@Override @Override
protected void onCompleteSuccess() protected void onCompleted(Throwable cause)
{ {
// The iteration completed successfully. // The iteration completed.
getEndPoint().close();
}
@Override
protected void onCompleteFailure(Throwable cause)
{
// The iteration completed with a failure.
getEndPoint().close(cause); getEndPoint().close(cause);
buffer = Retainable.release(buffer);
} }
@Override @Override

6
documentation/jetty/modules/code/examples/src/main/java/org/eclipse/jetty/docs/programming/WebSocketDocs.java
View File

@ -523,14 +523,16 @@ public class WebSocketDocs
@Override @Override
public void succeed() public void succeed()
{ {
// When the send succeeds, succeed this IteratingCallback. // Map the o.e.j.websocket.api.Callback to o.e.j.util.Callback.
// When the send() succeeds, succeed this IteratingCallback.
succeeded(); succeeded();
} }
@Override @Override
public void fail(Throwable x) public void fail(Throwable x)
{ {
// When the send fails, fail this IteratingCallback. // Map the o.e.j.websocket.api.Callback to o.e.j.util.Callback.
// When the send() fails, fail this IteratingCallback.
failed(x); failed(x);
} }

17
documentation/jetty/modules/code/examples/src/main/java/org/eclipse/jetty/docs/programming/server/ServerDocs.java
View File

@ -176,6 +176,7 @@ public class ServerDocs
@Override @Override
public void onFillable() public void onFillable()
{ {
// Called from fillInterested() in onOpen() to start iteration.
callback.iterate(); callback.iterate();
} }
@ -206,11 +207,8 @@ public class ServerDocs
// the application completed the request processing. // the application completed the request processing.
return Action.SCHEDULED; return Action.SCHEDULED;
} }
else // Did not receive enough JSON bytes to complete the
{ // JSON parsing, loop around to try to read more bytes.
// Did not receive enough JSON bytes,
// loop around to try to read more.
}
} }
else if (filled == 0) else if (filled == 0)
{ {
@ -218,12 +216,11 @@ public class ServerDocs
// don't keep it around while we are idle. // don't keep it around while we are idle.
buffer = null; buffer = null;
// No more bytes to read, declare // No more bytes to read, declare again interest for fill events.
// again interest for fill events. fillInterested(this);
fillInterested();
// Signal that the iteration is now IDLE. // Signal that the iteration is now SCHEDULED for fill interest callback.
return Action.IDLE; return Action.SCHEDULED;
} }
else else
{ {

15
documentation/jetty/modules/programming-guide/pages/arch/io.adoc
View File

@ -187,7 +187,7 @@ In turn, this calls `IteratingCallback.process()`, an abstract method that must
Method `process()` must return: Method `process()` must return:
* `Action.SCHEDULED`, to indicate whether the loop has performed a non-blocking, possibly asynchronous, operation * `Action.SCHEDULED`, to indicate whether the loop has performed a non-blocking, possibly asynchronous, operation
* `Action.IDLE`, to indicate that the loop should temporarily be suspended to be resumed later * `Action.IDLE`, to indicate that the loop should temporarily be suspended to be resumed later with another call to iterate
* `Action.SUCCEEDED` to indicate that the loop exited successfully * `Action.SUCCEEDED` to indicate that the loop exited successfully
Any exception thrown within `process()` exits the loops with a failure. Any exception thrown within `process()` exits the loops with a failure.
@ -209,13 +209,18 @@ If this was the only active network connection, the system would now be idle, wi
Eventually, the Jetty I/O system will notify that the `write()` completed; this notifies the `IteratingCallback` that can now resume the loop and call `process()` again. Eventually, the Jetty I/O system will notify that the `write()` completed; this notifies the `IteratingCallback` that can now resume the loop and call `process()` again.
When `process()` is called, it is possible that zero bytes are read from the network; in this case, you want to deallocate the buffer since the other peer may never send more bytes for the `Connection` to read, or it may send them after a long pause -- in both cases we do not want to retain the memory allocated by the buffer; next, you want to call `fillInterested()` to declare again interest for read events, and return `Action.IDLE` since there is nothing to write back and therefore the loop may be suspended. When `process()` is called, it is possible that zero bytes are read from the network; in this case, you want to deallocate the buffer since the other peer may never send more bytes for the `Connection` to read, or it may send them after a long pause -- in both cases we do not want to retain the memory allocated by the buffer; next, you want to call `fillInterested(this)` to declare again interest for read events, and return `Action.SCHEDULED` since a callback is scheduled to occur once filling is possible.
When more bytes are again available to be read from the network, `onFillable()` will be called again and that will start the iteration again.
Another possibility is that during `process()` the read returns `-1` indicating that the other peer has closed the connection; this means that there will not be more bytes to read and the loop can be exited, so you return `Action.SUCCEEDED`; `IteratingCallback` will then call `onCompleteSuccess()` where you can close the `EndPoint`. Another possibility is that during `process()` the read returns `-1` indicating that the other peer has closed the connection; this means that there will not be more bytes to read and the loop can be exited, so you return `Action.SUCCEEDED`; `IteratingCallback` will then call `onCompleteSuccess()` where you can close the `EndPoint`.
The last case is that during `process()` an exception is thrown, for example by `EndPoint.fill(ByteBuffer)` or, in more advanced implementations, by code that parses the bytes that have been read and finds them unacceptable; any exception thrown within `process()` will be caught by `IteratingCallback` that will exit the loop with a failure and call `onCompleteFailure(Throwable)` with the exception that has been thrown, where you can close the `EndPoint`, passing the exception that is the reason for closing prematurely the `EndPoint`. The last case is that during `process()` an exception is thrown, for example by `EndPoint.fill(ByteBuffer)` or, in more advanced implementations, by code that parses the bytes that have been read and finds them unacceptable; any exception thrown within `process()` will be caught by `IteratingCallback` that will exit the loop with a failure and call `onCompleteFailure(Throwable)` with the exception that has been thrown, where you can close the `EndPoint`, passing the exception that is the reason for closing prematurely the `EndPoint`.
Note that some failures may occur whilst a scheduled operation is in progress.
Such failures are notified immediately via the `onFailure(Throwable)` method, but care must be taken to not release any resources that may still be in use by the scheduled operation.
The `onCompleteFailure(Throwable)` method is called when both a failure has occurred and any scheduled operation has completed.
An example of this issue is that a buffer used for a write operation cannot be returned to a pool in `onFailure(Throwable)` as the write may still be progressing.
Either the buffer must be removed from the pool in `onFailure(Throwable)` or the release of the buffer deferred until `onCompleteFailure(Throwable)` is called.
[IMPORTANT] [IMPORTANT]
==== ====
Asynchronous programming is hard. Asynchronous programming is hard.
@ -356,9 +361,9 @@ You must initiate a second write only when the first is finished, for example:
include::code:example$src/main/java/org/eclipse/jetty/docs/programming/ContentDocs.java[tags=sinkMany] include::code:example$src/main/java/org/eclipse/jetty/docs/programming/ContentDocs.java[tags=sinkMany]
---- ----
When you need to perform an unknown number of writes, you must use an `IteratingCallback`, explained in <<echo,this section>>, to avoid ``StackOverFlowError``s. When you need to perform an unknown number of writes, you may use an `IteratingCallback`, explained in <<echo,this section>>, to avoid ``StackOverFlowError``s.
For example, to copy from a `Content.Source` to a `Content.Sink` you should use the convenience method `Content.copy(Content.Source, Content.Sink, Callback)`. For example, to copy from a `Content.Source` to a `Content.Sink` you could use the convenience method `Content.copy(Content.Source, Content.Sink, Callback)`.
For illustrative purposes, below you can find the implementation of `copy(Content.Source, Content.Sink, Callback)` that uses an `IteratingCallback`: For illustrative purposes, below you can find the implementation of `copy(Content.Source, Content.Sink, Callback)` that uses an `IteratingCallback`:
[,java,indent=0] [,java,indent=0]

16
jetty-core/jetty-client/src/main/java/org/eclipse/jetty/client/transport/HttpSender.java
View File

@ -617,14 +617,8 @@ public abstract class HttpSender
} }
@Override @Override
protected void onCompleteFailure(Throwable x) protected void onFailure(Throwable x)
{ {
if (chunk != null)
{
chunk.release();
chunk = Content.Chunk.next(chunk);
}
failRequest(x); failRequest(x);
internalAbort(exchange, x); internalAbort(exchange, x);
@ -633,6 +627,14 @@ public abstract class HttpSender
promise.succeeded(true); promise.succeeded(true);
} }
@Override
protected void onCompleteFailure(Throwable x)
{
if (chunk != null)
chunk.release();
chunk = Content.Chunk.next(chunk);
}
@Override @Override
public InvocationType getInvocationType() public InvocationType getInvocationType()
{ {

27
jetty-core/jetty-client/src/main/java/org/eclipse/jetty/client/transport/internal/HttpSenderOverHTTP.java
View File

@ -26,6 +26,7 @@ import org.eclipse.jetty.http.MetaData;
import org.eclipse.jetty.io.ByteBufferPool; import org.eclipse.jetty.io.ByteBufferPool;
import org.eclipse.jetty.io.Content; import org.eclipse.jetty.io.Content;
import org.eclipse.jetty.io.EndPoint; import org.eclipse.jetty.io.EndPoint;
import org.eclipse.jetty.io.Retainable;
import org.eclipse.jetty.io.RetainableByteBuffer; import org.eclipse.jetty.io.RetainableByteBuffer;
import org.eclipse.jetty.util.BufferUtil; import org.eclipse.jetty.util.BufferUtil;
import org.eclipse.jetty.util.Callback; import org.eclipse.jetty.util.Callback;
@ -237,7 +238,9 @@ public class HttpSenderOverHTTP extends HttpSender
@Override @Override
protected void onSuccess() protected void onSuccess()
{ {
release(); headerBuffer = Retainable.release(headerBuffer);
chunkBuffer = Retainable.release(chunkBuffer);
contentByteBuffer = null;
} }
@Override @Override
@ -248,21 +251,16 @@ public class HttpSenderOverHTTP extends HttpSender
} }
@Override @Override
protected void onCompleteFailure(Throwable cause) protected void onFailure(Throwable cause)
{ {
super.onCompleteFailure(cause);
release();
callback.failed(cause); callback.failed(cause);
} }
private void release() @Override
protected void onCompleteFailure(Throwable cause)
{ {
if (headerBuffer != null) headerBuffer = Retainable.release(headerBuffer);
headerBuffer.release(); chunkBuffer = Retainable.release(chunkBuffer);
headerBuffer = null;
if (chunkBuffer != null)
chunkBuffer.release();
chunkBuffer = null;
contentByteBuffer = null; contentByteBuffer = null;
} }
} }
@ -334,11 +332,16 @@ public class HttpSenderOverHTTP extends HttpSender
} }
} }
@Override
protected void onFailure(Throwable cause)
{
callback.failed(cause);
}
@Override @Override
protected void onCompleteFailure(Throwable cause) protected void onCompleteFailure(Throwable cause)
{ {
release(); release();
callback.failed(cause);
} }
private void release() private void release()

50
jetty-core/jetty-fcgi/jetty-fcgi-client/src/main/java/org/eclipse/jetty/fcgi/generator/Flusher.java
View File

@ -15,6 +15,7 @@ package org.eclipse.jetty.fcgi.generator;
import java.nio.ByteBuffer; import java.nio.ByteBuffer;
import java.util.ArrayDeque; import java.util.ArrayDeque;
import java.util.ArrayList;
import java.util.List; import java.util.List;
import java.util.Queue; import java.util.Queue;
@ -104,43 +105,60 @@ public class Flusher
protected void onSuccess() protected void onSuccess()
{ {
if (active != null) if (active != null)
{
active.release();
active.succeeded(); active.succeeded();
active = null; active = null;
}
} }
@Override @Override
public void onCompleteFailure(Throwable x) public void onFailure(Throwable cause)
{ {
if (active != null) if (active != null)
active.failed(x); active.failed(cause);
active = null; List<Entry> entries;
try (AutoLock ignored = lock.lock())
while (true)
{ {
Entry entry = poll(); entries = new ArrayList<>(queue);
if (entry == null)
break;
entry.failed(x);
} }
entries.forEach(entry -> entry.failed(cause));
}
@Override
protected void onCompleteFailure(Throwable cause)
{
if (active != null)
{
active.release();
active = null;
}
List<Entry> entries;
try (AutoLock ignored = lock.lock())
{
entries = new ArrayList<>(queue);
queue.clear();
}
entries.forEach(Entry::release);
} }
} }
private record Entry(ByteBufferPool.Accumulator accumulator, Callback callback) implements Callback private record Entry(ByteBufferPool.Accumulator accumulator, Callback callback)
{ {
@Override
public void succeeded() public void succeeded()
{ {
if (accumulator != null)
accumulator.release();
callback.succeeded(); callback.succeeded();
} }
@Override
public void failed(Throwable x) public void failed(Throwable x)
{
callback.failed(x);
}
private void release()
{ {
if (accumulator != null) if (accumulator != null)
accumulator.release(); accumulator.release();
callback.failed(x);
} }
} }
} }

3
jetty-core/jetty-fcgi/jetty-fcgi-server/src/main/java/org/eclipse/jetty/fcgi/server/internal/ServerFCGIConnection.java
View File

@ -259,8 +259,7 @@ public class ServerFCGIConnection extends AbstractMetaDataConnection implements
boolean released = inputBuffer.release(); boolean released = inputBuffer.release();
if (LOG.isDebugEnabled()) if (LOG.isDebugEnabled())
LOG.debug("releaseInputBuffer {} {}", released, this); LOG.debug("releaseInputBuffer {} {}", released, this);
if (released) inputBuffer = null;
inputBuffer = null;
} }
private int fillInputBuffer() private int fillInputBuffer()

10
jetty-core/jetty-http2/jetty-http2-common/src/main/java/org/eclipse/jetty/http2/internal/HTTP2Flusher.java
View File

@ -344,10 +344,8 @@ public class HTTP2Flusher extends IteratingCallback implements Dumpable
} }
@Override @Override
protected void onCompleteFailure(Throwable x) protected void onFailure(Throwable x)
{ {
release();
Throwable closed; Throwable closed;
Set<HTTP2Session.Entry> allEntries; Set<HTTP2Session.Entry> allEntries;
try (AutoLock ignored = lock.lock()) try (AutoLock ignored = lock.lock())
@ -376,6 +374,12 @@ public class HTTP2Flusher extends IteratingCallback implements Dumpable
session.onWriteFailure(x); session.onWriteFailure(x);
} }
@Override
protected void onCompleteFailure(Throwable x)
{
release();
}
public void terminate(Throwable cause) public void terminate(Throwable cause)
{ {
Throwable closed; Throwable closed;

4
jetty-core/jetty-http2/jetty-http2-tests/src/test/java/org/eclipse/jetty/http2/tests/RawHTTP2ProxyTest.java
View File

@ -519,7 +519,7 @@ public class RawHTTP2ProxyTest
} }
@Override @Override
protected void onCompleteFailure(Throwable cause) protected void onFailure(Throwable cause)
{ {
frameInfo.callback.failed(cause); frameInfo.callback.failed(cause);
} }
@ -673,7 +673,7 @@ public class RawHTTP2ProxyTest
} }
@Override @Override
protected void onCompleteFailure(Throwable cause) protected void onFailure(Throwable cause)
{ {
frameInfo.callback.failed(cause); frameInfo.callback.failed(cause);
} }

10
jetty-core/jetty-http3/jetty-http3-common/src/main/java/org/eclipse/jetty/http3/ControlFlusher.java
View File

@ -122,13 +122,11 @@ public class ControlFlusher extends IteratingCallback
} }
@Override @Override
protected void onCompleteFailure(Throwable failure) protected void onFailure(Throwable failure)
{ {
if (LOG.isDebugEnabled()) if (LOG.isDebugEnabled())
LOG.debug("failed to write {} on {}", entries, this, failure); LOG.debug("failed to write {} on {}", entries, this, failure);
accumulator.release();
List<Entry> allEntries = new ArrayList<>(entries); List<Entry> allEntries = new ArrayList<>(entries);
entries.clear(); entries.clear();
try (AutoLock ignored = lock.lock()) try (AutoLock ignored = lock.lock())
@ -147,6 +145,12 @@ public class ControlFlusher extends IteratingCallback
endPoint.getQuicSession().getProtocolSession().outwardClose(error, "control_stream_failure"); endPoint.getQuicSession().getProtocolSession().outwardClose(error, "control_stream_failure");
} }
@Override
protected void onCompleteFailure(Throwable cause)
{
accumulator.release();
}
@Override @Override
public InvocationType getInvocationType() public InvocationType getInvocationType()
{ {

10
jetty-core/jetty-http3/jetty-http3-common/src/main/java/org/eclipse/jetty/http3/InstructionFlusher.java
View File

@ -118,13 +118,11 @@ public class InstructionFlusher extends IteratingCallback
} }
@Override @Override
protected void onCompleteFailure(Throwable failure) protected void onFailure(Throwable failure)
{ {
if (LOG.isDebugEnabled()) if (LOG.isDebugEnabled())
LOG.debug("failed to write buffers on {}", this, failure); LOG.debug("failed to write buffers on {}", this, failure);
accumulator.release();
try (AutoLock ignored = lock.lock()) try (AutoLock ignored = lock.lock())
{ {
terminated = failure; terminated = failure;
@ -138,6 +136,12 @@ public class InstructionFlusher extends IteratingCallback
endPoint.getQuicSession().getProtocolSession().outwardClose(error, "instruction_stream_failure"); endPoint.getQuicSession().getProtocolSession().outwardClose(error, "instruction_stream_failure");
} }
@Override
protected void onCompleteFailure(Throwable cause)
{
accumulator.release();
}
@Override @Override
public InvocationType getInvocationType() public InvocationType getInvocationType()
{ {

10
jetty-core/jetty-http3/jetty-http3-common/src/main/java/org/eclipse/jetty/http3/MessageFlusher.java
View File

@ -118,13 +118,11 @@ public class MessageFlusher extends IteratingCallback
} }
@Override @Override
protected void onCompleteFailure(Throwable cause) protected void onFailure(Throwable cause)
{ {
if (LOG.isDebugEnabled()) if (LOG.isDebugEnabled())
LOG.debug("failed to write {} on {}", entry, this, cause); LOG.debug("failed to write {} on {}", entry, this, cause);
accumulator.release();
if (entry != null) if (entry != null)
{ {
entry.callback.failed(cause); entry.callback.failed(cause);
@ -132,6 +130,12 @@ public class MessageFlusher extends IteratingCallback
} }
} }
@Override
protected void onCompleteFailure(Throwable cause)
{
accumulator.release();
}
@Override @Override
public InvocationType getInvocationType() public InvocationType getInvocationType()
{ {

14
jetty-core/jetty-io/src/main/java/org/eclipse/jetty/io/AbstractConnection.java
View File

@ -145,6 +145,20 @@ public abstract class AbstractConnection implements Connection, Invocable
getEndPoint().fillInterested(_readCallback); getEndPoint().fillInterested(_readCallback);
} }
/**
* <p>Utility method to be called to register read interest.</p>
* <p>After a call to this method, {@link #onFillable()} or {@link #onFillInterestedFailed(Throwable)}
* will be called back as appropriate.</p>
*
* @see #onFillable()
*/
public void fillInterested(Callback callback)
{
if (LOG.isDebugEnabled())
LOG.debug("fillInterested {} {}", callback, this);
getEndPoint().fillInterested(callback);
}
public void tryFillInterested(Callback callback) public void tryFillInterested(Callback callback)
{ {
getEndPoint().tryFillInterested(callback); getEndPoint().tryFillInterested(callback);

4
jetty-core/jetty-io/src/main/java/org/eclipse/jetty/io/ArrayByteBufferPool.java
View File

@ -224,7 +224,7 @@ public class ArrayByteBufferPool implements ByteBufferPool, Dumpable
} }
@Override @Override
public boolean removeAndRelease(RetainableByteBuffer buffer) public boolean releaseAndRemove(RetainableByteBuffer buffer)
{ {
RetainableByteBuffer actual = buffer; RetainableByteBuffer actual = buffer;
while (actual instanceof RetainableByteBuffer.Wrapper wrapper) while (actual instanceof RetainableByteBuffer.Wrapper wrapper)
@ -244,7 +244,7 @@ public class ArrayByteBufferPool implements ByteBufferPool, Dumpable
return buffer.release(); return buffer.release();
} }
return ByteBufferPool.super.removeAndRelease(buffer); return ByteBufferPool.super.releaseAndRemove(buffer);
} }
private void reserve(RetainedBucket bucket, ByteBuffer byteBuffer) private void reserve(RetainedBucket bucket, ByteBuffer byteBuffer)

12
jetty-core/jetty-io/src/main/java/org/eclipse/jetty/io/ByteBufferPool.java
View File

@ -64,11 +64,9 @@ public interface ByteBufferPool
* If the buffer is not in a pool, calling this method is equivalent to calling {@link RetainableByteBuffer#release()}. * If the buffer is not in a pool, calling this method is equivalent to calling {@link RetainableByteBuffer#release()}.
* Calling this method satisfies any contract that requires a call to {@link RetainableByteBuffer#release()}. * Calling this method satisfies any contract that requires a call to {@link RetainableByteBuffer#release()}.
* @return {@code true} if a call to {@link RetainableByteBuffer#release()} would have returned {@code true}. * @return {@code true} if a call to {@link RetainableByteBuffer#release()} would have returned {@code true}.
* @see RetainableByteBuffer#release() * @see RetainableByteBuffer#releaseAndRemove()
* @deprecated This API is experimental and may be removed in future releases
*/ */
@Deprecated default boolean releaseAndRemove(RetainableByteBuffer buffer)
default boolean removeAndRelease(RetainableByteBuffer buffer)
{ {
return buffer != null && buffer.release(); return buffer != null && buffer.release();
} }
@ -96,6 +94,12 @@ public interface ByteBufferPool
return wrapped; return wrapped;
} }
@Override
public boolean releaseAndRemove(RetainableByteBuffer buffer)
{
return getWrapped().releaseAndRemove(buffer);
}
@Override @Override
public RetainableByteBuffer.Mutable acquire(int size, boolean direct) public RetainableByteBuffer.Mutable acquire(int size, boolean direct)
{ {

21
jetty-core/jetty-io/src/main/java/org/eclipse/jetty/io/Content.java
View File

@ -1080,6 +1080,27 @@ public class Content
return null; return null;
} }
/**
* Convenience method to release a chunk and return {@link #next(Chunk)}.
* Equivalent to:
* <pre>{@code
* if (chunk != null)
* {
* chunk.release();
* chunk = Chunk.next(chunk);
* }
* }</pre>
* @param chunk The chunk to release or {@code null}
* @return The {@link #next(Chunk)} chunk;
*/
static Chunk releaseAndNext(Chunk chunk)
{
if (chunk == null)
return null;
chunk.release();
return next(chunk);
}
/** /**
* @param chunk The chunk to test for an {@link Chunk#getFailure() failure}. * @param chunk The chunk to test for an {@link Chunk#getFailure() failure}.
* @return True if the chunk is non-null and {@link Chunk#getFailure() chunk.getError()} returns non-null. * @return True if the chunk is non-null and {@link Chunk#getFailure() chunk.getError()} returns non-null.

12
jetty-core/jetty-io/src/main/java/org/eclipse/jetty/io/IOResources.java
View File

@ -321,12 +321,18 @@ public class IOResources
} }
@Override @Override
protected void onCompleteFailure(Throwable x) protected void onFailure(Throwable x)
{
IO.close(channel);
super.onFailure(x);
}
@Override
protected void onCompleteFailure(Throwable cause)
{ {
if (retainableByteBuffer != null) if (retainableByteBuffer != null)
retainableByteBuffer.release(); retainableByteBuffer.release();
IO.close(channel); super.onCompleteFailure(cause);
super.onCompleteFailure(x);
} }
} }
} }

26
jetty-core/jetty-io/src/main/java/org/eclipse/jetty/io/Retainable.java
View File

@ -247,4 +247,30 @@ public interface Retainable
return String.format("%s@%x[r=%d]", getClass().getSimpleName(), hashCode(), get()); return String.format("%s@%x[r=%d]", getClass().getSimpleName(), hashCode(), get());
} }
} }
/**
* Convenience method that replaces code like:
* <pre>{@code
* if (buffer != null)
* {
* buffer.release();
* buffer = null;
* }
* }
* </pre>
* with:
* <pre>{@code
* buffer = Retainable.release(buffer);
* }
* </pre>
* @param retainable The retainable to release, if not {@code null}.
* @param <R> The type of the retainable
* @return always returns {@code null}
*/
static <R extends Retainable> R release(R retainable)
{
if (retainable != null)
retainable.release();
return null;
}
} }

59
jetty-core/jetty-io/src/main/java/org/eclipse/jetty/io/RetainableByteBuffer.java
View File

@ -164,6 +164,19 @@ public interface RetainableByteBuffer extends Retainable
throw new ReadOnlyBufferException(); throw new ReadOnlyBufferException();
} }
/**
* {@link #release() Releases} the buffer in a way that ensures it will not be recycled in a buffer pool.
* This method should be used in cases where it is unclear if operations on the buffer have completed
* (for example, when a write operation has been aborted asynchronously or timed out, but the write
* operation may still be pending).
* @return whether if the buffer was released.
* @see ByteBufferPool#releaseAndRemove(RetainableByteBuffer)
*/
default boolean releaseAndRemove()
{
return release();
}
/** /**
* Appends and consumes the contents of this buffer to the passed buffer, limited by the capacity of the target buffer. * Appends and consumes the contents of this buffer to the passed buffer, limited by the capacity of the target buffer.
* @param buffer The buffer to append bytes to, whose limit will be updated. * @param buffer The buffer to append bytes to, whose limit will be updated.
@ -657,6 +670,12 @@ public interface RetainableByteBuffer extends Retainable
return (RetainableByteBuffer)super.getWrapped(); return (RetainableByteBuffer)super.getWrapped();
} }
@Override
public boolean releaseAndRemove()
{
return getWrapped().releaseAndRemove();
}
@Override @Override
public boolean isRetained() public boolean isRetained()
{ {
@ -1301,6 +1320,12 @@ public interface RetainableByteBuffer extends Retainable
_pool = pool; _pool = pool;
} }
@Override
public boolean releaseAndRemove()
{
return _pool.releaseAndRemove(this);
}
@Override @Override
public RetainableByteBuffer slice(long length) public RetainableByteBuffer slice(long length)
{ {
@ -1939,6 +1964,22 @@ public interface RetainableByteBuffer extends Retainable
return false; return false;
} }
@Override
public boolean releaseAndRemove()
{
if (LOG.isDebugEnabled())
LOG.debug("release {}", this);
if (super.release())
{
for (RetainableByteBuffer buffer : _buffers)
buffer.releaseAndRemove();
_buffers.clear();
_aggregate = null;
return true;
}
return false;
}
@Override @Override
public void clear() public void clear()
{ {
@ -2333,10 +2374,6 @@ public interface RetainableByteBuffer extends Retainable
@Override @Override
protected Action process() protected Action process()
{ {
// release the last buffer written
if (_buffer != null)
_buffer.release();
// write next buffer // write next buffer
if (_index < _buffers.size()) if (_index < _buffers.size())
{ {
@ -2357,6 +2394,20 @@ public interface RetainableByteBuffer extends Retainable
_buffers.clear(); _buffers.clear();
return Action.SUCCEEDED; return Action.SUCCEEDED;
} }
@Override
protected void onSuccess()
{
// release the last buffer written
_buffer = Retainable.release(_buffer);
}
@Override
protected void onCompleteFailure(Throwable x)
{
// release the last buffer written
_buffer = Retainable.release(_buffer);
}
}.iterate(); }.iterate();
} }
} }

2
jetty-core/jetty-io/src/main/java/org/eclipse/jetty/io/internal/ByteBufferChunk.java
View File

@ -145,7 +145,7 @@ public abstract class ByteBufferChunk extends RetainableByteBuffer.FixedCapacity
public WithRetainable(ByteBuffer byteBuffer, boolean last, Retainable retainable) public WithRetainable(ByteBuffer byteBuffer, boolean last, Retainable retainable)
{ {
super(byteBuffer, last); super(byteBuffer, last);
this.retainable = retainable; this.retainable = Objects.requireNonNull(retainable);
} }
@Override @Override

38
jetty-core/jetty-io/src/main/java/org/eclipse/jetty/io/internal/ContentCopier.java
View File

@ -27,7 +27,7 @@ public class ContentCopier extends IteratingNestedCallback
private final Content.Source source; private final Content.Source source;
private final Content.Sink sink; private final Content.Sink sink;
private final Content.Chunk.Processor chunkProcessor; private final Content.Chunk.Processor chunkProcessor;
private Content.Chunk current; private Content.Chunk chunk;
private boolean terminated; private boolean terminated;
public ContentCopier(Content.Source source, Content.Sink sink, Content.Chunk.Processor chunkProcessor, Callback callback) public ContentCopier(Content.Source source, Content.Sink sink, Content.Chunk.Processor chunkProcessor, Callback callback)
@ -47,43 +47,47 @@ public class ContentCopier extends IteratingNestedCallback
@Override @Override
protected Action process() throws Throwable protected Action process() throws Throwable
{ {
if (current != null)
current.release();
if (terminated) if (terminated)
return Action.SUCCEEDED; return Action.SUCCEEDED;
current = source.read(); chunk = source.read();
if (current == null) if (chunk == null)
{ {
source.demand(this::succeeded); source.demand(this::succeeded);
return Action.SCHEDULED; return Action.SCHEDULED;
} }
if (chunkProcessor != null && chunkProcessor.process(current, this)) if (chunkProcessor != null && chunkProcessor.process(chunk, this))
return Action.SCHEDULED; return Action.SCHEDULED;
terminated = current.isLast(); terminated = chunk.isLast();
if (Content.Chunk.isFailure(current)) if (Content.Chunk.isFailure(chunk))
{ {
failed(current.getFailure()); failed(chunk.getFailure());
return Action.SCHEDULED; return Action.SCHEDULED;
} }
sink.write(current.isLast(), current.getByteBuffer(), this); sink.write(chunk.isLast(), chunk.getByteBuffer(), this);
return Action.SCHEDULED; return Action.SCHEDULED;
} }
@Override
protected void onSuccess()
{
chunk = Content.Chunk.releaseAndNext(chunk);
}
@Override
protected void onFailure(Throwable cause)
{
ExceptionUtil.callAndThen(cause, source::fail, super::onFailure);
}
@Override @Override
protected void onCompleteFailure(Throwable x) protected void onCompleteFailure(Throwable x)
{ {
if (current != null) chunk = Content.Chunk.releaseAndNext(chunk);
{
current.release();
current = Content.Chunk.next(current);
}
ExceptionUtil.callAndThen(x, source::fail, super::onCompleteFailure);
} }
} }

8
jetty-core/jetty-io/src/test/java/org/eclipse/jetty/io/ArrayByteBufferPoolTest.java
View File

@ -448,7 +448,7 @@ public class ArrayByteBufferPoolTest
} }
@Test @Test
public void testRemoveAndRelease() public void testReleaseAndRemove()
{ {
ArrayByteBufferPool pool = new ArrayByteBufferPool(); ArrayByteBufferPool pool = new ArrayByteBufferPool();
@ -471,9 +471,9 @@ public class ArrayByteBufferPoolTest
retained1 = pool.acquire(1024, false); retained1 = pool.acquire(1024, false);
retained1.retain(); retained1.retain();
assertTrue(pool.removeAndRelease(reserved1)); assertTrue(reserved1.releaseAndRemove());
assertTrue(pool.removeAndRelease(acquired1)); assertTrue(acquired1.releaseAndRemove());
assertFalse(pool.removeAndRelease(retained1)); assertFalse(retained1.releaseAndRemove());
assertTrue(retained1.release()); assertTrue(retained1.release());
assertThat(pool.getHeapByteBufferCount(), is(2L)); assertThat(pool.getHeapByteBufferCount(), is(2L));

2
jetty-core/jetty-quic/jetty-quic-common/src/main/java/org/eclipse/jetty/quic/common/QuicConnection.java
View File

@ -377,7 +377,7 @@ public abstract class QuicConnection extends AbstractConnection
} }
@Override @Override
protected void onCompleteFailure(Throwable cause) protected void onFailure(Throwable cause)
{ {
entry.callback.failed(cause); entry.callback.failed(cause);
QuicConnection.this.close(); QuicConnection.this.close();

21
jetty-core/jetty-quic/jetty-quic-common/src/main/java/org/eclipse/jetty/quic/common/QuicSession.java
View File

@ -38,6 +38,7 @@ import org.eclipse.jetty.io.ByteBufferPool;
import org.eclipse.jetty.io.Connection; import org.eclipse.jetty.io.Connection;
import org.eclipse.jetty.io.CyclicTimeout; import org.eclipse.jetty.io.CyclicTimeout;
import org.eclipse.jetty.io.EndPoint; import org.eclipse.jetty.io.EndPoint;
import org.eclipse.jetty.io.Retainable;
import org.eclipse.jetty.io.RetainableByteBuffer; import org.eclipse.jetty.io.RetainableByteBuffer;
import org.eclipse.jetty.quic.quiche.QuicheConnection; import org.eclipse.jetty.quic.quiche.QuicheConnection;
import org.eclipse.jetty.quic.quiche.QuicheConnectionId; import org.eclipse.jetty.quic.quiche.QuicheConnectionId;
@ -533,7 +534,7 @@ public abstract class QuicSession extends ContainerLifeCycle
{ {
if (LOG.isDebugEnabled()) if (LOG.isDebugEnabled())
LOG.debug("written cipher bytes on {}", QuicSession.this); LOG.debug("written cipher bytes on {}", QuicSession.this);
cipherBuffer.release(); cipherBuffer = Retainable.release(cipherBuffer);
} }
@Override @Override
@ -547,23 +548,25 @@ public abstract class QuicSession extends ContainerLifeCycle
{ {
if (LOG.isDebugEnabled()) if (LOG.isDebugEnabled())
LOG.debug("connection closed {}", QuicSession.this); LOG.debug("connection closed {}", QuicSession.this);
finish(new ClosedChannelException()); cipherBuffer = Retainable.release(cipherBuffer);
finishOutwardClose(new ClosedChannelException());
timeout.destroy();
} }
@Override @Override
protected void onCompleteFailure(Throwable failure) protected void onFailure(Throwable failure)
{ {
if (LOG.isDebugEnabled()) if (LOG.isDebugEnabled())
LOG.debug("failed to write cipher bytes, closing session on {}", QuicSession.this, failure); LOG.debug("failed to write cipher bytes, closing session on {}", QuicSession.this, failure);
finish(failure);
}
private void finish(Throwable failure)
{
cipherBuffer.release();
finishOutwardClose(failure); finishOutwardClose(failure);
timeout.destroy(); timeout.destroy();
} }
@Override
protected void onCompleteFailure(Throwable cause)
{
cipherBuffer = Retainable.release(cipherBuffer);
}
} }
/** /**

31
jetty-core/jetty-server/src/main/java/org/eclipse/jetty/server/handler/ConnectHandler.java
View File

@ -37,6 +37,7 @@ import org.eclipse.jetty.io.ByteBufferPool;
import org.eclipse.jetty.io.Connection; import org.eclipse.jetty.io.Connection;
import org.eclipse.jetty.io.EndPoint; import org.eclipse.jetty.io.EndPoint;
import org.eclipse.jetty.io.ManagedSelector; import org.eclipse.jetty.io.ManagedSelector;
import org.eclipse.jetty.io.Retainable;
import org.eclipse.jetty.io.RetainableByteBuffer; import org.eclipse.jetty.io.RetainableByteBuffer;
import org.eclipse.jetty.io.SelectorManager; import org.eclipse.jetty.io.SelectorManager;
import org.eclipse.jetty.io.SocketChannelEndPoint; import org.eclipse.jetty.io.SocketChannelEndPoint;
@ -736,18 +737,17 @@ public class ConnectHandler extends Handler.Wrapper
write(connection.getEndPoint(), byteBuffer, this); write(connection.getEndPoint(), byteBuffer, this);
return Action.SCHEDULED; return Action.SCHEDULED;
} }
else if (filled == 0)
buffer = Retainable.release(buffer);
if (filled == 0)
{ {
buffer.release(); fillInterested(this);
fillInterested(); return Action.SCHEDULED;
return Action.IDLE;
}
else
{
buffer.release();
connection.getEndPoint().shutdownOutput();
return Action.SUCCEEDED;
} }
connection.getEndPoint().shutdownOutput();
return Action.SUCCEEDED;
} }
catch (IOException x) catch (IOException x)
{ {
@ -764,18 +764,23 @@ public class ConnectHandler extends Handler.Wrapper
{ {
if (LOG.isDebugEnabled()) if (LOG.isDebugEnabled())
LOG.debug("Wrote {} bytes {}", filled, TunnelConnection.this); LOG.debug("Wrote {} bytes {}", filled, TunnelConnection.this);
buffer.release(); buffer = Retainable.release(buffer);
} }
@Override @Override
protected void onCompleteFailure(Throwable x) protected void onFailure(Throwable x)
{ {
if (LOG.isDebugEnabled()) if (LOG.isDebugEnabled())
LOG.debug("Failed to write {} bytes {}", filled, TunnelConnection.this, x); LOG.debug("Failed to write {} bytes {}", filled, TunnelConnection.this, x);
buffer.release();
disconnect(x); disconnect(x);
} }
@Override
protected void onCompleteFailure(Throwable cause)
{
buffer = Retainable.release(buffer);
}
private void disconnect(Throwable x) private void disconnect(Throwable x)
{ {
TunnelConnection.this.close(x); TunnelConnection.this.close(x);

18
jetty-core/jetty-server/src/main/java/org/eclipse/jetty/server/handler/ErrorHandler.java
View File

@ -254,14 +254,14 @@ public class ErrorHandler implements Request.Handler
} }
response.getHeaders().put(type.getContentTypeField(charset)); response.getHeaders().put(type.getContentTypeField(charset));
response.write(true, buffer.getByteBuffer(), new WriteErrorCallback(callback, byteBufferPool, buffer)); response.write(true, buffer.getByteBuffer(), new WriteErrorCallback(callback, buffer));
return true; return true;
} }
catch (Throwable x) catch (Throwable x)
{ {
if (buffer != null) if (buffer != null)
byteBufferPool.removeAndRelease(buffer); buffer.releaseAndRemove();
throw x; throw x;
} }
} }
@ -586,13 +586,11 @@ public class ErrorHandler implements Request.Handler
private static class WriteErrorCallback implements Callback private static class WriteErrorCallback implements Callback
{ {
private final AtomicReference<Callback> _callback; private final AtomicReference<Callback> _callback;
private final ByteBufferPool _pool;
private final RetainableByteBuffer _buffer; private final RetainableByteBuffer _buffer;
public WriteErrorCallback(Callback callback, ByteBufferPool pool, RetainableByteBuffer retainable) public WriteErrorCallback(Callback callback, RetainableByteBuffer retainable)
{ {
_callback = new AtomicReference<>(callback); _callback = new AtomicReference<>(callback);
_pool = pool;
_buffer = retainable; _buffer = retainable;
} }
@ -600,7 +598,9 @@ public class ErrorHandler implements Request.Handler
public void succeeded() public void succeeded()
{ {
Callback callback = _callback.getAndSet(null); Callback callback = _callback.getAndSet(null);
if (callback != null) if (callback == null)
_buffer.release();
else
ExceptionUtil.callAndThen(_buffer::release, callback::succeeded); ExceptionUtil.callAndThen(_buffer::release, callback::succeeded);
} }
@ -608,8 +608,10 @@ public class ErrorHandler implements Request.Handler
public void failed(Throwable x) public void failed(Throwable x)
{ {
Callback callback = _callback.getAndSet(null); Callback callback = _callback.getAndSet(null);
if (callback != null) if (callback == null)
ExceptionUtil.callAndThen(x, t -> _pool.removeAndRelease(_buffer), callback::failed); _buffer.releaseAndRemove();
else
ExceptionUtil.callAndThen(x, t -> _buffer.releaseAndRemove(), callback::failed);
} }
} }
} }

14
jetty-core/jetty-server/src/main/java/org/eclipse/jetty/server/handler/gzip/GzipResponseAndCallback.java
View File

@ -321,13 +321,6 @@ public class GzipResponseAndCallback extends Response.Wrapper implements Callbac
LOG.debug("GzipBufferCB(complete={}, callback={}, content={})", complete, callback, BufferUtil.toDetailString(content)); LOG.debug("GzipBufferCB(complete={}, callback={}, content={})", complete, callback, BufferUtil.toDetailString(content));
} }
@Override
protected void onCompleteFailure(Throwable x)
{
cleanup();
super.onCompleteFailure(x);
}
@Override @Override
protected Action process() throws Exception protected Action process() throws Exception
{ {
@ -373,6 +366,13 @@ public class GzipResponseAndCallback extends Response.Wrapper implements Callbac
}; };
} }
@Override
protected void onCompleteFailure(Throwable x)
{
cleanup();
super.onCompleteFailure(x);
}
private void cleanup() private void cleanup()
{ {
if (_deflaterEntry != null) if (_deflaterEntry != null)

23
jetty-core/jetty-server/src/main/java/org/eclipse/jetty/server/internal/HttpConnection.java
View File

@ -878,15 +878,19 @@ public class HttpConnection extends AbstractMetaDataConnection implements Runnab
} }
} }
private Callback release() private Callback resetCallback()
{ {
Callback complete = _callback; Callback complete = _callback;
_callback = null; _callback = null;
_info = null; _info = null;
_content = null; _content = null;
return complete;
}
private void release()
{
releaseHeader(); releaseHeader();
releaseChunk(); releaseChunk();
return complete;
} }
private void releaseHeader() private void releaseHeader()
@ -906,13 +910,22 @@ public class HttpConnection extends AbstractMetaDataConnection implements Runnab
@Override @Override
protected void onCompleteSuccess() protected void onCompleteSuccess()
{ {
release().succeeded(); Callback callback = resetCallback();
release();
callback.succeeded();
} }
@Override @Override
public void onCompleteFailure(final Throwable x) public void onFailure(final Throwable x)
{ {
failedCallback(release(), x); Callback callback = resetCallback();
failedCallback(callback, x);
}
@Override
protected void onCompleteFailure(Throwable cause)
{
release();
} }
@Override @Override

11
jetty-core/jetty-tests/jetty-test-client-transports/src/test/java/org/eclipse/jetty/test/client/transport/AbstractTest.java
View File

@ -26,7 +26,6 @@ import java.util.concurrent.TimeUnit;
import java.util.stream.Stream; import java.util.stream.Stream;
import javax.management.MBeanServer; import javax.management.MBeanServer;
import org.awaitility.Awaitility;
import org.eclipse.jetty.alpn.server.ALPNServerConnectionFactory; import org.eclipse.jetty.alpn.server.ALPNServerConnectionFactory;
import org.eclipse.jetty.client.HttpClient; import org.eclipse.jetty.client.HttpClient;
import org.eclipse.jetty.client.HttpClientTransport; import org.eclipse.jetty.client.HttpClientTransport;
@ -66,7 +65,6 @@ import org.eclipse.jetty.util.SocketAddressResolver;
import org.eclipse.jetty.util.component.LifeCycle; import org.eclipse.jetty.util.component.LifeCycle;
import org.eclipse.jetty.util.ssl.SslContextFactory; import org.eclipse.jetty.util.ssl.SslContextFactory;
import org.eclipse.jetty.util.thread.QueuedThreadPool; import org.eclipse.jetty.util.thread.QueuedThreadPool;
import org.hamcrest.Matchers;
import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.Tag; import org.junit.jupiter.api.Tag;
import org.junit.jupiter.api.Tags; import org.junit.jupiter.api.Tags;
@ -75,6 +73,9 @@ import org.junit.jupiter.api.extension.BeforeTestExecutionCallback;
import org.junit.jupiter.api.extension.ExtendWith; import org.junit.jupiter.api.extension.ExtendWith;
import org.junit.jupiter.api.extension.RegisterExtension; import org.junit.jupiter.api.extension.RegisterExtension;
import static org.awaitility.Awaitility.await;
import static org.hamcrest.MatcherAssert.assertThat;
import static org.hamcrest.Matchers.is;
import static org.junit.jupiter.api.Assertions.fail; import static org.junit.jupiter.api.Assertions.fail;
@ExtendWith(WorkDirExtension.class) @ExtendWith(WorkDirExtension.class)
@ -129,9 +130,9 @@ public class AbstractTest
try try
{ {
if (serverBufferPool != null && !isLeakTrackingDisabled(testInfo, "server")) if (serverBufferPool != null && !isLeakTrackingDisabled(testInfo, "server"))
assertNoLeaks(serverBufferPool, testInfo, "server-", "\n---\nServer Leaks: " + serverBufferPool.dumpLeaks() + "---\n"); assertNoLeaks(serverBufferPool, testInfo, "server-", "Server Leaks: ");
if (clientBufferPool != null && !isLeakTrackingDisabled(testInfo, "client")) if (clientBufferPool != null && !isLeakTrackingDisabled(testInfo, "client"))
assertNoLeaks(clientBufferPool, testInfo, "client-", "\n---\nClient Leaks: " + clientBufferPool.dumpLeaks() + "---\n"); assertNoLeaks(clientBufferPool, testInfo, "client-", "Client Leaks: ");
} }
finally finally
{ {
@ -149,7 +150,7 @@ public class AbstractTest
{ {
try try
{ {
Awaitility.await().atMost(3, TimeUnit.SECONDS).until(() -> bufferPool.getLeaks().size(), Matchers.is(0)); await().atMost(3, TimeUnit.SECONDS).untilAsserted(() -> assertThat("\n---\n" + msg + bufferPool.dumpLeaks(), bufferPool.getLeaks().size(), is(0)));
} }
catch (Exception e) catch (Exception e)
{ {

4
jetty-core/jetty-tests/jetty-test-client-transports/src/test/java/org/eclipse/jetty/test/client/transport/CustomTransportTest.java
View File

@ -335,7 +335,7 @@ public class CustomTransportTest
} }
@Override @Override
protected void onCompleteFailure(Throwable cause) protected void onFailure(Throwable cause)
{ {
// There was a write error, close the Gateway Channel. // There was a write error, close the Gateway Channel.
channel.close(cause); channel.close(cause);
@ -378,7 +378,7 @@ public class CustomTransportTest
} }
@Override @Override
protected void onCompleteFailure(Throwable cause) protected void onFailure(Throwable cause)
{ {
// There was a write error, close the Gateway Channel. // There was a write error, close the Gateway Channel.
channel.close(cause); channel.close(cause);

663
jetty-core/jetty-util/src/main/java/org/eclipse/jetty/util/IteratingCallback.java
View File

@ -14,8 +14,12 @@
package org.eclipse.jetty.util; package org.eclipse.jetty.util;
import java.io.IOException; import java.io.IOException;
import java.util.Objects;
import java.util.function.Consumer;
import org.eclipse.jetty.util.thread.AutoLock; import org.eclipse.jetty.util.thread.AutoLock;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/** /**
* This specialized callback implements a pattern that allows * This specialized callback implements a pattern that allows
@ -51,10 +55,12 @@ import org.eclipse.jetty.util.thread.AutoLock;
*/ */
public abstract class IteratingCallback implements Callback public abstract class IteratingCallback implements Callback
{ {
private static final Logger LOG = LoggerFactory.getLogger(IteratingCallback.class);
/** /**
* The internal states of this callback. * The internal states of this callback.
*/ */
private enum State enum State
{ {
/** /**
* This callback is idle, ready to iterate. * This callback is idle, ready to iterate.
@ -64,48 +70,35 @@ public abstract class IteratingCallback implements Callback
/** /**
* This callback is just about to call {@link #process()}, * This callback is just about to call {@link #process()},
* or within it, or just exited from it, either normally * or within it, or just exited from it, either normally
* or by throwing. * or by throwing. Further actions are waiting for the
* {@link #process()} method to return.
*/ */
PROCESSING, PROCESSING,
/**
* The asynchronous sub-task was completed either with
* a call to {@link #succeeded()} or {@link #failed(Throwable)}, whilst in
* {@link #PROCESSING} state. Further actions are waiting for the
* {@link #process()} method to return.
*/
PROCESSING_CALLED,
/** /**
* Method {@link #process()} returned {@link Action#SCHEDULED} * Method {@link #process()} returned {@link Action#SCHEDULED}
* and this callback is waiting for the asynchronous sub-task * and this callback is waiting for the asynchronous sub-task
* to complete. * to complete via a callback to {@link #succeeded()} or {@link #failed(Throwable)}
*/ */
PENDING, PENDING,
/** /**
* The asynchronous sub-task was completed successfully * This callback is complete.
* via a call to {@link #succeeded()} while in
* {@link #PROCESSING} state.
*/ */
CALLED, COMPLETE,
/** /**
* The iteration terminated successfully as indicated by * Complete and can't be reset.
* {@link Action#SUCCEEDED} returned from
* {@link IteratingCallback#process()}.
*/ */
SUCCEEDED, CLOSED
/**
* The iteration terminated with a failure via a call
* to {@link IteratingCallback#failed(Throwable)}.
*/
FAILED,
/**
* This callback has been {@link #close() closed} and
* cannot be {@link #reset() reset}.
*/
CLOSED,
/**
* This callback has been {@link #abort(Throwable) aborted},
* and cannot be {@link #reset() reset}.
*/
ABORTED
} }
/** /**
@ -120,6 +113,7 @@ public abstract class IteratingCallback implements Callback
* for additional events to trigger more work. * for additional events to trigger more work.
*/ */
IDLE, IDLE,
/** /**
* Indicates that {@link #process()} has initiated an asynchronous * Indicates that {@link #process()} has initiated an asynchronous
* sub-task, where the execution has started but the callback * sub-task, where the execution has started but the callback
@ -127,6 +121,7 @@ public abstract class IteratingCallback implements Callback
* may have not yet been invoked. * may have not yet been invoked.
*/ */
SCHEDULED, SCHEDULED,
/** /**
* Indicates that {@link #process()} has completed the whole * Indicates that {@link #process()} has completed the whole
* iteration successfully. * iteration successfully.
@ -135,9 +130,13 @@ public abstract class IteratingCallback implements Callback
} }
private final AutoLock _lock = new AutoLock(); private final AutoLock _lock = new AutoLock();
private final Runnable _onSuccess = this::onSuccess;
private final Runnable _processing = this::processing;
private final Consumer<Throwable> _onCompleted = this::onCompleted;
private State _state; private State _state;
private Throwable _failure; private Throwable _failure;
private boolean _iterate; private boolean _reprocess;
private boolean _aborted;
protected IteratingCallback() protected IteratingCallback()
{ {
@ -146,7 +145,7 @@ public abstract class IteratingCallback implements Callback
protected IteratingCallback(boolean needReset) protected IteratingCallback(boolean needReset)
{ {
_state = needReset ? State.SUCCEEDED : State.IDLE; _state = needReset ? State.COMPLETE : State.IDLE;
} }
/** /**
@ -180,7 +179,32 @@ public abstract class IteratingCallback implements Callback
} }
/** /**
* Invoked when the overall task has completed successfully. * Invoked when the overall task has been {@link #abort(Throwable) aborted} or {@link #failed(Throwable) failed}.
* <p>
* Calls to this method are serialized with respect to {@link #onAborted(Throwable)}, {@link #process()},
* {@link #onCompleteFailure(Throwable)} and {@link #onCompleted(Throwable)}.
* <p>
* Because {@code onFailure} can be called due to an {@link #abort(Throwable)} or {@link #close()} operation, it is
* possible that any resources passed to a {@link Action#SCHEDULED} operation may still be in use, and thus should not
* be recycled by this call. For example any buffers passed to a write operation should not be returned to a buffer
* pool by implementations of {@code onFailure}. Such resources may be discarded here, or safely recycled in a
* subsequent call to {@link #onCompleted(Throwable)} or {@link #onCompleteFailure(Throwable)}, when
* the {@link Action#SCHEDULED} operation has completed.
* @param cause The cause of the failure or abort
* @see #onCompleted(Throwable)
* @see #onCompleteFailure(Throwable)
*/
protected void onFailure(Throwable cause)
{
}
/**
* Invoked when the overall task has completed successfully, specifically after any {@link Action#SCHEDULED} operations
* have {@link Callback#succeeded()} and {@link #process()} has returned {@link Action#SUCCEEDED}.
* <p>
* Calls to this method are serialized with respect to {@link #process()}, {@link #onAborted(Throwable)}
* and {@link #onCompleted(Throwable)}.
* If this method is called, then {@link #onCompleteFailure(Throwable)} ()} will never be called.
* *
* @see #onCompleteFailure(Throwable) * @see #onCompleteFailure(Throwable)
*/ */
@ -190,6 +214,10 @@ public abstract class IteratingCallback implements Callback
/** /**
* Invoked when the overall task has completed with a failure. * Invoked when the overall task has completed with a failure.
* <p>
* Calls to this method are serialized with respect to {@link #process()}, {@link #onAborted(Throwable)}
* and {@link #onCompleted(Throwable)}.
* If this method is called, then {@link #onCompleteSuccess()} will never be called.
* *
* @param cause the throwable to indicate cause of failure * @param cause the throwable to indicate cause of failure
* @see #onCompleteSuccess() * @see #onCompleteSuccess()
@ -198,6 +226,101 @@ public abstract class IteratingCallback implements Callback
{ {
} }
/**
* Invoked when the overall task has been aborted.
* <p>
* Calls to this method are serialized with respect to {@link #process()}, {@link #onCompleteFailure(Throwable)}
* and {@link #onCompleted(Throwable)}.
* If this method is called, then {@link #onCompleteSuccess()} will never be called.
* <p>
* The default implementation of this method calls {@link #failed(Throwable)}. Overridden implementations of
* this method SHOULD NOT call {@code super.onAborted(Throwable)}.
* <p>
* Because {@code onAborted} can be called due to an {@link #abort(Throwable)} or {@link #close()} operation, it is
* possible that any resources passed to a {@link Action#SCHEDULED} operation may still be in use, and thus should not
* be recycled by this call. For example any buffers passed to a write operation should not be returned to a buffer
* pool by implementations of {@code onFailure}. Such resources may be discarded here, or safely recycled in a
* subsequent call to {@link #onCompleted(Throwable)} or {@link #onCompleteFailure(Throwable)}, when
* the {@link Action#SCHEDULED} operation has completed.
* @param cause The cause of the abort
* @see #onCompleted(Throwable)
* @see #onCompleteFailure(Throwable)
*/
protected void onAborted(Throwable cause)
{
}
/**
* Invoked when the overall task has completed.
* <p>
* Calls to this method are serialized with respect to {@link #process()} and {@link #onAborted(Throwable)}.
* The default implementation of this method will call either {@link #onCompleteSuccess()} or {@link #onCompleteFailure(Throwable)}
* thus implementations of this method should always call {@code super.onCompleted(Throwable)}.
*
* @param causeOrNull the cause of any {@link #abort(Throwable) abort} or {@link #failed(Throwable) failure},
* else {@code null} for {@link #succeeded() success}.
*/
protected void onCompleted(Throwable causeOrNull)
{
if (causeOrNull == null)
onCompleteSuccess();
else
onCompleteFailure(causeOrNull);
}
private void doOnSuccessProcessing()
{
ExceptionUtil.callAndThen(_onSuccess, _processing);
}
private void doCompleteSuccess()
{
onCompleted(null);
}
private void doOnCompleted(Throwable cause)
{
ExceptionUtil.call(cause, _onCompleted);
}
private void doOnFailureOnCompleted(Throwable cause)
{
ExceptionUtil.callAndThen(cause, this::onFailure, _onCompleted);
}
private void doOnAbortedOnFailure(Throwable cause)
{
ExceptionUtil.callAndThen(cause, this::onAborted, this::onFailure);
}
private void doOnAbortedOnFailureOnCompleted(Throwable cause)
{
ExceptionUtil.callAndThen(cause, this::doOnAbortedOnFailure, _onCompleted);
}
private void doOnAbortedOnFailureIfNotPendingDoCompleted(Throwable cause)
{
ExceptionUtil.callAndThen(cause, this::doOnAbortedOnFailure, this::ifNotPendingDoCompleted);
}
private void ifNotPendingDoCompleted()
{
Throwable completeFailure = null;
try (AutoLock ignored = _lock.lock())
{
_failure = _failure.getCause();
if (Objects.requireNonNull(_state) != State.PENDING)
{
// the callback completed, one way or another, so it is up to us to do the completion
completeFailure = _failure;
}
}
if (completeFailure != null)
doOnCompleted(completeFailure);
}
/** /**
* This method must be invoked by applications to start the processing * This method must be invoked by applications to start the processing
* of asynchronous sub-tasks. * of asynchronous sub-tasks.
@ -215,28 +338,18 @@ public abstract class IteratingCallback implements Callback
{ {
switch (_state) switch (_state)
{ {
case PENDING:
case CALLED:
// process will be called when callback is handled
break;
case IDLE: case IDLE:
_state = State.PROCESSING; _state = State.PROCESSING;
process = true; process = true;
break; break;
case PROCESSING: case PROCESSING:
_iterate = true; case PROCESSING_CALLED:
_reprocess = true;
break; break;
case FAILED:
case SUCCEEDED:
break;
case CLOSED:
case ABORTED:
default: default:
throw new IllegalStateException(toString()); break;
} }
} }
if (process) if (process)
@ -248,21 +361,24 @@ public abstract class IteratingCallback implements Callback
// This should only ever be called when in processing state, however a failed or close call // This should only ever be called when in processing state, however a failed or close call
// may happen concurrently, so state is not assumed. // may happen concurrently, so state is not assumed.
boolean notifyCompleteSuccess = false; boolean completeSuccess = false;
Throwable notifyCompleteFailure = null; Throwable onAbortedOnFailureOnCompleted = null;
Throwable onFailureOnCompleted = null;
Throwable onAbortedOnFailureIfNotPendingDoCompleted = null;
// While we are processing // While we are processing
processing: processing:
while (true) while (true)
{ {
// Call process to get the action that we have to take. // Call process to get the action that we have to take.
Action action = null; Action action;
try try
{ {
action = process(); action = process();
} }
catch (Throwable x) catch (Throwable x)
{ {
action = null;
failed(x); failed(x);
// Fall through to possibly invoke onCompleteFailure(). // Fall through to possibly invoke onCompleteFailure().
} }
@ -271,72 +387,104 @@ public abstract class IteratingCallback implements Callback
// acted on the action we have just received // acted on the action we have just received
try (AutoLock ignored = _lock.lock()) try (AutoLock ignored = _lock.lock())
{ {
if (LOG.isDebugEnabled())
LOG.debug("processing {} {}", action, this);
switch (_state) switch (_state)
{ {
case PROCESSING: case PROCESSING:
{ {
if (action != null) if (action == null)
break processing;
switch (action)
{ {
switch (action) case IDLE:
{ {
case IDLE: if (_aborted)
{ {
// Has iterate been called while we were processing? _state = _failure instanceof ClosedException ? State.CLOSED : State.COMPLETE;
if (_iterate) onAbortedOnFailureOnCompleted = _failure;
{ break processing;
// yes, so skip idle and keep processing }
_iterate = false;
continue;
}
// No, so we can go idle // Has iterate been called while we were processing?
_state = State.IDLE; if (_reprocess)
break processing;
}
case SCHEDULED:
{ {
// we won the race against the callback, so the callback has to process and we can break processing // yes, so skip idle and keep processing
_state = State.PENDING; _reprocess = false;
break processing; continue;
} }
case SUCCEEDED:
// No, so we can go idle
_state = State.IDLE;
break processing;
}
case SCHEDULED:
{
// we won the race against the callback, so the callback has to process and we can break processing
_state = State.PENDING;
if (_aborted)
{ {
// we lost the race against the callback, onAbortedOnFailureIfNotPendingDoCompleted = _failure;
_iterate = false; _failure = new AbortingException(onAbortedOnFailureIfNotPendingDoCompleted);
_state = State.SUCCEEDED;
notifyCompleteSuccess = true;
break processing;
} }
default: break processing;
}
case SUCCEEDED:
{
// we lost the race against the callback,
_reprocess = false;
if (_aborted)
{ {
break; _state = _failure instanceof ClosedException ? State.CLOSED : State.COMPLETE;
onAbortedOnFailureOnCompleted = _failure;
} }
else
{
_state = State.COMPLETE;
completeSuccess = true;
}
break processing;
}
default:
{
break;
} }
} }
throw new IllegalStateException(String.format("%s[action=%s]", this, action)); throw new IllegalStateException(String.format("%s[action=%s]", this, action));
} }
case CALLED: case PROCESSING_CALLED:
{ {
if (action != Action.SCHEDULED && action != null)
{
_state = State.CLOSED;
onAbortedOnFailureOnCompleted = new IllegalStateException("Action not scheduled");
if (_failure == null)
{
_failure = onAbortedOnFailureOnCompleted;
}
else
{
ExceptionUtil.addSuppressedIfNotAssociated(_failure, onAbortedOnFailureIfNotPendingDoCompleted);
onAbortedOnFailureOnCompleted = _failure;
}
break processing;
}
if (_failure != null)
{
if (_aborted)
onAbortedOnFailureOnCompleted = _failure;
else
onFailureOnCompleted = _failure;
_state = _failure instanceof ClosedException ? State.CLOSED : State.COMPLETE;
break processing;
}
callOnSuccess = true; callOnSuccess = true;
if (action != Action.SCHEDULED)
throw new IllegalStateException(String.format("%s[action=%s]", this, action));
// we lost the race, so we have to keep processing
_state = State.PROCESSING; _state = State.PROCESSING;
continue; break;
} }
case FAILED:
case CLOSED:
case ABORTED:
notifyCompleteFailure = _failure;
break processing;
case SUCCEEDED:
break processing;
case IDLE:
case PENDING:
default: default:
throw new IllegalStateException(String.format("%s[action=%s]", this, action)); throw new IllegalStateException(String.format("%s[action=%s]", this, action));
} }
@ -347,47 +495,74 @@ public abstract class IteratingCallback implements Callback
onSuccess(); onSuccess();
} }
} }
if (onAbortedOnFailureOnCompleted != null)
if (notifyCompleteSuccess) doOnAbortedOnFailureOnCompleted(onAbortedOnFailureOnCompleted);
onCompleteSuccess(); else if (completeSuccess)
else if (notifyCompleteFailure != null) doCompleteSuccess();
onCompleteFailure(notifyCompleteFailure); else if (onFailureOnCompleted != null)
doOnFailureOnCompleted(onFailureOnCompleted);
else if (onAbortedOnFailureIfNotPendingDoCompleted != null)
doOnAbortedOnFailureIfNotPendingDoCompleted(onAbortedOnFailureIfNotPendingDoCompleted);
} }
/** /**
* Method to invoke when the asynchronous sub-task succeeds. * Method to invoke when the asynchronous sub-task succeeds.
* <p> * <p>
* This method should be considered final for all practical purposes. * For most purposes, this method should be considered {@code final} and should only be
* <p> * overridden in extraordinary circumstances.
* Subclasses that override this method must always call {@code super.succeeded()}.
* Such overridden methods are not serialized with respect to {@link #process()}, {@link #onCompleteSuccess()},
* {@link #onCompleteFailure(Throwable)}, nor {@link #onAborted(Throwable)}. They should not act on nor change any
* fields that may be used by those methods.
* Eventually, {@link #onSuccess()} is * Eventually, {@link #onSuccess()} is
* called, either by the caller thread or by the processing * called, either by the caller thread or by the processing
* thread. * thread.
*/ */
@Override @Override
public void succeeded() public final void succeeded()
{ {
boolean process = false; boolean onSuccessProcessing = false;
Throwable onCompleted = null;
try (AutoLock ignored = _lock.lock()) try (AutoLock ignored = _lock.lock())
{ {
if (LOG.isDebugEnabled())
LOG.debug("succeeded {}", this);
switch (_state) switch (_state)
{ {
case PROCESSING: case PROCESSING:
{ {
_state = State.CALLED; // Another thread is processing, so we just tell it the state
_state = State.PROCESSING_CALLED;
break; break;
} }
case PENDING: case PENDING:
{ {
_state = State.PROCESSING; if (_aborted)
process = true; {
if (_failure instanceof AbortingException)
{
// Another thread is still calling onAborted, so we will let it do the completion
_state = _failure.getCause() instanceof ClosedException ? State.CLOSED : State.COMPLETE;
}
else
{
// The onAborted call is complete, so we must do the completion
_state = _failure instanceof ClosedException ? State.CLOSED : State.COMPLETE;
onCompleted = _failure;
}
}
else
{
// No other thread is processing, so we will do the processing
_state = State.PROCESSING;
onSuccessProcessing = true;
}
break; break;
} }
case FAILED: case COMPLETE, CLOSED:
case CLOSED:
case ABORTED:
{ {
// Too late! // Too late
break; return;
} }
default: default:
{ {
@ -395,10 +570,13 @@ public abstract class IteratingCallback implements Callback
} }
} }
} }
if (process) if (onSuccessProcessing)
{ {
onSuccess(); doOnSuccessProcessing();
processing(); }
else if (onCompleted != null)
{
doOnCompleted(onCompleted);
} }
} }
@ -407,47 +585,84 @@ public abstract class IteratingCallback implements Callback
* or to fail the overall asynchronous task and therefore * or to fail the overall asynchronous task and therefore
* terminate the iteration. * terminate the iteration.
* <p> * <p>
* This method should be considered final for all practical purposes.
* <p>
* Eventually, {@link #onCompleteFailure(Throwable)} is * Eventually, {@link #onCompleteFailure(Throwable)} is
* called, either by the caller thread or by the processing * called, either by the caller thread or by the processing
* thread. * thread.
* * <p>
* For most purposes, this method should be considered {@code final} and should only be
* overridden in extraordinary circumstances.
* Subclasses that override this method must always call {@code super.succeeded()}.
* Such overridden methods are not serialized with respect to {@link #process()}, {@link #onCompleteSuccess()},
* {@link #onCompleteFailure(Throwable)}, nor {@link #onAborted(Throwable)}. They should not act on nor change any
* fields that may be used by those methods.
* @see #isFailed() * @see #isFailed()
*/ */
@Override @Override
public void failed(Throwable x) public final void failed(Throwable cause)
{ {
boolean failure = false; cause = Objects.requireNonNullElseGet(cause, IOException::new);
Throwable onFailureOnCompleted = null;
Throwable onCompleted = null;
try (AutoLock ignored = _lock.lock()) try (AutoLock ignored = _lock.lock())
{ {
if (LOG.isDebugEnabled())
LOG.debug("failed {}", this, cause);
switch (_state) switch (_state)
{ {
case CALLED:
case SUCCEEDED:
case FAILED:
case CLOSED:
case ABORTED:
// Too late!
break;
case PENDING:
{
_state = State.FAILED;
failure = true;
break;
}
case PROCESSING: case PROCESSING:
{ {
_state = State.FAILED; // Another thread is processing, so we just tell it the state
_failure = x; _state = State.PROCESSING_CALLED;
if (_failure == null)
_failure = cause;
else
ExceptionUtil.addSuppressedIfNotAssociated(_failure, cause);
break; break;
} }
case PENDING:
{
if (_aborted)
{
if (_failure instanceof AbortingException)
{
// Another thread is still calling onAborted, so we will let it do the completion
ExceptionUtil.addSuppressedIfNotAssociated(_failure.getCause(), cause);
_state = _failure.getCause() instanceof ClosedException ? State.CLOSED : State.COMPLETE;
}
else
{
// The onAborted call is complete, so we must do the completion
ExceptionUtil.addSuppressedIfNotAssociated(_failure, cause);
_state = _failure instanceof ClosedException ? State.CLOSED : State.COMPLETE;
onCompleted = _failure;
}
}
else
{
// No other thread is processing, so we will do the processing
_state = State.COMPLETE;
_failure = cause;
onFailureOnCompleted = _failure;
}
break;
}
case COMPLETE, CLOSED:
{
// Too late
ExceptionUtil.addSuppressedIfNotAssociated(_failure, cause);
return;
}
default: default:
{
throw new IllegalStateException(toString()); throw new IllegalStateException(toString());
}
} }
} }
if (failure) if (onFailureOnCompleted != null)
onCompleteFailure(x); doOnFailureOnCompleted(onFailureOnCompleted);
else if (onCompleted != null)
doOnCompleted(onCompleted);
} }
/** /**
@ -459,37 +674,63 @@ public abstract class IteratingCallback implements Callback
* *
* @see #isClosed() * @see #isClosed()
*/ */
public void close() public final void close()
{ {
String failure = null; Throwable onAbortedOnFailureIfNotPendingDoCompleted = null;
Throwable onAbortedOnFailureOnCompleted = null;
try (AutoLock ignored = _lock.lock()) try (AutoLock ignored = _lock.lock())
{ {
if (LOG.isDebugEnabled())
LOG.debug("close {}", this);
switch (_state) switch (_state)
{ {
case IDLE: case IDLE ->
case SUCCEEDED: {
case FAILED: // Nothing happening so we can abort and complete
_state = State.CLOSED; _state = State.CLOSED;
break; _failure = new ClosedException();
onAbortedOnFailureOnCompleted = _failure;
}
case PROCESSING, PROCESSING_CALLED ->
{
// Another thread is processing, so we just tell it the state and let it handle it
if (_aborted)
{
ExceptionUtil.addSuppressedIfNotAssociated(_failure, new ClosedException());
}
else
{
_aborted = true;
_failure = new ClosedException();
}
}
case PROCESSING: case PENDING ->
_failure = new IOException(String.format("Close %s in state %s", this, _state)); {
// We are waiting for the callback, so we can only call onAbort and then keep waiting
onAbortedOnFailureIfNotPendingDoCompleted = new ClosedException();
_failure = new AbortingException(onAbortedOnFailureIfNotPendingDoCompleted);
_aborted = true;
}
case COMPLETE ->
{
_state = State.CLOSED; _state = State.CLOSED;
break; }
case CLOSED: case CLOSED ->
case ABORTED: {
break; // too late
return;
default: }
failure = String.format("Close %s in state %s", this, _state);
_state = State.CLOSED;
break;
} }
} }
if (failure != null) if (onAbortedOnFailureIfNotPendingDoCompleted != null)
onCompleteFailure(new IOException(failure)); doOnAbortedOnFailureIfNotPendingDoCompleted(onAbortedOnFailureIfNotPendingDoCompleted);
else if (onAbortedOnFailureOnCompleted != null)
doOnAbortedOnFailureOnCompleted(onAbortedOnFailureOnCompleted);
} }
/** /**
@ -498,49 +739,83 @@ public abstract class IteratingCallback implements Callback
* ultimately be invoked, either during this call or later after * ultimately be invoked, either during this call or later after
* any call to {@link #process()} has returned.</p> * any call to {@link #process()} has returned.</p>
* *
* @param failure the cause of the abort * @param cause the cause of the abort
* @return {@code true} if abort was called before the callback was complete.
* @see #isAborted() * @see #isAborted()
*/ */
public void abort(Throwable failure) public final boolean abort(Throwable cause)
{ {
boolean abort = false; cause = Objects.requireNonNullElseGet(cause, Throwable::new);
boolean onAbort = false;
boolean onAbortedOnFailureOnCompleted = false;
try (AutoLock ignored = _lock.lock()) try (AutoLock ignored = _lock.lock())
{ {
if (LOG.isDebugEnabled())
LOG.debug("abort {}", this, cause);
// Are we already aborted?
if (_aborted)
{
ExceptionUtil.addSuppressedIfNotAssociated(_failure, cause);
return false;
}
switch (_state) switch (_state)
{ {
case SUCCEEDED:
case FAILED:
case CLOSED:
case ABORTED:
{
// Too late.
break;
}
case IDLE: case IDLE:
case PENDING:
{ {
_failure = failure; // Nothing happening so we can abort and complete
_state = State.ABORTED; _state = State.COMPLETE;
abort = true; _failure = cause;
_aborted = true;
onAbortedOnFailureOnCompleted = true;
break; break;
} }
case PROCESSING: case PROCESSING:
case CALLED:
{ {
_failure = failure; // Another thread is processing, so we just tell it the state and let it handle everything
_state = State.ABORTED; _failure = cause;
_aborted = true;
break; break;
} }
default: case PROCESSING_CALLED:
throw new IllegalStateException(toString()); {
// Another thread is processing, but we have already succeeded or failed.
if (_failure == null)
_failure = cause;
else
ExceptionUtil.addSuppressedIfNotAssociated(_failure, cause);
_aborted = true;
break;
}
case PENDING:
{
// We are waiting for the callback, so we can only call onAbort and then keep waiting
onAbort = true;
_failure = new AbortingException(cause);
_aborted = true;
break;
}
case COMPLETE, CLOSED:
{
// too late
ExceptionUtil.addSuppressedIfNotAssociated(_failure, cause);
return false;
}
} }
} }
if (abort) if (onAbortedOnFailureOnCompleted)
onCompleteFailure(failure); doOnAbortedOnFailureOnCompleted(cause);
else if (onAbort)
doOnAbortedOnFailureIfNotPendingDoCompleted(cause);
return true;
} }
/** /**
@ -561,7 +836,7 @@ public abstract class IteratingCallback implements Callback
{ {
try (AutoLock ignored = _lock.lock()) try (AutoLock ignored = _lock.lock())
{ {
return _state == State.CLOSED; return _state == State.CLOSED || _failure instanceof ClosedException;
} }
} }
@ -572,7 +847,7 @@ public abstract class IteratingCallback implements Callback
{ {
try (AutoLock ignored = _lock.lock()) try (AutoLock ignored = _lock.lock())
{ {
return _state == State.FAILED; return _failure != null;
} }
} }
@ -585,7 +860,7 @@ public abstract class IteratingCallback implements Callback
{ {
try (AutoLock ignored = _lock.lock()) try (AutoLock ignored = _lock.lock())
{ {
return _state == State.SUCCEEDED; return _state == State.COMPLETE && _failure == null;
} }
} }
@ -596,7 +871,7 @@ public abstract class IteratingCallback implements Callback
{ {
try (AutoLock ignored = _lock.lock()) try (AutoLock ignored = _lock.lock())
{ {
return _state == State.ABORTED; return _aborted;
} }
} }
@ -618,11 +893,10 @@ public abstract class IteratingCallback implements Callback
case IDLE: case IDLE:
return true; return true;
case SUCCEEDED: case COMPLETE:
case FAILED:
_state = State.IDLE; _state = State.IDLE;
_failure = null; _failure = null;
_iterate = false; _reprocess = false;
return true; return true;
default: default:
@ -634,6 +908,31 @@ public abstract class IteratingCallback implements Callback
@Override @Override
public String toString() public String toString()
{ {
return String.format("%s@%x[%s]", getClass().getSimpleName(), hashCode(), _state); try (AutoLock ignored = _lock.lock())
{
return String.format("%s@%x[%s, %b, %s]", getClass().getSimpleName(), hashCode(), _state, _aborted, _failure);
}
}
private static class ClosedException extends Exception
{
ClosedException()
{
super("Closed");
}
ClosedException(Throwable suppressed)
{
this();
ExceptionUtil.addSuppressedIfNotAssociated(this, suppressed);
}
}
private static class AbortingException extends Exception
{
AbortingException(Throwable cause)
{
super(cause.getMessage(), cause);
}
} }
} }

2
jetty-core/jetty-util/src/main/java/org/eclipse/jetty/util/IteratingNestedCallback.java
View File

@ -53,7 +53,7 @@ public abstract class IteratingNestedCallback extends IteratingCallback
} }
@Override @Override
protected void onCompleteFailure(Throwable x) protected void onFailure(Throwable x)
{ {
_callback.failed(x); _callback.failed(x);
} }

674
jetty-core/jetty-util/src/test/java/org/eclipse/jetty/util/IteratingCallbackTest.java
View File

@ -13,19 +13,37 @@
package org.eclipse.jetty.util; package org.eclipse.jetty.util;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CountDownLatch; import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicMarkableReference;
import java.util.concurrent.atomic.AtomicReference;
import java.util.stream.Stream;
import org.awaitility.Awaitility;
import org.eclipse.jetty.util.thread.ScheduledExecutorScheduler; import org.eclipse.jetty.util.thread.ScheduledExecutorScheduler;
import org.eclipse.jetty.util.thread.Scheduler; import org.eclipse.jetty.util.thread.Scheduler;
import org.hamcrest.Matchers;
import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test; import org.junit.jupiter.api.Test;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.Arguments;
import org.junit.jupiter.params.provider.MethodSource;
import static org.hamcrest.MatcherAssert.assertThat;
import static org.hamcrest.Matchers.containsString;
import static org.hamcrest.Matchers.instanceOf;
import static org.hamcrest.Matchers.is;
import static org.hamcrest.Matchers.not;
import static org.hamcrest.Matchers.notNullValue;
import static org.hamcrest.Matchers.nullValue;
import static org.hamcrest.Matchers.sameInstance;
import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse; import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue; import static org.junit.jupiter.api.Assertions.assertTrue;
public class IteratingCallbackTest public class IteratingCallbackTest
@ -202,46 +220,31 @@ public class IteratingCallbackTest
{ {
processed++; processed++;
switch (i--) return switch (i--)
{ {
case 5: case 5, 2 ->
{
succeeded(); succeeded();
return Action.SCHEDULED; yield Action.SCHEDULED;
}
case 4: case 4, 1 ->
{
scheduler.schedule(successTask, 5, TimeUnit.MILLISECONDS); scheduler.schedule(successTask, 5, TimeUnit.MILLISECONDS);
return Action.SCHEDULED; yield Action.SCHEDULED;
}
case 3: case 3 ->
scheduler.schedule(new Runnable() {
{ scheduler.schedule(idle::countDown, 5, TimeUnit.MILLISECONDS);
@Override yield Action.IDLE;
public void run() }
{ case 0 -> Action.SUCCEEDED;
idle.countDown(); default -> throw new IllegalStateException();
} };
}, 5, TimeUnit.MILLISECONDS);
return Action.IDLE;
case 2:
succeeded();
return Action.SCHEDULED;
case 1:
scheduler.schedule(successTask, 5, TimeUnit.MILLISECONDS);
return Action.SCHEDULED;
case 0:
return Action.SUCCEEDED;
default:
throw new IllegalStateException();
}
} }
}; };
cb.iterate(); cb.iterate();
idle.await(10, TimeUnit.SECONDS); assertTrue(idle.await(10, TimeUnit.SECONDS));
assertTrue(cb.isIdle()); assertTrue(cb.isIdle());
cb.iterate(); cb.iterate();
@ -252,25 +255,53 @@ public class IteratingCallbackTest
@Test @Test
public void testCloseDuringProcessingReturningScheduled() throws Exception public void testCloseDuringProcessingReturningScheduled() throws Exception
{ {
testCloseDuringProcessing(IteratingCallback.Action.SCHEDULED); final CountDownLatch abortLatch = new CountDownLatch(1);
final CountDownLatch failureLatch = new CountDownLatch(1);
IteratingCallback callback = new IteratingCallback()
{
@Override
protected Action process()
{
close();
return Action.SCHEDULED;
}
@Override
protected void onAborted(Throwable cause)
{
abortLatch.countDown();
}
@Override
protected void onCompleteFailure(Throwable cause)
{
failureLatch.countDown();
}
};
callback.iterate();
assertFalse(failureLatch.await(100, TimeUnit.MILLISECONDS));
assertTrue(abortLatch.await(1000000000, TimeUnit.SECONDS));
assertTrue(callback.isClosed());
callback.succeeded();
assertTrue(failureLatch.await(1, TimeUnit.SECONDS));
assertTrue(callback.isFailed());
assertTrue(callback.isClosed());
} }
@Test @Test
public void testCloseDuringProcessingReturningSucceeded() throws Exception public void testCloseDuringProcessingReturningSucceeded() throws Exception
{
testCloseDuringProcessing(IteratingCallback.Action.SUCCEEDED);
}
private void testCloseDuringProcessing(final IteratingCallback.Action action) throws Exception
{ {
final CountDownLatch failureLatch = new CountDownLatch(1); final CountDownLatch failureLatch = new CountDownLatch(1);
IteratingCallback callback = new IteratingCallback() IteratingCallback callback = new IteratingCallback()
{ {
@Override @Override
protected Action process() throws Exception protected Action process()
{ {
close(); close();
return action; return Action.SUCCEEDED;
} }
@Override @Override
@ -287,22 +318,8 @@ public class IteratingCallbackTest
private abstract static class TestCB extends IteratingCallback private abstract static class TestCB extends IteratingCallback
{ {
protected Runnable successTask = new Runnable() protected Runnable successTask = this::succeeded;
{ protected Runnable failTask = () -> failed(new Exception("testing failure"));
@Override
public void run()
{
succeeded();
}
};
protected Runnable failTask = new Runnable()
{
@Override
public void run()
{
failed(new Exception("testing failure"));
}
};
protected CountDownLatch completed = new CountDownLatch(1); protected CountDownLatch completed = new CountDownLatch(1);
protected int processed = 0; protected int processed = 0;
@ -320,8 +337,7 @@ public class IteratingCallbackTest
boolean waitForComplete() throws InterruptedException boolean waitForComplete() throws InterruptedException
{ {
completed.await(10, TimeUnit.SECONDS); return completed.await(10, TimeUnit.SECONDS) && isSucceeded();
return isSucceeded();
} }
} }
@ -342,7 +358,6 @@ public class IteratingCallbackTest
@Override @Override
protected void onCompleteFailure(Throwable cause) protected void onCompleteFailure(Throwable cause)
{ {
super.onCompleteFailure(cause);
failure.incrementAndGet(); failure.incrementAndGet();
} }
}; };
@ -390,57 +405,544 @@ public class IteratingCallbackTest
assertEquals(1, count.get()); assertEquals(1, count.get());
// Aborting should not iterate.
icb.abort(new Exception()); icb.abort(new Exception());
assertTrue(ocfLatch.await(5, TimeUnit.SECONDS)); assertTrue(ocfLatch.await(5, TimeUnit.SECONDS));
assertTrue(icb.isFailed());
assertTrue(icb.isAborted()); assertTrue(icb.isAborted());
assertEquals(1, count.get()); assertEquals(1, count.get());
} }
@Test @Test
public void testWhenProcessingAbortSerializesOnCompleteFailure() throws Exception public void testWhenPendingAbortSerializesOnCompleteFailure() throws Exception
{ {
AtomicInteger count = new AtomicInteger(); AtomicReference<Throwable> aborted = new AtomicReference<>();
CountDownLatch ocfLatch = new CountDownLatch(1); CountDownLatch abortLatch = new CountDownLatch(1);
AtomicReference<Throwable> failure = new AtomicReference<>();
AtomicMarkableReference<Throwable> completed = new AtomicMarkableReference<>(null, false);
IteratingCallback icb = new IteratingCallback() IteratingCallback icb = new IteratingCallback()
{ {
@Override @Override
protected Action process() throws Throwable protected Action process() throws Throwable
{ {
count.incrementAndGet();
abort(new Exception());
// After calling abort, onCompleteFailure() must not be called yet.
assertFalse(ocfLatch.await(1, TimeUnit.SECONDS));
return Action.SCHEDULED; return Action.SCHEDULED;
} }
@Override
protected void onAborted(Throwable cause)
{
aborted.set(cause);
ExceptionUtil.call(abortLatch::await, Throwable::printStackTrace);
}
@Override
protected void onCompleteSuccess()
{
completed.set(null, true);
}
@Override @Override
protected void onCompleteFailure(Throwable cause) protected void onCompleteFailure(Throwable cause)
{ {
ocfLatch.countDown(); completed.set(cause, true);
failure.set(cause);
} }
}; };
icb.iterate(); icb.iterate();
assertEquals(1, count.get()); assertThat(icb.toString(), containsString("[PENDING, false,"));
assertTrue(ocfLatch.await(5, TimeUnit.SECONDS)); Throwable cause = new Throwable("test abort");
assertTrue(icb.isAborted()); new Thread(() -> icb.abort(cause)).start();
Awaitility.waitAtMost(5, TimeUnit.SECONDS).until(() -> icb.toString().contains("[PENDING, true,"));
Awaitility.waitAtMost(5, TimeUnit.SECONDS).until(() -> aborted.get() != null);
// Calling succeeded() won't cause further iterations.
icb.succeeded(); icb.succeeded();
assertEquals(1, count.get()); // We are now complete, but callbacks have not yet been done
assertThat(icb.toString(), containsString("[COMPLETE, true,"));
assertThat(failure.get(), nullValue());
assertFalse(completed.isMarked());
abortLatch.countDown();
Awaitility.waitAtMost(5, TimeUnit.SECONDS).until(completed::isMarked);
assertThat(failure.get(), sameInstance(cause));
assertThat(completed.getReference(), sameInstance(cause));
}
public enum Event
{
PROCESSED,
ABORTED,
SUCCEEDED,
FAILED
}
public static Stream<List<Event>> serializedEvents()
{
return Stream.of(
List.of(Event.PROCESSED, Event.ABORTED, Event.SUCCEEDED),
List.of(Event.PROCESSED, Event.SUCCEEDED, Event.ABORTED),
List.of(Event.SUCCEEDED, Event.PROCESSED, Event.ABORTED),
List.of(Event.SUCCEEDED, Event.ABORTED, Event.PROCESSED),
List.of(Event.ABORTED, Event.SUCCEEDED, Event.PROCESSED),
List.of(Event.ABORTED, Event.PROCESSED, Event.SUCCEEDED),
List.of(Event.PROCESSED, Event.ABORTED, Event.FAILED),
List.of(Event.PROCESSED, Event.FAILED, Event.ABORTED),
List.of(Event.FAILED, Event.PROCESSED, Event.ABORTED),
List.of(Event.FAILED, Event.ABORTED, Event.PROCESSED),
List.of(Event.ABORTED, Event.FAILED, Event.PROCESSED),
List.of(Event.ABORTED, Event.PROCESSED, Event.FAILED)
);
}
@ParameterizedTest
@MethodSource("serializedEvents")
public void testSerializesProcessAbortCompletion(List<Event> events) throws Exception
{
AtomicReference<Throwable> aborted = new AtomicReference<>();
CountDownLatch processingLatch = new CountDownLatch(1);
CountDownLatch abortLatch = new CountDownLatch(1);
AtomicReference<Throwable> failure = new AtomicReference<>();
AtomicMarkableReference<Throwable> completed = new AtomicMarkableReference<>(null, false);
Throwable cause = new Throwable("test abort");
IteratingCallback icb = new IteratingCallback()
{
@Override
protected Action process() throws Throwable
{
abort(cause);
ExceptionUtil.call(processingLatch::await, Throwable::printStackTrace);
return Action.SCHEDULED;
}
@Override
protected void onAborted(Throwable cause)
{
aborted.set(cause);
ExceptionUtil.call(abortLatch::await, Throwable::printStackTrace);
}
@Override
protected void onCompleteSuccess()
{
completed.set(null, true);
}
@Override
protected void onCompleteFailure(Throwable cause)
{
completed.set(cause, true);
failure.set(cause);
}
};
new Thread(icb::iterate).start();
Awaitility.waitAtMost(5, TimeUnit.SECONDS).until(() -> icb.toString().contains("[PROCESSING, true,"));
// we have aborted, but onAborted not yet called
assertThat(aborted.get(), nullValue());
int count = 0;
for (Event event : events)
{
switch (event)
{
case PROCESSED ->
{
processingLatch.countDown();
// We can call aborted
Awaitility.waitAtMost(5, TimeUnit.SECONDS).pollInterval(10, TimeUnit.MILLISECONDS).until(() -> aborted.get() != null);
}
case ABORTED ->
{
abortLatch.countDown();
Awaitility.waitAtMost(5, TimeUnit.SECONDS).pollInterval(10, TimeUnit.MILLISECONDS).until(() -> !icb.toString().contains("AbortingException"));
}
case SUCCEEDED -> icb.succeeded();
case FAILED -> icb.failed(new Throwable("failure"));
}
if (++count < 3)
{
// Not complete yet
assertThat(failure.get(), nullValue());
assertFalse(completed.isMarked());
}
// Extra aborts ignored
assertFalse(icb.abort(new Throwable("ignored")));
}
// When the callback is succeeded, the completion events can be called
Awaitility.waitAtMost(5, TimeUnit.SECONDS).pollInterval(10, TimeUnit.MILLISECONDS).until(completed::isMarked);
assertThat(failure.get(), sameInstance(cause));
assertThat(completed.getReference(), sameInstance(cause));
}
@Test
public void testICBSuccess() throws Exception
{
TestIteratingCB callback = new TestIteratingCB();
callback.iterate();
callback.succeeded();
assertTrue(callback._completed.await(1, TimeUnit.SECONDS));
assertThat(callback._onFailure.get(), nullValue());
assertThat(callback._completion.getReference(), Matchers.nullValue());
assertTrue(callback._completion.isMarked());
// Everything now a noop
assertFalse(callback.abort(new Throwable()));
callback.failed(new Throwable());
assertThat(callback._completion.getReference(), Matchers.nullValue());
assertThat(callback._completed.getCount(), is(0L));
callback.checkNoBadCalls();
}
@Test
public void testICBFailure() throws Exception
{
Throwable failure = new Throwable();
TestIteratingCB callback = new TestIteratingCB();
callback.iterate();
callback.failed(failure);
assertTrue(callback._completed.await(1, TimeUnit.SECONDS));
assertThat(callback._onFailure.get(), sameInstance(failure));
assertThat(callback._completion.getReference(), Matchers.sameInstance(failure));
assertTrue(callback._completion.isMarked());
// Everything now a noop, other than suppression
callback.succeeded();
Throwable late = new Throwable();
assertFalse(callback.abort(late));
assertFalse(ExceptionUtil.areNotAssociated(failure, late));
assertThat(callback._completion.getReference(), Matchers.sameInstance(failure));
assertThat(callback._completed.getCount(), is(0L));
callback.checkNoBadCalls();
}
@Test
public void testICBAbortSuccess() throws Exception
{
TestIteratingCB callback = new TestIteratingCB();
callback.iterate();
Throwable abort = new Throwable();
callback.abort(abort);
assertFalse(callback._completed.await(100, TimeUnit.MILLISECONDS));
assertThat(callback._onFailure.get(), sameInstance(abort));
assertThat(callback._completion.getReference(), Matchers.sameInstance(abort));
assertFalse(callback._completion.isMarked());
callback.succeeded();
assertThat(callback._completion.getReference(), Matchers.sameInstance(abort));
assertThat(callback._completed.getCount(), is(0L));
Throwable late = new Throwable();
callback.failed(late);
assertFalse(callback.abort(late));
assertTrue(ExceptionUtil.areAssociated(abort, late));
assertTrue(ExceptionUtil.areAssociated(callback._onFailure.get(), late));
assertThat(callback._completion.getReference(), Matchers.sameInstance(abort));
assertThat(callback._completed.getCount(), is(0L));
callback.checkNoBadCalls();
}
public static Stream<Arguments> abortTests()
{
List<Arguments> tests = new ArrayList<>();
for (IteratingCallback.State state : IteratingCallback.State.values())
{
String name = state.name();
if (name.contains("PROCESSING"))
{
for (IteratingCallback.Action action : IteratingCallback.Action.values())
{
if (name.contains("CALLED"))
{
if (action == IteratingCallback.Action.SCHEDULED)
{
tests.add(Arguments.of(name, action.toString(), Boolean.TRUE));
tests.add(Arguments.of(name, action.toString(), Boolean.FALSE));
}
}
else if (action == IteratingCallback.Action.SCHEDULED)
{
tests.add(Arguments.of(name, action.toString(), Boolean.TRUE));
tests.add(Arguments.of(name, action.toString(), Boolean.FALSE));
}
else
{
tests.add(Arguments.of(name, action.toString(), null));
}
}
}
else if (name.equals("COMPLETE") || name.contains("PENDING"))
{
tests.add(Arguments.of(name, null, Boolean.TRUE));
tests.add(Arguments.of(name, null, Boolean.FALSE));
}
else
{
tests.add(Arguments.of(name, null, null));
}
}
return tests.stream();
}
@ParameterizedTest
@MethodSource("abortTests")
public void testAbortInEveryState(String state, String action, Boolean success) throws Exception
{
CountDownLatch processLatch = new CountDownLatch(1);
AtomicReference<Throwable> onAbort = new AtomicReference<>();
AtomicReference<Throwable> onFailure = new AtomicReference<>(null);
AtomicMarkableReference<Throwable> onCompleted = new AtomicMarkableReference<>(null, false);
Throwable cause = new Throwable("abort");
Throwable failure = new Throwable("failure");
AtomicInteger badCalls = new AtomicInteger(0);
IteratingCallback callback = new IteratingCallback()
{
@Override
protected Action process() throws Throwable
{
if (state.contains("CALLED"))
{
if (success)
succeeded();
else
failed(failure);
}
if (state.contains("PENDING"))
return Action.SCHEDULED;
if (state.equals("COMPLETE"))
{
if (success)
return Action.SUCCEEDED;
failed(new Throwable("Complete Failure"));
return Action.SCHEDULED;
}
if (state.equals("CLOSED"))
{
close();
return Action.SUCCEEDED;
}
processLatch.await();
return IteratingCallback.Action.valueOf(action);
}
@Override
protected void onFailure(Throwable cause)
{
if (!onFailure.compareAndSet(null, cause))
badCalls.incrementAndGet();
}
@Override
protected void onAborted(Throwable cause)
{
if (!onAbort.compareAndSet(null, cause))
badCalls.incrementAndGet();
}
@Override
protected void onCompleteSuccess()
{
onCompleted.set(null, true);
}
@Override
protected void onCompleteFailure(Throwable cause)
{
onCompleted.set(cause, true);
}
};
if (!state.equals("IDLE"))
{
new Thread(callback::iterate).start();
}
Awaitility.waitAtMost(5, TimeUnit.SECONDS).pollInterval(10, TimeUnit.MILLISECONDS).until(() -> callback.toString().contains(state));
assertThat(callback.toString(), containsString("[" + state + ","));
onAbort.set(null);
if (success == Boolean.FALSE && (state.equals("COMPLETE") || state.equals("CLOSED")))
{
// We must be failed already
assertThat(onFailure.get(), notNullValue());
}
boolean aborted = callback.abort(cause);
// Check abort in completed state
if (state.equals("COMPLETE") || state.equals("CLOSED"))
{
assertThat(aborted, is(false));
assertThat(onAbort.get(), nullValue());
assertTrue(onCompleted.isMarked());
if (success == Boolean.TRUE)
assertThat(onCompleted.getReference(), nullValue());
else
assertThat(onCompleted.getReference(), notNullValue());
return;
}
// Check abort in non completed state
assertThat(aborted, is(true));
if (state.contains("PROCESSING"))
{
processLatch.countDown();
Awaitility.waitAtMost(5, TimeUnit.SECONDS).pollInterval(10, TimeUnit.MILLISECONDS).until(() -> !callback.toString().contains("PROCESSING"));
if (action.equals("SCHEDULED"))
{
if (success)
{
callback.succeeded();
}
else
{
Throwable failureAfterAbort = new Throwable("failure after abort");
callback.failed(failureAfterAbort);
assertThat(onFailure.get(), not(sameInstance(failureAfterAbort)));
assertTrue(ExceptionUtil.areAssociated(onFailure.get(), failureAfterAbort));
}
}
}
else if (state.contains("PENDING"))
{
if (success)
callback.succeeded();
else
callback.failed(new Throwable("failure after abort"));
}
assertTrue(onCompleted.isMarked());
if (state.contains("CALLED") && !success)
{
assertThat(onCompleted.getReference(), sameInstance(failure));
assertThat(onAbort.get(), sameInstance(failure));
}
else
{
assertThat(onCompleted.getReference(), sameInstance(cause));
assertThat(onAbort.get(), sameInstance(cause));
}
assertThat(badCalls.get(), is(0));
}
private static class TestIteratingCB extends IteratingCallback
{
final AtomicInteger _count;
final AtomicInteger _badCalls = new AtomicInteger(0);
final AtomicBoolean _onSuccess = new AtomicBoolean();
final AtomicReference<Throwable> _onFailure = new AtomicReference<>();
final AtomicMarkableReference<Throwable> _completion = new AtomicMarkableReference<>(null, false);
final CountDownLatch _completed = new CountDownLatch(1);
private TestIteratingCB()
{
this(1);
}
private TestIteratingCB(int count)
{
_count = new AtomicInteger(count);
}
@Override
protected Action process()
{
return _count.getAndDecrement() == 0 ? Action.SUCCEEDED : Action.SCHEDULED;
}
@Override
protected void onAborted(Throwable cause)
{
_completion.compareAndSet(null, cause, false, false);
}
@Override
protected void onSuccess()
{
if (!_onSuccess.compareAndSet(false, true))
_badCalls.incrementAndGet();
}
@Override
protected void onFailure(Throwable cause)
{
if (!_onFailure.compareAndSet(null, cause))
_badCalls.incrementAndGet();
}
@Override
protected void onCompleteSuccess()
{
if (_completion.isMarked())
_badCalls.incrementAndGet();
if (_completion.compareAndSet(null, null, false, true))
_completed.countDown();
}
@Override
protected void onCompleteFailure(Throwable cause)
{
if (_completion.isMarked())
_badCalls.incrementAndGet();
if (_completion.compareAndSet(null, cause, false, true))
_completed.countDown();
// Try again the CAS if there was a call to onAborted().
Throwable failure = _completion.getReference();
if (failure != null && _completion.compareAndSet(failure, failure, false, true))
_completed.countDown();
}
public void checkNoBadCalls()
{
assertThat(_badCalls.get(), is(0));
}
} }
@Test @Test
public void testOnSuccessCalledDespiteISE() throws Exception public void testOnSuccessCalledDespiteISE() throws Exception
{ {
CountDownLatch latch = new CountDownLatch(1); CountDownLatch latch = new CountDownLatch(1);
AtomicReference<Throwable> aborted = new AtomicReference<>();
IteratingCallback icb = new IteratingCallback() IteratingCallback icb = new IteratingCallback()
{ {
@Override @Override
@ -451,13 +953,27 @@ public class IteratingCallbackTest
} }
@Override @Override
protected void onSuccess() protected void onAborted(Throwable cause)
{
aborted.set(cause);
super.onAborted(cause);
}
@Override
protected void onCompleteSuccess()
{
latch.countDown();
}
@Override
protected void onCompleteFailure(Throwable cause)
{ {
latch.countDown(); latch.countDown();
} }
}; };
assertThrows(IllegalStateException.class, icb::iterate); icb.iterate();
assertTrue(latch.await(5, TimeUnit.SECONDS)); assertTrue(latch.await(5, TimeUnit.SECONDS));
assertThat(aborted.get(), instanceOf(IllegalStateException.class));
} }
} }

4
jetty-core/jetty-websocket/jetty-websocket-core-common/src/main/java/org/eclipse/jetty/websocket/core/WebSocketConnection.java
View File

@ -635,10 +635,10 @@ public class WebSocketConnection extends AbstractConnection implements Connectio
} }
@Override @Override
public void onCompleteFailure(Throwable x) public void onFailure(Throwable x)
{ {
coreSession.processConnectionError(x, NOOP); coreSession.processConnectionError(x, NOOP);
super.onCompleteFailure(x); super.onFailure(x);
} }
} }
} }

21
jetty-core/jetty-websocket/jetty-websocket-core-common/src/main/java/org/eclipse/jetty/websocket/core/internal/FrameFlusher.java
View File

@ -405,11 +405,8 @@ public class FrameFlusher extends IteratingCallback
} }
@Override @Override
public void onCompleteFailure(Throwable failure) public void onFailure(Throwable failure)
{ {
if (batchBuffer != null)
batchBuffer.clear();
releaseAggregate();
try (AutoLock l = lock.lock()) try (AutoLock l = lock.lock())
{ {
failedEntries.addAll(queue); failedEntries.addAll(queue);
@ -418,9 +415,6 @@ public class FrameFlusher extends IteratingCallback
failedEntries.addAll(entries); failedEntries.addAll(entries);
entries.clear(); entries.clear();
releasableBuffers.forEach(RetainableByteBuffer::release);
releasableBuffers.clear();
if (closedCause == null) if (closedCause == null)
closedCause = failure; closedCause = failure;
else if (closedCause != failure) else if (closedCause != failure)
@ -436,6 +430,19 @@ public class FrameFlusher extends IteratingCallback
endPoint.close(closedCause); endPoint.close(closedCause);
} }
@Override
protected void onCompleteFailure(Throwable cause)
{
if (batchBuffer != null)
batchBuffer.clear();
releaseAggregate();
try (AutoLock l = lock.lock())
{
releasableBuffers.forEach(RetainableByteBuffer::release);
releasableBuffers.clear();
}
}
private void releaseAggregate() private void releaseAggregate()
{ {
if (batchBuffer != null && batchBuffer.isEmpty()) if (batchBuffer != null && batchBuffer.isEmpty())

1
jetty-core/jetty-websocket/jetty-websocket-core-common/src/main/java/org/eclipse/jetty/websocket/core/internal/PerMessageDeflateExtension.java
View File

@ -497,7 +497,6 @@ public class PerMessageDeflateExtension extends AbstractExtension implements Dem
protected void onCompleteFailure(Throwable cause) protected void onCompleteFailure(Throwable cause)
{ {
releasePayload(_payloadRef); releasePayload(_payloadRef);
super.onCompleteFailure(cause);
} }
private void releasePayload(AtomicReference<RetainableByteBuffer> reference) private void releasePayload(AtomicReference<RetainableByteBuffer> reference)

6
jetty-core/jetty-websocket/jetty-websocket-core-common/src/main/java/org/eclipse/jetty/websocket/core/util/DemandingFlusher.java
View File

@ -153,7 +153,7 @@ public abstract class DemandingFlusher extends IteratingCallback implements Dema
throw failure; throw failure;
if (!_demand.get()) if (!_demand.get())
break; return Action.IDLE;
if (_needContent) if (_needContent)
{ {
@ -173,12 +173,10 @@ public abstract class DemandingFlusher extends IteratingCallback implements Dema
_callback = null; _callback = null;
} }
} }
return Action.IDLE;
} }
@Override @Override
protected void onCompleteFailure(Throwable cause) protected void onFailure(Throwable cause)
{ {
Throwable suppressed = _failure.getAndSet(cause); Throwable suppressed = _failure.getAndSet(cause);
if (suppressed != null && suppressed != cause) if (suppressed != null && suppressed != cause)

4
jetty-core/jetty-websocket/jetty-websocket-core-common/src/main/java/org/eclipse/jetty/websocket/core/util/TransformingFlusher.java
View File

@ -170,7 +170,7 @@ public abstract class TransformingFlusher
} }
@Override @Override
protected void onCompleteFailure(Throwable t) protected void onFailure(Throwable t)
{ {
if (log.isDebugEnabled()) if (log.isDebugEnabled())
log.debug("onCompleteFailure {}", t.toString()); log.debug("onCompleteFailure {}", t.toString());
@ -180,7 +180,7 @@ public abstract class TransformingFlusher
notifyCallbackFailure(current.callback, t); notifyCallbackFailure(current.callback, t);
current = null; current = null;
} }
onFailure(t); TransformingFlusher.this.onFailure(t);
} }
} }

4
jetty-core/jetty-websocket/jetty-websocket-core-tests/src/test/java/org/eclipse/jetty/websocket/core/internal/FrameFlusherTest.java
View File

@ -166,11 +166,11 @@ public class FrameFlusherTest
FrameFlusher frameFlusher = new FrameFlusher(bufferPool, scheduler, generator, endPoint, bufferSize, maxGather) FrameFlusher frameFlusher = new FrameFlusher(bufferPool, scheduler, generator, endPoint, bufferSize, maxGather)
{ {
@Override @Override
public void onCompleteFailure(Throwable failure) public void onFailure(Throwable failure)
{ {
error.set(failure); error.set(failure);
flusherFailure.countDown(); flusherFailure.countDown();
super.onCompleteFailure(failure); super.onFailure(failure);
} }
}; };

2
jetty-ee10/jetty-ee10-proxy/src/main/java/org/eclipse/jetty/ee10/proxy/AsyncMiddleManServlet.java
View File

@ -400,7 +400,7 @@ public class AsyncMiddleManServlet extends AbstractProxyServlet
} }
@Override @Override
protected void onCompleteFailure(Throwable x) protected void onFailure(Throwable x)
{ {
onError(x); onError(x);
} }

2
jetty-ee10/jetty-ee10-proxy/src/main/java/org/eclipse/jetty/ee10/proxy/AsyncProxyServlet.java
View File

@ -192,7 +192,7 @@ public class AsyncProxyServlet extends ProxyServlet
} }
@Override @Override
protected void onCompleteFailure(Throwable cause) protected void onFailure(Throwable cause)
{ {
onError(cause); onError(cause);
} }

55
jetty-ee10/jetty-ee10-servlet/src/main/java/org/eclipse/jetty/ee10/servlet/HttpOutput.java
View File

@ -132,7 +132,6 @@ public class HttpOutput extends ServletOutputStream implements Runnable
private State _state = State.OPEN; private State _state = State.OPEN;
private boolean _softClose = false; private boolean _softClose = false;
private long _written; private long _written;
private long _flushed;
private long _firstByteNanoTime = -1; private long _firstByteNanoTime = -1;
private ByteBufferPool.Sized _pool; private ByteBufferPool.Sized _pool;
private RetainableByteBuffer _aggregate; private RetainableByteBuffer _aggregate;
@ -222,7 +221,8 @@ public class HttpOutput extends ServletOutputStream implements Runnable
_state = State.CLOSED; _state = State.CLOSED;
closedCallback = _closedCallback; closedCallback = _closedCallback;
_closedCallback = null; _closedCallback = null;
lockedReleaseBuffer(failure != null); if (failure == null)
lockedReleaseBuffer();
wake = updateApiState(failure); wake = updateApiState(failure);
} }
else if (_state == State.CLOSE) else if (_state == State.CLOSE)
@ -444,7 +444,7 @@ public class HttpOutput extends ServletOutputStream implements Runnable
try (AutoLock ignored = _channelState.lock()) try (AutoLock ignored = _channelState.lock())
{ {
_state = State.CLOSED; _state = State.CLOSED;
lockedReleaseBuffer(failure != null); lockedReleaseBuffer();
} }
} }
@ -598,18 +598,13 @@ public class HttpOutput extends ServletOutputStream implements Runnable
return _aggregate; return _aggregate;
} }
private void lockedReleaseBuffer(boolean failure) private void lockedReleaseBuffer()
{ {
assert _channelState.isLockHeldByCurrentThread(); assert _channelState.isLockHeldByCurrentThread();
if (_aggregate != null) if (_aggregate != null)
{ {
if (failure && _pool != null) _aggregate.release();
_pool.removeAndRelease(_aggregate);
else
_aggregate.release();
_aggregate = null; _aggregate = null;
_pool = null;
} }
} }
@ -1251,7 +1246,7 @@ public class HttpOutput extends ServletOutputStream implements Runnable
{ {
try (AutoLock ignored = _channelState.lock()) try (AutoLock ignored = _channelState.lock())
{ {
lockedReleaseBuffer(_state != State.CLOSED); lockedReleaseBuffer();
_state = State.OPEN; _state = State.OPEN;
_apiState = ApiState.BLOCKING; _apiState = ApiState.BLOCKING;
_softClose = true; // Stay closed until next request _softClose = true; // Stay closed until next request
@ -1264,7 +1259,6 @@ public class HttpOutput extends ServletOutputStream implements Runnable
_writeListener = null; _writeListener = null;
_onError = null; _onError = null;
_firstByteNanoTime = -1; _firstByteNanoTime = -1;
_flushed = 0;
_closedCallback = null; _closedCallback = null;
} }
} }
@ -1404,10 +1398,19 @@ public class HttpOutput extends ServletOutputStream implements Runnable
} }
@Override @Override
public void onCompleteFailure(Throwable e) protected void onFailure(Throwable e)
{ {
onWriteComplete(_last, e); onWriteComplete(_last, e);
} }
@Override
protected void onCompleteFailure(Throwable cause)
{
try (AutoLock ignored = _channelState.lock())
{
lockedReleaseBuffer();
}
}
} }
private abstract class NestedChannelWriteCB extends ChannelWriteCB private abstract class NestedChannelWriteCB extends ChannelWriteCB
@ -1440,11 +1443,11 @@ public class HttpOutput extends ServletOutputStream implements Runnable
} }
@Override @Override
public void onCompleteFailure(Throwable e) protected void onFailure(Throwable e)
{ {
try try
{ {
super.onCompleteFailure(e); super.onFailure(e);
} }
catch (Throwable t) catch (Throwable t)
{ {
@ -1467,7 +1470,7 @@ public class HttpOutput extends ServletOutputStream implements Runnable
} }
@Override @Override
protected Action process() throws Exception protected Action process()
{ {
if (_aggregate != null && _aggregate.hasRemaining()) if (_aggregate != null && _aggregate.hasRemaining())
{ {
@ -1518,7 +1521,7 @@ public class HttpOutput extends ServletOutputStream implements Runnable
} }
@Override @Override
protected Action process() throws Exception protected Action process()
{ {
// flush any content from the aggregate // flush any content from the aggregate
if (_aggregate != null && _aggregate.hasRemaining()) if (_aggregate != null && _aggregate.hasRemaining())
@ -1641,15 +1644,18 @@ public class HttpOutput extends ServletOutputStream implements Runnable
{ {
_buffer.release(); _buffer.release();
IO.close(_in); IO.close(_in);
super.onCompleteSuccess(); }
@Override
protected void onFailure(Throwable cause)
{
IO.close(_in);
} }
@Override @Override
public void onCompleteFailure(Throwable x) public void onCompleteFailure(Throwable x)
{ {
_buffer.release(); _buffer.release();
IO.close(_in);
super.onCompleteFailure(x);
} }
} }
@ -1714,15 +1720,18 @@ public class HttpOutput extends ServletOutputStream implements Runnable
{ {
_buffer.release(); _buffer.release();
IO.close(_in); IO.close(_in);
super.onCompleteSuccess(); }
@Override
protected void onFailure(Throwable cause)
{
IO.close(_in);
} }
@Override @Override
public void onCompleteFailure(Throwable x) public void onCompleteFailure(Throwable x)
{ {
_buffer.release(); _buffer.release();
IO.close(_in);
super.onCompleteFailure(x);
} }
} }

3
jetty-ee10/jetty-ee10-webapp/src/main/java/org/eclipse/jetty/ee10/webapp/WebInfConfiguration.java
View File

@ -18,11 +18,8 @@ import java.io.IOException;
import java.net.URI; import java.net.URI;
import java.nio.file.Files; import java.nio.file.Files;
import java.nio.file.Path; import java.nio.file.Path;
import java.util.List;
import jakarta.servlet.ServletContext; import jakarta.servlet.ServletContext;
import org.eclipse.jetty.server.Connector;
import org.eclipse.jetty.server.NetworkConnector;
import org.eclipse.jetty.server.Server; import org.eclipse.jetty.server.Server;
import org.eclipse.jetty.util.FileID; import org.eclipse.jetty.util.FileID;
import org.eclipse.jetty.util.IO; import org.eclipse.jetty.util.IO;

2
jetty-ee11/jetty-ee11-proxy/src/main/java/org/eclipse/jetty/ee11/proxy/AsyncMiddleManServlet.java
View File

@ -400,7 +400,7 @@ public class AsyncMiddleManServlet extends AbstractProxyServlet
} }
@Override @Override
protected void onCompleteFailure(Throwable x) protected void onFailure(Throwable x)
{ {
onError(x); onError(x);
} }

3
jetty-ee11/jetty-ee11-proxy/src/main/java/org/eclipse/jetty/ee11/proxy/AsyncProxyServlet.java
View File

@ -192,9 +192,8 @@ public class AsyncProxyServlet extends ProxyServlet
} }
@Override @Override
public void failed(Throwable x) public void onFailure(Throwable x)
{ {
super.failed(x);
onError(x); onError(x);
} }
} }

58
jetty-ee11/jetty-ee11-servlet/src/main/java/org/eclipse/jetty/ee11/servlet/HttpOutput.java
View File

@ -132,7 +132,6 @@ public class HttpOutput extends ServletOutputStream implements Runnable
private State _state = State.OPEN; private State _state = State.OPEN;
private boolean _softClose = false; private boolean _softClose = false;
private long _written; private long _written;
private long _flushed;
private long _firstByteNanoTime = -1; private long _firstByteNanoTime = -1;
private ByteBufferPool.Sized _pool; private ByteBufferPool.Sized _pool;
private RetainableByteBuffer _aggregate; private RetainableByteBuffer _aggregate;
@ -222,7 +221,8 @@ public class HttpOutput extends ServletOutputStream implements Runnable
_state = State.CLOSED; _state = State.CLOSED;
closedCallback = _closedCallback; closedCallback = _closedCallback;
_closedCallback = null; _closedCallback = null;
lockedReleaseBuffer(failure != null); if (failure == null)
lockedReleaseBuffer();
wake = updateApiState(failure); wake = updateApiState(failure);
} }
else if (_state == State.CLOSE) else if (_state == State.CLOSE)
@ -325,10 +325,9 @@ public class HttpOutput extends ServletOutputStream implements Runnable
{ {
if (_state != State.OPEN) if (_state != State.OPEN)
throw new IllegalStateException(stateString()); throw new IllegalStateException(stateString());
// TODO avoid this copy.
ByteBuffer content = _aggregate != null && _aggregate.hasRemaining() ? BufferUtil.copy(_aggregate.getByteBuffer()) : BufferUtil.EMPTY_BUFFER; ByteBuffer content = _aggregate != null && _aggregate.hasRemaining() ? BufferUtil.copy(_aggregate.getByteBuffer()) : BufferUtil.EMPTY_BUFFER;
_state = State.CLOSED; _state = State.CLOSED;
lockedReleaseBuffer(false); lockedReleaseBuffer();
return content; return content;
} }
} }
@ -458,7 +457,7 @@ public class HttpOutput extends ServletOutputStream implements Runnable
try (AutoLock ignored = _channelState.lock()) try (AutoLock ignored = _channelState.lock())
{ {
_state = State.CLOSED; _state = State.CLOSED;
lockedReleaseBuffer(failure != null); lockedReleaseBuffer();
} }
} }
@ -612,18 +611,13 @@ public class HttpOutput extends ServletOutputStream implements Runnable
return _aggregate; return _aggregate;
} }
private void lockedReleaseBuffer(boolean failure) private void lockedReleaseBuffer()
{ {
assert _channelState.isLockHeldByCurrentThread(); assert _channelState.isLockHeldByCurrentThread();
if (_aggregate != null) if (_aggregate != null)
{ {
if (failure && _pool != null) _aggregate.release();
_pool.removeAndRelease(_aggregate);
else
_aggregate.release();
_aggregate = null; _aggregate = null;
_pool = null;
} }
} }
@ -1265,7 +1259,7 @@ public class HttpOutput extends ServletOutputStream implements Runnable
{ {
try (AutoLock ignored = _channelState.lock()) try (AutoLock ignored = _channelState.lock())
{ {
lockedReleaseBuffer(_state != State.CLOSED); lockedReleaseBuffer();
_state = State.OPEN; _state = State.OPEN;
_apiState = ApiState.BLOCKING; _apiState = ApiState.BLOCKING;
_softClose = true; // Stay closed until next request _softClose = true; // Stay closed until next request
@ -1278,7 +1272,6 @@ public class HttpOutput extends ServletOutputStream implements Runnable
_writeListener = null; _writeListener = null;
_onError = null; _onError = null;
_firstByteNanoTime = -1; _firstByteNanoTime = -1;
_flushed = 0;
_closedCallback = null; _closedCallback = null;
} }
} }
@ -1418,10 +1411,19 @@ public class HttpOutput extends ServletOutputStream implements Runnable
} }
@Override @Override
public void onCompleteFailure(Throwable e) protected void onFailure(Throwable e)
{ {
onWriteComplete(_last, e); onWriteComplete(_last, e);
} }
@Override
protected void onCompleteFailure(Throwable cause)
{
try (AutoLock ignored = _channelState.lock())
{
lockedReleaseBuffer();
}
}
} }
private abstract class NestedChannelWriteCB extends ChannelWriteCB private abstract class NestedChannelWriteCB extends ChannelWriteCB
@ -1454,11 +1456,11 @@ public class HttpOutput extends ServletOutputStream implements Runnable
} }
@Override @Override
public void onCompleteFailure(Throwable e) protected void onFailure(Throwable e)
{ {
try try
{ {
super.onCompleteFailure(e); super.onFailure(e);
} }
catch (Throwable t) catch (Throwable t)
{ {
@ -1481,7 +1483,7 @@ public class HttpOutput extends ServletOutputStream implements Runnable
} }
@Override @Override
protected Action process() throws Exception protected Action process()
{ {
if (_aggregate != null && _aggregate.hasRemaining()) if (_aggregate != null && _aggregate.hasRemaining())
{ {
@ -1532,7 +1534,7 @@ public class HttpOutput extends ServletOutputStream implements Runnable
} }
@Override @Override
protected Action process() throws Exception protected Action process()
{ {
// flush any content from the aggregate // flush any content from the aggregate
if (_aggregate != null && _aggregate.hasRemaining()) if (_aggregate != null && _aggregate.hasRemaining())
@ -1655,15 +1657,18 @@ public class HttpOutput extends ServletOutputStream implements Runnable
{ {
_buffer.release(); _buffer.release();
IO.close(_in); IO.close(_in);
super.onCompleteSuccess(); }
@Override
protected void onFailure(Throwable cause)
{
IO.close(_in);
} }
@Override @Override
public void onCompleteFailure(Throwable x) public void onCompleteFailure(Throwable x)
{ {
_buffer.release(); _buffer.release();
IO.close(_in);
super.onCompleteFailure(x);
} }
} }
@ -1728,15 +1733,18 @@ public class HttpOutput extends ServletOutputStream implements Runnable
{ {
_buffer.release(); _buffer.release();
IO.close(_in); IO.close(_in);
super.onCompleteSuccess(); }
@Override
protected void onFailure(Throwable cause)
{
IO.close(_in);
} }
@Override @Override
public void onCompleteFailure(Throwable x) public void onCompleteFailure(Throwable x)
{ {
_buffer.release(); _buffer.release();
IO.close(_in);
super.onCompleteFailure(x);
} }
} }

2
jetty-ee9/jetty-ee9-nested/src/main/java/org/eclipse/jetty/ee9/nested/FileBufferedResponseHandler.java
View File

@ -221,7 +221,7 @@ public class FileBufferedResponseHandler extends BufferedResponseHandler
} }
@Override @Override
protected void onCompleteFailure(Throwable cause) protected void onFailure(Throwable cause)
{ {
dispose(); dispose();
callback.failed(cause); callback.failed(cause);

59
jetty-ee9/jetty-ee9-nested/src/main/java/org/eclipse/jetty/ee9/nested/HttpOutput.java
View File

@ -196,7 +196,6 @@ public class HttpOutput extends ServletOutputStream implements Runnable
private boolean _softClose = false; private boolean _softClose = false;
private Interceptor _interceptor; private Interceptor _interceptor;
private long _written; private long _written;
private long _flushed;
private long _firstByteNanoTime = -1; private long _firstByteNanoTime = -1;
private ByteBufferPool.Sized _pool; private ByteBufferPool.Sized _pool;
private RetainableByteBuffer _aggregate; private RetainableByteBuffer _aggregate;
@ -300,7 +299,8 @@ public class HttpOutput extends ServletOutputStream implements Runnable
_state = State.CLOSED; _state = State.CLOSED;
closedCallback = _closedCallback; closedCallback = _closedCallback;
_closedCallback = null; _closedCallback = null;
lockedReleaseBuffer(failure != null); if (failure == null)
lockedReleaseBuffer();
wake = updateApiState(failure); wake = updateApiState(failure);
} }
else if (_state == State.CLOSE) else if (_state == State.CLOSE)
@ -521,7 +521,7 @@ public class HttpOutput extends ServletOutputStream implements Runnable
try (AutoLock l = _channelState.lock()) try (AutoLock l = _channelState.lock())
{ {
_state = State.CLOSED; _state = State.CLOSED;
lockedReleaseBuffer(failure != null); lockedReleaseBuffer();
} }
} }
@ -672,16 +672,13 @@ public class HttpOutput extends ServletOutputStream implements Runnable
return _aggregate; return _aggregate;
} }
private void lockedReleaseBuffer(boolean failure) private void lockedReleaseBuffer()
{ {
assert _channelState.isLockHeldByCurrentThread(); assert _channelState.isLockHeldByCurrentThread();
if (_aggregate != null) if (_aggregate != null)
{ {
if (failure && _pool != null) _aggregate.release();
_pool.removeAndRelease(_aggregate);
else
_aggregate.release();
_aggregate = null; _aggregate = null;
_pool = null; _pool = null;
} }
@ -1455,7 +1452,7 @@ public class HttpOutput extends ServletOutputStream implements Runnable
{ {
try (AutoLock l = _channelState.lock()) try (AutoLock l = _channelState.lock())
{ {
lockedReleaseBuffer(_state != State.CLOSED); lockedReleaseBuffer();
_state = State.OPEN; _state = State.OPEN;
_apiState = ApiState.BLOCKING; _apiState = ApiState.BLOCKING;
_softClose = true; // Stay closed until next request _softClose = true; // Stay closed until next request
@ -1469,7 +1466,6 @@ public class HttpOutput extends ServletOutputStream implements Runnable
_writeListener = null; _writeListener = null;
_onError = null; _onError = null;
_firstByteNanoTime = -1; _firstByteNanoTime = -1;
_flushed = 0;
_closedCallback = null; _closedCallback = null;
} }
} }
@ -1616,10 +1612,19 @@ public class HttpOutput extends ServletOutputStream implements Runnable
} }
@Override @Override
public void onCompleteFailure(Throwable e) protected void onFailure(Throwable e)
{ {
onWriteComplete(_last, e); onWriteComplete(_last, e);
} }
@Override
protected void onCompleteFailure(Throwable cause)
{
try (AutoLock l = _channelState.lock())
{
lockedReleaseBuffer();
}
}
} }
private abstract class NestedChannelWriteCB extends ChannelWriteCB private abstract class NestedChannelWriteCB extends ChannelWriteCB
@ -1652,11 +1657,11 @@ public class HttpOutput extends ServletOutputStream implements Runnable
} }
@Override @Override
public void onCompleteFailure(Throwable e) protected void onFailure(Throwable e)
{ {
try try
{ {
super.onCompleteFailure(e); super.onFailure(e);
} }
catch (Throwable t) catch (Throwable t)
{ {
@ -1679,7 +1684,7 @@ public class HttpOutput extends ServletOutputStream implements Runnable
} }
@Override @Override
protected Action process() throws Exception protected Action process()
{ {
if (_aggregate != null && _aggregate.hasRemaining()) if (_aggregate != null && _aggregate.hasRemaining())
{ {
@ -1730,7 +1735,7 @@ public class HttpOutput extends ServletOutputStream implements Runnable
} }
@Override @Override
protected Action process() throws Exception protected Action process()
{ {
// flush any content from the aggregate // flush any content from the aggregate
if (_aggregate != null && _aggregate.hasRemaining()) if (_aggregate != null && _aggregate.hasRemaining())
@ -1850,17 +1855,23 @@ public class HttpOutput extends ServletOutputStream implements Runnable
@Override @Override
protected void onCompleteSuccess() protected void onCompleteSuccess()
{ {
super.onCompleteSuccess();
_buffer.release(); _buffer.release();
IO.close(_in); IO.close(_in);
super.onCompleteSuccess(); }
@Override
protected void onFailure(Throwable cause)
{
super.onFailure(cause);
IO.close(_in);
} }
@Override @Override
public void onCompleteFailure(Throwable x) public void onCompleteFailure(Throwable x)
{ {
_buffer.release();
IO.close(_in);
super.onCompleteFailure(x); super.onCompleteFailure(x);
_buffer.release();
} }
} }
@ -1921,17 +1932,23 @@ public class HttpOutput extends ServletOutputStream implements Runnable
@Override @Override
protected void onCompleteSuccess() protected void onCompleteSuccess()
{ {
super.onCompleteSuccess();
_buffer.release(); _buffer.release();
IO.close(_in); IO.close(_in);
super.onCompleteSuccess(); }
@Override
protected void onFailure(Throwable cause)
{
super.onFailure(cause);
IO.close(_in);
} }
@Override @Override
public void onCompleteFailure(Throwable x) public void onCompleteFailure(Throwable x)
{ {
_buffer.release();
IO.close(_in);
super.onCompleteFailure(x); super.onCompleteFailure(x);
_buffer.release();
} }
} }

2
jetty-ee9/jetty-ee9-proxy/src/main/java/org/eclipse/jetty/ee9/proxy/AsyncMiddleManServlet.java
View File

@ -400,7 +400,7 @@ public class AsyncMiddleManServlet extends AbstractProxyServlet
} }
@Override @Override
protected void onCompleteFailure(Throwable x) protected void onFailure(Throwable x)
{ {
onError(x); onError(x);
} }

2
jetty-ee9/jetty-ee9-proxy/src/main/java/org/eclipse/jetty/ee9/proxy/AsyncProxyServlet.java
View File

@ -192,7 +192,7 @@ public class AsyncProxyServlet extends ProxyServlet
} }
@Override @Override
protected void onCompleteFailure(Throwable cause) protected void onFailure(Throwable cause)
{ {
onError(cause); onError(cause);
} }

1
jetty-ee9/jetty-ee9-servlet/src/test/java/org/eclipse/jetty/ee9/servlet/ContextScopeListenerTest.java
View File

@ -103,6 +103,7 @@ public class ContextScopeListenerTest
{ {
throw new RuntimeException(e); throw new RuntimeException(e);
} }
Awaitility.waitAtMost(5, TimeUnit.SECONDS).until(() -> _history.size() == 5);
} }
}), "/"); }), "/");

3
jetty-ee9/jetty-ee9-webapp/src/main/java/org/eclipse/jetty/ee9/webapp/WebInfConfiguration.java
View File

@ -20,12 +20,9 @@ import java.nio.file.Files;
import java.nio.file.Path; import java.nio.file.Path;
import jakarta.servlet.ServletContext; import jakarta.servlet.ServletContext;
import org.eclipse.jetty.server.Connector;
import org.eclipse.jetty.server.NetworkConnector;
import org.eclipse.jetty.server.Server; import org.eclipse.jetty.server.Server;
import org.eclipse.jetty.util.FileID; import org.eclipse.jetty.util.FileID;
import org.eclipse.jetty.util.IO; import org.eclipse.jetty.util.IO;
import org.eclipse.jetty.util.StringUtil;
import org.eclipse.jetty.util.URIUtil; import org.eclipse.jetty.util.URIUtil;
import org.eclipse.jetty.util.resource.MountedPathResource; import org.eclipse.jetty.util.resource.MountedPathResource;
import org.eclipse.jetty.util.resource.Resource; import org.eclipse.jetty.util.resource.Resource;

8
jetty-home/pom.xml
View File

@ -236,14 +236,6 @@
<type>jar</type> <type>jar</type>
<optional>true</optional> <optional>true</optional>
</dependency> </dependency>
<dependency>
<groupId>org.eclipse.jetty.demos</groupId>
<artifactId>jetty-servlet5-demo-jndi-webapp</artifactId>
<version>${project.version}</version>
<classifier>config</classifier>
<type>jar</type>
<optional>true</optional>
</dependency>
<dependency> <dependency>
<groupId>org.eclipse.jetty.demos</groupId> <groupId>org.eclipse.jetty.demos</groupId>
<artifactId>jetty-servlet5-demo-jsp-webapp</artifactId> <artifactId>jetty-servlet5-demo-jsp-webapp</artifactId>