1515 */
1616package org .asynchttpclient .netty .timeout ;
1717
18+ import io .netty .util .Timeout ;
1819import io .netty .util .TimerTask ;
20+ import io .netty .util .concurrent .ScheduledFuture ;
1921import org .asynchttpclient .netty .NettyResponseFuture ;
2022import org .asynchttpclient .netty .request .NettyRequestSender ;
23+ import org .jetbrains .annotations .Nullable ;
2124import org .slf4j .Logger ;
2225import org .slf4j .LoggerFactory ;
2326
2427import java .net .InetSocketAddress ;
28+ import java .util .concurrent .RejectedExecutionException ;
2529import java .util .concurrent .TimeoutException ;
2630import java .util .concurrent .atomic .AtomicBoolean ;
2731
32+ /**
33+ * A timeout that can be armed either on a {@link io.netty.util.Timer} or on an event loop; which one an
34+ * exchange uses is {@link org.asynchttpclient.AsyncHttpClientConfig#isUseEventLoopTimeouts()}. An event loop
35+ * schedules {@link Runnable}s, so each subclass implements that as a second entry point into the same body.
36+ * This class stays a {@link TimerTask} alone: {@code Runnable#run} declares no checked exception, so a
37+ * {@code run()} here would have to catch what {@link TimerTask#run(Timeout)} declares and no subclass throws.
38+ */
2839public abstract class TimeoutTimerTask implements TimerTask {
2940
3041 private static final Logger LOGGER = LoggerFactory .getLogger (TimeoutTimerTask .class );
@@ -33,13 +44,67 @@ public abstract class TimeoutTimerTask implements TimerTask {
3344 protected final NettyRequestSender requestSender ;
3445 final TimeoutsHolder timeoutsHolder ;
3546 volatile NettyResponseFuture <?> nettyResponseFuture ;
47+ // The scheduled entry this task is armed on, one field per scheduler so that a scheduler changing its
48+ // return type is a compile error rather than a cancellation that silently stops working. At most one is
49+ // ever set. Held here rather than in a wrapper so arming allocates nothing beyond what the scheduler needs.
50+ private volatile @ Nullable Timeout timerHandle ;
51+ private volatile @ Nullable ScheduledFuture <?> loopHandle ;
3652
3753 TimeoutTimerTask (NettyResponseFuture <?> nettyResponseFuture , NettyRequestSender requestSender , TimeoutsHolder timeoutsHolder ) {
3854 this .nettyResponseFuture = nettyResponseFuture ;
3955 this .requestSender = requestSender ;
4056 this .timeoutsHolder = timeoutsHolder ;
4157 }
4258
59+ void armedOn (Timeout handle ) {
60+ // Each clears the other, so a handle left over from a previous arming cannot mask the live one and
61+ // leave its entry sitting in a scheduler, holding the future until a deadline nobody is waiting for.
62+ loopHandle = null ;
63+ timerHandle = handle ;
64+ }
65+
66+ void armedOn (ScheduledFuture <?> handle ) {
67+ timerHandle = null ;
68+ loopHandle = handle ;
69+ }
70+
71+ /**
72+ * Cancels the scheduled entry this task was armed on, if any. Never interrupts: on the event-loop path the
73+ * task may be running on the very thread this is called from, and nothing in it answers interruption.
74+ *
75+ * @return whether an entry was taken back out of its scheduler before it could run
76+ */
77+ boolean cancelArmed () {
78+ Timeout timer = timerHandle ;
79+ if (timer != null ) {
80+ timerHandle = null ;
81+ return timer .cancel ();
82+ }
83+ ScheduledFuture <?> scheduled = loopHandle ;
84+ if (scheduled != null ) {
85+ loopHandle = null ;
86+ try {
87+ return scheduled .cancel (false );
88+ } catch (RejectedExecutionException e ) {
89+ // Cancelling from off the loop enqueues the removal, which a loop that is already shutting down
90+ // rejects. The entry dies with the loop either way, and this runs under
91+ // ListenableFuture#cancel, which has never thrown for a client that is closing.
92+ LOGGER .debug ("Event loop rejected a timeout cancellation" , e );
93+ return false ;
94+ }
95+ }
96+ return false ;
97+ }
98+
99+ /**
100+ * Whether this task has been claimed, either by firing or by {@link #clean()}. Stands in for the
101+ * scheduler's own already-expired flag, which the two schedulers spell differently, and is if anything the
102+ * more precise of the two: it flips when {@code run} is entered rather than when the entry is marked.
103+ */
104+ boolean isClaimed () {
105+ return done .get ();
106+ }
107+
43108 void expire (String message , long time ) {
44109 LOGGER .debug ("{} for {} after {} ms" , message , nettyResponseFuture , time );
45110 requestSender .abort (nettyResponseFuture .channel (), nettyResponseFuture , new TimeoutException (message ));
0 commit comments