Google Git
Sign in
apache / geode / 7038f08ea88ec42f6c58764525460a1c7fe1abcf / . / geode-core / src / main / java / org / apache / geode / internal / cache / BucketAdvisor.java
blob: 3a55af97341c3ab69d31ce79a0dd52dbc2014a76 [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 static org.apache.geode.internal.cache.CacheServerImpl.CACHE_SERVER_BIND_ADDRESS_NOT_AVAILABLE_EXCEPTION_MESSAGE;
import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
import java.io.ObjectInputStream;
import java.io.Serializable;
import java.util.AbstractSet;
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.Queue;
import java.util.Random;
import java.util.Set;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;
import org.apache.logging.log4j.Logger;
import org.apache.geode.CancelException;
import org.apache.geode.DataSerializer;
import org.apache.geode.annotations.Immutable;
import org.apache.geode.cache.RegionDestroyedException;
import org.apache.geode.cache.client.internal.locator.SerializationHelper;
import org.apache.geode.cache.partition.PartitionListener;
import org.apache.geode.cache.server.CacheServer;
import org.apache.geode.distributed.DistributedLockService;
import org.apache.geode.distributed.DistributedSystemDisconnectedException;
import org.apache.geode.distributed.LockNotHeldException;
import org.apache.geode.distributed.LockServiceDestroyedException;
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.DLockService;
import org.apache.geode.distributed.internal.locks.DistributedMemberLock;
import org.apache.geode.distributed.internal.membership.InternalDistributedMember;
import org.apache.geode.internal.Assert;
import org.apache.geode.internal.cache.partitioned.Bucket;
import org.apache.geode.internal.cache.partitioned.BucketProfileUpdateMessage;
import org.apache.geode.internal.cache.partitioned.DeposePrimaryBucketMessage;
import org.apache.geode.internal.cache.partitioned.DeposePrimaryBucketMessage.DeposePrimaryBucketResponse;
import org.apache.geode.internal.cache.partitioned.RegionAdvisor;
import org.apache.geode.internal.logging.log4j.LogMarker;
import org.apache.geode.internal.serialization.DeserializationContext;
import org.apache.geode.internal.serialization.SerializationContext;
import org.apache.geode.internal.util.StopWatch;
import org.apache.geode.logging.internal.log4j.api.LogService;
/**
* Specialized {@link CacheDistributionAdvisor} for {@link BucketRegion BucketRegions}. The
* <code>BucketAdvisor</code> is owned by a {@link ProxyBucketRegion} and may outlive a
* <code>BucketRegion</code>.
*/
@SuppressWarnings("synthetic-access")
public class BucketAdvisor extends CacheDistributionAdvisor {
private static final Logger logger = LogService.getLogger();
private static final boolean ENFORCE_SAFE_CLOSE = false;
// TODO: Boolean.getBoolean("gemfire.BucketAdvisor.debug.enforceSafeClose");
/** Reference to the InternalDistributedMember that is primary. */
private final AtomicReference<InternalDistributedMember> primaryMember = new AtomicReference<>();
/**
* Advice requests for {@link #adviseProfileUpdate()} delegate to the partitioned region's
* <code>RegionAdvisor</code> to include members with {@link ProxyBucketRegion}s as well as real
* {@link BucketRegion}s.
*/
private final RegionAdvisor regionAdvisor;
private final BucketRedundancyTracker redundancyTracker;
/**
* The bucket primary will be holding this distributed lock. Protected by synchronized(this).
*/
private DistributedMemberLock primaryLock;
// private static final byte MASK_HOSTING = 1; // 0001
// private static final byte MASK_VOLUNTEERING = 2; // 0010
// private static final byte MASK_OTHER_PRIMARY = 4; // 0100
// private static final byte MASK_IS_PRIMARY = 8; // 1000
private static final byte NO_PRIMARY_NOT_HOSTING = 0; // 0000_0000
private static final byte NO_PRIMARY_HOSTING = 1; // 0000_0001
private static final byte OTHER_PRIMARY_NOT_HOSTING = 4; // 0000_0100
private static final byte OTHER_PRIMARY_HOSTING = 5; // 0000_0101
private static final byte VOLUNTEERING_HOSTING = 3; // 0000_0011
private static final byte BECOMING_HOSTING = 15; // 0000_1111
private static final byte IS_PRIMARY_HOSTING = 9; // 0000_1001
private static final byte CLOSED = 16; // 0001_0000
/**
* The current state of this BucketAdvisor which tracks which member is primary and whether or not
* this member is hosting a real Bucket.
*/
private volatile byte primaryState = NO_PRIMARY_NOT_HOSTING;
/**
* This delegate handles all volunteering for primary status. Lazily created. Protected by
* synchronization(this).
*/
private VolunteeringDelegate volunteeringDelegate;
/**
* A random number generator
*
* @see #getPreferredNode()
*/
@Immutable
private static final Random myRand = new Random();
/**
* A read/write lock to prevent making this bucket not primary while a write is in progress on the
* bucket.
*/
private final ReadWriteLock primaryMoveReadWriteLock = new ReentrantReadWriteLock();
private final Lock primaryMoveReadLock = primaryMoveReadWriteLock.readLock();
private final Lock primaryMoveWriteLock = primaryMoveReadWriteLock.writeLock();
/**
* The advisor for the bucket region that we are colocated with, if this region is a colocated
* region.
*/
private BucketAdvisor parentAdvisor;
/**
* The member that is responsible for choosing the primary for this bucket. While this field is
* set and this member exists, this bucket won't try to become primary.
*/
private volatile InternalDistributedMember primaryElector;
private volatile BucketProfile localProfile;
private volatile boolean everHadPrimary = false;
private BucketAdvisor startingBucketAdvisor;
private PartitionedRegion pRegion;
private volatile boolean shadowBucketDestroyed;
/**
* Constructs a new BucketAdvisor for the Bucket owned by RegionAdvisor.
*
* @param bucket the bucket to provide metadata and advice for
* @param regionAdvisor advisor for the PartitionedRegion
*/
private BucketAdvisor(Bucket bucket, RegionAdvisor regionAdvisor) {
super(bucket);
this.regionAdvisor = regionAdvisor;
pRegion = this.regionAdvisor.getPartitionedRegion();
redundancyTracker =
new BucketRedundancyTracker(pRegion.getRedundantCopies(), pRegion.getRedundancyTracker());
resetParentAdvisor(bucket.getId());
}
public static BucketAdvisor createBucketAdvisor(Bucket bucket, RegionAdvisor regionAdvisor) {
BucketAdvisor advisor = new BucketAdvisor(bucket, regionAdvisor);
advisor.initialize();
return advisor;
}
private void resetParentAdvisor(int bucketId) {
PartitionedRegion colocatedRegion = ColocationHelper.getColocatedRegion(pRegion);
if (colocatedRegion != null) {
if (colocatedRegion.isFixedPartitionedRegion()) {
List<FixedPartitionAttributesImpl> fpas = colocatedRegion.getFixedPartitionAttributesImpl();
if (fpas != null) {
for (FixedPartitionAttributesImpl fpa : fpas) {
if (fpa.hasBucket(bucketId)) {
parentAdvisor =
colocatedRegion.getRegionAdvisor().getBucketAdvisor(fpa.getStartingBucketID());
break;
}
}
}
} else {
parentAdvisor = colocatedRegion.getRegionAdvisor().getBucketAdvisor(bucketId);
}
} else {
parentAdvisor = null;
}
}
private void assignStartingBucketAdvisorIfFixedPartitioned() {
if (startingBucketAdvisor != null) {
// already assigned
return;
}
if (pRegion.isFixedPartitionedRegion()) {
List<FixedPartitionAttributesImpl> fpas = pRegion.getFixedPartitionAttributesImpl();
if (fpas != null) {
int bucketId = getBucket().getId();
for (FixedPartitionAttributesImpl fpa : fpas) {
if (fpa.hasBucket(bucketId) && bucketId != fpa.getStartingBucketID()) {
startingBucketAdvisor = regionAdvisor.getBucketAdvisor(fpa.getStartingBucketID());
break;
}
}
}
}
}
/**
* Returns the lock that prevents the primary from moving while active writes are in progress.
* This should be locked before checking if the local bucket is primary.
*
* @return the lock for in-progress write operations
*/
public Lock getPrimaryMoveReadLock() {
return primaryMoveReadLock;
}
/**
* Returns the lock that prevents the parent's primary from moving while active writes are in
* progress. This should be locked before checking if the local bucket is primary.
*
* @return the lock for in-progress write operations
*/
Lock getParentPrimaryMoveReadLock() {
if (parentAdvisor != null) {
return parentAdvisor.getPrimaryMoveReadLock();
}
return null;
}
/**
* Try to lock the primary bucket to make sure no operation is on-going at current bucket.
*
*/
void tryLockIfPrimary() {
if (isPrimary()) {
try {
primaryMoveWriteLock.lock();
} finally {
primaryMoveWriteLock.unlock();
}
}
}
/**
* Makes this <code>BucketAdvisor</code> give up being a primary and become a secondary. Does
* nothing if not currently the primary.
*
* @return true if this advisor has been deposed as primary
*/
public boolean deposePrimary() {
if (isPrimary()) {
primaryMoveWriteLock.lock();
boolean needToSendProfileUpdate = false;
try {
removePrimary(getDistributionManager().getId());
synchronized (this) {
if (!isPrimary()) {
// releasePrimaryLock();
needToSendProfileUpdate = true;
return true;
} else {
return false; // failed for some reason
}
}
} finally {
primaryMoveWriteLock.unlock();
if (needToSendProfileUpdate) {
sendProfileUpdate();
}
}
} else {
sendProfileUpdate();
return true;
}
}
/**
* This calls deposePrimary on every colocated child that is directly colocated to this bucket's
* PR. Those each in turn do the same to their child buckets and so on before returning. Each
* depose will send a dlock release message to the grantor, wait for reply, and then also send a
* profile update.
* <p>
* Caller must synchronize on this BucketAdvisor.
*
*/
private void deposePrimaryForColocatedChildren() {
boolean deposedChildPrimaries = true;
List<PartitionedRegion> colocatedChildPRs = ColocationHelper.getColocatedChildRegions(pRegion);
if (colocatedChildPRs != null) {
for (PartitionedRegion pr : colocatedChildPRs) {
Bucket b = pr.getRegionAdvisor().getBucket(getBucket().getId());
if (b != null) {
BucketAdvisor ba = b.getBucketAdvisor();
deposedChildPrimaries = ba.deposePrimary() && deposedChildPrimaries;
if (b instanceof BucketRegionQueue) {
BucketRegionQueue brq = (BucketRegionQueue) b;
brq.decQueueSize(brq.size());
brq.incSecondaryQueueSize(brq.size());
}
}
}
}
}
private void deposeOtherPrimaryBucketForFixedPartition() {
boolean deposedOtherPrimaries = true;
int bucketId = getBucket().getId();
List<FixedPartitionAttributesImpl> fpas = pRegion.getFixedPartitionAttributesImpl();
if (fpas != null) {
for (FixedPartitionAttributesImpl fpa : fpas) {
if (fpa.getStartingBucketID() == bucketId) {
for (int i = (bucketId + 1); i <= fpa.getLastBucketID(); i++) {
Bucket b = regionAdvisor.getBucket(i);
if (b != null) {
BucketAdvisor ba = b.getBucketAdvisor();
deposedOtherPrimaries = ba.deposePrimary() && deposedOtherPrimaries;
}
}
}
}
}
}
void removeBucket() {
setHosting(false);
}
/**
* Return (and possibly choose) a thread-sticky member from whose data store this bucket's values
* should be read
*
* @return member to use for reads, null if none available
*/
public InternalDistributedMember getPreferredNode() {
if (isHosting()) {
getPartitionedRegionStats().incPreferredReadLocal();
return getDistributionManager().getId();
}
Profile[] locProfiles = profiles; // volatile read
if (locProfiles.length == 0) {
return null;
}
Profile selectedProfile = selectNotInitializingProfile(locProfiles);
if (selectedProfile == null) {
return null;
}
getPartitionedRegionStats().incPreferredReadRemote();
return selectedProfile.peerMemberId;
}
/**
* Get the random Profile that is not initializing BucketProfile.
*
*/
private Profile selectNotInitializingProfile(Profile[] inProfiles) {
int index = 0;
int offset = 0;
if (inProfiles.length > 1) {
// Pick random offset.
offset = myRand.nextInt(inProfiles.length);
}
for (int i = 0; i < inProfiles.length; i++) {
index = (offset + i) % inProfiles.length;
BucketProfile bp = (BucketProfile) inProfiles[index];
if (!bp.isInitializing) {
return inProfiles[index];
}
}
return null;
}
/**
* Returns the thread-safe queue of primary volunteering tasks for the parent Partitioned Region.
*
* @return the queue of primary volunteering tasks
*/
private Queue<Runnable> getVolunteeringQueue() {
return regionAdvisor.getVolunteeringQueue();
}
/**
* Returns the semaphore which controls the number of threads allowed to consume from the
* {@link #getVolunteeringQueue volunteering queue}.
*
* @return the semaphore which controls the number of volunteering threads
*/
private Semaphore getVolunteeringSemaphore() {
return regionAdvisor.getVolunteeringSemaphore();
}
/**
* Returns the PartitionedRegionStats.
*
* @return the PartitionedRegionStats
*/
private PartitionedRegionStats getPartitionedRegionStats() {
return regionAdvisor.getPartitionedRegionStats();
}
/**
* Concurrency: protected by synchronizing on *this*
*/
@Override
protected void profileCreated(Profile profile) {
regionAdvisor.incrementBucketCount(profile);
super.profileCreated(profile);
if (updateRedundancy() > 0) {
// wake up any threads in waitForRedundancy or waitForPrimary
notifyAll();
}
regionAdvisor.updateBucketStatus(getBucket().getId(), profile.peerMemberId, false);
if (logger.isDebugEnabled()) {
logger.debug("Profile added {} Profile : {}", getBucket().getFullPath(), profile);
}
synchronized (this) {
updateServerBucketProfile();
}
}
/**
* Concurrency: protected by synchronizing on *this*
*/
@Override
protected void profileUpdated(Profile profile) {
super.profileUpdated(profile);
if (updateRedundancy() > 0) {
// wake up any threads in waitForRedundancy or waitForPrimary
notifyAll();
}
regionAdvisor.updateBucketStatus(getBucket().getId(), profile.peerMemberId, false);
if (logger.isDebugEnabled()) {
logger.debug("Profile updated {} Profile : {}", getBucket().getFullPath(), profile);
}
synchronized (this) {
updateServerBucketProfile();
}
}
/**
* Concurrency: protected by synchronizing on *this*
*/
@Override
protected void profileRemoved(Profile profile) {
if (profile != null) {
regionAdvisor.updateBucketStatus(getBucket().getId(),
profile.getDistributedMember(), true);
regionAdvisor.decrementsBucketCount(profile);
}
updateRedundancy();
if (logger.isDebugEnabled()) {
logger.debug("Profile removed {} the member lost {} Profile : {}", getBucket().getFullPath(),
profile != null ? profile.getDistributedMember() : null, profile);
}
synchronized (this) {
updateServerBucketProfile();
}
}
@Override
public boolean shouldSyncForCrashedMember(InternalDistributedMember id) {
BucketProfile profile = (BucketProfile) getProfile(id);
return (profile != null) && (profile.isPrimary);
}
@Override
public DistributedRegion getRegionForDeltaGII() {
DistributedRegion result = super.getRegionForDeltaGII();
if (result == null && getAdvisee() instanceof ProxyBucketRegion) {
result = ((ProxyBucketRegion) getAdvisee()).getHostedBucketRegion();
}
return result;
}
/**
* Called by the RegionAdvisor.profileRemoved, this method tests to see if the missing member is
* the primary elector for this bucket.
*
* We can't call this method from BucketAdvisor.profileRemoved, because the primaryElector may not
* actually host the bucket.
*
*/
public void checkForLostPrimaryElector(Profile profile) {
// If the member that went away was in the middle of creating
// the bucket, finish the bucket creation.
ProfileId elector = primaryElector;
if (elector != null && elector.equals(profile.getDistributedMember())) {
if (logger.isDebugEnabled()) {
logger.debug(
"Bucket {} lost the member responsible for electing the primary. Finishing bucket creation",
getBucket().getFullPath());
}
primaryElector = getBucket().getDistributionManager().getId();
getBucket().getDistributionManager().getExecutors().getWaitingThreadPool().execute(
() -> getBucket().getPartitionedRegion().getRedundancyProvider()
.finishIncompleteBucketCreation(getBucket().getId()));
}
}
/**
* Only allows profiles that actually hosting this bucket. If the profile is primary, then
* primaryMember will be set to that member but only if we are not already the primary.
*
* @param profile the profile to add (must be a BucketProfile)
* @param forceProfile true will force profile to be added even if member is not in distributed
* view
*
* @see #adviseProfileUpdate()
*/
@Override
public boolean putProfile(Profile profile, boolean forceProfile) {
assert profile instanceof BucketProfile;
BucketProfile bp = (BucketProfile) profile;
// Only hosting buckets will be initializing, the isInitializing boolean is to
// allow for early entry into the advisor for GII purposes
if (!bp.isHosting && !bp.isInitializing) {
if (logger.isTraceEnabled(LogMarker.DISTRIBUTION_ADVISOR_VERBOSE)) {
logger.trace(LogMarker.DISTRIBUTION_ADVISOR_VERBOSE, "BucketAdvisor#putProfile early out");
}
return false; // Do not allow introduction of proxy profiles, they don't provide anything
// useful
// isHosting = false, isInitializing = false
}
if (logger.isTraceEnabled(LogMarker.DISTRIBUTION_ADVISOR_VERBOSE)) {
logger.trace(LogMarker.DISTRIBUTION_ADVISOR_VERBOSE,
"BucketAdvisor#putProfile profile=<{}> force={}; profile = {}", profile, forceProfile,
bp);
}
// isHosting = false, isInitializing = true
// isHosting = true, isInitializing = false
// isHosting = true, isInitializing = true... (false state)
final boolean applied;
synchronized (this) {
// force new membership version in the advisor so that the
// state flush mechanism can capture any updates to the bucket
// MIN_VALUE is intended as a somewhat unique value for potential debug purposes
profile.initialMembershipVersion = Long.MIN_VALUE;
applied = super.putProfile(profile, forceProfile);
// skip following block if isPrimary to avoid race where we process late
// arriving OTHER_PRIMARY profile after we've already become primary
if (applied && !isPrimary()) {
if (bp.isPrimary) {
setPrimaryMember(bp.getDistributedMember());
} else {
notPrimary(bp.getDistributedMember());
}
} // if: !isPrimary
} // synchronized
return applied;
}
private static <E> Set<E> newSetFromMap(Map<E, Boolean> map) {
if (map.isEmpty()) {
return new SetFromMap<>(map);
}
throw new IllegalArgumentException();
}
@SuppressWarnings("NullableProblems")
private static class SetFromMap<E> extends AbstractSet<E> implements Serializable {
private static final long serialVersionUID = 2454657854757543876L;
// must named as it, to pass serialization compatibility test.
private Map<E, Boolean> m;
private transient Set<E> backingSet;
SetFromMap(final Map<E, Boolean> map) {
super();
m = map;
backingSet = map.keySet();
}
@Override
public boolean equals(Object object) {
return backingSet.equals(object);
}
@Override
public int hashCode() {
return backingSet.hashCode();
}
@Override
public boolean add(E object) {
return m.put(object, Boolean.TRUE) == null;
}
@Override
public void clear() {
m.clear();
}
@Override
public String toString() {
return backingSet.toString();
}
@Override
public boolean contains(Object object) {
return backingSet.contains(object);
}
@Override
public boolean containsAll(Collection<?> collection) {
return backingSet.containsAll(collection);
}
@Override
public boolean isEmpty() {
return m.isEmpty();
}
@Override
public boolean remove(Object object) {
return m.remove(object) != null;
}
@Override
public boolean retainAll(Collection<?> collection) {
return backingSet.retainAll(collection);
}
@Override
public Object[] toArray() {
return backingSet.toArray();
}
@Override
public <T> T[] toArray(T[] contents) {
return backingSet.toArray(contents);
}
@Override
public Iterator<E> iterator() {
return backingSet.iterator();
}
@Override
public int size() {
return m.size();
}
private void readObject(ObjectInputStream stream) throws IOException, ClassNotFoundException {
stream.defaultReadObject();
backingSet = m == null ? Collections.emptySet() : m.keySet();
}
}
/**
* repopulates the RegionAdvisor's location information for this bucket
*/
private void updateServerBucketProfile() {
int bucketId = getBucket().getId();
Set<ServerBucketProfile> serverProfiles =
newSetFromMap(new HashMap<>());
for (Profile p : profiles) {
if (p instanceof ServerBucketProfile) {
serverProfiles.add((ServerBucketProfile) p);
}
}
regionAdvisor.setClientBucketProfiles(bucketId, serverProfiles);
}
/**
* Only for local profile.
*
*/
public synchronized void updateServerBucketProfile(BucketProfile p) {
localProfile = p;
}
public BucketProfile getLocalProfile() {
return localProfile;
}
@Override
public boolean removeId(ProfileId memberId, boolean crashed, boolean destroyed,
boolean fromMembershipListener) {
boolean hadBucketRegion = super.removeId(memberId, crashed, destroyed, fromMembershipListener);
if (hadBucketRegion) {
// do NOT call notPrimary under synchronization
try {
notPrimary((InternalDistributedMember) memberId);
} catch (CancelException e) {
// must be closing the cache - no need to try to become primary
}
}
return hadBucketRegion;
}
/**
* Removes the profile for the specified member. If that profile is marked as primary, this will
* call {@link #notPrimary(InternalDistributedMember)}.
*
* @param memberId the member to remove the profile for
* @param serialNum specific serial number to remove
* @return true if a matching profile for the member was found
*/
@Override
public boolean removeIdWithSerial(InternalDistributedMember memberId, int serialNum,
boolean regionDestroyed) {
boolean hadBucketRegion = super.removeIdWithSerial(memberId, serialNum, regionDestroyed);
if (hadBucketRegion) {
// do NOT call notPrimary under synchronization
notPrimary(memberId);
}
forceNewMembershipVersion();
return hadBucketRegion;
}
@Override
public Set<InternalDistributedMember> adviseProfileExchange() {
// delegate up to RegionAdvisor to include members that might have
// ProxyBucketRegion without a real BucketRegion
Assert.assertTrue(regionAdvisor.isInitialized());
return regionAdvisor.adviseBucketProfileExchange();
}
@Override
public Set<InternalDistributedMember> adviseProfileUpdate() {
// delegate up to RegionAdvisor to include members that might have
// ProxyBucketRegion without a real BucketRegion
return regionAdvisor.adviseGeneric();
}
/**
* Sets hosting to false and returns without closing. Calling closeAdvisor will actually close
* this advisor.
*/
@Override
public void close() {
// ok, so BucketRegion extends DistributedRegion which calls close on the
// advisor, BUT the ProxyBucketRegion truly owns the advisor and only it
// should be able to close the advisor...
//
// if a BucketRegion closes, it will call this method and we want to change
// our state to NOT_HOSTING
//
// see this.closeAdvisor()
setHosting(false);
}
/**
* Blocks until there is a known primary and return that member, but only if there are real bucket
* regions that exist. If there are no real bucket regions within the distribution config's
* member-timeout setting * 3 (time required to eject a member) + 15000, then this returns null.
*
* kbanks: reworked this method to avoid JIT issue #40639
*
* @return the member who is primary for this bucket
*/
public InternalDistributedMember getPrimary() {
InternalDistributedMember primary = getExistingPrimary();
if (primary == null) {
primary = waitForNewPrimary();
}
return primary;
}
/**
* This method was split out from getPrimary() due to bug #40639 and is only intended to be called
* from within that method.
*
* @see #getPrimary()
* @return the existing primary (if it is still in the view) otherwise null
*/
private InternalDistributedMember getExistingPrimary() {
return basicGetPrimaryMember();
}
/**
* This method was split out from getPrimary() due to bug #40639 and is only intended to be called
* from within that method.
*
* @see #getPrimary()
* @return the new primary
*/
private InternalDistributedMember waitForNewPrimary() {
DistributionManager dm = regionAdvisor.getDistributionManager();
DistributionConfig config = dm.getConfig();
// failure detection period
long timeout = config.getMemberTimeout() * 3L;
// plus time for a new member to become primary
timeout += Long.getLong(DistributionConfig.GEMFIRE_PREFIX + "BucketAdvisor.getPrimaryTimeout",
15000L);
return waitForPrimaryMember(timeout);
}
/**
* Marks member as not primary. Initiates volunteerForPrimary if this member is hosting a real
* bucket. This method does nothing if the member parameter is the current member.
*
* @param member the member who is not primary
*/
public void notPrimary(InternalDistributedMember member) {
// Fix for 43569. Only the deposePrimary call should
// make the local member drop the primary lock.
if (!member.equals(getDistributionManager().getId())) {
removePrimary(member);
}
}
/**
* Marks member as not primary. Initiates volunteerForPrimary if this member is hosting a real
* bucket.
*
* @param member the member who is not primary
*/
private void removePrimary(InternalDistributedMember member) {
boolean needToVolunteerForPrimary = false;
if (!isClosed()) { // hole: requestPrimaryState not hosting
initializationGate();
}
boolean lostPrimary = false;
try {
synchronized (this) {
boolean wasPrimary = isPrimary() && getDistributionManager().getId().equals(member);
final InternalDistributedMember currentPrimary = primaryMember.get();
if (currentPrimary != null && currentPrimary.equals(member)) {
if (logger.isDebugEnabled()) {
logger.debug("[BucketAdvisor.notPrimary] {} for {}", member, this);
}
primaryMember.set(null);
} else {
return;
}
if (isClosed()) {
// possibly closed if caller comes from outside this advisor
return; // return quietly
}
// member is primary... need to change state to NO_PRIMARY_xxx
if (isHosting()) {
requestPrimaryState(NO_PRIMARY_HOSTING);
if (pRegion.isFixedPartitionedRegion()) {
InternalDistributedMember primaryMember =
regionAdvisor.adviseFixedPrimaryPartitionDataStore(getBucket().getId());
needToVolunteerForPrimary = primaryMember == null || primaryMember.equals(member);
} else {
needToVolunteerForPrimary = true;
}
} else {
requestPrimaryState(NO_PRIMARY_NOT_HOSTING);
}
if (wasPrimary) {
lostPrimary = true;
}
// try to fix yet another cause of bug 36881
findAndSetPrimaryMember();
}
} finally {
if (lostPrimary) {
invokeAfterSecondaryInPartitionListeners();
Bucket br = regionAdvisor.getBucket(getBucket().getId());
if (br instanceof BucketRegion) {
((BucketRegion) br).beforeReleasingPrimaryLockDuringDemotion();
}
releasePrimaryLock();
// this was a deposePrimary call so we need to depose children as well
deposePrimaryForColocatedChildren();
if (pRegion.isFixedPartitionedRegion()) {
deposeOtherPrimaryBucketForFixedPartition();
}
}
}
if (needToVolunteerForPrimary) {
volunteerForPrimary();
}
}
/**
* Returns the ProxyBucketRegion which owns this advisor.
*
* @return the ProxyBucketRegion which owns this advisor
*/
public ProxyBucketRegion getProxyBucketRegion() {
return (ProxyBucketRegion) getAdvisee();
}
/**
* Actually close this advisor for real. Called by ProxyBucketRegion only. Calling this method
* actually closes this advisor whereas {@link #close()} only sets hosting to false.
*/
void closeAdvisor() {
boolean wasPrimary;
synchronized (this) {
if (isClosed()) {
return;
}
wasPrimary = isPrimary();
super.close();
requestPrimaryState(CLOSED);
redundancyTracker.closeBucket();
localProfile = null;
}
if (wasPrimary) {
releasePrimaryLock();
}
}
/**
* Returns true if this advisor has been closed.
*
* @return true if this advisor has been closed
*/
protected boolean isClosed() {
return primaryState == CLOSED;
}
/**
* Returns true if this member is currently marked as primary.
*
* @return true if this member is currently marked as primary
*/
public boolean isPrimary() {
return primaryState == IS_PRIMARY_HOSTING;
}
/**
* Returns true if this member is currently volunteering for primary.
*
* @return true if this member is currently volunteering for primary
*/
private boolean isVolunteering() {
return primaryState == VOLUNTEERING_HOSTING;
}
/**
* Returns true if this member is currently attempting to become primary.
*
* @return true if this member is currently attempting to become primary
*/
private boolean isBecomingPrimary() {
synchronized (this) {
return primaryState == BECOMING_HOSTING && volunteeringDelegate != null
&& volunteeringDelegate.isAggressive();
}
}
/**
* Returns true if this member is currently hosting real bucket.
*
* @return true if this member is currently hosting real bucket
*/
public boolean isHosting() {
final byte primaryState = this.primaryState;
return primaryState == NO_PRIMARY_HOSTING || primaryState == OTHER_PRIMARY_HOSTING
|| primaryState == VOLUNTEERING_HOSTING || primaryState == BECOMING_HOSTING
|| primaryState == IS_PRIMARY_HOSTING;
}
/**
* Attempt to acquire lock for primary until a primary exists. Caller hands off responsibility to
* an executor (waiting pool) and returns early.
*/
public void volunteerForPrimary() {
InternalDistributedMember elector = primaryElector;
if (elector != null && regionAdvisor.hasPartitionedRegion(elector)) {
// another server will determine the primary node
return;
}
primaryElector = null;
initializationGate();
synchronized (this) {
if (isVolunteering() || isClosed() || !isHosting()) {
// only one thread should be attempting to volunteer at one time
return;
}
if (volunteeringDelegate == null) {
setVolunteeringDelegate(new VolunteeringDelegate());
}
volunteeringDelegate.volunteerForPrimary();
}
}
protected void setVolunteeringDelegate(VolunteeringDelegate delegate) {
volunteeringDelegate = delegate;
}
/**
* Makes this <code>BucketAdvisor</code> become the primary if it is already a secondary.
*
* @param isRebalance true if directed to become primary by rebalancing
* @return true if this advisor succeeds in becoming the primary
*/
public boolean becomePrimary(boolean isRebalance) {
initializationGate();
long startTime = getPartitionedRegionStats().startPrimaryTransfer(isRebalance);
try {
long waitTime = 2000; // time each iteration will wait
while (!isPrimary()) {
getAdvisee().getCancelCriterion().checkCancelInProgress(null);
boolean attemptToBecomePrimary = false;
boolean attemptToDeposePrimary = false;
if (Thread.currentThread().isInterrupted()) {
if (logger.isDebugEnabled()) {
logger.debug("Breaking from becomePrimary loop due to thread interrupt flag being set");
}
break;
}
if (isClosed() || !isHosting()) {
if (logger.isDebugEnabled()) {
logger.debug("Breaking from becomePrimary loop because {} is closed or not hosting",
this);
}
break;
}
VolunteeringDelegate vDelegate = null;
synchronized (this) {
if (isVolunteering()) {
// standard volunteering attempt already in progress...
if (logger.isDebugEnabled()) {
logger.debug("Waiting for volunteering thread {}. Time left: {} ms", this, waitTime);
}
wait(waitTime); // spurious wakeup ok
continue;
} else if (isBecomingPrimary()) {
// reattempt to depose otherPrimary...
attemptToDeposePrimary = true;
} else {
// invoke becomePrimary AFTER sync is released in this thread...
vDelegate = volunteeringDelegate;
if (vDelegate == null) {
vDelegate = new VolunteeringDelegate();
volunteeringDelegate = vDelegate;
}
} // else
} // synchronized
if (vDelegate != null) {
// Use the snapshot 'vDelegate' instead of 'this.volunteeringDelegate' since we are not
// synced here.
attemptToBecomePrimary = vDelegate.reserveForBecomePrimary(); // no sync!
}
// release synchronization and then call becomePrimary
if (attemptToBecomePrimary) {
synchronized (this) {
if (volunteeringDelegate == null) {
volunteeringDelegate = new VolunteeringDelegate();
}
volunteeringDelegate.volunteerForPrimary();
attemptToDeposePrimary = true;
} // synchronized
Thread.sleep(10);
} // attemptToBecomePrimary
// RACE: slight race condition with thread that's actually requesting the lock
if (attemptToDeposePrimary) {
InternalDistributedMember otherPrimary = getPrimary();
if (otherPrimary != null && !getDistributionManager().getId().equals(otherPrimary)) {
if (logger.isDebugEnabled()) {
logger.debug("Attempting to depose primary on {} for {}", otherPrimary, this);
}
DeposePrimaryBucketResponse response =
DeposePrimaryBucketMessage.send(otherPrimary, pRegion, getBucket().getId());
if (response != null) {
response.waitForRepliesUninterruptibly();
if (logger.isDebugEnabled()) {
logger.debug("Deposed primary on {}", otherPrimary);
}
}
}
Thread.sleep(10);
} // attemptToDeposePrimary
} // while
} catch (InterruptedException e) {
// abort and return null
Thread.currentThread().interrupt();
} finally {
getPartitionedRegionStats().endPrimaryTransfer(startTime, isPrimary(), isRebalance);
}
return isPrimary();
}
/**
* Check the primary member shortcut. Does not query the advisor. Should only be used when the
* advisor should not be consulted directly.
*
* @return the member or null if no primary exists
*/
public InternalDistributedMember basicGetPrimaryMember() {
return primaryMember.get();
}
/**
* Invoked when the primary lock has been acquired by this VM.
*
* @return true if successfully changed state to IS_PRIMARY
*/
private boolean acquiredPrimaryLock() {
if (logger.isDebugEnabled()) {
logger.debug("Acquired primary lock for BucketID {} PR : {}", getBucket().getId(),
regionAdvisor.getPartitionedRegion().getFullPath());
}
boolean changedStateToIsPrimary = false;
// Hold the primary move lock until we send a
// profile update. This will prevent writes
// from occurring until all members know that
// this member is now the primary.
boolean shouldInvokeListeners = false;
primaryMoveWriteLock.lock();
try {
synchronized (this) {
if (isHosting() && (isVolunteering() || isBecomingPrimary())) {
Bucket br = regionAdvisor.getBucket(getBucket().getId());
if (br instanceof BucketRegion) {
((BucketRegion) br).beforeAcquiringPrimaryState();
}
if (requestPrimaryState(IS_PRIMARY_HOSTING)) {
if (logger.isDebugEnabled()) {
logger.debug("Acquired primary lock for setting primary now BucketID {} PR : {}",
getBucket().getId(), regionAdvisor.getPartitionedRegion().getFullPath());
}
setPrimaryMember(getDistributionManager().getId());
changedStateToIsPrimary = true;
if (hasPrimary() && isPrimary()) {
shouldInvokeListeners = true;
}
}
}
}
if (shouldInvokeListeners) {
invokePartitionListeners();
}
return changedStateToIsPrimary;
} finally {
try {
if (changedStateToIsPrimary) {
// send profile update AFTER releasing sync
sendProfileUpdate();
Bucket br = regionAdvisor.getBucket(getBucket().getId());
if (br instanceof BucketRegion) {
((BucketRegion) br).processPendingSecondaryExpires();
}
if (br instanceof BucketRegionQueue) { // Shouldn't it be AbstractBucketRegionQueue
BucketRegionQueue brq = (BucketRegionQueue) br;
brq.incQueueSize(brq.size());
brq.decSecondaryQueueSize(brq.size());
}
if (br instanceof BucketRegion) {
((BucketRegion) br).afterAcquiringPrimaryState();
}
} else {
// release primary lock AFTER releasing sync
releasePrimaryLock();
}
} finally {
primaryMoveWriteLock.unlock();
}
}
}
private void invokePartitionListeners() {
PartitionListener[] listeners = pRegion.getPartitionListeners();
if (listeners == null || listeners.length == 0) {
return;
}
for (PartitionListener listener : listeners) {
if (listener != null) {
listener.afterPrimary(getBucket().getId());
}
}
}
private void invokeAfterSecondaryInPartitionListeners() {
PartitionListener[] listeners = pRegion.getPartitionListeners();
if (listeners == null || listeners.length == 0) {
return;
}
for (PartitionListener listener : listeners) {
if (listener != null) {
listener.afterSecondary(getBucket().getId());
}
}
}
/**
* Lazily gets the lock for acquiring primary lock. Caller must handle null. If DLS, Cache, or
* DistributedSystem are shutting down then null will be returned. If DLS does not yet exist and
* createDLS is false then null will be returned.
*
* @param createDLS true will create DLS if it does not exist
* @return distributed lock indicating primary member or null
*/
private DistributedMemberLock getPrimaryLock(boolean createDLS) {
synchronized (this) {
if (primaryLock == null) {
DistributedLockService dls = DistributedLockService
.getServiceNamed(PartitionedRegionHelper.PARTITION_LOCK_SERVICE_NAME);
if (dls == null) {
if (!createDLS || getProxyBucketRegion().getCache().isClosed()) {
return null; // cache closure has destroyed the DLS
}
try { // TODO: call GemFireCache#getPartitionedRegionLockService
dls = DLockService.create(PartitionedRegionHelper.PARTITION_LOCK_SERVICE_NAME,
getAdvisee().getSystem(), true /* distributed */, true /* destroyOnDisconnect */,
true /* automateFreeResources */);
} catch (IllegalArgumentException e) {
// indicates that the DLS is already created
dls = DistributedLockService
.getServiceNamed(PartitionedRegionHelper.PARTITION_LOCK_SERVICE_NAME);
if (dls == null) {
// another thread destroyed DLS after this thread called create
return null; // ok, caller will loop if necessary
}
}
// TODO: we need a good NotConnectedException to replace
// IllegalStateException and ShutdownException
// perhaps: DistributedSystemUnavailableException
catch (IllegalStateException | DistributedSystemDisconnectedException e) {
// create still throws IllegalStateException if isDisconnecting is true
return null;
} // this would certainly prevent us from creating a DLS... messy
}
primaryLock = new DistributedMemberLock(dls, getAdvisee().getName(),
DistributedMemberLock.NON_EXPIRING_LEASE,
DistributedMemberLock.LockReentryPolicy.PREVENT_SILENTLY);
}
return primaryLock;
}
}
private void acquirePrimaryRecursivelyForColocated() {
final List<PartitionedRegion> colocatedWithList =
ColocationHelper.getColocatedChildRegions(regionAdvisor.getPartitionedRegion());
if (colocatedWithList != null) {
for (PartitionedRegion childPR : colocatedWithList) {
Bucket b = childPR.getRegionAdvisor().getBucket(getBucket().getId());
BucketAdvisor childBA = b.getBucketAdvisor();
Assert.assertHoldsLock(childBA, false);
boolean acquireForChild = false;
if (logger.isDebugEnabled()) {
logger.debug(
"BucketAdvisor.acquirePrimaryRecursivelyForColocated: about to take lock for bucket: {} of PR: {} with isHosting={}",
getBucket().getId(), childPR.getFullPath(), childBA.isHosting());
}
childBA.primaryMoveWriteLock.lock();
try {
if (childBA.isHosting()) {
if (isPrimary()) {
if (!childBA.isPrimary()) {
childBA.setVolunteering();
boolean acquired = childBA.acquiredPrimaryLock();
acquireForChild = true;
if (acquired && pRegion.isFixedPartitionedRegion()) {
childBA.acquirePrimaryForRestOfTheBucket();
}
} else {
acquireForChild = true;
}
}
} // if isHosting
if (acquireForChild) {
childBA.acquirePrimaryRecursivelyForColocated();
}
} finally {
childBA.primaryMoveWriteLock.unlock();
}
}
}
}
private void acquirePrimaryForRestOfTheBucket() {
List<FixedPartitionAttributesImpl> fpas = pRegion.getFixedPartitionAttributesImpl();
if (fpas != null) {
int bucketId = getBucket().getId();
for (FixedPartitionAttributesImpl fpa : fpas) {
if (fpa.getStartingBucketID() == bucketId) {
for (int i = bucketId + 1; i <= fpa.getLastBucketID();) {
Bucket b = regionAdvisor.getBucket(i++);
if (b != null) {
BucketAdvisor ba = b.getBucketAdvisor();
ba.primaryMoveWriteLock.lock();
try {
if (ba.isHosting()) {
if (!ba.isPrimary()) {
ba.setVolunteering();
ba.acquiredPrimaryLock();
}
}
} finally {
ba.primaryMoveWriteLock.unlock();
}
}
}
}
}
}
}
/**
* Sets volunteering to true. Returns true if the state of volunteering was changed. Returns false
* if voluntering was already equal to true. Caller should do nothing if false is returned.
*/
private boolean setVolunteering() {
synchronized (this) {
return requestPrimaryState(VOLUNTEERING_HOSTING);
}
}
/**
* Sets becoming primary to true. Returns true if the state of becoming was changed. Returns false
* if becoming was already equal to true. Caller should do nothing if false is returned.
*/
private boolean setBecoming() {
synchronized (this) {
return requestPrimaryState(BECOMING_HOSTING);
}
}
/**
* Wait briefly for a primary member to be identified.
*
* @param timeout time in milliseconds to wait for a primary
* @return the primary bucket host
*/
private InternalDistributedMember waitForPrimaryMember(long timeout) {
synchronized (this) {
// let's park this thread and wait for a primary!
StopWatch timer = new StopWatch(true);
long warnTime = getDistributionManager().getConfig().getAckWaitThreshold() * 1000L;
boolean loggedWarning = false;
try {
for (;;) {
// bail out if the system starts closing
getAdvisee().getCancelCriterion().checkCancelInProgress(null);
final InternalCache cache = getBucket().getCache();
if (cache != null && cache.isCacheAtShutdownAll()) {
throw cache.getCacheClosedException("Cache is shutting down");
}
if (getBucketRedundancy() == -1) {
// there are no real buckets in other vms... no reason to wait
return null;
}
getProxyBucketRegion().getPartitionedRegion().checkReadiness();
if (isClosed()) {
break;
}
long elapsed = timer.elapsedTimeMillis();
long timeLeft = timeout - elapsed;
if (timeLeft <= 0) {
break;
}
if (getBucketRedundancy() == -1 || isClosed()) {
break; // early out... all bucket regions are gone or we closed
}
InternalDistributedMember primary = basicGetPrimaryMember();
if (primary != null) {
return primary;
}
if (logger.isDebugEnabled()) {
logger.debug("Waiting for bucket {}. Time left :{} ms", this, timeLeft);
}
// Log a warning if we have waited for the ack wait threshold time.
if (!loggedWarning) {
long timeUntilWarning = warnTime - elapsed;
if (timeUntilWarning <= 0) {
logger
.warn(
"{} secs have elapsed waiting for a primary for bucket {}. Current bucket owners {}",
new Object[] {warnTime / 1000L, this, adviseInitialized()});
// log a warning;
loggedWarning = true;
} else {
timeLeft = timeLeft > timeUntilWarning ? timeUntilWarning : timeLeft;
}
}
wait(timeLeft); // spurious wakeup ok
}
} catch (InterruptedException e) {
// abort and return null
Thread.currentThread().interrupt();
} finally {
if (loggedWarning) {
logger.info(
"Wait for primary completed");
}
}
return null;
}
}
/**
* How long to wait, in millisecs, for redundant buckets to exist
*/
private static final long BUCKET_REDUNDANCY_WAIT = 15000L; // 15 seconds
/**
* Wait the desired redundancy to be met.
*
* @param minRedundancy the amount of desired redundancy.
* @return true if desired redundancy is detected
*/
public boolean waitForRedundancy(int minRedundancy) {
synchronized (this) {
// let's park this thread and wait for redundancy!
StopWatch timer = new StopWatch(true);
try {
for (;;) {
if (getBucketRedundancy() >= minRedundancy) {
return true;
}
getProxyBucketRegion().getPartitionedRegion().checkReadiness();
if (isClosed()) {
return false;
}
long timeLeft = BUCKET_REDUNDANCY_WAIT - timer.elapsedTimeMillis();
if (timeLeft <= 0) {
return false;
}
if (logger.isDebugEnabled()) {
logger.debug("Waiting for bucket {}", this);
}
wait(timeLeft); // spurious wakeup ok
}
} catch (InterruptedException e) {
// abort and return null
Thread.currentThread().interrupt();
}
return false;
}
}
synchronized void clearPrimaryElector() {
primaryElector = null;
}
synchronized void setPrimaryElector(InternalDistributedMember newPrimaryElector) {
// Only set the new primary elector if we have not yet seen
// a primary for this bucket.
if (primaryElector != null) {
if (newPrimaryElector != null && !regionAdvisor.hasPartitionedRegion(newPrimaryElector)) {
// no longer a participant - don't use it
primaryElector = null;
} else {
primaryElector = newPrimaryElector;
}
}
}
public synchronized void initializePrimaryElector(InternalDistributedMember newPrimaryElector) {
// For child buckets, we want the parent bucket to take care'
// of finishing an incomplete bucket creation, so only set the elector for
// the leader region.
if (parentAdvisor == null) {
if (newPrimaryElector != null && !regionAdvisor.hasPartitionedRegion(newPrimaryElector)) {
// no longer a participant - don't use it
primaryElector = null;
} else {
primaryElector = newPrimaryElector;
}
}
}
/**
* Invoked when real bucket is created for hosting in this VM.
*
* @param value true to begin hosting; false to end hosting
*/
protected void setHosting(boolean value) {
// boolean needToNotPrimarySelf = false;
boolean needToVolunteerForPrimary = false;
boolean wasPrimary;
synchronized (this) {
wasPrimary = isPrimary();
if (isClosed()) {
return;
}
if (value) { // setting to HOSTING...
if (hasPrimary()) {
requestPrimaryState(OTHER_PRIMARY_HOSTING);
} else {
requestPrimaryState(NO_PRIMARY_HOSTING);
needToVolunteerForPrimary = true;
}
}
else { // setting to NOT_HOSTING...
if (hasPrimary()) { // has primary...
if (isPrimary()) {
requestPrimaryState(NO_PRIMARY_NOT_HOSTING);
primaryMember.set(null);
findAndSetPrimaryMember();
} else {
requestPrimaryState(OTHER_PRIMARY_NOT_HOSTING);
}
} else { // no primary...
// acquiredPrimaryLock will check isHosting and release if not hosting
requestPrimaryState(NO_PRIMARY_NOT_HOSTING);
}
}
volunteeringDelegate = null;
// Note - checkRedundancy has the side effect that it updates the stats.
// We need to invoke checkRedundancy here, regardless of whether we
// need this notify.
if (updateRedundancy() > 0 && isHosting()) {
// wake up any threads in waitForRedundancy or waitForPrimary
notifyAll();
}
}
if (wasPrimary) {
releasePrimaryLock();
}
if (logger.isTraceEnabled()) {
logger.trace("setHosting: {} needToVolunteerForPrimary={} primaryElector: {}", this,
needToVolunteerForPrimary, primaryElector);
}
/*
* if (needToNotPrimarySelf) { notPrimary(getAdvisee().getDistributionManager().getId()); }
*/
if (needToVolunteerForPrimary) {
volunteerForPrimary();
}
sendProfileUpdate();
}
/**
* Sends updated profile for this member to every member with the <code>PartitionedRegion</code>.
* <p>
* Never call this method while synchronized on this BucketAdvisor. This will result in
* distributed deadlocks.
*/
private void sendProfileUpdate() {
if (getDistributionManager().getSystem().isLoner()) {
// no one to send the profile update... return to prevent bug 39760
return;
}
// make sure caller is not synchronized or we'll deadlock
Assert.assertTrue(!Thread.holdsLock(this),
"Attempting to sendProfileUpdate while synchronized may result in deadlock");
// NOTE: if this assert fails, you COULD use the WaitingThreadPool in DM
final int partitionedRegionId = pRegion.getPRId();
final int bucketId = ((ProxyBucketRegion) getAdvisee()).getBucketId();
BucketProfile bp = (BucketProfile) createProfile();
updateServerBucketProfile(bp);
InternalDistributedMember primary = basicGetPrimaryMember();
HashSet<InternalDistributedMember> hostsAndProxyMembers = new HashSet<>();
if (primary != null && !primary.equals(getDistributionManager().getId())) {
hostsAndProxyMembers.add(primary); // Add the primary
}
hostsAndProxyMembers.addAll(adviseGeneric()); // Add all members hosting the bucket
hostsAndProxyMembers.addAll(adviseProfileUpdate()); // Add all proxy instances that could use
// the bucket
ReplyProcessor21 reply = BucketProfileUpdateMessage.send(hostsAndProxyMembers,
getDistributionManager(), partitionedRegionId, bucketId, bp, true);
if (reply != null) {
reply.waitForRepliesUninterruptibly();
}
}
/**
* Returns true if the a primary is known.
*/
private boolean hasPrimary() {
final byte primaryState = this.primaryState;
return primaryState == OTHER_PRIMARY_NOT_HOSTING || primaryState == OTHER_PRIMARY_HOSTING
|| primaryState == IS_PRIMARY_HOSTING;
}
@Override
protected Profile instantiateProfile(InternalDistributedMember memberId, int version) {
if (!pRegion.isShadowPR()) {
Set<BucketServerLocation66> serverLocations = getBucketServerLocations(version);
if (!serverLocations.isEmpty()) {
return new ServerBucketProfile(memberId, version, getBucket(), (HashSet) serverLocations);
}
}
return new BucketProfile(memberId, version, getBucket());
}
protected Set<BucketServerLocation66> getBucketServerLocations(int version) {
InternalCache cache = getProxyBucketRegion().getCache();
List<CacheServer> servers = cache.getCacheServers();
if (servers.isEmpty()) {
return Collections.emptySet();
}
HashSet<BucketServerLocation66> serverLocations = new HashSet<>();
for (CacheServer cacheServer : servers) {
CacheServerImpl server = (CacheServerImpl) cacheServer;
try {
String serverExternalAddress;
if (server.isRunning() && ((serverExternalAddress = server.getExternalAddress()) != null)) {
BucketServerLocation66 location = new BucketServerLocation66(getBucket().getId(),
server.getPort(), serverExternalAddress, getBucket().isPrimary(),
Integer.valueOf(version).byteValue(), server.getCombinedGroups());
serverLocations.add(location);
}
} catch (IllegalStateException e) {
if (!(e.getMessage() != null
&& e.getMessage().equals(CACHE_SERVER_BIND_ADDRESS_NOT_AVAILABLE_EXCEPTION_MESSAGE))) {
throw e;
}
}
}
return serverLocations;
}
/**
* Sets primaryMember and notifies all. Caller must be synced on this.
*
* @param id the member to use as primary for this bucket
*/
private void setPrimaryMember(InternalDistributedMember id) {
if (!getDistributionManager().getId().equals(id)) {
// volunteerForPrimary handles primary state change if its our id
if (isHosting()) {
requestPrimaryState(OTHER_PRIMARY_HOSTING);
} else {
requestPrimaryState(OTHER_PRIMARY_NOT_HOSTING);
}
}
primaryMember.set(id);
everHadPrimary = true;
if (id != null && id.equals(primaryElector)) {
primaryElector = null;
}
notifyAll(); // wake up any threads in waitForPrimaryMember
}
void setHadPrimary() {
everHadPrimary = true;
}
boolean getHadPrimary() {
return everHadPrimary;
}
public InternalDistributedMember getPrimaryElector() {
return primaryElector;
}
/**
* Get the current number of bucket hosts and update the redundancy statistics for the region
*
* @return number of current bucket hosts
*/
private int updateRedundancy() {
int numBucketHosts = getNumInitializedBuckets();
if (!isClosed()) {
redundancyTracker.updateStatistics(numBucketHosts);
}
return numBucketHosts;
}
/**
* Returns all {@link InternalDistributedMember}s currently flagged as primary.
* <p>
* Since profile messages may arrive out of order from different members, more than one member may
* temporarily be flagged as primary.
* <p>
* The user of this BucketAdvisor should simply assume that the first profile is primary until the
* dust settles, leaving only one primary profile.
*
* @return zero or greater array of primary members
*/
private InternalDistributedMember[] findPrimaryMembers() {
Set<InternalDistributedMember> primaryMembers = adviseFilter(profile -> {
assert profile instanceof BucketProfile;
BucketProfile srp = (BucketProfile) profile;
return srp.isPrimary;
});
if (primaryMembers.size() > 1 && logger.isDebugEnabled()) {
logger.debug("[findPrimaryProfiles] found the following primary members for {}: {}",
getAdvisee().getName(), primaryMembers);
}
return primaryMembers.toArray(new InternalDistributedMember[0]);
}
/**
* Searches through profiles to find first profile that is flagged as primary and sets
* {@link #primaryMember} to it. Caller must synchronize on this BucketAdvisor.
*
* @return true if a primary member was found and used
* @see #findAndSetPrimaryMember()
*/
private boolean findAndSetPrimaryMember() {
if (isPrimary()) {
setPrimaryMember(getDistributionManager().getDistributionManagerId());
return true;
}
InternalDistributedMember[] primaryMembers = findPrimaryMembers();
if (primaryMembers.length > 0) {
setPrimaryMember(primaryMembers[0]);
return true;
} else {
return false;
}
}
/**
* Returns the current redundancy of the this bucket, including the locally hosted bucket if it
* exists.
*
* @return current number of hosts of this bucket ; -1 if there are no hosts
*/
public int getBucketRedundancy() {
return redundancyTracker.getCurrentRedundancy();
}
Set<InternalDistributedMember> adviseInitialized() {
return adviseFilter(profile -> {
assert profile instanceof BucketProfile;
BucketProfile bucketProfile = (BucketProfile) profile;
return bucketProfile.isHosting;
});
}
Set<InternalDistributedMember> adviseRecoveredFromDisk() {
return regionAdvisor.adviseInitializedDataStore();
}
/**
* Get the number of members that are hosting the bucket, and have finished initialization.
*
* This method is currently only used to check the bucket redundancy just before creating the
* bucket. If it is used more frequently, it might be better to cache this count.
*/
private int getNumInitializedBuckets() {
Profile[] locProfiles = profiles; // grab current profiles
int count = 0;
for (Profile profile : locProfiles) {
BucketProfile bucketProfile = (BucketProfile) profile;
if (bucketProfile.isHosting) {
count++;
}
}
if (isHosting()) {
count++;
}
return count;
}
private Bucket getBucket() {
return (Bucket) getAdvisee();
}
/**
* Releases the primary lock for this bucket.
*/
private void releasePrimaryLock() {
// We don't have a lock if we have a parent advisor
if (parentAdvisor != null) {
return;
}
assignStartingBucketAdvisorIfFixedPartitioned();
if (startingBucketAdvisor != null) {
return;
}
try {
DistributedMemberLock thePrimaryLock = getPrimaryLock(false);
if (thePrimaryLock != null) {
thePrimaryLock.unlock();
}
} catch (LockNotHeldException e) {
Assert.assertTrue(!isHosting(), "Got LockNotHeldException for Bucket = " + this);
} catch (LockServiceDestroyedException e) {
Assert.assertTrue(isClosed(),
"BucketAdvisor was not closed before destroying PR lock service");
}
}
private String primaryStateToString() {
return primaryStateToString(primaryState);
}
/**
* Returns string representation of the primary state value.
*
* @param value the primary state to return string for
* @return string representation of primaryState
*/
private String primaryStateToString(byte value) {
switch (value) {
case NO_PRIMARY_NOT_HOSTING:
return "NO_PRIMARY_NOT_HOSTING";
case NO_PRIMARY_HOSTING:
return "NO_PRIMARY_HOSTING";
case OTHER_PRIMARY_NOT_HOSTING:
return "OTHER_PRIMARY_NOT_HOSTING";
case OTHER_PRIMARY_HOSTING:
return "OTHER_PRIMARY_HOSTING";
case VOLUNTEERING_HOSTING:
return "VOLUNTEERING_HOSTING";
case BECOMING_HOSTING:
return "BECOMING_HOSTING";
case IS_PRIMARY_HOSTING:
return "IS_PRIMARY_HOSTING";
case CLOSED:
return "CLOSED";
default:
return "<unhandled primaryState " + value + " >";
}
}
/**
* Requests change to the requested primary state. Controls all state changes pertaining to
* primary state. Caller must be synchronized on this.
*
* @param requestedState primaryState to change to
* @return true if the requestedState change was completed
* @throws IllegalStateException if an illegal state change was attempted
*/
private boolean requestPrimaryState(byte requestedState) {
final byte fromState = primaryState;
switch (fromState) {
case NO_PRIMARY_NOT_HOSTING:
switch (requestedState) {
case NO_PRIMARY_NOT_HOSTING:
// race condition ok, return false
return false;
case NO_PRIMARY_HOSTING:
primaryState = requestedState;
break;
case OTHER_PRIMARY_NOT_HOSTING:
primaryState = requestedState;
break;
case OTHER_PRIMARY_HOSTING:
primaryState = requestedState;
break;
case BECOMING_HOSTING:
// race condition during close is ok, return false
return false;
case VOLUNTEERING_HOSTING:
// race condition during close is ok, return false
return false;
case CLOSED:
primaryState = requestedState;
break;
default:
throw new IllegalStateException(String.format("Cannot change from %s to %s",
primaryStateToString(), primaryStateToString(requestedState)));
}
break;
case NO_PRIMARY_HOSTING:
switch (requestedState) {
case NO_PRIMARY_NOT_HOSTING:
primaryState = requestedState;
break;
// case OTHER_PRIMARY_NOT_HOSTING: -- enable for bucket migration
// this.primaryState = requestedState;
// break;
case NO_PRIMARY_HOSTING:
// race condition ok, return false
return false;
case VOLUNTEERING_HOSTING:
primaryState = requestedState; {
PartitionedRegionStats stats = getPartitionedRegionStats();
stats.putStartTime(this, stats.startVolunteering());
}
break;
case BECOMING_HOSTING:
primaryState = requestedState; {
PartitionedRegionStats stats = getPartitionedRegionStats();
stats.putStartTime(this, stats.startVolunteering());
}
break;
case OTHER_PRIMARY_HOSTING:
primaryState = requestedState;
break;
case CLOSED:
primaryState = requestedState;
break;
default:
throw new IllegalStateException(String.format("Cannot change from %s to %s",
primaryStateToString(),
primaryStateToString(requestedState)));
}
break;
case OTHER_PRIMARY_NOT_HOSTING:
switch (requestedState) {
case NO_PRIMARY_NOT_HOSTING:
primaryState = requestedState;
break;
case OTHER_PRIMARY_NOT_HOSTING:
// race condition ok, return false
return false;
case OTHER_PRIMARY_HOSTING:
primaryState = requestedState;
break;
case BECOMING_HOSTING:
// race condition during close is ok, return false
return false;
case VOLUNTEERING_HOSTING:
// race condition during close is ok, return false
return false;
case CLOSED:
primaryState = requestedState;
break;
default:
throw new IllegalStateException(String.format("Cannot change from %s to %s",
primaryStateToString(),
primaryStateToString(requestedState)));
}
break;
case OTHER_PRIMARY_HOSTING:
switch (requestedState) {
// case NO_PRIMARY_NOT_HOSTING: -- enable for bucket migration
// this.primaryState = requestedState;
// break;
case OTHER_PRIMARY_NOT_HOSTING:
// May occur when setHosting(false) is called
primaryState = requestedState;
break;
case OTHER_PRIMARY_HOSTING:
// race condition ok, return false
return false;
case NO_PRIMARY_HOSTING:
primaryState = requestedState;
break;
case CLOSED:
primaryState = requestedState;
break;
case VOLUNTEERING_HOSTING:
// race condition ok, return false to abort volunteering
return false;
case BECOMING_HOSTING:
primaryState = requestedState; {
PartitionedRegionStats stats = getPartitionedRegionStats();
stats.putStartTime(this, stats.startVolunteering());
}
break;
case IS_PRIMARY_HOSTING:
// race condition ok, probably race in HA where other becomes
// primary and immediately leaves while we have try-lock message
// enroute to grantor
primaryState = requestedState;
break;
default:
throw new IllegalStateException(String.format("Cannot change from %s to %s",
primaryStateToString(),
primaryStateToString(requestedState)));
}
break;
case VOLUNTEERING_HOSTING:
switch (requestedState) {
case NO_PRIMARY_NOT_HOSTING:
// May occur when setHosting(false) is called
primaryState = requestedState; {
PartitionedRegionStats stats = getPartitionedRegionStats();
stats.endVolunteeringClosed(stats.removeStartTime(this));
}
break;
case OTHER_PRIMARY_NOT_HOSTING:
// May occur when setHosting(false) is called
// Profile update for other primary may have slipped in
primaryState = requestedState; {
PartitionedRegionStats stats = getPartitionedRegionStats();
stats.endVolunteeringClosed(stats.removeStartTime(this));
}
break;
case NO_PRIMARY_HOSTING:
// race condition occurred, return false and stay in volunteering
return false;
case IS_PRIMARY_HOSTING:
primaryState = requestedState; {
PartitionedRegionStats stats = getPartitionedRegionStats();
stats.incPrimaryBucketCount(1);
stats.endVolunteeringBecamePrimary(stats.removeStartTime(this));
}
break;
case OTHER_PRIMARY_HOSTING:
primaryState = requestedState; {
PartitionedRegionStats stats = getPartitionedRegionStats();
stats.endVolunteeringOtherPrimary(stats.removeStartTime(this));
}
break;
case VOLUNTEERING_HOSTING:
// race condition ok, return false to abort volunteering
return false;
case BECOMING_HOSTING:
// race condition ok, return false to abort volunteering
return false;
case CLOSED:
primaryState = requestedState; {
PartitionedRegionStats stats = getPartitionedRegionStats();
stats.endVolunteeringClosed(stats.removeStartTime(this));
}
break;
default:
throw new IllegalStateException(String.format("Cannot change from %s to %s",
primaryStateToString(),
primaryStateToString(requestedState)));
}
break;
case BECOMING_HOSTING:
switch (requestedState) {
case NO_PRIMARY_NOT_HOSTING:
// May occur when setHosting(false) is called
primaryState = requestedState; {
PartitionedRegionStats stats = getPartitionedRegionStats();
stats.endVolunteeringClosed(stats.removeStartTime(this));
}
break;
case OTHER_PRIMARY_NOT_HOSTING:
// May occur when setHosting(false) is called
// Profile update for other primary may have slipped in
primaryState = requestedState; {
PartitionedRegionStats stats = getPartitionedRegionStats();
stats.endVolunteeringClosed(stats.removeStartTime(this));
}
break;
case NO_PRIMARY_HOSTING:
// race condition occurred, return false and stay in volunteering
return false;
case IS_PRIMARY_HOSTING:
primaryState = requestedState; {
PartitionedRegionStats stats = getPartitionedRegionStats();
stats.incPrimaryBucketCount(1);
stats.endVolunteeringBecamePrimary(stats.removeStartTime(this));
}
break;
case OTHER_PRIMARY_HOSTING:
return false;
case VOLUNTEERING_HOSTING:
// race condition ok, return false to abort volunteering
return false;
case BECOMING_HOSTING:
// race condition ok, return false to abort volunteering
return false;
case CLOSED:
primaryState = requestedState; {
PartitionedRegionStats stats = getPartitionedRegionStats();
stats.endVolunteeringClosed(stats.removeStartTime(this));
}
break;
default:
throw new IllegalStateException(String.format("Cannot change from %s to %s",
primaryStateToString(),
primaryStateToString(requestedState)));
}
break;
case IS_PRIMARY_HOSTING:
switch (requestedState) {
case NO_PRIMARY_HOSTING:
// rebalancing must have moved the primary
changeFromPrimaryTo(requestedState);
break;
// case OTHER_PRIMARY_HOSTING: -- enable for bucket migration
// // rebalancing must have moved the primary
// changeFromPrimaryTo(requestedState);
// break;
case OTHER_PRIMARY_NOT_HOSTING:
// rebalancing must have moved the primary and primary
changeFromPrimaryTo(requestedState);
break;
case NO_PRIMARY_NOT_HOSTING:
// May occur when setHosting(false) is called due to closing
changeFromPrimaryTo(requestedState);
break;
case VOLUNTEERING_HOSTING:
// race condition ok, return false to abort volunteering
return false;
case BECOMING_HOSTING:
// race condition ok, return false to abort volunteering
return false;
case CLOSED:
changeFromPrimaryTo(requestedState);
break;
default:
throw new IllegalStateException("Cannot change from " + primaryStateToString()
+ " to " + primaryStateToString(requestedState));
}
break;
case CLOSED:
switch (requestedState) {
case CLOSED:
Exception e = new Exception(
"Attempted to close BucketAdvisor that is already CLOSED");
logger.warn("Attempted to close BucketAdvisor that is already CLOSED",
e);
break;
case VOLUNTEERING_HOSTING:
// race condition ok, return false to abort volunteering
return false;
case BECOMING_HOSTING:
// race condition ok, return false to abort volunteering
return false;
case IS_PRIMARY_HOSTING:
// Commonly occurs when closing and volunteering thread is still running
return false;
case OTHER_PRIMARY_NOT_HOSTING:
// Commonly occurs when a putProfile occurs during closure
return false;
default:
throw new IllegalStateException(
String.format("Cannot change from %s to %s for bucket %s",
primaryStateToString(),
primaryStateToString(requestedState), getAdvisee().getName()));
}
}
return primaryState == requestedState;
}
private void changeFromPrimaryTo(byte requestedState) {
try {
primaryState = requestedState;
} finally {
getPartitionedRegionStats().incPrimaryBucketCount(-1);
}
}
@Override
public Set adviseDestroyRegion() {
// fix for bug 37604 - tell all owners of the pr that the bucket is being
// destroyed. This is needed when bucket cleanup is performed
return regionAdvisor.adviseAllPRNodes();
}
/**
* returns the set of all the members in the system which require both DistributedCacheOperation
* messages and notification-only partition messages
*
* @return a set of recipients requiring both cache-op and notification messages
* @since GemFire 5.7
*/
public Set<InternalDistributedMember> adviseRequiresTwoMessages() {
return adviseNotInitialized();
}
private Set<InternalDistributedMember> adviseNotInitialized() {
return adviseFilter(profile -> {
assert profile instanceof CacheProfile;
CacheProfile cp = (CacheProfile) profile;
return !cp.regionInitialized;
});
}
@Override
public Set adviseNetWrite() {
return regionAdvisor.adviseNetWrite();
}
@Override
public String toString() {
final StringBuilder sb = new StringBuilder("[BucketAdvisor ").append(getAdvisee().getFullPath())
.append(':').append(getAdvisee().getSerialNumber()).append(": ");
sb.append("state=").append(primaryStateToString());
sb.append("]");
return sb.toString();
}
// A listener for events on this bucket, and also for the entire PR
@Override
public void addMembershipAndProxyListener(MembershipListener listener) {
super.addMembershipAndProxyListener(listener);
regionAdvisor.addMembershipListener(listener);
}
@Override
public void removeMembershipAndProxyListener(MembershipListener listener) {
regionAdvisor.removeMembershipListener(listener);
super.removeMembershipAndProxyListener(listener);
}
/**
* Called from endBucket creation. We send out a profile to notify others that the persistence is
* initialized.
*/
void endBucketCreation() {
sendProfileUpdate();
}
/**
* Profile information for a remote bucket counterpart.
*/
public static class BucketProfile extends CacheProfile {
/** True if this profile's member is attempting to initialize the bucket */
public boolean isInitializing;
/** True if this profile's member is the primary for this bucket */
public boolean isPrimary;
/**
* True if the profile is coming from a real BucketRegion acceptible states hosting = false,
* init = true hosting = true, init = false hosting = false, init = false unacceptible states
* hosting = true, init = true
*/
public boolean isHosting;
/**
* True if the bucket has been removed from this host. (
*/
public boolean removed;
public BucketProfile() {}
public BucketProfile(InternalDistributedMember memberId, int version, Bucket bucket) {
super(memberId, version);
isPrimary = bucket.isPrimary();
isHosting = bucket.isHosting();
}
@Override
public StringBuilder getToStringHeader() {
return new StringBuilder("BucketAdvisor.BucketProfile");
}
@Override
public void fillInToString(StringBuilder sb) {
super.fillInToString(sb);
sb.append("; isPrimary=").append(isPrimary);
sb.append("; isHosting=").append(isHosting);
sb.append("; isInitializing=").append(isInitializing);
}
@Override
public void fromData(DataInput in,
DeserializationContext context) throws IOException, ClassNotFoundException {
super.fromData(in, context);
isPrimary = in.readBoolean();
isHosting = in.readBoolean();
isInitializing = in.readBoolean();
}
@Override
public void toData(DataOutput out,
SerializationContext context) throws IOException {
super.toData(out, context);
out.writeBoolean(isPrimary);
out.writeBoolean(isHosting);
out.writeBoolean(isInitializing);
}
@Override
public int getDSFID() {
return BUCKET_PROFILE;
}
}
/**
* Profile information for a remote bucket hosted by cache servers.
*/
public static class ServerBucketProfile extends BucketProfile {
public Set<BucketServerLocation66> bucketServerLocations;
private int bucketId;
public ServerBucketProfile() {}
public ServerBucketProfile(InternalDistributedMember memberId, int version, Bucket bucket,
HashSet<BucketServerLocation66> serverLocations) {
super(memberId, version, bucket);
bucketId = bucket.getId();
bucketServerLocations = serverLocations;
}
@Override
public StringBuilder getToStringHeader() {
return new StringBuilder("BucketAdvisor.ServerBucketProfile");
}
@Override
public void fillInToString(StringBuilder sb) {
super.fillInToString(sb);
for (BucketServerLocation66 location : bucketServerLocations) {
sb.append("; hostName=").append(location.getHostName());
sb.append("; port=").append(location.getPort());
}
}
@Override
public void fromData(DataInput in,
DeserializationContext context) throws IOException, ClassNotFoundException {
super.fromData(in, context);
bucketServerLocations = SerializationHelper.readBucketServerLocationSet(in);
bucketId = DataSerializer.readPrimitiveInt(in);
}
@Override
public void toData(DataOutput out,
SerializationContext context) throws IOException {
super.toData(out, context);
SerializationHelper.writeBucketServerLocationSet(bucketServerLocations, out);
DataSerializer.writePrimitiveInt(bucketId, out);
}
public Set<BucketServerLocation66> getBucketServerLocations() {
return bucketServerLocations;
}
@Override
public int getDSFID() {
return SERVER_BUCKET_PROFILE;
}
@Override
public int hashCode() {
BucketServerLocation66 sl = (BucketServerLocation66) bucketServerLocations.toArray()[0];
return 31 * bucketId + sl.getPort();
}
@Override
public boolean equals(Object obj) {
if (this == obj)
return true;
if (obj == null)
return false;
if (!(obj instanceof ServerBucketProfile))
return false;
final ServerBucketProfile other = (ServerBucketProfile) obj;
if (other.bucketId != bucketId) {
return false;
}
if (other.bucketServerLocations.size() != bucketServerLocations.size()) {
return false;
}
return other.bucketServerLocations.containsAll(bucketServerLocations);
}
}
/**
* Handles the actual volunteering to become primary bucket. Ensures that only one thread is ever
* volunteering at one time.
*
*/
class VolunteeringDelegate {
/**
* Reference to the Thread that is currently volunteering. Protected by
* synchronized(volunteeringLock).
*/
private Thread volunteeringThread;
private boolean aggressive = false;
/**
* Returns true if this delegate is aggressively trying to become the primary even if another
* member is already the primary.
*
* @return true if this aggressively trying to become the primary
*/
boolean isAggressive() {
synchronized (BucketAdvisor.this) {
return aggressive;
}
}
/**
* Initiates volunteering for primary. Repeated calls are harmless. Invoked by the
* BucketAdvisor. Caller must be synchronized on BucketAdvisor.
*/
void volunteerForPrimary() {
boolean handedOff = false;
while (!handedOff) {
getAdvisee().getCancelCriterion().checkCancelInProgress(null);
boolean interrupted = Thread.interrupted();
try {
execute(this::doVolunteerForPrimary);
handedOff = true;
} catch (InterruptedException e) {
interrupted = true;
getAdvisee().getCancelCriterion().checkCancelInProgress(e);
} finally {
if (interrupted) {
Thread.currentThread().interrupt();
}
}
}
}
/**
* Reserves this delegate for the current thread to call becomePrimary. Necessary because caller
* of doVolunteerForPrimary must not be synchronized on BucketAdvisor.
*
* @return true if successfully reserved for becomePrimary
*/
boolean reserveForBecomePrimary() {
synchronized (BucketAdvisor.this) {
if (volunteeringThread != null) {
return false;
}
aggressive = true;
return true;
}
}
/**
* Invoked by the thread that performs the actual volunteering work.
*/
void doVolunteerForPrimary() {
if (!beginVolunteering()) {
return;
}
boolean dlsDestroyed = false;
try {
if (logger.isDebugEnabled()) {
logger.debug("Begin volunteerForPrimary for {}", BucketAdvisor.this);
}
DistributedMemberLock thePrimaryLock = null;
while (continueVolunteering()) {
// Fix for 41865 - We can't send out profiles while holding the
// sync on this advisor, because that will cause a deadlock.
// Holding the primaryMoveWriteLock here instead prevents any
// operations from being performed on this primary until the child regions
// are synced up. It also prevents a depose from happening until then.
BucketAdvisor parentBA = parentAdvisor;
primaryMoveWriteLock.lock();
try {
boolean acquiredLock;
getAdvisee().getCancelCriterion().checkCancelInProgress(null);
// Check our parent advisor and set our state
// accordingly
if (parentBA != null) {
// Fix for 44350 - we don't want to get a primary move lock on
// the advisor, because that might deadlock with a user thread.
// However, since all depose/elect operations on the parent bucket
// cascade to the child bucket and get the child bucket move lock,
// if should be safe to check this without the lock here.
if (parentBA.isPrimary() && !isPrimary()) {
acquiredLock = acquiredPrimaryLock();
} else {
return;
}
} else {
// we're not colocated, need to get the dlock
assignStartingBucketAdvisorIfFixedPartitioned();
if (startingBucketAdvisor != null) {
Assert.assertHoldsLock(this, false);
synchronized (startingBucketAdvisor) {
if (startingBucketAdvisor.isPrimary() && !isPrimary()) {
acquiredLock = acquiredPrimaryLock();
} else {
return;
}
}
} else {
if (thePrimaryLock == null) {
thePrimaryLock = getPrimaryLock(true);
if (thePrimaryLock == null) {
// InternalDistributedSystem.isDisconnecting probably
// prevented us from
// creating the DLS... hope there's a thread closing this
// advisor but
// it's probably not safe to assert that it already happened
return;
}
}
Assert.assertTrue(!thePrimaryLock.holdsLock());
if (isAggressive()) {
acquiredLock = thePrimaryLock.tryLock(Long.MAX_VALUE, TimeUnit.MILLISECONDS);
} else {
acquiredLock = thePrimaryLock.tryLock();
}
if (acquiredLock) {
acquiredLock = acquiredPrimaryLock();
}
} // parentAdvisor == null
}
if (acquiredLock) {
// if the lock has been acquired then try to do the same for colocated PR's too
// Here if somehow a bucket can't acquire a lock
// we assume that it is in the process of becoming primary through
// BucketAdvisor.volunteerForPrimary()
// Either way we have to guarantee that the bucket becomes primary for sure.(How to
// guarantee?)
acquirePrimaryRecursivelyForColocated();
acquirePrimaryForRestOfTheBucket();
return;
}
} finally {
primaryMoveWriteLock.unlock();
}
// else: acquiredPrimaryLock released thePrimaryLock
if (!continueVolunteering()) {
// this avoids calling the wait below...
return;
}
waitIfNoPrimaryMemberFound();
} // while
} catch (LockServiceDestroyedException e) {
dlsDestroyed = true;
handleException(e, true);
} catch (RegionDestroyedException | CancelException e) {
handleException(e, false);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
handleException(e, false);
} finally {
if (logger.isDebugEnabled()) {
logger.debug("Exit volunteerForPrimary for {}; dlsDestroyed={}", BucketAdvisor.this,
dlsDestroyed);
}
endVolunteering();
// if (isPrimary()) {
// Bucket bucket = getBucket();
// if (bucket instanceof ProxyBucketRegion) {
// bucket = ((ProxyBucketRegion)bucket).getHostedBucketRegion();
// }
// }
}
}
/**
* Called from catch blocks in {@link #doVolunteerForPrimary()}. Handles the exception properly
* based on advisor settings and shutdown condition.
*
* @param e the RuntimeException that was caught while volunteering
* @param loggit true if message should be logged if shutdown condition is not met
*/
private void handleException(Exception e, boolean loggit) {
boolean safe = isClosed() || getAdvisee().getCancelCriterion().isCancelInProgress();
if (!safe) {
if (ENFORCE_SAFE_CLOSE) {
Assert.assertTrue(safe,
"BucketAdvisor was not closed properly.");
} else if (loggit) {
logger.warn("BucketAdvisor was not closed properly.", e);
}
}
}
private boolean beginVolunteering() {
synchronized (BucketAdvisor.this) {
if (Thread.currentThread().equals(volunteeringThread)) {
return true; // this thread is already volunteering or reserved
}
if (volunteeringThread != null) {
// another thread is already volunteering
return false;
}
volunteeringThread = Thread.currentThread();
boolean changedState = false;
try {
if (isAggressive()) {
changedState = setBecoming();
} else {
changedState = setVolunteering();
}
return changedState;
} finally {
if (!changedState) {
aggressive = false;
volunteeringThread = null;
}
}
}
}
private boolean continueVolunteering() {
synchronized (BucketAdvisor.this) {
// false if caller is not the volunteeringThread
if (!Thread.currentThread().equals(volunteeringThread)) {
return false;
}
if (!isVolunteering() && !isBecomingPrimary()) {
return false;
}
// false if primaryMember is not null
if (!isAggressive() && basicGetPrimaryMember() != null) {
return false;
}
// false if this member is already primary
if (isPrimary()) {
return false;
}
// false if closed
if (isClosed()) {
return false;
}
// false if no longer hosting
return isHosting();
// must be true... need to continue volunteering
}
}
private void endVolunteering() {
if (Thread.currentThread().equals(volunteeringThread)) {
volunteeringThread = null;
aggressive = false;
}
}
private void waitIfNoPrimaryMemberFound() {
synchronized (BucketAdvisor.this) {
if (basicGetPrimaryMember() == null) {
waitForPrimaryMember(100);
if (basicGetPrimaryMember() == null) {
findAndSetPrimaryMember();
}
}
}
}
/**
* Executes the primary volunteering task after queuing it in the
* {@link BucketAdvisor#getVolunteeringQueue()}. A number of threads equal to
* {@link RegionAdvisor#VOLUNTEERING_THREAD_COUNT} are permitted to consume from the queue by
* acquiring permits from {@link BucketAdvisor#getVolunteeringSemaphore()}.
*
* @param volunteeringTask the task to queue and then execute in waiting thread pool
*
*/
private void execute(Runnable volunteeringTask) throws InterruptedException {
// @todo: instead of having a semaphore and queue on RegionAdvisor
// we should have an executor which limits its max threads to
// VOLUNTEERING_THREAD_COUNT.
if (Thread.interrupted())
throw new InterruptedException();
Queue<Runnable> volunteeringQueue = getVolunteeringQueue();
synchronized (volunteeringQueue) {
// add the volunteering task
volunteeringQueue.add(volunteeringTask);
if (getVolunteeringSemaphore().tryAcquire()) {
// ensure there is a thread consuming the queue
boolean handedOff = false;
try {
getDistributionManager().getExecutors().getWaitingThreadPool().execute(consumeQueue());
handedOff = true;
} finally {
if (!handedOff) {
getVolunteeringSemaphore().release();
}
}
}
}
}
/**
* Returns the runnable used to consume the volunteering queue. The executing thread(s) will
* consume from the queue until it is empty.
*
* @return runnable for consuming the volunteering queue
*/
private Runnable consumeQueue() {
return () -> {
getPartitionedRegionStats().incVolunteeringThreads(1);
boolean releaseSemaphore = true;
try {
Queue<Runnable> volunteeringQueue = getVolunteeringQueue();
Runnable queuedWork;
while (true) {
// SystemFailure.checkFailure();
getAdvisee().getCancelCriterion().checkCancelInProgress(null);
synchronized (volunteeringQueue) {
// synchronized volunteeringQueue for coordination between threads adding
// work to the queue and checking for a consuming thread and the existing
// consuming thread to determine if it can exit since the queue is empty.
queuedWork = volunteeringQueue.poll();
if (queuedWork == null) {
// the queue is empty... no more work... so return
// @todo why release the semaphore here are sync'ed?
// we could just let the finally block do it.
getVolunteeringSemaphore().release();
releaseSemaphore = false;
return;
}
// still more work in the queue so let's run it
}
try {
queuedWork.run();
} catch (CancelException e) {
return;
} catch (RuntimeException e) {
// log and continue consuming queue
logger.error(e.getMessage(), e);
}
}
} finally {
getPartitionedRegionStats().incVolunteeringThreads(-1);
if (releaseSemaphore) {
// Clean up, just in case
getVolunteeringSemaphore().release();
}
}
};
}
}
void setShadowBucketDestroyed(boolean destroyed) {
shadowBucketDestroyed = destroyed;
}
public boolean getShadowBucketDestroyed() {
return shadowBucketDestroyed;
}
}