Method AsyncFetchReadBufferBlock

be/src/runtime/tmp-file-mgr.cc:1089–1151

}

void TmpFileRemote::AsyncFetchReadBufferBlock(io::DiskFile* read_buffer_file,
    io::MemBlock* read_buffer_block, int read_buffer_idx, bool* fetched) {
  DCHECK(fetched != nullptr);
  *fetched = false;
  {
    shared_lock<shared_mutex> read_file_lock(*(read_buffer_file->GetFileLock()));
    unique_lock<SpinLock> mem_bloc_lock(*(read_buffer_block->GetLock()));
    // Check the block status.
    // If the block is disabled, the caller won't be able to use this buffer block.
    // If the block is written, the block is already fetched, set the fetched flag and
    // return immediately.
    // If the block is uninitialized, we will fetch the block immediately but without
    // waiting for the fetch, so that it won't block the current page reading.
    // If the block is in reserved or alloc status, means one other thread is handling
    // the block, here we don't wait because the blocking could be expensive.
    if (read_buffer_file->IsReadBufferBlockStatus(read_buffer_block,
            io::MemBlockStatus::DISABLED, read_file_lock, &mem_bloc_lock)) {
      return;
    } else if (read_buffer_file->IsReadBufferBlockStatus(read_buffer_block,
                   io::MemBlockStatus::WRITTEN, read_file_lock, &mem_bloc_lock)) {
      *fetched = true;
      return;
    } else if (read_buffer_file->IsReadBufferBlockStatus(read_buffer_block,
                   io::MemBlockStatus::UNINIT, read_file_lock, &mem_bloc_lock)) {
      bool dofetch = true;
      int64_t mem_size_limit =
          file_group_->tmp_file_mgr()->GetRemoteMaxTotalReadBufferSize();
      auto read_mem_counter =
          file_group_->tmp_file_mgr()->scratch_read_memory_buffer_used_metric_;
      if (read_mem_counter->Increment(read_buffer_file->read_buffer_block_size())
          > mem_size_limit) {
        read_mem_counter->Increment(-1 * read_buffer_file->read_buffer_block_size());
        dofetch = false;
      }
      if (dofetch) {
        read_buffer_file->SetReadBufferBlockStatus(read_buffer_block,
            io::MemBlockStatus::RESERVED, read_file_lock, &mem_bloc_lock);
        RemoteOperRange::RemoteOperDoneCallback fetch_callback =
            [read_buffer_block, tmp_file = this](const Status& fetch_status) {
              if (!fetch_status.ok()) {
                // Disable the read buffer if fails to fetch.
                tmp_file->TryDeleteReadBufferExcl(read_buffer_block->block_id());
              }
            };
        fetch_ranges_[read_buffer_idx].reset(new RemoteOperRange(disk_file_.get(),
            read_buffer_file, file_group_->tmp_file_mgr()->GetRemoteTmpBlockSize(),
            disk_id(true), RequestType::FILE_FETCH, file_group_->io_mgr_, fetch_callback,
            GetReadBuffStartOffset(read_buffer_idx)));
        Status add_status = file_group_->io_ctx_->AddRemoteOperRange(
            fetch_ranges_[read_buffer_idx].get());
        if (!add_status.ok()) {
          read_buffer_file->SetReadBufferBlockStatus(read_buffer_block,
              io::MemBlockStatus::DISABLED, read_file_lock, &mem_bloc_lock);
        }
      } else {
        read_buffer_file->SetReadBufferBlockStatus(read_buffer_block,
            io::MemBlockStatus::DISABLED, read_file_lock, &mem_bloc_lock);
      }