Method EnqueueWriteRange

be/src/runtime/tmp-file-mgr.cc:2313–2342

}

Status TmpFileBufferPool::EnqueueWriteRange(io::WriteRange* range, TmpFile* tmp_file) {
  Status status = Status::OK();
  {
    unique_lock<mutex> write_range_list_lock(lock_);
    DCHECK(range != nullptr);
    DCHECK(range->disk_file() != nullptr);
    DCHECK(range->io_ctx() != nullptr);
    if (range->disk_file()->IsSpaceReserved()) {
      // If the space is reserved, send the range to the DiskQueue.
      return range->io_ctx()->AddWriteRange(range);
    } else if (range->io_ctx()->IsCancelled()) {
      // If the io_ctx is cancelled, nofity the caller to cancel the query.
      return TMP_FILE_BUFFER_POOL_CONTEXT_CANCELLED;
    } else {
      io_ctx_to_file_set_map_[range->io_ctx()].insert(range->disk_file());
      write_ranges_to_add_[range->disk_file()].emplace_back(range);
    }
    // Put the first range of a file to the queue for waiting for the available space,
    // the ranges in the queue would be popped one by one, when the space is reserved,
    // all ranges of the file are added to the DiskQueue by io_ctx.
    if (range->offset() == 0) {
      write_ranges_.emplace_back(range);
      DCHECK(tmp_file != nullptr);
      write_ranges_iterator_[range] =
          std::make_pair(prev(write_ranges_.cend()), tmp_file);
    }
  }
  work_available_.NotifyAll();
  return status;
}

void TmpFileBufferPool::RemoveWriteRangesInternal(
    RequestContext* io_ctx, vector<TmpFileMgr::WriteDoneCallback>* write_callbacks) {