| /* |
| * 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 org.apache.ignite.IgniteCheckedException; |
| import org.apache.ignite.IgniteException; |
| import org.apache.ignite.configuration.TransactionConfiguration; |
| import org.apache.ignite.internal.GridKernalContext; |
| import org.apache.ignite.internal.cluster.ClusterTopologyServerNotFoundException; |
| import org.apache.ignite.internal.pagemem.PageMemory; |
| import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion; |
| import org.apache.ignite.internal.processors.cache.CacheGroupContext; |
| import org.apache.ignite.internal.processors.cache.GridCacheContext; |
| import org.apache.ignite.internal.processors.cache.distributed.near.GridNearTxLocal; |
| import org.apache.ignite.internal.processors.cache.mvcc.txlog.TxState; |
| import org.apache.ignite.internal.processors.cache.persistence.tree.io.DataPageIO; |
| import org.apache.ignite.internal.processors.cache.query.IgniteQueryErrorCode; |
| import org.apache.ignite.internal.processors.cache.transactions.IgniteInternalTx; |
| import org.apache.ignite.internal.processors.cache.transactions.IgniteTxManager; |
| import org.apache.ignite.internal.processors.cache.version.GridCacheVersion; |
| import org.apache.ignite.internal.processors.query.IgniteSQLException; |
| import org.apache.ignite.internal.transactions.IgniteTxMvccVersionCheckedException; |
| import org.apache.ignite.internal.util.typedef.internal.CU; |
| import org.apache.ignite.transactions.TransactionState; |
| import org.jetbrains.annotations.NotNull; |
| import org.jetbrains.annotations.Nullable; |
| |
| import static org.apache.ignite.cache.CacheWriteSynchronizationMode.FULL_SYNC; |
| import static org.apache.ignite.internal.pagemem.PageIdUtils.itemId; |
| import static org.apache.ignite.internal.pagemem.PageIdUtils.pageId; |
| import static org.apache.ignite.internal.processors.cache.persistence.tree.io.DataPageIO.MVCC_INFO_SIZE; |
| import static org.apache.ignite.internal.processors.cache.query.IgniteQueryErrorCode.TRANSACTION_COMPLETED; |
| import static org.apache.ignite.transactions.TransactionConcurrency.PESSIMISTIC; |
| import static org.apache.ignite.transactions.TransactionIsolation.REPEATABLE_READ; |
| |
| /** |
| * Utils for MVCC. |
| */ |
| public class MvccUtils { |
| /** */ |
| public static final int MVCC_KEY_ABSENT_BEFORE_OFF = 29; |
| |
| /** */ |
| public static final int MVCC_KEY_ABSENT_BEFORE_MASK = 0b0001 << MVCC_KEY_ABSENT_BEFORE_OFF; |
| |
| /** */ |
| public static final int MVCC_HINTS_BIT_OFF = MVCC_KEY_ABSENT_BEFORE_OFF + 1; |
| |
| /** */ |
| public static final int MVCC_HINTS_MASK = 0b0011 << MVCC_HINTS_BIT_OFF; |
| |
| /** Mask for all masked high bits in operation counter field. */ |
| public static final int MVCC_OP_COUNTER_MASK = 0b0111 << MVCC_KEY_ABSENT_BEFORE_OFF; |
| |
| /** */ |
| public static final long MVCC_CRD_COUNTER_NA = 0L; |
| |
| /** */ |
| public static final long MVCC_CRD_START_CNTR = 1L; |
| |
| /** */ |
| public static final long MVCC_COUNTER_NA = 0L; |
| |
| /** */ |
| public static final long MVCC_INITIAL_CNTR = 1L; |
| |
| /** */ |
| public static final long MVCC_START_CNTR = 3L; |
| |
| /** */ |
| public static final int MVCC_OP_COUNTER_NA = 0; |
| |
| /** */ |
| public static final int MVCC_START_OP_CNTR = 1; |
| |
| /** */ |
| public static final int MVCC_READ_OP_CNTR = ~MVCC_HINTS_MASK; |
| |
| /** */ |
| public static final int MVCC_INVISIBLE = 0; |
| |
| /** */ |
| public static final int MVCC_VISIBLE_REMOVED = 1; |
| |
| /** */ |
| public static final int MVCC_VISIBLE = 2; |
| |
| /** */ |
| public static final MvccVersion INITIAL_VERSION = |
| mvccVersion(MVCC_CRD_START_CNTR, MVCC_INITIAL_CNTR, MVCC_START_OP_CNTR); |
| |
| /** */ |
| public static final MvccVersion MVCC_VERSION_NA = |
| mvccVersion(MVCC_CRD_COUNTER_NA, MVCC_COUNTER_NA, MVCC_OP_COUNTER_NA); |
| |
| /** */ |
| private static final MvccClosure<Integer> getVisibleState = new GetVisibleState(); |
| |
| /** */ |
| private static final MvccClosure<Boolean> isVisible = new IsVisible(); |
| |
| /** */ |
| private static final MvccClosure<MvccVersion> getNewVer = new GetNewVersion(); |
| |
| /** |
| * |
| */ |
| private MvccUtils(){ |
| } |
| |
| /** |
| * @param cctx Cache context. |
| * @param mvccCrd Mvcc coordinator version. |
| * @param mvccCntr Mvcc counter. |
| * @param snapshot Snapshot. |
| * @return {@code True} if transaction is active. |
| * @see TxState |
| * @throws IgniteCheckedException If failed. |
| */ |
| public static boolean isActive(GridCacheContext cctx, long mvccCrd, long mvccCntr, MvccSnapshot snapshot) |
| throws IgniteCheckedException { |
| if (isVisible(cctx, snapshot, mvccCrd, mvccCntr, MVCC_OP_COUNTER_NA, false)) |
| return false; |
| |
| byte state = state(cctx, mvccCrd, mvccCntr, 0); |
| |
| return state != TxState.COMMITTED && state != TxState.ABORTED |
| || cctx.kernalContext().coordinators().hasLocalTransaction(mvccCrd, mvccCntr); |
| } |
| |
| /** |
| * @param cctx Cache context. |
| * @param mvccCrd Mvcc coordinator version. |
| * @param mvccCntr Mvcc counter. |
| * @param mvccOpCntr Mvcc operation counter. |
| * @return TxState |
| * @see TxState |
| * @throws IgniteCheckedException If failed. |
| */ |
| public static byte state(GridCacheContext cctx, long mvccCrd, long mvccCntr, int mvccOpCntr) throws IgniteCheckedException { |
| return state(cctx.kernalContext().coordinators(), mvccCrd, mvccCntr, mvccOpCntr); |
| } |
| |
| /** |
| * @param grp Cache group context. |
| * @param mvccCrd Mvcc coordinator version. |
| * @param mvccCntr Mvcc counter. |
| * @param mvccOpCntr Mvcc operation counter. |
| * @return TxState |
| * @see TxState |
| * @throws IgniteCheckedException If failed. |
| */ |
| public static byte state(CacheGroupContext grp, long mvccCrd, long mvccCntr, int mvccOpCntr) throws IgniteCheckedException { |
| return state(grp.shared().coordinators(), mvccCrd, mvccCntr, mvccOpCntr); |
| } |
| |
| /** |
| * @param proc Mvcc processor. |
| * @param mvccCrd Mvcc coordinator version. |
| * @param mvccCntr Mvcc counter. |
| * @return TxState |
| * @see TxState |
| * @throws IgniteCheckedException If failed. |
| */ |
| private static byte state(MvccProcessor proc, long mvccCrd, long mvccCntr, int mvccOpCntr) throws IgniteCheckedException { |
| if (compare(INITIAL_VERSION, mvccCrd, mvccCntr, mvccOpCntr) == 0) |
| return TxState.COMMITTED; // Initial version is always committed; |
| |
| if ((mvccOpCntr & MVCC_HINTS_MASK) != 0) |
| return (byte)(mvccOpCntr >>> MVCC_HINTS_BIT_OFF); |
| |
| byte state = proc.state(mvccCrd, mvccCntr); |
| |
| if ((state == TxState.NA || state == TxState.PREPARED) |
| && (proc.currentCoordinator() == null // Recovery from WAL. |
| || mvccCrd < proc.currentCoordinator().coordinatorVersion())) |
| state = TxState.ABORTED; |
| |
| return state; |
| } |
| |
| /** |
| * Checks if version is visible from the given snapshot. |
| * |
| * @param cctx Cache context. |
| * @param snapshot Snapshot. |
| * @param mvccCrd Mvcc coordinator. |
| * @param mvccCntr Mvcc counter. |
| * @param opCntr Operation counter. |
| * @return {@code True} if visible. |
| * @throws IgniteCheckedException If failed. |
| */ |
| public static boolean isVisible(GridCacheContext cctx, MvccSnapshot snapshot, long mvccCrd, long mvccCntr, |
| int opCntr) throws IgniteCheckedException { |
| return isVisible(cctx, snapshot, mvccCrd, mvccCntr, opCntr, true); |
| } |
| |
| /** |
| * Checks if version is visible from the given snapshot. |
| * |
| * @param cctx Cache context. |
| * @param snapshot Snapshot. |
| * @param mvccCrd Mvcc coordinator. |
| * @param mvccCntr Mvcc counter. |
| * @param opCntr Operation counter. |
| * @param useTxLog {@code True} if TxLog should be used. |
| * @return {@code True} if visible. |
| * @throws IgniteCheckedException If failed. |
| */ |
| public static boolean isVisible(GridCacheContext cctx, MvccSnapshot snapshot, long mvccCrd, long mvccCntr, |
| int opCntr, boolean useTxLog) throws IgniteCheckedException { |
| if (mvccCrd == MVCC_CRD_COUNTER_NA) { |
| assert mvccCntr == MVCC_COUNTER_NA && opCntr == MVCC_OP_COUNTER_NA |
| : "rowVer=" + mvccVersion(mvccCrd, mvccCntr, opCntr) + ", snapshot=" + snapshot; |
| |
| return false; // Unassigned version is always invisible |
| } |
| |
| if (compare(INITIAL_VERSION, mvccCrd, mvccCntr, opCntr) == 0) |
| return true; // Initial version is always visible |
| |
| long snapshotCrd = snapshot.coordinatorVersion(); |
| |
| long snapshotCntr = snapshot.counter(); |
| int snapshotOpCntr = snapshot.operationCounter(); |
| |
| if (mvccCrd > snapshotCrd) |
| return false; // Rows in the future are never visible. |
| |
| if (mvccCrd < snapshotCrd) |
| // Don't check the row with TxLog if the row is expected to be committed. |
| return !useTxLog || isCommitted(cctx, mvccCrd, mvccCntr, opCntr); |
| |
| if (mvccCntr > snapshotCntr) // we don't see future updates |
| return false; |
| |
| // Basically we can make fast decision about visibility if found rows from the same transaction. |
| // But we can't make such decision for read-only queries, |
| // because read-only queries use last committed version in it's snapshot which could be actually aborted |
| // (during transaction recovery we do not know whether recovered transaction was committed or aborted). |
| if (mvccCntr == snapshotCntr && snapshotOpCntr != MVCC_READ_OP_CNTR) { |
| assert opCntr <= snapshotOpCntr : "rowVer=" + mvccVersion(mvccCrd, mvccCntr, opCntr) + ", snapshot=" + snapshot; |
| |
| return opCntr < snapshotOpCntr; // we don't see own pending updates |
| } |
| |
| if (snapshot.activeTransactions().contains(mvccCntr)) // we don't see of other transactions' pending updates |
| return false; |
| |
| if (!useTxLog) |
| return true; // The checking row is expected to be committed. |
| |
| byte state = state(cctx, mvccCrd, mvccCntr, opCntr); |
| |
| if (state != TxState.COMMITTED && state != TxState.ABORTED) |
| throw unexpectedStateException(cctx, state, mvccCrd, mvccCntr, opCntr, snapshot); |
| |
| return state == TxState.COMMITTED; |
| } |
| |
| /** |
| * |
| * @param grp Cache group context. |
| * @param state State. |
| * @param crd Mvcc coordinator counter. |
| * @param cntr Mvcc counter. |
| * @param opCntr Mvcc operation counter. |
| * @return State exception. |
| */ |
| public static IgniteTxMvccVersionCheckedException unexpectedStateException( |
| CacheGroupContext grp, byte state, long crd, long cntr, |
| int opCntr) { |
| return unexpectedStateException(grp.shared().kernalContext(), state, crd, cntr, opCntr, null); |
| } |
| |
| /** |
| * |
| * @param cctx Cache context. |
| * @param state State. |
| * @param crd Mvcc coordinator counter. |
| * @param cntr Mvcc counter. |
| * @param opCntr Mvcc operation counter. |
| * @param snapshot Mvcc snapshot |
| * @return State exception. |
| */ |
| public static IgniteTxMvccVersionCheckedException unexpectedStateException( |
| GridCacheContext cctx, byte state, long crd, long cntr, |
| int opCntr, MvccSnapshot snapshot) { |
| return unexpectedStateException(cctx.kernalContext(), state, crd, cntr, opCntr, snapshot); |
| } |
| |
| /** */ |
| private static IgniteTxMvccVersionCheckedException unexpectedStateException(GridKernalContext ctx, byte state, long crd, long cntr, |
| int opCntr, MvccSnapshot snapshot) { |
| String msg = "Unexpected state: [state=" + state + ", rowVer=" + crd + ":" + cntr + ":" + opCntr; |
| |
| if (snapshot != null) |
| msg += ", txVer=" + snapshot.coordinatorVersion() + ":" + snapshot.counter() + ":" + snapshot.operationCounter(); |
| |
| msg += ", localNodeId=" + ctx.localNodeId() + "]"; |
| |
| return new IgniteTxMvccVersionCheckedException(msg); |
| } |
| |
| /** |
| * Checks visibility of the given row versions from the given snapshot. |
| * |
| * @param cctx Context. |
| * @param snapshot Snapshot. |
| * @param crd Mvcc coordinator counter. |
| * @param cntr Mvcc counter. |
| * @param opCntr Mvcc operation counter. |
| * @param link Link to data row (new version is located there). |
| * @return Visibility status. |
| * @throws IgniteCheckedException If failed. |
| */ |
| public static boolean isVisible(GridCacheContext cctx, MvccSnapshot snapshot, long crd, long cntr, |
| int opCntr, long link) throws IgniteCheckedException { |
| return isVisible(cctx, snapshot, crd, cntr, opCntr, false) |
| && isVisible(cctx, link, snapshot); |
| } |
| |
| /** |
| * Checks if a row has not empty new version (xid_max). |
| * |
| * @param row Row. |
| * @return {@code True} if row has a new version. |
| */ |
| public static boolean hasNewVersion(MvccUpdateVersionAware row) { |
| assert row.newMvccCoordinatorVersion() == MVCC_CRD_COUNTER_NA |
| || mvccVersionIsValid(row.newMvccCoordinatorVersion(), row.newMvccCounter(), row.newMvccOperationCounter()); |
| |
| return row.newMvccCoordinatorVersion() > MVCC_CRD_COUNTER_NA; |
| } |
| |
| /** |
| * Checks if a row's new version is visible for the given snapshot. |
| * |
| * @param cctx Cache context. |
| * @param link Link to the row. |
| * @param snapshot Mvcc snapshot. |
| * @return {@code True} if row is visible for the given snapshot. |
| * @throws IgniteCheckedException If failed. |
| */ |
| public static int getVisibleState(GridCacheContext cctx, long link, MvccSnapshot snapshot) |
| throws IgniteCheckedException { |
| return invoke(cctx, link, getVisibleState, snapshot); |
| } |
| |
| /** |
| * Returns new version of row (xid_max) if any. |
| * |
| * @param cctx Cache context. |
| * @param link Link to the row. |
| * @return New {@code MvccVersion} if row has xid_max, or null if doesn't. |
| * @throws IgniteCheckedException If failed. |
| */ |
| public static MvccVersion getNewVersion(GridCacheContext cctx, long link) |
| throws IgniteCheckedException { |
| return invoke(cctx, link, getNewVer, null); |
| } |
| |
| /** |
| * Compares row version (xid_min) with the given version. |
| * |
| * @param row Row. |
| * @param ver Version. |
| * @return Comparison result, see {@link Comparable}. |
| */ |
| public static int compare(MvccVersionAware row, MvccVersion ver) { |
| return compare(row.mvccCoordinatorVersion(), row.mvccCounter(), row.mvccOperationCounter(), |
| ver.coordinatorVersion(), ver.counter(), ver.operationCounter()); |
| } |
| |
| /** |
| * Compares to pairs of MVCC versions. See {@link Comparable}. |
| * |
| * @param mvccVerLeft First MVCC version. |
| * @param mvccCrdRight Second coordinator version. |
| * @param mvccCntrRight Second counter. |
| * @return Comparison result, see {@link Comparable}. |
| */ |
| public static int compare(MvccVersion mvccVerLeft, long mvccCrdRight, long mvccCntrRight) { |
| return compare(mvccVerLeft.coordinatorVersion(), mvccVerLeft.counter(), mvccCrdRight, mvccCntrRight); |
| } |
| |
| /** |
| * Compares to pairs of MVCC versions. See {@link Comparable}. |
| * |
| * @param row First MVCC version. |
| * @param mvccCrdRight Second coordinator version. |
| * @param mvccCntrRight Second counter. |
| * @return Comparison result, see {@link Comparable}. |
| */ |
| public static int compare(MvccVersionAware row, long mvccCrdRight, long mvccCntrRight) { |
| return compare(row.mvccCoordinatorVersion(), row.mvccCounter(), mvccCrdRight, mvccCntrRight); |
| } |
| |
| /** |
| * Compares to pairs of MVCC versions. See {@link Comparable}. |
| * |
| * @param mvccVerLeft First MVCC version. |
| * @param mvccCrdRight Second coordinator version. |
| * @param mvccCntrRight Second counter. |
| * @param mvccOpCntrRight Second operation counter. |
| * @return Comparison result, see {@link Comparable}. |
| */ |
| public static int compare(MvccVersion mvccVerLeft, long mvccCrdRight, long mvccCntrRight, int mvccOpCntrRight) { |
| return compare(mvccVerLeft.coordinatorVersion(), mvccVerLeft.counter(), |
| mvccVerLeft.operationCounter(), mvccCrdRight, mvccCntrRight, mvccOpCntrRight); |
| } |
| |
| /** |
| * Compares to pairs of coordinator/counter versions. See {@link Comparable}. |
| * |
| * @param mvccCrdLeft First coordinator version. |
| * @param mvccCntrLeft First counter version. |
| * @param mvccOpCntrLeft First operation counter. |
| * @param other The object to compare with. |
| * @return Comparison result, see {@link Comparable}. |
| */ |
| public static int compare(long mvccCrdLeft, long mvccCntrLeft, int mvccOpCntrLeft, MvccVersionAware other) { |
| return compare(mvccCrdLeft, mvccCntrLeft, mvccOpCntrLeft, |
| other.mvccCoordinatorVersion(), other.mvccCounter(), other.mvccOperationCounter()); |
| } |
| |
| /** |
| * Compares to pairs of coordinator/counter versions. See {@link Comparable}. |
| * |
| * @param mvccCrdLeft First coordinator version. |
| * @param mvccCntrLeft First counter version. |
| * @param mvccCrdRight Second coordinator version. |
| * @param mvccCntrRight Second counter version. |
| * @return Comparison result, see {@link Comparable}. |
| */ |
| public static int compare(long mvccCrdLeft, long mvccCntrLeft, long mvccCrdRight, long mvccCntrRight) { |
| return compare(mvccCrdLeft, mvccCntrLeft, 0, mvccCrdRight, mvccCntrRight, 0); |
| } |
| |
| /** |
| * Compares to pairs of coordinator/counter versions. See {@link Comparable}. |
| * |
| * @param mvccCrdLeft First coordinator version. |
| * @param mvccCntrLeft First counter version. |
| * @param mvccOpCntrLeft First operation counter. |
| * @param mvccCrdRight Second coordinator version. |
| * @param mvccCntrRight Second counter version. |
| * @param mvccOpCntrRight Second operation counter. |
| * @return Comparison result, see {@link Comparable}. |
| */ |
| public static int compare(long mvccCrdLeft, long mvccCntrLeft, int mvccOpCntrLeft, long mvccCrdRight, |
| long mvccCntrRight, int mvccOpCntrRight) { |
| int cmp; |
| |
| if ((cmp = Long.compare(mvccCrdLeft, mvccCrdRight)) != 0 |
| || (cmp = Long.compare(mvccCntrLeft, mvccCntrRight)) != 0 |
| || (cmp = Integer.compare(mvccOpCntrLeft & ~MVCC_HINTS_MASK, mvccOpCntrRight & ~MVCC_HINTS_MASK)) != 0) |
| return cmp; |
| |
| return 0; |
| } |
| |
| /** |
| * Compares left version (xid_min) with the given version ignoring operation counter. |
| * |
| * @param left Version. |
| * @param right Version. |
| * @return Comparison result, see {@link Comparable}. |
| */ |
| public static int compareIgnoreOpCounter(MvccVersion left, MvccVersion right) { |
| return compare(left.coordinatorVersion(), left.counter(), 0, right.coordinatorVersion(), right.counter(), 0); |
| } |
| |
| /** |
| * Compares new row version (xid_max) with the given counter and coordinator versions. |
| * |
| * @param row Row. |
| * @param mvccCrd Mvcc coordinator. |
| * @param mvccCntr Mvcc counter. |
| * @return Comparison result, see {@link Comparable}. |
| */ |
| public static int compareNewVersion(MvccUpdateVersionAware row, long mvccCrd, long mvccCntr) { |
| return compare(row.newMvccCoordinatorVersion(), row.newMvccCounter(), mvccCrd, mvccCntr); |
| } |
| |
| /** |
| * Compares new row version (xid_max) with the given counter and coordinator versions. |
| * |
| * @param row Row. |
| * @param mvccCrd Mvcc coordinator. |
| * @param mvccCntr Mvcc counter. |
| * @param opCntr Mvcc operation counter. |
| * @return Comparison result, see {@link Comparable}. |
| */ |
| public static int compareNewVersion(MvccUpdateVersionAware row, long mvccCrd, long mvccCntr, int opCntr) { |
| return compare(row.newMvccCoordinatorVersion(), row.newMvccCounter(), row.newMvccOperationCounter(), mvccCrd, mvccCntr, opCntr); |
| } |
| |
| /** |
| * Compares new row version (xid_max) with the given version. |
| * |
| * @param row Row. |
| * @param ver Version. |
| * @return Comparison result, see {@link Comparable}. |
| */ |
| public static int compareNewVersion(MvccUpdateVersionAware row, MvccVersion ver) { |
| return compare(row.newMvccCoordinatorVersion(), row.newMvccCounter(), row.newMvccOperationCounter(), |
| ver.coordinatorVersion(), ver.counter(), ver.operationCounter()); |
| } |
| |
| /** |
| * @param crdVer Mvcc coordinator version. |
| * @param cntr Counter. |
| * @param opCntr Operation counter. |
| * @return Always {@code true}. |
| */ |
| public static boolean mvccVersionIsValid(long crdVer, long cntr, int opCntr) { |
| return mvccVersionIsValid(crdVer, cntr) && opCntr != MVCC_OP_COUNTER_NA; |
| } |
| |
| /** |
| * @param crdVer Mvcc coordinator version. |
| * @param cntr Counter. |
| * @return {@code True} if version is valid. |
| */ |
| public static boolean mvccVersionIsValid(long crdVer, long cntr) { |
| return crdVer > MVCC_CRD_COUNTER_NA && cntr != MVCC_COUNTER_NA; |
| } |
| |
| /** |
| * @param topVer Topology version for cache operation. |
| * @return Error. |
| */ |
| public static ClusterTopologyServerNotFoundException noCoordinatorError(AffinityTopologyVersion topVer) { |
| return new ClusterTopologyServerNotFoundException("Mvcc coordinator is not assigned for " + |
| "topology version: " + topVer); |
| } |
| |
| /** |
| * @return Error. |
| */ |
| public static ClusterTopologyServerNotFoundException noCoordinatorError() { |
| return new ClusterTopologyServerNotFoundException("Mvcc coordinator is not assigned."); |
| } |
| |
| /** |
| * @param cctx Cache context. |
| * @param link Link to the row. |
| * @param snapshot Mvcc snapshot. |
| * @return {@code True} if row is updated for given snapshot. |
| * @throws IgniteCheckedException If failed. |
| */ |
| private static boolean isVisible(GridCacheContext cctx, long link, |
| MvccSnapshot snapshot) |
| throws IgniteCheckedException { |
| return invoke(cctx, link, isVisible, snapshot); |
| } |
| |
| /** |
| * Encapsulates common logic for working with row mvcc info: page locking/unlocking, checks and other. |
| * Strategy pattern. |
| * |
| * @param cctx Cache group. |
| * @param link Row link. |
| * @param clo Closure to apply. |
| * @param snapshot Mvcc snapshot. |
| * @param <R> Return type. |
| * @return Result. |
| * @throws IgniteCheckedException If failed. |
| */ |
| private static <R> R invoke(GridCacheContext cctx, long link, MvccClosure<R> clo, MvccSnapshot snapshot) |
| throws IgniteCheckedException { |
| assert cctx.mvccEnabled(); |
| |
| PageMemory pageMem = cctx.dataRegion().pageMemory(); |
| int grpId = cctx.groupId(); |
| |
| long pageId = pageId(link); |
| long page = pageMem.acquirePage(grpId, pageId); |
| |
| try { |
| long pageAddr = pageMem.readLock(grpId, pageId, page); |
| |
| try{ |
| DataPageIO dataIo = DataPageIO.VERSIONS.forPage(pageAddr); |
| |
| int offset = dataIo.getPayloadOffset(pageAddr, itemId(link), pageMem.realPageSize(grpId), |
| MVCC_INFO_SIZE); |
| |
| long mvccCrd = dataIo.mvccCoordinator(pageAddr, offset); |
| long mvccCntr = dataIo.mvccCounter(pageAddr, offset); |
| int mvccOpCntr = dataIo.mvccOperationCounter(pageAddr, offset) & ~MVCC_KEY_ABSENT_BEFORE_MASK; |
| |
| assert mvccVersionIsValid(mvccCrd, mvccCntr, mvccOpCntr) : mvccVersion(mvccCrd, mvccCntr, mvccOpCntr); |
| |
| long newMvccCrd = dataIo.newMvccCoordinator(pageAddr, offset); |
| long newMvccCntr = dataIo.newMvccCounter(pageAddr, offset); |
| int newMvccOpCntr = dataIo.newMvccOperationCounter(pageAddr, offset) & ~MVCC_KEY_ABSENT_BEFORE_MASK; |
| |
| assert newMvccCrd == MVCC_CRD_COUNTER_NA || mvccVersionIsValid(newMvccCrd, newMvccCntr, newMvccOpCntr) |
| : mvccVersion(newMvccCrd, newMvccCntr, newMvccOpCntr); |
| |
| return clo.apply(cctx, snapshot, mvccCrd, mvccCntr, mvccOpCntr, newMvccCrd, newMvccCntr, newMvccOpCntr); |
| } |
| finally { |
| pageMem.readUnlock(grpId, pageId, page); |
| } |
| } |
| finally { |
| pageMem.releasePage(grpId, pageId, page); |
| } |
| } |
| |
| /** |
| * |
| * @param cctx Cache context. |
| * @param mvccCrd Coordinator version. |
| * @param mvccCntr Counter. |
| * @return {@code True} in case the corresponding transaction is in {@code TxState.COMMITTED} state. |
| * @throws IgniteCheckedException If failed. |
| */ |
| private static boolean isCommitted(GridCacheContext cctx, long mvccCrd, long mvccCntr, int mvccOpCntr) throws IgniteCheckedException { |
| return state(cctx, mvccCrd, mvccCntr, mvccOpCntr) == TxState.COMMITTED; |
| } |
| |
| /** |
| * Throw an {@link UnsupportedOperationException} if this cache is transactional and MVCC is enabled with |
| * appropriate message about corresponding operation type. |
| * @param cctx Cache context. |
| * @param opType operation type to mention in error message. |
| */ |
| public static void verifyMvccOperationSupport(GridCacheContext<?, ?> cctx, String opType) { |
| if (cctx.mvccEnabled()) |
| throw new UnsupportedOperationException(opType + " operations are not supported on transactional " + |
| "caches when MVCC is enabled."); |
| } |
| |
| /** |
| * Checks transaction state. |
| * @param tx Transaction. |
| * @return Checked transaction. |
| */ |
| public static GridNearTxLocal checkActive(GridNearTxLocal tx) { |
| if (tx != null && tx.state() != TransactionState.ACTIVE) |
| throw new IgniteSQLException("Transaction is already completed.", TRANSACTION_COMPLETED); |
| |
| return tx; |
| } |
| |
| |
| /** |
| * @param ctx Grid kernal context. |
| * @return Currently started user transaction, or {@code null} if none started. |
| */ |
| @Nullable public static GridNearTxLocal tx(GridKernalContext ctx) { |
| return tx(ctx, null); |
| } |
| |
| /** |
| * @param ctx Grid kernal context. |
| * @param txId Transaction ID. |
| * @return Currently started user transaction, or {@code null} if none started. |
| */ |
| @Nullable public static GridNearTxLocal tx(GridKernalContext ctx, @Nullable GridCacheVersion txId) { |
| IgniteTxManager tm = ctx.cache().context().tm(); |
| |
| IgniteInternalTx tx0 = txId == null ? tm.tx() : tm.tx(txId); |
| |
| GridNearTxLocal tx = tx0 != null && tx0.user() ? (GridNearTxLocal)tx0 : null; |
| |
| if (tx != null) { |
| if (!tx.pessimistic() || !tx.repeatableRead()) { |
| tx.setRollbackOnly(); |
| |
| throw new IgniteSQLException("Only pessimistic repeatable read transactions are supported at the moment.", |
| IgniteQueryErrorCode.UNSUPPORTED_OPERATION); |
| |
| } |
| |
| if (!tx.isOperationAllowed(true)) { |
| tx.setRollbackOnly(); |
| |
| throw new IgniteSQLException("SQL queries and cache operations " + |
| "may not be used in the same transaction.", IgniteQueryErrorCode.TRANSACTION_TYPE_MISMATCH); |
| } |
| } |
| |
| return tx; |
| } |
| |
| |
| /** |
| * @param ctx Grid kernal context. |
| * @param timeout Transaction timeout. |
| * @return Newly started SQL transaction. |
| */ |
| public static GridNearTxLocal txStart(GridKernalContext ctx, long timeout) { |
| return txStart(ctx, null, timeout); |
| } |
| |
| /** |
| * @param cctx Cache context. |
| * @param timeout Transaction timeout. |
| * @return Newly started SQL transaction. |
| */ |
| public static GridNearTxLocal txStart(GridCacheContext cctx, long timeout) { |
| return txStart(cctx.kernalContext(), cctx, timeout); |
| } |
| |
| /** |
| * @param ctx Grid kernal context. |
| * @param cctx Cache context. |
| * @param timeout Transaction timeout. |
| * @return Newly started SQL transaction. |
| */ |
| private static GridNearTxLocal txStart(GridKernalContext ctx, @Nullable GridCacheContext cctx, long timeout) { |
| if (timeout == 0) { |
| TransactionConfiguration tcfg = CU.transactionConfiguration(cctx, ctx.config()); |
| |
| if (tcfg != null) |
| timeout = tcfg.getDefaultTxTimeout(); |
| } |
| |
| GridNearTxLocal tx = ctx.cache().context().tm().newTx( |
| false, |
| false, |
| cctx != null && cctx.systemTx() ? cctx : null, |
| PESSIMISTIC, |
| REPEATABLE_READ, |
| timeout, |
| cctx == null || !cctx.skipStore(), |
| true, |
| 0, |
| null |
| ); |
| |
| tx.syncMode(FULL_SYNC); |
| |
| return tx; |
| } |
| |
| /** |
| * @param ctx Grid kernal context. |
| * @return Whether MVCC is enabled or not. |
| */ |
| public static boolean mvccEnabled(GridKernalContext ctx) { |
| return ctx.coordinators().mvccEnabled(); |
| } |
| |
| /** |
| * Initialises MVCC filter and returns MVCC query tracker if needed. |
| * @param cctx Cache context. |
| * @param startTx Start transaction flag. |
| * @return MVCC query tracker. |
| * @throws IgniteCheckedException If failed. |
| */ |
| @NotNull public static MvccQueryTracker mvccTracker(GridCacheContext cctx, boolean startTx) throws IgniteCheckedException { |
| assert cctx != null && cctx.mvccEnabled(); |
| |
| GridNearTxLocal tx = tx(cctx.kernalContext()); |
| |
| if (tx == null && startTx) |
| tx = txStart(cctx, 0); |
| |
| return mvccTracker(cctx, tx); |
| } |
| |
| /** |
| * Initialises MVCC filter and returns MVCC query tracker if needed. |
| * @param cctx Cache context. |
| * @param tx Transaction. |
| * @return MVCC query tracker. |
| * @throws IgniteCheckedException If failed. |
| */ |
| @NotNull public static MvccQueryTracker mvccTracker(GridCacheContext cctx, |
| GridNearTxLocal tx) throws IgniteCheckedException { |
| MvccQueryTracker tracker; |
| |
| if (tx == null) |
| tracker = new MvccQueryTrackerImpl(cctx); |
| else if ((tracker = tx.mvccQueryTracker()) == null) |
| tracker = new StaticMvccQueryTracker(cctx, requestSnapshot(cctx, tx)); |
| |
| if (tracker.snapshot() == null) |
| // TODO IGNITE-7388 |
| tracker.requestSnapshot().get(); |
| |
| return tracker; |
| } |
| |
| /** |
| * @param cctx Cache context. |
| * @param tx Transaction. |
| * @throws IgniteCheckedException If failed. |
| * @return Mvcc snapshot. |
| */ |
| public static MvccSnapshot requestSnapshot(GridCacheContext cctx, |
| GridNearTxLocal tx) throws IgniteCheckedException { |
| MvccSnapshot snapshot; |
| |
| tx = checkActive(tx); |
| |
| if ((snapshot = tx.mvccSnapshot()) == null) { |
| MvccProcessor prc = cctx.shared().coordinators(); |
| |
| snapshot = prc.tryRequestSnapshotLocal(tx); |
| |
| if (snapshot == null) |
| // TODO IGNITE-7388 |
| snapshot = prc.requestSnapshotAsync(tx).get(); |
| |
| tx.mvccSnapshot(snapshot); |
| } |
| |
| return snapshot; |
| } |
| |
| /** |
| * Throws atomicity modes compatibility validation exception. |
| * |
| * @param ctx1 Cache context. |
| * @param ctx2 Another cache context. |
| */ |
| public static void throwAtomicityModesMismatchException(GridCacheContext ctx1, GridCacheContext ctx2) { |
| throw new IgniteException("Caches with transactional_snapshot atomicity mode cannot participate in the same" + |
| " transaction with caches having another atomicity mode. [cacheName=" + ctx1.name() + |
| ", cacheMode=" + ctx1.config().getAtomicityMode() + |
| ", anotherCacheName=" + ctx2.name() + " anotherCacheMode=" + ctx2.config().getAtomicityMode() + ']'); |
| } |
| |
| /** */ |
| private static MvccVersion mvccVersion(long crd, long cntr, int opCntr) { |
| return new MvccVersionImpl(crd, cntr, opCntr); |
| } |
| |
| /** |
| * Mvcc closure interface. |
| * @param <R> Return type. |
| */ |
| private interface MvccClosure<R> { |
| /** |
| * Runs closure over the Mvcc info. |
| * @param snapshot Mvcc snapshot. |
| * @param mvccCrd Coordinator version. |
| * @param mvccCntr Counter. |
| * @param mvccOpCntr Operation counter. |
| * @param newMvccCrd New mvcc coordinator |
| * @param newMvccCntr New mvcc counter. |
| * @param newMvccOpCntr New mvcc operation counter. |
| * @return Result. |
| */ |
| public R apply(GridCacheContext cctx, MvccSnapshot snapshot, long mvccCrd, long mvccCntr, int mvccOpCntr, |
| long newMvccCrd, long newMvccCntr, int newMvccOpCntr) throws IgniteCheckedException; |
| } |
| |
| /** |
| * Closure for checking row visibility for snapshot. |
| */ |
| private static class GetVisibleState implements MvccClosure<Integer> { |
| /** {@inheritDoc} */ |
| @Override public Integer apply(GridCacheContext cctx, MvccSnapshot snapshot, long mvccCrd, long mvccCntr, |
| int mvccOpCntr, long newMvccCrd, long newMvccCntr, int newMvccOpCntr) throws IgniteCheckedException { |
| |
| if (!isVisible(cctx, snapshot, mvccCrd, mvccCntr, mvccOpCntr)) |
| return MVCC_INVISIBLE; |
| |
| if (newMvccCrd == MVCC_CRD_COUNTER_NA) |
| return MVCC_VISIBLE; |
| |
| assert mvccVersionIsValid(newMvccCrd, newMvccCntr, newMvccOpCntr); |
| |
| if (mvccCrd == newMvccCrd && mvccCntr == newMvccCntr) // Double-changed in scope of one transaction. |
| return MVCC_VISIBLE_REMOVED; |
| |
| return isVisible(cctx, snapshot, newMvccCrd, newMvccCntr, newMvccOpCntr) ? MVCC_VISIBLE_REMOVED : |
| MVCC_VISIBLE; |
| } |
| } |
| |
| /** |
| * Closure for checking whether the row is visible for given snapshot. |
| */ |
| private static class IsVisible implements MvccClosure<Boolean> { |
| /** {@inheritDoc} */ |
| @Override public Boolean apply(GridCacheContext cctx, MvccSnapshot snapshot, long mvccCrd, long mvccCntr, |
| int mvccOpCntr, long newMvccCrd, long newMvccCntr, int newMvccOpCntr) throws IgniteCheckedException { |
| |
| if (!isVisible(cctx, snapshot, mvccCrd, mvccCntr, mvccOpCntr)) |
| return false; |
| |
| if (newMvccCrd == MVCC_CRD_COUNTER_NA) |
| return true; |
| |
| assert mvccVersionIsValid(newMvccCrd, newMvccCntr, newMvccOpCntr); |
| |
| if (mvccCrd == newMvccCrd && mvccCntr == newMvccCntr) // Double-changed in scope of one transaction. |
| return false; |
| |
| return !isVisible(cctx, snapshot, newMvccCrd, newMvccCntr, newMvccOpCntr); |
| } |
| } |
| |
| /** |
| * Closure for getting xid_max version of row. |
| */ |
| private static class GetNewVersion implements MvccClosure<MvccVersion> { |
| /** {@inheritDoc} */ |
| @Override public MvccVersion apply(GridCacheContext cctx, MvccSnapshot snapshot, long mvccCrd, long mvccCntr, |
| int mvccOpCntr, long newMvccCrd, long newMvccCntr, int newMvccOpCntr) { |
| return newMvccCrd == MVCC_CRD_COUNTER_NA ? null : mvccVersion(newMvccCrd, newMvccCntr, newMvccOpCntr); |
| } |
| } |
| } |
