1 //
2 // ========================================================================
3 // Copyright (c) 1995-2015 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.nio.channels.ClosedChannelException;
22 import java.util.concurrent.atomic.AtomicReference;
23
24 /**
25 * This specialized callback implements a pattern that allows
26 * a large job to be broken into smaller tasks using iteration
27 * rather than recursion.
28 * <p/>
29 * A typical example is the write of a large content to a socket,
30 * divided in chunks. Chunk C1 is written by thread T1, which
31 * also invokes the callback, which writes chunk C2, which invokes
32 * the callback again, which writes chunk C3, and so forth.
33 * <p/>
34 * The problem with the example is that if the callback thread
35 * is the same that performs the I/O operation, then the process
36 * is recursive and may result in a stack overflow.
37 * To avoid the stack overflow, a thread dispatch must be performed,
38 * causing context switching and cache misses, affecting performance.
39 * <p/>
40 * To avoid this issue, this callback uses an AtomicReference to
41 * record whether success callback has been called during the processing
42 * of a sub task, and if so then the processing iterates rather than
43 * recurring.
44 * <p/>
45 * Subclasses must implement method {@link #process()} where the sub
46 * task is executed and a suitable {@link IteratingCallback.Action} is
47 * returned to this callback to indicate the overall progress of the job.
48 * This callback is passed to the asynchronous execution of each sub
49 * task and a call the {@link #succeeded()} on this callback represents
50 * the completion of the sub task.
51 */
52 public abstract class IteratingCallback implements Callback
53 {
54 /**
55 * The internal states of this callback
56 */
57 private enum State
58 {
59 /**
60 * This callback is IDLE, ready to iterate.
61 */
62 IDLE,
63
64 /**
65 * This callback is iterating calls to {@link #process()} and is dealing with
66 * the returns. To get into processing state, it much of held the lock state
67 * and set iterating to true.
68 */
69 PROCESSING,
70
71 /**
72 * Waiting for a schedule callback
73 */
74 PENDING,
75
76 /**
77 * Called by a schedule callback
78 */
79 CALLED,
80
81 /**
82 * The overall job has succeeded as indicated by a {@link Action#SUCCEEDED} return
83 * from {@link IteratingCallback#process()}
84 */
85 SUCCEEDED,
86
87 /**
88 * The overall job has failed as indicated by a call to {@link IteratingCallback#failed(Throwable)}
89 */
90 FAILED,
91
92 /**
93 * This callback has been closed and cannot be reset.
94 */
95 CLOSED,
96
97 /**
98 * State is locked while leaving processing state to check the iterate boolean
99 */
100 LOCKED
101 }
102
103 /**
104 * The indication of the overall progress of the overall job that
105 * implementations of {@link #process()} must return.
106 */
107 protected enum Action
108 {
109 /**
110 * Indicates that {@link #process()} has no more work to do,
111 * but the overall job is not completed yet, probably waiting
112 * for additional events to trigger more work.
113 */
114 IDLE,
115 /**
116 * Indicates that {@link #process()} is executing asynchronously
117 * a sub task, where the execution has started but the callback
118 * may have not yet been invoked.
119 */
120 SCHEDULED,
121
122 /**
123 * Indicates that {@link #process()} has completed the overall job.
124 */
125 SUCCEEDED
126 }
127
128 private final AtomicReference<State> _state;
129 private boolean _iterate;
130
131
132 protected IteratingCallback()
133 {
134 _state = new AtomicReference<>(State.IDLE);
135 }
136
137 protected IteratingCallback(boolean needReset)
138 {
139 _state = new AtomicReference<>(needReset ? State.SUCCEEDED : State.IDLE);
140 }
141
142 /**
143 * Method called by {@link #iterate()} to process the sub task.
144 * <p/>
145 * Implementations must start the asynchronous execution of the sub task
146 * (if any) and return an appropriate action:
147 * <ul>
148 * <li>{@link Action#IDLE} when no sub tasks are available for execution
149 * but the overall job is not completed yet</li>
150 * <li>{@link Action#SCHEDULED} when the sub task asynchronous execution
151 * has been started</li>
152 * <li>{@link Action#SUCCEEDED} when the overall job is completed</li>
153 * </ul>
154 *
155 * @throws Exception if the sub task processing throws
156 */
157 protected abstract Action process() throws Exception;
158
159 /**
160 * @deprecated Use {@link #onCompleteSuccess()} instead.
161 */
162 @Deprecated
163 protected void completed()
164 {
165 }
166
167 /**
168 * Invoked when the overall task has completed successfully.
169 *
170 * @see #onCompleteFailure(Throwable)
171 */
172 protected void onCompleteSuccess()
173 {
174 completed();
175 }
176
177 /**
178 * Invoked when the overall task has completed with a failure.
179 *
180 * @see #onCompleteSuccess()
181 */
182 protected void onCompleteFailure(Throwable x)
183 {
184 }
185
186 /**
187 * This method must be invoked by applications to start the processing
188 * of sub tasks. It can be called at any time by any thread, and it's
189 * contract is that when called, then the {@link #process()} method will
190 * be called during or soon after, either by the calling thread or by
191 * another thread.
192 */
193 public void iterate()
194 {
195 loop: while (true)
196 {
197 State state=_state.get();
198 switch (state)
199 {
200 case PENDING:
201 case CALLED:
202 // process will be called when callback is handled
203 break loop;
204
205 case IDLE:
206 if (!_state.compareAndSet(state,State.PROCESSING))
207 continue;
208 processing();
209 break loop;
210
211 case PROCESSING:
212 if (!_state.compareAndSet(state,State.LOCKED))
213 continue;
214 // Tell the thread that is processing that it must iterate again
215 _iterate=true;
216 _state.set(State.PROCESSING);
217 break loop;
218
219 case LOCKED:
220 Thread.yield();
221 continue loop;
222
223 case FAILED:
224 case SUCCEEDED:
225 break loop;
226
227 case CLOSED:
228 default:
229 throw new IllegalStateException("state="+state);
230 }
231 }
232 }
233
234 private void processing()
235 {
236 // This should only ever be called when in processing state, however a failed or close call
237 // may happen concurrently, so state is not assumed.
238
239 // While we are processing
240 processing: while (true)
241 {
242 // Call process to get the action that we have to take.
243 Action action;
244 try
245 {
246 action = process();
247 }
248 catch (Throwable x)
249 {
250 failed(x);
251 break processing;
252 }
253
254 // loop until we have successfully acted on the action we have just received
255 acting: while(true)
256 {
257 // action handling needs to know the state
258 State state=_state.get();
259
260 switch (state)
261 {
262 case PROCESSING:
263 {
264 switch (action)
265 {
266 case IDLE:
267 {
268 // lock the state
269 if (!_state.compareAndSet(state,State.LOCKED))
270 continue acting;
271
272 // Has iterate been called while we were processing?
273 if (_iterate)
274 {
275 // yes, so skip idle and keep processing
276 _iterate=false;
277 _state.set(State.PROCESSING);
278 continue processing;
279 }
280
281 // No, so we can go idle
282 _state.set(State.IDLE);
283 break processing;
284 }
285
286 case SCHEDULED:
287 {
288 if (!_state.compareAndSet(state, State.PENDING))
289 continue acting;
290 // we won the race against the callback, so the callback has to process and we can break processing
291 break processing;
292 }
293
294 case SUCCEEDED:
295 {
296 if (!_state.compareAndSet(state, State.LOCKED))
297 continue acting;
298 _iterate=false;
299 _state.set(State.SUCCEEDED);
300 onCompleteSuccess();
301 break processing;
302 }
303
304 default:
305 throw new IllegalStateException("state="+state+" action="+action);
306 }
307 }
308
309 case CALLED:
310 {
311 switch (action)
312 {
313 case SCHEDULED:
314 {
315 if (!_state.compareAndSet(state, State.PROCESSING))
316 continue acting;
317 // we lost the race, so we have to keep processing
318 continue processing;
319 }
320
321 default:
322 throw new IllegalStateException("state="+state+" action="+action);
323 }
324 }
325
326 case LOCKED:
327 Thread.yield();
328 continue acting;
329
330 case SUCCEEDED:
331 case FAILED:
332 case CLOSED:
333 break processing;
334
335 case IDLE:
336 case PENDING:
337 default:
338 throw new IllegalStateException("state="+state+" action="+action);
339 }
340 }
341 }
342 }
343
344 /**
345 * Invoked when the sub task succeeds.
346 * Subclasses that override this method must always remember to call
347 * {@code super.succeeded()}.
348 */
349 @Override
350 public void succeeded()
351 {
352 loop: while (true)
353 {
354 State state = _state.get();
355 switch (state)
356 {
357 case PROCESSING:
358 {
359 if (!_state.compareAndSet(state, State.CALLED))
360 continue loop;
361 break loop;
362 }
363 case PENDING:
364 {
365 if (!_state.compareAndSet(state, State.PROCESSING))
366 continue loop;
367 processing();
368 break loop;
369 }
370 case CLOSED:
371 case FAILED:
372 {
373 // Too late!
374 break loop;
375 }
376 case LOCKED:
377 {
378 Thread.yield();
379 continue loop;
380 }
381 default:
382 {
383 throw new IllegalStateException("state="+state);
384 }
385 }
386 }
387 }
388
389 /**
390 * Invoked when the sub task fails.
391 * Subclasses that override this method must always remember to call
392 * {@code super.failed(Throwable)}.
393 */
394 @Override
395 public void failed(Throwable x)
396 {
397 loop: while (true)
398 {
399 State state = _state.get();
400 switch (state)
401 {
402 case SUCCEEDED:
403 case FAILED:
404 case IDLE:
405 case CLOSED:
406 case CALLED:
407 {
408 // too late!.
409 break loop;
410 }
411 case LOCKED:
412 {
413 Thread.yield();
414 continue loop;
415 }
416 case PENDING:
417 case PROCESSING:
418 {
419 if (!_state.compareAndSet(state, State.FAILED))
420 continue loop;
421
422 onCompleteFailure(x);
423 break loop;
424 }
425 default:
426 throw new IllegalStateException("state="+state);
427 }
428 }
429 }
430
431 public void close()
432 {
433 loop: while (true)
434 {
435 State state = _state.get();
436 switch (state)
437 {
438 case IDLE:
439 case SUCCEEDED:
440 case FAILED:
441 {
442 if (!_state.compareAndSet(state, State.CLOSED))
443 continue loop;
444 break loop;
445 }
446 case CLOSED:
447 {
448 break loop;
449 }
450 case LOCKED:
451 {
452 Thread.yield();
453 continue loop;
454 }
455 default:
456 {
457 if (!_state.compareAndSet(state, State.CLOSED))
458 continue loop;
459 onCompleteFailure(new ClosedChannelException());
460 break loop;
461 }
462 }
463 }
464 }
465
466 /*
467 * only for testing
468 * @return whether this callback is idle and {@link #iterate()} needs to be called
469 */
470 boolean isIdle()
471 {
472 return _state.get() == State.IDLE;
473 }
474
475 public boolean isClosed()
476 {
477 return _state.get() == State.CLOSED;
478 }
479
480 /**
481 * @return whether this callback has failed
482 */
483 public boolean isFailed()
484 {
485 return _state.get() == State.FAILED;
486 }
487
488 /**
489 * @return whether this callback has succeeded
490 */
491 public boolean isSucceeded()
492 {
493 return _state.get() == State.SUCCEEDED;
494 }
495
496 /**
497 * Resets this callback.
498 * <p/>
499 * A callback can only be reset to IDLE from the
500 * SUCCEEDED or FAILED states or if it is already IDLE.
501 *
502 * @return true if the reset was successful
503 */
504 public boolean reset()
505 {
506 while (true)
507 {
508 State state=_state.get();
509 switch(state)
510 {
511 case IDLE:
512 return true;
513
514 case SUCCEEDED:
515 if (!_state.compareAndSet(state, State.LOCKED))
516 continue;
517 _iterate=false;
518 _state.set(State.IDLE);
519 return true;
520
521 case FAILED:
522 if (!_state.compareAndSet(state, State.LOCKED))
523 continue;
524 _iterate=false;
525 _state.set(State.IDLE);
526 return true;
527
528 case LOCKED:
529 Thread.yield();
530 continue;
531
532 default:
533 return false;
534 }
535 }
536 }
537
538 @Override
539 public String toString()
540 {
541 return String.format("%s[%s]", super.toString(), _state);
542 }
543 }