1 //
2 // ========================================================================
3 // Copyright (c) 1995-2014 Mort Bay Consulting Pty. Ltd.
4 // ------------------------------------------------------------------------
5 // All rights reserved. This program and the accompanying materials
6 // are made available under the terms of the Eclipse Public License v1.0
7 // and Apache License v2.0 which accompanies this distribution.
8 //
9 // The Eclipse Public License is available at
10 // http://www.eclipse.org/legal/epl-v10.html
11 //
12 // The Apache License v2.0 is available at
13 // http://www.opensource.org/licenses/apache2.0.php
14 //
15 // You may elect to redistribute this code under either of these licenses.
16 // ========================================================================
17 //
18
19 package org.eclipse.jetty.util;
20
21 import java.util.concurrent.atomic.AtomicReference;
22
23 /**
24 * This specialized callback implements a pattern that allows
25 * a large job to be broken into smaller tasks using iteration
26 * rather than recursion.
27 * <p/>
28 * A typical example is the write of a large content to a socket,
29 * divided in chunks. Chunk C1 is written by thread T1, which
30 * also invokes the callback, which writes chunk C2, which invokes
31 * the callback again, which writes chunk C3, and so forth.
32 * <p/>
33 * The problem with the example is that if the callback thread
34 * is the same that performs the I/O operation, then the process
35 * is recursive and may result in a stack overflow.
36 * To avoid the stack overflow, a thread dispatch must be performed,
37 * causing context switching and cache misses, affecting performance.
38 * <p/>
39 * To avoid this issue, this callback uses an AtomicReference to
40 * record whether success callback has been called during the processing
41 * of a sub task, and if so then the processing iterates rather than
42 * recurring.
43 * <p/>
44 * Subclasses must implement method {@link #process()} where the sub
45 * task is executed and a suitable {@link IteratingCallback.Action} is
46 * returned to this callback to indicate the overall progress of the job.
47 * This callback is passed to the asynchronous execution of each sub
48 * task and a call the {@link #succeeded()} on this callback represents
49 * the completion of the sub task.
50 */
51 public abstract class IteratingCallback implements Callback
52 {
53 /**
54 * The internal states of this callback
55 */
56 private enum State
57 {
58 /**
59 * This callback is inactive, ready to iterate.
60 */
61 INACTIVE,
62 /**
63 * This callback is iterating and {@link #process()} has scheduled an
64 * asynchronous operation by returning {@link Action#SCHEDULED}, but
65 * the operation is still undergoing.
66 */
67 ACTIVE,
68 /**
69 * This callback is iterating and {@link #process()} has been called
70 * but not returned yet.
71 */
72 ITERATING,
73 /**
74 * While this callback was iterating, another request for iteration
75 * has been issued, so the iteration must continue even if a previous
76 * call to {@link #process()} returned {@link Action#IDLE}.
77 */
78 ITERATE_AGAIN,
79 /**
80 * The overall job has succeeded.
81 */
82 SUCCEEDED,
83 /**
84 * The overall job has failed.
85 */
86 FAILED,
87 /**
88 * This callback has been closed and cannot be reset.
89 */
90 CLOSED
91 }
92
93 /**
94 * The indication of the overall progress of the overall job that
95 * implementations of {@link #process()} must return.
96 */
97 protected enum Action
98 {
99 /**
100 * Indicates that {@link #process()} has no more work to do,
101 * but the overall job is not completed yet, probably waiting
102 * for additional events to trigger more work.
103 */
104 IDLE,
105 /**
106 * Indicates that {@link #process()} is executing asynchronously
107 * a sub task, where the execution has started but the callback
108 * may have not yet been invoked.
109 */
110 SCHEDULED,
111 /**
112 * Indicates that {@link #process()} has completed the overall job.
113 */
114 SUCCEEDED
115 }
116
117 private final AtomicReference<State> _state;
118
119 protected IteratingCallback()
120 {
121 _state = new AtomicReference<>(State.INACTIVE);
122 }
123
124 protected IteratingCallback(boolean needReset)
125 {
126 _state = new AtomicReference<>(needReset ? State.SUCCEEDED : State.INACTIVE);
127 }
128
129 /**
130 * Method called by {@link #iterate()} to process the sub task.
131 * <p/>
132 * Implementations must start the asynchronous execution of the sub task
133 * (if any) and return an appropriate action:
134 * <ul>
135 * <li>{@link Action#IDLE} when no sub tasks are available for execution
136 * but the overall job is not completed yet</li>
137 * <li>{@link Action#SCHEDULED} when the sub task asynchronous execution
138 * has been started</li>
139 * <li>{@link Action#SUCCEEDED} when the overall job is completed</li>
140 * </ul>
141 *
142 * @throws Exception if the sub task processing throws
143 */
144 protected abstract Action process() throws Exception;
145
146 /**
147 * @deprecated Use {@link #onCompleteSuccess()} instead.
148 */
149 @Deprecated
150 protected void completed()
151 {
152 }
153
154 /**
155 * Invoked when the overall task has completed successfully.
156 *
157 * @see #onCompleteFailure(Throwable)
158 */
159 protected void onCompleteSuccess()
160 {
161 completed();
162 }
163
164 /**
165 * Invoked when the overall task has completed with a failure.
166 *
167 * @see #onCompleteSuccess()
168 */
169 protected void onCompleteFailure(Throwable x)
170 {
171 }
172
173 /**
174 * This method must be invoked by applications to start the processing
175 * of sub tasks.
176 * <p/>
177 * If {@link #process()} returns {@link Action#IDLE}, then this method
178 * should be called again to restart processing.
179 * It is safe to call iterate multiple times from multiple threads since only
180 * the first thread to move the state out of INACTIVE will actually do any iteration
181 * and processing.
182 */
183 public void iterate()
184 {
185 try
186 {
187 while (true)
188 {
189 switch (_state.get())
190 {
191 case INACTIVE:
192 {
193 if (processIterations())
194 return;
195 break;
196 }
197 case ITERATING:
198 {
199 if (_state.compareAndSet(State.ITERATING, State.ITERATE_AGAIN))
200 return;
201 break;
202 }
203 default:
204 {
205 return;
206 }
207 }
208 }
209 }
210 catch (Throwable x)
211 {
212 failed(x);
213 }
214 }
215
216 private boolean processIterations() throws Exception
217 {
218 // Keeps iterating as long as succeeded() is called during process().
219 // If we are in INACTIVE state, either this is the first iteration or
220 // succeeded()/failed() were called already.
221 while (_state.compareAndSet(State.INACTIVE, State.ITERATING))
222 {
223 // Method process() can only be called by one thread at a time because
224 // it is guarded by the CaS above. However, the case blocks below may
225 // be executed concurrently in this case: T1 calls process() which
226 // executes the asynchronous sub task, which calls succeeded(), which
227 // moves the state into INACTIVE, then returns SCHEDULED; T2 calls
228 // iterate(), state is now INACTIVE and process() is called again and
229 // returns another action. Now we have 2 threads that may execute the
230 // action case blocks below concurrently; therefore each case block
231 // has to be prepared to fail the CaS it's doing.
232
233 Action action = process();
234 switch (action)
235 {
236 case IDLE:
237 {
238 // No more progress can be made.
239 if (_state.compareAndSet(State.ITERATING, State.INACTIVE))
240 return true;
241
242 // Was iterate() called again since we already decided to go INACTIVE ?
243 // If so, try another iteration as more work may have been added
244 // while the previous call to process() was returning.
245 if (_state.compareAndSet(State.ITERATE_AGAIN, State.INACTIVE))
246 continue;
247
248 // State may have changed concurrently, try again.
249 continue;
250 }
251 case SCHEDULED:
252 {
253 // The sub task is executing, and the callback for it may or
254 // may not have already been called yet, which we figure out below.
255 // Can double CaS here because state never changes directly ITERATING_AGAIN --> ITERATE.
256 if (_state.compareAndSet(State.ITERATING, State.ACTIVE) ||
257 _state.compareAndSet(State.ITERATE_AGAIN, State.ACTIVE))
258 // Not called back yet, so wait.
259 return true;
260 // Call back must have happened, so iterate.
261 continue;
262 }
263 case SUCCEEDED:
264 {
265 // The overall job has completed.
266 while (true)
267 {
268 State current = _state.get();
269 switch(current)
270 {
271 case SUCCEEDED:
272 case FAILED:
273 // Already complete!.
274 return true;
275 case CLOSED:
276 throw new IllegalStateException();
277 default:
278 if (_state.compareAndSet(current, State.SUCCEEDED))
279 {
280 onCompleteSuccess();
281 return true;
282 }
283 }
284 }
285 }
286 default:
287 {
288 throw new IllegalStateException(toString());
289 }
290 }
291 }
292 return false;
293 }
294
295 /**
296 * Invoked when the sub task succeeds.
297 * Subclasses that override this method must always remember to call
298 * {@code super.succeeded()}.
299 */
300 @Override
301 public void succeeded()
302 {
303 while (true)
304 {
305 State current = _state.get();
306 switch (current)
307 {
308 case ITERATE_AGAIN:
309 case ITERATING:
310 {
311 if (_state.compareAndSet(current, State.INACTIVE))
312 return;
313 continue;
314 }
315 case ACTIVE:
316 {
317 // If we can move from ACTIVE to INACTIVE
318 // then we are responsible to call iterate().
319 if (_state.compareAndSet(current, State.INACTIVE))
320 iterate();
321 // If we can't CaS, then failed() must have been
322 // called, and we just return.
323 return;
324 }
325 case INACTIVE:
326 {
327 // Support the case where the callback is scheduled
328 // externally without a call to iterate().
329 iterate();
330 return;
331 }
332 case CLOSED:
333 {
334 // Too late!
335 return;
336 }
337 default:
338 {
339 throw new IllegalStateException(toString());
340 }
341 }
342 }
343 }
344
345 /**
346 * Invoked when the sub task fails.
347 * Subclasses that override this method must always remember to call
348 * {@code super.failed(Throwable)}.
349 */
350 @Override
351 public void failed(Throwable x)
352 {
353 while (true)
354 {
355 State current = _state.get();
356 switch (current)
357 {
358 case SUCCEEDED:
359 case FAILED:
360 case INACTIVE:
361 case CLOSED:
362 {
363 // Already complete!.
364 return;
365 }
366 default:
367 {
368 if (_state.compareAndSet(current, State.FAILED))
369 {
370 onCompleteFailure(x);
371 return;
372 }
373 }
374 }
375 }
376 }
377
378 public void close()
379 {
380 while (true)
381 {
382 State current = _state.get();
383 switch (current)
384 {
385 case INACTIVE:
386 case SUCCEEDED:
387 case FAILED:
388 {
389 if (_state.compareAndSet(current, State.CLOSED))
390 return;
391 break;
392 }
393 default:
394 {
395 if (_state.compareAndSet(current, State.CLOSED))
396 {
397 onCompleteFailure(new IllegalStateException("Closed with pending callback " + this));
398 return;
399 }
400 }
401 }
402 }
403 }
404
405 /*
406 * only for testing
407 * @return whether this callback is idle and {@link #iterate()} needs to be called
408 */
409 boolean isIdle()
410 {
411 return _state.get() == State.INACTIVE;
412 }
413
414 public boolean isClosed()
415 {
416 return _state.get() == State.CLOSED;
417 }
418
419 /**
420 * @return whether this callback has failed
421 */
422 public boolean isFailed()
423 {
424 return _state.get() == State.FAILED;
425 }
426
427 /**
428 * @return whether this callback has succeeded
429 */
430 public boolean isSucceeded()
431 {
432 return _state.get() == State.SUCCEEDED;
433 }
434
435 /**
436 * Resets this callback.
437 * <p/>
438 * A callback can only be reset to INACTIVE from the
439 * SUCCEEDED or FAILED states or if it is already INACTIVE.
440 *
441 * @return true if the reset was successful
442 */
443 public boolean reset()
444 {
445 while (true)
446 {
447 switch(_state.get())
448 {
449 case INACTIVE:
450 return true;
451
452 case SUCCEEDED:
453 if (_state.compareAndSet(State.SUCCEEDED, State.INACTIVE))
454 return true;
455 break;
456
457 case FAILED:
458 if (_state.compareAndSet(State.FAILED, State.INACTIVE))
459 return true;
460 break;
461
462 default:
463 return false;
464 }
465 }
466 }
467
468 @Override
469 public String toString()
470 {
471 return String.format("%s[%s]", super.toString(), _state);
472 }
473 }