Method AddRangeToDisk

be/src/runtime/io/request-context.cc:372–418

}

void RequestContext::AddRangeToDisk(const unique_lock<mutex>& lock,
    RequestRange* range, ScheduleMode schedule_mode) {
  DCHECK(lock.mutex() == &lock_ && lock.owns_lock());
  DCHECK_EQ(state_, Active) << DebugString();
  PerDiskState* disk_state = &disk_states_[range->disk_id()];
  if (disk_state->done()) {
    DCHECK_EQ(disk_state->num_remaining_ranges(), 0);
    disk_state->set_done(false);
    ++num_disks_with_ranges_;
  }
  if (range->request_type() == RequestType::READ) {
    ScanRange* scan_range = static_cast<ScanRange*>(range);
    if (schedule_mode == ScheduleMode::IMMEDIATELY) {
      ScheduleScanRange(lock, scan_range);
    } else if (schedule_mode != ScheduleMode::BY_CALLER) {
      if (schedule_mode == ScheduleMode::UPON_GETNEXT_TAIL) {
        disk_state->unstarted_scan_ranges()->Enqueue(scan_range);
      } else {
        DCHECK_ENUM_EQ(schedule_mode, ScheduleMode::UPON_GETNEXT_HEAD);
        disk_state->unstarted_scan_ranges()->PushFront(scan_range);
      }
      num_unstarted_scan_ranges_.Add(1);
      // If there's no 'next_scan_range_to_start', schedule this RequestContext so that
      // one of the 'unstarted_scan_ranges' will become the 'next_scan_range_to_start'.
      if (disk_state->next_scan_range_to_start() == nullptr) {
        disk_state->ScheduleContext(lock, this, range->disk_id());
      }
    }
  } else if (range->request_type() == RequestType::WRITE) {
    DCHECK(schedule_mode == ScheduleMode::IMMEDIATELY) << static_cast<int>(schedule_mode);
    WriteRange* write_range = static_cast<WriteRange*>(range);
    disk_state->unstarted_write_ranges()->Enqueue(write_range);

    // Ensure that the context is scheduled so that the write range gets picked up.
    // ScheduleContext() has no effect if already scheduled, so this is safe to do always.
    disk_state->ScheduleContext(lock, this, range->disk_id());
  } else {
    DCHECK(range->request_type() == RequestType::FILE_UPLOAD
        || range->request_type() == RequestType::FILE_FETCH);
    DCHECK(schedule_mode == ScheduleMode::IMMEDIATELY) << static_cast<int>(schedule_mode);
    RemoteOperRange* oper_range = static_cast<RemoteOperRange*>(range);
    disk_state->unstarted_remote_file_oper_ranges()->Enqueue(oper_range);
    disk_state->ScheduleContext(lock, this, range->disk_id());
  }

  ++disk_state->num_remaining_ranges();
}

Status RequestContext::AddScanRanges(
    const vector<ScanRange*>& ranges, EnqueueLocation enqueue_location) {