Method DiskThreadLoop

be/src/runtime/io/disk-io-mgr.cc:828–887

}

void DiskQueue::DiskThreadLoop(DiskIoMgr* io_mgr) {
  // The thread waits until there is work or the queue is shut down. If there is work,
  // performs the read or write requested. Locks are not taken when reading from or
  // writing to disk.
  while (true) {
    RequestContext* worker_context = nullptr;
    RequestRange* range = GetNextRequestRange(&worker_context);
    if (range == nullptr) {
      DCHECK(shut_down_);
      return;
    }
    // We are now working on behalf of a query, so set thread state appropriately.
    // See also IMPALA-6254 and IMPALA-6417.
    ScopedThreadContext tdi_scope(GetThreadDebugInfo(), worker_context->query_id(),
        worker_context->instance_id());

    switch (range->request_type()) {
      case RequestType::READ: {
        ScanRange* scan_range = static_cast<ScanRange*>(range);
        ReadOutcome outcome = scan_range->DoRead(this, disk_id_);
        worker_context->ReadDone(disk_id_, outcome, scan_range);
        break;
      }
      case RequestType::WRITE: {
        WriteRange* write_range = static_cast<WriteRange*>(range);
        Status status = write_range->DoWrite();
        worker_context->OperDone(write_range, status);
        break;
      }
      case RequestType::FILE_UPLOAD: {
        RemoteOperRange* oper_range = static_cast<RemoteOperRange*>(range);
        int64_t size = oper_range->block_size();
        // Use malloc to get the memory in case there is no available space
        // in the buffer pool because spilling to disk happens when scarcity
        // of memory in the buffer pool. Be better to preserve memory than
        // malloc.
        uint8_t* buffer = static_cast<uint8_t*>(malloc(size));
        if (UNLIKELY(buffer == nullptr)) {
          worker_context->OperDone(oper_range,
              Status(Substitute("Couldn't allocate memory for remote file operations, "
                                "block size: '$0'",
                  size)));
        } else {
          Status oper_status = oper_range->DoUpload(buffer, size);
          worker_context->OperDone(oper_range, oper_status);
          free(buffer);
        }
        break;
      }
      case RequestType::FILE_FETCH: {
        RemoteOperRange* oper_range = static_cast<RemoteOperRange*>(range);
        Status oper_status = oper_range->DoFetch();
        worker_context->OperDone(oper_range, oper_status);
        break;
      }
      default:
        DCHECK(false) << "Invalid request type: " << range->request_type();
    }