Enforces the total timeout for for exchanges that are still in the queue.
+ *The total timeout for exchanges that are not in the destination queue + * is enforced in {@link HttpChannel} by {@link TimeoutCompleteListener}.
*/ - private class TimeoutTask extends CyclicTimeout + private class RequestTimeouts extends CyclicTimeout { - private final AtomicLong nextTimeout = new AtomicLong(Long.MAX_VALUE); + private final AtomicLong earliestTimeout = new AtomicLong(Long.MAX_VALUE); - private TimeoutTask(Scheduler scheduler) + private RequestTimeouts(Scheduler scheduler) { super(scheduler); } @@ -564,14 +564,17 @@ private TimeoutTask(Scheduler scheduler) public void onTimeoutExpired() { if (LOG.isDebugEnabled()) - LOG.debug("{} timeout expired", this); + LOG.debug("{} timeouts check", this); - nextTimeout.set(Long.MAX_VALUE); long now = System.nanoTime(); - long nextExpiresAt = Long.MAX_VALUE; - // Check all queued exchanges for those that have expired - // and to determine when the next check must be. + // Reset the earliest timeout so we can expire again. + // A concurrent call to schedule(long) may lose an earliest + // value, but the corresponding exchange is already enqueued + // and will be seen by scanning the exchange queue below. + earliestTimeout.set(Long.MAX_VALUE); + + long earliest = Long.MAX_VALUE; for (HttpExchange exchange : exchanges) { HttpRequest request = exchange.getRequest(); @@ -580,34 +583,27 @@ public void onTimeoutExpired() continue; if (expiresAt <= now) request.abort(new TimeoutException("Total timeout " + request.getTimeout() + " ms elapsed")); - else if (expiresAt < nextExpiresAt) - nextExpiresAt = expiresAt; + else if (expiresAt < earliest) + earliest = expiresAt; } - if (nextExpiresAt < Long.MAX_VALUE && client.isRunning()) - schedule(nextExpiresAt); + if (earliest < Long.MAX_VALUE && client.isRunning()) + schedule(earliest); } private void schedule(long expiresAt) { - // Schedule a timeout for the soonest any known exchange can expire. - // If subsequently that exchange is removed from the queue, the - // timeout is not cancelled, instead the entire queue is swept - // for expired exchanges and a new timeout is set. - long timeoutAt = nextTimeout.getAndUpdate(e -> Math.min(e, expiresAt)); - if (timeoutAt != expiresAt) + // Schedule a timeout for the earliest exchange that may expire. + // When the timeout expires, scan the exchange queue for the next + // earliest exchange that may expire, and reschedule a new timeout. + long earliest = earliestTimeout.getAndUpdate(t -> Math.min(t, expiresAt)); + if (expiresAt != earliest) { - long delay = expiresAt - System.nanoTime(); - if (delay <= 0) - { - onTimeoutExpired(); - } - else - { - schedule(delay, TimeUnit.NANOSECONDS); - if (LOG.isDebugEnabled()) - LOG.debug("{} scheduled timeout in {} ms", this, TimeUnit.NANOSECONDS.toMillis(delay)); - } + // A new request expires earlier than previous requests, schedule it. + long delay = Math.max(0, expiresAt - System.nanoTime()); + if (LOG.isDebugEnabled()) + LOG.debug("{} scheduling timeout in {} ms", this, TimeUnit.NANOSECONDS.toMillis(delay)); + schedule(delay, TimeUnit.NANOSECONDS); } } } diff --git a/jetty-client/src/main/java/org/eclipse/jetty/client/TimeoutCompleteListener.java b/jetty-client/src/main/java/org/eclipse/jetty/client/TimeoutCompleteListener.java index 7457fbd32e1c..464178857797 100644 --- a/jetty-client/src/main/java/org/eclipse/jetty/client/TimeoutCompleteListener.java +++ b/jetty-client/src/main/java/org/eclipse/jetty/client/TimeoutCompleteListener.java @@ -34,7 +34,7 @@ public class TimeoutCompleteListener extends CyclicTimeout implements Response.C { private static final Logger LOG = Log.getLogger(TimeoutCompleteListener.class); - private final AtomicReferenceSubclasses should implement {@link #onTimeoutExpired()}.
*This implementation is optimised assuming that the timeout * will mostly be cancelled and then reused with a similar value.
+ *The typical scenario to use this class is when you have events + * that postpone (by re-scheduling), or cancel then re-schedule, a + * timeout for a single entity. + * For example: connection idleness, where for each connection there + * is a CyclicTimeout and a read/write postpones the timeout; when + * the timeout expires, the implementation checks against a timestamp + * if the connection is really idle. + * Another example: HTTP session expiration, where for each HTTP + * session there is a CyclicTimeout and at the beginning of the + * request processing the timeout is canceled (via cancel()), but at + * the end of the request processing the timeout is re-scheduled.
+ *Another typical scenario is for a parent entity to manage + * the timeouts of many children entities; the timeout is scheduled + * for the child entity that expires the earlier; when the timeout + * expires, the implementation scans the children entities to find + * the expired child entities and to find the next child entity + * that expires the earlier.
*This implementation has a {@link Timeout} holding the time * at which the scheduled task should fire, and a linked list of * {@link Wakeup}, each holding the actual scheduled task.
@@ -102,8 +119,9 @@ public boolean schedule(long delay, TimeUnit units) if (_timeout.compareAndSet(timeout, new Timeout(newTimeoutAt, wakeup))) { if (LOG.isDebugEnabled()) - LOG.debug("Installed timeout in {} ms, waking up in {} ms", + LOG.debug("Installed timeout in {} ms, {} wake up in {} ms", units.toMillis(delay), + newWakeup != null ? "new" : "existing", TimeUnit.NANOSECONDS.toMillis(wakeup._at - now)); break; }