Method Cancel

be/src/runtime/io/request-context.cc:283–346

Cancellation of a RequestContext requires coordination from multiple threads that may hold references to the context: 1. Disk threads that are currently processing a range for this context. 2. Caller threads that are waiting in GetNext(). Each thread that currently has a reference to the request context must notice the cancel, cancel any pending operations involving the context and remove the con

// DecrementDiskRefCount(). After the last disk thread has called DecrementDiskRefCount(),
// cancellation is done and it is safe to unregister the context.
void RequestContext::Cancel() {
  // Callbacks are collected in this vector and invoked while no lock is held.
  vector<WriteRange::WriteDoneCallback> write_callbacks;
  vector<RemoteOperRange::RemoteOperDoneCallback> remote_oper_callbacks;
  {
    unique_lock<mutex> lock(lock_);
    DCHECK(Validate()) << endl << DebugString();

    // Already being cancelled
    if (state_ == RequestContext::Cancelled) return;

    // The reader will be put into a cancelled state until call cleanup is complete.
    state_ = RequestContext::Cancelled;

    // Clear out all request ranges from queues for this reader. Cancel the scan
    // ranges and invoke the write range callbacks to propagate the cancellation.
    for (ScanRange* range : active_scan_ranges_) {
      range->CancelInternal(CONTEXT_CANCELLED, false);
    }
    active_scan_ranges_.clear();
    for (PerDiskState& disk_state : disk_states_) {
      RequestRange* range;
      while ((range = disk_state.in_flight_ranges()->Dequeue()) != nullptr) {
        if (range->request_type() == RequestType::WRITE) {
          write_callbacks.push_back(static_cast<WriteRange*>(range)->callback());
        }
      }
      while (disk_state.unstarted_scan_ranges()->Dequeue() != nullptr);
      WriteRange* write_range;
      while ((write_range = disk_state.unstarted_write_ranges()->Dequeue()) != nullptr) {
        write_callbacks.push_back(write_range->callback());
      }

      RemoteOperRange* oper_range;
      while ((oper_range = disk_state.unstarted_remote_file_oper_ranges()->Dequeue())
          != nullptr) {
        remote_oper_callbacks.push_back(oper_range->callback());
      }
    }
    // Clear out the lists of scan ranges.
    while (ready_to_start_ranges_.Dequeue() != nullptr);
    while (cached_ranges_.Dequeue() != nullptr);

    // Ensure that the reader is scheduled on all disks (it may already be scheduled on
    // some). The disk threads will notice that the context is cancelled and do any
    // required cleanup for the disk state.
    for (int i = 0; i < disk_states_.size(); ++i) {
      disk_states_[i].ScheduleContext(lock, this, i);
    }
  }

  for (const WriteRange::WriteDoneCallback& write_callback: write_callbacks) {
    write_callback(CONTEXT_CANCELLED);
  }

  for (const RemoteOperRange::RemoteOperDoneCallback& oper_callback :
      remote_oper_callbacks) {
    oper_callback(CONTEXT_CANCELLED);