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.io;
20
21 import java.io.IOException;
22 import java.nio.ByteBuffer;
23 import java.nio.channels.ClosedChannelException;
24 import java.nio.channels.WritePendingException;
25 import java.util.EnumMap;
26 import java.util.EnumSet;
27 import java.util.Set;
28 import java.util.concurrent.atomic.AtomicReference;
29
30 import org.eclipse.jetty.util.BufferUtil;
31 import org.eclipse.jetty.util.Callback;
32 import org.eclipse.jetty.util.log.Log;
33 import org.eclipse.jetty.util.log.Logger;
34
35
36 /**
37 * A Utility class to help implement {@link EndPoint#write(Callback, ByteBuffer...)} by calling
38 * {@link EndPoint#flush(ByteBuffer...)} until all content is written.
39 * The abstract method {@link #onIncompleteFlushed()} is called when not all content has been written after a call to
40 * flush and should organise for the {@link #completeWrite()} method to be called when a subsequent call to flush
41 * should be able to make more progress.
42 * <p>
43 */
44 abstract public class WriteFlusher
45 {
46 private static final Logger LOG = Log.getLogger(WriteFlusher.class);
47 private static final boolean DEBUG = LOG.isDebugEnabled(); // Easy for the compiler to remove the code if DEBUG==false
48 private static final ByteBuffer[] EMPTY_BUFFERS = new ByteBuffer[0];
49 private static final EnumMap<StateType, Set<StateType>> __stateTransitions = new EnumMap<>(StateType.class);
50 private static final State __IDLE = new IdleState();
51 private static final State __WRITING = new WritingState();
52 private static final State __COMPLETING = new CompletingState();
53 private final EndPoint _endPoint;
54 private final AtomicReference<State> _state = new AtomicReference<>();
55
56 static
57 {
58 // fill the state machine
59 __stateTransitions.put(StateType.IDLE, EnumSet.of(StateType.WRITING));
60 __stateTransitions.put(StateType.WRITING, EnumSet.of(StateType.IDLE, StateType.PENDING, StateType.FAILED));
61 __stateTransitions.put(StateType.PENDING, EnumSet.of(StateType.COMPLETING,StateType.IDLE));
62 __stateTransitions.put(StateType.COMPLETING, EnumSet.of(StateType.IDLE, StateType.PENDING, StateType.FAILED));
63 __stateTransitions.put(StateType.FAILED, EnumSet.of(StateType.IDLE));
64 }
65
66 // A write operation may either complete immediately:
67 // IDLE-->WRITING-->IDLE
68 // Or it may not completely flush and go via the PENDING state
69 // IDLE-->WRITING-->PENDING-->COMPLETING-->IDLE
70 // Or it may take several cycles to complete
71 // IDLE-->WRITING-->PENDING-->COMPLETING-->PENDING-->COMPLETING-->IDLE
72 //
73 // If a failure happens while in IDLE, it is a noop since there is no operation to tell of the failure.
74 // If a failure happens while in WRITING, but the the write has finished successfully or with an IOExceptions,
75 // the callback's complete or respectively failed methods will be called.
76 // If a failure happens in PENDING state, then the fail method calls the pending callback and moves to IDLE state
77 //
78 // IDLE--(fail)-->IDLE
79 // IDLE-->WRITING--(fail)-->FAILED-->IDLE
80 // IDLE-->WRITING-->PENDING--(fail)-->IDLE
81 // IDLE-->WRITING-->PENDING-->COMPLETING--(fail)-->FAILED-->IDLE
82 //
83 // So a call to fail in the PENDING state will be directly handled and the state changed to IDLE
84 // A call to fail in the WRITING or COMPLETING states will just set the state to FAILED and the failure will be
85 // handled with the write or completeWrite methods try to move the state from what they thought it was.
86 //
87
88 protected WriteFlusher(EndPoint endPoint)
89 {
90 _state.set(__IDLE);
91 _endPoint = endPoint;
92 }
93
94 private enum StateType
95 {
96 IDLE,
97 WRITING,
98 PENDING,
99 COMPLETING,
100 FAILED
101 }
102
103 /**
104 * Tries to update the current state to the given new state.
105 * @param previous the expected current state
106 * @param next the desired new state
107 * @return the previous state or null if the state transition failed
108 * @throws WritePendingException if currentState is WRITING and new state is WRITING (api usage error)
109 */
110 private boolean updateState(State previous,State next)
111 {
112 if (!isTransitionAllowed(previous,next))
113 throw new IllegalStateException();
114
115 boolean updated = _state.compareAndSet(previous, next);
116 if (DEBUG)
117 LOG.debug("update {}:{}{}{}", this, previous, updated?"-->":"!->",next);
118 return updated;
119 }
120
121 private void fail(PendingState pending)
122 {
123 State current = _state.get();
124 if (current.getType()==StateType.FAILED)
125 {
126 FailedState failed=(FailedState)current;
127 if (updateState(failed,__IDLE))
128 {
129 pending.fail(failed.getCause());
130 return;
131 }
132 }
133 throw new IllegalStateException();
134 }
135
136 private void ignoreFail()
137 {
138 State current = _state.get();
139 while (current.getType()==StateType.FAILED)
140 {
141 if (updateState(current,__IDLE))
142 return;
143 current = _state.get();
144 }
145 }
146
147 private boolean isTransitionAllowed(State currentState, State newState)
148 {
149 Set<StateType> allowedNewStateTypes = __stateTransitions.get(currentState.getType());
150 if (!allowedNewStateTypes.contains(newState.getType()))
151 {
152 LOG.warn("{}: {} -> {} not allowed", this, currentState, newState);
153 return false;
154 }
155 return true;
156 }
157
158 /**
159 * State represents a State of WriteFlusher.
160 */
161 private static class State
162 {
163 private final StateType _type;
164
165 private State(StateType stateType)
166 {
167 _type = stateType;
168 }
169
170 public StateType getType()
171 {
172 return _type;
173 }
174
175 @Override
176 public String toString()
177 {
178 return String.format("%s", _type);
179 }
180 }
181
182 /**
183 * In IdleState WriteFlusher is idle and accepts new writes
184 */
185 private static class IdleState extends State
186 {
187 private IdleState()
188 {
189 super(StateType.IDLE);
190 }
191 }
192
193 /**
194 * In WritingState WriteFlusher is currently writing.
195 */
196 private static class WritingState extends State
197 {
198 private WritingState()
199 {
200 super(StateType.WRITING);
201 }
202 }
203
204 /**
205 * In FailedState no more operations are allowed. The current implementation will never recover from this state.
206 */
207 private static class FailedState extends State
208 {
209 private final Throwable _cause;
210 private FailedState(Throwable cause)
211 {
212 super(StateType.FAILED);
213 _cause=cause;
214 }
215
216 public Throwable getCause()
217 {
218 return _cause;
219 }
220 }
221
222 /**
223 * In CompletingState WriteFlusher is flushing buffers that have not been fully written in write(). If write()
224 * didn't flush all buffers in one go, it'll switch the State to PendingState. completeWrite() will then switch to
225 * this state and try to flush the remaining buffers.
226 */
227 private static class CompletingState extends State
228 {
229 private CompletingState()
230 {
231 super(StateType.COMPLETING);
232 }
233 }
234
235 /**
236 * In PendingState not all buffers could be written in one go. Then write() will switch to PendingState() and
237 * preserve the state by creating a new PendingState object with the given parameters.
238 */
239 private class PendingState extends State
240 {
241 private final Callback _callback;
242 private final ByteBuffer[] _buffers;
243
244 private PendingState(ByteBuffer[] buffers, Callback callback)
245 {
246 super(StateType.PENDING);
247 _buffers = compact(buffers);
248 _callback = callback;
249 }
250
251 public ByteBuffer[] getBuffers()
252 {
253 return _buffers;
254 }
255
256 protected void fail(Throwable cause)
257 {
258 if (_callback!=null)
259 _callback.failed(cause);
260 }
261
262 protected void complete()
263 {
264 if (_callback!=null)
265 _callback.succeeded();
266 }
267
268 /**
269 * Compacting the buffers is needed because the semantic of WriteFlusher is
270 * to write the buffers and if the caller sees that the buffer is consumed,
271 * then it can recycle it.
272 * If we do not compact, then it is possible that we store a consumed buffer,
273 * which is then recycled and refilled; when the WriteFlusher is invoked to
274 * complete the write, it will write the refilled bytes, garbling the content.
275 *
276 * @param buffers the buffers to compact
277 * @return the compacted buffers
278 */
279 private ByteBuffer[] compact(ByteBuffer[] buffers)
280 {
281 int length = buffers.length;
282
283 // Just one element, no need to compact
284 if (length < 2)
285 return buffers;
286
287 // How many still have content ?
288 int consumed = 0;
289 while (consumed < length && BufferUtil.isEmpty(buffers[consumed]))
290 ++consumed;
291
292 // All of them still have content, no need to compact
293 if (consumed == 0)
294 return buffers;
295
296 // None has content, return empty
297 if (consumed == length)
298 return EMPTY_BUFFERS;
299
300 int newLength = length - consumed;
301 ByteBuffer[] result = new ByteBuffer[newLength];
302 System.arraycopy(buffers, consumed, result, 0, newLength);
303 return result;
304 }
305 }
306
307 /**
308 * Abstract call to be implemented by specific WriteFlushers. It should schedule a call to {@link #completeWrite()}
309 * or {@link #onFail(Throwable)} when appropriate.
310 */
311 abstract protected void onIncompleteFlushed();
312
313 /**
314 * Tries to switch state to WRITING. If successful it writes the given buffers to the EndPoint. If state transition
315 * fails it'll fail the callback.
316 *
317 * If not all buffers can be written in one go it creates a new {@link PendingState} object to preserve the state
318 * and then calls {@link #onIncompleteFlushed()}. The remaining buffers will be written in {@link #completeWrite()}.
319 *
320 * If all buffers have been written it calls callback.complete().
321 *
322 * @param callback the callback to call on either failed or complete
323 * @param buffers the buffers to flush to the endpoint
324 */
325 public void write(Callback callback, ByteBuffer... buffers) throws WritePendingException
326 {
327 if (DEBUG)
328 LOG.debug("write: {} {}", this, BufferUtil.toDetailString(buffers));
329
330 if (!updateState(__IDLE,__WRITING))
331 throw new WritePendingException();
332
333 try
334 {
335 boolean flushed=_endPoint.flush(buffers);
336 if (DEBUG)
337 LOG.debug("flushed {}", flushed);
338
339 // Are we complete?
340 for (ByteBuffer b : buffers)
341 {
342 if (!flushed||BufferUtil.hasContent(b))
343 {
344 PendingState pending=new PendingState(buffers, callback);
345 if (updateState(__WRITING,pending))
346 onIncompleteFlushed();
347 else
348 fail(pending);
349 return;
350 }
351 }
352
353 // If updateState didn't succeed, we don't care as our buffers have been written
354 if (!updateState(__WRITING,__IDLE))
355 ignoreFail();
356 if (callback!=null)
357 callback.succeeded();
358 }
359 catch (IOException e)
360 {
361 if (DEBUG)
362 LOG.debug("write exception", e);
363 if (updateState(__WRITING,__IDLE))
364 {
365 if (callback!=null)
366 callback.failed(e);
367 }
368 else
369 fail(new PendingState(buffers, callback));
370 }
371 }
372
373
374 /**
375 * Complete a write that has not completed and that called {@link #onIncompleteFlushed()} to request a call to this
376 * method when a call to {@link EndPoint#flush(ByteBuffer...)} is likely to be able to progress.
377 *
378 * It tries to switch from PENDING to COMPLETING. If state transition fails, then it does nothing as the callback
379 * should have been already failed. That's because the only way to switch from PENDING outside this method is
380 * {@link #onFail(Throwable)} or {@link #onClose()}
381 */
382 public void completeWrite()
383 {
384 if (DEBUG)
385 LOG.debug("completeWrite: {}", this);
386
387 State previous = _state.get();
388
389 if (previous.getType()!=StateType.PENDING)
390 return; // failure already handled.
391
392 PendingState pending = (PendingState)previous;
393 if (!updateState(pending,__COMPLETING))
394 return; // failure already handled.
395
396 try
397 {
398 ByteBuffer[] buffers = pending.getBuffers();
399
400 boolean flushed=_endPoint.flush(buffers);
401 if (DEBUG)
402 LOG.debug("flushed {}", flushed);
403
404 // Are we complete?
405 for (ByteBuffer b : buffers)
406 {
407 if (!flushed || BufferUtil.hasContent(b))
408 {
409 if (updateState(__COMPLETING,pending))
410 onIncompleteFlushed();
411 else
412 fail(pending);
413 return;
414 }
415 }
416
417 // If updateState didn't succeed, we don't care as our buffers have been written
418 if (!updateState(__COMPLETING,__IDLE))
419 ignoreFail();
420 pending.complete();
421 }
422 catch (IOException e)
423 {
424 if (DEBUG)
425 LOG.debug("completeWrite exception", e);
426 if(updateState(__COMPLETING,__IDLE))
427 pending.fail(e);
428 else
429 fail(pending);
430 }
431 }
432
433 public void onFail(Throwable cause)
434 {
435 if (DEBUG)
436 LOG.debug("failed: {} {}", this, cause);
437
438 // Keep trying to handle the failure until we get to IDLE or FAILED state
439 while(true)
440 {
441 State current=_state.get();
442 switch(current.getType())
443 {
444 case IDLE:
445 case FAILED:
446 return;
447
448 case PENDING:
449 PendingState pending = (PendingState)current;
450 if (updateState(pending,__IDLE))
451 {
452 pending.fail(cause);
453 return;
454 }
455 break;
456
457 default:
458 if (updateState(current,new FailedState(cause)))
459 return;
460 break;
461 }
462 }
463 }
464
465 public void onClose()
466 {
467 if (_state.get()==__IDLE)
468 return;
469 onFail(new ClosedChannelException());
470 }
471
472 boolean isIdle()
473 {
474 return _state.get().getType() == StateType.IDLE;
475 }
476
477 public boolean isInProgress()
478 {
479 switch(_state.get().getType())
480 {
481 case WRITING:
482 case PENDING:
483 case COMPLETING:
484 return true;
485 default:
486 return false;
487 }
488 }
489
490 @Override
491 public String toString()
492 {
493 return String.format("WriteFlusher@%x{%s}", hashCode(), _state.get());
494 }
495 }