Google Git
Sign in
apache / geode / 7038f08ea88ec42f6c58764525460a1c7fe1abcf / . / geode-core / src / main / java / org / apache / geode / internal / cache / DistributedRegion.java
blob: a89949dad71ff2d8e63eabdb2bf663ae3abd7658 [file] [log] [blame]
/*
* 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 java.io.IOException;
import java.io.InputStream;
import java.util.ArrayList;
import java.util.Collection;
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.Set;
import java.util.TreeMap;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantReadWriteLock;
import org.apache.logging.log4j.Logger;
import org.apache.geode.CancelException;
import org.apache.geode.InternalGemFireError;
import org.apache.geode.InvalidDeltaException;
import org.apache.geode.SystemFailure;
import org.apache.geode.annotations.VisibleForTesting;
import org.apache.geode.annotations.internal.MakeNotStatic;
import org.apache.geode.annotations.internal.MutableForTesting;
import org.apache.geode.cache.CacheClosedException;
import org.apache.geode.cache.CacheEvent;
import org.apache.geode.cache.CacheListener;
import org.apache.geode.cache.CacheLoader;
import org.apache.geode.cache.CacheLoaderException;
import org.apache.geode.cache.CacheWriter;
import org.apache.geode.cache.CacheWriterException;
import org.apache.geode.cache.DataPolicy;
import org.apache.geode.cache.DiskAccessException;
import org.apache.geode.cache.EntryNotFoundException;
import org.apache.geode.cache.LossAction;
import org.apache.geode.cache.MembershipAttributes;
import org.apache.geode.cache.Operation;
import org.apache.geode.cache.RegionAccessException;
import org.apache.geode.cache.RegionAttributes;
import org.apache.geode.cache.RegionDestroyedException;
import org.apache.geode.cache.RegionDistributionException;
import org.apache.geode.cache.RegionMembershipListener;
import org.apache.geode.cache.ResumptionAction;
import org.apache.geode.cache.RoleException;
import org.apache.geode.cache.TimeoutException;
import org.apache.geode.cache.asyncqueue.internal.AsyncEventQueueImpl;
import org.apache.geode.cache.execute.Function;
import org.apache.geode.cache.execute.FunctionException;
import org.apache.geode.cache.execute.ResultCollector;
import org.apache.geode.cache.execute.ResultSender;
import org.apache.geode.cache.persistence.PersistentReplicatesOfflineException;
import org.apache.geode.cache.wan.GatewaySender;
import org.apache.geode.distributed.DistributedLockService;
import org.apache.geode.distributed.DistributedMember;
import org.apache.geode.distributed.LockServiceDestroyedException;
import org.apache.geode.distributed.Role;
import org.apache.geode.distributed.internal.DistributionAdvisee;
import org.apache.geode.distributed.internal.DistributionAdvisor;
import org.apache.geode.distributed.internal.DistributionAdvisor.Profile;
import org.apache.geode.distributed.internal.DistributionAdvisor.ProfileVisitor;
import org.apache.geode.distributed.internal.DistributionConfig;
import org.apache.geode.distributed.internal.DistributionManager;
import org.apache.geode.distributed.internal.MembershipListener;
import org.apache.geode.distributed.internal.ReplyProcessor21;
import org.apache.geode.distributed.internal.locks.DLockRemoteToken;
import org.apache.geode.distributed.internal.locks.DLockService;
import org.apache.geode.distributed.internal.membership.InternalDistributedMember;
import org.apache.geode.internal.Assert;
import org.apache.geode.internal.SystemTimer;
import org.apache.geode.internal.cache.CacheDistributionAdvisor.CacheProfile;
import org.apache.geode.internal.cache.InitialImageOperation.GIIStatus;
import org.apache.geode.internal.cache.RegionMap.ARMLockTestHook;
import org.apache.geode.internal.cache.control.InternalResourceManager.ResourceType;
import org.apache.geode.internal.cache.control.MemoryEvent;
import org.apache.geode.internal.cache.event.DistributedEventTracker;
import org.apache.geode.internal.cache.event.EventTracker;
import org.apache.geode.internal.cache.eviction.EvictableEntry;
import org.apache.geode.internal.cache.execute.DistributedRegionFunctionExecutor;
import org.apache.geode.internal.cache.execute.DistributedRegionFunctionResultSender;
import org.apache.geode.internal.cache.execute.DistributedRegionFunctionResultWaiter;
import org.apache.geode.internal.cache.execute.LocalResultCollector;
import org.apache.geode.internal.cache.execute.RegionFunctionContextImpl;
import org.apache.geode.internal.cache.execute.ServerToClientFunctionResultSender;
import org.apache.geode.internal.cache.execute.metrics.FunctionStats;
import org.apache.geode.internal.cache.execute.metrics.FunctionStatsManager;
import org.apache.geode.internal.cache.persistence.CreatePersistentRegionProcessor;
import org.apache.geode.internal.cache.persistence.PersistenceAdvisor;
import org.apache.geode.internal.cache.persistence.PersistenceAdvisorImpl;
import org.apache.geode.internal.cache.persistence.PersistentMemberID;
import org.apache.geode.internal.cache.persistence.PersistentMemberManager;
import org.apache.geode.internal.cache.persistence.PersistentMemberView;
import org.apache.geode.internal.cache.tier.sockets.ClientProxyMembershipID;
import org.apache.geode.internal.cache.tier.sockets.VersionedObjectList;
import org.apache.geode.internal.cache.tx.RemoteClearMessage;
import org.apache.geode.internal.cache.tx.RemoteDestroyMessage;
import org.apache.geode.internal.cache.tx.RemoteFetchVersionMessage;
import org.apache.geode.internal.cache.tx.RemoteFetchVersionMessage.FetchVersionResponse;
import org.apache.geode.internal.cache.tx.RemoteInvalidateMessage;
import org.apache.geode.internal.cache.tx.RemotePutMessage;
import org.apache.geode.internal.cache.versions.ConcurrentCacheModificationException;
import org.apache.geode.internal.cache.versions.RegionVersionHolder;
import org.apache.geode.internal.cache.versions.RegionVersionVector;
import org.apache.geode.internal.cache.versions.VersionSource;
import org.apache.geode.internal.cache.versions.VersionTag;
import org.apache.geode.internal.cache.wan.AsyncEventQueueConfigurationException;
import org.apache.geode.internal.cache.wan.GatewaySenderConfigurationException;
import org.apache.geode.internal.offheap.annotations.Released;
import org.apache.geode.internal.offheap.annotations.Retained;
import org.apache.geode.internal.sequencelog.RegionLogger;
import org.apache.geode.internal.statistics.StatisticsClock;
import org.apache.geode.internal.util.concurrent.StoppableCountDownLatch;
import org.apache.geode.logging.internal.executors.LoggingThread;
import org.apache.geode.logging.internal.log4j.api.LogService;
@SuppressWarnings("deprecation")
public class DistributedRegion extends LocalRegion implements InternalDistributedRegion {
private static final Logger logger = LogService.getLogger();
/** causes cache profile to be added to afterRemoteRegionCreate notification for testing */
@MutableForTesting
public static boolean TEST_HOOK_ADD_PROFILE = false;
/** Used to sync accesses to this.dlockService to allow lazy construction */
private final Object dlockMonitor = new Object();
final CacheDistributionAdvisor distAdvisor;
private final SenderIdMonitor senderIdMonitor;
/**
* GuardedBy {@link #dlockMonitor}
*/
private DistributedLockService dlockService;
final AdvisorListener advisorListener = new AdvisorListener();
/** Set of currently missing required roles */
private final HashSet<Role> missingRequiredRoles = new HashSet<>();
/** True if this region is currently missing any required roles */
private volatile boolean isMissingRequiredRoles = false;
/**
* True if this region is has any required roles defined and the LossAction is either NO_ACCESS or
* LIMITED_ACCESS. Reliability checks will only happen if this is true.
*/
private final boolean requiresReliabilityCheck;
/**
* Latch that is opened after initialization waits for required roles up to the
* <a href="DistributedSystem#member-timeout">member-timeout </a>.
*/
private final StoppableCountDownLatch initializationLatchAfterMemberTimeout;
private final PersistenceAdvisor persistenceAdvisor;
private final PersistentMemberID persistentId;
/**
* This boolean is set to false when this region is non-persistent, but there are persistent
* members in the distributed system to which all region modifications should be forwarded see bug
* 45186
*/
private volatile boolean generateVersionTag = true;
/** Tests can set this to true and ignore reliability triggered reconnects */
@MutableForTesting
public static boolean ignoreReconnect = false;
/**
* Lock to prevent multiple threads on this member from performing a clear at the same time.
*/
private final Object clearLock = new Object();
private final ReentrantReadWriteLock failedInitialImageLock = new ReentrantReadWriteLock(true);
@MakeNotStatic
private static final AtomicBoolean loggedNetworkPartitionWarning = new AtomicBoolean(false);
/** Creates a new instance of DistributedRegion */
protected DistributedRegion(String regionName, RegionAttributes attrs, LocalRegion parentRegion,
InternalCache cache, InternalRegionArguments internalRegionArgs,
StatisticsClock statisticsClock) {
super(regionName, attrs, parentRegion, cache, internalRegionArgs, statisticsClock);
initializationLatchAfterMemberTimeout =
new StoppableCountDownLatch(getCancelCriterion(), 1);
distAdvisor = createDistributionAdvisor(internalRegionArgs);
senderIdMonitor = createSenderIdMonitor();
if (getDistributionManager().getConfig().getEnableNetworkPartitionDetection()
&& !isInternalRegion() && !attrs.getScope().isAck() && !doesNotDistribute()
&& attrs.getDataPolicy().withStorage()) {
logger.warn(
"Region {} is being created with scope {} but enable-network-partition-detection is enabled in the distributed system. This can lead to cache inconsistencies if there is a network failure.",
new Object[] {regionName, attrs.getScope()});
}
if (!getDistributionManager().getConfig().getEnableNetworkPartitionDetection()
&& attrs.getDataPolicy().withPersistence()
&& !loggedNetworkPartitionWarning.getAndSet(true)) {
logger.warn(
"Creating persistent region {}, but enable-network-partition-detection is set to false. Running with network partition detection disabled can lead to an unrecoverable system in the event of a network split.",
regionName);
}
boolean setRequiresReliabilityCheck = attrs.getMembershipAttributes().hasRequiredRoles() &&
// note that the following includes NO_ACCESS, LIMITED_ACCESS,
!attrs.getMembershipAttributes().getLossAction().isAllAccess()
&& !attrs.getMembershipAttributes().getLossAction().isReconnect();
// this optimization is safe for as long as Roles and Required Roles are
// immutable
// if this VM fulfills all required roles, make requiresReliabilityCheck
// false
Set<Role> reqRoles = new HashSet<>(attrs.getMembershipAttributes().getRequiredRoles());
reqRoles.removeAll(getSystem().getDistributedMember().getRoles());
if (reqRoles.isEmpty()) {
setRequiresReliabilityCheck = false;
}
requiresReliabilityCheck = setRequiresReliabilityCheck;
if (internalRegionArgs.isUsedForPartitionedRegionBucket()) {
persistenceAdvisor = internalRegionArgs.getPersistenceAdvisor();
} else if (allowsPersistence()) {
// TODO: prpersist - using this lock service is a hack. Maybe? Or maybe
// it's ok if we have one (rarely used) lock service for many operations?
// What does the resource manager do?
DistributedLockService dl = cache.getPartitionedRegionLockService();
try {
// TODO: prpersist - this is just a quick and dirty storage mechanism so that
// I can test the storage.
DiskRegionStats diskStats;
PersistentMemberView storage;
if (getDataPolicy().withPersistence()) {
storage = getDiskRegion();
diskStats = getDiskRegion().getStats();
} else {
storage = new InMemoryPersistentMemberView();
diskStats = null;
}
PersistentMemberManager memberManager = cache.getPersistentMemberManager();
persistenceAdvisor = new PersistenceAdvisorImpl(distAdvisor, dl, storage,
getFullPath(), diskStats, memberManager);
} catch (Exception ignore) {
throw new InternalGemFireError("Couldn't recover persistence");
}
} else {
persistenceAdvisor = null;
}
if (persistenceAdvisor != null) {
persistentId = persistenceAdvisor.generatePersistentID();
} else {
persistentId = null;
}
}
@Override
protected EventTracker createEventTracker() {
EventTracker tracker = new DistributedEventTracker(cache, getCancelCriterion(), getName());
tracker.start();
return tracker;
}
/**
* Record the event state from image provider
*
* @param provider the member that provided the initial image and event state
*/
protected void recordEventStateFromImageProvider(InternalDistributedMember provider) {
// No Op. Only Bucket region will initiate event states
}
/**
* Intended for used during construction of a DistributedRegion
*
* @return the advisor to be used by the region
*/
protected CacheDistributionAdvisor createDistributionAdvisor(
InternalRegionArguments internalRegionArgs) {
// Warning: potential early escape of object before full construction
return CacheDistributionAdvisor.createCacheDistributionAdvisor(this);
}
/**
* Does this region support persistence?
*/
public boolean allowsPersistence() {
return true;
}
@Override
boolean requiresOneHopForMissingEntry(EntryEventImpl event) {
// received from another member - don't use one-hop
if (event.isOriginRemote()) {
return false;
}
// local ops aren't distributed
if (event.getOperation().isLocal()) {
return false;
}
// if it already has a valid version tag it can go out with a DistributedCacheOperation
if (event.getVersionTag() != null && event.getVersionTag().getRegionVersion() > 0) {
return false;
}
// if we're not allowed to generate a version tag we need to send it to someone who can
if (!generateVersionTag) {
return true;
}
return getConcurrencyChecksEnabled() && (getServerProxy() == null) && !isTX()
&& scope.isDistributed() && !getDataPolicy().withReplication();
}
@Override
public boolean virtualPut(EntryEventImpl event, boolean ifNew, boolean ifOld,
Object expectedOldValue, boolean requireOldValue, long lastModified,
boolean overwriteDestroyed) throws TimeoutException, CacheWriterException {
final boolean isTraceEnabled = logger.isTraceEnabled();
Lock dlock = null;
if (scope.isGlobal() && // lock only applies to global scope
!event.isOriginRemote() && // only if operation originating locally
!event.isNetSearch() && // search and load processor handles own locking
!event.isNetLoad() &&
// @todo darrel/kirk: what about putAll?
!event.isLocalLoad() && !event.isSingleHopPutOp()) {
// Single Hop Op means dlock is already taken at origin node.
dlock = getDistributedLockIfGlobal(event.getKey());
}
if (isTraceEnabled) {
logger.trace("virtualPut invoked for event {}", event);
}
try {
if (!hasSeenEvent(event)) {
if (requiresOneHopForMissingEntry(event)) {
// bug #45704: see if a one-hop must be done for this operation
RegionEntry re = getRegionEntry(event.getKey());
if (re == null /* || re.isTombstone() */ || !generateVersionTag) {
if (!event.isBulkOpInProgress() || getDataPolicy().withStorage()) {
// putAll will send a single one-hop for empty regions. for other missing entries
// we need to get a valid version number before modifying the local cache
boolean didDistribute = RemotePutMessage.distribute(event, lastModified, false, false,
expectedOldValue, requireOldValue, !generateVersionTag);
if (!didDistribute && isTraceEnabled) {
logger.trace("Unable to perform one-hop messaging");
}
if (!generateVersionTag && !didDistribute) {
throw new PersistentReplicatesOfflineException();
}
if (didDistribute) {
if (isTraceEnabled) {
logger.trace("Event after remotePut operation: {}", event);
}
if (event.getVersionTag() == null) {
// if the event wasn't applied by the one-hop replicate it will not have a version
// tag
// and so should not be applied to this cache
return false;
}
}
}
}
}
return super.virtualPut(event, ifNew, ifOld, expectedOldValue, requireOldValue,
lastModified, overwriteDestroyed);
} else {
if (event.getDeltaBytes() != null && event.getRawNewValue() == null) {
// This means that this event has delta bytes but no full value.
// Request the full value of this event.
// The value in this vm may not be same as this event's value.
throw new InvalidDeltaException(
"Cache encountered replay of event containing delta bytes for key " + event.getKey());
}
// if the listeners have already seen this event, then it has already
// been successfully applied to the cache. Distributed messages and
// return
if (logger.isDebugEnabled()) {
logger.debug("DR.virtualPut: this cache has already seen this event {}", event);
}
// Fix 39014: when hasSeenEvent, put will still distribute
// event, but putAll did not. We add the logic back here, not to put
// back into DR.distributeUpdate() because we moved this part up into
// LR.basicPutPart3 in purpose.
if (event.isBulkOpInProgress() && !event.isOriginRemote()) {
event.getPutAllOperation().addEntry(event, true);
}
/*
* doing this so that other VMs will apply this no matter what. If it is an "update" they
* will not apply it if they don't have the key. Because this is probably a retry, it will
* never get applied to this local AbstractRegionMap, and so will never be flipped to a
* 'create'
*/
event.makeCreate();
if (!getConcurrencyChecksEnabled() || event.hasValidVersionTag()) {
distributeUpdate(event, lastModified, ifNew, ifOld, expectedOldValue, requireOldValue);
event.invokeCallbacks(this, true, true);
}
return true;
}
} finally {
if (dlock != null) {
dlock.unlock();
}
}
}
@Override
RegionEntry basicPutEntry(EntryEventImpl event, long lastModified)
throws TimeoutException, CacheWriterException {
final boolean isTraceEnabled = logger.isTraceEnabled();
if (isTraceEnabled) {
logger.trace("basicPutEntry invoked for event {}", event);
}
if (requiresOneHopForMissingEntry(event)) {
// bug #45704: see if a one-hop must be done for this operation
RegionEntry re = getRegionEntry(event.getKey());
if (re == null /* || re.isTombstone() */ || !generateVersionTag) {
final boolean ifNew = false;
final boolean ifOld = false;
boolean didDistribute = RemotePutMessage.distribute(event, lastModified, ifNew, ifOld, null,
false, !generateVersionTag);
if (!generateVersionTag && !didDistribute) {
throw new PersistentReplicatesOfflineException();
}
if (didDistribute && isTraceEnabled) {
logger.trace("Event after remotePut for basicPutEntry: {}", event);
}
}
}
return super.basicPutEntry(event, lastModified);
}
@Override
void performPutAllEntry(EntryEventImpl event) {
/*
* force shared data view so that we just do the virtual op, accruing things in the put all
* operation for later
*/
if (isTX()) {
event.getPutAllOperation().addEntry(event);
} else {
getSharedDataView().putEntry(event, false, false, null, false, 0L, false);
}
}
@Override
void performRemoveAllEntry(EntryEventImpl event) {
// force shared data view so that we just do the virtual op, accruing things in the bulk
// operation for later
if (isTX()) {
event.getRemoveAllOperation().addEntry(event);
} else {
basicDestroy(event, true, null);
}
}
/**
* distribution and listener notification
*/
@Override
public void basicPutPart3(EntryEventImpl event, RegionEntry entry, boolean isInitialized,
long lastModified, boolean invokeCallbacks, boolean ifNew, boolean ifOld,
Object expectedOldValue, boolean requireOldValue) {
distributeUpdate(event, lastModified, false, false, null, false);
super.basicPutPart3(event, entry, isInitialized, lastModified, invokeCallbacks, ifNew, ifOld,
expectedOldValue, requireOldValue);
}
/** distribute an update operation */
protected void distributeUpdate(EntryEventImpl event, long lastModified, boolean ifNew,
boolean ifOld, Object expectedOldValue, boolean requireOldValue) {
// an update from a netSearch is not distributed
if (!event.isOriginRemote() && !event.isNetSearch() && !event.isBulkOpInProgress()) {
boolean distribute = true;
if (event.getInhibitDistribution()) {
// this has already been distributed by a one-hop operation
distribute = false;
}
if (distribute) {
// before distribute: DR's put, it has notified gateway sender earlier
UpdateOperation op = new UpdateOperation(event, lastModified);
if (logger.isTraceEnabled()) {
logger.trace("distributing operation for event : {} : for region : {}", event,
getName());
}
op.distribute();
}
}
}
@Override
public boolean hasSeenEvent(EntryEventImpl event) {
boolean isDuplicate;
isDuplicate = getEventTracker().hasSeenEvent(event);
if (isDuplicate) {
markEventAsDuplicate(event);
} else {
// bug #48205 - a retried PR operation may already have a version assigned to it
// in another VM
if (event.isPossibleDuplicate() && event.getRegion().getConcurrencyChecksEnabled()
&& event.getVersionTag() == null && event.getEventId() != null) {
boolean isBulkOp = event.getOperation().isPutAll() || event.getOperation().isRemoveAll();
VersionTag tag =
FindVersionTagOperation.findVersionTag(event.getRegion(), event.getEventId(), isBulkOp);
event.setVersionTag(tag);
}
}
return isDuplicate;
}
private void markEventAsDuplicate(EntryEventImpl event) {
event.setPossibleDuplicate(true);
if (getConcurrencyChecksEnabled() && event.getVersionTag() == null) {
if (event.isBulkOpInProgress()) {
event.setVersionTag(getEventTracker().findVersionTagForBulkOp(event.getEventId()));
} else {
event.setVersionTag(getEventTracker().findVersionTagForSequence(event.getEventId()));
}
}
}
void setGeneratedVersionTag(boolean generateVersionTag) {
// there is at-least one other persistent member, so turn on concurrencyChecks
enableConcurrencyChecks();
this.generateVersionTag = generateVersionTag;
}
protected boolean getGenerateVersionTag() {
return generateVersionTag;
}
@Override
boolean shouldGenerateVersionTag(RegionEntry entry, EntryEventImpl event) {
if (logger.isTraceEnabled()) {
logger.trace(
"shouldGenerateVersionTag this.generateVersionTag={} ccenabled={} dataPolicy={} event:{}",
generateVersionTag, getConcurrencyChecksEnabled(), getDataPolicy(), event);
}
if (!getConcurrencyChecksEnabled() || getDataPolicy() == DataPolicy.EMPTY
|| !generateVersionTag) {
return false;
}
if (getServerProxy() != null) { // client
return false;
}
if (event.getVersionTag() != null && !event.getVersionTag().isGatewayTag()) {
if (logger.isDebugEnabled()) {
logger.debug("Not to create a new version tag for retried event {}", event);
}
return false;
}
if (event.getOperation().isLocal()) { // bug #45402 - localDestroy generated a version tag
return false;
}
if (!event.isOriginRemote() && getDataPolicy().withReplication()) {
return true;
}
if (!getDataPolicy().withReplication() && !getDataPolicy().withPersistence()) {
// do not generate a version stamp in a region that has no replication if it's not based
// on an existing version from a replicate region
return entry.getVersionStamp().hasValidVersion();
}
if (!event.isOriginRemote() && event.getDistributedMember() != null) {
if (!event.getDistributedMember().equals(getMyId())) {
return event.getVersionTag() == null; // one-hop remote message
}
}
return false;
}
/**
* Throws RegionAccessException if required roles are missing and the LossAction is NO_ACCESS
*
* @throws RegionAccessException if required roles are missing and the LossAction is NO_ACCESS
*/
@Override
protected void checkForNoAccess() {
if (requiresReliabilityCheck && isMissingRequiredRoles) {
if (getMembershipAttributes().getLossAction().isNoAccess()) {
synchronized (missingRequiredRoles) {
if (!isMissingRequiredRoles)
return;
Set<Role> roles = Collections.unmodifiableSet(new HashSet<>(missingRequiredRoles));
throw new RegionAccessException(
String.format(
"Operation is disallowed by LossAction %s because these required roles are missing: %s.",
getMembershipAttributes().getLossAction(), roles),
getFullPath(), roles);
}
}
}
}
/**
* Throws RegionAccessException is required roles are missing and the LossAction is either
* NO_ACCESS or LIMITED_ACCESS.
*
* @throws RegionAccessException if required roles are missing and the LossAction is either
* NO_ACCESS or LIMITED_ACCESS
*/
@Override
public void checkForLimitedOrNoAccess() {
if (requiresReliabilityCheck && isMissingRequiredRoles) {
if (getMembershipAttributes().getLossAction().isNoAccess()
|| getMembershipAttributes().getLossAction().isLimitedAccess()) {
synchronized (missingRequiredRoles) {
if (!isMissingRequiredRoles)
return;
Set<Role> roles = Collections.unmodifiableSet(new HashSet<>(missingRequiredRoles));
Assert.assertTrue(!roles.isEmpty());
throw new RegionAccessException(
String.format(
"Operation is disallowed by LossAction %s because these required roles are missing: %s.",
getMembershipAttributes().getLossAction(), roles),
getFullPath(), roles);
}
}
}
}
@Override
public void handleReliableDistribution(Set successfulRecipients) {
handleReliableDistribution(successfulRecipients, Collections.emptySet(),
Collections.emptySet());
}
private void handleReliableDistribution(Set successfulRecipients, Set otherRecipients1,
Set otherRecipients2) {
if (requiresReliabilityCheck) {
MembershipAttributes ra = getMembershipAttributes();
// determine the successful roles
Set<Role> roles = new HashSet<>();
for (Object successfulRecipient : successfulRecipients) {
InternalDistributedMember mbr = (InternalDistributedMember) successfulRecipient;
if (mbr != null) {
roles.addAll(mbr.getRoles());
}
}
for (Object anOtherRecipients1 : otherRecipients1) {
InternalDistributedMember mbr = (InternalDistributedMember) anOtherRecipients1;
if (mbr != null) {
roles.addAll(mbr.getRoles());
}
}
for (Object anOtherRecipients2 : otherRecipients2) {
InternalDistributedMember mbr = (InternalDistributedMember) anOtherRecipients2;
if (mbr != null) {
roles.addAll(mbr.getRoles());
}
}
// determine the missing roles
Set<Role> failedRoles = new HashSet<>(ra.getRequiredRoles());
failedRoles.removeAll(roles);
if (failedRoles.isEmpty()) {
return;
}
throw new RegionDistributionException(
String.format("Operation distribution may have failed to notify these required roles: %s",
failedRoles),
getFullPath(), failedRoles);
}
}
/**
* Called when we do a distributed operation and don't have anyone to distributed it too. Since
* this is only called when no distribution was done (i.e. no recipients) we do not check
* isMissingRequiredRoles because it might not longer be true due to race conditions
*
* @return false if this region has at least one required role and queuing is configured. Returns
* true if sending to no one is ok.
* @throws RoleException if a required role is missing and the LossAction is either NO_ACCESS or
* LIMITED_ACCESS.
* @since GemFire 5.0
*/
boolean isNoDistributionOk() {
if (requiresReliabilityCheck) {
MembershipAttributes ra = getMembershipAttributes();
Set<Role> failedRoles = ra.getRequiredRoles();
throw new RegionDistributionException(
String.format("Operation distribution was not done to these required roles: %s",
failedRoles),
getFullPath(), failedRoles);
}
return true;
}
/**
* returns true if this Region does not distribute its operations to other members.
*
* @since GemFire 6.0
* @see HARegion#localDestroyNoCallbacks(Object)
*/
public boolean doesNotDistribute() {
return false;
}
@Override
public boolean shouldSyncForCrashedMember(InternalDistributedMember id) {
return !doesNotDistribute() && super.shouldSyncForCrashedMember(id);
}
@Override
public boolean requiresReliabilityCheck() {
return requiresReliabilityCheck;
}
/**
* Returns true if the ExpiryTask is currently allowed to expire.
* <p>
* If the region is in NO_ACCESS due to reliability configuration, then no expiration actions are
* allowed.
* <p>
* If the region is in LIMITED_ACCESS due to reliability configuration, then only non-distributed
* expiration actions are allowed.
*/
@Override
protected boolean isExpirationAllowed(ExpiryTask expiry) {
if (requiresReliabilityCheck && isMissingRequiredRoles) {
if (getMembershipAttributes().getLossAction().isNoAccess()) {
return false;
}
return !getMembershipAttributes().getLossAction().isLimitedAccess()
|| !expiry.isDistributedAction();
}
return true;
}
/**
* Performs the resumption action when reliability is resumed.
*
* @return true if asynchronous resumption is triggered
*/
private boolean resumeReliability(InternalDistributedMember id, Set newlyAcquiredRoles) {
boolean async = false;
try {
ResumptionAction ra = getMembershipAttributes().getResumptionAction();
if (ra.isNone()) {
if (logger.isDebugEnabled()) {
logger.debug("Reliability resumption for action of none");
}
resumeExpiration();
} else if (ra.isReinitialize()) {
async = true;
asyncResumeReliability(id, newlyAcquiredRoles);
}
} catch (Exception e) {
logger.fatal("Unexpected exception:",
e);
}
return async;
}
/**
* Handles asynchronous ResumptionActions such as region reinitialize.
*/
private void asyncResumeReliability(final InternalDistributedMember id,
final Set newlyAcquiredRoles) throws RejectedExecutionException {
final ResumptionAction ra = getMembershipAttributes().getResumptionAction();
getDistributionManager().getExecutors().getWaitingThreadPool().execute(() -> {
try {
if (ra.isReinitialize()) {
if (logger.isDebugEnabled()) {
logger.debug("Reliability resumption for action of reinitialize");
}
if (!isDestroyed() && !cache.isClosed()) {
RegionEventImpl event = new RegionEventImpl(this, Operation.REGION_REINITIALIZE, null,
false, getMyId(), generateEventID());
reinitialize(null, event);
}
synchronized (missingRequiredRoles) {
// any number of threads may be waiting on missingRequiredRoles
missingRequiredRoles.notifyAll();
if (hasListener() && id != null) {
// fire afterRoleGain event
RoleEventImpl relEvent = new RoleEventImpl(this, Operation.REGION_CREATE, null, true,
id, newlyAcquiredRoles);
dispatchListenerEvent(EnumListenerEvent.AFTER_ROLE_GAIN, relEvent);
}
}
}
} catch (Exception e) {
logger.fatal("Unexpected exception:", e);
}
});
}
/** Reschedules expiry tasks when reliability is resumed. */
private void resumeExpiration() {
boolean isNoAccess = getMembershipAttributes().getLossAction().isNoAccess();
boolean isLimitedAccess = getMembershipAttributes().getLossAction().isLimitedAccess();
if (!(isNoAccess || isLimitedAccess)) {
return; // early out: expiration was never affected by reliability
}
if (getEntryTimeToLive().getTimeout() > 0 && (isNoAccess || getEntryTimeToLive().getAction()
.isDistributed())) {
rescheduleEntryExpiryTasks();
} else if (getEntryIdleTimeout().getTimeout() > 0 && (isNoAccess || getEntryIdleTimeout()
.getAction().isDistributed())) {
rescheduleEntryExpiryTasks();
} else if (getCustomEntryTimeToLive() != null || getCustomEntryIdleTimeout() != null) {
// Force all entries to be rescheduled
rescheduleEntryExpiryTasks();
}
if (getRegionTimeToLive().getTimeout() > 0 && (isNoAccess || getRegionTimeToLive().getAction()
.isDistributed())) {
addTTLExpiryTask();
}
if (getRegionIdleTimeout().getTimeout() > 0 && (isNoAccess || getRegionIdleTimeout().getAction()
.isDistributed())) {
addIdleExpiryTask();
}
}
/**
* A boolean used to indicate if its the intialization time i.e the distributed Region is created
* for the first time. The variable is used at the time of lost reliablility.
*/
private boolean isInitializingThread = false;
/**
* Called when reliability is lost. If MembershipAttributes are configured with
* {@link LossAction#RECONNECT}then DistributedSystem reconnect will be called asynchronously.
*
* @return true if asynchronous resumption is triggered
*/
private boolean lostReliability(final InternalDistributedMember id, final Set newlyMissingRoles) {
if (DistributedRegion.ignoreReconnect) { // test hook
return false;
}
boolean async = false;
try {
if (getMembershipAttributes().getLossAction().isReconnect()) {
async = true;
doLostReliability(isInitializingThread, id, newlyMissingRoles);
}
} catch (CancelException cce) {
throw cce;
} catch (Exception e) {
logger.fatal("Unexpected exception:",
e);
}
return async;
}
private void doLostReliability(boolean isInitializing, final InternalDistributedMember id,
final Set newlyMissingRoles) {
try {
if (!isInitializing) {
// moved code to a new thread.
Thread thread = new LoggingThread("Reconnect Distributed System", () -> {
try {
logger.debug(
"Reliability loss with policy of reconnect and membership thread doing reconnect");
initializationLatchAfterMemberTimeout.await();
getSystem().tryReconnect(false, "Role Loss", getCache());
synchronized (missingRequiredRoles) {
// any number of threads may be waiting on missingRequiredRoles
missingRequiredRoles.notifyAll();
// need to fire an event if id is not null
if (hasListener() && id != null) {
RoleEventImpl relEvent = new RoleEventImpl(this,
Operation.CACHE_RECONNECT, null, true, id, newlyMissingRoles);
dispatchListenerEvent(EnumListenerEvent.AFTER_ROLE_LOSS, relEvent);
}
}
} catch (Exception e) {
logger.fatal("Unexpected exception:", e);
}
});
thread.start();
} else {
getSystem().tryReconnect(false, "Role Loss", getCache()); // added for
// reconnect.
synchronized (missingRequiredRoles) {
// any number of threads may be waiting on missingRequiredRoles
missingRequiredRoles.notifyAll();
// need to fire an event if id is not null
if (hasListener() && id != null) {
RoleEventImpl relEvent = new RoleEventImpl(this,
Operation.CACHE_RECONNECT, null, true, id, newlyMissingRoles);
dispatchListenerEvent(EnumListenerEvent.AFTER_ROLE_LOSS, relEvent);
}
}
}
} catch (CancelException ignor) {
throw ignor;
} catch (Exception e) {
logger.fatal("Unexpected exception:",
e);
}
}
private void lockCheckReadiness() {
cache.getCancelCriterion().checkCancelInProgress(null);
checkReadiness();
}
@Override
Object validatedDestroy(Object key, EntryEventImpl event)
throws TimeoutException, EntryNotFoundException, CacheWriterException {
Lock dlock = getDistributedLockIfGlobal(key);
try {
return super.validatedDestroy(key, event);
} finally {
if (dlock != null) {
dlock.unlock();
}
}
}
@Override
void localDestroyNoCallbacks(Object key) {
super.localDestroyNoCallbacks(key);
if (getScope().isGlobal()) {
try {
getLockService().freeResources(key);
} catch (LockServiceDestroyedException ignore) {
}
}
}
@Override
public void localDestroy(Object key, Object aCallbackArgument) throws EntryNotFoundException {
super.localDestroy(key, aCallbackArgument);
if (getScope().isGlobal()) {
try {
getLockService().freeResources(key);
} catch (LockServiceDestroyedException ignore) {
}
}
}
@Override
public void invalidate(Object key, Object aCallbackArgument)
throws TimeoutException, EntryNotFoundException {
validateKey(key);
checkReadiness();
checkForLimitedOrNoAccess();
Lock dlock = getDistributedLockIfGlobal(key);
try {
validatedInvalidate(key, aCallbackArgument);
} finally {
if (dlock != null)
dlock.unlock();
}
}
@Override
public Lock getRegionDistributedLock() throws IllegalStateException {
lockCheckReadiness();
checkForLimitedOrNoAccess();
if (!scope.isGlobal()) {
throw new IllegalStateException(
String.format(
"Distribution locks are only supported for regions with GLOBAL scope, not %s",
scope));
}
return new RegionDistributedLock();
}
@Override
public Lock getDistributedLock(Object key) throws IllegalStateException {
validateKey(key);
lockCheckReadiness();
checkForLimitedOrNoAccess();
if (!scope.isGlobal()) {
throw new IllegalStateException(
String.format(
"Distribution locks are only supported for regions with GLOBAL scope, not %s",
scope));
}
if (isLockingSuspendedByCurrentThread()) {
throw new IllegalStateException(
"This thread has suspended all locking for this region");
}
return new DistributedLock(key);
}
@Override
public void preInitialize() {
Set<String> allGatewaySenderIds = getAllGatewaySenderIds();
if (!allGatewaySenderIds.isEmpty()) {
for (GatewaySender sender : cache.getAllGatewaySenders()) {
if (sender.isParallel() && allGatewaySenderIds.contains(sender.getId())) {
// Once decided to support REPLICATED regions with parallel
// gateway-sender/asynchronous-event-queue, ShadowPartitionedRegionForUserRR should be
// called and this validation should be removed.
if (sender.getId().contains(AsyncEventQueueImpl.ASYNC_EVENT_QUEUE_PREFIX)) {
throw new AsyncEventQueueConfigurationException(
String.format(
"Parallel Async Event Queue %s can not be used with replicated region %s",
AsyncEventQueueImpl.getAsyncEventQueueIdFromSenderId(sender.getId()),
getFullPath()));
} else {
throw new GatewaySenderConfigurationException(
String.format(
"Parallel gateway sender %s can not be used with replicated region %s",
sender.getId(), getFullPath()));
}
}
}
}
}
/**
* Called while NOT holding lock on parent's subregions
*
* @throws IllegalStateException if region is not compatible with a region in another VM.
*/
@Override
public void initialize(InputStream snapshotInputStream, InternalDistributedMember imageTarget,
InternalRegionArguments internalRegionArgs)
throws TimeoutException, IOException, ClassNotFoundException {
Assert.assertTrue(!isInitialized());
if (logger.isDebugEnabled()) {
logger.debug("DistributedRegion.initialize BEGIN: {}", getFullPath());
}
if (scope.isGlobal()) {
getLockService(); // create lock service eagerly now
}
try {
try {
PersistentMemberID persistentMemberId = null;
boolean recoverFromDisk = isRecoveryNeeded();
DiskRegion dskRgn = getDiskRegion();
if (recoverFromDisk) {
if (logger.isDebugEnabled()) {
logger.debug("DistributedRegion.getInitialImageAndRecovery: Starting Recovery");
}
dskRgn.initializeOwner(this); // do recovery
if (logger.isDebugEnabled()) {
logger.debug("DistributedRegion.getInitialImageAndRecovery: Finished Recovery");
}
persistentMemberId = dskRgn.getMyPersistentID();
}
// Create OQL indexes before starting GII.
createOQLIndexes(internalRegionArgs, recoverFromDisk);
if (getDataPolicy().withReplication() || getDataPolicy().withPreloaded()) {
getInitialImageAndRecovery(snapshotInputStream, imageTarget, internalRegionArgs,
recoverFromDisk, persistentMemberId);
} else {
new CreateRegionProcessor(this).initializeRegion();
if (snapshotInputStream != null) {
releaseBeforeGetInitialImageLatch();
loadSnapshotDuringInitialization(snapshotInputStream);
}
}
} catch (DiskAccessException dae) {
handleDiskAccessException(dae, true);
throw dae;
}
initMembershipRoles();
isInitializingThread = false;
// makes sure all latches are released if they haven't been already
super.initialize(null, null, null);
} finally {
getEventTracker().setInitialized();
}
}
@Override
void initialized() {
new UpdateAttributesProcessor(this).distribute(false);
}
/** True if GII was impacted by missing required roles */
private boolean giiMissingRequiredRoles = false;
/**
* A reference counter to protected the memoryThresholdReached boolean
*/
private final Set<DistributedMember> memoryThresholdReachedMembers = new HashSet<>();
// TODO: cleanup getInitialImageAndRecovery
private void getInitialImageAndRecovery(InputStream snapshotInputStream,
InternalDistributedMember imageSrc, InternalRegionArguments internalRegionArgs,
boolean recoverFromDisk, PersistentMemberID persistentId) throws TimeoutException {
logger.info("Initializing region {}",
getName());
ImageState imgState = getImageState();
imgState.init();
boolean targetRecreated = internalRegionArgs.getRecreateFlag();
if (recoverFromDisk && snapshotInputStream != null) {
throw new InternalGemFireError(
String.format(
"if loading a snapshot, then should not be recovering; isRecovering= %s ,snapshotStream= %s",
true, snapshotInputStream));
}
ProfileExchangeProcessor targetProvider;
if (getDataPolicy().withPersistence()) {
targetProvider =
new CreatePersistentRegionProcessor(this, getPersistenceAdvisor(), recoverFromDisk);
} else {
// this will go in the advisor profile
targetProvider = new CreateRegionProcessor(this);
}
imgState.setInRecovery(false);
RegionVersionVector recovered_rvv = null;
if (getDataPolicy().withPersistence()) {
recovered_rvv = getVersionVector() == null ? null
: getVersionVector().getCloneForTransmission();
}
// initializeRegion will send out our profile
targetProvider.initializeRegion();
if (persistenceAdvisor != null) {
persistenceAdvisor.initialize();
}
// Register listener here so that the remote members are known
// since registering calls initializeCriticalMembers (which needs to know about
// remote members
if (!isInternalRegion()) {
if (!isDestroyed) {
cache.getInternalResourceManager().addResourceListener(ResourceType.MEMORY, this);
}
}
releaseBeforeGetInitialImageLatch();
// allow GII to invoke test hooks. Do this just after releasing the
// before-gii latch for bug #48962. See ConcurrentLeaveDuringGIIDUnitTest
InitialImageOperation.beforeGetInitialImage(this);
if (snapshotInputStream != null) {
try {
if (logger.isDebugEnabled()) {
logger.debug(
"DistributedRegion.getInitialImageAndRecovery: About to load snapshot, isInitialized={}; {}",
isInitialized(), getFullPath());
}
loadSnapshotDuringInitialization(snapshotInputStream);
} catch (IOException | ClassNotFoundException e) {
throw new RuntimeException(e);
}
cleanUpDestroyedTokensAndMarkGIIComplete(GIIStatus.NO_GII);
return;
}
// No snapshot provided, use the imageTarget(s)
// if we were given a recommended imageTarget, use that first, and
// treat it like it is a replicate (regardless of whether it actually is
// or not)
InitialImageOperation iiop = new InitialImageOperation(this, entries);
CacheDistributionAdvisor.InitialImageAdvice advice = null;
while (!isDestroyed()) {
advice = targetProvider.getInitialImageAdvice(advice);
boolean attemptGetFromOne = imageSrc != null // we were given a specific member
|| getDataPolicy().withPreloaded() && !advice.preloaded.isEmpty() // this is a
// preloaded
// region
|| (!advice.replicates.isEmpty());
// That is: if we have 0 or 1 giiProvider then we can do a getFromOne gii;
// if we have 2 or more giiProviders then we must do a getFromAll gii.
if (attemptGetFromOne) {
if (recoverFromDisk) {
if (LocalRegion.ISSUE_CALLBACKS_TO_CACHE_OBSERVER) {
CacheObserverHolder.getInstance().afterMarkingGIIStarted();
}
}
{
// If we have an imageSrc and the target is reinitializing mark the
// getInitialImage so that it will wait until the target region is fully initialized
// before responding to the get image request. Otherwise, the
// source may respond with no data because it is still initializing,
// e.g. loading a snapshot.
// Plan A: use specified imageSrc, if specified
if (imageSrc != null) {
try {
GIIStatus ret = iiop.getFromOne(Collections.singleton(imageSrc), targetRecreated,
advice, recoverFromDisk, recovered_rvv);
if (GIIStatus.didGII(ret)) {
giiMissingRequiredRoles = false;
cleanUpDestroyedTokensAndMarkGIIComplete(ret);
return;
}
} finally {
imageSrc = null;
}
}
// Plan C: use a replicate, if one exists
GIIStatus ret =
iiop.getFromOne(advice.replicates, false, advice, recoverFromDisk, recovered_rvv);
if (GIIStatus.didGII(ret)) {
cleanUpDestroyedTokensAndMarkGIIComplete(ret);
return;
}
// Plan D: if this is a PRELOADED region, fetch from another PRELOADED
if (getDataPolicy().isPreloaded()) {
GIIStatus ret_preload =
iiop.getFromOne(advice.preloaded, false, advice, recoverFromDisk, recovered_rvv);
if (GIIStatus.didGII(ret_preload)) {
cleanUpDestroyedTokensAndMarkGIIComplete(ret_preload);
return;
}
} // isPreloaded
}
// If we got to this point, we failed in the GII. Cleanup
// any partial image we received
cleanUpAfterFailedGII(recoverFromDisk);
} // attemptGetFromOne
else {
if (!isDestroyed()) {
if (recoverFromDisk) {
logger.info(
"Region {} recovered from the local disk. Old persistent ID: {}, new persistent ID {}",
getFullPath(), persistentId, getPersistentID());
if (persistentId != null) {
RegionLogger.logRecovery(getFullPath(), persistentId,
getDistributionManager().getDistributionManagerId());
}
} else {
RegionLogger.logCreate(getFullPath(),
getDistributionManager().getDistributionManagerId());
if (getPersistentID() != null) {
RegionLogger.logPersistence(getFullPath(),
getDistributionManager().getDistributionManagerId(), getPersistentID());
logger.info("Region {} was created on this member with the persistent id {}.",
new Object[] {getFullPath(), getPersistentID()});
}
}
cleanUpDestroyedTokensAndMarkGIIComplete(GIIStatus.NO_GII);
return;
}
break;
}
}
}
public void scheduleSynchronizeForLostMember(InternalDistributedMember member,
VersionSource lostVersionID, long delay) {
getGemFireCache().getCCPTimer().schedule(new SystemTimer.SystemTimerTask() {
@Override
public void run2() {
performSynchronizeForLostMemberTask(member, lostVersionID);
}
}, delay);
}
void performSynchronizeForLostMemberTask(InternalDistributedMember member,
VersionSource lostVersionID) {
if (!isInitializedWithWait()) {
return;
}
synchronizeForLostMember(member, lostVersionID);
}
/**
* If this region has concurrency controls enabled this will pull any missing changes from other
* replicates using InitialImageOperation and a filtered chunking protocol.
*/
void synchronizeForLostMember(InternalDistributedMember lostMember,
VersionSource lostVersionID) {
if (!getConcurrencyChecksEnabled()) {
return;
}
CacheDistributionAdvisor advisor = getCacheDistributionAdvisor();
Set<InternalDistributedMember> targets = advisor.adviseInitializedReplicates();
for (InternalDistributedMember target : targets) {
synchronizeWith(target, lostVersionID, lostMember);
}
}
/**
* synchronize with another member wrt messages from the given "lost" member. This can be used
* when a primary bucket crashes to ensure that interrupted message distribution is mended.
*/
private void synchronizeWith(InternalDistributedMember target, VersionSource versionMember,
InternalDistributedMember lostMember) {
InitialImageOperation op = new InitialImageOperation(this, entries);
op.synchronizeWith(target, versionMember, lostMember);
}
/**
* This is invoked by syncForCrashedMember when scheduling region synchronization
* triggered by member departed event. It sets the regionSynchronizeScheduledOrDone
* flag in region version holder to true. This indicates that no additional region sync for
* the lost member is needed, when it receives requests for region sync for the lost member.
*/
public void setRegionSynchronizeScheduled(VersionSource lostMemberVersionID) {
RegionVersionHolder regionVersionHolder =
getVersionVector().getHolderForMember(lostMemberVersionID);
if (regionVersionHolder != null) {
regionVersionHolder.setRegionSynchronizeScheduled();
}
}
/**
* This method checks region version holder to see if regionSynchronizeScheduledOrDone is
* set to true for the lost member. If it is not, the regionSynchronizeScheduledOrDone variable
* is set to true and returns true. If it is already set to true, do nothing and returns false.
*/
public boolean setRegionSynchronizedWithIfNotScheduled(VersionSource lostMemberVersionID) {
RegionVersionHolder regionVersionHolder =
getVersionVector().getHolderForMember(lostMemberVersionID);
if (regionVersionHolder != null) {
return regionVersionHolder.setRegionSynchronizeScheduledOrDoneIfNot();
}
return false;
}
public boolean isInitializedWithWait() {
while (!isInitialized()) {
if (isDestroyed()) {
return false;
} else {
try {
if (logger.isDebugEnabled()) {
logger.debug(
"da.syncForCrashedMember waiting for region to finish initializing: {}", this);
}
Thread.sleep(100);
} catch (InterruptedException e) {
return false;
}
}
}
return true;
}
/** remove any partial entries received in a failed GII */
void cleanUpAfterFailedGII(boolean recoverFromDisk) {
DiskRegion dskRgn = getDiskRegion();
// if we have a persistent region, instead of deleting everything on disk,
// we will just reset the "recovered from disk" flag. After
// the next GII we will delete these entries if they do not come
// in as part of the GII.
if (recoverFromDisk && dskRgn != null && dskRgn.isBackup()) {
dskRgn.resetRecoveredEntries(this);
return;
}
if (!getRegionMap().isEmpty()) {
lockFailedInitialImageWriteLock();
try {
closeEntries();
if (getDiskRegion() != null) {
getDiskRegion().clear(this, null);
}
// clear the left-members and version-tags sets in imageState
getImageState().getLeftMembers();
getImageState().getVersionTags();
// Clear OQL indexes
if (indexManager != null) {
try {
indexManager.rerunIndexCreationQuery();
} catch (Exception ex) {
if (logger.isDebugEnabled()) {
logger.debug("Exception while clearing indexes after GII failure.", ex);
}
}
}
} finally {
unlockFailedInitialImageWriteLock();
}
}
}
void lockFailedInitialImageWriteLock() {
failedInitialImageLock.writeLock().lock();
}
void unlockFailedInitialImageWriteLock() {
failedInitialImageLock.writeLock().unlock();
}
void lockFailedInitialImageReadLock() {
failedInitialImageLock.readLock().lock();
}
private void unlockFailedInitialImageReadLock() {
failedInitialImageLock.readLock().unlock();
}
@Override
public boolean lockWhenRegionIsInitializing() {
if (!isInitialized()) {
lockFailedInitialImageReadLock();
return true;
}
return false;
}
@Override
public void unlockWhenRegionIsInitializing() {
unlockFailedInitialImageReadLock();
}
private void initMembershipRoles() {
synchronized (advisorListener) {
// hold sync to prevent listener from changing initial members
Set<InternalDistributedMember> others =
distAdvisor.addMembershipListenerAndAdviseGeneric(advisorListener);
advisorListener.addMembers(others);
// initialize missing required roles with initial member info
if (getMembershipAttributes().hasRequiredRoles()) {
// AdvisorListener will also sync on missingRequiredRoles
synchronized (missingRequiredRoles) {
missingRequiredRoles.addAll(getMembershipAttributes().getRequiredRoles());
// remove all the roles we are playing since they will never be
// missing
missingRequiredRoles.removeAll(getSystem().getDistributedMember().getRoles());
for (Object other1 : others) {
DistributedMember other = (DistributedMember) other1;
missingRequiredRoles.removeAll(other.getRoles());
}
}
}
}
if (getMembershipAttributes().hasRequiredRoles()) {
// wait up to memberTimeout for required roles...
// boolean requiredRolesAreMissing = false;
int memberTimeout = getSystem().getConfig().getMemberTimeout();
if (logger.isDebugEnabled()) {
logger.debug("Waiting up to {} for required roles.", memberTimeout);
}
try {
if (giiMissingRequiredRoles) {
// force reliability loss and possibly resumption
isInitializingThread = true;
synchronized (advisorListener) {
synchronized (missingRequiredRoles) {
// forcing state of loss because of bad GII
isMissingRequiredRoles = true;
getCachePerfStats().incReliableRegionsMissing(1);
if (getMembershipAttributes().getLossAction().isAllAccess())
getCachePerfStats().incReliableRegionsMissingFullAccess(1); // rahul
else if (getMembershipAttributes().getLossAction().isLimitedAccess())
getCachePerfStats().incReliableRegionsMissingLimitedAccess(1);
else if (getMembershipAttributes().getLossAction().isNoAccess())
getCachePerfStats().incReliableRegionsMissingNoAccess(1);
// pur code to increment the stats.
if (logger.isDebugEnabled()) {
logger.debug("GetInitialImage had missing required roles.");
}
// TODO: will this work with RECONNECT and REINITIALIZE?
isInitializingThread = true;
lostReliability(null, null);
if (missingRequiredRoles.isEmpty()) {
// all required roles are present so force resumption
isMissingRequiredRoles = false;
getCachePerfStats().incReliableRegionsMissing(-1);
if (getMembershipAttributes().getLossAction().isAllAccess())
getCachePerfStats().incReliableRegionsMissingFullAccess(-1); // rahul
else if (getMembershipAttributes().getLossAction().isLimitedAccess())
getCachePerfStats().incReliableRegionsMissingLimitedAccess(-1);
else if (getMembershipAttributes().getLossAction().isNoAccess())
getCachePerfStats().incReliableRegionsMissingNoAccess(-1);
// pur code to increment the stats.
boolean async = resumeReliability(null, null);
if (async) {
advisorListener.destroyed = true;
}
}
}
}
} else {
if (!getSystem().isLoner()) {
waitForRequiredRoles(memberTimeout);
}
boolean initiateLossAction = false;
synchronized (advisorListener) {
synchronized (missingRequiredRoles) {
if (missingRequiredRoles.isEmpty()) {
Assert.assertTrue(!isMissingRequiredRoles);
if (logger.isDebugEnabled()) {
logger.debug("Initialization completed with all required roles present.");
}
} else {
// starting in state of loss...
isMissingRequiredRoles = true;
getCachePerfStats().incReliableRegionsMissing(1);
if (getMembershipAttributes().getLossAction().isAllAccess())
getCachePerfStats().incReliableRegionsMissingFullAccess(1); // rahul
else if (getMembershipAttributes().getLossAction().isLimitedAccess())
getCachePerfStats().incReliableRegionsMissingLimitedAccess(1);
else if (getMembershipAttributes().getLossAction().isNoAccess())
getCachePerfStats().incReliableRegionsMissingNoAccess(1);
if (logger.isDebugEnabled()) {
logger.debug("Initialization completed with missing required roles: {}",
missingRequiredRoles);
}
isInitializingThread = true;
initiateLossAction = true;
}
}
}
if (initiateLossAction) {
lostReliability(null, null);
}
}
} catch (RegionDestroyedException ignore) {
// ignore to fix bug 34639 may be thrown by waitForRequiredRoles
} catch (CancelException e) {
// ignore to fix bug 34639 may be thrown by waitForRequiredRoles
if (isInitializingThread) {
throw e;
}
} catch (Exception e) {
logger.fatal("Unexpected exception:", e);
}
}
// open latch which will allow any threads in lostReliability to proceed
initializationLatchAfterMemberTimeout.countDown();
}
private boolean isRecoveryNeeded() {
return getDataPolicy().withPersistence() && getDiskRegion().isRecreated();
}
// called by InitialImageOperation to clean up destroyed tokens
// release afterGetInitialImageInitializationLatch before unlocking
// cleanUpLock
@SuppressWarnings("UL_UNRELEASED_LOCK")
protected void cleanUpDestroyedTokensAndMarkGIIComplete(GIIStatus giiStatus) {
// We need to clean up the disk before we release the after get initial image latch
DiskRegion dskRgn = getDiskRegion();
if (dskRgn != null && dskRgn.isBackup()) {
dskRgn.finishInitializeOwner(this, giiStatus);
}
ImageState is = getImageState();
is.lockGII();
// clear the version tag and left-members sets
is.getVersionTags();
is.getLeftMembers();
// remove DESTROYED tokens
RegionVersionVector rvv = is.getClearRegionVersionVector();
try {
Iterator<Object> keysIt = getImageState().getDestroyedEntries();
while (keysIt.hasNext()) {
entries.removeIfDestroyed(keysIt.next());
}
if (rvv != null) {
// clear any entries received in the GII that are older than the RVV versions.
// this can happen if entry chunks were received prior to the clear() being
// processed
clearEntries(rvv);
}
// need to do this before we release the afterGetInitialImageLatch
if (persistenceAdvisor != null) {
persistenceAdvisor.setOnline(GIIStatus.didGII(giiStatus), false, getPersistentID());
}
} finally {
// release after gii lock first so basicDestroy will see isInitialized()
// be true
// when they get the cleanUp lock.
try {
releaseAfterGetInitialImageLatch();
} finally { // make sure unlockGII is done for bug 40001
is.unlockGII();
}
}
if (LocalRegion.ISSUE_CALLBACKS_TO_CACHE_OBSERVER) {
CacheObserverHolder.getInstance().afterMarkingGIICompleted();
}
// "Initializing region {0}" which is not acompanied by a completed message. Users think thread
// is stuck in some operation. Hence adding this log
logger.info("Initialization of region {} completed", getName());
}
@Override
public void basicBridgeRemove(Object key, Object expectedOldValue, Object p_callbackArg,
ClientProxyMembershipID memberId, boolean fromClient, EntryEventImpl clientEvent)
throws TimeoutException, EntryNotFoundException, CacheWriterException {
Lock lock = getDistributedLockIfGlobal(key);
try {
super.basicBridgeRemove(key, expectedOldValue, p_callbackArg, memberId, fromClient,
clientEvent);
} finally {
if (lock != null) {
logger.debug("releasing distributed lock on {}", key);
lock.unlock();
getLockService().freeResources(key);
}
}
}
@Override
public void basicBridgeDestroy(Object key, Object p_callbackArg, ClientProxyMembershipID memberId,
boolean fromClient, EntryEventImpl clientEvent)
throws TimeoutException, EntryNotFoundException, CacheWriterException {
Lock lock = getDistributedLockIfGlobal(key);
try {
super.basicBridgeDestroy(key, p_callbackArg, memberId, fromClient, clientEvent);
} finally {
if (lock != null) {
logger.debug("releasing distributed lock on {}", key);
lock.unlock();
getLockService().freeResources(key);
}
}
}
@Override
public void basicBridgeInvalidate(Object key, Object p_callbackArg,
ClientProxyMembershipID memberId, boolean fromClient, EntryEventImpl clientEvent)
throws TimeoutException, EntryNotFoundException, CacheWriterException {
Lock lock = getDistributedLockIfGlobal(key);
try {
super.basicBridgeInvalidate(key, p_callbackArg, memberId, fromClient, clientEvent);
} finally {
if (lock != null) {
logger.debug("releasing distributed lock on {}", key);
lock.unlock();
}
}
}
@Override
public void basicDestroy(EntryEventImpl event, boolean cacheWrite, Object expectedOldValue)
throws EntryNotFoundException, CacheWriterException, TimeoutException {
// disallow local destruction for mirrored keysvalues regions
boolean hasSeen = false;
if (hasSeenEvent(event)) {
hasSeen = true;
}
checkIfReplicatedAndLocalDestroy(event);
try {
boolean invokeWriter = cacheWrite;
if (requiresOneHopForMissingEntry(event)) {
// bug #45704: see if a one-hop must be done for this operation
RegionEntry re = getRegionEntry(event.getKey());
if (re == null /* || re.isTombstone() */ || !generateVersionTag) {
if (getServerProxy() == null) {
// only assert for non-client regions.
Assert.assertTrue(!getDataPolicy().withReplication() || !generateVersionTag);
}
if (!event.isBulkOpInProgress() || getDataPolicy().withStorage()) {
// removeAll will send a single one-hop for empty regions. for other missing entries
// we need to get a valid version number before modifying the local cache
// TODO: deltaGII: verify that delegating to a peer when this region is also a client is
// acceptable
boolean didDistribute =
RemoteDestroyMessage.distribute(event, expectedOldValue, !generateVersionTag);
if (!generateVersionTag && !didDistribute) {
throw new PersistentReplicatesOfflineException();
}
if (didDistribute) {
if (logger.isTraceEnabled()) {
logger.trace("Event after remoteDestroy operation: {}", event);
}
invokeWriter = false; // remote cache invoked the writer
if (event.getVersionTag() == null) {
// if the event wasn't applied by the one-hop replicate it will not have a version
// tag
// and so should not be applied to this cache
return;
}
}
}
}
}
super.basicDestroy(event, invokeWriter, expectedOldValue);
// if this is a destroy coming in from remote source, free up lock resources
// if this is a local origin destroy, this will happen after lock is
// released
if (scope.isGlobal() && event.isOriginRemote()) {
try {
getLockService().freeResources(event.getKey());
} catch (LockServiceDestroyedException ignore) {
}
}
} finally {
if (hasSeen) {
if (event.isBulkOpInProgress() && !event.isOriginRemote()) {
event.getRemoveAllOperation().addEntry(event, true);
}
if (!getConcurrencyChecksEnabled() || event.hasValidVersionTag()) {
// before distribute: DR.destroy, hasSeenEvent. no to notifyGateway
distributeDestroy(event, expectedOldValue);
event.invokeCallbacks(this, true, false);
}
}
}
}
@Override
public void basicDestroyPart3(RegionEntry re, EntryEventImpl event, boolean inTokenMode,
boolean duringRI, boolean invokeCallbacks, Object expectedOldValue) {
distributeDestroy(event, expectedOldValue);
super.basicDestroyPart3(re, event, inTokenMode, duringRI, invokeCallbacks, expectedOldValue);
}
void distributeDestroy(EntryEventImpl event, Object expectedOldValue) {
if (event.isDistributed() && !event.isOriginRemote() && !event.isBulkOpInProgress()) {
boolean distribute = !event.getInhibitDistribution();
if (distribute) {
// before distribute: DR.destroy, it has notifiedGatewaySender ealier
DestroyOperation op = new DestroyOperation(event);
op.distribute();
}
}
}
@Override
boolean evictDestroy(EvictableEntry entry) {
boolean evictDestroyWasDone = super.evictDestroy(entry);
if (evictDestroyWasDone) {
if (scope.isGlobal()) {
try {
getLockService().freeResources(entry.getKey());
} catch (LockServiceDestroyedException ignore) {
}
}
}
return evictDestroyWasDone;
}
@Override
void basicInvalidateRegion(RegionEventImpl event) {
// disallow local invalidation for replicated regions
if (!event.getOperation().isDistributed() && getScope().isDistributed()
&& getDataPolicy().withReplication()) {
throw new IllegalStateException(
"Not allowed to do a local invalidation on a replicated region");
}
if (shouldDistributeInvalidateRegion(event)) {
distributeInvalidateRegion(event);
}
super.basicInvalidateRegion(event);
}
/**
* decide if InvalidateRegionOperation should be sent to peers. broken out so that BucketRegion
* can override
*
* @return true if {@link InvalidateRegionOperation} should be distributed, false otherwise
*/
protected boolean shouldDistributeInvalidateRegion(RegionEventImpl event) {
return event.getOperation().isDistributed() && !event.isOriginRemote();
}
/**
* Distribute the invalidate of a region given its event. This implementation sends the invalidate
* to peers.
*
* @since GemFire 5.7
*/
protected void distributeInvalidateRegion(RegionEventImpl event) {
new InvalidateRegionOperation(event).distribute();
}
@Override
void basicDestroyRegion(RegionEventImpl event, boolean cacheWrite, boolean lock,
boolean callbackEvents) throws CacheWriterException, TimeoutException {
final String path = getFullPath();
// Keep track of regions that are being destroyed. This helps avoid a race
// when another member concurrently creates this region. See bug 42051.
boolean isClose = event.getOperation().isClose();
if (!isClose) {
cache.beginDestroy(path, this);
}
try {
super.basicDestroyRegion(event, cacheWrite, lock, callbackEvents);
// send destroy region operation even if this is a localDestroyRegion (or
// close)
if (!event.isOriginRemote()) {
distributeDestroyRegion(event, true);
} else {
if (!event.isReinitializing()) {
RegionEventImpl localEvent =
new RegionEventImpl(this, Operation.REGION_LOCAL_DESTROY, event.getCallbackArgument(),
false, getMyId(), generateEventID()/* generate EventID */);
distributeDestroyRegion(localEvent, false/* fixes bug 41111 */);
}
}
notifyBridgeClients(event);
} catch (CancelException e) {
if (logger.isDebugEnabled()) {
logger.debug("basicDestroyRegion short-circuited due to cancellation");
}
} finally {
if (!isClose) {
cache.endDestroy(path, this);
}
RegionLogger.logDestroy(path, getMyId(), getPersistentID(), isClose);
}
}
@Override
void distributeDestroyRegion(RegionEventImpl event, boolean notifyOfRegionDeparture) {
if (persistenceAdvisor != null) {
persistenceAdvisor.releaseTieLock();
}
new DestroyRegionOperation(event, notifyOfRegionDeparture).distribute();
}
/**
* Return true if invalidation occurred; false if it did not, for example if it was already
* invalidated
*/
@Override
public void basicInvalidate(EntryEventImpl event) throws EntryNotFoundException {
boolean hasSeen = false;
if (hasSeenEvent(event)) {
hasSeen = true;
}
try {
// disallow local invalidation for replicated regions
if (event.isLocalInvalid() && !event.getOperation().isLocal() && getScope().isDistributed()
&& getDataPolicy().withReplication()) {
throw new IllegalStateException(
"Not allowed to do a local invalidation on a replicated region");
}
if (requiresOneHopForMissingEntry(event)) {
// bug #45704: see if a one-hop must be done for this operation
RegionEntry re = getRegionEntry(event.getKey());
if (re == null/* || re.isTombstone() */ || !generateVersionTag) {
if (getServerProxy() == null) {
// only assert for non-client regions.
Assert.assertTrue(!getDataPolicy().withReplication() || !generateVersionTag);
}
// TODO: deltaGII: verify that delegating to a peer when this region is also a client is
// acceptable
boolean didDistribute =
RemoteInvalidateMessage.distribute(event, !generateVersionTag);
if (!generateVersionTag && !didDistribute) {
throw new PersistentReplicatesOfflineException();
}
if (didDistribute) {
if (logger.isDebugEnabled()) {
logger.debug("Event after remoteInvalidate operation: {}", event);
}
if (event.getVersionTag() == null) {
// if the event wasn't applied by the one-hop replicate it will not have a version tag
// and so should not be applied to this cache
return;
}
}
}
}
super.basicInvalidate(event);
} finally {
if (hasSeen) {
if (!getConcurrencyChecksEnabled() || event.hasValidVersionTag()) {
distributeInvalidate(event);
event.invokeCallbacks(this, true, false);
}
}
}
}
@Override
void basicInvalidatePart3(RegionEntry re, EntryEventImpl event, boolean invokeCallbacks) {
distributeInvalidate(event);
super.basicInvalidatePart3(re, event, invokeCallbacks);
}
void distributeInvalidate(EntryEventImpl event) {
if (!isRegionInvalid() && event.isDistributed() && !event.isOriginRemote()
&& !isTX() /* only distribute if non-tx */) {
if (event.isDistributed() && !event.isOriginRemote()) {
boolean distribute = !event.getInhibitDistribution();
if (distribute) {
// before distribute: DR.invalidate, it has triggered callback earlier
InvalidateOperation op = new InvalidateOperation(event);
op.distribute();
}
}
}
}
@Override
void basicUpdateEntryVersion(EntryEventImpl event) throws EntryNotFoundException {
InternalRegion internalRegion = event.getRegion();
AbstractRegionMap regionMap = (AbstractRegionMap) internalRegion.getRegionMap();
regionMap.lockForCacheModification(internalRegion, event);
final boolean locked = internalRegion.lockWhenRegionIsInitializing();
try {
try {
if (!hasSeenEvent(event)) {
super.basicUpdateEntryVersion(event);
}
} finally {
if (!getConcurrencyChecksEnabled() || event.hasValidVersionTag()) {
distributeUpdateEntryVersion(event);
}
}
} finally {
if (locked) {
internalRegion.unlockWhenRegionIsInitializing();
}
regionMap.releaseCacheModificationLock(internalRegion, event);
}
}
void distributeUpdateEntryVersion(EntryEventImpl event) {
if (!isRegionInvalid() && event.isDistributed() && !event.isOriginRemote()
&& !isTX() /* only distribute if non-tx */) {
if (event.isDistributed() && !event.isOriginRemote()) {
// before distribute: DR has sent callback earlier
UpdateEntryVersionOperation op = new UpdateEntryVersionOperation(event);
op.distribute();
}
}
}
@Override
protected void basicClear(RegionEventImpl regionEvent) {
Lock dlock = getRegionDistributedLockIfGlobal();
try {
super.basicClear(regionEvent);
} finally {
if (dlock != null)
dlock.unlock();
}
}
@Override
void basicClear(RegionEventImpl regionEvent, boolean cacheWrite) {
if (getConcurrencyChecksEnabled() && !getDataPolicy().withReplication()) {
boolean retry = false;
do {
// non-replicate regions must defer to a replicate for clear/invalidate of region
Set<InternalDistributedMember> repls = distAdvisor.adviseReplicates();
if (!repls.isEmpty()) {
InternalDistributedMember mbr = repls.iterator().next();
RemoteClearMessage op = RemoteClearMessage.create(mbr, this);
try {
op.distribute();
return;
} catch (CancelException | RegionDestroyedException | RemoteOperationException e) {
getCancelCriterion().checkCancelInProgress(e);
retry = true;
}
}
} while (retry);
}
// if no version vector or if no replicates are around, use the default mechanism
super.basicClear(regionEvent, cacheWrite);
}
@Override
void cmnClearRegion(RegionEventImpl regionEvent, boolean cacheWrite, boolean useRVV) {
boolean enableRVV = useRVV && getDataPolicy().withReplication()
&& getConcurrencyChecksEnabled() && !getDistributionManager().isLoner();
// Fix for 46338 - apparently multiple threads from the same VM are allowed
// to suspend locking, which is what distributedLockForClear() does. We don't
// want that to happen, so we'll synchronize to make sure only one thread on
// this member performs a clear.
synchronized (clearLock) {
if (enableRVV) {
distributedLockForClear();
try {
Set<InternalDistributedMember> participants =
getCacheDistributionAdvisor().adviseInvalidateRegion();
// pause all generation of versions and flush from the other members to this one
try {
obtainWriteLocksForClear(regionEvent, participants);
clearRegionLocally(regionEvent, cacheWrite, null);
if (!regionEvent.isOriginRemote() && regionEvent.getOperation().isDistributed()) {
DistributedClearOperation.clear(regionEvent, null, participants);
}
} finally {
releaseWriteLocksForClear(regionEvent, participants);
}
} finally {
distributedUnlockForClear();
}
} else {
Set<InternalDistributedMember> participants =
getCacheDistributionAdvisor().adviseInvalidateRegion();
clearRegionLocally(regionEvent, cacheWrite, null);
if (!regionEvent.isOriginRemote() && regionEvent.getOperation().isDistributed()) {
DistributedClearOperation.clear(regionEvent, null, participants);
}
}
}
// since clients do not maintain RVVs except for tombstone GC
// we need to ensure that current ops reach the client queues
// before queuing a clear, but there is no infrastructure for doing so
notifyBridgeClients(regionEvent);
}
/**
* Obtain a distributed lock for the clear operation.
*/
private void distributedLockForClear() {
if (!scope.isGlobal()) { // non-global regions must lock when using RVV
try {
getLockService().lock("_clearOperation", -1, -1);
} catch (IllegalStateException e) {
lockCheckReadiness();
throw e;
}
}
}
/**
* Release the distributed lock for the clear operation.
*/
private void distributedUnlockForClear() {
if (!scope.isGlobal()) {
try {
getLockService().unlock("_clearOperation");
} catch (IllegalStateException e) {
lockCheckReadiness();
throw e;
}
}
}
/**
* obtain locks preventing generation of new versions in other members
*/
private void obtainWriteLocksForClear(RegionEventImpl regionEvent,
Set<InternalDistributedMember> participants) {
lockLocallyForClear(getDistributionManager(), getMyId(), regionEvent);
DistributedClearOperation.lockAndFlushToOthers(regionEvent, participants);
}
/**
* pause local operations so that a clear() can be performed and flush comm channels to the given
* member
*/
void lockLocallyForClear(DistributionManager dm, InternalDistributedMember locker,
CacheEvent event) {
RegionVersionVector rvv = getVersionVector();
ARMLockTestHook armLockTestHook = getRegionMap().getARMLockTestHook();
if (armLockTestHook != null) {
armLockTestHook.beforeLock(this, event);
}
if (rvv != null) {
// block new operations from being applied to the region map
rvv.lockForClear(getFullPath(), dm, locker);
// Check for region destroyed after we have locked, to make sure
// we don't continue a clear if the region has been destroyed.
checkReadiness();
// Only need to flush if NOACK at this point
if (getAttributes().getScope().isDistributedNoAck()) {
Set<InternalDistributedMember> members = getDistributionAdvisor().adviseCacheOp();
StateFlushOperation.flushTo(members, this);
}
}
if (armLockTestHook != null) {
armLockTestHook.afterLock(this, null);
}
}
/**
* releases the locks obtained in obtainWriteLocksForClear
*/
private void releaseWriteLocksForClear(RegionEventImpl regionEvent,
Set<InternalDistributedMember> participants) {
ARMLockTestHook armLockTestHook = getRegionMap().getARMLockTestHook();
if (armLockTestHook != null) {
armLockTestHook.beforeRelease(this, regionEvent);
}
getVersionVector().unlockForClear(getMyId());
DistributedClearOperation.releaseLocks(regionEvent, participants);
if (armLockTestHook != null) {
armLockTestHook.afterRelease(this, regionEvent);
}
}
/**
* Wait for in progress clears that were initiated by this member.
*/
private void waitForInProgressClear() {
RegionVersionVector rvv = getVersionVector();
if (rvv != null) {
synchronized (clearLock) {
// do nothing;
// I'm a little scared that the compiler might optimize
// away this synchronization if we really do nothing. Hence
// my fine log message below. This might not be necessary.
if (logger.isDebugEnabled()) {
logger.debug("Done waiting for clear");
}
}
}
}
/**
* Distribute Tombstone garbage-collection information to all peers with storage
*/
protected EventID distributeTombstoneGC(Set<Object> keysRemoved) {
getCachePerfStats().incTombstoneGCCount();
EventID eventId = new EventID(getSystem());
DistributedTombstoneOperation gc = DistributedTombstoneOperation.gc(this, eventId);
gc.distribute();
notifyClientsOfTombstoneGC(getVersionVector().getTombstoneGCVector(), keysRemoved, eventId,
null);
return eventId;
}
@Override
void basicLocalClear(RegionEventImpl rEvent) {
if (getScope().isDistributed() && getDataPolicy().withReplication()) {
throw new UnsupportedOperationException(
"localClear is not supported on distributed replicated regions.");
}
super.basicLocalClear(rEvent);
}
public DistributionConfig getDistributionConfig() {
return getSystem().getDistributionManager().getConfig();
}
@Override
public CacheDistributionAdvisor getDistributionAdvisor() {
return distAdvisor;
}
@Override
public CacheDistributionAdvisor getCacheDistributionAdvisor() {
return distAdvisor;
}
public PersistenceAdvisor getPersistenceAdvisor() {
return persistenceAdvisor;
}
public PersistentMemberID getPersistentID() {
return persistentId;
}
/** Returns the distribution profile; lazily creates one if needed */
@Override
public Profile getProfile() {
return distAdvisor.createProfile();
}
@Override
public void fillInProfile(Profile profile) {
assert profile instanceof CacheProfile;
CacheProfile cacheProfile = (CacheProfile) profile;
cacheProfile.dataPolicy = getDataPolicy();
cacheProfile.hasCacheLoader = basicGetLoader() != null;
cacheProfile.hasCacheWriter = basicGetWriter() != null;
cacheProfile.hasCacheListener = hasListener();
Assert.assertTrue(scope.isDistributed());
cacheProfile.scope = scope;
cacheProfile.inRecovery = getImageState().getInRecovery();
cacheProfile.isPersistent = getDataPolicy().withPersistence();
cacheProfile.setSubscriptionAttributes(getSubscriptionAttributes());
// Below PDX check is added for rolling upgrade support. We are
// removing Old wan in this checkin. PDX region are always gatewayEnabled
// irrespective whether gatewayHub is configured or not.
// Old version Pdx region always has this attribute true so to avoid region
// attribute comparison across member we are setting it to true.
cacheProfile.isGatewayEnabled = isPdxTypesRegion();
cacheProfile.serialNumber = getSerialNumber();
cacheProfile.regionInitialized = isInitialized();
cacheProfile.persistentID = getPersistentID();
if (getPersistenceAdvisor() != null) {
cacheProfile.persistenceInitialized = getPersistenceAdvisor().isOnline();
}
cacheProfile.hasCacheServer = !cache.getCacheServers().isEmpty();
cacheProfile.requiresOldValueInEvents = getDataPolicy().withReplication()
&& filterProfile != null && filterProfile.hasCQs();
cacheProfile.gatewaySenderIds = getGatewaySenderIds();
cacheProfile.asyncEventQueueIds = getVisibleAsyncEventQueueIds();
cacheProfile.isOffHeap = getOffHeap();
}
/**
* Return the DistributedLockService associated with this Region. This method will lazily create
* that service the first time it is invoked on this region.
*/
public DistributedLockService getLockService() {
synchronized (dlockMonitor) {
String dlsName = getFullPath();
if (dlockService == null) {
dlockService = DistributedLockService.getServiceNamed(dlsName);
if (dlockService == null) {
// region destroy will destroy dls and manual freeResources only
dlockService = DLockService.create(getFullPath(), getSystem(), true, false, false);
}
// handle is-lock-grantor region attribute...
if (isLockGrantor) {
dlockService.becomeLockGrantor();
}
if (logger.isDebugEnabled()) {
logger.debug("LockService for {} is using LockLease={}, LockTimeout={}", dlsName,
getCache().getLockLease(), getCache().getLockTimeout());
}
}
return dlockService;
}
}
@Override
boolean isCurrentlyLockGrantor() {
return scope.isGlobal() && getLockService().isLockGrantor();
}
@Override
public boolean isLockGrantor() {
return scope.isGlobal() && isLockGrantor;
}
@Override
public void becomeLockGrantor() {
checkReadiness();
checkForLimitedOrNoAccess();
if (!scope.isGlobal()) {
throw new IllegalStateException(
String.format(
"Distribution locks are only supported for regions with GLOBAL scope, not %s",
scope));
}
DistributedLockService svc = getLockService();
try {
super.becomeLockGrantor();
if (!svc.isLockGrantor()) {
svc.becomeLockGrantor();
}
} finally {
if (!svc.isLockGrantor()) {
if (logger.isDebugEnabled()) {
logger.debug("isLockGrantor is false after becomeLockGrantor for {}", getFullPath());
}
}
}
}
/** @return the deserialized value */
@Override
@Retained
Object findObjectInSystem(KeyInfo keyInfo, boolean isCreate, TXStateInterface txState,
boolean generateCallbacks, Object localValue, boolean disableCopyOnRead, boolean preferCD,
ClientProxyMembershipID requestingClient, EntryEventImpl clientEvent,
boolean returnTombstones) throws CacheLoaderException, TimeoutException {
@Released
EntryEventImpl event = null;
checkForLimitedOrNoAccess();
final Operation op = isCreate ? Operation.CREATE : Operation.UPDATE;
long lastModified = 0L;
try {
event = findOnServer(keyInfo, op, generateCallbacks, clientEvent);
if (event == null) {
event = createEventForLoad(keyInfo, generateCallbacks, requestingClient, op);
lastModified =
findUsingSearchLoad(txState, localValue, clientEvent, keyInfo, event, preferCD);
}
// Update region with new value.
if (event.hasNewValue() && !isMemoryThresholdReachedForLoad()) {
putNewValueInRegion(isCreate, clientEvent, lastModified, event);
} else if (isCreate) {
recordMiss(null, event.getKey());
}
return determineResult(preferCD, event);
} finally {
if (event != null) {
event.release();
}
}
}
private EntryEventImpl createEventForLoad(KeyInfo keyInfo, boolean generateCallbacks,
ClientProxyMembershipID requestingClient, Operation op) {
// Do not generate Event ID
EntryEventImpl event = EntryEventImpl.create(this, op, keyInfo.getKey(), null /* newValue */,
keyInfo.getCallbackArg(), false, getMyId(), generateCallbacks);
if (requestingClient != null) {
event.setContext(requestingClient);
}
return event;
}
private Object determineResult(boolean preferCD, EntryEventImpl event) {
if (preferCD) {
return event.getRawNewValueAsHeapObject();
}
return event.getNewValue();
}
private void putNewValueInRegion(boolean isCreate, EntryEventImpl clientEvent, long lastModified,
EntryEventImpl event) {
RegionEntry re = null;
// Set eventId. Required for interested clients.
event.setNewEventId(cache.getDistributedSystem());
long startPut = getStatisticsClock().getTime();
validateKey(event.getKey());
// this next step also distributes the object to other processes, if necessary
try {
re = basicPutEntry(event, lastModified);
// Update client event with latest version tag from re.
if (re != null && clientEvent != null) {
clientEvent.setVersionTag(event.getVersionTag());
}
if (!isTX()) {
getCachePerfStats().endPut(startPut, event.isOriginRemote());
}
} catch (ConcurrentCacheModificationException e) {
// the cache was modified while we were searching for this entry and
// the netsearch result was elided. Return the current value from the cache
updateEventWithCurrentRegionEntry(event, clientEvent);
} catch (CacheWriterException cwe) {
if (logger.isDebugEnabled()) {
logger.debug("findObjectInSystem: writer exception putting entry {} : {}", event, cwe);
}
}
if (isCreate) {
recordMiss(re, event.getKey());
}
}
private void updateEventWithCurrentRegionEntry(EntryEventImpl event, EntryEventImpl clientEvent) {
// defer the lruUpdateCallback to prevent a deadlock (see bug 51121).
final boolean disabled = entries.disableLruUpdateCallback();
try {
RegionEntry re = getRegionEntry(event.getKey());
if (re != null) {
synchronized (re) {
// value & version must be obtained atomically
// Update client event with latest version tag from re
if (clientEvent != null) {
clientEvent.setVersionTag(re.getVersionStamp().asVersionTag());
}
// OFFHEAP: need to incrc, copy to heap to setNewValue, decrc
event.setNewValue(re.getValue(this));
}
}
} finally {
if (disabled) {
entries.enableLruUpdateCallback();
}
try {
entries.lruUpdateCallback();
} catch (DiskAccessException dae) {
handleDiskAccessException(dae);
throw dae;
}
}
}
/**
* If its client, get the value from server.
*/
private EntryEventImpl findOnServer(KeyInfo keyInfo, Operation op, boolean generateCallbacks,
EntryEventImpl clientEvent) {
if (getServerProxy() == null) {
return null;
}
EntryEventImpl event = null;
VersionTagHolder holder = new VersionTagHolder();
Object aCallbackArgument = keyInfo.getCallbackArg();
Object value = getServerProxy().get(keyInfo.getKey(), aCallbackArgument, holder);
if (value != null) {
event = EntryEventImpl.create(this, op, keyInfo.getKey(), value, aCallbackArgument, false,
getMyId(), generateCallbacks);
event.setVersionTag(holder.getVersionTag());
event.setFromServer(true); // fix for bug 39358
if (clientEvent != null && clientEvent.getVersionTag() == null) {
clientEvent.setVersionTag(holder.getVersionTag());
}
}
return event;
}
private long findUsingSearchLoad(TXStateInterface txState, Object localValue,
EntryEventImpl clientEvent, final KeyInfo keyInfo, EntryEventImpl event, boolean preferCD) {
long lastModified = 0L;
// If this event is because of a register interest call, don't invoke the CacheLoader
boolean getForRegisterInterest = clientEvent != null && clientEvent.getOperation() != null
&& clientEvent.getOperation().isGetForRegisterInterest();
if (!getForRegisterInterest) {
SearchLoadAndWriteProcessor processor = SearchLoadAndWriteProcessor.getProcessor();
try {
processor.initialize(this, keyInfo.getKey(), keyInfo.getCallbackArg());
// processor fills in event
processor.doSearchAndLoad(event, txState, localValue, preferCD);
if (clientEvent != null && clientEvent.getVersionTag() == null) {
clientEvent.setVersionTag(event.getVersionTag());
}
lastModified = processor.getLastModified();
} finally {
processor.release();
}
} else {
if (logger.isDebugEnabled()) {
logger.debug("DistributedRegion.findObjectInSystem skipping loader for region="
+ getFullPath() + "; key=" + keyInfo.getKey());
}
}
return lastModified;
}
/**
* @return true if cacheWrite was performed
*/
@Override
public boolean cacheWriteBeforeDestroy(EntryEventImpl event, Object expectedOldValue)
throws CacheWriterException, EntryNotFoundException, TimeoutException {
boolean result = false;
if (event.isDistributed()) {
CacheWriter localWriter = basicGetWriter();
Set netWriteRecipients = localWriter == null ? distAdvisor.adviseNetWrite() : null;
if ((localWriter != null || (netWriteRecipients != null && !netWriteRecipients.isEmpty()))
&& !event.inhibitAllNotifications()) {
final long start = getCachePerfStats().startCacheWriterCall();
try {
event.setOldValueFromRegion();
SearchLoadAndWriteProcessor processor = SearchLoadAndWriteProcessor.getProcessor();
try {
processor.initialize(this, event.getKey(), null);
processor.doNetWrite(event, netWriteRecipients, localWriter,
SearchLoadAndWriteProcessor.BEFOREDESTROY);
result = true;
} finally {
processor.release();
}
} finally {
getCachePerfStats().endCacheWriterCall(start);
}
}
serverDestroy(event, expectedOldValue);
}
return result;
}
@Override
boolean cacheWriteBeforeRegionDestroy(RegionEventImpl event)
throws CacheWriterException, TimeoutException {
boolean result = false;
if (event.getOperation().isDistributed()) {
CacheWriter localWriter = basicGetWriter();
Set netWriteRecipients = localWriter == null ? distAdvisor.adviseNetWrite() : null;
if (localWriter != null || netWriteRecipients != null && !netWriteRecipients.isEmpty()) {
final long start = getCachePerfStats().startCacheWriterCall();
try {
SearchLoadAndWriteProcessor processor = SearchLoadAndWriteProcessor.getProcessor();
try {
processor.initialize(this, "preDestroyRegion", null);
processor.doNetWrite(event, netWriteRecipients, localWriter,
SearchLoadAndWriteProcessor.BEFOREREGIONDESTROY);
result = true;
} finally {
processor.release();
}
} finally {
getCachePerfStats().endCacheWriterCall(start);
}
}
serverRegionDestroy(event);
}
return result;
}
protected void distributedRegionCleanup(RegionEventImpl event) {
if (event == null || event.getOperation() != Operation.REGION_REINITIALIZE) {
// only perform this if reinitialize is not due to resumption
// (REGION_REINITIALIZE)
// or if event is null then this was a failed initialize (create)
// wake up any threads in waitForRequiredRoles... they will checkReadiness
synchronized (missingRequiredRoles) {
missingRequiredRoles.notifyAll();
}
}
if (persistenceAdvisor != null) {
persistenceAdvisor.close(); // fix for bug 41094
}
distAdvisor.close();
// Fix for bug 46338. Wait for in progress clears before destroying the
// lock service, because destroying the service immediately releases the dlock
waitForInProgressClear();
DLockService dls = null;
synchronized (dlockMonitor) {
if (dlockService != null) {
dls = (DLockService) dlockService;
}
}
if (dls != null) {
try {
dls.destroyAndRemove();
} catch (CancelException e) {
// bug 37118
if (logger.isDebugEnabled()) {
logger.debug("DLS destroy abridged due to shutdown", e);
}
} catch (Exception ex) {
logger.warn("DLS destroy may have failed for " +
getFullPath(),
ex);
}
}
waitForCurrentOperations();
}
private void waitForCurrentOperations() {
// Fix for #48066 - make sure that region operations are completely
// distributed to peers before destroying the region.
boolean flushOnClose =
!Boolean.getBoolean(DistributionConfig.GEMFIRE_PREFIX + "no-flush-on-close"); // test hook
if (!cache.forcedDisconnect() && flushOnClose
&& getDistributionManager().getMembershipManager() != null
&& getDistributionManager().getMembershipManager().isConnected()) {
getDistributionAdvisor().forceNewMembershipVersion();
try {
getDistributionAdvisor().waitForCurrentOperations();
} catch (Exception e) {
// log this but try to close the region so that listeners are invoked
logger.warn(String.format("%s: error closing region %s", this, getFullPath()), e);
}
}
}
/**
* In addition to inherited code this method also invokes RegionMembershipListeners
*/
@Override
public void postCreateRegion() {
super.postCreateRegion();
// should we sync on this.distAdvisor first to prevent bug 44369?
synchronized (advisorListener) {
Set<InternalDistributedMember> others = advisorListener.getInitialMembers();
CacheListener[] listeners = fetchCacheListenersField();
if (listeners != null) {
for (CacheListener listener : listeners) {
if (listener instanceof RegionMembershipListener) {
RegionMembershipListener regionMembershipListener = (RegionMembershipListener) listener;
try {
DistributedMember[] otherDms = others.toArray(new DistributedMember[0]);
regionMembershipListener.initialMembers(this, otherDms);
} catch (VirtualMachineError err) {
SystemFailure.initiateFailure(err);
// If this ever returns, rethrow the error. We're poisoned
// now, so don't let this thread continue.
throw err;
} catch (Throwable t) {
// Whenever you catch Error or Throwable, you must also
// catch VirtualMachineError (see above). However, there is
// _still_ a possibility that you are dealing with a cascading
// error condition, so you also need to check to see if the JVM
// is still usable:
SystemFailure.checkFailure();
logger.error("Exception occurred in RegionMembershipListener",
t);
}
}
}
}
}
}
/**
* Free resources held by this region. This method is invoked after isDestroyed has been set to
* true.
*/
@Override
protected void postDestroyRegion(boolean destroyDiskRegion, RegionEventImpl event) {
distributedRegionCleanup(event);
try {
super.postDestroyRegion(destroyDiskRegion, event);
} catch (CancelException e) {
// I don't think this should ever happens: bulletproofing for bug 39454
logger.warn("postDestroyRegion: encountered cancellation", e);
}
}
@Override
public void cleanupFailedInitialization() {
super.cleanupFailedInitialization();
try {
RegionEventImpl ev = new RegionEventImpl(this, Operation.REGION_CLOSE, null, false, getMyId(),
generateEventID());
distributeDestroyRegion(ev, true);
distributedRegionCleanup(null);
} catch (RegionDestroyedException ignore) {
// someone else must have concurrently destroyed the region (maybe a distributed destroy)
} catch (CancelException ignore) {
// cache or DS is closed, ignore
} catch (VirtualMachineError e) {
SystemFailure.initiateFailure(e);
throw e;
} catch (Throwable t) {
logger.warn("Error cleaning up after failed region initialization of region " + this,
t);
}
}
@Override
public void handleCacheClose(Operation operation) {
try {
super.handleCacheClose(operation);
} finally {
distributedRegionCleanup(null);
}
}
/**
* invoke a cache writer before a put is performed elsewhere
*/
@Override
public void cacheWriteBeforePut(EntryEventImpl event, Set netWriteRecipients,
CacheWriter localWriter, boolean requireOldValue, Object expectedOldValue)
throws CacheWriterException, TimeoutException {
if ((localWriter != null || (netWriteRecipients != null && !netWriteRecipients.isEmpty()))
&& !event.inhibitAllNotifications()) {
final boolean isNewKey = event.getOperation().isCreate();
final long start = getCachePerfStats().startCacheWriterCall();
try {
SearchLoadAndWriteProcessor processor = SearchLoadAndWriteProcessor.getProcessor();
processor.initialize(this, "preUpdate", null);
try {
if (!isNewKey) {
processor.doNetWrite(event, netWriteRecipients, localWriter,
SearchLoadAndWriteProcessor.BEFOREUPDATE);
} else {
processor.doNetWrite(event, netWriteRecipients, localWriter,
SearchLoadAndWriteProcessor.BEFORECREATE);
}
} finally {
processor.release();
}
} finally {
getCachePerfStats().endCacheWriterCall(start);
}
}
serverPut(event, requireOldValue, expectedOldValue);
}
@Override
protected void cacheListenersChanged(boolean nowHasListener) {
if (nowHasListener) {
advisorListener.initRMLWrappers();
}
new UpdateAttributesProcessor(this).distribute();
}
@Override
protected void cacheWriterChanged(CacheWriter oldWriter) {
super.cacheWriterChanged(oldWriter);
if (oldWriter == null ^ basicGetWriter() == null) {
new UpdateAttributesProcessor(this).distribute();
}
}
@Override
protected void cacheLoaderChanged(CacheLoader oldLoader) {
super.cacheLoaderChanged(oldLoader);
if (oldLoader == null ^ basicGetLoader() == null) {
new UpdateAttributesProcessor(this).distribute();
}
}
@Override
public void addGatewaySenderId(String gatewaySenderId) {
super.addGatewaySenderId(gatewaySenderId);
new UpdateAttributesProcessor(this).distribute();
updateSenderIdMonitor();
}
@Override
public void removeGatewaySenderId(String gatewaySenderId) {
super.removeGatewaySenderId(gatewaySenderId);
new UpdateAttributesProcessor(this).distribute();
updateSenderIdMonitor();
}
@Override
public void addAsyncEventQueueId(String asyncEventQueueId) {
super.addAsyncEventQueueId(asyncEventQueueId);
new UpdateAttributesProcessor(this).distribute();
updateSenderIdMonitor();
}
@Override
public void removeAsyncEventQueueId(String asyncEventQueueId) {
super.removeAsyncEventQueueId(asyncEventQueueId);
new UpdateAttributesProcessor(this).distribute();
updateSenderIdMonitor();
}
SenderIdMonitor createSenderIdMonitor() {
return SenderIdMonitor.createSenderIdMonitor(this, this.distAdvisor);
}
void updateSenderIdMonitor() {
this.senderIdMonitor.update();
}
@Override
void checkSameSenderIdsAvailableOnAllNodes() {
this.senderIdMonitor.checkSenderIds();
}
/**
* Wraps call to dlock service in order to throw RegionDestroyedException if dlock service throws
* IllegalStateException and isDestroyed is true.
*/
private boolean isLockingSuspendedByCurrentThread() {
try {
return getLockService().isLockingSuspendedByCurrentThread();
} catch (IllegalStateException e) {
lockCheckReadiness();
throw e;
}
}
/**
* If this region's scope is GLOBAL, get a distributed lock on the given key, and return the Lock.
* The sender is responsible for unlocking.
*
* @return the acquired Lock if the region is GLOBAL, otherwise null.
*
* @throws NullPointerException if key is null
*/
private Lock getDistributedLockIfGlobal(Object key) throws TimeoutException {
if (getScope().isGlobal()) {
if (isLockingSuspendedByCurrentThread())
return null;
long start = System.currentTimeMillis();
long timeLeft = getCache().getLockTimeout();
long lockTimeout = timeLeft;
String msg = null;
Object[] msgArgs = null;
while (timeLeft > 0 || lockTimeout == -1) {
cache.getCancelCriterion().checkCancelInProgress(null);
boolean interrupted = Thread.interrupted();
try {
Lock dlock = getDistributedLock(key);
if (!dlock.tryLock(timeLeft, TimeUnit.SECONDS)) {
msg = "Attempt to acquire distributed lock for %s failed after waiting %s seconds.";
msgArgs = new Object[] {key, (System.currentTimeMillis() - start) / 1000L};
break;
}
return dlock;
} catch (InterruptedException ex) {
interrupted = true;
cache.getCancelCriterion().checkCancelInProgress(ex);
// TODO: Why is it OK to keep going?
if (lockTimeout > -1) {
timeLeft = getCache().getLockTimeout() - (System.currentTimeMillis() - start) / 1000L;
}
} finally {
if (interrupted) {
Thread.currentThread().interrupt();
}
}
} // while
if (msg == null) {
msg = "Timed out after waiting %s seconds for the distributed lock for %s.";
msgArgs = new Object[] {getCache().getLockTimeout(), key};
}
throw new TimeoutException(String.format(msg, msgArgs));
} else {
return null;
}
}
/**
* Checks if the entry is a valid entry
*
* @return true if entry not null or entry is not removed
*/
protected boolean checkEntryNotValid(RegionEntry mapEntry) {
return mapEntry == null || mapEntry.isRemoved() && !mapEntry.isTombstone();
}
/**
* Get the best iterator for iterating over the contents of this region. This method will either
* an iterator that uses hash ordering from the entry map, or, in the case of an overflow region,
* an iterator that iterates over the entries in disk order.
*/
@Override
public Iterator<RegionEntry> getBestIterator(boolean includeValues) {
DiskRegion dr = getDiskRegion();
if (DiskPage.DISK_PAGE_SIZE > 0 && includeValues && dr != null) {
// Wait for the disk region to recover values first.
dr.waitForAsyncRecovery();
if (dr.getNumOverflowOnDisk() > 0) {
return new DiskSavvyIterator();
}
}
return entries.regionEntries().iterator();
}
/**
* Should only be used if this region has entries on disk that are not in memory. This currently
* happens for overflow and for recovery when values are not recovered. The first iteration does a
* normal iteration of the regionEntries. But if it finds an entry that is currently only on disk
* it saves it in a list sorted by the location on disk. Once the regionEntries iterator has
* nothing more to iterate it starts iterating over, in disk order, the entries on disk.
*/
private class DiskSavvyIterator implements Iterator<RegionEntry> {
private boolean usingIt = true;
private Iterator<?> it = entries.regionEntries().iterator();
// iterator for nested ArrayLists
private Iterator<RegionEntry> subIt = null;
private final TreeMap<DiskPage, Object> diskMap = new TreeMap<>();
DiskSavvyIterator() {}
@Override
public boolean hasNext() {
boolean result;
if (subIt != null) {
result = subIt.hasNext();
if (!result) {
subIt = null;
} else {
return result;
}
}
result = it.hasNext();
if (usingIt && !result) {
usingIt = false;
it = diskMap.values().iterator();
result = it.hasNext();
}
return result;
}
@Override
public RegionEntry next() {
for (;;) {
if (subIt != null) {
return subIt.next();
}
if (usingIt) {
RegionEntry regionEntry = (RegionEntry) it.next();
DiskPosition diskPosition = new DiskPosition();
if (regionEntry.isOverflowedToDisk(DistributedRegion.this, diskPosition)) {
// add diskPosition to sorted list
DiskPage dPage = new DiskPage(diskPosition);
Object value = diskMap.get(dPage);
if (value == null) {
diskMap.put(dPage, regionEntry);
} else if (value instanceof ArrayList) {
List<Object> list = (ArrayList<Object>) value;
list.add(regionEntry);
} else {
List<Object> list = new ArrayList<>();
list.add(value);
list.add(regionEntry);
diskMap.put(dPage, list);
}
if (!hasNext()) {
throw new NoSuchElementException();
}
} else {
return regionEntry;
}
} else {
Object value = it.next();
if (value instanceof ArrayList) {
List<RegionEntry> list = (List<RegionEntry>) value;
subIt = list.iterator();
return subIt.next();
} else {
return (RegionEntry) value;
}
}
}
}
@Override
public void remove() {
throw new UnsupportedOperationException();
}
}
public static class DiskPosition implements Comparable<DiskPosition> {
long oplogId; // package-private to avoid synthetic accessor
long offset; // package-private to avoid synthetic accessor
DiskPosition() {}
public void setPosition(long oplogId, long offset) {
this.oplogId = oplogId;
this.offset = offset;
}
@Override
public int hashCode() {
// TODO: Object instantiation inside 'hashCode()' is bad
return Long.valueOf(oplogId ^ offset).hashCode();
}
@Override
public boolean equals(Object obj) {
if (obj instanceof DiskPosition) {
DiskPosition other = (DiskPosition) obj;
return oplogId == other.oplogId && offset == other.offset;
} else {
return false;
}
}
@Override
public int compareTo(DiskPosition o) {
int result = Long.signum(oplogId - o.oplogId);
if (result == 0) {
result = Long.signum(offset - o.offset);
}
return result;
}
@Override
public String toString() {
return "<" + oplogId + ':' + offset + '>';
}
}
static class DiskPage extends DiskPosition {
static final long DISK_PAGE_SIZE =
Long.getLong(DistributionConfig.GEMFIRE_PREFIX + "DISK_PAGE_SIZE", 8 << 10);
DiskPage(DiskPosition diskPosition) {
setPosition(diskPosition.oplogId, diskPosition.offset / DISK_PAGE_SIZE);
}
}
/**
* Returns the lock lease value to use for DistributedLock and RegionDistributedLock. -1 is
* supported as non-expiring lock.
*/
private long getLockLeaseForLock() { // package-private to avoid synthetic accessor
if (getCache().getLockLease() == -1) {
return -1;
}
return (getCache().getLockLease()) * 1000L;
}
/**
* Returns the lock timeout value to use for DistributedLock and RegionDistributedLock. -1 is
* supported as a lock that never times out.
*/
private long getLockTimeoutForLock(long time, TimeUnit unit) { // package-private to avoid
// synthetic
// accessor
if (time == -1) {
return -1;
}
return TimeUnit.MILLISECONDS.convert(time, unit);
}
private class DistributedLock implements Lock {
private final Object key;
DistributedLock(Object key) {
this.key = key;
}
@Override
public void lock() {
try {
boolean locked = basicTryLock(-1, TimeUnit.MILLISECONDS, false);
if (!locked) {
lockCheckReadiness();
}
Assert.assertTrue(locked, "Failed to acquire DistributedLock");
} catch (IllegalStateException ex) {
lockCheckReadiness();
throw ex;
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
lockCheckReadiness();
Assert.assertTrue(false, "Failed to acquire DistributedLock");
}
}
@Override
public void lockInterruptibly() throws InterruptedException {
try {
boolean locked = basicTryLock(-1, TimeUnit.MILLISECONDS, true);
if (!locked) {
lockCheckReadiness();
}
Assert.assertTrue(locked, "Failed to acquire DistributedLock");
} catch (IllegalStateException ex) {
lockCheckReadiness();
throw ex;
}
}
@Override
public boolean tryLock() {
try {
ReplyProcessor21.forceSevereAlertProcessing();
return getLockService().lock(key, 0, getLockLeaseForLock());
} catch (IllegalStateException ex) {
lockCheckReadiness();
throw ex;
} finally {
ReplyProcessor21.unforceSevereAlertProcessing();
}
}
@Override
public boolean tryLock(long time, TimeUnit unit) throws InterruptedException {
return basicTryLock(time, unit, true);
}
// TODO: basicTryLock method is too complex for IDE to analyze
private boolean basicTryLock(long time, TimeUnit unit, boolean interruptible)
throws InterruptedException {
// if (Thread.interrupted()) throw new InterruptedException(); not necessary lockInterruptibly
// does this
final DistributionManager dm = getDistributionManager();
long start = System.currentTimeMillis();
long timeoutMS = getLockTimeoutForLock(time, unit);
long end;
if (timeoutMS < 0) {
timeoutMS = Long.MAX_VALUE;
end = Long.MAX_VALUE;
} else {
end = start + timeoutMS;
}
long ackSAThreshold = getSystem().getConfig().getAckSevereAlertThreshold() * 1000L;
long waitInterval;
long ackWaitThreshold;
if (ackSAThreshold > 0) {
ackWaitThreshold = getSystem().getConfig().getAckWaitThreshold() * 1000L;
waitInterval = ackWaitThreshold;
} else {
waitInterval = timeoutMS;
ackWaitThreshold = 0;
}
boolean suspected = false;
boolean severeAlertIssued = false;
DistributedMember lockHolder = null;
do {
try {
waitInterval = Math.min(end - System.currentTimeMillis(), waitInterval);
ReplyProcessor21.forceSevereAlertProcessing();
final boolean gotLock;
if (interruptible) {
gotLock =
getLockService().lockInterruptibly(key, waitInterval, getLockLeaseForLock());
} else {
gotLock = getLockService().lock(key, waitInterval, getLockLeaseForLock());
}
if (gotLock) {
return true;
}
if (ackSAThreshold > 0) {
long elapsed = System.currentTimeMillis() - start;
if (elapsed > ackWaitThreshold) {
if (!suspected) {
// start suspect processing on the holder of the lock
suspected = true;
severeAlertIssued = false; // in case this is a new lock holder
waitInterval = ackSAThreshold;
DLockRemoteToken remoteToken = ((DLockService) getLockService()).queryLock(key);
lockHolder = remoteToken.getLessee();
if (lockHolder != null) {
dm.getMembershipManager().suspectMember((InternalDistributedMember) lockHolder,
"Has not released a global region entry lock in over "
+ ackWaitThreshold / 1000 + " seconds");
}
} else if (elapsed > ackSAThreshold) {
DLockRemoteToken remoteToken = ((DLockService) getLockService()).queryLock(key);
if (lockHolder != null && remoteToken.getLessee() != null
&& lockHolder.equals(remoteToken.getLessee())) {
if (!severeAlertIssued) {
severeAlertIssued = true;
logger.fatal(
"{} seconds have elapsed waiting for global region entry lock held by {}",
ackWaitThreshold + ackSAThreshold, lockHolder);
}
} else {
// the lock holder has changed
suspected = false;
waitInterval = ackWaitThreshold;
lockHolder = null;
}
}
}
} // ackSAThreshold processing
} catch (IllegalStateException ex) {
lockCheckReadiness();
throw ex;
} finally {
ReplyProcessor21.unforceSevereAlertProcessing();
}
} while (System.currentTimeMillis() < end);
return false;
}
@Override
public void unlock() {
try {
ReplyProcessor21.forceSevereAlertProcessing();
getLockService().unlock(key);
if (!entries.containsKey(key)) {
getLockService().freeResources(key);
}
} catch (IllegalStateException ex) {
lockCheckReadiness();
throw ex;
} finally {
ReplyProcessor21.unforceSevereAlertProcessing();
}
}
@Override
public Condition newCondition() {
throw new UnsupportedOperationException(
"newCondition unsupported");
}
}
private class RegionDistributedLock implements Lock {
RegionDistributedLock() {}
@Override
public void lock() {
try {
boolean locked = getLockService().suspendLocking(-1);
Assert.assertTrue(locked, "Failed to acquire RegionDistributedLock");
} catch (IllegalStateException ex) {
lockCheckReadiness();
throw ex;
}
}
@Override
public void lockInterruptibly() throws InterruptedException {
// suspendLockingInterruptibly checks Thread.interrupted() to throw InterruptedException
try {
boolean locked = getLockService().suspendLockingInterruptibly(-1);
Assert.assertTrue(locked, "Failed to acquire RegionDistributedLock");
} catch (IllegalStateException ex) {
lockCheckReadiness();
throw ex;
}
}
@Override
public boolean tryLock() {
try {
return getLockService().suspendLocking(0);
} catch (IllegalStateException ex) {
lockCheckReadiness();
throw ex;
}
}
@Override
public boolean tryLock(long time, TimeUnit unit) throws InterruptedException {
// suspendLockingInterruptibly checks Thread.interrupted() to throw InterruptedException
try {
return getLockService().suspendLockingInterruptibly(getLockTimeoutForLock(time, unit));
} catch (IllegalStateException ex) {
lockCheckReadiness();
throw ex;
}
}
@Override
public void unlock() {
try {
getLockService().resumeLocking();
} catch (IllegalStateException ex) {
lockCheckReadiness();
throw ex;
}
}
@Override
public Condition newCondition() {
throw new UnsupportedOperationException("newCondition unsupported");
}
}
/**
* If this region's scope is GLOBAL, get the region distributed lock. The sender is responsible
* for unlocking.
*
* @return the acquired Lock if the region is GLOBAL and not already suspend, otherwise null.
*/
private Lock getRegionDistributedLockIfGlobal() throws TimeoutException {
if (getScope().isGlobal()) {
if (isLockingSuspendedByCurrentThread())
return null;
Lock dlock = getRegionDistributedLock();
dlock.lock(); // caller is expected to use a try-finally to unlock
return dlock;
}
return null;
}
/**
* Distribute the PutAllOp. This implementation distributes it to peers.
*
* @return token >0 means startOperation finished distribution
* @since GemFire 5.7
*/
@Override
public long postPutAllSend(DistributedPutAllOperation putAllOp,
VersionedObjectList successfulPuts) {
long token = -1;
if (putAllOp.putAllDataSize > 0) {
token = putAllOp.startOperation();
} else {
if (logger.isDebugEnabled()) {
logger.debug("DR.postPutAll: no data to distribute");
}
}
return token;
}
@Override
public long postRemoveAllSend(DistributedRemoveAllOperation op,
VersionedObjectList successfulOps) {
long token = -1;
if (op.removeAllDataSize > 0) {
token = op.startOperation();
} else {
getCache().getLogger().fine("DR.postRemoveAll: no data to distribute");
}
return token;
}
@Override
VersionedObjectList basicPutAll(final Map<?, ?> map,
final DistributedPutAllOperation putAllOp, final Map<Object, VersionTag> retryVersions) {
Lock dlock = getRegionDistributedLockIfGlobal();
try {
return super.basicPutAll(map, putAllOp, retryVersions);
} finally {
if (dlock != null) {
dlock.unlock();
}
}
}
@Override
public VersionedObjectList basicRemoveAll(final Collection<Object> keys,
final DistributedRemoveAllOperation removeAllOp, final List<VersionTag> retryVersions) {
Lock dlock = getRegionDistributedLockIfGlobal();
try {
return super.basicRemoveAll(keys, removeAllOp, retryVersions);
} finally {
if (dlock != null) {
dlock.unlock();
}
}
}
/**
* Returns the missing required roles after waiting up to the timeout
*
* @throws IllegalStateException if region is not configured with required roles
*/
public Set waitForRequiredRoles(long timeout) throws InterruptedException {
if (Thread.interrupted()) {
throw new InterruptedException();
}
checkReadiness();
if (!getMembershipAttributes().hasRequiredRoles()) {
throw new IllegalStateException(
"Region has not been configured with required roles.");
}
if (!isMissingRequiredRoles) { // should we delete this check?
if (logger.isDebugEnabled()) {
logger.debug("No missing required roles to wait for.");
}
return Collections.emptySet(); // early-out: no missing required roles
}
if (timeout != 0) { // if timeout is zero then fall through past waits
if (timeout == -1) { // infinite timeout
while (isMissingRequiredRoles) {
checkReadiness();
// bail if distribution has stopped
cache.getCancelCriterion().checkCancelInProgress(null);
synchronized (missingRequiredRoles) {
// one more check while synced
if (isMissingRequiredRoles) {
if (logger.isDebugEnabled()) {
logger.debug("About to wait for missing required roles.");
}
// TODO an infinite wait here might be a problem...
missingRequiredRoles.wait(); // spurious wakeup ok
}
}
}
} else { // use the timeout
long endTime = System.currentTimeMillis() + timeout;
while (isMissingRequiredRoles) {
checkReadiness();
// bail if distribution has stopped
cache.getCancelCriterion().checkCancelInProgress(null);
synchronized (missingRequiredRoles) {
// one more check while synced
if (isMissingRequiredRoles) {
long timeToWait = endTime - System.currentTimeMillis();
if (timeToWait > 0) {
if (logger.isDebugEnabled()) {
logger.debug("About to wait up to {} milliseconds for missing required roles.",
timeToWait);
}
missingRequiredRoles.wait(timeToWait); // spurious wakeup ok
} else {
break;
}
}
}
}
}
}
// check readiness again: thread may have been notified at destroy time
checkReadiness();
if (isMissingRequiredRoles) {
// sync on missingRequiredRoles to prevent mods to required role status...
synchronized (missingRequiredRoles) {
return Collections.unmodifiableSet(new HashSet<>(missingRequiredRoles));
}
} else {
return Collections.emptySet();
}
}
/** Returns true if the role is currently present this region's membership. */
public boolean isRoleInRegionMembership(Role role) {
checkReadiness();
return basicIsRoleInRegionMembership(role);
}
private boolean basicIsRoleInRegionMembership(Role role) {
if (getSystem().getDistributedMember().getRoles().contains(role)) {
// since we are playing the role
return true;
}
for (DistributedMember member : distAdvisor.adviseGeneric()) {
Set<Role> roles = member.getRoles();
if (roles.contains(role)) {
return true;
}
}
return false;
}
@Override
public void remoteRegionInitialized(CacheProfile profile) {
synchronized (advisorListener) {
if (advisorListener.members == null && hasListener()) {
Object callback = TEST_HOOK_ADD_PROFILE ? profile : null;
RegionEventImpl event = new RegionEventImpl(this, Operation.REGION_CREATE, callback, true,
profile.peerMemberId);
dispatchListenerEvent(EnumListenerEvent.AFTER_REMOTE_REGION_CREATE, event);
}
}
}
@Override
void removeSenderFromAdvisor(InternalDistributedMember sender, int serial,
boolean regionDestroyed) {
getDistributionAdvisor().removeIdWithSerial(sender, serial, regionDestroyed);
}
/** doesn't throw RegionDestroyedException, used by CacheDistributionAdvisor */
@Override
public DistributionAdvisee getParentAdvisee() {
return (DistributionAdvisee) basicGetParentRegion();
}
/**
* Used to get membership events from our advisor to implement RegionMembershipListener
* invocations.
*
* @since GemFire 5.0
*/
protected class AdvisorListener implements MembershipListener {
Set<InternalDistributedMember> members = new HashSet<>(); // package-private to avoid synthetic
// accessor
protected boolean destroyed = false;
synchronized void addMembers(Set<InternalDistributedMember> newMembers) {
members.addAll(newMembers);
}
protected synchronized Set<InternalDistributedMember> getInitialMembers() {
Set<InternalDistributedMember> initMembers = members;
members = null;
return initMembers;
}
@Override
public void quorumLost(DistributionManager distributionManager,
Set<InternalDistributedMember> failures, List<InternalDistributedMember> remaining) {
// do nothing
}
@Override
public void memberSuspect(DistributionManager distributionManager, InternalDistributedMember id,
InternalDistributedMember whoSuspected, String reason) {
// do nothing
}
/** called when membership listeners are added after region creation */
synchronized void initRMLWrappers() {
Set<InternalDistributedMember> membersWithThisRegion =
distAdvisor.adviseGeneric();
initPostCreateRegionMembershipListeners(membersWithThisRegion);
}
@Override
public synchronized void memberJoined(DistributionManager distributionManager,
InternalDistributedMember id) {
if (destroyed) {
return;
}
if (members != null) {
members.add(id);
}
// bug #44684 - do not notify listener of create until remote member is initialized
if (getMembershipAttributes().hasRequiredRoles()) {
// newlyAcquiredRoles is used for intersection and RoleEvent
Set<Role> newlyAcquiredRoles = Collections.emptySet();
synchronized (missingRequiredRoles) {
if (isMissingRequiredRoles) {
Set<Role> roles = id.getRoles();
newlyAcquiredRoles = new HashSet<>(missingRequiredRoles);
newlyAcquiredRoles.retainAll(roles); // find the intersection
if (!newlyAcquiredRoles.isEmpty()) {
missingRequiredRoles.removeAll(newlyAcquiredRoles);
if (members == null && missingRequiredRoles.isEmpty()) {
isMissingRequiredRoles = false;
getCachePerfStats().incReliableRegionsMissing(-1);
if (getMembershipAttributes().getLossAction().isAllAccess())
getCachePerfStats().incReliableRegionsMissingFullAccess(-1); // rahul
else if (getMembershipAttributes().getLossAction().isLimitedAccess())
getCachePerfStats().incReliableRegionsMissingLimitedAccess(-1);
else if (getMembershipAttributes().getLossAction().isNoAccess())
getCachePerfStats().incReliableRegionsMissingNoAccess(-1);
boolean async = resumeReliability(id, newlyAcquiredRoles);
if (async) {
destroyed = true;
}
}
}
}
if (!destroyed) {
// any number of threads may be waiting on missingRequiredRoles
missingRequiredRoles.notifyAll();
}
}
if (!destroyed && members == null && hasListener()) {
if (!newlyAcquiredRoles.isEmpty()) {
// fire afterRoleGain event
RoleEventImpl relEvent = new RoleEventImpl(DistributedRegion.this,
Operation.REGION_CREATE, null, true, id, newlyAcquiredRoles);
dispatchListenerEvent(EnumListenerEvent.AFTER_ROLE_GAIN, relEvent);
}
}
}
}
@Override
public synchronized void memberDeparted(DistributionManager distributionManager,
InternalDistributedMember id, boolean crashed) {
if (destroyed) {
return;
}
if (members != null) {
members.remove(id);
}
if (members == null && hasListener()) {
RegionEventImpl event =
new RegionEventImpl(DistributedRegion.this, Operation.REGION_CLOSE, null, true, id);
if (crashed) {
dispatchListenerEvent(EnumListenerEvent.AFTER_REMOTE_REGION_CRASH, event);
} else {
// @todo darrel: it would be nice to know if what actual op was done
// could be close, local destroy, or destroy (or load snap?)
if (DestroyRegionOperation.isRegionDepartureNotificationOk()) {
dispatchListenerEvent(EnumListenerEvent.AFTER_REMOTE_REGION_DEPARTURE, event);
}
}
}
if (getMembershipAttributes().hasRequiredRoles()) {
Set<Role> newlyMissingRoles = Collections.emptySet();
synchronized (missingRequiredRoles) {
Set<Role> roles = id.getRoles();
for (Role role : roles) {
if (getMembershipAttributes().getRequiredRoles().contains(role)
&& !basicIsRoleInRegionMembership(role)) {
if (newlyMissingRoles.isEmpty()) {
newlyMissingRoles = new HashSet<>();
}
newlyMissingRoles.add(role);
if (members == null && !isMissingRequiredRoles) {
isMissingRequiredRoles = true;
getCachePerfStats().incReliableRegionsMissing(1);
if (getMembershipAttributes().getLossAction().isAllAccess()) {
getCachePerfStats().incReliableRegionsMissingFullAccess(1); // rahul
} else if (getMembershipAttributes().getLossAction().isLimitedAccess()) {
getCachePerfStats().incReliableRegionsMissingLimitedAccess(1);
} else if (getMembershipAttributes().getLossAction().isNoAccess()) {
getCachePerfStats().incReliableRegionsMissingNoAccess(1);
}
boolean async = lostReliability(id, newlyMissingRoles);
if (async) {
destroyed = true;
}
}
}
}
if (!destroyed) {
missingRequiredRoles.addAll(newlyMissingRoles);
// any number of threads may be waiting on missingRequiredRoles...
missingRequiredRoles.notifyAll();
}
}
if (!destroyed && members == null && hasListener()) {
if (!newlyMissingRoles.isEmpty()) {
// fire afterRoleLoss event
RoleEventImpl relEvent = new RoleEventImpl(DistributedRegion.this,
Operation.REGION_CLOSE, null, true, id, newlyMissingRoles);
dispatchListenerEvent(EnumListenerEvent.AFTER_ROLE_LOSS, relEvent);
}
}
}
}
}
/**
* Used to bootstrap txState.
*
* @return member with primary bucket for partitionedRegions
*/
@Override
public DistributedMember getOwnerForKey(KeyInfo key) {
assert !isInternalRegion() || isMetaRegionWithTransactions();
if (!getAttributes().getDataPolicy().withStorage() || (getConcurrencyChecksEnabled()
&& getAttributes().getDataPolicy() == DataPolicy.NORMAL)) {
// execute on random replicate
return getRandomReplicate();
}
// if we are non-persistent, forward transactions to
// a persistent member
if (getConcurrencyChecksEnabled() && !generateVersionTag) {
return getRandomPersistentReplicate();
}
return super.getOwnerForKey(key);
}
/**
* Execute the provided named function in all locations that contain the given keys. So function
* can be executed on just one fabric node, executed in parallel on a subset of nodes in parallel
* across all the nodes.
*
* @since GemFire 5.8
*/
@Override
public ResultCollector executeFunction(final DistributedRegionFunctionExecutor execution,
final Function function, final Object args, final ResultCollector rc, final Set filter,
final ServerToClientFunctionResultSender sender) {
DistributedMember target = getTransactionalNode();
if (target != null) {
if (target.equals(getMyId())) {
return executeLocally(execution, function, args, 0, rc, filter, sender);
}
return executeOnReplicate(execution, function, args, rc, filter, target);
} else if (getAttributes().getDataPolicy().withReplication()
|| getAttributes().getDataPolicy().withPreloaded()) {
// execute locally
final Set<InternalDistributedMember> singleMember = Collections.singleton(getMyId());
execution.validateExecution(function, singleMember);
execution.setExecutionNodes(singleMember);
return executeLocally(execution, function, args, 0, rc, filter, sender);
} else {
// select a random replicate
target = getRandomReplicate();
if (target == null) {
throw new FunctionException(
String.format("No Replicated Region found for executing function : %s.",
function.getId()));
}
}
final LocalResultCollector<?, ?> localRC = execution.getLocalResultCollector(function, rc);
return executeOnReplicate(execution, function, args, localRC, filter, target);
}
private ResultCollector executeOnReplicate(final DistributedRegionFunctionExecutor execution,
final Function function, final Object args, ResultCollector rc, final Set filter,
final DistributedMember target) {
final Set<InternalDistributedMember> singleMember =
Collections.singleton((InternalDistributedMember) target);
execution.validateExecution(function, singleMember);
execution.setExecutionNodes(singleMember);
HashMap<InternalDistributedMember, Object> memberArgs = new HashMap<>();
memberArgs.put((InternalDistributedMember) target,
execution.getArgumentsForMember(target.getId()));
ResultSender resultSender = new DistributedRegionFunctionResultSender(null, rc, function,
execution.getServerResultSender());
DistributedRegionFunctionResultWaiter waiter =
new DistributedRegionFunctionResultWaiter(getSystem(), getFullPath(), rc,
function, filter, Collections.singleton(target), memberArgs, resultSender);
rc = waiter.getFunctionResultFrom(Collections.singleton(target), function, execution);
return rc;
}
/**
* @return the node which a transaction is already is progress, null otherwise
*/
private DistributedMember getTransactionalNode() {
if (cache.getTxManager().getTXState() != null) {
return cache.getTxManager().getTXState().getTarget();
}
return null;
}
/**
* Implementation of {@link ProfileVisitor} that selects a random replicated member from the
* available ones for this region.
*/
static class GetRandomReplicate implements ProfileVisitor<DistributedMember> {
private boolean onlyPersistent = false;
InternalDistributedMember member = null;
private int randIndex = -1;
GetRandomReplicate() {}
GetRandomReplicate(boolean onlyPersistent) {
this.onlyPersistent = onlyPersistent;
}
@Override
public boolean visit(DistributionAdvisor advisor, Profile profile, int profileIndex,
int numProfiles, DistributedMember member) {
final CacheProfile cp = (CacheProfile) profile;
if (randIndex < 0) {
randIndex = PartitionedRegion.RANDOM.nextInt(numProfiles);
}
if (cp.dataPolicy.withReplication() && cp.regionInitialized) {
if (onlyPersistent && !cp.dataPolicy.withPersistence()) {
return true;
}
// store the last replicated member in any case since in the worst case
// there may be no replicated node after "randIndex" in which case the
// last visited member will be used
this.member = cp.getDistributedMember();
return profileIndex < randIndex;
}
return true;
}
}
/**
* @return a random replicate, null if there are none
*/
private InternalDistributedMember getRandomReplicate() {
/*
* [sumedh] The old code causes creation of a unnecessary HashSet and population with all
* replicates (which may be large), then copy into an array and then selection of a random one
* from that. The new approach uses a much more efficient visitor instead. Set replicates =
* this.getCacheDistributionAdvisor().adviseReplicates(); if (replicates.isEmpty()) { return
* null; } return (InternalDistributedMember)(replicates .toArray()[new
* Random().nextInt(replicates.size())]);
*/
final GetRandomReplicate getReplicate = new GetRandomReplicate();
getCacheDistributionAdvisor().accept(getReplicate, null);
return getReplicate.member;
}
/**
* @return a random persistent replicate, null if there is none
*/
private InternalDistributedMember getRandomPersistentReplicate() {
final GetRandomReplicate getPersistentReplicate = new GetRandomReplicate(true);
getCacheDistributionAdvisor().accept(getPersistentReplicate, null);
return getPersistentReplicate.member;
}
void executeOnRegion(DistributedRegionFunctionStreamingMessage msg, final Function function,
final Object args, int prid, final Set filter, boolean isReExecute) throws IOException {
final DistributionManager dm = getDistributionManager();
ResultSender resultSender = new DistributedRegionFunctionResultSender(dm, msg, function);
final RegionFunctionContextImpl context = new RegionFunctionContextImpl(cache, function.getId(),
this, args, filter, null, null, resultSender, isReExecute);
FunctionStats stats = FunctionStatsManager.getFunctionStats(function.getId(), dm.getSystem());
long start = stats.startFunctionExecution(function.hasResult());
try {
function.execute(context);
stats.endFunctionExecution(start, function.hasResult());
} catch (FunctionException functionException) {
if (logger.isDebugEnabled()) {
logger.debug("FunctionException occurred on remote node while executing Function: {}",
function.getId(), functionException);
}
stats.endFunctionExecutionWithException(start, function.hasResult());
throw functionException;
} catch (CacheClosedException cacheClosedexception) {
if (logger.isDebugEnabled()) {
logger.debug("CacheClosedException occurred on remote node while executing Function: {}",
function.getId(), cacheClosedexception);
}
throw cacheClosedexception;
} catch (Exception exception) {
if (logger.isDebugEnabled()) {
logger.debug("Exception occurred on remote node while executing Function: {}",
function.getId(), exception);
}
stats.endFunctionExecutionWithException(start, function.hasResult());
throw new FunctionException(exception);
}
}
private ResultCollector executeLocally(final DistributedRegionFunctionExecutor execution,
final Function function, final Object args, int prid,
final ResultCollector rc,
final Set filter,
final ServerToClientFunctionResultSender sender) {
final LocalResultCollector<?, ?> localRC = execution.getLocalResultCollector(function, rc);
final DistributionManager dm = getDistributionManager();
final DistributedRegionFunctionResultSender resultSender =
new DistributedRegionFunctionResultSender(dm, localRC, function, sender);
final RegionFunctionContextImpl context = new RegionFunctionContextImpl(cache, function.getId(),
this, args, filter, null, null, resultSender, execution.isReExecute());
execution.executeFunctionOnLocalNode(function, context, resultSender, dm, isTX());
return localRC;
}
@Override
void setMemoryThresholdFlag(MemoryEvent event) {
Set<InternalDistributedMember> others = getCacheDistributionAdvisor().adviseGeneric();
if (event.isLocal() || others.contains(event.getMember())) {
if (event.getState().isCritical() && !event.getPreviousState().isCritical()
&& (event.getType() == ResourceType.HEAP_MEMORY
|| (event.getType() == ResourceType.OFFHEAP_MEMORY && getOffHeap()))) {
addCriticalMember(event.getMember());
} else if (!event.getState().isCritical() && event.getPreviousState().isCritical()
&& (event.getType() == ResourceType.HEAP_MEMORY
|| (event.getType() == ResourceType.OFFHEAP_MEMORY && getOffHeap()))) {
removeCriticalMember(event.getMember());
}
}
}
@Override
public void removeCriticalMember(DistributedMember member) {
if (logger.isDebugEnabled()) {
logger.debug("DR: removing member {} from critical member list", member);
}
synchronized (memoryThresholdReachedMembers) {
memoryThresholdReachedMembers.remove(member);
if (memoryThresholdReachedMembers.isEmpty()) {
setMemoryThresholdReached(false);
}
}
}
@Override
void initialCriticalMembers(boolean localMemoryIsCritical,
Set<InternalDistributedMember> criticalMembers) {
Set<InternalDistributedMember> others = getCacheDistributionAdvisor().adviseGeneric();
for (InternalDistributedMember idm : criticalMembers) {
if (others.contains(idm)) {
addCriticalMember(idm);
}
}
}
/**
* @param idm member whose threshold has been exceeded
*/
protected void addCriticalMember(final DistributedMember idm) {
synchronized (memoryThresholdReachedMembers) {
if (memoryThresholdReachedMembers.isEmpty()) {
setMemoryThresholdReached(true);
}
memoryThresholdReachedMembers.add(idm);
}
}
@Override
public MemoryThresholdInfo getAtomicThresholdInfo() {
if (!isMemoryThresholdReached()) {
return MemoryThresholdInfo.getNotReached();
}
synchronized (memoryThresholdReachedMembers) {
return new MemoryThresholdInfo(isMemoryThresholdReached(),
new HashSet<>(memoryThresholdReachedMembers));
}
}
/**
* Fetch Version for the given key from a remote replicate member.
*
* @throws EntryNotFoundException if the entry is not found on replicate member
* @return VersionTag for the key
*/
protected VersionTag fetchRemoteVersionTag(Object key) {
VersionTag tag = null;
assert getDataPolicy() != DataPolicy.REPLICATE;
final TXStateProxy tx = cache.getTXMgr().pauseTransaction();
try {
boolean retry = true;
InternalDistributedMember member = getRandomReplicate();
while (retry) {
try {
if (member == null) {
break;
}
FetchVersionResponse response = RemoteFetchVersionMessage.send(member, this, key);
tag = response.waitForResponse();
retry = false;
} catch (RemoteOperationException e) {
member = getRandomReplicate();
if (member != null) {
if (logger.isDebugEnabled()) {
logger.debug("Retrying RemoteFetchVersionMessage on member:{}", member);
}
}
}
}
} finally {
cache.getTXMgr().unpauseTransaction(tx);
}
return tag;
}
/**
* Test hook for bug 48578. Returns true if it sees a net loader. Returns false if it does not
* have one.
*/
public boolean hasNetLoader() {
return hasNetLoader(getCacheDistributionAdvisor());
}
@Override
long getLatestLastAccessTimeFromOthers(Object key) {
LatestLastAccessTimeOperation op = new LatestLastAccessTimeOperation<>(this, key);
return op.getLatestLastAccessTime();
}
@Override
public Set adviseNetWrite() {
return getCacheDistributionAdvisor().adviseNetWrite();
}
@VisibleForTesting
public SenderIdMonitor getSenderIdMonitor() {
return senderIdMonitor;
}
}