Method read

lib/cpp/src/thrift/transport/TSocket.cpp:522–644

}

uint32_t TSocket::read(uint8_t* buf, uint32_t len) {
  checkReadBytesAvailable(len);
  if (socket_ == THRIFT_INVALID_SOCKET) {
    throw TTransportException(TTransportException::NOT_OPEN, "Called read on non-open socket");
  }

  int32_t retries = 0;

  // THRIFT_EAGAIN can be signalled both when a timeout has occurred and when
  // the system is out of resources (an awesome undocumented feature).
  // The following is an approximation of the time interval under which
  // THRIFT_EAGAIN is taken to indicate an out of resources error.
  uint32_t eagainThresholdMicros = 0;
  if (recvTimeout_) {
    // if a readTimeout is specified along with a max number of recv retries, then
    // the threshold will ensure that the read timeout is not exceeded even in the
    // case of resource errors
    eagainThresholdMicros = (recvTimeout_ * 1000) / ((maxRecvRetries_ > 0) ? maxRecvRetries_ : 2);
  }

try_again:
  // Read from the socket
  struct timeval begin;
  if (recvTimeout_ > 0) {
    THRIFT_GETTIMEOFDAY(&begin, nullptr);
  } else {
    // if there is no read timeout we don't need the TOD to determine whether
    // an THRIFT_EAGAIN is due to a timeout or an out-of-resource condition.
    begin.tv_sec = begin.tv_usec = 0;
  }

  int got = 0;

  if (interruptListener_) {
    struct THRIFT_POLLFD fds[2];
    std::memset(fds, 0, sizeof(fds));
    fds[0].fd = socket_;
    fds[0].events = THRIFT_POLLIN;
    fds[1].fd = *(interruptListener_.get());
    fds[1].events = THRIFT_POLLIN;

    int ret = THRIFT_POLL(fds, 2, (recvTimeout_ == 0) ? -1 : recvTimeout_);
    int errno_copy = THRIFT_GET_SOCKET_ERROR;
    if (ret < 0) {
      // error cases
      if (errno_copy == THRIFT_EINTR && (retries++ < maxRecvRetries_)) {
        goto try_again;
      }
      TOutput::instance().perror("TSocket::read() THRIFT_POLL() ", errno_copy);
      throw TTransportException(TTransportException::UNKNOWN, "Unknown", errno_copy);
    } else if (ret > 0) {
      // Check the interruptListener
      if (fds[1].revents & THRIFT_POLLIN) {
        throw TTransportException(TTransportException::INTERRUPTED, "Interrupted");
      }
    } else /* ret == 0 */ {
      TOutput::instance().printf("TSocket::read() THRIFT_EAGAIN (timed out) after %d ms", recvTimeout_);
      throw TTransportException(TTransportException::TIMED_OUT, "THRIFT_EAGAIN (timed out)");