Reads from the current chunk or polls a next chunk from the queue. This is synchronized to
allow reads to be used on different thread while still guaranteeing their sequentiality.
778232
@Override
  public synchronized int read(byte[] buf, int offset, int length) throws IOException {
    if (buf == null) {
      throw new IllegalArgumentException("buf must not be null");
    }

    if (!(offset >= 0 && length > 0 && offset < buf.length && length <= (buf.length - offset))) {
      throw new IllegalArgumentException("invalid offset and lengeth");
    }

    // Return early if closed is true; throw the saved exception if needed.
    if (closed) {
      if (exception != null) {
        throw exception;
      }
      return -1;
    }

    // Check if there is already a chunk that hasn't been exhausted; call take() if not.
    if (currentChunk == null) {
      if (currentChunkReadPos != -1) {
        throw new IllegalStateException("currentChunk is null but currentChunkReadPos is not -1");
      }

      byte[] next = null;
      try {
        if (timeoutMillis >= 0) {
          next = queue.poll(timeoutMillis, TimeUnit.MILLISECONDS);
        } else {
          next = queue.take();
        }
      } catch (InterruptedException e) {
        // Unlikely to happen.
        throw new AssertionError(e);
      }
      if (next == null) {
        closed = true;
        SocketTimeoutException timeoutException = new SocketTimeoutException();

        if (exception == null) {
          exception = timeoutException;
          // It is still possible that an endOffer(Exception) races and set exception at this point,
          // but timeoutException is still being thrown, and it's ok for subsequent reads to
          // throw the exception set by endOffer(Exception).
        }

        throw timeoutException;
      }

      // If it's the end marker, acknowledge that so that the buffer will mark itself as closed
      // for reading.
      if (next == CLOSED) {
        closed = true;
        if (exception != null) {
          throw exception;
        }
        return -1;
      }
      currentChunk = next;
      currentChunkReadPos = 0;
    }

    int available = currentChunk.length - currentChunkReadPos;
    if (length < available) {
      // Copy from the currentChunk.
      System.arraycopy(currentChunk, currentChunkReadPos, buf, offset, length);
      currentChunkReadPos += length;
      return length;
    } else {
      // Copy the entire currentChunk.
      System.arraycopy(currentChunk, currentChunkReadPos, buf, offset, available);
      currentChunk = null;
      currentChunkReadPos = -1;
      return available;
    }
  }
645741
def _handle_read_chunk(self):
        """"""
        new_data = b''
        buffer_length = len(self.read_buffer)
        try:
            while buffer_length < self.MAX_BUFFER_SIZE:
                try:
                    piece = self.recv(4096)
                except OSError as e:
                    if e.errno == errno.EAGAIN:
                        # End of the available data
                        break
                    elif e.errno == errno.EIO and new_data:
                        # Hopefully we could read an error message before the
                        # actual termination
                        break
                    else:
                        raise

                if not piece:
                    # A closed connection is indicated by signaling a read
                    # condition, and having recv() return 0.
                    break

                new_data += piece
                buffer_length += len(piece)

        finally:
            new_data = new_data.replace(b'\r', b'\n')
            self.read_buffer += new_data
        return new_data
760796
@Override
    protected void doRead(final Action<ByteBuffer> chunkAction) {
        // Using one of Play's thread pools may be a better way to go?
        // https://www.playframework.com/documentation/2.3.x/ThreadPools
        // TODO https://github.com/vibe-project/vibe-java-platform/issues/6
        new Thread(new Runnable() {
            @Override
            public void run() {
                chunkAction.on(ByteBuffer.wrap(request.body().asRaw().asBytes()));
                endActions.fire();
            }
        })
        .start();
    }
108971
private Thread fullGetFirstQueuedThread() {
        /*
         * The first node is normally head.next. Try to get its
         * thread field, ensuring consistent reads: If thread
         * field is nulled out or s.prev is no longer head, then
         * some other thread(s) concurrently performed setHead in
         * between some of our reads. We try this twice before
         * resorting to traversal.
         */
        Node h, s;
        Thread st;
        if (((h = head) != null && (s = h.next) != null &&
             s.prev == head && (st = s.thread) != null) ||
            ((h = head) != null && (s = h.next) != null &&
             s.prev == head && (st = s.thread) != null))
            return st;

        /*
         * Head's next field might not have been set yet, or may have
         * been unset after setHead. So we must check to see if tail
         * is actually first node. If not, we continue on, safely
         * traversing from tail back to head to find first,
         * guaranteeing termination.
         */

        Thread firstThread = null;
        for (Node p = tail; p != null && p != head; p = p.prev) {
            Thread t = p.thread;
            if (t != null)
                firstThread = t;
        }
        return firstThread;
    }
1038688
private void getRead() {
		// Execute on background thread as we don't know how large this is
		new AsyncTask<Void, Void, Void>() {
			@Override
			protected Void doInBackground(Void... params) {
				int size = mPrefs.getInt("READ_ARRAY_SIZE", 0);
				boolean value;
				if(size < 1)
					return null;

				for(int i = 0, key; i < size; i++) {
					key = mPrefs.getInt("READ_ARRAY_KEY_" + i, 0);
					value = mPrefs.getBoolean("READ_ARRAY_VALUE_" + i, false);

					readList.put(key, value);
				}
				return null;
			}
		}.executeOnExecutor(AsyncTask.SERIAL_EXECUTOR);
	}
659461
int receive(byte[] b, int offset, int length) throws IOException, InterruptedException {

    synchronized (readLock) {
      if (remaining > 0) {
        int bytesToRead = Math.min(remaining, length);
        int result = stdin.read(b, offset, bytesToRead);
        remaining -= result;
        return result;
      }
      if (eof) {
        return -1;
      }
    }

    // make client know we want more data!
    sendSendInput();

    synchronized (readLock) {
      if (remaining == 0 && !eof) {
        readLock.wait();
      }

      // at this point we should have data so call itself recursively to utilize
      // reentrant lock
      return receive(b, offset, length);
    }
  }
428583
async def read(self):
        
        payload = await self._queue.get()
        self._queue.task_done()
        return payload
935545
def _read_select_poll(poll):
        
        npipes = len(NonBlockingStreamReader._streams)
        for stream_descr in NonBlockingStreamReader._streams:
            if not stream_descr.has_error:
                poll.register(stream_descr.stream,
                              select.POLLIN | select.POLLERR | select.POLLHUP | select.POLLNVAL)

        while NonBlockingStreamReader._run_flag:
            for (file_descr, event) in poll.poll(500):
                if event == select.POLLIN:
                    NonBlockingStreamReader._read_fd(file_descr)
                else:
                    # Because event != select.POLLIN, the pipe is closed
                    # but we still want to read all bytes
                    while NonBlockingStreamReader._read_fd(file_descr) != 0:
                        pass
                    # Dut died, signal the processing thread so it notices that no lines coming in
                    stream_descr = NonBlockingStreamReader._get_sd(file_descr)
                    if stream_descr is None:
                        return # PIPE closed but DUT already disappeared
                    stream_descr.has_error = True
                    if stream_descr.callback is not None:
                        stream_descr.callback()
                        return # Force poll object to reregister only alive descriptors

            # Check if new pipes added, don't need mutext just for reading the size
            # If we will not get it right now, we will at next time
            if npipes != len(NonBlockingStreamReader._streams):
                return
620236
private void readChunk() throws IOException {
    if (mDataReader == null) {
      mDataReader = mDataReaderFactory.create(mPos, mLength - mPos);
    }

    if (mCurrentChunk != null && mCurrentChunk.readableBytes() == 0) {
      mCurrentChunk.release();
      mCurrentChunk = null;
    }
    if (mCurrentChunk == null) {
      mCurrentChunk = mDataReader.readChunk();
    }
  }
212180
private boolean readMore(int wanted) throws IOException {
    if (endIndex == index) {
      index = 0;
      endIndex = 0;
    }
    int canFit = buffer.length - endIndex;
    if (canFit < wanted) {
      // would the wanted bytes fit if we compacted the buffer
      // and still leave some slack
      if (index + canFit > wanted + MINIMUM_READ) {
        compact();
      } else {
        doubleBuffer();
      }
      canFit = buffer.length - endIndex;
    }
    int read = wrapped.read(buffer, endIndex, canFit);
    if (read < 0) {
      return false;
    }
    endIndex += read;
    return true;
  }
135458
protected synchronized int read() throws IOException {
		try {
			block();
		} catch (InterruptedException ex) {
			throw new InterruptedIOException(
					"The blocking operation was interrupted");
		}

		if (closed && available() <= 0) {
			return -1;
		}

		return buf[readpos++];

	}
663061
private void reads() {
    int index = random.nextInt();
    for (;;) {
      Integer key = ints[index++ & MASK];
      cache.get(key);
      calls.increment();
    }
  }
138145
private boolean readNextChunkFromFile() throws IOException {
        if (readOffset >= filledUpToWhenStarted || readOffset < 0) {
            return false;
        }

        bufferedReader.clear();
        dataFile.read(readOffset, bufferedReader);
        bufferedReader.position(0);
        if (bufferedReader.limit() > 0) {
            readOffset += bufferedReader.limit();
            while (bufferedReader.remaining() > 0) {
                byte[] dst = new byte[elementSize];
                bufferedReader.get(dst);
                pendingElements.add(dst);
            }
        }
        return true;
    }
852793
@Override
    public Chunk<byte[], BytesReference> readChunk() throws IOException {
        Chunk<byte[], BytesReference> chunk = internalReadChunk();
        if (chunk != null) {
            processChunk(chunk);
        }
        return chunk;
    }
547627
def acquire_read(self):
        
        self.monitor.acquire()
        while self.rwlock < 0 or self.writers_waiting:
            self.readers_ok.wait()
        self.rwlock += 1
        self.monitor.release()
1083130
def release_read_lock
      while(true)
        c = @counter.value
        if @counter.compare_and_swap(c,c-1)
          if waiting_writer?(c) && running_readers(c) == 1
            @writer_mutex.synchronize { @writer_q.signal }
          end
          break
        end
      end
      true
    end
329347
def read(self):
        
        """
        # Read until we have a \0 in our buffer.
        while b'\0' not in self._recv_buffer:
            self._recv_buffer += yield From(_read_chunk_from_socket(self.socket))

        # Split on the first separator.
        pos = self._recv_buffer.index(b'\0')

        packet = self._recv_buffer[:pos]
        self._recv_buffer = self._recv_buffer[pos + 1:]

        raise Return(packet)