/*
 * 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;

import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
import java.util.AbstractSet;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.NoSuchElementException;
import java.util.Objects;
import java.util.Queue;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.Semaphore;
import java.util.concurrent.atomic.AtomicLong;
import java.util.function.Predicate;

import org.apache.logging.log4j.Logger;

import org.apache.geode.DataSerializable;
import org.apache.geode.DataSerializer;
import org.apache.geode.annotations.Immutable;
import org.apache.geode.annotations.VisibleForTesting;
import org.apache.geode.cache.InterestPolicy;
import org.apache.geode.cache.LowMemoryException;
import org.apache.geode.cache.Region;
import org.apache.geode.distributed.DistributedMember;
import org.apache.geode.distributed.internal.DistributionConfig;
import org.apache.geode.distributed.internal.ProfileListener;
import org.apache.geode.distributed.internal.membership.InternalDistributedMember;
import org.apache.geode.internal.InternalDataSerializer;
import org.apache.geode.internal.cache.BucketAdvisor;
import org.apache.geode.internal.cache.BucketAdvisor.BucketProfile;
import org.apache.geode.internal.cache.BucketAdvisor.ServerBucketProfile;
import org.apache.geode.internal.cache.BucketPersistenceAdvisor;
import org.apache.geode.internal.cache.BucketRegion;
import org.apache.geode.internal.cache.BucketServerLocation66;
import org.apache.geode.internal.cache.CacheDistributionAdvisor;
import org.apache.geode.internal.cache.FixedPartitionAttributesImpl;
import org.apache.geode.internal.cache.InternalRegionArguments;
import org.apache.geode.internal.cache.PartitionedRegion;
import org.apache.geode.internal.cache.PartitionedRegionStats;
import org.apache.geode.internal.cache.ProxyBucketRegion;
import org.apache.geode.internal.cache.control.MemoryThresholds;
import org.apache.geode.internal.cache.control.ResourceAdvisor;
import org.apache.geode.internal.logging.log4j.LogMarker;
import org.apache.geode.internal.serialization.DeserializationContext;
import org.apache.geode.internal.serialization.SerializationContext;
import org.apache.geode.logging.internal.log4j.api.LogService;
import org.apache.geode.util.internal.GeodeGlossary;

public class RegionAdvisor extends CacheDistributionAdvisor {
  private static final Logger logger = LogService.getLogger();

  /**
   * Number of threads allowed to concurrently volunteer for bucket primary.
   */
  public static final short VOLUNTEERING_THREAD_COUNT = Integer
      .getInteger(GeodeGlossary.GEMFIRE_PREFIX + "RegionAdvisor.volunteeringThreadCount", 1)
      .shortValue();

  /**
   * Non-thread safe queue for volunteering for primary bucket. Each BucketAdvisor for this PR uses
   * this queue. The thread that uses this queue is a waiting pool thread. Any thread using this
   * queue must synchronize on this queue.
   */
  private final Queue<Runnable> volunteeringQueue = new ConcurrentLinkedQueue<>();

  /**
   * Semaphore with {@link #VOLUNTEERING_THREAD_COUNT} number of permits to control number of
   * threads volunteering for bucket primaries.
   */
  private final Semaphore volunteeringSemaphore = new Semaphore(VOLUNTEERING_THREAD_COUNT);

  private volatile int lastActiveProfiles = 0;
  private volatile int numDataStores = 0;
  protected volatile ProxyBucketRegion[] buckets;

  @VisibleForTesting
  protected Queue<QueuedBucketProfile> preInitQueue;
  private final Object preInitQueueMonitor = new Object();

  private final ConcurrentHashMap<Integer, Set<ServerBucketProfile>> clientBucketProfilesMap;

  @VisibleForTesting
  protected RegionAdvisor(PartitionedRegion region) {
    super(region);
    synchronized (preInitQueueMonitor) {
      preInitQueue = new ConcurrentLinkedQueue<>();
    }
    clientBucketProfilesMap = new ConcurrentHashMap<>();
  }

  public static RegionAdvisor createRegionAdvisor(PartitionedRegion region) {
    RegionAdvisor advisor = new RegionAdvisor(region);
    advisor.initialize();
    return advisor;
  }

  public PartitionedRegionStats getPartitionedRegionStats() {
    return getPartitionedRegion().getPrStats();
  }

  public synchronized void initializeRegionAdvisor() {
    if (buckets != null) {
      return;
    }
    PartitionedRegion p = getPartitionedRegion();
    int numBuckets = p.getAttributes().getPartitionAttributes().getTotalNumBuckets();
    ProxyBucketRegion[] bucs = new ProxyBucketRegion[numBuckets];

    InternalRegionArguments args = new InternalRegionArguments();
    args.setPartitionedRegionAdvisor(this);
    for (int i = 0; i < bucs.length; i++) {
      bucs[i] = new ProxyBucketRegion(i, p, args);
      bucs[i].initialize();
    }
    buckets = bucs;
  }

  /**
   * Process those profiles which were received during the initialization period. It is safe to
   * process these profiles potentially out of order due to the profiles version which is
   * established on the sender.
   */
  public void processProfilesQueuedDuringInitialization() {
    synchronized (preInitQueueMonitor) {
      Iterator pi = preInitQueue.iterator();
      boolean finishedInitQueue = false;
      try {
        while (pi.hasNext()) {
          Object o = pi.next();
          QueuedBucketProfile qbp = (QueuedBucketProfile) o;
          if (!qbp.isRemoval) {
            if (logger.isTraceEnabled(LogMarker.DISTRIBUTION_ADVISOR_VERBOSE)) {
              logger.trace(LogMarker.DISTRIBUTION_ADVISOR_VERBOSE,
                  "applying queued profile addition for bucket {}", qbp.bucketId);
            }
            getBucket(qbp.bucketId).getBucketAdvisor().putProfile(qbp.bucketProfile);
          } else if (qbp.memberDeparted
              || !getDistributionManager().isCurrentMember(qbp.memberId)) {
            boolean crashed;
            if (qbp.memberDeparted) {
              crashed = qbp.crashed;
            } else {
              // TODO not necessarily accurate, but how important is this?
              crashed = !stillInView(qbp.memberId);
            }
            if (logger.isTraceEnabled(LogMarker.DISTRIBUTION_ADVISOR_VERBOSE)) {
              logger.trace(LogMarker.DISTRIBUTION_ADVISOR_VERBOSE,
                  "applying queued member departure for all buckets for {}", qbp.memberId);
            }
            for (ProxyBucketRegion bucket : buckets) {
              BucketAdvisor ba = bucket.getBucketAdvisor();
              ba.removeId(qbp.memberId, crashed, qbp.destroyed, qbp.fromMembershipListener);
            } // for
          } else { // apply removal for member still in the view
            if (logger.isTraceEnabled(LogMarker.DISTRIBUTION_ADVISOR_VERBOSE)) {
              logger.trace(LogMarker.DISTRIBUTION_ADVISOR_VERBOSE,
                  "applying queued profile removal for all buckets for {}", qbp.memberId);
            }
            if (qbp.serials.length == 1 && qbp.serials[0] != ILLEGAL_SERIAL) {
              buckets[qbp.bucketId].getBucketAdvisor().removeIdWithSerial(qbp.memberId,
                  qbp.serials[0], qbp.destroyed);
            } else {
              for (int i = 0; i < buckets.length; i++) {
                BucketAdvisor ba = buckets[i].getBucketAdvisor();
                int serial = qbp.serials[i];
                if (serial != ILLEGAL_SERIAL) {
                  ba.removeIdWithSerial(qbp.memberId, serial, qbp.destroyed);
                }
              } // for
            }
          } // apply removal for member still in the view
        } // while
        finishedInitQueue = true;
      } finally {
        preInitQueue = null; // prevent further additions to the queue
        if (!finishedInitQueue && !getAdvisee().getCancelCriterion().isCancelInProgress()) {
          logger.error("Failed to process all queued BucketProfiles for {}",
              getAdvisee());
        }
      }
    }
  }

  @Override
  protected Profile instantiateProfile(InternalDistributedMember memberId, int version) {
    return new PartitionProfile(memberId, version);
  }

  /**
   * Returns the {@link #volunteeringQueue} used to queue primary volunteering tasks by this PR's
   * BucketAdvisors.
   *
   * @return the volunteering queue for use by this PR's BucketAdvisors
   */
  public Queue<Runnable> getVolunteeringQueue() {
    return volunteeringQueue;
  }

  /**
   * Returns the {@link #volunteeringSemaphore} for controlling the number of threads that this PR's
   * BucketAdvisors are allowed to use for volunteering to be primary.
   *
   * @return the semaphore for controlling number of volunteering threads
   */
  public Semaphore getVolunteeringSemaphore() {
    return volunteeringSemaphore;
  }

  /**
   * Returns an unmodifiable map of bucket IDs to locations hosting the bucket.
   */
  public Map<Integer, List<BucketServerLocation66>> getAllClientBucketProfiles() {
    Map<Integer, List<BucketServerLocation66>> bucketToServerLocations = new HashMap<>();
    for (Integer bucketId : clientBucketProfilesMap.keySet()) {
      ArrayList<BucketServerLocation66> clientBucketProfiles = new ArrayList<>();
      for (BucketProfile profile : clientBucketProfilesMap.get(bucketId)) {
        if (profile.isHosting) {
          ServerBucketProfile cProfile = (ServerBucketProfile) profile;
          Set<BucketServerLocation66> bucketServerLocations = cProfile.getBucketServerLocations();
          // Either we can make BucketServeLocation having ServerLocation with them
          // Or we can create bucketServerLocation as it is by iterating over the set of servers
          clientBucketProfiles.addAll(bucketServerLocations);
        }
      }
      bucketToServerLocations.put(bucketId, clientBucketProfiles);
    }

    if (getPartitionedRegion().isDataStore()) {
      for (Integer bucketId : getPartitionedRegion().getDataStore().getAllLocalBucketIds()) {
        BucketProfile profile = getBucketAdvisor(bucketId).getLocalProfile();

        if (logger.isDebugEnabled()) {
          logger.debug("The local profile is : {}", profile);
        }

        if (profile != null) {
          List<BucketServerLocation66> clientBucketProfiles =
              bucketToServerLocations.computeIfAbsent(bucketId, k -> new ArrayList<>());
          if ((profile instanceof ServerBucketProfile) && profile.isHosting) {
            ServerBucketProfile cProfile = (ServerBucketProfile) profile;
            Set<BucketServerLocation66> bucketServerLocations = cProfile.getBucketServerLocations();
            // Either we can make BucketServeLocation having ServerLocation with
            // them
            // Or we can create bucketServerLocation as it is by iterating over
            // the set of servers
            clientBucketProfiles.removeAll(bucketServerLocations);
            clientBucketProfiles.addAll(bucketServerLocations);
          }
        }
      }
    }
    return bucketToServerLocations;
  }

  @VisibleForTesting
  public ConcurrentHashMap<Integer, Set<ServerBucketProfile>> getAllClientBucketProfilesTest() {
    ConcurrentHashMap<Integer, Set<ServerBucketProfile>> map = new ConcurrentHashMap<>();
    Map<Integer, List<BucketServerLocation66>> testMap =
        new HashMap<>(getAllClientBucketProfiles());
    for (Integer bucketId : testMap.keySet()) {
      Set<ServerBucketProfile> parr = new HashSet<>(clientBucketProfilesMap.get(bucketId));
      map.put(bucketId, parr);
    }

    if (getPartitionedRegion().isDataStore()) {
      for (Integer bucketId : getPartitionedRegion().getDataStore().getAllLocalBucketIds()) {
        BucketProfile profile = getBucketAdvisor(bucketId).getLocalProfile();
        if ((profile instanceof ServerBucketProfile) && profile.isHosting) {
          map.get(bucketId).add((ServerBucketProfile) profile);
        }
      }
    }
    if (logger.isDebugEnabled()) {
      logger.debug("This maps is sksk {} and size is {}", map, map.keySet().size());
    }
    return map;
  }


  public Set<ServerBucketProfile> getClientBucketProfiles(Integer bucketId) {
    return clientBucketProfilesMap.get(bucketId);
  }

  public void setClientBucketProfiles(Integer bucketId, Set<ServerBucketProfile> profiles) {
    clientBucketProfilesMap.put(bucketId, Collections.unmodifiableSet(profiles));
  }

  /**
   * Close the bucket advisors, releasing any locks for primary buckets
   */
  public void closeBucketAdvisors() {
    if (buckets != null) {
      for (ProxyBucketRegion pbr : buckets) {
        pbr.close();
      }
    }
  }

  /**
   * Close the adviser and all bucket advisors.
   */
  @Override
  public void close() {
    super.close();
    if (buckets != null) {
      for (ProxyBucketRegion bucket : buckets) {
        bucket.close();
      }
    }
  }

  @Override
  public boolean removeId(ProfileId memberId, boolean crashed, boolean regionDestroyed,
      boolean fromMembershipListener) {
    // It's important that we remove member from the bucket advisors first
    // Calling super.removeId triggers redundancy satisfaction, so the bucket
    // advisors must have up to data information at that point.
    boolean removeBuckets = true;
    synchronized (preInitQueueMonitor) {
      if (preInitQueue != null) {
        // Queue profile during pre-initialization
        QueuedBucketProfile qbf = new QueuedBucketProfile((InternalDistributedMember) memberId,
            crashed, regionDestroyed, fromMembershipListener);
        preInitQueue.add(qbf);
        removeBuckets = false;
      }
    } // synchronized
    if (removeBuckets && buckets != null) {
      for (ProxyBucketRegion pbr : buckets) {
        BucketAdvisor pbra = pbr.getBucketAdvisor();
        boolean shouldSync = false;
        Profile profile = null;
        InternalDistributedMember mbr = null;
        if (memberId instanceof InternalDistributedMember) {
          mbr = (InternalDistributedMember) memberId;
          shouldSync = pbra.shouldSyncForCrashedMember(mbr);
          if (shouldSync) {
            profile = pbr.getBucketAdvisor().getProfile(memberId);
          }
        }
        boolean removed = pbr.getBucketAdvisor().removeId(memberId, crashed, regionDestroyed,
            fromMembershipListener);
        if (removed && shouldSync) {
          pbra.syncForCrashedMember(mbr, profile);
        }
      }
    }

    boolean removedId;
    removedId = super.removeId(memberId, crashed, regionDestroyed, fromMembershipListener);
    if (logger.isTraceEnabled(LogMarker.DISTRIBUTION_ADVISOR_VERBOSE)) {
      logger.trace(LogMarker.DISTRIBUTION_ADVISOR_VERBOSE,
          "RegionAdvisor#removeId: removing member from region {}: {}; removed = {}; crashed = {}",
          getPartitionedRegion().getName(), memberId, removedId, crashed);
    }

    return removedId;
  }

  /**
   * Clear the knowledge of given member from this advisor. In particular, clear the knowledge of
   * remote Bucket locations so that we avoid sending partition messages to buckets that will soon
   * be destroyed.
   *
   * @param memberId member that has closed the region
   * @param prSerial serial number of this partitioned region
   * @param serials serial numbers of buckets that need to be removed
   */
  public void removeIdAndBuckets(InternalDistributedMember memberId, int prSerial, int[] serials,
      boolean regionDestroyed) {
    if (logger.isTraceEnabled(LogMarker.DISTRIBUTION_ADVISOR_VERBOSE)) {
      logger.trace(LogMarker.DISTRIBUTION_ADVISOR_VERBOSE,
          "RegionAdvisor#removeIdAndBuckets: removing member from region {}: {}; buckets = ({}) serials",
          getPartitionedRegion().getName(), memberId,
          (serials == null ? "null" : serials.length));
    }

    synchronized (preInitQueueMonitor) {
      if (preInitQueue != null) {
        // Queue profile during pre-initialization
        QueuedBucketProfile qbf = new QueuedBucketProfile(memberId, serials, regionDestroyed);
        preInitQueue.add(qbf);
        return;
      }
    }

    // OK, apply the update NOW
    if (buckets != null) {
      Objects.requireNonNull(serials);
      if (logger.isTraceEnabled(LogMarker.DISTRIBUTION_ADVISOR_VERBOSE)) {
        logger.trace(LogMarker.DISTRIBUTION_ADVISOR_VERBOSE,
            "RegionAdvisor#removeIdAndBuckets: removing buckets for member{};{}", memberId, this);
      }
      for (int i = 0; i < buckets.length; i++) {
        int s = serials[i];
        if (s != ILLEGAL_SERIAL) {
          if (logger.isTraceEnabled(LogMarker.DISTRIBUTION_ADVISOR_VERBOSE)) {
            logger.trace(LogMarker.DISTRIBUTION_ADVISOR_VERBOSE,
                "RegionAdvisor#removeIdAndBuckets: removing bucket #{} serial {}", i, s);
          }
          buckets[i].getBucketAdvisor().removeIdWithSerial(memberId, s, regionDestroyed);
        }
      }

      super.removeIdWithSerial(memberId, prSerial, regionDestroyed);
    }
  }

  public void removeIdAndBucket(int bucketId, InternalDistributedMember memberId, int serial,
      boolean regionDestroyed) {
    synchronized (preInitQueueMonitor) {
      if (preInitQueue != null) {
        // Queue profile during pre-initialization
        QueuedBucketProfile qbf =
            new QueuedBucketProfile(bucketId, memberId, serial, regionDestroyed);
        preInitQueue.add(qbf);
        return;
      }
    }

    if (buckets != null) {
      buckets[bucketId].getBucketAdvisor().removeIdWithSerial(memberId, serial, regionDestroyed);
    }
  }

  /**
   * Iterates over all buckets and marks them sick if the given member hosts the bucket.
   *
   * @param sick true if the bucket should be marked sick, false if healthy
   */
  public void markBucketsOnMember(DistributedMember member, boolean sick) {
    // The health profile exchange at cache level should take care of preInitQueue
    if (buckets == null) {
      return;
    }
    for (int i = 0; i < buckets.length; i++) {
      if (sick && !buckets[i].getBucketOwners().contains(member)) {
        continue;
      }
      buckets[i].setBucketSick(member, sick);
      if (logger.isDebugEnabled()) {
        logger.debug("Marked bucket ({}) {}", getPartitionedRegion().bucketStringForLogs(i),
            (buckets[i].isBucketSick() ? "sick" : "healthy"));
      }
    }
  }

  public void updateBucketStatus(int bucketId, DistributedMember member, boolean profileRemoved) {
    if (profileRemoved) {
      buckets[bucketId].setBucketSick(member, false);

    } else {
      ResourceAdvisor advisor = getPartitionedRegion().getCache().getResourceAdvisor();
      boolean sick = advisor.adviseCriticalMembers().contains(member);
      if (logger.isDebugEnabled()) {
        logger.debug("updateBucketStatus:({}):member:{}:sick:{}",
            getPartitionedRegion().bucketStringForLogs(bucketId), member, sick);
      }
      buckets[bucketId].setBucketSick(member, sick);
    }
  }

  /**
   * throws LowMemoryException if the given bucket is hosted on a member which has crossed the
   * ResourceManager threshold.
   *
   * @param key for bucketId used in exception
   */
  public void checkIfBucketSick(final int bucketId, final Object key) throws LowMemoryException {
    if (MemoryThresholds.isLowMemoryExceptionDisabled()) {
      return;
    }
    if (buckets[bucketId].isBucketSick()) {
      Set<DistributedMember> sm = buckets[bucketId].getSickMembers();
      if (sm.isEmpty()) {
        // check again as this list is obtained under synchronization
        // fixes bug 50845
        return;
      }
      if (logger.isDebugEnabled()) {
        logger.debug("For bucket {} sick members are {}.",
            getPartitionedRegion().bucketStringForLogs(bucketId), sm);
      }
      throw new LowMemoryException(String.format(
          "PartitionedRegion: %s cannot process operation on key %s because members %s are running low on memory",
          getPartitionedRegion().getFullPath(), key, sm), sm);
    }
  }

  /**
   * Profile information for a remote counterpart.
   */
  public static class PartitionProfile extends CacheProfile {

    /**
     * The number of Mb the VM is allowed to use for the PR
     * {@link PartitionedRegion#getLocalMaxMemory()}
     */
    public int localMaxMemory;

    /**
     * A data store is a VM that has a non-zero local max memory, Since the localMaxMemory is
     * already sent, there is no need to send this state as it's implied in localMaxMemory
     */
    public transient boolean isDataStore = false;

    /**
     * requiresNotification determines whether a member needs to be notified of cache operations so
     * that cache listeners and other hooks can be engaged
     *
     * @since GemFire 5.1
     */
    public boolean requiresNotification = false;

    /**
     * Track the number of buckets this data store may have, implies isDataStore == true This value
     * is NOT sent directly but updated when {@link org.apache.geode.internal.cache.BucketAdvisor}s
     * receive updates
     */
    public transient short numBuckets = 0;

    /**
     * represents the list of the fixed partitions defined for this region.
     */
    public List<FixedPartitionAttributesImpl> fixedPAttrs;


    // Indicate the status of shutdown request
    public int shutDownAllStatus = PartitionedRegion.RUNNING_MODE;

    /** for internal use, required for DataSerializer.readObject */
    public PartitionProfile() {}

    public PartitionProfile(InternalDistributedMember memberId, int version) {
      super(memberId, version);
      isPartitioned = true;
    }

    @Override
    protected int getIntInfo() {
      int s = super.getIntInfo();
      if (requiresNotification) {
        s |= REQUIRES_NOTIFICATION_MASK;
      }
      return s;
    }

    @Override
    protected void setIntInfo(int s) {
      super.setIntInfo(s);
      requiresNotification = (s & REQUIRES_NOTIFICATION_MASK) != 0;
    }

    @Override
    public void fromData(DataInput in,
        DeserializationContext context) throws IOException, ClassNotFoundException {
      super.fromData(in, context);
      localMaxMemory = in.readInt();
      isDataStore = localMaxMemory > 0;
      fixedPAttrs = DataSerializer.readObject(in);
      shutDownAllStatus = in.readInt();
    }

    @Override
    public void toData(DataOutput out,
        SerializationContext context) throws IOException {
      super.toData(out, context);
      out.writeInt(localMaxMemory);
      DataSerializer.writeObject(fixedPAttrs, out);
      out.writeInt(shutDownAllStatus);
    }

    @Override
    public StringBuilder getToStringHeader() {
      return new StringBuilder("RegionAdvisor.PartitionProfile");
    }

    @Override
    public void fillInToString(StringBuilder sb) {
      super.fillInToString(sb);
      sb.append("; isDataStore=").append(isDataStore).append("; requiresNotification=")
          .append(requiresNotification).append("; localMaxMemory=").append(localMaxMemory)
          .append("; numBuckets=").append(numBuckets);
      if (fixedPAttrs != null) {
        sb.append("; FixedPartitionAttributes=").append(fixedPAttrs);
      }
      sb.append("; filterProfile=").append(filterProfile);
      sb.append("; shutDownAllStatus=").append(shutDownAllStatus);
    }

    @Override
    public int getDSFID() {
      return PARTITION_PROFILE;
    }


  } // end class PartitionProfile

  public int getNumDataStores() {
    final int numProfs = getNumProfiles();
    if (lastActiveProfiles != numProfs) {
      numDataStores = adviseDataStore().size();
      lastActiveProfiles = numProfs;
    }
    return numDataStores;
  }

  public Set<InternalDistributedMember> adviseDataStore() {
    return adviseDataStore(false);
  }

  /**
   * Returns the set of data stores that have finished initialization.
   */
  public Set<InternalDistributedMember> adviseInitializedDataStore() {
    return adviseFilter(profile -> {
      // probably not needed as all profiles for a partitioned region are Partition profiles
      if (profile instanceof PartitionProfile) {
        PartitionProfile p = (PartitionProfile) profile;
        return p.isDataStore && (!p.dataPolicy.withPersistence() || p.regionInitialized);
      }
      return false;
    });
  }

  /**
   * Returns the set of members that are not arrived at specified shutDownAll status
   */
  private Set<InternalDistributedMember> adviseNotAtShutDownAllStatus(final int status) {
    return adviseFilter(profile -> {
      // probably not needed as all profiles for a partitioned region are Partition profiles
      if (profile instanceof PartitionProfile) {
        PartitionProfile p = (PartitionProfile) profile;
        return p.isDataStore && p.shutDownAllStatus < status;
      }
      return false;
    });
  }

  public void waitForProfileStatus(int status) {
    ProfileShutdownListener listener = new ProfileShutdownListener();
    addProfileChangeListener(listener);
    try {
      int memberNum;
      String regionName = getPartitionedRegion().getFullPath();
      do {
        Region pr = getPartitionedRegion().getCache().getRegion(regionName);
        if (pr == null || pr.isDestroyed()) {
          break;
        }
        Set members = adviseNotAtShutDownAllStatus(status);
        memberNum = members.size();
        if (memberNum > 0) {
          if (logger.isDebugEnabled()) {
            logger.debug("waitForProfileStatus {} at PR:{}, expecting {} members: {}", status,
                getPartitionedRegion().getFullPath(), memberNum, members);
          }
          listener.waitForChange();
        }
      } while (memberNum > 0);
    } finally {
      removeProfileChangeListener(listener);
    }
  }

  /**
   * Return a real Set if set to true, which can be modified
   *
   * @param realHashSet true if a real set is needed
   * @return depending on the realHashSet value may be a HashSet
   */
  public Set<InternalDistributedMember> adviseDataStore(boolean realHashSet) {
    Set<InternalDistributedMember> s = adviseFilter(profile -> {
      // probably not needed as all profiles for a partitioned region are Partition profiles
      if (profile instanceof PartitionProfile) {
        PartitionProfile p = (PartitionProfile) profile;
        return p.isDataStore;
      }
      return false;
    });

    if (realHashSet) {
      if (s == Collections.EMPTY_SET) {
        s = new HashSet<>();
      }
    }

    if (logger.isTraceEnabled(LogMarker.DISTRIBUTION_ADVISOR_VERBOSE)) {
      logger.trace(LogMarker.DISTRIBUTION_ADVISOR_VERBOSE, "adviseDataStore returning {} from {}",
          s, toStringWithProfiles());
    }
    return s;
  }

  /**
   * return the set of the distributed members on which the given partition name is defined.
   *
   */

  public Set<InternalDistributedMember> adviseFixedPartitionDataStores(final String partitionName) {
    Set<InternalDistributedMember> s = adviseFilter(profile -> {
      // probably not needed as all profiles for a partitioned region are
      // Partition profiles
      if (profile instanceof PartitionProfile) {
        PartitionProfile p = (PartitionProfile) profile;
        if (p.fixedPAttrs != null) {
          for (FixedPartitionAttributesImpl fpa : p.fixedPAttrs) {
            if (fpa.getPartitionName().equals(partitionName)) {
              return true;
            }
          }
        }
      }
      return false;
    });

    if (s == Collections.EMPTY_SET) {
      s = new HashSet<>();
    }

    if (logger.isTraceEnabled(LogMarker.DISTRIBUTION_ADVISOR_VERBOSE)) {
      logger.trace(LogMarker.DISTRIBUTION_ADVISOR_VERBOSE,
          "adviseFixedPartitionDataStore returning {} from {}", s, toStringWithProfiles());
    }
    return s;
  }

  /**
   * return a distributed members on which the primary partition for given bucket is defined
   *
   */
  public InternalDistributedMember adviseFixedPrimaryPartitionDataStore(final int bucketId) {
    final List<InternalDistributedMember> fixedPartitionDataStore = new ArrayList<>(1);
    fetchProfiles(profile -> {
      if (profile instanceof PartitionProfile) {
        PartitionProfile p = (PartitionProfile) profile;
        if (p.fixedPAttrs != null) {
          for (FixedPartitionAttributesImpl fpa : p.fixedPAttrs) {
            if (fpa.isPrimary() && fpa.hasBucket(bucketId)) {
              fixedPartitionDataStore.add(0, p.getDistributedMember());
              return true;
            }
          }
        }
      }
      return false;
    });

    if (logger.isTraceEnabled(LogMarker.DISTRIBUTION_ADVISOR_VERBOSE)) {
      logger.trace(LogMarker.DISTRIBUTION_ADVISOR_VERBOSE,
          "adviseFixedPartitionDataStore returning {} from {}", fixedPartitionDataStore,
          toStringWithProfiles());
    }
    if (fixedPartitionDataStore.isEmpty()) {
      return null;
    }
    return fixedPartitionDataStore.get(0);
  }

  /**
   * Returns the list of all remote FixedPartitionAttributes defined across all members for the
   * given partitioned region
   *
   * @return list of all partitions(primary as well as secondary) defined on remote nodes
   */
  public List<FixedPartitionAttributesImpl> adviseAllFixedPartitionAttributes() {
    final List<FixedPartitionAttributesImpl> allFPAs = new ArrayList<>();
    fetchProfiles(profile -> {
      if (profile instanceof PartitionProfile) {
        final PartitionProfile pp = (PartitionProfile) profile;
        if (pp.fixedPAttrs != null) {
          allFPAs.addAll(pp.fixedPAttrs);
          return true;
        }
      }
      return false;
    });
    return allFPAs;
  }

  /**
   * Returns the list of all FixedPartitionAttributes defined across all members of given
   * partitioned region for a given FixedPartitionAttributes
   *
   * @return the list of same partitions defined on other nodes(can be primary or secondary)
   */
  public List<FixedPartitionAttributesImpl> adviseSameFPAs(final FixedPartitionAttributesImpl fpa) {
    final List<FixedPartitionAttributesImpl> sameFPAs = new ArrayList<>();

    fetchProfiles(profile -> {
      if (profile instanceof PartitionProfile) {
        final PartitionProfile pp = (PartitionProfile) profile;
        List<FixedPartitionAttributesImpl> fpaList = pp.fixedPAttrs;
        if (fpaList != null) {
          int index = fpaList.indexOf(fpa);
          if (index != -1) {
            sameFPAs.add(fpaList.get(index));
          }
          return true;
        }
      }
      return false;
    });
    return sameFPAs;
  }

  /**
   * Returns the list of all remote primary FixedPartitionAttributes defined across members for the
   * given partitioned region
   *
   * @return list of all primary partitions defined on remote nodes
   */
  public List<FixedPartitionAttributesImpl> adviseRemotePrimaryFPAs() {
    final List<FixedPartitionAttributesImpl> remotePrimaryFPAs = new ArrayList<>();

    fetchProfiles(profile -> {
      if (profile instanceof PartitionProfile) {
        final PartitionProfile pp = (PartitionProfile) profile;
        List<FixedPartitionAttributesImpl> fpaList = pp.fixedPAttrs;
        if (fpaList != null) {
          for (FixedPartitionAttributesImpl fpa : fpaList) {
            if (fpa.isPrimary()) {
              remotePrimaryFPAs.add(fpa);
              return true;
            }
          }
        }
      }
      return false;
    });
    return remotePrimaryFPAs;
  }

  public Set<InternalDistributedMember> adviseAllPRNodes() {
    return adviseFilter(profile -> {
      CacheProfile prof = (CacheProfile) profile;
      return prof.isPartitioned;
    });
  }

  Set adviseAllServersWithInterest() {
    return adviseFilter(profile -> {
      CacheProfile prof = (CacheProfile) profile;
      return prof.hasCacheServer && prof.filterProfile != null
          && prof.filterProfile.hasInterest();
    });
  }

  @Immutable
