This function returns the next scan range the reader should work on, checking for eos and error cases. If there isn't already a cached scan range or a scan range prepared by the disk threads, the caller waits on the disk threads.
| 452 | // for eos and error cases. If there isn't already a cached scan range or a scan |
| 453 | // range prepared by the disk threads, the caller waits on the disk threads. |
| 454 | Status RequestContext::GetNextUnstartedRange(ScanRange** range, bool* needs_buffers) { |
| 455 | DCHECK(range != nullptr); |
| 456 | *range = nullptr; |
| 457 | *needs_buffers = false; |
| 458 | |
| 459 | unique_lock<mutex> lock(lock_); |
| 460 | DCHECK(Validate()) << endl << DebugString(); |
| 461 | while (true) { |
| 462 | if (state_ == RequestContext::Cancelled) return CONTEXT_CANCELLED; |
| 463 | |
| 464 | if (num_unstarted_scan_ranges_.Load() == 0 && ready_to_start_ranges_.empty() |
| 465 | && cached_ranges_.empty()) { |
| 466 | // All ranges are done, just return. |
| 467 | return Status::OK(); |
| 468 | } |
| 469 | |
| 470 | if (!cached_ranges_.empty()) { |
| 471 | // We have a cached range. |
| 472 | *range = cached_ranges_.Dequeue(); |
| 473 | DCHECK((*range)->UseHdfsCache()); |
| 474 | bool cached_read_succeeded; |
| 475 | RETURN_IF_ERROR(TryReadFromCache(lock, *range, &cached_read_succeeded, |
| 476 | needs_buffers)); |
| 477 | if (cached_read_succeeded) return Status::OK(); |
| 478 | |
| 479 | // This range ended up not being cached. Loop again and pick up a new range. |
| 480 | AddRangeToDisk(lock, *range, ScheduleMode::UPON_GETNEXT_TAIL); |
| 481 | DCHECK(Validate()) << endl << DebugString(); |
| 482 | *range = nullptr; |
| 483 | continue; |
| 484 | } |
| 485 | |
| 486 | if (ready_to_start_ranges_.empty()) { |
| 487 | ready_to_start_ranges_cv_.Wait(lock); |
| 488 | } else { |
| 489 | *range = ready_to_start_ranges_.Dequeue(); |
| 490 | DCHECK(*range != nullptr); |
| 491 | int disk_id = (*range)->disk_id(); |
| 492 | DCHECK_EQ(*range, disk_states_[disk_id].next_scan_range_to_start()); |
| 493 | // Set this to nullptr, the next time this disk runs for this reader, it will |
| 494 | // get another range ready. |
| 495 | disk_states_[disk_id].set_next_scan_range_to_start(nullptr); |
| 496 | if ((*range)->buffer_manager_->is_internal_buffer()) { |
| 497 | // We can't schedule this range until the client gives us buffers. The context |
| 498 | // must be rescheduled regardless to ensure that 'next_scan_range_to_start' is |
| 499 | // refilled. |
| 500 | disk_states_[disk_id].ScheduleContext(lock, this, disk_id); |
| 501 | (*range)->SetBlockedOnBuffer(); |
| 502 | *needs_buffers = true; |
| 503 | } else { |
| 504 | ScheduleScanRange(lock, *range); |
| 505 | } |
| 506 | return Status::OK(); |
| 507 | } |
| 508 | } |
| 509 | } |
| 510 | |
| 511 | Status RequestContext::StartScanRange(ScanRange* range, bool* needs_buffers) { |