001
002 /*
003 * Copyright (C) 2011 Archie L. Cobbs. All rights reserved.
004 *
005 * $Id: AsyncOutputStream.java 162 2011-10-24 20:42:37Z archie.cobbs $
006 */
007
008 package org.dellroad.stuff.io;
009
010 import java.io.FilterOutputStream;
011 import java.io.IOException;
012 import java.io.OutputStream;
013
014 import org.dellroad.stuff.java.CheckedExceptionWrapper;
015 import org.dellroad.stuff.java.Predicate;
016 import org.dellroad.stuff.java.TimedWait;
017 import org.slf4j.Logger;
018 import org.slf4j.LoggerFactory;
019
020 /**
021 * An {@link OutputStream} that performs writes using a background thread, so that
022 * write, flush, and close operations never block.
023 * <p/>
024 *
025 * <p>
026 * If the underlying output stream throws an {@link IOException} during any operation,
027 * this instance will re-throw the exception for all subsequent operations.
028 * </p>
029 *
030 * <p>
031 * Instances use an internal buffer whose size is configured at construction time;
032 * if the buffer overflows, a {@link BufferOverflowException} is thrown.
033 * </p>
034 *
035 * <p>
036 * Instances of this class are thread safe, and moreover writes are atomic: if multiple threads are writing
037 * at the same time the bytes written in any single method invocation are written contiguously to the
038 * underlying output.
039 * </p>
040 */
041 public class AsyncOutputStream extends FilterOutputStream {
042
043 protected final Logger log = LoggerFactory.getLogger(getClass());
044
045 private final String name;
046 private final byte[] buf; // output buffer
047 private int count; // number of bytes in output buffer ready to be written
048 private int flushMark = -1; // buffer byte at which a flush is requested, or -1 if none
049 private Thread thread; // async writer thread
050 private IOException exception; // exception caught by async thread
051 private boolean closed; // this instance has been close()'d
052
053 /**
054 * Constructor.
055 *
056 * @param out underlying output stream
057 * @param bufsize maximum number of bytes we can buffer
058 * @param name name for this instance; used to create the name of the background thread
059 */
060 public AsyncOutputStream(OutputStream out, int bufsize, String name) {
061 super(out);
062 if (out == null)
063 throw new IllegalArgumentException("null output");
064 this.name = name;
065 this.buf = new byte[bufsize];
066
067 // Start worker thread
068 this.thread = new Thread(this.name) {
069 @Override
070 public void run() {
071 AsyncOutputStream.this.threadMain();
072 }
073 };
074 this.thread.setDaemon(true);
075 this.thread.start();
076 }
077
078 /**
079 * Write data.
080 *
081 * <p>
082 * This method will never block. To effect a normal blocking write, use {@link #waitForSpace} first.
083 * </p>
084 *
085 * @param b byte to write (lower 8 bits)
086 * @throws IOException if an exception has been thrown by the underlying stream
087 * @throws IOException if this instance has been closed
088 * @throws BufferOverflowException if the buffer does not have room for the new byte
089 */
090 @Override
091 public void write(int b) throws IOException {
092 this.write(new byte[] { (byte)b }, 0, 1);
093 }
094
095 /**
096 * Write data.
097 *
098 * <p>
099 * This method will never block. To effect a normal blocking write, invoke {@link #waitForSpace} first.
100 * </p>
101 *
102 * @param data bytes to write
103 * @param off starting offset in buffer
104 * @param len number of bytes to write
105 * @throws IOException if an exception has been thrown by the underlying stream
106 * @throws IOException if this instance has been closed
107 * @throws BufferOverflowException if the buffer does not have room for the new data
108 * @throws IllegalArgumentException if {@code len} is negative
109 */
110 @Override
111 public synchronized void write(byte[] data, int off, int len) throws IOException {
112
113 // Check exception conditions
114 this.checkExceptions();
115 if (this.count + len > this.buf.length)
116 throw new BufferOverflowException(len + " more byte(s) would exceed the " + this.buf.length + " byte buffer");
117 if (len < 0)
118 throw new IllegalArgumentException("len = " + len);
119 if (len == 0)
120 return;
121
122 // Add data to buffer
123 System.arraycopy(data, off, this.buf, this.count, len);
124 this.count += len;
125
126 // Wakeup writer thread
127 this.notifyAll();
128 }
129
130 /**
131 * Flush output. This method will cause the underlying stream to be flushed once all of the data written to this
132 * instance at the time this method is invoked has been written to it.
133 *
134 * <p>
135 * If additional data is written and then a second flush is requested before the first flush has actually occurred,
136 * the first flush will be canceled and only the second flush will be applied. Normally this is not a problem because
137 * the act of writing more data and then flushing forces earlier data to be flushed as well.
138 * </p>
139 *
140 * <p>
141 * This method will never block. To block until the underlying flush operation completes, invoke {@link #waitForIdle}.
142 * </p>
143 *
144 * @throws IOException if this instance has been closed
145 * @throws IOException if an exception has been detected on the underlying stream
146 * @throws IOException if the current thread is interrupted; the nested exception will an {@link InterruptedException}
147 */
148 @Override
149 public synchronized void flush() throws IOException {
150 this.checkExceptions();
151 this.flushMark = this.count;
152 this.notifyAll(); // wake up writer thread
153 }
154
155 /**
156 * Close this instance. This will (eventually) close the underlying output stream.
157 *
158 * <p>
159 * If this instance has already been closed, nothing happens.
160 * </p>
161 *
162 * <p>
163 * This method will never block. To block until the underlying close operation completes, invoke {@link #waitForIdle}.
164 * </p>
165 *
166 * @throws IOException if an exception has been detected on the underlying stream
167 */
168 @Override
169 public synchronized void close() throws IOException {
170 if (this.closed)
171 return;
172 this.closed = true;
173 this.notifyAll(); // wake up writer thread
174 }
175
176 /**
177 * Get the exception thrown by the underlying output stream, if any.
178 *
179 * @return thrown exception, or {@code null} if none has been thrown by the underlying stream
180 */
181 public synchronized IOException getException() {
182 return this.exception;
183 }
184
185 /**
186 * Get the capacity of this instance's output buffer.
187 *
188 * @return output buffer capacity configured at construction time
189 */
190 public synchronized int getBufferSize() {
191 return this.buf.length;
192 }
193
194 /**
195 * Get the number of free bytes remaining in the output buffer.
196 *
197 * @return current number of available bytes in the output buffer
198 * @throws IOException if this instance is or has been closed
199 * @throws IOException if an exception has been detected on the underlying stream
200 * @see #waitForSpace
201 */
202 public synchronized int availableBufferSpace() throws IOException {
203 this.checkExceptions();
204 return this.buf.length - this.count;
205 }
206
207 /**
208 * Determine if there is outstanding work still to be performed (writes, flushes, and/or close operations)
209 * by the background thread.
210 *
211 * @throws IOException if this instance is or has been closed
212 * @throws IOException if an exception has been detected on the underlying stream
213 * @see #waitForIdle
214 */
215 public synchronized boolean isWorkOutstanding() throws IOException {
216 this.checkExceptions();
217 return this.threadHasWork();
218 }
219
220 /**
221 * Wait for buffer space availability.
222 *
223 * @param numBytes amount of buffer space required
224 * @param timeout maximum time to wait in milliseconds, or zero for infinite
225 * @return true if space was found, false if time expired
226 * @throws IOException if this instance is or has been closed
227 * @throws IOException if an exception has been detected on the underlying stream
228 * @throws IllegalArgumentException if {@code numBytes} is greater than the configured buffer size
229 * @throws IllegalArgumentException if {@code timeout} is negative
230 * @throws InterruptedException if the current thread is interrupted
231 * @see #availableBufferSpace
232 */
233 public boolean waitForSpace(final int numBytes, long timeout) throws IOException, InterruptedException {
234 if (numBytes > this.buf.length)
235 throw new IllegalArgumentException("numBytes (" + numBytes + ") > buffer size (" + this.buf.length + ")");
236 return this.waitForPredicate(timeout, new Predicate() {
237 @Override
238 public boolean test() {
239 return AsyncOutputStream.this.buf.length - AsyncOutputStream.this.count >= numBytes;
240 }
241 });
242 }
243
244 /**
245 * Wait for all outstanding work to complete.
246 *
247 * @param timeout maximum time to wait in milliseconds, or zero for infinite
248 * @return true for success, false if time expired
249 * @throws IOException if this instance is or has been closed
250 * @throws IOException if an exception has been detected on the underlying stream
251 * @throws IllegalArgumentException if {@code timeout} is negative
252 * @throws InterruptedException if the current thread is interrupted
253 * @see #isWorkOutstanding
254 */
255 public synchronized boolean waitForIdle(long timeout) throws IOException, InterruptedException {
256 return this.waitForPredicate(timeout, new Predicate() {
257 @Override
258 public boolean test() {
259 return !AsyncOutputStream.this.threadHasWork();
260 }
261 });
262 }
263
264 /**
265 * Check for exceptions.
266 *
267 * @throws IOException if this instance has been closed
268 * @throws IOException if an exception has been detected on the underlying stream
269 */
270 private void checkExceptions() throws IOException {
271 if (this.closed)
272 throw new IOException("instance has been closed");
273 if (this.exception != null)
274 throw new IOException("exception from underlying output stream", this.exception);
275 }
276
277 /**
278 * Determine if there is outstanding work still to be performed (writes, flushes, and/or close operations)
279 * by the background thread.
280 */
281 private boolean threadHasWork() {
282 return this.count > 0 || this.flushMark != -1 || this.closed;
283 }
284
285 /**
286 * Writer thread main entry point.
287 */
288 private void threadMain() {
289 try {
290 this.runLoop();
291 } catch (Throwable t) {
292 synchronized (this) {
293 this.exception = t instanceof IOException ? (IOException)t : new IOException("caught unexpected exception", t);
294 this.notifyAll(); // wake up sleepers in waitForSpace() and waitForIdle()
295 }
296 } finally {
297 synchronized (this) {
298 this.thread = null;
299 this.notifyAll(); // wake up sleepers in waitForIdle()
300 }
301 }
302 }
303
304 /**
305 * Async writer thread main loop.
306 */
307 private void runLoop() throws IOException, InterruptedException {
308 while (true) {
309
310 // Wait for something to do
311 synchronized (this) {
312 while (!this.threadHasWork())
313 this.wait(); // will be woken up by write(), flush(), or close()
314 }
315
316 // Determine what needs to be done
317 int wlen;
318 boolean flush;
319 boolean close;
320 synchronized (this) {
321 wlen = this.count;
322 flush = this.flushMark == 0;
323 close = this.closed;
324 }
325
326 // First priority: any data to write?
327 if (wlen > 0) {
328
329 // Write data
330 this.out.write(this.buf, 0, wlen);
331
332 // Shift data in buffer
333 synchronized (this) {
334 System.arraycopy(this.buf, wlen, this.buf, 0, this.count - wlen);
335 this.count -= wlen;
336 if (this.flushMark != -1)
337 this.flushMark = Math.max(0, this.flushMark - wlen);
338 this.notifyAll(); // wake up sleepers in waitForSpace() and waitForIdle()
339 }
340 continue;
341 }
342
343 // Second priority: is a flush required?
344 if (flush) {
345
346 // Flush output
347 this.out.flush();
348
349 // Update flush mark
350 synchronized (this) {
351 if (this.flushMark == 0) {
352 this.flushMark = -1;
353 this.notifyAll(); // wake up sleepers in waitForIdle()
354 }
355 }
356 continue;
357 }
358
359 // Third priority: is a close required?
360 if (close) {
361 this.out.close();
362 break;
363 }
364 }
365 }
366
367 /**
368 * Wait for some condition to become true. Of course somebody has to wake us up when it becomes true.
369 */
370 private synchronized boolean waitForPredicate(long timeout, final Predicate predicate)
371 throws IOException, InterruptedException {
372 try {
373 return TimedWait.wait(this, timeout, new Predicate() {
374 @Override
375 public boolean test() {
376 try {
377 checkExceptions();
378 } catch (IOException e) {
379 throw new CheckedExceptionWrapper(e);
380 }
381 return predicate.test();
382 }
383 });
384 } catch (CheckedExceptionWrapper e) {
385 throw (IOException)e.getException();
386 }
387 }
388 }
389