/*
 * Licensed to the Apache Software Foundation (ASF) under one or more contributor license
 * agreements. See the NOTICE file distributed with this work for additional information regarding
 * copyright ownership. The ASF licenses this file to You under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance with the License. You may obtain a
 * copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software distributed under the License
 * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
 * or implied. See the License for the specific language governing permissions and limitations under
 * the License.
 */
package org.apache.geode.internal.cache.partitioned.rebalance.model;

import java.util.Collection;
import java.util.Comparator;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;
import java.util.SortedSet;
import java.util.TreeSet;

import org.apache.logging.log4j.Logger;

import org.apache.geode.annotations.Immutable;
import org.apache.geode.cache.partition.PartitionMemberInfo;
import org.apache.geode.distributed.internal.membership.InternalDistributedMember;
import org.apache.geode.internal.Assert;
import org.apache.geode.internal.cache.FixedPartitionAttributesImpl;
import org.apache.geode.internal.cache.PartitionedRegion;
import org.apache.geode.internal.cache.partitioned.InternalPartitionDetails;
import org.apache.geode.internal.cache.partitioned.OfflineMemberDetails;
import org.apache.geode.internal.cache.partitioned.PRLoad;
import org.apache.geode.internal.cache.partitioned.PartitionMemberInfoImpl;
import org.apache.geode.internal.cache.partitioned.rebalance.BucketOperator;
import org.apache.geode.internal.cache.persistence.PersistentMemberID;
import org.apache.geode.logging.internal.log4j.api.LogService;

/**
 * A model of the load on all of the members for a partitioned region. This model is used to find
 * the best members to create buckets on or move buckets or primaries too. All of the actual work of
 * creating a copy, moving a primary, etc. Is performed by the BucketOperator that is passed to the
 * constructor.
 *
 * To use, create a model and populate it using the addMember method. addMember takes a region
 * argument, to indicate which region the data is for. All of the regions added to a single model
 * are assumed to be colocated, and the model adds together the load from each of the individual
 * regions to balance all of the regions together.
 *
 * Reblancing operations are performed by repeatedly calling model.nextStep until it returns false.
 * Each call to nextStep should perform another operation. The model will make callbacks to the
 * BucketOperator you provide to the contructor perform the actual create or move.
 *
 * While creating redundant copies our moving buckets, this model tries to minimize the standard
 * deviation in the weighted loads for the members. The weighted load for the member is the sum of
 * the load for all of the buckets on the member divided by that members weight.
 *
 * This model is not threadsafe.
 *
 * @since GemFire 6.0
 */
@SuppressWarnings("synthetic-access")
public class PartitionedRegionLoadModel {
  private static final Logger logger = LogService.getLogger();

  /**
   * A comparator that is used to sort buckets in the order that we should satisfy redundancy - most
   * needy buckets first.
   */
  @Immutable
  private static final Comparator<Bucket> REDUNDANCY_COMPARATOR = (o1, o2) -> {
    // put the buckets with the lowest redundancy first
    int result = o1.getRedundancy() - o2.getRedundancy();
    if (result == 0) {
      // put the bucket with the largest load first. This should give us a
      // better chance of finding a place to put it
      result = Float.compare(o2.getLoad(), o1.getLoad());
    }
    if (result == 0) {
      // finally, just use the id so the comparator doesn't swallow buckets
      // with the same load
      result = o1.getId() - o2.getId();
    }

    return result;
  };

  private static final long MEGABYTES = 1024 * 1024;

  /**
   * A member to represent inconsistent data. For example, if two members think they are the primary
   * for a bucket, we will set the primary to invalid, so it won't be a candidate for rebalancing.
   */
  @Immutable
  public static final MemberRollup INVALID_MEMBER = new MemberRollup(null, null, false, false);

  private final BucketRollup[] buckets;

  /**
   * A map of all members that host this partitioned region
   */
  private final Map<InternalDistributedMember, MemberRollup> members = new HashMap<>();

  /**
   * The set of all regions that are colocated in this model.
   */
  private final Set<String> allColocatedRegions = new HashSet<>();

  /**
   * The list of buckets that have low redundancy
   */
  private SortedSet<BucketRollup> lowRedundancyBuckets = null;
  private SortedSet<BucketRollup> overRedundancyBuckets = null;
  private final Collection<Move> attemptedPrimaryMoves = new HashSet<>();
  private final Collection<Move> attemptedBucketMoves = new HashSet<>();
  private final Collection<Move> attemptedBucketCreations = new HashSet<>();
  private final Collection<Move> attemptedBucketRemoves = new HashSet<>();

  private final BucketOperator operator;
  private final int requiredRedundancy;

  /** The average primary load on a member */
  private float primaryAverage = -1;
  /** The average bucket load on a member */
  private float averageLoad = -1;
  /**
   * The minimum improvement in variance that we'll consider worth moving a primary
   */
  private double minPrimaryImprovement = -1;
  /**
   * The minimum improvement in variance that we'll consider worth moving a bucket
   */
  private double minImprovement = -1;

  private final AddressComparor addressComparor;

  private final Set<InternalDistributedMember> criticalMembers;

  private final PartitionedRegion partitionedRegion;

  /**
   * Create a new model
   *
   * @param operator the operator which performs the actual creates/moves for buckets
   * @param redundancyLevel The expected redundancy level for the region
   */
  public PartitionedRegionLoadModel(BucketOperator operator, int redundancyLevel, int numBuckets,
      AddressComparor addressComparor, Set<InternalDistributedMember> criticalMembers,
      PartitionedRegion region) {
    this.operator = operator;
    this.requiredRedundancy = redundancyLevel;
    this.buckets = new BucketRollup[numBuckets];
    this.addressComparor = addressComparor;
    this.criticalMembers = criticalMembers;
    this.partitionedRegion = region;
  }

  /**
   * Add a region to the model. All regions that are added are assumed to be colocated. The first
   * region added to the model should be the parent region. The parent region is expected to have at
   * least as many members as child regions; it may have more. If the parent has more members than
   * child regions those members will be considered invalid.
   *
   * @param memberDetailSet A set of details about each member.
   */
  public void addRegion(String region,
      Collection<? extends InternalPartitionDetails> memberDetailSet,
      OfflineMemberDetails offlineDetails, boolean enforceLocalMaxMemory) {
    this.allColocatedRegions.add(region);
    // build up a list of members and an array of buckets for this
    // region. Each bucket has a reference to all of the members
    // that host it and each member has a reference to all of the buckets
    // it hosts
    Map<InternalDistributedMember, Member> regionMember = new HashMap<>();
    Bucket[] regionBuckets = new Bucket[this.buckets.length];
    for (InternalPartitionDetails memberDetails : memberDetailSet) {
      InternalDistributedMember memberId =
          (InternalDistributedMember) memberDetails.getDistributedMember();

      boolean isCritical = criticalMembers.contains(memberId);
      Member member = new Member(addressComparor, memberId, memberDetails.getPRLoad().getWeight(),
          memberDetails.getConfiguredMaxMemory(), isCritical, enforceLocalMaxMemory);
      regionMember.put(memberId, member);

      PRLoad load = memberDetails.getPRLoad();
      for (int i = 0; i < regionBuckets.length; i++) {
        if (load.getReadLoad(i) > 0) {
          Bucket bucket = regionBuckets[i];
          if (bucket == null) {
            Set<PersistentMemberID> offlineMembers = offlineDetails.getOfflineMembers(i);
            bucket =
                new Bucket(i, load.getReadLoad(i), memberDetails.getBucketSize(i), offlineMembers);
            regionBuckets[i] = bucket;
          }
          bucket.addMember(member);
          if (load.getWriteLoad(i) > 0) {
            if (bucket.getPrimary() == null) {
              bucket.setPrimary(member, load.getWriteLoad(i));
            } else if (!bucket.getPrimary().equals(member)) {
              bucket.setPrimary(INVALID_MEMBER, 1);
            }
          }
        }
      }
    }

    // add each member for this region to a rollup of all colocated
    // regions
    for (Member member : regionMember.values()) {
      InternalDistributedMember memberId = member.getDistributedMember();
      MemberRollup memberSum = this.members.get(memberId);
      boolean isCritical = criticalMembers.contains(memberId);
      if (memberSum == null) {
        memberSum = new MemberRollup(addressComparor, memberId, isCritical, enforceLocalMaxMemory);
        this.members.put(memberId, memberSum);
      }

      memberSum.addColocatedMember(region, member);
    }

    // Now, add the region to the rollups of the colocated
    // regions and buckets
    for (int i = 0; i < this.buckets.length; i++) {
      if (regionBuckets[i] == null) {
        // do nothing, this bucket is not hosted for this region.
        // [sumedh] remove from buckets array too to be consistent since
        // this method will be invoked repeatedly for all colocated regions,
        // and then we may miss some colocated regions for a bucket leading
        // to all kinds of issues later
        this.buckets[i] = null;
        continue;
      }
      if (this.buckets[i] == null) {
        // If this is the first region we have seen that is hosting this bucket, create a bucket
        // rollup
        this.buckets[i] = new BucketRollup(i);
      }

      // Add all of the members hosting the bucket to the rollup
      for (Member member : regionBuckets[i].getMembersHosting()) {
        InternalDistributedMember memberId = member.getDistributedMember();
        this.buckets[i].addMember(this.members.get(memberId));
      }

      // set the primary for the rollup
      if (regionBuckets[i].getPrimary() != null) {
        if (this.buckets[i].getPrimary() == null) {
          InternalDistributedMember memberId = regionBuckets[i].getPrimary().getDistributedMember();
          this.buckets[i].setPrimary(this.members.get(memberId), 0);
        } else {
          if (!(this.buckets[i].getPrimary() == INVALID_MEMBER)) {
            if (!this.buckets[i].getPrimary().getDistributedMember()
                .equals(regionBuckets[i].getPrimary().getDistributedMember())) {
              if (logger.isDebugEnabled()) {
                logger.debug(
                    "PartitionedRegionLoadModel - Setting bucket {} to INVALID because it is the primary on two members.This could just be a race in the collocation of data. member1={} member2={}",
                    this.buckets[i], this.buckets[i].getPrimary(), regionBuckets[i].getPrimary());
              }
              this.buckets[i].setPrimary(INVALID_MEMBER, 0);
            }
          }
        }
      }
      this.buckets[i].addColocatedBucket(region, regionBuckets[i]);
    }

    // TODO rebalance - there is a possibility of adding members
    // back here, which I don't like. I think maybe all of the regions should be in the
    // constructor for the load model, and then when the constructor is done
    // we can do with validation.
    // If any members don't have this new region, remove them.
    for (Iterator<Entry<InternalDistributedMember, MemberRollup>> itr =
        members.entrySet().iterator(); itr.hasNext();) {
      MemberRollup memberRollup = itr.next().getValue();
      if (!memberRollup.getColocatedMembers().keySet().equals(this.allColocatedRegions)) {
        itr.remove();
        if (logger.isDebugEnabled()) {
          logger.debug(
              "PartitionedRegionLoadModel - removing member {} from the consideration because it doesn't have all of the colocated regions. Expected={}, was={}",
              memberRollup, allColocatedRegions, memberRollup.getColocatedMembers());
        }
        // This state should never happen
        if (!memberRollup.getBuckets().isEmpty()) {
          logger.warn(
              "PartitionedRegionLoadModel - member {} has incomplete colocation, but it has buckets for some regions. Should have colocated regions {} but had {} and contains buckets {}",
              new Object[] {memberRollup, this.allColocatedRegions,
                  memberRollup.getColocatedMembers().keySet(), memberRollup.getBuckets()});
        }
        for (Bucket bucket : new HashSet<Bucket>(memberRollup.getBuckets())) {
          bucket.removeMember(memberRollup);
        }
      }
    }
  }

  public void initialize() {
    resetAverages();
    initOverRedundancyBuckets();
    initLowRedundancyBuckets();
  }

  public SortedSet<BucketRollup> getLowRedundancyBuckets() {
    return lowRedundancyBuckets;
  }

  public SortedSet<BucketRollup> getOverRedundancyBuckets() {
    return overRedundancyBuckets;
  }

  public boolean enforceUniqueZones() {
    return addressComparor.enforceUniqueZones();
  }

  public void ignoreLowRedundancyBucket(BucketRollup first) {
    this.lowRedundancyBuckets.remove(first);
  }

  public void ignoreOverRedundancyBucket(BucketRollup first) {
    this.overRedundancyBuckets.remove(first);
  }

  public MemberRollup getMember(InternalDistributedMember target) {
    return members.get(target);
  }

  public BucketRollup[] getBuckets() {
    return buckets;
  }

  public String getName() {
    return getPartitionedRegion().getFullPath();
  }

  public PartitionedRegion getPartitionedRegion() {
    // TODO - this model really should not have
    // a reference to the partitioned region object.
    // The fixed PR code currently depends on this
    // partitioned region object and needs
    // refactoring.
    return partitionedRegion;
  }

  private Map<String, Long> getColocatedRegionSizes(BucketRollup bucket) {
    Map<String, Long> colocatedRegionSizes = new HashMap<>();

    for (Map.Entry<String, Bucket> entry : bucket.getColocatedBuckets().entrySet()) {
      colocatedRegionSizes.put(entry.getKey(), entry.getValue().getBytes());
    }
    return colocatedRegionSizes;
  }

  /**
   * Trigger the creation of a redundant bucket, potentially asynchronously.
   *
   * This method will find the best node to create a redundant bucket and invoke the bucket operator
   * to create a bucket on that node. Because the bucket operator is asynchronous, the bucket may
   * not be created immediately, but the model will be updated regardless. Invoke
   * {@link #waitForOperations()} to wait for those operations to actually complete
   */
  public void createRedundantBucket(final BucketRollup bucket, final Member targetMember) {
    Map<String, Long> colocatedRegionSizes = getColocatedRegionSizes(bucket);
    final Move move = new Move(null, targetMember, bucket);

    this.lowRedundancyBuckets.remove(bucket);
    bucket.addMember(targetMember);
    // put the bucket back into the list if we still need to satisfy redundancy for
    // this bucket
    if (bucket.getRedundancy() < this.requiredRedundancy) {
      this.lowRedundancyBuckets.add(bucket);
    }
    resetAverages();

    this.operator.createRedundantBucket(targetMember.getMemberId(), bucket.getId(),
        colocatedRegionSizes, new BucketOperator.Completion() {
          @Override
          public void onSuccess() {}

          @Override
          public void onFailure() {
            // If the bucket creation failed, we need to undo the changes
            // we made to the model
            attemptedBucketCreations.add(move);
            // remove the bucket from lowRedundancyBuckets before mutating the state
            lowRedundancyBuckets.remove(bucket);
            bucket.removeMember(targetMember);
            if (bucket.getRedundancy() < requiredRedundancy) {
              lowRedundancyBuckets.add(bucket);
            }
            resetAverages();
          }
        });
  }

  public void remoteOverRedundancyBucket(BucketRollup bucket, Member targetMember) {
    Move bestMove = new Move(null, targetMember, bucket);
    Map<String, Long> colocatedRegionSizes = getColocatedRegionSizes(bucket);

    if (!this.operator.removeBucket(targetMember.getMemberId(), bucket.getId(),
        colocatedRegionSizes)) {
      this.attemptedBucketRemoves.add(bestMove);
    } else {
      this.overRedundancyBuckets.remove(bucket);
      bucket.removeMember(targetMember);
      // put the bucket back into the list if we still need to satisfy redundancy for
      // this bucket
      if (bucket.getOnlineRedundancy() > this.requiredRedundancy) {
        this.overRedundancyBuckets.add(bucket);
      }
      resetAverages();
    }
  }

  private void initLowRedundancyBuckets() {
    this.lowRedundancyBuckets = new TreeSet<>(REDUNDANCY_COMPARATOR);
    for (BucketRollup b : this.buckets) {
      if (b != null && b.getRedundancy() >= 0 && b.getRedundancy() < this.requiredRedundancy) {
        this.lowRedundancyBuckets.add(b);
      }
    }
  }

  private void initOverRedundancyBuckets() {
    this.overRedundancyBuckets = new TreeSet<>(REDUNDANCY_COMPARATOR);
    for (BucketRollup b : this.buckets) {
      if (b != null && b.getOnlineRedundancy() > this.requiredRedundancy) {
        this.overRedundancyBuckets.add(b);
      }
    }
  }

  /**
   * Find the best member to put a new bucket on.
   *
   * @param bucket the bucket we want to create
   * @param checkIPAddress true if we should only consider members that do not have the same IP
   *        Address as a member that already hosts the bucket
   */
  public Move findBestTarget(Bucket bucket, boolean checkIPAddress) {
    float leastCost = Float.MAX_VALUE;
    Move bestMove = null;

    for (Member member : this.members.values()) {
      if (member.willAcceptBucket(bucket, null, checkIPAddress).willAccept()) {
        float cost = (member.getTotalLoad() + bucket.getLoad()) / member.getWeight();
        if (cost < leastCost) {
          Move move = new Move(null, member, bucket);
          if (!this.attemptedBucketCreations.contains(move)) {
            leastCost = cost;
            bestMove = move;
          }
        }
      }
    }
    return bestMove;
  }

  /**
   * Find the best member to remove a bucket from
   *
   * @param bucket the bucket we want to create
   */
  public Move findBestRemove(Bucket bucket) {
    float mostLoaded = Float.MIN_VALUE;
    Move bestMove = null;

    for (Member member : bucket.getMembersHosting()) {
      float newLoad = (member.getTotalLoad() - bucket.getLoad()) / member.getWeight();
      if (newLoad > mostLoaded && !member.equals(bucket.getPrimary())) {
        Move move = new Move(null, member, bucket);
        if (!this.attemptedBucketRemoves.contains(move)) {
          mostLoaded = newLoad;
          bestMove = move;
        }
      }
    }
    return bestMove;
  }

  public Move findBestTargetForFPR(Bucket bucket, boolean checkIPAddress) {
    InternalDistributedMember targetMemberID = null;
    Member targetMember = null;
    List<FixedPartitionAttributesImpl> fpas =
        this.partitionedRegion.getFixedPartitionAttributesImpl();

    if (fpas != null) {
      for (FixedPartitionAttributesImpl fpaImpl : fpas) {
        if (fpaImpl.hasBucket(bucket.getId())) {
          targetMemberID =
              this.partitionedRegion.getDistributionManager().getDistributionManagerId();
          if (this.members.containsKey(targetMemberID)) {
            targetMember = this.members.get(targetMemberID);
            if (targetMember.willAcceptBucket(bucket, null, checkIPAddress).willAccept()) {
              // We should have just one move for creating
              // all the buckets for a FPR on this node.
              return new Move(null, targetMember, bucket);
            }
          }
        }
      }
    }

    return null;
  }

  public boolean movePrimary(Move bestMove) {
    Member bestSource = bestMove.getSource();
    Member bestTarget = bestMove.getTarget();
    Bucket bestBucket = bestMove.getBucket();
    boolean successfulMove = this.operator.movePrimary(bestSource.getDistributedMember(),
        bestTarget.getDistributedMember(), bestBucket.getId());

    if (successfulMove) {
      bestBucket.setPrimary(bestTarget, bestBucket.getPrimaryLoad());
    }

    boolean entryAdded = this.attemptedPrimaryMoves.add(bestMove);
    Assert.assertTrue(entryAdded,
        "PartitionedRegionLoadModel.movePrimarys - excluded set is not growing, so we probably would have an infinite loop here");

    return successfulMove;
  }

  public Move findBestPrimaryMove() {
    Move bestMove = null;
    double bestImprovement = 0;
    for (Member source : this.members.values()) {
      for (Bucket bucket : source.getPrimaryBuckets()) {
        for (Member target : bucket.getMembersHosting()) {
          if (source.equals(target)) {
            continue;
          }
          double improvement =
              improvement(source.getPrimaryLoad(), source.getWeight(), target.getPrimaryLoad(),
                  target.getWeight(), bucket.getPrimaryLoad(), getPrimaryAverage());
          if (improvement > bestImprovement && improvement > getMinPrimaryImprovement()) {
            Move move = new Move(source, target, bucket);
            if (!this.attemptedPrimaryMoves.contains(move)) {
              bestImprovement = improvement;
              bestMove = move;
            }
          }
        }
      }
    }
    return bestMove;
  }

  /**
   * Calculate the target weighted number of primaries on each node.
   */
  private float getPrimaryAverage() {
    if (this.primaryAverage == -1) {
      float totalWeight = 0;
      float totalPrimaryCount = 0;
      for (Member member : this.members.values()) {
        totalPrimaryCount += member.getPrimaryLoad();
        totalWeight += member.getWeight();
      }

      this.primaryAverage = totalPrimaryCount / totalWeight;
    }

    return this.primaryAverage;
  }

  /**
   * Calculate the target weighted amount of data on each node.
   */
  private float getAverageLoad() {
    if (this.averageLoad == -1) {
      float totalWeight = 0;
      float totalLoad = 0;
      for (Member member : this.members.values()) {
        totalLoad += member.getTotalLoad();
        totalWeight += member.getWeight();
      }

      this.averageLoad = totalLoad / totalWeight;
    }

    return this.averageLoad;
  }

  /**
   * Calculate the minimum improvement in variance that will we consider worth while. Currently this
   * is calculated as the improvement in variance that would occur by removing the smallest bucket
   * from the member with the largest weight.
   */
  private double getMinPrimaryImprovement() {
    if ((this.minPrimaryImprovement + 1.0) < .0000001) { // i.e. == -1
      float largestWeight = 0;
      float smallestBucket = 0;
      for (Member member : this.members.values()) {
        if (member.getWeight() > largestWeight) {
          largestWeight = member.getWeight();
        }
        for (Bucket bucket : member.getPrimaryBuckets()) {
          if (bucket.getPrimaryLoad() < smallestBucket || smallestBucket == 0) {
            smallestBucket = bucket.getPrimaryLoad();
          }
        }
      }
      double before = variance(getPrimaryAverage() * largestWeight + smallestBucket, largestWeight,
          getPrimaryAverage());
      double after =
          variance(getPrimaryAverage() * largestWeight, largestWeight, getPrimaryAverage());
      this.minPrimaryImprovement = (before - after) / smallestBucket;
    }
    return this.minPrimaryImprovement;
  }

  /**
   * Calculate the minimum improvement in variance that will we consider worth while. Currently this
   * is calculated as the improvement in variance that would occur by removing the smallest bucket
   * from the member with the largest weight.
   */
  private double getMinImprovement() {
    if ((this.minImprovement + 1.0) < .0000001) { // i.e. == -1
      float largestWeight = 0;
      float smallestBucket = 0;
      for (Member member : this.members.values()) {
        if (member.getWeight() > largestWeight) {
          largestWeight = member.getWeight();
        }
        // find the smallest bucket, ignoring empty buckets.
        for (Bucket bucket : member.getBuckets()) {
          if (smallestBucket == 0 || (bucket.getLoad() < smallestBucket && bucket.getBytes() > 0)) {
            smallestBucket = bucket.getLoad();
          }
        }
      }
      double before = variance(getAverageLoad() * largestWeight + smallestBucket, largestWeight,
          getAverageLoad());
      double after = variance(getAverageLoad() * largestWeight, largestWeight, getAverageLoad());
      this.minImprovement = (before - after) / smallestBucket;
    }
    return this.minImprovement;
  }

  private void resetAverages() {
    this.primaryAverage = -1;
    this.averageLoad = -1;
    this.minPrimaryImprovement = -1;
    this.minImprovement = -1;
  }

  /**
   * Calculate how much the variance in load will decrease for a given move.
   *
   * @param sLoad the current load on the source member
   * @param sWeight the weight of the source member
   * @param tLoad the current load on the target member
   * @param tWeight the weight of the target member
   * @param bucketSize the size of the bucket we're considering moving
   * @param average the target weighted load for all members.
   * @return the change in variance that would occur by making this move. Essentially
   *         variance_before - variance_after, so a positive change is a means the variance is
   *         decreasing.
   */
  private double improvement(float sLoad, float sWeight, float tLoad, float tWeight,
      float bucketSize, float average) {

    double vSourceBefore = variance(sLoad, sWeight, average);
    double vSourceAfter = variance(sLoad - bucketSize, sWeight, average);
    double vTargetBefore = variance(tLoad, tWeight, average);
    double vTargetAfter = variance(tLoad + bucketSize, tWeight, average);

    double improvement = vSourceBefore - vSourceAfter + vTargetBefore - vTargetAfter;
    return improvement / bucketSize;
  }

  private double variance(double load, double weight, double average) {
    double deviation = (load / weight - average);
    return deviation * deviation;
  }

  public Move findBestBucketMove() {
    Move bestMove = null;
    double bestImprovement = 0;
    for (Member source : this.members.values()) {
      for (Bucket bucket : source.getBuckets()) {
        for (Member target : this.members.values()) {
          if (bucket.getMembersHosting().contains(target)) {
            continue;
          }
          if (!target.willAcceptBucket(bucket, source, true).willAccept()) {
            continue;
          }
          double improvement = improvement(source.getTotalLoad(), source.getWeight(),
              target.getTotalLoad(), target.getWeight(), bucket.getLoad(), getAverageLoad());
          if (improvement > bestImprovement && improvement > getMinImprovement()) {
            Move move = new Move(source, target, bucket);
            if (!this.attemptedBucketMoves.contains(move)) {
              bestImprovement = improvement;
              bestMove = move;
            }
          }
        }
      }
    }
    return bestMove;
  }

  public boolean moveBucket(Move bestMove) {
    Member bestSource = bestMove.getSource();
    Member bestTarget = bestMove.getTarget();
    BucketRollup bestBucket = (BucketRollup) bestMove.getBucket();

    Map<String, Long> colocatedRegionSizes = getColocatedRegionSizes(bestBucket);

    boolean successfulMove = this.operator.moveBucket(bestSource.getDistributedMember(),
        bestTarget.getDistributedMember(), bestBucket.getId(), colocatedRegionSizes);

    if (successfulMove) {
      bestBucket.addMember(bestTarget);
      if (bestSource.equals(bestBucket.getPrimary())) {
        bestBucket.setPrimary(bestTarget, bestBucket.getPrimaryLoad());
      }
      bestBucket.removeMember(bestSource);
    }

    boolean entryAdded = this.attemptedBucketMoves.add(bestMove);
    Assert.assertTrue(entryAdded,
        "PartitionedRegionLoadModel.moveBuckets - excluded set is not growing, so we probably would have an infinite loop here");

    return successfulMove;
  }

  /**
   * Return a snapshot of what the partitioned member details look like.
   *
   * @return a set of partitioned member details.
   */
  public Set<PartitionMemberInfo> getPartitionedMemberDetails(String region) {
    TreeSet<PartitionMemberInfo> result = new TreeSet<>();
    for (MemberRollup member : this.members.values()) {
      Member colocatedMember = member.getColocatedMember(region);
      if (colocatedMember != null) {
        result.add(new PartitionMemberInfoImpl(colocatedMember.getDistributedMember(),
            colocatedMember.getConfiguredMaxMemory(), colocatedMember.getSize(),
            colocatedMember.getBucketCount(), colocatedMember.getPrimaryCount()));
      }
    }
    return result;
  }

  /**
   * For testing only, calculate the total variance of the members
   */
  public float getVarianceForTest() {
    float variance = 0;

    for (Member member : this.members.values()) {
      variance += variance(member.getTotalLoad(), member.getWeight(), getAverageLoad());
    }

    return variance;
  }

  /**
   * For testing only, calculate the total variance of the members
   */
  public float getPrimaryVarianceForTest() {
    float variance = 0;

    for (Member member : this.members.values()) {
      variance += variance(member.getPrimaryLoad(), member.getWeight(), getPrimaryAverage());
    }

    return variance;
  }

  /**
   * Wait for the bucket operator to complete any pending asynchronous operations.
   */
  public void waitForOperations() {
    operator.waitForOperations();
  }

  @Override
  public String toString() {
    StringBuilder result = new StringBuilder();
    TreeSet<Bucket> allBucketIds = new TreeSet<>(Comparator.comparingInt(Bucket::getId));
    if (this.members.isEmpty()) {
      return "";
    }
    int longestMemberId = 0;
    for (Member member : this.members.values()) {
      allBucketIds.addAll(member.getBuckets());
      int memberIdLength = member.getDistributedMember().toString().length();
      if (longestMemberId < memberIdLength) {
        longestMemberId = memberIdLength;
      }
    }
    result
        .append(String.format("%" + longestMemberId + "s primaries size(MB)  max(MB)", "MemberId"));
    for (Bucket bucket : allBucketIds) {
      result.append(String.format("%4s", bucket.getId()));
    }
    for (Member member : this.members.values()) {
      result.append(String.format("\n%" + longestMemberId + "s %9.0f %8.2f %8.2f",
          member.getDistributedMember(), member.getPrimaryLoad(),
          member.getSize() / (float) MEGABYTES,
          member.getConfiguredMaxMemory() / (float) MEGABYTES));
      for (Bucket bucket : allBucketIds) {
        char symbol;
        if (member.getPrimaryBuckets().contains(bucket)) {
          symbol = 'P';
        } else if (member.getBuckets().contains(bucket)) {
          symbol = 'R';
        } else {
          symbol = 'X';
        }
        result.append("   ").append(symbol);
      }
    }

    result.append(String.format("\n%" + longestMemberId + "s                            ",
        "#offline"));
    for (Bucket bucket : allBucketIds) {
      result.append(String.format("%4s", bucket.getOfflineMembers().size()));
    }

    return result.toString();
  }

}