| /* |
| * Licensed to the Apache Software Foundation (ASF) under one or more contributor license |
| * agreements. See the NOTICE file distributed with this work for additional information regarding |
| * copyright ownership. The ASF licenses this file to You under the Apache License, Version 2.0 (the |
| * "License"); you may not use this file except in compliance with the License. You may obtain a |
| * copy of the License at |
| * |
| * http://www.apache.org/licenses/LICENSE-2.0 |
| * |
| * Unless required by applicable law or agreed to in writing, software distributed under the License |
| * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express |
| * or implied. See the License for the specific language governing permissions and limitations under |
| * the License. |
| */ |
| package org.apache.geode.internal.cache; |
| |
| import java.util.ArrayList; |
| import java.util.HashMap; |
| import java.util.HashSet; |
| import java.util.Map; |
| import java.util.Map.Entry; |
| import java.util.Set; |
| import java.util.TreeMap; |
| import java.util.TreeSet; |
| |
| import org.apache.geode.cache.CommitConflictException; |
| import org.apache.geode.cache.Operation; |
| import org.apache.geode.cache.TransactionInDoubtException; |
| import org.apache.geode.cache.UnsupportedOperationInTransactionException; |
| import org.apache.geode.distributed.DistributedMember; |
| import org.apache.geode.distributed.internal.DistributionManager; |
| import org.apache.geode.distributed.internal.membership.InternalDistributedMember; |
| import org.apache.geode.internal.cache.DistTXPrecommitMessage.DistTxPrecommitResponse; |
| import org.apache.geode.internal.cache.TXEntryState.DistTxThinEntryState; |
| import org.apache.geode.internal.cache.tier.sockets.VersionedObjectList; |
| import org.apache.geode.internal.cache.tx.DistClientTXStateStub; |
| import org.apache.geode.internal.cache.tx.DistTxEntryEvent; |
| import org.apache.geode.internal.statistics.StatisticsClock; |
| |
| public class DistTXStateProxyImplOnCoordinator extends DistTXStateProxyImpl { |
| |
| /** |
| * A map of distributed system member to either {@link DistPeerTXStateStub} or |
| * {@link DistTXStateOnCoordinator} (in case of TX coordinator is also a data node) |
| */ |
| private final HashMap<DistributedMember, DistTXCoordinatorInterface> target2realDeals = |
| new HashMap<>(); |
| |
| private HashMap<InternalRegion, DistributedMember> rrTargets; |
| |
| private Set<DistributedMember> txRemoteParticpants = null; // other than local |
| |
| private HashMap<String, ArrayList<DistTxThinEntryState>> txEntryEventMap = null; |
| |
| public DistTXStateProxyImplOnCoordinator(InternalCache cache, TXManagerImpl managerImpl, TXId id, |
| InternalDistributedMember clientMember, StatisticsClock statisticsClock) { |
| super(cache, managerImpl, id, clientMember, statisticsClock); |
| } |
| |
| public DistTXStateProxyImplOnCoordinator(InternalCache cache, TXManagerImpl managerImpl, TXId id, |
| boolean isjta, StatisticsClock statisticsClock) { |
| super(cache, managerImpl, id, isjta, statisticsClock); |
| } |
| |
| /* |
| * (non-Javadoc) |
| * |
| * @see org.apache.geode.internal.cache.TXStateInterface#commit() |
| * |
| * [DISTTX] TODO Catch all exceptions in precommit and rollback and make sure these messages reach |
| * all |
| */ |
| @Override |
| public void commit() throws CommitConflictException { |
| boolean preserveTx = false; |
| boolean precommitResult = false; |
| try { |
| // create a map of secondary(for PR) / replica(for RR) to stubs to send |
| // commit message to those |
| HashMap<DistributedMember, DistTXCoordinatorInterface> otherTargets2realDeals = |
| getSecondariesAndReplicasForTxOps(); |
| // add it to the existing map and then send commit to all copies |
| target2realDeals.putAll(otherTargets2realDeals); |
| if (logger.isDebugEnabled()) { |
| logger.debug( |
| "DistTXStateProxyImplOnCoordinator.commit target2realDeals = " + target2realDeals); |
| } |
| |
| precommitResult = doPrecommit(); |
| if (precommitResult) { |
| if (logger.isDebugEnabled()) { |
| logger.debug("DistTXStateProxyImplOnCoordinator.commit Going for commit "); |
| } |
| boolean phase2commitDone = doCommit(); |
| if (logger.isDebugEnabled()) { |
| logger.debug("DistTXStateProxyImplOnCoordinator.commit Commit " |
| + (phase2commitDone ? "Done" : "Failed")); |
| } |
| if (!phase2commitDone) { |
| throw new TransactionInDoubtException( |
| "Commit failed on cache server"); |
| } |
| } else { |
| if (logger.isDebugEnabled()) { |
| logger.debug( |
| "DistTXStateProxyImplOnCoordinator.commit precommitResult = " + precommitResult); |
| } |
| } |
| } catch (UnsupportedOperationInTransactionException e) { |
| // fix for #42490 |
| preserveTx = true; |
| throw e; |
| } finally { |
| if (!precommitResult) { |
| rollback(); |
| } |
| |
| inProgress = preserveTx; |
| } |
| } |
| |
| /** |
| * creates a map of all secondaries(for PR) / replicas(for RR) to stubs to send commit message to |
| * those |
| */ |
| private HashMap<DistributedMember, DistTXCoordinatorInterface> getSecondariesAndReplicasForTxOps() { |
| InternalDistributedMember currentNode = |
| getCache().getInternalDistributedSystem().getDistributedMember(); |
| |
| HashMap<DistributedMember, DistTXCoordinatorInterface> secondaryTarget2realDeals = |
| new HashMap<>(); |
| for (Entry<DistributedMember, DistTXCoordinatorInterface> e : target2realDeals.entrySet()) { |
| DistributedMember originalTarget = e.getKey(); |
| DistTXCoordinatorInterface distPeerTxStateStub = e.getValue(); |
| |
| ArrayList<DistTxEntryEvent> primaryTxOps = |
| distPeerTxStateStub.getPrimaryTransactionalOperations(); |
| for (DistTxEntryEvent dtop : primaryTxOps) { |
| InternalRegion internalRegion = dtop.getRegion(); |
| // replicas or secondaries |
| Set<InternalDistributedMember> otherNodes = null; |
| if (internalRegion instanceof PartitionedRegion) { |
| Set<InternalDistributedMember> allNodes = ((PartitionedRegion) dtop.getRegion()) |
| .getRegionAdvisor().getBucketOwners(dtop.getKeyInfo().getBucketId()); |
| allNodes.remove(originalTarget); |
| otherNodes = allNodes; |
| } else if (internalRegion instanceof DistributedRegion) { |
| otherNodes = ((DistributedRegion) internalRegion).getCacheDistributionAdvisor() |
| .adviseInitializedReplicates(); |
| otherNodes.remove(originalTarget); |
| } |
| |
| if (otherNodes != null) { |
| for (InternalDistributedMember dm : otherNodes) { |
| // whether the target already exists due to other Tx op on the node |
| DistTXCoordinatorInterface existingDistPeerTXStateStub = target2realDeals.get(dm); |
| if (existingDistPeerTXStateStub == null) { |
| existingDistPeerTXStateStub = secondaryTarget2realDeals.get(dm); |
| if (existingDistPeerTXStateStub == null) { |
| DistTXCoordinatorInterface newTxStub = null; |
| if (currentNode.equals(dm)) { |
| // [DISTTX] TODO add a test case for this condition? |
| newTxStub = new DistTXStateOnCoordinator(this, false, getStatisticsClock()); |
| } else { |
| newTxStub = new DistPeerTXStateStub(this, dm, onBehalfOfClientMember); |
| } |
| newTxStub.addSecondaryTransactionalOperations(dtop); |
| secondaryTarget2realDeals.put(dm, newTxStub); |
| } else { |
| existingDistPeerTXStateStub.addSecondaryTransactionalOperations(dtop); |
| } |
| } else { |
| existingDistPeerTXStateStub.addSecondaryTransactionalOperations(dtop); |
| } |
| } |
| } |
| } |
| } |
| return secondaryTarget2realDeals; |
| } |
| |
| @Override |
| public void rollback() { |
| if (logger.isDebugEnabled()) { |
| logger.debug("DistTXStateProxyImplOnCoordinator.rollback Going for rollback "); |
| } |
| |
| boolean finalResult = false; |
| final DistributionManager dm = getCache().getDistributionManager(); |
| try { |
| // Create Tx Participants |
| Set<DistributedMember> txRemoteParticpants = getTxRemoteParticpants(dm); |
| |
| // create processor and rollback message |
| DistTXRollbackMessage.DistTxRollbackReplyProcessor processor = |
| new DistTXRollbackMessage.DistTxRollbackReplyProcessor(this.getTxId(), dm, |
| txRemoteParticpants, target2realDeals); |
| // TODO [DISTTX} whats ack threshold? |
| processor.enableSevereAlertProcessing(); |
| final DistTXRollbackMessage rollbackMsg = |
| new DistTXRollbackMessage(this.getTxId(), this.onBehalfOfClientMember, processor); |
| |
| // send rollback message to remote nodes |
| for (DistributedMember remoteNode : txRemoteParticpants) { |
| DistTXCoordinatorInterface remoteTXStateStub = target2realDeals.get(remoteNode); |
| if (remoteTXStateStub.isTxState()) { |
| throw new UnsupportedOperationInTransactionException( |
| String.format("Expected %s during a distributed transaction but got %s", |
| "DistPeerTXStateStub", |
| remoteTXStateStub.getClass().getSimpleName())); |
| } |
| try { |
| remoteTXStateStub.setRollbackMessage(rollbackMsg, dm); |
| remoteTXStateStub.rollback(); |
| } finally { |
| remoteTXStateStub.setRollbackMessage(null, null); |
| remoteTXStateStub.finalCleanup(); |
| } |
| if (logger.isDebugEnabled()) { // TODO - make this trace level |
| logger.debug("DistTXStateProxyImplOnCoordinator.rollback target = " + remoteNode); |
| } |
| } |
| |
| // Do rollback on local node |
| DistTXCoordinatorInterface localTXState = target2realDeals.get(dm.getId()); |
| if (localTXState != null) { |
| if (!localTXState.isTxState()) { |
| throw new UnsupportedOperationInTransactionException( |
| String.format("Expected %s during a distributed transaction but got %s", |
| "DistTXStateOnCoordinator", |
| localTXState.getClass().getSimpleName())); |
| } |
| localTXState.rollback(); |
| boolean localResult = localTXState.getRollbackResponse(); |
| if (logger.isDebugEnabled()) { |
| logger.debug("DistTXStateProxyImplOnCoordinator.rollback local = " + dm.getId() |
| + " ,result= " + localResult + " ,finalResult-old= " + finalResult); |
| } |
| finalResult = finalResult && localResult; |
| } |
| |
| /* |
| * [DISTTX] TODO Any test hooks |
| */ |
| // if (internalAfterIndividualSend != null) { |
| // internalAfterIndividualSend.run(); |
| // } |
| |
| /* |
| * [DISTTX] TODO see how to handle exception |
| */ |
| |
| /* |
| * [DISTTX] TODO Any test hooks |
| */ |
| // if (internalAfterIndividualCommitProcess != null) { |
| // // Testing callback |
| // internalAfterIndividualCommitProcess.run(); |
| // } |
| |
| { // Wait for results |
| dm.getCancelCriterion().checkCancelInProgress(null); |
| processor.waitForPrecommitCompletion(); |
| |
| // [DISTTX} TODO Handle stats |
| // dm.getStats().incCommitWaits(); |
| |
| Map<DistributedMember, Boolean> remoteResults = processor.getRollbackResponseMap(); |
| for (Entry<DistributedMember, Boolean> e : remoteResults.entrySet()) { |
| DistributedMember target = e.getKey(); |
| Boolean remoteResult = e.getValue(); |
| if (logger.isDebugEnabled()) { // TODO - make this trace level |
| logger.debug("DistTXStateProxyImplOnCoordinator.rollback target = " + target |
| + " ,result= " + remoteResult + " ,finalResult-old= " + finalResult); |
| } |
| finalResult = finalResult && remoteResult; |
| } |
| } |
| |
| } finally { |
| inProgress = false; |
| } |
| |
| /* |
| * [DISTTX] TODO Write similar method to take out exception |
| * |
| * [DISTTX] TODO Handle Reliable regions |
| */ |
| // if (this.hasReliableRegions) { |
| // checkDistributionReliability(distMap, processor); |
| // } |
| |
| if (logger.isDebugEnabled()) { |
| logger.debug("DistTXStateProxyImplOnCoordinator.rollback finalResult= " + finalResult); |
| } |
| } |
| |
| /** |
| * {@inheritDoc} |
| */ |
| @Override |
| public TXStateInterface getRealDeal(KeyInfo key, InternalRegion r) { |
| if (r != null) { |
| target = null; |
| // wait for the region to be initialized fixes bug 44652 |
| r.waitOnInitialization(r.getInitializationLatchBeforeGetInitialImage()); |
| if (r instanceof PartitionedRegion) { |
| target = getOwnerForKey(r, key); |
| } else if (r instanceof BucketRegion) { |
| target = ((BucketRegion) r).getBucketAdvisor().getPrimary(); |
| // target = r.getMyId(); |
| } else { // replicated region |
| target = getRRTarget(key, r); |
| } |
| this.realDeal = target2realDeals.get(target); |
| } |
| if (this.realDeal == null) { |
| // assert (r != null); |
| if (r == null) { // TODO: stop gap to get tests working |
| this.realDeal = new DistTXStateOnCoordinator(this, false, getStatisticsClock()); |
| target = this.txMgr.getDM().getId(); |
| } else { |
| // Code to keep going forward |
| if (r.hasServerProxy()) { |
| // TODO [DISTTX] See what we need for client? |
| this.realDeal = |
| new DistClientTXStateStub(r.getCache(), r.getDistributionManager(), this, target, r); |
| if (r.getScope().isDistributed()) { |
| if (txDistributedClientWarningIssued.compareAndSet(false, true)) { |
| logger.warn( |
| "Distributed region {} is being used in a client-initiated transaction. The transaction will only affect servers and this client. To keep from seeing this message use 'local' scope in client regions used in transactions.", |
| r.getFullPath()); |
| } |
| } |
| } else { |
| // (r != null) code block above |
| if (target == null || target.equals(this.txMgr.getDM().getId())) { |
| this.realDeal = new DistTXStateOnCoordinator(this, false, getStatisticsClock()); |
| } else { |
| this.realDeal = new DistPeerTXStateStub(this, target, onBehalfOfClientMember); |
| } |
| } |
| } |
| if (logger.isDebugEnabled()) { |
| logger.debug( |
| "DistTXStateProxyImplOnCoordinator::getRealDeal Built a new TXState: {} txMge:{} proxy {} target {}", |
| this.realDeal, this.txMgr.getDM().getId(), this, target/* , new Throwable() */); |
| } |
| target2realDeals.put(target, (DistTXCoordinatorInterface) realDeal); |
| if (logger.isDebugEnabled()) { |
| logger |
| .debug("DistTXStateProxyImplOnCoordinator.getRealDeal added TxState target2realDeals = " |
| + target2realDeals); |
| } |
| } else { |
| if (logger.isDebugEnabled()) { |
| logger.debug( |
| "DistTXStateProxyImplOnCoordinator::getRealDeal Found TXState: {} proxy {} target {} target2realDeals {}", |
| this.realDeal, this, target, target2realDeals); |
| } |
| } |
| return this.realDeal; |
| } |
| |
| @Override |
| public TXStateInterface getRealDeal(DistributedMember t) { |
| assert t != null; |
| this.realDeal = target2realDeals.get(target); |
| if (this.realDeal == null) { |
| this.target = t; |
| this.realDeal = new DistPeerTXStateStub(this, target, onBehalfOfClientMember); |
| if (logger.isDebugEnabled()) { |
| logger.debug( |
| "DistTXStateProxyImplOnCoordinator::getRealDeal(t) Built a new TXState: {} me:{}", |
| this.realDeal, this.txMgr.getDM().getId()); |
| } |
| if (!this.realDeal.isDistTx() || this.realDeal.isCreatedOnDistTxCoordinator() |
| || !this.realDeal.isTxState()) { |
| throw new UnsupportedOperationInTransactionException( |
| String.format("Expected %s during a distributed transaction but got %s", |
| "DistPeerTXStateStub", this.realDeal.getClass().getSimpleName())); |
| } |
| target2realDeals.put(target, (DistPeerTXStateStub) realDeal); |
| if (logger.isDebugEnabled()) { |
| logger.debug( |
| "DistTXStateProxyImplOnCoordinator.getRealDeal(t) added TxState target2realDeals = " |
| + target2realDeals); |
| } |
| } else { |
| if (logger.isDebugEnabled()) { |
| logger.debug( |
| "DistTXStateProxyImplOnCoordinator::getRealDeal(t) Found TXState: {} proxy {} target {} target2realDeals {}", |
| this.realDeal, this, target, target2realDeals); |
| } |
| } |
| return this.realDeal; |
| } |
| |
| /* |
| * [DISTTX] TODO Do some optimization |
| */ |
| private DistributedMember getRRTarget(KeyInfo key, InternalRegion r) { |
| if (this.rrTargets == null) { |
| this.rrTargets = new HashMap(); |
| } |
| DistributedMember m = this.rrTargets.get(r); |
| if (m == null) { |
| m = getOwnerForKey(r, key); |
| this.rrTargets.put(r, m); |
| } |
| return m; |
| } |
| |
| private Set<DistributedMember> getTxRemoteParticpants(final DistributionManager dm) { |
| if (this.txRemoteParticpants == null) { |
| Set<DistributedMember> txParticpants = target2realDeals.keySet(); |
| this.txRemoteParticpants = new HashSet<DistributedMember>(txParticpants); |
| // Remove local member from remote participant list |
| this.txRemoteParticpants.remove(dm.getId()); |
| if (logger.isDebugEnabled()) { |
| logger.debug("DistTXStateProxyImplOnCoordinator.doPrecommit txParticpants = " |
| + txParticpants + " ,txRemoteParticpants=" + this.txRemoteParticpants + " ,originator=" |
| + dm.getId()); |
| } |
| } |
| return txRemoteParticpants; |
| } |
| |
| private boolean doPrecommit() { |
| boolean finalResult = true; |
| final DistributionManager dm = getCache().getDistributionManager(); |
| Set<DistributedMember> txRemoteParticpants = getTxRemoteParticpants(dm); |
| |
| // create processor and precommit message |
| DistTXPrecommitMessage.DistTxPrecommitReplyProcessor processor = |
| new DistTXPrecommitMessage.DistTxPrecommitReplyProcessor(this.getTxId(), dm, |
| txRemoteParticpants, target2realDeals); |
| // TODO [DISTTX} whats ack threshold? |
| processor.enableSevereAlertProcessing(); |
| final DistTXPrecommitMessage precommitMsg = |
| new DistTXPrecommitMessage(this.getTxId(), this.onBehalfOfClientMember, processor); |
| |
| // send precommit message to remote nodes |
| for (DistributedMember remoteNode : txRemoteParticpants) { |
| DistTXCoordinatorInterface remoteTXStateStub = target2realDeals.get(remoteNode); |
| if (remoteTXStateStub.isTxState()) { |
| throw new UnsupportedOperationInTransactionException( |
| String.format("Expected %s during a distributed transaction but got %s", |
| "DistPeerTXStateStub", |
| remoteTXStateStub.getClass().getSimpleName())); |
| } |
| try { |
| remoteTXStateStub.setPrecommitMessage(precommitMsg, dm); |
| remoteTXStateStub.precommit(); |
| } finally { |
| remoteTXStateStub.setPrecommitMessage(null, null); |
| } |
| if (logger.isDebugEnabled()) { |
| logger.debug( |
| "DistTXStateProxyImplOnCoordinator.doPrecommit Sent Message to target = " + remoteNode); |
| } |
| } |
| |
| // Do precommit on local node |
| TreeSet<String> sortedRegionName = new TreeSet<>(); |
| DistTXCoordinatorInterface localTXState = target2realDeals.get(dm.getId()); |
| if (localTXState != null) { |
| if (!localTXState.isTxState()) { |
| throw new UnsupportedOperationInTransactionException( |
| String.format("Expected %s during a distributed transaction but got %s", |
| "DistTXStateOnCoordinator", |
| localTXState.getClass().getSimpleName())); |
| } |
| localTXState.precommit(); |
| boolean localResult = localTXState.getPreCommitResponse(); |
| TreeMap<String, ArrayList<DistTxThinEntryState>> entryStateSortedMap = |
| new TreeMap<String, ArrayList<DistTxThinEntryState>>(); |
| ArrayList<ArrayList<DistTxThinEntryState>> entryEventList = null; |
| if (localResult) { |
| localResult = ((DistTXStateOnCoordinator) localTXState) |
| .populateDistTxEntryStateList(entryStateSortedMap); |
| if (localResult) { |
| entryEventList = |
| new ArrayList<ArrayList<DistTxThinEntryState>>(entryStateSortedMap.values()); |
| populateEntryEventMap(dm.getId(), entryEventList, sortedRegionName); |
| } |
| } |
| |
| if (logger.isDebugEnabled()) { |
| logger.debug("DistTXStateProxyImplOnCoordinator.doPrecommit local = " + dm.getId() |
| + " ,entryEventList=" + printEntryEventList(entryEventList) + " ,txRegionVersionsMap=" |
| + printEntryEventMap(this.txEntryEventMap) + " ,result= " + localResult |
| + " ,finalResult-old= " + finalResult); |
| } |
| finalResult = finalResult && localResult; |
| } |
| |
| /* |
| * [DISTTX] TODO Any test hooks |
| */ |
| // if (internalAfterIndividualSend != null) { |
| // internalAfterIndividualSend.run(); |
| // } |
| |
| /* |
| * [DISTTX] TODO see how to handle exception |
| */ |
| |
| /* |
| * [DISTTX] TODO Any test hooks |
| */ |
| // if (internalAfterIndividualCommitProcess != null) { |
| // // Testing callback |
| // internalAfterIndividualCommitProcess.run(); |
| // } |
| |
| { // Wait for results |
| dm.getCancelCriterion().checkCancelInProgress(null); |
| processor.waitForPrecommitCompletion(); |
| |
| // [DISTTX} TODO Handle stats |
| // dm.getStats().incCommitWaits(); |
| |
| Map<DistributedMember, DistTxPrecommitResponse> remoteResults = |
| processor.getCommitResponseMap(); |
| for (Entry<DistributedMember, DistTxPrecommitResponse> e : remoteResults.entrySet()) { |
| DistributedMember target = e.getKey(); |
| DistTxPrecommitResponse remoteResponse = e.getValue(); |
| ArrayList<ArrayList<DistTxThinEntryState>> entryEventList = |
| remoteResponse.getDistTxEntryEventList(); |
| populateEntryEventMap(target, entryEventList, sortedRegionName); |
| if (logger.isDebugEnabled()) { |
| logger.debug("DistTXStateProxyImplOnCoordinator.doPrecommit got reply from target = " |
| + target + " ,sortedRegions" + sortedRegionName + " ,entryEventList=" |
| + printEntryEventList(entryEventList) + " ,txEntryEventMap=" |
| + printEntryEventMap(this.txEntryEventMap) + " ,result= " |
| + remoteResponse.getCommitState() + " ,finalResult-old= " + finalResult); |
| } |
| finalResult = finalResult && remoteResponse.getCommitState(); |
| } |
| } |
| |
| /* |
| * [DISTTX] TODO Write similar method to take out exception |
| * |
| * [DISTTX] TODO Handle Reliable regions |
| */ |
| // if (this.hasReliableRegions) { |
| // checkDistributionReliability(distMap, processor); |
| // } |
| |
| if (logger.isDebugEnabled()) { |
| logger.debug("DistTXStateProxyImplOnCoordinator.doPrecommit finalResult= " + finalResult); |
| } |
| return finalResult; |
| } |
| |
| /* |
| * Handle response of precommit reply |
| * |
| * Go over list of region versions for this target and fill map |
| */ |
| private void populateEntryEventMap(DistributedMember target, |
| ArrayList<ArrayList<DistTxThinEntryState>> entryEventList, TreeSet<String> sortedRegionName) { |
| if (this.txEntryEventMap == null) { |
| this.txEntryEventMap = new HashMap<String, ArrayList<DistTxThinEntryState>>(); |
| } |
| |
| DistTXCoordinatorInterface distTxIface = target2realDeals.get(target); |
| if (distTxIface.getPrimaryTransactionalOperations() != null |
| && distTxIface.getPrimaryTransactionalOperations().size() > 0) { |
| sortedRegionName.clear(); |
| distTxIface.gatherAffectedRegionsName(sortedRegionName, true, false); |
| |
| if (sortedRegionName.size() != entryEventList.size()) { |
| throw new UnsupportedOperationInTransactionException( |
| String.format("Expected %s during a distributed transaction but got %s", |
| "size of " + sortedRegionName.size() + " {" + sortedRegionName + "}" |
| + " for target=" + target, |
| entryEventList.size() + " {" + entryEventList + "}")); |
| } |
| |
| int index = 0; |
| // Get region as per sorted order of region path |
| for (String rName : sortedRegionName) { |
| txEntryEventMap.put(rName, entryEventList.get(index++)); |
| } |
| } |
| } |
| |
| /* |
| * Populate list of regions for this target, while sending commit messages |
| */ |
| private void populateEntryEventList(DistributedMember target, |
| ArrayList<ArrayList<DistTxThinEntryState>> entryEventList, TreeSet<String> sortedRegionMap) { |
| DistTXCoordinatorInterface distTxItem = target2realDeals.get(target); |
| sortedRegionMap.clear(); |
| distTxItem.gatherAffectedRegionsName(sortedRegionMap, false, true); |
| |
| // Get region as per sorted order of region path |
| entryEventList.clear(); |
| for (String rName : sortedRegionMap) { |
| ArrayList<DistTxThinEntryState> entryStates = this.txEntryEventMap.get(rName); |
| if (entryStates == null) { |
| throw new UnsupportedOperationInTransactionException( |
| String.format("Expected %s during a distributed transaction but got %s", |
| "entryStates for " + rName + " at target " + target, "null")); |
| } |
| entryEventList.add(entryStates); |
| } |
| } |
| |
| /* |
| * [DISTTX] TODO - Handle result TXMessage |
| */ |
| private boolean doCommit() { |
| boolean finalResult = true; |
| final DistributionManager dm = getCache().getDistributionManager(); |
| |
| // Create Tx Participants |
| Set<DistributedMember> txRemoteParticpants = getTxRemoteParticpants(dm); |
| |
| // create processor and commit message |
| DistTXCommitMessage.DistTxCommitReplyProcessor processor = |
| new DistTXCommitMessage.DistTxCommitReplyProcessor(this.getTxId(), dm, txRemoteParticpants, |
| target2realDeals); |
| // TODO [DISTTX} whats ack threshold? |
| processor.enableSevereAlertProcessing(); |
| final DistTXCommitMessage commitMsg = |
| new DistTXCommitMessage(this.getTxId(), this.onBehalfOfClientMember, processor); |
| |
| // send commit message to remote nodes |
| ArrayList<ArrayList<DistTxThinEntryState>> entryEventList = new ArrayList<>(); |
| TreeSet<String> sortedRegionName = new TreeSet<>(); |
| for (DistributedMember remoteNode : txRemoteParticpants) { |
| DistTXCoordinatorInterface remoteTXStateStub = target2realDeals.get(remoteNode); |
| if (remoteTXStateStub.isTxState()) { |
| throw new UnsupportedOperationInTransactionException( |
| String.format("Expected %s during a distributed transaction but got %s", |
| "DistPeerTXStateStub", |
| remoteTXStateStub.getClass().getSimpleName())); |
| } |
| try { |
| populateEntryEventList(remoteNode, entryEventList, sortedRegionName); |
| commitMsg.setEntryStateList(entryEventList); |
| remoteTXStateStub.setCommitMessage(commitMsg, dm); |
| remoteTXStateStub.commit(); |
| } finally { |
| remoteTXStateStub.setCommitMessage(null, null); |
| remoteTXStateStub.finalCleanup(); |
| } |
| if (logger.isDebugEnabled()) { |
| logger.debug("DistTXStateProxyImplOnCoordinator.doCommit Sent Message target = " |
| + remoteNode + " ,sortedRegions=" + sortedRegionName + " ,entryEventList=" |
| + printEntryEventList(entryEventList) + " ,txEntryEventMap=" |
| + printEntryEventMap(this.txEntryEventMap)); |
| } |
| } |
| |
| // Do commit on local node |
| DistTXCoordinatorInterface localTXState = target2realDeals.get(dm.getId()); |
| if (localTXState != null) { |
| if (!localTXState.isTxState()) { |
| throw new UnsupportedOperationInTransactionException( |
| String.format("Expected %s during a distributed transaction but got %s", |
| "DistTXStateOnCoordinator", |
| localTXState.getClass().getSimpleName())); |
| } |
| populateEntryEventList(dm.getId(), entryEventList, sortedRegionName); |
| ((DistTXStateOnCoordinator) localTXState).setDistTxEntryStates(entryEventList); |
| localTXState.commit(); |
| TXCommitMessage localResultMsg = localTXState.getCommitMessage(); |
| if (logger.isDebugEnabled()) { |
| logger.debug( |
| "DistTXStateProxyImplOnCoordinator.doCommit local = " + dm.getId() + " ,sortedRegions=" |
| + sortedRegionName + " ,entryEventList=" + printEntryEventList(entryEventList) |
| + " ,txEntryEventMap=" + printEntryEventMap(this.txEntryEventMap) + " ,result= " |
| + (localResultMsg != null) + " ,finalResult-old= " + finalResult); |
| } |
| finalResult = finalResult && (localResultMsg != null); |
| } |
| |
| /* |
| * [DISTTX] TODO Any test hooks |
| */ |
| // if (internalAfterIndividualSend != null) { |
| // internalAfterIndividualSend.run(); |
| // } |
| |
| /* |
| * [DISTTX] TODO see how to handle exception |
| */ |
| |
| /* |
| * [DISTTX] TODO Any test hooks |
| */ |
| // if (internalAfterIndividualCommitProcess != null) { |
| // // Testing callback |
| // internalAfterIndividualCommitProcess.run(); |
| // } |
| |
| { // Wait for results |
| dm.getCancelCriterion().checkCancelInProgress(null); |
| processor.waitForPrecommitCompletion(); |
| |
| // [DISTTX} TODO Handle stats |
| dm.getStats().incCommitWaits(); |
| |
| Map<DistributedMember, TXCommitMessage> remoteResults = processor.getCommitResponseMap(); |
| for (Entry<DistributedMember, TXCommitMessage> e : remoteResults.entrySet()) { |
| DistributedMember target = e.getKey(); |
| TXCommitMessage remoteResultMsg = e.getValue(); |
| if (logger.isDebugEnabled()) { // TODO - make this trace level |
| logger.debug( |
| "DistTXStateProxyImplOnCoordinator.doCommit got results from target = " + target |
| + " ,result= " + (remoteResultMsg != null) + " ,finalResult-old= " + finalResult); |
| } |
| finalResult = finalResult && remoteResultMsg != null; |
| } |
| } |
| |
| /* |
| * [DISTTX] TODO Write similar method to take out exception |
| * |
| * [DISTTX] TODO Handle Reliable regions |
| */ |
| // if (this.hasReliableRegions) { |
| // checkDistributionReliability(distMap, processor); |
| // } |
| |
| if (logger.isDebugEnabled()) { |
| logger.debug("DistTXStateProxyImplOnCoordinator.doCommit finalResult= " + finalResult); |
| } |
| return finalResult; |
| } |
| |
| /** |
| * For distributed transactions, this divides the user's putAll operation into multiple per bucket |
| * putAll ops(with entries to be put in that bucket) and then fires those using using appropriate |
| * TXStateStub (for target that host the corresponding bucket) |
| */ |
| @Override |
| public void postPutAll(DistributedPutAllOperation putallOp, VersionedObjectList successfulPuts, |
| InternalRegion reg) { |
| if (putallOp.putAllData.length == 0) { |
| return; |
| } |
| if (reg instanceof DistributedRegion) { |
| super.postPutAll(putallOp, successfulPuts, reg); |
| } else { |
| reg.getCancelCriterion().checkCancelInProgress(null); // fix for bug |
| // #43651 |
| |
| if (logger.isDebugEnabled()) { |
| logger.debug( |
| "DistTXStateProxyImplOnCoordinator.postPutAll " |
| + "processing putAll op for region {}, size of putAllOp " + "is {}", |
| reg, putallOp.putAllData.length); |
| } |
| |
| |
| // map of bucketId to putall op for this bucket |
| HashMap<Integer, DistributedPutAllOperation> bucketToPutallMap = |
| new HashMap<Integer, DistributedPutAllOperation>(); |
| // map of bucketId to TXStateStub for target that hosts this bucket |
| HashMap<Integer, DistTXCoordinatorInterface> bucketToTxStateStubMap = |
| new HashMap<Integer, DistTXCoordinatorInterface>(); |
| |
| // separate the putall op per bucket |
| for (int i = 0; i < putallOp.putAllData.length; i++) { |
| assert (putallOp.putAllData[i] != null); |
| Object key = putallOp.putAllData[i].key; |
| int bucketId = putallOp.putAllData[i].getBucketId(); |
| |
| DistributedPutAllOperation putAllForBucket = bucketToPutallMap.get(bucketId);; |
| if (putAllForBucket == null) { |
| // TODO DISTTX: event is never released |
| EntryEventImpl event = EntryEventImpl.createPutAllEvent(null, reg, |
| Operation.PUTALL_CREATE, key, putallOp.putAllData[i].getValue(reg.getCache())); |
| event.setEventId(putallOp.putAllData[i].getEventID()); |
| putAllForBucket = |
| new DistributedPutAllOperation(event, putallOp.putAllDataSize, putallOp.isBridgeOp); |
| bucketToPutallMap.put(bucketId, putAllForBucket); |
| } |
| putallOp.putAllData[i].setFakeEventID(); |
| putAllForBucket.addEntry(putallOp.putAllData[i]); |
| |
| KeyInfo ki = new KeyInfo(key, null, null); |
| DistTXCoordinatorInterface tsi = (DistTXCoordinatorInterface) getRealDeal(ki, reg); |
| bucketToTxStateStubMap.put(bucketId, tsi); |
| } |
| |
| // fire a putAll operation for each bucket using appropriate TXStateStub |
| // (for target that host this bucket) |
| |
| // [DISTTX] [TODO] Perf: Can this be further optimized? |
| // This sends putAll in a loop to each target bucket (and waits for ack) |
| // one after another.Could we send respective putAll messages to all |
| // targets using same reply processor and wait on it? |
| for (Entry<Integer, DistTXCoordinatorInterface> e : bucketToTxStateStubMap.entrySet()) { |
| Integer bucketId = e.getKey(); |
| DistTXCoordinatorInterface dtsi = e.getValue(); |
| DistributedPutAllOperation putAllForBucket = bucketToPutallMap.get(bucketId); |
| |
| if (logger.isDebugEnabled()) { |
| logger.debug( |
| "DistTXStateProxyImplOnCoordinator.postPutAll processing" |
| + " putAll for ##bucketId = {}, ##txStateStub = {}, " + "##putAllOp = {}", |
| bucketId, dtsi, putAllForBucket); |
| } |
| dtsi.postPutAll(putAllForBucket, successfulPuts, reg); |
| } |
| } |
| } |
| |
| /** |
| * For distributed transactions, this divides the user's removeAll operation into multiple per |
| * bucket removeAll ops(with entries to be removed from that bucket) and then fires those using |
| * using appropriate TXStateStub (for target that host the corresponding bucket) |
| */ |
| @Override |
| public void postRemoveAll(DistributedRemoveAllOperation op, VersionedObjectList successfulOps, |
| InternalRegion reg) { |
| if (op.removeAllData.length == 0) { |
| return; |
| } |
| if (reg instanceof DistributedRegion) { |
| super.postRemoveAll(op, successfulOps, reg); |
| } else { |
| reg.getCancelCriterion().checkCancelInProgress(null); // fix for bug |
| // #43651 |
| if (logger.isDebugEnabled()) { |
| logger.debug( |
| "DistTXStateProxyImplOnCoordinator.postRemoveAll " |
| + "processing removeAll op for region {}, size of removeAll " + "is {}", |
| reg, op.removeAllDataSize); |
| } |
| |
| // map of bucketId to removeAll op for this bucket |
| HashMap<Integer, DistributedRemoveAllOperation> bucketToRemoveAllMap = |
| new HashMap<Integer, DistributedRemoveAllOperation>(); |
| // map of bucketId to TXStateStub for target that hosts this bucket |
| HashMap<Integer, DistTXCoordinatorInterface> bucketToTxStateStubMap = |
| new HashMap<Integer, DistTXCoordinatorInterface>(); |
| |
| // separate the removeAll op per bucket |
| for (int i = 0; i < op.removeAllData.length; i++) { |
| assert (op.removeAllData[i] != null); |
| Object key = op.removeAllData[i].key; |
| int bucketId = op.removeAllData[i].getBucketId(); |
| |
| DistributedRemoveAllOperation removeAllForBucket = bucketToRemoveAllMap.get(bucketId); |
| if (removeAllForBucket == null) { |
| // TODO DISTTX: event is never released |
| EntryEventImpl event = EntryEventImpl.createRemoveAllEvent(op, reg, key); |
| event.setEventId(op.removeAllData[i].getEventID()); |
| removeAllForBucket = |
| new DistributedRemoveAllOperation(event, op.removeAllDataSize, op.isBridgeOp); |
| bucketToRemoveAllMap.put(bucketId, removeAllForBucket); |
| } |
| op.removeAllData[i].setFakeEventID(); |
| removeAllForBucket.addEntry(op.removeAllData[i]); |
| |
| KeyInfo ki = new KeyInfo(key, null, null); |
| DistTXCoordinatorInterface tsi = (DistTXCoordinatorInterface) getRealDeal(ki, reg); |
| bucketToTxStateStubMap.put(bucketId, tsi); |
| } |
| |
| // fire a removeAll operation for each bucket using appropriate TXStateStub |
| // (for target that host this bucket) |
| |
| // [DISTTX] [TODO] Perf: Can this be further optimized? |
| // This sends putAll in a loop to each target bucket (and waits for ack) |
| // one after another.Could we send respective putAll messages to all |
| // targets using same reply processor and wait on it? |
| for (Entry<Integer, DistTXCoordinatorInterface> e : bucketToTxStateStubMap.entrySet()) { |
| Integer bucketId = e.getKey(); |
| DistTXCoordinatorInterface dtsi = e.getValue(); |
| DistributedRemoveAllOperation removeAllForBucket = bucketToRemoveAllMap.get(bucketId); |
| |
| if (logger.isDebugEnabled()) { |
| logger.debug( |
| "DistTXStateProxyImplOnCoordinator.postRemoveAll processing" |
| + " removeAll for ##bucketId = {}, ##txStateStub = {}, " + "##removeAllOp = {}", |
| bucketId, dtsi, removeAllForBucket); |
| } |
| dtsi.postRemoveAll(removeAllForBucket, successfulOps, reg); |
| } |
| |
| } |
| } |
| |
| @Override |
| public boolean isCreatedOnDistTxCoordinator() { |
| return true; |
| } |
| |
| public static String printEntryEventMap( |
| HashMap<String, ArrayList<DistTxThinEntryState>> txRegionVersionsMap) { |
| StringBuilder str = new StringBuilder(); |
| str.append(" ("); |
| str.append(txRegionVersionsMap.size()); |
| str.append(")=[ "); |
| for (Map.Entry<String, ArrayList<DistTxThinEntryState>> entry : txRegionVersionsMap |
| .entrySet()) { |
| str.append(" {").append(entry.getKey()); |
| str.append(":").append("size(").append(entry.getValue().size()).append(")"); |
| str.append("=").append(entry.getValue()).append("}, "); |
| } |
| str.append(" } "); |
| return str.toString(); |
| } |
| |
| public static String printEntryEventList( |
| ArrayList<ArrayList<DistTxThinEntryState>> entryEventList) { |
| StringBuilder str = new StringBuilder(); |
| str.append(" ("); |
| str.append(entryEventList.size()); |
| str.append(")=[ "); |
| for (ArrayList<DistTxThinEntryState> entry : entryEventList) { |
| str.append(" ( "); |
| str.append(entry.size()); |
| str.append(" )={").append(entry); |
| str.append(" } "); |
| } |
| str.append(" ] "); |
| return str.toString(); |
| } |
| |
| /* |
| * Do not return null |
| */ |
| public DistributedMember getOwnerForKey(InternalRegion r, KeyInfo key) { |
| DistributedMember m = r.getOwnerForKey(key); |
| if (m == null) { |
| m = getCache().getDistributedSystem().getDistributedMember(); |
| } |
| return m; |
| } |
| } |
