Method CancelInternal

be/src/runtime/io/scan-range.cc:397–437

}

void ScanRange::CancelInternal(const Status& status, bool read_error) {
  DCHECK(io_mgr_ != nullptr);
  DCHECK(!status.ok());
  FileReader* file_reader = nullptr;
  {
    // Grab both locks to make sure that we don't change 'cancel_status_' while other
    // threads are in critical sections.
    unique_lock<mutex> scan_range_lock(lock_);
    {
      file_reader = use_local_buffer_ ? local_buffer_reader_.get() : file_reader_.get();
      unique_lock<SpinLock> fs_lock(file_reader->lock());
      DCHECK(Validate(scan_range_lock)) << DebugString();
      // If already cancelled, preserve the original reason for cancellation. Most of the
      // cleanup is not required if already cancelled, but we need to set
      // 'read_in_flight_' to false.
      if (cancel_status_.ok()) cancel_status_ = status;
    }

    /// Clean up ready buffers (i.e., 'buffer_manager_->ready_buffers_') while still
    /// holding 'lock_' to prevent other threads from seeing inconsistent state.
    buffer_manager_->CleanUpReadyBuffers(scan_range_lock);

    /// Clean up buffers that we don't need any more because we won't read any more data.
    buffer_manager_->CleanUpUnusedBuffers(scan_range_lock);
    if (read_error) {
      DCHECK(read_in_flight_);
      read_in_flight_ = false;
    }
  }
  buffer_ready_cv_.NotifyAll();

  // For cached buffers, we can't close the range until the cached buffer is returned.
  // Close() is called from ScanRange::CleanUpBufferLocked().
  // TODO: IMPALA-4249 - this Close() call makes it unsafe to reuse a cancelled scan
  // range, because there is no synchronisation between this Close() call and the
  // client adding the ScanRange back into the IoMgr.
  if (!buffer_manager_->is_cached()) {
    DCHECK(file_reader != nullptr);
    file_reader->Close();
  }
}

void ScanRange::WaitForInFlightRead() {
  unique_lock<mutex> scan_range_lock(lock_);