| /* |
| * 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.Collection; |
| 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 java.util.concurrent.Executor; |
| import java.util.concurrent.locks.ReentrantLock; |
| |
| import javax.transaction.Status; |
| |
| import org.apache.logging.log4j.Logger; |
| |
| import org.apache.geode.CancelCriterion; |
| import org.apache.geode.CancelException; |
| import org.apache.geode.InternalGemFireError; |
| import org.apache.geode.SystemFailure; |
| import org.apache.geode.annotations.Immutable; |
| import org.apache.geode.cache.CommitConflictException; |
| import org.apache.geode.cache.DiskAccessException; |
| import org.apache.geode.cache.EntryNotFoundException; |
| import org.apache.geode.cache.FailedSynchronizationException; |
| import org.apache.geode.cache.Operation; |
| import org.apache.geode.cache.Region; |
| import org.apache.geode.cache.Region.Entry; |
| import org.apache.geode.cache.RegionDestroyedException; |
| import org.apache.geode.cache.SynchronizationCommitConflictException; |
| import org.apache.geode.cache.TransactionDataRebalancedException; |
| import org.apache.geode.cache.TransactionException; |
| import org.apache.geode.cache.TransactionId; |
| import org.apache.geode.cache.TransactionWriter; |
| import org.apache.geode.cache.TransactionWriterException; |
| import org.apache.geode.cache.UnsupportedOperationInTransactionException; |
| import org.apache.geode.cache.client.internal.ServerRegionDataAccess; |
| import org.apache.geode.distributed.DistributedMember; |
| import org.apache.geode.distributed.TXManagerCancelledException; |
| import org.apache.geode.distributed.internal.membership.InternalDistributedMember; |
| import org.apache.geode.internal.Assert; |
| import org.apache.geode.internal.cache.control.MemoryThresholds; |
| import org.apache.geode.internal.cache.entries.AbstractRegionEntry; |
| import org.apache.geode.internal.cache.partitioned.PutAllPRMessage; |
| import org.apache.geode.internal.cache.partitioned.RemoveAllPRMessage; |
| import org.apache.geode.internal.cache.tier.sockets.ClientProxyMembershipID; |
| import org.apache.geode.internal.cache.tier.sockets.VersionedObjectList; |
| import org.apache.geode.internal.cache.tx.TransactionalOperation.ServerRegionOperation; |
| import org.apache.geode.internal.logging.LogService; |
| import org.apache.geode.internal.offheap.annotations.Released; |
| import org.apache.geode.internal.offheap.annotations.Retained; |
| |
| /** |
| * TXState is the entity that tracks the transaction state on a per thread basis, noting changes to |
| * Region entries on a per operation basis. It lives on the node where transaction data exists. |
| * |
| * @since GemFire 4.0 |
| * @see TXManagerImpl |
| */ |
| public class TXState implements TXStateInterface { |
| protected static final Logger logger = LogService.getLogger(); |
| |
| // The nano-timestamp of when the transaction began |
| private final long beginTime; |
| // A map of transaction state by Region |
| final IdentityHashMap<InternalRegion, TXRegionState> regions; |
| |
| /** whether completion has been started */ |
| protected boolean completionStarted; |
| |
| /** whether the transaction has been completed and cleaned up */ |
| protected boolean closed = false; |
| |
| /** guards the completionStarted boolean and the closed boolean */ |
| protected final Object completionGuard = new Object(); |
| |
| protected TXLockRequest locks = null; |
| // Used for jta commit lifetime |
| private long jtaLifeTime; |
| /** |
| * Used to hand out modification serial numbers used to preserve the order of operation done by |
| * this transaction. |
| */ |
| private int modSerialNum; |
| private final List<EntryEventImpl> pendingCallbacks = new ArrayList<EntryEventImpl>(); |
| // Access this variable should be in synchronized block. |
| private boolean beforeCompletionCalled; |
| |
| /** |
| * for client/server JTA transactions we need to have a single thread handle both beforeCompletion |
| * and afterCompletion so that beforeCompletion can obtain locks for the afterCompletion step. |
| * This is that thread |
| */ |
| private final SingleThreadJTAExecutor singleThreadJTAExecutor; |
| |
| // Internal testing hooks |
| private Runnable internalAfterReservation; |
| protected Runnable internalAfterConflictCheck; |
| protected Runnable internalDuringApplyChanges; |
| protected Runnable internalAfterApplyChanges; |
| protected Runnable internalAfterReleaseLocalLocks; |
| Runnable internalDuringIndividualSend; // package scope allows TXCommitMessage use |
| Runnable internalAfterIndividualSend; // package scope allows TXCommitMessage use |
| Runnable internalDuringIndividualCommitProcess; // package scope allows TXCommitMessage use |
| Runnable internalAfterIndividualCommitProcess; // package scope allows TXCommitMessage use |
| protected Runnable internalAfterSend; |
| protected Runnable internalBeforeSend; |
| |
| /** |
| * Used to generate eventIDs |
| */ |
| private byte[] baseMembershipId; |
| /** |
| * Used to generate eventIDs |
| */ |
| private long baseThreadId; |
| /** |
| * Used to generate eventIDs |
| */ |
| private long baseSequenceId; |
| protected final TXStateProxy proxy; |
| protected boolean firedWriter = false; |
| protected final boolean onBehalfOfRemoteStub; |
| protected boolean gotBucketLocks = false; |
| protected TXCommitMessage commitMessage = null; |
| ClientProxyMembershipID bridgeContext = null; |
| /** keeps track of events, so as not to re-apply events */ |
| protected Set<EventID> seenEvents = new HashSet<EventID>(); |
| /** keeps track of results of txPutEntry */ |
| private Map<EventID, Boolean> seenResults = new HashMap<EventID, Boolean>(); |
| |
| @Immutable |
| static final TXEntryState ENTRY_EXISTS = new TXEntryState(); |
| |
| private volatile DistributedMember proxyServer; |
| |
| public TXState(TXStateProxy proxy, boolean onBehalfOfRemoteStub) { |
| this(proxy, onBehalfOfRemoteStub, new SingleThreadJTAExecutor()); |
| } |
| |
| public TXState(TXStateProxy proxy, boolean onBehalfOfRemoteStub, |
| SingleThreadJTAExecutor singleThreadJTAExecutor) { |
| this.beginTime = CachePerfStats.getStatTime(); |
| this.regions = new IdentityHashMap<>(); |
| |
| this.internalAfterConflictCheck = null; |
| this.internalAfterApplyChanges = null; |
| this.internalAfterReleaseLocalLocks = null; |
| this.internalDuringIndividualSend = null; |
| this.internalAfterIndividualSend = null; |
| this.internalBeforeSend = null; |
| this.internalAfterSend = null; |
| this.proxy = proxy; |
| this.onBehalfOfRemoteStub = onBehalfOfRemoteStub; |
| this.singleThreadJTAExecutor = singleThreadJTAExecutor; |
| } |
| |
| private boolean hasSeenEvent(EntryEventImpl event) { |
| assert event != null; |
| if (event.getEventId() == null) { |
| return false; |
| } |
| return this.seenEvents.contains(event.getEventId()); |
| } |
| |
| private void recordEvent(EntryEventImpl event) { |
| assert event != null; |
| if (event.getEventId() != null) { |
| this.seenEvents.add(event.getEventId()); |
| } |
| } |
| |
| private void recordEventAndResult(EntryEventImpl event, boolean result) { |
| recordEvent(event); |
| if (event.getEventId() != null) { |
| this.seenResults.put(event.getEventId(), result); |
| } |
| } |
| |
| private boolean getRecordedResult(EntryEventImpl event) { |
| assert event != null; |
| assert this.seenResults.containsKey(event.getEventId()); |
| return this.seenResults.get(event.getEventId()); |
| } |
| |
| @Override |
| public String toString() { |
| StringBuilder builder = new StringBuilder(); |
| builder.append(this.getClass()).append("@").append(System.identityHashCode(this)) |
| .append(" onBehalfOfRemoteStub:").append(this.onBehalfOfRemoteStub); |
| return builder.toString(); |
| } |
| |
| /* |
| * (non-Javadoc) |
| * |
| * @see org.apache.geode.internal.cache.TXStateInterface#getTransactionId() |
| */ |
| @Override |
| public TransactionId getTransactionId() { |
| return this.proxy.getTxId(); |
| } |
| |
| public void firePendingCallbacks() { |
| for (EntryEventImpl ee : getPendingCallbacks()) { |
| 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); |
| } |
| } |
| } |
| |
| public void freePendingCallbacks() { |
| for (EntryEventImpl ee : getPendingCallbacks()) { |
| ee.release(); |
| } |
| } |
| |
| public List<EntryEventImpl> getPendingCallbacks() { |
| return pendingCallbacks; |
| } |
| |
| |
| /* |
| * (non-Javadoc) |
| * |
| * @see |
| * org.apache.geode.internal.cache.TXStateInterface#readRegion(org.apache.geode.internal.cache. |
| * LocalRegion) |
| */ |
| @Override |
| public TXRegionState readRegion(InternalRegion r) { |
| return this.regions.get(r); |
| } |
| |
| @Override |
| public void rmRegion(LocalRegion r) { |
| TXRegionState txr = this.regions.remove(r); |
| if (txr != null) { |
| txr.cleanup(r); |
| } |
| } |
| |
| /* |
| * (non-Javadoc) |
| * |
| * @see |
| * org.apache.geode.internal.cache.TXStateInterface#writeRegion(org.apache.geode.internal.cache. |
| * LocalRegion) |
| */ |
| @Override |
| public TXRegionState writeRegion(InternalRegion r) { |
| TXRegionState result = readRegion(r); |
| if (result == null) { |
| if (r instanceof BucketRegion) { |
| result = new TXBucketRegionState((BucketRegion) r, this); |
| } else { |
| result = new TXRegionState(r, this); |
| } |
| this.regions.put(r, result); |
| } |
| if (logger.isDebugEnabled()) { |
| logger.debug("TXState writeRegion flag {} region-state {} ", false, |
| result/* , new Throwable() */); |
| } |
| return result; |
| } |
| |
| /* |
| * (non-Javadoc) |
| * |
| * @see org.apache.geode.internal.cache.TXStateInterface#getBeginTime() |
| */ |
| @Override |
| public long getBeginTime() { |
| return this.beginTime; |
| } |
| |
| /* |
| * (non-Javadoc) |
| * |
| * @see org.apache.geode.internal.cache.TXStateInterface#getChanges() |
| */ |
| @Override |
| public int getChanges() { |
| int changes = 0; |
| Iterator<TXRegionState> it = this.regions.values().iterator(); |
| while (it.hasNext()) { |
| TXRegionState txrs = it.next(); |
| changes += txrs.getChanges(); |
| } |
| return changes; |
| } |
| |
| /* |
| * (non-Javadoc) |
| * |
| * @see org.apache.geode.internal.cache.TXStateInterface#isInProgress() |
| */ |
| @Override |
| public boolean isInProgress() { |
| return !this.closed; |
| } |
| |
| /* |
| * (non-Javadoc) |
| * |
| * @see org.apache.geode.internal.cache.TXStateInterface#nextModSerialNum() |
| */ |
| @Override |
| public int nextModSerialNum() { |
| this.modSerialNum += 1; |
| return this.modSerialNum; |
| } |
| |
| /* |
| * (non-Javadoc) |
| * |
| * @see org.apache.geode.internal.cache.TXStateInterface#needsLargeModCount() |
| */ |
| @Override |
| public boolean needsLargeModCount() { |
| return this.modSerialNum > Byte.MAX_VALUE; |
| } |
| |
| protected void reserveAndCheck() throws CommitConflictException { |
| if (this.closed) { |
| return; |
| } |
| |
| final long conflictStart = CachePerfStats.getStatTime(); |
| this.locks = createLockRequest(); |
| this.locks.obtain(getCache().getInternalDistributedSystem()); |
| // for now check account the dlock service time |
| // later this stat end should be moved to a finally block |
| if (CachePerfStats.enableClockStats) |
| this.proxy.getTxMgr().getCachePerfStats() |
| .incTxConflictCheckTime(CachePerfStats.getStatTime() - conflictStart); |
| if (this.internalAfterReservation != null) { |
| this.internalAfterReservation.run(); |
| } |
| checkForConflicts(); |
| } |
| |
| byte[] getBaseMembershipId() { |
| return this.baseMembershipId; |
| } |
| |
| long getBaseThreadId() { |
| return this.baseThreadId; |
| } |
| |
| long getBaseSequenceId() { |
| return this.baseSequenceId; |
| } |
| |
| @Override |
| public void precommit() |
| throws CommitConflictException, UnsupportedOperationInTransactionException { |
| throw new UnsupportedOperationInTransactionException( |
| String.format("precommit() operation %s meant for Dist Tx is not supported", |
| "precommit")); |
| } |
| |
| /* |
| * (non-Javadoc) |
| * |
| * @see org.apache.geode.internal.cache.TXStateInterface#commit() |
| */ |
| @Override |
| public void commit() throws CommitConflictException { |
| if (this.closed) { |
| return; |
| } |
| |
| if (logger.isDebugEnabled()) { |
| logger.debug("committing transaction {}", getTransactionId()); |
| } |
| synchronized (this.completionGuard) { |
| this.completionStarted = true; |
| } |
| |
| if (onBehalfOfRemoteStub && !proxy.isCommitOnBehalfOfRemoteStub()) { |
| throw new UnsupportedOperationInTransactionException( |
| "Cannot commit a transaction being run on behalf of a remote thread"); |
| } |
| cleanupNonDirtyRegions(); |
| try { |
| /* |
| * Lock buckets so they can't be rebalanced then perform the conflict check to fix #43489 |
| */ |
| try { |
| lockBucketRegions(); |
| } catch (PrimaryBucketException pbe) { |
| // not sure what to do here yet |
| RuntimeException re = new TransactionDataRebalancedException( |
| "Transactional data moved, due to rebalancing."); |
| re.initCause(pbe); |
| throw re; |
| } |
| |
| if (this.locks == null) { |
| reserveAndCheck(); |
| } |
| |
| // For internal testing |
| if (this.internalAfterConflictCheck != null) { |
| this.internalAfterConflictCheck.run(); |
| } |
| |
| /* |
| * If there is a TransactionWriter plugged in, we need to to give it an opportunity to cleanup |
| * the transaction. |
| */ |
| TransactionWriter writer = this.proxy.getTxMgr().getWriter(); |
| if (!firedWriter && writer != null) { |
| try { |
| firedWriter = true; |
| TXEvent event = getEvent(); |
| if (!event.hasOnlyInternalEvents()) { |
| writer.beforeCommit(event); |
| } |
| } catch (TransactionWriterException twe) { |
| cleanup(); |
| throw new CommitConflictException(twe); |
| } catch (VirtualMachineError err) { |
| // cleanup(); this allocates objects so I don't think we can do it - that leaves the TX |
| // open, but we are poison pilling so we should be ok?? |
| |
| 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) { |
| cleanup(); // rollback the transaction! |
| // 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(); |
| throw new CommitConflictException(t); |
| } |
| } |
| |
| List/* <TXEntryStateWithRegionAndKey> */ entries = generateEventOffsets(); |
| TXCommitMessage msg = null; |
| try { |
| |
| /* |
| * In order to preserve data consistency, we need to: 1. Modify the cache first |
| * (applyChanges) 2. Ask for advice on who to send to (buildMessage) 3. Send out to other |
| * members. |
| * |
| * If this is done out of order, we will have problems with GII, split brain, and HA. |
| */ |
| |
| attachFilterProfileInformation(entries); |
| |
| lockTXRegions(regions); |
| |
| try { |
| // apply changes to the cache |
| applyChanges(entries); |
| // For internal testing |
| if (this.internalAfterApplyChanges != null) { |
| this.internalAfterApplyChanges.run(); |
| } |
| |
| // build and send the message |
| msg = buildMessage(); |
| this.commitMessage = msg; |
| if (this.internalBeforeSend != null) { |
| this.internalBeforeSend.run(); |
| } |
| |
| msg.send(this.locks.getDistributedLockId()); |
| // For internal testing |
| if (this.internalAfterSend != null) { |
| this.internalAfterSend.run(); |
| } |
| |
| firePendingCallbacks(); |
| /* |
| * This is to prepare the commit message for the caller, make sure all events are in |
| * there. |
| */ |
| this.commitMessage = buildCompleteMessage(); |
| } finally { |
| unlockTXRegions(regions); |
| } |
| } finally { |
| if (msg != null) { |
| msg.releaseViewVersions(); |
| } |
| this.locks.releaseLocal(); |
| // For internal testing |
| if (this.internalAfterReleaseLocalLocks != null) { |
| this.internalAfterReleaseLocalLocks.run(); |
| } |
| } |
| } finally { |
| cleanup(); |
| } |
| } |
| |
| private void lockTXRegions(IdentityHashMap<InternalRegion, TXRegionState> regions) { |
| Iterator<Map.Entry<InternalRegion, TXRegionState>> it = regions.entrySet().iterator(); |
| while (it.hasNext()) { |
| Map.Entry<InternalRegion, TXRegionState> me = it.next(); |
| InternalRegion r = me.getKey(); |
| r.getRegionMap().lockRegionForAtomicTX(r); |
| } |
| } |
| |
| private void unlockTXRegions(IdentityHashMap<InternalRegion, TXRegionState> regions) { |
| Iterator<Map.Entry<InternalRegion, TXRegionState>> it = regions.entrySet().iterator(); |
| while (it.hasNext()) { |
| Map.Entry<InternalRegion, TXRegionState> me = it.next(); |
| InternalRegion r = me.getKey(); |
| r.getRegionMap().unlockRegionForAtomicTX(r); |
| } |
| } |
| |
| protected void attachFilterProfileInformation(List entries) { |
| { |
| Iterator/* <TXEntryStateWithRegionAndKey> */ it = entries.iterator(); |
| while (it.hasNext()) { |
| TXEntryStateWithRegionAndKey o = (TXEntryStateWithRegionAndKey) it.next(); |
| try { |
| if (o.r.isUsedForPartitionedRegionBucket()) { |
| BucketRegion bucket = (BucketRegion) o.r; |
| /* |
| * The event must contain the bucket region |
| */ |
| @Released |
| EntryEventImpl ev = |
| (EntryEventImpl) o.es.getEvent(o.r, o.key, o.es.getTXRegionState().getTXState()); |
| try { |
| /* |
| * The routing information is derived from the PR advisor, not the bucket advisor. |
| */ |
| FilterRoutingInfo fri = bucket.getPartitionedRegion().getRegionAdvisor() |
| .adviseFilterRouting(ev, Collections.EMPTY_SET); |
| o.es.setFilterRoutingInfo(fri); |
| Set set = bucket.getAdjunctReceivers(ev, Collections.EMPTY_SET, new HashSet(), fri); |
| o.es.setAdjunctRecipients(set); |
| } finally { |
| ev.release(); |
| } |
| } |
| } catch (RegionDestroyedException ex) { |
| // region was destroyed out from under us; after conflict checking |
| // passed. So act as if the region destroy happened right after the |
| // commit. We act this way by doing nothing; including distribution |
| // of this region's commit data. |
| } catch (CancelException ex) { |
| // cache was closed out from under us; after conflict checking |
| // passed. So do nothing. |
| } |
| } |
| } |
| |
| } |
| |
| /* |
| * (non-Javadoc) |
| * |
| * @see org.apache.geode.internal.cache.TXStateInterface#rollback() |
| */ |
| @Override |
| public void rollback() { |
| if (this.closed) { |
| return; |
| } |
| synchronized (this.completionGuard) { |
| this.completionStarted = true; |
| } |
| cleanup(); |
| } |
| |
| /** |
| * This is a fix for bug #42228 where a client fails over from one server to another but gets a |
| * conflict on completion because completion had already been initiated and had not yet completed |
| * |
| * @return true if a previous completion was in progress |
| */ |
| public boolean waitForPreviousCompletion() { |
| synchronized (this.completionGuard) {// should have already been done, but just to be sure |
| if (!this.completionStarted) { |
| return false; |
| } |
| while (this.commitMessage == null && !this.closed) { |
| if (logger.isDebugEnabled()) { |
| logger.debug("Waiting for previous completion for transaction {}", getTransactionId()); |
| } |
| try { |
| this.completionGuard.wait(); |
| } catch (InterruptedException e) { |
| this.proxy.getCache().getCancelCriterion().checkCancelInProgress(e); |
| Thread.currentThread().interrupt(); |
| return true; |
| } |
| } // while |
| } |
| return true; |
| } |
| |
| /** |
| * Generate an event id for each operation that will be done by this tx during the application |
| * phase of its commit. |
| * |
| * @return a sorted list of TXEntryStateWithRegionAndKey that will be used to apply the ops on the |
| * nearside in the correct order. |
| */ |
| protected List/* <TXEntryStateWithRegionAndKey> */ generateEventOffsets() { |
| this.baseMembershipId = EventID.getMembershipId(this.proxy.getTxMgr().getDM().getSystem()); |
| this.baseThreadId = EventID.getThreadId(); |
| this.baseSequenceId = EventID.getSequenceId(); |
| |
| List/* <TXEntryStateWithRegionAndKey> */ entries = getSortedEntries(); |
| if (logger.isDebugEnabled()) { |
| logger |
| .debug("generateEventOffsets() entries " + entries + " RegionState Map=" + this.regions); |
| } |
| Iterator it = entries.iterator(); |
| while (it.hasNext()) { |
| TXEntryStateWithRegionAndKey o = (TXEntryStateWithRegionAndKey) it.next(); |
| o.es.generateEventOffsets(this); |
| } |
| return entries; |
| } |
| |
| private TXLockRequest createLockRequest() { |
| TXLockRequest result = new TXLockRequest(); |
| Iterator<Map.Entry<InternalRegion, TXRegionState>> it = this.regions.entrySet().iterator(); |
| while (it.hasNext()) { |
| Map.Entry<InternalRegion, TXRegionState> me = it.next(); |
| InternalRegion r = me.getKey(); |
| TXRegionState txrs = me.getValue(); |
| txrs.createLockRequest(r, result); |
| } |
| return result; |
| } |
| |
| private void checkForConflicts() throws CommitConflictException, PrimaryBucketException { |
| Iterator<Map.Entry<InternalRegion, TXRegionState>> it = this.regions.entrySet().iterator(); |
| while (it.hasNext()) { |
| Map.Entry<InternalRegion, TXRegionState> me = it.next(); |
| InternalRegion r = me.getKey(); |
| TXRegionState txrs = me.getValue(); |
| try { |
| txrs.checkForConflicts(r); |
| } catch (DiskAccessException dae) { |
| r.handleDiskAccessException(dae); |
| throw dae; |
| } |
| } |
| } |
| |
| |
| protected void lockBucketRegions() throws PrimaryBucketException { |
| |
| boolean lockingSucceeded; |
| do { |
| lockingSucceeded = true; |
| Iterator<Map.Entry<InternalRegion, TXRegionState>> it = this.regions.entrySet().iterator(); |
| Set<BucketRegion> obtained = new HashSet<BucketRegion>(); |
| while (it.hasNext()) { |
| Map.Entry<InternalRegion, TXRegionState> me = it.next(); |
| InternalRegion r = me.getKey(); |
| if (r instanceof BucketRegion) { |
| if (isDistTx() && !((BucketRegion) r).getBucketAdvisor().isPrimary()) { |
| // For distTx we skip for taking locks on secondary. |
| continue; |
| } |
| BucketRegion b = (BucketRegion) r; |
| /* |
| * Lock the primary bucket so it doesnt get rebalanced until we cleanup! |
| */ |
| boolean lockObtained = false; |
| try { |
| // use tryLocks to avoid hanging (bug #41708) |
| boolean locked = b.doLockForPrimary(true); |
| if (locked) { |
| obtained.add(b); |
| lockObtained = true; |
| } else { |
| // if we can't get locks then someone has a write-lock. To prevent |
| // deadlock (see bug #41708) we release locks and re-acquire them |
| r.getCancelCriterion().checkCancelInProgress(null); |
| if (logger.isDebugEnabled()) { |
| logger.debug("tryLock failed for commit on {}. Releasing locks and retrying", |
| r.getFullPath()); |
| } |
| // release locks and start over |
| break; |
| } |
| } catch (RegionDestroyedException rde) { |
| if (logger.isDebugEnabled()) { |
| logger.debug("RegionDestroyedException while locking bucket region {}", |
| r.getFullPath(), rde); |
| } |
| throw new TransactionDataRebalancedException( |
| "Bucket rebalanced during commit: " + r.getFullPath()); |
| } finally { |
| if (!lockObtained) { |
| // fix for bug #41708 - unlock operation-locks already obtained |
| if (logger.isDebugEnabled()) { |
| logger.debug("Unexpected exception while locking bucket {}", r.getFullPath()); |
| } |
| for (BucketRegion br : obtained) { |
| br.doUnlockForPrimary(); |
| } |
| try { |
| Thread.sleep(50); |
| } catch (InterruptedException e) { |
| Thread.currentThread().interrupt(); |
| return; |
| } |
| lockingSucceeded = false; |
| } |
| } |
| } |
| } |
| } while (!lockingSucceeded); |
| gotBucketLocks = true; |
| } |
| |
| |
| protected void cleanupNonDirtyRegions() { |
| Iterator<Map.Entry<InternalRegion, TXRegionState>> it = this.regions.entrySet().iterator(); |
| while (it.hasNext()) { |
| Map.Entry<InternalRegion, TXRegionState> me = it.next(); |
| InternalRegion r = me.getKey(); |
| TXRegionState txrs = me.getValue(); |
| txrs.cleanupNonDirtyEntries(r); |
| } |
| } |
| |
| /** |
| * this builds a new TXCommitMessage and returns it |
| * |
| * @return the new message |
| */ |
| protected TXCommitMessage buildMessage() { |
| TXCommitMessage msg = |
| new TXCommitMessage(this.proxy.getTxId(), this.proxy.getTxMgr().getDM(), this); |
| Iterator<Map.Entry<InternalRegion, TXRegionState>> it = this.regions.entrySet().iterator(); |
| while (it.hasNext()) { |
| Map.Entry<InternalRegion, TXRegionState> me = it.next(); |
| InternalRegion r = me.getKey(); |
| TXRegionState txrs = me.getValue(); |
| txrs.buildMessage(r, msg); |
| } |
| return msg; |
| } |
| |
| |
| /** |
| * this builds a new TXCommitMessage and returns it |
| * |
| * @return the new message |
| */ |
| protected TXCommitMessage buildCompleteMessage() { |
| TXCommitMessage msg = |
| new TXCommitMessage(this.proxy.getTxId(), this.proxy.getTxMgr().getDM(), this); |
| Iterator<Map.Entry<InternalRegion, TXRegionState>> it = this.regions.entrySet().iterator(); |
| while (it.hasNext()) { |
| Map.Entry<InternalRegion, TXRegionState> me = it.next(); |
| InternalRegion r = me.getKey(); |
| TXRegionState txrs = me.getValue(); |
| txrs.buildCompleteMessage(r, msg); |
| // rcl.add(r); |
| } |
| return msg; |
| } |
| |
| /** |
| * applies this transaction to the cache. |
| */ |
| protected void applyChanges(List/* <TXEntryStateWithRegionAndKey> */ entries) { |
| // applyChangesStart for each region |
| for (Map.Entry<InternalRegion, TXRegionState> me : this.regions.entrySet()) { |
| InternalRegion r = me.getKey(); |
| TXRegionState txrs = me.getValue(); |
| txrs.applyChangesStart(r, this); |
| } |
| |
| // serializePendingValue for each entry |
| for (Object entry : entries) { |
| TXEntryStateWithRegionAndKey o = (TXEntryStateWithRegionAndKey) entry; |
| o.es.serializePendingValue(); |
| } |
| |
| // applyChanges for each entry |
| for (Object entry : entries) { |
| TXEntryStateWithRegionAndKey o = (TXEntryStateWithRegionAndKey) entry; |
| if (this.internalDuringApplyChanges != null) { |
| this.internalDuringApplyChanges.run(); |
| } |
| try { |
| o.es.applyChanges(o.r, o.key, this); |
| } catch (RegionDestroyedException ex) { |
| // region was destroyed out from under us; after conflict checking |
| // passed. So act as if the region destroy happened right after the |
| // commit. We act this way by doing nothing; including distribution |
| // of this region's commit data. |
| } catch (CancelException ex) { |
| // cache was closed out from under us; after conflict checking |
| // passed. So do nothing. |
| } |
| } |
| |
| // applyChangesEnd for each region |
| for (Map.Entry<InternalRegion, TXRegionState> me : this.regions.entrySet()) { |
| InternalRegion r = me.getKey(); |
| TXRegionState txrs = me.getValue(); |
| txrs.applyChangesEnd(r, this); |
| } |
| } |
| |
| @Override |
| public TXEvent getEvent() { |
| return new TXEvent(this, getCache()); |
| } |
| |
| /** |
| * Note that cleanup does more than is needed in this method. This method only needs to do stuff |
| * that is required when a Cache close is done and we have txs that are still in progress. |
| * Currently the only thing that is needed is to decrement off-heap refcounts since off-heap |
| * memory lives after a cache close. |
| */ |
| @Override |
| public void close() { |
| if (!this.closed) { |
| if (locks != null) { |
| cleanup(); |
| return; |
| } |
| this.closed = true; |
| for (TXRegionState r : this.regions.values()) { |
| r.close(); |
| } |
| } |
| } |
| |
| protected void cleanup() { |
| if (singleThreadJTAExecutor.shouldDoCleanup()) { |
| singleThreadJTAExecutor.cleanup(); |
| } else { |
| doCleanup(); |
| } |
| } |
| |
| void doCleanup() { |
| RuntimeException exception = null; |
| try { |
| this.closed = true; |
| this.seenEvents.clear(); |
| this.seenResults.clear(); |
| freePendingCallbacks(); |
| if (this.locks != null) { |
| final long conflictStart = CachePerfStats.getStatTime(); |
| try { |
| this.locks.cleanup(getCache().getInternalDistributedSystem()); |
| } catch (IllegalArgumentException | IllegalMonitorStateException e) { |
| exception = e; |
| } |
| if (CachePerfStats.enableClockStats) |
| this.proxy.getTxMgr().getCachePerfStats() |
| .incTxConflictCheckTime(CachePerfStats.getStatTime() - conflictStart); |
| } |
| Iterator<Map.Entry<InternalRegion, TXRegionState>> it = this.regions.entrySet().iterator(); |
| while (it.hasNext()) { |
| Map.Entry<InternalRegion, TXRegionState> me = it.next(); |
| InternalRegion r = me.getKey(); |
| TXRegionState txrs = me.getValue(); |
| /* |
| * Need to unlock the primary lock for rebalancing so that rebalancing can resume. |
| */ |
| if (gotBucketLocks) { |
| if (r instanceof BucketRegion && (((BucketRegion) r).getBucketAdvisor().isPrimary())) { |
| try { |
| ((BucketRegion) r).doUnlockForPrimary(); |
| } catch (RegionDestroyedException rde) { |
| // ignore |
| if (logger.isDebugEnabled()) { |
| logger.debug("RegionDestroyedException while unlocking bucket region {}", |
| r.getFullPath(), rde); |
| } |
| } catch (Exception rde) { |
| // ignore |
| if (logger.isDebugEnabled()) { |
| logger.debug( |
| "Exception while unlocking bucket region {} this is probably because the bucket was destroyed and never locked initially.", |
| r.getFullPath(), rde); |
| } |
| } |
| } |
| } |
| txrs.cleanup(r); |
| } |
| } finally { |
| synchronized (this.completionGuard) { |
| this.completionGuard.notifyAll(); |
| } |
| |
| if (exception != null && !this.proxy.getCache().isClosed()) { |
| throw exception; |
| } |
| } |
| } |
| |
| /* |
| * (non-Javadoc) |
| * |
| * @see org.apache.geode.internal.cache.TXStateInterface#getEvents() |
| */ |
| @Override |
| public List getEvents() { |
| ArrayList events = new ArrayList(); |
| Iterator<Map.Entry<InternalRegion, TXRegionState>> it = this.regions.entrySet().iterator(); |
| while (it.hasNext()) { |
| Map.Entry me = it.next(); |
| InternalRegion r = (InternalRegion) me.getKey(); |
| TXRegionState txrs = (TXRegionState) me.getValue(); |
| txrs.getEvents(r, events, this); |
| } |
| if (events.isEmpty()) { |
| return Collections.EMPTY_LIST; |
| } else { |
| Collections.sort(events); |
| return Collections.unmodifiableList(events); |
| } |
| } |
| |
| private List/* <TXEntryStateWithRegionAndKey> */ getSortedEntries() { |
| ArrayList/* <TXEntryStateWithRegionAndKey> */ entries = new ArrayList(); |
| Iterator it = this.regions.entrySet().iterator(); |
| while (it.hasNext()) { |
| Map.Entry me = (Map.Entry) it.next(); |
| InternalRegion r = (InternalRegion) me.getKey(); |
| TXRegionState txrs = (TXRegionState) me.getValue(); |
| txrs.getEntries(entries, r); |
| } |
| if (entries.isEmpty()) { |
| return Collections.EMPTY_LIST; |
| } else { |
| Collections.sort(entries); |
| return entries; |
| } |
| } |
| |
| /** |
| * Used to keep track of the region and key associated with a TXEntryState. Also used to sort the |
| * entries into the order in which they will be applied. |
| * |
| * @since GemFire 5.7 |
| */ |
| static class TXEntryStateWithRegionAndKey implements Comparable { |
| public final TXEntryState es; |
| public final InternalRegion r; |
| public final Object key; |
| |
| public TXEntryStateWithRegionAndKey(TXEntryState es, InternalRegion r, Object key) { |
| this.es = es; |
| this.r = r; |
| this.key = key; |
| } |
| |
| private int getSortValue() { |
| return this.es.getSortValue(); |
| } |
| |
| @Override |
| public int compareTo(Object o) { |
| TXEntryStateWithRegionAndKey other = (TXEntryStateWithRegionAndKey) o; |
| return getSortValue() - other.getSortValue(); |
| } |
| |
| @Override |
| public boolean equals(Object o) { |
| if (o == null || !(o instanceof TXEntryStateWithRegionAndKey)) |
| return false; |
| return compareTo(o) == 0; |
| } |
| |
| @Override |
| public int hashCode() { |
| return getSortValue(); |
| } |
| } |
| |
| ////////////////////////////////////////////////////////////////// |
| // JTA Synchronization implementation // |
| ////////////////////////////////////////////////////////////////// |
| /* |
| * (non-Javadoc) |
| * |
| * @see org.apache.geode.internal.cache.TXStateInterface#beforeCompletion() |
| */ |
| @Override |
| public synchronized void beforeCompletion() throws SynchronizationCommitConflictException { |
| proxy.getTxMgr().setTXState(null); |
| if (this.closed) { |
| throw new TXManagerCancelledException(); |
| } |
| |
| if (beforeCompletionCalled) { |
| // do not re-execute beforeCompletion again |
| return; |
| } |
| beforeCompletionCalled = true; |
| singleThreadJTAExecutor.executeBeforeCompletion(this, |
| getExecutor(), getCancelCriterion()); |
| } |
| |
| private Executor getExecutor() { |
| return getCache().getDistributionManager().getWaitingThreadPool(); |
| } |
| |
| private CancelCriterion getCancelCriterion() { |
| return getCache().getCancelCriterion(); |
| } |
| |
| void doBeforeCompletion() { |
| final long opStart = CachePerfStats.getStatTime(); |
| this.jtaLifeTime = opStart - getBeginTime(); |
| |
| try { |
| reserveAndCheck(); |
| /* |
| * If there is a TransactionWriter plugged in, we need to to give it an opportunity to cleanup |
| * the transaction. |
| */ |
| TransactionWriter writer = this.proxy.getTxMgr().getWriter(); |
| if (writer != null) { |
| try { |
| // need to mark this so we don't fire again in commit |
| firedWriter = true; |
| TXEvent event = getEvent(); |
| if (!event.hasOnlyInternalEvents()) { |
| writer.beforeCommit(event); |
| } |
| } catch (TransactionWriterException twe) { |
| throw new CommitConflictException(twe); |
| } catch (VirtualMachineError err) { |
| // cleanup(); this allocates objects so I don't think we can do it - that leaves the TX |
| // open, but we are poison pilling so we should be ok?? |
| |
| 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(); |
| throw new CommitConflictException(t); |
| } |
| } |
| } catch (CommitConflictException commitConflict) { |
| cleanup(); |
| proxy.getTxMgr().noteCommitFailure(opStart, this.jtaLifeTime, this); |
| throw new SynchronizationCommitConflictException( |
| String.format("Conflict detected in GemFire transaction %s", |
| getTransactionId()), |
| commitConflict); |
| } |
| } |
| |
| /* |
| * (non-Javadoc) |
| * |
| * @see org.apache.geode.internal.cache.TXStateInterface#afterCompletion(int) |
| */ |
| @Override |
| public synchronized void afterCompletion(int status) { |
| proxy.getTxMgr().setTXState(null); |
| // if there was a beforeCompletion call then there will be a thread |
| // sitting in the waiting pool to execute afterCompletion. Otherwise |
| // throw FailedSynchronizationException(). |
| if (wasBeforeCompletionCalled()) { |
| switch (status) { |
| case Status.STATUS_COMMITTED: |
| singleThreadJTAExecutor.executeAfterCompletionCommit(); |
| break; |
| case Status.STATUS_ROLLEDBACK: |
| singleThreadJTAExecutor.executeAfterCompletionRollback(); |
| break; |
| default: |
| throw new TransactionException("Unknown JTA Synchronization status " + status); |
| } |
| } else { |
| // rollback does not run beforeCompletion. |
| if (status != Status.STATUS_ROLLEDBACK) { |
| throw new FailedSynchronizationException( |
| "Could not execute afterCompletion when beforeCompletion was not executed"); |
| } |
| doAfterCompletionRollback(); |
| } |
| } |
| |
| void doAfterCompletionCommit() { |
| final long opStart = CachePerfStats.getStatTime(); |
| try { |
| Assert.assertTrue(this.locks != null, |
| "Gemfire Transaction afterCompletion called with illegal state."); |
| try { |
| commit(); |
| saveTXCommitMessageForClientFailover(); |
| } catch (CommitConflictException error) { |
| Assert.assertTrue(false, "Gemfire Transaction " + getTransactionId() |
| + " afterCompletion failed.due to CommitConflictException: " + error); |
| } |
| this.proxy.getTxMgr().noteCommitSuccess(opStart, this.jtaLifeTime, this); |
| this.locks = null; |
| |
| } catch (InternalGemFireError error) { |
| throw new TransactionException(error); |
| } |
| } |
| |
| void doAfterCompletionRollback() { |
| final long opStart = CachePerfStats.getStatTime(); |
| this.jtaLifeTime = opStart - getBeginTime(); |
| try { |
| rollback(); |
| saveTXCommitMessageForClientFailover(); |
| this.proxy.getTxMgr().noteRollbackSuccess(opStart, this.jtaLifeTime, this); |
| } catch (InternalGemFireError error) { |
| throw new TransactionException(error); |
| } |
| } |
| |
| boolean wasBeforeCompletionCalled() { |
| return beforeCompletionCalled; |
| } |
| |
| void saveTXCommitMessageForClientFailover() { |
| proxy.getTxMgr().saveTXStateForClientFailover(proxy); |
| } |
| |
| |
| /** |
| * Add an internal callback which is run after the reservation/lock is returned from the Grantor |
| * but before the local identity/conflict check. This is the first callback to be called during |
| * the commit. |
| */ |
| public void setAfterReservation(Runnable afterReservation) { |
| this.internalAfterReservation = afterReservation; |
| } |
| |
| /** |
| * Add an internal callback which is run after the local identity/conflict check has completed but |
| * before the changes have been applied to committed state. |
| */ |
| public void setAfterConflictCheck(Runnable afterConflictCheck) { |
| this.internalAfterConflictCheck = afterConflictCheck; |
| } |
| |
| /** |
| * Add an internal callback which is run as each transaction change is applied. |
| */ |
| public void setDuringApplyChanges(Runnable duringApplyChanges) { |
| this.internalDuringApplyChanges = duringApplyChanges; |
| } |
| |
| /** |
| * Add an internal callback which is run after the transaction changes have been applied to |
| * committed state (locally) but before local locks are released (occurs for regions of Local and |
| * Distributed No Ack scope). |
| */ |
| public void setAfterApplyChanges(Runnable afterApplyChanges) { |
| this.internalAfterApplyChanges = afterApplyChanges; |
| } |
| |
| /** |
| * Add an internal callback which is run after the the local locks are released (which occurs for |
| * regions of Local and Distributed No Ack scope) but before commit data is sent to recipients aka |
| * Far Siders (only for Distributed Scope regions). |
| */ |
| public void setAfterReleaseLocalLocks(Runnable afterReleaseLocalLocks) { |
| this.internalAfterReleaseLocalLocks = afterReleaseLocalLocks; |
| } |
| |
| /** |
| * Add an internal callback which is run once for each recipient (aka Far Sider) of commit data, |
| * prior to actually sending the data. This is called prior to calling |
| * <code>setAfterIndividualSend</code>. |
| */ |
| public void setDuringIndividualSend(Runnable duringIndividualSend) { |
| this.internalDuringIndividualSend = duringIndividualSend; |
| } |
| |
| /** |
| * Add an internal callback which is run once after all the commit data has been sent to each |
| * recipient but before the "commit process" message is sent (only sent in the case there regions |
| * with Distributed Ack scope) |
| */ |
| public void setAfterIndividualSend(Runnable afterIndividualSend) { |
| this.internalAfterIndividualSend = afterIndividualSend; |
| } |
| |
| /** |
| * Add an internal callback which is run once for each recipient (aka Far Sider) of the "commit |
| * process" message (only for recipients with Distributed Ack regions), prior to actually sending |
| * the message. |
| */ |
| public void setDuringIndividualCommitProcess(Runnable duringIndividualCommitProcess) { |
| this.internalDuringIndividualCommitProcess = duringIndividualCommitProcess; |
| } |
| |
| /** |
| * Add an internal callback which is run once after all the "commit process" messages (only for |
| * recipients with Distributed Ack regions) have been sent but before <code>setAfterSend</code> |
| * callback has been called. |
| */ |
| public void setAfterIndividualCommitProcess(Runnable afterIndividualCommitProcess) { |
| this.internalAfterIndividualCommitProcess = afterIndividualCommitProcess; |
| } |
| |
| |
| /** |
| * Add an internal callback which is run after all data has been sent (for Distributed scope |
| * regions) and any acknowledgements have been received (for Distributed Ack scope regions) a but |
| * before the transaction has been cleaned up. |
| */ |
| public void setAfterSend(Runnable afterSend) { |
| this.internalAfterSend = afterSend; |
| } |
| |
| /** |
| * Add an internal callback which is run after the commit message is formed but before it is sent. |
| */ |
| public void setBeforeSend(Runnable r) { |
| this.internalBeforeSend = r; |
| } |
| |
| |
| /* |
| * (non-Javadoc) |
| * |
| * @see org.apache.geode.internal.cache.TXStateInterface#getCache() |
| */ |
| @Override |
| public InternalCache getCache() { |
| return this.proxy.getCache(); |
| } |
| |
| /* |
| * (non-Javadoc) |
| * |
| * @see org.apache.geode.internal.cache.TXStateInterface#getRegions() |
| */ |
| @Override |
| public Collection<InternalRegion> getRegions() { |
| return this.regions.keySet(); |
| } |
| |
| @Override |
| public TXRegionState txWriteRegion(final InternalRegion internalRegion, final KeyInfo entryKey) { |
| InternalRegion ir = internalRegion.getDataRegionForWrite(entryKey); |
| return writeRegion(ir); |
| } |
| |
| @Override |
| public TXRegionState txReadRegion(InternalRegion internalRegion) { |
| return readRegion(internalRegion); |
| } |
| |
| /** |
| * @param ifNew only write the entry if it currently does not exist |
| * @param requireOldValue if true set the old value in the event, even if ifNew and entry doesn't |
| * currently exist (this is needed for putIfAbsent). |
| * @param expectedOldValue the required old value or null |
| */ |
| TXEntryState txWriteEntry(InternalRegion region, EntryEventImpl event, boolean ifNew, |
| boolean requireOldValue, Object expectedOldValue) throws EntryNotFoundException { |
| boolean createIfAbsent = true; |
| if (event.getOperation() == Operation.REPLACE) { |
| // replace(K,V) and replace(K,V,V) cannot create an entry |
| createIfAbsent = false; |
| } |
| TXEntryState tx = |
| txReadEntry(event.getKeyInfo(), region, true, expectedOldValue, createIfAbsent); |
| if (tx != null) { |
| if (requireOldValue && tx.existsLocally()) { |
| event.setOldValue(tx.getNearSidePendingValue(), true); |
| } |
| boolean existsLocally = tx.existsLocally(); |
| if (!existsLocally && event.getOperation() == Operation.REPLACE) { |
| throw new EntryNotFoundException("No previously created Entry to be updated"); |
| } |
| if (existsLocally && ifNew) { |
| // Since "ifNew" is true then let caller know entry exists |
| // in tx state or cmt state |
| return ENTRY_EXISTS; |
| } else { |
| tx.updateForWrite(nextModSerialNum()); |
| } |
| } else { |
| if (!createIfAbsent) { |
| throw new EntryNotFoundException("No previously created Entry to be updated"); |
| } |
| } |
| return tx; |
| } |
| |
| /** |
| * this version of txPutEntry takes a ConcurrentMap expectedOldValue parameter. If not null, this |
| * value must match the current value of the entry or false is returned |
| */ |
| @Override |
| public boolean txPutEntry(final EntryEventImpl event, boolean ifNew, boolean requireOldValue, |
| boolean checkResources, Object expectedOldValue) { |
| |
| InternalRegion region = event.getRegion(); |
| if (checkResources) { |
| if (!MemoryThresholds.isLowMemoryExceptionDisabled()) { |
| region.checkIfAboveThreshold(event); |
| } |
| } |
| |
| if (bridgeContext == null) { |
| bridgeContext = event.getContext(); |
| } |
| |
| if (hasSeenEvent(event)) { |
| return getRecordedResult(event); |
| } |
| |
| // if requireOldValue then oldValue gets set in event |
| // (even if ifNew and entry exists) |
| // !!!:ezoerner:20080813 need to handle ifOld for transactional on |
| // PRs when PRs become transactional |
| TXEntryState tx = null; |
| boolean result = false; |
| try { |
| tx = txWriteEntry(region, event, ifNew, requireOldValue, expectedOldValue); |
| if (tx == TXState.ENTRY_EXISTS) { |
| result = false; |
| } else { |
| result = tx.basicPut(event, ifNew, isOriginRemoteForEvents()); |
| } |
| } catch (EntryNotFoundException e) { |
| result = false; |
| } finally { |
| recordEventAndResult(event, result); |
| } |
| return result; |
| } |
| |
| |
| /* |
| * (non-Javadoc) |
| * |
| * @see org.apache.geode.internal.cache.TXStateInterface#containsValueForKey(java.lang.Object, |
| * org.apache.geode.internal.cache.LocalRegion) |
| */ |
| @Override |
| public boolean containsValueForKey(KeyInfo keyInfo, LocalRegion region) { |
| TXEntryState tx = txReadEntry(keyInfo, region, true, true/* create txEntry is absent */); |
| if (tx != null) { |
| /* |
| * Note that we don't consult this.getDataPolicy().isProxy() when setting this because in this |
| * context we don't want proxies to pretend they have a value. |
| */ |
| boolean isProxy = false; |
| return tx.isLocallyValid(isProxy); |
| } else { |
| return region.nonTXContainsValueForKey(keyInfo); |
| } |
| } |
| |
| /* |
| * (non-Javadoc) |
| * |
| * @see |
| * org.apache.geode.internal.cache.TXStateInterface#destroyExistingEntry(org.apache.geode.internal |
| * .cache.EntryEventImpl, boolean, java.lang.Object) |
| */ |
| @Override |
| public void destroyExistingEntry(final EntryEventImpl event, final boolean cacheWrite, |
| Object expectedOldValue) { |
| if (bridgeContext == null) { |
| bridgeContext = event.getContext(); |
| } |
| if (hasSeenEvent(event)) { |
| return; |
| } |
| TXEntryState tx = txWriteExistingEntry(event, expectedOldValue); |
| final InternalRegion region = event.getRegion(); |
| if (tx.destroy(event, cacheWrite, isOriginRemoteForEvents())) { |
| Object key = event.getKey(); |
| LocalRegion rr = region.getDataRegionForRead(event.getKeyInfo()); |
| txReadRegion(rr).rmEntryUserAttr(key); |
| recordEvent(event); |
| } |
| } |
| |
| /* |
| * (non-Javadoc) |
| * |
| * @see org.apache.geode.internal.cache.TXStateInterface#invalidateExistingEntry(org.apache.geode. |
| * internal.cache.EntryEventImpl, boolean, boolean) |
| */ |
| @Override |
| public void invalidateExistingEntry(final EntryEventImpl event, boolean invokeCallbacks, |
| boolean forceNewEntry) { |
| if (bridgeContext == null) { |
| bridgeContext = event.getContext(); |
| } |
| if (hasSeenEvent(event)) { |
| return; |
| } |
| TXEntryState tx = txWriteExistingEntry(event, null); |
| assert invokeCallbacks && !forceNewEntry; |
| tx.invalidate(event); |
| recordEvent(event); |
| } |
| |
| /** |
| * Write an existing entry. This form takes an expectedOldValue which, if not null, must be equal |
| * to the current value of the entry. If it is not, an EntryNotFoundException is thrown. |
| * |
| * @return the tx entry object |
| */ |
| private TXEntryState txWriteExistingEntry(final EntryEventImpl event, Object expectedOldValue) |
| throws EntryNotFoundException { |
| assert !event.isExpiration(); |
| final Object entryKey = event.getKey(); |
| final InternalRegion region = event.getRegion(); |
| final Operation op = event.getOperation(); |
| TXEntryState tx = txReadEntry(event.getKeyInfo(), region, true, expectedOldValue, |
| true/* create txEntry is absent */); |
| assert tx != null; |
| if (tx.existsLocally()) { |
| final boolean invalidatingInvalidEntry = |
| op.isInvalidate() && Token.isInvalid(tx.getValueInVM(entryKey)); |
| // Ignore invalidating an invalid entry |
| if (!invalidatingInvalidEntry) { |
| tx.updateForWrite(nextModSerialNum()); |
| } |
| } else if (region.isProxy() && !op.isLocal() && !tx.hasOp()) { |
| // Distributed operations on proxy regions need to be done |
| // even if the entry does not exist locally. |
| // But only if we don't already have a tx operation (once we have an op |
| // then we honor tx.existsLocally since the tx has storage unlike the proxy). |
| // We must not throw EntryNotFoundException in this case |
| tx.updateForWrite(nextModSerialNum()); |
| } else { |
| throw new EntryNotFoundException(entryKey.toString()); |
| } |
| return tx; |
| } |
| |
| /* |
| * (non-Javadoc) |
| * |
| * @see org.apache.geode.internal.cache.TXStateInterface#getEntry(java.lang.Object, |
| * org.apache.geode.internal.cache.LocalRegion) |
| */ |
| @Override |
| public Entry getEntry(final KeyInfo keyInfo, final LocalRegion region, boolean allowTombstones) { |
| TXEntryState tx = txReadEntry(keyInfo, region, true, true/* create txEntry is absent */); |
| if (tx != null && tx.existsLocally()) { |
| return new TXEntry(region, keyInfo, getProxy()); |
| } else { |
| return null; |
| } |
| } |
| |
| @Override |
| public Entry accessEntry(KeyInfo keyInfo, LocalRegion localRegion) { |
| return getEntry(keyInfo, localRegion, false); |
| } |
| |
| private TXStateInterface getProxy() { |
| return this.proxy; |
| } |
| |
| /** |
| * @param rememberRead true if the value read from committed state needs to be remembered in tx |
| * state for repeatable read. |
| * @param createIfAbsent should a transactional entry be created if not present. |
| * @return a txEntryState or null if the entry doesn't exist in the transaction and/or committed |
| * state. |
| */ |
| @Override |
| public TXEntryState txReadEntry(KeyInfo keyInfo, LocalRegion localRegion, boolean rememberRead, |
| boolean createIfAbsent) { |
| localRegion.cache.getCancelCriterion().checkCancelInProgress(null); |
| return txReadEntry(keyInfo, localRegion, rememberRead, null, createIfAbsent); |
| } |
| |
| /** |
| * This form of txReadEntry takes a concurrent-map argument, expectedOldValue. If this parameter |
| * is not null it must match the current value of the entry or an EntryNotFoundException is |
| * thrown. |
| */ |
| protected TXEntryState txReadEntry(KeyInfo keyInfo, InternalRegion internalRegion, |
| boolean rememberRead, Object expectedOldValue, boolean createIfAbsent) |
| throws EntryNotFoundException { |
| InternalRegion dataReg = internalRegion.getDataRegionForWrite(keyInfo); |
| TXRegionState txr = txReadRegion(dataReg); |
| TXEntryState result = null; |
| if (txr != null) { |
| result = txr.readEntry(keyInfo.getKey()); |
| } |
| if (result == null && rememberRead) { |
| // to support repeatable read create an tx entry that reflects current committed state |
| if (txr == null) { |
| txr = txWriteRegion(internalRegion, keyInfo); |
| } |
| result = dataReg.createReadEntry(txr, keyInfo, createIfAbsent); |
| if (result == null) { |
| // createReadEntry will only returns null if createIfAbsent is false. |
| // CreateIfAbsent will only be false when this method is called by set operations. |
| // In that case we do not want the TXState to have a TXEntryState. |
| assert !createIfAbsent; |
| return result; |
| } |
| } |
| |
| if (result != null) { |
| if (expectedOldValue != null) { |
| Object val = result.getNearSidePendingValue(); |
| if (!AbstractRegionEntry.checkExpectedOldValue(expectedOldValue, val, internalRegion)) { |
| throw new EntryNotFoundException( |
| "The current value was not equal to expected value."); |
| } |
| } |
| } else { |
| /* |
| * This means it isn't in the cache and rememberReads = false. This should only happen from |
| * test hooks at this point. |
| * |
| */ |
| if (txr != null) { |
| txr.cleanupNonDirtyEntries(dataReg); |
| } |
| if (expectedOldValue == null) { |
| /* |
| * They were expecting non-existence. |
| */ |
| return result; |
| } else { |
| /* |
| * If they pass in null to expectedOldValue, we will have it as Token.INVALID here |
| */ |
| if (!Token.isInvalid(expectedOldValue)) { |
| throw new EntryNotFoundException( |
| "The current value was not equal to expected value."); |
| } |
| } |
| } |
| return result; |
| } |
| |
| /* |
| * (non-Javadoc) |
| * |
| * @see org.apache.geode.internal.cache.TXStateInterface#getDeserializedValue(java.lang.Object, |
| * org.apache.geode.internal.cache.LocalRegion, boolean) |
| */ |
| @Override |
| public Object getDeserializedValue(KeyInfo keyInfo, LocalRegion localRegion, boolean updateStats, |
| boolean disableCopyOnRead, boolean preferCD, EntryEventImpl clientEvent, |
| boolean returnTombstones, boolean retainResult, boolean createIfAbsent) { |
| TXEntryState tx = txReadEntry(keyInfo, localRegion, true, createIfAbsent); |
| if (tx != null) { |
| Object v = tx.getValue(keyInfo, localRegion, preferCD); |
| if (!disableCopyOnRead) { |
| v = localRegion.conditionalCopy(v); |
| } |
| return v; |
| } else { |
| return localRegion.getDeserializedValue(null, keyInfo, updateStats, disableCopyOnRead, |
| preferCD, clientEvent, returnTombstones, retainResult); |
| } |
| } |
| |
| /* |
| * (non-Javadoc) |
| * |
| * @see |
| * org.apache.geode.internal.cache.InternalDataView#getSerializedValue(org.apache.geode.internal. |
| * cache.LocalRegion, java.lang.Object, java.lang.Object) |
| */ |
| @Override |
| @Retained |
| public Object getSerializedValue(LocalRegion localRegion, KeyInfo keyInfo, boolean doNotLockEntry, |
| ClientProxyMembershipID requestingClient, EntryEventImpl clientEvent, |
| boolean returnTombstones) throws DataLocationException { |
| final Object key = keyInfo.getKey(); |
| TXEntryState tx = txReadEntry(keyInfo, localRegion, true, true/* create txEntry is absent */); |
| if (tx != null) { |
| Object val = tx.getPendingValue(); |
| if (val == null || Token.isInvalidOrRemoved(val)) { |
| val = findObject(keyInfo, localRegion, val != Token.INVALID, true, val, false, false, |
| requestingClient, clientEvent, false); |
| } |
| return val; |
| } else { |
| // rememberRead is always true for now, |
| // so we should never come here |
| assert localRegion instanceof PartitionedRegion; |
| PartitionedRegion pr = (PartitionedRegion) localRegion; |
| return pr.getDataStore().getSerializedLocally(keyInfo, doNotLockEntry, null, null, |
| returnTombstones); |
| } |
| } |
| |
| /* |
| * (non-Javadoc) |
| * |
| * @see |
| * org.apache.geode.internal.cache.InternalDataView#entryCount(org.apache.geode.internal.cache. |
| * LocalRegion) |
| */ |
| @Override |
| public int entryCount(LocalRegion localRegion) { |
| int result = localRegion.getRegionSize(); |
| TXRegionState txr = txReadRegion(localRegion); |
| if (txr != null) { |
| result += txr.entryCountMod(); |
| } |
| if (result > 0) { |
| return result; |
| } else { |
| // This is to work around bug #40946. |
| // Other threads can destroy all the keys, and so our entryModCount |
| // can bring us below 0 |
| return 0; |
| } |
| } |
| |
| /* |
| * (non-Javadoc) |
| * |
| * @see org.apache.geode.internal.cache.TXStateInterface#containsKey(java.lang.Object, |
| * org.apache.geode.internal.cache.LocalRegion) |
| */ |
| @Override |
| public boolean containsKey(KeyInfo keyInfo, LocalRegion localRegion) { |
| TXEntryState tx = txReadEntry(keyInfo, localRegion, true, true/* create txEntry is absent */); |
| if (tx != null) { |
| return tx.existsLocally(); |
| } else { |
| return localRegion.nonTXContainsKey(keyInfo); |
| } |
| } |
| |
| /* |
| * (non-Javadoc) |
| * |
| * @see org.apache.geode.internal.cache.TXStateInterface#getValueInVM(java.lang.Object, |
| * org.apache.geode.internal.cache.LocalRegion, boolean) |
| */ |
| @Override |
| @Retained |
| public Object getValueInVM(KeyInfo keyInfo, LocalRegion localRegion, boolean rememberRead) { |
| TXEntryState tx = |
| txReadEntry(keyInfo, localRegion, rememberRead, true/* create txEntry is absent */); |
| if (tx != null) { |
| return tx.getValueInVM(keyInfo); |
| } |
| return localRegion.nonTXbasicGetValueInVM(keyInfo); |
| } |
| |
| /* |
| * (non-Javadoc) |
| * |
| * @see org.apache.geode.internal.cache.TXStateInterface#putEntry(org.apache.geode.internal.cache. |
| * EntryEventImpl, boolean, boolean, java.lang.Object, boolean, long, boolean) |
| */ |
| @Override |
| public boolean putEntry(EntryEventImpl event, boolean ifNew, boolean ifOld, |
| Object expectedOldValue, boolean requireOldValue, long lastModified, |
| boolean overwriteDestroyed) { |
| validateDelta(event); |
| return txPutEntry(event, ifNew, requireOldValue, true, expectedOldValue); |
| } |
| |
| /** |
| * throws an exception when cloning is disabled while using delta |
| */ |
| private void validateDelta(EntryEventImpl event) { |
| if (event.getDeltaBytes() != null && !event.getRegion().getAttributes().getCloningEnabled()) { |
| throw new UnsupportedOperationInTransactionException( |
| "Delta without cloning cannot be used in transaction"); |
| } |
| } |
| |
| /* |
| * (non-Javadoc) |
| * |
| * @see org.apache.geode.internal.cache.InternalDataView#isStatsDeferred() |
| */ |
| @Override |
| public boolean isDeferredStats() { |
| return true; |
| } |
| |
| /* |
| * (non-Javadoc) |
| * |
| * @see |
| * org.apache.geode.internal.cache.TXStateInterface#findObject(org.apache.geode.internal.cache. |
| * LocalRegion, java.lang.Object, java.lang.Object, boolean, boolean, java.lang.Object) |
| */ |
| @Override |
| public Object findObject(KeyInfo key, LocalRegion r, boolean isCreate, boolean generateCallbacks, |
| Object value, boolean disableCopyOnRead, boolean preferCD, |
| ClientProxyMembershipID requestingClient, EntryEventImpl clientEvent, |
| boolean returnTombstones) { |
| return r.findObjectInSystem(key, isCreate, this, generateCallbacks, value, disableCopyOnRead, |
| preferCD, requestingClient, clientEvent, returnTombstones); |
| } |
| |
| private TXEntryState readEntryAndCheckIfDestroyed(KeyInfo keyInfo, LocalRegion localRegion, |
| boolean rememberReads, boolean createIfAbsent) { |
| TXEntryState txEntryState = |
| txReadEntry(keyInfo, localRegion, rememberReads, createIfAbsent); |
| if (txEntryState != null) { |
| if (!txEntryState.existsLocally()) { |
| // It was destroyed by the transaction so skip |
| // this key and try the next one |
| return null; // fix for bug 34583 |
| } |
| } |
| return txEntryState; |
| } |
| |
| /* |
| * (non-Javadoc) |
| * |
| * @see |
| * org.apache.geode.internal.cache.TXStateInterface#getEntryForIterator(org.apache.geode.internal. |
| * cache.LocalRegion, java.lang.Object, boolean) |
| */ |
| @Override |
| public Object getEntryForIterator(KeyInfo curr, LocalRegion currRgn, boolean rememberReads, |
| boolean allowTombstones) { |
| if (currRgn instanceof PartitionedRegion) { |
| PartitionedRegion pr = (PartitionedRegion) currRgn; |
| if (!pr.getBucketPrimary(curr.getBucketId()).equals(pr.cache.getMyId())) { |
| // to fix bug 47893 suspend the tx before calling nonTXGetEntry |
| final TXManagerImpl txmgr = pr.getGemFireCache().getTXMgr(); |
| final TXStateProxy tx = txmgr.pauseTransaction(); |
| try { |
| return pr.nonTXGetEntry(curr, false, allowTombstones); |
| } finally { |
| txmgr.unpauseTransaction(tx); |
| } |
| } |
| } |
| TXEntryState txEntryState = |
| readEntryAndCheckIfDestroyed(curr, currRgn, rememberReads, allowTombstones); |
| if (txEntryState != null) { |
| // need to create KeyInfo since higher level iterator may reuse KeyInfo |
| return new TXEntry(currRgn, |
| new KeyInfo(curr.getKey(), curr.getCallbackArg(), curr.getBucketId()), proxy, |
| rememberReads); |
| } |
| return null; |
| } |
| |
| /* |
| * (non-Javadoc) |
| * |
| * @see org.apache.geode.internal.cache.InternalDataView#getKeyForIterator(java.lang.Object, |
| * org.apache.geode.internal.cache.LocalRegion, boolean) |
| */ |
| @Override |
| public Object getKeyForIterator(KeyInfo curr, LocalRegion currRgn, boolean rememberReads, |
| boolean allowTombstones) { |
| assert !(curr.getKey() instanceof RegionEntry); |
| TXEntryState txEntryState = |
| readEntryAndCheckIfDestroyed(curr, currRgn, rememberReads, allowTombstones); |
| if (txEntryState != null) { |
| // txEntry is created/read into txState. |
| return curr.getKey(); |
| } |
| return null; |
| } |
| |
| /* |
| * (non-Javadoc) |
| * |
| * @see |
| * org.apache.geode.internal.cache.InternalDataView#getAdditionalKeysForIterator(org.apache.geode. |
| * internal.cache.LocalRegion) |
| */ |
| @Override |
| public Set getAdditionalKeysForIterator(LocalRegion currRgn) { |
| if (currRgn instanceof PartitionedRegion) { |
| final HashSet ret = new HashSet(); |
| for (TXRegionState rs : this.regions.values()) { |
| if (rs instanceof TXBucketRegionState) { |
| TXBucketRegionState brs = (TXBucketRegionState) rs; |
| if (brs.getPartitionedRegion() == currRgn) { |
| brs.fillInCreatedEntryKeys(ret); |
| } |
| } |
| } |
| return ret; |
| } else { |
| TXRegionState txr = txReadRegion(currRgn); |
| if (txr != null) { |
| final HashSet ret = new HashSet(); |
| txr.fillInCreatedEntryKeys(ret); |
| return ret; |
| } else { |
| return null; |
| } |
| } |
| } |
| |
| |
| /* |
| * (non-Javadoc) |
| * |
| * @see org.apache.geode.internal.cache.TXStateInterface#isInProgressAndSameAs(org.apache.geode. |
| * internal.cache.TXStateInterface) |
| */ |
| @Override |
| public boolean isInProgressAndSameAs(TXStateInterface otherState) { |
| return isInProgress() && otherState == this; |
| } |
| |
| |
| /* |
| * (non-Javadoc) |
| * |
| * @see |
| * org.apache.geode.internal.cache.InternalDataView#putEntryOnRemote(org.apache.geode.internal. |
| * cache.EntryEventImpl, boolean, boolean, java.lang.Object, boolean, long, boolean) |
| */ |
| @Override |
| public boolean putEntryOnRemote(EntryEventImpl event, boolean ifNew, boolean ifOld, |
| Object expectedOldValue, boolean requireOldValue, long lastModified, |
| boolean overwriteDestroyed) throws DataLocationException { |
| /* |
| * Need to flip OriginRemote to true because it is certain that this came from a remote TxStub |
| */ |
| event.setOriginRemote(true); |
| |
| return txPutEntry(event, ifNew, requireOldValue, true, expectedOldValue); |
| } |
| |
| |
| @Override |
| public boolean isFireCallbacks() { |
| return !getEvent().hasOnlyInternalEvents(); |
| } |
| |
| public boolean isOriginRemoteForEvents() { |
| return onBehalfOfRemoteStub || this.proxy.isOnBehalfOfClient(); |
| } |
| |
| @Override |
| public void destroyOnRemote(EntryEventImpl event, boolean cacheWrite, Object expectedOldValue) |
| throws DataLocationException { |
| event.setOriginRemote(true); |
| destroyExistingEntry(event, cacheWrite, expectedOldValue); |
| } |
| |
| @Override |
| public void invalidateOnRemote(EntryEventImpl event, boolean invokeCallbacks, |
| boolean forceNewEntry) throws DataLocationException { |
| event.setOriginRemote(true); |
| invalidateExistingEntry(event, invokeCallbacks, forceNewEntry); |
| } |
| |
| @Override |
| public void checkSupportsRegionDestroy() throws UnsupportedOperationInTransactionException { |
| throw new UnsupportedOperationInTransactionException( |
| "destroyRegion() is not supported while in a transaction"); |
| } |
| |
| @Override |
| public void checkSupportsRegionInvalidate() throws UnsupportedOperationInTransactionException { |
| throw new UnsupportedOperationInTransactionException( |
| "invalidateRegion() is not supported while in a transaction"); |
| } |
| |
| @Override |
| public void checkSupportsRegionClear() throws UnsupportedOperationInTransactionException { |
| throw new UnsupportedOperationInTransactionException( |
| "clear() is not supported while in a transaction"); |
| } |
| |
| /* |
| * (non-Javadoc) |
| * |
| * @see |
| * org.apache.geode.internal.cache.InternalDataView#getBucketKeys(org.apache.geode.internal.cache. |
| * LocalRegion, int) |
| */ |
| @Override |
| public Set getBucketKeys(LocalRegion localRegion, int bucketId, boolean allowTombstones) { |
| PartitionedRegion pr = (PartitionedRegion) localRegion; |
| return pr.getBucketKeys(bucketId, allowTombstones); |
| } |
| |
| /* |
| * (non-Javadoc) |
| * |
| * @see org.apache.geode.internal.cache.InternalDataView#getEntryOnRemote(java.lang.Object, |
| * org.apache.geode.internal.cache.LocalRegion) |
| */ |
| @Override |
| public Entry getEntryOnRemote(KeyInfo key, LocalRegion localRegion, boolean allowTombstones) |
| throws DataLocationException { |
| PartitionedRegion pr = (PartitionedRegion) localRegion; |
| Region.Entry txval = getEntry(key, pr, allowTombstones); |
| if (txval == null) { |
| throw new EntryNotFoundException( |
| "entry not found"); |
| } else { |
| NonLocalRegionEntry nlre = new NonLocalRegionEntry(txval, localRegion); |
| LocalRegion dataReg = localRegion.getDataRegionForRead(key); |
| return new EntrySnapshot(nlre, dataReg, (LocalRegion) txval.getRegion(), allowTombstones); |
| } |
| } |
| |
| /* |
| * (non-Javadoc) |
| * |
| * @see org.apache.geode.internal.cache.TXStateInterface#getSemaphore() |
| */ |
| @Override |
| public ReentrantLock getLock() { |
| return proxy.getLock(); |
| } |
| |
| /* |
| * (non-Javadoc) |
| * |
| * @see |
| * org.apache.geode.internal.cache.InternalDataView#getRegionKeysForIteration(org.apache.geode. |
| * internal.cache.LocalRegion) |
| */ |
| @Override |
| public Set getRegionKeysForIteration(LocalRegion currRegion) { |
| return currRegion.getRegionKeysForIteration(); |
| } |
| |
| |
| @Override |
| public boolean isRealDealLocal() { |
| return true; |
| } |
| |
| @Override |
| public InternalDistributedMember getOriginatingMember() { |
| // During set operations we need to forward it to other nodes to avoid |
| // wrong transaction is masqueraded if the transaction is on behalf of a client. |
| // This needs to be set to the clients member id if the client originated the tx. |
| return proxy.getOnBehalfOfClientMember(); |
| } |
| |
| @Override |
| public boolean isMemberIdForwardingRequired() { |
| /* |
| * State will never fwd on to other nodes so this is not relevant |
| */ |
| return false; |
| } |
| |
| @Override |
| public TXCommitMessage getCommitMessage() { |
| return commitMessage; |
| } |
| |
| |
| /* |
| * For TX this needs to be a PR passed in as region |
| * |
| * @see |
| * org.apache.geode.internal.cache.InternalDataView#postPutAll(org.apache.geode.internal.cache. |
| * DistributedPutAllOperation, java.util.Map, org.apache.geode.internal.cache.LocalRegion) |
| */ |
| @Override |
| public void postPutAll(final DistributedPutAllOperation putallOp, |
| final VersionedObjectList successfulPuts, InternalRegion reg) { |
| |
| final InternalRegion theRegion; |
| if (reg instanceof BucketRegion) { |
| theRegion = ((BucketRegion) reg).getPartitionedRegion(); |
| } else { |
| theRegion = reg; |
| } |
| /* |
| * Don't fire events here. |
| */ |
| /* |
| * We are on the data store, we don't need to do anything here. Commit will push them out. |
| */ |
| /* |
| * We need to put this into the tx state. |
| */ |
| theRegion.syncBulkOp(new Runnable() { |
| @Override |
| public void run() { |
| // final boolean requiresRegionContext = theRegion.keyRequiresRegionContext(); |
| InternalDistributedMember myId = |
| theRegion.getDistributionManager().getDistributionManagerId(); |
| for (int i = 0; i < putallOp.putAllDataSize; ++i) { |
| @Released |
| EntryEventImpl ev = PutAllPRMessage.getEventFromEntry(theRegion, myId, myId, i, |
| putallOp.putAllData, false, putallOp.getBaseEvent().getContext(), false, |
| !putallOp.getBaseEvent().isGenerateCallbacks()); |
| try { |
| ev.setPutAllOperation(putallOp); |
| if (theRegion.basicPut(ev, false, false, null, false)) { |
| successfulPuts.addKeyAndVersion(putallOp.putAllData[i].key, null); |
| } |
| } finally { |
| ev.release(); |
| } |
| } |
| } |
| }, putallOp.getBaseEvent().getEventId()); |
| |
| } |
| |
| @Override |
| public void postRemoveAll(final DistributedRemoveAllOperation op, |
| final VersionedObjectList successfulOps, InternalRegion reg) { |
| final InternalRegion theRegion; |
| if (reg instanceof BucketRegion) { |
| theRegion = ((BucketRegion) reg).getPartitionedRegion(); |
| } else { |
| theRegion = reg; |
| } |
| /* |
| * Don't fire events here. We are on the data store, we don't need to do anything here. Commit |
| * will push them out. We need to put this into the tx state. |
| */ |
| theRegion.syncBulkOp(new Runnable() { |
| @Override |
| public void run() { |
| InternalDistributedMember myId = |
| theRegion.getDistributionManager().getDistributionManagerId(); |
| for (int i = 0; i < op.removeAllDataSize; ++i) { |
| @Released |
| EntryEventImpl ev = RemoveAllPRMessage.getEventFromEntry(theRegion, myId, myId, i, |
| op.removeAllData, false, op.getBaseEvent().getContext(), false, |
| !op.getBaseEvent().isGenerateCallbacks()); |
| ev.setRemoveAllOperation(op); |
| try { |
| theRegion.basicDestroy(ev, true/* should we invoke cacheWriter? */, null); |
| } catch (EntryNotFoundException ignore) { |
| } finally { |
| ev.release(); |
| } |
| successfulOps.addKeyAndVersion(op.removeAllData[i].key, null); |
| } |
| } |
| }, op.getBaseEvent().getEventId()); |
| |
| } |
| |
| @Override |
| public void suspend() { |
| // no special tasks to perform |
| } |
| |
| @Override |
| public void resume() { |
| // no special tasks to perform |
| } |
| |
| @Override |
| public void recordTXOperation(ServerRegionDataAccess region, ServerRegionOperation op, Object key, |
| Object arguments[]) { |
| // no-op here |
| } |
| |
| @Override |
| public void updateEntryVersion(EntryEventImpl event) throws EntryNotFoundException { |
| // Do nothing. Not applicable for transactions. |
| } |
| |
| @Override |
| public boolean isTxState() { |
| return true; |
| } |
| |
| @Override |
| public boolean isTxStateStub() { |
| return false; |
| } |
| |
| @Override |
| public boolean isTxStateProxy() { |
| return false; |
| } |
| |
| @Override |
| public boolean isDistTx() { |
| return false; |
| } |
| |
| @Override |
| public boolean isCreatedOnDistTxCoordinator() { |
| return false; |
| } |
| |
| public void setProxyServer(DistributedMember proxyServer) { |
| this.proxyServer = proxyServer; |
| } |
| |
| public DistributedMember getProxyServer() { |
| return this.proxyServer; |
| } |
| |
| boolean isClosed() { |
| return closed; |
| } |
| } |
