Method AllocateInternal

be/src/runtime/bufferpool/buffer-allocator.cc:274–371

}

Status BufferPool::BufferAllocator::AllocateInternal(
    BufferPool::Client* client, int64_t len, BufferHandle* buffer) {
  DCHECK(!buffer->is_open());
  DCHECK_GE(len, min_buffer_len_);
  DCHECK(BitUtil::IsPowerOf2(len)) << len;

  if (UNLIKELY(len > MAX_BUFFER_BYTES)) {
    return Status(Substitute(
        "Tried to allocate buffer of $0 bytes > max of $1 bytes", len, MAX_BUFFER_BYTES));
  }
  if (UNLIKELY(len > system_bytes_limit_)) {
    return Status(Substitute("Tried to allocate buffer of $0 bytes > buffer pool limit "
        "of $1 bytes", len, system_bytes_limit_));
  }

  const int current_core = CpuInfo::GetCurrentCore();
  // Fast path: recycle a buffer of the correct size from this core's arena.
  FreeBufferArena* current_core_arena = per_core_arenas_[current_core].get();
  if (current_core_arena->PopFreeBuffer(len, buffer)) {
    current_core_arena->local_arena_free_buffer_hits()->Increment(1);
    return Status::OK();
  }

  // Fast-ish path: allocate a new buffer if there is room in 'system_bytes_remaining_'.
  int64_t delta = DecreaseBytesRemaining(len, true, &system_bytes_remaining_);
  // Whether to record stats about the system alloc (we don't want to do this every
  // allocation because of the overhead).
  bool sample_sys_alloc_stats = false;
  if (delta == len) {
    int64_t count = current_core_arena->direct_alloc_count()->Increment(1);
    sample_sys_alloc_stats = count % ALLOC_STAT_SAMPLE_RATE == 0;
  } else {
    DCHECK_EQ(0, delta);
    const vector<int>& numa_node_cores = CpuInfo::GetCoresOfSameNumaNode(current_core);
    const int numa_node_core_idx = CpuInfo::GetNumaNodeCoreIdx(current_core);

    // Fast-ish path: find a buffer of the right size from another core on the same
    // NUMA node. Avoid getting a buffer from another NUMA node - prefer reclaiming
    // a clean page on this NUMA node or scavenging then reallocating a new buffer.
    // We don't want to get into a state where allocations between the nodes are
    // unbalanced and one node is stuck reusing memory allocated on the other node.
    for (int i = 1; i < numa_node_cores.size(); ++i) {
      // Each core should start searching from a different point to avoid hot-spots.
      int other_core = numa_node_cores[(numa_node_core_idx + i) % numa_node_cores.size()];
      FreeBufferArena* other_core_arena = per_core_arenas_[other_core].get();
      if (other_core_arena->PopFreeBuffer(len, buffer)) {
        current_core_arena->numa_arena_free_buffer_hits()->Increment(1);
        return Status::OK();
      }
    }

    // Fast-ish path: evict a clean page of the right size from the current NUMA node.
    for (int i = 0; i < numa_node_cores.size(); ++i) {
      int other_core = numa_node_cores[(numa_node_core_idx + i) % numa_node_cores.size()];
      FreeBufferArena* other_core_arena = per_core_arenas_[other_core].get();
      if (other_core_arena->EvictCleanPage(len, buffer)) {
        current_core_arena->clean_page_hits()->Increment(1);
        return Status::OK();