| /* |
| * 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.mvcc; |
| |
| import java.util.ArrayList; |
| import java.util.Collection; |
| import java.util.HashMap; |
| import java.util.HashSet; |
| import java.util.List; |
| import java.util.Map; |
| import java.util.Set; |
| import java.util.TreeMap; |
| import java.util.UUID; |
| import java.util.concurrent.BlockingQueue; |
| import java.util.concurrent.ConcurrentHashMap; |
| import java.util.concurrent.LinkedBlockingQueue; |
| import java.util.concurrent.atomic.AtomicInteger; |
| import java.util.stream.Collectors; |
| import org.apache.ignite.IgniteCheckedException; |
| import org.apache.ignite.IgniteException; |
| import org.apache.ignite.IgniteLogger; |
| import org.apache.ignite.cluster.ClusterNode; |
| import org.apache.ignite.configuration.CacheConfiguration; |
| import org.apache.ignite.configuration.DataRegionConfiguration; |
| import org.apache.ignite.configuration.DataStorageConfiguration; |
| import org.apache.ignite.events.DiscoveryEvent; |
| import org.apache.ignite.events.Event; |
| import org.apache.ignite.internal.GridKernalContext; |
| import org.apache.ignite.internal.IgniteDiagnosticPrepareContext; |
| import org.apache.ignite.internal.IgniteFutureTimeoutCheckedException; |
| import org.apache.ignite.internal.IgniteInternalFuture; |
| import org.apache.ignite.internal.IgniteInterruptedCheckedException; |
| import org.apache.ignite.internal.NodeStoppingException; |
| import org.apache.ignite.internal.cluster.ClusterTopologyCheckedException; |
| import org.apache.ignite.internal.managers.communication.GridMessageListener; |
| import org.apache.ignite.internal.managers.discovery.DiscoCache; |
| import org.apache.ignite.internal.managers.discovery.DiscoveryCustomMessage; |
| import org.apache.ignite.internal.managers.eventstorage.GridLocalEventListener; |
| import org.apache.ignite.internal.processors.GridProcessorAdapter; |
| import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion; |
| import org.apache.ignite.internal.processors.cache.CacheGroupContext; |
| import org.apache.ignite.internal.processors.cache.DynamicCacheChangeBatch; |
| import org.apache.ignite.internal.processors.cache.DynamicCacheChangeRequest; |
| import org.apache.ignite.internal.processors.cache.ExchangeContext; |
| import org.apache.ignite.internal.processors.cache.GridCacheContext; |
| import org.apache.ignite.internal.processors.cache.GridCacheEntryEx; |
| import org.apache.ignite.internal.processors.cache.KeyCacheObject; |
| import org.apache.ignite.internal.processors.cache.distributed.dht.topology.GridDhtLocalPartition; |
| import org.apache.ignite.internal.processors.cache.distributed.near.GridNearTxLocal; |
| import org.apache.ignite.internal.processors.cache.mvcc.msg.MvccAckRequestQueryCntr; |
| import org.apache.ignite.internal.processors.cache.mvcc.msg.MvccAckRequestQueryId; |
| import org.apache.ignite.internal.processors.cache.mvcc.msg.MvccAckRequestTx; |
| import org.apache.ignite.internal.processors.cache.mvcc.msg.MvccAckRequestTxAndQueryCntr; |
| import org.apache.ignite.internal.processors.cache.mvcc.msg.MvccAckRequestTxAndQueryId; |
| import org.apache.ignite.internal.processors.cache.mvcc.msg.MvccActiveQueriesMessage; |
| import org.apache.ignite.internal.processors.cache.mvcc.msg.MvccFutureResponse; |
| import org.apache.ignite.internal.processors.cache.mvcc.msg.MvccMessage; |
| import org.apache.ignite.internal.processors.cache.mvcc.msg.MvccQuerySnapshotRequest; |
| import org.apache.ignite.internal.processors.cache.mvcc.msg.MvccRecoveryFinishedMessage; |
| import org.apache.ignite.internal.processors.cache.mvcc.msg.MvccSnapshotResponse; |
| import org.apache.ignite.internal.processors.cache.mvcc.msg.MvccTxSnapshotRequest; |
| import org.apache.ignite.internal.processors.cache.mvcc.msg.MvccWaitTxsRequest; |
| import org.apache.ignite.internal.processors.cache.mvcc.txlog.TxKey; |
| import org.apache.ignite.internal.processors.cache.mvcc.txlog.TxLog; |
| import org.apache.ignite.internal.processors.cache.mvcc.txlog.TxState; |
| import org.apache.ignite.internal.processors.cache.persistence.CacheDataRow; |
| import org.apache.ignite.internal.processors.cache.persistence.DatabaseLifecycleListener; |
| import org.apache.ignite.internal.processors.cache.persistence.IgniteCacheDatabaseSharedManager; |
| import org.apache.ignite.internal.processors.cache.transactions.IgniteInternalTx; |
| import org.apache.ignite.internal.processors.cache.tree.mvcc.data.MvccDataRow; |
| import org.apache.ignite.internal.processors.cache.tree.mvcc.search.MvccLinkAwareSearchRow; |
| import org.apache.ignite.internal.util.GridAtomicLong; |
| import org.apache.ignite.internal.util.GridLongList; |
| import org.apache.ignite.internal.util.future.GridCompoundFuture; |
| import org.apache.ignite.internal.util.future.GridCompoundIdentityFuture; |
| import org.apache.ignite.internal.util.future.GridFinishedFuture; |
| import org.apache.ignite.internal.util.future.GridFutureAdapter; |
| import org.apache.ignite.internal.util.lang.GridClosureException; |
| import org.apache.ignite.internal.util.lang.GridCursor; |
| import org.apache.ignite.internal.util.typedef.F; |
| 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.internal.util.worker.GridWorker; |
| import org.apache.ignite.lang.IgniteClosure; |
| import org.apache.ignite.lang.IgniteInClosure; |
| import org.apache.ignite.lang.IgniteProductVersion; |
| import org.apache.ignite.plugin.extensions.communication.Message; |
| import org.apache.ignite.spi.IgniteNodeValidationResult; |
| import org.apache.ignite.thread.IgniteThread; |
| import org.jetbrains.annotations.NotNull; |
| import org.jetbrains.annotations.Nullable; |
| |
| import static org.apache.ignite.cache.CacheAtomicityMode.TRANSACTIONAL_SNAPSHOT; |
| import static org.apache.ignite.events.EventType.EVT_CLIENT_NODE_DISCONNECTED; |
| import static org.apache.ignite.events.EventType.EVT_NODE_FAILED; |
| import static org.apache.ignite.events.EventType.EVT_NODE_LEFT; |
| import static org.apache.ignite.events.EventType.EVT_NODE_METRICS_UPDATED; |
| import static org.apache.ignite.events.EventType.EVT_NODE_SEGMENTED; |
| import static org.apache.ignite.internal.GridTopic.TOPIC_CACHE_COORDINATOR; |
| import static org.apache.ignite.internal.events.DiscoveryCustomEvent.EVT_DISCOVERY_CUSTOM_EVT; |
| import static org.apache.ignite.internal.managers.communication.GridIoPolicy.SYSTEM_POOL; |
| import static org.apache.ignite.internal.processors.cache.mvcc.MvccQueryTracker.MVCC_TRACKER_ID_NA; |
| import static org.apache.ignite.internal.processors.cache.mvcc.MvccUtils.MVCC_COUNTER_NA; |
| import static org.apache.ignite.internal.processors.cache.mvcc.MvccUtils.MVCC_CRD_COUNTER_NA; |
| import static org.apache.ignite.internal.processors.cache.mvcc.MvccUtils.MVCC_HINTS_BIT_OFF; |
| import static org.apache.ignite.internal.processors.cache.mvcc.MvccUtils.MVCC_INITIAL_CNTR; |
| import static org.apache.ignite.internal.processors.cache.mvcc.MvccUtils.MVCC_READ_OP_CNTR; |
| import static org.apache.ignite.internal.processors.cache.mvcc.MvccUtils.MVCC_START_CNTR; |
| import static org.apache.ignite.internal.processors.cache.mvcc.MvccUtils.MVCC_START_OP_CNTR; |
| import static org.apache.ignite.internal.processors.cache.mvcc.MvccUtils.compare; |
| import static org.apache.ignite.internal.processors.cache.mvcc.MvccUtils.hasNewVersion; |
| import static org.apache.ignite.internal.processors.cache.mvcc.MvccUtils.isVisible; |
| import static org.apache.ignite.internal.processors.cache.mvcc.MvccUtils.noCoordinatorError; |
| import static org.apache.ignite.internal.processors.cache.mvcc.txlog.TxLog.TX_LOG_CACHE_ID; |
| import static org.apache.ignite.internal.processors.cache.mvcc.txlog.TxLog.TX_LOG_CACHE_NAME; |
| import static org.apache.ignite.internal.processors.cache.persistence.CacheDataRowAdapter.RowData.KEY_ONLY; |
| |
| /** |
| * MVCC processor. |
| */ |
| @SuppressWarnings("serial") |
| public class MvccProcessorImpl extends GridProcessorAdapter implements MvccProcessor, DatabaseLifecycleListener { |
| /** */ |
| private static final IgniteProductVersion MVCC_SUPPORTED_SINCE = IgniteProductVersion.fromString("2.7.0"); |
| |
| /** */ |
| private static final Waiter LOCAL_TRANSACTION_MARKER = new LocalTransactionMarker(); |
| |
| /** Dummy tx for vacuum. */ |
| private static final IgniteInternalTx DUMMY_TX = new GridNearTxLocal(); |
| |
| /** For tests only. */ |
| private static IgniteClosure<Collection<ClusterNode>, ClusterNode> crdC; |
| |
| /** |
| * For testing only. |
| * |
| * @param crdC Closure assigning coordinator. |
| */ |
| static void coordinatorAssignClosure(IgniteClosure<Collection<ClusterNode>, ClusterNode> crdC) { |
| MvccProcessorImpl.crdC = crdC; |
| } |
| |
| /** Topology version when local node was assigned as coordinator. */ |
| private long crdVer; |
| |
| /** */ |
| private volatile MvccCoordinator curCrd; |
| |
| /** */ |
| private volatile MvccCoordinator assignedCrd; |
| |
| /** */ |
| private TxLog txLog; |
| |
| /** */ |
| private List<GridWorker> vacuumWorkers; |
| |
| /** */ |
| private BlockingQueue<VacuumTask> cleanupQueue; |
| |
| /** |
| * Vacuum mutex. Prevents concurrent vacuum while start/stop operations |
| */ |
| private final Object mux = new Object(); |
| |
| /** */ |
| private final GridAtomicLong futIdCntr = new GridAtomicLong(0); |
| |
| /** */ |
| private final GridAtomicLong mvccCntr = new GridAtomicLong(MVCC_START_CNTR); |
| |
| /** */ |
| private final GridAtomicLong committedCntr = new GridAtomicLong(MVCC_INITIAL_CNTR); |
| |
| /** |
| * Contains active transactions on mvcc coordinator. Key is mvcc counter. |
| * Access is protected by "this" monitor. |
| */ |
| private final Map<Long, ActiveTx> activeTxs = new HashMap<>(); |
| |
| /** Active query trackers. */ |
| private final Map<Long, MvccQueryTracker> activeTrackers = new ConcurrentHashMap<>(); |
| |
| /** */ |
| private final Map<UUID, Map<Long, MvccSnapshotResponseListener>> snapLsnrs = new ConcurrentHashMap<>(); |
| |
| /** */ |
| private final Map<Long, WaitAckFuture> ackFuts = new ConcurrentHashMap<>(); |
| |
| /** */ |
| private final Map<Long, GridFutureAdapter> waitTxFuts = new ConcurrentHashMap<>(); |
| |
| /** */ |
| private final Map<TxKey, Waiter> waitMap = new ConcurrentHashMap<>(); |
| |
| /** */ |
| private final ActiveQueries activeQueries = new ActiveQueries(); |
| |
| /** */ |
| private final MvccPreviousCoordinatorQueries prevCrdQueries = new MvccPreviousCoordinatorQueries(); |
| |
| /** */ |
| private final GridFutureAdapter<Void> initFut = new GridFutureAdapter<>(); |
| |
| /** Flag whether at least one cache with {@code CacheAtomicityMode.TRANSACTIONAL_SNAPSHOT} mode is registered. */ |
| private volatile boolean mvccEnabled; |
| |
| /** Flag whether all nodes in cluster support MVCC. */ |
| private volatile boolean mvccSupported = true; |
| |
| /** |
| * Maps failed node id to votes accumulator for that node. |
| */ |
| private final ConcurrentHashMap<UUID, RecoveryBallotBox> recoveryBallotBoxes = new ConcurrentHashMap<>(); |
| |
| /** |
| * @param ctx Context. |
| */ |
| public MvccProcessorImpl(GridKernalContext ctx) { |
| super(ctx); |
| |
| ctx.internalSubscriptionProcessor().registerDatabaseListener(this); |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public void start() throws IgniteCheckedException { |
| ctx.event().addLocalEventListener(new CacheCoordinatorNodeFailListener(), |
| EVT_NODE_FAILED, EVT_NODE_LEFT); |
| |
| ctx.io().addMessageListener(TOPIC_CACHE_COORDINATOR, new CoordinatorMessageListener()); |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public boolean mvccEnabled() { |
| return mvccEnabled; |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public void preProcessCacheConfiguration(CacheConfiguration ccfg) { |
| if (ccfg.getAtomicityMode() == TRANSACTIONAL_SNAPSHOT) { |
| if (!mvccSupported) |
| throw new IgniteException("Cannot start MVCC transactional cache. " + |
| "MVCC is unsupported by the cluster."); |
| |
| mvccEnabled = true; |
| } |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public void validateCacheConfiguration(CacheConfiguration ccfg) { |
| if (ccfg.getAtomicityMode() == TRANSACTIONAL_SNAPSHOT) { |
| if (!mvccSupported) |
| throw new IgniteException("Cannot start MVCC transactional cache. " + |
| "MVCC is unsupported by the cluster."); |
| |
| mvccEnabled = true; |
| } |
| } |
| |
| /** {@inheritDoc} */ |
| @Nullable @Override public IgniteNodeValidationResult validateNode(ClusterNode node) { |
| if (mvccEnabled && node.version().compareToIgnoreTimestamp(MVCC_SUPPORTED_SINCE) < 0) { |
| String errMsg = "Failed to add node to topology. MVCC is enabled on the cluster, but " + |
| "the node doesn't support MVCC [nodeId=" + node.id() + ']'; |
| |
| return new IgniteNodeValidationResult(node.id(), errMsg, errMsg); |
| } |
| |
| return null; |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public void ensureStarted() throws IgniteCheckedException { |
| if (!ctx.clientNode()) { |
| assert mvccEnabled && mvccSupported; |
| |
| if (txLog == null) |
| txLog = new TxLog(ctx, ctx.cache().context().database()); |
| |
| startVacuumWorkers(); |
| |
| if (log.isInfoEnabled()) |
| log.info("Mvcc processor started."); |
| } |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public void beforeStop(IgniteCacheDatabaseSharedManager mgr) { |
| stopVacuumWorkers(); |
| |
| txLog = null; |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public void onInitDataRegions(IgniteCacheDatabaseSharedManager mgr) throws IgniteCheckedException { |
| // We have to always init txLog data region. |
| DataStorageConfiguration dscfg = dataStorageConfiguration(); |
| |
| mgr.addDataRegion( |
| dscfg, |
| createTxLogRegion(dscfg), |
| CU.isPersistenceEnabled(ctx.config())); |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public void afterInitialise(IgniteCacheDatabaseSharedManager mgr) { |
| // No-op. |
| } |
| |
| /** {@inheritDoc} */ |
| @SuppressWarnings("ConstantConditions") |
| @Override public void beforeMemoryRestore(IgniteCacheDatabaseSharedManager mgr) throws IgniteCheckedException { |
| assert CU.isPersistenceEnabled(ctx.config()); |
| assert txLog == null; |
| |
| ctx.cache().context().pageStore().initialize(TX_LOG_CACHE_ID, 1, |
| TX_LOG_CACHE_NAME, mgr.dataRegion(TX_LOG_CACHE_NAME).memoryMetrics()); |
| |
| boolean hasMvccCaches = ctx.cache().cacheGroups().stream().filter(CacheGroupContext::persistenceEnabled) |
| .anyMatch(g -> g.config().getAtomicityMode() == TRANSACTIONAL_SNAPSHOT); |
| |
| if (hasMvccCaches) { |
| txLog = new TxLog(ctx, mgr); |
| |
| mvccEnabled = true; |
| } |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public void afterMemoryRestore(IgniteCacheDatabaseSharedManager mgr) { |
| // No-op. |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public void onDiscoveryEvent(int evtType, Collection<ClusterNode> nodes, long topVer, |
| @Nullable DiscoveryCustomMessage customMsg) { |
| if (evtType == EVT_NODE_METRICS_UPDATED) |
| return; |
| |
| if (evtType == EVT_DISCOVERY_CUSTOM_EVT) |
| checkMvccCacheStarted(customMsg); |
| else |
| assignMvccCoordinator(evtType, nodes, topVer); |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public void onExchangeStart(MvccCoordinator mvccCrd, ExchangeContext exchCtx, ClusterNode exchCrd) { |
| if (!exchCtx.newMvccCoordinator()) |
| return; |
| |
| GridLongList activeQryTrackers = collectActiveQueryTrackers(); |
| |
| exchCtx.addActiveQueries(ctx.localNodeId(), activeQryTrackers); |
| |
| if (exchCrd == null || !mvccCrd.nodeId().equals(exchCrd.id())) { |
| try { |
| sendMessage(mvccCrd.nodeId(), new MvccActiveQueriesMessage(activeQryTrackers)); |
| } |
| catch (IgniteCheckedException e) { |
| U.error(log, "Failed to send active queries to mvcc coordinator: " + e); |
| } |
| } |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public void onExchangeDone(boolean newCrd, DiscoCache discoCache, Map<UUID, GridLongList> activeQueries) { |
| if (!newCrd) { |
| if (curCrd != null && ctx.localNodeId().equals(curCrd.nodeId()) && discoCache != null) |
| cleanupOrphanedServerTransactions(discoCache.serverNodes()); |
| |
| return; |
| } |
| |
| ctx.cache().context().tm().rollbackMvccTxOnCoordinatorChange(); |
| |
| if (ctx.localNodeId().equals(curCrd.nodeId())) { |
| assert ctx.localNodeId().equals(curCrd.nodeId()); |
| |
| MvccCoordinator crd = discoCache.mvccCoordinator(); |
| |
| assert crd != null; |
| |
| // No need to re-initialize if coordinator version hasn't changed (e.g. it was cluster activation). |
| if (crdVer == crd.coordinatorVersion()) |
| return; |
| |
| crdVer = crd.coordinatorVersion(); |
| |
| if (log.isInfoEnabled()) |
| log.info("Initialize local node as mvcc coordinator [node=" + ctx.localNodeId() + |
| ", crdVer=" + crdVer + ']'); |
| |
| prevCrdQueries.init(activeQueries, F.view(discoCache.allNodes(), this::supportsMvcc), ctx.discovery()); |
| |
| initFut.onDone(); |
| } |
| } |
| |
| /** |
| * Cleans up active transacitons lost near node which is server. Executed on coordinator. |
| * @param liveSrvs Live server nodes at the moment of cleanup. |
| */ |
| private void cleanupOrphanedServerTransactions(Collection<ClusterNode> liveSrvs) { |
| Set<UUID> ids = liveSrvs.stream() |
| .map(ClusterNode::id) |
| .collect(Collectors.toSet()); |
| |
| List<Long> forRmv = new ArrayList<>(); |
| |
| synchronized (this) { |
| for (Map.Entry<Long, ActiveTx> entry : activeTxs.entrySet()) { |
| // If node started tx is not known as live then remove such tx from active list |
| ActiveTx activeTx = entry.getValue(); |
| |
| if (activeTx.getClass() == ActiveServerTx.class && !ids.contains(activeTx.nearNodeId)) |
| forRmv.add(entry.getKey()); |
| } |
| } |
| |
| for (Long txCntr : forRmv) |
| // Committed counter is increased because it is not known if transaction was committed or not and we must |
| // bump committed counter for committed transaction as it is used in (read-only) query snapshot. |
| onTxDone(txCntr, true); |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public void processClientActiveQueries(UUID nodeId, @Nullable GridLongList activeQueries) { |
| prevCrdQueries.addNodeActiveQueries(nodeId, activeQueries); |
| } |
| |
| /** {@inheritDoc} */ |
| @Override @Nullable public MvccCoordinator currentCoordinator() { |
| return currentCoordinator(AffinityTopologyVersion.NONE); |
| } |
| |
| /** {@inheritDoc} */ |
| @Override @Nullable public MvccCoordinator currentCoordinator(AffinityTopologyVersion topVer) { |
| MvccCoordinator crd = curCrd; |
| |
| // Assert coordinator did not already change. |
| assert crd == null |
| || topVer == AffinityTopologyVersion.NONE |
| || crd.topologyVersion().compareTo(topVer) <= 0 : "Invalid coordinator [crd=" + crd + ", topVer=" + topVer + ']'; |
| |
| return crd; |
| } |
| |
| /** {@inheritDoc} */ |
| @Override @Nullable public MvccCoordinator assignedCoordinator() { |
| return assignedCrd; |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public UUID currentCoordinatorId() { |
| MvccCoordinator curCrd = this.curCrd; |
| |
| return curCrd != null ? curCrd.nodeId() : null; |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public void updateCoordinator(MvccCoordinator curCrd) { |
| this.curCrd = curCrd; |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public byte state(MvccVersion ver) throws IgniteCheckedException { |
| return state(ver.coordinatorVersion(), ver.counter()); |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public byte state(long crdVer, long cntr) throws IgniteCheckedException { |
| assert txLog != null && mvccEnabled; |
| |
| return txLog.get(crdVer, cntr); |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public void updateState(MvccVersion ver, byte state) throws IgniteCheckedException { |
| updateState(ver, state, true); |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public void updateState(MvccVersion ver, byte state, boolean primary) throws IgniteCheckedException { |
| assert txLog != null && mvccEnabled; |
| |
| TxKey key = new TxKey(ver.coordinatorVersion(), ver.counter()); |
| |
| txLog.put(key, state, primary); |
| |
| Waiter waiter; |
| |
| if (primary && (state == TxState.ABORTED || state == TxState.COMMITTED) |
| && (waiter = waitMap.remove(key)) != null) |
| waiter.run(ctx); |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public void registerLocalTransaction(long crd, long cntr) { |
| Waiter old = waitMap.putIfAbsent(new TxKey(crd, cntr), LOCAL_TRANSACTION_MARKER); |
| |
| assert old == null || old.hasLocalTransaction(); |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public boolean hasLocalTransaction(long crd, long cntr) { |
| Waiter waiter = waitMap.get(new TxKey(crd, cntr)); |
| |
| return waiter != null && waiter.hasLocalTransaction(); |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public IgniteInternalFuture<Void> waitFor(GridCacheContext cctx, MvccVersion locked) throws IgniteCheckedException { |
| TxKey key = new TxKey(locked.coordinatorVersion(), locked.counter()); |
| |
| LockFuture fut = new LockFuture(cctx.ioPolicy()); |
| |
| Waiter waiter = waitMap.merge(key, fut, Waiter::concat); |
| |
| byte state = txLog.get(key); |
| |
| if ((state == TxState.ABORTED || state == TxState.COMMITTED) |
| && !waiter.hasLocalTransaction() && (waiter = waitMap.remove(key)) != null) |
| waiter.run(ctx); |
| |
| return fut; |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public void addQueryTracker(MvccQueryTracker tracker) { |
| assert tracker.id() != MVCC_TRACKER_ID_NA; |
| |
| MvccQueryTracker tr = activeTrackers.put(tracker.id(), tracker); |
| |
| assert tr == null; |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public void removeQueryTracker(Long id) { |
| activeTrackers.remove(id); |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public MvccSnapshot tryRequestSnapshotLocal() throws ClusterTopologyCheckedException { |
| return tryRequestSnapshotLocal(null); |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public MvccSnapshot tryRequestSnapshotLocal(@Nullable IgniteInternalTx tx) throws ClusterTopologyCheckedException { |
| MvccCoordinator crd = currentCoordinator(); |
| |
| if (crd == null) |
| throw noCoordinatorError(); |
| |
| if (tx != null) { |
| AffinityTopologyVersion topVer = ctx.cache().context().lockedTopologyVersion(null); |
| |
| if (topVer != null && topVer.compareTo(crd.topologyVersion()) < 0) |
| throw new ClusterTopologyCheckedException("Mvcc coordinator is outdated " + |
| "for the locked topology version. [crd=" + crd + ", tx=" + tx + ']'); |
| } |
| |
| if (!ctx.localNodeId().equals(crd.nodeId()) || !initFut.isDone()) |
| return null; |
| else if (tx != null) |
| return assignTxSnapshot(0L, ctx.localNodeId(), false); |
| else |
| return activeQueries.assignQueryCounter(ctx.localNodeId(), 0L); |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public IgniteInternalFuture<MvccSnapshot> requestSnapshotAsync(IgniteInternalTx tx) { |
| MvccSnapshotFuture fut = new MvccSnapshotFuture(); |
| |
| requestSnapshotAsync(tx, fut); |
| |
| return fut; |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public void requestSnapshotAsync(MvccSnapshotResponseListener lsnr) { |
| requestSnapshotAsync(null, lsnr); |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public void requestSnapshotAsync(IgniteInternalTx tx, MvccSnapshotResponseListener lsnr) { |
| MvccCoordinator crd = currentCoordinator(); |
| |
| if (crd == null) { |
| lsnr.onError(noCoordinatorError()); |
| |
| return; |
| } |
| |
| if (tx != null) { |
| AffinityTopologyVersion topVer = ctx.cache().context().lockedTopologyVersion(null); |
| |
| if (topVer != null && topVer.compareTo(crd.topologyVersion()) < 0) { |
| lsnr.onError(new ClusterTopologyCheckedException("Mvcc coordinator is outdated " + |
| "for the locked topology version. [crd=" + crd + ", tx=" + tx + ']')); |
| |
| return; |
| } |
| } |
| |
| if (ctx.localNodeId().equals(crd.nodeId())) { |
| if (!initFut.isDone()) { |
| // Wait for the local coordinator init. |
| initFut.listen(new IgniteInClosure<IgniteInternalFuture>() { |
| @Override public void apply(IgniteInternalFuture fut) { |
| requestSnapshotAsync(tx, lsnr); |
| } |
| }); |
| } |
| else if (tx != null) |
| lsnr.onResponse(assignTxSnapshot(0L, ctx.localNodeId(), false)); |
| else |
| lsnr.onResponse(activeQueries.assignQueryCounter(ctx.localNodeId(), 0L)); |
| |
| return; |
| } |
| |
| // Send request to the remote coordinator. |
| UUID nodeId = crd.nodeId(); |
| |
| long id = futIdCntr.incrementAndGet(); |
| |
| Map<Long, MvccSnapshotResponseListener> map = snapLsnrs.get(nodeId), map0; |
| |
| if (map == null && (map0 = snapLsnrs.putIfAbsent(nodeId, map = new ConcurrentHashMap<>())) != null) |
| map = map0; |
| |
| map.put(id, lsnr); |
| |
| try { |
| sendMessage(nodeId, tx != null ? new MvccTxSnapshotRequest(id) : new MvccQuerySnapshotRequest(id)); |
| } |
| catch (IgniteCheckedException e) { |
| if (map.remove(id) != null) |
| lsnr.onError(e); |
| } |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public IgniteInternalFuture<Void> ackTxCommit(MvccSnapshot updateVer) { |
| return ackTxCommit(updateVer, null, 0L); |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public IgniteInternalFuture<Void> ackTxCommit(MvccVersion updateVer, MvccSnapshot readSnapshot, |
| long qryId) { |
| assert updateVer != null; |
| |
| MvccCoordinator crd = curCrd; |
| |
| if (updateVer.coordinatorVersion() == crd.coordinatorVersion()) |
| return sendTxCommit(crd, createTxAckMessage(futIdCntr.incrementAndGet(), updateVer, readSnapshot, qryId)); |
| else if (readSnapshot != null) |
| ackQueryDone(readSnapshot, qryId); |
| |
| return new GridFinishedFuture<>(); |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public void ackTxRollback(MvccVersion updateVer) { |
| assert updateVer != null; |
| |
| MvccCoordinator crd = curCrd; |
| |
| if (crd.coordinatorVersion() != updateVer.coordinatorVersion()) |
| return; |
| |
| MvccAckRequestTx msg = createTxAckMessage(-1, updateVer, null, 0L); |
| |
| msg.skipResponse(true); |
| |
| try { |
| sendMessage(crd.nodeId(), msg); |
| } |
| catch (ClusterTopologyCheckedException e) { |
| if (log.isDebugEnabled()) |
| log.debug("Failed to send tx rollback ack, node left [msg=" + msg + ", node=" + crd.nodeId() + ']'); |
| } |
| catch (IgniteCheckedException e) { |
| U.error(log, "Failed to send tx rollback ack [msg=" + msg + ", node=" + crd.nodeId() + ']', e); |
| } |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public void ackTxRollback(MvccVersion updateVer, MvccSnapshot readSnapshot, long qryTrackerId) { |
| assert updateVer != null; |
| |
| MvccCoordinator crd = curCrd; |
| |
| if (crd.coordinatorVersion() != updateVer.coordinatorVersion()) { |
| if (readSnapshot != null) |
| ackQueryDone(readSnapshot, qryTrackerId); |
| |
| return; |
| } |
| |
| MvccAckRequestTx msg = createTxAckMessage(-1, updateVer, readSnapshot, qryTrackerId); |
| |
| msg.skipResponse(true); |
| |
| try { |
| sendMessage(crd.nodeId(), msg); |
| } |
| catch (ClusterTopologyCheckedException e) { |
| if (log.isDebugEnabled()) |
| log.debug("Failed to send tx rollback ack, node left [msg=" + msg + ", node=" + crd.nodeId() + ']'); |
| } |
| catch (IgniteCheckedException e) { |
| U.error(log, "Failed to send tx rollback ack [msg=" + msg + ", node=" + crd.nodeId() + ']', e); |
| } |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public void ackQueryDone(MvccSnapshot snapshot, long qryId) { |
| assert snapshot != null; |
| |
| MvccCoordinator crd = currentCoordinator(); |
| |
| if (crd == null || crd.coordinatorVersion() == snapshot.coordinatorVersion() |
| && sendQueryDone(crd, new MvccAckRequestQueryCntr(queryTrackCounter(snapshot)))) |
| return; |
| |
| Message msg = new MvccAckRequestQueryId(qryId); |
| |
| do { |
| crd = currentCoordinator(); |
| } |
| while (!sendQueryDone(crd, msg)); |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public IgniteInternalFuture<Void> waitTxsFuture(UUID crdId, GridLongList txs) { |
| assert crdId != null; |
| assert txs != null && !txs.isEmpty(); |
| |
| WaitAckFuture fut = new WaitAckFuture(futIdCntr.incrementAndGet(), crdId, false); |
| |
| ackFuts.put(fut.id, fut); |
| |
| try { |
| sendMessage(crdId, new MvccWaitTxsRequest(fut.id, txs)); |
| } |
| catch (IgniteCheckedException e) { |
| if (ackFuts.remove(fut.id) != null) { |
| if (e instanceof ClusterTopologyCheckedException) |
| fut.onDone(); // No need to wait, new coordinator will be assigned, finish without error. |
| else |
| fut.onDone(e); |
| } |
| } |
| |
| return fut; |
| } |
| |
| /** {@inheritDoc} */ |
| // TODO: Proper use of diagnostic context. |
| @Override public void dumpDebugInfo(IgniteLogger log, @Nullable IgniteDiagnosticPrepareContext diagCtx) { |
| boolean first = true; |
| |
| for (Map<Long, MvccSnapshotResponseListener> map : snapLsnrs.values()) { |
| if (first) { |
| U.warn(log, "Pending mvcc listener: "); |
| |
| first = false; |
| } |
| |
| for (MvccSnapshotResponseListener lsnr : map.values()) |
| U.warn(log, ">>> " + lsnr.toString()); |
| } |
| |
| first = true; |
| |
| for (WaitAckFuture waitAckFut : ackFuts.values()) { |
| if (first) { |
| U.warn(log, "Pending mvcc wait ack futures: "); |
| |
| first = false; |
| } |
| |
| U.warn(log, ">>> " + waitAckFut.toString()); |
| } |
| } |
| |
| /** |
| * Removes all less or equals to the given one records from Tx log. |
| * |
| * @param ver Version. |
| * @throws IgniteCheckedException If fails. |
| */ |
| void removeUntil(MvccVersion ver) throws IgniteCheckedException { |
| txLog.removeUntil(ver.coordinatorVersion(), ver.counter()); |
| } |
| |
| /** |
| * TODO IGNITE-7966 |
| * |
| * @return Data region configuration. |
| */ |
| private DataRegionConfiguration createTxLogRegion(DataStorageConfiguration dscfg) { |
| DataRegionConfiguration cfg = new DataRegionConfiguration(); |
| |
| cfg.setName(TX_LOG_CACHE_NAME); |
| cfg.setInitialSize(dscfg.getSystemRegionInitialSize()); |
| cfg.setMaxSize(dscfg.getSystemRegionMaxSize()); |
| cfg.setPersistenceEnabled(CU.isPersistenceEnabled(dscfg)); |
| return cfg; |
| } |
| |
| /** |
| * @return Data storage configuration. |
| */ |
| private DataStorageConfiguration dataStorageConfiguration() { |
| return ctx.config().getDataStorageConfiguration(); |
| } |
| |
| /** */ |
| private void assignMvccCoordinator(int evtType, Collection<ClusterNode> nodes, long topVer) { |
| checkMvccSupported(nodes); |
| |
| MvccCoordinator crd; |
| |
| if (evtType == EVT_NODE_SEGMENTED || evtType == EVT_CLIENT_NODE_DISCONNECTED) |
| crd = null; |
| else { |
| crd = assignedCrd; |
| |
| if (crd == null || |
| ((evtType == EVT_NODE_FAILED || evtType == EVT_NODE_LEFT) && !F.nodeIds(nodes).contains(crd.nodeId()))) { |
| ClusterNode crdNode = null; |
| |
| if (crdC != null) { |
| crdNode = crdC.apply(nodes); |
| |
| if (log.isInfoEnabled()) |
| log.info("Assigned coordinator using test closure: " + crd); |
| } |
| else { |
| // Expect nodes are sorted by order. |
| for (ClusterNode node : nodes) { |
| if (!node.isClient() && supportsMvcc(node)) { |
| crdNode = node; |
| |
| break; |
| } |
| } |
| } |
| |
| crd = crdNode != null ? new MvccCoordinator(crdNode.id(), coordinatorVersion(crdNode), |
| new AffinityTopologyVersion(topVer, 0)) : null; |
| |
| if (log.isInfoEnabled() && crd != null) |
| log.info("Assigned mvcc coordinator [crd=" + crd + ", crdNode=" + crdNode + ']'); |
| else if (crd == null) |
| U.warn(log, "New mvcc coordinator was not assigned [topVer=" + topVer + ']'); |
| } |
| } |
| |
| assignedCrd = crd; |
| } |
| |
| /** |
| * @param crdNode Assigned coordinator node. |
| * @return Coordinator version. |
| */ |
| private long coordinatorVersion(ClusterNode crdNode) { |
| return crdNode.order() + ctx.discovery().gridStartTime(); |
| } |
| |
| /** */ |
| private void checkMvccSupported(Collection<ClusterNode> nodes) { |
| if (mvccEnabled) { |
| assert mvccSupported; |
| |
| return; |
| } |
| |
| boolean res = true, was = mvccSupported; |
| |
| for (ClusterNode node : nodes) { |
| if (!supportsMvcc(node)) { |
| res = false; |
| |
| break; |
| } |
| } |
| |
| if (was != res) |
| mvccSupported = res; |
| } |
| |
| /** */ |
| private boolean supportsMvcc(ClusterNode node) { |
| return node.version().compareToIgnoreTimestamp(MVCC_SUPPORTED_SINCE) >= 0; |
| } |
| |
| /** */ |
| private void checkMvccCacheStarted(@Nullable DiscoveryCustomMessage customMsg) { |
| assert customMsg != null; |
| |
| if (!mvccEnabled && customMsg instanceof DynamicCacheChangeBatch) { |
| for (DynamicCacheChangeRequest req : ((DynamicCacheChangeBatch)customMsg).requests()) { |
| CacheConfiguration ccfg = req.startCacheConfiguration(); |
| |
| if (ccfg == null) |
| continue; |
| |
| if (ccfg.getAtomicityMode() == TRANSACTIONAL_SNAPSHOT) { |
| assert mvccSupported; |
| |
| mvccEnabled = true; |
| } |
| } |
| } |
| } |
| |
| /** |
| * @return Active queries list. |
| */ |
| private GridLongList collectActiveQueryTrackers() { |
| assert curCrd != null; |
| |
| GridLongList activeQryTrackers = new GridLongList(); |
| |
| for (MvccQueryTracker tracker : activeTrackers.values()) { |
| long trackerId = tracker.onMvccCoordinatorChange(curCrd); |
| |
| if (trackerId != MVCC_TRACKER_ID_NA) |
| activeQryTrackers.add(trackerId); |
| } |
| |
| return activeQryTrackers; |
| } |
| |
| /** */ |
| private MvccSnapshotResponse assignTxSnapshot(long futId, UUID nearId, boolean client) { |
| assert initFut.isDone(); |
| assert crdVer != 0; |
| assert ctx.localNodeId().equals(currentCoordinatorId()); |
| |
| MvccSnapshotResponse res = new MvccSnapshotResponse(); |
| |
| long ver, cleanup, tracking; |
| |
| synchronized (this) { |
| ver = mvccCntr.incrementAndGet(); |
| tracking = ver; |
| cleanup = committedCntr.get() + 1; |
| |
| for (Map.Entry<Long, ActiveTx> entry : activeTxs.entrySet()) { |
| cleanup = Math.min(entry.getValue().tracking, cleanup); |
| tracking = Math.min(entry.getKey(), tracking); |
| |
| res.addTx(entry.getKey()); |
| } |
| |
| ActiveTx activeTx = client ? new ActiveTx(tracking, nearId) : new ActiveServerTx(tracking, nearId); |
| |
| boolean add = activeTxs.put(ver, activeTx) == null; |
| |
| assert add : ver; |
| } |
| |
| long minQry = activeQueries.minimalQueryCounter(); |
| |
| if (minQry != -1) |
| cleanup = Math.min(cleanup, minQry); |
| |
| cleanup = prevCrdQueries.previousQueriesDone() ? cleanup - 1 : MVCC_COUNTER_NA; |
| |
| res.init(futId, crdVer, ver, MVCC_START_OP_CNTR, cleanup, tracking); |
| |
| return res; |
| } |
| |
| /** */ |
| private void onTxDone(Long txCntr, boolean increaseCommittedCntr) { |
| assert initFut.isDone(); |
| |
| GridFutureAdapter fut; |
| |
| synchronized (this) { |
| activeTxs.remove(txCntr); |
| |
| if (increaseCommittedCntr) |
| committedCntr.setIfGreater(txCntr); |
| } |
| |
| fut = waitTxFuts.remove(txCntr); |
| |
| if (fut != null) |
| fut.onDone(); |
| } |
| |
| /** |
| * @param mvccCntr Query counter. |
| */ |
| private void onQueryDone(UUID nodeId, Long mvccCntr) { |
| activeQueries.onQueryDone(nodeId, mvccCntr); |
| } |
| |
| /** |
| * @param futId Future ID. |
| * @param updateVer Update version. |
| * @param readSnapshot Optional read version. |
| * @param qryTrackerId Query tracker id. |
| * @return Message. |
| */ |
| private MvccAckRequestTx createTxAckMessage(long futId, MvccVersion updateVer, MvccSnapshot readSnapshot, |
| long qryTrackerId) { |
| if (readSnapshot == null) |
| return new MvccAckRequestTx(futId, updateVer.counter()); |
| else if (readSnapshot.coordinatorVersion() == updateVer.coordinatorVersion()) |
| return new MvccAckRequestTxAndQueryCntr(futId, updateVer.counter(), queryTrackCounter(readSnapshot)); |
| else |
| return new MvccAckRequestTxAndQueryId(futId, updateVer.counter(), qryTrackerId); |
| } |
| |
| /** |
| * @param mvccVer Read version. |
| * @return Tracker counter. |
| */ |
| private long queryTrackCounter(MvccSnapshot mvccVer) { |
| long trackCntr = mvccVer.counter(); |
| |
| MvccLongList txs = mvccVer.activeTransactions(); |
| |
| int size = txs.size(); |
| |
| for (int i = 0; i < size; i++) { |
| long txVer = txs.get(i); |
| |
| if (txVer < trackCntr) |
| trackCntr = txVer; |
| } |
| |
| return trackCntr; |
| } |
| |
| /** |
| * Launches vacuum workers and scheduler. |
| */ |
| void startVacuumWorkers() { |
| if (!ctx.clientNode()) { |
| synchronized (mux) { |
| if (vacuumWorkers == null) { |
| assert cleanupQueue == null; |
| |
| cleanupQueue = new LinkedBlockingQueue<>(); |
| |
| vacuumWorkers = new ArrayList<>(ctx.config().getMvccVacuumThreadCount() + 1); |
| |
| vacuumWorkers.add(new VacuumScheduler(ctx, log, this)); |
| |
| for (int i = 0; i < ctx.config().getMvccVacuumThreadCount(); i++) { |
| vacuumWorkers.add(new VacuumWorker(ctx, log, cleanupQueue)); |
| } |
| |
| for (GridWorker worker : vacuumWorkers) { |
| new IgniteThread(worker).start(); |
| } |
| |
| return; |
| } |
| } |
| |
| U.warn(log, "Attempting to start active vacuum."); |
| } |
| } |
| |
| /** |
| * Stops vacuum worker and scheduler. |
| */ |
| void stopVacuumWorkers() { |
| if (!ctx.clientNode()) { |
| List<GridWorker> workers; |
| BlockingQueue<VacuumTask> queue; |
| |
| synchronized (mux) { |
| workers = vacuumWorkers; |
| queue = cleanupQueue; |
| |
| vacuumWorkers = null; |
| cleanupQueue = null; |
| } |
| |
| if (workers == null) { |
| if (log.isDebugEnabled() && mvccEnabled()) |
| log.debug("Attempting to stop inactive vacuum."); |
| |
| return; |
| } |
| |
| assert queue != null; |
| |
| // Stop vacuum workers outside mutex to prevent deadlocks. |
| U.cancel(workers); |
| U.join(workers, log); |
| |
| if (!queue.isEmpty()) { |
| IgniteCheckedException ex = vacuumCancelledException(); |
| |
| for (VacuumTask task : queue) { |
| task.onDone(ex); |
| } |
| } |
| } |
| } |
| |
| /** |
| * Runs vacuum process. |
| * |
| * @return {@code Future} with {@link VacuumMetrics}. |
| */ |
| IgniteInternalFuture<VacuumMetrics> runVacuum() { |
| assert !ctx.clientNode(); |
| |
| MvccCoordinator crd0 = currentCoordinator(); |
| |
| if (Thread.currentThread().isInterrupted() || |
| crd0 == null || |
| crdVer == 0 && ctx.localNodeId().equals(crd0.nodeId())) |
| return new GridFinishedFuture<>(new VacuumMetrics()); |
| |
| final GridFutureAdapter<VacuumMetrics> res = new GridFutureAdapter<>(); |
| |
| MvccSnapshot snapshot; |
| |
| try { |
| // TODO IGNITE-8974 create special method for getting cleanup version only. |
| snapshot = tryRequestSnapshotLocal(DUMMY_TX); |
| } |
| catch (ClusterTopologyCheckedException e) { |
| throw new AssertionError(e); |
| } |
| |
| if (snapshot != null) |
| continueRunVacuum(res, snapshot); |
| else |
| requestSnapshotAsync(DUMMY_TX, new MvccSnapshotResponseListener() { |
| @Override public void onResponse(MvccSnapshot s) { |
| continueRunVacuum(res, s); |
| } |
| |
| @Override public void onError(IgniteCheckedException e) { |
| if (!(e instanceof ClusterTopologyCheckedException)) |
| completeWithException(res, e); |
| else { |
| if (log.isDebugEnabled()) |
| log.debug("Vacuum failed to receive an Mvcc snapshot. " + |
| "Need to retry on the stable topology. " + e.getMessage()); |
| |
| res.onDone(new VacuumMetrics()); |
| } |
| } |
| }); |
| |
| return res; |
| } |
| |
| /** |
| * @param res Result. |
| * @param snapshot Snapshot. |
| */ |
| private void continueRunVacuum(GridFutureAdapter<VacuumMetrics> res, MvccSnapshot snapshot) { |
| ackTxCommit(snapshot) |
| .listen(new IgniteInClosure<IgniteInternalFuture>() { |
| @Override public void apply(IgniteInternalFuture fut) { |
| Throwable err; |
| |
| if ((err = fut.error()) != null) { |
| U.error(log, "Vacuum error.", err); |
| |
| res.onDone(err); |
| } |
| else if (snapshot.cleanupVersion() <= MVCC_COUNTER_NA) |
| res.onDone(new VacuumMetrics()); |
| else { |
| try { |
| if (log.isDebugEnabled()) |
| log.debug("Started vacuum with cleanup version=" + snapshot.cleanupVersion() + '.'); |
| |
| synchronized (mux) { |
| if (cleanupQueue == null) { |
| res.onDone(vacuumCancelledException()); |
| |
| return; |
| } |
| |
| GridCompoundIdentityFuture<VacuumMetrics> res0 = |
| new GridCompoundIdentityFuture<VacuumMetrics>(new VacuumMetricsReducer()) { |
| /** {@inheritDoc} */ |
| @Override protected void logError(IgniteLogger log, String msg, Throwable e) { |
| // no-op |
| } |
| |
| /** {@inheritDoc} */ |
| @Override protected void logDebug(IgniteLogger log, String msg) { |
| // no-op |
| } |
| }; |
| |
| for (CacheGroupContext grp : ctx.cache().cacheGroups()) { |
| if (grp.mvccEnabled()) { |
| grp.topology().readLock(); |
| |
| try { |
| for (GridDhtLocalPartition part : grp.topology().localPartitions()) { |
| VacuumTask task = new VacuumTask(snapshot, part); |
| |
| cleanupQueue.offer(task); |
| |
| res0.add(task); |
| } |
| } |
| finally { |
| grp.topology().readUnlock(); |
| } |
| } |
| } |
| |
| res0.markInitialized(); |
| |
| res0.listen(future -> { |
| VacuumMetrics metrics = null; Throwable ex = null; |
| |
| try { |
| metrics = future.get(); |
| |
| if (U.assertionsEnabled()) { |
| MvccCoordinator crd = currentCoordinator(); |
| |
| assert crd != null |
| && crd.coordinatorVersion() >= snapshot.coordinatorVersion(); |
| |
| for (TxKey key : waitMap.keySet()) { |
| if (!( key.major() == snapshot.coordinatorVersion() |
| && key.minor() > snapshot.cleanupVersion() |
| || key.major() > snapshot.coordinatorVersion())) { |
| byte state = state(key.major(), key.minor()); |
| |
| assert state == TxState.ABORTED : "tx state=" + state; |
| } |
| } |
| } |
| |
| txLog.removeUntil(snapshot.coordinatorVersion(), snapshot.cleanupVersion()); |
| |
| if (log.isDebugEnabled()) |
| log.debug("Vacuum completed. " + metrics); |
| } catch (Throwable e) { |
| if (X.hasCause(e, NodeStoppingException.class)) { |
| if (log.isDebugEnabled()) |
| log.debug("Cannot complete vacuum (node is stopping)."); |
| |
| metrics = new VacuumMetrics(); |
| } else |
| ex = new GridClosureException(e); |
| } |
| |
| res.onDone(metrics, ex); |
| }); |
| } |
| |
| } |
| catch (Throwable e) { |
| completeWithException(res, e); |
| } |
| } |
| } |
| }); |
| } |
| |
| /** */ |
| private void completeWithException(GridFutureAdapter fut, Throwable e) { |
| fut.onDone(e); |
| |
| if (e instanceof Error) |
| throw (Error)e; |
| } |
| |
| /** */ |
| @NotNull private IgniteCheckedException vacuumCancelledException() { |
| return new NodeStoppingException("Operation has been cancelled (node is stopping)."); |
| } |
| |
| /** |
| * @param nodeId Node ID. |
| * @param msg Message. |
| */ |
| private void sendFutureResponse(UUID nodeId, MvccWaitTxsRequest msg) { |
| try { |
| sendMessage(nodeId, new MvccFutureResponse(msg.futureId())); |
| } |
| catch (ClusterTopologyCheckedException e) { |
| if (log.isDebugEnabled()) |
| log.debug("Failed to send tx ack response, node left [msg=" + msg + ", node=" + nodeId + ']'); |
| } |
| catch (IgniteCheckedException e) { |
| U.error(log, "Failed to send tx ack response [msg=" + msg + ", node=" + nodeId + ']', e); |
| } |
| } |
| |
| /** */ |
| @NotNull private IgniteInternalFuture<Void> sendTxCommit(MvccCoordinator crd, MvccAckRequestTx msg) { |
| WaitAckFuture fut = new WaitAckFuture(msg.futureId(), crd.nodeId(), true); |
| |
| ackFuts.put(fut.id, fut); |
| |
| try { |
| sendMessage(crd.nodeId(), msg); |
| } |
| catch (IgniteCheckedException e) { |
| if (ackFuts.remove(fut.id) != null) { |
| if (e instanceof ClusterTopologyCheckedException) { |
| if (log.isDebugEnabled()) |
| log.debug("Failed to send tx ack, node left [crd=" + crd + ", msg=" + msg + ']'); |
| |
| fut.onDone(); // No need to ack, finish without error. |
| } |
| else |
| |
| fut.onDone(e); |
| } |
| } |
| |
| return fut; |
| } |
| |
| /** |
| * @param crd Mvcc coordinator. |
| * @param msg Message. |
| * @return {@code True} if no need to resend the message to a new coordinator. |
| */ |
| private boolean sendQueryDone(MvccCoordinator crd, Message msg) { |
| if (crd == null) |
| return true; // no need to send ack; |
| |
| try { |
| sendMessage(crd.nodeId(), msg); |
| |
| return true; |
| } |
| catch (ClusterTopologyCheckedException e) { |
| if (log.isDebugEnabled()) |
| log.debug("Failed to send query ack, node left [crd=" + crd + ", msg=" + msg + ']'); |
| |
| MvccCoordinator crd0 = currentCoordinator(); |
| |
| // Coordinator is unassigned or still the same. |
| return crd0 == null || crd.coordinatorVersion() == crd0.coordinatorVersion(); |
| } |
| catch (IgniteCheckedException e) { |
| U.error(log, "Failed to send query ack [crd=" + crd + ", msg=" + msg + ']', e); |
| |
| return true; |
| } |
| } |
| |
| /** |
| * Send IO message. |
| * |
| * @param nodeId Node ID. |
| * @param msg Message. |
| */ |
| private void sendMessage(UUID nodeId, Message msg) throws IgniteCheckedException { |
| ctx.io().sendToGridTopic(nodeId, TOPIC_CACHE_COORDINATOR, msg, SYSTEM_POOL); |
| } |
| |
| /** |
| * @param nodeId Sender node ID. |
| * @param msg Message. |
| */ |
| private void processCoordinatorTxSnapshotRequest(UUID nodeId, MvccTxSnapshotRequest msg) { |
| ClusterNode node = ctx.discovery().node(nodeId); |
| |
| if (node == null) { |
| if (log.isDebugEnabled()) |
| log.debug("Ignore tx snapshot request processing, node left [msg=" + msg + ", node=" + nodeId + ']'); |
| |
| return; |
| } |
| |
| MvccSnapshotResponse res = assignTxSnapshot(msg.futureId(), nodeId, node.isClient()); |
| |
| boolean finishFailed = true; |
| |
| try { |
| sendMessage(node.id(), res); |
| |
| finishFailed = false; |
| } |
| catch (ClusterTopologyCheckedException e) { |
| if (log.isDebugEnabled()) |
| log.debug("Failed to send tx snapshot response, node left [msg=" + msg + ", node=" + nodeId + ']'); |
| } |
| catch (IgniteCheckedException e) { |
| U.error(log, "Failed to send tx snapshot response [msg=" + msg + ", node=" + nodeId + ']', e); |
| } |
| |
| if (finishFailed) |
| onTxDone(res.counter(), false); |
| } |
| |
| /** |
| * @param nodeId Sender node ID. |
| * @param msg Message. |
| */ |
| private void processCoordinatorQuerySnapshotRequest(UUID nodeId, MvccQuerySnapshotRequest msg) { |
| ClusterNode node = ctx.discovery().node(nodeId); |
| |
| if (node == null) { |
| if (log.isDebugEnabled()) |
| log.debug("Ignore query counter request processing, node left [msg=" + msg + ", node=" + nodeId + ']'); |
| |
| return; |
| } |
| |
| MvccSnapshotResponse res = activeQueries.assignQueryCounter(nodeId, msg.futureId()); |
| |
| try { |
| sendMessage(node.id(), res); |
| } |
| catch (ClusterTopologyCheckedException e) { |
| if (log.isDebugEnabled()) |
| log.debug("Failed to send query counter response, node left [msg=" + msg + ", node=" + nodeId + ']'); |
| } |
| catch (IgniteCheckedException e) { |
| onQueryDone(nodeId, res.tracking()); |
| |
| U.error(log, "Failed to send query counter response [msg=" + msg + ", node=" + nodeId + ']', e); |
| } |
| } |
| |
| /** |
| * @param nodeId Sender node ID. |
| * @param msg Message. |
| */ |
| private void processCoordinatorSnapshotResponse(UUID nodeId, MvccSnapshotResponse msg) { |
| Map<Long, MvccSnapshotResponseListener> map = snapLsnrs.get(nodeId); |
| |
| MvccSnapshotResponseListener lsnr; |
| |
| if (map != null && (lsnr = map.remove(msg.futureId())) != null) |
| lsnr.onResponse(msg); |
| else { |
| if (ctx.discovery().alive(nodeId)) |
| U.warn(log, "Failed to find query version future [node=" + nodeId + ", msg=" + msg + ']'); |
| else if (log.isDebugEnabled()) |
| log.debug("Failed to find query version future [node=" + nodeId + ", msg=" + msg + ']'); |
| } |
| } |
| |
| /** |
| * @param nodeId Node ID. |
| * @param msg Message. |
| */ |
| private void processCoordinatorQueryAckRequest(UUID nodeId, MvccAckRequestQueryCntr msg) { |
| onQueryDone(nodeId, msg.counter()); |
| } |
| |
| /** |
| * @param nodeId Node ID. |
| * @param msg Message. |
| */ |
| private void processNewCoordinatorQueryAckRequest(UUID nodeId, MvccAckRequestQueryId msg) { |
| prevCrdQueries.onQueryDone(nodeId, msg.queryTrackerId()); |
| } |
| |
| /** |
| * @param nodeId Sender node ID. |
| * @param msg Message. |
| */ |
| private void processCoordinatorTxAckRequest(UUID nodeId, MvccAckRequestTx msg) { |
| onTxDone(msg.txCounter(), msg.futureId() >= 0); |
| |
| if (msg.queryCounter() != MVCC_COUNTER_NA) |
| onQueryDone(nodeId, msg.queryCounter()); |
| else if (msg.queryTrackerId() != MVCC_TRACKER_ID_NA) |
| prevCrdQueries.onQueryDone(nodeId, msg.queryTrackerId()); |
| |
| if (!msg.skipResponse()) { |
| try { |
| sendMessage(nodeId, new MvccFutureResponse(msg.futureId())); |
| } |
| catch (ClusterTopologyCheckedException e) { |
| if (log.isDebugEnabled()) |
| log.debug("Failed to send tx ack response, node left [msg=" + msg + ", node=" + nodeId + ']'); |
| } |
| catch (IgniteCheckedException e) { |
| U.error(log, "Failed to send tx ack response [msg=" + msg + ", node=" + nodeId + ']', e); |
| } |
| } |
| } |
| |
| /** |
| * @param nodeId Sender node ID. |
| * @param msg Message. |
| */ |
| private void processCoordinatorAckResponse(UUID nodeId, MvccFutureResponse msg) { |
| WaitAckFuture fut = ackFuts.remove(msg.futureId()); |
| |
| if (fut != null) |
| fut.onResponse(); |
| else { |
| if (ctx.discovery().alive(nodeId)) |
| U.warn(log, "Failed to find tx ack future [node=" + nodeId + ", msg=" + msg + ']'); |
| else if (log.isDebugEnabled()) |
| log.debug("Failed to find tx ack future [node=" + nodeId + ", msg=" + msg + ']'); |
| } |
| } |
| |
| /** |
| * @param nodeId Node ID. |
| * @param msg Message. |
| */ |
| @SuppressWarnings("unchecked") |
| private void processCoordinatorWaitTxsRequest(final UUID nodeId, final MvccWaitTxsRequest msg) { |
| GridLongList txs = msg.transactions(); |
| |
| GridCompoundFuture resFut = null; |
| |
| for (int i = 0; i < txs.size(); i++) { |
| Long txId = txs.get(i); |
| |
| GridFutureAdapter fut = waitTxFuts.get(txId); |
| |
| if (fut == null) { |
| GridFutureAdapter old = waitTxFuts.putIfAbsent(txId, fut = new GridFutureAdapter()); |
| |
| if (old != null) |
| fut = old; |
| } |
| |
| boolean isDone; |
| |
| synchronized (this) { |
| isDone = !activeTxs.containsKey(txId); |
| } |
| |
| if (isDone) |
| fut.onDone(); |
| |
| if (!fut.isDone()) { |
| if (resFut == null) |
| resFut = new GridCompoundFuture(); |
| |
| resFut.add(fut); |
| } |
| } |
| |
| if (resFut != null) |
| resFut.markInitialized(); |
| |
| if (resFut == null || resFut.isDone()) |
| sendFutureResponse(nodeId, msg); |
| else { |
| resFut.listen(new IgniteInClosure<IgniteInternalFuture>() { |
| @Override public void apply(IgniteInternalFuture fut) { |
| sendFutureResponse(nodeId, msg); |
| } |
| }); |
| } |
| } |
| |
| /** |
| * @param nodeId Node ID. |
| * @param msg Message. |
| */ |
| private void processCoordinatorActiveQueriesMessage(UUID nodeId, MvccActiveQueriesMessage msg) { |
| prevCrdQueries.addNodeActiveQueries(nodeId, msg.activeQueries()); |
| } |
| |
| /** |
| * |
| */ |
| private class ActiveQueries { |
| /** */ |
| private final Map<UUID, TreeMap<Long, AtomicInteger>> activeQueries = new HashMap<>(); |
| |
| /** */ |
| private Long minQry; |
| |
| /** */ |
| private synchronized long minimalQueryCounter() { |
| return minQry == null ? -1 : minQry; |
| } |
| |
| /** */ |
| private synchronized MvccSnapshotResponse assignQueryCounter(UUID nodeId, long futId) { |
| MvccSnapshotResponse res = new MvccSnapshotResponse(); |
| |
| long ver, tracking; |
| |
| synchronized (MvccProcessorImpl.this) { |
| ver = committedCntr.get(); |
| tracking = ver; |
| |
| for (Long txVer : activeTxs.keySet()) { |
| if (txVer < ver) { |
| tracking = Math.min(txVer, tracking); |
| res.addTx(txVer); |
| } |
| } |
| } |
| |
| TreeMap<Long, AtomicInteger> nodeMap = activeQueries.get(nodeId); |
| |
| if (nodeMap == null) { |
| activeQueries.put(nodeId, nodeMap = new TreeMap<>()); |
| |
| nodeMap.put(tracking, new AtomicInteger(1)); |
| } |
| else { |
| AtomicInteger cntr = nodeMap.get(tracking); |
| |
| if (cntr == null) |
| nodeMap.put(tracking, new AtomicInteger(1)); |
| else |
| cntr.incrementAndGet(); |
| } |
| |
| if (minQry == null) |
| minQry = tracking; |
| |
| res.init(futId, crdVer, ver, MVCC_READ_OP_CNTR, MVCC_COUNTER_NA, tracking); |
| |
| return res; |
| } |
| |
| /** */ |
| private synchronized void onQueryDone(UUID nodeId, Long ver) { |
| TreeMap<Long, AtomicInteger> nodeMap = activeQueries.get(nodeId); |
| |
| if (nodeMap == null) |
| return; |
| |
| assert minQry != null; |
| |
| AtomicInteger cntr = nodeMap.get(ver); |
| |
| assert cntr != null && cntr.get() > 0 : "onQueryDone ver=" + ver; |
| |
| if (cntr.decrementAndGet() == 0) { |
| nodeMap.remove(ver); |
| |
| if (nodeMap.isEmpty()) |
| activeQueries.remove(nodeId); |
| |
| if (ver.equals(minQry)) |
| minQry = activeMinimal(); |
| } |
| } |
| |
| /** */ |
| private synchronized void onNodeFailed(UUID nodeId) { |
| activeQueries.remove(nodeId); |
| |
| minQry = activeMinimal(); |
| } |
| |
| /** */ |
| private Long activeMinimal() { |
| Long min = null; |
| |
| for (TreeMap<Long, AtomicInteger> s : activeQueries.values()) { |
| Long first = s.firstKey(); |
| |
| if (min == null || first < min) |
| min = first; |
| } |
| |
| return min; |
| } |
| } |
| |
| /** |
| * |
| */ |
| private class WaitAckFuture extends MvccFuture<Void> { |
| /** */ |
| private final long id; |
| |
| /** */ |
| final boolean ackTx; |
| |
| /** |
| * @param id Future ID. |
| * @param nodeId Coordinator node ID. |
| * @param ackTx {@code True} if ack tx commit, {@code false} if waits for previous txs. |
| */ |
| WaitAckFuture(long id, UUID nodeId, boolean ackTx) { |
| super(nodeId); |
| |
| this.id = id; |
| this.ackTx = ackTx; |
| } |
| |
| /** |
| * |
| */ |
| void onResponse() { |
| onDone(); |
| } |
| |
| /** |
| * @param nodeId Failed node ID. |
| */ |
| void onNodeLeft(UUID nodeId) { |
| if (crdId.equals(nodeId) && ackFuts.remove(id) != null) |
| onDone(); |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public String toString() { |
| return S.toString(WaitAckFuture.class, this, super.toString()); |
| } |
| } |
| |
| /** |
| * |
| */ |
| private class CacheCoordinatorNodeFailListener implements GridLocalEventListener { |
| /** {@inheritDoc} */ |
| @Override public void onEvent(Event evt) { |
| assert evt instanceof DiscoveryEvent : evt; |
| |
| DiscoveryEvent discoEvt = (DiscoveryEvent)evt; |
| |
| UUID nodeId = discoEvt.eventNode().id(); |
| |
| Map<Long, MvccSnapshotResponseListener> map = snapLsnrs.remove(nodeId); |
| |
| if (map != null) { |
| ClusterTopologyCheckedException ex = new ClusterTopologyCheckedException("Failed to request mvcc " + |
| "version, coordinator failed: " + nodeId); |
| |
| MvccSnapshotResponseListener lsnr; |
| |
| for (Long id : map.keySet()) { |
| if ((lsnr = map.remove(id)) != null) |
| lsnr.onError(ex); |
| } |
| } |
| |
| for (WaitAckFuture fut : ackFuts.values()) |
| fut.onNodeLeft(nodeId); |
| |
| activeQueries.onNodeFailed(nodeId); |
| |
| prevCrdQueries.onNodeFailed(nodeId); |
| |
| recoveryBallotBoxes.forEach((nearNodeId, ballotBox) -> { |
| // Put synthetic vote from another failed node |
| ballotBox.vote(nodeId); |
| |
| tryFinishRecoveryVoting(nearNodeId, ballotBox); |
| }); |
| |
| if (discoEvt.eventNode().isClient()) { |
| RecoveryBallotBox ballotBox = recoveryBallotBoxes |
| .computeIfAbsent(nodeId, uuid -> new RecoveryBallotBox()); |
| |
| ballotBox |
| .voters(discoEvt.topologyNodes().stream().map(ClusterNode::id).collect(Collectors.toList())); |
| |
| tryFinishRecoveryVoting(nodeId, ballotBox); |
| } |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public String toString() { |
| return "CacheCoordinatorDiscoveryListener[]"; |
| } |
| } |
| |
| /** |
| * |
| */ |
| private class CoordinatorMessageListener implements GridMessageListener { |
| /** {@inheritDoc} */ |
| @Override public void onMessage(UUID nodeId, Object msg, byte plc) { |
| MvccMessage msg0 = (MvccMessage)msg; |
| |
| if (msg0.waitForCoordinatorInit() && !initFut.isDone()) { |
| initFut.listen(new IgniteInClosure<IgniteInternalFuture<Void>>() { |
| @Override public void apply(IgniteInternalFuture<Void> future) { |
| assert crdVer != 0L; |
| |
| processMessage(nodeId, msg); |
| } |
| }); |
| } |
| else |
| processMessage(nodeId, msg); |
| } |
| |
| /** |
| * Processes mvcc message. |
| * |
| * @param nodeId Node id. |
| * @param msg Message. |
| */ |
| private void processMessage(UUID nodeId, Object msg) { |
| if (msg instanceof MvccTxSnapshotRequest) |
| processCoordinatorTxSnapshotRequest(nodeId, (MvccTxSnapshotRequest)msg); |
| else if (msg instanceof MvccAckRequestTx) |
| processCoordinatorTxAckRequest(nodeId, (MvccAckRequestTx)msg); |
| else if (msg instanceof MvccFutureResponse) |
| processCoordinatorAckResponse(nodeId, (MvccFutureResponse)msg); |
| else if (msg instanceof MvccAckRequestQueryCntr) |
| processCoordinatorQueryAckRequest(nodeId, (MvccAckRequestQueryCntr)msg); |
| else if (msg instanceof MvccQuerySnapshotRequest) |
| processCoordinatorQuerySnapshotRequest(nodeId, (MvccQuerySnapshotRequest)msg); |
| else if (msg instanceof MvccSnapshotResponse) |
| processCoordinatorSnapshotResponse(nodeId, (MvccSnapshotResponse)msg); |
| else if (msg instanceof MvccWaitTxsRequest) |
| processCoordinatorWaitTxsRequest(nodeId, (MvccWaitTxsRequest)msg); |
| else if (msg instanceof MvccAckRequestQueryId) |
| processNewCoordinatorQueryAckRequest(nodeId, (MvccAckRequestQueryId)msg); |
| else if (msg instanceof MvccActiveQueriesMessage) |
| processCoordinatorActiveQueriesMessage(nodeId, (MvccActiveQueriesMessage)msg); |
| else if (msg instanceof MvccRecoveryFinishedMessage) |
| processRecoveryFinishedMessage(nodeId, ((MvccRecoveryFinishedMessage)msg)); |
| else |
| U.warn(log, "Unexpected message received [node=" + nodeId + ", msg=" + msg + ']'); |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public String toString() { |
| return "CoordinatorMessageListener[]"; |
| } |
| } |
| |
| /** |
| * Accumulates transaction recovery votes for a node left the cluster. |
| * Transactions started by the left node are considered not active |
| * when each cluster server node aknowledges that is has finished transactions for the left node. |
| */ |
| private static class RecoveryBallotBox { |
| /** */ |
| private List<UUID> voters; |
| /** */ |
| private final Set<UUID> ballots = new HashSet<>(); |
| |
| /** |
| * @param voters Nodes which can have transaction started by the left node. |
| */ |
| private synchronized void voters(List<UUID> voters) { |
| this.voters = voters; |
| } |
| |
| /** |
| * @param nodeId Voting node id. |
| * |
| */ |
| private synchronized void vote(UUID nodeId) { |
| ballots.add(nodeId); |
| } |
| |
| /** |
| * @return {@code True} if all nodes expected to vote done it. |
| */ |
| private synchronized boolean isVotingDone() { |
| if (voters == null) |
| return false; |
| |
| return ballots.containsAll(voters); |
| } |
| } |
| |
| /** |
| * Process message that one node has finished with transactions for the left node. |
| * @param nodeId Node sent the message. |
| * @param msg Message. |
| */ |
| private void processRecoveryFinishedMessage(UUID nodeId, MvccRecoveryFinishedMessage msg) { |
| UUID nearNodeId = msg.nearNodeId(); |
| |
| RecoveryBallotBox ballotBox = recoveryBallotBoxes.computeIfAbsent(nearNodeId, uuid -> new RecoveryBallotBox()); |
| |
| ballotBox.vote(nodeId); |
| |
| tryFinishRecoveryVoting(nearNodeId, ballotBox); |
| } |
| |
| /** |
| * Finishes recovery on coordinator by removing transactions started by the left node |
| * @param nearNodeId Left node. |
| * @param ballotBox Votes accumulator for the left node. |
| */ |
| private void tryFinishRecoveryVoting(UUID nearNodeId, RecoveryBallotBox ballotBox) { |
| if (ballotBox.isVotingDone()) { |
| List<Long> recoveredTxs; |
| |
| synchronized (this) { |
| recoveredTxs = activeTxs.entrySet().stream() |
| .filter(e -> e.getValue().nearNodeId.equals(nearNodeId)) |
| .map(Map.Entry::getKey) |
| .collect(Collectors.toList()); |
| } |
| |
| // Committed counter is increased because it is not known if transaction was committed or not and we must |
| // bump committed counter for committed transaction as it is used in (read-only) query snapshot. |
| recoveredTxs.forEach(txCntr -> onTxDone(txCntr, true)); |
| |
| recoveryBallotBoxes.remove(nearNodeId); |
| } |
| } |
| |
| /** */ |
| private interface Waiter { |
| /** |
| * @param ctx Grid kernal context. |
| */ |
| void run(GridKernalContext ctx); |
| |
| /** |
| * @param other Another waiter. |
| * @return New compound waiter. |
| */ |
| Waiter concat(Waiter other); |
| |
| /** |
| * @return {@code True} if there is an active local transaction |
| */ |
| boolean hasLocalTransaction(); |
| |
| /** |
| * @return {@code True} if it is a compound waiter. |
| */ |
| boolean compound(); |
| } |
| |
| /** */ |
| private static class LockFuture extends GridFutureAdapter<Void> implements Waiter, Runnable { |
| /** */ |
| private final byte plc; |
| |
| /** |
| * @param plc Pool policy. |
| */ |
| LockFuture(byte plc) { |
| this.plc = plc; |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public void run() { |
| onDone(); |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public void run(GridKernalContext ctx) { |
| try { |
| ctx.pools().poolForPolicy(plc).execute(this); |
| } |
| catch (IgniteCheckedException e) { |
| U.error(ctx.log(LockFuture.class), e); |
| } |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public Waiter concat(Waiter other) { |
| return new CompoundWaiterNoLocal(this, other); |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public boolean hasLocalTransaction() { |
| return false; |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public boolean compound() { |
| return false; |
| } |
| } |
| |
| /** */ |
| private static class LocalTransactionMarker implements Waiter { |
| /** {@inheritDoc} */ |
| @Override public void run(GridKernalContext ctx) { |
| // No-op |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public Waiter concat(Waiter other) { |
| return new CompoundWaiter(other); |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public boolean hasLocalTransaction() { |
| return true; |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public boolean compound() { |
| return false; |
| } |
| } |
| |
| /** */ |
| @SuppressWarnings("unchecked") |
| private static class CompoundWaiter implements Waiter { |
| /** */ |
| private final Object inner; |
| |
| /** |
| * @param waiter Waiter to wrap. |
| */ |
| private CompoundWaiter(Waiter waiter) { |
| inner = waiter.compound() ? ((CompoundWaiter)waiter).inner : waiter; |
| } |
| |
| /** |
| * @param first First waiter. |
| * @param second Second waiter. |
| */ |
| private CompoundWaiter(Waiter first, Waiter second) { |
| ArrayList<Waiter> list = new ArrayList<>(); |
| |
| add(list, first); |
| add(list, second); |
| |
| inner = list; |
| } |
| |
| /** */ |
| private void add(List<Waiter> to, Waiter waiter) { |
| if (!waiter.compound()) |
| to.add(waiter); |
| else if (((CompoundWaiter)waiter).inner.getClass() == ArrayList.class) |
| to.addAll((List<Waiter>)((CompoundWaiter)waiter).inner); |
| else |
| to.add((Waiter)((CompoundWaiter)waiter).inner); |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public void run(GridKernalContext ctx) { |
| if (inner.getClass() == ArrayList.class) { |
| for (Waiter waiter : (List<Waiter>)inner) { |
| waiter.run(ctx); |
| } |
| } |
| else |
| ((Waiter)inner).run(ctx); |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public Waiter concat(Waiter other) { |
| return new CompoundWaiter(this, other); |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public boolean hasLocalTransaction() { |
| return true; |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public boolean compound() { |
| return true; |
| } |
| } |
| |
| /** */ |
| private static class CompoundWaiterNoLocal extends CompoundWaiter { |
| /** |
| * @param first First waiter. |
| * @param second Second waiter. |
| */ |
| private CompoundWaiterNoLocal(Waiter first, Waiter second) { |
| super(first, second); |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public Waiter concat(Waiter other) { |
| return new CompoundWaiterNoLocal(this, other); |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public boolean hasLocalTransaction() { |
| return false; |
| } |
| } |
| |
| /** |
| * Mvcc garbage collection scheduler. |
| */ |
| private static class VacuumScheduler extends GridWorker { |
| /** */ |
| private final static long VACUUM_TIMEOUT = 60_000; |
| |
| /** */ |
| private final long interval; |
| |
| /** */ |
| private final MvccProcessorImpl prc; |
| |
| /** |
| * @param ctx Kernal context. |
| * @param log Logger. |
| * @param prc Mvcc processor. |
| */ |
| VacuumScheduler(GridKernalContext ctx, IgniteLogger log, MvccProcessorImpl prc) { |
| super(ctx.igniteInstanceName(), "vacuum-scheduler", log); |
| |
| this.interval = ctx.config().getMvccVacuumFrequency(); |
| this.prc = prc; |
| } |
| |
| /** {@inheritDoc} */ |
| @Override protected void body() throws InterruptedException, IgniteInterruptedCheckedException { |
| U.sleep(interval); // initial delay |
| |
| while (!isCancelled()) { |
| long nextScheduledTime = U.currentTimeMillis() + interval; |
| |
| try { |
| IgniteInternalFuture<VacuumMetrics> fut = prc.runVacuum(); |
| |
| if (log.isDebugEnabled()) |
| log.debug("Vacuum started by scheduler."); |
| |
| while (true) { |
| try { |
| fut.get(VACUUM_TIMEOUT); |
| |
| break; |
| } |
| catch (IgniteFutureTimeoutCheckedException e) { |
| U.warn(log, "Failed to wait for vacuum complete. Consider increasing vacuum workers count."); |
| } |
| } |
| } |
| catch (IgniteInterruptedCheckedException e) { |
| throw e; // Cancelled. |
| } |
| catch (Throwable e) { |
| if (e instanceof Error) |
| throw (Error) e; |
| |
| U.error(log, "Vacuum error.", e); |
| } |
| |
| long delay = nextScheduledTime - U.currentTimeMillis(); |
| |
| if (delay > 0) |
| U.sleep(delay); |
| } |
| } |
| } |
| |
| /** |
| * Vacuum worker. |
| */ |
| private static class VacuumWorker extends GridWorker { |
| /** */ |
| private final BlockingQueue<VacuumTask> cleanupQueue; |
| |
| /** |
| * @param ctx Kernal context. |
| * @param log Logger. |
| * @param cleanupQueue Cleanup tasks queue. |
| */ |
| VacuumWorker(GridKernalContext ctx, IgniteLogger log, BlockingQueue<VacuumTask> cleanupQueue) { |
| super(ctx.igniteInstanceName(), "vacuum-cleaner", log); |
| |
| this.cleanupQueue = cleanupQueue; |
| } |
| |
| /** {@inheritDoc} */ |
| @Override protected void body() throws InterruptedException, IgniteInterruptedCheckedException { |
| while (!isCancelled()) { |
| VacuumTask task = cleanupQueue.take(); |
| |
| try { |
| switch (task.part().state()) { |
| case EVICTED: |
| case RENTING: |
| task.onDone(new VacuumMetrics()); |
| |
| break; |
| case MOVING: |
| task.part().group().preloader().rebalanceFuture().listen(f -> cleanupQueue.add(task)); |
| |
| break; |
| case OWNING: |
| task.onDone(processPartition(task)); |
| |
| break; |
| case LOST: |
| task.onDone(new IgniteCheckedException("Partition is lost.")); |
| |
| break; |
| } |
| } |
| catch (IgniteInterruptedCheckedException e) { |
| task.onDone(e); |
| |
| throw e; // Cancelled. |
| } |
| catch (Throwable e) { |
| task.onDone(e); |
| |
| if (e instanceof Error) |
| throw (Error) e; |
| } |
| } |
| } |
| |
| /** |
| * Process partition. |
| * |
| * @param task VacuumTask. |
| * @throws IgniteCheckedException If failed. |
| */ |
| private VacuumMetrics processPartition(VacuumTask task) throws IgniteCheckedException { |
| long startNanoTime = System.nanoTime(); |
| |
| GridDhtLocalPartition part = task.part(); |
| |
| VacuumMetrics metrics = new VacuumMetrics(); |
| |
| if (!part.reserve()) |
| return metrics; |
| |
| int curCacheId = CU.UNDEFINED_CACHE_ID; |
| |
| try { |
| GridCursor<? extends CacheDataRow> cursor = part.dataStore().cursor(KEY_ONLY); |
| |
| KeyCacheObject prevKey = null; |
| |
| Object rest = null; |
| |
| List<MvccLinkAwareSearchRow> cleanupRows = null; |
| |
| MvccSnapshot snapshot = task.snapshot(); |
| |
| GridCacheContext cctx = null; |
| |
| boolean shared = part.group().sharedGroup(); |
| |
| if (!shared && (cctx = F.first(part.group().caches())) == null) |
| return metrics; |
| |
| while (cursor.next()) { |
| if (isCancelled()) |
| throw new IgniteInterruptedCheckedException("Operation has been cancelled."); |
| |
| MvccDataRow row = (MvccDataRow)cursor.get(); |
| |
| if (prevKey == null) |
| prevKey = row.key(); |
| |
| if (cctx == null) { |
| cctx = part.group().shared().cacheContext(curCacheId = row.cacheId()); |
| |
| if (cctx == null) |
| return metrics; |
| } |
| |
| if (!prevKey.equals(row.key()) || (shared && curCacheId != row.cacheId())) { |
| if (rest != null || !F.isEmpty(cleanupRows)) |
| cleanup(part, prevKey, cleanupRows, rest, cctx, metrics); |
| |
| cleanupRows = null; |
| |
| rest = null; |
| |
| if (shared && curCacheId != row.cacheId()) { |
| cctx = part.group().shared().cacheContext(curCacheId = row.cacheId()); |
| |
| if (cctx == null) |
| return metrics; |
| } |
| |
| prevKey = row.key(); |
| } |
| |
| if (canClean(row, snapshot, cctx)) |
| cleanupRows = addRow(cleanupRows, row); |
| else if (actualize(cctx, row, snapshot)) |
| rest = addRest(rest, row); |
| |
| metrics.addScannedRowsCount(1); |
| } |
| |
| if (rest != null || !F.isEmpty(cleanupRows)) |
| cleanup(part, prevKey, cleanupRows, rest, cctx, metrics); |
| |
| metrics.addSearchNanoTime(System.nanoTime() - startNanoTime - metrics.cleanupNanoTime()); |
| |
| return metrics; |
| } |
| finally { |
| part.release(); |
| } |
| } |
| |
| /** */ |
| @SuppressWarnings("unchecked") |
| @NotNull private Object addRest(@Nullable Object rest, MvccDataRow row) { |
| if (rest == null) |
| rest = row; |
| else if (rest.getClass() == ArrayList.class) |
| ((List)rest).add(row); |
| else { |
| ArrayList list = new ArrayList(); |
| |
| list.add(rest); |
| list.add(row); |
| |
| rest = list; |
| } |
| |
| return rest; |
| } |
| |
| /** |
| * @param rows Collection of rows. |
| * @param row Row to add. |
| * @return Collection of rows. |
| */ |
| @NotNull private List<MvccLinkAwareSearchRow> addRow(@Nullable List<MvccLinkAwareSearchRow> rows, MvccDataRow row) { |
| if (rows == null) |
| rows = new ArrayList<>(); |
| |
| rows.add(new MvccLinkAwareSearchRow(row.cacheId(), row.key(), row.mvccCoordinatorVersion(), |
| row.mvccCounter(), row.mvccOperationCounter(), row.link())); |
| |
| return rows; |
| } |
| |
| /** |
| * @param row Mvcc row to check. |
| * @param snapshot Cleanup version to compare with. |
| * @param cctx Cache context. |
| * @throws IgniteCheckedException If failed. |
| */ |
| private boolean canClean(MvccDataRow row, MvccSnapshot snapshot, |
| GridCacheContext cctx) throws IgniteCheckedException { |
| // Row can be safely cleaned if it has ABORTED min version or COMMITTED and less than cleanup one max version. |
| return compare(row, snapshot.coordinatorVersion(), snapshot.cleanupVersion()) <= 0 |
| && hasNewVersion(row) && MvccUtils.compareNewVersion(row, snapshot.coordinatorVersion(), snapshot.cleanupVersion()) <= 0 |
| && MvccUtils.state(cctx, row.newMvccCoordinatorVersion(), row.newMvccCounter(), |
| row.newMvccOperationCounter() | (row.newMvccTxState() << MVCC_HINTS_BIT_OFF)) == TxState.COMMITTED |
| || MvccUtils.state(cctx, row.mvccCoordinatorVersion(), row.mvccCounter(), |
| row.mvccOperationCounter() | (row.mvccTxState() << MVCC_HINTS_BIT_OFF)) == TxState.ABORTED; |
| } |
| |
| /** */ |
| private boolean actualize(GridCacheContext cctx, MvccDataRow row, |
| MvccSnapshot snapshot) throws IgniteCheckedException { |
| return isVisible(cctx, snapshot, row.mvccCoordinatorVersion(), row.mvccCounter(), row.mvccOperationCounter(), false) |
| && (row.mvccTxState() == TxState.NA || (row.newMvccCoordinatorVersion() != MVCC_CRD_COUNTER_NA && row.newMvccTxState() == TxState.NA)); |
| } |
| |
| /** |
| * @param part Local partition. |
| * @param key Key. |
| * @param cleanupRows Cleanup rows. |
| * @param cctx Cache context. |
| * @param metrics Vacuum metrics. |
| * @throws IgniteCheckedException If failed. |
| */ |
| @SuppressWarnings("unchecked") |
| private void cleanup(GridDhtLocalPartition part, KeyCacheObject key, List<MvccLinkAwareSearchRow> cleanupRows, |
| Object rest, GridCacheContext cctx, VacuumMetrics metrics) throws IgniteCheckedException { |
| assert key != null && cctx != null && (!F.isEmpty(cleanupRows) || rest != null); |
| |
| cctx.gate().enter(); |
| |
| try { |
| long cleanupStartNanoTime = System.nanoTime(); |
| |
| GridCacheEntryEx entry = cctx.cache().entryEx(key); |
| |
| while (true) { |
| entry.lockEntry(); |
| |
| if (!entry.obsolete()) |
| break; |
| |
| entry.unlockEntry(); |
| |
| entry = cctx.cache().entryEx(key); |
| } |
| |
| int cleaned = 0; |
| |
| try { |
| cctx.shared().database().checkpointReadLock(); |
| |
| try { |
| if (cleanupRows != null) |
| cleaned = part.dataStore().cleanup(cctx, cleanupRows); |
| |
| if (rest != null) { |
| if (rest.getClass() == ArrayList.class) { |
| for (MvccDataRow row : ((List<MvccDataRow>) rest)) { |
| part.dataStore().updateTxState(cctx, row); |
| } |
| } else |
| part.dataStore().updateTxState(cctx, (MvccDataRow) rest); |
| } |
| } finally { |
| cctx.shared().database().checkpointReadUnlock(); |
| } |
| } finally { |
| entry.unlockEntry(); |
| cctx.evicts().touch(entry, AffinityTopologyVersion.NONE); |
| |
| metrics.addCleanupNanoTime(System.nanoTime() - cleanupStartNanoTime); |
| metrics.addCleanupRowsCnt(cleaned); |
| } |
| } |
| finally { |
| cctx.gate().leave(); |
| } |
| } |
| } |
| |
| /** */ |
| private static class ActiveTx { |
| /** */ |
| private final long tracking; |
| /** */ |
| private final UUID nearNodeId; |
| |
| /** */ |
| private ActiveTx(long tracking, UUID nearNodeId) { |
| this.tracking = tracking; |
| this.nearNodeId = nearNodeId; |
| } |
| } |
| |
| /** */ |
| private static class ActiveServerTx extends ActiveTx { |
| /** */ |
| private ActiveServerTx(long tracking, UUID nearNodeId) { |
| super(tracking, nearNodeId); |
| } |
| } |
| } |
