Method InsertIntoCache

be/src/runtime/io/data-cache.cc:898–956

}

bool DataCache::Partition::InsertIntoCache(const Slice& key, CacheFile* cache_file,
    int64_t insertion_offset, const uint8_t* buffer, int64_t buffer_len) {
  if (UNLIKELY(trace_replay_)) {
    DCHECK(buffer == nullptr);
    DCHECK(cache_file == nullptr);
  }
  DCHECK_EQ(insertion_offset % PAGE_SIZE, 0);
  const int64_t charge_len = BitUtil::RoundUp(buffer_len, PAGE_SIZE);

  // Allocate a cache handle
  Cache::UniquePendingHandle pending_handle(
      meta_cache_->Allocate(key, sizeof(CacheEntry), charge_len));
  if (UNLIKELY(pending_handle.get() == nullptr)) return false;

  int64_t checksum = 0;
  // Trace replays have no data and cannot do checksums
  if (LIKELY(!trace_replay_)) {
    // Compute checksum if necessary.
    checksum = FLAGS_data_cache_checksum ? Checksum(buffer, buffer_len) : 0;

    // Write to backing file.
    VLOG(3) << Substitute("Storing file $0 offset $1 len $2 checksum $3 ",
        cache_file->path(), insertion_offset, buffer_len, checksum);
    bool write_success;
    {
      ScopedHistogramTimer write_timer(write_latency_);
      write_success = cache_file->Write(insertion_offset, buffer, buffer_len);
    }
    if (UNLIKELY(!write_success)) {
      return false;
    }
  }

  // IMPALA-10971: These metrics need to be incremented prior to the Insert(), because
  // the Insert() can fail and instantly evict the entry. When that happens,
  // the total bytes and num entries are decremented. Without this corresponding
  // increment, the counts will be incorrect.
  // Trace replays do not keep metrics
  if (LIKELY(!trace_replay_)) {
    ImpaladMetrics::IO_MGR_REMOTE_DATA_CACHE_TOTAL_BYTES->Increment(charge_len);
    ImpaladMetrics::IO_MGR_REMOTE_DATA_CACHE_NUM_ENTRIES->Increment(1);
    ImpaladMetrics::IO_MGR_REMOTE_DATA_CACHE_NUM_WRITES->Increment(1);
  }

  // Insert the new entry into the cache.
  CacheEntry entry(cache_file, insertion_offset, buffer_len, checksum);
  memcpy(meta_cache_->MutableValue(&pending_handle), &entry, sizeof(CacheEntry));
  Cache::UniqueHandle handle(meta_cache_->Insert(std::move(pending_handle), this));
  // Check for failure of Insert(), which means the entry was evicted during Insert()
  if (UNLIKELY(handle.get() == nullptr)){
    // Trace replays do not keep metrics
    if (LIKELY(!trace_replay_)) {
      // EvictedEntry() already ran and decremented the other counters.
      ImpaladMetrics::IO_MGR_REMOTE_DATA_CACHE_INSTANT_EVICTIONS->Increment(1);
    }
    return false;
  }
  return true;