| /* |
| * 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; |
| |
| import static org.apache.geode.internal.cache.LocalRegion.InitializationLevel.ANY_INIT; |
| |
| import java.io.DataInput; |
| import java.io.DataOutput; |
| import java.io.IOException; |
| 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.Set; |
| |
| import org.apache.logging.log4j.Logger; |
| |
| import org.apache.geode.DataSerializer; |
| import org.apache.geode.cache.CacheEvent; |
| import org.apache.geode.cache.DataPolicy; |
| import org.apache.geode.cache.InterestPolicy; |
| import org.apache.geode.cache.RegionDestroyedException; |
| import org.apache.geode.cache.Scope; |
| import org.apache.geode.cache.SubscriptionAttributes; |
| import org.apache.geode.distributed.internal.ClusterDistributionManager; |
| import org.apache.geode.distributed.internal.DistributionAdvisor; |
| import org.apache.geode.distributed.internal.MembershipListener; |
| import org.apache.geode.distributed.internal.membership.InternalDistributedMember; |
| import org.apache.geode.internal.Assert; |
| import org.apache.geode.internal.DSCODE; |
| import org.apache.geode.internal.InternalDataSerializer; |
| import org.apache.geode.internal.cache.LocalRegion.InitializationLevel; |
| import org.apache.geode.internal.cache.partitioned.PRLocallyDestroyedException; |
| import org.apache.geode.internal.cache.persistence.DiskStoreID; |
| import org.apache.geode.internal.cache.persistence.PersistentMemberID; |
| import org.apache.geode.internal.logging.LogService; |
| import org.apache.geode.internal.logging.log4j.LogMarker; |
| |
| /** |
| * Adds bookkeeping info and cache-specific behavior to DistributionAdvisor. Adds bit-encoded flags |
| * in addition to object info. |
| * |
| */ |
| @SuppressWarnings("deprecation") |
| public class CacheDistributionAdvisor extends DistributionAdvisor { |
| |
| private static final Logger logger = LogService.getLogger(); |
| |
| // moved ROLLOVER_* constants to DistributionAdvisor |
| |
| /** bit masks */ |
| private static final int INTEREST_MASK = 0x01; |
| private static final int REPLICATE_MASK = 0x02; |
| private static final int LOADER_MASK = 0x04; |
| private static final int WRITER_MASK = 0x08; |
| private static final int LISTENER_MASK = 0x10; |
| private static final int DIST_ACK_MASK = 0x20; |
| private static final int GLOBAL_MASK = 0x40; |
| private static final int IN_RECOVERY_MASK = 0x80; |
| private static final int PERSISTENT_MASK = 0x100; |
| private static final int PROXY_MASK = 0x200; |
| private static final int PRELOADED_MASK = 0x400; |
| private static final int IS_PARTITIONED_MASK = 0x800; |
| private static final int REGION_INITIALIZED_MASK = 0x1000; |
| private static final int IS_GATEWAY_ENABLED_MASK = 0x2000; |
| // provider is no longer a supported attribute. |
| // private static final int IS_GII_PROVIDER_MASK = 0x4000; |
| private static final int PERSISTENT_ID_MASK = 0x4000; |
| /** does this member require operation notification (PartitionedRegions) */ |
| protected static final int REQUIRES_NOTIFICATION_MASK = 0x8000; |
| private static final int HAS_CACHE_SERVER_MASK = 0x10000; |
| private static final int REQUIRES_OLD_VALUE_MASK = 0x20000; |
| // unused 0x40000; |
| private static final int PERSISTENCE_INITIALIZED_MASK = 0x80000; |
| // Important below mentioned bit masks are not available |
| /** |
| * Using following masks for gatewaysender queue startup polic informations. |
| */ |
| // private static final int HUB_STARTUP_POLICY_MASK = 0x07<<20; |
| |
| private static final int GATEWAY_SENDER_IDS_MASK = 0x200000; |
| |
| private static final int ASYNC_EVENT_QUEUE_IDS_MASK = 0x400000; |
| private static final int IS_OFF_HEAP_MASK = 0x800000; |
| private static final int CACHE_SERVICE_PROFILES_MASK = 0x1000000; |
| |
| |
| // moved initializ* to DistributionAdvisor |
| |
| // moved membershipVersion to DistributionAdvisor |
| |
| // moved previousVersionOpCount to DistributionAdvisor |
| // moved currentVersionOpCount to DistributionAdvisor |
| |
| // moved removedProfiles to DistributionAdvisor |
| |
| /** Creates a new instance of CacheDistributionAdvisor */ |
| protected CacheDistributionAdvisor(CacheDistributionAdvisee region) { |
| super(region); |
| } |
| |
| static CacheDistributionAdvisor createCacheDistributionAdvisor( |
| CacheDistributionAdvisee region) { |
| CacheDistributionAdvisor advisor = new CacheDistributionAdvisor(region); |
| advisor.initialize(); |
| return advisor; |
| } |
| |
| @Override |
| public String toString() { |
| return "CacheDistributionAdvisor for region " + getAdvisee().getFullPath(); |
| } |
| |
| // moved toStringWithProfiles to DistributionAdvisor |
| |
| // moved initializationGate to DistributionAdvisor |
| |
| // moved isInitialized to DistributionAdvisor |
| |
| // moved addMembershipListenerAndAdviseGeneric to DistributionAdvisor |
| |
| /** |
| * Returns a the set of members that either want all events or are caching data. |
| * |
| * @param excludeInRecovery if true then members in recovery are excluded |
| */ |
| private Set<InternalDistributedMember> adviseAllEventsOrCached(final boolean excludeInRecovery) |
| throws IllegalStateException { |
| getAdvisee().getCancelCriterion().checkCancelInProgress(null); |
| return adviseFilter(profile -> { |
| assert profile instanceof CacheProfile; |
| CacheProfile cp = (CacheProfile) profile; |
| if (excludeInRecovery && cp.inRecovery) { |
| return false; |
| } |
| return cp.cachedOrAllEventsWithListener(); |
| }); |
| } |
| |
| /** |
| * Provide recipient information for an update or create operation. |
| * |
| */ |
| Set adviseUpdate(final EntryEventImpl event) throws IllegalStateException { |
| if (event.hasNewValue() || event.getOperation().isPutAll()) { |
| // only need to distribute it to members that want all events or cache data |
| return adviseAllEventsOrCached(true/* fixes 41147 */); |
| } else { |
| // The new value is null so this is a create with a null value, |
| // in which case we only need to distribute this message to replicates |
| // or all events that are not a proxy or if a proxy has a listener |
| return adviseFilter(profile -> { |
| assert profile instanceof CacheProfile; |
| CacheProfile cp = (CacheProfile) profile; |
| DataPolicy dp = cp.dataPolicy; |
| return dp.withReplication() |
| || (cp.allEvents() && (dp.withStorage() || cp.hasCacheListener)); |
| }); |
| } |
| } |
| |
| /** |
| * Provide recipient information for TX lock and commit. |
| * |
| * @return Set of Serializable members that the current transaction will be distributed to. |
| * Currently this is any other member who has this region defined. No reference to Set |
| * kept by advisor so caller is free to modify it |
| */ |
| Set<InternalDistributedMember> adviseTX() throws IllegalStateException { |
| |
| boolean isMetaDataWithTransactions = getAdvisee() instanceof LocalRegion |
| && ((LocalRegion) getAdvisee()).isMetaRegionWithTransactions(); |
| |
| Set<InternalDistributedMember> badList = Collections.emptySet(); |
| if (!TXManagerImpl.ALLOW_PERSISTENT_TRANSACTIONS && !isMetaDataWithTransactions) { |
| badList = adviseFilter(profile -> { |
| assert profile instanceof CacheProfile; |
| CacheProfile prof = (CacheProfile) profile; |
| return (prof.isPersistent()); |
| }); |
| } |
| if (badList.isEmpty()) { |
| return adviseFilter(profile -> { |
| assert profile instanceof CacheProfile; |
| CacheProfile cp = (CacheProfile) profile; |
| return cp.cachedOrAllEvents(); |
| }); |
| } else { |
| StringBuilder badIds = new StringBuilder(); |
| Iterator biI = badList.iterator(); |
| while (biI.hasNext()) { |
| badIds.append(biI.next().toString()); |
| if (biI.hasNext()) |
| badIds.append(", "); |
| } |
| throw new IllegalStateException( |
| String.format("Illegal Region Configuration for members: %s", |
| badIds.toString())); |
| } |
| } |
| |
| /** |
| * Provide recipient information for netLoad |
| * |
| * @return Set of Serializable members that have a CacheLoader installed; no reference to Set kept |
| * by advisor so caller is free to modify it |
| */ |
| public Set adviseNetLoad() { |
| return adviseFilter(profile -> { |
| assert profile instanceof CacheProfile; |
| CacheProfile prof = (CacheProfile) profile; |
| |
| // if region in cache is not yet initialized, exclude |
| if (!prof.regionInitialized) { // fix for bug 41102 |
| return false; |
| } |
| |
| return prof.hasCacheLoader; |
| }); |
| } |
| |
| public FilterRoutingInfo adviseFilterRouting(CacheEvent event, Set cacheOpRecipients) { |
| FilterProfile fp = ((LocalRegion) event.getRegion()).getFilterProfile(); |
| if (fp != null) { |
| return fp.getFilterRoutingInfoPart1(event, profiles, cacheOpRecipients); |
| } |
| return null; |
| } |
| |
| |
| /** |
| * Same as adviseGeneric except in recovery excluded. |
| */ |
| public Set<InternalDistributedMember> adviseCacheOp() { |
| return adviseAllEventsOrCached(true); |
| } |
| |
| /* |
| * * Same as adviseGeneric but excludes if cache profile is in recovery |
| */ |
| Set<InternalDistributedMember> adviseInvalidateRegion() { |
| return adviseFilter(profile -> { |
| assert profile instanceof CacheProfile; |
| CacheProfile cp = (CacheProfile) profile; |
| return !cp.inRecovery; |
| }); |
| } |
| |
| |
| /** |
| * Same as adviseGeneric |
| */ |
| public Set adviseDestroyRegion() { |
| return adviseGeneric(); |
| } |
| |
| /** |
| * Provide recipient information for netWrite |
| * |
| * @return Set of Serializable member ids that have a CacheWriter installed; no reference to Set |
| * kept by advisor so caller is free to modify it |
| */ |
| public Set adviseNetWrite() { |
| return adviseFilter(profile -> { |
| assert profile instanceof CacheProfile; |
| CacheProfile prof = (CacheProfile) profile; |
| // if region in cache is in recovery, exclude |
| if (prof.inRecovery) { |
| return false; |
| } |
| |
| return prof.hasCacheWriter; |
| }); |
| } |
| |
| public Set<InternalDistributedMember> adviseInitializedReplicates() { |
| return adviseFilter(profile -> { |
| assert profile instanceof CacheProfile; |
| CacheProfile cp = (CacheProfile) profile; |
| return cp.dataPolicy.withReplication() && cp.regionInitialized; |
| }); |
| } |
| |
| /** |
| * Provide recipient information for netSearch |
| * |
| * @return Set of Serializable member ids that have the region and are have storage (no need to |
| * search an empty cache) |
| */ |
| Set adviseNetSearch() { |
| return adviseFilter(profile -> { |
| assert profile instanceof CacheProfile; |
| CacheProfile cp = (CacheProfile) profile; |
| // if region in cache is not yet initialized, exclude |
| if (!cp.regionInitialized) { |
| return false; |
| } |
| DataPolicy dp = cp.dataPolicy; |
| return dp.withStorage(); |
| }); |
| } |
| |
| // moved dumpProfiles to DistributionAdvisor |
| |
| public InitialImageAdvice adviseInitialImage(InitialImageAdvice previousAdvice) { |
| return adviseInitialImage(previousAdvice, false); |
| } |
| |
| @SuppressWarnings("synthetic-access") |
| public InitialImageAdvice adviseInitialImage(InitialImageAdvice previousAdvice, |
| boolean persistent) { |
| initializationGate(); |
| |
| if (logger.isTraceEnabled(LogMarker.DISTRIBUTION_ADVISOR_VERBOSE)) { |
| dumpProfiles("AdviseInitialImage"); |
| } |
| |
| Profile[] allProfiles = profiles; // volatile read |
| if (allProfiles.length == 0) { |
| return new InitialImageAdvice(); |
| } |
| |
| Set<InternalDistributedMember> replicates = new HashSet<>(); |
| Set<InternalDistributedMember> others = new HashSet<>(); |
| Set<InternalDistributedMember> preloaded = new HashSet<>(); |
| Set<InternalDistributedMember> empties = new HashSet<>(); |
| Set<InternalDistributedMember> uninitialized = new HashSet<>(); |
| Set<InternalDistributedMember> nonPersistent = new HashSet<>(); |
| |
| Map<InternalDistributedMember, CacheProfile> memberProfiles = |
| new HashMap<>(); |
| |
| for (Profile allProfile : allProfiles) { |
| CacheProfile profile = (CacheProfile) allProfile; |
| |
| // Make sure that we don't return a member that was in the previous initial image advice. |
| // Unless that member has changed it's profile since the last time we checked. |
| if (previousAdvice != null) { |
| CacheProfile previousProfile = |
| previousAdvice.memberProfiles.get(profile.getDistributedMember()); |
| if (previousProfile != null |
| && previousProfile.getSerialNumber() == profile.getSerialNumber() |
| && previousProfile.getVersion() == profile.getVersion()) { |
| continue; |
| } |
| } |
| |
| // if region in cache is in recovery, exclude |
| if (profile.inRecovery) { |
| uninitialized.add(profile.getDistributedMember()); |
| continue; |
| } |
| // No need to do a GII from uninitialized member. |
| if (!profile.regionInitialized) { |
| uninitialized.add(profile.getDistributedMember()); |
| continue; |
| } |
| |
| if (profile.dataPolicy.withReplication()) { |
| if (!persistent || profile.dataPolicy.withPersistence()) { |
| // If the local member is persistent, we only want |
| // to include persistent replicas in the set of replicates. |
| replicates.add(profile.getDistributedMember()); |
| } else { |
| nonPersistent.add(profile.getDistributedMember()); |
| } |
| memberProfiles.put(profile.getDistributedMember(), profile); |
| } else if (profile.dataPolicy.isPreloaded()) { |
| preloaded.add(profile.getDistributedMember()); |
| memberProfiles.put(profile.getDistributedMember(), profile); |
| } else if (profile.dataPolicy.withStorage()) { |
| // don't bother asking proxy members for initial image |
| others.add(profile.getDistributedMember()); |
| memberProfiles.put(profile.getDistributedMember(), profile); |
| } else { |
| empties.add(profile.getDistributedMember()); |
| } |
| } |
| |
| InitialImageAdvice advice = new InitialImageAdvice(replicates, others, preloaded, empties, |
| uninitialized, nonPersistent, memberProfiles); |
| |
| if (logger.isDebugEnabled()) { |
| logger.debug(advice); |
| } |
| return advice; |
| } |
| |
| /** |
| * returns the set of all the members in the system which requires old values and are not yet |
| * finished with initialization (including GII). |
| * |
| * @since GemFire 5.5 |
| */ |
| Set adviseRequiresOldValueInCacheOp() { |
| return adviseFilter(profile -> { |
| assert profile instanceof CacheProfile; |
| CacheProfile cp = (CacheProfile) profile; |
| return cp.requiresOldValueInEvents && !cp.regionInitialized; |
| }); |
| } |
| |
| |
| // moved adviseProfileExchange to DistributionAdvisor |
| |
| // moved getProfile to DistributionAdvisor |
| |
| // moved exchangeProfiles to DistributionAdvisor |
| |
| // moved getDistributionManager to DistributionAdvisor |
| |
| /** Instantiate new distribution profile for this member */ |
| @Override |
| protected Profile instantiateProfile(InternalDistributedMember memberId, int version) { |
| return new CacheProfile(memberId, version); |
| } |
| |
| @Override |
| protected boolean evaluateProfiles(Profile newProfile, Profile oldProfile) { |
| boolean result = super.evaluateProfiles(newProfile, oldProfile); |
| if (result) { |
| CacheProfile newCP = (CacheProfile) newProfile; |
| CacheProfile oldCP = (CacheProfile) oldProfile; |
| if ((oldCP == null || !oldCP.regionInitialized) && newCP.regionInitialized) { |
| // invoke membership listeners, if any |
| CacheDistributionAdvisee advisee = (CacheDistributionAdvisee) getAdvisee(); |
| advisee.remoteRegionInitialized(newCP); |
| } |
| } |
| return result; |
| } |
| |
| /** |
| * Profile information for a remote counterpart. |
| */ |
| public static class CacheProfile extends DistributionAdvisor.Profile { |
| public DataPolicy dataPolicy = DataPolicy.REPLICATE; |
| public InterestPolicy interestPolicy = InterestPolicy.DEFAULT; |
| public boolean hasCacheLoader = false; |
| public boolean hasCacheWriter = false; |
| public boolean hasCacheListener = false; |
| public Scope scope = Scope.DISTRIBUTED_NO_ACK; |
| public boolean inRecovery = false; |
| public Set<String> gatewaySenderIds = Collections.emptySet(); |
| public Set<String> asyncEventQueueIds = Collections.emptySet(); |
| /** |
| * Will be null if the profile doesn't need to have the attributes |
| */ |
| public SubscriptionAttributes subscriptionAttributes = null; |
| |
| public boolean isPartitioned = false; |
| public boolean isGatewayEnabled = false; |
| public boolean isPersistent = false; |
| |
| public boolean isOffHeap = false; |
| |
| // moved initialMembershipVersion to DistributionAdvisor.Profile |
| // moved serialNumber to DistributionAdvisor.Profile |
| |
| /** |
| * this member's client interest / continuous query profile. This is used for event processing |
| * to reduce the number of times CQs are executed and to have the originating member for an |
| * event pay the cpu cost of executing filters on the event. |
| */ |
| public FilterProfile filterProfile; |
| |
| /** |
| * Some cache listeners require old values in cache operation messages, at least during GII |
| */ |
| boolean requiresOldValueInEvents; |
| |
| /** |
| * Whether the region has completed initialization, including GII. This information may be |
| * incorrect for a PartitionedRegion, but may be relied upon for DistributedRegions (including |
| * BucketRegions) |
| * |
| * @since GemFire prpersist this field is now overloaded for partitioned regions with |
| * persistence. In the case of pr persistence, this field indicates that the region has |
| * finished recovery from disk. |
| */ |
| public boolean regionInitialized; |
| |
| |
| /** |
| * True when a members persistent store is initialized. Note that regionInitialized may be true |
| * when this is false in the case of createBucketAtomically. With createBucketAtomically, the |
| * peristent store is not created until the EndBucketCreationMessage is sent. |
| */ |
| public boolean persistenceInitialized; |
| |
| public PersistentMemberID persistentID; |
| |
| /** |
| * This member has any cache servers. This is not actively maintained for local profiles (i.e., |
| * a profile representing this vm) |
| */ |
| public boolean hasCacheServer = false; |
| |
| List<CacheServiceProfile> cacheServiceProfiles = new ArrayList<>(); |
| |
| /** for internal use, required for DataSerializer.readObject */ |
| public CacheProfile() {} |
| |
| /** used for routing computation */ |
| public CacheProfile(FilterProfile localProfile) { |
| filterProfile = localProfile; |
| } |
| |
| public CacheProfile(InternalDistributedMember memberId, int version) { |
| super(memberId, version); |
| } |
| |
| public CacheProfile(CacheProfile toCopy) { |
| super(toCopy.getDistributedMember(), toCopy.version); |
| setIntInfo(toCopy.getIntInfo()); |
| } |
| |
| /** Return the profile data information that can be stored in an int */ |
| protected int getIntInfo() { |
| int s = 0; |
| if (dataPolicy.withReplication()) { |
| s |= REPLICATE_MASK; |
| if (dataPolicy.isPersistentReplicate()) { |
| s |= PERSISTENT_MASK; |
| } |
| } else { |
| if (dataPolicy.isEmpty()) |
| s |= PROXY_MASK; |
| if (dataPolicy.isPreloaded()) |
| s |= PRELOADED_MASK; |
| } |
| if (subscriptionAttributes != null |
| && subscriptionAttributes.getInterestPolicy().isAll()) { |
| s |= INTEREST_MASK; |
| } |
| if (hasCacheLoader) |
| s |= LOADER_MASK; |
| if (hasCacheWriter) |
| s |= WRITER_MASK; |
| if (hasCacheListener) |
| s |= LISTENER_MASK; |
| if (scope.isDistributedAck()) |
| s |= DIST_ACK_MASK; |
| if (scope.isGlobal()) |
| s |= GLOBAL_MASK; |
| if (inRecovery) |
| s |= IN_RECOVERY_MASK; |
| if (isPartitioned) |
| s |= IS_PARTITIONED_MASK; |
| if (isGatewayEnabled) |
| s |= IS_GATEWAY_ENABLED_MASK; |
| if (isPersistent) |
| s |= PERSISTENT_MASK; |
| if (regionInitialized) |
| s |= REGION_INITIALIZED_MASK; |
| if (persistentID != null) |
| s |= PERSISTENT_ID_MASK; |
| if (hasCacheServer) |
| s |= HAS_CACHE_SERVER_MASK; |
| if (requiresOldValueInEvents) |
| s |= REQUIRES_OLD_VALUE_MASK; |
| if (persistenceInitialized) |
| s |= PERSISTENCE_INITIALIZED_MASK; |
| if (!gatewaySenderIds.isEmpty()) |
| s |= GATEWAY_SENDER_IDS_MASK; |
| if (!asyncEventQueueIds.isEmpty()) |
| s |= ASYNC_EVENT_QUEUE_IDS_MASK; |
| if (isOffHeap) |
| s |= IS_OFF_HEAP_MASK; |
| if (!cacheServiceProfiles.isEmpty()) |
| s |= CACHE_SERVICE_PROFILES_MASK; |
| Assert.assertTrue(!scope.isLocal()); |
| return s; |
| } |
| |
| private boolean hasGatewaySenderIds(int bits) { |
| return (bits & GATEWAY_SENDER_IDS_MASK) != 0; |
| } |
| |
| private boolean hasAsyncEventQueueIds(int bits) { |
| return (bits & ASYNC_EVENT_QUEUE_IDS_MASK) != 0; |
| } |
| |
| /** |
| * @return true if the serialized message has a persistentID |
| */ |
| private boolean hasPersistentID(int bits) { |
| return (bits & PERSISTENT_ID_MASK) != 0; |
| } |
| |
| public boolean isPersistent() { |
| return dataPolicy.withPersistence(); |
| } |
| |
| /** Set the profile data information that is stored in a short */ |
| protected void setIntInfo(int s) { |
| if ((s & REPLICATE_MASK) != 0) { |
| if ((s & PERSISTENT_MASK) != 0) { |
| dataPolicy = DataPolicy.PERSISTENT_REPLICATE; |
| } else { |
| dataPolicy = DataPolicy.REPLICATE; |
| } |
| } else if ((s & PROXY_MASK) != 0) { |
| dataPolicy = DataPolicy.EMPTY; |
| } else if ((s & PRELOADED_MASK) != 0) { |
| dataPolicy = DataPolicy.PRELOADED; |
| } else { // CACHED |
| dataPolicy = DataPolicy.NORMAL; |
| } |
| |
| if ((s & IS_PARTITIONED_MASK) != 0) { |
| if ((s & PERSISTENT_MASK) != 0) { |
| dataPolicy = DataPolicy.PERSISTENT_PARTITION; |
| } else { |
| dataPolicy = DataPolicy.PARTITION; |
| } |
| } |
| |
| if ((s & INTEREST_MASK) != 0) { |
| subscriptionAttributes = new SubscriptionAttributes(InterestPolicy.ALL); |
| } else { |
| subscriptionAttributes = new SubscriptionAttributes(InterestPolicy.CACHE_CONTENT); |
| } |
| hasCacheLoader = (s & LOADER_MASK) != 0; |
| hasCacheWriter = (s & WRITER_MASK) != 0; |
| hasCacheListener = (s & LISTENER_MASK) != 0; |
| scope = (s & DIST_ACK_MASK) != 0 ? Scope.DISTRIBUTED_ACK |
| : ((s & GLOBAL_MASK) != 0 ? Scope.GLOBAL : Scope.DISTRIBUTED_NO_ACK); |
| inRecovery = (s & IN_RECOVERY_MASK) != 0; |
| isPartitioned = (s & IS_PARTITIONED_MASK) != 0; |
| isGatewayEnabled = (s & IS_GATEWAY_ENABLED_MASK) != 0; |
| isPersistent = (s & PERSISTENT_MASK) != 0; |
| regionInitialized = ((s & REGION_INITIALIZED_MASK) != 0); |
| hasCacheServer = ((s & HAS_CACHE_SERVER_MASK) != 0); |
| requiresOldValueInEvents = ((s & REQUIRES_OLD_VALUE_MASK) != 0); |
| persistenceInitialized = (s & PERSISTENCE_INITIALIZED_MASK) != 0; |
| isOffHeap = (s & IS_OFF_HEAP_MASK) != 0; |
| } |
| |
| /** |
| * Sets the SubscriptionAttributes for the region that this profile is on |
| * |
| * @since GemFire 5.0 |
| */ |
| public void setSubscriptionAttributes(SubscriptionAttributes sa) { |
| subscriptionAttributes = sa; |
| } |
| |
| /** |
| * Return true if cached or allEvents and a listener |
| */ |
| boolean cachedOrAllEventsWithListener() { |
| // to fix bug 36804 to ignore hasCacheListener |
| // return this.dataPolicy.withStorage() || |
| // (allEvents() && this.hasCacheListener); |
| return cachedOrAllEvents(); |
| } |
| |
| /** |
| * Return true if cached or allEvents |
| */ |
| boolean cachedOrAllEvents() { |
| return dataPolicy.withStorage() || allEvents(); |
| } |
| |
| /** |
| * Return true if subscribed to all events |
| */ |
| boolean allEvents() { |
| return subscriptionAttributes.getInterestPolicy().isAll(); |
| } |
| |
| public void addCacheServiceProfile(CacheServiceProfile profile) { |
| cacheServiceProfiles.add(profile); |
| } |
| |
| private boolean hasCacheServiceProfiles(int bits) { |
| return (bits & CACHE_SERVICE_PROFILES_MASK) != 0; |
| } |
| |
| /** |
| * Used to process an incoming cache profile. |
| */ |
| @Override |
| public void processIncoming(ClusterDistributionManager dm, String adviseePath, |
| boolean removeProfile, boolean exchangeProfiles, final List<Profile> replyProfiles) { |
| try { |
| Assert.assertTrue(adviseePath != null, "adviseePath was null"); |
| |
| InternalRegion lclRgn; |
| final InitializationLevel oldLevel = LocalRegion.setThreadInitLevelRequirement(ANY_INIT); |
| try { |
| InternalCache cache = dm.getCache(); |
| lclRgn = cache == null ? null : cache.getRegionByPath(adviseePath); |
| } finally { |
| LocalRegion.setThreadInitLevelRequirement(oldLevel); |
| } |
| |
| if (lclRgn instanceof CacheDistributionAdvisee) { |
| if (lclRgn.isUsedForPartitionedRegionBucket()) { |
| if (!((BucketRegion) lclRgn).isPartitionedRegionOpen()) { |
| return; |
| } |
| } |
| handleCacheDistributionAdvisee((CacheDistributionAdvisee) lclRgn, adviseePath, |
| removeProfile, exchangeProfiles, true, replyProfiles); |
| } else { |
| if (lclRgn == null) { |
| handleCacheDistributionAdvisee( |
| PartitionedRegionHelper.getProxyBucketRegion(dm.getCache(), adviseePath, false), |
| adviseePath, removeProfile, exchangeProfiles, false, replyProfiles); |
| } else { |
| if (logger.isDebugEnabled()) { |
| logger.debug("While processing UpdateAttributes message, region has local scope: {}", |
| adviseePath); |
| } |
| } |
| } |
| } catch (PRLocallyDestroyedException fre) { |
| if (logger.isDebugEnabled()) { |
| logger.debug("<Region Locally destroyed> /// {}", this); |
| } |
| } catch (RegionDestroyedException e) { |
| if (logger.isDebugEnabled()) { |
| logger.debug("<region destroyed> /// {}", this); |
| } |
| } |
| } |
| |
| @Override |
| public void cleanUp() { |
| if (filterProfile != null) { |
| filterProfile.cleanUp(); |
| } |
| } |
| |
| /** |
| * Attempts to process this message with the specified <code>CacheDistributionAdvisee</code>. |
| * |
| * @param cda the CacheDistributionAdvisee to apply this profile to |
| * @param isRealRegion true if CacheDistributionAdvisee is a real region |
| */ |
| private void handleCacheDistributionAdvisee(CacheDistributionAdvisee cda, String adviseePath, |
| boolean removeProfile, boolean exchangeProfiles, boolean isRealRegion, |
| final List<Profile> replyProfiles) { |
| if (cda != null) { |
| handleDistributionAdvisee(cda, removeProfile, isRealRegion && exchangeProfiles, |
| replyProfiles); |
| if (logger.isDebugEnabled()) { |
| logger.debug("While processing UpdateAttributes message, handled advisee: {}", cda); |
| } |
| } else { |
| if (logger.isDebugEnabled()) { |
| logger.debug("While processing UpdateAttributes message, region not found: {}", |
| adviseePath); |
| } |
| } |
| } |
| |
| @Override |
| public int getDSFID() { |
| return CACHE_PROFILE; |
| } |
| |
| @Override |
| public void toData(DataOutput out) throws IOException { |
| super.toData(out); |
| out.writeInt(getIntInfo()); |
| if (persistentID != null) { |
| InternalDataSerializer.invokeToData(persistentID, out); |
| } |
| if (!gatewaySenderIds.isEmpty()) { |
| writeSet(gatewaySenderIds, out); |
| } |
| if (!asyncEventQueueIds.isEmpty()) { |
| writeSet(asyncEventQueueIds, out); |
| } |
| DataSerializer.writeObject(filterProfile, out); |
| if (!cacheServiceProfiles.isEmpty()) { |
| DataSerializer.writeObject(cacheServiceProfiles, out); |
| } |
| } |
| |
| private void writeSet(Set<?> set, DataOutput out) throws IOException { |
| // to fix bug 47205 always serialize the Set as a HashSet. |
| out.writeByte(DSCODE.HASH_SET.toByte()); |
| InternalDataSerializer.writeSet(set, out); |
| } |
| |
| @Override |
| public void fromData(DataInput in) throws IOException, ClassNotFoundException { |
| super.fromData(in); |
| int bits = in.readInt(); |
| setIntInfo(bits); |
| if (hasPersistentID(bits)) { |
| persistentID = new PersistentMemberID(); |
| InternalDataSerializer.invokeFromData(persistentID, in); |
| } |
| if (hasGatewaySenderIds(bits)) { |
| gatewaySenderIds = DataSerializer.readObject(in); |
| } |
| if (hasAsyncEventQueueIds(bits)) { |
| asyncEventQueueIds = DataSerializer.readObject(in); |
| } |
| filterProfile = DataSerializer.readObject(in); |
| if (hasCacheServiceProfiles(bits)) { |
| cacheServiceProfiles = DataSerializer.readObject(in); |
| } |
| } |
| |
| @Override |
| public StringBuilder getToStringHeader() { |
| return new StringBuilder("CacheProfile"); |
| } |
| |
| @Override |
| public void fillInToString(StringBuilder sb) { |
| super.fillInToString(sb); |
| sb.append("; dataPolicy=").append(dataPolicy); |
| sb.append("; hasCacheLoader=").append(hasCacheLoader); |
| sb.append("; hasCacheWriter=").append(hasCacheWriter); |
| sb.append("; hasCacheListener=").append(hasCacheListener); |
| sb.append("; hasCacheServer=").append(hasCacheServer); |
| sb.append("; scope=").append(scope); |
| sb.append("; regionInitialized=").append(regionInitialized); |
| sb.append("; inRecovery=").append(inRecovery); |
| sb.append("; subcription=").append(subscriptionAttributes); |
| sb.append("; isPartitioned=").append(isPartitioned); |
| sb.append("; isGatewayEnabled=").append(isGatewayEnabled); |
| sb.append("; isPersistent=").append(isPersistent); |
| sb.append("; persistentID=").append(persistentID); |
| if (filterProfile != null) { |
| sb.append("; ").append(filterProfile); |
| } |
| sb.append("; gatewaySenderIds =").append(gatewaySenderIds); |
| sb.append("; asyncEventQueueIds =").append(asyncEventQueueIds); |
| sb.append("; IsOffHeap=").append(isOffHeap); |
| sb.append("; cacheServiceProfiles=").append(cacheServiceProfiles); |
| } |
| } |
| |
| /** Recipient information used for getInitialImage operation */ |
| public static class InitialImageAdvice { |
| public Set<InternalDistributedMember> getOthers() { |
| return others; |
| } |
| |
| public void setOthers(Set<InternalDistributedMember> others) { |
| this.others = others; |
| } |
| |
| public Set<InternalDistributedMember> getReplicates() { |
| return replicates; |
| } |
| |
| public Set<InternalDistributedMember> getNonPersistent() { |
| return nonPersistent; |
| } |
| |
| public Set<InternalDistributedMember> getPreloaded() { |
| return preloaded; |
| } |
| |
| public Set<InternalDistributedMember> getEmpties() { |
| return empties; |
| } |
| |
| public Set<InternalDistributedMember> getUninitialized() { |
| return uninitialized; |
| } |
| |
| /** Set of replicate recipients */ |
| protected final Set<InternalDistributedMember> replicates; |
| |
| /** Set of peers that are preloaded */ |
| protected final Set<InternalDistributedMember> preloaded; |
| |
| /** |
| * Set of tertiary recipients which are not replicates, in which case they should all be queried |
| * and a superset taken of their images. To be used only if the image cannot be obtained from |
| * the replicates set. |
| */ |
| protected Set<InternalDistributedMember> others; |
| |
| /** Set of members that might be data feeds and have EMPTY data policy */ |
| protected final Set<InternalDistributedMember> empties; |
| |
| /** Set of members that may not have finished initializing their caches */ |
| protected final Set<InternalDistributedMember> uninitialized; |
| |
| /** Set of members that are replicates but not persistent */ |
| final Set<InternalDistributedMember> nonPersistent; |
| |
| private final Map<InternalDistributedMember, CacheProfile> memberProfiles; |
| |
| |
| public InitialImageAdvice(Set<InternalDistributedMember> replicates, |
| Set<InternalDistributedMember> others, Set<InternalDistributedMember> preloaded, |
| Set<InternalDistributedMember> empties, Set<InternalDistributedMember> uninitialized, |
| Set<InternalDistributedMember> nonPersistent, |
| Map<InternalDistributedMember, CacheProfile> memberProfiles) { |
| this.replicates = replicates; |
| this.others = others; |
| this.preloaded = preloaded; |
| this.empties = empties; |
| this.uninitialized = uninitialized; |
| this.nonPersistent = nonPersistent; |
| this.memberProfiles = memberProfiles; |
| } |
| |
| public InitialImageAdvice() { |
| this(Collections.emptySet(), Collections.emptySet(), Collections.emptySet(), |
| Collections.emptySet(), Collections.emptySet(), Collections.emptySet(), |
| Collections.emptyMap()); |
| } |
| |
| @Override |
| public String toString() { |
| return "InitialImageAdvice(" + "replicates=" + replicates + "; others=" + others |
| + "; preloaded=" + preloaded + "; empty=" + empties + "; initializing=" |
| + uninitialized + ")"; |
| } |
| |
| } |
| |
| /** |
| * Provide all the replicates including persistent replicates. |
| * |
| * @return the set of replicate's memberIds |
| * @since GemFire 5.8 |
| */ |
| public Set<InternalDistributedMember> adviseReplicates() { |
| return adviseFilter(profile -> { |
| assert profile instanceof CacheProfile; |
| CacheProfile cp = (CacheProfile) profile; |
| return cp.dataPolicy.withReplication(); |
| }); |
| } |
| |
| /** |
| * Provide only the preloadeds given a set of existing memberIds |
| * |
| * @return the set of preloaded's memberIds |
| * @since GemFire prPersistSprint1 |
| */ |
| public Set advisePreloadeds() { |
| return adviseFilter(profile -> { |
| assert profile instanceof CacheProfile; |
| CacheProfile cp = (CacheProfile) profile; |
| return cp.dataPolicy.withPreloaded(); |
| }); |
| } |
| |
| /** |
| * Provide only the empty's (having DataPolicy.EMPTY) given a set of existing memberIds |
| * |
| * @return the set of replicate's memberIds |
| * @since GemFire 5.8 |
| */ |
| Set adviseEmptys() { |
| return adviseFilter(profile -> { |
| assert profile instanceof CacheProfile; |
| CacheProfile cp = (CacheProfile) profile; |
| return cp.dataPolicy.isEmpty(); |
| }); |
| } |
| |
| @Override |
| protected void profileRemoved(Profile profile) { |
| if (logger.isDebugEnabled()) { |
| logger.debug("CDA: removing profile {}", profile); |
| } |
| if (getAdvisee() instanceof LocalRegion && profile != null) { |
| ((LocalRegion) getAdvisee()).removeCriticalMember(profile.getDistributedMember()); |
| } |
| } |
| |
| /** |
| * Returns the list of all persistent members. For most cases, adviseInitializedPersistentMembers |
| * is more appropriate. These method includes members that are still in the process of GII. |
| */ |
| public Map<InternalDistributedMember, PersistentMemberID> advisePersistentMembers() { |
| initializationGate(); |
| |
| Map<InternalDistributedMember, PersistentMemberID> result = |
| new HashMap<>(); |
| Profile[] snapshot = profiles; |
| for (Profile profile : snapshot) { |
| CacheProfile cp = (CacheProfile) profile; |
| if (cp.persistentID != null) { |
| result.put(cp.getDistributedMember(), cp.persistentID); |
| } |
| } |
| |
| return result; |
| } |
| |
| public Map<InternalDistributedMember, PersistentMemberID> adviseInitializedPersistentMembers() { |
| initializationGate(); |
| |
| Map<InternalDistributedMember, PersistentMemberID> result = |
| new HashMap<>(); |
| Profile[] snapshot = profiles; |
| for (Profile profile : snapshot) { |
| CacheProfile cp = (CacheProfile) profile; |
| if (cp.persistentID != null && cp.persistenceInitialized) { |
| result.put(cp.getDistributedMember(), cp.persistentID); |
| } |
| } |
| |
| return result; |
| } |
| |
| Set adviseCacheServers() { |
| getAdvisee().getCancelCriterion().checkCancelInProgress(null); |
| return adviseFilter(profile -> { |
| assert profile instanceof CacheProfile; |
| CacheProfile cp = (CacheProfile) profile; |
| return cp.hasCacheServer; |
| }); |
| } |
| |
| // Overrided for bucket regions. This listener also receives events |
| // about PR joins and leaves. |
| public void addMembershipAndProxyListener(MembershipListener listener) { |
| addMembershipListener(listener); |
| } |
| |
| public void removeMembershipAndProxyListener(MembershipListener listener) { |
| removeMembershipListener(listener); |
| } |
| |
| @Override |
| public boolean removeId(ProfileId memberId, boolean crashed, boolean destroyed, |
| boolean fromMembershipListener) { |
| |
| boolean isPersistent = false; |
| DiskStoreID persistentId = null; |
| CacheDistributionAdvisee advisee = (CacheDistributionAdvisee) getAdvisee(); |
| if (advisee.getAttributes().getDataPolicy().withPersistence()) { |
| isPersistent = true; |
| CacheProfile profile = (CacheProfile) getProfile(memberId); |
| if (profile != null && profile.persistentID != null) { |
| persistentId = ((CacheProfile) getProfile(memberId)).persistentID.getDiskStoreId(); |
| } |
| } |
| |
| boolean result = super.removeId(memberId, crashed, destroyed, fromMembershipListener); |
| |
| // bug #48962 - record members that leave during GII so IIOp knows about them |
| if (advisee instanceof DistributedRegion) { |
| DistributedRegion r = (DistributedRegion) advisee; |
| if (!r.isInitialized() && !r.isUsedForPartitionedRegionBucket()) { |
| if (logger.isDebugEnabled()) { |
| logger.debug("recording that {} has left during initialization of {}", memberId, |
| r.getName()); |
| } |
| ImageState state = r.getImageState(); |
| if (isPersistent) { |
| if (persistentId != null) { |
| state.addLeftMember(persistentId); |
| } |
| } else { |
| state.addLeftMember((InternalDistributedMember) memberId); |
| } |
| } |
| } |
| return result; |
| } |
| } |
