Method CleanPages

be/src/runtime/bufferpool/buffer-pool.cc:663–717

}

Status BufferPool::Client::CleanPages(
    unique_lock<mutex>* client_lock, int64_t len, bool lazy_flush) {
  DCHECK_GE(len, 0);
  DCHECK_LE(len, reservation_.GetReservation());
  DCheckHoldsLock(*client_lock);
  // If another thread is in CleanPages() for this client (and has dropped the lock while
  // waiting on 'write_complete_cv_', wait for it to flush its pages before proceeding so
  // that we don't overcommit memory.
  while (cleaning_pages_) clean_pages_done_cv_.Wait(*client_lock);

  DCHECK_CONSISTENCY();
  // Work out what we need to get bytes of dirty unpinned + in flight pages down to
  // in order to satisfy the eviction policy.
  int64_t target_dirty_bytes = reservation_.GetReservation() - buffers_allocated_bytes_
      - pinned_pages_.bytes() - len;
  if (VLOG_IS_ON(3)) {
    VLOG(3)   << "target_dirty_bytes=" << target_dirty_bytes
              << " reservation=" << reservation_.GetReservation()
              << " buffers_allocated_bytes_=" << buffers_allocated_bytes_
              << " pinned_pages_.bytes()=" << pinned_pages_.bytes()
              << " len=" << len << "\n"
              << DebugStringLocked();
  }
  // Start enough writes to ensure that the loop condition below will eventually become
  // false (or a write error will be encountered).
  int64_t min_bytes_to_write =
      max<int64_t>(0, dirty_unpinned_pages_.bytes() - target_dirty_bytes);
  if (lazy_flush
      && dirty_unpinned_pages_.bytes() + in_flight_write_pages_.bytes()
          <= target_dirty_bytes) {
    return Status::OK();
  }
  cleaning_pages_ = true;
  auto exit_trigger = MakeScopeExitTrigger([this, client_lock]() {
    DCheckHoldsLock(*client_lock);
    cleaning_pages_ = false;
    clean_pages_done_cv_.NotifyAll();
  });
  WriteDirtyPagesAsync(min_bytes_to_write);

  // One of the writes we initiated, or an earlier in-flight write may have hit an error.
  RETURN_IF_ERROR(write_status_);

  // Wait until enough writes have finished so that we can make the allocation without
  // violating the eviction policy. I.e. so that other clients can immediately get the
  // memory they're entitled to without waiting for this client's write to complete.
  DCHECK_GE(in_flight_write_pages_.bytes(), min_bytes_to_write) << DebugStringLocked();
  while (dirty_unpinned_pages_.bytes() + in_flight_write_pages_.bytes()
      > target_dirty_bytes) {
    SCOPED_TIMER(counters().write_wait_time);
    write_complete_cv_.Wait(*client_lock);
    RETURN_IF_ERROR(write_status_); // Check if error occurred while waiting.
  }
  return Status::OK();
}

void BufferPool::Client::WriteDirtyPagesAsync(int64_t min_bytes_to_write) {
  DCHECK_GE(min_bytes_to_write, 0) << DebugStringLocked();