< prev index next >

src/java.base/share/classes/java/io/BufferedInputStream.java

Print this page
rev 13994 : 8153334: Replace BufferedInputStreams use of AtomicReferenceFieldUpdater with Unsafe
Reviewed-by: alanb, forax


   7  * published by the Free Software Foundation.  Oracle designates this
   8  * particular file as subject to the "Classpath" exception as provided
   9  * by Oracle in the LICENSE file that accompanied this code.
  10  *
  11  * This code is distributed in the hope that it will be useful, but WITHOUT
  12  * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
  13  * FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General Public License
  14  * version 2 for more details (a copy is included in the LICENSE file that
  15  * accompanied this code).
  16  *
  17  * You should have received a copy of the GNU General Public License version
  18  * 2 along with this work; if not, write to the Free Software Foundation,
  19  * Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA.
  20  *
  21  * Please contact Oracle, 500 Oracle Parkway, Redwood Shores, CA 94065 USA
  22  * or visit www.oracle.com if you need additional information or have any
  23  * questions.
  24  */
  25 
  26 package java.io;
  27 import java.util.concurrent.atomic.AtomicReferenceFieldUpdater;

  28 
  29 /**
  30  * A <code>BufferedInputStream</code> adds
  31  * functionality to another input stream-namely,
  32  * the ability to buffer the input and to
  33  * support the <code>mark</code> and <code>reset</code>
  34  * methods. When  the <code>BufferedInputStream</code>
  35  * is created, an internal buffer array is
  36  * created. As bytes  from the stream are read
  37  * or skipped, the internal buffer is refilled
  38  * as necessary  from the contained input stream,
  39  * many bytes at a time. The <code>mark</code>
  40  * operation  remembers a point in the input
  41  * stream and the <code>reset</code> operation
  42  * causes all the  bytes read since the most
  43  * recent <code>mark</code> operation to be
  44  * reread before new bytes are  taken from
  45  * the contained input stream.
  46  *
  47  * @author  Arthur van Hoff


  51 class BufferedInputStream extends FilterInputStream {
  52 
  53     private static int DEFAULT_BUFFER_SIZE = 8192;
  54 
  55     /**
  56      * The maximum size of array to allocate.
  57      * Some VMs reserve some header words in an array.
  58      * Attempts to allocate larger arrays may result in
  59      * OutOfMemoryError: Requested array size exceeds VM limit
  60      */
  61     private static int MAX_BUFFER_SIZE = Integer.MAX_VALUE - 8;
  62 
  63     /**
  64      * The internal buffer array where the data is stored. When necessary,
  65      * it may be replaced by another array of
  66      * a different size.
  67      */
  68     protected volatile byte buf[];
  69 
  70     /**
  71      * Atomic updater to provide compareAndSet for buf. This is
  72      * necessary because closes can be asynchronous. We use nullness
  73      * of buf[] as primary indicator that this stream is closed. (The
  74      * "in" field is also nulled out on close.)
  75      */
  76     private static final
  77         AtomicReferenceFieldUpdater<BufferedInputStream, byte[]> bufUpdater =
  78         AtomicReferenceFieldUpdater.newUpdater
  79         (BufferedInputStream.class,  byte[].class, "buf");






  80 
  81     /**
  82      * The index one greater than the index of the last valid byte in
  83      * the buffer.
  84      * This value is always
  85      * in the range <code>0</code> through <code>buf.length</code>;
  86      * elements <code>buf[0]</code>  through <code>buf[count-1]
  87      * </code>contain buffered input data obtained
  88      * from the underlying  input stream.
  89      */
  90     protected int count;
  91 
  92     /**
  93      * The current position in the buffer. This is the index of the next
  94      * character to be read from the <code>buf</code> array.
  95      * <p>
  96      * This value is always in the range <code>0</code>
  97      * through <code>count</code>. If it is less
  98      * than <code>count</code>, then  <code>buf[pos]</code>
  99      * is the next byte to be supplied as input;


 195      * @param   size   the buffer size.
 196      * @exception IllegalArgumentException if {@code size <= 0}.
 197      */
 198     public BufferedInputStream(InputStream in, int size) {
 199         super(in);
 200         if (size <= 0) {
 201             throw new IllegalArgumentException("Buffer size <= 0");
 202         }
 203         buf = new byte[size];
 204     }
 205 
 206     /**
 207      * Fills the buffer with more data, taking into account
 208      * shuffling and other tricks for dealing with marks.
 209      * Assumes that it is being called by a synchronized method.
 210      * This method also assumes that all data has already been read in,
 211      * hence pos > count.
 212      */
 213     private void fill() throws IOException {
 214         byte[] buffer = getBufIfOpen();
 215         if (markpos < 0)
 216             pos = 0;            /* no mark: throw away the buffer */
 217         else if (pos >= buffer.length)  /* no room left in buffer */
 218             if (markpos > 0) {  /* can throw away early part of the buffer */
 219                 int sz = pos - markpos;
 220                 System.arraycopy(buffer, markpos, buffer, 0, sz);
 221                 pos = sz;
 222                 markpos = 0;
 223             } else if (buffer.length >= marklimit) {
 224                 markpos = -1;   /* buffer got too big, invalidate mark */
 225                 pos = 0;        /* drop buffer contents */
 226             } else if (buffer.length >= MAX_BUFFER_SIZE) {
 227                 throw new OutOfMemoryError("Required array size too large");
 228             } else {            /* grow buffer */
 229                 int nsz = (pos <= MAX_BUFFER_SIZE - pos) ?
 230                         pos * 2 : MAX_BUFFER_SIZE;
 231                 if (nsz > marklimit)
 232                     nsz = marklimit;
 233                 byte nbuf[] = new byte[nsz];
 234                 System.arraycopy(buffer, 0, nbuf, 0, pos);
 235                 if (!bufUpdater.compareAndSet(this, buffer, nbuf)) {

 236                     // Can't replace buf if there was an async close.
 237                     // Note: This would need to be changed if fill()
 238                     // is ever made accessible to multiple threads.
 239                     // But for now, the only way CAS can fail is via close.
 240                     // assert buf == null;
 241                     throw new IOException("Stream closed");
 242                 }
 243                 buffer = nbuf;
 244             }

 245         count = pos;
 246         int n = getInIfOpen().read(buffer, pos, buffer.length - pos);
 247         if (n > 0)
 248             count = n + pos;
 249     }
 250 
 251     /**
 252      * See
 253      * the general contract of the <code>read</code>
 254      * method of <code>InputStream</code>.
 255      *
 256      * @return     the next byte of data, or <code>-1</code> if the end of the
 257      *             stream is reached.
 258      * @exception  IOException  if this input stream has been closed by
 259      *                          invoking its {@link #close()} method,
 260      *                          or an I/O error occurs.
 261      * @see        java.io.FilterInputStream#in
 262      */
 263     public synchronized int read() throws IOException {
 264         if (pos >= count) {


 459      *          the <code>mark</code> and <code>reset</code> methods.
 460      * @see     java.io.InputStream#mark(int)
 461      * @see     java.io.InputStream#reset()
 462      */
 463     public boolean markSupported() {
 464         return true;
 465     }
 466 
 467     /**
 468      * Closes this input stream and releases any system resources
 469      * associated with the stream.
 470      * Once the stream has been closed, further read(), available(), reset(),
 471      * or skip() invocations will throw an IOException.
 472      * Closing a previously closed stream has no effect.
 473      *
 474      * @exception  IOException  if an I/O error occurs.
 475      */
 476     public void close() throws IOException {
 477         byte[] buffer;
 478         while ( (buffer = buf) != null) {
 479             if (bufUpdater.compareAndSet(this, buffer, null)) {
 480                 InputStream input = in;
 481                 in = null;
 482                 if (input != null)
 483                     input.close();
 484                 return;
 485             }
 486             // Else retry in case a new buf was CASed in fill()
 487         }
 488     }
 489 }


   7  * published by the Free Software Foundation.  Oracle designates this
   8  * particular file as subject to the "Classpath" exception as provided
   9  * by Oracle in the LICENSE file that accompanied this code.
  10  *
  11  * This code is distributed in the hope that it will be useful, but WITHOUT
  12  * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
  13  * FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General Public License
  14  * version 2 for more details (a copy is included in the LICENSE file that
  15  * accompanied this code).
  16  *
  17  * You should have received a copy of the GNU General Public License version
  18  * 2 along with this work; if not, write to the Free Software Foundation,
  19  * Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA.
  20  *
  21  * Please contact Oracle, 500 Oracle Parkway, Redwood Shores, CA 94065 USA
  22  * or visit www.oracle.com if you need additional information or have any
  23  * questions.
  24  */
  25 
  26 package java.io;
  27 
  28 import jdk.internal.misc.Unsafe;
  29 
  30 /**
  31  * A <code>BufferedInputStream</code> adds
  32  * functionality to another input stream-namely,
  33  * the ability to buffer the input and to
  34  * support the <code>mark</code> and <code>reset</code>
  35  * methods. When  the <code>BufferedInputStream</code>
  36  * is created, an internal buffer array is
  37  * created. As bytes  from the stream are read
  38  * or skipped, the internal buffer is refilled
  39  * as necessary  from the contained input stream,
  40  * many bytes at a time. The <code>mark</code>
  41  * operation  remembers a point in the input
  42  * stream and the <code>reset</code> operation
  43  * causes all the  bytes read since the most
  44  * recent <code>mark</code> operation to be
  45  * reread before new bytes are  taken from
  46  * the contained input stream.
  47  *
  48  * @author  Arthur van Hoff


  52 class BufferedInputStream extends FilterInputStream {
  53 
  54     private static int DEFAULT_BUFFER_SIZE = 8192;
  55 
  56     /**
  57      * The maximum size of array to allocate.
  58      * Some VMs reserve some header words in an array.
  59      * Attempts to allocate larger arrays may result in
  60      * OutOfMemoryError: Requested array size exceeds VM limit
  61      */
  62     private static int MAX_BUFFER_SIZE = Integer.MAX_VALUE - 8;
  63 
  64     /**
  65      * The internal buffer array where the data is stored. When necessary,
  66      * it may be replaced by another array of
  67      * a different size.
  68      */
  69     protected volatile byte buf[];
  70 
  71     /**
  72      * Get Unsafe and offset of buf to provide compareAndSet functionality.
  73      * This is necessary because closes can be asynchronous. We use nullness
  74      * of buf[] as primary indicator that this stream is closed. (The
  75      * "in" field is also nulled out on close.)
  76      */
  77     private static final Unsafe UNSAFE = Unsafe.getUnsafe();
  78     private static final long BUF_OFFSET;
  79     static {
  80         try {
  81             BUF_OFFSET = UNSAFE.objectFieldOffset(
  82                     BufferedInputStream.class.getDeclaredField("buf"));
  83         } catch (ReflectiveOperationException e) {
  84             throw new Error(e);
  85         }
  86     }
  87 
  88     /**
  89      * The index one greater than the index of the last valid byte in
  90      * the buffer.
  91      * This value is always
  92      * in the range <code>0</code> through <code>buf.length</code>;
  93      * elements <code>buf[0]</code>  through <code>buf[count-1]
  94      * </code>contain buffered input data obtained
  95      * from the underlying  input stream.
  96      */
  97     protected int count;
  98 
  99     /**
 100      * The current position in the buffer. This is the index of the next
 101      * character to be read from the <code>buf</code> array.
 102      * <p>
 103      * This value is always in the range <code>0</code>
 104      * through <code>count</code>. If it is less
 105      * than <code>count</code>, then  <code>buf[pos]</code>
 106      * is the next byte to be supplied as input;


 202      * @param   size   the buffer size.
 203      * @exception IllegalArgumentException if {@code size <= 0}.
 204      */
 205     public BufferedInputStream(InputStream in, int size) {
 206         super(in);
 207         if (size <= 0) {
 208             throw new IllegalArgumentException("Buffer size <= 0");
 209         }
 210         buf = new byte[size];
 211     }
 212 
 213     /**
 214      * Fills the buffer with more data, taking into account
 215      * shuffling and other tricks for dealing with marks.
 216      * Assumes that it is being called by a synchronized method.
 217      * This method also assumes that all data has already been read in,
 218      * hence pos > count.
 219      */
 220     private void fill() throws IOException {
 221         byte[] buffer = getBufIfOpen();
 222         if (markpos < 0) {
 223             pos = 0;            /* no mark: throw away the buffer */
 224         } else if (pos >= buffer.length) { /* no room left in buffer */
 225             if (markpos > 0) {  /* can throw away early part of the buffer */
 226                 int sz = pos - markpos;
 227                 System.arraycopy(buffer, markpos, buffer, 0, sz);
 228                 pos = sz;
 229                 markpos = 0;
 230             } else if (buffer.length >= marklimit) {
 231                 markpos = -1;   /* buffer got too big, invalidate mark */
 232                 pos = 0;        /* drop buffer contents */
 233             } else if (buffer.length >= MAX_BUFFER_SIZE) {
 234                 throw new OutOfMemoryError("Required array size too large");
 235             } else {            /* grow buffer */
 236                 int nsz = (pos <= MAX_BUFFER_SIZE - pos) ?
 237                         pos * 2 : MAX_BUFFER_SIZE;
 238                 if (nsz > marklimit)
 239                     nsz = marklimit;
 240                 byte nbuf[] = new byte[nsz];
 241                 System.arraycopy(buffer, 0, nbuf, 0, pos);
 242                 if (!UNSAFE.compareAndSwapObject(this, BUF_OFFSET, buffer,
 243                         nbuf)) {
 244                     // Can't replace buf if there was an async close.
 245                     // Note: This would need to be changed if fill()
 246                     // is ever made accessible to multiple threads.
 247                     // But for now, the only way CAS can fail is via close.
 248                     // assert buf == null;
 249                     throw new IOException("Stream closed");
 250                 }
 251                 buffer = nbuf;
 252             }
 253         }
 254         count = pos;
 255         int n = getInIfOpen().read(buffer, pos, buffer.length - pos);
 256         if (n > 0)
 257             count = n + pos;
 258     }
 259 
 260     /**
 261      * See
 262      * the general contract of the <code>read</code>
 263      * method of <code>InputStream</code>.
 264      *
 265      * @return     the next byte of data, or <code>-1</code> if the end of the
 266      *             stream is reached.
 267      * @exception  IOException  if this input stream has been closed by
 268      *                          invoking its {@link #close()} method,
 269      *                          or an I/O error occurs.
 270      * @see        java.io.FilterInputStream#in
 271      */
 272     public synchronized int read() throws IOException {
 273         if (pos >= count) {


 468      *          the <code>mark</code> and <code>reset</code> methods.
 469      * @see     java.io.InputStream#mark(int)
 470      * @see     java.io.InputStream#reset()
 471      */
 472     public boolean markSupported() {
 473         return true;
 474     }
 475 
 476     /**
 477      * Closes this input stream and releases any system resources
 478      * associated with the stream.
 479      * Once the stream has been closed, further read(), available(), reset(),
 480      * or skip() invocations will throw an IOException.
 481      * Closing a previously closed stream has no effect.
 482      *
 483      * @exception  IOException  if an I/O error occurs.
 484      */
 485     public void close() throws IOException {
 486         byte[] buffer;
 487         while ( (buffer = buf) != null) {
 488             if (UNSAFE.compareAndSwapObject(this, BUF_OFFSET, buffer, null)) {
 489                 InputStream input = in;
 490                 in = null;
 491                 if (input != null)
 492                     input.close();
 493                 return;
 494             }
 495             // Else retry in case a new buf was CASed in fill()
 496         }
 497     }
 498 }
< prev index next >