| 509 | } |
| 510 | |
| 511 | Status RequestContext::StartScanRange(ScanRange* range, bool* needs_buffers) { |
| 512 | RETURN_IF_ERROR(parent_->ValidateScanRange(range)); |
| 513 | range->InitInternal(parent_, this); |
| 514 | |
| 515 | unique_lock<mutex> lock(lock_); |
| 516 | DCHECK(Validate()) << endl << DebugString(); |
| 517 | if (state_ == RequestContext::Cancelled) return CONTEXT_CANCELLED; |
| 518 | |
| 519 | DCHECK_NE(range->bytes_to_read(), 0); |
| 520 | if (range->UseHdfsCache()) { |
| 521 | bool cached_read_succeeded; |
| 522 | RETURN_IF_ERROR(TryReadFromCache(lock, range, &cached_read_succeeded, |
| 523 | needs_buffers)); |
| 524 | if (cached_read_succeeded) return Status::OK(); |
| 525 | // Cached read failed, fall back to normal read path. |
| 526 | } |
| 527 | // If we don't have a buffer yet, the caller must allocate buffers for the range. |
| 528 | *needs_buffers = |
| 529 | range->buffer_manager_->is_internal_buffer(); |
| 530 | if (*needs_buffers) range->SetBlockedOnBuffer(); |
| 531 | AddActiveScanRangeLocked(lock, range); |
| 532 | AddRangeToDisk(lock, range, |
| 533 | *needs_buffers ? ScheduleMode::BY_CALLER : ScheduleMode::IMMEDIATELY); |
| 534 | DCHECK(Validate()) << endl << DebugString(); |
| 535 | return Status::OK(); |
| 536 | } |
| 537 | |
| 538 | Status RequestContext::TryReadFromCache(const unique_lock<mutex>& lock, |
| 539 | ScanRange* range, bool* read_succeeded, bool* needs_buffers) { |