Method TmpFileSpaceReserveThreadLoop

be/src/runtime/tmp-file-mgr.cc:2221–2279

}

void TmpFileBufferPool::TmpFileSpaceReserveThreadLoop() {
  while (true) {
    {
      unique_lock<mutex> l(lock_);
      while (!shut_down_ && write_ranges_.empty()) {
        // Wait if there are no ranges in the queue.
        work_available_.Wait(l);
      }
      if (shut_down_) return;
      DCHECK(!write_ranges_.empty());
      cur_write_range_ = write_ranges_.front();
      write_ranges_.pop_front();
      DCHECK(cur_write_range_ != nullptr);

      // Find out the TmpFile which the current range is associated with, and store the
      // shared_ptr of the file to cur_tmp_file_ in case it is deconstructed while waiting
      // for reservation.
      auto it = write_ranges_iterator_.find(cur_write_range_);
      DCHECK(it != write_ranges_iterator_.end());
      TmpFile* tmp_file = it->second.second;
      cur_tmp_file_ = tmp_file->FileGroup()->FindTmpFileSharedPtr(tmp_file);
      DCHECK(cur_tmp_file_ != nullptr);
      DCHECK_EQ(cur_write_range_->disk_file(), cur_tmp_file_->GetWriteFile());
      write_ranges_iterator_.erase(it);
    }

    // Reserve space from the tmp_files_avail_pool_. The process could need a long wait.
    Status status = tmp_file_mgr_->ReserveLocalBufferSpace(false);
    vector<TmpFileMgr::WriteDoneCallback> write_callbacks;
    {
      unique_lock<mutex> lock(lock_);
      if (status.ok()) {
        DCHECK(cur_tmp_file_ != nullptr);
        cur_tmp_file_->GetWriteFile()->SetSpaceReserved();
        if (cur_write_range_ != nullptr) {
          // Send all of the writes of the same disk file to the disk queue.
          status = MoveWriteRangesHelper(
              cur_write_range_->disk_file(), &write_callbacks, false);
        } else {
          // If the current range becomes a nullptr, it must be set by
          // RemoveWriteRanges(). In this case, the io_ctx which the range belongs to is
          // cancelled, and all the writes using that io_ctx are already cancelled. So, we
          // are safe to return the TmpFile to the pool to recycle the buffer space.
          EnqueueTmpFilesPool(cur_tmp_file_, true);
        }
      } else if (!status.ok() && cur_write_range_ != nullptr) {
        // Cancel the spilling if fails to reserve the buffer.
        RemoveWriteRangesInternal(cur_write_range_->io_ctx(), &write_callbacks);
        status = Status::CancelledInternal(
            Substitute("TmpFileBufferPool because: $0", status.GetDetail()).c_str());
      }
      cur_write_range_ = nullptr;
      cur_tmp_file_.reset();
    }
    for (const TmpFileMgr::WriteDoneCallback& write_callback : write_callbacks) {
      write_callback(status);
    }
  }