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/06/05 08:57:05 UTC
[1/5] ignite git commit: Fixed DiscoverySpi methods return types.
Repository: ignite
Updated Branches:
refs/heads/ignite-5293 [created] 282d0d6d1
Fixed DiscoverySpi methods return types.
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/87708161
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/87708161
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/87708161
Branch: refs/heads/ignite-5293
Commit: 87708161edac26caa6b6cad0c41fbf40e09aac00
Parents: bc30dc7
Author: Yakov Zhdanov <yz...@gridgain.com>
Authored: Thu Jun 1 20:01:58 2017 +0300
Committer: Yakov Zhdanov <yz...@gridgain.com>
Committed: Thu Jun 1 20:01:58 2017 +0300
----------------------------------------------------------------------
.../ignite/codegen/MessageCodeGenerator.java | 20 +-
.../processors/cache/GridCacheIoManager.java | 2 +
.../distributed/GridDistributedLockRequest.java | 4 +-
.../GridDistributedLockResponse.java | 48 ++-
.../GridDistributedTxPrepareRequest.java | 7 +
.../distributed/dht/GridDhtLockFuture.java | 298 ++++++++++---------
.../distributed/dht/GridDhtLockRequest.java | 12 +-
.../distributed/dht/GridDhtLockResponse.java | 90 ++----
.../dht/GridDhtTransactionalCacheAdapter.java | 47 ++-
.../distributed/dht/GridDhtTxPrepareFuture.java | 28 +-
.../dht/colocated/GridDhtColocatedCache.java | 11 +-
.../colocated/GridDhtColocatedLockFuture.java | 1 -
.../distributed/near/GridNearLockResponse.java | 30 +-
.../GridNearPessimisticTxPrepareFuture.java | 2 +-
.../processors/job/GridJobProcessor.java | 1 +
.../ignite/internal/util/StripedExecutor.java | 2 +
16 files changed, 298 insertions(+), 305 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ignite/blob/87708161/modules/codegen/src/main/java/org/apache/ignite/codegen/MessageCodeGenerator.java
----------------------------------------------------------------------
diff --git a/modules/codegen/src/main/java/org/apache/ignite/codegen/MessageCodeGenerator.java b/modules/codegen/src/main/java/org/apache/ignite/codegen/MessageCodeGenerator.java
index 99ec08a..24f5087 100644
--- a/modules/codegen/src/main/java/org/apache/ignite/codegen/MessageCodeGenerator.java
+++ b/modules/codegen/src/main/java/org/apache/ignite/codegen/MessageCodeGenerator.java
@@ -44,8 +44,12 @@ import org.apache.ignite.internal.GridDirectCollection;
import org.apache.ignite.internal.GridDirectMap;
import org.apache.ignite.internal.GridDirectTransient;
import org.apache.ignite.internal.IgniteCodeGeneratingFail;
-import org.apache.ignite.internal.processors.cache.version.GridCacheVersion;
-import org.apache.ignite.internal.processors.cache.version.GridCacheVersionEx;
+import org.apache.ignite.internal.processors.cache.distributed.GridDistributedLockRequest;
+import org.apache.ignite.internal.processors.cache.distributed.GridDistributedLockResponse;
+import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtLockRequest;
+import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtLockResponse;
+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.util.typedef.internal.SB;
import org.apache.ignite.internal.util.typedef.internal.U;
import org.apache.ignite.lang.IgniteUuid;
@@ -177,12 +181,12 @@ public class MessageCodeGenerator {
// gen.generateAndWrite(GridMessageCollection.class);
// gen.generateAndWrite(DataStreamerEntry.class);
-// gen.generateAndWrite(GridDistributedLockRequest.class);
-// gen.generateAndWrite(GridDistributedLockResponse.class);
-// gen.generateAndWrite(GridNearLockRequest.class);
-// gen.generateAndWrite(GridNearLockResponse.class);
-// gen.generateAndWrite(GridDhtLockRequest.class);
-// gen.generateAndWrite(GridDhtLockResponse.class);
+ gen.generateAndWrite(GridDistributedLockRequest.class);
+ gen.generateAndWrite(GridDistributedLockResponse.class);
+ gen.generateAndWrite(GridNearLockRequest.class);
+ gen.generateAndWrite(GridNearLockResponse.class);
+ gen.generateAndWrite(GridDhtLockRequest.class);
+ gen.generateAndWrite(GridDhtLockResponse.class);
//
// gen.generateAndWrite(GridDistributedTxPrepareRequest.class);
// gen.generateAndWrite(GridDistributedTxPrepareResponse.class);
http://git-wip-us.apache.org/repos/asf/ignite/blob/87708161/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheIoManager.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheIoManager.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheIoManager.java
index bb31645..df1a747 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheIoManager.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheIoManager.java
@@ -531,6 +531,7 @@ public class GridCacheIoManager extends GridCacheSharedManagerAdapter {
req.version(),
req.futureId(),
req.miniId(),
+ req.partition(),
0,
ctx.deploymentEnabled());
@@ -670,6 +671,7 @@ public class GridCacheIoManager extends GridCacheSharedManagerAdapter {
req.futureId(),
req.miniId(),
false,
+ req.partition(),
0,
req.classError(),
null,
http://git-wip-us.apache.org/repos/asf/ignite/blob/87708161/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridDistributedLockRequest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridDistributedLockRequest.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridDistributedLockRequest.java
index 74f34a9..14c2519 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridDistributedLockRequest.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridDistributedLockRequest.java
@@ -565,7 +565,9 @@ public class GridDistributedLockRequest extends GridDistributedBaseMessage {
/** {@inheritDoc} */
@Override public String toString() {
- return S.toString(GridDistributedLockRequest.class, this, "keysCnt", retVals.length,
+ return S.toString(GridDistributedLockRequest.class, this,
+ "keysCnt", retVals.length,
+ "part", partition(),
"super", super.toString());
}
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/87708161/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridDistributedLockResponse.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridDistributedLockResponse.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridDistributedLockResponse.java
index 4b21896..eb65239 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridDistributedLockResponse.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridDistributedLockResponse.java
@@ -48,6 +48,9 @@ public class GridDistributedLockResponse extends GridDistributedBaseMessage {
/** Future ID. */
private IgniteUuid futId;
+ /** Partition. */
+ private int part;
+
/** Error. */
@GridDirectTransient
private Throwable err;
@@ -74,17 +77,21 @@ public class GridDistributedLockResponse extends GridDistributedBaseMessage {
* @param cnt Key count.
* @param addDepInfo Deployment info.
*/
- public GridDistributedLockResponse(int cacheId,
+ public GridDistributedLockResponse(
+ int cacheId,
GridCacheVersion lockVer,
IgniteUuid futId,
+ int part,
int cnt,
- boolean addDepInfo) {
+ boolean addDepInfo
+ ) {
super(lockVer, cnt, addDepInfo);
assert futId != null;
this.cacheId = cacheId;
this.futId = futId;
+ this.part = part;
vals = new ArrayList<>(cnt);
}
@@ -93,20 +100,25 @@ public class GridDistributedLockResponse extends GridDistributedBaseMessage {
* @param cacheId Cache ID.
* @param lockVer Lock ID.
* @param futId Future ID.
+ * @param part Partition.
* @param err Error.
* @param addDepInfo Deployment info.
*/
- public GridDistributedLockResponse(int cacheId,
+ public GridDistributedLockResponse(
+ int cacheId,
GridCacheVersion lockVer,
IgniteUuid futId,
+ int part,
Throwable err,
- boolean addDepInfo) {
+ boolean addDepInfo
+ ) {
super(lockVer, 0, addDepInfo);
assert futId != null;
this.cacheId = cacheId;
this.futId = futId;
+ this.part = part;
this.err = err;
}
@@ -114,13 +126,16 @@ public class GridDistributedLockResponse extends GridDistributedBaseMessage {
* @param cacheId Cache ID.
* @param lockVer Lock ID.
* @param futId Future ID.
+ * @param part Partition.
* @param cnt Count.
* @param err Error.
* @param addDepInfo Deployment info.
*/
- public GridDistributedLockResponse(int cacheId,
+ public GridDistributedLockResponse(
+ int cacheId,
GridCacheVersion lockVer,
IgniteUuid futId,
+ int part,
int cnt,
Throwable err,
boolean addDepInfo) {
@@ -131,12 +146,17 @@ public class GridDistributedLockResponse extends GridDistributedBaseMessage {
this.cacheId = cacheId;
this.futId = futId;
this.err = err;
+ this.part = part;
vals = new ArrayList<>(cnt);
}
+ /** {@inheritDoc} */
+ @Override public int partition() {
+ return part;
+ }
+
/**
- *
* @return Future ID.
*/
public IgniteUuid futureId() {
@@ -234,6 +254,12 @@ public class GridDistributedLockResponse extends GridDistributedBaseMessage {
writer.incrementState();
case 9:
+ if (!writer.writeInt("part", part))
+ return false;
+
+ writer.incrementState();
+
+ case 10:
if (!writer.writeCollection("vals", vals, MessageCollectionItemType.MSG))
return false;
@@ -272,6 +298,14 @@ public class GridDistributedLockResponse extends GridDistributedBaseMessage {
reader.incrementState();
case 9:
+ part = reader.readInt("part");
+
+ if (!reader.isLastRead())
+ return false;
+
+ reader.incrementState();
+
+ case 10:
vals = reader.readCollection("vals", MessageCollectionItemType.MSG);
if (!reader.isLastRead())
@@ -291,7 +325,7 @@ public class GridDistributedLockResponse extends GridDistributedBaseMessage {
/** {@inheritDoc} */
@Override public byte fieldsCount() {
- return 10;
+ return 11;
}
/** {@inheritDoc} */
http://git-wip-us.apache.org/repos/asf/ignite/blob/87708161/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridDistributedTxPrepareRequest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridDistributedTxPrepareRequest.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridDistributedTxPrepareRequest.java
index 4b11414f..78aa217 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridDistributedTxPrepareRequest.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridDistributedTxPrepareRequest.java
@@ -360,6 +360,12 @@ public class GridDistributedTxPrepareRequest extends GridDistributedBaseMessage
this.txState = txState;
}
+ /** {@inheritDoc} */
+ @Override public int partition() {
+ return writes != null && !writes.isEmpty() ? F.first(writes).key().partition() :
+ reads != null && !reads.isEmpty() ? F.first(reads).key().partition() : -1;
+ }
+
/** {@inheritDoc}
* @param ctx*/
@Override public void prepareMarshal(GridCacheSharedContext ctx) throws IgniteCheckedException {
@@ -701,6 +707,7 @@ public class GridDistributedTxPrepareRequest extends GridDistributedBaseMessage
return GridToStringBuilder.toString(GridDistributedTxPrepareRequest.class, this,
"flags", flags.toString(),
+ "part", partition(),
"super", super.toString());
}
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/87708161/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtLockFuture.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtLockFuture.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtLockFuture.java
index 388a434..9365259 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtLockFuture.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtLockFuture.java
@@ -31,19 +31,17 @@ import java.util.concurrent.atomic.AtomicReference;
import org.apache.ignite.IgniteCheckedException;
import org.apache.ignite.IgniteLogger;
import org.apache.ignite.cluster.ClusterNode;
-import org.apache.ignite.internal.IgniteInternalFuture;
import org.apache.ignite.internal.NodeStoppingException;
import org.apache.ignite.internal.cluster.ClusterTopologyCheckedException;
import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
-import org.apache.ignite.internal.processors.cache.CacheEntryPredicate;
import org.apache.ignite.internal.processors.cache.CacheLockCandidates;
import org.apache.ignite.internal.processors.cache.CacheObject;
import org.apache.ignite.internal.processors.cache.GridCacheAdapter;
-import org.apache.ignite.internal.processors.cache.GridCacheCompoundIdentityFuture;
import org.apache.ignite.internal.processors.cache.GridCacheContext;
import org.apache.ignite.internal.processors.cache.GridCacheEntryEx;
import org.apache.ignite.internal.processors.cache.GridCacheEntryInfo;
import org.apache.ignite.internal.processors.cache.GridCacheEntryRemovedException;
+import org.apache.ignite.internal.processors.cache.GridCacheFutureAdapter;
import org.apache.ignite.internal.processors.cache.GridCacheMvccCandidate;
import org.apache.ignite.internal.processors.cache.GridCacheMvccFuture;
import org.apache.ignite.internal.processors.cache.GridCacheOperation;
@@ -53,11 +51,9 @@ import org.apache.ignite.internal.processors.cache.distributed.GridDistributedCa
import org.apache.ignite.internal.processors.cache.distributed.GridDistributedLockCancelledException;
import org.apache.ignite.internal.processors.cache.distributed.near.GridNearCacheAdapter;
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.version.GridCacheVersion;
import org.apache.ignite.internal.processors.dr.GridDrType;
import org.apache.ignite.internal.processors.timeout.GridTimeoutObjectAdapter;
-import org.apache.ignite.internal.util.future.GridFutureAdapter;
import org.apache.ignite.internal.util.tostring.GridToStringExclude;
import org.apache.ignite.internal.util.tostring.GridToStringInclude;
import org.apache.ignite.internal.util.typedef.C1;
@@ -79,11 +75,8 @@ import static org.apache.ignite.internal.processors.dr.GridDrType.DR_PRELOAD;
/**
* Cache lock future.
*/
-public final class GridDhtLockFuture extends GridCacheCompoundIdentityFuture<Boolean>
+public final class GridDhtLockFuture extends GridCacheFutureAdapter<Boolean>
implements GridCacheMvccFuture<Boolean>, GridDhtFuture<Boolean>, GridCacheMappedVersion {
- /** */
- private static final long serialVersionUID = 0L;
-
/** Logger reference. */
private static final AtomicReference<IgniteLogger> logRef = new AtomicReference<>();
@@ -140,9 +133,6 @@ public final class GridDhtLockFuture extends GridCacheCompoundIdentityFuture<Boo
/** Lock timeout. */
private final long timeout;
- /** Filter. */
- private CacheEntryPredicate[] filter;
-
/** Transaction. */
private GridDhtTxLocalAdapter tx;
@@ -173,6 +163,15 @@ public final class GridDhtLockFuture extends GridCacheCompoundIdentityFuture<Boo
/** Keep binary. */
private final boolean keepBinary;
+ /** */
+ private List<MiniFuture> miniFuts;
+
+ /** */
+ private int completedMiniCnt;
+
+ /** */
+ private boolean inited;
+
/**
* @param cctx Cache context.
* @param nearNodeId Near node ID.
@@ -185,7 +184,6 @@ public final class GridDhtLockFuture extends GridCacheCompoundIdentityFuture<Boo
* @param tx Transaction.
* @param threadId Thread ID.
* @param accessTtl TTL for read operation.
- * @param filter Filter.
* @param skipStore Skip store flag.
*/
public GridDhtLockFuture(
@@ -201,11 +199,9 @@ public final class GridDhtLockFuture extends GridCacheCompoundIdentityFuture<Boo
long threadId,
long createTtl,
long accessTtl,
- CacheEntryPredicate[] filter,
boolean skipStore,
- boolean keepBinary) {
- super(CU.boolReducer());
-
+ boolean keepBinary
+ ) {
assert nearNodeId != null;
assert nearLockVer != null;
assert topVer.topologyVersion() > 0;
@@ -218,7 +214,6 @@ public final class GridDhtLockFuture extends GridCacheCompoundIdentityFuture<Boo
this.read = read;
this.needReturnVal = needReturnVal;
this.timeout = timeout;
- this.filter = filter;
this.tx = tx;
this.createTtl = createTtl;
this.accessTtl = accessTtl;
@@ -482,13 +477,15 @@ public final class GridDhtLockFuture extends GridCacheCompoundIdentityFuture<Boo
@Override public boolean onNodeLeft(UUID nodeId) {
boolean found = false;
- for (IgniteInternalFuture<?> fut : futures()) {
- MiniFuture f = (MiniFuture)fut;
-
- if (f.node().id().equals(nodeId)) {
- f.onResult();
+ synchronized (this) {
+ if (miniFuts != null) {
+ for (MiniFuture miniFut : miniFuts) {
+ if (miniFut.node().id().equals(nodeId)) {
+ miniFut.onResult();
- found = true;
+ found = true;
+ }
+ }
}
}
@@ -527,21 +524,15 @@ public final class GridDhtLockFuture extends GridCacheCompoundIdentityFuture<Boo
* @return Mini future.
*/
@SuppressWarnings("ForLoopReplaceableByForEach")
- private MiniFuture miniFuture(IgniteUuid miniId) {
- // We iterate directly over the futs collection here to avoid copy.
+ private MiniFuture miniFuture(int miniId) {
synchronized (this) {
- int size = futuresCountNoLock();
+ if (miniFuts != null) {
+ MiniFuture mini = miniFuts.get(miniId);
- // Avoid iterator creation.
- for (int i = 0; i < size; i++) {
- MiniFuture mini = (MiniFuture) future(i);
-
- if (mini.futureId().equals(miniId)) {
- if (!mini.isDone())
- return mini;
- else
- return null;
- }
+ if (!mini.isDone())
+ return mini;
+ else
+ return null;
}
}
@@ -610,30 +601,6 @@ public final class GridDhtLockFuture extends GridCacheCompoundIdentityFuture<Boo
}
/**
- * @param cached Entry to check.
- * @return {@code True} if filter passed.
- */
- private boolean filter(GridCacheEntryEx cached) {
- try {
- if (!cctx.isAll(cached, filter)) {
- if (log.isDebugEnabled())
- log.debug("Filter didn't pass for entry (will fail lock): " + cached);
-
- onFailed(true);
-
- return false;
- }
-
- return true;
- }
- catch (IgniteCheckedException e) {
- onError(e);
-
- return false;
- }
- }
-
- /**
* Callback for whenever entry lock ownership changes.
*
* @param entry Entry whose lock ownership changed.
@@ -685,8 +652,9 @@ public final class GridDhtLockFuture extends GridCacheCompoundIdentityFuture<Boo
}
assert err == null || !success;
- assert !success || (initialized() && !hasPending()) : "Invalid done callback [success=" + success +
- ", fut=" + this + ']';
+// TODO:
+// assert !success || (initialized() && !hasPending()) : "Invalid done callback [success=" + success +
+// ", fut=" + this + ']';
if (log.isDebugEnabled())
log.debug("Received onDone(..) callback [success=" + success + ", err=" + err + ", fut=" + this + ']');
@@ -832,7 +800,8 @@ public final class GridDhtLockFuture extends GridCacheCompoundIdentityFuture<Boo
}
if (log.isDebugEnabled())
- log.debug("Mapped DHT lock future [dhtMap=" + F.nodeIds(dhtMap.keySet()) + ", dhtLockFut=" + this + ']');
+ log.debug("Mapped DHT lock future [dhtMap=" + F.nodeIds(dhtMap.keySet()) +
+ ", dhtLockFut=" + this + ']');
long timeout = inTx() ? tx.remainingTime() : this.timeout;
@@ -850,31 +819,8 @@ public final class GridDhtLockFuture extends GridCacheCompoundIdentityFuture<Boo
if (inTx() && tx.remainingTime() == -1)
return;
- MiniFuture fut = new MiniFuture(n, dhtMapping);
-
- GridDhtLockRequest req = new GridDhtLockRequest(
- cctx.cacheId(),
- nearNodeId,
- inTx() ? tx.nearXidVersion() : null,
- threadId,
- futId,
- fut.futureId(),
- lockVer,
- topVer,
- inTx(),
- read,
- isolation(),
- isInvalidate(),
- timeout,
- cnt,
- 0,
- inTx() ? tx.size() : cnt,
- inTx() ? tx.subjectId() : null,
- inTx() ? tx.taskNameHash() : 0,
- read ? accessTtl : -1L,
- skipStore,
- keepBinary,
- cctx.deploymentEnabled());
+ MiniFuture fut = null;
+ GridDhtLockRequest req = null;
try {
for (ListIterator<GridDhtCacheEntry> it = dhtMapping.listIterator(); it.hasNext();) {
@@ -906,6 +852,34 @@ public final class GridDhtLockFuture extends GridCacheCompoundIdentityFuture<Boo
if (tx != null && !needVal)
continue;
+ if (fut == null) {
+ fut = new MiniFuture(nextMiniId(), n, dhtMapping);
+
+ req = new GridDhtLockRequest(
+ cctx.cacheId(),
+ nearNodeId,
+ inTx() ? tx.nearXidVersion() : null,
+ threadId,
+ futId,
+ fut.futureId(),
+ lockVer,
+ topVer,
+ inTx(),
+ read,
+ isolation(),
+ isInvalidate(),
+ timeout,
+ cnt,
+ 0,
+ inTx() ? tx.size() : cnt,
+ inTx() ? tx.subjectId() : null,
+ inTx() ? tx.taskNameHash() : 0,
+ read ? accessTtl : -1L,
+ skipStore,
+ keepBinary,
+ cctx.deploymentEnabled());
+ }
+
boolean invalidateRdr = e.readerId(n.id()) != null;
req.addDhtKey(e.key(), invalidateRdr, cctx);
@@ -926,7 +900,7 @@ public final class GridDhtLockFuture extends GridCacheCompoundIdentityFuture<Boo
it.set(addOwned(req, e));
}
- if (!F.isEmpty(req.keys())) {
+ if (req != null && !F.isEmpty(req.keys())) {
if (tx != null)
tx.addLockTransactionNode(n);
@@ -967,6 +941,44 @@ public final class GridDhtLockFuture extends GridCacheCompoundIdentityFuture<Boo
}
/**
+ * @return Next mini ID.
+ */
+ private synchronized int nextMiniId() {
+ return miniFuts != null ? miniFuts.size() : 0;
+ }
+
+ /**
+ * Marks future as initialized.
+ */
+ private void markInitialized() {
+ boolean completed;
+
+ synchronized (this) {
+ if (inited)
+ return;
+
+ inited = true;
+
+ completed = miniFuts == null || completedMiniCnt == miniFuts.size();
+ }
+
+ if (completed)
+ onDone(true);
+ }
+
+ /**
+ * @param fut Mini future to add.
+ */
+ private synchronized void add(MiniFuture fut) {
+ if (miniFuts == null)
+ miniFuts = new ArrayList<>();
+
+ assert fut.futureId() == miniFuts.size();
+
+ miniFuts.add(fut);
+ }
+
+ /**
* @param req Request.
* @param e Entry.
* @return Entry.
@@ -1001,24 +1013,10 @@ public final class GridDhtLockFuture extends GridCacheCompoundIdentityFuture<Boo
}
/** {@inheritDoc} */
- @Override public String toString() {
- Collection<String> futs = F.viewReadOnly(futures(), new C1<IgniteInternalFuture<?>, String>() {
- @Override public String apply(IgniteInternalFuture<?> f) {
- MiniFuture m = (MiniFuture)f;
-
- return "[node=" + m.node().id() + ", loc=" + m.node().isLocal() + ", done=" + f.isDone() + "]";
- }
- });
-
- Collection<KeyCacheObject> locks;
-
- synchronized (this) {
- locks = new HashSet<>(pendingLocks);
- }
-
+ @Override public synchronized String toString() {
return S.toString(GridDhtLockFuture.class, this,
- "innerFuts", futs,
- "pendingLocks", locks,
+ "innerFuts", miniFuts,
+ "pendingLocks", pendingLocks,
"super", super.toString());
}
@@ -1123,8 +1121,6 @@ public final class GridDhtLockFuture extends GridCacheCompoundIdentityFuture<Boo
// Stop locks and responses processing.
pendingLocks.clear();
-
- clear();
}
boolean releaseLocks = !(inTx() && cctx.tm().deadlockDetectionEnabled());
@@ -1142,9 +1138,9 @@ public final class GridDhtLockFuture extends GridCacheCompoundIdentityFuture<Boo
* Mini-future for get operations. Mini-futures are only waiting on a single
* node as opposed to multiple nodes.
*/
- private class MiniFuture extends GridFutureAdapter<Boolean> {
+ private class MiniFuture {
/** */
- private final IgniteUuid futId = IgniteUuid.randomUuid();
+ private final int miniId;
/** Node. */
@GridToStringExclude
@@ -1154,13 +1150,18 @@ public final class GridDhtLockFuture extends GridCacheCompoundIdentityFuture<Boo
@GridToStringInclude
private List<GridDhtCacheEntry> dhtMapping;
+ /** */
+ private volatile boolean done;
+
/**
+ * @param miniId Mini ID.
* @param node Node.
* @param dhtMapping Mapping.
*/
- MiniFuture(ClusterNode node, List<GridDhtCacheEntry> dhtMapping) {
+ MiniFuture(int miniId, ClusterNode node, List<GridDhtCacheEntry> dhtMapping) {
assert node != null;
+ this.miniId = miniId;
this.node = node;
this.dhtMapping = dhtMapping;
}
@@ -1168,8 +1169,8 @@ public final class GridDhtLockFuture extends GridCacheCompoundIdentityFuture<Boo
/**
* @return Future ID.
*/
- IgniteUuid futureId() {
- return futId;
+ int futureId() {
+ return miniId;
}
/**
@@ -1187,7 +1188,7 @@ public final class GridDhtLockFuture extends GridCacheCompoundIdentityFuture<Boo
log.debug("Failed to get future result [fut=" + this + ", err=" + e + ']');
// Fail.
- onDone(e);
+ GridDhtLockFuture.this.onDone(e);
}
/**
@@ -1260,13 +1261,25 @@ public final class GridDhtLockFuture extends GridCacheCompoundIdentityFuture<Boo
replicate ? DR_PRELOAD : DR_NONE,
false)) {
if (rec && !entry.isInternal())
- cctx.events().addEvent(entry.partition(), entry.key(), cctx.localNodeId(),
- (IgniteUuid)null, null, EVT_CACHE_REBALANCE_OBJECT_LOADED, info.value(), true, null,
- false, null, null, null, false);
+ cctx.events().addEvent(
+ entry.partition(),
+ entry.key(),
+ cctx.localNodeId(),
+ (IgniteUuid)null,
+ null,
+ EVT_CACHE_REBALANCE_OBJECT_LOADED,
+ info.value(),
+ true,
+ null,
+ false,
+ null,
+ null,
+ null,
+ false);
}
}
catch (IgniteCheckedException e) {
- onDone(e);
+ GridDhtLockFuture.this.onDone(e);
return;
}
@@ -1282,42 +1295,31 @@ public final class GridDhtLockFuture extends GridCacheCompoundIdentityFuture<Boo
}
/**
- * @param cacheCtx Context.
- * @param keys Keys to evict readers for.
- * @param nodeId Node ID.
- * @param msgId Message ID.
- * @param entries Entries to check.
+ * @param res Result (always {@code true}).
*/
- @SuppressWarnings({"ForLoopReplaceableByForEach"})
- private void evictReaders(GridCacheContext<?, ?> cacheCtx, Collection<IgniteTxKey> keys, UUID nodeId, long msgId,
- @Nullable List<GridDhtCacheEntry> entries) {
- if (entries == null || keys == null || entries.isEmpty() || keys.isEmpty())
- return;
+ void onDone(boolean res) {
+ assert res;
- for (ListIterator<GridDhtCacheEntry> it = entries.listIterator(); it.hasNext(); ) {
- GridDhtCacheEntry cached = it.next();
+ boolean complete;
- if (keys.contains(cached.txKey())) {
- while (true) {
- try {
- cached.removeReader(nodeId, msgId);
+ synchronized (GridDhtLockFuture.this) {
+ if (done)
+ return;
- if (tx != null)
- tx.removeNearMapping(nodeId, cached);
+ done = true;
- break;
- }
- catch (GridCacheEntryRemovedException ignore) {
- GridDhtCacheEntry e = cacheCtx.dht().peekExx(cached.key());
+ complete = (++completedMiniCnt == miniFuts.size()) && inited;
+ }
- if (e == null)
- break;
+ if (complete)
+ GridDhtLockFuture.this.onDone(true);
+ }
- it.set(e);
- }
- }
- }
- }
+ /**
+ * @return {@code True} if done.
+ */
+ public boolean isDone() {
+ return done;
}
/** {@inheritDoc} */
http://git-wip-us.apache.org/repos/asf/ignite/blob/87708161/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtLockRequest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtLockRequest.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtLockRequest.java
index 1f81764..bfc67ae 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtLockRequest.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtLockRequest.java
@@ -62,7 +62,7 @@ public class GridDhtLockRequest extends GridDistributedLockRequest {
private BitSet invalidateEntries;
/** Mini future ID. */
- private IgniteUuid miniId;
+ private int miniId;
/** Owner mapped version, if any. */
@GridToStringInclude
@@ -129,7 +129,7 @@ public class GridDhtLockRequest extends GridDistributedLockRequest {
GridCacheVersion nearXidVer,
long threadId,
IgniteUuid futId,
- IgniteUuid miniId,
+ int miniId,
GridCacheVersion lockVer,
@NotNull AffinityTopologyVersion topVer,
boolean isInTx,
@@ -169,8 +169,6 @@ public class GridDhtLockRequest extends GridDistributedLockRequest {
nearKeys = nearCnt == 0 ? Collections.<KeyCacheObject>emptyList() : new ArrayList<KeyCacheObject>(nearCnt);
invalidateEntries = new BitSet(dhtCnt == 0 ? nearCnt : dhtCnt);
- assert miniId != null;
-
this.miniId = miniId;
this.subjId = subjId;
this.taskNameHash = taskNameHash;
@@ -294,7 +292,7 @@ public class GridDhtLockRequest extends GridDistributedLockRequest {
/**
* @return Mini ID.
*/
- public IgniteUuid miniId() {
+ public int miniId() {
return miniId;
}
@@ -372,7 +370,7 @@ public class GridDhtLockRequest extends GridDistributedLockRequest {
writer.incrementState();
case 22:
- if (!writer.writeIgniteUuid("miniId", miniId))
+ if (!writer.writeInt("miniId", miniId))
return false;
writer.incrementState();
@@ -452,7 +450,7 @@ public class GridDhtLockRequest extends GridDistributedLockRequest {
reader.incrementState();
case 22:
- miniId = reader.readIgniteUuid("miniId");
+ miniId = reader.readInt("miniId");
if (!reader.isLastRead())
return false;
http://git-wip-us.apache.org/repos/asf/ignite/blob/87708161/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtLockResponse.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtLockResponse.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtLockResponse.java
index ea6ca06..ee116c8 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtLockResponse.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtLockResponse.java
@@ -30,7 +30,6 @@ import org.apache.ignite.internal.processors.cache.GridCacheContext;
import org.apache.ignite.internal.processors.cache.GridCacheEntryInfo;
import org.apache.ignite.internal.processors.cache.GridCacheSharedContext;
import org.apache.ignite.internal.processors.cache.distributed.GridDistributedLockResponse;
-import org.apache.ignite.internal.processors.cache.transactions.IgniteTxKey;
import org.apache.ignite.internal.processors.cache.version.GridCacheVersion;
import org.apache.ignite.internal.util.tostring.GridToStringInclude;
import org.apache.ignite.internal.util.typedef.internal.S;
@@ -46,13 +45,8 @@ public class GridDhtLockResponse extends GridDistributedLockResponse {
/** */
private static final long serialVersionUID = 0L;
- /** Evicted readers. */
- @GridToStringInclude
- @GridDirectCollection(IgniteTxKey.class)
- private Collection<IgniteTxKey> nearEvicted;
-
/** Mini ID. */
- private IgniteUuid miniId;
+ private int miniId;
/** Invalid partitions. */
@GridToStringInclude
@@ -74,14 +68,20 @@ public class GridDhtLockResponse extends GridDistributedLockResponse {
* @param lockVer Lock version.
* @param futId Future ID.
* @param miniId Mini future ID.
+ * @param part Partition.
* @param cnt Key count.
* @param addDepInfo Deployment info.
*/
- public GridDhtLockResponse(int cacheId, GridCacheVersion lockVer, IgniteUuid futId, IgniteUuid miniId, int cnt,
- boolean addDepInfo) {
- super(cacheId, lockVer, futId, cnt, addDepInfo);
-
- assert miniId != null;
+ public GridDhtLockResponse(
+ int cacheId,
+ GridCacheVersion lockVer,
+ IgniteUuid futId,
+ int miniId,
+ int part,
+ int cnt,
+ boolean addDepInfo
+ ) {
+ super(cacheId, lockVer, futId, part, cnt, addDepInfo);
this.miniId = miniId;
}
@@ -90,36 +90,28 @@ public class GridDhtLockResponse extends GridDistributedLockResponse {
* @param lockVer Lock ID.
* @param futId Future ID.
* @param miniId Mini future ID.
+ * @param part Partition.
* @param err Error.
- * @param addDepInfo
+ * @param addDepInfo Deployment info.
*/
- public GridDhtLockResponse(int cacheId, GridCacheVersion lockVer, IgniteUuid futId, IgniteUuid miniId,
- Throwable err, boolean addDepInfo) {
- super(cacheId, lockVer, futId, err, addDepInfo);
-
- assert miniId != null;
+ public GridDhtLockResponse(
+ int cacheId,
+ GridCacheVersion lockVer,
+ IgniteUuid futId,
+ int miniId,
+ int part,
+ Throwable err,
+ boolean addDepInfo
+ ) {
+ super(cacheId, lockVer, futId, part, err, addDepInfo);
this.miniId = miniId;
}
/**
- * @return Evicted readers.
- */
- public Collection<IgniteTxKey> nearEvicted() {
- return nearEvicted;
- }
-
- /**
- * @param nearEvicted Evicted readers.
- */
- public void nearEvicted(Collection<IgniteTxKey> nearEvicted) {
- this.nearEvicted = nearEvicted;
- }
-
- /**
* @return Mini future ID.
*/
- public IgniteUuid miniId() {
+ public int miniId() {
return miniId;
}
@@ -168,11 +160,6 @@ public class GridDhtLockResponse extends GridDistributedLockResponse {
GridCacheContext cctx = ctx.cacheContext(cacheId);
- if (nearEvicted != null) {
- for (IgniteTxKey key : nearEvicted)
- key.prepareMarshal(cctx);
- }
-
if (preloadEntries != null)
marshalInfos(preloadEntries, cctx);
}
@@ -183,11 +170,6 @@ public class GridDhtLockResponse extends GridDistributedLockResponse {
GridCacheContext cctx = ctx.cacheContext(cacheId);
- if (nearEvicted != null) {
- for (IgniteTxKey key : nearEvicted)
- key.finishUnmarshal(cctx, ldr);
- }
-
if (preloadEntries != null)
unmarshalInfos(preloadEntries, ctx.cacheContext(cacheId), ldr);
}
@@ -207,20 +189,14 @@ public class GridDhtLockResponse extends GridDistributedLockResponse {
}
switch (writer.state()) {
- case 10:
- if (!writer.writeCollection("invalidParts", invalidParts, MessageCollectionItemType.INT))
- return false;
-
- writer.incrementState();
-
case 11:
- if (!writer.writeIgniteUuid("miniId", miniId))
+ if (!writer.writeCollection("invalidParts", invalidParts, MessageCollectionItemType.INT))
return false;
writer.incrementState();
case 12:
- if (!writer.writeCollection("nearEvicted", nearEvicted, MessageCollectionItemType.MSG))
+ if (!writer.writeInt("miniId", miniId))
return false;
writer.incrementState();
@@ -247,16 +223,8 @@ public class GridDhtLockResponse extends GridDistributedLockResponse {
return false;
switch (reader.state()) {
- case 10:
- invalidParts = reader.readCollection("invalidParts", MessageCollectionItemType.INT);
-
- if (!reader.isLastRead())
- return false;
-
- reader.incrementState();
-
case 11:
- miniId = reader.readIgniteUuid("miniId");
+ invalidParts = reader.readCollection("invalidParts", MessageCollectionItemType.INT);
if (!reader.isLastRead())
return false;
@@ -264,7 +232,7 @@ public class GridDhtLockResponse extends GridDistributedLockResponse {
reader.incrementState();
case 12:
- nearEvicted = reader.readCollection("nearEvicted", MessageCollectionItemType.MSG);
+ miniId = reader.readInt("miniId");
if (!reader.isLastRead())
return false;
http://git-wip-us.apache.org/repos/asf/ignite/blob/87708161/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTransactionalCacheAdapter.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTransactionalCacheAdapter.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTransactionalCacheAdapter.java
index c91eb7a..1cec45b 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTransactionalCacheAdapter.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTransactionalCacheAdapter.java
@@ -65,7 +65,6 @@ import org.apache.ignite.internal.util.GridLeanSet;
import org.apache.ignite.internal.util.future.GridFinishedFuture;
import org.apache.ignite.internal.util.lang.GridClosureException;
import org.apache.ignite.internal.util.lang.IgnitePair;
-import org.apache.ignite.internal.util.typedef.C1;
import org.apache.ignite.internal.util.typedef.C2;
import org.apache.ignite.internal.util.typedef.CI1;
import org.apache.ignite.internal.util.typedef.CI2;
@@ -432,6 +431,7 @@ public abstract class GridDhtTransactionalCacheAdapter<K, V> extends GridDhtCach
req.version(),
req.futureId(),
req.miniId(),
+ req.partition(),
e,
ctx.deploymentEnabled());
@@ -467,35 +467,30 @@ public abstract class GridDhtTransactionalCacheAdapter<K, V> extends GridDhtCach
boolean cancelled = false;
try {
- res = new GridDhtLockResponse(ctx.cacheId(), req.version(), req.futureId(), req.miniId(), cnt,
+ res = new GridDhtLockResponse(
+ ctx.cacheId(),
+ req.version(),
+ req.futureId(),
+ req.miniId(),
+ req.partition(),
+ cnt,
ctx.deploymentEnabled());
dhtTx = startRemoteTx(nodeId, req, res);
nearTx = isNearEnabled(cacheCfg) ? near().startRemoteTx(nodeId, req) : null;
-
- if (nearTx != null && !nearTx.empty())
- res.nearEvicted(nearTx.evicted());
- else {
- if (!F.isEmpty(req.nearKeys())) {
- Collection<IgniteTxKey> nearEvicted = new ArrayList<>(req.nearKeys().size());
-
- nearEvicted.addAll(F.viewReadOnly(req.nearKeys(), new C1<KeyCacheObject, IgniteTxKey>() {
- @Override public IgniteTxKey apply(KeyCacheObject k) {
- return ctx.txKey(k);
- }
- }));
-
- res.nearEvicted(nearEvicted);
- }
- }
}
catch (IgniteTxRollbackCheckedException e) {
String err = "Failed processing DHT lock request (transaction has been completed): " + req;
U.error(log, err, e);
- res = new GridDhtLockResponse(ctx.cacheId(), req.version(), req.futureId(), req.miniId(),
- new IgniteTxRollbackCheckedException(err, e), ctx.deploymentEnabled());
+ res = new GridDhtLockResponse(ctx.cacheId(),
+ req.version(),
+ req.futureId(),
+ req.miniId(),
+ req.partition(),
+ new IgniteTxRollbackCheckedException(err, e),
+ ctx.deploymentEnabled());
fail = true;
}
@@ -508,7 +503,9 @@ public abstract class GridDhtTransactionalCacheAdapter<K, V> extends GridDhtCach
req.version(),
req.futureId(),
req.miniId(),
- new IgniteCheckedException(err, e), ctx.deploymentEnabled());
+ req.partition(),
+ new IgniteCheckedException(err, e),
+ ctx.deploymentEnabled());
fail = true;
}
@@ -626,7 +623,8 @@ public abstract class GridDhtTransactionalCacheAdapter<K, V> extends GridDhtCach
ClusterNode nearNode = ctx.discovery().node(nodeId);
if (nearNode == null) {
- U.warn(txLockMsgLog, "Received near lock request from unknown node (will ignore) [txId=" + req.version() +
+ U.warn(txLockMsgLog, "Received near lock request from unknown node (will ignore) " +
+ "[txId=" + req.version() +
", inTx=" + req.inTx() +
", node=" + nodeId + ']');
@@ -743,7 +741,6 @@ public abstract class GridDhtTransactionalCacheAdapter<K, V> extends GridDhtCach
tx.threadId(),
createTtl,
accessTtl,
- filter,
skipStore,
keepBinary);
@@ -865,7 +862,6 @@ public abstract class GridDhtTransactionalCacheAdapter<K, V> extends GridDhtCach
req.threadId(),
req.createTtl(),
req.accessTtl(),
- filter,
req.skipStore(),
req.keepBinary());
@@ -1114,6 +1110,7 @@ public abstract class GridDhtTransactionalCacheAdapter<K, V> extends GridDhtCach
req.futureId(),
req.miniId(),
false,
+ req.partition(),
0,
null,
topVer,
@@ -1160,6 +1157,7 @@ public abstract class GridDhtTransactionalCacheAdapter<K, V> extends GridDhtCach
req.futureId(),
req.miniId(),
tx != null && tx.onePhaseCommit(),
+ req.partition(),
entries.size(),
err,
null,
@@ -1270,6 +1268,7 @@ public abstract class GridDhtTransactionalCacheAdapter<K, V> extends GridDhtCach
req.futureId(),
req.miniId(),
false,
+ req.partition(),
entries.size(),
e,
null,
http://git-wip-us.apache.org/repos/asf/ignite/blob/87708161/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxPrepareFuture.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxPrepareFuture.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxPrepareFuture.java
index e2b7803..8478ebd 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxPrepareFuture.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxPrepareFuture.java
@@ -43,11 +43,11 @@ import org.apache.ignite.internal.processors.cache.CacheInvokeEntry;
import org.apache.ignite.internal.processors.cache.CacheLockCandidates;
import org.apache.ignite.internal.processors.cache.CacheObject;
import org.apache.ignite.internal.processors.cache.GridCacheAdapter;
-import org.apache.ignite.internal.processors.cache.GridCacheCompoundFuture;
import org.apache.ignite.internal.processors.cache.GridCacheContext;
import org.apache.ignite.internal.processors.cache.GridCacheEntryEx;
import org.apache.ignite.internal.processors.cache.GridCacheEntryInfo;
import org.apache.ignite.internal.processors.cache.GridCacheEntryRemovedException;
+import org.apache.ignite.internal.processors.cache.GridCacheFutureAdapter;
import org.apache.ignite.internal.processors.cache.GridCacheMvccCandidate;
import org.apache.ignite.internal.processors.cache.GridCacheMvccFuture;
import org.apache.ignite.internal.processors.cache.GridCacheOperation;
@@ -82,7 +82,6 @@ import org.apache.ignite.internal.util.typedef.internal.CU;
import org.apache.ignite.internal.util.typedef.internal.S;
import org.apache.ignite.internal.util.typedef.internal.U;
import org.apache.ignite.lang.IgniteFutureCancelledException;
-import org.apache.ignite.lang.IgniteReducer;
import org.apache.ignite.lang.IgniteUuid;
import org.jetbrains.annotations.Nullable;
@@ -99,11 +98,8 @@ import static org.apache.ignite.transactions.TransactionState.PREPARED;
*
*/
@SuppressWarnings("unchecked")
-public final class GridDhtTxPrepareFuture extends GridCacheCompoundFuture<IgniteInternalTx, GridNearTxPrepareResponse>
+public final class GridDhtTxPrepareFuture extends GridCacheFutureAdapter<GridNearTxPrepareResponse>
implements GridCacheMvccFuture<GridNearTxPrepareResponse> {
- /** */
- private static final long serialVersionUID = 0L;
-
/** Logger reference. */
private static final AtomicReference<IgniteLogger> logRef = new AtomicReference<>();
@@ -111,19 +107,6 @@ public final class GridDhtTxPrepareFuture extends GridCacheCompoundFuture<Ignite
private static final AtomicReferenceFieldUpdater<GridDhtTxPrepareFuture, Throwable> ERR_UPD =
AtomicReferenceFieldUpdater.newUpdater(GridDhtTxPrepareFuture.class, Throwable.class, "err");
- /** */
- private static final IgniteReducer<IgniteInternalTx, GridNearTxPrepareResponse> REDUCER =
- new IgniteReducer<IgniteInternalTx, GridNearTxPrepareResponse>() {
- @Override public boolean collect(IgniteInternalTx e) {
- return true;
- }
-
- @Override public GridNearTxPrepareResponse reduce() {
- // Nothing to aggregate.
- return null;
- }
- };
-
/** Replied flag updater. */
private static final AtomicIntegerFieldUpdater<GridDhtTxPrepareFuture> REPLIED_UPD =
AtomicIntegerFieldUpdater.newUpdater(GridDhtTxPrepareFuture.class, "replied");
@@ -230,8 +213,6 @@ public final class GridDhtTxPrepareFuture extends GridCacheCompoundFuture<Ignite
boolean last,
boolean retVal
) {
- super(REDUCER);
-
this.cctx = cctx;
this.tx = tx;
this.dhtVerMap = dhtVerMap;
@@ -844,7 +825,8 @@ public final class GridDhtTxPrepareFuture extends GridCacheCompoundFuture<Ignite
assert F.isEmpty(tx.invalidPartitions());
GridNearTxPrepareResponse res = new GridNearTxPrepareResponse(
- -1,
+ tx.writeEntries() != null && !tx.writeEntries().isEmpty() ?
+ F.first(tx.writeEntries()).key().partition() : -1,
tx.nearXidVersion(),
tx.colocated() ? tx.xid() : tx.nearFutureId(),
nearMiniId,
@@ -1800,8 +1782,6 @@ public final class GridDhtTxPrepareFuture extends GridCacheCompoundFuture<Ignite
/** {@inheritDoc} */
@Override public void onTimeout() {
synchronized (GridDhtTxPrepareFuture.this) {
- clear();
-
lockKeys.clear();
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/87708161/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/colocated/GridDhtColocatedCache.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/colocated/GridDhtColocatedCache.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/colocated/GridDhtColocatedCache.java
index 12a3912..815800c 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/colocated/GridDhtColocatedCache.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/colocated/GridDhtColocatedCache.java
@@ -29,7 +29,6 @@ import org.apache.ignite.cluster.ClusterNode;
import org.apache.ignite.internal.IgniteInternalFuture;
import org.apache.ignite.internal.cluster.ClusterTopologyCheckedException;
import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
-import org.apache.ignite.internal.processors.cache.CacheEntryPredicate;
import org.apache.ignite.internal.processors.cache.CacheObject;
import org.apache.ignite.internal.processors.cache.CacheOperationContext;
import org.apache.ignite.internal.processors.cache.EntryGetResult;
@@ -919,7 +918,6 @@ public class GridDhtColocatedCache<K, V> extends GridDhtTransactionalCacheAdapte
* @param timeout Lock timeout.
* @param createTtl TTL for create operation.
* @param accessTtl TTL for read operation.
- * @param filter filter Optional filter.
* @param skipStore Skip store flag.
* @return Lock future.
*/
@@ -935,7 +933,6 @@ public class GridDhtColocatedCache<K, V> extends GridDhtTransactionalCacheAdapte
final long timeout,
final long createTtl,
final long accessTtl,
- @Nullable final CacheEntryPredicate[] filter,
final boolean skipStore,
final boolean keepBinary
) {
@@ -960,7 +957,6 @@ public class GridDhtColocatedCache<K, V> extends GridDhtTransactionalCacheAdapte
timeout,
createTtl,
accessTtl,
- filter,
skipStore,
keepBinary);
}
@@ -982,7 +978,6 @@ public class GridDhtColocatedCache<K, V> extends GridDhtTransactionalCacheAdapte
timeout,
createTtl,
accessTtl,
- filter,
skipStore,
keepBinary);
}
@@ -1003,7 +998,6 @@ public class GridDhtColocatedCache<K, V> extends GridDhtTransactionalCacheAdapte
* @param timeout Lock timeout.
* @param createTtl TTL for create operation.
* @param accessTtl TTL for read operation.
- * @param filter filter Optional filter.
* @param skipStore Skip store flag.
* @return Lock future.
*/
@@ -1019,9 +1013,9 @@ public class GridDhtColocatedCache<K, V> extends GridDhtTransactionalCacheAdapte
final long timeout,
final long createTtl,
final long accessTtl,
- @Nullable final CacheEntryPredicate[] filter,
boolean skipStore,
- boolean keepBinary) {
+ boolean keepBinary
+ ) {
int cnt = keys.size();
if (tx == null) {
@@ -1037,7 +1031,6 @@ public class GridDhtColocatedCache<K, V> extends GridDhtTransactionalCacheAdapte
threadId,
createTtl,
accessTtl,
- filter,
skipStore,
keepBinary);
http://git-wip-us.apache.org/repos/asf/ignite/blob/87708161/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 4442b3a..cff5e0c 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
@@ -1135,7 +1135,6 @@ public final class GridDhtColocatedLockFuture extends GridCacheCompoundIdentityF
timeout,
createTtl,
accessTtl,
- filter,
skipStore,
keepBinary);
http://git-wip-us.apache.org/repos/asf/ignite/blob/87708161/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearLockResponse.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearLockResponse.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearLockResponse.java
index e88f0a0..56bda1c 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearLockResponse.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearLockResponse.java
@@ -76,6 +76,7 @@ public class GridNearLockResponse extends GridDistributedLockResponse {
* @param futId Future ID.
* @param miniId Mini future ID.
* @param filterRes {@code True} if need to allocate array for filter evaluation results.
+ * @param part Partition.
* @param cnt Count.
* @param err Error.
* @param clientRemapVer {@code True} if client node should remap lock request.
@@ -87,12 +88,13 @@ public class GridNearLockResponse extends GridDistributedLockResponse {
IgniteUuid futId,
int miniId,
boolean filterRes,
+ int part,
int cnt,
Throwable err,
AffinityTopologyVersion clientRemapVer,
boolean addDepInfo
) {
- super(cacheId, lockVer, futId, cnt, err, addDepInfo);
+ super(cacheId, lockVer, futId, part, cnt, err, addDepInfo);
assert miniId != 0;
@@ -208,37 +210,37 @@ public class GridNearLockResponse extends GridDistributedLockResponse {
}
switch (writer.state()) {
- case 10:
+ case 11:
if (!writer.writeMessage("clientRemapVer", clientRemapVer))
return false;
writer.incrementState();
- case 11:
+ case 12:
if (!writer.writeObjectArray("dhtVers", dhtVers, MessageCollectionItemType.MSG))
return false;
writer.incrementState();
- case 12:
+ case 13:
if (!writer.writeBooleanArray("filterRes", filterRes))
return false;
writer.incrementState();
- case 13:
+ case 14:
if (!writer.writeObjectArray("mappedVers", mappedVers, MessageCollectionItemType.MSG))
return false;
writer.incrementState();
- case 14:
+ case 15:
if (!writer.writeInt("miniId", miniId))
return false;
writer.incrementState();
- case 15:
+ case 16:
if (!writer.writeCollection("pending", pending, MessageCollectionItemType.MSG))
return false;
@@ -260,7 +262,7 @@ public class GridNearLockResponse extends GridDistributedLockResponse {
return false;
switch (reader.state()) {
- case 10:
+ case 11:
clientRemapVer = reader.readMessage("clientRemapVer");
if (!reader.isLastRead())
@@ -268,7 +270,7 @@ public class GridNearLockResponse extends GridDistributedLockResponse {
reader.incrementState();
- case 11:
+ case 12:
dhtVers = reader.readObjectArray("dhtVers", MessageCollectionItemType.MSG, GridCacheVersion.class);
if (!reader.isLastRead())
@@ -276,7 +278,7 @@ public class GridNearLockResponse extends GridDistributedLockResponse {
reader.incrementState();
- case 12:
+ case 13:
filterRes = reader.readBooleanArray("filterRes");
if (!reader.isLastRead())
@@ -284,7 +286,7 @@ public class GridNearLockResponse extends GridDistributedLockResponse {
reader.incrementState();
- case 13:
+ case 14:
mappedVers = reader.readObjectArray("mappedVers", MessageCollectionItemType.MSG, GridCacheVersion.class);
if (!reader.isLastRead())
@@ -292,7 +294,7 @@ public class GridNearLockResponse extends GridDistributedLockResponse {
reader.incrementState();
- case 14:
+ case 15:
miniId = reader.readInt("miniId");
if (!reader.isLastRead())
@@ -300,7 +302,7 @@ public class GridNearLockResponse extends GridDistributedLockResponse {
reader.incrementState();
- case 15:
+ case 16:
pending = reader.readCollection("pending", MessageCollectionItemType.MSG);
if (!reader.isLastRead())
@@ -320,7 +322,7 @@ public class GridNearLockResponse extends GridDistributedLockResponse {
/** {@inheritDoc} */
@Override public byte fieldsCount() {
- return 16;
+ return 17;
}
/** {@inheritDoc} */
http://git-wip-us.apache.org/repos/asf/ignite/blob/87708161/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearPessimisticTxPrepareFuture.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearPessimisticTxPrepareFuture.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearPessimisticTxPrepareFuture.java
index e934319..faa0eea 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearPessimisticTxPrepareFuture.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearPessimisticTxPrepareFuture.java
@@ -426,7 +426,7 @@ public class GridNearPessimisticTxPrepareFuture extends GridNearTxPrepareFutureA
/**
*
*/
- private class MiniFuture extends GridFutureAdapter<GridNearTxPrepareResponse> {
+ private class MiniFuture {
/** */
private final int futId;
http://git-wip-us.apache.org/repos/asf/ignite/blob/87708161/modules/core/src/main/java/org/apache/ignite/internal/processors/job/GridJobProcessor.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/job/GridJobProcessor.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/job/GridJobProcessor.java
index e0bc4d2..3224c792 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/job/GridJobProcessor.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/job/GridJobProcessor.java
@@ -621,6 +621,7 @@ public class GridJobProcessor extends GridProcessorAdapter {
cancelPassiveJob(job);
}
}
+
for (GridJobWorker job : activeJobs.values()) {
if (idsMatch.apply(job))
cancelActiveJob(job, sys);
http://git-wip-us.apache.org/repos/asf/ignite/blob/87708161/modules/core/src/main/java/org/apache/ignite/internal/util/StripedExecutor.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/util/StripedExecutor.java b/modules/core/src/main/java/org/apache/ignite/internal/util/StripedExecutor.java
index 6c85b32..709f454 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/util/StripedExecutor.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/util/StripedExecutor.java
@@ -479,6 +479,8 @@ public class StripedExecutor implements ExecutorService {
if (cmd != null) {
active = true;
+// U.debug(log, "Running [r=" + cmd + ']');
+
try {
cmd.run();
}
[4/5] ignite git commit: Merge remote-tracking branch
'remotes/origin/master' into ignite-5293
Posted by sb...@apache.org.
Merge remote-tracking branch 'remotes/origin/master' into ignite-5293
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/fa2e7f7e
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/fa2e7f7e
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/fa2e7f7e
Branch: refs/heads/ignite-5293
Commit: fa2e7f7e25216d84bfe97b5abe9e41ed2757a7e3
Parents: dd1310f e273f2e
Author: sboikov <sb...@gridgain.com>
Authored: Mon Jun 5 10:32:59 2017 +0300
Committer: sboikov <sb...@gridgain.com>
Committed: Mon Jun 5 10:32:59 2017 +0300
----------------------------------------------------------------------
assembly/libs/README.txt | 1 +
.../jdbc2/JdbcDynamicIndexAbstractSelfTest.java | 2 +
.../jdbc/suite/IgniteJdbcDriverTestSuite.java | 40 +-
.../jdbc/thin/JdbcConnectionSelfTest.java | 195 --
.../JdbcThinAbstractDmlStatementSelfTest.java | 251 +++
.../jdbc/thin/JdbcThinAbstractSelfTest.java | 41 +
...JdbcThinAbstractUpdateStatementSelfTest.java | 40 +
.../jdbc/thin/JdbcThinComplexQuerySelfTest.java | 330 ++++
.../jdbc/thin/JdbcThinConnectionSelfTest.java | 237 +++
.../thin/JdbcThinDeleteStatementSelfTest.java | 49 +
.../JdbcThinDynamicIndexAbstractSelfTest.java | 328 ++++
...namicIndexAtomicPartitionedNearSelfTest.java | 26 +
...inDynamicIndexAtomicPartitionedSelfTest.java | 39 +
...hinDynamicIndexAtomicReplicatedSelfTest.java | 39 +
...dexTransactionalPartitionedNearSelfTest.java | 26 +
...icIndexTransactionalPartitionedSelfTest.java | 39 +
...micIndexTransactionalReplicatedSelfTest.java | 39 +
.../jdbc/thin/JdbcThinEmptyCacheSelfTest.java | 146 ++
.../thin/JdbcThinInsertStatementSelfTest.java | 185 ++
.../thin/JdbcThinMergeStatementSelfTest.java | 130 ++
.../jdbc/thin/JdbcThinMetadataSelfTest.java | 357 ++++
.../jdbc/thin/JdbcThinNoDefaultSchemaTest.java | 160 ++
.../thin/JdbcThinPreparedStatementSelfTest.java | 771 ++++++++
.../jdbc/thin/JdbcThinResultSetSelfTest.java | 762 ++++++++
.../jdbc/thin/JdbcThinStatementSelfTest.java | 343 ++++
.../thin/JdbcThinUpdateStatementSelfTest.java | 50 +
.../org/apache/ignite/IgniteJdbcDriver.java | 3 +
.../org/apache/ignite/IgniteJdbcThinDriver.java | 138 +-
.../cache/store/jdbc/CacheJdbcBlobStore.java | 5 +-
.../ignite/internal/binary/BinaryFieldImpl.java | 10 +-
.../internal/binary/BinaryObjectImpl.java | 10 +-
.../binary/BinaryObjectOffheapImpl.java | 10 +-
.../ignite/internal/binary/BinaryTypeProxy.java | 3 +
.../stream/v2/DirectByteBufferStreamImplV2.java | 1 +
.../internal/jdbc/thin/JdbcConnection.java | 529 ------
.../ignite/internal/jdbc/thin/JdbcTcpIo.java | 207 ---
.../internal/jdbc/thin/JdbcThinConnection.java | 657 +++++++
.../jdbc/thin/JdbcThinPreparedStatement.java | 485 +++++
.../internal/jdbc/thin/JdbcThinResultSet.java | 1704 ++++++++++++++++++
.../jdbc/thin/JdbcThinResultSetMetadata.java | 161 ++
.../internal/jdbc/thin/JdbcThinStatement.java | 495 +++++
.../internal/jdbc/thin/JdbcThinTcpIo.java | 365 ++++
.../internal/jdbc/thin/JdbcThinUtils.java | 208 +++
.../checkpoint/GridCheckpointManager.java | 45 +-
.../checkpoint/GridCheckpointRequest.java | 2 +
.../processors/cache/GridCacheAdapter.java | 51 +-
.../cache/GridCacheDeploymentManager.java | 12 +-
.../processors/cache/IgniteCacheFutureImpl.java | 8 +-
.../distributed/dht/GridDhtCacheAdapter.java | 6 +-
.../distributed/dht/GridDhtTxPrepareFuture.java | 9 +-
...arOptimisticSerializableTxPrepareFuture.java | 6 +-
.../query/GridCacheQueryFutureAdapter.java | 18 +-
.../cache/query/GridCacheSqlQuery.java | 23 +
.../store/GridCacheStoreManagerAdapter.java | 4 +-
.../cache/store/GridCacheWriteBehindStore.java | 13 +-
.../cache/transactions/TxDeadlock.java | 11 +-
.../odbc/OdbcQueryGetColumnsMetaRequest.java | 74 -
.../odbc/OdbcQueryGetColumnsMetaResult.java | 42 -
.../odbc/OdbcQueryGetParamsMetaRequest.java | 60 -
.../odbc/OdbcQueryGetParamsMetaResult.java | 40 -
.../odbc/OdbcQueryGetTablesMetaRequest.java | 85 -
.../odbc/OdbcQueryGetTablesMetaResult.java | 42 -
.../internal/processors/odbc/OdbcTableMeta.java | 88 -
.../odbc/SqlListenerAbstractMessageParser.java | 265 ---
.../odbc/SqlListenerAbstractObjectReader.java | 137 --
.../odbc/SqlListenerAbstractObjectWriter.java | 111 --
.../processors/odbc/SqlListenerColumnMeta.java | 110 --
.../processors/odbc/SqlListenerNioListener.java | 27 +-
.../processors/odbc/SqlListenerProcessor.java | 1 +
.../odbc/SqlListenerQueryCloseRequest.java | 49 -
.../odbc/SqlListenerQueryCloseResult.java | 40 -
.../odbc/SqlListenerQueryExecuteRequest.java | 78 -
.../odbc/SqlListenerQueryExecuteResult.java | 54 -
.../odbc/SqlListenerQueryFetchRequest.java | 61 -
.../odbc/SqlListenerQueryFetchResult.java | 66 -
.../processors/odbc/SqlListenerRequest.java | 37 +-
.../odbc/SqlListenerRequestHandler.java | 8 +
.../odbc/SqlListenerRequestHandlerImpl.java | 494 -----
.../processors/odbc/SqlListenerResponse.java | 42 +-
.../processors/odbc/SqlListenerUtils.java | 250 +++
.../processors/odbc/jdbc/JdbcColumnMeta.java | 129 ++
.../processors/odbc/jdbc/JdbcMessageParser.java | 53 +-
.../processors/odbc/jdbc/JdbcObjectReader.java | 33 -
.../processors/odbc/jdbc/JdbcObjectWriter.java | 33 -
.../odbc/jdbc/JdbcQueryCloseRequest.java | 72 +
.../processors/odbc/jdbc/JdbcQueryCursor.java | 132 ++
.../odbc/jdbc/JdbcQueryExecuteRequest.java | 147 ++
.../odbc/jdbc/JdbcQueryExecuteResult.java | 150 ++
.../odbc/jdbc/JdbcQueryFetchRequest.java | 87 +
.../odbc/jdbc/JdbcQueryFetchResult.java | 84 +
.../odbc/jdbc/JdbcQueryMetadataRequest.java | 73 +
.../odbc/jdbc/JdbcQueryMetadataResult.java | 93 +
.../odbc/jdbc/JdbcRawBinarylizable.java | 44 +
.../processors/odbc/jdbc/JdbcRequest.java | 108 ++
.../odbc/jdbc/JdbcRequestHandler.java | 285 +++
.../processors/odbc/jdbc/JdbcResponse.java | 104 ++
.../processors/odbc/jdbc/JdbcResult.java | 91 +
.../processors/odbc/jdbc/JdbcUtils.java | 73 +
.../processors/odbc/odbc/OdbcColumnMeta.java | 111 ++
.../processors/odbc/odbc/OdbcMessageParser.java | 213 ++-
.../processors/odbc/odbc/OdbcObjectReader.java | 33 -
.../processors/odbc/odbc/OdbcObjectWriter.java | 32 -
.../odbc/odbc/OdbcQueryCloseRequest.java | 49 +
.../odbc/odbc/OdbcQueryCloseResult.java | 40 +
.../odbc/odbc/OdbcQueryExecuteRequest.java | 78 +
.../odbc/odbc/OdbcQueryExecuteResult.java | 54 +
.../odbc/odbc/OdbcQueryFetchRequest.java | 61 +
.../odbc/odbc/OdbcQueryFetchResult.java | 66 +
.../odbc/OdbcQueryGetColumnsMetaRequest.java | 74 +
.../odbc/OdbcQueryGetColumnsMetaResult.java | 42 +
.../odbc/OdbcQueryGetParamsMetaRequest.java | 60 +
.../odbc/odbc/OdbcQueryGetParamsMetaResult.java | 40 +
.../odbc/OdbcQueryGetTablesMetaRequest.java | 85 +
.../odbc/odbc/OdbcQueryGetTablesMetaResult.java | 42 +
.../processors/odbc/odbc/OdbcRequest.java | 63 +
.../odbc/odbc/OdbcRequestHandler.java | 505 ++++++
.../processors/odbc/odbc/OdbcResponse.java | 69 +
.../processors/odbc/odbc/OdbcTableMeta.java | 89 +
.../processors/query/GridQueryProcessor.java | 57 +-
.../internal/processors/query/QueryUtils.java | 6 +
.../resource/GridResourceProcessor.java | 9 +-
.../handlers/query/QueryCommandHandler.java | 7 +-
.../processors/task/GridTaskWorker.java | 4 +-
.../util/future/AsyncFutureListener.java | 57 +
.../internal/util/future/IgniteFutureImpl.java | 33 +-
.../util/tostring/GridToStringBuilder.java | 419 +++++
.../util/tostring/GridToStringThreadLocal.java | 6 +-
.../org/apache/ignite/lang/IgniteFuture.java | 24 +-
.../internal/SensitiveInfoTestLoggerProxy.java | 493 +++++
.../SensitiveInfoTestLoggerProxy_Exclude.txt | 20 +
.../SensitiveInfoTestLoggerProxy_Include.txt | 18 +
.../binary/BinaryFieldExtractionSelfTest.java | 52 +
.../binary/BinaryMarshallerSelfTest.java | 36 +-
.../query/GridQueryCommandHandlerTest.java | 3 +-
.../util/future/IgniteCacheFutureImplTest.java | 46 +
.../util/future/IgniteFutureImplTest.java | 466 ++++-
.../tostring/IncludeSensitiveAtomicTest.java | 29 +
.../util/tostring/IncludeSensitiveTest.java | 149 ++
.../IncludeSensitiveTransactionalTest.java | 43 +
.../junits/IgniteTestResources.java | 11 +-
.../testsuites/IgniteLangSelfTestSuite.java | 3 +
modules/flink/pom.xml | 2 +-
.../query/h2/H2IndexingAbstractGeoSelfTest.java | 27 +-
.../cache/query/CacheQueryPartitionInfo.java | 110 ++
.../cache/query/GridCacheTwoStepQuery.java | 18 +
.../processors/query/h2/IgniteH2Indexing.java | 69 +-
.../query/h2/ddl/DdlStatementsProcessor.java | 43 +-
.../query/h2/sql/GridSqlCreateTable.java | 48 +-
.../query/h2/sql/GridSqlQueryParser.java | 105 +-
.../query/h2/sql/GridSqlQuerySplitter.java | 228 ++-
.../query/h2/twostep/GridMapQueryExecutor.java | 75 +-
.../h2/twostep/GridReduceQueryExecutor.java | 28 +-
.../h2/twostep/msg/GridH2QueryRequest.java | 37 +-
.../cache/IgniteCacheAbstractQuerySelfTest.java | 4 +
.../DynamicIndexAbstractBasicSelfTest.java | 22 +-
.../cache/index/H2DynamicTableSelfTest.java | 222 ++-
.../processors/query/IgniteSqlRoutingTest.java | 552 ++++++
.../query/IgniteSqlSchemaIndexingTest.java | 3 +-
.../processors/query/SqlSchemaSelfTest.java | 238 +++
.../query/h2/sql/GridQueryParsingTest.java | 23 +-
.../IgniteCacheQuerySelfTestSuite.java | 6 +-
.../impls/storage/matrix/MapWrapperStorage.java | 17 +
.../ml/math/impls/vector/MapWrapperVector.java | 17 +
.../ml/clustering/ClusteringTesetSuite.java | 17 +
.../Apache.Ignite.Core.Tests.NuGet/NuGet.config | 2 +-
.../StartupTest.cs | 1 -
.../install-package.ps1 | 31 +-
.../Apache.Ignite.Core.Tests.TestDll.csproj | 1 +
.../TestExtensions.cs | 35 +
.../Apache.Ignite.Core.Tests.csproj | 2 +
.../Binary/BinarySelfTest.cs | 2 +-
.../Binary/EnumsTest.cs | 299 ++-
.../Binary/EnumsTestOnline.cs | 45 +
.../Deployment/PeerAssemblyLoadingTest.cs | 13 +
.../Deployment/RuntimeDependencyFunc.cs | 34 +
.../Impl/Binary/BinaryReader.cs | 9 +-
.../Impl/Binary/BinaryReflectiveActions.cs | 11 +-
.../Impl/Binary/BinaryUtils.cs | 2 +
.../Impl/Binary/BinaryWriter.cs | 4 +-
.../Impl/Compute/ComputeJobHolder.cs | 6 +-
.../Impl/Deployment/GetAssemblyFunc.cs | 28 +
.../Impl/Deployment/PeerAssemblyResolver.cs | 26 +-
.../Apache.Ignite.Core/Impl/ExceptionUtils.cs | 1 +
modules/platforms/dotnet/build.ps1 | 5 +-
.../processors/schedule/ScheduleFutureImpl.java | 51 +-
.../schedule/GridScheduleSelfTest.java | 84 +-
modules/web-console/backend/app/mongo.js | 79 +-
modules/web-console/backend/services/caches.js | 2 +
modules/web-console/backend/services/igfss.js | 5 +
modules/web-console/frontend/app/app.config.js | 3 +-
modules/web-console/frontend/app/app.js | 12 +-
.../cluster-select/cluster-select.pug | 2 +-
.../list-of-registered-users.column-defs.js | 2 +-
.../app/components/version-picker/component.js | 52 +
.../app/components/version-picker/index.js | 25 +
.../app/components/version-picker/style.scss | 41 +
.../app/components/version-picker/template.pug | 33 +
.../frontend/app/data/event-groups.json | 14 +
.../ui-ace-docker/ui-ace-docker.controller.js | 8 +-
.../ui-ace-java/ui-ace-java.controller.js | 29 +-
.../ui-ace-java/ui-ace-java.directive.js | 12 +-
.../ui-ace-pom/ui-ace-pom.controller.js | 8 +-
.../ui-ace-spring/ui-ace-spring.controller.js | 29 +-
.../ui-ace-spring/ui-ace-spring.directive.js | 12 +-
.../frontend/app/helpers/jade/mixins.pug | 4 +-
.../app/modules/agent/AgentManager.service.js | 15 +-
.../frontend/app/modules/cluster/Cache.js | 4 +
.../app/modules/cluster/CacheMetrics.js | 4 +
.../modules/configuration/Version.service.js | 180 +-
.../generator/AbstractTransformer.js | 110 +-
.../generator/ConfigurationGenerator.js | 570 ++++--
.../configuration/generator/Docker.service.js | 12 +-
.../generator/JavaTransformer.service.js | 32 +-
.../configuration/generator/Maven.service.js | 38 +-
.../generator/PlatformGenerator.js | 19 +-
.../generator/SpringTransformer.service.js | 23 +-
.../defaults/Cache.platform.service.js | 9 +
.../generator/defaults/Cache.service.js | 23 +-
.../generator/defaults/Cluster.service.js | 46 +-
.../generator/defaults/IGFS.service.js | 9 +-
.../frontend/app/modules/sql/sql.controller.js | 6 +-
.../app/modules/states/configuration.state.js | 21 +-
.../states/configuration/caches/affinity.pug | 23 +-
.../states/configuration/caches/concurrency.pug | 14 +
.../states/configuration/caches/general.pug | 71 +-
.../states/configuration/caches/memory.pug | 99 +-
.../states/configuration/caches/node-filter.pug | 27 +-
.../states/configuration/caches/query.pug | 25 +-
.../states/configuration/caches/store.pug | 5 +
.../configuration/clusters/deployment.pug | 8 +-
.../states/configuration/clusters/discovery.pug | 14 +-
.../states/configuration/clusters/events.pug | 26 +-
.../states/configuration/clusters/failover.pug | 9 +
.../states/configuration/clusters/general.pug | 2 +-
.../states/configuration/clusters/hadoop.pug | 119 ++
.../configuration/clusters/marshaller.pug | 37 +-
.../states/configuration/clusters/memory.pug | 114 ++
.../states/configuration/clusters/metrics.pug | 2 +-
.../states/configuration/clusters/misc.pug | 60 +
.../states/configuration/clusters/service.pug | 88 +
.../states/configuration/clusters/swap.pug | 72 +
.../states/configuration/clusters/thread.pug | 54 +-
.../states/configuration/clusters/time.pug | 13 +-
.../states/configuration/domains/query.pug | 2 +-
.../modules/states/configuration/igfs/dual.pug | 42 +
.../modules/states/configuration/igfs/misc.pug | 25 +-
.../summary/summary-zipper.service.js | 4 +-
.../configuration/summary/summary.controller.js | 2 +-
.../configuration/summary/summary.worker.js | 13 +-
.../frontend/app/primitives/btn-group/index.pug | 12 +-
.../frontend/app/primitives/btn/index.scss | 6 +
.../frontend/app/primitives/dropdown/index.pug | 3 +-
.../frontend/app/primitives/dropdown/index.scss | 19 +-
.../frontend/controllers/caches-controller.js | 71 +-
.../frontend/controllers/clusters-controller.js | 190 +-
.../frontend/controllers/domains-controller.js | 7 +-
.../frontend/controllers/igfs-controller.js | 9 +-
.../frontend/public/stylesheets/style.scss | 19 +-
.../frontend/test/unit/Version.test.js | 95 +-
.../frontend/views/configuration/caches.tpl.pug | 2 +-
.../views/configuration/clusters.tpl.pug | 13 +-
.../views/configuration/domains.tpl.pug | 2 +-
.../frontend/views/configuration/igfs.tpl.pug | 6 +-
.../views/configuration/sidebar.tpl.pug | 4 +-
.../frontend/views/includes/header-left.pug | 3 -
.../frontend/views/includes/header-right.pug | 1 -
.../frontend/views/settings/admin.tpl.pug | 2 +-
.../frontend/views/settings/profile.tpl.pug | 2 +-
.../web-console/frontend/views/signin.tpl.pug | 6 +-
.../web-console/frontend/views/sql/sql.tpl.pug | 8 +-
.../frontend/webpack/webpack.common.js | 8 +-
.../frontend/webpack/webpack.dev.babel.js | 5 +-
.../console/agent/handlers/ClusterListener.java | 12 +-
273 files changed, 19133 insertions(+), 4112 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ignite/blob/fa2e7f7e/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxPrepareFuture.java
----------------------------------------------------------------------
[3/5] ignite git commit: wip
Posted by sb...@apache.org.
wip
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/dd1310f6
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/dd1310f6
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/dd1310f6
Branch: refs/heads/ignite-5293
Commit: dd1310f664194d9948eac5a6483e71163821746a
Parents: 87a79f1
Author: Yakov Zhdanov <yz...@gridgain.com>
Authored: Fri Jun 2 17:27:05 2017 +0300
Committer: Yakov Zhdanov <yz...@gridgain.com>
Committed: Fri Jun 2 17:27:05 2017 +0300
----------------------------------------------------------------------
.../distributed/dht/GridDhtLockFuture.java | 19 +-
.../distributed/dht/GridDhtTxFinishFuture.java | 185 ++++++++++++-------
.../distributed/dht/GridDhtTxPrepareFuture.java | 169 ++++++++++-------
3 files changed, 236 insertions(+), 137 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ignite/blob/dd1310f6/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtLockFuture.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtLockFuture.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtLockFuture.java
index 9365259..a6cec4f 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtLockFuture.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtLockFuture.java
@@ -477,15 +477,20 @@ public final class GridDhtLockFuture extends GridCacheFutureAdapter<Boolean>
@Override public boolean onNodeLeft(UUID nodeId) {
boolean found = false;
+ List<MiniFuture> miniFuts0;
+
synchronized (this) {
- if (miniFuts != null) {
- for (MiniFuture miniFut : miniFuts) {
- if (miniFut.node().id().equals(nodeId)) {
- miniFut.onResult();
+ if (miniFuts == null)
+ return false;
- found = true;
- }
- }
+ miniFuts0 = new ArrayList<>(miniFuts);
+ }
+
+ for (MiniFuture miniFut : miniFuts0) {
+ if (miniFut.node().id().equals(nodeId)) {
+ miniFut.onResult();
+
+ found = true;
}
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/dd1310f6/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 8a31bac..98442bf 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
@@ -19,6 +19,7 @@ package org.apache.ignite.internal.processors.cache.distributed.dht;
import java.util.ArrayList;
import java.util.Collection;
+import java.util.List;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.atomic.AtomicReference;
@@ -26,19 +27,16 @@ import java.util.concurrent.atomic.AtomicReferenceFieldUpdater;
import org.apache.ignite.IgniteCheckedException;
import org.apache.ignite.IgniteLogger;
import org.apache.ignite.cluster.ClusterNode;
-import org.apache.ignite.internal.IgniteInternalFuture;
import org.apache.ignite.internal.NodeStoppingException;
import org.apache.ignite.internal.cluster.ClusterTopologyCheckedException;
-import org.apache.ignite.internal.processors.cache.GridCacheCompoundIdentityFuture;
import org.apache.ignite.internal.processors.cache.GridCacheFuture;
+import org.apache.ignite.internal.processors.cache.GridCacheFutureAdapter;
import org.apache.ignite.internal.processors.cache.GridCacheSharedContext;
import org.apache.ignite.internal.processors.cache.distributed.GridDistributedTxMapping;
import org.apache.ignite.internal.processors.cache.transactions.IgniteInternalTx;
import org.apache.ignite.internal.processors.cache.transactions.IgniteTxEntry;
-import org.apache.ignite.internal.util.future.GridFutureAdapter;
import org.apache.ignite.internal.util.tostring.GridToStringExclude;
import org.apache.ignite.internal.util.tostring.GridToStringInclude;
-import org.apache.ignite.internal.util.typedef.C1;
import org.apache.ignite.internal.util.typedef.F;
import org.apache.ignite.internal.util.typedef.X;
import org.apache.ignite.internal.util.typedef.internal.S;
@@ -52,11 +50,9 @@ import static org.apache.ignite.transactions.TransactionState.COMMITTING;
/**
*
*/
-public final class GridDhtTxFinishFuture<K, V> extends GridCacheCompoundIdentityFuture<IgniteInternalTx>
+public final class GridDhtTxFinishFuture<K, V>
+ extends GridCacheFutureAdapter<IgniteInternalTx>
implements GridCacheFuture<IgniteInternalTx> {
- /** */
- private static final long serialVersionUID = 0L;
-
/** Logger reference. */
private static final AtomicReference<IgniteLogger> logRef = new AtomicReference<>();
@@ -94,14 +90,21 @@ public final class GridDhtTxFinishFuture<K, V> extends GridCacheCompoundIdentity
/** Near mappings. */
private Map<UUID, GridDistributedTxMapping> nearMap;
+ /** */
+ private List<MiniFuture> miniFuts;
+
+ /** */
+ private int completedMiniCnt;
+
+ /** */
+ private volatile boolean inited;
+
/**
* @param cctx Context.
* @param tx Transaction.
* @param commit Commit flag.
*/
public GridDhtTxFinishFuture(GridCacheSharedContext<K, V> cctx, GridDhtTxLocalAdapter tx, boolean commit) {
- super(F.<IgniteInternalTx>identityReducer(tx));
-
this.cctx = cctx;
this.tx = tx;
this.commit = commit;
@@ -118,6 +121,18 @@ public final class GridDhtTxFinishFuture<K, V> extends GridCacheCompoundIdentity
}
/**
+ * @param fut Mini future to add.
+ */
+ private synchronized void add(MiniFuture fut) {
+ if (miniFuts == null)
+ miniFuts = new ArrayList<>();
+
+ assert fut.futureId() == miniFuts.size();
+
+ miniFuts.add(fut);
+ }
+
+ /**
* @return Transaction.
*/
public GridDhtTxLocalAdapter tx() {
@@ -130,18 +145,23 @@ public final class GridDhtTxFinishFuture<K, V> extends GridCacheCompoundIdentity
}
/** {@inheritDoc} */
- @SuppressWarnings("unchecked")
@Override public boolean onNodeLeft(UUID nodeId) {
- for (IgniteInternalFuture<?> fut : futures())
- if (isMini(fut)) {
- MiniFuture f = (MiniFuture)fut;
+ List<MiniFuture> miniFuts0;
- if (f.node().id().equals(nodeId)) {
- f.onNodeLeft();
+ synchronized (this) {
+ if (miniFuts == null)
+ return false;
- return true;
- }
+ miniFuts0 = new ArrayList<>(miniFuts);
+ }
+
+ for (MiniFuture miniFut : miniFuts0) {
+ if (miniFut.node().id().equals(nodeId)) {
+ miniFut.onNodeLeft();
+
+ return true;
}
+ }
return false;
}
@@ -173,25 +193,37 @@ public final class GridDhtTxFinishFuture<K, V> extends GridCacheCompoundIdentity
}
/**
+ * Finds pending mini future by the given mini ID.
+ *
+ * @param miniId Mini ID to find.
+ * @return Mini future.
+ */
+ @SuppressWarnings("ForLoopReplaceableByForEach")
+ private MiniFuture miniFuture(int miniId) {
+ synchronized (this) {
+ if (miniFuts != null && !miniFuts.get(miniId).isDone())
+ return miniFuts.get(miniId);
+ }
+
+ return null;
+ }
+
+ /**
* @param nodeId Sender.
* @param res Result.
*/
public void onResult(UUID nodeId, GridDhtTxFinishResponse res) {
if (!isDone()) {
- boolean found = false;
+ boolean found;
- for (IgniteInternalFuture<IgniteInternalTx> fut : futures()) {
- if (isMini(fut)) {
- MiniFuture f = (MiniFuture)fut;
+ MiniFuture mini = miniFuture(res.miniId());
- if (f.futureId() == res.miniId()) {
- found = true;
+ if (mini == null)
+ found = false;
+ else {
+ found = true;
- assert f.node().id().equals(nodeId);
-
- f.onResult(res);
- }
- }
+ mini.onResult(res);
}
if (!found) {
@@ -217,7 +249,7 @@ public final class GridDhtTxFinishFuture<K, V> extends GridCacheCompoundIdentity
/** {@inheritDoc} */
@Override public boolean onDone(IgniteInternalTx tx, Throwable err) {
- if (initialized() || err != null) {
+ if (inited || err != null) {
Throwable e = this.err;
if (this.tx.onePhaseCommit() && (this.tx.state() == COMMITTING)) {
@@ -255,14 +287,6 @@ public final class GridDhtTxFinishFuture<K, V> extends GridCacheCompoundIdentity
}
/**
- * @param f Future.
- * @return {@code True} if mini-future.
- */
- private boolean isMini(IgniteInternalFuture<?> f) {
- return f.getClass().equals(MiniFuture.class);
- }
-
- /**
* Completeness callback.
*/
private void onComplete() {
@@ -314,7 +338,7 @@ public final class GridDhtTxFinishFuture<K, V> extends GridCacheCompoundIdentity
for (ClusterNode n : nodes) {
assert !n.isLocal();
- MiniFuture fut = new MiniFuture(++miniId, n);
+ MiniFuture fut = new MiniFuture(miniId++, n);
add(fut); // Append new future.
@@ -361,7 +385,7 @@ public final class GridDhtTxFinishFuture<K, V> extends GridCacheCompoundIdentity
catch (IgniteCheckedException e) {
// Fail the whole thing.
if (e instanceof ClusterTopologyCheckedException)
- fut.onNodeLeft((ClusterTopologyCheckedException)e);
+ fut.onNodeLeft();
else {
if (msgLog.isDebugEnabled()) {
msgLog.debug("DHT finish fut, failed to send request lock tx [txId=" + tx.nearXidVersion() +
@@ -379,6 +403,25 @@ public final class GridDhtTxFinishFuture<K, V> extends GridCacheCompoundIdentity
}
/**
+ * Marks future as initialized.
+ */
+ private void markInitialized() {
+ boolean completed;
+
+ synchronized (this) {
+ if (inited)
+ return;
+
+ inited = true;
+
+ completed = miniFuts == null || completedMiniCnt == miniFuts.size();
+ }
+
+ if (completed)
+ onDone();
+ }
+
+ /**
* @param commit Commit flag.
* @param dhtMap DHT map.
* @param nearMap Near map.
@@ -411,7 +454,7 @@ public final class GridDhtTxFinishFuture<K, V> extends GridCacheCompoundIdentity
// Nothing to send.
continue;
- MiniFuture fut = new MiniFuture(++miniId, dhtMapping, nearMapping);
+ MiniFuture fut = new MiniFuture(miniId++, dhtMapping, nearMapping);
add(fut); // Append new future.
@@ -466,7 +509,7 @@ public final class GridDhtTxFinishFuture<K, V> extends GridCacheCompoundIdentity
catch (IgniteCheckedException e) {
// Fail the whole thing.
if (e instanceof ClusterTopologyCheckedException)
- fut.onNodeLeft((ClusterTopologyCheckedException)e);
+ fut.onNodeLeft();
else {
if (msgLog.isDebugEnabled()) {
msgLog.debug("DHT finish fut, failed to send request dht [txId=" + tx.nearXidVersion() +
@@ -535,7 +578,7 @@ public final class GridDhtTxFinishFuture<K, V> extends GridCacheCompoundIdentity
catch (IgniteCheckedException e) {
// Fail the whole thing.
if (e instanceof ClusterTopologyCheckedException)
- fut.onNodeLeft((ClusterTopologyCheckedException)e);
+ fut.onNodeLeft();
else {
if (msgLog.isDebugEnabled()) {
msgLog.debug("DHT finish fut, failed to send request near [txId=" + tx.nearXidVersion() +
@@ -554,19 +597,10 @@ public final class GridDhtTxFinishFuture<K, V> extends GridCacheCompoundIdentity
}
/** {@inheritDoc} */
- @Override public String toString() {
- Collection<String> futs = F.viewReadOnly(futures(), new C1<IgniteInternalFuture<?>, String>() {
- @SuppressWarnings("unchecked")
- @Override public String apply(IgniteInternalFuture<?> f) {
- return "[node=" + ((MiniFuture)f).node().id() +
- ", loc=" + ((MiniFuture)f).node().isLocal() +
- ", done=" + f.isDone() + "]";
- }
- });
-
+ @Override public synchronized String toString() {
return S.toString(GridDhtTxFinishFuture.class, this,
"xidVer", tx.xidVersion(),
- "innerFuts", futs,
+ "innerFuts", miniFuts,
"super", super.toString());
}
@@ -574,7 +608,7 @@ public final class GridDhtTxFinishFuture<K, V> extends GridCacheCompoundIdentity
* Mini-future for get operations. Mini-futures are only waiting on a single
* node as opposed to multiple nodes.
*/
- private class MiniFuture extends GridFutureAdapter<IgniteInternalTx> {
+ private class MiniFuture {
/** */
private final int futId;
@@ -590,6 +624,9 @@ public final class GridDhtTxFinishFuture<K, V> extends GridCacheCompoundIdentity
@GridToStringInclude
private ClusterNode node;
+ /** */
+ private volatile boolean done;
+
/**
* @param futId Future ID.
* @param node Node.
@@ -634,14 +671,7 @@ public final class GridDhtTxFinishFuture<K, V> extends GridCacheCompoundIdentity
log.debug("Failed to get future result [fut=" + this + ", err=" + e + ']');
// Fail.
- onDone(e);
- }
-
- /**
- * @param e Node failure.
- */
- void onNodeLeft(ClusterTopologyCheckedException e) {
- onNodeLeft();
+ GridDhtTxFinishFuture.this.onDone(e);
}
/**
@@ -654,7 +684,7 @@ public final class GridDhtTxFinishFuture<K, V> extends GridCacheCompoundIdentity
}
// If node left, then there is nothing to commit on it.
- onDone(tx);
+ onDone();
}
/**
@@ -667,6 +697,33 @@ public final class GridDhtTxFinishFuture<K, V> extends GridCacheCompoundIdentity
onDone();
}
+ /**
+ * @return {@code True} if done.
+ */
+ public boolean isDone() {
+ return done;
+ }
+
+ /**
+ *
+ */
+ public void onDone() {
+ boolean complete;
+
+ synchronized (GridDhtTxFinishFuture.this) {
+ if (done)
+ return;
+
+ done = true;
+
+ complete = (++completedMiniCnt == miniFuts.size()) && inited;
+ }
+
+ if (complete)
+ GridDhtTxFinishFuture.this.onDone(tx);
+ }
+
+
/** {@inheritDoc} */
@Override public String toString() {
return S.toString(MiniFuture.class, this, "done", isDone(), "cancelled", isCancelled(), "err", error());
http://git-wip-us.apache.org/repos/asf/ignite/blob/dd1310f6/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxPrepareFuture.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxPrepareFuture.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxPrepareFuture.java
index 02df645..ec2bd06 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxPrepareFuture.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxPrepareFuture.java
@@ -72,7 +72,6 @@ import org.apache.ignite.internal.util.future.GridCompoundFuture;
import org.apache.ignite.internal.util.lang.IgnitePair;
import org.apache.ignite.internal.util.tostring.GridToStringExclude;
import org.apache.ignite.internal.util.tostring.GridToStringInclude;
-import org.apache.ignite.internal.util.typedef.C1;
import org.apache.ignite.internal.util.typedef.CI1;
import org.apache.ignite.internal.util.typedef.CIX1;
import org.apache.ignite.internal.util.typedef.F;
@@ -195,6 +194,15 @@ public final class GridDhtTxPrepareFuture
/** Timeout object. */
private final PrepareTimeoutObject timeoutObj;
+ /** */
+ private List<MiniFuture> miniFuts;
+
+ /** */
+ private int completedMiniCnt;
+
+ /** */
+ private boolean inited;
+
/**
* @param cctx Context.
* @param tx Transaction.
@@ -238,6 +246,37 @@ public final class GridDhtTxPrepareFuture
timeoutObj = timeout > 0 ? new PrepareTimeoutObject(timeout) : null;
}
+ /**
+ * Marks future as initialized.
+ */
+ private void markInitialized() {
+ boolean completed;
+
+ synchronized (this) {
+ if (inited)
+ return;
+
+ inited = true;
+
+ completed = miniFuts == null || completedMiniCnt == miniFuts.size();
+ }
+
+ if (completed)
+ onDone();
+ }
+
+ /**
+ * @param fut Mini future to add.
+ */
+ private synchronized void add(MiniFuture fut) {
+ if (miniFuts == null)
+ miniFuts = new ArrayList<>();
+
+ assert fut.futureId() == miniFuts.size();
+
+ miniFuts.add(fut);
+ }
+
/** {@inheritDoc} */
@Override public IgniteUuid futureId() {
return futId;
@@ -300,16 +339,22 @@ public final class GridDhtTxPrepareFuture
/** {@inheritDoc} */
@Override public boolean onNodeLeft(UUID nodeId) {
- for (IgniteInternalFuture<?> fut : futures())
- if (isMini(fut)) {
- MiniFuture f = (MiniFuture)fut;
+ List<MiniFuture> miniFuts0;
- if (f.node().id().equals(nodeId)) {
- f.onNodeLeft();
+ synchronized (this) {
+ if (miniFuts == null)
+ return false;
- return true;
- }
+ miniFuts0 = new ArrayList<>(miniFuts);
+ }
+
+ for (MiniFuture miniFut : miniFuts0) {
+ if (miniFut.node().id().equals(nodeId)) {
+ miniFut.onNodeLeft();
+
+ return true;
}
+ }
return false;
}
@@ -354,7 +399,8 @@ public final class GridDhtTxPrepareFuture
if (readOld) {
boolean readThrough = !txEntry.skipStore() &&
- (txEntry.op() == TRANSFORM || ((retVal || hasFilters) && cacheCtx.config().isLoadPreviousValue()));
+ (txEntry.op() == TRANSFORM || ((retVal || hasFilters) &&
+ cacheCtx.config().isLoadPreviousValue()));
boolean evt = retVal || txEntry.op() == TRANSFORM;
@@ -519,11 +565,8 @@ public final class GridDhtTxPrepareFuture
MiniFuture mini = miniFuture(res.miniId());
- if (mini != null) {
- assert mini.node().id().equals(nodeId);
-
+ if (mini != null)
mini.onResult(res);
- }
else {
if (msgLog.isDebugEnabled()) {
msgLog.debug("DHT prepare fut, failed to find mini future [txId=" + tx.nearXidVersion() +
@@ -543,26 +586,9 @@ public final class GridDhtTxPrepareFuture
*/
@SuppressWarnings("ForLoopReplaceableByForEach")
private MiniFuture miniFuture(int miniId) {
- // We iterate directly over the futs collection here to avoid copy.
synchronized (this) {
- int size = futuresCountNoLock();
-
- // Avoid iterator creation.
- for (int i = 0; i < size; i++) {
- IgniteInternalFuture<IgniteInternalTx> fut = future(i);
-
- if (!isMini(fut))
- continue;
-
- MiniFuture mini = (MiniFuture)fut;
-
- if (mini.futureId() == miniId) {
- if (!mini.isDone())
- return mini;
- else
- return null;
- }
- }
+ if (miniFuts != null && !miniFuts.get(miniId).isDone())
+ return miniFuts.get(miniId);
}
return null;
@@ -671,8 +697,8 @@ public final class GridDhtTxPrepareFuture
/** {@inheritDoc} */
@Override public boolean onDone(GridNearTxPrepareResponse res0, Throwable err) {
- assert err != null || (initialized() && !hasPending()) : "On done called for prepare future that has " +
- "pending mini futures: " + this;
+// TODO assert err != null || (initialized() && !hasPending()) : "On done called for prepare future that has " +
+// "pending mini futures: " + this;
ERR_UPD.compareAndSet(this, null, err);
@@ -926,20 +952,11 @@ public final class GridDhtTxPrepareFuture
}
/**
- * @param f Future.
- * @return {@code True} if mini-future.
- */
- private boolean isMini(IgniteInternalFuture<?> f) {
- return f.getClass().equals(MiniFuture.class);
- }
-
- /**
* Completeness callback.
*
* @param res Response.
- * @return {@code True} if {@code done} flag was changed as a result of this call.
*/
- private boolean onComplete(@Nullable GridNearTxPrepareResponse res) {
+ private void onComplete(@Nullable GridNearTxPrepareResponse res) {
if (last || tx.isSystemInvalidate())
tx.state(PREPARED);
@@ -949,11 +966,7 @@ public final class GridDhtTxPrepareFuture
if (timeoutObj != null)
cctx.time().removeTimeoutObject(timeoutObj);
-
- return true;
}
-
- return false;
}
/**
@@ -1232,7 +1245,7 @@ public final class GridDhtTxPrepareFuture
if (tx.remainingTime() == -1)
return;
- MiniFuture fut = new MiniFuture(n.id(), ++miniId, dhtMapping, nearMapping);
+ MiniFuture fut = new MiniFuture(n.id(), miniId++, dhtMapping, nearMapping);
add(fut); // Append new future.
@@ -1374,11 +1387,11 @@ public final class GridDhtTxPrepareFuture
GridCacheMvccCandidate added = entry.cached().candidate(version());
assert added != null : "Null candidate for non-group-lock entry " +
- "[added=" + added + ", entry=" + entry + ']';
+ "[added=null, entry=" + entry + ']';
assert added.dhtLocal() : "Got non-dht-local candidate for prepare future" +
"[added=" + added + ", entry=" + entry + ']';
- if (added != null && added.ownerVersion() != null)
+ if (added.ownerVersion() != null)
req.owned(entry.txKey(), added.ownerVersion());
}
@@ -1544,18 +1557,10 @@ public final class GridDhtTxPrepareFuture
}
/** {@inheritDoc} */
- @Override public String toString() {
- Collection<String> futs = F.viewReadOnly(futures(), new C1<IgniteInternalFuture<?>, String>() {
- @Override public String apply(IgniteInternalFuture<?> f) {
- return "[node=" + ((MiniFuture)f).node().id() +
- ", loc=" + ((MiniFuture)f).node().isLocal() +
- ", done=" + f.isDone() + "]";
- }
- });
-
+ @Override public synchronized String toString() {
return S.toString(GridDhtTxPrepareFuture.class, this,
"xid", tx.xidVersion(),
- "innerFuts", futs,
+ "innerFuts", miniFuts,
"super", super.toString());
}
@@ -1578,6 +1583,9 @@ public final class GridDhtTxPrepareFuture
@GridToStringInclude
private GridDistributedTxMapping nearMapping;
+ /** */
+ private volatile boolean done;
+
/**
* @param nodeId Node ID.
* @param futId Future ID.
@@ -1620,7 +1628,7 @@ public final class GridDhtTxPrepareFuture
log.debug("Failed to get future result [fut=" + this + ", err=" + e + ']');
// Fail.
- onDone(e);
+ GridDhtTxPrepareFuture.this.onDone(e);
}
/**
@@ -1635,7 +1643,7 @@ public final class GridDhtTxPrepareFuture
if (tx != null)
tx.removeMapping(nodeId);
- onDone(tx);
+ onDone();
}
/**
@@ -1752,7 +1760,7 @@ public final class GridDhtTxPrepareFuture
}
catch (IgniteCheckedException e) {
// Fail the whole thing.
- onDone(e);
+ GridDhtTxPrepareFuture.this.onDone(e);
return;
}
@@ -1765,13 +1773,42 @@ public final class GridDhtTxPrepareFuture
}
// Finish mini future.
- onDone(tx);
+ onDone();
}
}
+ /**
+ * @return {@code True} if done.
+ */
+ public boolean isDone() {
+ return done;
+ }
+
+ /**
+ *
+ */
+ public void onDone() {
+ boolean complete;
+
+ synchronized (GridDhtTxPrepareFuture.this) {
+ if (done)
+ return;
+
+ done = true;
+
+ complete = (++completedMiniCnt == miniFuts.size()) && inited;
+ }
+
+ if (complete)
+ GridDhtTxPrepareFuture.this.onDone();
+ }
+
/** {@inheritDoc} */
@Override public String toString() {
- return S.toString(MiniFuture.class, this, "done", isDone(), "cancelled", isCancelled(), "err", error());
+ return S.toString(MiniFuture.class, this,
+ "done", isDone(),
+ "cancelled", isCancelled(),
+ "err", error());
}
}
[5/5] ignite git commit: ignite-5293
Posted by sb...@apache.org.
ignite-5293
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/282d0d6d
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/282d0d6d
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/282d0d6d
Branch: refs/heads/ignite-5293
Commit: 282d0d6d1e0b91cc690e8d1dd7e8be753220ad17
Parents: fa2e7f7
Author: sboikov <sb...@gridgain.com>
Authored: Mon Jun 5 11:40:47 2017 +0300
Committer: sboikov <sb...@gridgain.com>
Committed: Mon Jun 5 11:56:49 2017 +0300
----------------------------------------------------------------------
.../ignite/codegen/MessageCodeGenerator.java | 20 +++---
.../GridDistributedLockResponse.java | 1 +
.../distributed/dht/GridDhtLockFuture.java | 49 +++++++++-----
.../distributed/dht/GridDhtLockResponse.java | 2 +
.../distributed/dht/GridDhtTxFinishFuture.java | 28 ++++----
.../distributed/dht/GridDhtTxFinishRequest.java | 2 +-
.../dht/GridDhtTxFinishResponse.java | 2 +-
.../distributed/dht/GridDhtTxPrepareFuture.java | 68 +++++++++++---------
.../dht/GridDhtTxPrepareRequest.java | 2 +-
.../dht/GridDhtTxPrepareResponse.java | 2 +-
.../dht/colocated/GridDhtColocatedCache.java | 2 +
.../ignite/internal/util/StripedExecutor.java | 2 -
12 files changed, 104 insertions(+), 76 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ignite/blob/282d0d6d/modules/codegen/src/main/java/org/apache/ignite/codegen/MessageCodeGenerator.java
----------------------------------------------------------------------
diff --git a/modules/codegen/src/main/java/org/apache/ignite/codegen/MessageCodeGenerator.java b/modules/codegen/src/main/java/org/apache/ignite/codegen/MessageCodeGenerator.java
index 24f5087..99ec08a 100644
--- a/modules/codegen/src/main/java/org/apache/ignite/codegen/MessageCodeGenerator.java
+++ b/modules/codegen/src/main/java/org/apache/ignite/codegen/MessageCodeGenerator.java
@@ -44,12 +44,8 @@ import org.apache.ignite.internal.GridDirectCollection;
import org.apache.ignite.internal.GridDirectMap;
import org.apache.ignite.internal.GridDirectTransient;
import org.apache.ignite.internal.IgniteCodeGeneratingFail;
-import org.apache.ignite.internal.processors.cache.distributed.GridDistributedLockRequest;
-import org.apache.ignite.internal.processors.cache.distributed.GridDistributedLockResponse;
-import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtLockRequest;
-import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtLockResponse;
-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.version.GridCacheVersion;
+import org.apache.ignite.internal.processors.cache.version.GridCacheVersionEx;
import org.apache.ignite.internal.util.typedef.internal.SB;
import org.apache.ignite.internal.util.typedef.internal.U;
import org.apache.ignite.lang.IgniteUuid;
@@ -181,12 +177,12 @@ public class MessageCodeGenerator {
// gen.generateAndWrite(GridMessageCollection.class);
// gen.generateAndWrite(DataStreamerEntry.class);
- gen.generateAndWrite(GridDistributedLockRequest.class);
- gen.generateAndWrite(GridDistributedLockResponse.class);
- gen.generateAndWrite(GridNearLockRequest.class);
- gen.generateAndWrite(GridNearLockResponse.class);
- gen.generateAndWrite(GridDhtLockRequest.class);
- gen.generateAndWrite(GridDhtLockResponse.class);
+// gen.generateAndWrite(GridDistributedLockRequest.class);
+// gen.generateAndWrite(GridDistributedLockResponse.class);
+// gen.generateAndWrite(GridNearLockRequest.class);
+// gen.generateAndWrite(GridNearLockResponse.class);
+// gen.generateAndWrite(GridDhtLockRequest.class);
+// gen.generateAndWrite(GridDhtLockResponse.class);
//
// gen.generateAndWrite(GridDistributedTxPrepareRequest.class);
// gen.generateAndWrite(GridDistributedTxPrepareResponse.class);
http://git-wip-us.apache.org/repos/asf/ignite/blob/282d0d6d/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridDistributedLockResponse.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridDistributedLockResponse.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridDistributedLockResponse.java
index eb65239..a6ac426 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridDistributedLockResponse.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridDistributedLockResponse.java
@@ -74,6 +74,7 @@ public class GridDistributedLockResponse extends GridDistributedBaseMessage {
* @param cacheId Cache ID.
* @param lockVer Lock version.
* @param futId Future ID.
+ * @param part Partition.
* @param cnt Key count.
* @param addDepInfo Deployment info.
*/
http://git-wip-us.apache.org/repos/asf/ignite/blob/282d0d6d/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtLockFuture.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtLockFuture.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtLockFuture.java
index a6cec4f..8bb0def 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtLockFuture.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtLockFuture.java
@@ -80,6 +80,9 @@ public final class GridDhtLockFuture extends GridCacheFutureAdapter<Boolean>
/** Logger reference. */
private static final AtomicReference<IgniteLogger> logRef = new AtomicReference<>();
+ /** */
+ private static final List<MiniFuture> CLEARED = Collections.emptyList();
+
/** Logger. */
private static IgniteLogger log;
@@ -184,7 +187,9 @@ public final class GridDhtLockFuture extends GridCacheFutureAdapter<Boolean>
* @param tx Transaction.
* @param threadId Thread ID.
* @param accessTtl TTL for read operation.
+ * @param createTtl TTL for create operation.
* @param skipStore Skip store flag.
+ * @param keepBinary Keep binary flag.
*/
public GridDhtLockFuture(
GridCacheContext<?, ?> cctx,
@@ -475,26 +480,30 @@ public final class GridDhtLockFuture extends GridCacheFutureAdapter<Boolean>
*/
@SuppressWarnings({"ThrowableInstanceNeverThrown"})
@Override public boolean onNodeLeft(UUID nodeId) {
- boolean found = false;
-
- List<MiniFuture> miniFuts0;
+ MiniFuture miniFut0 = null;
synchronized (this) {
if (miniFuts == null)
return false;
- miniFuts0 = new ArrayList<>(miniFuts);
- }
+ for (int i = 0; i < miniFuts.size(); i++) {
+ MiniFuture miniFut = miniFuts.get(i);
- for (MiniFuture miniFut : miniFuts0) {
- if (miniFut.node().id().equals(nodeId)) {
- miniFut.onResult();
+ if (miniFut.node().id().equals(nodeId)) {
+ miniFut0 = miniFut;
- found = true;
+ break;
+ }
}
}
- return found;
+ if (miniFut0 != null) {
+ miniFut0.onResult();
+
+ return true;
+ }
+
+ return false;
}
/**
@@ -657,9 +666,6 @@ public final class GridDhtLockFuture extends GridCacheFutureAdapter<Boolean>
}
assert err == null || !success;
-// TODO:
-// assert !success || (initialized() && !hasPending()) : "Invalid done callback [success=" + success +
-// ", fut=" + this + ']';
if (log.isDebugEnabled())
log.debug("Received onDone(..) callback [success=" + success + ", err=" + err + ", fut=" + this + ']');
@@ -909,7 +915,8 @@ public final class GridDhtLockFuture extends GridCacheFutureAdapter<Boolean>
if (tx != null)
tx.addLockTransactionNode(n);
- add(fut); // Append new future.
+ if (!add(fut))
+ return;
cctx.io().send(n, req, cctx.ioPolicy());
@@ -959,7 +966,7 @@ public final class GridDhtLockFuture extends GridCacheFutureAdapter<Boolean>
boolean completed;
synchronized (this) {
- if (inited)
+ if (inited || timedOut)
return;
inited = true;
@@ -973,14 +980,20 @@ public final class GridDhtLockFuture extends GridCacheFutureAdapter<Boolean>
/**
* @param fut Mini future to add.
+ * @return {@code False} if future is already completed on timeout.
*/
- private synchronized void add(MiniFuture fut) {
+ private synchronized boolean add(MiniFuture fut) {
+ if (miniFuts == CLEARED)
+ return false;
+
if (miniFuts == null)
miniFuts = new ArrayList<>();
assert fut.futureId() == miniFuts.size();
miniFuts.add(fut);
+
+ return true;
}
/**
@@ -1126,6 +1139,8 @@ public final class GridDhtLockFuture extends GridCacheFutureAdapter<Boolean>
// Stop locks and responses processing.
pendingLocks.clear();
+
+ miniFuts = CLEARED;
}
boolean releaseLocks = !(inTx() && cctx.tm().deadlockDetectionEnabled());
@@ -1329,7 +1344,7 @@ public final class GridDhtLockFuture extends GridCacheFutureAdapter<Boolean>
/** {@inheritDoc} */
@Override public String toString() {
- return S.toString(MiniFuture.class, this, "nodeId", node.id(), "super", super.toString());
+ return "MiniFuture [node=" + node.id() + ", loc=" + node.isLocal() + ", done=" + done + ']';
}
}
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/282d0d6d/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtLockResponse.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtLockResponse.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtLockResponse.java
index ee116c8..19b3f97 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtLockResponse.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtLockResponse.java
@@ -65,6 +65,7 @@ public class GridDhtLockResponse extends GridDistributedLockResponse {
}
/**
+ * @param cacheId Cache ID.
* @param lockVer Lock version.
* @param futId Future ID.
* @param miniId Mini future ID.
@@ -87,6 +88,7 @@ public class GridDhtLockResponse extends GridDistributedLockResponse {
}
/**
+ * @param cacheId Cache ID.
* @param lockVer Lock ID.
* @param futId Future ID.
* @param miniId Mini future ID.
http://git-wip-us.apache.org/repos/asf/ignite/blob/282d0d6d/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 98442bf..31cec19 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
@@ -50,8 +50,7 @@ import static org.apache.ignite.transactions.TransactionState.COMMITTING;
/**
*
*/
-public final class GridDhtTxFinishFuture<K, V>
- extends GridCacheFutureAdapter<IgniteInternalTx>
+public final class GridDhtTxFinishFuture<K, V> extends GridCacheFutureAdapter<IgniteInternalTx>
implements GridCacheFuture<IgniteInternalTx> {
/** Logger reference. */
private static final AtomicReference<IgniteLogger> logRef = new AtomicReference<>();
@@ -146,23 +145,29 @@ public final class GridDhtTxFinishFuture<K, V>
/** {@inheritDoc} */
@Override public boolean onNodeLeft(UUID nodeId) {
- List<MiniFuture> miniFuts0;
+ MiniFuture miniFut0 = null;
synchronized (this) {
if (miniFuts == null)
return false;
- miniFuts0 = new ArrayList<>(miniFuts);
- }
+ for (int i = 0; i < miniFuts.size(); i++) {
+ MiniFuture miniFut = miniFuts.get(i);
- for (MiniFuture miniFut : miniFuts0) {
- if (miniFut.node().id().equals(nodeId)) {
- miniFut.onNodeLeft();
+ if (miniFut.node().id().equals(nodeId)) {
+ miniFut0 = miniFut;
- return true;
+ break;
+ }
}
}
+ if (miniFut0 != null) {
+ miniFut0.onNodeLeft();
+
+ return true;
+ }
+
return false;
}
@@ -198,7 +203,6 @@ public final class GridDhtTxFinishFuture<K, V>
* @param miniId Mini ID to find.
* @return Mini future.
*/
- @SuppressWarnings("ForLoopReplaceableByForEach")
private MiniFuture miniFuture(int miniId) {
synchronized (this) {
if (miniFuts != null && !miniFuts.get(miniId).isDone())
@@ -529,7 +533,7 @@ public final class GridDhtTxFinishFuture<K, V>
// Nothing to send.
continue;
- MiniFuture fut = new MiniFuture(++miniId, null, nearMapping);
+ MiniFuture fut = new MiniFuture(miniId++, null, nearMapping);
add(fut); // Append new future.
@@ -726,7 +730,7 @@ public final class GridDhtTxFinishFuture<K, V>
/** {@inheritDoc} */
@Override public String toString() {
- return S.toString(MiniFuture.class, this, "done", isDone(), "cancelled", isCancelled(), "err", error());
+ return "MiniFuture [node=" + node.id() + ", loc=" + node.isLocal() + ", done=" + done + ']';
}
}
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/282d0d6d/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxFinishRequest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxFinishRequest.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxFinishRequest.java
index 90f3687..f74533d 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxFinishRequest.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxFinishRequest.java
@@ -142,7 +142,7 @@ public class GridDhtTxFinishRequest extends GridDistributedTxFinishRequest {
txSize,
addDepInfo);
- assert miniId != 0;
+ assert miniId >= 0;
assert nearNodeId != null;
assert isolation != null;
http://git-wip-us.apache.org/repos/asf/ignite/blob/282d0d6d/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxFinishResponse.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxFinishResponse.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxFinishResponse.java
index d777a22..67be2dd 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxFinishResponse.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxFinishResponse.java
@@ -71,7 +71,7 @@ public class GridDhtTxFinishResponse extends GridDistributedTxFinishResponse {
public GridDhtTxFinishResponse(int part, GridCacheVersion xid, IgniteUuid futId, int miniId) {
super(part, xid, futId);
- assert miniId != 0;
+ assert miniId >= 0;
this.miniId = miniId;
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/282d0d6d/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxPrepareFuture.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxPrepareFuture.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxPrepareFuture.java
index d764ea5..474219d 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxPrepareFuture.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxPrepareFuture.java
@@ -19,6 +19,7 @@ package org.apache.ignite.internal.processors.cache.distributed.dht;
import java.util.ArrayList;
import java.util.Collection;
+import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
@@ -96,12 +97,14 @@ import static org.apache.ignite.transactions.TransactionState.PREPARED;
*
*/
@SuppressWarnings("unchecked")
-public final class GridDhtTxPrepareFuture
- extends GridCacheFutureAdapter<GridNearTxPrepareResponse>
+public final class GridDhtTxPrepareFuture extends GridCacheFutureAdapter<GridNearTxPrepareResponse>
implements GridCacheMvccFuture<GridNearTxPrepareResponse> {
/** Logger reference. */
private static final AtomicReference<IgniteLogger> logRef = new AtomicReference<>();
+ /** */
+ private static final List<MiniFuture> CLEARED = Collections.emptyList();
+
/** Error updater. */
private static final AtomicReferenceFieldUpdater<GridDhtTxPrepareFuture, Throwable> ERR_UPD =
AtomicReferenceFieldUpdater.newUpdater(GridDhtTxPrepareFuture.class, Throwable.class, "err");
@@ -149,10 +152,10 @@ public final class GridDhtTxPrepareFuture
private volatile int mapped;
/** Prepare reads. */
- private Iterable<IgniteTxEntry> reads;
+ private Collection<IgniteTxEntry> reads;
/** Prepare writes. */
- private Iterable<IgniteTxEntry> writes;
+ private Collection<IgniteTxEntry> writes;
/** Tx nodes. */
private Map<UUID, Collection<UUID>> txNodes;
@@ -253,7 +256,7 @@ public final class GridDhtTxPrepareFuture
boolean completed;
synchronized (this) {
- if (inited)
+ if (inited || miniFuts == CLEARED)
return;
inited = true;
@@ -267,14 +270,20 @@ public final class GridDhtTxPrepareFuture
/**
* @param fut Mini future to add.
+ * @return {@code False} if future is already completed on timeout.
*/
- private synchronized void add(MiniFuture fut) {
+ private synchronized boolean add(MiniFuture fut) {
+ if (miniFuts == CLEARED)
+ return false;
+
if (miniFuts == null)
miniFuts = new ArrayList<>();
assert fut.futureId() == miniFuts.size();
miniFuts.add(fut);
+
+ return true;
}
/** {@inheritDoc} */
@@ -339,23 +348,29 @@ public final class GridDhtTxPrepareFuture
/** {@inheritDoc} */
@Override public boolean onNodeLeft(UUID nodeId) {
- List<MiniFuture> miniFuts0;
+ MiniFuture miniFut0 = null;
synchronized (this) {
if (miniFuts == null)
return false;
- miniFuts0 = new ArrayList<>(miniFuts);
- }
+ for (int i = 0; i < miniFuts.size(); i++) {
+ MiniFuture miniFut = miniFuts.get(i);
- for (MiniFuture miniFut : miniFuts0) {
- if (miniFut.node().id().equals(nodeId)) {
- miniFut.onNodeLeft();
+ if (miniFut.node().id().equals(nodeId)) {
+ miniFut0 = miniFut;
- return true;
+ break;
+ }
}
}
+ if (miniFut0 != null) {
+ miniFut0.onNodeLeft();
+
+ return true;
+ }
+
return false;
}
@@ -400,7 +415,7 @@ public final class GridDhtTxPrepareFuture
if (readOld) {
boolean readThrough = !txEntry.skipStore() &&
(txEntry.op() == TRANSFORM || ((retVal || hasFilters) &&
- cacheCtx.config().isLoadPreviousValue()));
+ cacheCtx.config().isLoadPreviousValue()));
boolean evt = retVal || txEntry.op() == TRANSFORM;
@@ -584,7 +599,6 @@ public final class GridDhtTxPrepareFuture
* @param miniId Mini ID to find.
* @return Mini future.
*/
- @SuppressWarnings("ForLoopReplaceableByForEach")
private MiniFuture miniFuture(int miniId) {
synchronized (this) {
if (miniFuts != null && !miniFuts.get(miniId).isDone())
@@ -697,9 +711,6 @@ public final class GridDhtTxPrepareFuture
/** {@inheritDoc} */
@Override public boolean onDone(GridNearTxPrepareResponse res0, Throwable err) {
-// TODO assert err != null || (initialized() && !hasPending()) : "On done called for prepare future that has " +
-// "pending mini futures: " + this;
-
ERR_UPD.compareAndSet(this, null, err);
// Must clear prepare future before response is sent or listeners are notified.
@@ -851,8 +862,7 @@ public final class GridDhtTxPrepareFuture
assert F.isEmpty(tx.invalidPartitions());
GridNearTxPrepareResponse res = new GridNearTxPrepareResponse(
- tx.writeEntries() != null && !tx.writeEntries().isEmpty() ?
- F.first(tx.writeEntries()).key().partition() : -1,
+ writes != null && !writes.isEmpty() ? F.first(writes).key().partition() : -1,
tx.nearXidVersion(),
tx.colocated() ? tx.xid() : tx.nearFutureId(),
nearMiniId,
@@ -1250,7 +1260,8 @@ public final class GridDhtTxPrepareFuture
MiniFuture fut = new MiniFuture(n.id(), miniId++, dhtMapping, nearMapping);
- add(fut); // Append new future.
+ if (!add(fut))
+ return;
assert txNodes != null;
@@ -1362,9 +1373,10 @@ public final class GridDhtTxPrepareFuture
if (tx.remainingTime() == -1)
return;
- MiniFuture fut = new MiniFuture(nearMapping.primary().id(), ++miniId, null, nearMapping);
+ MiniFuture fut = new MiniFuture(nearMapping.primary().id(), miniId++, null, nearMapping);
- add(fut); // Append new future.
+ if (!add(fut))
+ return;
GridDhtTxPrepareRequest req = new GridDhtTxPrepareRequest(
futId,
@@ -1568,8 +1580,7 @@ public final class GridDhtTxPrepareFuture
}
/**
- * Mini-future for get operations. Mini-futures are only waiting on a single
- * node as opposed to multiple nodes.
+ *
*/
private class MiniFuture {
/** */
@@ -1808,10 +1819,7 @@ public final class GridDhtTxPrepareFuture
/** {@inheritDoc} */
@Override public String toString() {
- return S.toString(MiniFuture.class, this,
- "done", isDone(),
- "cancelled", isCancelled(),
- "err", error());
+ return "MiniFuture [node=" + nodeId + ", loc=" + node().isLocal() + ", done=" + done + ']';
}
}
@@ -1834,6 +1842,8 @@ public final class GridDhtTxPrepareFuture
/** {@inheritDoc} */
@Override public void onTimeout() {
synchronized (GridDhtTxPrepareFuture.this) {
+ miniFuts = CLEARED;
+
lockKeys.clear();
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/282d0d6d/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxPrepareRequest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxPrepareRequest.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxPrepareRequest.java
index c7c4280..8c1f74d 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxPrepareRequest.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxPrepareRequest.java
@@ -145,7 +145,7 @@ public class GridDhtTxPrepareRequest extends GridDistributedTxPrepareRequest {
addDepInfo);
assert futId != null;
- assert miniId != 0;
+ assert miniId >= 0;
this.topVer = topVer;
this.futId = futId;
http://git-wip-us.apache.org/repos/asf/ignite/blob/282d0d6d/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxPrepareResponse.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxPrepareResponse.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxPrepareResponse.java
index 5dcb98c..aa74e5e 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxPrepareResponse.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxPrepareResponse.java
@@ -91,7 +91,7 @@ public class GridDhtTxPrepareResponse extends GridDistributedTxPrepareResponse {
super(part, xid, addDepInfo);
assert futId != null;
- assert miniId != 0;
+ assert miniId >= 0;
this.futId = futId;
this.miniId = miniId;
http://git-wip-us.apache.org/repos/asf/ignite/blob/282d0d6d/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/colocated/GridDhtColocatedCache.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/colocated/GridDhtColocatedCache.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/colocated/GridDhtColocatedCache.java
index 815800c..84d3ea1 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/colocated/GridDhtColocatedCache.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/colocated/GridDhtColocatedCache.java
@@ -919,6 +919,7 @@ public class GridDhtColocatedCache<K, V> extends GridDhtTransactionalCacheAdapte
* @param createTtl TTL for create operation.
* @param accessTtl TTL for read operation.
* @param skipStore Skip store flag.
+ * @param keepBinary Keep binary flag.
* @return Lock future.
*/
IgniteInternalFuture<Exception> lockAllAsync(
@@ -999,6 +1000,7 @@ public class GridDhtColocatedCache<K, V> extends GridDhtTransactionalCacheAdapte
* @param createTtl TTL for create operation.
* @param accessTtl TTL for read operation.
* @param skipStore Skip store flag.
+ * @param keepBinary Keep binary flag.
* @return Lock future.
*/
private IgniteInternalFuture<Exception> lockAllAsync0(
http://git-wip-us.apache.org/repos/asf/ignite/blob/282d0d6d/modules/core/src/main/java/org/apache/ignite/internal/util/StripedExecutor.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/util/StripedExecutor.java b/modules/core/src/main/java/org/apache/ignite/internal/util/StripedExecutor.java
index 709f454..6c85b32 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/util/StripedExecutor.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/util/StripedExecutor.java
@@ -479,8 +479,6 @@ public class StripedExecutor implements ExecutorService {
if (cmd != null) {
active = true;
-// U.debug(log, "Running [r=" + cmd + ']');
-
try {
cmd.run();
}
[2/5] ignite git commit: wip
Posted by sb...@apache.org.
wip
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/87a79f1f
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/87a79f1f
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/87a79f1f
Branch: refs/heads/ignite-5293
Commit: 87a79f1fd4c59ba7eb4deba7a83a0b88f50a03f5
Parents: 8770816
Author: Yakov Zhdanov <yz...@gridgain.com>
Authored: Thu Jun 1 20:15:51 2017 +0300
Committer: Yakov Zhdanov <yz...@gridgain.com>
Committed: Thu Jun 1 20:15:51 2017 +0300
----------------------------------------------------------------------
.../distributed/dht/GridDhtTxPrepareFuture.java | 24 +++++++++++++++-----
.../GridNearPessimisticTxPrepareFuture.java | 2 +-
2 files changed, 19 insertions(+), 7 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ignite/blob/87a79f1f/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxPrepareFuture.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxPrepareFuture.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxPrepareFuture.java
index 8478ebd..02df645 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxPrepareFuture.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxPrepareFuture.java
@@ -69,7 +69,6 @@ import org.apache.ignite.internal.transactions.IgniteTxOptimisticCheckedExceptio
import org.apache.ignite.internal.transactions.IgniteTxTimeoutCheckedException;
import org.apache.ignite.internal.util.GridLeanSet;
import org.apache.ignite.internal.util.future.GridCompoundFuture;
-import org.apache.ignite.internal.util.future.GridFutureAdapter;
import org.apache.ignite.internal.util.lang.IgnitePair;
import org.apache.ignite.internal.util.tostring.GridToStringExclude;
import org.apache.ignite.internal.util.tostring.GridToStringInclude;
@@ -98,7 +97,8 @@ import static org.apache.ignite.transactions.TransactionState.PREPARED;
*
*/
@SuppressWarnings("unchecked")
-public final class GridDhtTxPrepareFuture extends GridCacheFutureAdapter<GridNearTxPrepareResponse>
+public final class GridDhtTxPrepareFuture
+ extends GridCacheFutureAdapter<GridNearTxPrepareResponse>
implements GridCacheMvccFuture<GridNearTxPrepareResponse> {
/** Logger reference. */
private static final AtomicReference<IgniteLogger> logRef = new AtomicReference<>();
@@ -1563,7 +1563,7 @@ public final class GridDhtTxPrepareFuture extends GridCacheFutureAdapter<GridNea
* Mini-future for get operations. Mini-futures are only waiting on a single
* node as opposed to multiple nodes.
*/
- private class MiniFuture extends GridFutureAdapter<IgniteInternalTx> {
+ private class MiniFuture {
/** */
private final int futId;
@@ -1728,9 +1728,21 @@ public final class GridDhtTxPrepareFuture extends GridCacheFutureAdapter<GridNea
drType,
false)) {
if (rec && !entry.isInternal())
- cacheCtx.events().addEvent(entry.partition(), entry.key(), cctx.localNodeId(),
- (IgniteUuid)null, null, EVT_CACHE_REBALANCE_OBJECT_LOADED, info.value(), true, null,
- false, null, null, null, false);
+ cacheCtx.events().addEvent(
+ entry.partition(),
+ entry.key(),
+ cctx.localNodeId(),
+ (IgniteUuid)null,
+ null,
+ EVT_CACHE_REBALANCE_OBJECT_LOADED,
+ info.value(),
+ true,
+ null,
+ false,
+ null,
+ null,
+ null,
+ false);
if (retVal && !invoke)
ret.value(cacheCtx, info.value(), false);
http://git-wip-us.apache.org/repos/asf/ignite/blob/87a79f1f/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearPessimisticTxPrepareFuture.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearPessimisticTxPrepareFuture.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearPessimisticTxPrepareFuture.java
index faa0eea..e934319 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearPessimisticTxPrepareFuture.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearPessimisticTxPrepareFuture.java
@@ -426,7 +426,7 @@ public class GridNearPessimisticTxPrepareFuture extends GridNearTxPrepareFutureA
/**
*
*/
- private class MiniFuture {
+ private class MiniFuture extends GridFutureAdapter<GridNearTxPrepareResponse> {
/** */
private final int futId;