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 2017/09/20 12:04:10 UTC
[1/7] ignite git commit: IGNITE-6376 Web console: Enable task and job
events in demo mode by default.
Repository: ignite
Updated Branches:
refs/heads/ignite-6181-2 685f051ce -> dcb7511a6
IGNITE-6376 Web console: Enable task and job events in demo mode by default.
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/ff177838
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/ff177838
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/ff177838
Branch: refs/heads/ignite-6181-2
Commit: ff177838f80f068345876149bf44b01990e32529
Parents: 2299589
Author: vsisko <vs...@gridgain.com>
Authored: Wed Sep 20 17:19:58 2017 +0700
Committer: Alexey Kuznetsov <ak...@apache.org>
Committed: Wed Sep 20 17:19:58 2017 +0700
----------------------------------------------------------------------
.../internal/visor/node/VisorNodeDataCollectorTask.java | 2 +-
.../visor/node/VisorNodeDataCollectorTaskResult.java | 4 ++--
modules/web-console/backend/routes/demo/domains.json | 5 +++++
.../configuration/generator/JavaTransformer.service.js | 5 +++--
.../org/apache/ignite/console/demo/AgentClusterDemo.java | 9 ++++++++-
5 files changed, 19 insertions(+), 6 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ignite/blob/ff177838/modules/core/src/main/java/org/apache/ignite/internal/visor/node/VisorNodeDataCollectorTask.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/visor/node/VisorNodeDataCollectorTask.java b/modules/core/src/main/java/org/apache/ignite/internal/visor/node/VisorNodeDataCollectorTask.java
index 80664a1..abd9ce2 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/visor/node/VisorNodeDataCollectorTask.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/visor/node/VisorNodeDataCollectorTask.java
@@ -89,7 +89,7 @@ public class VisorNodeDataCollectorTask extends VisorMultiNodeTask<VisorNodeData
taskRes.getTopologyVersions().put(nid, jobRes.getTopologyVersion());
- taskRes.isTaskMonitoringEnabled().put(nid, jobRes.isTaskMonitoringEnabled());
+ taskRes.getTaskMonitoringEnabled().put(nid, jobRes.isTaskMonitoringEnabled());
taskRes.getErrorCounts().put(nid, jobRes.getErrorCount());
http://git-wip-us.apache.org/repos/asf/ignite/blob/ff177838/modules/core/src/main/java/org/apache/ignite/internal/visor/node/VisorNodeDataCollectorTaskResult.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/visor/node/VisorNodeDataCollectorTaskResult.java b/modules/core/src/main/java/org/apache/ignite/internal/visor/node/VisorNodeDataCollectorTaskResult.java
index c1e2f1f..6e10b84 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/visor/node/VisorNodeDataCollectorTaskResult.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/visor/node/VisorNodeDataCollectorTaskResult.java
@@ -169,7 +169,7 @@ public class VisorNodeDataCollectorTaskResult extends VisorDataTransferObject {
/**
* @return All task monitoring state collected from nodes.
*/
- public Map<UUID, Boolean> isTaskMonitoringEnabled() {
+ public Map<UUID, Boolean> getTaskMonitoringEnabled() {
return taskMonitoringEnabled;
}
@@ -283,7 +283,7 @@ public class VisorNodeDataCollectorTaskResult extends VisorDataTransferObject {
unhandledEx.putAll(res.getUnhandledEx());
gridNames.putAll(res.getGridNames());
topVersions.putAll(res.getTopologyVersions());
- taskMonitoringEnabled.putAll(res.isTaskMonitoringEnabled());
+ taskMonitoringEnabled.putAll(res.getTaskMonitoringEnabled());
errCnts.putAll(res.getErrorCounts());
evts.addAll(res.getEvents());
evtsEx.putAll(res.getEventsEx());
http://git-wip-us.apache.org/repos/asf/ignite/blob/ff177838/modules/web-console/backend/routes/demo/domains.json
----------------------------------------------------------------------
diff --git a/modules/web-console/backend/routes/demo/domains.json b/modules/web-console/backend/routes/demo/domains.json
index ae46758..25f5019 100644
--- a/modules/web-console/backend/routes/demo/domains.json
+++ b/modules/web-console/backend/routes/demo/domains.json
@@ -17,6 +17,7 @@
"className": "Integer"
}
],
+ "keyFieldName": "id",
"valueFields": [
{
"databaseFieldName": "NAME",
@@ -60,6 +61,7 @@
"className": "String"
}
],
+ "keyFieldName": "id",
"valueFields": [
{
"databaseFieldName": "COUNTRY_ID",
@@ -156,6 +158,7 @@
"className": "Double"
}
],
+ "keyFieldName": "id",
"valueFields": [
{
"databaseFieldName": "DEPARTMENT_ID",
@@ -241,6 +244,7 @@
"className": "Integer"
}
],
+ "keyFieldName": "id",
"valueFields": [
{
"databaseFieldName": "NAME",
@@ -284,6 +288,7 @@
"className": "String"
}
],
+ "keyFieldName": "id",
"valueFields": [
{
"databaseFieldName": "PARKING_ID",
http://git-wip-us.apache.org/repos/asf/ignite/blob/ff177838/modules/web-console/frontend/app/modules/configuration/generator/JavaTransformer.service.js
----------------------------------------------------------------------
diff --git a/modules/web-console/frontend/app/modules/configuration/generator/JavaTransformer.service.js b/modules/web-console/frontend/app/modules/configuration/generator/JavaTransformer.service.js
index 64fc172..4e6204f 100644
--- a/modules/web-console/frontend/app/modules/configuration/generator/JavaTransformer.service.js
+++ b/modules/web-console/frontend/app/modules/configuration/generator/JavaTransformer.service.js
@@ -1606,6 +1606,9 @@ export default class IgniteJavaTransformer extends AbstractTransformer {
shortFactoryCls = this.javaTypes.shortClassName(factoryCls);
}
+ if ((_.nonEmpty(clientNearCaches) || demo) && shortFactoryCls)
+ imports.push('org.apache.ignite.Ignite');
+
sb.append(`package ${pkg};`)
.emptyLine();
@@ -1651,8 +1654,6 @@ export default class IgniteJavaTransformer extends AbstractTransformer {
}
if ((_.nonEmpty(clientNearCaches) || demo) && shortFactoryCls) {
- imports.push('org.apache.ignite.Ignite');
-
sb.append(`Ignite ignite = Ignition.start(${cfgRef});`);
_.forEach(clientNearCaches, (cache, idx) => {
http://git-wip-us.apache.org/repos/asf/ignite/blob/ff177838/modules/web-console/web-agent/src/main/java/org/apache/ignite/console/demo/AgentClusterDemo.java
----------------------------------------------------------------------
diff --git a/modules/web-console/web-agent/src/main/java/org/apache/ignite/console/demo/AgentClusterDemo.java b/modules/web-console/web-agent/src/main/java/org/apache/ignite/console/demo/AgentClusterDemo.java
index 73577b5..886888b 100644
--- a/modules/web-console/web-agent/src/main/java/org/apache/ignite/console/demo/AgentClusterDemo.java
+++ b/modules/web-console/web-agent/src/main/java/org/apache/ignite/console/demo/AgentClusterDemo.java
@@ -55,6 +55,7 @@ import static org.apache.ignite.IgniteSystemProperties.IGNITE_QUIET;
import static org.apache.ignite.IgniteSystemProperties.IGNITE_UPDATE_NOTIFIER;
import static org.apache.ignite.console.demo.AgentDemoUtils.newScheduledThreadPool;
import static org.apache.ignite.events.EventType.EVTS_DISCOVERY;
+import static org.apache.ignite.internal.visor.util.VisorTaskUtils.VISOR_TASK_EVTS;
import static org.apache.ignite.internal.IgniteNodeAttributes.ATTR_REST_JETTY_ADDRS;
import static org.apache.ignite.internal.IgniteNodeAttributes.ATTR_REST_JETTY_PORT;
@@ -92,7 +93,13 @@ public class AgentClusterDemo {
cfg.setIgniteInstanceName((client ? "demo-client-" : "demo-server-" ) + gridIdx);
cfg.setLocalHost("127.0.0.1");
cfg.setEventStorageSpi(new MemoryEventStorageSpi());
- cfg.setIncludeEventTypes(EVTS_DISCOVERY);
+
+ int[] evts = new int[EVTS_DISCOVERY.length + VISOR_TASK_EVTS.length];
+
+ System.arraycopy(EVTS_DISCOVERY, 0, evts, 0, EVTS_DISCOVERY.length);
+ System.arraycopy(VISOR_TASK_EVTS, 0, evts, EVTS_DISCOVERY.length, VISOR_TASK_EVTS.length);
+
+ cfg.setIncludeEventTypes(evts);
cfg.getConnectorConfiguration().setPort(basePort);
[5/7] ignite git commit: ignite-6181
Posted by sb...@apache.org.
ignite-6181
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/56dbdc2f
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/56dbdc2f
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/56dbdc2f
Branch: refs/heads/ignite-6181-2
Commit: 56dbdc2fc25f5a53570e59a87767a97fb7108e49
Parents: 2299589
Author: sboikov <sb...@gridgain.com>
Authored: Wed Sep 20 10:58:24 2017 +0300
Committer: sboikov <sb...@gridgain.com>
Committed: Wed Sep 20 14:56:00 2017 +0300
----------------------------------------------------------------------
.../IgniteDiagnosticPrepareContext.java | 4 +-
.../processors/cache/GridCacheAdapter.java | 2 +-
.../processors/cache/GridCacheMapEntry.java | 9 +-
.../GridDistributedTxRemoteAdapter.java | 2 +-
.../distributed/dht/GridDhtTxFinishFuture.java | 2 +-
.../cache/distributed/dht/GridDhtTxLocal.java | 6 +-
.../distributed/dht/GridDhtTxLocalAdapter.java | 4 +-
.../colocated/GridDhtColocatedLockFuture.java | 56 +-
.../near/GridNearFastFinishFuture.java | 79 +++
.../distributed/near/GridNearLockFuture.java | 56 +-
.../near/GridNearOptimisticTxPrepareFuture.java | 3 +-
.../near/GridNearTxFinishFuture.java | 15 +-
.../cache/distributed/near/GridNearTxLocal.java | 276 +++++---
.../distributed/near/NearTxFinishFuture.java | 31 +
.../cache/transactions/IgniteInternalTx.java | 2 +-
.../cache/transactions/IgniteTxAdapter.java | 13 +-
.../transactions/IgniteTxLocalAdapter.java | 8 +-
.../cache/transactions/IgniteTxLocalEx.java | 4 +-
.../cache/transactions/IgniteTxManager.java | 79 ++-
.../timeout/GridTimeoutProcessor.java | 21 +-
.../cache/IgniteTxConfigCacheSelfTest.java | 14 +
.../IgniteCacheThreadLocalTxTest.java | 223 +++++++
.../IgniteOptimisticTxSuspendResumeTest.java | 6 +-
...ionedMultiNodeLongTxTimeout2FullApiTest.java | 34 +
.../TxRollbackOnTimeoutNearCacheTest.java | 28 +
...ollbackOnTimeoutNoDeadlockDetectionTest.java | 47 ++
.../transactions/TxRollbackOnTimeoutTest.java | 658 +++++++++++++++++++
.../IgniteCacheFullApiSelfTestSuite.java | 2 +
.../ignite/testsuites/IgniteCacheTestSuite.java | 9 +
.../hadoop/impl/HadoopTxConfigCacheTest.java | 4 +-
30 files changed, 1502 insertions(+), 195 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ignite/blob/56dbdc2f/modules/core/src/main/java/org/apache/ignite/internal/IgniteDiagnosticPrepareContext.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/IgniteDiagnosticPrepareContext.java b/modules/core/src/main/java/org/apache/ignite/internal/IgniteDiagnosticPrepareContext.java
index 378dc74..ed8d35e 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/IgniteDiagnosticPrepareContext.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/IgniteDiagnosticPrepareContext.java
@@ -75,7 +75,7 @@ public class IgniteDiagnosticPrepareContext {
* @param keys Entry keys.
* @param msg Initial message.
*/
- public void txKeyInfo(UUID nodeId, int cacheId, Collection<KeyCacheObject> keys, String msg) {
+ public void txKeyInfo(UUID nodeId, int cacheId, Collection<KeyCacheObject> keys, String msg) {
closure(nodeId).add(msg, new TxEntriesInfoClosure(cacheId, keys));
}
@@ -280,4 +280,4 @@ public class IgniteDiagnosticPrepareContext {
}
}
}
-}
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/ignite/blob/56dbdc2f/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheAdapter.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheAdapter.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheAdapter.java
index 92a8245..05c8cc3 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheAdapter.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheAdapter.java
@@ -1857,7 +1857,7 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
return new GridFinishedFuture<>(e);
}
- tx = ctx.tm().threadLocalTx(ctx.systemTx() ? ctx : null);
+ tx = ctx.tm().threadLocalTx(ctx);
}
if (tx == null || tx.implicit()) {
http://git-wip-us.apache.org/repos/asf/ignite/blob/56dbdc2f/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheMapEntry.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheMapEntry.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheMapEntry.java
index 958f156..7b60b9c 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheMapEntry.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheMapEntry.java
@@ -2463,7 +2463,7 @@ public abstract class GridCacheMapEntry extends GridMetadataAwareAdapter impleme
/** {@inheritDoc} */
@Nullable @Override public CacheObject peek(@Nullable IgniteCacheExpiryPolicy plc)
throws GridCacheEntryRemovedException, IgniteCheckedException {
- IgniteInternalTx tx = cctx.tm().localTxx();
+ IgniteInternalTx tx = cctx.tm().localTx();
AffinityTopologyVersion topVer = tx != null ? tx.topologyVersion() : cctx.affinity().affinityTopologyVersion();
@@ -3110,10 +3110,7 @@ public abstract class GridCacheMapEntry extends GridMetadataAwareAdapter impleme
* @return Current transaction.
*/
private IgniteTxLocalAdapter currentTx() {
- if (cctx.isDht())
- return cctx.dht().near().context().tm().localTx();
- else
- return cctx.tm().localTx();
+ return cctx.tm().localTx();
}
/** {@inheritDoc} */
@@ -3491,7 +3488,7 @@ public abstract class GridCacheMapEntry extends GridMetadataAwareAdapter impleme
}
}
- IgniteInternalTx tx = cctx.tm().localTxx();
+ IgniteInternalTx tx = cctx.tm().localTx();
if (tx != null) {
IgniteTxEntry e = tx.entry(txKey());
http://git-wip-us.apache.org/repos/asf/ignite/blob/56dbdc2f/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridDistributedTxRemoteAdapter.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridDistributedTxRemoteAdapter.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridDistributedTxRemoteAdapter.java
index ea6461d..e5bcc46 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridDistributedTxRemoteAdapter.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridDistributedTxRemoteAdapter.java
@@ -855,7 +855,7 @@ public abstract class GridDistributedTxRemoteAdapter extends IgniteTxAdapter
// Note that we don't evict near entries here -
// they will be deleted by their corresponding transactions.
if (state(ROLLING_BACK) || state() == UNKNOWN) {
- cctx.tm().rollbackTx(this);
+ cctx.tm().rollbackTx(this, false);
state(ROLLED_BACK);
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/56dbdc2f/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxFinishFuture.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxFinishFuture.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxFinishFuture.java
index 5311ddc..6380710 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxFinishFuture.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxFinishFuture.java
@@ -227,7 +227,7 @@ public final class GridDhtTxFinishFuture<K, V> extends GridCacheCompoundIdentity
try {
boolean nodeStop = err != null && X.hasCause(err, NodeStoppingException.class);
- this.tx.tmFinish(err == null, nodeStop);
+ this.tx.tmFinish(err == null, nodeStop, false);
}
catch (IgniteCheckedException finishErr) {
U.error(log, "Failed to finish tx: " + tx, e);
http://git-wip-us.apache.org/repos/asf/ignite/blob/56dbdc2f/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxLocal.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxLocal.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxLocal.java
index ab5631e..28cc018 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxLocal.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxLocal.java
@@ -436,7 +436,7 @@ public class GridDhtTxLocal extends GridDhtTxLocalAdapter implements GridCacheMa
if (prepFut != null)
prepFut.get(); // Check for errors.
- boolean finished = localFinish(commit);
+ boolean finished = localFinish(commit, false);
if (!finished)
err = new IgniteCheckedException("Failed to finish transaction [commit=" + commit +
@@ -542,7 +542,7 @@ public class GridDhtTxLocal extends GridDhtTxLocalAdapter implements GridCacheMa
/** {@inheritDoc} */
@SuppressWarnings({"CatchGenericClass", "ThrowableInstanceNeverThrown"})
- @Override public boolean localFinish(boolean commit) throws IgniteCheckedException {
+ @Override public boolean localFinish(boolean commit, boolean clearThreadMap) throws IgniteCheckedException {
assert nearFinFutId != null || isInvalidate() || !commit || isSystemInvalidate()
|| onePhaseCommit() || state() == PREPARED :
"Invalid state [nearFinFutId=" + nearFinFutId + ", isInvalidate=" + isInvalidate() + ", commit=" + commit +
@@ -550,7 +550,7 @@ public class GridDhtTxLocal extends GridDhtTxLocalAdapter implements GridCacheMa
assert nearMiniId != 0;
- return super.localFinish(commit);
+ return super.localFinish(commit, clearThreadMap);
}
/** {@inheritDoc} */
http://git-wip-us.apache.org/repos/asf/ignite/blob/56dbdc2f/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxLocalAdapter.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxLocalAdapter.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxLocalAdapter.java
index 86eac42..e4a7141 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxLocalAdapter.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxLocalAdapter.java
@@ -734,7 +734,7 @@ public abstract class GridDhtTxLocalAdapter extends IgniteTxLocalAdapter {
/** {@inheritDoc} */
@SuppressWarnings({"CatchGenericClass", "ThrowableInstanceNeverThrown"})
- @Override public boolean localFinish(boolean commit) throws IgniteCheckedException {
+ @Override public boolean localFinish(boolean commit, boolean clearThreadMap) throws IgniteCheckedException {
if (log.isDebugEnabled())
log.debug("Finishing dht local tx [tx=" + this + ", commit=" + commit + "]");
@@ -773,7 +773,7 @@ public abstract class GridDhtTxLocalAdapter extends IgniteTxLocalAdapter {
if (commit && !isRollbackOnly())
userCommit();
else
- userRollback();
+ userRollback(clearThreadMap);
}
catch (IgniteCheckedException e) {
err = e;
http://git-wip-us.apache.org/repos/asf/ignite/blob/56dbdc2f/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/colocated/GridDhtColocatedLockFuture.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/colocated/GridDhtColocatedLockFuture.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/colocated/GridDhtColocatedLockFuture.java
index 58c6319..b3f4c0d 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/colocated/GridDhtColocatedLockFuture.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/colocated/GridDhtColocatedLockFuture.java
@@ -54,6 +54,7 @@ import org.apache.ignite.internal.processors.cache.distributed.near.GridNearLock
import org.apache.ignite.internal.processors.cache.distributed.near.GridNearLockRequest;
import org.apache.ignite.internal.processors.cache.distributed.near.GridNearLockResponse;
import org.apache.ignite.internal.processors.cache.distributed.near.GridNearTxLocal;
+import org.apache.ignite.internal.processors.cache.transactions.IgniteInternalTx;
import org.apache.ignite.internal.processors.cache.transactions.IgniteTxEntry;
import org.apache.ignite.internal.processors.cache.transactions.IgniteTxKey;
import org.apache.ignite.internal.processors.cache.transactions.TxDeadlock;
@@ -226,12 +227,6 @@ public final class GridDhtColocatedLockFuture extends GridCacheCompoundIdentityF
log = U.logger(cctx.kernalContext(), logRef, GridDhtColocatedLockFuture.class);
}
- if (timeout > 0) {
- timeoutObj = new LockTimeoutObject();
-
- cctx.time().addTimeoutObject(timeoutObj);
- }
-
valMap = new ConcurrentHashMap8<>();
}
@@ -399,7 +394,7 @@ public final class GridDhtColocatedLockFuture extends GridCacheCompoundIdentityF
* @param success Success flag.
*/
public void complete(boolean success) {
- onComplete(success, true);
+ onComplete(success, true, true);
}
/**
@@ -533,7 +528,7 @@ public final class GridDhtColocatedLockFuture extends GridCacheCompoundIdentityF
/** {@inheritDoc} */
@Override public boolean cancel() {
if (onCancelled())
- onComplete(false, true);
+ onComplete(false, true, true);
return isCancelled();
}
@@ -556,7 +551,7 @@ public final class GridDhtColocatedLockFuture extends GridCacheCompoundIdentityF
if (err != null)
success = false;
- return onComplete(success, true);
+ return onComplete(success, true, true);
}
/**
@@ -564,9 +559,10 @@ public final class GridDhtColocatedLockFuture extends GridCacheCompoundIdentityF
*
* @param success {@code True} if lock was acquired.
* @param distribute {@code True} if need to distribute lock removal in case of failure.
+ * @param restoreTimeout {@code True} if need restore tx timeout callback.
* @return {@code True} if complete by this operation.
*/
- private boolean onComplete(boolean success, boolean distribute) {
+ private boolean onComplete(boolean success, boolean distribute, boolean restoreTimeout) {
if (log.isDebugEnabled())
log.debug("Received onComplete(..) callback [success=" + success + ", distribute=" + distribute +
", fut=" + this + ']');
@@ -577,6 +573,12 @@ public final class GridDhtColocatedLockFuture extends GridCacheCompoundIdentityF
if (tx != null)
cctx.tm().txContext(tx);
+ if (restoreTimeout && tx != null && tx.trackTimeout()) {
+ // Need restore timeout before onDone is called, but onComplete can be called concurrently,
+ // thus need ignore duplicated timeout objects.
+ tx.addTimeoutHandler(true);
+ }
+
if (super.onDone(success, err)) {
if (log.isDebugEnabled())
log.debug("Completing future: " + this);
@@ -675,6 +677,30 @@ public final class GridDhtColocatedLockFuture extends GridCacheCompoundIdentityF
* part. Note that if primary node leaves grid, the future will fail and transaction will be rolled back.
*/
void map() {
+ if (tx != null && tx.trackTimeout()) {
+ if (!tx.removeTimeoutHandler()) {
+ tx.finishFuture().listen(new IgniteInClosure<IgniteInternalFuture<IgniteInternalTx>>() {
+ @Override public void apply(IgniteInternalFuture<IgniteInternalTx> fut) {
+ IgniteTxTimeoutCheckedException err = new IgniteTxTimeoutCheckedException("Failed to " +
+ "acquire lock, transaction was rolled back on timeout [timeout=" + tx.timeout() +
+ ", tx=" + tx + ']');
+
+ onError(err);
+
+ onComplete(false, false, false);
+ }
+ });
+
+ return;
+ }
+ }
+
+ if (timeout > 0) {
+ timeoutObj = new LockTimeoutObject();
+
+ cctx.time().addTimeoutObject(timeoutObj);
+ }
+
// Obtain the topology version to use.
AffinityTopologyVersion topVer = cctx.mvcc().lastExplicitLockTopologyVersion(threadId);
@@ -930,7 +956,7 @@ public final class GridDhtColocatedLockFuture extends GridCacheCompoundIdentityF
if (log.isDebugEnabled())
log.debug("Entry being locked did not pass filter (will not lock): " + entry);
- onComplete(false, false);
+ onComplete(false, false, true);
return;
}
@@ -1307,7 +1333,7 @@ public final class GridDhtColocatedLockFuture extends GridCacheCompoundIdentityF
if (log.isDebugEnabled())
log.debug("Entry being locked did not pass filter (will not lock): " + entry);
- onComplete(false, false);
+ onComplete(false, false, true);
return false;
}
@@ -1419,12 +1445,12 @@ public final class GridDhtColocatedLockFuture extends GridCacheCompoundIdentityF
U.warn(log, "Failed to detect deadlock.", e);
}
- onComplete(false, true);
+ onComplete(false, true, true);
}
});
}
else
- onComplete(false, true);
+ onComplete(false, true, true);
}
/** {@inheritDoc} */
@@ -1673,4 +1699,4 @@ public final class GridDhtColocatedLockFuture extends GridCacheCompoundIdentityF
return S.toString(MiniFuture.class, this, "node", node.id(), "super", super.toString());
}
}
-}
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/ignite/blob/56dbdc2f/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearFastFinishFuture.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearFastFinishFuture.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearFastFinishFuture.java
new file mode 100644
index 0000000..158d85d
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearFastFinishFuture.java
@@ -0,0 +1,79 @@
+/*
+ * 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.internal.processors.cache.transactions.IgniteInternalTx;
+import org.apache.ignite.internal.util.future.GridFutureAdapter;
+
+import static org.apache.ignite.transactions.TransactionState.COMMITTED;
+import static org.apache.ignite.transactions.TransactionState.COMMITTING;
+import static org.apache.ignite.transactions.TransactionState.PREPARED;
+import static org.apache.ignite.transactions.TransactionState.PREPARING;
+import static org.apache.ignite.transactions.TransactionState.ROLLED_BACK;
+import static org.apache.ignite.transactions.TransactionState.ROLLING_BACK;
+
+/**
+ *
+ */
+public class GridNearFastFinishFuture extends GridFutureAdapter<IgniteInternalTx> implements NearTxFinishFuture {
+ /** */
+ private final GridNearTxLocal tx;
+
+ /** */
+ private final boolean commit;
+
+ /**
+ * @param tx Transaction.
+ * @param commit Commit flag.
+ */
+ GridNearFastFinishFuture(GridNearTxLocal tx, boolean commit) {
+ this.tx = tx;
+ this.commit = commit;
+ }
+
+ /** {@inheritDoc} */
+ @Override public boolean commit() {
+ return commit;
+ }
+
+ /**
+ *
+ */
+ public void finish() {
+ if (commit) {
+ tx.state(PREPARING);
+ tx.state(PREPARED);
+ tx.state(COMMITTING);
+
+ tx.context().tm().fastFinishTx(tx, true);
+
+ tx.state(COMMITTED);
+ }
+ else {
+ tx.state(PREPARING);
+ tx.state(PREPARED);
+ tx.state(ROLLING_BACK);
+
+ tx.context().tm().fastFinishTx(tx, false);
+
+ tx.state(ROLLED_BACK);
+ }
+
+ onDone(tx);
+ }
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/56dbdc2f/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearLockFuture.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearLockFuture.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearLockFuture.java
index c947715..c999351 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearLockFuture.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearLockFuture.java
@@ -50,6 +50,7 @@ import org.apache.ignite.internal.processors.cache.distributed.GridDistributedCa
import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtCacheEntry;
import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtTopologyFuture;
import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtTransactionalCacheAdapter;
+import org.apache.ignite.internal.processors.cache.transactions.IgniteInternalTx;
import org.apache.ignite.internal.processors.cache.transactions.IgniteTxEntry;
import org.apache.ignite.internal.processors.cache.transactions.IgniteTxKey;
import org.apache.ignite.internal.processors.cache.transactions.TxDeadlock;
@@ -230,12 +231,6 @@ public final class GridNearLockFuture extends GridCacheCompoundIdentityFuture<Bo
if (log == null)
log = U.logger(cctx.kernalContext(), logRef, GridNearLockFuture.class);
- if (timeout > 0) {
- timeoutObj = new LockTimeoutObject();
-
- cctx.time().addTimeoutObject(timeoutObj);
- }
-
valMap = new ConcurrentHashMap8<>();
}
@@ -434,7 +429,7 @@ public final class GridNearLockFuture extends GridCacheCompoundIdentityFuture<Bo
* @param success Success flag.
*/
public void complete(boolean success) {
- onComplete(success, true);
+ onComplete(success, true, true);
}
/**
@@ -655,7 +650,7 @@ public final class GridNearLockFuture extends GridCacheCompoundIdentityFuture<Bo
log.debug("Local lock acquired for entries [fut=" + this + ", entries=" + entries + "]");
}
- onComplete(true, true);
+ onComplete(true, true, true);
return true;
}
@@ -666,7 +661,7 @@ public final class GridNearLockFuture extends GridCacheCompoundIdentityFuture<Bo
/** {@inheritDoc} */
@Override public boolean cancel() {
if (onCancelled())
- onComplete(false, true);
+ onComplete(false, true, true);
return isCancelled();
}
@@ -690,7 +685,7 @@ public final class GridNearLockFuture extends GridCacheCompoundIdentityFuture<Bo
if (err != null)
success = false;
- return onComplete(success, true);
+ return onComplete(success, true, true);
}
/**
@@ -698,9 +693,10 @@ public final class GridNearLockFuture extends GridCacheCompoundIdentityFuture<Bo
*
* @param success {@code True} if lock was acquired.
* @param distribute {@code True} if need to distribute lock removal in case of failure.
+ * @param restoreTimeout {@code True} if need restore tx timeout callback.
* @return {@code True} if complete by this operation.
*/
- private boolean onComplete(boolean success, boolean distribute) {
+ private boolean onComplete(boolean success, boolean distribute, boolean restoreTimeout) {
if (log.isDebugEnabled())
log.debug("Received onComplete(..) callback [success=" + success + ", distribute=" + distribute +
", fut=" + this + ']');
@@ -711,6 +707,12 @@ public final class GridNearLockFuture extends GridCacheCompoundIdentityFuture<Bo
if (tx != null)
cctx.tm().txContext(tx);
+ if (restoreTimeout && tx != null && tx.trackTimeout()) {
+ // Need restore timeout before onDone is called, but onComplete can be called concurrently,
+ // thus need ignore duplicated timeout objects.
+ tx.addTimeoutHandler(true);
+ }
+
if (super.onDone(success, err)) {
if (log.isDebugEnabled())
log.debug("Completing future: " + this);
@@ -770,6 +772,30 @@ public final class GridNearLockFuture extends GridCacheCompoundIdentityFuture<Bo
* part. Note that if primary node leaves grid, the future will fail and transaction will be rolled back.
*/
void map() {
+ if (tx != null && tx.trackTimeout()) {
+ if (!tx.removeTimeoutHandler()) {
+ tx.finishFuture().listen(new IgniteInClosure<IgniteInternalFuture<IgniteInternalTx>>() {
+ @Override public void apply(IgniteInternalFuture<IgniteInternalTx> fut) {
+ IgniteTxTimeoutCheckedException err = new IgniteTxTimeoutCheckedException("Failed to " +
+ "acquire lock, transaction was rolled back on timeout [timeout=" + tx.timeout() +
+ ", tx=" + tx + ']');
+
+ onError(err);
+
+ onComplete(false, false, false);
+ }
+ });
+
+ return;
+ }
+ }
+
+ if (timeout > 0) {
+ timeoutObj = new LockTimeoutObject();
+
+ cctx.time().addTimeoutObject(timeoutObj);
+ }
+
// Obtain the topology version to use.
long threadId = Thread.currentThread().getId();
@@ -981,9 +1007,7 @@ public final class GridNearLockFuture extends GridCacheCompoundIdentityFuture<Bo
if (log.isDebugEnabled())
log.debug("Entry being locked did not pass filter (will not lock): " + entry);
- onComplete(
- false,
- false);
+ onComplete(false, false, true);
return;
}
@@ -1476,12 +1500,12 @@ public final class GridNearLockFuture extends GridCacheCompoundIdentityFuture<Bo
U.warn(log, "Failed to detect deadlock.", e);
}
- onComplete(false, true);
+ onComplete(false, true, true);
}
});
}
else
- onComplete(false, true);
+ onComplete(false, true, true);
}
/** {@inheritDoc} */
http://git-wip-us.apache.org/repos/asf/ignite/blob/56dbdc2f/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearOptimisticTxPrepareFuture.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearOptimisticTxPrepareFuture.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearOptimisticTxPrepareFuture.java
index 77c68bd..6d7a862 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearOptimisticTxPrepareFuture.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearOptimisticTxPrepareFuture.java
@@ -913,8 +913,7 @@ public class GridNearOptimisticTxPrepareFuture extends GridNearOptimisticTxPrepa
return;
if (RCV_RES_UPD.compareAndSet(this, 0, 1)) {
- if (parent.cctx.tm().deadlockDetectionEnabled() &&
- (parent.tx.remainingTime() == -1 || res.error() instanceof IgniteTxTimeoutCheckedException)) {
+ if (parent.tx.remainingTime() == -1 || res.error() instanceof IgniteTxTimeoutCheckedException) {
parent.onTimeout();
return;
http://git-wip-us.apache.org/repos/asf/ignite/blob/56dbdc2f/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxFinishFuture.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxFinishFuture.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxFinishFuture.java
index c45eb7b..b4bc977 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxFinishFuture.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxFinishFuture.java
@@ -67,7 +67,7 @@ import static org.apache.ignite.transactions.TransactionState.UNKNOWN;
*
*/
public final class GridNearTxFinishFuture<K, V> extends GridCacheCompoundIdentityFuture<IgniteInternalTx>
- implements GridCacheFuture<IgniteInternalTx> {
+ implements GridCacheFuture<IgniteInternalTx>, NearTxFinishFuture {
/** */
private static final long serialVersionUID = 0L;
@@ -136,6 +136,11 @@ public final class GridNearTxFinishFuture<K, V> extends GridCacheCompoundIdentit
}
/** {@inheritDoc} */
+ @Override public boolean commit() {
+ return commit;
+ }
+
+ /** {@inheritDoc} */
@Override public IgniteUuid futureId() {
return futId;
}
@@ -310,7 +315,7 @@ public final class GridNearTxFinishFuture<K, V> extends GridCacheCompoundIdentit
err = new TransactionRollbackException("Failed to commit transaction.", err);
try {
- tx.localFinish(err == null);
+ tx.localFinish(err == null, true);
}
catch (IgniteCheckedException e) {
if (err != null)
@@ -327,7 +332,7 @@ public final class GridNearTxFinishFuture<K, V> extends GridCacheCompoundIdentit
finishOnePhase(commit);
try {
- tx.tmFinish(commit, nodeStop);
+ tx.tmFinish(commit, nodeStop, true);
}
catch (IgniteCheckedException e) {
U.error(log, "Failed to finish tx: " + tx, e);
@@ -388,7 +393,7 @@ public final class GridNearTxFinishFuture<K, V> extends GridCacheCompoundIdentit
* @param commit Commit flag.
*/
@SuppressWarnings("ForLoopReplaceableByForEach")
- void finish(boolean commit) {
+ public void finish(boolean commit, boolean clearThreadMap) {
if (tx.onNeedCheckBackup()) {
assert tx.onePhaseCommit();
@@ -402,7 +407,7 @@ public final class GridNearTxFinishFuture<K, V> extends GridCacheCompoundIdentit
}
try {
- if (tx.localFinish(commit) || (!commit && tx.state() == UNKNOWN)) {
+ if (tx.localFinish(commit, clearThreadMap) || (!commit && tx.state() == UNKNOWN)) {
if ((tx.onePhaseCommit() && needFinishOnePhase(commit)) || (!tx.onePhaseCommit() && mappings != null)) {
if (mappings.single()) {
GridDistributedTxMapping mapping = mappings.singleMapping();
http://git-wip-us.apache.org/repos/asf/ignite/blob/56dbdc2f/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxLocal.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxLocal.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxLocal.java
index 058a3ff..a08d693 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxLocal.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxLocal.java
@@ -67,6 +67,7 @@ import org.apache.ignite.internal.processors.cache.transactions.IgniteTxKey;
import org.apache.ignite.internal.processors.cache.transactions.TransactionProxy;
import org.apache.ignite.internal.processors.cache.transactions.TransactionProxyImpl;
import org.apache.ignite.internal.processors.cache.version.GridCacheVersion;
+import org.apache.ignite.internal.processors.timeout.GridTimeoutObject;
import org.apache.ignite.internal.transactions.IgniteTxOptimisticCheckedException;
import org.apache.ignite.internal.transactions.IgniteTxRollbackCheckedException;
import org.apache.ignite.internal.transactions.IgniteTxTimeoutCheckedException;
@@ -89,6 +90,7 @@ import org.apache.ignite.internal.util.typedef.internal.S;
import org.apache.ignite.internal.util.typedef.internal.U;
import org.apache.ignite.lang.IgniteBiClosure;
import org.apache.ignite.lang.IgniteClosure;
+import org.apache.ignite.lang.IgniteInClosure;
import org.apache.ignite.lang.IgniteUuid;
import org.apache.ignite.plugin.security.SecurityPermission;
import org.apache.ignite.transactions.TransactionConcurrency;
@@ -107,6 +109,7 @@ import static org.apache.ignite.internal.processors.cache.transactions.IgniteTxE
import static org.apache.ignite.internal.processors.cache.transactions.IgniteTxEntry.SER_READ_NOT_EMPTY_VER;
import static org.apache.ignite.transactions.TransactionState.COMMITTED;
import static org.apache.ignite.transactions.TransactionState.COMMITTING;
+import static org.apache.ignite.transactions.TransactionState.MARKED_ROLLBACK;
import static org.apache.ignite.transactions.TransactionState.PREPARED;
import static org.apache.ignite.transactions.TransactionState.PREPARING;
import static org.apache.ignite.transactions.TransactionState.ROLLED_BACK;
@@ -117,7 +120,7 @@ import static org.apache.ignite.transactions.TransactionState.UNKNOWN;
* Replicated user transaction.
*/
@SuppressWarnings("unchecked")
-public class GridNearTxLocal extends GridDhtTxLocalAdapter implements AutoCloseable {
+public class GridNearTxLocal extends GridDhtTxLocalAdapter implements GridTimeoutObject, AutoCloseable {
/** */
private static final long serialVersionUID = 0L;
@@ -126,12 +129,8 @@ public class GridNearTxLocal extends GridDhtTxLocalAdapter implements AutoClosea
AtomicReferenceFieldUpdater.newUpdater(GridNearTxLocal.class, IgniteInternalFuture.class, "prepFut");
/** Prepare future updater. */
- private static final AtomicReferenceFieldUpdater<GridNearTxLocal, GridNearTxFinishFuture> COMMIT_FUT_UPD =
- AtomicReferenceFieldUpdater.newUpdater(GridNearTxLocal.class, GridNearTxFinishFuture.class, "commitFut");
-
- /** Rollback future updater. */
- private static final AtomicReferenceFieldUpdater<GridNearTxLocal, GridNearTxFinishFuture> ROLLBACK_FUT_UPD =
- AtomicReferenceFieldUpdater.newUpdater(GridNearTxLocal.class, GridNearTxFinishFuture.class, "rollbackFut");
+ private static final AtomicReferenceFieldUpdater<GridNearTxLocal, NearTxFinishFuture> FINISH_FUT_UPD =
+ AtomicReferenceFieldUpdater.newUpdater(GridNearTxLocal.class, NearTxFinishFuture.class, "finishFut");
/** DHT mappings. */
private IgniteTxMappings mappings;
@@ -144,12 +143,7 @@ public class GridNearTxLocal extends GridDhtTxLocalAdapter implements AutoClosea
/** Commit future. */
@SuppressWarnings("UnusedDeclaration")
@GridToStringExclude
- private volatile GridNearTxFinishFuture commitFut;
-
- /** Rollback future. */
- @SuppressWarnings("UnusedDeclaration")
- @GridToStringExclude
- private volatile GridNearTxFinishFuture rollbackFut;
+ private volatile NearTxFinishFuture finishFut;
/** True if transaction contains near cache entries mapped to local node. */
private boolean nearLocallyMapped;
@@ -170,6 +164,9 @@ public class GridNearTxLocal extends GridDhtTxLocalAdapter implements AutoClosea
protected boolean transform;
/** */
+ private boolean trackTimeout;
+
+ /** */
@GridToStringExclude
private TransactionProxyImpl proxy;
@@ -229,6 +226,9 @@ public class GridNearTxLocal extends GridDhtTxLocalAdapter implements AutoClosea
mappings = implicitSingle ? new IgniteTxMappingsSingleImpl() : new IgniteTxMappingsImpl();
initResult();
+
+ if (timeout() > 0 && !implicit())
+ trackTimeout = cctx.time().addTimeoutObject(this, false);
}
/** {@inheritDoc} */
@@ -3044,7 +3044,7 @@ public class GridNearTxLocal extends GridDhtTxLocalAdapter implements AutoClosea
/** {@inheritDoc} */
@SuppressWarnings({"CatchGenericClass", "ThrowableInstanceNeverThrown"})
- @Override public boolean localFinish(boolean commit) throws IgniteCheckedException {
+ @Override public boolean localFinish(boolean commit, boolean clearThreadMap) throws IgniteCheckedException {
if (log.isDebugEnabled())
log.debug("Finishing near local tx [tx=" + this + ", commit=" + commit + "]");
@@ -3080,7 +3080,7 @@ public class GridNearTxLocal extends GridDhtTxLocalAdapter implements AutoClosea
if (commit && !isRollbackOnly())
userCommit();
else
- userRollback();
+ userRollback(clearThreadMap);
}
catch (IgniteCheckedException e) {
err = e;
@@ -3146,6 +3146,9 @@ public class GridNearTxLocal extends GridDhtTxLocalAdapter implements AutoClosea
if (!PREP_FUT_UPD.compareAndSet(this, null, fut))
return prepFut;
+ if (trackTimeout)
+ removeTimeoutHandler();
+
if (timeout == -1) {
fut.onDone(this, timeoutException());
@@ -3178,21 +3181,51 @@ public class GridNearTxLocal extends GridDhtTxLocalAdapter implements AutoClosea
if (awaitLastFuture)
txState().awaitLastFuture(cctx);
- prepareAsync().get();
+ prepareNearTxLocal().get();
}
/**
- * @return Prepare future.
+ * @throws IgniteCheckedException If failed.
*/
- private IgniteInternalFuture<?> prepareAsync() {
- return prepareNearTxLocal();
+ public void commit() throws IgniteCheckedException {
+ commitNearTxLocalAsync().get();
}
/**
- * @throws IgniteCheckedException If failed.
+ * @param fut Already started finish future.
+ * @param commit Commit flag.
+ * @return Finish future.
*/
- public void commit() throws IgniteCheckedException {
- commitNearTxLocalAsync().get();
+ private IgniteInternalFuture<IgniteInternalTx> chainFinishFuture(NearTxFinishFuture fut, final boolean commit) {
+ assert fut != null;
+
+ if (fut.commit() != commit) {
+ final GridNearTxLocal tx = this;
+
+ if (!commit) {
+ if (log.isDebugEnabled())
+ log.debug("Failed to rollback transaction, commit already started: " + tx);
+
+ return fut;
+ }
+
+ final GridFutureAdapter<IgniteInternalTx> fut0 = new GridFutureAdapter<>();
+
+ fut.listen(new IgniteInClosure<IgniteInternalFuture<IgniteInternalTx>>() {
+ @Override public void apply(IgniteInternalFuture<IgniteInternalTx> fut) {
+ if (timedOut())
+ fut0.onDone(new IgniteTxTimeoutCheckedException("Failed to commit transaction, " +
+ "transaction is concurrently rolled back on timeout: " + tx));
+ else
+ fut0.onDone(new IgniteCheckedException("Failed to commit transaction, " +
+ "transaction is concurrently rolled back: " + tx));
+ }
+ });
+
+ return fut0;
+ }
+
+ return fut;
}
/**
@@ -3202,42 +3235,43 @@ public class GridNearTxLocal extends GridDhtTxLocalAdapter implements AutoClosea
if (log.isDebugEnabled())
log.debug("Committing near local tx: " + this);
+ NearTxFinishFuture fut = finishFut;
+
+ if (fut != null)
+ return chainFinishFuture(fut, true);
+
if (fastFinish()) {
- state(PREPARING);
- state(PREPARED);
- state(COMMITTING);
+ GridNearFastFinishFuture fut0;
- cctx.tm().fastFinishTx(this, true);
+ if (!FINISH_FUT_UPD.compareAndSet(this, null, fut0 = new GridNearFastFinishFuture(this, true)))
+ return chainFinishFuture(finishFut, true);
- state(COMMITTED);
+ fut0.finish();
- return new GridFinishedFuture<>((IgniteInternalTx)this);
+ return fut0;
}
- final IgniteInternalFuture<?> prepareFut = prepareNearTxLocal();
+ final GridNearTxFinishFuture fut0;
- GridNearTxFinishFuture fut = commitFut;
+ if (!FINISH_FUT_UPD.compareAndSet(this, null, fut0 = new GridNearTxFinishFuture<>(cctx, this, true)))
+ return chainFinishFuture(finishFut, true);
- if (fut != null ||
- !COMMIT_FUT_UPD.compareAndSet(this, null, fut = new GridNearTxFinishFuture<>(cctx, this, true)))
- return commitFut;
+ cctx.mvcc().addFuture(fut0, fut0.futureId());
- cctx.mvcc().addFuture(fut, fut.futureId());
+ final IgniteInternalFuture<?> prepareFut = prepareNearTxLocal();
prepareFut.listen(new CI1<IgniteInternalFuture<?>>() {
@Override public void apply(IgniteInternalFuture<?> f) {
- GridNearTxFinishFuture fut0 = commitFut;
-
try {
// Make sure that here are no exceptions.
prepareFut.get();
- fut0.finish(true);
+ fut0.finish(true, true);
}
catch (Error | RuntimeException e) {
COMMIT_ERR_UPD.compareAndSet(GridNearTxLocal.this, null, e);
- fut0.finish(false);
+ fut0.finish(false, true);
throw e;
}
@@ -3245,12 +3279,12 @@ public class GridNearTxLocal extends GridDhtTxLocalAdapter implements AutoClosea
COMMIT_ERR_UPD.compareAndSet(GridNearTxLocal.this, null, e);
if (!(e instanceof NodeStoppingException))
- fut0.finish(false);
+ fut0.finish(false, true);
}
}
});
- return fut;
+ return fut0;
}
/** {@inheritDoc} */
@@ -3269,30 +3303,42 @@ public class GridNearTxLocal extends GridDhtTxLocalAdapter implements AutoClosea
* @return Rollback future.
*/
public IgniteInternalFuture<IgniteInternalTx> rollbackNearTxLocalAsync() {
+ return rollbackNearTxLocalAsync(false);
+ }
+
+ /**
+ * @param onTimeout {@code True} if rolled back asynchronously on timeout.
+ * @return Rollback future.
+ */
+ private IgniteInternalFuture<IgniteInternalTx> rollbackNearTxLocalAsync(final boolean onTimeout) {
if (log.isDebugEnabled())
log.debug("Rolling back near tx: " + this);
+ if (!onTimeout && trackTimeout)
+ removeTimeoutHandler();
+
+ NearTxFinishFuture fut = finishFut;
+
+ if (fut != null)
+ return chainFinishFuture(finishFut, false);
+
if (fastFinish()) {
- state(PREPARING);
- state(PREPARED);
- state(ROLLING_BACK);
+ GridNearFastFinishFuture fut0;
- cctx.tm().fastFinishTx(this, false);
+ if (!FINISH_FUT_UPD.compareAndSet(this, null, fut0 = new GridNearFastFinishFuture(this, false)))
+ return chainFinishFuture(finishFut, false);
- state(ROLLED_BACK);
+ fut0.finish();
- return new GridFinishedFuture<>((IgniteInternalTx)this);
+ return fut0;
}
- GridNearTxFinishFuture fut = rollbackFut;
-
- if (fut != null)
- return fut;
+ final GridNearTxFinishFuture fut0;
- if (!ROLLBACK_FUT_UPD.compareAndSet(this, null, fut = new GridNearTxFinishFuture<>(cctx, this, false)))
- return rollbackFut;
+ if (!FINISH_FUT_UPD.compareAndSet(this, null, fut0 = new GridNearTxFinishFuture<>(cctx, this, false)))
+ return chainFinishFuture(finishFut, false);
- cctx.mvcc().addFuture(fut, fut.futureId());
+ cctx.mvcc().addFuture(fut0, fut0.futureId());
IgniteInternalFuture<?> prepFut = this.prepFut;
@@ -3307,7 +3353,7 @@ public class GridNearTxLocal extends GridDhtTxLocalAdapter implements AutoClosea
log.debug("Got optimistic tx failure [tx=" + this + ", err=" + e + ']');
}
- fut.finish(false);
+ fut0.finish(false, !onTimeout);
}
else {
prepFut.listen(new CI1<IgniteInternalFuture<?>>() {
@@ -3321,14 +3367,12 @@ public class GridNearTxLocal extends GridDhtTxLocalAdapter implements AutoClosea
log.debug("Got optimistic tx failure [tx=" + this + ", err=" + e + ']');
}
- GridNearTxFinishFuture fut0 = rollbackFut;
-
- fut0.finish(false);
+ fut0.finish(false, !onTimeout);
}
});
}
- return fut;
+ return fut0;
}
/** {@inheritDoc} */
@@ -3693,38 +3737,52 @@ public class GridNearTxLocal extends GridDhtTxLocalAdapter implements AutoClosea
@Override public void close() throws IgniteCheckedException {
TransactionState state = state();
- if (state != ROLLING_BACK && state != ROLLED_BACK && state != COMMITTING && state != COMMITTED)
- rollback();
+ try {
+ if (state == COMMITTED || state == ROLLED_BACK)
+ return;
- synchronized (this) {
- try {
- while (!done())
- wait();
- }
- catch (InterruptedException e) {
- Thread.currentThread().interrupt();
+ if (trackTimeout)
+ removeTimeoutHandler();
- if (!done())
- throw new IgniteCheckedException("Got interrupted while waiting for transaction to complete: " +
- this, e);
+ if (state != ROLLING_BACK && state != COMMITTING)
+ rollback();
+
+ synchronized (this) {
+ try {
+ while (!done())
+ wait();
+ }
+ catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+
+ if (!done())
+ throw new IgniteCheckedException("Got interrupted while waiting for transaction to complete: " +
+ this, e);
+ }
}
}
+ finally {
+ boolean cleanup = state == ROLLED_BACK && timedOut();
- if (accessMap != null) {
- assert optimistic();
+ if (cleanup)
+ cctx.tm().clearThreadMap(this);
- for (Map.Entry<IgniteTxKey, IgniteCacheExpiryPolicy> e : accessMap.entrySet()) {
- if (e.getValue().entries() != null) {
- GridCacheContext cctx0 = cctx.cacheContext(e.getKey().cacheId());
+ if (accessMap != null) {
+ assert optimistic();
- if (cctx0.isNear())
- cctx0.near().dht().sendTtlUpdateRequest(e.getValue());
- else
- cctx0.dht().sendTtlUpdateRequest(e.getValue());
+ for (Map.Entry<IgniteTxKey, IgniteCacheExpiryPolicy> e : accessMap.entrySet()) {
+ if (e.getValue().entries() != null) {
+ GridCacheContext cctx0 = cctx.cacheContext(e.getKey().cacheId());
+
+ if (cctx0.isNear())
+ cctx0.near().dht().sendTtlUpdateRequest(e.getValue());
+ else
+ cctx0.dht().sendTtlUpdateRequest(e.getValue());
+ }
}
- }
- accessMap = null;
+ accessMap = null;
+ }
}
}
@@ -4000,13 +4058,67 @@ public class GridNearTxLocal extends GridDhtTxLocalAdapter implements AutoClosea
/**
* @param threadId new owner of transaction.
- * @throws IgniteCheckedException if method executed not in the middle of resume or suspend.
*/
public void threadId(long threadId) {
this.threadId = threadId;
}
/**
+ * @return {@code True} if need register callback which cancels tx on timeout.
+ */
+ public boolean trackTimeout() {
+ return trackTimeout;
+ }
+
+ /**
+ * Removes timeout handler.
+ *
+ * @return {@code True} if handler was removed.
+ */
+ public boolean removeTimeoutHandler() {
+ assert trackTimeout;
+
+ return cctx.time().removeTimeoutObject(this);
+ }
+
+ /**
+ * @param ignoreDuplicated {@code True} if do need check for duplicated timeout object.
+ */
+ public void addTimeoutHandler(boolean ignoreDuplicated) {
+ assert trackTimeout;
+
+ cctx.time().addTimeoutObject(this, ignoreDuplicated);
+ }
+
+ /** {@inheritDoc} */
+ @Override public IgniteUuid timeoutId() {
+ return xid();
+ }
+
+ /** {@inheritDoc} */
+ @Override public long endTime() {
+ return startTime() + timeout();
+ }
+
+ /** {@inheritDoc} */
+ @Override public void onTimeout() {
+ if (state(MARKED_ROLLBACK, true) || (state() == MARKED_ROLLBACK)) {
+ if (log.isDebugEnabled())
+ log.debug("Will rollback tx on timeout: " + this);
+
+ cctx.kernalContext().closure().runLocalSafe(new Runnable() {
+ @Override public void run() {
+ rollbackNearTxLocalAsync(true);
+ }
+ });
+ }
+ else {
+ if (log.isDebugEnabled())
+ log.debug("Skip rollback tx on timeout: " + this);
+ }
+ }
+
+ /**
* Post-lock closure.
*
* @param <T> Return type.
http://git-wip-us.apache.org/repos/asf/ignite/blob/56dbdc2f/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/NearTxFinishFuture.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/NearTxFinishFuture.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/NearTxFinishFuture.java
new file mode 100644
index 0000000..132c754
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/NearTxFinishFuture.java
@@ -0,0 +1,31 @@
+/*
+ * 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.internal.IgniteInternalFuture;
+import org.apache.ignite.internal.processors.cache.transactions.IgniteInternalTx;
+
+/**
+ *
+ */
+public interface NearTxFinishFuture extends IgniteInternalFuture<IgniteInternalTx> {
+ /**
+ * @return Commit flag.
+ */
+ boolean commit();
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/56dbdc2f/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 7598003..9e06d9d 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
@@ -634,4 +634,4 @@ public interface IgniteInternalTx {
* @param e Commit error.
*/
public void commitError(Throwable e);
-}
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/ignite/blob/56dbdc2f/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 c447436..b5178b5 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
@@ -370,6 +370,13 @@ public abstract class IgniteTxAdapter extends GridMetadataAwareAdapter implement
consistentIdMapper = new ConsistentIdMapper(cctx.discovery());
}
+ /**
+ * @return Shared cache context.
+ */
+ public GridCacheSharedContext<?, ?> context() {
+ return cctx;
+ }
+
/** {@inheritDoc} */
@Override public boolean localResult() {
assert originatingNodeId() != null;
@@ -987,7 +994,7 @@ public abstract class IgniteTxAdapter extends GridMetadataAwareAdapter implement
* @return {@code True} if state changed.
*/
@SuppressWarnings({"TooBroadScope"})
- protected boolean state(TransactionState state, boolean timedOut) {
+ protected final boolean state(TransactionState state, boolean timedOut) {
boolean valid = false;
TransactionState prev;
@@ -1068,7 +1075,9 @@ public abstract class IgniteTxAdapter extends GridMetadataAwareAdapter implement
if (valid) {
this.state = state;
- this.timedOut = timedOut;
+
+ if (timedOut)
+ this.timedOut = true;
if (log.isDebugEnabled())
log.debug("Changed transaction state [prev=" + prev + ", new=" + this.state + ", tx=" + this + ']');
http://git-wip-us.apache.org/repos/asf/ignite/blob/56dbdc2f/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 7ab921c..9fa0ff7 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
@@ -901,7 +901,7 @@ public abstract class IgniteTxLocalAdapter extends IgniteTxAdapter implements Ig
* @param nodeStop If {@code true} tx is cancelled on node stop.
* @throws IgniteCheckedException If failed.
*/
- public void tmFinish(boolean commit, boolean nodeStop) throws IgniteCheckedException {
+ public void tmFinish(boolean commit, boolean nodeStop, boolean clearThreadMap) throws IgniteCheckedException {
assert onePhaseCommit();
if (DONE_FLAG_UPD.compareAndSet(this, 0, 1)) {
@@ -910,7 +910,7 @@ public abstract class IgniteTxLocalAdapter extends IgniteTxAdapter implements Ig
if (commit)
cctx.tm().commitTx(this);
else
- cctx.tm().rollbackTx(this);
+ cctx.tm().rollbackTx(this, clearThreadMap);
}
state(commit ? COMMITTED : ROLLED_BACK);
@@ -957,7 +957,7 @@ public abstract class IgniteTxLocalAdapter extends IgniteTxAdapter implements Ig
}
/** {@inheritDoc} */
- @Override public void userRollback() throws IgniteCheckedException {
+ @Override public void userRollback(boolean clearThreadMap) throws IgniteCheckedException {
TransactionState state = state();
if (state != ROLLING_BACK && state != ROLLED_BACK) {
@@ -975,7 +975,7 @@ public abstract class IgniteTxLocalAdapter extends IgniteTxAdapter implements Ig
}
if (DONE_FLAG_UPD.compareAndSet(this, 0, 1)) {
- cctx.tm().rollbackTx(this);
+ cctx.tm().rollbackTx(this, clearThreadMap);
if (!internal()) {
Collection<CacheStoreManager> stores = txState.stores(cctx);
http://git-wip-us.apache.org/repos/asf/ignite/blob/56dbdc2f/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 307c348..e0aa7a5 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
@@ -43,7 +43,7 @@ public interface IgniteTxLocalEx extends IgniteInternalTx {
/**
* @throws IgniteCheckedException If rollback failed.
*/
- public void userRollback() throws IgniteCheckedException;
+ public void userRollback(boolean clearThreadMap) throws IgniteCheckedException;
/**
* Finishes transaction (either commit or rollback).
@@ -52,5 +52,5 @@ public interface IgniteTxLocalEx extends IgniteInternalTx {
* @return {@code True} if state has been changed.
* @throws IgniteCheckedException If finish failed.
*/
- public boolean localFinish(boolean commit) throws IgniteCheckedException;
+ public boolean localFinish(boolean commit, boolean clearThreadMap) throws IgniteCheckedException;
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/56dbdc2f/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 9a8280f..208bd61 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
@@ -291,7 +291,7 @@ public class IgniteTxManager extends GridCacheSharedManagerAdapter {
public void rollbackTransactionsForCache(int cacheId) {
rollbackTransactionsForCache(cacheId, nearIdMap);
- rollbackTransactionsForCache(cacheId, threadMap);
+ rollbackTransactionsForCache(cacheId, idMap);
}
/**
@@ -304,7 +304,7 @@ public class IgniteTxManager extends GridCacheSharedManagerAdapter {
for (IgniteTxEntry entry : tx.allEntries()) {
if (entry.cacheId() == cacheId) {
- rollbackTx(tx);
+ rollbackTx(tx, false);
break;
}
@@ -316,8 +316,10 @@ public class IgniteTxManager extends GridCacheSharedManagerAdapter {
@Override public void onDisconnected(IgniteFuture reconnectFut) {
txFinishSync.onDisconnected(reconnectFut);
- for (Map.Entry<Long, IgniteInternalTx> e : threadMap.entrySet())
- rollbackTx(e.getValue());
+ for (IgniteInternalTx tx : idMap.values())
+ rollbackTx(tx, true);
+ for (IgniteInternalTx tx : nearIdMap.values())
+ rollbackTx(tx, true);
IgniteClientDisconnectedException err =
new IgniteClientDisconnectedException(reconnectFut, "Client node disconnected.");
@@ -378,6 +380,7 @@ public class IgniteTxManager extends GridCacheSharedManagerAdapter {
X.println(">>> Transaction manager memory stats [igniteInstanceName=" + cctx.igniteInstanceName() + ']');
X.println(">>> threadMapSize: " + threadMap.size());
X.println(">>> idMap [size=" + idMap.size() + ']');
+ X.println(">>> nearIdMap [size=" + nearIdMap.size() + ']');
X.println(">>> completedVersSortedSize: " + completedVersSorted.size());
X.println(">>> completedVersHashMapSize: " + completedVersHashMap.sizex());
}
@@ -490,14 +493,15 @@ public class IgniteTxManager extends GridCacheSharedManagerAdapter {
IgniteInternalTx t;
if ((t = txIdMap.putIfAbsent(tx.xidVersion(), tx)) == null) {
- // Add both, explicit and implicit transactions.
- // Do not add remote and dht local transactions as remote node may have the same thread ID
- // and overwrite local transaction.
if (tx.local() && !tx.dht()) {
- if (cacheCtx == null || !cacheCtx.systemTx())
- threadMap.put(tx.threadId(), tx);
- else
- sysThreadMap.put(new TxThreadKey(tx.threadId(), cacheCtx.cacheId()), tx);
+ assert tx instanceof GridNearTxLocal : tx;
+
+ if (!tx.implicit()) {
+ if (cacheCtx == null || !cacheCtx.systemTx())
+ threadMap.put(tx.threadId(), tx);
+ else
+ sysThreadMap.put(new TxThreadKey(tx.threadId(), cacheCtx.cacheId()), tx);
+ }
}
// Handle mapped versions.
@@ -633,11 +637,10 @@ public class IgniteTxManager extends GridCacheSharedManagerAdapter {
/**
* @return Local transaction.
*/
- @SuppressWarnings({"unchecked"})
- @Nullable public <T> T localTx() {
- IgniteInternalTx tx = tx();
+ @Nullable public IgniteTxLocalAdapter localTx() {
+ IgniteTxLocalAdapter tx = tx();
- return tx != null && tx.local() ? (T)tx : null;
+ return tx != null && tx.local() ? tx : null;
}
/**
@@ -719,15 +722,6 @@ public class IgniteTxManager extends GridCacheSharedManagerAdapter {
}
/**
- * @return Local transaction.
- */
- @Nullable public IgniteInternalTx localTxx() {
- IgniteInternalTx tx = tx();
-
- return tx != null && tx.local() ? tx : null;
- }
-
- /**
* @return User transaction for current thread.
*/
@Nullable public GridNearTxLocal userTx() {
@@ -1215,32 +1209,32 @@ public class IgniteTxManager extends GridCacheSharedManagerAdapter {
collectPendingVersions(dhtTxLoc);
}
- // 4. Unlock write resources.
+ // 3. Unlock write resources.
unlockMultiple(tx, tx.writeEntries());
- // 5. Unlock read resources if required.
+ // 4. Unlock read resources if required.
if (unlockReadEntries(tx))
unlockMultiple(tx, tx.readEntries());
- // 6. Notify evictions.
+ // 5. Notify evictions.
notifyEvictions(tx);
- // 7. Remove obsolete entries from cache.
+ // 6. Remove obsolete entries from cache.
removeObsolete(tx);
- // 8. Assign transaction number at the end of transaction.
+ // 7. Assign transaction number at the end of transaction.
tx.endVersion(cctx.versions().next(tx.topologyVersion()));
- // 9. Remove from per-thread storage.
+ // 8. Remove from per-thread storage.
clearThreadMap(tx);
- // 10. Unregister explicit locks.
+ // 9. Unregister explicit locks.
if (!tx.alternateVersions().isEmpty()) {
for (GridCacheVersion ver : tx.alternateVersions())
idMap.remove(ver);
}
- // 11. Remove Near-2-DHT mappings.
+ // 10. Remove Near-2-DHT mappings.
if (tx instanceof GridCacheMappedVersion) {
GridCacheVersion mapped = ((GridCacheMappedVersion)tx).mappedVersion();
@@ -1248,10 +1242,10 @@ public class IgniteTxManager extends GridCacheSharedManagerAdapter {
mappedVers.remove(mapped);
}
- // 12. Clear context.
+ // 11. Clear context.
resetContext();
- // 14. Update metrics.
+ // 12. Update metrics.
if (!tx.dht() && tx.local()) {
if (!tx.system())
cctx.txMetrics().onTxCommit();
@@ -1276,7 +1270,7 @@ public class IgniteTxManager extends GridCacheSharedManagerAdapter {
*
* @param tx Transaction to rollback.
*/
- public void rollbackTx(IgniteInternalTx tx) {
+ public void rollbackTx(IgniteInternalTx tx, boolean clearThreadMap) {
assert tx != null;
if (log.isDebugEnabled())
@@ -1302,7 +1296,8 @@ public class IgniteTxManager extends GridCacheSharedManagerAdapter {
removeObsolete(tx);
// 6. Remove from per-thread storage.
- clearThreadMap(tx);
+ if (clearThreadMap)
+ clearThreadMap(tx);
// 7. Unregister explicit locks.
if (!tx.alternateVersions().isEmpty())
@@ -1427,8 +1422,10 @@ public class IgniteTxManager extends GridCacheSharedManagerAdapter {
/**
* @param tx Transaction to clear.
*/
- private void clearThreadMap(IgniteInternalTx tx) {
+ public void clearThreadMap(IgniteInternalTx tx) {
if (tx.local() && !tx.dht()) {
+ assert tx instanceof GridNearTxLocal : tx;
+
if (!tx.system())
threadMap.remove(tx.threadId(), tx);
else {
@@ -2257,6 +2254,7 @@ public class IgniteTxManager extends GridCacheSharedManagerAdapter {
* @see #resumeTx(GridNearTxLocal)
* @see GridNearTxLocal#suspend()
* @see GridNearTxLocal#resume()
+ * @throws IgniteCheckedException If failed to suspend transaction.
*/
public void suspendTx(final GridNearTxLocal tx) throws IgniteCheckedException {
assert tx != null && !tx.system() : tx;
@@ -2280,6 +2278,7 @@ public class IgniteTxManager extends GridCacheSharedManagerAdapter {
* @see #suspendTx(GridNearTxLocal)
* @see GridNearTxLocal#suspend()
* @see GridNearTxLocal#resume()
+ * @throws IgniteCheckedException If failed to resume tx.
*/
public void resumeTx(GridNearTxLocal tx) throws IgniteCheckedException {
assert tx != null && !tx.system() : tx;
@@ -2287,7 +2286,7 @@ public class IgniteTxManager extends GridCacheSharedManagerAdapter {
assert !transactionMap(tx).containsValue(tx) : tx;
assert !haveSystemTxForThread(Thread.currentThread().getId());
- if(!tx.state(ACTIVE)) {
+ if (!tx.state(ACTIVE)) {
throw new IgniteCheckedException("Trying to resume transaction with incorrect state "
+ "[expected=" + SUSPENDED + ", actual=" + tx.state() + ']');
}
@@ -2295,10 +2294,10 @@ public class IgniteTxManager extends GridCacheSharedManagerAdapter {
long threadId = Thread.currentThread().getId();
if (threadMap.putIfAbsent(threadId, tx) != null)
- throw new IgniteCheckedException("Thread already start a transaction.");
+ throw new IgniteCheckedException("Thread already has started a transaction.");
if (transactionMap(tx).putIfAbsent(tx.xidVersion(), tx) != null)
- throw new IgniteCheckedException("Thread already start a transaction.");
+ throw new IgniteCheckedException("Thread already has started a transaction.");
tx.threadId(threadId);
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/56dbdc2f/modules/core/src/main/java/org/apache/ignite/internal/processors/timeout/GridTimeoutProcessor.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/timeout/GridTimeoutProcessor.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/timeout/GridTimeoutProcessor.java
index 8c71f76..9317391 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/timeout/GridTimeoutProcessor.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/timeout/GridTimeoutProcessor.java
@@ -85,21 +85,31 @@ public class GridTimeoutProcessor extends GridProcessorAdapter {
/**
* @param timeoutObj Timeout object.
*/
- @SuppressWarnings({"NakedNotify", "CallToNotifyInsteadOfNotifyAll"})
public void addTimeoutObject(GridTimeoutObject timeoutObj) {
+ addTimeoutObject(timeoutObj, false);
+ }
+
+ /**
+ * @param timeoutObj Timeout object.
+ * @param ignoreDuplicated {@code True} if do need check for duplicated object.
+ */
+ @SuppressWarnings({"NakedNotify", "CallToNotifyInsteadOfNotifyAll"})
+ public boolean addTimeoutObject(GridTimeoutObject timeoutObj, boolean ignoreDuplicated) {
if (timeoutObj.endTime() <= 0 || timeoutObj.endTime() == Long.MAX_VALUE)
// Timeout will never happen.
- return;
+ return false;
boolean added = timeoutObjs.add(timeoutObj);
- assert added : "Duplicate timeout object found: " + timeoutObj;
+ assert ignoreDuplicated || added : "Duplicate timeout object found: " + timeoutObj;
if (timeoutObjs.firstx() == timeoutObj) {
synchronized (mux) {
mux.notify(); // No need to notifyAll since we only have one thread.
}
}
+
+ return true;
}
/**
@@ -124,9 +134,10 @@ public class GridTimeoutProcessor extends GridProcessorAdapter {
/**
* @param timeoutObj Timeout object.
+ * @return {@code True} if timeout object was removed.
*/
- public void removeTimeoutObject(GridTimeoutObject timeoutObj) {
- timeoutObjs.remove(timeoutObj);
+ public boolean removeTimeoutObject(GridTimeoutObject timeoutObj) {
+ return timeoutObjs.remove(timeoutObj);
}
/**
http://git-wip-us.apache.org/repos/asf/ignite/blob/56dbdc2f/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteTxConfigCacheSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteTxConfigCacheSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteTxConfigCacheSelfTest.java
index 2efa0cb..f2e17e4 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteTxConfigCacheSelfTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteTxConfigCacheSelfTest.java
@@ -185,6 +185,8 @@ public class IgniteTxConfigCacheSelfTest extends GridCommonAbstractTest {
try (final Transaction tx = ignite.transactions().txStart()) {
assert tx != null;
+ cache.put("key0", "val0");
+
sleepForTxFailure();
cache.put("key", "val");
@@ -195,7 +197,19 @@ public class IgniteTxConfigCacheSelfTest extends GridCommonAbstractTest {
assert e.getCause() instanceof TransactionTimeoutException;
}
+ assertNull(ignite.transactions().tx());
+
+ assert !cache.containsKey("key0");
assert !cache.containsKey("key");
+
+ // New transaction must succeed.
+ try (final Transaction tx = ignite.transactions().txStart()) {
+ cache.put("key", "val");
+
+ tx.commit();
+ }
+
+ assert cache.containsKey("key");
}
/**
http://git-wip-us.apache.org/repos/asf/ignite/blob/56dbdc2f/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/IgniteCacheThreadLocalTxTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/IgniteCacheThreadLocalTxTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/IgniteCacheThreadLocalTxTest.java
new file mode 100644
index 0000000..c8eac20
--- /dev/null
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/IgniteCacheThreadLocalTxTest.java
@@ -0,0 +1,223 @@
+/*
+ * 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;
+
+import java.util.concurrent.ThreadLocalRandom;
+import org.apache.ignite.Ignite;
+import org.apache.ignite.IgniteCache;
+import org.apache.ignite.IgniteTransactions;
+import org.apache.ignite.configuration.CacheConfiguration;
+import org.apache.ignite.configuration.IgniteConfiguration;
+import org.apache.ignite.internal.IgniteKernal;
+import org.apache.ignite.internal.util.typedef.G;
+import org.apache.ignite.lang.IgniteFuture;
+import org.apache.ignite.spi.discovery.tcp.TcpDiscoverySpi;
+import org.apache.ignite.spi.discovery.tcp.ipfinder.TcpDiscoveryIpFinder;
+import org.apache.ignite.spi.discovery.tcp.ipfinder.vm.TcpDiscoveryVmIpFinder;
+import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
+import org.apache.ignite.transactions.Transaction;
+import org.apache.ignite.transactions.TransactionConcurrency;
+import org.apache.ignite.transactions.TransactionIsolation;
+
+import static org.apache.ignite.cache.CacheAtomicityMode.TRANSACTIONAL;
+
+/**
+ *
+ */
+public class IgniteCacheThreadLocalTxTest extends GridCommonAbstractTest {
+ /** */
+ private static final TcpDiscoveryIpFinder ipFinder = new TcpDiscoveryVmIpFinder(true);
+
+ /** */
+ private boolean client;
+
+ /** {@inheritDoc} */
+ @Override protected IgniteConfiguration getConfiguration(String igniteInstanceName) throws Exception {
+ IgniteConfiguration cfg = super.getConfiguration(igniteInstanceName);
+
+ ((TcpDiscoverySpi)cfg.getDiscoverySpi()).setIpFinder(ipFinder);
+
+ cfg.setClientMode(client);
+
+ return cfg;
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void afterTest() throws Exception {
+ stopAllGrids();
+
+ super.afterTest();
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testSingleNode() throws Exception {
+ threadLocalTx(startGrid(0));
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testMultiNode() throws Exception {
+ startGridsMultiThreaded(4);
+
+ client = true;
+
+ startGrid(4);
+
+ for (Ignite node : G.allGrids())
+ threadLocalTx(node);
+ }
+
+ /**
+ * @param node Node.
+ * @throws Exception If failed.
+ */
+ @SuppressWarnings("unchecked")
+ private void threadLocalTx(Ignite node) throws Exception {
+ CacheConfiguration ccfg = new CacheConfiguration(DEFAULT_CACHE_NAME);
+
+ ccfg.setAtomicityMode(TRANSACTIONAL);
+ ccfg.setBackups(2);
+
+ IgniteCache<Object, Object> cache = node.getOrCreateCache(ccfg);
+
+ checkNoTx(node);
+
+ boolean[] reads = {true, false};
+ boolean[] writes = {true, false};
+ int endOps = 5;
+
+ for (TransactionConcurrency concurrency : TransactionConcurrency.values()) {
+ for (TransactionIsolation isolation : TransactionIsolation.values()) {
+ for (boolean read : reads) {
+ for (boolean write : writes) {
+ for (int i = 0; i < endOps; i++)
+ checkTx(concurrency, isolation, node, cache, read, write, i);
+ }
+ }
+ }
+ }
+
+ checkNoTx(node);
+
+ cache.put(1, 1);
+
+ checkNoTx(node);
+ }
+
+ /**
+ * @param concurrency Tx concurrency.
+ * @param isolation Tx isolation.
+ * @param node Node.
+ * @param cache Cache.
+ * @param read {@code True} if read in tx.
+ * @param write {@code True} if write in tx.
+ * @param endOp Operation to test.
+ */
+ private void checkTx(TransactionConcurrency concurrency,
+ TransactionIsolation isolation,
+ Ignite node,
+ IgniteCache<Object, Object> cache,
+ boolean read,
+ boolean write,
+ int endOp) {
+ IgniteTransactions txs = node.transactions();
+
+ checkNoTx(node);
+
+ Transaction tx = txs.txStart(concurrency, isolation);
+
+ assertEquals(tx, txs.tx());
+
+ try {
+ txs.txStart(concurrency, isolation);
+
+ fail();
+ }
+ catch (IllegalStateException expected) {
+ // No-op.
+ }
+
+ if (read)
+ cache.get(ThreadLocalRandom.current().nextInt(100_000));
+
+ if (write)
+ cache.put(ThreadLocalRandom.current().nextInt(100_000), 1);
+
+
+ try {
+ txs.txStart(concurrency, isolation);
+
+ fail();
+ }
+ catch (IllegalStateException expected) {
+ // No-op.
+ }
+
+ assertEquals(tx, txs.tx());
+
+ IgniteFuture fut = null;
+
+ switch (endOp) {
+ case 0:
+ tx.commit();
+
+ break;
+
+ case 1:
+ fut = tx.commitAsync();
+
+ break;
+
+ case 2:
+ tx.rollback();
+
+ break;
+
+ case 3:
+ fut = tx.rollbackAsync();
+
+ break;
+
+ case 4:
+ tx.close();
+
+ break;
+
+ default:
+ fail();
+ }
+
+ if (fut != null)
+ fut.get();
+
+ checkNoTx(node);
+ }
+
+ /**
+ * @param node Node.
+ */
+ private void checkNoTx(Ignite node) {
+ IgniteTransactions txs = node.transactions();
+
+ assertNull(txs.tx());
+ assertNull(((IgniteKernal)node).context().cache().context().tm().tx());
+ }
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/56dbdc2f/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/IgniteOptimisticTxSuspendResumeTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/IgniteOptimisticTxSuspendResumeTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/IgniteOptimisticTxSuspendResumeTest.java
index 37003a7..86c0fa4 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/IgniteOptimisticTxSuspendResumeTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/IgniteOptimisticTxSuspendResumeTest.java
@@ -46,7 +46,6 @@ import static org.apache.ignite.cache.CacheWriteSynchronizationMode.FULL_SYNC;
import static org.apache.ignite.transactions.TransactionConcurrency.OPTIMISTIC;
import static org.apache.ignite.transactions.TransactionState.ACTIVE;
import static org.apache.ignite.transactions.TransactionState.COMMITTED;
-import static org.apache.ignite.transactions.TransactionState.MARKED_ROLLBACK;
import static org.apache.ignite.transactions.TransactionState.ROLLED_BACK;
import static org.apache.ignite.transactions.TransactionState.SUSPENDED;
@@ -60,6 +59,7 @@ public class IgniteOptimisticTxSuspendResumeTest extends GridCommonAbstractTest
/** Future timeout */
private static final int FUT_TIMEOUT = 5000;
+ /** */
private boolean client = false;
/**
@@ -442,7 +442,7 @@ public class IgniteOptimisticTxSuspendResumeTest extends GridCommonAbstractTest
}
}, TransactionTimeoutException.class);
- assertEquals(MARKED_ROLLBACK, tx.state());
+ assertEquals(ROLLED_BACK, tx.state());
tx.close();
}
@@ -476,7 +476,7 @@ public class IgniteOptimisticTxSuspendResumeTest extends GridCommonAbstractTest
}
}, TransactionTimeoutException.class);
- assertEquals(MARKED_ROLLBACK, tx.state());
+ assertEquals(ROLLED_BACK, tx.state());
tx.close();
http://git-wip-us.apache.org/repos/asf/ignite/blob/56dbdc2f/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/near/CachePartitionedMultiNodeLongTxTimeout2FullApiTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/near/CachePartitionedMultiNodeLongTxTimeout2FullApiTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/near/CachePartitionedMultiNodeLongTxTimeout2FullApiTest.java
new file mode 100644
index 0000000..aef63d0
--- /dev/null
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/near/CachePartitionedMultiNodeLongTxTimeout2FullApiTest.java
@@ -0,0 +1,34 @@
+/*
+ * 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.configuration.IgniteConfiguration;
+
+/**
+ *
+ */
+public class CachePartitionedMultiNodeLongTxTimeout2FullApiTest extends GridCachePartitionedMultiNodeFullApiSelfTest {
+ /** {@inheritDoc} */
+ @Override protected IgniteConfiguration getConfiguration(String igniteInstanceName) throws Exception {
+ IgniteConfiguration cfg = super.getConfiguration(igniteInstanceName);
+
+ cfg.getTransactionConfiguration().setDefaultTxTimeout(Long.MAX_VALUE / 2);
+
+ return cfg;
+ }
+}
[6/7] ignite git commit: Merge remote-tracking branch
'remotes/origin/master' into ignite-6181-2
Posted by sb...@apache.org.
Merge remote-tracking branch 'remotes/origin/master' into ignite-6181-2
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/274b87fb
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/274b87fb
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/274b87fb
Branch: refs/heads/ignite-6181-2
Commit: 274b87fbf0326d9cdd638c3a4f228df1f42ebccd
Parents: 56dbdc2 06d2978
Author: sboikov <sb...@gridgain.com>
Authored: Wed Sep 20 14:57:20 2017 +0300
Committer: sboikov <sb...@gridgain.com>
Committed: Wed Sep 20 14:57:20 2017 +0300
----------------------------------------------------------------------
.../visor/node/VisorNodeDataCollectorTask.java | 2 +-
.../node/VisorNodeDataCollectorTaskResult.java | 4 +-
modules/platforms/dotnet/build.ps1 | 122 +++++++++++++------
.../commands/alert/VisorAlertCommand.scala | 29 ++---
.../backend/routes/demo/domains.json | 5 +
.../generator/JavaTransformer.service.js | 5 +-
.../ignite/console/demo/AgentClusterDemo.java | 9 +-
7 files changed, 117 insertions(+), 59 deletions(-)
----------------------------------------------------------------------
[7/7] ignite git commit: Merge remote-tracking branch
'origin/ignite-6181-2' into ignite-6181-2
Posted by sb...@apache.org.
Merge remote-tracking branch 'origin/ignite-6181-2' into ignite-6181-2
# Conflicts:
# modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheAdapter.java
# modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxLocal.java
# modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxManager.java
# modules/core/src/main/java/org/apache/ignite/internal/processors/timeout/GridTimeoutProcessor.java
# modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/IgniteCacheThreadLocalTxTest.java
# modules/core/src/test/java/org/apache/ignite/internal/processors/cache/transactions/TxRollbackOnTimeoutTest.java
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/dcb7511a
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/dcb7511a
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/dcb7511a
Branch: refs/heads/ignite-6181-2
Commit: dcb7511a689aa3b780b5f267f7f416a538f1a432
Parents: 274b87f 685f051
Author: sboikov <sb...@gridgain.com>
Authored: Wed Sep 20 15:03:52 2017 +0300
Committer: sboikov <sb...@gridgain.com>
Committed: Wed Sep 20 15:03:52 2017 +0300
----------------------------------------------------------------------
.../main/java/org/apache/ignite/cache/CacheMetrics.java | 2 ++
.../internal/processors/cache/CacheMetricsImpl.java | 4 ++--
.../processors/cache/GridCacheSharedContext.java | 11 ++++++-----
.../distributed/dht/colocated/GridDhtColocatedCache.java | 4 ++--
.../dht/colocated/GridDhtColocatedLockFuture.java | 2 +-
.../cache/distributed/near/GridNearLockFuture.java | 2 +-
.../near/GridNearOptimisticTxPrepareFutureAdapter.java | 2 +-
.../distributed/near/GridNearTransactionalCache.java | 2 +-
.../processors/cache/transactions/IgniteTxAdapter.java | 2 +-
.../processors/cache/transactions/IgniteTxHandler.java | 4 ++--
.../IgniteCacheSystemTransactionsSelfTest.java | 4 ++--
11 files changed, 21 insertions(+), 18 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ignite/blob/dcb7511a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxAdapter.java
----------------------------------------------------------------------
[4/7] ignite git commit: ignite-6181
Posted by sb...@apache.org.
http://git-wip-us.apache.org/repos/asf/ignite/blob/56dbdc2f/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/transactions/TxRollbackOnTimeoutNearCacheTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/transactions/TxRollbackOnTimeoutNearCacheTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/transactions/TxRollbackOnTimeoutNearCacheTest.java
new file mode 100644
index 0000000..7c1a6dd
--- /dev/null
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/transactions/TxRollbackOnTimeoutNearCacheTest.java
@@ -0,0 +1,28 @@
+/*
+ * 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.transactions;
+
+/**
+ *
+ */
+public class TxRollbackOnTimeoutNearCacheTest extends TxRollbackOnTimeoutTest {
+ /** {@inheritDoc} */
+ @Override protected boolean nearCacheEnabled() {
+ return true;
+ }
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/56dbdc2f/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/transactions/TxRollbackOnTimeoutNoDeadlockDetectionTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/transactions/TxRollbackOnTimeoutNoDeadlockDetectionTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/transactions/TxRollbackOnTimeoutNoDeadlockDetectionTest.java
new file mode 100644
index 0000000..5123329
--- /dev/null
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/transactions/TxRollbackOnTimeoutNoDeadlockDetectionTest.java
@@ -0,0 +1,47 @@
+/*
+ * 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.transactions;
+
+import org.apache.ignite.transactions.TransactionTimeoutException;
+
+import static org.apache.ignite.IgniteSystemProperties.IGNITE_TX_DEADLOCK_DETECTION_MAX_ITERS;
+
+/**
+ * Tests an ability to eagerly rollback timed out transactions.
+ */
+public class TxRollbackOnTimeoutNoDeadlockDetectionTest extends TxRollbackOnTimeoutTest {
+ /** {@inheritDoc} */
+ @Override protected void beforeTestsStarted() throws Exception {
+ System.setProperty(IGNITE_TX_DEADLOCK_DETECTION_MAX_ITERS, "0");
+
+ super.beforeTestsStarted();
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void afterTestsStopped() throws Exception {
+ super.afterTestsStopped();
+
+ System.clearProperty(IGNITE_TX_DEADLOCK_DETECTION_MAX_ITERS);
+ }
+
+ /** */
+ @Override protected void validateDeadlockException(Exception e) {
+ assertEquals("TimeoutException is expected",
+ TransactionTimeoutException.class, e.getCause().getClass());
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/ignite/blob/56dbdc2f/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/transactions/TxRollbackOnTimeoutTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/transactions/TxRollbackOnTimeoutTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/transactions/TxRollbackOnTimeoutTest.java
new file mode 100644
index 0000000..84cda23
--- /dev/null
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/transactions/TxRollbackOnTimeoutTest.java
@@ -0,0 +1,658 @@
+/*
+ * 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.transactions;
+
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ThreadLocalRandom;
+import javax.cache.CacheException;
+import org.apache.ignite.Ignite;
+import org.apache.ignite.IgniteCache;
+import org.apache.ignite.IgniteCheckedException;
+import org.apache.ignite.IgniteException;
+import org.apache.ignite.configuration.CacheConfiguration;
+import org.apache.ignite.configuration.IgniteConfiguration;
+import org.apache.ignite.configuration.NearCacheConfiguration;
+import org.apache.ignite.internal.IgniteEx;
+import org.apache.ignite.internal.IgniteInternalFuture;
+import org.apache.ignite.internal.processors.cache.distributed.near.GridNearTxLocal;
+import org.apache.ignite.internal.util.GridConcurrentSkipListSet;
+import org.apache.ignite.internal.util.typedef.F;
+import org.apache.ignite.internal.util.typedef.G;
+import org.apache.ignite.internal.util.typedef.X;
+import org.apache.ignite.internal.util.typedef.internal.U;
+import org.apache.ignite.lang.IgniteInClosure;
+import org.apache.ignite.spi.discovery.tcp.TcpDiscoverySpi;
+import org.apache.ignite.spi.discovery.tcp.ipfinder.vm.TcpDiscoveryVmIpFinder;
+import org.apache.ignite.testframework.GridTestUtils;
+import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
+import org.apache.ignite.transactions.Transaction;
+import org.apache.ignite.transactions.TransactionConcurrency;
+import org.apache.ignite.transactions.TransactionDeadlockException;
+import org.apache.ignite.transactions.TransactionIsolation;
+import org.apache.ignite.transactions.TransactionTimeoutException;
+
+import static org.apache.ignite.cache.CacheAtomicityMode.TRANSACTIONAL;
+import static org.apache.ignite.cache.CacheWriteSynchronizationMode.FULL_SYNC;
+import static org.apache.ignite.transactions.TransactionConcurrency.PESSIMISTIC;
+import static org.apache.ignite.transactions.TransactionIsolation.REPEATABLE_READ;
+
+/**
+ * Tests an ability to eagerly rollback timed out transactions.
+ */
+public class TxRollbackOnTimeoutTest extends GridCommonAbstractTest {
+ /** */
+ private static final long TX_MIN_TIMEOUT = 1;
+
+ /** */
+ private static final String CACHE_NAME = "test";
+
+ /** IP finder. */
+ private static final TcpDiscoveryVmIpFinder IP_FINDER = new TcpDiscoveryVmIpFinder(true);
+
+ /** */
+ private static final int GRID_CNT = 3;
+
+ /** {@inheritDoc} */
+ @Override protected IgniteConfiguration getConfiguration(String igniteInstanceName) throws Exception {
+ IgniteConfiguration cfg = super.getConfiguration(igniteInstanceName);
+
+ ((TcpDiscoverySpi)cfg.getDiscoverySpi()).setIpFinder(IP_FINDER);
+
+ boolean client = "client".equals(igniteInstanceName);
+
+ cfg.setClientMode(client);
+
+ if (!client) {
+ CacheConfiguration ccfg = new CacheConfiguration(CACHE_NAME);
+
+ if (nearCacheEnabled())
+ ccfg.setNearConfiguration(new NearCacheConfiguration());
+
+ ccfg.setAtomicityMode(TRANSACTIONAL);
+ ccfg.setBackups(2);
+ ccfg.setWriteSynchronizationMode(FULL_SYNC);
+
+ cfg.setCacheConfiguration(ccfg);
+ }
+
+ return cfg;
+ }
+
+ /**
+ * @return Near cache flag.
+ */
+ protected boolean nearCacheEnabled() {
+ return false;
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void beforeTest() throws Exception {
+ super.beforeTest();
+
+ startGridsMultiThreaded(GRID_CNT);
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void afterTest() throws Exception {
+ super.afterTest();
+
+ stopAllGrids();
+ }
+
+ /**
+ * @throws Exception If f nodeailed.
+ * @return Started client.
+ */
+ private Ignite startClient() throws Exception {
+ Ignite client = startGrid("client");
+
+ assertTrue(client.configuration().isClientMode());
+
+ if (nearCacheEnabled())
+ client.createNearCache(CACHE_NAME, new NearCacheConfiguration<>());
+ else
+ assertNotNull(client.cache(CACHE_NAME));
+
+ return client;
+ }
+
+ /**
+ * @param e Exception.
+ */
+ protected void validateDeadlockException(Exception e) {
+ assertEquals("Deadlock report is expected",
+ TransactionDeadlockException.class, e.getCause().getCause().getClass());
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testLockAndConcurrentTimeout() throws Exception {
+ startClient();
+
+ for (Ignite node : G.allGrids()) {
+ log.info("Test with node: " + node.name());
+
+ lock(node, false);
+
+ lock(node, false);
+
+ lock(node, true);
+ }
+ }
+
+ /**
+ * @param node Node.
+ * @param retry {@code True}
+ * @throws Exception If failed.
+ */
+ private void lock(final Ignite node, final boolean retry) throws Exception {
+ final IgniteCache<Object, Object> cache = node.cache(CACHE_NAME);
+
+ final int KEYS_PER_THREAD = 10_000;
+
+ GridTestUtils.runMultiThreaded(new IgniteInClosure<Integer>() {
+ @Override public void apply(Integer idx) {
+ int start = idx * KEYS_PER_THREAD;
+ int end = start + KEYS_PER_THREAD;
+
+ int locked = 0;
+
+ try {
+ try (Transaction tx = node.transactions().txStart(PESSIMISTIC, REPEATABLE_READ, 500, 0)) {
+ for (int i = start; i < end; i++) {
+ cache.get(i);
+
+ locked++;
+ }
+
+ tx.commit();
+ }
+ }
+ catch (Exception e) {
+ info("Expected error: " + e);
+ }
+
+ info("Done, locked: " + locked);
+
+ if (retry) {
+ try (Transaction tx = node.transactions().txStart(PESSIMISTIC, REPEATABLE_READ, 10 * 60_000, 0)) {
+ for (int i = start; i < end; i++) {
+ cache.get(i);
+
+ locked++;
+ }
+
+ cache.put(start, 0);
+
+ tx.commit();
+ }
+ }
+ }
+ }, Runtime.getRuntime().availableProcessors(), "tx-thread");
+ }
+
+ /**
+ * Tests if timeout on first tx unblocks second tx waiting for the locked key.
+ *
+ * @throws Exception If failed.
+ */
+ public void testWaitingTxUnblockedOnTimeout() throws Exception {
+ waitingTxUnblockedOnTimeout(grid(0), grid(0));
+
+ waitingTxUnblockedOnTimeout(grid(0), grid(1));
+
+ Ignite client = startClient();
+
+ waitingTxUnblockedOnTimeout(grid(0), client);
+
+ waitingTxUnblockedOnTimeout(grid(1), client);
+
+ waitingTxUnblockedOnTimeout(client, grid(0));
+
+ waitingTxUnblockedOnTimeout(client, grid(1));
+
+ waitingTxUnblockedOnTimeout(client, client);
+ }
+
+ /**
+ * Tests if timeout on first tx unblocks second tx waiting for the locked key.
+ *
+ * @throws Exception If failed.
+ */
+ public void testWaitingTxUnblockedOnThreadDeath() throws Exception {
+ waitingTxUnblockedOnThreadDeath(grid(0), grid(0));
+
+ waitingTxUnblockedOnThreadDeath(grid(0), grid(1));
+
+ Ignite client = startClient();
+
+ waitingTxUnblockedOnThreadDeath(grid(0), client);
+
+ waitingTxUnblockedOnThreadDeath(grid(1), client);
+
+ waitingTxUnblockedOnThreadDeath(client, grid(0));
+
+ waitingTxUnblockedOnThreadDeath(client, grid(1));
+
+ waitingTxUnblockedOnThreadDeath(client, client);
+ }
+
+ /**
+ * Tests if deadlock is resolved on timeout with correct message.
+ *
+ * @throws Exception If failed.
+ */
+ public void testDeadlockUnblockedOnTimeout() throws Exception {
+ deadlockUnblockedOnTimeout(ignite(0), ignite(1));
+
+ deadlockUnblockedOnTimeout(ignite(0), ignite(0));
+
+ Ignite client = startClient();
+
+ deadlockUnblockedOnTimeout(ignite(0), client);
+
+ deadlockUnblockedOnTimeout(client, ignite(0));
+ }
+
+ /**
+ * Tests if deadlock is resolved on timeout with correct message.
+ *
+ * @param node1 First node.
+ * @param node2 Second node.
+ * @throws Exception If failed.
+ */
+ private void deadlockUnblockedOnTimeout(final Ignite node1, final Ignite node2) throws Exception {
+ info("Start test [node1=" + node1.name() + ", node2=" + node2.name() + ']');
+
+ final CountDownLatch l = new CountDownLatch(2);
+
+ IgniteInternalFuture<?> fut1 = GridTestUtils.runAsync(new Runnable() {
+ @Override public void run() {
+ try {
+ try (Transaction tx = node1.transactions().txStart(PESSIMISTIC, REPEATABLE_READ, 5000, 2)) {
+ node1.cache(CACHE_NAME).put(1, 10);
+
+ l.countDown();
+
+ U.awaitQuiet(l);
+
+ node1.cache(CACHE_NAME).put(2, 20);
+
+ tx.commit();
+
+ fail();
+ }
+ }
+ catch (CacheException e) {
+ // No-op.
+ validateDeadlockException(e);
+ }
+ }
+ }, "First");
+
+ IgniteInternalFuture<?> fut2 = GridTestUtils.runAsync(new Runnable() {
+ @Override public void run() {
+ try (Transaction tx = node2.transactions().txStart(PESSIMISTIC, REPEATABLE_READ, 0, 2)) {
+ node2.cache(CACHE_NAME).put(2, 2);
+
+ l.countDown();
+
+ U.awaitQuiet(l);
+
+ node2.cache(CACHE_NAME).put(1, 1);
+
+ tx.commit();
+ }
+ }
+ }, "Second");
+
+ fut1.get();
+ fut2.get();
+
+ assertTrue("Expecting committed key 2", node1.cache(CACHE_NAME).get(2) != null);
+ assertTrue("Expecting committed key 1", node1.cache(CACHE_NAME).get(1) != null);
+
+ node1.cache(CACHE_NAME).removeAll(F.asSet(1, 2));
+ }
+
+ /**
+ * Tests timeout object cleanup on tx commit.
+ *
+ * @throws Exception If failed.
+ */
+ public void testTimeoutRemoval() throws Exception {
+ IgniteEx client = (IgniteEx)startClient();
+
+ final long TX_TIMEOUT = 250;
+
+ final int modesCnt = 5;
+
+ for (int i = 0; i < modesCnt; i++)
+ testTimeoutRemoval0(grid(0), i, TX_TIMEOUT);
+
+ for (int i = 0; i < modesCnt; i++)
+ testTimeoutRemoval0(client, i, TX_TIMEOUT);
+
+ for (int i = 0; i < modesCnt; i++)
+ testTimeoutRemoval0(grid(0), i, TX_MIN_TIMEOUT);
+
+ for (int i = 0; i < modesCnt; i++)
+ testTimeoutRemoval0(client, i, TX_MIN_TIMEOUT);
+
+ ThreadLocalRandom rnd = ThreadLocalRandom.current();
+
+ // Repeat with more iterations to make sure everything is cleared.
+ for (int i = 0; i < 500; i++)
+ testTimeoutRemoval0(client, rnd.nextInt(modesCnt), TX_MIN_TIMEOUT);
+ }
+
+ /**
+ * Tests timeouts in all tx configurations.
+ *
+ * @throws Exception If failed.
+ */
+ public void testSimple() throws Exception {
+ for (TransactionConcurrency concurrency : TransactionConcurrency.values())
+ for (TransactionIsolation isolation : TransactionIsolation.values()) {
+ for (int op = 0; op < 4; op++)
+ testSimple0(concurrency, isolation, op);
+ }
+ }
+
+ /**
+ * @param concurrency Concurrency.
+ * @param isolation Isolation.
+ * @param op Operation to test.
+ * @throws Exception If failed.
+ */
+ private void testSimple0(TransactionConcurrency concurrency, TransactionIsolation isolation, int op) throws Exception {
+ Ignite near = grid(0);
+
+ final int key = 1, val = 1;
+
+ final long TX_TIMEOUT = 250;
+
+ IgniteCache<Object, Object> cache = near.cache(CACHE_NAME);
+
+ try (Transaction tx = near.transactions().txStart(concurrency, isolation, TX_TIMEOUT, 1)) {
+ cache.put(key, val);
+
+ U.sleep(TX_TIMEOUT * 2);
+
+ try {
+ switch (op) {
+ case 0:
+ cache.put(key + 1, val);
+
+ break;
+
+ case 1:
+ cache.remove(key + 1);
+
+ break;
+
+ case 2:
+ cache.get(key + 1);
+
+ break;
+
+ case 3:
+ tx.commit();
+
+ break;
+
+ default:
+ fail();
+ }
+
+ fail("Tx must timeout");
+ }
+ catch (CacheException | IgniteException e) {
+ assertTrue("Expected exception: " + e, X.hasCause(e, TransactionTimeoutException.class));
+ }
+ }
+
+ assertFalse("Must be removed by rollback on timeout", near.cache(CACHE_NAME).containsKey(key));
+ assertFalse("Must be removed by rollback on timeout", near.cache(CACHE_NAME).containsKey(key + 1));
+
+ assertNull(near.transactions().tx());
+ }
+
+ /**
+ * @param near Node.
+ * @param mode Test mode.
+ *
+ * @param timeout Tx timeout.
+ * @throws Exception If failed.
+ */
+ private void testTimeoutRemoval0(IgniteEx near, int mode, long timeout) throws Exception {
+ Throwable saved = null;
+
+ try (Transaction tx = near.transactions().txStart(PESSIMISTIC, REPEATABLE_READ, timeout, 1)) {
+ near.cache(CACHE_NAME).put(1, 1);
+
+ switch (mode) {
+ case 0:
+ tx.commit();
+ break;
+
+ case 1:
+ tx.commitAsync().get();
+ break;
+
+ case 2:
+ tx.rollback();
+ break;
+
+ case 3:
+ tx.rollbackAsync().get();
+ break;
+
+ case 4:
+ break;
+
+ default:
+ fail();
+ }
+ }
+ catch (Throwable t) {
+ saved = t;
+ }
+
+ GridConcurrentSkipListSet set = U.field(near.context().cache().context().time(), "timeoutObjs");
+
+ for (Object obj : set) {
+ if (obj.getClass().isAssignableFrom(GridNearTxLocal.class)) {
+ log.error("Last saved exception: " + saved, saved);
+
+ fail("Not remove for mode=" + mode + " and timeout=" + timeout);
+ }
+ }
+ }
+
+ /**
+ * @param near Node starting tx which is timed out.
+ * @param other Node starting second tx.
+ * @throws Exception If failed.
+ */
+ private void waitingTxUnblockedOnTimeout(final Ignite near, final Ignite other) throws Exception {
+ waitingTxUnblockedOnTimeout(near, other, 1000);
+
+ waitingTxUnblockedOnTimeout(near, other, 50);
+ }
+
+ /**
+ * @param near Node starting tx which is timed out.
+ * @param other Node starting second tx.
+ * @param timeout Timeout.
+ * @throws Exception If failed.
+ */
+ private void waitingTxUnblockedOnTimeout(final Ignite near, final Ignite other, final long timeout) throws Exception {
+ info("Start test [node1=" + near.name() + ", node2=" + other.name() + ']');
+
+ final CountDownLatch blocked = new CountDownLatch(1);
+
+ final CountDownLatch unblocked = new CountDownLatch(1);
+
+ final int recordsCnt = 5;
+
+ IgniteInternalFuture<?> fut1 = GridTestUtils.runAsync(new Runnable() {
+ @Override public void run() {
+ try (Transaction tx = near.transactions().txStart(PESSIMISTIC, REPEATABLE_READ, timeout, 0)) {
+ try {
+ for (int i = 0; i < recordsCnt; i++)
+ near.cache(CACHE_NAME).put(i, i);
+
+ info("Locked all keys.");
+ }
+ catch (CacheException e) {
+ info("Failed to lock keys: " + e);
+ }
+ finally {
+ blocked.countDown();
+ }
+
+ // Will be unblocked after tx timeout occurs.
+ U.awaitQuiet(unblocked);
+
+ try {
+ near.cache(CACHE_NAME).put(0, 0);
+
+ fail();
+ }
+ catch (CacheException e) {
+ log.info("Expecting error: " + e.getMessage());
+ }
+
+ try {
+ tx.commit();
+
+ fail();
+ }
+ catch (IgniteException e) {
+ log.info("Expecting error: " + e.getMessage());
+ }
+ }
+
+ // Check thread is able to start new tx.
+ try (Transaction tx = near.transactions().txStart(PESSIMISTIC, REPEATABLE_READ, 60_000, 0)) {
+ for (int i = 0; i < recordsCnt; i++)
+ near.cache(CACHE_NAME).put(i, i);
+
+ tx.commit();
+ }
+ }
+ }, "First");
+
+ IgniteInternalFuture<?> fut2 = GridTestUtils.runAsync(new Runnable() {
+ @Override public void run() {
+ U.awaitQuiet(blocked);
+
+ try (Transaction tx = other.transactions().txStart(PESSIMISTIC, REPEATABLE_READ, 0, 1)) {
+ for (int i = 0; i < recordsCnt; i++)
+ other.cache(CACHE_NAME).put(i, i);
+
+ // Will wait until timeout on first tx will unblock put.
+ tx.commit();
+ }
+ }
+ }, "Second");
+
+ fut2.get();
+
+ unblocked.countDown();
+
+ fut1.get();
+ }
+
+ /**
+ * @param near Node starting tx which is timed out.
+ * @param other Node starting second tx.
+ * @throws Exception If failed.
+ */
+ private void waitingTxUnblockedOnThreadDeath(final Ignite near, final Ignite other) throws Exception {
+ waitingTxUnblockedOnThreadDeath0(near, other, 10, 1000); // Try provoke timeout after all keys are locked.
+
+ waitingTxUnblockedOnThreadDeath0(near, other, 1000, 100); // Try provoke timeout while trying to lock keys.
+ }
+
+ /**
+ * @param near Node starting tx which is timed out.
+ * @param other Node starting second tx.
+ * @param recordsCnt Number of records to locks.
+ * @param timeout Transaction timeout.
+ * @throws Exception If failed.
+ */
+ private void waitingTxUnblockedOnThreadDeath0(final Ignite near,
+ final Ignite other,
+ final int recordsCnt,
+ final long timeout)
+ throws Exception
+ {
+ info("Start test [node1=" + near.name() + ", node2=" + other.name() + ']');
+
+ final CountDownLatch blocked = new CountDownLatch(1);
+
+ IgniteInternalFuture<?> fut1 = multithreadedAsync(new Runnable() {
+ @Override public void run() {
+ near.transactions().txStart(PESSIMISTIC, REPEATABLE_READ, timeout, recordsCnt);
+
+ try {
+ for (int i = 0; i < recordsCnt; i++)
+ near.cache(CACHE_NAME).put(i, i);
+
+ log.info("Locked all records.");
+ }
+ catch (Exception e) {
+ log.info("Failed to locked all records: " + e);
+ }
+ finally {
+ blocked.countDown();
+ }
+
+ throw new IgniteException("Failure");
+ }
+ }, 1, "First");
+
+ IgniteInternalFuture<?> fut2 = multithreadedAsync(new Runnable() {
+ @Override public void run() {
+ U.awaitQuiet(blocked);
+
+ try (Transaction tx = other.transactions().txStart(PESSIMISTIC, REPEATABLE_READ, 0, recordsCnt)) {
+ for (int i = 0; i < recordsCnt; i++)
+ other.cache(CACHE_NAME).put(i, i);
+
+ // Will wait until timeout on first tx will unblock put.
+ tx.commit();
+ }
+ }
+ }, 1, "Second");
+
+ try {
+ fut1.get();
+
+ fail();
+ }
+ catch (IgniteCheckedException e) {
+ // No-op.
+ }
+
+ fut2.get();
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/ignite/blob/56dbdc2f/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheFullApiSelfTestSuite.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheFullApiSelfTestSuite.java b/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheFullApiSelfTestSuite.java
index 164ff6a..b380ebc 100644
--- a/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheFullApiSelfTestSuite.java
+++ b/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheFullApiSelfTestSuite.java
@@ -31,6 +31,7 @@ import org.apache.ignite.internal.processors.cache.distributed.dht.GridCachePart
import org.apache.ignite.internal.processors.cache.distributed.dht.GridCachePartitionedNearDisabledMultiNodeWithGroupFullApiSelfTest;
import org.apache.ignite.internal.processors.cache.distributed.dht.GridCachePartitionedNearDisabledOnheapFullApiSelfTest;
import org.apache.ignite.internal.processors.cache.distributed.dht.GridCachePartitionedNearDisabledOnheapMultiNodeFullApiSelfTest;
+import org.apache.ignite.internal.processors.cache.distributed.near.CachePartitionedMultiNodeLongTxTimeout2FullApiTest;
import org.apache.ignite.internal.processors.cache.distributed.near.CachePartitionedMultiNodeLongTxTimeoutFullApiTest;
import org.apache.ignite.internal.processors.cache.distributed.near.CachePartitionedNearEnabledMultiNodeLongTxTimeoutFullApiTest;
import org.apache.ignite.internal.processors.cache.distributed.near.GridCacheAtomicClientOnlyMultiNodeFullApiSelfTest;
@@ -123,6 +124,7 @@ public class IgniteCacheFullApiSelfTestSuite extends TestSuite {
suite.addTestSuite(GridCacheAtomicMultiNodeP2PDisabledFullApiSelfTest.class);
suite.addTestSuite(GridCacheAtomicNearEnabledMultiNodeFullApiSelfTest.class);
suite.addTestSuite(CachePartitionedMultiNodeLongTxTimeoutFullApiTest.class);
+ suite.addTestSuite(CachePartitionedMultiNodeLongTxTimeout2FullApiTest.class);
suite.addTestSuite(CachePartitionedNearEnabledMultiNodeLongTxTimeoutFullApiTest.class);
suite.addTestSuite(GridCachePartitionedNearDisabledMultiNodeFullApiSelfTest.class);
http://git-wip-us.apache.org/repos/asf/ignite/blob/56dbdc2f/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheTestSuite.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheTestSuite.java b/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheTestSuite.java
index e8810bb..0df600c 100755
--- a/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheTestSuite.java
+++ b/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheTestSuite.java
@@ -124,6 +124,7 @@ import org.apache.ignite.internal.processors.cache.distributed.IgniteCacheConnec
import org.apache.ignite.internal.processors.cache.distributed.IgniteCacheMessageRecoveryIdleConnectionTest;
import org.apache.ignite.internal.processors.cache.distributed.IgniteCacheMessageWriteTimeoutTest;
import org.apache.ignite.internal.processors.cache.distributed.IgniteCacheSystemTransactionsSelfTest;
+import org.apache.ignite.internal.processors.cache.distributed.IgniteCacheThreadLocalTxTest;
import org.apache.ignite.internal.processors.cache.distributed.IgniteCacheTxMessageRecoveryTest;
import org.apache.ignite.internal.processors.cache.distributed.IgniteCrossCacheTxStoreSelfTest;
import org.apache.ignite.internal.processors.cache.distributed.dht.GridCacheAtomicNearCacheSelfTest;
@@ -135,6 +136,9 @@ import org.apache.ignite.internal.processors.cache.distributed.replicated.GridCa
import org.apache.ignite.internal.processors.cache.local.GridCacheLocalTxExceptionSelfTest;
import org.apache.ignite.internal.processors.cache.query.continuous.CacheEntryProcessorExternalizableFailedTest;
import org.apache.ignite.internal.processors.cache.query.continuous.CacheEntryProcessorNonSerializableTest;
+import org.apache.ignite.internal.processors.cache.transactions.TxRollbackOnTimeoutNearCacheTest;
+import org.apache.ignite.internal.processors.cache.transactions.TxRollbackOnTimeoutNoDeadlockDetectionTest;
+import org.apache.ignite.internal.processors.cache.transactions.TxRollbackOnTimeoutTest;
import org.apache.ignite.internal.processors.datastreamer.DataStreamProcessorSelfTest;
import org.apache.ignite.internal.processors.datastreamer.DataStreamerImplSelfTest;
import org.apache.ignite.internal.processors.datastreamer.DataStreamerMultiThreadedSelfTest;
@@ -317,6 +321,11 @@ public class IgniteCacheTestSuite extends TestSuite {
suite.addTestSuite(GridStoreLoadCacheTest.class);
+ suite.addTestSuite(TxRollbackOnTimeoutTest.class);
+ suite.addTestSuite(TxRollbackOnTimeoutNearCacheTest.class);
+ suite.addTestSuite(TxRollbackOnTimeoutNoDeadlockDetectionTest.class);
+ suite.addTestSuite(IgniteCacheThreadLocalTxTest.class);
+
return suite;
}
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/56dbdc2f/modules/hadoop/src/test/java/org/apache/ignite/internal/processors/hadoop/impl/HadoopTxConfigCacheTest.java
----------------------------------------------------------------------
diff --git a/modules/hadoop/src/test/java/org/apache/ignite/internal/processors/hadoop/impl/HadoopTxConfigCacheTest.java b/modules/hadoop/src/test/java/org/apache/ignite/internal/processors/hadoop/impl/HadoopTxConfigCacheTest.java
index e85baed..f404732 100644
--- a/modules/hadoop/src/test/java/org/apache/ignite/internal/processors/hadoop/impl/HadoopTxConfigCacheTest.java
+++ b/modules/hadoop/src/test/java/org/apache/ignite/internal/processors/hadoop/impl/HadoopTxConfigCacheTest.java
@@ -25,11 +25,11 @@ import org.apache.ignite.internal.util.typedef.internal.CU;
/**
* Test checks whether hadoop system cache doesn't use user defined TX config.
*/
-public class HadoopTxConfigCacheTest extends IgniteTxConfigCacheSelfTest {
+public class HadoopTxConfigCacheTest extends IgniteTxConfigCacheSelfTest {
/**
* Success if system caches weren't timed out.
*
- * @throws Exception
+ * @throws Exception If failed.
*/
public void testSystemCacheTx() throws Exception {
final Ignite ignite = grid(0);
[3/7] ignite git commit: IGNITE-6209 .NET: Improve build script for
release builds on TC
Posted by sb...@apache.org.
IGNITE-6209 .NET: Improve build script for release builds on TC
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/06d29782
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/06d29782
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/06d29782
Branch: refs/heads/ignite-6181-2
Commit: 06d297823a3c229f570a3132f1bb222c0349db90
Parents: d1f178f
Author: Pavel Tupitsyn <pt...@apache.org>
Authored: Wed Sep 20 13:37:29 2017 +0300
Committer: Pavel Tupitsyn <pt...@apache.org>
Committed: Wed Sep 20 13:37:29 2017 +0300
----------------------------------------------------------------------
modules/platforms/dotnet/build.ps1 | 122 ++++++++++++++++++++++----------
1 file changed, 83 insertions(+), 39 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ignite/blob/06d29782/modules/platforms/dotnet/build.ps1
----------------------------------------------------------------------
diff --git a/modules/platforms/dotnet/build.ps1 b/modules/platforms/dotnet/build.ps1
index d9757ed..c037574 100644
--- a/modules/platforms/dotnet/build.ps1
+++ b/modules/platforms/dotnet/build.ps1
@@ -33,6 +33,9 @@ Requirements:
.PARAMETER skipJava
Skip Java build.
+.PARAMETER skipDotNet
+Skip .NET build.
+
.PARAMETER skipNuGet
Skip NuGet packaging.
@@ -51,6 +54,15 @@ Build configuration ("Release", "Debug").
.PARAMETER mavenOpts
Custom Maven options, default is "-U -P-lgpl,-scala,-examples,-test,-benchmarks -Dmaven.javadoc.skip=true".
+.PARAMETER jarDirs
+Java jar files source folders, default is "modules\indexing\target,modules\core\target,modules\spring\target"
+
+.PARAMETER asmDirs
+.NET assembly directories to copy pre-build binaries from. Default is "", which means no copy.
+
+.PARAMETER nugetPath
+Path to nuget.exe.
+
.EXAMPLE
.\build.ps1 -clean
# Full rebuild of Java, .NET and NuGet packages.
@@ -63,6 +75,7 @@ Custom Maven options, default is "-U -P-lgpl,-scala,-examples,-test,-benchmarks
param (
[switch]$skipJava,
+ [switch]$skipDotNet,
[switch]$skipNuGet,
[switch]$skipCodeAnalysis,
[switch]$clean,
@@ -70,15 +83,22 @@ param (
[string]$platform="Any CPU",
[ValidateSet("Release", "Debug")]
[string]$configuration="Release",
- [string]$mavenOpts="-U -P-lgpl,-scala,-examples,-test,-benchmarks -Dmaven.javadoc.skip=true"
+ [string]$mavenOpts="-U -P-lgpl,-scala,-examples,-test,-benchmarks -Dmaven.javadoc.skip=true",
+ [string]$jarDirs="modules\indexing\target,modules\core\target,modules\spring\target",
+ [string]$asmDirs="",
+ [string]$nugetPath=""
)
# 1) Build Java (Maven)
# Detect Ignite root directory
cd $PSScriptRoot\..
-while (!((Test-Path bin) -and (Test-Path examples) -and ((Test-Path modules) -or (Test-Path platforms))))
-{ cd .. }
+while (!((Test-Path bin) -and (Test-Path examples) -and ((Test-Path modules) -or (Test-Path platforms)))) {
+ cd ..
+ if ((Get-Location).Drive.Root -eq (Get-Location).Path) {
+ break
+ }
+}
echo "Ignite home detected at '$pwd'."
@@ -116,56 +136,80 @@ else {
$libsDir = "$PSScriptRoot\bin\Libs"
mkdir -Force $libsDir; del -Force $libsDir\*.*
-ls modules\indexing\target,modules\core\target,modules\spring\target *.jar -recurse `
+ls $jarDirs.Split(',') *.jar -recurse `
-include "ignite-core*","ignite-indexing*","ignite-shmem*","ignite-spring*","lucene*","h2*","cache-api*","commons-*","spring*" `
- -exclude "*-sources*","*-javadoc*","*-tests*" `
+ -exclude "*-sources*","*-javadoc*","*-tests*","*optional*" `
| % { copy -Force $_ $libsDir }
-
+
# Restore directory
cd $PSScriptRoot
# 2) Build .NET
-# Detect MSBuild 4.0+
-for ($i=20; $i -ge 4; $i--) {
- $regKey = "HKLM:\software\Microsoft\MSBuild\ToolsVersions\$i.0"
- if (Test-Path $regKey) { break }
+
+# Detect NuGet
+$ng = if ($nugetPath) { $nugetPath } else { "nuget" }
+
+if ((Get-Command $ng -ErrorAction SilentlyContinue) -eq $null) {
+ $ng = ".\nuget.exe"
+
+ if (-not (Test-Path $ng)) {
+ echo "Downloading NuGet..."
+ (New-Object System.Net.WebClient).DownloadFile("https://dist.nuget.org/win-x86-commandline/v3.3.0/nuget.exe", "nuget.exe");
+ }
}
-if (!(Test-Path $regKey)) {
- echo "Failed to detect MSBuild path, exiting."
- exit -1
+echo "Using NuGet from: $ng"
+
+if (!$skipDotNet) {
+ # Detect MSBuild 4.0+
+ for ($i=20; $i -ge 4; $i--) {
+ $regKey = "HKLM:\software\Microsoft\MSBuild\ToolsVersions\$i.0"
+ if (Test-Path $regKey) { break }
+ }
+
+ if (!(Test-Path $regKey)) {
+ echo "Failed to detect MSBuild path, exiting."
+ exit -1
+ }
+
+ $msbuildExe = (join-path -path (Get-ItemProperty $regKey)."MSBuildToolsPath" -childpath "msbuild.exe")
+ echo "MSBuild detected at '$msbuildExe'."
+
+ # Restore NuGet packages
+ echo "Restoring NuGet..."
+ & $ng restore
+
+ # Build
+ $targets = if ($clean) {"Clean;Rebuild"} else {"Build"}
+ $codeAnalysis = if ($skipCodeAnalysis) {"/p:RunCodeAnalysis=false"} else {""}
+ $msBuildCommand = "`"$msBuildExe`" Apache.Ignite.sln /target:$targets /p:Configuration=$configuration /p:Platform=`"$platform`" $codeAnalysis /p:UseSharedCompilation=false"
+ echo "Starting MsBuild: '$msBuildCommand'"
+ cmd /c $msBuildCommand
+
+ # Check result
+ if ($LastExitCode -ne 0) {
+ echo ".NET build failed."
+ exit -1
+ }
}
-$msbuildExe = (join-path -path (Get-ItemProperty $regKey)."MSBuildToolsPath" -childpath "msbuild.exe")
-echo "MSBuild detected at '$msbuildExe'."
+if ($asmDirs) {
+ ls $asmDirs.Split(',') | % `
+ {
+ $projName = [System.IO.Path]::GetFileNameWithoutExtension($_.Name);
-# Detect NuGet
-$ng = "nuget"
-if ((Get-Command $ng -ErrorAction SilentlyContinue) -eq $null) {
- $ng = ".\nuget.exe"
+ if ($_.Name.EndsWith(".exe.config")) {
+ $projName = [System.IO.Path]::GetFileNameWithoutExtension($projName);
+ }
- if (-not (Test-Path $ng)) {
- echo "Downloading NuGet..."
- (New-Object System.Net.WebClient).DownloadFile("https://dist.nuget.org/win-x86-commandline/v3.3.0/nuget.exe", "nuget.exe");
- }
-}
+ if ($projName.StartsWith("Apache.Ignite")) {
+ $target = "$projName\bin\Release"
+ mkdir -Force $target
-# Restore NuGet packages
-echo "Restoring NuGet..."
-& $ng restore
-
-# Build
-$targets = if ($clean) {"Clean;Rebuild"} else {"Build"}
-$codeAnalysis = if ($skipCodeAnalysis) {"/p:RunCodeAnalysis=false"} else {""}
-$msBuildCommand = "`"$msBuildExe`" Apache.Ignite.sln /target:$targets /p:Configuration=$configuration /p:Platform=`"$platform`" $codeAnalysis /p:UseSharedCompilation=false"
-echo "Starting MsBuild: '$msBuildCommand'"
-cmd /c $msBuildCommand
-
-# Check result
-if ($LastExitCode -ne 0) {
- echo ".NET build failed."
- exit -1
+ xcopy /s /y $_.FullName $target
+ }
+ }
}
# Copy binaries
[2/7] ignite git commit: IGNITE-6304 Fixed script execution after
throttling interval.
Posted by sb...@apache.org.
IGNITE-6304 Fixed script execution after throttling interval.
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/d1f178f3
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/d1f178f3
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/d1f178f3
Branch: refs/heads/ignite-6181-2
Commit: d1f178f3ca339d590be3031db167ecc716ffd33f
Parents: ff17783
Author: vsisko <vs...@gridgain.com>
Authored: Wed Sep 20 17:27:14 2017 +0700
Committer: Alexey Kuznetsov <ak...@apache.org>
Committed: Wed Sep 20 17:27:14 2017 +0700
----------------------------------------------------------------------
.../commands/alert/VisorAlertCommand.scala | 29 ++++++++++----------
1 file changed, 15 insertions(+), 14 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ignite/blob/d1f178f3/modules/visor-console/src/main/scala/org/apache/ignite/visor/commands/alert/VisorAlertCommand.scala
----------------------------------------------------------------------
diff --git a/modules/visor-console/src/main/scala/org/apache/ignite/visor/commands/alert/VisorAlertCommand.scala b/modules/visor-console/src/main/scala/org/apache/ignite/visor/commands/alert/VisorAlertCommand.scala
index fb60a1e..254dc2d 100644
--- a/modules/visor-console/src/main/scala/org/apache/ignite/visor/commands/alert/VisorAlertCommand.scala
+++ b/modules/visor-console/src/main/scala/org/apache/ignite/visor/commands/alert/VisorAlertCommand.scala
@@ -85,7 +85,7 @@ import scala.util.control.Breaks._
* 2) Alert condition as string.
* 3, ...) Values of alert conditions ordered as in alert command.
* -i
- * Configure alert notification minimal throttling interval in seconds. Default is 60 seconds.
+ * Configure alert notification minimal throttling interval in seconds. Default is 0 seconds.
*
* -<metric>
* This defines a mnemonic for the metric that will be measured:
@@ -267,7 +267,7 @@ class VisorAlertCommand extends VisorConsoleCommand {
var script: Option[String] = None
val conditions = mutable.ArrayBuffer.empty[VisorAlertCondition]
var freq = DFLT_FREQ
- var interval = DFLT_FREQ
+ var interval = 0L
try {
args.foreach(arg => {
@@ -653,25 +653,26 @@ class VisorAlertCommand extends VisorConsoleCommand {
private def executeAlertScript(alert: VisorAlert, node: ClusterNode, values: Seq[String]) {
val n = alert.notification
+ if (n.notified && System.currentTimeMillis() - n.notifiedTime < n.throttleInterval)
+ return
+
try {
n.script.foreach(script => {
- if (!n.notified && (n.notifiedTime < 0 || (System.currentTimeMillis() - n.notifiedTime) > n.throttleInterval)) {
- val scriptFile = new File(script)
+ val scriptFile = new File(script)
- if (!scriptFile.exists())
- throw new FileNotFoundException("Script/executable not found: " + script)
+ if (!scriptFile.exists())
+ throw new FileNotFoundException("Script/executable not found: " + script)
- val scriptFolder = scriptFile.getParentFile
+ val scriptFolder = scriptFile.getParentFile
- val p = Process(Seq(script, alert.name.getOrElse(alert.id), alert.conditionSpec) ++ values,
- Some(scriptFolder))
+ val p = Process(Seq(script, alert.name.getOrElse(alert.id), alert.conditionSpec) ++ values,
+ Some(scriptFolder))
- p.run(ProcessLogger((fn: String) => {}))
+ p.run(ProcessLogger((fn: String) => {}))
- n.notifiedTime = System.currentTimeMillis()
+ n.notifiedTime = System.currentTimeMillis()
- n.notified = true
- }
+ n.notified = true
})
}
catch {
@@ -865,7 +866,7 @@ object VisorAlertCommand {
" 2) Alert condition as string.",
" 3, ...) Values of alert conditions ordered as in alert command."
),
- "-i" -> "Configure alert notification minimal throttling interval in seconds. Default is 60 seconds.",
+ "-i" -> "Configure alert notification minimal throttling interval in seconds. Default is 0 seconds.",
"-<metric>" -> Seq(
"This defines a mnemonic for the metric that will be measured:",
"",