You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by sb...@apache.org on 2015/05/16 09:07:39 UTC
[4/8] incubator-ignite git commit: Revert "ignite-471: fixed NPE in
PortableMarshaller"
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/8a6dd007/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxPrepareFutureAdapter.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxPrepareFutureAdapter.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxPrepareFutureAdapter.java
deleted file mode 100644
index 60b918c..0000000
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxPrepareFutureAdapter.java
+++ /dev/null
@@ -1,226 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.ignite.internal.processors.cache.distributed.near;
-
-import org.apache.ignite.*;
-import org.apache.ignite.internal.processors.cache.*;
-import org.apache.ignite.internal.processors.cache.distributed.*;
-import org.apache.ignite.internal.processors.cache.distributed.dht.*;
-import org.apache.ignite.internal.processors.cache.distributed.dht.colocated.*;
-import org.apache.ignite.internal.processors.cache.transactions.*;
-import org.apache.ignite.internal.processors.cache.version.*;
-import org.apache.ignite.internal.util.future.*;
-import org.apache.ignite.internal.util.tostring.*;
-import org.apache.ignite.internal.util.typedef.*;
-import org.apache.ignite.internal.util.typedef.internal.*;
-import org.apache.ignite.lang.*;
-
-import javax.cache.expiry.*;
-import java.util.*;
-import java.util.concurrent.atomic.*;
-
-import static org.apache.ignite.internal.processors.cache.GridCacheOperation.*;
-
-/**
- * Common code for tx prepare in optimistic and pessimistic modes.
- */
-public abstract class GridNearTxPrepareFutureAdapter extends GridCompoundIdentityFuture<IgniteInternalTx>
- implements GridCacheFuture<IgniteInternalTx> {
- /** Logger reference. */
- protected static final AtomicReference<IgniteLogger> logRef = new AtomicReference<>();
-
- /** */
- private static final IgniteReducer<IgniteInternalTx, IgniteInternalTx> REDUCER =
- new IgniteReducer<IgniteInternalTx, IgniteInternalTx>() {
- @Override public boolean collect(IgniteInternalTx e) {
- return true;
- }
-
- @Override public IgniteInternalTx reduce() {
- // Nothing to aggregate.
- return null;
- }
- };
-
- /** Logger. */
- protected static IgniteLogger log;
-
- /** Context. */
- protected GridCacheSharedContext<?, ?> cctx;
-
- /** Future ID. */
- protected IgniteUuid futId;
-
- /** Transaction. */
- @GridToStringInclude
- protected GridNearTxLocal tx;
-
- /** Error. */
- @GridToStringExclude
- protected AtomicReference<Throwable> err = new AtomicReference<>(null);
-
- /** Trackable flag. */
- protected boolean trackable = true;
-
- /** Full information about transaction nodes mapping. */
- protected GridDhtTxMapping txMapping;
-
- /**
- * @param cctx Context.
- * @param tx Transaction.
- */
- public GridNearTxPrepareFutureAdapter(GridCacheSharedContext cctx, final GridNearTxLocal tx) {
- super(cctx.kernalContext(), REDUCER);
-
- assert cctx != null;
- assert tx != null;
-
- this.cctx = cctx;
- this.tx = tx;
-
- futId = IgniteUuid.randomUuid();
-
- if (log == null)
- log = U.logger(cctx.kernalContext(), logRef, GridNearTxPrepareFutureAdapter.class);
- }
-
- /** {@inheritDoc} */
- @Override public IgniteUuid futureId() {
- return futId;
- }
-
- /** {@inheritDoc} */
- @Override public GridCacheVersion version() {
- return tx.xidVersion();
- }
-
- /** {@inheritDoc} */
- @Override public void markNotTrackable() {
- trackable = false;
- }
-
- /** {@inheritDoc} */
- @Override public boolean trackable() {
- return trackable;
- }
-
- /**
- * Prepares transaction.
- */
- public abstract void prepare();
-
- /**
- * @param nodeId Sender.
- * @param res Result.
- */
- public abstract void onResult(UUID nodeId, GridNearTxPrepareResponse res);
-
- /**
- * Checks if mapped transaction can be committed on one phase.
- * One-phase commit can be done if transaction maps to one primary node and not more than one backup.
- */
- protected final void checkOnePhase() {
- if (tx.storeUsed())
- return;
-
- Map<UUID, Collection<UUID>> map = txMapping.transactionNodes();
-
- if (map.size() == 1) {
- Map.Entry<UUID, Collection<UUID>> entry = F.firstEntry(map);
-
- assert entry != null;
-
- Collection<UUID> backups = entry.getValue();
-
- if (backups.size() <= 1)
- tx.onePhaseCommit(true);
- }
- }
-
- /**
- * @param m Mapping.
- * @param res Response.
- */
- protected final void onPrepareResponse(GridDistributedTxMapping m, GridNearTxPrepareResponse res) {
- if (res == null)
- return;
-
- assert res.error() == null : res;
- assert F.isEmpty(res.invalidPartitions()) : res;
-
- for (Map.Entry<IgniteTxKey, CacheVersionedValue> entry : res.ownedValues().entrySet()) {
- IgniteTxEntry txEntry = tx.entry(entry.getKey());
-
- assert txEntry != null;
-
- GridCacheContext cacheCtx = txEntry.context();
-
- while (true) {
- try {
- if (cacheCtx.isNear()) {
- GridNearCacheEntry nearEntry = (GridNearCacheEntry)txEntry.cached();
-
- CacheVersionedValue tup = entry.getValue();
-
- nearEntry.resetFromPrimary(tup.value(), tx.xidVersion(),
- tup.version(), m.node().id(), tx.topologyVersion());
- }
- else if (txEntry.cached().detached()) {
- GridDhtDetachedCacheEntry detachedEntry = (GridDhtDetachedCacheEntry)txEntry.cached();
-
- CacheVersionedValue tup = entry.getValue();
-
- detachedEntry.resetFromPrimary(tup.value(), tx.xidVersion());
- }
-
- break;
- }
- catch (GridCacheEntryRemovedException ignored) {
- // Retry.
- }
- }
- }
-
- tx.implicitSingleResult(res.returnValue());
-
- for (IgniteTxKey key : res.filterFailedKeys()) {
- IgniteTxEntry txEntry = tx.entry(key);
-
- assert txEntry != null : "Missing tx entry for write key: " + key;
-
- txEntry.op(NOOP);
-
- assert txEntry.context() != null;
-
- ExpiryPolicy expiry = txEntry.context().expiryForTxEntry(txEntry);
-
- if (expiry != null)
- txEntry.ttl(CU.toTtl(expiry.getExpiryForAccess()));
- }
-
- if (!m.empty()) {
- // Register DHT version.
- tx.addDhtVersion(m.node().id(), res.dhtVersion());
-
- m.dhtVersion(res.dhtVersion());
-
- if (m.near())
- tx.readyNearLocks(m, res.pending(), res.committedVersions(), res.rolledbackVersions());
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/8a6dd007/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxPrepareRequest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxPrepareRequest.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxPrepareRequest.java
index a08637d..f0587ac 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxPrepareRequest.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxPrepareRequest.java
@@ -88,6 +88,8 @@ public class GridNearTxPrepareRequest extends GridDistributedTxPrepareRequest {
* @param tx Transaction.
* @param reads Read entries.
* @param writes Write entries.
+ * @param grpLockKey Group lock key if preparing group-lock transaction.
+ * @param partLock {@code True} if preparing group-lock transaction with partition lock.
* @param near {@code True} if mapping is for near caches.
* @param txNodes Transaction nodes mapping.
* @param last {@code True} if this last prepare request for node.
@@ -101,6 +103,8 @@ public class GridNearTxPrepareRequest extends GridDistributedTxPrepareRequest {
IgniteInternalTx tx,
Collection<IgniteTxEntry> reads,
Collection<IgniteTxEntry> writes,
+ IgniteTxKey grpLockKey,
+ boolean partLock,
boolean near,
Map<UUID, Collection<UUID>> txNodes,
boolean last,
@@ -112,7 +116,7 @@ public class GridNearTxPrepareRequest extends GridDistributedTxPrepareRequest {
@Nullable UUID subjId,
int taskNameHash
) {
- super(tx, reads, writes, txNodes, onePhaseCommit);
+ super(tx, reads, writes, grpLockKey, partLock, txNodes, onePhaseCommit);
assert futId != null;
@@ -266,67 +270,67 @@ public class GridNearTxPrepareRequest extends GridDistributedTxPrepareRequest {
}
switch (writer.state()) {
- case 23:
+ case 25:
if (!writer.writeBoolean("explicitLock", explicitLock))
return false;
writer.incrementState();
- case 24:
+ case 26:
if (!writer.writeIgniteUuid("futId", futId))
return false;
writer.incrementState();
- case 25:
+ case 27:
if (!writer.writeBoolean("implicitSingle", implicitSingle))
return false;
writer.incrementState();
- case 26:
+ case 28:
if (!writer.writeBoolean("last", last))
return false;
writer.incrementState();
- case 27:
+ case 29:
if (!writer.writeCollection("lastBackups", lastBackups, MessageCollectionItemType.UUID))
return false;
writer.incrementState();
- case 28:
+ case 30:
if (!writer.writeIgniteUuid("miniId", miniId))
return false;
writer.incrementState();
- case 29:
+ case 31:
if (!writer.writeBoolean("near", near))
return false;
writer.incrementState();
- case 30:
+ case 32:
if (!writer.writeBoolean("retVal", retVal))
return false;
writer.incrementState();
- case 31:
+ case 33:
if (!writer.writeUuid("subjId", subjId))
return false;
writer.incrementState();
- case 32:
+ case 34:
if (!writer.writeInt("taskNameHash", taskNameHash))
return false;
writer.incrementState();
- case 33:
+ case 35:
if (!writer.writeMessage("topVer", topVer))
return false;
@@ -348,7 +352,7 @@ public class GridNearTxPrepareRequest extends GridDistributedTxPrepareRequest {
return false;
switch (reader.state()) {
- case 23:
+ case 25:
explicitLock = reader.readBoolean("explicitLock");
if (!reader.isLastRead())
@@ -356,7 +360,7 @@ public class GridNearTxPrepareRequest extends GridDistributedTxPrepareRequest {
reader.incrementState();
- case 24:
+ case 26:
futId = reader.readIgniteUuid("futId");
if (!reader.isLastRead())
@@ -364,7 +368,7 @@ public class GridNearTxPrepareRequest extends GridDistributedTxPrepareRequest {
reader.incrementState();
- case 25:
+ case 27:
implicitSingle = reader.readBoolean("implicitSingle");
if (!reader.isLastRead())
@@ -372,7 +376,7 @@ public class GridNearTxPrepareRequest extends GridDistributedTxPrepareRequest {
reader.incrementState();
- case 26:
+ case 28:
last = reader.readBoolean("last");
if (!reader.isLastRead())
@@ -380,7 +384,7 @@ public class GridNearTxPrepareRequest extends GridDistributedTxPrepareRequest {
reader.incrementState();
- case 27:
+ case 29:
lastBackups = reader.readCollection("lastBackups", MessageCollectionItemType.UUID);
if (!reader.isLastRead())
@@ -388,7 +392,7 @@ public class GridNearTxPrepareRequest extends GridDistributedTxPrepareRequest {
reader.incrementState();
- case 28:
+ case 30:
miniId = reader.readIgniteUuid("miniId");
if (!reader.isLastRead())
@@ -396,7 +400,7 @@ public class GridNearTxPrepareRequest extends GridDistributedTxPrepareRequest {
reader.incrementState();
- case 29:
+ case 31:
near = reader.readBoolean("near");
if (!reader.isLastRead())
@@ -404,7 +408,7 @@ public class GridNearTxPrepareRequest extends GridDistributedTxPrepareRequest {
reader.incrementState();
- case 30:
+ case 32:
retVal = reader.readBoolean("retVal");
if (!reader.isLastRead())
@@ -412,7 +416,7 @@ public class GridNearTxPrepareRequest extends GridDistributedTxPrepareRequest {
reader.incrementState();
- case 31:
+ case 33:
subjId = reader.readUuid("subjId");
if (!reader.isLastRead())
@@ -420,7 +424,7 @@ public class GridNearTxPrepareRequest extends GridDistributedTxPrepareRequest {
reader.incrementState();
- case 32:
+ case 34:
taskNameHash = reader.readInt("taskNameHash");
if (!reader.isLastRead())
@@ -428,7 +432,7 @@ public class GridNearTxPrepareRequest extends GridDistributedTxPrepareRequest {
reader.incrementState();
- case 33:
+ case 35:
topVer = reader.readMessage("topVer");
if (!reader.isLastRead())
@@ -448,7 +452,7 @@ public class GridNearTxPrepareRequest extends GridDistributedTxPrepareRequest {
/** {@inheritDoc} */
@Override public byte fieldsCount() {
- return 34;
+ return 36;
}
/** {@inheritDoc} */
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/8a6dd007/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxRemote.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxRemote.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxRemote.java
index 49283cb..b6b6017 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxRemote.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxRemote.java
@@ -51,6 +51,9 @@ public class GridNearTxRemote extends GridDistributedTxRemoteAdapter {
/** Owned versions. */
private Map<IgniteTxKey, GridCacheVersion> owned;
+ /** Group lock flag. */
+ private boolean grpLock;
+
/**
* Empty constructor required for {@link Externalizable}.
*/
@@ -75,6 +78,7 @@ public class GridNearTxRemote extends GridDistributedTxRemoteAdapter {
* @param writeEntries Write entries.
* @param ctx Cache registry.
* @param txSize Expected transaction size.
+ * @param grpLockKey Group lock key if this is a group-lock transaction.
* @throws IgniteCheckedException If unmarshalling failed.
*/
public GridNearTxRemote(
@@ -93,11 +97,12 @@ public class GridNearTxRemote extends GridDistributedTxRemoteAdapter {
long timeout,
Collection<IgniteTxEntry> writeEntries,
int txSize,
+ @Nullable IgniteTxKey grpLockKey,
@Nullable UUID subjId,
int taskNameHash
) throws IgniteCheckedException {
super(ctx, nodeId, rmtThreadId, xidVer, commitVer, sys, plc, concurrency, isolation, invalidate, timeout,
- txSize, subjId, taskNameHash);
+ txSize, grpLockKey, subjId, taskNameHash);
assert nearNodeId != null;
@@ -133,6 +138,7 @@ public class GridNearTxRemote extends GridDistributedTxRemoteAdapter {
* @param timeout Timeout.
* @param ctx Cache registry.
* @param txSize Expected transaction size.
+ * @param grpLockKey Collection of group lock keys if this is a group-lock transaction.
*/
public GridNearTxRemote(
GridCacheSharedContext ctx,
@@ -149,11 +155,12 @@ public class GridNearTxRemote extends GridDistributedTxRemoteAdapter {
boolean invalidate,
long timeout,
int txSize,
+ @Nullable IgniteTxKey grpLockKey,
@Nullable UUID subjId,
int taskNameHash
) {
super(ctx, nodeId, rmtThreadId, xidVer, commitVer, sys, plc, concurrency, isolation, invalidate, timeout,
- txSize, subjId, taskNameHash);
+ txSize, grpLockKey, subjId, taskNameHash);
assert nearNodeId != null;
@@ -185,6 +192,19 @@ public class GridNearTxRemote extends GridDistributedTxRemoteAdapter {
}
/**
+ * Marks near local transaction as group lock. Note that near remote transaction may be
+ * marked as group lock even if it does not contain any locked key.
+ */
+ public void markGroupLock() {
+ grpLock = true;
+ }
+
+ /** {@inheritDoc} */
+ @Override public boolean groupLock() {
+ return grpLock || super.groupLock();
+ }
+
+ /**
* @return Near transaction ID.
*/
@Override public GridCacheVersion nearXidVersion() {
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/8a6dd007/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/local/GridLocalCacheEntry.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/local/GridLocalCacheEntry.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/local/GridLocalCacheEntry.java
index ea59f1f..84d4c90 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/local/GridLocalCacheEntry.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/local/GridLocalCacheEntry.java
@@ -29,9 +29,6 @@ import static org.apache.ignite.events.EventType.*;
*/
@SuppressWarnings({"NonPrivateFieldAccessedInSynchronizedContext", "TooBroadScope"})
public class GridLocalCacheEntry extends GridCacheMapEntry {
- /** Off-heap value pointer. */
- private long valPtr;
-
/**
* @param ctx Cache registry.
* @param key Cache key.
@@ -387,19 +384,4 @@ public class GridLocalCacheEntry extends GridCacheMapEntry {
return doomed != null;
}
-
- /** {@inheritDoc} */
- @Override protected boolean hasOffHeapPointer() {
- return valPtr != 0;
- }
-
- /** {@inheritDoc} */
- @Override protected long offHeapPointer() {
- return valPtr;
- }
-
- /** {@inheritDoc} */
- @Override protected void offHeapPointer(long valPtr) {
- this.valPtr = valPtr;
- }
}
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/8a6dd007/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteInternalTx.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteInternalTx.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteInternalTx.java
index 5f877ec..8dc07cc 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteInternalTx.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteInternalTx.java
@@ -284,6 +284,16 @@ public interface IgniteInternalTx extends AutoCloseable, GridTimeoutObject {
public boolean empty();
/**
+ * @return {@code True} if transaction group-locked.
+ */
+ public boolean groupLock();
+
+ /**
+ * @return Group lock key if {@link #groupLock()} is {@code true}.
+ */
+ @Nullable public IgniteTxKey groupLockKey();
+
+ /**
* @return {@code True} if preparing flag was set with this call.
*/
public boolean markPreparing();
@@ -541,7 +551,7 @@ public interface IgniteInternalTx extends AutoCloseable, GridTimeoutObject {
*
* @return Future for prepare step.
*/
- public IgniteInternalFuture<?> prepareAsync();
+ public IgniteInternalFuture<IgniteInternalTx> prepareAsync();
/**
* @param endVer End version (a.k.a. <tt>'tnc'</tt> or <tt>'transaction number counter'</tt>)
@@ -570,7 +580,7 @@ public interface IgniteInternalTx extends AutoCloseable, GridTimeoutObject {
/**
* @return Future for transaction prepare if prepare is in progress.
*/
- @Nullable public IgniteInternalFuture<?> currentPrepareFuture();
+ @Nullable public IgniteInternalFuture<IgniteInternalTx> currentPrepareFuture();
/**
* @param state Transaction state.
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/8a6dd007/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTransactionsImpl.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTransactionsImpl.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTransactionsImpl.java
index 99907e4..044c3d7 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTransactionsImpl.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTransactionsImpl.java
@@ -160,7 +160,9 @@ public class IgniteTransactionsImpl<K, V> implements IgniteTransactionsEx {
isolation,
timeout,
true,
- txSize
+ txSize,
+ /** group lock keys */null,
+ /** partition lock */false
);
assert tx != null;
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/8a6dd007/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxAdapter.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxAdapter.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxAdapter.java
index eb8825e..82d68b3 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxAdapter.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxAdapter.java
@@ -179,6 +179,9 @@ public abstract class IgniteTxAdapter extends GridMetadataAwareAdapter
/** */
protected int txSize;
+ /** Group lock key, if any. */
+ protected IgniteTxKey grpLockKey;
+
/** */
@GridToStringExclude
private AtomicReference<GridFutureAdapter<IgniteInternalTx>> finFut = new AtomicReference<>();
@@ -230,6 +233,7 @@ public abstract class IgniteTxAdapter extends GridMetadataAwareAdapter
* @param isolation Isolation.
* @param timeout Timeout.
* @param txSize Transaction size.
+ * @param grpLockKey Group lock key if this is group-lock transaction.
*/
protected IgniteTxAdapter(
GridCacheSharedContext<?, ?> cctx,
@@ -245,6 +249,7 @@ public abstract class IgniteTxAdapter extends GridMetadataAwareAdapter
boolean invalidate,
boolean storeEnabled,
int txSize,
+ @Nullable IgniteTxKey grpLockKey,
@Nullable UUID subjId,
int taskNameHash
) {
@@ -264,6 +269,7 @@ public abstract class IgniteTxAdapter extends GridMetadataAwareAdapter
this.invalidate = invalidate;
this.storeEnabled = storeEnabled;
this.txSize = txSize;
+ this.grpLockKey = grpLockKey;
this.subjId = subjId;
this.taskNameHash = taskNameHash;
@@ -288,6 +294,7 @@ public abstract class IgniteTxAdapter extends GridMetadataAwareAdapter
* @param isolation Isolation.
* @param timeout Timeout.
* @param txSize Transaction size.
+ * @param grpLockKey Group lock key if this is group-lock transaction.
*/
protected IgniteTxAdapter(
GridCacheSharedContext<?, ?> cctx,
@@ -301,6 +308,7 @@ public abstract class IgniteTxAdapter extends GridMetadataAwareAdapter
TransactionIsolation isolation,
long timeout,
int txSize,
+ @Nullable IgniteTxKey grpLockKey,
@Nullable UUID subjId,
int taskNameHash
) {
@@ -315,6 +323,7 @@ public abstract class IgniteTxAdapter extends GridMetadataAwareAdapter
this.isolation = isolation;
this.timeout = timeout;
this.txSize = txSize;
+ this.grpLockKey = grpLockKey;
this.subjId = subjId;
this.taskNameHash = taskNameHash;
@@ -378,7 +387,30 @@ public abstract class IgniteTxAdapter extends GridMetadataAwareAdapter
/** {@inheritDoc} */
@Override public Collection<IgniteTxEntry> optimisticLockEntries() {
- return writeEntries();
+ if (!groupLock())
+ return writeEntries();
+ else {
+ if (!F.isEmpty(invalidParts)) {
+ assert invalidParts.size() == 1 : "Only one partition expected for group lock transaction " +
+ "[tx=" + this + ", invalidParts=" + invalidParts + ']';
+ assert groupLockEntry() == null : "Group lock key should be rejected " +
+ "[tx=" + this + ", groupLockEntry=" + groupLockEntry() + ']';
+ assert F.isEmpty(writeMap()) : "All entries should be rejected for group lock transaction " +
+ "[tx=" + this + ", writes=" + writeMap() + ']';
+
+ return Collections.emptyList();
+ }
+
+ IgniteTxEntry grpLockEntry = groupLockEntry();
+
+ assert grpLockEntry != null || (near() && !local()):
+ "Group lock entry was not enlisted into transaction [tx=" + this +
+ ", grpLockKey=" + groupLockKey() + ']';
+
+ return grpLockEntry == null ?
+ Collections.<IgniteTxEntry>emptyList() :
+ Collections.singletonList(grpLockEntry);
+ }
}
/** {@inheritDoc} */
@@ -450,6 +482,16 @@ public abstract class IgniteTxAdapter extends GridMetadataAwareAdapter
cctx.tm().uncommitTx(this);
}
+ /**
+ * This method uses unchecked assignment to cast group lock key entry to transaction generic signature.
+ *
+ * @return Group lock tx entry.
+ */
+ @SuppressWarnings("unchecked")
+ public IgniteTxEntry groupLockEntry() {
+ return this.entry(groupLockKey());
+ }
+
/** {@inheritDoc} */
@Override public UUID otherNodeId() {
return null;
@@ -561,6 +603,16 @@ public abstract class IgniteTxAdapter extends GridMetadataAwareAdapter
public abstract boolean isStarted();
/** {@inheritDoc} */
+ @Override public boolean groupLock() {
+ return grpLockKey != null;
+ }
+
+ /** {@inheritDoc} */
+ @Override public IgniteTxKey groupLockKey() {
+ return grpLockKey;
+ }
+
+ /** {@inheritDoc} */
@Override public int size() {
return txSize;
}
@@ -746,6 +798,9 @@ public abstract class IgniteTxAdapter extends GridMetadataAwareAdapter
GridCacheVersion explicit = txEntry == null ? null : txEntry.explicitVersion();
+ assert !txEntry.groupLockEntry() || groupLock() : "Can not have group-locked tx entries in " +
+ "non-group-lock transactions [txEntry=" + txEntry + ", tx=" + this + ']';
+
return local() && !cacheCtx.isDht() ?
entry.lockedByThread(threadId()) || (explicit != null && entry.lockedBy(explicit)) :
// If candidate is not there, then lock was explicit.
@@ -762,6 +817,9 @@ public abstract class IgniteTxAdapter extends GridMetadataAwareAdapter
GridCacheVersion explicit = txEntry == null ? null : txEntry.explicitVersion();
+ assert !txEntry.groupLockEntry() || groupLock() : "Can not have group-locked tx entries in " +
+ "non-group-lock transactions [txEntry=" + txEntry + ", tx=" + this + ']';
+
return local() && !cacheCtx.isDht() ?
entry.lockedByThreadUnsafe(threadId()) || (explicit != null && entry.lockedByUnsafe(explicit)) :
// If candidate is not there, then lock was explicit.
@@ -950,7 +1008,7 @@ public abstract class IgniteTxAdapter extends GridMetadataAwareAdapter
}
/** {@inheritDoc} */
- @Nullable @Override public IgniteInternalFuture<?> currentPrepareFuture() {
+ @Nullable @Override public IgniteInternalFuture<IgniteInternalTx> currentPrepareFuture() {
return null;
}
@@ -1496,7 +1554,7 @@ public abstract class IgniteTxAdapter extends GridMetadataAwareAdapter
/** {@inheritDoc} */
@Override public String toString() {
return GridToStringBuilder.toString(IgniteTxAdapter.class, this,
- "duration", (U.currentTimeMillis() - startTime) + "ms",
+ "duration", (U.currentTimeMillis() - startTime) + "ms", "grpLock", groupLock(),
"onePhaseCommit", onePhaseCommit);
}
@@ -1721,6 +1779,16 @@ public abstract class IgniteTxAdapter extends GridMetadataAwareAdapter
}
/** {@inheritDoc} */
+ @Override public boolean groupLock() {
+ return false;
+ }
+
+ /** {@inheritDoc} */
+ @Nullable @Override public IgniteTxKey groupLockKey() {
+ throw new IllegalStateException("Deserialized transaction can only be used as read-only.");
+ }
+
+ /** {@inheritDoc} */
@Override public boolean markPreparing() {
throw new IllegalStateException("Deserialized transaction can only be used as read-only.");
}
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/8a6dd007/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxEntry.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxEntry.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxEntry.java
index aff2f40..6d4fed6 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxEntry.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxEntry.java
@@ -137,6 +137,9 @@ public class IgniteTxEntry implements GridPeerDeployAware, Message {
@GridDirectTransient
private boolean locMapped;
+ /** Group lock entry flag. */
+ private boolean grpLock;
+
/** Expiry policy. */
@GridDirectTransient
private ExpiryPolicy expiryPlc;
@@ -274,6 +277,22 @@ public class IgniteTxEntry implements GridPeerDeployAware, Message {
}
/**
+ * @return {@code True} if this entry was added in group lock transaction and
+ * this is not a group lock entry.
+ */
+ public boolean groupLockEntry() {
+ return grpLock;
+ }
+
+ /**
+ * @param grpLock {@code True} if this entry was added in group lock transaction and
+ * this is not a group lock entry.
+ */
+ public void groupLockEntry(boolean grpLock) {
+ this.grpLock = grpLock;
+ }
+
+ /**
* @param ctx Context.
* @return Clean copy of this entry.
*/
@@ -292,6 +311,7 @@ public class IgniteTxEntry implements GridPeerDeployAware, Message {
cp.ttl = ttl;
cp.conflictExpireTime = conflictExpireTime;
cp.explicitVer = explicitVer;
+ cp.grpLock = grpLock;
cp.conflictVer = conflictVer;
cp.expiryPlc = expiryPlc;
cp.flags = flags;
@@ -831,24 +851,30 @@ public class IgniteTxEntry implements GridPeerDeployAware, Message {
writer.incrementState();
case 7:
- if (!writer.writeMessage("key", key))
+ if (!writer.writeBoolean("grpLock", grpLock))
return false;
writer.incrementState();
case 8:
- if (!writer.writeByteBuffer("transformClosBytes", transformClosBytes))
+ if (!writer.writeMessage("key", key))
return false;
writer.incrementState();
case 9:
- if (!writer.writeLong("ttl", ttl))
+ if (!writer.writeByteBuffer("transformClosBytes", transformClosBytes))
return false;
writer.incrementState();
case 10:
+ if (!writer.writeLong("ttl", ttl))
+ return false;
+
+ writer.incrementState();
+
+ case 11:
if (!writer.writeMessage("val", val))
return false;
@@ -924,7 +950,7 @@ public class IgniteTxEntry implements GridPeerDeployAware, Message {
reader.incrementState();
case 7:
- key = reader.readMessage("key");
+ grpLock = reader.readBoolean("grpLock");
if (!reader.isLastRead())
return false;
@@ -932,7 +958,7 @@ public class IgniteTxEntry implements GridPeerDeployAware, Message {
reader.incrementState();
case 8:
- transformClosBytes = reader.readByteBuffer("transformClosBytes");
+ key = reader.readMessage("key");
if (!reader.isLastRead())
return false;
@@ -940,7 +966,7 @@ public class IgniteTxEntry implements GridPeerDeployAware, Message {
reader.incrementState();
case 9:
- ttl = reader.readLong("ttl");
+ transformClosBytes = reader.readByteBuffer("transformClosBytes");
if (!reader.isLastRead())
return false;
@@ -948,6 +974,14 @@ public class IgniteTxEntry implements GridPeerDeployAware, Message {
reader.incrementState();
case 10:
+ ttl = reader.readLong("ttl");
+
+ if (!reader.isLastRead())
+ return false;
+
+ reader.incrementState();
+
+ case 11:
val = reader.readMessage("val");
if (!reader.isLastRead())
@@ -967,7 +1001,7 @@ public class IgniteTxEntry implements GridPeerDeployAware, Message {
/** {@inheritDoc} */
@Override public byte fieldsCount() {
- return 11;
+ return 12;
}
/** {@inheritDoc} */
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/8a6dd007/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxHandler.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxHandler.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxHandler.java
index f466bf2..af75fb8 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxHandler.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxHandler.java
@@ -31,6 +31,7 @@ import org.apache.ignite.internal.util.future.*;
import org.apache.ignite.internal.util.lang.*;
import org.apache.ignite.internal.util.typedef.*;
import org.apache.ignite.internal.util.typedef.internal.*;
+import org.apache.ignite.lang.*;
import org.jetbrains.annotations.*;
import java.util.*;
@@ -57,9 +58,9 @@ public class IgniteTxHandler {
* @param req Request.
* @return Prepare future.
*/
- public IgniteInternalFuture<?> processNearTxPrepareRequest(final UUID nearNodeId,
+ public IgniteInternalFuture<IgniteInternalTx> processNearTxPrepareRequest(final UUID nearNodeId,
final GridNearTxPrepareRequest req) {
- return prepareTx(nearNodeId, null, req);
+ return prepareTx(nearNodeId, null, req, null);
}
/**
@@ -137,28 +138,32 @@ public class IgniteTxHandler {
* @param nearNodeId Near node ID that initiated transaction.
* @param locTx Optional local transaction.
* @param req Near prepare request.
+ * @param completeCb Completion callback.
* @return Future for transaction.
*/
- public IgniteInternalFuture<GridNearTxPrepareResponse> prepareTx(
+ public IgniteInternalFuture<IgniteInternalTx> prepareTx(
UUID nearNodeId,
@Nullable GridNearTxLocal locTx,
- GridNearTxPrepareRequest req
+ GridNearTxPrepareRequest req,
+ @Nullable IgniteInClosure<GridNearTxPrepareResponse> completeCb
) {
assert nearNodeId != null;
assert req != null;
if (locTx != null) {
+ assert completeCb != null;
+
if (req.near()) {
// Make sure not to provide Near entries to DHT cache.
req.cloneEntries();
- return prepareNearTx(nearNodeId, req);
+ return prepareNearTx(nearNodeId, req, completeCb);
}
else
- return prepareColocatedTx(locTx, req);
+ return prepareColocatedTx(locTx, req, completeCb);
}
else
- return prepareNearTx(nearNodeId, req);
+ return prepareNearTx(nearNodeId, req, null);
}
/**
@@ -166,27 +171,30 @@ public class IgniteTxHandler {
*
* @param locTx Local transaction.
* @param req Near prepare request.
+ * @param completeCb Completion callback.
* @return Prepare future.
*/
- private IgniteInternalFuture<GridNearTxPrepareResponse> prepareColocatedTx(
+ private IgniteInternalFuture<IgniteInternalTx> prepareColocatedTx(
final GridNearTxLocal locTx,
- final GridNearTxPrepareRequest req
+ final GridNearTxPrepareRequest req,
+ final IgniteInClosure<GridNearTxPrepareResponse> completeCb
) {
IgniteInternalFuture<Object> fut = new GridFinishedFuture<>(); // TODO force preload keys.
return new GridEmbeddedFuture<>(
fut,
- new C2<Object, Exception, IgniteInternalFuture<GridNearTxPrepareResponse>>() {
- @Override public IgniteInternalFuture<GridNearTxPrepareResponse> apply(Object o, Exception ex) {
+ new C2<Object, Exception, IgniteInternalFuture<IgniteInternalTx>>() {
+ @Override public IgniteInternalFuture<IgniteInternalTx> apply(Object o, Exception ex) {
if (ex != null)
throw new GridClosureException(ex);
- IgniteInternalFuture<GridNearTxPrepareResponse> fut = locTx.prepareAsyncLocal(
+ IgniteInternalFuture<IgniteInternalTx> fut = locTx.prepareAsyncLocal(
req.reads(),
req.writes(),
req.transactionNodes(),
req.last(),
- req.lastBackups());
+ req.lastBackups(),
+ completeCb);
if (locTx.isRollbackOnly())
locTx.rollbackAsync();
@@ -194,16 +202,18 @@ public class IgniteTxHandler {
return fut;
}
},
- new C2<GridNearTxPrepareResponse, Exception, GridNearTxPrepareResponse>() {
- @Nullable @Override public GridNearTxPrepareResponse apply(GridNearTxPrepareResponse res, Exception e) {
+ new C2<IgniteInternalTx, Exception, IgniteInternalTx>() {
+ @Nullable @Override public IgniteInternalTx apply(IgniteInternalTx tx, Exception e) {
if (e != null) {
- locTx.setRollbackOnly(); // Just in case.
+ // tx can be null of exception occurred.
+ if (tx != null)
+ tx.setRollbackOnly(); // Just in case.
if (!(e instanceof IgniteTxOptimisticCheckedException))
- U.error(log, "Failed to prepare transaction: " + locTx, e);
+ U.error(log, "Failed to prepare DHT transaction: " + tx, e);
}
- return res;
+ return tx;
}
}
);
@@ -214,11 +224,13 @@ public class IgniteTxHandler {
*
* @param nearNodeId Near node ID that initiated transaction.
* @param req Near prepare request.
+ * @param completeCb Completion callback.
* @return Prepare future.
*/
- private IgniteInternalFuture<GridNearTxPrepareResponse> prepareNearTx(
+ private IgniteInternalFuture<IgniteInternalTx> prepareNearTx(
final UUID nearNodeId,
- final GridNearTxPrepareRequest req
+ final GridNearTxPrepareRequest req,
+ IgniteInClosure<GridNearTxPrepareResponse> completeCb
) {
ClusterNode nearNode = ctx.node(nearNodeId);
@@ -271,6 +283,8 @@ public class IgniteTxHandler {
req.isInvalidate(),
false,
req.txSize(),
+ req.groupLockKey(),
+ req.partitionLock(),
req.transactionNodes(),
req.subjectId(),
req.taskNameHash()
@@ -301,7 +315,7 @@ public class IgniteTxHandler {
if (req.returnValue())
tx.needReturnValue(true);
- IgniteInternalFuture<GridNearTxPrepareResponse> fut = tx.prepareAsync(
+ IgniteInternalFuture<IgniteInternalTx> fut = tx.prepareAsync(
req.reads(),
req.writes(),
req.dhtVersions(),
@@ -309,7 +323,8 @@ public class IgniteTxHandler {
req.miniId(),
req.transactionNodes(),
req.last(),
- req.lastBackups());
+ req.lastBackups(),
+ completeCb);
if (tx.isRollbackOnly()) {
try {
@@ -322,8 +337,8 @@ public class IgniteTxHandler {
final GridDhtTxLocal tx0 = tx;
- fut.listen(new CI1<IgniteInternalFuture<?>>() {
- @Override public void apply(IgniteInternalFuture<?> txFut) {
+ fut.listen(new CI1<IgniteInternalFuture<IgniteInternalTx>>() {
+ @Override public void apply(IgniteInternalFuture<IgniteInternalTx> txFut) {
try {
txFut.get();
}
@@ -339,7 +354,7 @@ public class IgniteTxHandler {
return fut;
}
else
- return new GridFinishedFuture<>((GridNearTxPrepareResponse)null);
+ return new GridFinishedFuture<>((IgniteInternalTx)null);
}
/**
@@ -347,7 +362,7 @@ public class IgniteTxHandler {
* @param res Response.
*/
private void processNearTxPrepareResponse(UUID nodeId, GridNearTxPrepareResponse res) {
- GridNearTxPrepareFutureAdapter fut = (GridNearTxPrepareFutureAdapter)ctx.mvcc()
+ GridNearTxPrepareFuture fut = (GridNearTxPrepareFuture)ctx.mvcc()
.<IgniteInternalTx>future(res.version(), res.futureId());
if (fut == null) {
@@ -384,7 +399,8 @@ public class IgniteTxHandler {
* @param res Response.
*/
private void processDhtTxPrepareResponse(UUID nodeId, GridDhtTxPrepareResponse res) {
- GridDhtTxPrepareFuture fut = (GridDhtTxPrepareFuture)ctx.mvcc().future(res.version(), res.futureId());
+ GridDhtTxPrepareFuture fut = (GridDhtTxPrepareFuture)ctx.mvcc().
+ <IgniteInternalTx>future(res.version(), res.futureId());
if (fut == null) {
if (log.isDebugEnabled())
@@ -552,6 +568,8 @@ public class IgniteTxHandler {
req.isInvalidate(),
req.storeEnabled(),
req.txSize(),
+ req.groupLockKey(),
+ false,
null,
req.subjectId(),
req.taskNameHash()));
@@ -998,6 +1016,7 @@ public class IgniteTxHandler {
req.isInvalidate(),
req.timeout(),
req.writes() != null ? Math.max(req.writes().size(), req.txSize()) : req.txSize(),
+ req.groupLockKey(),
req.nearXidVersion(),
req.transactionNodes(),
req.subjectId(),
@@ -1131,6 +1150,7 @@ public class IgniteTxHandler {
req.timeout(),
req.nearWrites(),
req.txSize(),
+ req.groupLockKey(),
req.subjectId(),
req.taskNameHash()
);
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/8a6dd007/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxLocalAdapter.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxLocalAdapter.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxLocalAdapter.java
index 609108f..3c855ec 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxLocalAdapter.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxLocalAdapter.java
@@ -86,6 +86,9 @@ public abstract class IgniteTxLocalAdapter extends IgniteTxAdapter
/** Base for completed versions. */
private GridCacheVersion completedBase;
+ /** Flag indicating partition lock in group lock transaction. */
+ private boolean partLock;
+
/** Flag indicating that transformed values should be sent to remote nodes. */
private boolean sndTransformedVals;
@@ -120,6 +123,8 @@ public abstract class IgniteTxLocalAdapter extends IgniteTxAdapter
* @param isolation Isolation.
* @param timeout Timeout.
* @param txSize Expected transaction size.
+ * @param grpLockKey Group lock key if this is a group-lock transaction.
+ * @param partLock {@code True} if this is a group-lock transaction and lock is acquired for whole partition.
*/
protected IgniteTxLocalAdapter(
GridCacheSharedContext cctx,
@@ -134,11 +139,17 @@ public abstract class IgniteTxLocalAdapter extends IgniteTxAdapter
boolean invalidate,
boolean storeEnabled,
int txSize,
+ @Nullable IgniteTxKey grpLockKey,
+ boolean partLock,
@Nullable UUID subjId,
int taskNameHash
) {
super(cctx, xidVer, implicit, implicitSingle, /*local*/true, sys, plc, concurrency, isolation, timeout,
- invalidate, storeEnabled, txSize, subjId, taskNameHash);
+ invalidate, storeEnabled, txSize, grpLockKey, subjId, taskNameHash);
+
+ assert !partLock || grpLockKey != null;
+
+ this.partLock = partLock;
minVer = xidVer;
}
@@ -171,6 +182,11 @@ public abstract class IgniteTxLocalAdapter extends IgniteTxAdapter
}
/** {@inheritDoc} */
+ @Override public boolean partitionLock() {
+ return partLock;
+ }
+
+ /** {@inheritDoc} */
@Override public Throwable commitError() {
return commitErr.get();
}
@@ -483,7 +499,7 @@ public abstract class IgniteTxLocalAdapter extends IgniteTxAdapter
CacheStoreManager store = store();
if (store != null && store.isWriteThrough() && storeEnabled() &&
- !internal() && (near() || store.isWriteToStoreFromDht())) {
+ (!internal() || groupLock()) && (near() || store.isWriteToStoreFromDht())) {
try {
if (writeEntries != null) {
Map<Object, IgniteBiTuple<Object, GridCacheVersion>> putMap = null;
@@ -663,6 +679,9 @@ public abstract class IgniteTxLocalAdapter extends IgniteTxAdapter
if (!empty || colocated())
cctx.tm().addCommittedTx(this);
+ if (groupLock())
+ addGroupTxMapping(writeSet());
+
if (!empty) {
batchStoreCommit(writeMap().values());
@@ -890,6 +909,10 @@ public abstract class IgniteTxLocalAdapter extends IgniteTxAdapter
log.debug("Ignoring READ entry when committing: " + txEntry);
}
else {
+ assert !groupLock() || txEntry.groupLockEntry() || ownsLock(txEntry.cached()):
+ "Transaction does not own lock for group lock entry during commit [tx=" +
+ this + ", txEntry=" + txEntry + ']';
+
if (conflictCtx == null || !conflictCtx.isUseOld()) {
if (txEntry.ttl() != CU.TTL_NOT_CHANGED)
cached.updateTtl(null, txEntry.ttl());
@@ -904,7 +927,7 @@ public abstract class IgniteTxLocalAdapter extends IgniteTxAdapter
// we are not changing obsolete entries.
// (innerSet and innerRemove will throw an exception
// if an entry is obsolete).
- if (txEntry.op() != READ)
+ if (txEntry.op() != READ && !txEntry.groupLockEntry())
checkCommitLocks(cached);
// Break out of while loop.
@@ -973,7 +996,7 @@ public abstract class IgniteTxLocalAdapter extends IgniteTxAdapter
else {
CacheStoreManager store = store();
- if (store != null && !internal()) {
+ if (store != null && (!internal() || groupLock())) {
try {
store.sessionEnd(this, true);
}
@@ -1079,7 +1102,7 @@ public abstract class IgniteTxLocalAdapter extends IgniteTxAdapter
CacheStoreManager store = store();
if (store != null && (near() || store.isWriteToStoreFromDht())) {
- if (!internal())
+ if (!internal() || groupLock())
store.sessionEnd(this, false);
}
}
@@ -1129,6 +1152,8 @@ public abstract class IgniteTxLocalAdapter extends IgniteTxAdapter
cacheCtx.checkSecurity(SecurityPermission.CACHE_READ);
+ groupLockSanityCheck(cacheCtx, keys);
+
boolean single = keysCnt == 1;
Collection<KeyCacheObject> lockKeys = null;
@@ -1160,7 +1185,7 @@ public abstract class IgniteTxLocalAdapter extends IgniteTxAdapter
cacheCtx.addResult(map, key, val, skipVals, keepCacheObjects, deserializePortable, false);
}
else {
- assert txEntry.op() == TRANSFORM;
+ assert txEntry.op() == TRANSFORM || (groupLock() && !txEntry.groupLockEntry());
while (true) {
try {
@@ -1238,7 +1263,7 @@ public abstract class IgniteTxLocalAdapter extends IgniteTxAdapter
CacheObject val = null;
- if (!pessimistic() || readCommitted() && !skipVals) {
+ if (!pessimistic() || readCommitted() || groupLock() && !skipVals) {
IgniteCacheExpiryPolicy accessPlc =
optimistic() ? accessPolicy(cacheCtx, txKey, expiryPlc) : null;
@@ -1286,6 +1311,8 @@ public abstract class IgniteTxLocalAdapter extends IgniteTxAdapter
null,
skipStore);
+ if (groupLock())
+ txEntry.groupLockEntry(true);
// As optimization, mark as checked immediately
// for non-pessimistic if value is not null.
@@ -1500,7 +1527,7 @@ public abstract class IgniteTxLocalAdapter extends IgniteTxAdapter
nextVer = cctx.versions().next(topologyVersion());
while (true) {
- assert txEntry != null || readCommitted() || skipVals;
+ assert txEntry != null || readCommitted() || groupLock() || skipVals;
GridCacheEntryEx e = txEntry == null ? entryEx(cacheCtx, txKey) : txEntry.cached();
@@ -1517,7 +1544,8 @@ public abstract class IgniteTxLocalAdapter extends IgniteTxAdapter
log.debug("Got removed entry in transaction getAll method " +
"(will try again): " + e);
- if (pessimistic() && !readCommitted() && !isRollbackOnly()) {
+ if (pessimistic() && !readCommitted() && !isRollbackOnly() &&
+ (!groupLock() || F.eq(e.key(), groupLockKey()))) {
U.error(log, "Inconsistent transaction state (entry got removed while " +
"holding lock) [entry=" + e + ", tx=" + IgniteTxLocalAdapter.this + "]");
@@ -1535,7 +1563,7 @@ public abstract class IgniteTxLocalAdapter extends IgniteTxAdapter
// In pessimistic mode, we should always be able to set.
assert set || !pessimistic();
- if (readCommitted() || skipVals) {
+ if (readCommitted() || groupLock() || skipVals) {
cacheCtx.evicts().touch(e, topologyVersion());
if (visibleVal != null) {
@@ -1626,7 +1654,7 @@ public abstract class IgniteTxLocalAdapter extends IgniteTxAdapter
return new GridFinishedFuture<>(retMap);
// Handle locks.
- if (pessimistic() && !readCommitted() && !skipVals) {
+ if (pessimistic() && !readCommitted() && !groupLock() && !skipVals) {
if (expiryPlc == null)
expiryPlc = cacheCtx.expiry();
@@ -1732,7 +1760,7 @@ public abstract class IgniteTxLocalAdapter extends IgniteTxAdapter
}
}
- if (!missed.isEmpty() && cacheCtx.isLocal()) {
+ if (!missed.isEmpty() && (cacheCtx.isReplicated() || cacheCtx.isLocal())) {
return checkMissed(cacheCtx,
retMap,
missed,
@@ -1783,7 +1811,7 @@ public abstract class IgniteTxLocalAdapter extends IgniteTxAdapter
}
}
else {
- assert optimistic() || readCommitted() || skipVals;
+ assert optimistic() || readCommitted() || groupLock() || skipVals;
final Collection<KeyCacheObject> redos = new ArrayList<>();
@@ -2008,6 +2036,8 @@ public abstract class IgniteTxLocalAdapter extends IgniteTxAdapter
if (invokeMap != null)
transform = true;
+ groupLockSanityCheck(cacheCtx, keys);
+
for (Object key : keys) {
if (key == null) {
rollback();
@@ -2164,9 +2194,12 @@ public abstract class IgniteTxLocalAdapter extends IgniteTxAdapter
if (!implicit() && readCommitted() && !cacheCtx.offheapTiered())
cacheCtx.evicts().touch(entry, topologyVersion());
+ if (groupLock() && !lockOnly)
+ txEntry.groupLockEntry(true);
+
enlisted.add(cacheKey);
- if (!pessimistic() && !implicit()) {
+ if ((!pessimistic() && !implicit()) || (groupLock() && !lockOnly)) {
txEntry.markValid();
if (old == null) {
@@ -2611,7 +2644,7 @@ public abstract class IgniteTxLocalAdapter extends IgniteTxAdapter
null,
opCtx != null && opCtx.skipStore());
- if (pessimistic()) {
+ if (pessimistic() && !groupLock()) {
// Loose all skipped.
final Set<KeyCacheObject> loaded = loadFut.get();
@@ -2834,7 +2867,7 @@ public abstract class IgniteTxLocalAdapter extends IgniteTxAdapter
// Acquire locks only after having added operation to the write set.
// Otherwise, during rollback we will not know whether locks need
// to be rolled back.
- if (pessimistic()) {
+ if (pessimistic() && !groupLock()) {
// Loose all skipped.
final Collection<KeyCacheObject> passedKeys = F.view(enlisted, F0.notIn(loadFut.get()));
@@ -2952,6 +2985,108 @@ public abstract class IgniteTxLocalAdapter extends IgniteTxAdapter
}
/**
+ * Adds key mapping to transaction.
+ * @param keys Keys to add.
+ */
+ protected void addGroupTxMapping(Collection<IgniteTxKey> keys) {
+ // No-op. This method is overriden in transactions that store key to remote node mapping
+ // for commit.
+ }
+
+ /**
+ * Checks that affinity keys are enlisted in group transaction on start.
+ *
+ * @param cacheCtx Cache context.
+ * @param keys Keys to check.
+ * @throws IgniteCheckedException If sanity check failed.
+ */
+ private <K> void groupLockSanityCheck(GridCacheContext cacheCtx, Iterable<? extends K> keys)
+ throws IgniteCheckedException
+ {
+ if (groupLock() && cctx.kernalContext().config().isCacheSanityCheckEnabled()) {
+ // Note that affinity is called without mapper on purpose.
+ int affinityPart = cacheCtx.config().getAffinity().partition(grpLockKey.key());
+
+ for (K key : keys) {
+ if (partitionLock()) {
+ int part = cacheCtx.affinity().partition(key);
+
+ if (affinityPart != part)
+ throw new IgniteCheckedException("Failed to enlist key into group-lock transaction (given " +
+ "key does not belong to locked partition) [key=" + key + ", affinityPart=" + affinityPart +
+ ", part=" + part + ", groupLockKey=" + grpLockKey + ']');
+ }
+ else {
+ KeyCacheObject cacheKey =
+ cacheCtx.toCacheKeyObject(cacheCtx.config().getAffinityMapper().affinityKey(key));
+
+ IgniteTxKey affinityKey = cacheCtx.txKey(cacheKey);
+
+ if (!grpLockKey.equals(affinityKey))
+ throw new IgniteCheckedException("Failed to enlist key into group-lock transaction (affinity key was " +
+ "not enlisted to transaction on start) [key=" + key + ", affinityKey=" + affinityKey +
+ ", groupLockKey=" + grpLockKey + ']');
+ }
+ }
+ }
+ }
+
+ /**
+ * Performs keys locking for affinity-based group lock transactions.
+ * @return Lock future.
+ */
+ @Override public <K> IgniteInternalFuture<?> groupLockAsync(GridCacheContext cacheCtx, Collection<K> keys) {
+ assert groupLock();
+
+ try {
+ init();
+
+ GridCacheReturn ret = new GridCacheReturn(localResult(), false);
+
+ Collection<KeyCacheObject> enlisted = new ArrayList<>();
+
+ Set<KeyCacheObject> skipped = enlistWrite(
+ cacheCtx,
+ keys,
+ /** cached entry */null,
+ /** expiry - leave unchanged */null,
+ /** implicit */false,
+ /** lookup map */null,
+ /** invoke map */null,
+ /** invoke arguments */null,
+ /** retval */false,
+ /** lock only */true,
+ CU.empty0(),
+ ret,
+ enlisted,
+ null,
+ null,
+ cacheCtx.skipStore()
+ ).get();
+
+ // No keys should be skipped with empty filter.
+ assert F.isEmpty(skipped);
+
+ // Lock group key in pessimistic mode only.
+ return pessimistic() ?
+ cacheCtx.cache().txLockAsync(enlisted,
+ lockTimeout(),
+ this,
+ false,
+ false,
+ isolation,
+ isInvalidate(),
+ -1L) :
+ new GridFinishedFuture<>();
+ }
+ catch (IgniteCheckedException e) {
+ setRollbackOnly();
+
+ return new GridFinishedFuture<Object>(e);
+ }
+ }
+
+ /**
* Initializes read map.
*
* @return {@code True} if transaction was successfully started.
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/8a6dd007/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxLocalEx.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxLocalEx.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxLocalEx.java
index 14562ab..61041e1 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxLocalEx.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxLocalEx.java
@@ -58,9 +58,14 @@ public interface IgniteTxLocalEx extends IgniteInternalTx {
public void userRollback() throws IgniteCheckedException;
/**
+ * @return Group lock entry if this is a group-lock transaction.
+ */
+ @Nullable public IgniteTxEntry groupLockEntry();
+
+ /**
* @param cacheCtx Cache context.
* @param keys Keys to get.
- * @param cached Cached entry if this method is called from entry wrapper
+ * @param cached Cached entry if this method is called from entry wrapper.
* Cached entry is passed if and only if there is only one key in collection of keys.
* @param deserializePortable Deserialize portable flag.
* @param skipVals Skip values flag.
@@ -139,6 +144,20 @@ public interface IgniteTxLocalEx extends IgniteInternalTx {
Map<KeyCacheObject, GridCacheVersion> drMap);
/**
+ * Performs keys locking for affinity-based group lock transactions.
+ *
+ * @param cacheCtx Cache context.
+ * @param keys Keys to lock.
+ * @return Lock future.
+ */
+ public <K> IgniteInternalFuture<?> groupLockAsync(GridCacheContext cacheCtx, Collection<K> keys);
+
+ /**
+ * @return {@code True} if keys from the same partition are allowed to be enlisted in group-lock transaction.
+ */
+ public boolean partitionLock();
+
+ /**
* @return Return value for
*/
public GridCacheReturn implicitSingleResult();
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/8a6dd007/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxManager.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxManager.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxManager.java
index 8e95a5d..8a1d490 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxManager.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxManager.java
@@ -347,6 +347,8 @@ public class IgniteTxManager extends GridCacheSharedManagerAdapter {
* @param isolation Isolation.
* @param timeout transaction timeout.
* @param txSize Expected transaction size.
+ * @param grpLockKey Group lock key if this is a group-lock transaction.
+ * @param partLock {@code True} if partition is locked.
* @return New transaction.
*/
public IgniteTxLocalAdapter newTx(
@@ -357,7 +359,9 @@ public class IgniteTxManager extends GridCacheSharedManagerAdapter {
TransactionIsolation isolation,
long timeout,
boolean storeEnabled,
- int txSize) {
+ int txSize,
+ @Nullable IgniteTxKey grpLockKey,
+ boolean partLock) {
assert sysCacheCtx == null || sysCacheCtx.systemTx();
UUID subjId = null; // TODO GG-9141 how to get subj ID?
@@ -375,6 +379,8 @@ public class IgniteTxManager extends GridCacheSharedManagerAdapter {
timeout,
storeEnabled,
txSize,
+ grpLockKey,
+ partLock,
subjId,
taskNameHash);
@@ -1201,10 +1207,13 @@ public class IgniteTxManager extends GridCacheSharedManagerAdapter {
cctx.kernalContext().dataStructures().onTxCommitted(tx);
// 4. Unlock write resources.
- unlockMultiple(tx, tx.writeEntries());
+ if (tx.groupLock())
+ unlockGroupLocks(tx);
+ else
+ unlockMultiple(tx, tx.writeEntries());
// 5. For pessimistic transaction, unlock read resources if required.
- if (tx.pessimistic() && !tx.readCommitted())
+ if (tx.pessimistic() && !tx.readCommitted() && !tx.groupLock())
unlockMultiple(tx, tx.readEntries());
// 6. Notify evictions.
@@ -1432,7 +1441,7 @@ public class IgniteTxManager extends GridCacheSharedManagerAdapter {
* @param tx Transaction to notify evictions for.
*/
private void notifyEvitions(IgniteInternalTx tx) {
- if (tx.internal())
+ if (tx.internal() && !tx.groupLock())
return;
for (IgniteTxEntry txEntry : tx.allEntries())
@@ -1608,6 +1617,51 @@ public class IgniteTxManager extends GridCacheSharedManagerAdapter {
}
/**
+ * Unlocks entries locked by group transaction.
+ *
+ * @param txx Transaction.
+ */
+ @SuppressWarnings("unchecked")
+ private void unlockGroupLocks(IgniteInternalTx txx) {
+ IgniteTxKey grpLockKey = txx.groupLockKey();
+
+ assert grpLockKey != null;
+
+ if (grpLockKey == null)
+ return;
+
+ IgniteTxEntry txEntry = txx.entry(grpLockKey);
+
+ assert txEntry != null || (txx.near() && !txx.local());
+
+ if (txEntry != null) {
+ GridCacheContext cacheCtx = txEntry.context();
+
+ // Group-locked entries must be locked.
+ while (true) {
+ try {
+ GridCacheEntryEx entry = txEntry.cached();
+
+ assert entry != null;
+
+ entry.txUnlock(txx);
+
+ break;
+ }
+ catch (GridCacheEntryRemovedException ignored) {
+ if (log.isDebugEnabled())
+ log.debug("Got removed entry in TM unlockGroupLocks(..) method (will retry): " + txEntry);
+
+ GridCacheAdapter cache = cacheCtx.cache();
+
+ // Renew cache entry.
+ txEntry.cached(cache.entryEx(txEntry.key()));
+ }
+ }
+ }
+ }
+
+ /**
* @param tx Owning transaction.
* @param entries Entries to unlock.
*/
@@ -1762,7 +1816,7 @@ public class IgniteTxManager extends GridCacheSharedManagerAdapter {
if (nearVer.equals(tx.nearXidVersion())) {
TransactionState state = tx.state();
- IgniteInternalFuture<?> prepFut = tx.currentPrepareFuture();
+ IgniteInternalFuture<IgniteInternalTx> prepFut = tx.currentPrepareFuture();
if (prepFut != null && !prepFut.isDone()) {
if (log.isDebugEnabled())
@@ -1774,8 +1828,8 @@ public class IgniteTxManager extends GridCacheSharedManagerAdapter {
final Collection<GridCacheVersion> processedVers0 = processedVers;
- prepFut.listen(new CI1<IgniteInternalFuture<?>>() {
- @Override public void apply(IgniteInternalFuture<?> prepFut) {
+ prepFut.listen(new CI1<IgniteInternalFuture<IgniteInternalTx>>() {
+ @Override public void apply(IgniteInternalFuture<IgniteInternalTx> prepFut) {
if (log.isDebugEnabled())
log.debug("Transaction prepare future finished: " + tx);
@@ -1975,11 +2029,11 @@ public class IgniteTxManager extends GridCacheSharedManagerAdapter {
if (tx.state() == PREPARED)
commitIfPrepared(tx);
else {
- IgniteInternalFuture<?> prepFut = tx.currentPrepareFuture();
+ IgniteInternalFuture<IgniteInternalTx> prepFut = tx.currentPrepareFuture();
if (prepFut != null) {
- prepFut.listen(new CI1<IgniteInternalFuture<?>>() {
- @Override public void apply(IgniteInternalFuture<?> fut) {
+ prepFut.listen(new CI1<IgniteInternalFuture<IgniteInternalTx>>() {
+ @Override public void apply(IgniteInternalFuture<IgniteInternalTx> fut) {
if (tx.state() == PREPARED)
commitIfPrepared(tx);
else if (tx.setRollbackOnly())
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/8a6dd007/modules/core/src/main/java/org/apache/ignite/internal/processors/resource/GridResourceField.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/resource/GridResourceField.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/resource/GridResourceField.java
index fed7ebd..6b921f4 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/resource/GridResourceField.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/resource/GridResourceField.java
@@ -22,7 +22,6 @@ import org.jetbrains.annotations.*;
import java.lang.annotation.*;
import java.lang.reflect.*;
-import java.util.*;
/**
* Wrapper for data where resource should be injected.
@@ -44,7 +43,10 @@ class GridResourceField {
* @param field Field where resource should be injected.
* @param ann Resource annotation.
*/
- GridResourceField(@NotNull Field field, @NotNull Annotation ann) {
+ GridResourceField(Field field, @Nullable Annotation ann) {
+ assert field != null;
+ assert ann != null || GridResourceUtils.mayRequireResources(field);
+
this.field = field;
this.ann = ann;
@@ -76,16 +78,6 @@ class GridResourceField {
return ann == null;
}
- /**
- * @param c Closure.
- */
- public static GridResourceField[] toArray(Collection<GridResourceField> c) {
- if (c.isEmpty())
- return EMPTY_ARRAY;
-
- return c.toArray(new GridResourceField[c.size()]);
- }
-
/** {@inheritDoc} */
@Override public String toString() {
return S.toString(GridResourceField.class, this);