Method TransferReservationTo

be/src/runtime/bufferpool/reservation-tracker.cc:284–369

}

bool ReservationTracker::TransferReservationTo(ReservationTracker* other, int64_t bytes) {
  if (other == this) return true;
  // Find the path to the root from both. The root is guaranteed to be a common ancestor.
  vector<ReservationTracker*> path_to_common = FindPathToRoot();
  vector<ReservationTracker*> other_path_to_common = other->FindPathToRoot();
  DCHECK_EQ(path_to_common.back(), other_path_to_common.back());
  ReservationTracker* common_ancestor = path_to_common.back();
  // Remove any common ancestors - they do not need to be updated for this transfer.
  while (!path_to_common.empty() && !other_path_to_common.empty()
      && path_to_common.back() == other_path_to_common.back()) {
    common_ancestor = path_to_common.back();
    path_to_common.pop_back();
    other_path_to_common.pop_back();
  }

  // At this point, we have three cases:
  // 1. 'common_ancestor' == 'other'. 'other_path_to_common' is empty because 'other' is
  //    the lowest common ancestor. To transfer, we decrease the reservation on the
  //    trackers under 'other', down to 'this'.
  // 2. 'common_ancestor' == 'this'. 'path_to_common' is empty because 'this' is the
  //    lowest common ancestor. To transfer, we increase the reservation on the trackers
  //    under 'this', down to 'other'.
  // 3. Neither is an ancestor of the other. Both 'other_path_to_common' and
  //    'path_to_common' are non-empty. We increase the reservation on trackers from
  //    'other' up to one below the common ancestor (checking limits as needed) and if
  //    successful, decrease reservations on trackers from 'this' up to one below the
  //    common ancestor.

  // Lock all of the trackers so we can do the update atomically. Need to be careful to
  // lock subtrees in the correct order.
  vector<unique_lock<SpinLock>> locks;
  bool lock_first = path_to_common.empty() || other_path_to_common.empty()
      || lock_sibling_subtree_first(path_to_common.back(), other_path_to_common.back());
  if (lock_first) {
    for (ReservationTracker* tracker : path_to_common) locks.emplace_back(tracker->lock_);
  }
  for (ReservationTracker* tracker : other_path_to_common) {
    locks.emplace_back(tracker->lock_);
  }
  if (!lock_first) {
    for (ReservationTracker* tracker : path_to_common) locks.emplace_back(tracker->lock_);
  }

  // Check reservation limits will not be violated before applying any updates.
  for (ReservationTracker* tracker : other_path_to_common) {
    if (tracker->reservation_.Load() + bytes > tracker->reservation_limit_.Load()) {
      return false;
    }
  }

  // Do the updates now that we have checked the limits. We're holding all the locks
  // so this is all atomic.
  for (ReservationTracker* tracker : other_path_to_common) {
    tracker->UpdateReservation(bytes);
    // We don't handle MemTrackers with limit in this function - this should always
    // succeed.
    DCHECK(tracker->mem_tracker_ == nullptr || !tracker->mem_tracker_->has_limit());
    bool success = tracker->TryConsumeFromMemTracker(bytes, MemLimit::HARD);