Mercurial Hosting > luan

comparison src/goodjava/io/BufferedInputStream.java @ 1480:1f41e5921090

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
input buffering
author Franklin Schmidt <fschmidt@gmail.com>
date Fri, 24 Apr 2020 14:32:20 -0600
parents bd13aaeaf6d4
children 97740900c820
comparison
equal deleted inserted replaced
1479:bd13aaeaf6d4 1480:1f41e5921090
1 /* 1 /*
2 * Copyright (c) 1994, 2013, Oracle and/or its affiliates. All rights reserved. 2 * Copyright (c) 1994, 2013, Oracle and/or its affiliates. All rights reserved.
3 * ORACLE PROPRIETARY/CONFIDENTIAL. Use is subject to license terms. 3 * ORACLE PROPRIETARY/CONFIDENTIAL. Use is subject to license terms.
4 *
5 *
6 *
7 *
8 *
9 *
10 *
11 *
12 *
13 *
14 *
15 *
16 *
17 *
18 *
19 *
20 *
21 *
22 *
23 *
24 */ 4 */
25 5
26 package goodjava.io; 6 package goodjava.io;
27 7
28 import java.io.InputStream; 8 import java.io.InputStream;
29 import java.io.FilterInputStream; 9 import java.io.FilterInputStream;
30 import java.io.IOException; 10 import java.io.IOException;
31 import java.util.concurrent.atomic.AtomicReferenceFieldUpdater;
32 11
33 /** 12 /**
34 * A <code>BufferedInputStream</code> adds 13 * A <code>BufferedInputStream</code> adds
35 * functionality to another input stream-namely, 14 * functionality to another input stream-namely,
36 * the ability to buffer the input and to 15 * the ability to buffer the input. When the <code>BufferedInputStream</code>
37 * support the <code>mark</code> and <code>reset</code>
38 * methods. When the <code>BufferedInputStream</code>
39 * is created, an internal buffer array is 16 * is created, an internal buffer array is
40 * created. As bytes from the stream are read 17 * created. As bytes from the stream are read
41 * or skipped, the internal buffer is refilled 18 * or skipped, the internal buffer is refilled
42 * as necessary from the contained input stream, 19 * as necessary from the contained input stream,
43 * many bytes at a time. The <code>mark</code> 20 * many bytes at a time.
44 * operation remembers a point in the input
45 * stream and the <code>reset</code> operation
46 * causes all the bytes read since the most
47 * recent <code>mark</code> operation to be
48 * reread before new bytes are taken from
49 * the contained input stream.
50 * 21 *
51 * @author Arthur van Hoff 22 * @author Arthur van Hoff
52 * @since JDK1.0 23 * @since JDK1.0
53 */ 24 */
54 public class BufferedInputStream extends FilterInputStream { 25 public final class BufferedInputStream extends FilterInputStream {
55
56 private static int DEFAULT_BUFFER_SIZE = 8192; 26 private static int DEFAULT_BUFFER_SIZE = 8192;
57 27 private final byte buf[];
58 /** 28 private boolean isClosed = false;
59 * The maximum size of array to allocate.
60 * Some VMs reserve some header words in an array.
61 * Attempts to allocate larger arrays may result in
62 * OutOfMemoryError: Requested array size exceeds VM limit
63 */
64 private static int MAX_BUFFER_SIZE = Integer.MAX_VALUE - 8;
65
66 /**
67 * The internal buffer array where the data is stored. When necessary,
68 * it may be replaced by another array of
69 * a different size.
70 */
71 protected volatile byte buf[];
72
73 /**
74 * Atomic updater to provide compareAndSet for buf. This is
75 * necessary because closes can be asynchronous. We use nullness
76 * of buf[] as primary indicator that this stream is closed. (The
77 * "in" field is also nulled out on close.)
78 */
79 private static final
80 AtomicReferenceFieldUpdater<BufferedInputStream, byte[]> bufUpdater =
81 AtomicReferenceFieldUpdater.newUpdater
82 (BufferedInputStream.class, byte[].class, "buf");
83 29
84 /** 30 /**
85 * The index one greater than the index of the last valid byte in 31 * The index one greater than the index of the last valid byte in
86 * the buffer. 32 * the buffer.
87 * This value is always 33 * This value is always
88 * in the range <code>0</code> through <code>buf.length</code>; 34 * in the range <code>0</code> through <code>buf.length</code>;
89 * elements <code>buf[0]</code> through <code>buf[count-1] 35 * elements <code>buf[0]</code> through <code>buf[count-1]
90 * </code>contain buffered input data obtained 36 * </code>contain buffered input data obtained
91 * from the underlying input stream. 37 * from the underlying input stream.
92 */ 38 */
93 protected int count; 39 private int count;
94 40
95 /** 41 /**
96 * The current position in the buffer. This is the index of the next 42 * The current position in the buffer. This is the index of the next
97 * character to be read from the <code>buf</code> array. 43 * character to be read from the <code>buf</code> array.
98 * <p> 44 * <p>
105 * operation will require more bytes to be 51 * operation will require more bytes to be
106 * read from the contained input stream. 52 * read from the contained input stream.
107 * 53 *
108 * @see java.io.BufferedInputStream#buf 54 * @see java.io.BufferedInputStream#buf
109 */ 55 */
110 protected int pos; 56 private int pos;
111 57
112 /** 58 private void checkClosed() throws IOException {
113 * The value of the <code>pos</code> field at the time the last 59 if (isClosed)
114 * <code>mark</code> method was called.
115 * <p>
116 * This value is always
117 * in the range <code>-1</code> through <code>pos</code>.
118 * If there is no marked position in the input
119 * stream, this field is <code>-1</code>. If
120 * there is a marked position in the input
121 * stream, then <code>buf[markpos]</code>
122 * is the first byte to be supplied as input
123 * after a <code>reset</code> operation. If
124 * <code>markpos</code> is not <code>-1</code>,
125 * then all bytes from positions <code>buf[markpos]</code>
126 * through <code>buf[pos-1]</code> must remain
127 * in the buffer array (though they may be
128 * moved to another place in the buffer array,
129 * with suitable adjustments to the values
130 * of <code>count</code>, <code>pos</code>,
131 * and <code>markpos</code>); they may not
132 * be discarded unless and until the difference
133 * between <code>pos</code> and <code>markpos</code>
134 * exceeds <code>marklimit</code>.
135 *
136 * @see java.io.BufferedInputStream#mark(int)
137 * @see java.io.BufferedInputStream#pos
138 */
139 protected int markpos = -1;
140
141 /**
142 * The maximum read ahead allowed after a call to the
143 * <code>mark</code> method before subsequent calls to the
144 * <code>reset</code> method fail.
145 * Whenever the difference between <code>pos</code>
146 * and <code>markpos</code> exceeds <code>marklimit</code>,
147 * then the mark may be dropped by setting
148 * <code>markpos</code> to <code>-1</code>.
149 *
150 * @see java.io.BufferedInputStream#mark(int)
151 * @see java.io.BufferedInputStream#reset()
152 */
153 protected int marklimit;
154
155 /**
156 * Check to make sure that underlying input stream has not been
157 * nulled out due to close; if not return it;
158 */
159 private InputStream getInIfOpen() throws IOException {
160 InputStream input = in;
161 if (input == null)
162 throw new IOException("Stream closed"); 60 throw new IOException("Stream closed");
163 return input;
164 }
165
166 /**
167 * Check to make sure that buffer has not been nulled out due to
168 * close; if not return it;
169 */
170 private byte[] getBufIfOpen() throws IOException {
171 byte[] buffer = buf;
172 if (buffer == null)
173 throw new IOException("Stream closed");
174 return buffer;
175 } 61 }
176 62
177 /** 63 /**
178 * Creates a <code>BufferedInputStream</code> 64 * Creates a <code>BufferedInputStream</code>
179 * and saves its argument, the input stream 65 * and saves its argument, the input stream
194 * buffer array of length <code>size</code> 80 * buffer array of length <code>size</code>
195 * is created and stored in <code>buf</code>. 81 * is created and stored in <code>buf</code>.
196 * 82 *
197 * @param in the underlying input stream. 83 * @param in the underlying input stream.
198 * @param size the buffer size. 84 * @param size the buffer size.
199 * @exception IllegalArgumentException if {@code size <= 0}.
200 */ 85 */
201 public BufferedInputStream(InputStream in, int size) { 86 public BufferedInputStream(InputStream in, int size) {
202 super(in); 87 super(in);
203 if (size <= 0) {
204 throw new IllegalArgumentException("Buffer size <= 0");
205 }
206 buf = new byte[size]; 88 buf = new byte[size];
207 } 89 }
208 90
209 /** 91 /**
210 * Fills the buffer with more data, taking into account 92 * Fills the buffer with more data.
211 * shuffling and other tricks for dealing with marks.
212 * Assumes that it is being called by a synchronized method. 93 * Assumes that it is being called by a synchronized method.
213 * This method also assumes that all data has already been read in, 94 * This method also assumes that all data has already been read in,
214 * hence pos > count. 95 * hence pos > count.
215 */ 96 */
216 private void fill() throws IOException { 97 private void fill() throws IOException {
217 byte[] buffer = getBufIfOpen(); 98 pos = 0;
218 if (markpos < 0) 99 count = 0;
219 pos = 0; /* no mark: throw away the buffer */ 100 int n = super.read(buf, 0, buf.length);
220 else if (pos >= buffer.length) /* no room left in buffer */
221 if (markpos > 0) { /* can throw away early part of the buffer */
222 int sz = pos - markpos;
223 System.arraycopy(buffer, markpos, buffer, 0, sz);
224 pos = sz;
225 markpos = 0;
226 } else if (buffer.length >= marklimit) {
227 markpos = -1; /* buffer got too big, invalidate mark */
228 pos = 0; /* drop buffer contents */
229 } else if (buffer.length >= MAX_BUFFER_SIZE) {
230 throw new OutOfMemoryError("Required array size too large");
231 } else { /* grow buffer */
232 int nsz = (pos <= MAX_BUFFER_SIZE - pos) ?
233 pos * 2 : MAX_BUFFER_SIZE;
234 if (nsz > marklimit)
235 nsz = marklimit;
236 byte nbuf[] = new byte[nsz];
237 System.arraycopy(buffer, 0, nbuf, 0, pos);
238 if (!bufUpdater.compareAndSet(this, buffer, nbuf)) {
239 // Can't replace buf if there was an async close.
240 // Note: This would need to be changed if fill()
241 // is ever made accessible to multiple threads.
242 // But for now, the only way CAS can fail is via close.
243 // assert buf == null;
244 throw new IOException("Stream closed");
245 }
246 buffer = nbuf;
247 }
248 count = pos;
249 int n = getInIfOpen().read(buffer, pos, buffer.length - pos);
250 if (n > 0) 101 if (n > 0)
251 count = n + pos; 102 count = n;
252 } 103 }
253 104
254 /** 105 /**
255 * See 106 * See
256 * the general contract of the <code>read</code> 107 * the general contract of the <code>read</code>
262 * invoking its {@link #close()} method, 113 * invoking its {@link #close()} method,
263 * or an I/O error occurs. 114 * or an I/O error occurs.
264 * @see java.io.FilterInputStream#in 115 * @see java.io.FilterInputStream#in
265 */ 116 */
266 public synchronized int read() throws IOException { 117 public synchronized int read() throws IOException {
118 checkClosed();
267 if (pos >= count) { 119 if (pos >= count) {
268 fill(); 120 fill();
269 if (pos >= count) 121 if (pos >= count)
270 return -1; 122 return -1;
271 } 123 }
272 return getBufIfOpen()[pos++] & 0xff; 124 return buf[pos++] & 0xff;
273 } 125 }
274 126
275 /** 127 /**
276 * Read characters into a portion of an array, reading from the underlying 128 * Read characters into a portion of an array, reading from the underlying
277 * stream at most once if necessary. 129 * stream at most once if necessary.
278 */ 130 */
279 private int read1(byte[] b, int off, int len) throws IOException { 131 public synchronized int read(byte[] b, int off, int len) throws IOException {
132 checkClosed();
280 int avail = count - pos; 133 int avail = count - pos;
281 if (avail <= 0) { 134 if (avail <= 0) {
282 /* If the requested length is at least as large as the buffer, and 135 /* If the requested length is at least as large as the buffer, do not bother to copy the
283 if there is no mark/reset activity, do not bother to copy the
284 bytes into the local buffer. In this way buffered streams will 136 bytes into the local buffer. In this way buffered streams will
285 cascade harmlessly. */ 137 cascade harmlessly. */
286 if (len >= getBufIfOpen().length && markpos < 0) { 138 if (len >= buf.length) {
287 return getInIfOpen().read(b, off, len); 139 return super.read(b, off, len);
288 } 140 }
289 fill(); 141 fill();
290 avail = count - pos; 142 avail = count - pos;
291 if (avail <= 0) return -1; 143 if (avail <= 0) return -1;
292 } 144 }
293 int cnt = (avail < len) ? avail : len; 145 int cnt = (avail < len) ? avail : len;
294 System.arraycopy(getBufIfOpen(), pos, b, off, cnt); 146 System.arraycopy(buf, pos, b, off, cnt);
295 pos += cnt; 147 pos += cnt;
296 return cnt; 148 return cnt;
297 }
298
299 /**
300 * Reads bytes from this byte-input stream into the specified byte array,
301 * starting at the given offset.
302 *
303 * <p> This method implements the general contract of the corresponding
304 * <code>{@link InputStream#read(byte[], int, int) read}</code> method of
305 * the <code>{@link InputStream}</code> class. As an additional
306 * convenience, it attempts to read as many bytes as possible by repeatedly
307 * invoking the <code>read</code> method of the underlying stream. This
308 * iterated <code>read</code> continues until one of the following
309 * conditions becomes true: <ul>
310 *
311 * <li> The specified number of bytes have been read,
312 *
313 * <li> The <code>read</code> method of the underlying stream returns
314 * <code>-1</code>, indicating end-of-file, or
315 *
316 * <li> The <code>available</code> method of the underlying stream
317 * returns zero, indicating that further input requests would block.
318 *
319 * </ul> If the first <code>read</code> on the underlying stream returns
320 * <code>-1</code> to indicate end-of-file then this method returns
321 * <code>-1</code>. Otherwise this method returns the number of bytes
322 * actually read.
323 *
324 * <p> Subclasses of this class are encouraged, but not required, to
325 * attempt to read as many bytes as possible in the same fashion.
326 *
327 * @param b destination buffer.
328 * @param off offset at which to start storing bytes.
329 * @param len maximum number of bytes to read.
330 * @return the number of bytes read, or <code>-1</code> if the end of
331 * the stream has been reached.
332 * @exception IOException if this input stream has been closed by
333 * invoking its {@link #close()} method,
334 * or an I/O error occurs.
335 */
336 public synchronized int read(byte b[], int off, int len)
337 throws IOException
338 {
339 getBufIfOpen(); // Check for closed stream
340 if ((off | len | (off + len) | (b.length - (off + len))) < 0) {
341 throw new IndexOutOfBoundsException();
342 } else if (len == 0) {
343 return 0;
344 }
345
346 int n = 0;
347 for (;;) {
348 int nread = read1(b, off + n, len - n);
349 if (nread <= 0)
350 return (n == 0) ? nread : n;
351 n += nread;
352 if (n >= len)
353 return n;
354 // if not closed but no bytes available, return
355 InputStream input = in;
356 if (input != null && input.available() <= 0)
357 return n;
358 }
359 } 149 }
360 150
361 /** 151 /**
362 * See the general contract of the <code>skip</code> 152 * See the general contract of the <code>skip</code>
363 * method of <code>InputStream</code>. 153 * method of <code>InputStream</code>.
366 * or if this input stream has been closed by 156 * or if this input stream has been closed by
367 * invoking its {@link #close()} method, or an 157 * invoking its {@link #close()} method, or an
368 * I/O error occurs. 158 * I/O error occurs.
369 */ 159 */
370 public synchronized long skip(long n) throws IOException { 160 public synchronized long skip(long n) throws IOException {
371 getBufIfOpen(); // Check for closed stream 161 checkClosed();
372 if (n <= 0) { 162 long avail = count - pos;
373 return 0; 163 if (avail <= 0) {
164 return super.skip(n);
374 } 165 }
375 long avail = count - pos;
376
377 if (avail <= 0) {
378 // If no mark position set then don't keep in buffer
379 if (markpos <0)
380 return getInIfOpen().skip(n);
381
382 // Fill in buffer to save bytes for reset
383 fill();
384 avail = count - pos;
385 if (avail <= 0)
386 return 0;
387 }
388
389 long skipped = (avail < n) ? avail : n; 166 long skipped = (avail < n) ? avail : n;
390 pos += skipped; 167 pos += skipped;
391 return skipped; 168 return skipped;
392 } 169 }
393 170
408 * invoking its {@link #close()} method, 185 * invoking its {@link #close()} method,
409 * or an I/O error occurs. 186 * or an I/O error occurs.
410 */ 187 */
411 public synchronized int available() throws IOException { 188 public synchronized int available() throws IOException {
412 int n = count - pos; 189 int n = count - pos;
413 int avail = getInIfOpen().available(); 190 int avail = super.available();
414 return n > (Integer.MAX_VALUE - avail) 191 return n > (Integer.MAX_VALUE - avail)
415 ? Integer.MAX_VALUE 192 ? Integer.MAX_VALUE
416 : n + avail; 193 : n + avail;
417 } 194 }
418 195
419 /** 196 public void mark(int readlimit) {}
420 * See the general contract of the <code>mark</code> 197
421 * method of <code>InputStream</code>. 198 public void reset() throws IOException {
422 * 199 throw new IOException("mark/reset not supported");
423 * @param readlimit the maximum limit of bytes that can be read before 200 }
424 * the mark position becomes invalid. 201
425 * @see java.io.BufferedInputStream#reset()
426 */
427 public synchronized void mark(int readlimit) {
428 marklimit = readlimit;
429 markpos = pos;
430 }
431
432 /**
433 * See the general contract of the <code>reset</code>
434 * method of <code>InputStream</code>.
435 * <p>
436 * If <code>markpos</code> is <code>-1</code>
437 * (no mark has been set or the mark has been
438 * invalidated), an <code>IOException</code>
439 * is thrown. Otherwise, <code>pos</code> is
440 * set equal to <code>markpos</code>.
441 *
442 * @exception IOException if this stream has not been marked or,
443 * if the mark has been invalidated, or the stream
444 * has been closed by invoking its {@link #close()}
445 * method, or an I/O error occurs.
446 * @see java.io.BufferedInputStream#mark(int)
447 */
448 public synchronized void reset() throws IOException {
449 getBufIfOpen(); // Cause exception if closed
450 if (markpos < 0)
451 throw new IOException("Resetting to invalid mark");
452 pos = markpos;
453 }
454
455 /**
456 * Tests if this input stream supports the <code>mark</code>
457 * and <code>reset</code> methods. The <code>markSupported</code>
458 * method of <code>BufferedInputStream</code> returns
459 * <code>true</code>.
460 *
461 * @return a <code>boolean</code> indicating if this stream type supports
462 * the <code>mark</code> and <code>reset</code> methods.
463 * @see java.io.InputStream#mark(int)
464 * @see java.io.InputStream#reset()
465 */
466 public boolean markSupported() { 202 public boolean markSupported() {
467 return true; 203 return false;
468 } 204 }
469 205
470 /** 206 /**
471 * Closes this input stream and releases any system resources 207 * Closes this input stream and releases any system resources
472 * associated with the stream. 208 * associated with the stream.
474 * or skip() invocations will throw an IOException. 210 * or skip() invocations will throw an IOException.
475 * Closing a previously closed stream has no effect. 211 * Closing a previously closed stream has no effect.
476 * 212 *
477 * @exception IOException if an I/O error occurs. 213 * @exception IOException if an I/O error occurs.
478 */ 214 */
479 public void close() throws IOException { 215 public synchronized void close() throws IOException {
480 byte[] buffer; 216 isClosed = true;
481 while ( (buffer = buf) != null) { 217 super.close();
482 if (bufUpdater.compareAndSet(this, buffer, null)) {
483 InputStream input = in;
484 in = null;
485 if (input != null)
486 input.close();
487 return;
488 }
489 // Else retry in case a new buf was CASed in fill()
490 }
491 } 218 }
492 } 219 }