1 //
2 // ========================================================================
3 // Copyright (c) 1995-2013 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 /* ------------------------------------------------------------ */
25 /** Iterating Callback.
26 * <p>This specialized callback is used when breaking up an
27 * asynchronous task into smaller asynchronous tasks. A typical pattern
28 * is that a successful callback is used to schedule the next sub task, but
29 * if that task completes quickly and uses the calling thread to callback
30 * the success notification, this can result in a growing stack depth.
31 * </p>
32 * <p>To avoid this issue, this callback uses an AtomicReference to note
33 * if the success callback has been called during the processing of a
34 * sub task, and if so then the processing iterates rather than recurses.
35 * </p>
36 * <p>This callback is passed to the asynchronous handling of each sub
37 * task and a call the {@link #succeeded()} on this call back represents
38 * completion of the subtask. Only once all the subtasks are completed is
39 * the {#completed()} method called.</p>
40 *
41 */
42 public abstract class IteratingCallback implements Callback
43 {
44 private enum State { WAITING, ITERATING, SUCCEEDED, FAILED };
45 private final AtomicReference<State> _state = new AtomicReference<>(State.WAITING);
46
47 public IteratingCallback()
48 {
49 }
50
51 /* ------------------------------------------------------------ */
52 /**
53 * Process a subtask.
54 * <p>Called by {@link #iterate()} to process a sub task of the overall task
55 * <p>
56 * @return True if the total task is complete. If false is returned
57 * then this Callback must be scheduled to receive either a call to
58 * {@link #succeeded()} or {@link #failed(Throwable)}.
59 * @throws Exception
60 */
61 abstract protected boolean process() throws Exception;
62
63 abstract protected void completed();
64
65 /* ------------------------------------------------------------ */
66 /** This method is called initially to start processing and
67 * is then called by subsequent sub task success to continue
68 * processing.
69 */
70 public void iterate()
71 {
72 try
73 {
74 // Keep iterating as long as succeeded() is called during process()
75 // If we are in WAITING state, either this is the first iteration or
76 // succeeded()/failed() were called already.
77 while(_state.compareAndSet(State.WAITING,State.ITERATING))
78 {
79 // Make some progress by calling process()
80 if (process())
81 {
82 // A true return indicates we are finished and no further callbacks
83 // are scheduled. So we must still be ITERATING.
84 if (_state.compareAndSet(State.ITERATING,State.SUCCEEDED))
85 completed();
86 else
87 throw new IllegalStateException("Already "+_state.get());
88 return;
89 }
90 // else a callback has been scheduled. If it has not happened yet,
91 // we will still be ITERATING
92 else if (_state.compareAndSet(State.ITERATING,State.WAITING))
93 // no callback yet, so break the loop and wait for it
94 break;
95
96 // The callback must have happened and we are either WAITING already or FAILED
97 // the loop test will work out which
98 }
99 }
100 catch(Exception e)
101 {
102 failed(e);
103 }
104 }
105
106 /* ------------------------------------------------------------ */
107 @Override
108 public void succeeded()
109 {
110 // Try a short cut for the fast method. If we are still iterating
111 if (_state.compareAndSet(State.ITERATING,State.WAITING))
112 // then next loop will continue processing, so nothing to do here
113 return;
114
115 // OK do it properly
116 loop: while(true)
117 {
118 switch(_state.get())
119 {
120 case ITERATING:
121 if (_state.compareAndSet(State.ITERATING,State.WAITING))
122 break loop;
123 continue;
124
125 case WAITING:
126 // we are really waiting, so use this callback thread to iterate some more
127 iterate();
128 break loop;
129
130 default:
131 throw new IllegalStateException("Already "+_state.get());
132 }
133 }
134 }
135
136 /* ------------------------------------------------------------ */
137 /**
138 * Derivations of this method should always call super.failed(x)
139 * to check the state before handling the failure.
140 * @see org.eclipse.jetty.util.Callback#failed(java.lang.Throwable)
141 */
142 @Override
143 public void failed(Throwable x)
144 {
145 loop: while(true)
146 {
147 switch(_state.get())
148 {
149 case ITERATING:
150 if (_state.compareAndSet(State.ITERATING,State.FAILED))
151 break loop;
152 continue;
153
154 case WAITING:
155 if (_state.compareAndSet(State.WAITING,State.FAILED))
156 break loop;
157 continue;
158
159 default:
160 throw new IllegalStateException("Already "+_state.get(),x);
161 }
162 }
163 }
164 }