| 418 | } |
| 419 | |
| 420 | Status RequestContext::AddScanRanges( |
| 421 | const vector<ScanRange*>& ranges, EnqueueLocation enqueue_location) { |
| 422 | DCHECK_GT(ranges.size(), 0); |
| 423 | // Validate and initialize all ranges |
| 424 | for (int i = 0; i < ranges.size(); ++i) { |
| 425 | RETURN_IF_ERROR(parent_->ValidateScanRange(ranges[i])); |
| 426 | ranges[i]->InitInternal(parent_, this); |
| 427 | } |
| 428 | |
| 429 | unique_lock<mutex> lock(lock_); |
| 430 | DCHECK(Validate()) << endl << DebugString(); |
| 431 | |
| 432 | if (state_ == RequestContext::Cancelled) return CONTEXT_CANCELLED; |
| 433 | |
| 434 | // Add each range to the queue of the disk the range is on |
| 435 | for (ScanRange* range : ranges) { |
| 436 | // Don't add empty ranges. |
| 437 | DCHECK_NE(range->bytes_to_read(), 0); |
| 438 | AddActiveScanRangeLocked(lock, range); |
| 439 | if (range->UseHdfsCache()) { |
| 440 | cached_ranges_.Enqueue(range); |
| 441 | } else { |
| 442 | AddRangeToDisk(lock, range, (enqueue_location == EnqueueLocation::HEAD) ? |
| 443 | ScheduleMode::UPON_GETNEXT_HEAD : |
| 444 | ScheduleMode::UPON_GETNEXT_TAIL); |
| 445 | } |
| 446 | } |
| 447 | DCHECK(Validate()) << endl << DebugString(); |
| 448 | return Status::OK(); |
| 449 | } |
| 450 | |
| 451 | // This function returns the next scan range the reader should work on, checking |
| 452 | // for eos and error cases. If there isn't already a cached scan range or a scan |