Google Git
Sign in
apache / geode / 7038f08ea88ec42f6c58764525460a1c7fe1abcf / . / geode-core / src / main / java / org / apache / geode / internal / cache / TXCommitMessage.java
blob: 893c117b56408ce219dc8b0718452350c0c890f9 [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.LocalRegion.InitializationLevel.BEFORE_INITIAL_IMAGE;
import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.IdentityHashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Set;
import org.apache.logging.log4j.Logger;
import org.apache.geode.CancelException;
import org.apache.geode.DataSerializer;
import org.apache.geode.SystemFailure;
import org.apache.geode.annotations.Immutable;
import org.apache.geode.annotations.internal.MakeNotStatic;
import org.apache.geode.cache.Cache;
import org.apache.geode.cache.CacheClosedException;
import org.apache.geode.cache.CacheRuntimeException;
import org.apache.geode.cache.CommitDistributionException;
import org.apache.geode.cache.CommitIncompleteException;
import org.apache.geode.cache.DataPolicy;
import org.apache.geode.cache.Operation;
import org.apache.geode.cache.RegionDestroyedException;
import org.apache.geode.cache.RegionDistributionException;
import org.apache.geode.cache.TransactionId;
import org.apache.geode.cache.TransactionListener;
import org.apache.geode.distributed.DistributedMember;
import org.apache.geode.distributed.internal.ClusterDistributionManager;
import org.apache.geode.distributed.internal.DistributionManager;
import org.apache.geode.distributed.internal.DistributionMessage;
import org.apache.geode.distributed.internal.MembershipListener;
import org.apache.geode.distributed.internal.MessageWithReply;
import org.apache.geode.distributed.internal.PooledDistributionMessage;
import org.apache.geode.distributed.internal.ReliableReplyProcessor21;
import org.apache.geode.distributed.internal.ReplyException;
import org.apache.geode.distributed.internal.ReplyMessage;
import org.apache.geode.distributed.internal.ReplyProcessor21;
import org.apache.geode.distributed.internal.membership.InternalDistributedMember;
import org.apache.geode.internal.Assert;
import org.apache.geode.internal.HeapDataOutputStream;
import org.apache.geode.internal.InternalDataSerializer;
import org.apache.geode.internal.cache.LocalRegion.InitializationLevel;
import org.apache.geode.internal.cache.locks.TXLockId;
import org.apache.geode.internal.cache.locks.TXLockIdImpl;
import org.apache.geode.internal.cache.locks.TXLockService;
import org.apache.geode.internal.cache.partitioned.Bucket;
import org.apache.geode.internal.cache.persistence.PersistentMemberID;
import org.apache.geode.internal.cache.tier.sockets.ClientProxyMembershipID;
import org.apache.geode.internal.cache.versions.VersionSource;
import org.apache.geode.internal.cache.versions.VersionTag;
import org.apache.geode.internal.offheap.annotations.Released;
import org.apache.geode.internal.serialization.DeserializationContext;
import org.apache.geode.internal.serialization.SerializationContext;
import org.apache.geode.internal.serialization.Version;
import org.apache.geode.logging.internal.executors.LoggingThread;
import org.apache.geode.logging.internal.log4j.api.LogService;
/**
* TXCommitMessage is the message that contains all the information that needs to be distributed, on
* commit, to other cache members.
*
* @since GemFire 4.0
*/
public class TXCommitMessage extends PooledDistributionMessage
implements MembershipListener, MessageWithReply {
private static final Logger logger = LogService.getLogger();
// Keep a 60 second history @ an estimated 1092 transactions/second ~= 16^4
@MakeNotStatic
protected static final TXFarSideCMTracker txTracker = new TXFarSideCMTracker((60 * 1092));
private ArrayList<RegionCommit> regions; // list of RegionCommit instances
protected TXId txIdent;
protected int processorId; // 0 unless needsAck is true
protected TXLockIdImpl lockId;
protected HashSet farSiders;
protected transient DistributionManager dm; // Used on the sending side of this message
private transient int sequenceNum = 0;
// Maps receiver Serializables to RegionCommitList instances
private transient HashMap<InternalDistributedMember, RegionCommitList> msgMap = null;
private transient RegionCommit currentRegion;
protected transient TXState txState = null;
private transient boolean wasProcessed;
private transient boolean isProcessing;
private transient boolean dontProcess;
private transient boolean departureNoticed = false;
private transient boolean lockNeedsUpdate = false;
private transient boolean ackRequired = true;
/**
* List of operations to do when processing this tx. Valid on farside only.
*/
protected transient ArrayList farSideEntryOps;
private byte[] farsideBaseMembershipId; // only available on farside
private long farsideBaseThreadId; // only available on farside
private long farsideBaseSequenceId; // only available on farside
/**
* (Nearside) true of any regions in this TX have required roles
*/
private transient boolean hasReliableRegions = false;
/**
* Set of all caching exceptions produced while processing this tx
*/
private transient Set processingExceptions = Collections.emptySet();
private ClientProxyMembershipID bridgeContext = null;
/**
* Version of the client that this TXCommitMessage is being sent to. Used for backwards
* compatibility
*/
private transient Version clientVersion;
/**
* A token to be put in TXManagerImpl#failoverMap to represent a CommitConflictException while
* committing a transaction
*/
@Immutable
public static final TXCommitMessage CMT_CONFLICT_MSG = new TXCommitMessage();
/**
* A token to be put in TXManagerImpl#failoverMap to represent a
* TransactionDataNodeHasDepartedException
* while committing a transaction
*/
@Immutable
public static final TXCommitMessage REBALANCE_MSG = new TXCommitMessage();
/**
* A token to be put in TXManagerImpl#failoverMap to represent an exception while committing a
* transaction
*/
@Immutable
public static final TXCommitMessage EXCEPTION_MSG = new TXCommitMessage();
/**
* A token to be put in TXManagerImpl#failoverMap to represent a rolled back transaction
*/
@Immutable
public static final TXCommitMessage ROLLBACK_MSG = new TXCommitMessage();
public TXCommitMessage(TXId txIdent, DistributionManager dm, TXState txState) {
this.dm = dm;
this.txIdent = txIdent;
this.lockId = null;
this.regions = null;
this.txState = txState;
this.wasProcessed = false;
this.isProcessing = false;
this.dontProcess = false;
this.farSiders = null;
this.bridgeContext = txState.bridgeContext;
}
public TXCommitMessage() {
// zero arg constructor for DataSerializer
}
public static TXFarSideCMTracker getTracker() {
return TXCommitMessage.txTracker;
}
/**
* Create and return an eventId given its offset.
*
* @since GemFire 5.7
*/
protected EventID getEventId(int eventOffset) {
return new EventID(this.farsideBaseMembershipId, this.farsideBaseThreadId,
this.farsideBaseSequenceId + eventOffset);
}
/**
* Return the TXCommitMessage we have already received that is associated with id. Note because of
* bug 37657 we may need to wait for it to show up.
*/
public static TXCommitMessage waitForMessage(Object id, DistributionManager dm) {
TXFarSideCMTracker map = getTracker();
return map.waitForMessage(id, dm);
}
void startRegion(InternalRegion r, int maxSize) {
this.currentRegion = new RegionCommit(this, r, maxSize);
if (r.requiresReliabilityCheck()) {
this.hasReliableRegions = true;
}
}
void finishRegion(Set<InternalDistributedMember> s) {
// make sure we have some changes and someone to send them to
if (!this.currentRegion.isEmpty() && s != null && !s.isEmpty()) {
// Get the persistent ids for the current region and save them
this.currentRegion.persistentIds = getPersistentIds(this.currentRegion.internalRegion);
if (this.msgMap == null) {
this.msgMap = new HashMap<>();
}
{
RegionCommitList newRCL = null;
Iterator<InternalDistributedMember> it = s.iterator();
while (it.hasNext()) {
InternalDistributedMember recipient = it.next();
if (!this.dm.getDistributionManagerIds().contains(recipient)) {
if (logger.isDebugEnabled()) {
logger.debug("Skipping member {} due to dist list absence", recipient);
}
// skip this member since the dm no longer knows about it
continue;
}
RegionCommitList rcl = this.msgMap.get(recipient);
if (rcl == null) {
if (newRCL == null) {
rcl = new RegionCommitList();
rcl.add(this.currentRegion);
newRCL = rcl;
} else {
rcl = newRCL;
}
this.msgMap.put(recipient, rcl);
} else if (rcl.get(rcl.size() - 1) != this.currentRegion) {
rcl.add(this.currentRegion);
}
}
}
// Now deal with each existing recipient that does not care
// about this region
Iterator<Map.Entry<InternalDistributedMember, RegionCommitList>> it =
this.msgMap.entrySet().iterator();
while (it.hasNext()) {
Map.Entry<InternalDistributedMember, RegionCommitList> me = it.next();
if (!s.contains(me.getKey())) {
RegionCommitList rcl = me.getValue();
RegionCommitList trimmedRcl = rcl.trim(this.currentRegion);
if (trimmedRcl != rcl) {
me.setValue(trimmedRcl);
}
}
}
}
this.currentRegion = null;
}
private Map<InternalDistributedMember, PersistentMemberID> getPersistentIds(InternalRegion r) {
if (r instanceof DistributedRegion) {
return ((CacheDistributionAdvisee) r).getCacheDistributionAdvisor().advisePersistentMembers();
} else {
return Collections.emptyMap();
}
}
void finishRegionComplete() {
// make sure we have some changes and someone to send them to
if (!this.currentRegion.isEmpty()) {
{
if (this.regions == null) {
this.regions = new RegionCommitList();
}
this.regions.add(this.currentRegion);
}
}
this.currentRegion = null;
}
Map viewVersions = new HashMap();
private Boolean needsLargeModCount;
private transient boolean disableListeners = false;
/**
* record CacheDistributionAdvisor.startOperation versions for later cleanup
*/
protected void addViewVersion(DistributedRegion dr, long version) {
viewVersions.put(dr, version);
}
protected void releaseViewVersions() {
RuntimeException rte = null;
for (Iterator it = viewVersions.entrySet().iterator(); it.hasNext();) {
Map.Entry e = (Map.Entry) it.next();
DistributedRegion dr = (DistributedRegion) e.getKey();
Long viewVersion = (Long) e.getValue();
// need to continue the iteration if one of the regions is destroyed
// since others may still be okay
try {
dr.getDistributionAdvisor().endOperation(viewVersion);
} catch (RuntimeException ex) {
rte = ex;
}
}
if (rte != null) {
throw rte;
}
}
private boolean isEmpty() {
return this.msgMap == null || this.msgMap.isEmpty();
}
void addOp(InternalRegion r, Object key, TXEntryState entry, Set otherRecipients) {
this.currentRegion.addOp(key, entry);
}
void send(TXLockId lockId) {
if (isEmpty()) {
if (logger.isDebugEnabled()) {
logger.debug("empty transaction - nothing to distribute");
}
return;
}
Assert.assertTrue(this.txState != null, "Send must have transaction state.");
this.lockId = (TXLockIdImpl) lockId;
updateLockMembers();
IdentityHashMap distMap = new IdentityHashMap(); // Map of RegionCommitList keys to Sets of
// receivers
HashSet ackReceivers = null;
{
Iterator it = this.msgMap.entrySet().iterator();
while (it.hasNext()) {
Map.Entry me = (Map.Entry) it.next();
RegionCommitList rcl = (RegionCommitList) me.getValue();
if (rcl.getNeedsAck()) {
if (ackReceivers == null) {
ackReceivers = new HashSet();
}
ackReceivers.add(me.getKey());
}
HashSet receivers = (HashSet) distMap.get(rcl);
if (receivers == null) {
receivers = new HashSet();
distMap.put(rcl, receivers);
}
receivers.add(me.getKey());
}
}
CommitReplyProcessor processor = null;
{
if (ackReceivers != null) {
processor = new CommitReplyProcessor(this.dm, ackReceivers, msgMap);
if (ackReceivers.size() > 1) {
this.farSiders = ackReceivers;
}
processor.enableSevereAlertProcessing();
}
{
Iterator it = distMap.entrySet().iterator();
while (it.hasNext()) {
Map.Entry me = (Map.Entry) it.next();
RegionCommitList rcl = (RegionCommitList) me.getKey();
HashSet recipients = (HashSet) me.getValue();
// now remove from the recipients any recipients that the dm no
// longer knows about
recipients.retainAll(this.dm.getDistributionManagerIds());
if (!recipients.isEmpty()) {
if (this.txState.internalDuringIndividualSend != null) {
// Run in test mode, splitting out individual recipients,
// so we can control who gets what
Iterator indivRecip = recipients.iterator();
while (indivRecip.hasNext()) {
this.txState.internalDuringIndividualSend.run();
setRecipientsSendData(Collections.singleton(indivRecip.next()), processor, rcl);
}
} else {
// Run in normal mode sending to multiple recipients in
// one shot
setRecipientsSendData(recipients, processor, rcl);
}
}
}
}
if (this.txState.internalAfterIndividualSend != null) {
this.txState.internalAfterIndividualSend.run();
}
}
if (processor != null) {
// Send the CommitProcessMessage
final CommitProcessMessage cpMsg;
if (this.lockId != null) {
cpMsg = new CommitProcessForLockIdMessage(this.lockId);
} else {
cpMsg = new CommitProcessForTXIdMessage(this.txIdent);
}
if (this.txState.internalDuringIndividualCommitProcess != null) {
// Run in test mode
Iterator<InternalDistributedMember> indivRecip = ackReceivers.iterator();
while (indivRecip.hasNext()) {
this.txState.internalDuringIndividualCommitProcess.run();
cpMsg.setRecipients(Collections.<InternalDistributedMember>singleton(indivRecip.next()));
this.dm.putOutgoing(cpMsg);
cpMsg.resetRecipients();
}
} else {
// Run in normal mode
cpMsg.setRecipients(ackReceivers);
this.dm.putOutgoing(cpMsg);
}
if (this.txState.internalAfterIndividualCommitProcess != null) {
// Testing callback
this.txState.internalAfterIndividualCommitProcess.run();
}
// for() loop removed for bug 36983 - you can't loop on waitForReplies()
dm.getCancelCriterion().checkCancelInProgress(null);
processor.waitForCommitCompletion();
this.dm.getStats().incCommitWaits();
}
if (this.hasReliableRegions) {
checkDistributionReliability(distMap, processor);
}
}
@Override
public boolean containsRegionContentChange() {
return true;
}
/**
* Checks reliable regions and throws CommitDistributionException if any required roles may not
* have received the commit message.
*
* @param distMap map of RegionCommitList keys to Sets of receivers
* @param processor the reply processor
* @throws CommitDistributionException if any required roles may not have received the commit
* message
*/
private void checkDistributionReliability(Map distMap, CommitReplyProcessor processor) {
// key=RegionCommit, value=Set of recipients
Map regionToRecipients = new IdentityHashMap();
// build up the keys in regionToRecipients and add all receivers
for (Iterator distIter = distMap.entrySet().iterator(); distIter.hasNext();) {
Map.Entry me = (Map.Entry) distIter.next();
RegionCommitList rcl = (RegionCommitList) me.getKey();
Set recipients = (Set) me.getValue();
for (Iterator rclIter = rcl.iterator(); rclIter.hasNext();) {
RegionCommit rc = (RegionCommit) rclIter.next();
// skip region if no required roles
if (!rc.internalRegion.requiresReliabilityCheck()) {
continue;
}
Set recipientsForRegion = (Set) regionToRecipients.get(rc);
if (recipientsForRegion == null) {
recipientsForRegion = new HashSet();
regionToRecipients.put(rc, recipientsForRegion);
}
// get the receiver Set for rcl and perform addAll
if (recipients != null) {
recipientsForRegion.addAll(recipients);
}
}
}
Set cacheClosedMembers =
(processor == null) ? Collections.emptySet() : processor.getCacheClosedMembers();
Set departedMembers =
(processor == null) ? Collections.emptySet() : processor.getDepartedMembers();
// check reliability on each region
Set regionDistributionExceptions = Collections.emptySet();
Set failedRegionNames = Collections.emptySet();
for (Iterator iter = regionToRecipients.entrySet().iterator(); iter.hasNext();) {
Map.Entry me = (Map.Entry) iter.next();
final RegionCommit rc = (RegionCommit) me.getKey();
final Set successfulRecipients = new HashSet(msgMap.keySet());
successfulRecipients.removeAll(departedMembers);
// remove members who destroyed that region or closed their cache
Set regionDestroyedMembers = (processor == null) ? Collections.emptySet()
: processor.getRegionDestroyedMembers(rc.internalRegion.getFullPath());
successfulRecipients.removeAll(cacheClosedMembers);
successfulRecipients.removeAll(regionDestroyedMembers);
try {
rc.internalRegion.handleReliableDistribution(successfulRecipients);
} catch (RegionDistributionException e) {
if (regionDistributionExceptions == Collections.emptySet()) {
regionDistributionExceptions = new HashSet();
failedRegionNames = new HashSet();
}
regionDistributionExceptions.add(e);
failedRegionNames.add(rc.internalRegion.getFullPath());
}
}
if (!regionDistributionExceptions.isEmpty()) {
throw new CommitDistributionException(
String.format(
"These regions experienced reliability failure during distribution of the operation: %s",
failedRegionNames),
regionDistributionExceptions);
}
}
/**
* Helper method for send
*/
private void setRecipientsSendData(Set recipients, ReplyProcessor21 processor,
RegionCommitList rcl) {
setRecipients(recipients);
this.regions = rcl;
if (rcl.getNeedsAck()) {
this.processorId = processor.getProcessorId();
} else {
this.processorId = 0;
}
this.dm.getStats().incSentCommitMessages(1L);
this.sequenceNum++;
this.dm.putOutgoing(this);
resetRecipients();
}
@Override
protected void process(ClusterDistributionManager dm) {
this.dm = dm;
// Remove this node from the set of recipients
if (this.farSiders != null) {
this.farSiders.remove(dm.getId());
}
if (this.processorId != 0) {
TXLockService.createDTLS(this.dm.getSystem()); // fix bug 38843; no-op if already created
synchronized (this) {
// Handle potential origin departure
this.dm.addMembershipListener(this);
// Assume ACK mode, defer processing until we receive a
// CommitProcess message
if (logger.isDebugEnabled()) {
final Object key = getTrackerKey();
logger.debug("Adding key:{} class{} to tracker list", key, key.getClass().getName());
}
txTracker.add(this);
}
if (!this.dm.getDistributionManagerIds().contains(getSender())) {
memberDeparted(this.dm, getSender(), false /* don't care */);
}
} else {
basicProcess();
}
}
/**
* Adds an entry op for this tx to do on the far side
*/
void addFarSideEntryOp(RegionCommit.FarSideEntryOp entryOp) {
this.farSideEntryOps.add(entryOp);
}
protected void addProcessingException(Exception e) {
// clear all previous exceptions if e is a CacheClosedException
if (this.processingExceptions == Collections.emptySet() || e instanceof CancelException) {
this.processingExceptions = new HashSet();
}
this.processingExceptions.add(e);
}
public void setDM(DistributionManager dm) {
this.dm = dm;
}
public void basicProcess() {
final DistributionManager dm = this.dm;
synchronized (this) {
if (isProcessing()) {
if (logger.isDebugEnabled()) {
logger.debug("TXCommitMessage {} is already in process, returning", this);
}
return;
} else {
setIsProcessing(true);
}
}
if (logger.isDebugEnabled()) {
logger.debug("begin processing TXCommitMessage for {}", this.txIdent);
}
final InitializationLevel oldLevel =
LocalRegion.setThreadInitLevelRequirement(BEFORE_INITIAL_IMAGE);
boolean forceListener = false; // this gets flipped if we need to fire tx listener
// it needs to default to false because we don't want to fire listeners on pr replicates
try {
TXRmtEvent txEvent = null;
final Cache cache = dm.getExistingCache();
if (cache == null) {
addProcessingException(new CacheClosedException());
// return ... this cache is closed so we can't do anything.
return;
}
final TransactionListener[] tls = cache.getCacheTransactionManager().getListeners();
if (tls.length > 0) {
txEvent = new TXRmtEvent(this.txIdent, cache);
}
try {
// Pre-process each Region in the tx
try {
Iterator it = this.regions.iterator();
while (it.hasNext()) {
boolean failedBeginProcess = true;
RegionCommit rc = (RegionCommit) it.next();
try {
failedBeginProcess = !rc.beginProcess(dm, this.txIdent, txEvent);
} catch (CacheRuntimeException problem) {
processCacheRuntimeException(problem);
} finally {
if (failedBeginProcess) {
rc.internalRegion = null; // Cause related FarSideEntryOps to skip processing
it.remove(); // Skip endProcessing as well
}
}
}
basicProcessOps();
} finally { // fix for bug 40001
// post-process each Region in the tx
Iterator it = this.regions.iterator();
while (it.hasNext()) {
try {
RegionCommit rc = (RegionCommit) it.next();
rc.endProcess();
if (rc.isForceFireEvent(dm)) {
forceListener = true;
}
} catch (CacheRuntimeException problem) {
processCacheRuntimeException(problem);
}
}
}
/*
* We need to make sure that we should fire a TX afterCommit event.
*/
boolean internalEvent = (txEvent != null && txEvent.hasOnlyInternalEvents());
if (!disableListeners && !internalEvent
&& (forceListener || (txEvent != null && !txEvent.isEmpty()))) {
for (int i = 0; i < tls.length; i++) {
try {
tls[i].afterCommit(txEvent);
} 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 TransactionListener",
t);
}
}
}
} catch (CancelException e) {
processCacheRuntimeException(e);
} finally {
if (txEvent != null) {
txEvent.freeOffHeapResources();
}
}
} finally {
LocalRegion.setThreadInitLevelRequirement(oldLevel);
if (isAckRequired()) {
ack();
}
if (!dm.getExistingCache().isClient() && bridgeContext != null) {
getTracker().saveTXForClientFailover(txIdent, this);
}
if (logger.isDebugEnabled()) {
logger.debug("completed processing TXCommitMessage for {}", this.txIdent);
}
}
}
public void basicProcessOps() {
List<EntryEventImpl> pendingCallbacks = new ArrayList<>(this.farSideEntryOps.size());
Collections.sort(this.farSideEntryOps);
Iterator it = this.farSideEntryOps.iterator();
while (it.hasNext()) {
try {
RegionCommit.FarSideEntryOp entryOp = (RegionCommit.FarSideEntryOp) it.next();
entryOp.process(pendingCallbacks);
} catch (CacheRuntimeException problem) {
processCacheRuntimeException(problem);
} catch (Exception e) {
addProcessingException(e);
}
}
firePendingCallbacks(pendingCallbacks);
}
private void firePendingCallbacks(List<EntryEventImpl> callbacks) {
Iterator<EntryEventImpl> ci = callbacks.iterator();
while (ci.hasNext()) {
EntryEventImpl ee = ci.next();
try {
if (ee.getOperation().isDestroy()) {
ee.getRegion().invokeTXCallbacks(EnumListenerEvent.AFTER_DESTROY, ee, true);
} else if (ee.getOperation().isInvalidate()) {
ee.getRegion().invokeTXCallbacks(EnumListenerEvent.AFTER_INVALIDATE, ee, true);
} else if (ee.getOperation().isCreate()) {
ee.getRegion().invokeTXCallbacks(EnumListenerEvent.AFTER_CREATE, ee, true);
} else {
ee.getRegion().invokeTXCallbacks(EnumListenerEvent.AFTER_UPDATE, ee, true);
}
} finally {
ee.release();
}
}
}
protected void processCacheRuntimeException(CacheRuntimeException problem) {
if (problem instanceof RegionDestroyedException) { // catch RegionDestroyedException
addProcessingException(problem);
} else if (problem instanceof CancelException) { // catch CacheClosedException
addProcessingException(problem);
throw problem;
} else { // catch CacheRuntimeException
addProcessingException(problem);
logger.error(
"Transaction message {} from sender {} failed processing, unknown transaction state: {}",
new Object[] {this, getSender(), problem});
}
}
private void ack() {
if (this.processorId != 0) {
CommitReplyException replyEx = null;
if (!this.processingExceptions.isEmpty()) {
replyEx = new CommitReplyException(
String.format("Commit operation generated one or more exceptions from %s",
this.getSender()),
this.processingExceptions);
}
ReplyMessage.send(getSender(), this.processorId, replyEx, this.dm);
}
}
@Override
public int getDSFID() {
// on near side send old TX_COMMIT_MESSAGE if there is at least one 7.0
// member in the system, otherwise send the new 7.0.1 message.
// 7.0.1 members will be able to deserialize either
// if (shouldSend701Message()) {
// this.shouldWriteShadowKey = true;
// return TX_COMMIT_MESSAGE_701;
return TX_COMMIT_MESSAGE;
/*
* } this.shouldWriteShadowKey = false; return TX_COMMIT_MESSAGE;
*/
}
/*
* /** Do not send shadowKey to clients or when there are member(s) older than 7.0.1.
*
* private boolean shouldSend701Message() { if (this.clientVersion == null &&
* this.getDM().getMembersWithOlderVersion("7.0.1").isEmpty()) { return true; } return false; }
*
* public boolean shouldReadShadowKey() { return this.shouldReadShadowKey; }
*
* public void setShouldReadShadowKey(boolean shouldReadShadowKey) { this.shouldReadShadowKey =
* shouldReadShadowKey; }
*
* public boolean shouldWriteShadowKey() { return this.shouldWriteShadowKey; }
*/
@Override
public void fromData(DataInput in,
DeserializationContext context) throws IOException, ClassNotFoundException {
int pId = in.readInt();
if (isAckRequired()) {
this.processorId = pId;
ReplyProcessor21.setMessageRPId(this.processorId);
} else {
this.processorId = -1;
}
this.txIdent = TXId.createFromData(in);
if (in.readBoolean()) {
this.lockId = TXLockIdImpl.createFromData(in);
}
int totalMaxSize = in.readInt();
this.farsideBaseMembershipId = DataSerializer.readByteArray(in);
this.farsideBaseThreadId = in.readLong();
this.farsideBaseSequenceId = in.readLong();
this.needsLargeModCount = in.readBoolean();
final boolean hasShadowKeys = hasFlagsField(in) ? in.readBoolean() : useShadowKey();
int regionsSize = in.readInt();
this.regions = new ArrayList(regionsSize);
this.farSideEntryOps = new ArrayList(totalMaxSize);
for (int i = 0; i < regionsSize; i++) {
RegionCommit rc = new RegionCommit(this);
try {
rc.fromData(in, hasShadowKeys);
} catch (CacheClosedException cce) {
addProcessingException(cce);
// return to avoid serialization error being sent in reply
return;
}
this.regions.add(rc);
}
this.bridgeContext = ClientProxyMembershipID.readCanonicalized(in);
this.farSiders = DataSerializer.readHashSet(in);
}
/**
* Return true if a distributed ack message is required. On the client side of a transaction, this
* returns false, while returning true elsewhere.
*
* @return requires ack message or not
*/
private boolean isAckRequired() {
return this.ackRequired;
}
/**
* Indicate whether an ack is required. Defaults to true.
*
* @param a true if we require an ack. false if not. false on clients.
*/
public void setAckRequired(boolean a) {
this.ackRequired = a;
if (!a) {
this.processorId = -1;
}
}
@Override
public void toData(DataOutput out,
SerializationContext context) throws IOException {
out.writeInt(this.processorId);
InternalDataSerializer.invokeToData(this.txIdent, out);
{
boolean hasLockId = this.lockId != null;
out.writeBoolean(hasLockId);
if (hasLockId) {
InternalDataSerializer.invokeToData(this.lockId, out);
}
}
int regionsSize = 0;
{
int totalMaxSize = 0;
if (this.regions != null) {
regionsSize = this.regions.size();
for (int i = 0; i < this.regions.size(); i++) {
RegionCommit rc = (RegionCommit) this.regions.get(i);
totalMaxSize += rc.maxSize;
}
}
out.writeInt(totalMaxSize);
}
if (this.txState != null) {
DataSerializer.writeByteArray(this.txState.getBaseMembershipId(), out);
out.writeLong(this.txState.getBaseThreadId());
out.writeLong(this.txState.getBaseSequenceId());
} else {
DataSerializer.writeByteArray(this.farsideBaseMembershipId, out);
out.writeLong(this.farsideBaseThreadId);
out.writeLong(this.farsideBaseSequenceId);
}
if (this.txState != null) {
DataSerializer.writeBoolean(this.txState.needsLargeModCount(), out);
} else {
DataSerializer.writeBoolean(this.needsLargeModCount, out);
}
final boolean useShadowKey = useShadowKey();
if (hasFlagsField(out)) {
out.writeBoolean(useShadowKey);
}
out.writeInt(regionsSize);
{
if (regionsSize > 0) {
for (int i = 0; i < this.regions.size(); i++) {
RegionCommit rc = (RegionCommit) this.regions.get(i);
rc.toData(out, context, useShadowKey);
}
}
}
DataSerializer.writeObject(bridgeContext, out);
DataSerializer.writeHashSet(this.farSiders, out);
}
private boolean hasFlagsField(final DataOutput out) {
return hasFlagsField(InternalDataSerializer.getVersionForDataStream(out));
}
private boolean hasFlagsField(final DataInput in) {
return hasFlagsField(InternalDataSerializer.getVersionForDataStream(in));
}
private boolean hasFlagsField(final Version version) {
return version.compareTo(Version.GEODE_1_7_0) >= 0;
}
private boolean useShadowKey() {
return null == clientVersion;
}
@Override
public String toString() {
StringBuilder result = new StringBuilder(256);
result.append("TXCommitMessage@").append(System.identityHashCode(this)).append("#")
.append(this.sequenceNum).append(" processorId=").append(this.processorId).append(" txId=")
.append(this.txIdent);
if (this.farSiders != null) {
Iterator fs = this.farSiders.iterator();
result.append(" farSiders=");
while (fs.hasNext()) {
result.append(fs.next());
if (fs.hasNext()) {
result.append(' ');
}
}
} else {
result.append(" farSiders=<null>");
}
if (this.regions != null) {
Iterator it = this.regions.iterator();
while (it.hasNext()) {
result.append(' ').append(it.next());
}
}
return result.toString();
}
/**
* Combines a set of small TXCommitMessages that belong to one transaction into a txCommitMessage
* that represents an entire transaction. At commit time the txCommitMessage sent to each node can
* be a subset of the transaction, this method will combine those subsets into a complete
* message.
*
* @return the complete txCommitMessage
*/
public static TXCommitMessage combine(Set<TXCommitMessage> msgSet) {
assert msgSet != null;
TXCommitMessage firstPart = null;
Iterator<TXCommitMessage> it = msgSet.iterator();
while (it.hasNext()) {
if (firstPart == null) {
firstPart = it.next();
continue;
}
firstPart.combine(it.next());
}
return firstPart;
}
/**
* Combines the other TXCommitMessage into this message. Used to compute complete TXCommitMessage
* from parts.
*/
public void combine(TXCommitMessage other) {
assert other != null;
Iterator it = other.regions.iterator();
Map<String, RegionCommit> regionCommits = new HashMap<>();
for (RegionCommit commit : regions) {
regionCommits.put(commit.getRegionPath(), commit);
}
for (RegionCommit commit : other.regions) {
if (!regionCommits.containsKey(commit.getRegionPath())) {
commit.msg = this;
this.regions.add(commit);
regionCommits.put(commit.getRegionPath(), commit);
}
}
}
public static class RegionCommitList extends ArrayList<RegionCommit> {
private static final long serialVersionUID = -8910813949027683641L;
private transient boolean needsAck = false;
private transient RegionCommit trimRC = null;
private transient RegionCommitList trimChild = null;
public RegionCommitList() {
super();
}
public RegionCommitList(RegionCommitList c) {
super(c);
}
public boolean getNeedsAck() {
return this.needsAck;
}
@Override // GemStoneAddition
public boolean add(RegionCommit o) {
RegionCommit rc = (RegionCommit) o;
rc.incRefCount();
if (!this.needsAck && rc.needsAck()) {
this.needsAck = true;
}
return super.add(o);
}
/**
* Creates a new list, if needed, that contains all the elements of the specified old list
* except the last one if it is 'rc'. Also recomputes needsAck field.
*/
public RegionCommitList trim(RegionCommit rc) {
if (get(size() - 1) != rc) {
// no need to trim because it does not contain rc
return this;
}
if (this.trimRC == rc) {
return this.trimChild;
}
RegionCommitList result = new RegionCommitList(this);
this.trimRC = rc;
this.trimChild = result;
result.remove(result.size() - 1);
{
Iterator it = result.iterator();
while (it.hasNext()) {
RegionCommit itrc = (RegionCommit) it.next();
itrc.incRefCount();
if (itrc.needsAck()) {
result.needsAck = true;
}
}
}
return result;
}
@Override
public String toString() {
StringBuilder result = new StringBuilder(256);
result.append('@').append(System.identityHashCode(this)).append(' ').append(super.toString());
return result.toString();
}
}
public static class RegionCommit {
private final TxCallbackEventFactory txCallbackEventFactory = new TxCallbackEventFactoryImpl();
/**
* The region that this commit represents. Valid on both nearside and farside.
*/
protected transient InternalRegion internalRegion;
/**
* Valid only on farside.
*/
private String regionPath;
private String parentRegionPath;
/**
* The message this region commit is a part of. Valid on both farside and nearside.
*/
private transient TXCommitMessage msg;
/**
* Number of RegionCommitList instances that have this RegionCommit in them Valid only on
* nearside.
*/
private transient int refCount = 0;
/**
* Valid only on nearside.
*/
private transient HeapDataOutputStream preserializedBuffer = null;
/**
* Upperbound on the number of operations this region could possibly have Valid only on
* nearside.
*/
transient int maxSize;
/**
* A list of Object; each one is the entry key for a distributed operation done by this
* transaction. The list must be kept in sync with opKeys. Valid only on nearside.
*/
private transient ArrayList opKeys;
/**
* A list of TXEntryState; each one is the entry info for a distributed operation done by this
* transaction. The list must be kept in sync with opKeys. Valid only on nearside.
*/
private transient ArrayList opEntries;
private transient VersionSource memberId;
/**
* The persistent ids of the peers for this region. Used to mark peers as offline if they do not
* apply the commit due to a cache close.
*/
public Map<InternalDistributedMember, PersistentMemberID> persistentIds;
/**
* Used on nearside
*/
RegionCommit(TXCommitMessage msg, InternalRegion r, int maxSize) {
this.msg = msg;
this.internalRegion = r;
this.maxSize = maxSize;
}
/**
* Used on farside who inits r later and never sets maxSize
*/
RegionCommit(TXCommitMessage msg) {
this.msg = msg;
}
public String getRegionPath() {
return regionPath;
}
public void incRefCount() {
this.refCount++;
}
/**
* Valid on farside after beginProcess. Used to remember what to do at region cleanup time
*/
private boolean needsUnlock;
/**
* Valid on farside after beginProcess. Used to remember what to do at region cleanup time
*/
private boolean needsLRUEnd;
/**
* Valid on farside after beginProcess This is the txEvent that should be used by this
* RegionCommit
*/
private TXRmtEvent txEvent;
/**
* Called to setup a region commit so its entryOps can be processed
*
* @return true if region should be processed; false if it can be ignored
* @throws CacheClosedException if the cache has been closed
*/
boolean beginProcess(DistributionManager dm, TransactionId txIdent, TXRmtEvent txEvent)
throws CacheClosedException {
if (logger.isDebugEnabled()) {
logger.debug("begin processing TXCommitMessage {} for region {}", txIdent, this.regionPath);
}
try {
if (!hookupRegion(dm)) {
return false;
}
if (msg.isAckRequired()
&& (this.internalRegion == null || !this.internalRegion.getScope().isDistributed())) {
if (logger.isDebugEnabled()) {
logger.debug("Received unneeded commit data for region {}", this.regionPath);
}
this.msg.addProcessingException(new RegionDestroyedException(
"Region not found",
this.regionPath));
this.internalRegion = null;
return false;
}
this.needsUnlock = this.internalRegion.lockGII();
this.internalRegion.txLRUStart();
this.needsLRUEnd = true;
if (this.internalRegion.isInitialized()) {
// We don't want the txEvent to know anything about our regions
// that are still doing gii.
this.txEvent = txEvent;
}
} catch (RegionDestroyedException e) {
this.msg.addProcessingException(e);
// Region destroyed: Update cancelled
if (logger.isDebugEnabled()) {
logger.debug(
"Received unneeded commit data for region {} because the region was destroyed.",
this.regionPath, e);
}
this.internalRegion = null;
}
return this.internalRegion != null;
}
private boolean hookupRegion(DistributionManager dm) {
this.internalRegion = getRegionByPath(dm, regionPath);
if (this.internalRegion == null && this.parentRegionPath != null) {
this.internalRegion = getRegionByPath(dm, this.parentRegionPath);
this.regionPath = this.parentRegionPath;
}
if (this.internalRegion == null && dm.getSystem().isLoner()) {
// If there are additional regions that the server enlisted in the tx,
// which the client does not have, the client can just ignore the region
// see bug 51922
return false;
}
return true;
}
LocalRegion getRegionByPath(DistributionManager dm, String regionPath) {
InternalCache cache = dm.getCache();
return cache == null ? null : (LocalRegion) cache.getRegionByPath(regionPath);
}
/**
* Called when processing is complete; only needs to be called if beginProcess returned true.
*/
void endProcess() {
if (this.internalRegion != null) {
try {
if (this.needsLRUEnd) {
this.needsLRUEnd = false;
this.internalRegion.txLRUEnd();
}
} finally {
if (this.needsUnlock) {
this.needsUnlock = false;
this.internalRegion.unlockGII();
}
}
}
}
/**
* Returns the eventId to use for the give farside entry op.
*
* @since GemFire 5.7
*/
private EventID getEventId(FarSideEntryOp entryOp) {
return this.msg.getEventId(entryOp.eventOffset);
}
/**
* Apply a single tx entry op on the far side
*/
@SuppressWarnings("synthetic-access")
protected void txApplyEntryOp(FarSideEntryOp entryOp, List<EntryEventImpl> pendingCallbacks) {
if (this.internalRegion == null) {
return;
}
EventID eventID = getEventId(entryOp);
boolean isDuplicate = this.internalRegion.hasSeenEvent(eventID);
boolean callbacksOnly =
(this.internalRegion.getDataPolicy() == DataPolicy.PARTITION) || isDuplicate;
if (this.internalRegion instanceof PartitionedRegion) {
/*
* This happens when we don't have the bucket and are getting adjunct notification
*/
// No need to release because it is added to pendingCallbacks and they will be released
// later
EntryEventImpl eei =
txCallbackEventFactory.createCallbackEvent(this.internalRegion, entryOp.op,
entryOp.key,
entryOp.value, this.msg.txIdent, txEvent, getEventId(entryOp), entryOp.callbackArg,
entryOp.filterRoutingInfo, this.msg.bridgeContext, null, entryOp.versionTag,
entryOp.tailKey);
if (entryOp.filterRoutingInfo != null) {
eei.setLocalFilterInfo(
entryOp.filterRoutingInfo.getFilterInfo(this.internalRegion.getCache().getMyId()));
}
if (isDuplicate) {
eei.setPossibleDuplicate(true);
}
if (logger.isDebugEnabled()) {
logger.debug("invoking transactional callbacks for {} key={} needsUnlock={} event={}",
entryOp.op, entryOp.key, this.needsUnlock, eei);
}
// we reach this spot because the event is either delivered to this member
// as an "adjunct" message or because the bucket was being created when
// the message was sent and already reflects the change caused by this event.
// In the latter case we need to invoke listeners
final boolean skipListeners = !isDuplicate;
eei.setInvokePRCallbacks(!skipListeners);
pendingCallbacks.add(eei);
return;
}
if (logger.isDebugEnabled()) {
logger.debug("applying transactional {} key={} needsUnlock={} eventId {} with routing {}",
entryOp.op, entryOp.key, this.needsUnlock, getEventId(entryOp),
entryOp.filterRoutingInfo);
}
if (entryOp.versionTag != null) {
entryOp.versionTag.replaceNullIDs(this.msg.getSender());
}
if (entryOp.op.isDestroy()) {
this.internalRegion.txApplyDestroy(entryOp.key, this.msg.txIdent, this.txEvent,
this.needsUnlock,
entryOp.op, getEventId(entryOp), entryOp.callbackArg, pendingCallbacks,
entryOp.filterRoutingInfo, this.msg.bridgeContext, false /* origin remote */,
null/* txEntryState */, entryOp.versionTag, entryOp.tailKey);
} else if (entryOp.op.isInvalidate()) {
this.internalRegion.txApplyInvalidate(entryOp.key, Token.INVALID, entryOp.didDestroy,
this.msg.txIdent,
this.txEvent, false /* localOp */, getEventId(entryOp), entryOp.callbackArg,
pendingCallbacks, entryOp.filterRoutingInfo, this.msg.bridgeContext,
null/* txEntryState */, entryOp.versionTag, entryOp.tailKey);
} else {
this.internalRegion.txApplyPut(entryOp.op, entryOp.key, entryOp.value, entryOp.didDestroy,
this.msg.txIdent, this.txEvent, getEventId(entryOp), entryOp.callbackArg,
pendingCallbacks, entryOp.filterRoutingInfo, this.msg.bridgeContext,
null/* txEntryState */, entryOp.versionTag, entryOp.tailKey);
}
}
/**
* Apply a single tx entry op on the far side
*/
@SuppressWarnings("synthetic-access")
protected void txApplyEntryOpAdjunctOnly(FarSideEntryOp entryOp) {
if (this.internalRegion == null) {
return;
}
EventID eventID = getEventId(entryOp);
boolean isDuplicate = this.internalRegion.hasSeenEvent(eventID);
boolean callbacksOnly =
(this.internalRegion.getDataPolicy() == DataPolicy.PARTITION) || isDuplicate;
if (this.internalRegion instanceof PartitionedRegion) {
PartitionedRegion pr = (PartitionedRegion) internalRegion;
BucketRegion br = pr.getBucketRegion(entryOp.key);
Set bucketOwners = br.getBucketOwners();
InternalDistributedMember thisMember = this.internalRegion.getDistributionManager().getId();
if (bucketOwners.contains(thisMember)) {
return;
}
/*
* This happens when we don't have the bucket and are getting adjunct notification
*/
@Released
EntryEventImpl eei =
txCallbackEventFactory.createCallbackEvent(this.internalRegion, entryOp.op,
entryOp.key,
entryOp.value, this.msg.txIdent, txEvent, getEventId(entryOp), entryOp.callbackArg,
entryOp.filterRoutingInfo, this.msg.bridgeContext, null, entryOp.versionTag,
entryOp.tailKey);
try {
if (entryOp.filterRoutingInfo != null) {
eei.setLocalFilterInfo(
entryOp.filterRoutingInfo.getFilterInfo(this.internalRegion.getCache().getMyId()));
}
if (isDuplicate) {
eei.setPossibleDuplicate(true);
}
if (logger.isDebugEnabled()) {
logger.debug("invoking transactional callbacks for {} key={} needsUnlock={} event={}",
entryOp.op, entryOp.key, this.needsUnlock, eei);
}
// we reach this spot because the event is either delivered to this member
// as an "adjunct" message or because the bucket was being created when
// the message was sent and already reflects the change caused by this event.
// In the latter case we need to invoke listeners
final boolean skipListeners = !isDuplicate;
eei.invokeCallbacks(this.internalRegion, skipListeners, true);
} finally {
eei.release();
}
return;
}
}
boolean isEmpty() {
return this.opKeys == null;
}
boolean needsAck() {
return this.internalRegion.getScope().isDistributedAck();
}
void addOp(Object key, TXEntryState entry) {
if (this.opKeys == null) {
this.opKeys = new ArrayList(this.maxSize);
this.opEntries = new ArrayList(this.maxSize);
}
this.opKeys.add(key);
this.opEntries.add(entry);
}
public boolean isForceFireEvent(DistributionManager dm) {
LocalRegion r = getRegionByPath(dm, regionPath);
if (r instanceof PartitionedRegion || (r != null && r.isUsedForPartitionedRegionBucket())) {
return false;
}
return true;
}
public void fromData(DataInput in, boolean hasShadowKey)
throws IOException, ClassNotFoundException {
this.regionPath = DataSerializer.readString(in);
this.parentRegionPath = DataSerializer.readString(in);
int size = in.readInt();
if (size > 0) {
this.opKeys = new ArrayList(size);
this.opEntries = new ArrayList(size);
final boolean largeModCount = in.readBoolean();
this.memberId = DataSerializer.readObject(in);
for (int i = 0; i < size; i++) {
FarSideEntryOp entryOp = new FarSideEntryOp();
// shadowkey is not being sent to clients
entryOp.fromData(in, largeModCount, hasShadowKey);
if (entryOp.versionTag != null && this.memberId != null) {
entryOp.versionTag.setMemberID(this.memberId);
}
this.msg.addFarSideEntryOp(entryOp);
this.opKeys.add(entryOp.key);
this.opEntries.add(entryOp);
}
}
}
@Override
public String toString() {
StringBuilder result = new StringBuilder(64);
if (this.regionPath != null) {
result.append(this.regionPath);
} else {
result.append(this.internalRegion.getFullPath());
}
if (this.refCount > 0) {
result.append(" refCount=").append(this.refCount);
}
return result.toString();
}
private void basicToData(DataOutput out,
SerializationContext context,
boolean useShadowKey) throws IOException {
if (this.internalRegion != null) {
DataSerializer.writeString(this.internalRegion.getFullPath(), out);
if (this.internalRegion instanceof BucketRegion) {
DataSerializer.writeString(
((Bucket) this.internalRegion).getPartitionedRegion().getFullPath(), out);
} else {
DataSerializer.writeString(null, out);
}
} else {
DataSerializer.writeString(this.regionPath, out);
DataSerializer.writeString(this.parentRegionPath, out);
}
if (isEmpty() || this.opKeys.size() == 0) {
out.writeInt(0);
} else {
int size = this.opKeys.size();
out.writeInt(size);
final boolean largeModCount;
if (this.msg.txState != null) {
largeModCount = this.msg.txState.needsLargeModCount();
} else {
largeModCount = this.msg.needsLargeModCount;
}
out.writeBoolean(largeModCount);
final boolean sendVersionTags =
this.msg.clientVersion == null || Version.GFE_70.compareTo(this.msg.clientVersion) <= 0;
if (sendVersionTags) {
VersionSource member = this.memberId;
if (member == null) {
if (this.internalRegion == null) {
Assert.assertTrue(this.msg.txState == null);
} else {
member = this.internalRegion.getVersionMember();
}
}
DataSerializer.writeObject(member, out);
}
for (int i = 0; i < size; i++) {
DataSerializer.writeObject(this.opKeys.get(i), out);
if (this.msg.txState != null) {
/* we are still on tx node and have the entry state */
((TXEntryState) this.opEntries.get(i)).toFarSideData(out, context, largeModCount,
sendVersionTags, useShadowKey);
} else {
((FarSideEntryOp) this.opEntries.get(i)).toData(out, largeModCount, sendVersionTags,
useShadowKey);
}
}
}
}
public void toData(DataOutput out, SerializationContext context, boolean useShadowKey)
throws IOException {
if (this.preserializedBuffer != null) {
this.preserializedBuffer.rewind();
this.preserializedBuffer.sendTo(out);
} else if (this.refCount > 1) {
Version v = InternalDataSerializer.getVersionForDataStream(out);
HeapDataOutputStream hdos = new HeapDataOutputStream(1024, v);
basicToData(hdos, context, useShadowKey);
this.preserializedBuffer = hdos;
this.preserializedBuffer.sendTo(out);
} else {
basicToData(out, context, useShadowKey);
}
}
/**
* Holds data that describes a tx entry op on the far side.
*
* @since GemFire 5.0
*/
public class FarSideEntryOp implements Comparable {
public Operation op;
public int modSerialNum;
public int eventOffset;
public Object key;
public Object value;
public boolean didDestroy;
public Object callbackArg;
private FilterRoutingInfo filterRoutingInfo;
private VersionTag versionTag;
private long tailKey;
/**
* Create a new representation of a tx entry op on the far side. All init will be done by a
* call to fromData
*/
public FarSideEntryOp() {}
/**
* Creates and returns a new instance of a tx entry op on the far side. The "toData" that this
* should match is {@link TXEntryState#toFarSideData}.
*
* @param in the data input that is used to read the data for this entry op
* @param largeModCount true if the mod count is a int instead of a byte.
* @param readShadowKey true if a long shadowKey should be read
*/
public void fromData(DataInput in, boolean largeModCount, boolean readShadowKey)
throws IOException, ClassNotFoundException {
this.key = DataSerializer.readObject(in);
this.op = Operation.fromOrdinal(in.readByte());
if (largeModCount) {
this.modSerialNum = in.readInt();
} else {
this.modSerialNum = in.readByte();
}
this.callbackArg = DataSerializer.readObject(in);
this.filterRoutingInfo = DataSerializer.readObject(in);
this.versionTag = DataSerializer.readObject(in);
if (readShadowKey) {
this.tailKey = in.readLong();
}
this.eventOffset = in.readInt();
if (!this.op.isDestroy()) {
this.didDestroy = in.readBoolean();
if (!this.op.isInvalidate()) {
boolean isTokenOrByteArray = in.readBoolean();
if (isTokenOrByteArray) {
// token or byte[]
this.value = DataSerializer.readObject(in);
} else {
// CachedDeserializable, Object, or PDX
this.value = CachedDeserializableFactory.create(DataSerializer.readByteArray(in),
GemFireCacheImpl.getInstance());
}
}
}
}
public void toData(DataOutput out, boolean largeModCount, boolean sendVersionTag,
boolean sendShadowKey) throws IOException {
// DataSerializer.writeObject(this.key,out);
/* Don't serialize key because caller did that already */
out.writeByte(this.op.ordinal);
if (largeModCount) {
out.writeInt(this.modSerialNum);
} else {
out.writeByte(this.modSerialNum);
}
DataSerializer.writeObject(this.callbackArg, out);
DataSerializer.writeObject(this.filterRoutingInfo, out);
if (sendVersionTag) {
DataSerializer.writeObject(this.versionTag, out);
}
if (sendShadowKey) {
out.writeLong(this.tailKey);
}
out.writeInt(this.eventOffset);
if (!this.op.isDestroy()) {
out.writeBoolean(this.didDestroy);
if (!this.op.isInvalidate()) {
boolean sendObject = Token.isInvalidOrRemoved(this.value);
sendObject = sendObject || this.value instanceof byte[];
out.writeBoolean(sendObject);
if (sendObject) {
DataSerializer.writeObject(this.value, out);
} else {
DataSerializer.writeObjectAsByteArray(this.value, out);
}
}
}
}
/**
* Performs this entryOp on the farside of a tx commit.
*/
public void process(List<EntryEventImpl> pendingCallbacks) {
txApplyEntryOp(this, pendingCallbacks);
}
public void processAdjunctOnly() {
txApplyEntryOpAdjunctOnly(this);
}
public RegionCommit getRegionCommit() {
return RegionCommit.this;
}
/**
* Returns the value to use to sort us
*/
private int getSortValue() {
return this.modSerialNum;
}
@Override
public int compareTo(Object o) {
FarSideEntryOp other = (FarSideEntryOp) o;
return getSortValue() - other.getSortValue();
}
@Override
public boolean equals(Object o) {
if (o == null || !(o instanceof FarSideEntryOp)) {
return false;
}
return compareTo(o) == 0;
}
@Override
public int hashCode() {
return getSortValue();
}
}
}
Object getTrackerKey() {
if (this.lockId != null) {
return this.lockId;
} else {
return this.txIdent;
}
}
/**
* Used to prevent processing of the message if we have reported to other FarSiders that we did
* not received the CommitProcessMessage
*/
boolean dontProcess() {
return this.dontProcess;
}
/**
* Indicate that this message should not be processed if we receive CommitProcessMessage (late)
*/
void setDontProcess() {
this.dontProcess = true;
}
boolean isProcessing() {
return this.isProcessing;
}
private void setIsProcessing(boolean isProcessing) {
this.isProcessing = isProcessing;
}
boolean wasProcessed() {
return this.wasProcessed;
}
void setProcessed(boolean wasProcessed) {
this.wasProcessed = wasProcessed;
}
/**
* The CommitProcessForLockIDMessaage is sent by the Distributed ACK TX origin to the recipients
* (aka FarSiders) to indicate that a previously received RegionCommit that contained a lockId
* should commence processing.
*/
public static class CommitProcessForLockIdMessage extends CommitProcessMessage {
private TXLockId lockId;
public CommitProcessForLockIdMessage() {
// Zero arg constructor for DataSerializer
}
public CommitProcessForLockIdMessage(TXLockId lockId) {
this.lockId = lockId;
Assert.assertTrue(this.lockId != null,
"CommitProcessForLockIdMessage must have a non-null lockid!");
}
@Override
protected void process(ClusterDistributionManager dm) {
final TXCommitMessage mess = waitForMessage(this.lockId, dm);
Assert.assertTrue(mess != null, "Commit data for TXLockId: " + this.lockId + " not found");
basicProcess(mess, dm);
}
@Override
public int getDSFID() {
return COMMIT_PROCESS_FOR_LOCKID_MESSAGE;
}
@Override
public void toData(DataOutput out,
SerializationContext context) throws IOException {
InternalDataSerializer.invokeToData(this.lockId, out);
}
@Override
public void fromData(DataInput in,
DeserializationContext context) throws IOException, ClassNotFoundException {
this.lockId = TXLockIdImpl.createFromData(in);
Assert.assertTrue(this.lockId != null,
"CommitProcessForLockIdMessage must have a non-null lockid!");
}
@Override
public String toString() {
StringBuilder result = new StringBuilder(128);
result.append("CommitProcessForLockIdMessage@").append(System.identityHashCode(this))
.append(" lockId=").append(this.lockId);
return result.toString();
}
}
/**
* The CommitProcessForTXIdMessaage is sent by the Distributed ACK TX origin to the recipients
* (aka FarSiders) to indicate that a previously received RegionCommit that contained a TXId
* should commence processing. RegionCommit messages that contain a TXId (and no TXLockId) are
* typically sent if all the TX changes are a result of load/netsearch/netload values (thus no
* lockid)
*/
public static class CommitProcessForTXIdMessage extends CommitProcessMessage {
private TXId txId;
public CommitProcessForTXIdMessage() {
// Zero arg constructor for DataSerializer
}
public CommitProcessForTXIdMessage(TXId txId) {
this.txId = txId;
Assert.assertTrue(this.txId != null,
"CommitProcessMessageForTXId must have a non-null txid!");
}
@Override
protected void process(ClusterDistributionManager dm) {
final TXCommitMessage mess = waitForMessage(this.txId, dm);
Assert.assertTrue(mess != null, "Commit data for TXId: " + this.txId + " not found");
basicProcess(mess, dm);
}
@Override
public int getDSFID() {
return COMMIT_PROCESS_FOR_TXID_MESSAGE;
}
@Override
public void toData(DataOutput out,
SerializationContext context) throws IOException {
InternalDataSerializer.invokeToData(this.txId, out);
}
@Override
public void fromData(DataInput in,
DeserializationContext context) throws IOException, ClassNotFoundException {
this.txId = TXId.createFromData(in);
Assert.assertTrue(this.txId != null,
"CommitProcessMessageForTXId must have a non-null txid!");
}
@Override
public String toString() {
StringBuilder result = new StringBuilder(128);
result.append("CommitProcessForTXIdMessage@").append(System.identityHashCode(this))
.append(" txId=").append(this.txId);
return result.toString();
}
}
public abstract static class CommitProcessMessage extends PooledDistributionMessage {
protected void basicProcess(final TXCommitMessage mess, final ClusterDistributionManager dm) {
dm.removeMembershipListener(mess);
synchronized (mess) {
if (mess.dontProcess()) {
return;
}
}
try {
mess.basicProcess();
} finally {
txTracker.processed(mess);
}
}
}
/**
* The CommitProcessQueryMessage is used to attempt to recover - in the Distributed ACK TXs - when
* the origin of the CommitProcess messages departed from the distributed system. The sender of
* this message is attempting to query other potential fellow FarSiders (aka recipients) who may
* have received the CommitProcess message.
*
* Since the occurance of this message will be rare (hopefully), it was decided to be general
* about the the tracker key - opting not to have specific messages for each type like
* CommitProcessFor<Lock/TX>Id - and take the performance penalty of an extra call to
* DataSerializer
*/
public static class CommitProcessQueryMessage extends PooledDistributionMessage {
private Object trackerKey; // Either a TXLockId or a TXId
private int processorId;
public CommitProcessQueryMessage() {
// Zero arg constructor for DataSerializer
}
public CommitProcessQueryMessage(Object trackerKey, int processorId) {
this.trackerKey = trackerKey;
this.processorId = processorId;
}
@Override
protected void process(ClusterDistributionManager dm) {
final boolean processMsgReceived = txTracker.commitProcessReceived(this.trackerKey);
if (!processMsgReceived) {
if (logger.isDebugEnabled()) {
logger.debug("CommitProcessQuery did not find {} in the history", this.trackerKey);
}
}
// Reply to the fellow FarSider as to whether the
// CommitProcess message was received
CommitProcessQueryReplyMessage resp = new CommitProcessQueryReplyMessage(processMsgReceived);
resp.setProcessorId(this.processorId);
resp.setRecipient(this.getSender());
dm.putOutgoing(resp);
}
@Override
public void toData(DataOutput out,
SerializationContext context) throws IOException {
DataSerializer.writeObject(this.trackerKey, out);
out.writeInt(this.processorId);
}
@Override
public int getDSFID() {
return COMMIT_PROCESS_QUERY_MESSAGE;
}
@Override
public void fromData(DataInput in,
DeserializationContext context) throws IOException, ClassNotFoundException {
this.trackerKey = DataSerializer.readObject(in);
this.processorId = in.readInt();
}
@Override
public String toString() {
StringBuilder result = new StringBuilder(128);
result.append("CommitProcessQueryMessage@").append(System.identityHashCode(this))
.append(" trackerKeyClass=").append(this.trackerKey.getClass().getName())
.append(" trackerKey=").append(this.trackerKey).append(" processorId=")
.append(this.processorId);
return result.toString();
}
}
/********************* Commit Process Query Response Message **********************************/
public static class CommitProcessQueryReplyMessage extends ReplyMessage {
private boolean wasReceived;
public CommitProcessQueryReplyMessage(boolean wasReceived) {
this.wasReceived = wasReceived;
}
public CommitProcessQueryReplyMessage() {
// zero arg constructor for DataSerializer
}
public boolean wasReceived() {
return wasReceived;
}
@Override
public int getDSFID() {
return COMMIT_PROCESS_QUERY_REPLY_MESSAGE;
}
@Override
public void fromData(DataInput in,
DeserializationContext context) throws IOException, ClassNotFoundException {
super.fromData(in, context);
this.wasReceived = in.readBoolean();
}
@Override
public void toData(DataOutput out,
SerializationContext context) throws IOException {
super.toData(out, context);
out.writeBoolean(this.wasReceived);
}
@Override
public String toString() {
StringBuilder result = new StringBuilder(128);
result.append("CommitProcessQueryReplyMessage@").append(System.identityHashCode(this))
.append(" wasReceived=").append(this.wasReceived).append(" processorId=")
.append(this.processorId).append(" from ").append(this.getSender());
return result.toString();
}
}
/********************* Commit Process Query Response Processor *********************************/
public static class CommitProcessQueryReplyProcessor extends ReplyProcessor21 {
public boolean receivedOnePositive;
CommitProcessQueryReplyProcessor(DistributionManager dm, Set members) {
super(dm, members);
this.receivedOnePositive = false;
}
@Override
public void process(DistributionMessage msg) {
CommitProcessQueryReplyMessage ccMess = (CommitProcessQueryReplyMessage) msg;
if (ccMess.wasReceived()) {
this.receivedOnePositive = true;
}
super.process(msg);
}
@Override
protected boolean canStopWaiting() {
return this.receivedOnePositive;
}
public boolean receivedACommitProcessMessage() {
return this.receivedOnePositive;
}
}
/********************* MembershipListener Implementation ***************************************/
@Override
public void memberJoined(DistributionManager distributionManager, InternalDistributedMember id) {
// do nothing
}
@Override
public void memberSuspect(DistributionManager distributionManager, InternalDistributedMember id,
InternalDistributedMember whoSuspected, String reason) {}
@Override
public void quorumLost(DistributionManager distributionManager,
Set<InternalDistributedMember> failures,
List<InternalDistributedMember> remaining) {}
/**
* return true if the member initiating this transaction has left the cluster
*/
public boolean isDepartureNoticed() {
return departureNoticed;
}
private void doOriginDepartedCommit() {
try {
// Set processor to zero to avoid the ack to the now departed origin
processorId = 0;
basicProcess();
} finally {
txTracker.processed(this);
}
}
@Override
public void memberDeparted(DistributionManager distributionManager,
final InternalDistributedMember id, boolean crashed) {
if (!getSender().equals(id)) {
return;
}
distributionManager.removeMembershipListener(this);
synchronized (this) {
if (isProcessing() || this.departureNoticed) {
if (logger.isDebugEnabled()) {
logger.debug("Member departed: Commit data is already being processed for lockid: {}",
lockId);
}
return;
}
this.departureNoticed = true;
}
// Send message to fellow FarSiders (aka recipients), if any, to
// determine if any one of them have received a CommitProcessMessage
if (getFarSiders() != null && !getFarSiders().isEmpty()) {
if (logger.isDebugEnabled()) {
logger.debug(
"Member departed: {} sending query for CommitProcess message to other recipients.", id);
}
// Create a new thread, send the CommitProcessQuery, wait for a response and potentially
// process
// Should I use a thread pool?, Darrel suggests look in DM somewhere or introduce a zero
// sized thread pool
Thread fellowFarSidersQuery = new LoggingThread("CommitProcessQuery Thread",
() -> doCommitProcessQuery(id));
fellowFarSidersQuery.start();
} else {
if (logger.isDebugEnabled()) {
logger.debug("Member departed: {}. Processing commit data.", getSender());
}
// Optimimal case where we are the only FarSider, assume we
// will never get the CommitProcess message, but it
// doesn't matter since we can commit anyway.
// Start a new thread to process the commit
Thread originDepartedCommit = new LoggingThread("Origin Departed Commit",
this::doOriginDepartedCommit);
originDepartedCommit.start();
}
}
HashSet getFarSiders() {
return farSiders;
}
DistributionManager getDistributionManager() {
return dm;
}
void doCommitProcessQuery(final InternalDistributedMember id) {
CommitProcessQueryReplyProcessor replyProcessor = createReplyProcessor();
CommitProcessQueryMessage queryMessage = createQueryMessage(replyProcessor);
queryMessage.setRecipients(this.farSiders);
getDistributionManager().putOutgoing(queryMessage);
// Wait for any one positive response or all negative responses.
// (while() loop removed for bug 36983 - you can't loop on waitForReplies()
getDistributionManager().getCancelCriterion().checkCancelInProgress(null);
try {
replyProcessor.waitForRepliesUninterruptibly();
} catch (ReplyException e) {
e.handleCause();
}
if (replyProcessor.receivedACommitProcessMessage()) {
if (logger.isDebugEnabled()) {
logger.debug(
"Transaction associated with lockID: {} from orign {} is processing due to a received \"commit process\" message",
lockId, id);
}
try {
// Set processor to zero to avoid the ack to the now departed origin
processorId = 0;
basicProcess();
} finally {
txTracker.processed(this);
}
} else {
if (logger.isDebugEnabled()) {
logger.debug(
"Transaction associated with lockID: {} from origin {} ignored. No other recipients received \"commit process\" message",
lockId, id);
}
txTracker.removeMessage(this);
}
}
CommitProcessQueryReplyProcessor createReplyProcessor() {
return new CommitProcessQueryReplyProcessor(dm, farSiders);
}
CommitProcessQueryMessage createQueryMessage(CommitProcessQueryReplyProcessor replyProcessor) {
return new CommitProcessQueryMessage(getTrackerKey(), replyProcessor.getProcessorId());
}
private DistributedMember getMemberFromTrackerKey(Object trackerKey) {
if (trackerKey instanceof TXId) {
TXId id1 = (TXId) trackerKey;
return id1.getMemberId();
} else if (trackerKey instanceof TXLockId) {
TXLockId id2 = (TXLockId) trackerKey;
return id2.getMemberId();
}
return null;
}
void setUpdateLockMembers() {
this.lockNeedsUpdate = true;
}
/**
* Intended to be called after TXState.applyChanges when the potential for a different set of TX
* members has been determined and it is safe to ignore any new members because the changes have
* been applied to committed state. This was added as the solution to bug 32999 and the recovery
* when the TXLock Lessor (the sending VM) crashes/departs before or while sending the
* TXCommitMessage.
*
* @see TXState#commit()
* @see org.apache.geode.internal.cache.locks.TXLockBatch#getBatchId()
*/
private void updateLockMembers() {
if (this.lockNeedsUpdate && this.lockId != null) {
TXLockService.createDTLS(this.dm.getSystem()).updateParticipants(this.lockId,
this.msgMap.keySet());
}
}
/**
* Reply processor which collects all CommitReplyExceptions and emits a detailed failure exception
* if problems occur
*
* @since GemFire 5.7
*/
private class CommitReplyProcessor extends ReliableReplyProcessor21 {
private HashMap msgMap;
public CommitReplyProcessor(DistributionManager dm, Set initMembers, HashMap msgMap) {
super(dm, initMembers);
this.msgMap = msgMap;
}
public void waitForCommitCompletion() {
try {
waitForRepliesUninterruptibly();
} catch (CommitExceptionCollectingException e) {
e.handlePotentialCommitFailure(msgMap);
}
}
@Override
protected void processException(DistributionMessage msg, ReplyException ex) {
if (msg instanceof ReplyMessage) {
synchronized (this) {
if (this.exception == null) {
// Exception Container
this.exception = new CommitExceptionCollectingException(txIdent);
}
CommitExceptionCollectingException cce =
(CommitExceptionCollectingException) this.exception;
if (ex instanceof CommitReplyException) {
CommitReplyException cre = (CommitReplyException) ex;
cce.addExceptionsFromMember(msg.getSender(), cre.getExceptions());
} else {
cce.addExceptionsFromMember(msg.getSender(), Collections.singleton(ex));
}
}
}
}
@Override
protected boolean stopBecauseOfExceptions() {
return false;
}
public Set getCacheClosedMembers() {
if (this.exception != null) {
CommitExceptionCollectingException cce =
(CommitExceptionCollectingException) this.exception;
return cce.getCacheClosedMembers();
} else {
return Collections.emptySet();
}
}
public Set getRegionDestroyedMembers(String regionFullPath) {
if (this.exception != null) {
CommitExceptionCollectingException cce =
(CommitExceptionCollectingException) this.exception;
return cce.getRegionDestroyedMembers(regionFullPath);
} else {
return Collections.emptySet();
}
}
}
/**
* An Exception that collects many remote CommitExceptions
*
* @since GemFire 5.7
*/
public static class CommitExceptionCollectingException extends ReplyException {
private static final long serialVersionUID = 589384721273797822L;
/**
* Set of members that threw CacheClosedExceptions
*/
private final Set<InternalDistributedMember> cacheExceptions;
/**
* key=region path, value=Set of members
*/
private final Map<String, Set<InternalDistributedMember>> regionExceptions;
/**
* List of exceptions that were unexpected and caused the tx to fail
*/
private final Map fatalExceptions;
private final TXId id;
public CommitExceptionCollectingException(TXId txIdent) {
this.cacheExceptions = new HashSet();
this.regionExceptions = new HashMap();
this.fatalExceptions = new HashMap();
this.id = txIdent;
}
/**
* Determine if the commit processing was incomplete, if so throw a detailed exception
* indicating the source of the problem
*/
public void handlePotentialCommitFailure(
HashMap<InternalDistributedMember, RegionCommitList> msgMap) {
if (fatalExceptions.size() > 0) {
StringBuilder errorMessage = new StringBuilder("Incomplete commit of transaction ")
.append(id).append(". Caused by the following exceptions: ");
for (Iterator i = fatalExceptions.entrySet().iterator(); i.hasNext();) {
Map.Entry me = (Map.Entry) i.next();
DistributedMember mem = (DistributedMember) me.getKey();
errorMessage.append(" From member: ").append(mem).append(" ");
List exceptions = (List) me.getValue();
for (Iterator ei = exceptions.iterator(); ei.hasNext();) {
Exception e = (Exception) ei.next();
errorMessage.append(e);
for (StackTraceElement ste : e.getStackTrace()) {
errorMessage.append("\n\tat ").append(ste);
}
if (ei.hasNext()) {
errorMessage.append("\nAND\n");
}
}
errorMessage.append(".");
}
throw new CommitIncompleteException(errorMessage.toString());
}
// Mark any persistent members as offline
handleClosedMembers(msgMap);
handleRegionDestroyed(msgMap);
}
/**
* Mark peers as offline for regions that the peer returned a RegionDestroyedException
*/
private void handleRegionDestroyed(
HashMap<InternalDistributedMember, RegionCommitList> msgMap) {
if (regionExceptions == null || regionExceptions.isEmpty()) {
return;
}
for (Map.Entry<InternalDistributedMember, RegionCommitList> memberMap : msgMap.entrySet()) {
InternalDistributedMember member = memberMap.getKey();
RegionCommitList rcl = memberMap.getValue();
for (RegionCommit region : rcl) {
Set<InternalDistributedMember> failedMembers =
regionExceptions.get(region.internalRegion.getFullPath());
if (failedMembers != null && failedMembers.contains(member)) {
markMemberOffline(member, region);
}
}
}
}
/**
* Mark peers as offline that returned a cache closed exception
*/
private void handleClosedMembers(HashMap<InternalDistributedMember, RegionCommitList> msgMap) {
for (InternalDistributedMember member : getCacheClosedMembers()) {
RegionCommitList rcl = msgMap.get(member);
for (RegionCommit region : rcl) {
markMemberOffline(member, region);
}
}
}
private void markMemberOffline(InternalDistributedMember member, RegionCommit region) {
if (region.persistentIds == null) {
return;
}
PersistentMemberID persistentId = region.persistentIds.get(member);
/// iterate over the list and mark the members offline
if (persistentId != null) {
// Fix for bug 42142 - In order for recovery to work,
// we must either
// 1) persistent the region operation successfully on the peer
// 2) record that the peer is offline
// or
// 3) fail the operation
// if we have started to shutdown, we don't want to mark the peer
// as offline, or we will think we have newer data when in fact we don't
region.internalRegion.getCancelCriterion().checkCancelInProgress(null);
// Otherwise, mark the peer as offline, because it didn't complete
// the operation.
((DistributedRegion) region.internalRegion).getPersistenceAdvisor().markMemberOffline(
member,
persistentId);
}
}
public Set<InternalDistributedMember> getCacheClosedMembers() {
return this.cacheExceptions;
}
public Set getRegionDestroyedMembers(String regionFullPath) {
Set members = (Set) this.regionExceptions.get(regionFullPath);
if (members == null) {
members = Collections.emptySet();
}
return members;
}
/**
* Protected by (this)
*/
public void addExceptionsFromMember(InternalDistributedMember member, Set exceptions) {
for (Iterator iter = exceptions.iterator(); iter.hasNext();) {
Exception ex = (Exception) iter.next();
if (ex instanceof CancelException) {
cacheExceptions.add(member);
} else if (ex instanceof RegionDestroyedException) {
String r = ((RegionDestroyedException) ex).getRegionFullPath();
Set<InternalDistributedMember> members = regionExceptions.get(r);
if (members == null) {
members = new HashSet();
regionExceptions.put(r, members);
}
members.add(member);
} else {
List el = (List) this.fatalExceptions.get(member);
if (el == null) {
el = new ArrayList(2);
this.fatalExceptions.put(member, el);
}
el.add(ex);
}
}
}
}
public void hookupRegions(DistributionManager dm) {
if (regions != null) {
Iterator it = regions.iterator();
while (it.hasNext()) {
RegionCommit rc = (RegionCommit) it.next();
rc.hookupRegion(dm);
}
}
}
/**
* Disable firing of TX Listeners. Currently on used on clients.
*
* @param b disable the listeners
*/
public void setDisableListeners(boolean b) {
disableListeners = true;
}
public Version getClientVersion() {
return clientVersion;
}
public void setClientVersion(Version clientVersion) {
this.clientVersion = clientVersion;
}
}