Method ScavengeBuffers

be/src/runtime/bufferpool/buffer-allocator.cc:386–435

}

int64_t BufferPool::BufferAllocator::ScavengeBuffers(
    bool slow_but_sure, int current_core, int64_t target_bytes) {
  // There are two strategies for scavenging buffers:
  // 1) Fast, opportunistic: Each arena is searched in succession. Although reservations
  //    guarantee that the memory we need is available somewhere, this may fail if we
  //    we race with another thread that returned buffers to an arena that we've already
  //    searched and took the buffers from an arena we haven't yet searched.
  // 2) Slow, guaranteed to succeed: In order to ensure that we can find the memory in a
  //    single pass, we hold locks for all arenas we've already examined. That way, other
  //    threads can't take the memory that we need from an arena that we haven't yet
  //    examined (or from 'system_bytes_available_') because in order to do so, it would
  //    have had to return the equivalent amount of memory to an earlier arena or added
  //    it back into 'systems_bytes_reamining_'. The former can't happen since we're
  //    still holding those locks, and the latter is solved by trying to decrease
  //    system_bytes_remaining_ with DecreaseBytesRemaining() at the end.
  DCHECK_GT(target_bytes, 0);
  // First make sure we've used up all the headroom in the buffer limit.
  int64_t bytes_found =
      DecreaseBytesRemaining(target_bytes, false, &system_bytes_remaining_);
  if (bytes_found == target_bytes) return bytes_found;

  // In 'slow_but_sure' mode, we will hold locks for multiple arenas at the same time and
  // therefore must start at 0 to respect the lock order. Otherwise we start with the
  // current core's arena for locality and to avoid excessive contention on arena 0.
  int start_core = slow_but_sure ? 0 : current_core;
  vector<std::unique_lock<SpinLock>> arena_locks;
  if (slow_but_sure) arena_locks.resize(per_core_arenas_.size());

  for (int i = 0; i < per_core_arenas_.size(); ++i) {
    int core_to_check = (start_core + i) % per_core_arenas_.size();
    FreeBufferArena* arena = per_core_arenas_[core_to_check].get();
    int64_t bytes_needed = target_bytes - bytes_found;
    bytes_found += arena->FreeSystemMemory(bytes_needed, bytes_needed,
         slow_but_sure ? &arena_locks[i] : nullptr).second;
    if (bytes_found == target_bytes) break;
  }
  DCHECK_LE(bytes_found, target_bytes);

  // Decrement 'system_bytes_remaining_' while still holding the arena locks to avoid
  // the window for a race with another thread that removes a buffer from a list and
  // then increments 'system_bytes_remaining_'. The race is prevented because the other
  // thread holds the lock while decrementing 'system_bytes_remaining_' in the cases
  // where it may not have reservation corresponding to that memory.
  if (slow_but_sure && bytes_found < target_bytes) {
    bytes_found += DecreaseBytesRemaining(
        target_bytes - bytes_found, true, &system_bytes_remaining_);
    DCHECK_EQ(bytes_found, target_bytes) << DebugString();
  }
  return bytes_found;
}

void BufferPool::BufferAllocator::Free(BufferHandle&& handle) {
  DCHECK(handle.is_open());