| /* |
| * 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.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 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.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.Assert; |
| 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.LogService; |
| import org.apache.geode.internal.logging.log4j.LogMarker; |
| |
| 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(DistributionConfig.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; |
| |
| private Queue<QueuedBucketProfile> preInitQueue; |
| private final Object preInitQueueMonitor = new Object(); |
| |
| private ConcurrentHashMap<Integer, Set<ServerBucketProfile>> clientBucketProfilesMap; |
| |
| private 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); |
| } |
| 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 |
| preInitQueueMonitor.notifyAll(); |
| 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; |
| } |
| |
| 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); |
| } |
| } |
| |
| /** |
| * 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; |
| } |
| assert buckets != null; |
| assert buckets[bucketId] != null; |
| 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) throws IOException, ClassNotFoundException { |
| super.fromData(in); |
| localMaxMemory = in.readInt(); |
| isDataStore = localMaxMemory > 0; |
| fixedPAttrs = DataSerializer.readObject(in); |
| shutDownAllStatus = in.readInt(); |
| } |
| |
| @Override |
| public void toData(DataOutput out) throws IOException { |
| super.toData(out); |
| 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 |
| private static final Filter prServerWithInterestFilter = profile -> { |
| CacheProfile prof = (CacheProfile) profile; |
| return prof.isPartitioned && prof.hasCacheServer && prof.filterProfile != null |
| && prof.filterProfile.hasInterest(); |
| }; |
| |
| public boolean hasPRServerWithInterest() { |
| return satisfiesFilter(prServerWithInterestFilter); |
| } |
| |
| /** |
| * return the set of all members who must receive operation notifications |
| * |
| * @since GemFire 5.1 |
| */ |
| public Set<InternalDistributedMember> adviseRequiresNotification() { |
| return adviseFilter(profile -> { |
| if (profile instanceof PartitionProfile) { |
| PartitionProfile prof = (PartitionProfile) profile; |
| if (prof.isPartitioned) { |
| if (prof.hasCacheListener) { |
| InterestPolicy pol = prof.subscriptionAttributes.getInterestPolicy(); |
| if (pol == InterestPolicy.ALL) { |
| return true; |
| } |
| } |
| return prof.requiresNotification; |
| } |
| } |
| return false; |
| }); |
| } |
| |
| @Override |
| public synchronized boolean putProfile(Profile p) { |
| assert p instanceof CacheProfile; |
| CacheProfile profile = (CacheProfile) p; |
| PartitionedRegion pr = getPartitionedRegion(); |
| if (profile.hasCacheLoader) { |
| pr.setHaveCacheLoader(); |
| } |
| // don't keep FilterProfiles around in accessors. They're needed only for |
| // routing messages in data stors |
| if (profile.filterProfile != null) { |
| if (!pr.isDataStore()) { |
| profile.filterProfile = null; |
| } |
| } |
| return super.putProfile(profile); |
| } |
| |
| public PartitionProfile getPartitionProfile(InternalDistributedMember id) { |
| return (PartitionProfile) getProfile(id); |
| } |
| |
| public boolean isPrimaryForBucket(int bucketId) { |
| if (buckets == null) { |
| return false; |
| } |
| return buckets[bucketId].isPrimary(); |
| } |
| |
| /** |
| * Returns true if the bucket is currently being hosted locally. Note that as soon as this call |
| * returns, this datastore may begin to host the bucket, thus two calls in a row may be different. |
| * |
| * @param bucketId the index of the bucket to check |
| * @return true if the bucket is currently being hosted locally |
| */ |
| public boolean isBucketLocal(int bucketId) { |
| if (buckets == null) { |
| return false; |
| } |
| return buckets[bucketId].getHostedBucketRegion() != null; |
| } |
| |
| public boolean areBucketsInitialized() { |
| return buckets != null; |
| } |
| |
| /** |
| * Returns the real BucketRegion if it's currently locally hosted. Otherwise the ProxyBucketRegion |
| * is returned. Note that this member may be in the process of hosting the real bucket. Until that |
| * has completed, getBucket will continue to return the ProxyBucketRegion. |
| * |
| * @param bucketId the index of the bucket to retrieve |
| * @return the bucket identified by bucketId |
| */ |
| public Bucket getBucket(int bucketId) { |
| Assert.assertTrue(buckets != null); |
| ProxyBucketRegion pbr = buckets[bucketId]; |
| Bucket ret = pbr.getHostedBucketRegion(); |
| if (ret != null) { |
| return ret; |
| } else { |
| return pbr; |
| } |
| } |
| |
| /** |
| * Returns the BucketAdvisor for the specified bucket. |
| * |
| * @param bucketId the index of the bucket to retrieve the advisor for |
| * @return the bucket advisor identified by bucketId |
| */ |
| public BucketAdvisor getBucketAdvisor(int bucketId) { |
| Assert.assertTrue(buckets != null); |
| ProxyBucketRegion pbr = buckets[bucketId]; |
| Bucket ret = pbr.getHostedBucketRegion(); |
| if (ret != null) { |
| return ret.getBucketAdvisor(); |
| } else { |
| return pbr.getBucketAdvisor(); |
| } |
| } |
| |
| public Map<Integer, BucketAdvisor> getAllBucketAdvisors() { |
| Assert.assertTrue(buckets != null); |
| Map<Integer, BucketAdvisor> map = new HashMap<>(); |
| for (ProxyBucketRegion pbr : buckets) { |
| Bucket ret = pbr.getHostedBucketRegion(); |
| if (ret != null) { |
| map.put(ret.getId(), ret.getBucketAdvisor()); |
| } |
| } |
| return map; |
| } |
| |
| /** |
| * |
| * @return array of serial numbers for buckets created locally |
| */ |
| public int[] getBucketSerials() { |
| if (buckets == null) { |
| return new int[0]; |
| } |
| int[] result = new int[buckets.length]; |
| |
| for (int i = 0; i < result.length; i++) { |
| ProxyBucketRegion pbr = buckets[i]; |
| Bucket b = pbr.getCreatedBucketRegion(); |
| if (b == null) { |
| result[i] = ILLEGAL_SERIAL; |
| } else { |
| result[i] = b.getSerialNumber(); |
| } |
| } |
| return result; |
| } |
| |
| /** |
| * Returns the bucket identified by bucketId after waiting for initialization to finish processing |
| * queued profiles. Call synchronizes and waits on {@link #preInitQueueMonitor}. |
| * |
| * @param bucketId the bucket identifier |
| * @return the bucket identified by bucketId |
| * @throws org.apache.geode.distributed.DistributedSystemDisconnectedException if interrupted for |
| * shutdown cancellation |
| */ |
| public Bucket getBucketPostInit(int bucketId) { |
| synchronized (preInitQueueMonitor) { |
| boolean interrupted = false; |
| try { |
| while (preInitQueue != null) { |
| try { |
| preInitQueueMonitor.wait(); // spurious wakeup ok |
| } catch (InterruptedException e) { |
| interrupted = true; |
| getAdvisee().getCancelCriterion().checkCancelInProgress(e); |
| } |
| } |
| } finally { |
| if (interrupted) { |
| Thread.currentThread().interrupt(); |
| } |
| } |
| } |
| return getBucket(bucketId); |
| } |
| |
| /** |
| * Get the most recent primary node for the bucketId. Returns null if no primary can be found |
| * within {@link DistributionConfig#getMemberTimeout}. |
| * |
| * @return the Node managing the primary copy of the bucket |
| */ |
| public InternalDistributedMember getPrimaryMemberForBucket(int bucketId) { |
| Assert.assertTrue(buckets != null); |
| Bucket b = buckets[bucketId]; |
| return b.getBucketAdvisor().getPrimary(); |
| } |
| |
| /** |
| * Return the node favored for reading for the given bucket |
| * |
| * @param bucketId the bucket we want to read |
| * @return the member, possibly null if no member is available |
| */ |
| public InternalDistributedMember getPreferredNode(int bucketId) { |
| Assert.assertTrue(buckets != null); |
| Bucket b = buckets[bucketId]; |
| return b.getBucketAdvisor().getPreferredNode(); |
| } |
| |
| public boolean isStorageAssignedForBucket(int bucketId) { |
| Assert.assertTrue(buckets != null); |
| return buckets[bucketId].getBucketRedundancy() >= 0; |
| } |
| |
| /** |
| * @param bucketId the bucket to check redundancy on |
| * @param minRedundancy the amount of expected redundancy; ignored if wait is false |
| * @param wait true if caller wants us to wait for redundancy |
| * @return true if redundancy on given bucket is detected |
| */ |
| public boolean isStorageAssignedForBucket(int bucketId, int minRedundancy, boolean wait) { |
| if (!wait) { |
| return isStorageAssignedForBucket(bucketId); |
| } else { |
| Assert.assertTrue(buckets != null); |
| return buckets[bucketId].getBucketAdvisor().waitForRedundancy(minRedundancy); |
| } |
| } |
| |
| /** |
| * Get the redundancy of the this bucket, taking into account the local bucket, if any. |
| * |
| * @return number of redundant copies for a given bucket, or -1 if there are no instances of the |
| * bucket. |
| */ |
| public int getBucketRedundancy(int bucketId) { |
| Assert.assertTrue(buckets != null); |
| return buckets[bucketId].getBucketRedundancy(); |
| } |
| |
| /** |
| * Return the set of all members who currently own the bucket, including the local owner, if |
| * applicable |
| * |
| * @return a set of {@link InternalDistributedMember}s that own the bucket |
| */ |
| public Set<InternalDistributedMember> getBucketOwners(int bucketId) { |
| Assert.assertTrue(buckets != null); |
| return buckets[bucketId].getBucketOwners(); |
| } |
| |
| /** |
| * Return the set of buckets which have storage assigned |
| * |
| * @return set of Integer bucketIds |
| */ |
| public Set<Integer> getBucketSet() { |
| Assert.assertTrue(buckets != null); |
| return new BucketSet(); |
| } |
| |
| |
| public ProxyBucketRegion[] getProxyBucketArray() { |
| return buckets; |
| } |
| |
| private class BucketSet extends AbstractSet<Integer> { |
| final ProxyBucketRegion[] pbrs; |
| |
| BucketSet() { |
| pbrs = buckets; |
| Assert.assertTrue(pbrs != null); |
| } |
| |
| /* |
| * Note: The consistency between size(), hasNext() and next() is weak, meaning that the state of |
| * the backing Set may change causing more or less elements to be available after calling size() |
| */ |
| @Override |
| public int size() { |
| return pbrs.length; |
| } |
| |
| @Override |
| public Iterator<Integer> iterator() { |
| return new BucketSetIterator(); |
| } |
| |
| class BucketSetIterator implements Iterator<Integer> { |
| private int currentItem = -1; |
| |
| |
| @Override |
| public void remove() { |
| throw new UnsupportedOperationException(); |
| } |
| |
| /* |
| * Note: The consistency guarantee between hasNext() and next() is weak. It's possible |
| * hasNext() will return true and a following call to next() may throw NoSuchElementException |
| * (due to loss of bucket storage). Its also equally possible for hasNext() to return false |
| * and a subsequent call to next() will return a valid bucketid. |
| */ |
| @Override |
| public boolean hasNext() { |
| if (getPartitionedRegion().isFixedPartitionedRegion()) { |
| if (currentItem + 1 < pbrs.length) { |
| int possibleBucketId = currentItem; |
| boolean bucketExists = false; |
| |
| List<FixedPartitionAttributesImpl> fpaList = adviseAllFixedPartitionAttributes(); |
| List<FixedPartitionAttributesImpl> localFpas = |
| getPartitionedRegion().getFixedPartitionAttributesImpl(); |
| if (localFpas != null) { |
| fpaList.addAll(localFpas); |
| } |
| while (++possibleBucketId < pbrs.length && !bucketExists) { |
| for (FixedPartitionAttributesImpl fpa : fpaList) { |
| if (fpa.hasBucket(possibleBucketId)) { |
| bucketExists = true; |
| break; |
| } |
| } |
| } |
| return bucketExists; |
| } else { |
| return false; |
| } |
| } else { |
| return currentItem + 1 < pbrs.length; |
| } |
| } |
| |
| @Override |
| public Integer next() { |
| if (++currentItem < pbrs.length) { |
| if (isStorageAssignedForBucket(currentItem)) { |
| return currentItem; |
| } else { |
| if (getPartitionedRegion().isFixedPartitionedRegion()) { |
| boolean bucketExists = false; |
| List<FixedPartitionAttributesImpl> fpaList = adviseAllFixedPartitionAttributes(); |
| List<FixedPartitionAttributesImpl> localFpas = |
| getPartitionedRegion().getFixedPartitionAttributesImpl(); |
| if (localFpas != null) { |
| fpaList.addAll(localFpas); |
| } |
| do { |
| for (FixedPartitionAttributesImpl fpa : fpaList) { |
| if (fpa.hasBucket(currentItem)) { |
| bucketExists = true; |
| break; |
| } |
| } |
| if (!bucketExists) { |
| currentItem++; |
| } |
| } while (currentItem < pbrs.length && !bucketExists); |
| |
| if (bucketExists) { |
| getPartitionedRegion().createBucket(currentItem, 0, null); |
| return currentItem; |
| } |
| } else { |
| getPartitionedRegion().createBucket(currentItem, 0, null); |
| return currentItem; |
| } |
| } |
| } |
| throw new NoSuchElementException(); |
| } |
| } |
| |
| } |
| |
| /** |
| * Obtain the ordered {@link ArrayList} of data stores limited to those specified in the provided |
| * memberFilter. |
| * |
| * @param memberFilter the set of members allowed to be in the list. |
| * @return a list of DataStoreBuckets |
| */ |
| public ArrayList<DataStoreBuckets> adviseFilteredDataStores( |
| final Set<InternalDistributedMember> memberFilter) { |
| final HashMap<InternalDistributedMember, Integer> memberToPrimaryCount = new HashMap<>(); |
| for (ProxyBucketRegion pbr : buckets) { |
| // quick dirty check |
| InternalDistributedMember p = pbr.getBucketAdvisor().basicGetPrimaryMember(); |
| if (p != null) { |
| memberToPrimaryCount.merge(p, 1, Integer::sum); |
| } |
| } |
| |
| final ArrayList<DataStoreBuckets> ds = new ArrayList<>(memberFilter.size()); |
| adviseFilter(profile -> { |
| if (profile instanceof PartitionProfile) { |
| PartitionProfile p = (PartitionProfile) profile; |
| if (memberFilter.contains(p.getDistributedMember())) { |
| Integer priCount = memberToPrimaryCount.get(p.getDistributedMember()); |
| int primaryCount = 0; |
| if (priCount != null) { |
| primaryCount = priCount; |
| } |
| ds.add(new DataStoreBuckets(p.getDistributedMember(), p.numBuckets, primaryCount, |
| p.localMaxMemory)); |
| } |
| } |
| return false; |
| }); |
| |
| |
| return ds; |
| } |
| |
| public void incrementBucketCount(Profile p) { |
| PartitionProfile pp = (PartitionProfile) getProfile(p.getDistributedMember()); |
| if (pp != null) { |
| Assert.assertTrue(pp.isDataStore); |
| pp.numBuckets++; |
| } |
| } |
| |
| public void decrementsBucketCount(Profile p) { |
| PartitionProfile pp = (PartitionProfile) getProfile(p.getDistributedMember()); |
| if (pp != null) { |
| Assert.assertTrue(pp.isDataStore); |
| pp.numBuckets--; |
| if (pp.numBuckets < 0) { |
| pp.numBuckets = 0; |
| } |
| } |
| } |
| |
| /** |
| * Dumps out all profiles in this advisor AND all buckets. Callers should check for debug enabled. |
| * |
| * @param infoMsg prefix message to log |
| */ |
| @Override |
| public void dumpProfiles(String infoMsg) { |
| if (logger.isDebugEnabled()) { |
| logger.debug("[dumpProfiles] dumping {}", toStringWithProfiles()); |
| } |
| |
| // 1st dump all profiles for this RegionAdvisor |
| super.dumpProfiles(infoMsg); |
| |
| // 2nd dump all profiles for each BucketAdvisor |
| ProxyBucketRegion[] pbrs = buckets; |
| if (pbrs == null) { |
| return; |
| } |
| for (ProxyBucketRegion pbr : pbrs) { |
| pbr.getBucketAdvisor().dumpProfiles(infoMsg); |
| BucketPersistenceAdvisor persistentAdvisor = pbr.getPersistenceAdvisor(); |
| if (persistentAdvisor != null) { |
| persistentAdvisor.dump(infoMsg); |
| } |
| } |
| } |
| |
| public void notPrimary(int bucketId, InternalDistributedMember wasPrimary) { |
| Assert.assertTrue(buckets != null); |
| ProxyBucketRegion b = buckets[bucketId]; |
| b.getBucketAdvisor().notPrimary(wasPrimary); |
| } |
| |
| /** |
| * Find the set of members which own primary buckets, including the local member |
| * |
| * @return set of InternalDistributedMember ids |
| */ |
| public Set advisePrimaryOwners() { |
| Assert.assertTrue(buckets != null); |
| ProxyBucketRegion[] bucs = buckets; |
| HashSet<InternalDistributedMember> hs = new HashSet<>(); |
| for (int i = 0; i < bucs.length; i++) { |
| if (isStorageAssignedForBucket(i)) { |
| InternalDistributedMember mem = bucs[i].getBucketAdvisor().getPrimary(); |
| if (mem != null) { |
| hs.add(mem); |
| } |
| } |
| } |
| return hs; |
| } |
| |
| /** |
| * A visitor interface for the buckets of this region used by |
| * {@link RegionAdvisor#accept(BucketVisitor, Object)}. |
| */ |
| public interface BucketVisitor<T> { |
| |
| /** |
| * Visit a given {@link ProxyBucketRegion} accumulating the results in the given aggregate. |
| * Returns false when the visit has to be terminated. |
| */ |
| boolean visit(RegionAdvisor advisor, ProxyBucketRegion pbr, T aggregate); |
| } |
| |
| /** |
| * Invoke the given {@link BucketVisitor} on all the {@link ProxyBucketRegion} s exiting when the |
| * {@link BucketVisitor#visit} method returns false. |
| * |
| * @param <T> the type of object used for aggregation of results |
| * @param visitor the {@link BucketVisitor} to use for the visit |
| * @param aggregate an aggregate object that will be used to for aggregation of results by the |
| * {@link BucketVisitor#visit} method; this allows the {@link BucketVisitor} to not |
| * maintain any state so that in most situations a global static object encapsulating the |
| * required behaviour will work |
| * |
| * @return true when the full visit completed, and false if it was terminated due to |
| * {@link BucketVisitor#visit} returning false |
| */ |
| public <T> boolean accept(BucketVisitor<T> visitor, T aggregate) { |
| final ProxyBucketRegion[] bucs = buckets; |
| Objects.requireNonNull(bucs); |
| for (ProxyBucketRegion pbr : bucs) { |
| if (!visitor.visit(this, pbr, aggregate)) { |
| return false; |
| } |
| } |
| return true; |
| } |
| |
| public PartitionedRegion getPartitionedRegion() { |
| return (PartitionedRegion) getAdvisee(); |
| } |
| |
| /** |
| * Update or create a bucket's meta-data If this advisor has not completed initialization, upon |
| * return the profile will be enqueued for processing during initialization, otherwise the profile |
| * will be immediately processed. This architecture limits the blockage of threads during |
| * initialization. |
| * |
| * @param bucketId the unique identifier of the bucket |
| * @param profile the bucket meta-data from a particular member with the bucket |
| */ |
| void putBucketProfile(int bucketId, BucketProfile profile) { |
| synchronized (preInitQueueMonitor) { |
| if (preInitQueue != null) { |
| // Queue profile during pre-initialization |
| QueuedBucketProfile qbf = new QueuedBucketProfile(bucketId, profile); |
| preInitQueue.add(qbf); |
| return; |
| } |
| } |
| |
| // Directly process profile post-initialization |
| getBucket(bucketId).getBucketAdvisor().putProfile(profile); |
| } |
| |
| static class QueuedBucketProfile { |
| protected final int bucketId; |
| final BucketProfile bucketProfile; |
| |
| /** true means that this member has departed the view */ |
| protected final boolean memberDeparted; |
| |
| /** true means that this profile needs to be removed */ |
| final boolean isRemoval; |
| |
| /** true means that the peer crashed */ |
| protected final boolean crashed; |
| |
| /** |
| * true means that this QueuedBucketProfile was created because of MembershipListener invocation |
| */ |
| final boolean fromMembershipListener; |
| |
| protected final boolean destroyed; |
| |
| protected final InternalDistributedMember memberId; |
| final int[] serials; |
| |
| /** |
| * Queue up an addition |
| * |
| * @param bId the bucket being added |
| * @param p the profile to add |
| */ |
| QueuedBucketProfile(int bId, BucketProfile p) { |
| bucketId = bId; |
| bucketProfile = p; |
| isRemoval = false; |
| crashed = false; |
| memberDeparted = false; |
| memberId = null; |
| serials = null; |
| destroyed = false; |
| fromMembershipListener = false; |
| } |
| |
| /** |
| * Queue up a removal due to member leaving the view |
| * |
| * @param mbr the member being removed |
| */ |
| QueuedBucketProfile(InternalDistributedMember mbr, boolean crashed, boolean destroyed, |
| boolean fromMembershipListener) { |
| bucketId = 0; |
| bucketProfile = null; |
| isRemoval = true; |
| this.crashed = crashed; |
| memberDeparted = true; |
| memberId = mbr; |
| serials = null; |
| this.destroyed = destroyed; |
| this.fromMembershipListener = fromMembershipListener; |
| } |
| |
| /** |
| * Queue up a removal due to region destroy |
| * |
| * @param mbr the member being removed |
| * @param serials the serials it had |
| */ |
| QueuedBucketProfile(InternalDistributedMember mbr, int[] serials, boolean destroyed) { |
| bucketId = 0; |
| bucketProfile = null; |
| isRemoval = true; |
| crashed = false; |
| memberDeparted = false; |
| memberId = mbr; |
| this.serials = serials; |
| this.destroyed = destroyed; |
| fromMembershipListener = false; |
| } |
| } |
| |
| public Set<InternalDistributedMember> adviseBucketProfileExchange() { |
| return adviseDataStore(); |
| } |
| |
| public long adviseTotalMemoryAllocation() { |
| final AtomicLong total = new AtomicLong(); |
| adviseFilter(profile -> { |
| // probably not needed as all profiles for a partitioned region are Partition profiles |
| if (profile instanceof PartitionProfile) { |
| PartitionProfile p = (PartitionProfile) profile; |
| total.addAndGet(p.localMaxMemory); |
| } |
| return false; |
| }); |
| return total.get(); |
| } |
| |
| /** |
| * Returns the total number of buckets created anywhere in the distributed system for this |
| * partitioned region. |
| * |
| * @return the total number of buckets created anywhere for this PR |
| */ |
| public int getCreatedBucketsCount() { |
| final ProxyBucketRegion[] bucs = buckets; |
| if (bucs == null) { |
| return 0; |
| } |
| int createdBucketsCount = 0; |
| for (ProxyBucketRegion buc : bucs) { |
| if (buc.getBucketOwnersCount() > 0) { |
| createdBucketsCount++; |
| } |
| } |
| return createdBucketsCount; |
| } |
| |
| /** |
| * Returns a possibly null list of this advisor's real bucket profiles. A real bucket profile is |
| * one that for a bucket that actually has storage in this vm. |
| * |
| * @return a list of BucketProfileAndId instances; may be null |
| * @since GemFire 5.5 |
| */ |
| public ArrayList getBucketRegionProfiles() { |
| final ProxyBucketRegion[] bucs = buckets; |
| if (bucs == null) { |
| return null; |
| } |
| ArrayList<BucketProfileAndId> result = new ArrayList<>(bucs.length); |
| for (int i = 0; i < bucs.length; i++) { |
| // Fix for 41436 - we need to include buckets that are still initializing here |
| // we must start including buckets in this list *before* those buckets exchange |
| // profiles. |
| BucketRegion br = bucs[i].getCreatedBucketRegion(); |
| if (br != null) { |
| result.add(new BucketProfileAndId(br.getProfile(), i)); |
| } |
| } |
| if (result.size() == 0) { |
| result = null; |
| } |
| return result; |
| } |
| |
| /** |
| * Takes a list of BucketProfileAndId and adds them to thsi advisors proxy buckets. |
| * |
| * @since GemFire 5.5 |
| */ |
| public void putBucketRegionProfiles(ArrayList<BucketProfileAndId> l) { |
| for (BucketProfileAndId bp : l) { |
| int id = bp.getId(); |
| getBucket(id).getBucketAdvisor().putProfile(bp.getBucketProfile()); |
| } |
| } |
| |
| /** |
| * return true if the given member has this advisor's partitioned region |
| */ |
| public boolean hasPartitionedRegion(InternalDistributedMember profileId) { |
| if (getDistributionManager().getId().equals(profileId)) { |
| return true; |
| } |
| return (getProfile(profileId) != null); |
| } |
| |
| @Override |
| protected void profileRemoved(Profile profile) { |
| if (logger.isDebugEnabled()) { |
| logger.debug("RA: removing profile {}", profile); |
| } |
| if (getAdvisee() instanceof PartitionedRegion) { |
| ((PartitionedRegion) getAdvisee()).removeCriticalMember(profile.peerMemberId); |
| } |
| |
| if (buckets != null) { |
| for (ProxyBucketRegion bucket : buckets) { |
| bucket.getBucketAdvisor().checkForLostPrimaryElector(profile); |
| } |
| } |
| } |
| |
| public static class BucketProfileAndId implements DataSerializable { |
| private static final long serialVersionUID = 332892607792421553L; |
| /* final */ private int id; |
| // bid = bucket id |
| /* final */ private BucketProfile bp; |
| private boolean isServerBucketProfile = false; |
| |
| public BucketProfileAndId(Profile bp, int id) { |
| this.id = id; |
| this.bp = (BucketProfile) bp; |
| if (bp instanceof ServerBucketProfile) |
| isServerBucketProfile = true; |
| } |
| |
| public BucketProfileAndId() {} |
| |
| public int getId() { |
| return id; |
| } |
| |
| BucketProfile getBucketProfile() { |
| return bp; |
| } |
| |
| @Override |
| public void fromData(DataInput in) throws IOException, ClassNotFoundException { |
| id = in.readInt(); |
| isServerBucketProfile = in.readBoolean(); |
| if (isServerBucketProfile) |
| bp = new ServerBucketProfile(); |
| else |
| bp = new BucketProfile(); |
| |
| InternalDataSerializer.invokeFromData(bp, in); |
| } |
| |
| @Override |
| public void toData(DataOutput out) throws IOException { |
| out.writeInt(id); |
| out.writeBoolean(isServerBucketProfile); |
| InternalDataSerializer.invokeToData(bp, out); |
| } |
| |
| @Override |
| public String toString() { |
| return "BucketProfileAndId (profile=" + bp + "; id=" + id + ")"; |
| } |
| } |
| |
| // profile listener to monitor remote member unexpected leave during shutdownAll |
| private class ProfileShutdownListener implements ProfileListener { |
| |
| ProfileShutdownListener() { |
| |
| } |
| |
| private boolean profileChanged = false; |
| |
| void waitForChange() { |
| Region pr = getPartitionedRegion(); |
| |
| synchronized (this) { |
| while (!profileChanged && pr != null && !pr.isDestroyed()) { |
| // the advisee might have been destroyed due to initialization failure |
| try { |
| wait(1000); |
| } catch (InterruptedException ignored) { |
| } |
| } |
| profileChanged = false; |
| } |
| } |
| |
| @Override |
| public void profileCreated(Profile profile) { |
| profileUpdated(profile); |
| } |
| |
| @Override |
| public void profileRemoved(Profile profile, boolean regionDestroyed) { |
| // if a profile is gone, notify |
| synchronized (this) { |
| profileChanged = true; |
| notifyAll(); |
| } |
| } |
| |
| @Override |
| public void profileUpdated(Profile profile) { |
| // when updated, notify the loop in GFC to check the list again |
| synchronized (this) { |
| profileChanged = true; |
| notifyAll(); |
| } |
| } |
| } |
| } |
