Method GetNextRequestRange

be/src/runtime/io/request-context.cc:605–677

This function gets the next RequestRange to work on for this RequestContext and disk combination this disk. It checks for cancellation and: a) Updates ready_to_start_ranges if there are no scan ranges queued for this disk. b) Adds an unstarted write range to in_flight_ranges_. The write range is processed immediately if there are no preceding scan ranges in in_flight_ranges_

// b) Adds an unstarted write range to in_flight_ranges_. The write range is processed
//    immediately if there are no preceding scan ranges in in_flight_ranges_
RequestRange* RequestContext::GetNextRequestRange(int disk_id) {
  PerDiskState* request_disk_state = &disk_states_[disk_id];
  // NOTE: no locks are held, so other threads could have modified the state of the reader
  // and disk state since this context was pulled off the queue. Only one disk thread can
  // be in this function for this reader, since the reader was removed from the queue and
  // has not be re-added. Other disk threads may be operating on this reader in other
  // functions though.
  unique_lock<mutex> request_lock(lock_);
  VLOG_FILE << "Disk (id=" << disk_id << ") reading for " << DebugString();

  // Check if reader has been cancelled
  if (state_ == RequestContext::Cancelled) {
    request_disk_state->DecrementDiskThread(request_lock, this);
    return nullptr;
  }
  DCHECK_EQ(state_, RequestContext::Active) << DebugString();
  if (request_disk_state->next_scan_range_to_start() == nullptr &&
      !request_disk_state->unstarted_scan_ranges()->empty()) {
    // We don't have a range queued for this disk for what the caller should
    // read next. Populate that.  We want to have one range waiting to minimize
    // wait time in GetNextUnstartedRange().
    ScanRange* new_range = request_disk_state->unstarted_scan_ranges()->Dequeue();
    num_unstarted_scan_ranges_.Add(-1);
    ready_to_start_ranges_.Enqueue(new_range);
    request_disk_state->set_next_scan_range_to_start(new_range);

    if (num_unstarted_scan_ranges_.Load() == 0) {
      // All the ranges have been started, notify everyone blocked on
      // GetNextUnstartedRange(). Only one of them will get work so make sure to return
      // nullptr to the other caller threads.
      ready_to_start_ranges_cv_.NotifyAll();
    } else {
      ready_to_start_ranges_cv_.NotifyOne();
    }
  }

  // Always enqueue a WriteRange to be processed into in_flight_ranges_.
  // This is done so in_flight_ranges_ does not exclusively contain ScanRanges.
  // For now, enqueuing a WriteRange on each invocation of GetNextRequestRange()
  // does not flood in_flight_ranges() with WriteRanges because the entire
  // WriteRange is processed and removed from the queue after GetNextRequestRange()
  // returns.
  if (!request_disk_state->unstarted_write_ranges()->empty()) {
    WriteRange* write_range = request_disk_state->unstarted_write_ranges()->Dequeue();
    request_disk_state->in_flight_ranges()->Enqueue(write_range);
  }

  // Do remote temporary files related work.
  if (!request_disk_state->unstarted_remote_file_oper_ranges()->empty()) {
    RemoteOperRange* oper_range;
    if (!request_disk_state->unstarted_remote_file_oper_ranges()->empty()) {
      oper_range = request_disk_state->unstarted_remote_file_oper_ranges()->Dequeue();
      request_disk_state->in_flight_ranges()->Enqueue(oper_range);
    }
  }

  // Get the next scan range to work on from the reader. Only in_flight_ranges
  // are eligible since the disk threads do not start new ranges on their own.