Function SockSendWithTimeout

src/common/io_util.cc:474–514

}

Status SockSendWithTimeout(int fd, const std::string &data, int timeout_ms) {
  // Fall back to blocking send if timeout is non-positive
  if (timeout_ms <= 0) {
    return SockSend(fd, data);
  }

  ssize_t n = 0;
  auto start = std::chrono::steady_clock::now();

  while (n < static_cast<ssize_t>(data.size())) {
    // Check if we've exceeded the timeout
    auto elapsed =
        std::chrono::duration_cast<std::chrono::milliseconds>(std::chrono::steady_clock::now() - start).count();
    if (elapsed >= timeout_ms) {
      return {Status::NotOK, fmt::format("send timeout after {} ms, sent {} of {} bytes", elapsed, n, data.size())};
    }

    // Calculate remaining timeout
    int remaining_ms = timeout_ms - static_cast<int>(elapsed);

    // Wait for socket to be writable with timeout
    int ready = AeWait(fd, AE_WRITABLE, remaining_ms);
    if (ready == 0) {
      return {Status::NotOK, fmt::format("send timeout waiting for socket, sent {} of {} bytes", n, data.size())};
    }
    if (ready < 0) {
      return Status::FromErrno("poll error while sending");
    }

    ssize_t nwritten = write(fd, data.data() + n, data.size() - n);
    if (nwritten == -1) {
      if (errno == EAGAIN || errno == EWOULDBLOCK) {
        // Socket buffer is full, continue waiting
        continue;
      }
      return Status::FromErrno();
    }
    n += nwritten;
  }
  return Status::OK();
}

Status SockSendWithTimeout(int fd, const std::string &data, [[maybe_unused]] bufferevent *bev, int timeout_ms) {
  // Fall back to blocking send if timeout is non-positive