| /* |
| * 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.ignite.internal.processors.cache.distributed.near; |
| |
| import java.util.ArrayDeque; |
| import java.util.Collection; |
| import java.util.HashMap; |
| import java.util.HashSet; |
| import java.util.List; |
| import java.util.Map; |
| import java.util.Queue; |
| import java.util.Set; |
| import java.util.UUID; |
| import java.util.concurrent.atomic.AtomicIntegerFieldUpdater; |
| import org.apache.ignite.IgniteCheckedException; |
| import org.apache.ignite.cluster.ClusterNode; |
| import org.apache.ignite.cluster.ClusterTopologyException; |
| import org.apache.ignite.internal.IgniteDiagnosticAware; |
| import org.apache.ignite.internal.IgniteDiagnosticPrepareContext; |
| import org.apache.ignite.internal.IgniteInternalFuture; |
| import org.apache.ignite.internal.cluster.ClusterTopologyCheckedException; |
| import org.apache.ignite.internal.cluster.ClusterTopologyServerNotFoundException; |
| import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion; |
| import org.apache.ignite.internal.processors.cache.GridCacheContext; |
| import org.apache.ignite.internal.processors.cache.GridCacheEntryEx; |
| import org.apache.ignite.internal.processors.cache.GridCacheEntryRemovedException; |
| import org.apache.ignite.internal.processors.cache.GridCacheMvccCandidate; |
| import org.apache.ignite.internal.processors.cache.GridCacheSharedContext; |
| import org.apache.ignite.internal.processors.cache.distributed.GridDistributedTxMapping; |
| import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtTxMapping; |
| import org.apache.ignite.internal.processors.cache.transactions.IgniteInternalTx; |
| import org.apache.ignite.internal.processors.cache.transactions.IgniteTxEntry; |
| import org.apache.ignite.internal.processors.cache.transactions.IgniteTxKey; |
| import org.apache.ignite.internal.processors.cache.transactions.TxDeadlock; |
| import org.apache.ignite.internal.processors.cache.version.GridCacheVersion; |
| import org.apache.ignite.internal.processors.tracing.MTC; |
| import org.apache.ignite.internal.transactions.IgniteTxRollbackCheckedException; |
| import org.apache.ignite.internal.transactions.IgniteTxTimeoutCheckedException; |
| import org.apache.ignite.internal.util.future.GridEmbeddedFuture; |
| import org.apache.ignite.internal.util.future.GridFutureAdapter; |
| import org.apache.ignite.internal.util.tostring.GridToStringInclude; |
| import org.apache.ignite.internal.util.typedef.C1; |
| import org.apache.ignite.internal.util.typedef.CI1; |
| import org.apache.ignite.internal.util.typedef.F; |
| import org.apache.ignite.internal.util.typedef.P1; |
| import org.apache.ignite.internal.util.typedef.X; |
| import org.apache.ignite.internal.util.typedef.internal.CU; |
| import org.apache.ignite.internal.util.typedef.internal.S; |
| import org.apache.ignite.internal.util.typedef.internal.U; |
| import org.apache.ignite.lang.IgniteBiClosure; |
| import org.apache.ignite.transactions.TransactionDeadlockException; |
| import org.apache.ignite.transactions.TransactionTimeoutException; |
| import org.jetbrains.annotations.Nullable; |
| |
| import static org.apache.ignite.internal.processors.cache.GridCacheOperation.TRANSFORM; |
| import static org.apache.ignite.internal.processors.tracing.MTC.TraceSurroundings; |
| import static org.apache.ignite.internal.processors.tracing.MTC.support; |
| import static org.apache.ignite.transactions.TransactionState.PREPARED; |
| import static org.apache.ignite.transactions.TransactionState.PREPARING; |
| |
| /** |
| * |
| */ |
| public class GridNearOptimisticTxPrepareFuture extends GridNearOptimisticTxPrepareFutureAdapter implements |
| IgniteDiagnosticAware { |
| /** */ |
| private int miniId; |
| |
| /** */ |
| private GridDhtTxMapping txMapping; |
| |
| /** |
| * @param cctx Context. |
| * @param tx Transaction. |
| */ |
| public GridNearOptimisticTxPrepareFuture(GridCacheSharedContext cctx, GridNearTxLocal tx) { |
| super(cctx, tx); |
| |
| assert tx.optimistic() && !tx.serializable() : tx; |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public boolean onOwnerChanged(GridCacheEntryEx entry, GridCacheMvccCandidate owner) { |
| if (log.isDebugEnabled()) |
| log.debug("Transaction future received owner changed callback: " + entry); |
| |
| if (tx.remainingTime() == -1) |
| return false; |
| |
| if (entry.context().isNear() && |
| owner != null && tx.hasWriteKey(entry.txKey())) { |
| if (keyLockFut != null) |
| keyLockFut.onKeyLocked(entry.txKey()); |
| |
| return true; |
| } |
| |
| return false; |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public boolean onNodeLeft(UUID nodeId) { |
| boolean found = false; |
| |
| for (IgniteInternalFuture<?> fut : futures()) { |
| if (isMini(fut)) { |
| MiniFuture f = (MiniFuture)fut; |
| |
| if (f.node().id().equals(nodeId)) { |
| ClusterTopologyCheckedException e = new ClusterTopologyCheckedException("Remote node left grid: " + |
| nodeId); |
| |
| e.retryReadyFuture(cctx.nextAffinityReadyFuture(tx.topologyVersion())); |
| |
| f.onNodeLeft(e, true); |
| |
| found = true; |
| } |
| } |
| } |
| |
| return found; |
| } |
| |
| /** |
| * @param e Error. |
| * @param discoThread {@code True} if executed from discovery thread. |
| */ |
| private void onError(Throwable e, boolean discoThread) { |
| try (TraceSurroundings ignored = support(span)) { |
| if (e instanceof IgniteTxTimeoutCheckedException) { |
| onTimeout(); |
| |
| return; |
| } |
| |
| if (X.hasCause(e, ClusterTopologyCheckedException.class) || X.hasCause(e, ClusterTopologyException.class)) { |
| if (tx.onePhaseCommit()) { |
| tx.markForBackupCheck(); |
| |
| onComplete(); |
| |
| return; |
| } |
| } |
| |
| if (ERR_UPD.compareAndSet(this, null, e)) |
| onComplete(); |
| } |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public void onResult(UUID nodeId, GridNearTxPrepareResponse res) { |
| if (!isDone()) { |
| MiniFuture mini = miniFuture(res.miniId()); |
| |
| if (mini != null) { |
| assert mini.node().id().equals(nodeId); |
| |
| mini.onResult(res); |
| } |
| else { |
| if (msgLog.isDebugEnabled()) { |
| msgLog.debug("Near optimistic prepare fut, failed to find mini future [txId=" + tx.nearXidVersion() + |
| ", node=" + nodeId + |
| ", res=" + res + |
| ", fut=" + this + ']'); |
| } |
| } |
| } |
| else { |
| if (msgLog.isDebugEnabled()) { |
| msgLog.debug("Near optimistic prepare fut, response for finished future [txId=" + tx.nearXidVersion() + |
| ", node=" + nodeId + |
| ", res=" + res + |
| ", fut=" + this + ']'); |
| } |
| } |
| } |
| |
| /** |
| * @return Keys for which {@code MiniFuture} isn't completed. |
| */ |
| public Set<IgniteTxKey> requestedKeys() { |
| compoundsReadLock(); |
| |
| try { |
| int size = futuresCountNoLock(); |
| |
| for (int i = 0; i < size; i++) { |
| IgniteInternalFuture fut = future(i); |
| |
| if (isMini(fut) && !fut.isDone()) { |
| MiniFuture miniFut = (MiniFuture)fut; |
| |
| Collection<IgniteTxEntry> entries = miniFut.mapping().entries(); |
| |
| Set<IgniteTxKey> keys = U.newHashSet(entries.size()); |
| |
| for (IgniteTxEntry entry : entries) |
| keys.add(entry.txKey()); |
| |
| return keys; |
| } |
| } |
| } |
| finally { |
| compoundsReadUnlock(); |
| } |
| |
| return null; |
| } |
| |
| /** |
| * Finds pending mini future by the given mini ID. |
| * |
| * @param miniId Mini ID to find. |
| * @return Mini future. |
| */ |
| private MiniFuture miniFuture(int miniId) { |
| // We iterate directly over the futs collection here to avoid copy. |
| compoundsReadLock(); |
| |
| try { |
| int size = futuresCountNoLock(); |
| |
| // Avoid iterator creation. |
| for (int i = size - 1; i >= 0; i--) { |
| IgniteInternalFuture fut = future(i); |
| |
| if (!isMini(fut)) |
| continue; |
| |
| MiniFuture mini = (MiniFuture)fut; |
| |
| if (mini.futureId() == miniId) { |
| if (!mini.isDone()) |
| return mini; |
| else |
| return null; |
| } |
| } |
| } |
| finally { |
| compoundsReadUnlock(); |
| } |
| |
| return null; |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public boolean onDone(IgniteInternalTx t, Throwable err) { |
| try (TraceSurroundings ignored = MTC.support(span)) { |
| if (isDone()) |
| return false; |
| |
| ERR_UPD.compareAndSet(this, null, err); |
| |
| return onComplete(); |
| } |
| } |
| |
| /** |
| * @param f Future. |
| * @return {@code True} if mini-future. |
| */ |
| private boolean isMini(IgniteInternalFuture<?> f) { |
| return f.getClass().equals(MiniFuture.class); |
| } |
| |
| /** |
| * Completeness callback. |
| * |
| * @return {@code True} if future was finished by this call. |
| */ |
| private boolean onComplete() { |
| Throwable err0 = err; |
| |
| if ((!tx.onePhaseCommit() || tx.mappings().get(cctx.localNodeId()) == null) && |
| (err0 == null || tx.needCheckBackup())) |
| tx.state(PREPARED); |
| |
| if (super.onDone(tx, err0)) { |
| // Don't forget to clean up. |
| cctx.mvcc().removeVersionedFuture(this); |
| |
| return true; |
| } |
| |
| return false; |
| } |
| |
| /** |
| * Initializes future. |
| * |
| * @param remap Remap flag. |
| * @param topLocked {@code True} if thread already acquired lock preventing topology change. |
| */ |
| @Override protected void prepare0(boolean remap, boolean topLocked) { |
| try { |
| boolean txStateCheck = remap ? tx.state() == PREPARING : tx.state(PREPARING); |
| |
| if (!txStateCheck) { |
| if (tx.isRollbackOnly() || tx.setRollbackOnly()) { |
| if (tx.remainingTime() == -1) |
| onDone(tx.timeoutException()); |
| else |
| onDone(tx.rollbackException()); |
| } |
| else |
| onDone(new IgniteCheckedException("Invalid transaction state for " + |
| "prepare [state=" + tx.state() + ", tx=" + this + ']')); |
| |
| return; |
| } |
| |
| IgniteTxEntry singleWrite = tx.singleWrite(); |
| |
| if (singleWrite != null) |
| prepareSingle(singleWrite, topLocked, remap); |
| else |
| prepare(tx.writeEntries(), topLocked, remap); |
| |
| markInitialized(); |
| } |
| catch (TransactionTimeoutException e) { |
| onError(e, false); |
| } |
| } |
| |
| /** |
| * @param write Write. |
| * @param topLocked {@code True} if thread already acquired lock preventing topology change. |
| * @param remap Remap flag. |
| */ |
| private void prepareSingle(IgniteTxEntry write, boolean topLocked, boolean remap) { |
| write.clearEntryReadVersion(); |
| |
| AffinityTopologyVersion topVer = tx.topologyVersion(); |
| |
| assert topVer.topologyVersion() > 0; |
| |
| txMapping = new GridDhtTxMapping(); |
| |
| GridDistributedTxMapping mapping = map(write, topVer, null, topLocked, remap); |
| |
| if (isDone()) { |
| if (log.isDebugEnabled()) |
| log.debug("Abandoning (re)map because future is done: " + this); |
| |
| return; |
| } |
| |
| if (mapping.primary().isLocal()) { |
| if (write.context().isNear()) |
| tx.nearLocallyMapped(true); |
| else if (write.context().isColocated()) |
| tx.colocatedLocallyMapped(true); |
| } |
| |
| if (keyLockFut != null) |
| keyLockFut.onAllKeysAdded(); |
| |
| tx.addSingleEntryMapping(mapping, write); |
| |
| cctx.mvcc().recheckPendingLocks(); |
| |
| mapping.last(true); |
| |
| tx.transactionNodes(txMapping.transactionNodes()); |
| |
| if (!write.context().isNear()) |
| checkOnePhase(txMapping); |
| |
| assert !(mapping.hasColocatedCacheEntries() && mapping.hasNearCacheEntries()) : mapping; |
| |
| proceedPrepare(mapping, null); |
| } |
| |
| /** |
| * @param writes Write entries. |
| * @param topLocked {@code True} if thread already acquired lock preventing topology change. |
| * @param remap Remap flag. |
| */ |
| private void prepare( |
| Iterable<IgniteTxEntry> writes, |
| boolean topLocked, |
| boolean remap |
| ) { |
| AffinityTopologyVersion topVer = tx.topologyVersion(); |
| |
| assert topVer.topologyVersion() > 0; |
| |
| txMapping = new GridDhtTxMapping(); |
| |
| Map<Object, GridDistributedTxMapping> map = new HashMap<>(); |
| |
| // Assign keys to primary nodes. |
| GridDistributedTxMapping cur = null; |
| |
| Queue<GridDistributedTxMapping> mappings = new ArrayDeque<>(); |
| |
| boolean hasNearCache = false; |
| |
| for (IgniteTxEntry write : writes) { |
| write.clearEntryReadVersion(); |
| |
| GridDistributedTxMapping updated = map(write, topVer, cur, topLocked, remap); |
| |
| if (updated == null) |
| // an exception occurred while transaction mapping, stop further processing |
| break; |
| |
| if (write.context().isNear()) |
| hasNearCache = true; |
| |
| if (cur != updated) { |
| mappings.offer(updated); |
| |
| updated.last(true); |
| |
| ClusterNode primary = updated.primary(); |
| |
| assert !primary.isLocal() || !cctx.kernalContext().clientNode(); |
| |
| // Minor optimization to not create MappingKey: on client node can not have mapping for local node. |
| Object key = cctx.kernalContext().clientNode() ? primary.id() : |
| new MappingKey(primary.id(), primary.isLocal() && updated.hasNearCacheEntries()); |
| |
| GridDistributedTxMapping prev = map.put(key, updated); |
| |
| if (prev != null) |
| prev.last(false); |
| |
| if (updated.primary().isLocal()) { |
| if (write.context().isNear()) |
| tx.nearLocallyMapped(true); |
| else if (write.context().isColocated()) |
| tx.colocatedLocallyMapped(true); |
| } |
| |
| cur = updated; |
| } |
| } |
| |
| if (isDone()) { |
| if (log.isDebugEnabled()) |
| log.debug("Abandoning (re)map because future is done: " + this); |
| |
| return; |
| } |
| |
| if (keyLockFut != null) |
| keyLockFut.onAllKeysAdded(); |
| |
| tx.addEntryMapping(mappings); |
| |
| cctx.mvcc().recheckPendingLocks(); |
| |
| tx.transactionNodes(txMapping.transactionNodes()); |
| |
| if (!hasNearCache) |
| checkOnePhase(txMapping); |
| |
| proceedPrepare(mappings); |
| } |
| |
| /** |
| * Continues prepare after previous mapping successfully finished. |
| * |
| * @param mappings Queue of mappings. |
| */ |
| private void proceedPrepare(final Queue<GridDistributedTxMapping> mappings) { |
| final GridDistributedTxMapping m = mappings.poll(); |
| |
| if (m == null) |
| return; |
| |
| proceedPrepare(m, mappings); |
| } |
| |
| /** |
| * Continues prepare after previous mapping successfully finished. |
| * |
| * @param m Mapping. |
| * @param mappings Queue of mappings. |
| */ |
| private void proceedPrepare(GridDistributedTxMapping m, @Nullable final Queue<GridDistributedTxMapping> mappings) { |
| if (isDone()) |
| return; |
| |
| boolean set = cctx.tm().setTxTopologyHint(tx.topologyVersionSnapshot()); |
| |
| try { |
| assert !m.empty(); |
| |
| final ClusterNode n = m.primary(); |
| |
| long timeout = tx.remainingTime(); |
| |
| if (timeout != -1) { |
| GridNearTxPrepareRequest req = new GridNearTxPrepareRequest( |
| futId, |
| tx.topologyVersion(), |
| tx, |
| timeout, |
| null, |
| m.writes(), |
| m.hasNearCacheEntries(), |
| txMapping.transactionNodes(), |
| m.last(), |
| tx.onePhaseCommit(), |
| tx.needReturnValue() && tx.implicit(), |
| tx.implicitSingle(), |
| m.explicitLock(), |
| tx.taskNameHash(), |
| m.clientFirst(), |
| txMapping.transactionNodes().size() == 1, |
| tx.activeCachesDeploymentEnabled(), |
| tx.txState().recovery()); |
| |
| for (IgniteTxEntry txEntry : m.entries()) { |
| if (txEntry.op() == TRANSFORM) |
| req.addDhtVersion(txEntry.txKey(), null); |
| } |
| |
| // Must lock near entries separately. |
| if (m.hasNearCacheEntries()) { |
| try { |
| cctx.tm().prepareTx(tx, m.nearCacheEntries()); |
| } |
| catch (IgniteCheckedException e) { |
| onError(e, false); |
| |
| return; |
| } |
| } |
| |
| final MiniFuture fut = new MiniFuture(this, m, ++miniId, mappings); |
| |
| req.miniId(fut.futureId()); |
| |
| add((IgniteInternalFuture)fut); // Append new future. |
| |
| if (n.isLocal()) { |
| assert !(m.hasColocatedCacheEntries() && m.hasNearCacheEntries()) : m; |
| |
| IgniteInternalFuture<GridNearTxPrepareResponse> prepFut = |
| m.hasNearCacheEntries() ? cctx.tm().txHandler().prepareNearTxLocal(tx, req) |
| : cctx.tm().txHandler().prepareColocatedTx(tx, req); |
| |
| prepFut.listen(new CI1<IgniteInternalFuture<GridNearTxPrepareResponse>>() { |
| @Override public void apply(IgniteInternalFuture<GridNearTxPrepareResponse> prepFut) { |
| try { |
| fut.onResult(prepFut.get()); |
| } |
| catch (IgniteCheckedException e) { |
| fut.onResult(e); |
| } |
| } |
| }); |
| } |
| else { |
| try { |
| cctx.io().send(n, req, tx.ioPolicy()); |
| |
| if (msgLog.isDebugEnabled()) { |
| msgLog.debug("Near optimistic prepare fut, sent request [txId=" + tx.nearXidVersion() + |
| ", node=" + n.id() + ']'); |
| } |
| } |
| catch (ClusterTopologyCheckedException e) { |
| e.retryReadyFuture(cctx.nextAffinityReadyFuture(tx.topologyVersion())); |
| |
| fut.onNodeLeft(e, false); |
| } |
| catch (IgniteCheckedException e) { |
| if (msgLog.isDebugEnabled()) { |
| msgLog.debug("Near optimistic prepare fut, failed to sent request [txId=" + tx.nearXidVersion() + |
| ", node=" + n.id() + |
| ", err=" + e + ']'); |
| } |
| |
| fut.onResult(e); |
| } |
| } |
| } |
| else |
| onTimeout(); |
| } |
| finally { |
| if (set) |
| cctx.tm().setTxTopologyHint(null); |
| } |
| } |
| |
| /** |
| * @param entry Transaction entry. |
| * @param topVer Topology version. |
| * @param cur Current mapping. |
| * @param topLocked {@code True} if thread already acquired lock preventing topology change. |
| * @param remap Remap flag. |
| * @return Mapping. |
| */ |
| private GridDistributedTxMapping map( |
| IgniteTxEntry entry, |
| AffinityTopologyVersion topVer, |
| @Nullable GridDistributedTxMapping cur, |
| boolean topLocked, |
| boolean remap |
| ) { |
| GridCacheContext cacheCtx = entry.context(); |
| |
| List<ClusterNode> nodes; |
| |
| GridCacheEntryEx cached0 = entry.cached(); |
| |
| if (cached0.isDht()) |
| nodes = cacheCtx.topology().nodes(cached0.partition(), topVer); |
| else |
| nodes = cacheCtx.topology().nodes(cacheCtx.affinity().partition(entry.key()), topVer); |
| |
| if (F.isEmpty(nodes)) { |
| ClusterTopologyServerNotFoundException e = new ClusterTopologyServerNotFoundException("Failed to map " + |
| "keys to nodes (partition is not mapped to any node) [key=" + entry.key() + |
| ", partition=" + cacheCtx.affinity().partition(entry.key()) + ", topVer=" + topVer + ']'); |
| |
| onDone(e); |
| |
| return null; |
| } |
| |
| txMapping.addMapping(nodes); |
| |
| ClusterNode primary = F.first(nodes); |
| |
| assert primary != null; |
| |
| if (log.isDebugEnabled()) { |
| log.debug("Mapped key to primary node [key=" + entry.key() + |
| ", part=" + cacheCtx.affinity().partition(entry.key()) + |
| ", primary=" + U.toShortString(primary) + ", topVer=" + topVer + ']'); |
| } |
| |
| // Must re-initialize cached entry while holding topology lock. |
| if (cacheCtx.isNear()) |
| entry.cached(cacheCtx.nearTx().entryExx(entry.key(), topVer)); |
| else |
| entry.cached(cacheCtx.colocated().entryExx(entry.key(), topVer, true)); |
| |
| if (cacheCtx.isNear()) { |
| if (entry.explicitVersion() == null && !remap) { |
| if (keyLockFut == null) { |
| keyLockFut = new KeyLockFuture(); |
| |
| add((IgniteInternalFuture)keyLockFut); |
| } |
| |
| keyLockFut.addLockKey(entry.txKey()); |
| } |
| } |
| |
| if (cur == null || !cur.primary().id().equals(primary.id()) || |
| (primary.isLocal() && cur.hasNearCacheEntries() != cacheCtx.isNear())) { |
| boolean clientFirst = cur == null && !topLocked && cctx.kernalContext().clientNode(); |
| |
| cur = new GridDistributedTxMapping(primary); |
| |
| cur.clientFirst(clientFirst); |
| } |
| |
| cur.add(entry); |
| |
| if (entry.explicitVersion() != null) { |
| tx.markExplicit(primary.id()); |
| |
| cur.markExplicitLock(); |
| } |
| |
| entry.nodeId(primary.id()); |
| |
| if (cacheCtx.isNear()) { |
| while (true) { |
| try { |
| GridNearCacheEntry cached = (GridNearCacheEntry)entry.cached(); |
| |
| cached.dhtNodeId(tx.xidVersion(), primary.id()); |
| |
| break; |
| } |
| catch (GridCacheEntryRemovedException ignore) { |
| entry.cached(cacheCtx.near().entryEx(entry.key(), topVer)); |
| } |
| } |
| } |
| |
| return cur; |
| } |
| |
| /** |
| * |
| */ |
| private void onTimeout() { |
| try (TraceSurroundings ignored = MTC.support(span)) { |
| if (cctx.tm().deadlockDetectionEnabled()) { |
| Set<IgniteTxKey> keys = null; |
| |
| if (keyLockFut != null) |
| keys = new HashSet<>(keyLockFut.lockKeys); |
| else { |
| compoundsReadLock(); |
| |
| try { |
| int size = futuresCountNoLock(); |
| |
| for (int i = 0; i < size; i++) { |
| IgniteInternalFuture fut = future(i); |
| |
| if (isMini(fut) && !fut.isDone()) { |
| MiniFuture miniFut = (MiniFuture)fut; |
| |
| Collection<IgniteTxEntry> entries = miniFut.mapping().entries(); |
| |
| keys = U.newHashSet(entries.size()); |
| |
| for (IgniteTxEntry entry : entries) |
| keys.add(entry.txKey()); |
| |
| break; |
| } |
| } |
| } |
| finally { |
| compoundsReadUnlock(); |
| } |
| } |
| |
| add(new GridEmbeddedFuture<>(new IgniteBiClosure<TxDeadlock, Exception, Object>() { |
| @Override public GridNearTxPrepareResponse apply(TxDeadlock deadlock, Exception e) { |
| if (e != null) |
| U.warn(log, "Failed to detect deadlock.", e); |
| else { |
| e = new IgniteTxTimeoutCheckedException("Failed to acquire lock within provided timeout for " + |
| "transaction [timeout=" + tx.timeout() + ", tx=" + CU.txString(tx) + ']', |
| deadlock != null ? new TransactionDeadlockException(deadlock.toString(cctx)) : null); |
| |
| if (!ERR_UPD.compareAndSet(GridNearOptimisticTxPrepareFuture.this, null, e) |
| && err instanceof IgniteTxTimeoutCheckedException) { |
| err = e; |
| } |
| } |
| |
| onDone(null, e); |
| |
| return null; |
| } |
| }, cctx.tm().detectDeadlock(tx, keys))); |
| } |
| else { |
| ERR_UPD.compareAndSet(this, null, new IgniteTxTimeoutCheckedException("Failed to acquire lock " + |
| "within provided timeout for transaction [timeout=" + tx.timeout() + ", tx=" + tx + ']')); |
| |
| onComplete(); |
| } |
| } |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public void addDiagnosticRequest(IgniteDiagnosticPrepareContext ctx) { |
| if (!isDone()) { |
| for (IgniteInternalFuture fut : futures()) { |
| if (!fut.isDone()) { |
| if (fut instanceof MiniFuture) { |
| MiniFuture miniFut = (MiniFuture)fut; |
| |
| UUID nodeId = miniFut.node().id(); |
| GridCacheVersion dhtVer = miniFut.m.dhtVersion(); |
| GridCacheVersion nearVer = tx.nearXidVersion(); |
| |
| if (dhtVer != null) { |
| ctx.remoteTxInfo( |
| nodeId, |
| dhtVer, |
| nearVer, |
| "GridNearOptimisticTxPrepareFuture waiting for remote node response [" + |
| "nodeId=" + nodeId + |
| ", topVer=" + tx.topologyVersion() + |
| ", dhtVer=" + dhtVer + |
| ", nearVer=" + nearVer + |
| ", futId=" + futId + |
| ", miniId=" + miniFut.futId + |
| ", tx=" + tx + ']'); |
| } |
| else { |
| ctx.basicInfo( |
| cctx.localNodeId(), |
| "GridNearOptimisticTxPrepareFuture waiting for remote node response [" + |
| "nodeId=" + nodeId + |
| ", topVer=" + tx.topologyVersion() + |
| ", dhtVer=" + dhtVer + |
| ", nearVer=" + nearVer + |
| ", futId=" + futId + |
| ", miniId=" + miniFut.futId + |
| ", tx=" + tx + ']'); |
| } |
| } |
| else if (fut instanceof KeyLockFuture) { |
| KeyLockFuture keyFut = (KeyLockFuture)fut; |
| |
| ctx.basicInfo( |
| cctx.localNodeId(), |
| "GridNearOptimisticTxPrepareFuture waiting for local keys lock [" + |
| "node=" + cctx.localNodeId() + |
| ", topVer=" + tx.topologyVersion() + |
| ", allKeysAdded=" + keyFut.allKeysAdded + |
| ", keys=" + keyFut.lockKeys + ']'); |
| } |
| } |
| } |
| } |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public String toString() { |
| Collection<String> futs = F.viewReadOnly(futures(), new C1<IgniteInternalFuture<?>, String>() { |
| @Override public String apply(IgniteInternalFuture<?> f) { |
| if (isMini(f)) { |
| return "[node=" + ((MiniFuture)f).node().id() + |
| ", loc=" + ((MiniFuture)f).node().isLocal() + |
| ", done=" + f.isDone() + "]"; |
| } |
| else |
| return f.toString(); |
| } |
| }, new P1<IgniteInternalFuture<Object>>() { |
| @Override public boolean apply(IgniteInternalFuture<Object> fut) { |
| return isMini(fut); |
| } |
| }); |
| |
| return S.toString(GridNearOptimisticTxPrepareFuture.class, this, |
| "innerFuts", futs, |
| "tx", tx, |
| "super", super.toString()); |
| } |
| |
| /** |
| * |
| */ |
| private static class MiniFuture extends GridFutureAdapter<GridNearTxPrepareResponse> { |
| /** Receive result flag updater. */ |
| private static final AtomicIntegerFieldUpdater<MiniFuture> RCV_RES_UPD = |
| AtomicIntegerFieldUpdater.newUpdater(MiniFuture.class, "rcvRes"); |
| |
| /** Parent future. */ |
| private final GridNearOptimisticTxPrepareFuture parent; |
| |
| /** */ |
| private final int futId; |
| |
| /** Keys. */ |
| @GridToStringInclude |
| private GridDistributedTxMapping m; |
| |
| /** Flag to signal some result being processed. */ |
| private volatile int rcvRes; |
| |
| /** Mappings to proceed prepare. */ |
| private Queue<GridDistributedTxMapping> mappings; |
| |
| /** |
| * @param parent Parent. |
| * @param m Mapping. |
| * @param futId Mini future ID. |
| * @param mappings Queue of mappings to proceed with. |
| */ |
| MiniFuture(GridNearOptimisticTxPrepareFuture parent, |
| GridDistributedTxMapping m, |
| int futId, |
| Queue<GridDistributedTxMapping> mappings) { |
| this.parent = parent; |
| this.m = m; |
| this.futId = futId; |
| this.mappings = mappings; |
| } |
| |
| /** |
| * @return Future ID. |
| */ |
| int futureId() { |
| return futId; |
| } |
| |
| /** |
| * @return Node ID. |
| */ |
| public ClusterNode node() { |
| return m.primary(); |
| } |
| |
| /** |
| * @return Keys. |
| */ |
| public GridDistributedTxMapping mapping() { |
| return m; |
| } |
| |
| /** |
| * @param e Error. |
| */ |
| void onResult(Throwable e) { |
| if (RCV_RES_UPD.compareAndSet(this, 0, 1)) { |
| if (log.isDebugEnabled()) |
| log.debug("Failed to get future result [fut=" + this + ", err=" + e + ']'); |
| |
| // Fail. |
| onDone(e); |
| } |
| else |
| U.warn(log, "Received error after another result has been processed [fut=" + |
| parent + ", mini=" + this + ']', e); |
| } |
| |
| /** |
| * @param e Node failure. |
| * @param discoThread {@code True} if executed from discovery thread. |
| */ |
| void onNodeLeft(ClusterTopologyCheckedException e, boolean discoThread) { |
| if (msgLog.isDebugEnabled()) { |
| msgLog.debug("Near optimistic prepare fut, mini future node left [txId=" + parent.tx.nearXidVersion() + |
| ", node=" + m.primary().id() + ']'); |
| } |
| |
| if (isDone()) |
| return; |
| |
| if (RCV_RES_UPD.compareAndSet(this, 0, 1)) { |
| if (log.isDebugEnabled()) |
| log.debug("Remote node left grid while sending or waiting for reply (will not retry): " + this); |
| |
| // Fail the whole future (make sure not to remap on different primary node |
| // to prevent multiple lock coordinators). |
| parent.onError(e, discoThread); |
| } |
| } |
| |
| /** |
| * @param res Result callback. |
| */ |
| void onResult(final GridNearTxPrepareResponse res) { |
| if (isDone()) |
| return; |
| |
| if (RCV_RES_UPD.compareAndSet(this, 0, 1)) { |
| if (parent.tx.remainingTime() == -1 || res.error() instanceof IgniteTxTimeoutCheckedException) { |
| parent.onTimeout(); |
| |
| return; |
| } |
| |
| if (res.error() != null) { |
| // Fail the whole compound future. |
| parent.onError(res.error(), false); |
| } |
| else { |
| if (res.clientRemapVersion() != null) { |
| assert parent.cctx.kernalContext().clientNode(); |
| assert m.clientFirst(); |
| |
| IgniteInternalFuture<?> affFut = |
| parent.cctx.exchange().affinityReadyFuture(res.clientRemapVersion()); |
| |
| parent.cctx.time().waitAsync(affFut, parent.tx.remainingTime(), (e, timedOut) -> { |
| if (parent.errorOrTimeoutOnTopologyVersion(e, timedOut)) |
| return; |
| |
| remap(); |
| }); |
| } |
| else { |
| parent.onPrepareResponse(m, res, m.hasNearCacheEntries()); |
| |
| // Proceed prepare before finishing mini future. |
| if (mappings != null) |
| parent.proceedPrepare(mappings); |
| |
| // Finish this mini future. |
| onDone((GridNearTxPrepareResponse)null); |
| } |
| } |
| } |
| } |
| |
| /** |
| * |
| */ |
| private void remap() { |
| if (parent.tx.isRollbackOnly()) { |
| onDone(new IgniteTxRollbackCheckedException( |
| "Failed to prepare the transaction, due to the transaction is marked as rolled back " + |
| "[tx=" + CU.txString(parent.tx) + ']')); |
| |
| return; |
| } |
| |
| parent.prepareOnTopology(true, new Runnable() { |
| @Override public void run() { |
| onDone((GridNearTxPrepareResponse)null); |
| } |
| }); |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public String toString() { |
| return S.toString(MiniFuture.class, this, "done", isDone(), "cancelled", isCancelled(), "err", error()); |
| } |
| } |
| |
| /** |
| * |
| */ |
| private static class MappingKey { |
| /** */ |
| private final UUID nodeId; |
| |
| /** */ |
| private final boolean nearEntries; |
| |
| /** |
| * @param nodeId Node ID. |
| * @param nearEntries Near cache entries flag (should be true only for local node). |
| */ |
| MappingKey(UUID nodeId, boolean nearEntries) { |
| this.nodeId = nodeId; |
| this.nearEntries = nearEntries; |
| } |
| |
| /** {@inheritDoc} */ |
| @SuppressWarnings("EqualsWhichDoesntCheckParameterClass") |
| @Override public boolean equals(Object o) { |
| MappingKey that = (MappingKey)o; |
| |
| return nearEntries == that.nearEntries && nodeId.equals(that.nodeId); |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public int hashCode() { |
| int res = nodeId.hashCode(); |
| res = 31 * res + (nearEntries ? 1 : 0); |
| return res; |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public String toString() { |
| return S.toString(MappingKey.class, this); |
| } |
| } |
| } |
