You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by sb...@apache.org on 2015/11/19 15:29:17 UTC
[1/3] ignite git commit: Optimization for single key cache 'get'
operation.
Repository: ignite
Updated Branches:
refs/heads/ignite-1.5 ba1d563ae -> 1f1030670
http://git-wip-us.apache.org/repos/asf/ignite/blob/1f103067/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearSingleGetResponse.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearSingleGetResponse.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearSingleGetResponse.java
new file mode 100644
index 0000000..ba0081c
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearSingleGetResponse.java
@@ -0,0 +1,321 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.internal.processors.cache.distributed.near;
+
+import java.nio.ByteBuffer;
+import org.apache.ignite.IgniteCheckedException;
+import org.apache.ignite.internal.GridDirectTransient;
+import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
+import org.apache.ignite.internal.processors.cache.CacheObject;
+import org.apache.ignite.internal.processors.cache.GridCacheContext;
+import org.apache.ignite.internal.processors.cache.GridCacheDeployable;
+import org.apache.ignite.internal.processors.cache.GridCacheEntryInfo;
+import org.apache.ignite.internal.processors.cache.GridCacheMessage;
+import org.apache.ignite.internal.processors.cache.GridCacheSharedContext;
+import org.apache.ignite.internal.util.typedef.internal.S;
+import org.apache.ignite.lang.IgniteUuid;
+import org.apache.ignite.plugin.extensions.communication.Message;
+import org.apache.ignite.plugin.extensions.communication.MessageReader;
+import org.apache.ignite.plugin.extensions.communication.MessageWriter;
+import org.jetbrains.annotations.Nullable;
+
+/**
+ *
+ */
+public class GridNearSingleGetResponse extends GridCacheMessage implements GridCacheDeployable {
+ /** */
+ private static final long serialVersionUID = 0L;
+
+ /** */
+ public static final int INVALID_PART_FLAG_MASK = 0x1;
+
+ /** */
+ public static final int CONTAINS_VAL_FLAG_MASK = 0x2;
+
+ /** Future ID. */
+ private IgniteUuid futId;
+
+ /** */
+ private Message res;
+
+ /** */
+ private AffinityTopologyVersion topVer;
+
+ /** Error. */
+ @GridDirectTransient
+ private IgniteCheckedException err;
+
+ /** Serialized error. */
+ private byte[] errBytes;
+
+ /** */
+ private byte flags;
+
+ /**
+ * Empty constructor required for {@link Message}.
+ */
+ public GridNearSingleGetResponse() {
+ // No-op.
+ }
+
+ /**
+ * @param cacheId Cache ID.
+ * @param futId Future ID.
+ * @param topVer Topology version.
+ * @param res Result.
+ * @param invalidPartitions {@code True} if invalid partitions error occurred.
+ * @param addDepInfo Deployment info.
+ */
+ public GridNearSingleGetResponse(
+ int cacheId,
+ IgniteUuid futId,
+ AffinityTopologyVersion topVer,
+ @Nullable Message res,
+ boolean invalidPartitions,
+ boolean addDepInfo
+ ) {
+ assert futId != null;
+
+ this.cacheId = cacheId;
+ this.futId = futId;
+ this.topVer = topVer;
+ this.res = res;
+ this.addDepInfo = addDepInfo;
+
+ if (invalidPartitions)
+ flags = (byte)(flags | INVALID_PART_FLAG_MASK);
+ }
+
+ /**
+ * @param err Error.
+ */
+ public void error(IgniteCheckedException err) {
+ this.err = err;
+ }
+
+ /** {@inheritDoc} */
+ @Override public IgniteCheckedException error() {
+ return err;
+ }
+
+ /**
+ * @return Topology version.
+ */
+ @Override public AffinityTopologyVersion topologyVersion() {
+ return topVer;
+ }
+
+ /**
+ * @return {@code True} if invalid partitions error occurred.
+ */
+ public boolean invalidPartitions() {
+ return (flags & INVALID_PART_FLAG_MASK) != 0;
+ }
+
+ /**
+ * @return Results for request with set flag {@link GridNearSingleGetRequest#skipValues()}.
+ */
+ public boolean containsValue() {
+ return (flags & CONTAINS_VAL_FLAG_MASK) != 0;
+ }
+
+ /**
+ *
+ */
+ public void setContainsValue() {
+ flags = (byte)(flags | CONTAINS_VAL_FLAG_MASK);
+ }
+
+ /**
+ * @return Result.
+ */
+ public Message result() {
+ return res;
+ }
+
+ /**
+ * @return Future ID.
+ */
+ public IgniteUuid futureId() {
+ return futId;
+ }
+
+ /** {@inheritDoc} */
+ @Override public void prepareMarshal(GridCacheSharedContext ctx) throws IgniteCheckedException {
+ super.prepareMarshal(ctx);
+
+ if (res != null) {
+ GridCacheContext cctx = ctx.cacheContext(cacheId);
+
+ if (res instanceof CacheObject)
+ prepareMarshalCacheObject((CacheObject) res, cctx);
+ else if (res instanceof CacheVersionedValue)
+ ((CacheVersionedValue)res).prepareMarshal(cctx.cacheObjectContext());
+ else if (res instanceof GridCacheEntryInfo)
+ ((GridCacheEntryInfo)res).marshal(cctx);
+ }
+
+ if (err != null)
+ errBytes = ctx.marshaller().marshal(err);
+ }
+
+ /** {@inheritDoc} */
+ @Override public void finishUnmarshal(GridCacheSharedContext ctx, ClassLoader ldr) throws IgniteCheckedException {
+ super.finishUnmarshal(ctx, ldr);
+
+ if (res != null) {
+ GridCacheContext cctx = ctx.cacheContext(cacheId());
+
+ if (res instanceof CacheObject)
+ ((CacheObject)res).finishUnmarshal(cctx.cacheObjectContext(), ldr);
+ else if (res instanceof CacheVersionedValue)
+ ((CacheVersionedValue)res).finishUnmarshal(cctx, ldr);
+ else if (res instanceof GridCacheEntryInfo)
+ ((GridCacheEntryInfo)res).unmarshal(cctx, ldr);
+ }
+
+ if (errBytes != null && err == null)
+ err = ctx.marshaller().unmarshal(errBytes, ldr);
+ }
+
+ /** {@inheritDoc} */
+ @Override public boolean writeTo(ByteBuffer buf, MessageWriter writer) {
+ writer.setBuffer(buf);
+
+ if (!super.writeTo(buf, writer))
+ return false;
+
+ if (!writer.isHeaderWritten()) {
+ if (!writer.writeHeader(directType(), fieldsCount()))
+ return false;
+
+ writer.onHeaderWritten();
+ }
+
+ switch (writer.state()) {
+ case 3:
+ if (!writer.writeByteArray("errBytes", errBytes))
+ return false;
+
+ writer.incrementState();
+
+ case 4:
+ if (!writer.writeByte("flags", flags))
+ return false;
+
+ writer.incrementState();
+
+ case 5:
+ if (!writer.writeIgniteUuid("futId", futId))
+ return false;
+
+ writer.incrementState();
+
+ case 6:
+ if (!writer.writeMessage("res", res))
+ return false;
+
+ writer.incrementState();
+
+ case 7:
+ if (!writer.writeMessage("topVer", topVer))
+ return false;
+
+ writer.incrementState();
+
+ }
+
+ return true;
+ }
+
+ /** {@inheritDoc} */
+ @Override public boolean readFrom(ByteBuffer buf, MessageReader reader) {
+ reader.setBuffer(buf);
+
+ if (!reader.beforeMessageRead())
+ return false;
+
+ if (!super.readFrom(buf, reader))
+ return false;
+
+ switch (reader.state()) {
+ case 3:
+ errBytes = reader.readByteArray("errBytes");
+
+ if (!reader.isLastRead())
+ return false;
+
+ reader.incrementState();
+
+ case 4:
+ flags = reader.readByte("flags");
+
+ if (!reader.isLastRead())
+ return false;
+
+ reader.incrementState();
+
+ case 5:
+ futId = reader.readIgniteUuid("futId");
+
+ if (!reader.isLastRead())
+ return false;
+
+ reader.incrementState();
+
+ case 6:
+ res = reader.readMessage("res");
+
+ if (!reader.isLastRead())
+ return false;
+
+ reader.incrementState();
+
+ case 7:
+ topVer = reader.readMessage("topVer");
+
+ if (!reader.isLastRead())
+ return false;
+
+ reader.incrementState();
+
+ }
+
+ return reader.afterMessageRead(GridNearSingleGetResponse.class);
+ }
+
+ /** {@inheritDoc} */
+ @Override public boolean addDeploymentInfo() {
+ return addDepInfo;
+ }
+
+ /** {@inheritDoc} */
+ @Override public byte directType() {
+ return 117;
+ }
+
+ /** {@inheritDoc} */
+ @Override public byte fieldsCount() {
+ return 8;
+ }
+
+ /** {@inheritDoc} */
+ @Override public String toString() {
+ return S.toString(GridNearSingleGetResponse.class, this);
+ }
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/1f103067/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTransactionalCache.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTransactionalCache.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTransactionalCache.java
index 1a26028..fd3d056 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTransactionalCache.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTransactionalCache.java
@@ -423,7 +423,7 @@ public class GridNearTransactionalCache<K, V> extends GridNearCacheAdapter<K, V>
assert nodeId != null;
assert res != null;
- GridNearLockFuture fut = (GridNearLockFuture)ctx.mvcc().<Boolean>future(res.version(),
+ GridNearLockFuture fut = (GridNearLockFuture)ctx.mvcc().<Boolean>mvccFuture(res.version(),
res.futureId());
if (fut != null)
http://git-wip-us.apache.org/repos/asf/ignite/blob/1f103067/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxFinishFuture.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxFinishFuture.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxFinishFuture.java
index 9c022b2..102cc4b 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxFinishFuture.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxFinishFuture.java
@@ -79,7 +79,7 @@ public final class GridNearTxFinishFuture<K, V> extends GridCompoundIdentityFutu
private GridCacheSharedContext<K, V> cctx;
/** Future ID. */
- private IgniteUuid futId;
+ private final IgniteUuid futId;
/** Transaction. */
@GridToStringInclude
@@ -125,26 +125,6 @@ public final class GridNearTxFinishFuture<K, V> extends GridCompoundIdentityFutu
}
/** {@inheritDoc} */
- @Override public GridCacheVersion version() {
- return tx.xidVersion();
- }
-
- /**
- * @return Involved nodes.
- */
- @Override public Collection<? extends ClusterNode> nodes() {
- return
- F.viewReadOnly(futures(), new IgniteClosure<IgniteInternalFuture<?>, ClusterNode>() {
- @Nullable @Override public ClusterNode apply(IgniteInternalFuture<?> f) {
- if (isMini(f))
- return ((MiniFuture)f).node();
-
- return cctx.discovery().localNode();
- }
- });
- }
-
- /** {@inheritDoc} */
@Override public boolean onNodeLeft(UUID nodeId) {
for (IgniteInternalFuture<?> fut : futures())
if (isMini(fut)) {
@@ -298,7 +278,7 @@ public final class GridNearTxFinishFuture<K, V> extends GridCompoundIdentityFutu
}
// Don't forget to clean up.
- cctx.mvcc().removeFuture(this);
+ cctx.mvcc().removeFuture(futId);
return true;
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/1f103067/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxLocal.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxLocal.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxLocal.java
index b92be31..1c01e4e 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxLocal.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxLocal.java
@@ -332,7 +332,7 @@ public class GridNearTxLocal extends GridDhtTxLocalAdapter {
boolean readThrough,
boolean async,
final Collection<KeyCacheObject> keys,
- boolean skipVals,
+ final boolean skipVals,
final boolean needVer,
final GridInClosure3<KeyCacheObject, Object, GridCacheVersion> c
) {
@@ -361,35 +361,70 @@ public class GridNearTxLocal extends GridDhtTxLocalAdapter {
});
}
else if (cacheCtx.isColocated()) {
- return cacheCtx.colocated().loadAsync(
- keys,
- readThrough,
- /*force primary*/needVer,
- topologyVersion(),
- CU.subjectId(this, cctx),
- resolveTaskName(),
- /*deserializePortable*/false,
- accessPolicy(cacheCtx, keys),
- skipVals,
- /*can remap*/true,
- needVer,
- /*keepCacheObject*/true
- ).chain(new C1<IgniteInternalFuture<Map<Object, Object>>, Void>() {
- @Override public Void apply(IgniteInternalFuture<Map<Object, Object>> f) {
- try {
- Map<Object, Object> map = f.get();
-
- processLoaded(map, keys, needVer, c);
-
- return null;
+ if (keys.size() == 1) {
+ final KeyCacheObject key = F.first(keys);
+
+ return cacheCtx.colocated().loadAsync(
+ key,
+ readThrough,
+ /*force primary*/needVer,
+ topologyVersion(),
+ CU.subjectId(this, cctx),
+ resolveTaskName(),
+ /*deserializePortable*/false,
+ accessPolicy(cacheCtx, keys),
+ skipVals,
+ /*can remap*/true,
+ needVer,
+ /*keepCacheObject*/true
+ ).chain(new C1<IgniteInternalFuture<Object>, Void>() {
+ @Override public Void apply(IgniteInternalFuture<Object> f) {
+ try {
+ Object val = f.get();
+
+ processLoaded(key, val, needVer, skipVals, c);
+
+ return null;
+ }
+ catch (Exception e) {
+ setRollbackOnly();
+
+ throw new GridClosureException(e);
+ }
}
- catch (Exception e) {
- setRollbackOnly();
-
- throw new GridClosureException(e);
+ });
+ }
+ else {
+ return cacheCtx.colocated().loadAsync(
+ keys,
+ readThrough,
+ /*force primary*/needVer,
+ topologyVersion(),
+ CU.subjectId(this, cctx),
+ resolveTaskName(),
+ /*deserializePortable*/false,
+ accessPolicy(cacheCtx, keys),
+ skipVals,
+ /*can remap*/true,
+ needVer,
+ /*keepCacheObject*/true
+ ).chain(new C1<IgniteInternalFuture<Map<Object, Object>>, Void>() {
+ @Override public Void apply(IgniteInternalFuture<Map<Object, Object>> f) {
+ try {
+ Map<Object, Object> map = f.get();
+
+ processLoaded(map, keys, needVer, c);
+
+ return null;
+ }
+ catch (Exception e) {
+ setRollbackOnly();
+
+ throw new GridClosureException(e);
+ }
}
- }
- });
+ });
+ }
}
else {
assert cacheCtx.isLocal();
@@ -409,29 +444,45 @@ public class GridNearTxLocal extends GridDhtTxLocalAdapter {
final Collection<KeyCacheObject> keys,
boolean needVer,
GridInClosure3<KeyCacheObject, Object, GridCacheVersion> c) {
- for (KeyCacheObject key : keys) {
- Object val = map.get(key);
+ for (KeyCacheObject key : keys)
+ processLoaded(key, map.get(key), needVer, false, c);
+ }
- if (val != null) {
- Object v;
- GridCacheVersion ver;
+ /**
+ * @param key Key.
+ * @param val Value.
+ * @param needVer If {@code true} version is required for loaded values.
+ * @param skipVals Skip values flag.
+ * @param c Closure.
+ */
+ private void processLoaded(
+ KeyCacheObject key,
+ @Nullable Object val,
+ boolean needVer,
+ boolean skipVals,
+ GridInClosure3<KeyCacheObject, Object, GridCacheVersion> c) {
+ if (val != null) {
+ Object v;
+ GridCacheVersion ver;
- if (needVer) {
- T2<Object, GridCacheVersion> t = (T2)val;
+ if (needVer) {
+ T2<Object, GridCacheVersion> t = (T2)val;
- v = t.get1();
- ver = t.get2();
- }
- else {
- v = val;
- ver = null;
- }
-
- c.apply(key, v, ver);
+ v = t.get1();
+ ver = t.get2();
}
- else
+ else {
+ v = val;
+ ver = null;
+ }
+
+ if (skipVals && v == Boolean.FALSE)
c.apply(key, null, IgniteTxEntry.SER_READ_EMPTY_ENTRY_VER);
+ else
+ c.apply(key, v, ver);
}
+ else
+ c.apply(key, null, IgniteTxEntry.SER_READ_EMPTY_ENTRY_VER);
}
/** {@inheritDoc} */
@@ -771,7 +822,7 @@ public class GridNearTxLocal extends GridDhtTxLocalAdapter {
if (fut == null && !commitFut.compareAndSet(null, fut = new GridNearTxFinishFuture<>(cctx, this, true)))
return commitFut.get();
- cctx.mvcc().addFuture(fut);
+ cctx.mvcc().addFuture(fut, fut.futureId());
final IgniteInternalFuture<?> prepareFut = prepFut.get();
@@ -816,7 +867,7 @@ public class GridNearTxLocal extends GridDhtTxLocalAdapter {
if (!rollbackFut.compareAndSet(null, fut = new GridNearTxFinishFuture<>(cctx, this, false)))
return rollbackFut.get();
- cctx.mvcc().addFuture(fut);
+ cctx.mvcc().addFuture(fut, fut.futureId());
IgniteInternalFuture<?> prepFut = this.prepFut.get();
@@ -957,7 +1008,7 @@ public class GridNearTxLocal extends GridDhtTxLocalAdapter {
final GridDhtTxFinishFuture fut = new GridDhtTxFinishFuture<>(cctx, this, /*commit*/true);
- cctx.mvcc().addFuture(fut);
+ cctx.mvcc().addFuture(fut, fut.futureId());
if (prep == null || prep.isDone()) {
assert prep != null || optimistic();
@@ -1016,7 +1067,7 @@ public class GridNearTxLocal extends GridDhtTxLocalAdapter {
final GridDhtTxFinishFuture fut = new GridDhtTxFinishFuture<>(cctx, this, /*commit*/false);
- cctx.mvcc().addFuture(fut);
+ cctx.mvcc().addFuture(fut, fut.futureId());
IgniteInternalFuture<?> prep = prepFut.get();
http://git-wip-us.apache.org/repos/asf/ignite/blob/1f103067/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxPrepareFutureAdapter.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxPrepareFutureAdapter.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxPrepareFutureAdapter.java
index cfaadc9..52cad91 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxPrepareFutureAdapter.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxPrepareFutureAdapter.java
@@ -25,7 +25,7 @@ import javax.cache.expiry.ExpiryPolicy;
import org.apache.ignite.IgniteLogger;
import org.apache.ignite.internal.processors.cache.GridCacheContext;
import org.apache.ignite.internal.processors.cache.GridCacheEntryRemovedException;
-import org.apache.ignite.internal.processors.cache.GridCacheFuture;
+import org.apache.ignite.internal.processors.cache.GridCacheMvccFuture;
import org.apache.ignite.internal.processors.cache.GridCacheSharedContext;
import org.apache.ignite.internal.processors.cache.distributed.GridDistributedTxMapping;
import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtTxMapping;
@@ -48,8 +48,8 @@ import static org.apache.ignite.internal.processors.cache.GridCacheOperation.NOO
/**
* Common code for tx prepare in optimistic and pessimistic modes.
*/
-public abstract class GridNearTxPrepareFutureAdapter extends GridCompoundFuture<GridNearTxPrepareResponse, IgniteInternalTx>
- implements GridCacheFuture<IgniteInternalTx> {
+public abstract class GridNearTxPrepareFutureAdapter extends
+ GridCompoundFuture<GridNearTxPrepareResponse, IgniteInternalTx> implements GridCacheMvccFuture<IgniteInternalTx> {
/** Logger reference. */
protected static final AtomicReference<IgniteLogger> logRef = new AtomicReference<>();
http://git-wip-us.apache.org/repos/asf/ignite/blob/1f103067/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/local/GridLocalCache.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/local/GridLocalCache.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/local/GridLocalCache.java
index 87da9a1..544d5b4 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/local/GridLocalCache.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/local/GridLocalCache.java
@@ -231,8 +231,8 @@ public class GridLocalCache<K, V> extends GridCacheAdapter<K, V> {
/**
* @param fut Clears future from cache.
*/
- void onFutureDone(GridCacheFuture<?> fut) {
- if (ctx.mvcc().removeFuture(fut)) {
+ void onFutureDone(GridLocalLockFuture fut) {
+ if (ctx.mvcc().removeMvccFuture(fut)) {
if (log().isDebugEnabled())
log().debug("Explicitly removed future from map of futures: " + fut);
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/1f103067/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/local/GridLocalLockFuture.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/local/GridLocalLockFuture.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/local/GridLocalLockFuture.java
index cb14b4c..d392d53 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/local/GridLocalLockFuture.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/local/GridLocalLockFuture.java
@@ -154,11 +154,6 @@ public final class GridLocalLockFuture<K, V> extends GridFutureAdapter<Boolean>
}
/** {@inheritDoc} */
- @Override public Collection<? extends ClusterNode> nodes() {
- return Collections.emptyList();
- }
-
- /** {@inheritDoc} */
@Override public boolean onNodeLeft(UUID nodeId) {
return false;
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/1f103067/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxHandler.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxHandler.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxHandler.java
index 0e5657b..a9846ef 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxHandler.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxHandler.java
@@ -477,7 +477,7 @@ public class IgniteTxHandler {
*/
private void processNearTxPrepareResponse(UUID nodeId, GridNearTxPrepareResponse res) {
GridNearTxPrepareFutureAdapter fut = (GridNearTxPrepareFutureAdapter)ctx.mvcc()
- .<IgniteInternalTx>future(res.version(), res.futureId());
+ .<IgniteInternalTx>mvccFuture(res.version(), res.futureId());
if (fut == null) {
U.warn(log, "Failed to find future for prepare response [sender=" + nodeId + ", res=" + res + ']');
@@ -495,8 +495,7 @@ public class IgniteTxHandler {
private void processNearTxFinishResponse(UUID nodeId, GridNearTxFinishResponse res) {
ctx.tm().onFinishedRemote(nodeId, res.threadId());
- GridNearTxFinishFuture fut = (GridNearTxFinishFuture)ctx.mvcc().<IgniteInternalTx>future(
- res.xid(), res.futureId());
+ GridNearTxFinishFuture fut = (GridNearTxFinishFuture)ctx.mvcc().<IgniteInternalTx>future(res.futureId());
if (fut == null) {
if (log.isDebugEnabled())
@@ -513,7 +512,7 @@ public class IgniteTxHandler {
* @param res Response.
*/
private void processDhtTxPrepareResponse(UUID nodeId, GridDhtTxPrepareResponse res) {
- GridDhtTxPrepareFuture fut = (GridDhtTxPrepareFuture)ctx.mvcc().future(res.version(), res.futureId());
+ GridDhtTxPrepareFuture fut = (GridDhtTxPrepareFuture)ctx.mvcc().mvccFuture(res.version(), res.futureId());
if (fut == null) {
if (log.isDebugEnabled())
@@ -534,8 +533,7 @@ public class IgniteTxHandler {
assert res != null;
if (res.checkCommitted()) {
- GridNearTxFinishFuture fut = (GridNearTxFinishFuture)ctx.mvcc().<IgniteInternalTx>future(
- res.xid(), res.futureId());
+ GridNearTxFinishFuture fut = (GridNearTxFinishFuture)ctx.mvcc().<IgniteInternalTx>future(res.futureId());
if (fut == null) {
if (log.isDebugEnabled())
@@ -547,8 +545,7 @@ public class IgniteTxHandler {
fut.onResult(nodeId, res);
}
else {
- GridDhtTxFinishFuture fut = (GridDhtTxFinishFuture)ctx.mvcc().<IgniteInternalTx>future(
- res.xid(), res.futureId());
+ GridDhtTxFinishFuture fut = (GridDhtTxFinishFuture)ctx.mvcc().<IgniteInternalTx>future(res.futureId());
if (fut == null) {
if (log.isDebugEnabled())
@@ -566,7 +563,8 @@ public class IgniteTxHandler {
* @param req Request.
* @return Future.
*/
- @Nullable public IgniteInternalFuture<IgniteInternalTx> processNearTxFinishRequest(UUID nodeId, GridNearTxFinishRequest req) {
+ @Nullable public IgniteInternalFuture<IgniteInternalTx> processNearTxFinishRequest(UUID nodeId,
+ GridNearTxFinishRequest req) {
return finish(nodeId, null, req);
}
@@ -1359,8 +1357,7 @@ public class IgniteTxHandler {
if (log.isDebugEnabled())
log.debug("Processing check prepared transaction response [nodeId=" + nodeId + ", res=" + res + ']');
- GridCacheTxRecoveryFuture fut = (GridCacheTxRecoveryFuture)ctx.mvcc().
- <Boolean>future(res.version(), res.futureId());
+ GridCacheTxRecoveryFuture fut = (GridCacheTxRecoveryFuture)ctx.mvcc().future(res.futureId());
if (fut == null) {
if (log.isDebugEnabled())
http://git-wip-us.apache.org/repos/asf/ignite/blob/1f103067/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxLocalAdapter.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxLocalAdapter.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxLocalAdapter.java
index 9e44b10..ecb0595 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxLocalAdapter.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxLocalAdapter.java
@@ -1372,7 +1372,6 @@ public abstract class IgniteTxLocalAdapter extends IgniteTxAdapter
if (txEntry != null) {
CacheObject val = txEntry.value();
- // Read value from locked entry in group-lock transaction as well.
if (txEntry.hasValue()) {
if (!F.isEmpty(txEntry.entryProcessors()))
val = txEntry.applyEntryProcessors(val);
@@ -2224,6 +2223,7 @@ public abstract class IgniteTxLocalAdapter extends IgniteTxAdapter
/**
* @param cacheCtx Cache context.
* @param keys Keys to load.
+ * @param filter Filter.
* @param ret Return value.
* @param needReadVer Read version flag.
* @param singleRmv {@code True} for single remove operation.
http://git-wip-us.apache.org/repos/asf/ignite/blob/1f103067/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxManager.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxManager.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxManager.java
index 67bca51..247ccaf 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxManager.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxManager.java
@@ -1764,7 +1764,7 @@ public class IgniteTxManager extends GridCacheSharedManagerAdapter {
tx.originatingNodeId(),
tx.transactionNodes());
- cctx.mvcc().addFuture(fut);
+ cctx.mvcc().addFuture(fut, fut.futureId());
if (log.isDebugEnabled())
log.debug("Checking optimistic transaction state on remote nodes [tx=" + tx + ", fut=" + fut + ']');
http://git-wip-us.apache.org/repos/asf/ignite/blob/1f103067/modules/core/src/test/java/org/apache/ignite/internal/IgniteClientReconnectCacheTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/IgniteClientReconnectCacheTest.java b/modules/core/src/test/java/org/apache/ignite/internal/IgniteClientReconnectCacheTest.java
index 6131f54..6cf10f4 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/IgniteClientReconnectCacheTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/IgniteClientReconnectCacheTest.java
@@ -53,6 +53,7 @@ import org.apache.ignite.internal.processors.cache.distributed.dht.atomic.GridNe
import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.GridDhtPartitionsFullMessage;
import org.apache.ignite.internal.processors.cache.distributed.near.GridNearGetResponse;
import org.apache.ignite.internal.processors.cache.distributed.near.GridNearLockResponse;
+import org.apache.ignite.internal.processors.cache.distributed.near.GridNearSingleGetResponse;
import org.apache.ignite.internal.processors.cache.distributed.near.GridNearTxFinishResponse;
import org.apache.ignite.internal.processors.cache.distributed.near.GridNearTxPrepareResponse;
import org.apache.ignite.internal.util.typedef.CI1;
@@ -765,6 +766,12 @@ public class IgniteClientReconnectCacheTest extends IgniteClientReconnectAbstrac
}
};
+ IgniteInClosure<IgniteCache<Object, Object>> getAllOp = new CI1<IgniteCache<Object, Object>>() {
+ @Override public void apply(IgniteCache<Object, Object> cache) {
+ cache.getAll(F.asSet(1, 2));
+ }
+ };
+
int cnt = 0;
for (CacheAtomicityMode atomicityMode : CacheAtomicityMode.values()) {
@@ -799,7 +806,9 @@ public class IgniteClientReconnectCacheTest extends IgniteClientReconnectAbstrac
log.info("Test cache get [atomicity=" + atomicityMode + ", syncMode=" + syncMode + ']');
- checkOperationInProgressFails(client, ccfg, GridNearGetResponse.class, getOp);
+ checkOperationInProgressFails(client, ccfg, GridNearSingleGetResponse.class, getOp);
+
+ checkOperationInProgressFails(client, ccfg, GridNearGetResponse.class, getAllOp);
client.destroyCache(ccfg.getName());
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/1f103067/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheAbstractFullApiSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheAbstractFullApiSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheAbstractFullApiSelfTest.java
index 1d79e20..e8e86e9 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheAbstractFullApiSelfTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheAbstractFullApiSelfTest.java
@@ -382,6 +382,81 @@ public abstract class GridCacheAbstractFullApiSelfTest extends GridCacheAbstract
/**
* @throws Exception If failed.
*/
+ public void testContainsKeyTx() throws Exception {
+ if (!txEnabled())
+ return;
+
+ IgniteCache<String, Integer> cache = jcache();
+
+ IgniteTransactions txs = ignite(0).transactions();
+
+ for (int i = 0; i < 10; i++) {
+ String key = String.valueOf(i);
+
+ try (Transaction tx = txs.txStart()) {
+ assertNull(key, cache.get(key));
+
+ assertFalse(cache.containsKey(key));
+
+ tx.commit();
+ }
+
+ try (Transaction tx = txs.txStart()) {
+ assertNull(key, cache.get(key));
+
+ cache.put(key, i);
+
+ assertTrue(cache.containsKey(key));
+
+ tx.commit();
+ }
+ }
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testContainsKeysTx() throws Exception {
+ if (!txEnabled())
+ return;
+
+ IgniteCache<String, Integer> cache = jcache();
+
+ IgniteTransactions txs = ignite(0).transactions();
+
+ Set<String> keys = new HashSet<>();
+
+ for (int i = 0; i < 10; i++) {
+ String key = String.valueOf(i);
+
+ keys.add(key);
+ }
+
+ try (Transaction tx = txs.txStart()) {
+ for (String key : keys)
+ assertNull(key, cache.get(key));
+
+ assertFalse(cache.containsKeys(keys));
+
+ tx.commit();
+ }
+
+ try (Transaction tx = txs.txStart()) {
+ for (String key : keys)
+ assertNull(key, cache.get(key));
+
+ for (String key : keys)
+ cache.put(key, 0);
+
+ assertTrue(cache.containsKeys(keys));
+
+ tx.commit();
+ }
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
public void testRemoveInExplicitLocks() throws Exception {
if (lockingEnabled()) {
IgniteCache<String, Integer> cache = jcache();
http://git-wip-us.apache.org/repos/asf/ignite/blob/1f103067/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheConcurrentTxMultiNodeTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheConcurrentTxMultiNodeTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheConcurrentTxMultiNodeTest.java
index 1ef77f2..45c8c2c 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheConcurrentTxMultiNodeTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheConcurrentTxMultiNodeTest.java
@@ -613,21 +613,6 @@ public class GridCacheConcurrentTxMultiNodeTest extends GridCommonAbstractTest {
X.println("Near entry [grid="+ g.name() + ", key=" + k + ", entry=" + nearEntry);
X.println("DHT entry [grid=" + g.name() + ", key=" + k + ", entry=" + dhtEntry);
-
- GridCacheMvccCandidate nearCand =
- nearEntry == null ? null : F.first(nearEntry.localCandidates());
-
- if (nearCand != null)
- X.println("Near futures: " +
- nearEntry.context().mvcc().futures(nearCand.version()));
-
- GridCacheMvccCandidate dhtCand =
- dhtEntry == null ? null : F.first(dhtEntry.localCandidates());
-
- if (dhtCand != null)
- X.println("Dht futures: " +
- dhtEntry.context().mvcc().futures(dhtCand.version()));
-
}
}
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/1f103067/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCachePartitionedGetSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCachePartitionedGetSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCachePartitionedGetSelfTest.java
index cbfc97b..2cb4a00 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCachePartitionedGetSelfTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCachePartitionedGetSelfTest.java
@@ -28,6 +28,7 @@ import org.apache.ignite.configuration.IgniteConfiguration;
import org.apache.ignite.internal.IgniteKernal;
import org.apache.ignite.internal.managers.communication.GridMessageListener;
import org.apache.ignite.internal.processors.cache.distributed.near.GridNearGetRequest;
+import org.apache.ignite.internal.processors.cache.distributed.near.GridNearSingleGetRequest;
import org.apache.ignite.spi.discovery.DiscoverySpi;
import org.apache.ignite.spi.discovery.tcp.TcpDiscoverySpi;
import org.apache.ignite.spi.discovery.tcp.ipfinder.TcpDiscoveryIpFinder;
@@ -228,7 +229,7 @@ public class GridCachePartitionedGetSelfTest extends GridCommonAbstractTest {
@Override public void onMessage(UUID nodeId, Object msg) {
info("Received message from node [nodeId=" + nodeId + ", msg=" + msg + ']');
- if (msg instanceof GridNearGetRequest) {
+ if (msg instanceof GridNearSingleGetRequest) {
info("Setting flag: " + System.identityHashCode(received));
received.set(true);
http://git-wip-us.apache.org/repos/asf/ignite/blob/1f103067/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteCacheAbstractStopBusySelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteCacheAbstractStopBusySelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteCacheAbstractStopBusySelfTest.java
index 8ddee0c..e5cff5a 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteCacheAbstractStopBusySelfTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteCacheAbstractStopBusySelfTest.java
@@ -20,6 +20,7 @@ package org.apache.ignite.internal.processors.cache;
import java.io.PrintWriter;
import java.io.StringWriter;
import java.util.List;
+import java.util.Set;
import java.util.concurrent.Callable;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
@@ -37,6 +38,8 @@ import org.apache.ignite.configuration.IgniteConfiguration;
import org.apache.ignite.internal.IgniteInternalFuture;
import org.apache.ignite.internal.managers.communication.GridIoMessage;
import org.apache.ignite.internal.processors.cache.distributed.near.GridNearGetRequest;
+import org.apache.ignite.internal.processors.cache.distributed.near.GridNearSingleGetRequest;
+import org.apache.ignite.internal.util.typedef.F;
import org.apache.ignite.internal.util.typedef.G;
import org.apache.ignite.lang.IgniteInClosure;
import org.apache.ignite.plugin.extensions.communication.Message;
@@ -218,7 +221,7 @@ public abstract class IgniteCacheAbstractStopBusySelfTest extends GridCommonAbst
* @throws Exception If failed.
*/
public void testGet() throws Exception {
- bannedMsg.set(GridNearGetRequest.class);
+ bannedMsg.set(GridNearSingleGetRequest.class);
executeTest(new Callable<Integer>() {
/** {@inheritDoc} */
@@ -235,6 +238,28 @@ public abstract class IgniteCacheAbstractStopBusySelfTest extends GridCommonAbst
}
/**
+ * @throws Exception If failed.
+ */
+ public void testGetAll() throws Exception {
+ bannedMsg.set(GridNearGetRequest.class);
+
+ executeTest(new Callable<Integer>() {
+ /** {@inheritDoc} */
+ @Override public Integer call() throws Exception {
+ info("Start operation.");
+
+ Set<Integer> keys = F.asSet(1, 2, 3);
+
+ clientCache().getAll(keys);
+
+ info("Stop operation.");
+
+ return null;
+ }
+ });
+ }
+
+ /**
*
* @param call Closure executing cache operation.
* @throws Exception If failed.
http://git-wip-us.apache.org/repos/asf/ignite/blob/1f103067/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteCacheP2pUnmarshallingErrorTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteCacheP2pUnmarshallingErrorTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteCacheP2pUnmarshallingErrorTest.java
index f4423f7..9c1abc7 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteCacheP2pUnmarshallingErrorTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteCacheP2pUnmarshallingErrorTest.java
@@ -21,6 +21,7 @@ import java.io.Externalizable;
import java.io.IOException;
import java.io.ObjectInput;
import java.io.ObjectOutput;
+import java.util.Set;
import java.util.concurrent.atomic.AtomicInteger;
import javax.cache.CacheException;
import org.apache.ignite.cache.CacheAtomicWriteOrderMode;
@@ -30,6 +31,7 @@ import org.apache.ignite.cache.query.annotations.QuerySqlField;
import org.apache.ignite.configuration.CacheConfiguration;
import org.apache.ignite.configuration.IgniteConfiguration;
import org.apache.ignite.configuration.NearCacheConfiguration;
+import org.apache.ignite.internal.util.typedef.F;
import org.apache.ignite.internal.util.typedef.X;
/**
@@ -37,7 +39,10 @@ import org.apache.ignite.internal.util.typedef.X;
*/
public class IgniteCacheP2pUnmarshallingErrorTest extends IgniteCacheAbstractTest {
/** Allows to change behavior of readExternal method. */
- protected static AtomicInteger readCnt = new AtomicInteger();
+ protected static final AtomicInteger readCnt = new AtomicInteger();
+
+ /** Allows to change behavior of readExternal method. */
+ protected static final AtomicInteger valReadCnt = new AtomicInteger();
/** Iterable key. */
protected static int key = 0;
@@ -86,71 +91,40 @@ public class IgniteCacheP2pUnmarshallingErrorTest extends IgniteCacheAbstractTes
return cfg;
}
- /** Test key 1. */
- protected static class TestKey implements Externalizable {
- /** Field. */
- @QuerySqlField(index = true)
- private String field;
-
- /**
- * @param field Test key 1.
- */
- public TestKey(String field) {
- this.field = field;
- }
-
- /** Test key 1. */
- public TestKey() {
- // No-op.
- }
-
- /** {@inheritDoc} */
- @Override public boolean equals(Object o) {
- if (this == o)
- return true;
- if (o == null || getClass() != o.getClass())
- return false;
-
- TestKey key = (TestKey)o;
-
- return !(field != null ? !field.equals(key.field) : key.field != null);
- }
+ /**
+ * Sends put atomically and handles fail.
+ *
+ * @param k Key.
+ */
+ protected void failAtomicPut(int k) {
+ try {
+ jcache(0).put(new TestKey(String.valueOf(k)), "");
- /** {@inheritDoc} */
- @Override public int hashCode() {
- return field != null ? field.hashCode() : 0;
+ assert false : "p2p marshalling failed, but error response was not sent";
}
-
- /** {@inheritDoc} */
- @Override public void writeExternal(ObjectOutput out) throws IOException {
- out.writeObject(field);
+ catch (CacheException e) {
+ assert X.hasCause(e, IOException.class);
}
- /** {@inheritDoc} */
- @Override public void readExternal(ObjectInput in) throws IOException, ClassNotFoundException {
- field = (String)in.readObject();
-
- if (readCnt.decrementAndGet() <= 0)
- throw new IOException("Class can not be unmarshalled.");
- }
+ assert readCnt.get() == 0; //ensure we have read count as expected.
}
/**
- * Sends put atomically and handles fail.
+ * Sends get atomically and handles fail.
*
* @param k Key.
*/
- protected void failAtomicPut(int k) {
+ protected void failGetAll(int k) {
try {
- jcache(0).put(new TestKey(String.valueOf(k)), "");
+ Set<Object> keys = F.<Object>asSet(new TestKey(String.valueOf(k)));
+
+ jcache(0).getAll(keys);
assert false : "p2p marshalling failed, but error response was not sent";
}
catch (CacheException e) {
assert X.hasCause(e, IOException.class);
}
-
- assert readCnt.get() == 0; //ensure we have read count as expected.
}
/**
@@ -158,7 +132,7 @@ public class IgniteCacheP2pUnmarshallingErrorTest extends IgniteCacheAbstractTes
*
* @param k Key.
*/
- protected void failAtomicGet(int k) {
+ protected void failGet(int k) {
try {
jcache(0).get(new TestKey(String.valueOf(k)));
@@ -175,38 +149,132 @@ public class IgniteCacheP2pUnmarshallingErrorTest extends IgniteCacheAbstractTes
* @throws Exception If failed.
*/
public void testResponseMessageOnUnmarshallingFailed() throws Exception {
- //GridNearAtomicUpdateRequest unmarshalling failed test
+ // GridNearAtomicUpdateRequest unmarshalling failed test.
readCnt.set(1);
failAtomicPut(++key);
- //Check that cache is empty.
+ // Check that cache is empty.
readCnt.set(Integer.MAX_VALUE);
assert jcache(0).get(new TestKey(String.valueOf(key))) == null;
- //GridDhtAtomicUpdateRequest unmarshalling failed test
+ // GridDhtAtomicUpdateRequest unmarshalling failed test.
readCnt.set(2);
failAtomicPut(++key);
- //Check that cache is not empty.
+ // Check that cache is not empty.
readCnt.set(Integer.MAX_VALUE);
assert jcache(0).get(new TestKey(String.valueOf(key))) != null;
- //GridNearGetRequest unmarshalling failed test
+ // GridNearGetRequest unmarshalling failed test.
readCnt.set(1);
- failAtomicGet(++key);
+ failGetAll(++key);
- //GridNearGetResponse unmarshalling failed test
+ // GridNearGetResponse unmarshalling failed test.
readCnt.set(Integer.MAX_VALUE);
jcache(0).put(new TestKey(String.valueOf(++key)), "");
readCnt.set(2);
- failAtomicGet(key);
+ failGetAll(key);
+
+ readCnt.set(Integer.MAX_VALUE);
+ valReadCnt.set(Integer.MAX_VALUE);
+
+ jcache(0).put(new TestKey(String.valueOf(++key)), new TestValue());
+
+ assertNotNull(new TestKey(String.valueOf(key)));
+
+ // GridNearSingleGetRequest unmarshalling failed.
+ readCnt.set(1);
+
+ failGet(key);
+
+ // GridNearSingleGetRequest unmarshalling failed.
+ valReadCnt.set(1);
+ readCnt.set(2);
+
+ failGet(key);
+ }
+
+ /**
+ * Test key.
+ */
+ protected static class TestKey implements Externalizable {
+ /** Field. */
+ @QuerySqlField(index = true)
+ private String field;
+
+ /**
+ * Required by {@link Externalizable}.
+ */
+ public TestKey() {
+ // No-op.
+ }
+
+ /**
+ * @param field Test key 1.
+ */
+ public TestKey(String field) {
+ this.field = field;
+ }
+
+ /** {@inheritDoc} */
+ @Override public boolean equals(Object o) {
+ if (this == o)
+ return true;
+ if (o == null || getClass() != o.getClass())
+ return false;
+
+ TestKey key = (TestKey)o;
+
+ return !(field != null ? !field.equals(key.field) : key.field != null);
+ }
+
+ /** {@inheritDoc} */
+ @Override public int hashCode() {
+ return field != null ? field.hashCode() : 0;
+ }
+
+ /** {@inheritDoc} */
+ @Override public void writeExternal(ObjectOutput out) throws IOException {
+ out.writeObject(field);
+ }
+
+ /** {@inheritDoc} */
+ @Override public void readExternal(ObjectInput in) throws IOException, ClassNotFoundException {
+ field = (String)in.readObject();
+
+ if (readCnt.decrementAndGet() <= 0)
+ throw new IOException("Class can not be unmarshalled.");
+ }
+ }
+
+ /**
+ * Test value.
+ */
+ protected static class TestValue implements Externalizable {
+ /**
+ * Required by {@link Externalizable}.
+ */
+ public TestValue() {
+ // No-op.
+ }
+
+ /** {@inheritDoc} */
+ @Override public void writeExternal(ObjectOutput out) throws IOException {
+ // No-op.
+ }
+
+ /** {@inheritDoc} */
+ @Override public void readExternal(ObjectInput in) throws IOException, ClassNotFoundException {
+ if (valReadCnt.decrementAndGet() <= 0)
+ throw new IOException("Class can not be unmarshalled.");
+ }
}
}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/ignite/blob/1f103067/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/CacheGetFutureHangsSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/CacheGetFutureHangsSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/CacheGetFutureHangsSelfTest.java
index 53ac648..c005945 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/CacheGetFutureHangsSelfTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/CacheGetFutureHangsSelfTest.java
@@ -19,6 +19,7 @@ package org.apache.ignite.internal.processors.cache.distributed;
import java.util.ArrayList;
import java.util.List;
+import java.util.Set;
import java.util.UUID;
import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.atomic.AtomicReferenceArray;
@@ -29,6 +30,7 @@ import org.apache.ignite.cache.CacheMode;
import org.apache.ignite.configuration.CacheConfiguration;
import org.apache.ignite.configuration.IgniteConfiguration;
import org.apache.ignite.internal.IgniteInternalFuture;
+import org.apache.ignite.internal.util.typedef.F;
import org.apache.ignite.internal.util.typedef.T2;
import org.apache.ignite.marshaller.optimized.OptimizedMarshaller;
import org.apache.ignite.spi.communication.tcp.TcpCommunicationSpi;
@@ -135,12 +137,16 @@ public class CacheGetFutureHangsSelfTest extends GridCommonAbstractTest {
@Override public void run() {
T2<Ignite, Integer> ignite;
+ Set<Integer> keys = F.asSet(1, 2, 3, 4, 5);
+
while ((ignite = randomNode()) != null) {
IgniteCache<Object, Object> cache = ignite.get1().cache(null);
for (int i = 0; i < 100; i++)
cache.containsKey(ThreadLocalRandom.current().nextInt(100_000));
+ cache.containsKeys(keys);
+
assertTrue(nodes.compareAndSet(ignite.get2(), null, ignite.get1()));
try {
http://git-wip-us.apache.org/repos/asf/ignite/blob/1f103067/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/GridCacheAbstractNodeRestartSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/GridCacheAbstractNodeRestartSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/GridCacheAbstractNodeRestartSelfTest.java
index 6f0565b..dbd8758 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/GridCacheAbstractNodeRestartSelfTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/GridCacheAbstractNodeRestartSelfTest.java
@@ -661,6 +661,8 @@ public abstract class GridCacheAbstractNodeRestartSelfTest extends GridCommonAbs
// It is ok if primary node leaves grid.
}
+ cache.get(key);
+
int c = putCntr.incrementAndGet();
if (c % logFreq == 0)
http://git-wip-us.apache.org/repos/asf/ignite/blob/1f103067/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/IgniteCacheSingleGetMessageTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/IgniteCacheSingleGetMessageTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/IgniteCacheSingleGetMessageTest.java
new file mode 100644
index 0000000..42b5ee3
--- /dev/null
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/IgniteCacheSingleGetMessageTest.java
@@ -0,0 +1,357 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.internal.processors.cache.distributed;
+
+import java.util.ArrayList;
+import java.util.List;
+import org.apache.ignite.Ignite;
+import org.apache.ignite.IgniteCache;
+import org.apache.ignite.IgniteException;
+import org.apache.ignite.cache.CacheAtomicityMode;
+import org.apache.ignite.cache.CacheMode;
+import org.apache.ignite.cache.CacheWriteSynchronizationMode;
+import org.apache.ignite.cluster.ClusterNode;
+import org.apache.ignite.configuration.CacheConfiguration;
+import org.apache.ignite.configuration.IgniteConfiguration;
+import org.apache.ignite.internal.managers.communication.GridIoMessage;
+import org.apache.ignite.internal.processors.cache.distributed.near.GridNearSingleGetRequest;
+import org.apache.ignite.internal.processors.cache.distributed.near.GridNearSingleGetResponse;
+import org.apache.ignite.lang.IgniteInClosure;
+import org.apache.ignite.plugin.extensions.communication.Message;
+import org.apache.ignite.spi.IgniteSpiException;
+import org.apache.ignite.spi.communication.tcp.TcpCommunicationSpi;
+import org.apache.ignite.spi.discovery.tcp.TcpDiscoverySpi;
+import org.apache.ignite.spi.discovery.tcp.ipfinder.TcpDiscoveryIpFinder;
+import org.apache.ignite.spi.discovery.tcp.ipfinder.vm.TcpDiscoveryVmIpFinder;
+import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
+import org.apache.ignite.transactions.Transaction;
+import org.jetbrains.annotations.Nullable;
+
+import static org.apache.ignite.cache.CacheAtomicityMode.ATOMIC;
+import static org.apache.ignite.cache.CacheAtomicityMode.TRANSACTIONAL;
+import static org.apache.ignite.cache.CacheMode.PARTITIONED;
+import static org.apache.ignite.cache.CacheMode.REPLICATED;
+import static org.apache.ignite.cache.CacheWriteSynchronizationMode.FULL_SYNC;
+import static org.apache.ignite.transactions.TransactionConcurrency.OPTIMISTIC;
+import static org.apache.ignite.transactions.TransactionIsolation.REPEATABLE_READ;
+
+/**
+ *
+ */
+public class IgniteCacheSingleGetMessageTest extends GridCommonAbstractTest {
+ /** */
+ private static TcpDiscoveryIpFinder ipFinder = new TcpDiscoveryVmIpFinder(true);
+
+ /** */
+ private static final int SRVS = 4;
+
+ /** */
+ private boolean client;
+
+ /** {@inheritDoc} */
+ @Override protected IgniteConfiguration getConfiguration(String gridName) throws Exception {
+ IgniteConfiguration cfg = super.getConfiguration(gridName);
+
+ cfg.setClientMode(client);
+
+ TestCommunicationSpi commSpi = new TestCommunicationSpi();
+
+ cfg.setCommunicationSpi(commSpi);
+
+ ((TcpDiscoverySpi)cfg.getDiscoverySpi()).setIpFinder(ipFinder);
+
+ return cfg;
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void beforeTestsStarted() throws Exception {
+ super.beforeTestsStarted();
+
+ startGridsMultiThreaded(SRVS);
+
+ client = true;
+
+ startGridsMultiThreaded(SRVS, 1);
+
+ client = false;
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void afterTestsStopped() throws Exception {
+ super.afterTestsStopped();
+
+ stopAllGrids();
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testSingleGetMessage() throws Exception {
+ assertFalse(ignite(0).configuration().isClientMode());
+ assertTrue(ignite(SRVS).configuration().isClientMode());
+
+ List<CacheConfiguration<Integer, Integer>> ccfgs = cacheConfigurations();
+
+ for (int i = 0; i < ccfgs.size(); i++) {
+ CacheConfiguration<Integer, Integer> ccfg = ccfgs.get(i);
+
+ ccfg.setName("cache-" + i);
+
+ log.info("Test cache: " + i);
+
+ ignite(0).createCache(ccfg);
+
+ try {
+ IgniteCache<Integer, Integer> srvCache = ignite(0).cache(ccfg.getName());
+ IgniteCache<Integer, Integer> clientCache = ignite(SRVS).cache(ccfg.getName());
+
+ Integer key = nearKey(clientCache);
+
+ checkSingleGetMessage(clientCache, key, false);
+
+ if (ccfg.getBackups() > 0) {
+ key = backupKeys(srvCache, 1, 100_000).get(0);
+
+ checkSingleGetMessage(srvCache, key, true);
+ }
+
+ if (ccfg.getCacheMode() != REPLICATED) {
+ key = nearKeys(srvCache, 1, 200_000).get(0);
+
+ checkSingleGetMessage(srvCache, key, false);
+ }
+ }
+ finally {
+ ignite(0).destroyCache(ccfg.getName());
+ }
+ }
+ }
+
+ /**
+ * @param cache Cache.
+ * @param key Key.
+ * @param backup {@code True} if given key is backup key.
+ * @throws Exception If failed.
+ */
+ @SuppressWarnings("unchecked")
+ private void checkSingleGetMessage(IgniteCache<Integer, Integer> cache,
+ Integer key,
+ boolean backup) throws Exception {
+ CacheConfiguration<Integer, Integer> ccfg = cache.getConfiguration(CacheConfiguration.class);
+
+ Ignite node = cache.unwrap(Ignite.class);
+
+ TestCommunicationSpi spi = (TestCommunicationSpi)node.configuration().getCommunicationSpi();
+
+ spi.record(GridNearSingleGetRequest.class);
+
+ Ignite primary = primaryNode(key, cache.getName());
+
+ assertNotSame(node, primary);
+
+ TestCommunicationSpi primarySpi = (TestCommunicationSpi)primary.configuration().getCommunicationSpi();
+
+ primarySpi.record(GridNearSingleGetResponse.class);
+
+ assertNull(cache.get(key));
+
+ checkMessages(spi, primarySpi);
+
+ assertFalse(cache.containsKey(key));
+
+ checkMessages(spi, primarySpi);
+
+ cache.put(key, 1);
+
+ assertNotNull(cache.get(key));
+
+ if (backup)
+ checkNoMessages(spi, primarySpi);
+ else
+ checkMessages(spi, primarySpi);
+
+ assertTrue(cache.containsKey(key));
+
+ if (backup)
+ checkNoMessages(spi, primarySpi);
+ else
+ checkMessages(spi, primarySpi);
+
+ if (ccfg.getAtomicityMode() == TRANSACTIONAL) {
+ cache.remove(key);
+
+ try (Transaction tx = node.transactions().txStart(OPTIMISTIC, REPEATABLE_READ)) {
+ assertNull(cache.get(key));
+
+ tx.commit();
+ }
+
+ checkMessages(spi, primarySpi);
+
+ try (Transaction tx = node.transactions().txStart(OPTIMISTIC, REPEATABLE_READ)) {
+ assertFalse(cache.containsKey(key));
+
+ tx.commit();
+ }
+
+ checkMessages(spi, primarySpi);
+
+ cache.put(key, 1);
+
+ try (Transaction tx = node.transactions().txStart(OPTIMISTIC, REPEATABLE_READ)) {
+ assertNotNull(cache.get(key));
+
+ tx.commit();
+ }
+
+ if (backup)
+ checkNoMessages(spi, primarySpi);
+ else
+ checkMessages(spi, primarySpi);
+
+ try (Transaction tx = node.transactions().txStart(OPTIMISTIC, REPEATABLE_READ)) {
+ assertTrue(cache.containsKey(key));
+
+ tx.commit();
+ }
+
+ if (backup)
+ checkNoMessages(spi, primarySpi);
+ else
+ checkMessages(spi, primarySpi);
+ }
+ }
+
+ /**
+ * @param spi Near node SPI.
+ * @param primarySpi Primary node SPI.
+ */
+ private void checkMessages(TestCommunicationSpi spi, TestCommunicationSpi primarySpi) {
+ List<Object> msgs = spi.recordedMessages();
+
+ assertEquals(1, msgs.size());
+ assertTrue(msgs.get(0) instanceof GridNearSingleGetRequest);
+
+ msgs = primarySpi.recordedMessages();
+
+ assertEquals(1, msgs.size());
+ assertTrue(msgs.get(0) instanceof GridNearSingleGetResponse);
+ }
+
+ /**
+ * @param spi Near node SPI.
+ * @param primarySpi Primary node SPI.
+ */
+ private void checkNoMessages(TestCommunicationSpi spi, TestCommunicationSpi primarySpi) {
+ List<Object> msgs = spi.recordedMessages();
+ assertEquals(0, msgs.size());
+
+ msgs = primarySpi.recordedMessages();
+ assertEquals(0, msgs.size());
+ }
+
+ /**
+ * @return Cache configurations to test.
+ */
+ private List<CacheConfiguration<Integer, Integer>> cacheConfigurations() {
+ List<CacheConfiguration<Integer, Integer>> ccfgs = new ArrayList<>();
+
+ ccfgs.add(cacheConfiguration(PARTITIONED, TRANSACTIONAL, FULL_SYNC, 0));
+ ccfgs.add(cacheConfiguration(PARTITIONED, TRANSACTIONAL, FULL_SYNC, 1));
+ ccfgs.add(cacheConfiguration(REPLICATED, TRANSACTIONAL, FULL_SYNC, 0));
+
+ ccfgs.add(cacheConfiguration(PARTITIONED, ATOMIC, FULL_SYNC, 0));
+ ccfgs.add(cacheConfiguration(PARTITIONED, ATOMIC, FULL_SYNC, 1));
+ ccfgs.add(cacheConfiguration(REPLICATED, ATOMIC, FULL_SYNC, 0));
+
+ return ccfgs;
+ }
+
+ /**
+ * @param cacheMode Cache mode.
+ * @param atomicityMode Cache atomicity mode.
+ * @param syncMode Write synchronization mode.
+ * @param backups Number of backups.
+ * @return Cache configuration.
+ */
+ private CacheConfiguration<Integer, Integer> cacheConfiguration(
+ CacheMode cacheMode,
+ CacheAtomicityMode atomicityMode,
+ CacheWriteSynchronizationMode syncMode,
+ int backups) {
+ CacheConfiguration<Integer, Integer> ccfg = new CacheConfiguration<>();
+
+ ccfg.setCacheMode(cacheMode);
+ ccfg.setAtomicityMode(atomicityMode);
+ ccfg.setAtomicityMode(TRANSACTIONAL);
+ ccfg.setWriteSynchronizationMode(syncMode);
+
+ if (cacheMode == PARTITIONED)
+ ccfg.setBackups(backups);
+
+ return ccfg;
+ }
+
+ /**
+ *
+ */
+ private static class TestCommunicationSpi extends TcpCommunicationSpi {
+ /** */
+ private Class<?> recordCls;
+
+ /** */
+ private List<Object> recordedMsgs = new ArrayList<>();
+
+ /** {@inheritDoc} */
+ @Override public void sendMessage(ClusterNode node, Message msg, IgniteInClosure<IgniteException> ackC)
+ throws IgniteSpiException {
+ if (msg instanceof GridIoMessage) {
+ Object msg0 = ((GridIoMessage)msg).message();
+
+ synchronized (this) {
+ if (recordCls != null && msg0.getClass().equals(recordCls))
+ recordedMsgs.add(msg0);
+ }
+ }
+
+ super.sendMessage(node, msg, ackC);
+ }
+
+ /**
+ * @param recordCls Message class to record.
+ */
+ void record(@Nullable Class<?> recordCls) {
+ synchronized (this) {
+ this.recordCls = recordCls;
+ }
+ }
+
+ /**
+ * @return Recorded messages.
+ */
+ List<Object> recordedMessages() {
+ synchronized (this) {
+ List<Object> msgs = recordedMsgs;
+
+ recordedMsgs = new ArrayList<>();
+
+ return msgs;
+ }
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/1f103067/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/replicated/GridCacheReplicatedMetricsSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/replicated/GridCacheReplicatedMetricsSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/replicated/GridCacheReplicatedMetricsSelfTest.java
index 319aa56..2fa43be 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/replicated/GridCacheReplicatedMetricsSelfTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/replicated/GridCacheReplicatedMetricsSelfTest.java
@@ -18,7 +18,6 @@
package org.apache.ignite.internal.processors.cache.distributed.replicated;
import org.apache.ignite.configuration.CacheConfiguration;
-import org.apache.ignite.configuration.IgniteConfiguration;
import org.apache.ignite.internal.processors.cache.GridCacheTransactionalAbstractMetricsSelfTest;
import static org.apache.ignite.cache.CacheMode.REPLICATED;
@@ -30,14 +29,6 @@ public class GridCacheReplicatedMetricsSelfTest extends GridCacheTransactionalAb
/** */
private static final int GRID_CNT = 2;
- @Override protected IgniteConfiguration getConfiguration(String gridName) throws Exception {
- IgniteConfiguration c = super.getConfiguration(gridName);
-
- c.getTransactionConfiguration().setTxSerializableEnabled(true);
-
- return c;
- }
-
/** {@inheritDoc} */
@Override protected CacheConfiguration cacheConfiguration(String gridName) throws Exception {
CacheConfiguration cfg = super.cacheConfiguration(gridName);
http://git-wip-us.apache.org/repos/asf/ignite/blob/1f103067/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheTestSuite4.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheTestSuite4.java b/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheTestSuite4.java
index b89bffd..fcc8d37 100644
--- a/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheTestSuite4.java
+++ b/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheTestSuite4.java
@@ -81,6 +81,7 @@ import org.apache.ignite.internal.processors.cache.distributed.CacheGetFutureHan
import org.apache.ignite.internal.processors.cache.distributed.CacheNoValueClassOnServerNodeTest;
import org.apache.ignite.internal.processors.cache.distributed.IgniteCacheCreatePutMultiNodeSelfTest;
import org.apache.ignite.internal.processors.cache.distributed.IgniteCacheCreatePutTest;
+import org.apache.ignite.internal.processors.cache.distributed.IgniteCacheSingleGetMessageTest;
import org.apache.ignite.internal.processors.cache.distributed.dht.GridCacheDhtTxPreloadSelfTest;
import org.apache.ignite.internal.processors.cache.distributed.dht.IgniteCacheLockFailoverSelfTest;
import org.apache.ignite.internal.processors.cache.distributed.dht.IgniteCacheMultiTxLockSelfTest;
@@ -283,6 +284,8 @@ public class IgniteCacheTestSuite4 extends TestSuite {
suite.addTestSuite(CacheGetFutureHangsSelfTest.class);
+ suite.addTestSuite(IgniteCacheSingleGetMessageTest.class);
+
return suite;
}
}
\ No newline at end of file
[2/3] ignite git commit: Optimization for single key cache 'get'
operation.
Posted by sb...@apache.org.
http://git-wip-us.apache.org/repos/asf/ignite/blob/1f103067/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridPartitionedSingleGetFuture.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridPartitionedSingleGetFuture.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridPartitionedSingleGetFuture.java
new file mode 100644
index 0000000..8f2357b
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridPartitionedSingleGetFuture.java
@@ -0,0 +1,697 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.internal.processors.cache.distributed.dht;
+
+import java.util.Collection;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.UUID;
+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.cluster.ClusterTopologyCheckedException;
+import org.apache.ignite.internal.cluster.ClusterTopologyServerNotFoundException;
+import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
+import org.apache.ignite.internal.processors.cache.CacheObject;
+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.GridCacheFuture;
+import org.apache.ignite.internal.processors.cache.GridCacheMessage;
+import org.apache.ignite.internal.processors.cache.IgniteCacheExpiryPolicy;
+import org.apache.ignite.internal.processors.cache.KeyCacheObject;
+import org.apache.ignite.internal.processors.cache.distributed.near.CacheVersionedValue;
+import org.apache.ignite.internal.processors.cache.distributed.near.GridNearGetRequest;
+import org.apache.ignite.internal.processors.cache.distributed.near.GridNearGetResponse;
+import org.apache.ignite.internal.processors.cache.distributed.near.GridNearSingleGetRequest;
+import org.apache.ignite.internal.processors.cache.distributed.near.GridNearSingleGetResponse;
+import org.apache.ignite.internal.processors.cache.version.GridCacheVersion;
+import org.apache.ignite.internal.util.future.GridFutureAdapter;
+import org.apache.ignite.internal.util.typedef.CI1;
+import org.apache.ignite.internal.util.typedef.CIX1;
+import org.apache.ignite.internal.util.typedef.F;
+import org.apache.ignite.internal.util.typedef.T2;
+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.IgniteProductVersion;
+import org.apache.ignite.lang.IgniteUuid;
+import org.apache.ignite.plugin.extensions.communication.Message;
+import org.jetbrains.annotations.Nullable;
+
+/**
+ *
+ */
+public class GridPartitionedSingleGetFuture extends GridFutureAdapter<Object> implements GridCacheFuture<Object>,
+ CacheGetFuture {
+ /** */
+ private static final long serialVersionUID = 0L;
+
+ /** */
+ public static final IgniteProductVersion SINGLE_GET_MSG_SINCE = IgniteProductVersion.fromString("1.5.0");
+
+ /** Logger reference. */
+ private static final AtomicReference<IgniteLogger> logRef = new AtomicReference<>();
+
+ /** Logger. */
+ private static IgniteLogger log;
+
+ /** Topology version. */
+ private AffinityTopologyVersion topVer;
+
+ /** Context. */
+ private final GridCacheContext cctx;
+
+ /** Key. */
+ private final KeyCacheObject key;
+
+ /** Read through flag. */
+ private final boolean readThrough;
+
+ /** Force primary flag. */
+ private final boolean forcePrimary;
+
+ /** Future ID. */
+ private final IgniteUuid futId;
+
+ /** Trackable flag. */
+ private boolean trackable;
+
+ /** Subject ID. */
+ private final UUID subjId;
+
+ /** Task name. */
+ private final String taskName;
+
+ /** Whether to deserialize portable objects. */
+ private boolean deserializePortable;
+
+ /** Skip values flag. */
+ private boolean skipVals;
+
+ /** Expiry policy. */
+ private IgniteCacheExpiryPolicy expiryPlc;
+
+ /** Flag indicating that get should be done on a locked topology version. */
+ private final boolean canRemap;
+
+ /** */
+ private final boolean needVer;
+
+ /** */
+ private final boolean keepCacheObjects;
+
+ /** */
+ private ClusterNode node;
+
+ /**
+ * @param cctx Context.
+ * @param key Key.
+ * @param topVer Topology version.
+ * @param readThrough Read through flag.
+ * @param forcePrimary If {@code true} then will force network trip to primary node even if called on backup node.
+ * @param subjId Subject ID.
+ * @param taskName Task name.
+ * @param deserializePortable Deserialize portable flag.
+ * @param expiryPlc Expiry policy.
+ * @param skipVals Skip values flag.
+ * @param canRemap Flag indicating whether future can be remapped on a newer topology version.
+ * @param needVer If {@code true} returns values as tuples containing value and version.
+ * @param keepCacheObjects Keep cache objects flag.
+ */
+ public GridPartitionedSingleGetFuture(
+ GridCacheContext cctx,
+ KeyCacheObject key,
+ AffinityTopologyVersion topVer,
+ boolean readThrough,
+ boolean forcePrimary,
+ @Nullable UUID subjId,
+ String taskName,
+ boolean deserializePortable,
+ @Nullable IgniteCacheExpiryPolicy expiryPlc,
+ boolean skipVals,
+ boolean canRemap,
+ boolean needVer,
+ boolean keepCacheObjects
+ ) {
+ assert key != null;
+
+ this.cctx = cctx;
+ this.key = key;
+ this.readThrough = readThrough;
+ this.forcePrimary = forcePrimary;
+ this.subjId = subjId;
+ this.taskName = taskName;
+ this.deserializePortable = deserializePortable;
+ this.expiryPlc = expiryPlc;
+ this.skipVals = skipVals;
+ this.canRemap = canRemap;
+ this.needVer = needVer;
+ this.keepCacheObjects = keepCacheObjects;
+ this.topVer = topVer;
+
+ futId = IgniteUuid.randomUuid();
+
+ if (log == null)
+ log = U.logger(cctx.kernalContext(), logRef, GridPartitionedSingleGetFuture.class);
+ }
+
+ /**
+ *
+ */
+ public void init() {
+ AffinityTopologyVersion topVer = this.topVer.topologyVersion() > 0 ? this.topVer :
+ canRemap ? cctx.affinity().affinityTopologyVersion() : cctx.shared().exchange().readyAffinityVersion();
+
+ map(topVer);
+ }
+
+ /**
+ * @param topVer Topology version.
+ */
+ @SuppressWarnings("unchecked")
+ private void map(AffinityTopologyVersion topVer) {
+ this.topVer = topVer;
+
+ ClusterNode node = mapKeyToNode(topVer);
+
+ if (node == null) {
+ assert isDone() : this;
+
+ return;
+ }
+
+ if (node.isLocal()) {
+ LinkedHashMap<KeyCacheObject, Boolean> map = U.newLinkedHashMap(1);
+
+ map.put(key, false);
+
+ final GridDhtFuture<Collection<GridCacheEntryInfo>> fut = cctx.dht().getDhtAsync(node.id(),
+ -1,
+ map,
+ readThrough,
+ topVer,
+ subjId,
+ taskName == null ? 0 : taskName.hashCode(),
+ expiryPlc,
+ skipVals);
+
+ final Collection<Integer> invalidParts = fut.invalidPartitions();
+
+ if (!F.isEmpty(invalidParts)) {
+ AffinityTopologyVersion updTopVer = cctx.discovery().topologyVersionEx();
+
+ assert updTopVer.compareTo(topVer) > 0 : "Got invalid partitions for local node but topology " +
+ "version did not change [topVer=" + topVer + ", updTopVer=" + updTopVer +
+ ", invalidParts=" + invalidParts + ']';
+
+ // Remap recursively.
+ map(updTopVer);
+ }
+ else {
+ fut.listen(new CI1<IgniteInternalFuture<Collection<GridCacheEntryInfo>>>() {
+ @Override public void apply(IgniteInternalFuture<Collection<GridCacheEntryInfo>> fut) {
+ try {
+ Collection<GridCacheEntryInfo> infos = fut.get();
+
+ assert F.isEmpty(infos) || infos.size() == 1 : infos;
+
+ setResult(F.first(infos));
+ }
+ catch (Exception e) {
+ U.error(log, "Failed to get values from dht cache [fut=" + fut + "]", e);
+
+ onDone(e);
+ }
+ }
+ });
+ }
+ }
+ else {
+ synchronized (this) {
+ this.node = node;
+ }
+
+ if (!trackable) {
+ trackable = true;
+
+ cctx.mvcc().addFuture(this, futId);
+ }
+
+ GridCacheMessage req;
+
+ if (node.version().compareTo(SINGLE_GET_MSG_SINCE) >= 0) {
+ req = new GridNearSingleGetRequest(cctx.cacheId(),
+ futId,
+ key,
+ readThrough,
+ topVer,
+ subjId,
+ taskName == null ? 0 : taskName.hashCode(),
+ expiryPlc != null ? expiryPlc.forAccess() : -1L,
+ skipVals,
+ /**add reader*/false,
+ needVer,
+ cctx.deploymentEnabled());
+ }
+ else {
+ LinkedHashMap<KeyCacheObject, Boolean> map = U.newLinkedHashMap(1);
+
+ map.put(key, false);
+
+ req = new GridNearGetRequest(
+ cctx.cacheId(),
+ futId,
+ futId,
+ cctx.versions().next(),
+ map,
+ readThrough,
+ topVer,
+ subjId,
+ taskName == null ? 0 : taskName.hashCode(),
+ expiryPlc != null ? expiryPlc.forAccess() : -1L,
+ skipVals,
+ cctx.deploymentEnabled());
+ }
+
+ try {
+ cctx.io().send(node, req, cctx.ioPolicy());
+ }
+ catch (IgniteCheckedException e) {
+ if (e instanceof ClusterTopologyCheckedException)
+ onNodeLeft(node.id());
+ else
+ onDone(e);
+ }
+ }
+ }
+
+ /**
+ * @param topVer Topology version.
+ * @return Primary node or {@code null} if future was completed.
+ */
+ @Nullable private ClusterNode mapKeyToNode(AffinityTopologyVersion topVer) {
+ ClusterNode primary = affinityNode(key, topVer);
+
+ if (primary == null) {
+ onDone(new ClusterTopologyServerNotFoundException("Failed to map keys for cache " +
+ "(all partition nodes left the grid) [topVer=" + topVer + ", cache=" + cctx.name() + ']'));
+
+ return null;
+ }
+
+ boolean allowLocRead = (cctx.affinityNode() && !forcePrimary) || primary.isLocal();
+
+ if (allowLocRead) {
+ GridDhtCacheAdapter colocated = cctx.dht();
+
+ while (true) {
+ GridCacheEntryEx entry;
+
+ try {
+ entry = colocated.context().isSwapOrOffheapEnabled() ? colocated.entryEx(key) :
+ colocated.peekEx(key);
+
+ // If our DHT cache do has value, then we peek it.
+ if (entry != null) {
+ boolean isNew = entry.isNewLocked();
+
+ CacheObject v = null;
+ GridCacheVersion ver = null;
+
+ if (needVer) {
+ T2<CacheObject, GridCacheVersion> res = entry.innerGetVersioned(
+ null,
+ /*swap*/true,
+ /*unmarshal*/true,
+ /**update-metrics*/false,
+ /*event*/!skipVals,
+ subjId,
+ null,
+ taskName,
+ expiryPlc);
+
+ if (res != null) {
+ v = res.get1();
+ ver = res.get2();
+ }
+ }
+ else {
+ v = entry.innerGet(null,
+ /*swap*/true,
+ /*read-through*/false,
+ /*fail-fast*/true,
+ /*unmarshal*/true,
+ /**update-metrics*/false,
+ /*event*/!skipVals,
+ /*temporary*/false,
+ subjId,
+ null,
+ taskName,
+ expiryPlc);
+ }
+
+ colocated.context().evicts().touch(entry, topVer);
+
+ // Entry was not in memory or in swap, so we remove it from cache.
+ if (v == null) {
+ if (isNew && entry.markObsoleteIfEmpty(ver))
+ colocated.removeIfObsolete(key);
+ }
+ else {
+ if (!skipVals && cctx.config().isStatisticsEnabled())
+ cctx.cache().metrics0().onRead(true);
+
+ if (!skipVals)
+ setResult(v, ver);
+ else
+ setSkipValueResult(true, ver);
+
+ return null;
+ }
+ }
+
+ break;
+ }
+ catch (GridDhtInvalidPartitionException ignored) {
+ break;
+ }
+ catch (IgniteCheckedException e) {
+ onDone(e);
+
+ return null;
+ }
+ catch (GridCacheEntryRemovedException ignored) {
+ // No-op, will retry.
+ }
+ }
+ }
+
+ return primary;
+ }
+
+ /**
+ * @param nodeId Node ID.
+ * @param res Result.
+ */
+ public void onResult(UUID nodeId, GridNearSingleGetResponse res) {
+ if (!processResponse(nodeId) || !checkError(res.error(), res.invalidPartitions(), res.topologyVersion()))
+ return;
+
+ Message res0 = res.result();
+
+ if (needVer) {
+ CacheVersionedValue verVal = (CacheVersionedValue)res0;
+
+ if (verVal != null) {
+ if (skipVals)
+ setSkipValueResult(true, verVal.version());
+ else
+ setResult(verVal.value() , verVal.version());
+ }
+ else {
+ if (skipVals)
+ setSkipValueResult(false, null);
+ else
+ setResult(null , null);
+ }
+ }
+ else {
+ if (skipVals)
+ setSkipValueResult(res.containsValue(), null);
+ else
+ setResult((CacheObject)res0, null);
+ }
+ }
+
+ /**
+ * @param nodeId Node ID.
+ * @param res Result.
+ */
+ public void onResult(UUID nodeId, GridNearGetResponse res) {
+ if (!processResponse(nodeId) ||
+ !checkError(res.error(), !F.isEmpty(res.invalidPartitions()), res.topologyVersion()))
+ return;
+
+ Collection<GridCacheEntryInfo> infos = res.entries();
+
+ assert F.isEmpty(infos) || infos.size() == 1 : infos;
+
+ setResult(F.first(infos));
+ }
+
+ /**
+ * @param nodeId Node ID.
+ * @return {@code True} if should process received response.
+ */
+ private boolean processResponse(UUID nodeId) {
+ synchronized (this) {
+ if (node != null && node.id().equals(nodeId)) {
+ node = null;
+
+ return true;
+ }
+ }
+
+ return false;
+ }
+
+ /**
+ * @param err Error.
+ * @param invalidParts Invalid partitions error flag.
+ * @param rmtTopVer Received topology version.
+ */
+ private boolean checkError(@Nullable IgniteCheckedException err,
+ boolean invalidParts,
+ AffinityTopologyVersion rmtTopVer) {
+ if (err != null) {
+ onDone(err);
+
+ return false;
+ }
+
+ if (invalidParts) {
+ assert !rmtTopVer.equals(AffinityTopologyVersion.ZERO);
+
+ if (rmtTopVer.compareTo(topVer) <= 0) {
+ // Fail the whole get future.
+ onDone(new IgniteCheckedException("Failed to process invalid partitions response (remote node reported " +
+ "invalid partitions but remote topology version does not differ from local) " +
+ "[topVer=" + topVer + ", rmtTopVer=" + rmtTopVer + ", part=" + cctx.affinity().partition(key) +
+ ", nodeId=" + node.id() + ']'));
+
+ return false;
+ }
+
+ if (canRemap) {
+ IgniteInternalFuture<Long> topFut = cctx.discovery().topologyFuture(rmtTopVer.topologyVersion());
+
+ topFut.listen(new CIX1<IgniteInternalFuture<Long>>() {
+ @Override public void applyx(IgniteInternalFuture<Long> fut) {
+ try {
+ AffinityTopologyVersion topVer = new AffinityTopologyVersion(fut.get());
+
+ remap(topVer);
+ }
+ catch (IgniteCheckedException e) {
+ onDone(e);
+ }
+ }
+ });
+
+ }
+ else
+ map(topVer);
+
+ return false;
+ }
+
+ return true;
+ }
+
+ /**
+ * @param info Entry info.
+ */
+ private void setResult(@Nullable GridCacheEntryInfo info) {
+ assert info == null || skipVals == (info.value() == null);
+
+ if (skipVals) {
+ if (info != null)
+ setSkipValueResult(true, info.version());
+ else
+ setSkipValueResult(false, null);
+ }
+ else {
+ if (info != null)
+ setResult(info.value(), info.version());
+ else
+ setResult(null, null);
+ }
+ }
+
+ /**
+ * @param res Result.
+ * @param ver Version.
+ */
+ private void setSkipValueResult(boolean res, @Nullable GridCacheVersion ver) {
+ assert skipVals;
+
+ if (needVer) {
+ assert ver != null || !res;
+
+ onDone(new T2<>(res, ver));
+ }
+ else
+ onDone(res);
+ }
+
+ /**
+ * @param val Value.
+ * @param ver Version.
+ */
+ private void setResult(@Nullable CacheObject val, @Nullable GridCacheVersion ver) {
+ try {
+ assert !skipVals;
+
+ if (val != null) {
+ if (needVer) {
+ assert ver != null;
+
+ onDone(new T2<>(val, ver));
+ }
+ else {
+ if (!keepCacheObjects) {
+ Object res = CU.value(val, cctx, true);
+
+ if (deserializePortable && !skipVals)
+ res = cctx.unwrapPortableIfNeeded(res, false);
+
+ onDone(res);
+ }
+ else
+ onDone(val);
+ }
+ }
+ else
+ onDone(null);
+ }
+ catch (Exception e) {
+ onDone(e);
+ }
+ }
+
+ /**
+ * Affinity node to send get request to.
+ *
+ * @param key Key to get.
+ * @param topVer Topology version.
+ * @return Affinity node to get key from.
+ */
+ private ClusterNode affinityNode(KeyCacheObject key, AffinityTopologyVersion topVer) {
+ if (!canRemap) {
+ List<ClusterNode> affNodes = cctx.affinity().nodes(key, topVer);
+
+ for (ClusterNode node : affNodes) {
+ if (cctx.discovery().alive(node))
+ return node;
+ }
+
+ return null;
+ }
+ else
+ return cctx.affinity().primary(key, topVer);
+ }
+
+ /** {@inheritDoc} */
+ @Override public IgniteUuid futureId() {
+ return futId;
+ }
+
+ /** {@inheritDoc} */
+ @Override public boolean onNodeLeft(UUID nodeId) {
+ if (!processResponse(nodeId))
+ return false;
+
+ if (canRemap) {
+ final AffinityTopologyVersion updTopVer = new AffinityTopologyVersion(
+ Math.max(topVer.topologyVersion() + 1, cctx.discovery().topologyVersion()));
+
+ cctx.affinity().affinityReadyFuture(updTopVer).listen(
+ new CI1<IgniteInternalFuture<AffinityTopologyVersion>>() {
+ @Override public void apply(IgniteInternalFuture<AffinityTopologyVersion> fut) {
+ try {
+ fut.get();
+
+ remap(updTopVer);
+ }
+ catch (IgniteCheckedException e) {
+ onDone(e);
+ }
+ }
+ });
+ }
+ else
+ remap(topVer);
+
+ return true;
+ }
+
+ /**
+ * @param topVer Topology version.
+ */
+ private void remap(final AffinityTopologyVersion topVer) {
+ cctx.closures().runLocalSafe(new Runnable() {
+ @Override public void run() {
+ map(topVer);
+ }
+ });
+ }
+
+ /** {@inheritDoc} */
+ @Override public boolean onDone(Object res, Throwable err) {
+ if (super.onDone(res, err)) {
+ // Don't forget to clean up.
+ if (trackable)
+ cctx.mvcc().removeFuture(futId);
+
+ cctx.dht().sendTtlUpdateRequest(expiryPlc);
+
+ return true;
+ }
+
+ return false;
+ }
+
+ /** {@inheritDoc} */
+ @Override public boolean trackable() {
+ return trackable;
+ }
+
+ /** {@inheritDoc} */
+ @Override public void markNotTrackable() {
+ // No-op.
+ }
+
+ /** {@inheritDoc} */
+ @Override public String toString() {
+ return S.toString(GridPartitionedSingleGetFuture.class, this);
+ }
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/1f103067/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicCache.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicCache.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicCache.java
index 7f9edb2..75f8c2f 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicCache.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicCache.java
@@ -65,11 +65,14 @@ import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtCacheE
import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtFuture;
import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtInvalidPartitionException;
import org.apache.ignite.internal.processors.cache.distributed.dht.GridPartitionedGetFuture;
+import org.apache.ignite.internal.processors.cache.distributed.dht.GridPartitionedSingleGetFuture;
import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.GridDhtPreloader;
import org.apache.ignite.internal.processors.cache.distributed.near.GridNearAtomicCache;
import org.apache.ignite.internal.processors.cache.distributed.near.GridNearCacheAdapter;
import org.apache.ignite.internal.processors.cache.distributed.near.GridNearGetRequest;
import org.apache.ignite.internal.processors.cache.distributed.near.GridNearGetResponse;
+import org.apache.ignite.internal.processors.cache.distributed.near.GridNearSingleGetRequest;
+import org.apache.ignite.internal.processors.cache.distributed.near.GridNearSingleGetResponse;
import org.apache.ignite.internal.processors.cache.dr.GridCacheDrExpirationInfo;
import org.apache.ignite.internal.processors.cache.dr.GridCacheDrInfo;
import org.apache.ignite.internal.processors.cache.transactions.IgniteTxLocalEx;
@@ -242,6 +245,12 @@ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K, V> {
}
});
+ ctx.io().addHandler(ctx.cacheId(), GridNearSingleGetRequest.class, new CI2<UUID, GridNearSingleGetRequest>() {
+ @Override public void apply(UUID nodeId, GridNearSingleGetRequest req) {
+ processNearSingleGetRequest(nodeId, req);
+ }
+ });
+
ctx.io().addHandler(ctx.cacheId(), GridNearAtomicUpdateRequest.class, new CI2<UUID, GridNearAtomicUpdateRequest>() {
@Override public void apply(UUID nodeId, GridNearAtomicUpdateRequest req) {
processNearAtomicUpdateRequest(nodeId, req);
@@ -279,6 +288,12 @@ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K, V> {
processNearGetResponse(nodeId, res);
}
});
+
+ ctx.io().addHandler(ctx.cacheId(), GridNearSingleGetResponse.class, new CI2<UUID, GridNearSingleGetResponse>() {
+ @Override public void apply(UUID nodeId, GridNearSingleGetResponse res) {
+ processNearSingleGetResponse(nodeId, res);
+ }
+ });
}
}
@@ -301,6 +316,45 @@ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K, V> {
}
/** {@inheritDoc} */
+ @Override protected IgniteInternalFuture<V> getAsync(final K key,
+ final boolean forcePrimary,
+ final boolean skipTx,
+ @Nullable UUID subjId,
+ final String taskName,
+ final boolean deserializePortable,
+ final boolean skipVals,
+ final boolean canRemap) {
+ ctx.checkSecurity(SecurityPermission.CACHE_READ);
+
+ if (keyCheck)
+ validateCacheKey(key);
+
+ CacheOperationContext opCtx = ctx.operationContextPerCall();
+
+ subjId = ctx.subjectIdPerCall(null, opCtx);
+
+ final UUID subjId0 = subjId;
+
+ final ExpiryPolicy expiryPlc = skipVals ? null : opCtx != null ? opCtx.expiry() : null;
+
+ final boolean skipStore = opCtx != null && opCtx.skipStore();
+
+ return asyncOp(new CO<IgniteInternalFuture<V>>() {
+ @Override public IgniteInternalFuture<V> apply() {
+ return getAsync0(ctx.toCacheKeyObject(key),
+ forcePrimary,
+ subjId0,
+ taskName,
+ deserializePortable,
+ expiryPlc,
+ skipVals,
+ skipStore,
+ canRemap);
+ }
+ });
+ }
+
+ /** {@inheritDoc} */
@Override public IgniteInternalFuture<Map<K, V>> getAllAsync(
@Nullable final Collection<? extends K> keys,
final boolean forcePrimary,
@@ -914,9 +968,57 @@ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K, V> {
}
/**
+ * Entry point to all public API single get methods.
+ *
+ * @param key Key.
+ * @param forcePrimary Force primary flag.
+ * @param subjId Subject ID.
+ * @param taskName Task name.
+ * @param deserializePortable Deserialize portable flag.
+ * @param expiryPlc Expiry policy.
+ * @param skipVals Skip values flag.
+ * @param skipStore Skip store flag.
+ * @param canRemap Can remap flag.
+ * @return Get future.
+ */
+ private IgniteInternalFuture<V> getAsync0(KeyCacheObject key,
+ boolean forcePrimary,
+ UUID subjId,
+ String taskName,
+ boolean deserializePortable,
+ @Nullable ExpiryPolicy expiryPlc,
+ boolean skipVals,
+ boolean skipStore,
+ boolean canRemap
+ ) {
+ AffinityTopologyVersion topVer = canRemap ? ctx.affinity().affinityTopologyVersion() :
+ ctx.shared().exchange().readyAffinityVersion();
+
+ IgniteCacheExpiryPolicy expiry = skipVals ? null : expiryPolicy(expiryPlc);
+
+ GridPartitionedSingleGetFuture fut = new GridPartitionedSingleGetFuture(ctx,
+ key,
+ topVer,
+ !skipStore,
+ forcePrimary,
+ subjId,
+ taskName,
+ deserializePortable,
+ expiry,
+ skipVals,
+ canRemap,
+ false,
+ false);
+
+ fut.init();
+
+ return (IgniteInternalFuture<V>)fut;
+ }
+
+ /**
* Entry point to all public API get methods.
*
- * @param keys Keys to remove.
+ * @param keys Keys.
* @param forcePrimary Force primary flag.
* @param subjId Subject ID.
* @param taskName Task name.
@@ -942,7 +1044,7 @@ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K, V> {
final IgniteCacheExpiryPolicy expiry = skipVals ? null : expiryPolicy(expiryPlc);
// Optimisation: try to resolve value locally and escape 'get future' creation.
- if (!forcePrimary) {
+ if (!forcePrimary && ctx.affinityNode()) {
Map<K, V> locVals = U.newHashMap(keys.size());
boolean success = true;
@@ -2409,27 +2511,6 @@ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K, V> {
/**
* @param nodeId Sender node ID.
- * @param res Near get response.
- */
- private void processNearGetResponse(UUID nodeId, GridNearGetResponse res) {
- if (log.isDebugEnabled())
- log.debug("Processing near get response [nodeId=" + nodeId + ", res=" + res + ']');
-
- GridPartitionedGetFuture<K, V> fut = (GridPartitionedGetFuture<K, V>)ctx.mvcc().<Map<K, V>>future(
- res.version(), res.futureId());
-
- if (fut == null) {
- if (log.isDebugEnabled())
- log.debug("Failed to find future for get response [sender=" + nodeId + ", res=" + res + ']');
-
- return;
- }
-
- fut.onResult(nodeId, res);
- }
-
- /**
- * @param nodeId Sender node ID.
* @param req Near atomic update request.
*/
private void processNearAtomicUpdateRequest(UUID nodeId, GridNearAtomicUpdateRequest req) {
http://git-wip-us.apache.org/repos/asf/ignite/blob/1f103067/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicUpdateFuture.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicUpdateFuture.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicUpdateFuture.java
index 4ace5c4..c34dcfd 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicUpdateFuture.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicUpdateFuture.java
@@ -22,7 +22,6 @@ import java.util.Collection;
import java.util.HashMap;
import java.util.Map;
import java.util.UUID;
-import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.atomic.AtomicReference;
import javax.cache.processor.EntryProcessor;
import org.apache.ignite.IgniteCheckedException;
@@ -48,7 +47,6 @@ import org.apache.ignite.internal.util.typedef.internal.S;
import org.apache.ignite.internal.util.typedef.internal.U;
import org.apache.ignite.lang.IgniteUuid;
import org.jetbrains.annotations.Nullable;
-import org.jsr166.ConcurrentHashMap8;
import static org.apache.ignite.cache.CacheWriteSynchronizationMode.FULL_SYNC;
@@ -67,39 +65,42 @@ public class GridDhtAtomicUpdateFuture extends GridFutureAdapter<Void>
protected static IgniteLogger log;
/** Cache context. */
- private GridCacheContext cctx;
+ private final GridCacheContext cctx;
/** Future version. */
- private GridCacheVersion futVer;
+ private final GridCacheVersion futVer;
/** Write version. */
- private GridCacheVersion writeVer;
+ private final GridCacheVersion writeVer;
/** Force transform backup flag. */
private boolean forceTransformBackups;
/** Completion callback. */
@GridToStringExclude
- private CI2<GridNearAtomicUpdateRequest, GridNearAtomicUpdateResponse> completionCb;
+ private final CI2<GridNearAtomicUpdateRequest, GridNearAtomicUpdateResponse> completionCb;
/** Mappings. */
@GridToStringInclude
- private ConcurrentMap<UUID, GridDhtAtomicUpdateRequest> mappings = new ConcurrentHashMap8<>();
+ private final Map<UUID, GridDhtAtomicUpdateRequest> mappings;
/** Entries with readers. */
private Map<KeyCacheObject, GridDhtCacheEntry> nearReadersEntries;
/** Update request. */
- private GridNearAtomicUpdateRequest updateReq;
+ private final GridNearAtomicUpdateRequest updateReq;
/** Update response. */
- private GridNearAtomicUpdateResponse updateRes;
+ private final GridNearAtomicUpdateResponse updateRes;
/** Future keys. */
- private Collection<KeyCacheObject> keys;
+ private final Collection<KeyCacheObject> keys;
/** */
- private boolean waitForExchange;
+ private final boolean waitForExchange;
+
+ /** Response count. */
+ private volatile int resCnt;
/**
* @param cctx Cache context.
@@ -128,6 +129,7 @@ public class GridDhtAtomicUpdateFuture extends GridFutureAdapter<Void>
log = U.logger(cctx.kernalContext(), logRef, GridDhtAtomicUpdateFuture.class);
keys = new ArrayList<>(updateReq.keys().size());
+ mappings = U.newHashMap(updateReq.keys().size());
boolean topLocked = updateReq.topologyLocked() || (updateReq.fastMap() && !updateReq.clientRequest());
@@ -145,22 +147,37 @@ public class GridDhtAtomicUpdateFuture extends GridFutureAdapter<Void>
}
/** {@inheritDoc} */
- @Override public Collection<? extends ClusterNode> nodes() {
- return F.view(F.viewReadOnly(mappings.keySet(), U.id2Node(cctx.kernalContext())), F.notNull());
- }
-
- /** {@inheritDoc} */
@Override public boolean onNodeLeft(UUID nodeId) {
if (log.isDebugEnabled())
log.debug("Processing node leave event [fut=" + this + ", nodeId=" + nodeId + ']');
+ return registerResponse(nodeId);
+ }
+
+ /**
+ * @param nodeId Node ID.
+ * @return {@code True} if request found.
+ */
+ private boolean registerResponse(UUID nodeId) {
+ int resCnt0;
+
GridDhtAtomicUpdateRequest req = mappings.get(nodeId);
if (req != null) {
- // Remove only after added keys to failed set.
- mappings.remove(nodeId);
+ synchronized (this) {
+ if (req.onResponse()) {
+ resCnt0 = resCnt;
+
+ resCnt0 += 1;
+
+ resCnt = resCnt0;
+ }
+ else
+ return false;
+ }
- checkComplete();
+ if (resCnt0 == mappings.size())
+ onDone();
return true;
}
@@ -343,18 +360,18 @@ public class GridDhtAtomicUpdateFuture extends GridFutureAdapter<Void>
U.warn(log, "Failed to send update request to backup node because it left grid: " +
req.nodeId());
- mappings.remove(req.nodeId());
+ registerResponse(req.nodeId());
}
catch (IgniteCheckedException e) {
U.error(log, "Failed to send update request to backup node (did node leave the grid?): "
+ req.nodeId(), e);
- mappings.remove(req.nodeId());
+ registerResponse(req.nodeId());
}
}
}
-
- checkComplete();
+ else
+ onDone();
// Send response right away if no ACKs from backup is required.
// Backups will send ACKs anyway, future will be completed after all backups have replied.
@@ -389,9 +406,7 @@ public class GridDhtAtomicUpdateFuture extends GridFutureAdapter<Void>
}
}
- mappings.remove(nodeId);
-
- checkComplete();
+ registerResponse(nodeId);
}
/**
@@ -403,22 +418,7 @@ public class GridDhtAtomicUpdateFuture extends GridFutureAdapter<Void>
if (log.isDebugEnabled())
log.debug("Received deferred DHT atomic update future result [nodeId=" + nodeId + ']');
- mappings.remove(nodeId);
-
- checkComplete();
- }
-
- /**
- * Checks if all required responses are received.
- */
- private void checkComplete() {
- // Always wait for replies from all backups.
- if (mappings.isEmpty()) {
- if (log.isDebugEnabled())
- log.debug("Completing DHT atomic update future: " + this);
-
- onDone();
- }
+ registerResponse(nodeId);
}
/** {@inheritDoc} */
http://git-wip-us.apache.org/repos/asf/ignite/blob/1f103067/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicUpdateRequest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicUpdateRequest.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicUpdateRequest.java
index e55cac9..1219f2f 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicUpdateRequest.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicUpdateRequest.java
@@ -139,6 +139,10 @@ public class GridDhtAtomicUpdateRequest extends GridCacheMessage implements Grid
/** Task name hash. */
private int taskNameHash;
+ /** On response flag. Access should be synced on future. */
+ @GridDirectTransient
+ private boolean onRes;
+
/**
* Empty constructor required by {@link Externalizable}.
*/
@@ -527,6 +531,13 @@ public class GridDhtAtomicUpdateRequest extends GridCacheMessage implements Grid
}
/**
+ * @return {@code True} if on response flag changed.
+ */
+ public boolean onResponse() {
+ return !onRes && (onRes = true);
+ }
+
+ /**
* @return Optional arguments for entry processor.
*/
@Nullable public Object[] invokeArguments() {
http://git-wip-us.apache.org/repos/asf/ignite/blob/1f103067/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridNearAtomicUpdateFuture.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridNearAtomicUpdateFuture.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridNearAtomicUpdateFuture.java
index ae662c8..a786803 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridNearAtomicUpdateFuture.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridNearAtomicUpdateFuture.java
@@ -238,11 +238,6 @@ public class GridNearAtomicUpdateFuture extends GridFutureAdapter<Object>
return state.futureVersion();
}
- /** {@inheritDoc} */
- @Override public Collection<? extends ClusterNode> nodes() {
- throw new UnsupportedOperationException();
- }
-
/**
* @return {@code True} if this future should block partition map exchange.
*/
http://git-wip-us.apache.org/repos/asf/ignite/blob/1f103067/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 7131aa5..47b7aea 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
@@ -53,8 +53,10 @@ import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtInvali
import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtLockFuture;
import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtTransactionalCacheAdapter;
import org.apache.ignite.internal.processors.cache.distributed.dht.GridPartitionedGetFuture;
+import org.apache.ignite.internal.processors.cache.distributed.dht.GridPartitionedSingleGetFuture;
import org.apache.ignite.internal.processors.cache.distributed.near.GridNearGetResponse;
import org.apache.ignite.internal.processors.cache.distributed.near.GridNearLockResponse;
+import org.apache.ignite.internal.processors.cache.distributed.near.GridNearSingleGetResponse;
import org.apache.ignite.internal.processors.cache.distributed.near.GridNearTransactionalCache;
import org.apache.ignite.internal.processors.cache.distributed.near.GridNearTxLocal;
import org.apache.ignite.internal.processors.cache.distributed.near.GridNearUnlockRequest;
@@ -67,6 +69,7 @@ import org.apache.ignite.internal.util.future.GridFinishedFuture;
import org.apache.ignite.internal.util.lang.IgnitePair;
import org.apache.ignite.internal.util.typedef.C2;
import org.apache.ignite.internal.util.typedef.CI2;
+import org.apache.ignite.internal.util.typedef.CX1;
import org.apache.ignite.internal.util.typedef.F;
import org.apache.ignite.internal.util.typedef.T2;
import org.apache.ignite.internal.util.typedef.internal.CU;
@@ -138,7 +141,13 @@ public class GridDhtColocatedCache<K, V> extends GridDhtTransactionalCacheAdapte
ctx.io().addHandler(ctx.cacheId(), GridNearGetResponse.class, new CI2<UUID, GridNearGetResponse>() {
@Override public void apply(UUID nodeId, GridNearGetResponse res) {
- processGetResponse(nodeId, res);
+ processNearGetResponse(nodeId, res);
+ }
+ });
+
+ ctx.io().addHandler(ctx.cacheId(), GridNearSingleGetResponse.class, new CI2<UUID, GridNearSingleGetResponse>() {
+ @Override public void apply(UUID nodeId, GridNearSingleGetResponse res) {
+ processNearSingleGetResponse(nodeId, res);
}
});
@@ -185,6 +194,80 @@ public class GridDhtColocatedCache<K, V> extends GridDhtTransactionalCacheAdapte
}
/** {@inheritDoc} */
+ @Override protected IgniteInternalFuture<V> getAsync(final K key,
+ boolean forcePrimary,
+ boolean skipTx,
+ @Nullable UUID subjId,
+ String taskName,
+ final boolean deserializePortable,
+ final boolean skipVals,
+ boolean canRemap) {
+ ctx.checkSecurity(SecurityPermission.CACHE_READ);
+
+ if (keyCheck)
+ validateCacheKey(key);
+
+ IgniteTxLocalAdapter tx = ctx.tm().threadLocalTx(ctx);
+
+ final CacheOperationContext opCtx = ctx.operationContextPerCall();
+
+ if (tx != null && !tx.implicit() && !skipTx) {
+ return asyncOp(tx, new AsyncOp<V>() {
+ @Override public IgniteInternalFuture<V> op(IgniteTxLocalAdapter tx) {
+ IgniteInternalFuture<Map<Object, Object>> fut = tx.getAllAsync(ctx,
+ Collections.singleton(ctx.toCacheKeyObject(key)),
+ deserializePortable,
+ skipVals,
+ false,
+ opCtx != null && opCtx.skipStore());
+
+ return fut.chain(new CX1<IgniteInternalFuture<Map<Object, Object>>, V>() {
+ @SuppressWarnings("unchecked")
+ @Override public V applyx(IgniteInternalFuture<Map<Object, Object>> e)
+ throws IgniteCheckedException {
+ Map<Object, Object> map = e.get();
+
+ assert map.isEmpty() || map.size() == 1 : map.size();
+
+ if (skipVals) {
+ Boolean val = map.isEmpty() ? false : (Boolean)F.firstValue(map);
+
+ return (V)(val);
+ }
+
+ return (V)map.get(key);
+ }
+ });
+ }
+ });
+ }
+
+ AffinityTopologyVersion topVer = tx == null ?
+ (canRemap ? ctx.affinity().affinityTopologyVersion() : ctx.shared().exchange().readyAffinityVersion()) :
+ tx.topologyVersion();
+
+ subjId = ctx.subjectIdPerCall(subjId, opCtx);
+
+ GridPartitionedSingleGetFuture fut = new GridPartitionedSingleGetFuture(ctx,
+ ctx.toCacheKeyObject(key),
+ topVer,
+ opCtx == null || !opCtx.skipStore(),
+ forcePrimary,
+ subjId,
+ taskName,
+ deserializePortable,
+ skipVals ? null : expiryPolicy(opCtx != null ? opCtx.expiry() : null),
+ skipVals,
+ canRemap,
+ /*needVer*/false,
+ /*keepCacheObjects*/false);
+
+ fut.init();
+
+ return (IgniteInternalFuture<V>)fut;
+ }
+
+ /** {@inheritDoc} */
@Override public IgniteInternalFuture<Map<K, V>> getAllAsync(
@Nullable final Collection<? extends K> keys,
boolean forcePrimary,
@@ -290,6 +373,54 @@ public class GridDhtColocatedCache<K, V> extends GridDhtTransactionalCacheAdapte
}
/**
+ * @param key Key to load.
+ * @param readThrough Read through flag.
+ * @param forcePrimary Force get from primary node flag.
+ * @param topVer Topology version.
+ * @param subjId Subject ID.
+ * @param taskName Task name.
+ * @param deserializePortable Deserialize portable flag.
+ * @param expiryPlc Expiry policy.
+ * @param skipVals Skip values flag.
+ * @param canRemap Flag indicating whether future can be remapped on a newer topology version.
+ * @param needVer If {@code true} returns values as tuples containing value and version.
+ * @param keepCacheObj Keep cache objects flag.
+ * @return Load future.
+ */
+ public final IgniteInternalFuture<Object> loadAsync(
+ KeyCacheObject key,
+ boolean readThrough,
+ boolean forcePrimary,
+ AffinityTopologyVersion topVer,
+ @Nullable UUID subjId,
+ String taskName,
+ boolean deserializePortable,
+ @Nullable IgniteCacheExpiryPolicy expiryPlc,
+ boolean skipVals,
+ boolean canRemap,
+ boolean needVer,
+ boolean keepCacheObj
+ ) {
+ GridPartitionedSingleGetFuture fut = new GridPartitionedSingleGetFuture(ctx,
+ ctx.toCacheKeyObject(key),
+ topVer,
+ readThrough,
+ forcePrimary,
+ subjId,
+ taskName,
+ deserializePortable,
+ expiryPlc,
+ skipVals,
+ canRemap,
+ needVer,
+ keepCacheObj);
+
+ fut.init();
+
+ return fut;
+ }
+
+ /**
* @param keys Keys to load.
* @param readThrough Read through flag.
* @param forcePrimary Force get from primary node flag.
@@ -299,9 +430,12 @@ public class GridDhtColocatedCache<K, V> extends GridDhtTransactionalCacheAdapte
* @param deserializePortable Deserialize portable flag.
* @param expiryPlc Expiry policy.
* @param skipVals Skip values flag.
- * @return Loaded values.
+ * @param canRemap Flag indicating whether future can be remapped on a newer topology version.
+ * @param needVer If {@code true} returns values as tuples containing value and version.
+ * @param keepCacheObj Keep cache objects flag.
+ * @return Load future.
*/
- public IgniteInternalFuture<Map<K, V>> loadAsync(
+ public final IgniteInternalFuture<Map<K, V>> loadAsync(
@Nullable Collection<KeyCacheObject> keys,
boolean readThrough,
boolean forcePrimary,
@@ -931,24 +1065,6 @@ public class GridDhtColocatedCache<K, V> extends GridDhtTransactionalCacheAdapte
}
/**
- * @param nodeId Sender ID.
- * @param res Response.
- */
- private void processGetResponse(UUID nodeId, GridNearGetResponse res) {
- GridPartitionedGetFuture<K, V> fut = (GridPartitionedGetFuture<K, V>)ctx.mvcc().<Map<K, V>>future(
- res.version(), res.futureId());
-
- if (fut == null) {
- if (log.isDebugEnabled())
- log.debug("Failed to find future for get response [sender=" + nodeId + ", res=" + res + ']');
-
- return;
- }
-
- fut.onResult(nodeId, res);
- }
-
- /**
* @param nodeId Node ID.
* @param res Response.
*/
@@ -957,7 +1073,7 @@ public class GridDhtColocatedCache<K, V> extends GridDhtTransactionalCacheAdapte
assert res != null;
GridDhtColocatedLockFuture fut = (GridDhtColocatedLockFuture)ctx.mvcc().
- <Boolean>future(res.version(), res.futureId());
+ <Boolean>mvccFuture(res.version(), res.futureId());
if (fut != null)
fut.onResult(nodeId, res);
http://git-wip-us.apache.org/repos/asf/ignite/blob/1f103067/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 abeb509..8245d88 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
@@ -35,10 +35,11 @@ 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.GridCacheContext;
+import org.apache.ignite.internal.processors.cache.GridCacheEntryEx;
import org.apache.ignite.internal.processors.cache.GridCacheEntryRemovedException;
-import org.apache.ignite.internal.processors.cache.GridCacheFuture;
import org.apache.ignite.internal.processors.cache.GridCacheLockTimeoutException;
import org.apache.ignite.internal.processors.cache.GridCacheMvccCandidate;
+import org.apache.ignite.internal.processors.cache.GridCacheMvccFuture;
import org.apache.ignite.internal.processors.cache.KeyCacheObject;
import org.apache.ignite.internal.processors.cache.distributed.GridDistributedCacheEntry;
import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtCacheEntry;
@@ -78,7 +79,7 @@ import static org.apache.ignite.events.EventType.EVT_CACHE_OBJECT_READ;
* Colocated cache lock future.
*/
public final class GridDhtColocatedLockFuture extends GridCompoundIdentityFuture<Boolean>
- implements GridCacheFuture<Boolean> {
+ implements GridCacheMvccFuture<Boolean> {
/** */
private static final long serialVersionUID = 0L;
@@ -198,25 +199,16 @@ public final class GridDhtColocatedLockFuture extends GridCompoundIdentityFuture
valMap = new ConcurrentHashMap8<>(keys.size(), 1f);
}
- /**
- * @return Participating nodes.
- */
- @Override public Collection<? extends ClusterNode> nodes() {
- return F.viewReadOnly(futures(), new IgniteClosure<IgniteInternalFuture<?>, ClusterNode>() {
- @Nullable @Override public ClusterNode apply(IgniteInternalFuture<?> f) {
- if (isMini(f))
- return ((MiniFuture)f).node();
-
- return cctx.discovery().localNode();
- }
- });
- }
-
/** {@inheritDoc} */
@Override public GridCacheVersion version() {
return lockVer;
}
+ /** {@inheritDoc} */
+ @Override public boolean onOwnerChanged(GridCacheEntryEx entry, GridCacheMvccCandidate owner) {
+ return false;
+ }
+
/**
* @return Future ID.
*/
@@ -538,7 +530,7 @@ public final class GridDhtColocatedLockFuture extends GridCompoundIdentityFuture
log.debug("Completing future: " + this);
// Clean up.
- cctx.mvcc().removeFuture(this);
+ cctx.mvcc().removeMvccFuture(this);
if (timeoutObj != null)
cctx.time().removeTimeoutObject(timeoutObj);
http://git-wip-us.apache.org/repos/asf/ignite/blob/1f103067/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/CacheVersionedValue.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/CacheVersionedValue.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/CacheVersionedValue.java
index 1559a91..c14621a 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/CacheVersionedValue.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/CacheVersionedValue.java
@@ -53,7 +53,7 @@ public class CacheVersionedValue implements Message {
* @param val Cache value.
* @param ver Cache version.
*/
- CacheVersionedValue(CacheObject val, GridCacheVersion ver) {
+ public CacheVersionedValue(CacheObject val, GridCacheVersion ver) {
this.val = val;
this.ver = ver;
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/1f103067/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearCacheAdapter.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearCacheAdapter.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearCacheAdapter.java
index 3c3527a..eb0b637 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearCacheAdapter.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearCacheAdapter.java
@@ -52,6 +52,7 @@ import org.apache.ignite.internal.processors.cache.GridCacheValueCollection;
import org.apache.ignite.internal.processors.cache.IgniteCacheExpiryPolicy;
import org.apache.ignite.internal.processors.cache.KeyCacheObject;
import org.apache.ignite.internal.processors.cache.distributed.GridDistributedCacheAdapter;
+import org.apache.ignite.internal.processors.cache.distributed.dht.CacheGetFuture;
import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtCacheAdapter;
import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtCacheEntry;
import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtLocalPartition;
@@ -292,8 +293,7 @@ public abstract class GridNearCacheAdapter<K, V> extends GridDistributedCacheAda
* @param res Response.
*/
protected void processGetResponse(UUID nodeId, GridNearGetResponse res) {
- GridNearGetFuture<K, V> fut = (GridNearGetFuture<K, V>)ctx.mvcc().<Map<K, V>>future(
- res.version(), res.futureId());
+ CacheGetFuture fut = (CacheGetFuture)ctx.mvcc().future(res.futureId());
if (fut == null) {
if (log.isDebugEnabled())
http://git-wip-us.apache.org/repos/asf/ignite/blob/1f103067/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearGetFuture.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearGetFuture.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearGetFuture.java
index ae1d43c..dfaa44e 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearGetFuture.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearGetFuture.java
@@ -170,24 +170,6 @@ public final class GridNearGetFuture<K, V> extends CacheDistributedGetFutureAdap
}
/** {@inheritDoc} */
- @Override public GridCacheVersion version() {
- return ver;
- }
-
- /** {@inheritDoc} */
- @Override public Collection<? extends ClusterNode> nodes() {
- return
- F.viewReadOnly(futures(), new IgniteClosure<IgniteInternalFuture<Map<K, V>>, ClusterNode>() {
- @Nullable @Override public ClusterNode apply(IgniteInternalFuture<Map<K, V>> f) {
- if (isMini(f))
- return ((MiniFuture)f).node();
-
- return cctx.discovery().localNode();
- }
- });
- }
-
- /** {@inheritDoc} */
@Override public boolean onNodeLeft(UUID nodeId) {
boolean found = false;
@@ -227,7 +209,7 @@ public final class GridNearGetFuture<K, V> extends CacheDistributedGetFutureAdap
if (super.onDone(res, err)) {
// Don't forget to clean up.
if (trackable)
- cctx.mvcc().removeFuture(this);
+ cctx.mvcc().removeFuture(futId);
cache().dht().sendTtlUpdateRequest(expiryPlc);
@@ -343,7 +325,7 @@ public final class GridNearGetFuture<K, V> extends CacheDistributedGetFutureAdap
if (!trackable) {
trackable = true;
- cctx.mvcc().addFuture(this);
+ cctx.mvcc().addFuture(this, futId);
}
MiniFuture fut = new MiniFuture(n, mappedKeys, saved, topVer);
@@ -386,6 +368,7 @@ public final class GridNearGetFuture<K, V> extends CacheDistributedGetFutureAdap
* @param saved Reserved near cache entries.
* @return Map.
*/
+ @SuppressWarnings("unchecked")
private Map<KeyCacheObject, GridNearCacheEntry> map(
KeyCacheObject key,
Map<ClusterNode, LinkedHashMap<KeyCacheObject, Boolean>> mappings,
@@ -538,11 +521,17 @@ public final class GridNearGetFuture<K, V> extends CacheDistributedGetFutureAdap
}
else {
K key0 = key.value(cctx.cacheObjectContext(), true);
- V val0 = v.value(cctx.cacheObjectContext(), true);
-
- val0 = (V)cctx.unwrapPortableIfNeeded(val0, !deserializePortable);
key0 = (K)cctx.unwrapPortableIfNeeded(key0, !deserializePortable);
+ V val0;
+
+ if (!skipVals) {
+ val0 = v.value(cctx.cacheObjectContext(), true);
+ val0 = (V)cctx.unwrapPortableIfNeeded(val0, !deserializePortable);
+ }
+ else
+ val0 = (V)Boolean.TRUE;
+
add(new GridFinishedFuture<>(Collections.singletonMap(key0, val0)));
}
}
@@ -618,28 +607,6 @@ public final class GridNearGetFuture<K, V> extends CacheDistributedGetFutureAdap
}
/**
- * Affinity node to send get request to.
- *
- * @param key Key to get.
- * @param topVer Topology version.
- * @return Affinity node to get key from.
- */
- private ClusterNode affinityNode(KeyCacheObject key, AffinityTopologyVersion topVer) {
- if (!canRemap) {
- List<ClusterNode> affNodes = cctx.affinity().nodes(key, topVer);
-
- for (ClusterNode node : affNodes) {
- if (cctx.discovery().alive(node))
- return node;
- }
-
- return null;
- }
- else
- return cctx.affinity().primary(key, topVer);
- }
-
- /**
* @return Near cache.
*/
private GridNearCacheAdapter<K, V> cache() {
http://git-wip-us.apache.org/repos/asf/ignite/blob/1f103067/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearGetRequest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearGetRequest.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearGetRequest.java
index 8482217..6d60298 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearGetRequest.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearGetRequest.java
@@ -133,7 +133,6 @@ public class GridNearGetRequest extends GridCacheMessage implements GridCacheDep
) {
assert futId != null;
assert miniId != null;
- assert ver != null;
assert keys != null;
this.cacheId = cacheId;
http://git-wip-us.apache.org/repos/asf/ignite/blob/1f103067/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearGetResponse.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearGetResponse.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearGetResponse.java
index fc06ab1..15a791f 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearGetResponse.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearGetResponse.java
@@ -100,8 +100,6 @@ public class GridNearGetResponse extends GridCacheMessage implements GridCacheDe
boolean addDepInfo
) {
assert futId != null;
- assert miniId != null;
- assert ver != null;
this.cacheId = cacheId;
this.futId = futId;
http://git-wip-us.apache.org/repos/asf/ignite/blob/1f103067/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearLockFuture.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearLockFuture.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearLockFuture.java
index 9c3701f..76f2fbe 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearLockFuture.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearLockFuture.java
@@ -209,20 +209,6 @@ public final class GridNearLockFuture extends GridCompoundIdentityFuture<Boolean
valMap = new ConcurrentHashMap8<>(keys.size(), 1f);
}
- /**
- * @return Participating nodes.
- */
- @Override public Collection<? extends ClusterNode> nodes() {
- return F.viewReadOnly(futures(), new IgniteClosure<IgniteInternalFuture<?>, ClusterNode>() {
- @Nullable @Override public ClusterNode apply(IgniteInternalFuture<?> f) {
- if (isMini(f))
- return ((MiniFuture)f).node();
-
- return cctx.discovery().localNode();
- }
- });
- }
-
/** {@inheritDoc} */
@Override public GridCacheVersion version() {
return lockVer;
@@ -672,7 +658,7 @@ public final class GridNearLockFuture extends GridCompoundIdentityFuture<Boolean
log.debug("Completing future: " + this);
// Clean up.
- cctx.mvcc().removeFuture(this);
+ cctx.mvcc().removeMvccFuture(this);
if (timeoutObj != null)
cctx.time().removeTimeoutObject(timeoutObj);
http://git-wip-us.apache.org/repos/asf/ignite/blob/1f103067/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearOptimisticSerializableTxPrepareFuture.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearOptimisticSerializableTxPrepareFuture.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearOptimisticSerializableTxPrepareFuture.java
index 1569b14..770c47a 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearOptimisticSerializableTxPrepareFuture.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearOptimisticSerializableTxPrepareFuture.java
@@ -73,8 +73,7 @@ import static org.apache.ignite.transactions.TransactionState.PREPARING;
/**
*
*/
-public class GridNearOptimisticSerializableTxPrepareFuture extends GridNearOptimisticTxPrepareFutureAdapter
- implements GridCacheMvccFuture<IgniteInternalTx> {
+public class GridNearOptimisticSerializableTxPrepareFuture extends GridNearOptimisticTxPrepareFutureAdapter {
/** */
public static final IgniteProductVersion SER_TX_SINCE = IgniteProductVersion.fromString("1.5.0");
@@ -148,18 +147,6 @@ public class GridNearOptimisticSerializableTxPrepareFuture extends GridNearOptim
}
/** {@inheritDoc} */
- @Override public Collection<? extends ClusterNode> nodes() {
- return F.viewReadOnly(futures(), new IgniteClosure<IgniteInternalFuture<?>, ClusterNode>() {
- @Nullable @Override public ClusterNode apply(IgniteInternalFuture<?> f) {
- if (isMini(f))
- return ((MiniFuture)f).node();
-
- return cctx.discovery().localNode();
- }
- });
- }
-
- /** {@inheritDoc} */
@Override public boolean onNodeLeft(UUID nodeId) {
boolean found = false;
@@ -287,7 +274,7 @@ public class GridNearOptimisticSerializableTxPrepareFuture extends GridNearOptim
tx.setRollbackOnly();
// Don't forget to clean up.
- cctx.mvcc().removeFuture(this);
+ cctx.mvcc().removeMvccFuture(this);
return true;
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/1f103067/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearOptimisticTxPrepareFuture.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearOptimisticTxPrepareFuture.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearOptimisticTxPrepareFuture.java
index 82e3868..eaf476c 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearOptimisticTxPrepareFuture.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearOptimisticTxPrepareFuture.java
@@ -35,7 +35,6 @@ import org.apache.ignite.internal.processors.cache.GridCacheContext;
import org.apache.ignite.internal.processors.cache.GridCacheEntryEx;
import org.apache.ignite.internal.processors.cache.GridCacheEntryRemovedException;
import org.apache.ignite.internal.processors.cache.GridCacheMvccCandidate;
-import org.apache.ignite.internal.processors.cache.GridCacheMvccFuture;
import org.apache.ignite.internal.processors.cache.GridCacheSharedContext;
import org.apache.ignite.internal.processors.cache.distributed.GridDistributedTxMapping;
import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtTxMapping;
@@ -54,7 +53,6 @@ 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;
import org.apache.ignite.internal.util.typedef.internal.U;
-import org.apache.ignite.lang.IgniteClosure;
import org.apache.ignite.lang.IgniteUuid;
import org.apache.ignite.transactions.TransactionTimeoutException;
import org.jetbrains.annotations.Nullable;
@@ -66,8 +64,7 @@ import static org.apache.ignite.transactions.TransactionState.PREPARING;
/**
*
*/
-public class GridNearOptimisticTxPrepareFuture extends GridNearOptimisticTxPrepareFutureAdapter
- implements GridCacheMvccFuture<IgniteInternalTx> {
+public class GridNearOptimisticTxPrepareFuture extends GridNearOptimisticTxPrepareFutureAdapter {
/** */
@GridToStringInclude
private Collection<IgniteTxKey> lockKeys = new GridConcurrentHashSet<>();
@@ -100,18 +97,6 @@ public class GridNearOptimisticTxPrepareFuture extends GridNearOptimisticTxPrepa
}
/** {@inheritDoc} */
- @Override public Collection<? extends ClusterNode> nodes() {
- return F.viewReadOnly(futures(), new IgniteClosure<IgniteInternalFuture<?>, ClusterNode>() {
- @Nullable @Override public ClusterNode apply(IgniteInternalFuture<?> f) {
- if (isMini(f))
- return ((MiniFuture)f).node();
-
- return cctx.discovery().localNode();
- }
- });
- }
-
- /** {@inheritDoc} */
@Override public boolean onNodeLeft(UUID nodeId) {
boolean found = false;
@@ -261,7 +246,7 @@ public class GridNearOptimisticTxPrepareFuture extends GridNearOptimisticTxPrepa
if (super.onDone(tx, err0)) {
// Don't forget to clean up.
- cctx.mvcc().removeFuture(this);
+ cctx.mvcc().removeMvccFuture(this);
return true;
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/1f103067/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 103105e..ffe5373 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
@@ -28,6 +28,8 @@ 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.GridCacheContext;
+import org.apache.ignite.internal.processors.cache.GridCacheEntryEx;
+import org.apache.ignite.internal.processors.cache.GridCacheMvccCandidate;
import org.apache.ignite.internal.processors.cache.GridCacheSharedContext;
import org.apache.ignite.internal.processors.cache.distributed.GridDistributedTxMapping;
import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtTxMapping;
@@ -42,7 +44,6 @@ import org.apache.ignite.internal.util.typedef.F;
import org.apache.ignite.internal.util.typedef.internal.S;
import org.apache.ignite.internal.util.typedef.internal.U;
import org.apache.ignite.lang.IgniteBiTuple;
-import org.apache.ignite.lang.IgniteClosure;
import org.apache.ignite.lang.IgniteUuid;
import org.jetbrains.annotations.Nullable;
@@ -68,15 +69,6 @@ public class GridNearPessimisticTxPrepareFuture extends GridNearTxPrepareFutureA
}
/** {@inheritDoc} */
- @Override public Collection<? extends ClusterNode> nodes() {
- return F.viewReadOnly(futures(), new IgniteClosure<IgniteInternalFuture<?>, ClusterNode>() {
- @Nullable @Override public ClusterNode apply(IgniteInternalFuture<?> f) {
- return ((MiniFuture)f).node();
- }
- });
- }
-
- /** {@inheritDoc} */
@Override public boolean onNodeLeft(UUID nodeId) {
boolean found = false;
@@ -280,6 +272,11 @@ public class GridNearPessimisticTxPrepareFuture extends GridNearTxPrepareFutureA
}
/** {@inheritDoc} */
+ @Override public boolean onOwnerChanged(GridCacheEntryEx entry, GridCacheMvccCandidate owner) {
+ return false;
+ }
+
+ /** {@inheritDoc} */
@Override public boolean onDone(@Nullable IgniteInternalTx res, @Nullable Throwable err) {
if (err != null)
this.err.compareAndSet(null, err);
@@ -290,7 +287,7 @@ public class GridNearPessimisticTxPrepareFuture extends GridNearTxPrepareFutureA
tx.state(PREPARED);
if (super.onDone(tx, err)) {
- cctx.mvcc().removeFuture(this);
+ cctx.mvcc().removeMvccFuture(this);
return true;
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/1f103067/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearSingleGetRequest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearSingleGetRequest.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearSingleGetRequest.java
new file mode 100644
index 0000000..a506007
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearSingleGetRequest.java
@@ -0,0 +1,396 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.internal.processors.cache.distributed.near;
+
+import java.nio.ByteBuffer;
+import java.util.UUID;
+import org.apache.ignite.IgniteCheckedException;
+import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
+import org.apache.ignite.internal.processors.cache.GridCacheContext;
+import org.apache.ignite.internal.processors.cache.GridCacheDeployable;
+import org.apache.ignite.internal.processors.cache.GridCacheMessage;
+import org.apache.ignite.internal.processors.cache.GridCacheSharedContext;
+import org.apache.ignite.internal.processors.cache.KeyCacheObject;
+import org.apache.ignite.internal.util.typedef.internal.S;
+import org.apache.ignite.lang.IgniteUuid;
+import org.apache.ignite.plugin.extensions.communication.Message;
+import org.apache.ignite.plugin.extensions.communication.MessageReader;
+import org.apache.ignite.plugin.extensions.communication.MessageWriter;
+import org.jetbrains.annotations.NotNull;
+
+import static org.apache.ignite.internal.processors.cache.GridCacheUtils.SKIP_STORE_FLAG_MASK;
+
+/**
+ *
+ */
+public class GridNearSingleGetRequest extends GridCacheMessage implements GridCacheDeployable {
+ /** */
+ private static final long serialVersionUID = 0L;
+
+ /** */
+ public static final int READ_THROUGH_FLAG_MASK = 0x01;
+
+ /** */
+ public static final int SKIP_VALS_FLAG_MASK = 0x02;
+
+ /** */
+ public static final int ADD_READER_FLAG_MASK = 0x04;
+
+ /** */
+ public static final int NEED_VER_FLAG_MASK = 0x08;
+
+ /** */
+ public static final int NEED_ENTRY_INFO_FLAG_MASK = 0x10;
+
+ /** Future ID. */
+ private IgniteUuid futId;
+
+ /** */
+ private KeyCacheObject key;
+
+ /** Flags. */
+ private byte flags;
+
+ /** Topology version. */
+ private AffinityTopologyVersion topVer;
+
+ /** Subject ID. */
+ private UUID subjId;
+
+ /** Task name hash. */
+ private int taskNameHash;
+
+ /** TTL for read operation. */
+ private long accessTtl;
+
+ /**
+ * Empty constructor required for {@link Message}.
+ */
+ public GridNearSingleGetRequest() {
+ // No-op.
+ }
+
+ /**
+ * @param cacheId Cache ID.
+ * @param futId Future ID.
+ * @param key Key.
+ * @param readThrough Read through flag.
+ * @param skipVals Skip values flag. When false, only boolean values will be returned indicating whether
+ * cache entry has a value.
+ * @param topVer Topology version.
+ * @param subjId Subject ID.
+ * @param taskNameHash Task name hash.
+ * @param accessTtl New TTL to set after entry is accessed, -1 to leave unchanged.
+ * @param addReader Add reader flag.
+ * @param needVer {@code True} if entry version is needed.
+ * @param addDepInfo Deployment info.
+ */
+ public GridNearSingleGetRequest(
+ int cacheId,
+ IgniteUuid futId,
+ KeyCacheObject key,
+ boolean readThrough,
+ @NotNull AffinityTopologyVersion topVer,
+ UUID subjId,
+ int taskNameHash,
+ long accessTtl,
+ boolean skipVals,
+ boolean addReader,
+ boolean needVer,
+ boolean addDepInfo
+ ) {
+ assert futId != null;
+ assert key != null;
+
+ this.cacheId = cacheId;
+ this.futId = futId;
+ this.key = key;
+ this.topVer = topVer;
+ this.subjId = subjId;
+ this.taskNameHash = taskNameHash;
+ this.accessTtl = accessTtl;
+ this.addDepInfo = addDepInfo;
+
+ if (readThrough)
+ flags = (byte)(flags | READ_THROUGH_FLAG_MASK);
+
+ if (skipVals)
+ flags = (byte)(flags | SKIP_VALS_FLAG_MASK);
+
+ if (addReader)
+ flags = (byte)(flags | ADD_READER_FLAG_MASK);
+
+ if (needVer)
+ flags = (byte)(flags | NEED_VER_FLAG_MASK);
+ }
+
+ /**
+ * @return Key.
+ */
+ public KeyCacheObject key() {
+ return key;
+ }
+
+ /**
+ * @return Future ID.
+ */
+ public IgniteUuid futureId() {
+ return futId;
+ }
+
+ /**
+ * @return Subject ID.
+ */
+ public UUID subjectId() {
+ return subjId;
+ }
+
+ /**
+ * Gets task name hash.
+ *
+ * @return Task name hash.
+ */
+ public int taskNameHash() {
+ return taskNameHash;
+ }
+
+ /**
+ * @return Topology version.
+ */
+ @Override public AffinityTopologyVersion topologyVersion() {
+ return topVer;
+ }
+
+ /**
+ * @return New TTL to set after entry is accessed, -1 to leave unchanged.
+ */
+ public long accessTtl() {
+ return accessTtl;
+ }
+
+ /**
+ * @return Read through flag.
+ */
+ public boolean readThrough() {
+ return (flags & SKIP_STORE_FLAG_MASK) != 0;
+ }
+
+ /**
+ * @return Read through flag.
+ */
+ public boolean skipValues() {
+ return (flags & SKIP_VALS_FLAG_MASK) != 0;
+ }
+
+ /**
+ * @return Add reader flag.
+ */
+ public boolean addReader() {
+ return (flags & ADD_READER_FLAG_MASK) != 0;
+ }
+
+ /**
+ * @return {@code True} if entry version is needed.
+ */
+ public boolean needVersion() {
+ return (flags & NEED_VER_FLAG_MASK) != 0;
+ }
+
+ /**
+ * @return {@code True} if full entry information is needed.
+ */
+ public boolean needEntryInfo() {
+ return (flags & NEED_ENTRY_INFO_FLAG_MASK) != 0;
+ }
+
+ /** {@inheritDoc} */
+ @Override public void prepareMarshal(GridCacheSharedContext ctx) throws IgniteCheckedException {
+ super.prepareMarshal(ctx);
+
+ assert key != null;
+
+ GridCacheContext cctx = ctx.cacheContext(cacheId);
+
+ prepareMarshalCacheObject(key, cctx);
+ }
+
+ /** {@inheritDoc} */
+ @Override public void finishUnmarshal(GridCacheSharedContext ctx, ClassLoader ldr) throws IgniteCheckedException {
+ super.finishUnmarshal(ctx, ldr);
+
+ assert key != null;
+
+ GridCacheContext cctx = ctx.cacheContext(cacheId);
+
+ key.finishUnmarshal(cctx.cacheObjectContext(), ldr);
+ }
+
+ /** {@inheritDoc} */
+ @Override public boolean readFrom(ByteBuffer buf, MessageReader reader) {
+ reader.setBuffer(buf);
+
+ if (!reader.beforeMessageRead())
+ return false;
+
+ if (!super.readFrom(buf, reader))
+ return false;
+
+ switch (reader.state()) {
+ case 3:
+ accessTtl = reader.readLong("accessTtl");
+
+ if (!reader.isLastRead())
+ return false;
+
+ reader.incrementState();
+
+ case 4:
+ flags = reader.readByte("flags");
+
+ if (!reader.isLastRead())
+ return false;
+
+ reader.incrementState();
+
+ case 5:
+ futId = reader.readIgniteUuid("futId");
+
+ if (!reader.isLastRead())
+ return false;
+
+ reader.incrementState();
+
+ case 6:
+ key = reader.readMessage("key");
+
+ if (!reader.isLastRead())
+ return false;
+
+ reader.incrementState();
+
+ case 7:
+ subjId = reader.readUuid("subjId");
+
+ if (!reader.isLastRead())
+ return false;
+
+ reader.incrementState();
+
+ case 8:
+ taskNameHash = reader.readInt("taskNameHash");
+
+ if (!reader.isLastRead())
+ return false;
+
+ reader.incrementState();
+
+ case 9:
+ topVer = reader.readMessage("topVer");
+
+ if (!reader.isLastRead())
+ return false;
+
+ reader.incrementState();
+
+ }
+
+ return reader.afterMessageRead(GridNearSingleGetRequest.class);
+ }
+
+ /** {@inheritDoc} */
+ @Override public boolean writeTo(ByteBuffer buf, MessageWriter writer) {
+ writer.setBuffer(buf);
+
+ if (!super.writeTo(buf, writer))
+ return false;
+
+ if (!writer.isHeaderWritten()) {
+ if (!writer.writeHeader(directType(), fieldsCount()))
+ return false;
+
+ writer.onHeaderWritten();
+ }
+
+ switch (writer.state()) {
+ case 3:
+ if (!writer.writeLong("accessTtl", accessTtl))
+ return false;
+
+ writer.incrementState();
+
+ case 4:
+ if (!writer.writeByte("flags", flags))
+ return false;
+
+ writer.incrementState();
+
+ case 5:
+ if (!writer.writeIgniteUuid("futId", futId))
+ return false;
+
+ writer.incrementState();
+
+ case 6:
+ if (!writer.writeMessage("key", key))
+ return false;
+
+ writer.incrementState();
+
+ case 7:
+ if (!writer.writeUuid("subjId", subjId))
+ return false;
+
+ writer.incrementState();
+
+ case 8:
+ if (!writer.writeInt("taskNameHash", taskNameHash))
+ return false;
+
+ writer.incrementState();
+
+ case 9:
+ if (!writer.writeMessage("topVer", topVer))
+ return false;
+
+ writer.incrementState();
+
+ }
+
+ return true;
+ }
+
+ /** {@inheritDoc} */
+ @Override public boolean addDeploymentInfo() {
+ return addDepInfo;
+ }
+
+ /** {@inheritDoc} */
+ @Override public byte directType() {
+ return 116;
+ }
+
+ /** {@inheritDoc} */
+ @Override public byte fieldsCount() {
+ return 10;
+ }
+
+ /** {@inheritDoc} */
+ @Override public String toString() {
+ return S.toString(GridNearSingleGetRequest.class, this);
+ }
+}
[3/3] ignite git commit: Optimization for single key cache 'get'
operation.
Posted by sb...@apache.org.
Optimization for single key cache 'get' operation.
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/1f103067
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/1f103067
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/1f103067
Branch: refs/heads/ignite-1.5
Commit: 1f1030670a6e7f9fbad1d939301c884f29b7885a
Parents: ba1d563
Author: sboikov <sb...@gridgain.com>
Authored: Thu Nov 19 17:29:04 2015 +0300
Committer: sboikov <sb...@gridgain.com>
Committed: Thu Nov 19 17:29:04 2015 +0300
----------------------------------------------------------------------
.../communication/GridIoMessageFactory.java | 12 +
.../processors/cache/GridCacheAdapter.java | 103 ++-
.../processors/cache/GridCacheAtomicFuture.java | 6 +
.../processors/cache/GridCacheFuture.java | 13 -
.../processors/cache/GridCacheIoManager.java | 50 +-
.../processors/cache/GridCacheMessage.java | 20 +-
.../processors/cache/GridCacheMvccFuture.java | 7 +
.../processors/cache/GridCacheMvccManager.java | 108 +--
.../distributed/GridCacheTxRecoveryFuture.java | 13 +-
.../dht/CacheDistributedGetFutureAdapter.java | 27 +-
.../cache/distributed/dht/CacheGetFuture.java | 32 +
.../distributed/dht/GridDhtCacheAdapter.java | 141 ++++
.../distributed/dht/GridDhtLockFuture.java | 16 +-
.../dht/GridDhtTransactionalCacheAdapter.java | 9 +-
.../distributed/dht/GridDhtTxFinishFuture.java | 24 +-
.../cache/distributed/dht/GridDhtTxLocal.java | 4 +-
.../distributed/dht/GridDhtTxPrepareFuture.java | 17 +-
.../dht/GridPartitionedGetFuture.java | 69 +-
.../dht/GridPartitionedSingleGetFuture.java | 697 +++++++++++++++++++
.../dht/atomic/GridDhtAtomicCache.java | 127 +++-
.../dht/atomic/GridDhtAtomicUpdateFuture.java | 84 +--
.../dht/atomic/GridDhtAtomicUpdateRequest.java | 11 +
.../dht/atomic/GridNearAtomicUpdateFuture.java | 5 -
.../dht/colocated/GridDhtColocatedCache.java | 160 ++++-
.../colocated/GridDhtColocatedLockFuture.java | 26 +-
.../distributed/near/CacheVersionedValue.java | 2 +-
.../distributed/near/GridNearCacheAdapter.java | 4 +-
.../distributed/near/GridNearGetFuture.java | 57 +-
.../distributed/near/GridNearGetRequest.java | 1 -
.../distributed/near/GridNearGetResponse.java | 2 -
.../distributed/near/GridNearLockFuture.java | 16 +-
...arOptimisticSerializableTxPrepareFuture.java | 17 +-
.../near/GridNearOptimisticTxPrepareFuture.java | 19 +-
.../GridNearPessimisticTxPrepareFuture.java | 19 +-
.../near/GridNearSingleGetRequest.java | 396 +++++++++++
.../near/GridNearSingleGetResponse.java | 321 +++++++++
.../near/GridNearTransactionalCache.java | 2 +-
.../near/GridNearTxFinishFuture.java | 24 +-
.../cache/distributed/near/GridNearTxLocal.java | 149 ++--
.../near/GridNearTxPrepareFutureAdapter.java | 6 +-
.../processors/cache/local/GridLocalCache.java | 4 +-
.../cache/local/GridLocalLockFuture.java | 5 -
.../cache/transactions/IgniteTxHandler.java | 19 +-
.../transactions/IgniteTxLocalAdapter.java | 2 +-
.../cache/transactions/IgniteTxManager.java | 2 +-
.../IgniteClientReconnectCacheTest.java | 11 +-
.../cache/GridCacheAbstractFullApiSelfTest.java | 75 ++
.../GridCacheConcurrentTxMultiNodeTest.java | 15 -
.../cache/GridCachePartitionedGetSelfTest.java | 3 +-
.../IgniteCacheAbstractStopBusySelfTest.java | 27 +-
.../IgniteCacheP2pUnmarshallingErrorTest.java | 184 +++--
.../CacheGetFutureHangsSelfTest.java | 6 +
.../GridCacheAbstractNodeRestartSelfTest.java | 2 +
.../IgniteCacheSingleGetMessageTest.java | 357 ++++++++++
.../GridCacheReplicatedMetricsSelfTest.java | 9 -
.../testsuites/IgniteCacheTestSuite4.java | 3 +
56 files changed, 2908 insertions(+), 632 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ignite/blob/1f103067/modules/core/src/main/java/org/apache/ignite/internal/managers/communication/GridIoMessageFactory.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/managers/communication/GridIoMessageFactory.java b/modules/core/src/main/java/org/apache/ignite/internal/managers/communication/GridIoMessageFactory.java
index 2503eda..3548aac 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/managers/communication/GridIoMessageFactory.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/managers/communication/GridIoMessageFactory.java
@@ -83,6 +83,8 @@ import org.apache.ignite.internal.processors.cache.distributed.near.GridNearGetR
import org.apache.ignite.internal.processors.cache.distributed.near.GridNearGetResponse;
import org.apache.ignite.internal.processors.cache.distributed.near.GridNearLockRequest;
import org.apache.ignite.internal.processors.cache.distributed.near.GridNearLockResponse;
+import org.apache.ignite.internal.processors.cache.distributed.near.GridNearSingleGetRequest;
+import org.apache.ignite.internal.processors.cache.distributed.near.GridNearSingleGetResponse;
import org.apache.ignite.internal.processors.cache.distributed.near.GridNearTxFinishRequest;
import org.apache.ignite.internal.processors.cache.distributed.near.GridNearTxFinishResponse;
import org.apache.ignite.internal.processors.cache.distributed.near.GridNearTxPrepareRequest;
@@ -696,6 +698,16 @@ public class GridIoMessageFactory implements MessageFactory {
break;
+ case 116:
+ msg = new GridNearSingleGetRequest();
+
+ break;
+
+ case 117:
+ msg = new GridNearSingleGetResponse();
+
+ break;
+
// [-3..114] - this
// [120..123] - DR
// [-4..-22] - SQL
http://git-wip-us.apache.org/repos/asf/ignite/blob/1f103067/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheAdapter.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheAdapter.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheAdapter.java
index cbb7486..562a0eb 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheAdapter.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheAdapter.java
@@ -599,11 +599,11 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
}
/** {@inheritDoc} */
- @Override public IgniteInternalFuture<Boolean> containsKeyAsync(K key) {
+ @Override public final IgniteInternalFuture<Boolean> containsKeyAsync(K key) {
A.notNull(key, "key");
- return getAllAsync(
- Collections.singletonList(key),
+ return (IgniteInternalFuture)getAsync(
+ key,
/*force primary*/false,
/*skip tx*/false,
/*subj id*/null,
@@ -611,15 +611,7 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
/*deserialize portable*/false,
/*skip values*/true,
/*can remap*/true
- ).chain(new CX1<IgniteInternalFuture<Map<K, V>>, Boolean>() {
- @Override public Boolean applyx(IgniteInternalFuture<Map<K, V>> fut) throws IgniteCheckedException {
- Map<K, V> map = fut.get();
-
- assert map.isEmpty() || map.size() == 1 : map.size();
-
- return map.isEmpty() ? false : map.values().iterator().next() != null;
- }
- });
+ );
}
/** {@inheritDoc} */
@@ -1473,6 +1465,52 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
}
/**
+ * @param key Key.
+ * @param forcePrimary Force primary.
+ * @param skipTx Skip tx.
+ * @param subjId Subj Id.
+ * @param taskName Task name.
+ * @param deserializePortable Deserialize portable.
+ * @param skipVals Skip values.
+ * @param canRemap Can remap flag.
+ * @return Future for the get operation.
+ */
+ protected IgniteInternalFuture<V> getAsync(
+ final K key,
+ boolean forcePrimary,
+ boolean skipTx,
+ @Nullable UUID subjId,
+ String taskName,
+ boolean deserializePortable,
+ final boolean skipVals,
+ boolean canRemap
+ ) {
+ return getAllAsync(Collections.singletonList(key),
+ forcePrimary,
+ skipTx,
+ subjId,
+ taskName,
+ deserializePortable,
+ skipVals,
+ canRemap).chain(
+ new CX1<IgniteInternalFuture<Map<K, V>>, V>() {
+ @Override public V applyx(IgniteInternalFuture<Map<K, V>> e) throws IgniteCheckedException {
+ Map<K, V> map = e.get();
+
+ assert map.isEmpty() || map.size() == 1 : map.size();
+
+ if (skipVals) {
+ Boolean val = map.isEmpty() ? false : (Boolean)F.firstValue(map);
+
+ return (V)(val);
+ }
+
+ return map.get(key);
+ }
+ });
+ }
+
+ /**
* @param keys Keys.
* @param forcePrimary Force primary.
* @param skipTx Skip tx.
@@ -1524,7 +1562,7 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
* @return Future for the get operation.
* @see GridCacheAdapter#getAllAsync(Collection)
*/
- public IgniteInternalFuture<Map<K, V>> getAllAsync(@Nullable final Collection<? extends K> keys,
+ public final IgniteInternalFuture<Map<K, V>> getAllAsync(@Nullable final Collection<? extends K> keys,
boolean readThrough,
boolean checkTx,
@Nullable final UUID subjId,
@@ -1605,11 +1643,20 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
final boolean storeEnabled = !skipVals && readThrough && ctx.readThrough();
+ final boolean needEntry = storeEnabled || ctx.isSwapOrOffheapEnabled();
+
Map<KeyCacheObject, GridCacheVersion> misses = null;
for (KeyCacheObject key : keys) {
while (true) {
- GridCacheEntryEx entry = entryEx(key);
+ GridCacheEntryEx entry = needEntry ? entryEx(key) : peekEx(key);
+
+ if (entry == null) {
+ if (!skipVals && ctx.config().isStatisticsEnabled())
+ ctx.cache().metrics0().onRead(false);
+
+ break;
+ }
try {
T2<CacheObject, GridCacheVersion> res = entry.innerGetVersioned(null,
@@ -4389,11 +4436,7 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
*/
@Nullable public V get(K key, boolean deserializePortable)
throws IgniteCheckedException {
- Map<K, V> map = getAllAsync(F.asList(key), deserializePortable).get();
-
- assert map.isEmpty() || map.size() == 1 : map.size();
-
- return map.get(key);
+ return getAsync(key, deserializePortable).get();
}
/**
@@ -4409,16 +4452,16 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
return new GridFinishedFuture<>(e);
}
- return getAllAsync(Collections.singletonList(key), deserializePortable).chain(
- new CX1<IgniteInternalFuture<Map<K, V>>, V>() {
- @Override public V applyx(IgniteInternalFuture<Map<K, V>> e) throws IgniteCheckedException {
- Map<K, V> map = e.get();
-
- assert map.isEmpty() || map.size() == 1 : map.size();
+ String taskName = ctx.kernalContext().job().currentTaskName();
- return map.get(key);
- }
- });
+ return getAsync(key,
+ !ctx.config().isReadFromBackup(),
+ /*skip tx*/false,
+ null,
+ taskName,
+ deserializePortable,
+ false,
+ /*can remap*/true);
}
/**
@@ -4445,10 +4488,10 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
return getAllAsync(keys,
!ctx.config().isReadFromBackup(),
/*skip tx*/false,
- null,
+ /*subject id*/null,
taskName,
deserializePortable,
- false,
+ /*skip vals*/false,
/*can remap*/true);
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/1f103067/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheAtomicFuture.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheAtomicFuture.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheAtomicFuture.java
index be35c5c..359909e 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheAtomicFuture.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheAtomicFuture.java
@@ -20,12 +20,18 @@ package org.apache.ignite.internal.processors.cache;
import java.util.Collection;
import org.apache.ignite.internal.IgniteInternalFuture;
import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
+import org.apache.ignite.internal.processors.cache.version.GridCacheVersion;
/**
* Update future for atomic cache.
*/
public interface GridCacheAtomicFuture<R> extends GridCacheFuture<R> {
/**
+ * @return Future version.
+ */
+ public GridCacheVersion version();
+
+ /**
* Gets future that will be completed when it is safe when update is finished on the given version of topology.
*
* @param topVer Topology version to finish.
http://git-wip-us.apache.org/repos/asf/ignite/blob/1f103067/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheFuture.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheFuture.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheFuture.java
index caa3d3f..8bf8d40 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheFuture.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheFuture.java
@@ -17,11 +17,8 @@
package org.apache.ignite.internal.processors.cache;
-import java.util.Collection;
import java.util.UUID;
-import org.apache.ignite.cluster.ClusterNode;
import org.apache.ignite.internal.IgniteInternalFuture;
-import org.apache.ignite.internal.processors.cache.version.GridCacheVersion;
import org.apache.ignite.lang.IgniteUuid;
/**
@@ -34,16 +31,6 @@ public interface GridCacheFuture<R> extends IgniteInternalFuture<R> {
public IgniteUuid futureId();
/**
- * @return Future version.
- */
- public GridCacheVersion version();
-
- /**
- * @return Involved nodes.
- */
- public Collection<? extends ClusterNode> nodes();
-
- /**
* Callback for when node left.
*
* @param nodeId Left node ID.
http://git-wip-us.apache.org/repos/asf/ignite/blob/1f103067/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 2334780..9afbca8 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
@@ -34,22 +34,24 @@ import org.apache.ignite.internal.cluster.ClusterTopologyCheckedException;
import org.apache.ignite.internal.managers.communication.GridMessageListener;
import org.apache.ignite.internal.managers.deployment.GridDeploymentInfo;
import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
+import org.apache.ignite.internal.processors.cache.distributed.dht.CacheGetFuture;
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.dht.GridDhtTxPrepareRequest;
import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtTxPrepareResponse;
-import org.apache.ignite.internal.processors.cache.distributed.dht.GridPartitionedGetFuture;
+import org.apache.ignite.internal.processors.cache.distributed.dht.GridPartitionedSingleGetFuture;
import org.apache.ignite.internal.processors.cache.distributed.dht.atomic.GridDhtAtomicUpdateRequest;
import org.apache.ignite.internal.processors.cache.distributed.dht.atomic.GridDhtAtomicUpdateResponse;
import org.apache.ignite.internal.processors.cache.distributed.dht.atomic.GridNearAtomicUpdateRequest;
import org.apache.ignite.internal.processors.cache.distributed.dht.atomic.GridNearAtomicUpdateResponse;
import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.GridDhtForceKeysRequest;
import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.GridDhtForceKeysResponse;
-import org.apache.ignite.internal.processors.cache.distributed.near.GridNearGetFuture;
import org.apache.ignite.internal.processors.cache.distributed.near.GridNearGetRequest;
import org.apache.ignite.internal.processors.cache.distributed.near.GridNearGetResponse;
import org.apache.ignite.internal.processors.cache.distributed.near.GridNearLockRequest;
import org.apache.ignite.internal.processors.cache.distributed.near.GridNearLockResponse;
+import org.apache.ignite.internal.processors.cache.distributed.near.GridNearSingleGetRequest;
+import org.apache.ignite.internal.processors.cache.distributed.near.GridNearSingleGetResponse;
import org.apache.ignite.internal.processors.cache.distributed.near.GridNearTxPrepareRequest;
import org.apache.ignite.internal.processors.cache.distributed.near.GridNearTxPrepareResponse;
import org.apache.ignite.internal.processors.cache.query.GridCacheQueryRequest;
@@ -437,7 +439,7 @@ public class GridCacheIoManager extends GridCacheSharedManagerAdapter {
case 50: {
GridNearGetResponse res = (GridNearGetResponse)msg;
- GridCacheFuture fut = ctx.mvcc().future(res.version(), res.futureId());
+ CacheGetFuture fut = (CacheGetFuture)ctx.mvcc().future(res.futureId());
if (fut == null) {
if (log.isDebugEnabled())
@@ -448,10 +450,7 @@ public class GridCacheIoManager extends GridCacheSharedManagerAdapter {
res.error(res.classError());
- if (fut instanceof GridNearGetFuture)
- ((GridNearGetFuture)fut).onResult(nodeId, res);
- else
- ((GridPartitionedGetFuture)fut).onResult(nodeId, res);
+ fut.onResult(nodeId, res);
}
break;
@@ -521,6 +520,43 @@ public class GridCacheIoManager extends GridCacheSharedManagerAdapter {
break;
+ case 116: {
+ GridNearSingleGetRequest req = (GridNearSingleGetRequest)msg;
+
+ GridNearSingleGetResponse res = new GridNearSingleGetResponse(
+ ctx.cacheId(),
+ req.futureId(),
+ req.topologyVersion(),
+ null,
+ false,
+ req.deployInfo() != null);
+
+ res.error(req.classError());
+
+ sendResponseOnFailedMessage(nodeId, res, cctx, ctx.ioPolicy());
+ }
+
+ break;
+
+ case 117: {
+ GridNearSingleGetResponse res = (GridNearSingleGetResponse)msg;
+
+ GridPartitionedSingleGetFuture fut = (GridPartitionedSingleGetFuture)ctx.mvcc().future(res.futureId());
+
+ if (fut == null) {
+ if (log.isDebugEnabled())
+ log.debug("Failed to find future for get response [sender=" + nodeId + ", res=" + res + ']');
+
+ return;
+ }
+
+ res.error(res.classError());
+
+ fut.onResult(nodeId, res);
+ }
+
+ break;
+
default:
throw new IgniteCheckedException("Failed to send response to node. Unsupported direct type [message="
+ msg + "]");
http://git-wip-us.apache.org/repos/asf/ignite/blob/1f103067/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheMessage.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheMessage.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheMessage.java
index bdd2118..61136bf 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheMessage.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheMessage.java
@@ -499,15 +499,21 @@ public abstract class GridCacheMessage implements Message {
int size = col.size();
- for (int i = 0 ; i < size; i++) {
- CacheObject obj = col.get(i);
+ for (int i = 0 ; i < size; i++)
+ prepareMarshalCacheObject(col.get(i), ctx);
+ }
- if (obj != null) {
- obj.prepareMarshal(ctx.cacheObjectContext());
+ /**
+ * @param obj Object.
+ * @param ctx Context.
+ * @throws IgniteCheckedException If failed.
+ */
+ protected final void prepareMarshalCacheObject(CacheObject obj, GridCacheContext ctx) throws IgniteCheckedException {
+ if (obj != null) {
+ obj.prepareMarshal(ctx.cacheObjectContext());
- if (addDepInfo)
- prepareObject(obj.value(ctx.cacheObjectContext(), false), ctx);
- }
+ if (addDepInfo)
+ prepareObject(obj.value(ctx.cacheObjectContext(), false), ctx);
}
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/1f103067/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheMvccFuture.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheMvccFuture.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheMvccFuture.java
index 67c1330..080a6f1 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheMvccFuture.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheMvccFuture.java
@@ -17,11 +17,18 @@
package org.apache.ignite.internal.processors.cache;
+import org.apache.ignite.internal.processors.cache.version.GridCacheVersion;
+
/**
* Distributed future aware of MVCC locking.
*/
public interface GridCacheMvccFuture<T> extends GridCacheFuture<T> {
/**
+ * @return Future version.
+ */
+ public GridCacheVersion version();
+
+ /**
* @param entry Entry which received new owner.
* @param owner Owner.
* @return {@code True} if future cares about this entry.
http://git-wip-us.apache.org/repos/asf/ignite/blob/1f103067/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheMvccManager.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheMvccManager.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheMvccManager.java
index 8562f37..9104acb 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheMvccManager.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheMvccManager.java
@@ -19,7 +19,6 @@ package org.apache.ignite.internal.processors.cache;
import java.util.ArrayList;
import java.util.Collection;
-import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
@@ -31,7 +30,6 @@ import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.ConcurrentMap;
import org.apache.ignite.IgniteCheckedException;
import org.apache.ignite.IgniteLogger;
-import org.apache.ignite.cluster.ClusterNode;
import org.apache.ignite.events.DiscoveryEvent;
import org.apache.ignite.events.Event;
import org.apache.ignite.internal.IgniteClientDisconnectedCheckedException;
@@ -101,12 +99,15 @@ public class GridCacheMvccManager extends GridCacheSharedManagerAdapter {
/** Active futures mapped by version ID. */
@GridToStringExclude
- private final ConcurrentMap<GridCacheVersion, Collection<GridCacheFuture<?>>> futs = newMap();
+ private final ConcurrentMap<GridCacheVersion, Collection<GridCacheMvccFuture<?>>> mvccFuts = newMap();
/** Pending atomic futures. */
private final ConcurrentMap<GridCacheVersion, GridCacheAtomicFuture<?>> atomicFuts =
new ConcurrentHashMap8<>();
+ /** */
+ private final ConcurrentMap<IgniteUuid, GridCacheFuture<?>> futs = new ConcurrentHashMap8<>();
+
/** Near to DHT version mapping. */
private final ConcurrentMap<GridCacheVersion, GridCacheVersion> near2dht = newMap();
@@ -137,12 +138,12 @@ public class GridCacheMvccManager extends GridCacheSharedManagerAdapter {
prev + ']');
if (owner != null && (owner.local() || owner.nearLocal())) {
- Collection<? extends GridCacheFuture> futCol = futs.get(owner.version());
+ Collection<GridCacheMvccFuture<?>> futCol = mvccFuts.get(owner.version());
if (futCol != null) {
synchronized (futCol) {
- for (GridCacheFuture fut : futCol) {
- if (fut instanceof GridCacheMvccFuture && !fut.isDone()) {
+ for (GridCacheMvccFuture<?> fut : futCol) {
+ if (!fut.isDone()) {
GridCacheMvccFuture<Boolean> mvccFut = (GridCacheMvccFuture<Boolean>)fut;
// Since this method is called outside of entry synchronization,
@@ -206,18 +207,14 @@ public class GridCacheMvccManager extends GridCacheSharedManagerAdapter {
for (GridCacheFuture<?> fut : activeFutures())
fut.onNodeLeft(discoEvt.eventNode().id());
- for (IgniteInternalFuture<?> fut : atomicFuts.values()) {
- if (fut instanceof GridCacheFuture) {
- GridCacheFuture cacheFut = (GridCacheFuture)fut;
+ for (GridCacheAtomicFuture<?> cacheFut : atomicFuts.values()) {
+ cacheFut.onNodeLeft(discoEvt.eventNode().id());
- cacheFut.onNodeLeft(discoEvt.eventNode().id());
+ if (cacheFut.isCancelled() || cacheFut.isDone()) {
+ GridCacheVersion futVer = cacheFut.version();
- if (cacheFut.isCancelled() || cacheFut.isDone()) {
- GridCacheVersion futVer = cacheFut.version();
-
- if (futVer != null)
- atomicFuts.remove(futVer, fut);
- }
+ if (futVer != null)
+ atomicFuts.remove(futVer, cacheFut);
}
}
}
@@ -261,12 +258,14 @@ public class GridCacheMvccManager extends GridCacheSharedManagerAdapter {
public Collection<GridCacheFuture<?>> activeFutures() {
ArrayList<GridCacheFuture<?>> col = new ArrayList<>();
- for (Collection<GridCacheFuture<?>> verFuts : futs.values()) {
- synchronized (verFuts) {
- col.addAll(verFuts);
+ for (Collection<GridCacheMvccFuture<?>> futs : mvccFuts.values()) {
+ synchronized (futs) {
+ col.addAll(futs);
}
}
+ col.addAll(futs.values());
+
return col;
}
@@ -420,13 +419,25 @@ public class GridCacheMvccManager extends GridCacheSharedManagerAdapter {
}
/**
+ * @param fut Future.
+ * @param futId Future ID.
+ */
+ public void addFuture(final GridCacheFuture<?> fut, final IgniteUuid futId) {
+ GridCacheFuture<?> old = futs.put(futId, fut);
+
+ assert old == null : old;
+
+ onFutureAdded(fut);
+ }
+
+ /**
* Adds future.
*
* @param fut Future.
* @return {@code True} if added.
*/
@SuppressWarnings({"SynchronizationOnLocalVariableOrMethodParameter"})
- public boolean addFuture(final GridCacheFuture<?> fut) {
+ public boolean addFuture(final GridCacheMvccFuture<?> fut) {
if (fut.isDone()) {
fut.markNotTrackable();
@@ -437,10 +448,10 @@ public class GridCacheMvccManager extends GridCacheSharedManagerAdapter {
return true;
while (true) {
- Collection<GridCacheFuture<?>> old = futs.get(fut.version());
+ Collection<GridCacheMvccFuture<?>> old = mvccFuts.get(fut.version());
if (old == null) {
- Collection<GridCacheFuture<?>> col = new HashSet<GridCacheFuture<?>>(U.capacity(4), 0.75f) {
+ Collection<GridCacheMvccFuture<?>> col = new HashSet<GridCacheMvccFuture<?>>(U.capacity(4), 0.75f) {
{
// Make sure that we add future to queue before
// adding queue to the map of futures.
@@ -456,7 +467,7 @@ public class GridCacheMvccManager extends GridCacheSharedManagerAdapter {
}
};
- old = futs.putIfAbsent(fut.version(), col);
+ old = mvccFuts.putIfAbsent(fut.version(), col);
}
if (old != null) {
@@ -471,7 +482,7 @@ public class GridCacheMvccManager extends GridCacheSharedManagerAdapter {
// Future is being removed, so we force-remove here and try again.
if (empty) {
- if (futs.remove(fut.version(), old)) {
+ if (mvccFuts.remove(fut.version(), old)) {
if (log.isDebugEnabled())
log.debug("Removed future list from futures map for lock version: " + fut.version());
}
@@ -501,16 +512,9 @@ public class GridCacheMvccManager extends GridCacheSharedManagerAdapter {
break;
}
- // Close window in case of node is gone before the future got added to
- // the map of futures.
- for (ClusterNode n : fut.nodes()) {
- if (cctx.discovery().node(n.id()) == null)
- fut.onNodeLeft(n.id());
- }
-
// Just in case if future was completed before it was added.
if (fut.isDone())
- removeFuture(fut);
+ removeMvccFuture(fut);
else
onFutureAdded(fut);
@@ -537,15 +541,22 @@ public class GridCacheMvccManager extends GridCacheSharedManagerAdapter {
}
/**
+ * @param futId Future ID.
+ */
+ public void removeFuture(IgniteUuid futId) {
+ futs.remove(futId);
+ }
+
+ /**
* @param fut Future to remove.
* @return {@code True} if removed.
*/
@SuppressWarnings({"SynchronizationOnLocalVariableOrMethodParameter"})
- public boolean removeFuture(GridCacheFuture<?> fut) {
+ public boolean removeMvccFuture(GridCacheMvccFuture<?> fut) {
if (!fut.trackable())
return true;
- Collection<GridCacheFuture<?>> cur = futs.get(fut.version());
+ Collection<GridCacheMvccFuture<?>> cur = mvccFuts.get(fut.version());
if (cur == null)
return false;
@@ -565,7 +576,7 @@ public class GridCacheMvccManager extends GridCacheSharedManagerAdapter {
else if (log.isDebugEnabled())
log.debug("Attempted to remove a non-registered future (has it been already removed?): " + fut);
- if (empty && futs.remove(fut.version(), cur))
+ if (empty && mvccFuts.remove(fut.version(), cur))
if (log.isDebugEnabled())
log.debug("Removed future list from futures map for lock version: " + fut.version());
@@ -580,12 +591,12 @@ public class GridCacheMvccManager extends GridCacheSharedManagerAdapter {
* @return Future.
*/
@SuppressWarnings({"unchecked"})
- @Nullable public GridCacheFuture future(GridCacheVersion ver, IgniteUuid futId) {
- Collection<? extends GridCacheFuture> futs = this.futs.get(ver);
+ @Nullable public GridCacheMvccFuture<?> mvccFuture(GridCacheVersion ver, IgniteUuid futId) {
+ Collection<GridCacheMvccFuture<?>> futs = this.mvccFuts.get(ver);
if (futs != null) {
synchronized (futs) {
- for (GridCacheFuture<?> fut : futs) {
+ for (GridCacheMvccFuture<?> fut : futs) {
if (fut.futureId().equals(futId)) {
if (log.isDebugEnabled())
log.debug("Found future in futures map: " + fut);
@@ -603,22 +614,11 @@ public class GridCacheMvccManager extends GridCacheSharedManagerAdapter {
}
/**
- * Gets all futures for given lock version, possibly empty collection.
- *
- * @param ver Version.
- * @return All futures for given lock version.
+ * @param futId Future ID.
+ * @return Found future.
*/
- @SuppressWarnings("unchecked")
- public <T> Collection<? extends IgniteInternalFuture<T>> futures(GridCacheVersion ver) {
- Collection c = futs.get(ver);
-
- if (c == null)
- return Collections.<IgniteInternalFuture<T>>emptyList();
- else {
- synchronized (c) {
- return new ArrayList<>((Collection<IgniteInternalFuture<T>>)c);
- }
- }
+ @Nullable public GridCacheFuture future(IgniteUuid futId) {
+ return futs.get(futId);
}
/**
@@ -913,7 +913,7 @@ public class GridCacheMvccManager extends GridCacheSharedManagerAdapter {
X.println(">>> Mvcc manager memory stats [grid=" + cctx.gridName() + ']');
X.println(">>> rmvLocksSize: " + rmvLocks.sizex());
X.println(">>> lockedSize: " + locked.size());
- X.println(">>> futsSize: " + futs.size());
+ X.println(">>> futsSize: " + (mvccFuts.size() + futs.size()));
X.println(">>> near2dhtSize: " + near2dht.size());
X.println(">>> finishFutsSize: " + finishFuts.sizex());
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/1f103067/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridCacheTxRecoveryFuture.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridCacheTxRecoveryFuture.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridCacheTxRecoveryFuture.java
index 01c4867..1648de0 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridCacheTxRecoveryFuture.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridCacheTxRecoveryFuture.java
@@ -29,7 +29,6 @@ import org.apache.ignite.internal.cluster.ClusterTopologyCheckedException;
import org.apache.ignite.internal.processors.cache.GridCacheFuture;
import org.apache.ignite.internal.processors.cache.GridCacheSharedContext;
import org.apache.ignite.internal.processors.cache.transactions.IgniteInternalTx;
-import org.apache.ignite.internal.processors.cache.version.GridCacheVersion;
import org.apache.ignite.internal.util.GridLeanMap;
import org.apache.ignite.internal.util.future.GridCompoundIdentityFuture;
import org.apache.ignite.internal.util.future.GridFutureAdapter;
@@ -389,16 +388,6 @@ public class GridCacheTxRecoveryFuture extends GridCompoundIdentityFuture<Boolea
}
/** {@inheritDoc} */
- @Override public GridCacheVersion version() {
- return tx.xidVersion();
- }
-
- /** {@inheritDoc} */
- @Override public Collection<? extends ClusterNode> nodes() {
- return nodes.values();
- }
-
- /** {@inheritDoc} */
@Override public boolean onNodeLeft(UUID nodeId) {
for (IgniteInternalFuture<?> fut : futures())
if (isMini(fut)) {
@@ -424,7 +413,7 @@ public class GridCacheTxRecoveryFuture extends GridCompoundIdentityFuture<Boolea
/** {@inheritDoc} */
@Override public boolean onDone(@Nullable Boolean res, @Nullable Throwable err) {
if (super.onDone(res, err)) {
- cctx.mvcc().removeFuture(this);
+ cctx.mvcc().removeFuture(futId);
if (err == null) {
assert res != null;
http://git-wip-us.apache.org/repos/asf/ignite/blob/1f103067/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/CacheDistributedGetFutureAdapter.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/CacheDistributedGetFutureAdapter.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/CacheDistributedGetFutureAdapter.java
index 721ba4e..245ffc6 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/CacheDistributedGetFutureAdapter.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/CacheDistributedGetFutureAdapter.java
@@ -18,9 +18,12 @@
package org.apache.ignite.internal.processors.cache.distributed.dht;
import java.util.Collection;
+import java.util.List;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.atomic.AtomicInteger;
+import org.apache.ignite.cluster.ClusterNode;
+import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
import org.apache.ignite.internal.processors.cache.GridCacheContext;
import org.apache.ignite.internal.processors.cache.GridCacheFuture;
import org.apache.ignite.internal.processors.cache.IgniteCacheExpiryPolicy;
@@ -40,7 +43,7 @@ import static org.apache.ignite.IgniteSystemProperties.getInteger;
*
*/
public abstract class CacheDistributedGetFutureAdapter<K, V> extends GridCompoundIdentityFuture<Map<K, V>>
- implements GridCacheFuture<Map<K, V>> {
+ implements GridCacheFuture<Map<K, V>>, CacheGetFuture {
/** Default max remap count value. */
public static final int DFLT_MAX_REMAP_CNT = 3;
@@ -155,4 +158,26 @@ public abstract class CacheDistributedGetFutureAdapter<K, V> extends GridCompoun
map.put(key, new T2<>(skipVals ? true : val, ver));
}
+
+ /**
+ * Affinity node to send get request to.
+ *
+ * @param key Key to get.
+ * @param topVer Topology version.
+ * @return Affinity node to get key from.
+ */
+ protected final ClusterNode affinityNode(KeyCacheObject key, AffinityTopologyVersion topVer) {
+ if (!canRemap) {
+ List<ClusterNode> affNodes = cctx.affinity().nodes(key, topVer);
+
+ for (ClusterNode node : affNodes) {
+ if (cctx.discovery().alive(node))
+ return node;
+ }
+
+ return null;
+ }
+ else
+ return cctx.affinity().primary(key, topVer);
+ }
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/1f103067/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/CacheGetFuture.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/CacheGetFuture.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/CacheGetFuture.java
new file mode 100644
index 0000000..ebe2cff
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/CacheGetFuture.java
@@ -0,0 +1,32 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.internal.processors.cache.distributed.dht;
+
+import java.util.UUID;
+import org.apache.ignite.internal.processors.cache.distributed.near.GridNearGetResponse;
+
+/**
+ *
+ */
+public interface CacheGetFuture {
+ /**
+ * @param nodeId Node ID.
+ * @param res Response.
+ */
+ public void onResult(UUID nodeId, GridNearGetResponse res);
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/1f103067/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtCacheAdapter.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtCacheAdapter.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtCacheAdapter.java
index bdd1140..8537357 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtCacheAdapter.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtCacheAdapter.java
@@ -57,9 +57,12 @@ import org.apache.ignite.internal.processors.cache.distributed.GridDistributedCa
import org.apache.ignite.internal.processors.cache.distributed.GridDistributedCacheEntry;
import org.apache.ignite.internal.processors.cache.distributed.dht.colocated.GridDhtDetachedCacheEntry;
import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.GridDhtPreloader;
+import org.apache.ignite.internal.processors.cache.distributed.near.CacheVersionedValue;
import org.apache.ignite.internal.processors.cache.distributed.near.GridNearCacheAdapter;
import org.apache.ignite.internal.processors.cache.distributed.near.GridNearGetRequest;
import org.apache.ignite.internal.processors.cache.distributed.near.GridNearGetResponse;
+import org.apache.ignite.internal.processors.cache.distributed.near.GridNearSingleGetRequest;
+import org.apache.ignite.internal.processors.cache.distributed.near.GridNearSingleGetResponse;
import org.apache.ignite.internal.processors.cache.version.GridCacheVersion;
import org.apache.ignite.internal.processors.platform.cache.PlatformCacheEntryFilter;
import org.apache.ignite.internal.util.future.GridCompoundFuture;
@@ -76,6 +79,7 @@ import org.apache.ignite.internal.util.typedef.internal.U;
import org.apache.ignite.lang.IgniteBiPredicate;
import org.apache.ignite.lang.IgniteBiTuple;
import org.apache.ignite.lang.IgniteUuid;
+import org.apache.ignite.plugin.extensions.communication.Message;
import org.jetbrains.annotations.NotNull;
import org.jetbrains.annotations.Nullable;
import org.jsr166.ConcurrentHashMap8;
@@ -110,6 +114,46 @@ public abstract class GridDhtCacheAdapter<K, V> extends GridDistributedCacheAdap
}
/**
+ * @param nodeId Sender node ID.
+ * @param res Near get response.
+ */
+ protected final void processNearGetResponse(UUID nodeId, GridNearGetResponse res) {
+ if (log.isDebugEnabled())
+ log.debug("Processing near get response [nodeId=" + nodeId + ", res=" + res + ']');
+
+ CacheGetFuture fut = (CacheGetFuture)ctx.mvcc().future(res.futureId());
+
+ if (fut == null) {
+ if (log.isDebugEnabled())
+ log.debug("Failed to find future for get response [sender=" + nodeId + ", res=" + res + ']');
+
+ return;
+ }
+
+ fut.onResult(nodeId, res);
+ }
+
+ /**
+ * @param nodeId Sender node ID.
+ * @param res Near get response.
+ */
+ protected void processNearSingleGetResponse(UUID nodeId, GridNearSingleGetResponse res) {
+ if (log.isDebugEnabled())
+ log.debug("Processing near get response [nodeId=" + nodeId + ", res=" + res + ']');
+
+ GridPartitionedSingleGetFuture fut = (GridPartitionedSingleGetFuture)ctx.mvcc().future(res.futureId());
+
+ if (fut == null) {
+ if (log.isDebugEnabled())
+ log.debug("Failed to find future for get response [sender=" + nodeId + ", res=" + res + ']');
+
+ return;
+ }
+
+ fut.onResult(nodeId, res);
+ }
+
+ /**
* @param ctx Context.
*/
protected GridDhtCacheAdapter(GridCacheContext<K, V> ctx) {
@@ -669,6 +713,103 @@ public abstract class GridDhtCacheAdapter<K, V> extends GridDistributedCacheAdap
* @param nodeId Node ID.
* @param req Get request.
*/
+ protected void processNearSingleGetRequest(final UUID nodeId, final GridNearSingleGetRequest req) {
+ assert ctx.affinityNode();
+
+ long ttl = req.accessTtl();
+
+ final CacheExpiryPolicy expiryPlc = CacheExpiryPolicy.forAccess(ttl);
+
+ LinkedHashMap<KeyCacheObject, Boolean> map = U.newLinkedHashMap(1);
+
+ map.put(req.key(), req.addReader());
+
+ IgniteInternalFuture<Collection<GridCacheEntryInfo>> fut =
+ getDhtAsync(nodeId,
+ req.messageId(),
+ map,
+ req.readThrough(),
+ req.topologyVersion(),
+ req.subjectId(),
+ req.taskNameHash(),
+ expiryPlc,
+ req.skipValues());
+
+ fut.listen(new CI1<IgniteInternalFuture<Collection<GridCacheEntryInfo>>>() {
+ @Override public void apply(IgniteInternalFuture<Collection<GridCacheEntryInfo>> f) {
+ GridNearSingleGetResponse res;
+
+ GridDhtFuture<Collection<GridCacheEntryInfo>> fut =
+ (GridDhtFuture<Collection<GridCacheEntryInfo>>)f;
+
+ try {
+ Collection<GridCacheEntryInfo> entries = fut.get();
+
+ if (F.isEmpty(fut.invalidPartitions())) {
+ GridCacheEntryInfo info = F.first(entries);
+
+ Message res0 = null;
+
+ if (info != null) {
+ if (req.needEntryInfo()) {
+ info.key(null);
+
+ res0 = info;
+ } else if (req.needVersion())
+ res0 = new CacheVersionedValue(info.value(), info.version());
+ else
+ res0 = info.value();
+ }
+
+ res = new GridNearSingleGetResponse(ctx.cacheId(),
+ req.futureId(),
+ req.topologyVersion(),
+ res0,
+ false,
+ req.addDeploymentInfo());
+
+ if (info != null && req.skipValues())
+ res.setContainsValue();
+ }
+ else {
+ res = new GridNearSingleGetResponse(ctx.cacheId(),
+ req.futureId(),
+ ctx.shared().exchange().readyAffinityVersion(),
+ null,
+ true,
+ req.addDeploymentInfo());
+ }
+ }
+ catch (IgniteCheckedException e) {
+ U.error(log, "Failed processing get request: " + req, e);
+
+ res = new GridNearSingleGetResponse(ctx.cacheId(),
+ req.futureId(),
+ req.topologyVersion(),
+ null,
+ false,
+ req.addDeploymentInfo());
+
+ res.error(e);
+ }
+
+ try {
+ ctx.io().send(nodeId, res, ctx.ioPolicy());
+ }
+ catch (IgniteCheckedException e) {
+ U.error(log, "Failed to send get response to node (is node still alive?) [nodeId=" + nodeId +
+ ",req=" + req + ", res=" + res + ']', e);
+ }
+
+ sendTtlUpdateRequest(expiryPlc);
+ }
+ });
+ }
+
+ /**
+ * @param nodeId Node ID.
+ * @param req Get request.
+ */
protected void processNearGetRequest(final UUID nodeId, final GridNearGetRequest req) {
assert ctx.affinityNode();
assert !req.reload() : req;
http://git-wip-us.apache.org/repos/asf/ignite/blob/1f103067/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 7284fd4..a7978c9 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
@@ -262,20 +262,6 @@ public final class GridDhtLockFuture extends GridCompoundIdentityFuture<Boolean>
log.debug("Added invalid partition to future [invalidPart=" + invalidPart + ", fut=" + this + ']');
}
- /**
- * @return Participating nodes.
- */
- @Override public Collection<? extends ClusterNode> nodes() {
- return F.viewReadOnly(futures(), new IgniteClosure<IgniteInternalFuture<?>, ClusterNode>() {
- @Nullable @Override public ClusterNode apply(IgniteInternalFuture<?> f) {
- if (isMini(f))
- return ((MiniFuture)f).node();
-
- return cctx.discovery().localNode();
- }
- });
- }
-
/** {@inheritDoc} */
@Override public GridCacheVersion version() {
return lockVer;
@@ -756,7 +742,7 @@ public final class GridDhtLockFuture extends GridCompoundIdentityFuture<Boolean>
log.debug("Completing future: " + this);
// Clean up.
- cctx.mvcc().removeFuture(this);
+ cctx.mvcc().removeMvccFuture(this);
if (timeoutObj != null)
cctx.time().removeTimeoutObject(timeoutObj);
http://git-wip-us.apache.org/repos/asf/ignite/blob/1f103067/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 35f63e3..2468cf0 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
@@ -50,6 +50,7 @@ import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.Gri
import org.apache.ignite.internal.processors.cache.distributed.near.GridNearGetRequest;
import org.apache.ignite.internal.processors.cache.distributed.near.GridNearLockRequest;
import org.apache.ignite.internal.processors.cache.distributed.near.GridNearLockResponse;
+import org.apache.ignite.internal.processors.cache.distributed.near.GridNearSingleGetRequest;
import org.apache.ignite.internal.processors.cache.distributed.near.GridNearTransactionalCache;
import org.apache.ignite.internal.processors.cache.distributed.near.GridNearTxRemote;
import org.apache.ignite.internal.processors.cache.distributed.near.GridNearUnlockRequest;
@@ -126,6 +127,12 @@ public abstract class GridDhtTransactionalCacheAdapter<K, V> extends GridDhtCach
}
});
+ ctx.io().addHandler(ctx.cacheId(), GridNearSingleGetRequest.class, new CI2<UUID, GridNearSingleGetRequest>() {
+ @Override public void apply(UUID nodeId, GridNearSingleGetRequest req) {
+ processNearSingleGetRequest(nodeId, req);
+ }
+ });
+
ctx.io().addHandler(ctx.cacheId(), GridNearLockRequest.class, new CI2<UUID, GridNearLockRequest>() {
@Override public void apply(UUID nodeId, GridNearLockRequest req) {
processNearLockRequest(nodeId, req);
@@ -566,7 +573,7 @@ public abstract class GridDhtTransactionalCacheAdapter<K, V> extends GridDhtCach
private void processDhtLockResponse(UUID nodeId, GridDhtLockResponse res) {
assert nodeId != null;
assert res != null;
- GridDhtLockFuture fut = (GridDhtLockFuture)ctx.mvcc().<Boolean>future(res.version(), res.futureId());
+ GridDhtLockFuture fut = (GridDhtLockFuture)ctx.mvcc().<Boolean>mvccFuture(res.version(), res.futureId());
if (fut == null) {
if (log.isDebugEnabled())
http://git-wip-us.apache.org/repos/asf/ignite/blob/1f103067/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 992bd66..bb370a5 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
@@ -66,7 +66,7 @@ public final class GridDhtTxFinishFuture<K, V> extends GridCompoundIdentityFutur
private GridCacheSharedContext<K, V> cctx;
/** Future ID. */
- private IgniteUuid futId;
+ private final IgniteUuid futId;
/** Transaction. */
@GridToStringExclude
@@ -115,26 +115,6 @@ public final class GridDhtTxFinishFuture<K, V> extends GridCompoundIdentityFutur
}
/** {@inheritDoc} */
- @Override public GridCacheVersion version() {
- return tx.xidVersion();
- }
-
- /**
- * @return Involved nodes.
- */
- @Override public Collection<? extends ClusterNode> nodes() {
- return
- F.viewReadOnly(futures(), new IgniteClosure<IgniteInternalFuture<?>, ClusterNode>() {
- @Nullable @Override public ClusterNode apply(IgniteInternalFuture<?> f) {
- if (isMini(f))
- return ((MiniFuture)f).node();
-
- return cctx.discovery().localNode();
- }
- });
- }
-
- /** {@inheritDoc} */
@Override public boolean onNodeLeft(UUID nodeId) {
for (IgniteInternalFuture<?> fut : futures())
if (isMini(fut)) {
@@ -228,7 +208,7 @@ public final class GridDhtTxFinishFuture<K, V> extends GridCompoundIdentityFutur
this.tx.sendFinishReply(commit, error());
// Don't forget to clean up.
- cctx.mvcc().removeFuture(this);
+ cctx.mvcc().removeFuture(futId);
return true;
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/1f103067/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxLocal.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxLocal.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxLocal.java
index 2bed931..f344d48 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxLocal.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxLocal.java
@@ -490,7 +490,7 @@ public class GridDhtTxLocal extends GridDhtTxLocalAdapter implements GridCacheMa
final GridDhtTxFinishFuture fut = new GridDhtTxFinishFuture<>(cctx, this, /*commit*/true);
- cctx.mvcc().addFuture(fut);
+ cctx.mvcc().addFuture(fut, fut.futureId());
GridDhtTxPrepareFuture prep = prepFut.get();
@@ -580,7 +580,7 @@ public class GridDhtTxLocal extends GridDhtTxLocalAdapter implements GridCacheMa
final GridDhtTxFinishFuture fut = new GridDhtTxFinishFuture<>(cctx, this, /*rollback*/false);
- cctx.mvcc().addFuture(fut);
+ cctx.mvcc().addFuture(fut, fut.futureId());
if (prepFut == null) {
try {
http://git-wip-us.apache.org/repos/asf/ignite/blob/1f103067/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 d081c0c..4cb5d05 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
@@ -243,21 +243,6 @@ public final class GridDhtTxPrepareFuture extends GridCompoundFuture<IgniteInter
return tx.xidVersion();
}
- /**
- * @return Involved nodes.
- */
- @Override public Collection<? extends ClusterNode> nodes() {
- return
- F.viewReadOnly(futures(), new IgniteClosure<IgniteInternalFuture<?>, ClusterNode>() {
- @Nullable @Override public ClusterNode apply(IgniteInternalFuture<?> f) {
- if (isMini(f))
- return ((MiniFuture)f).node();
-
- return cctx.discovery().localNode();
- }
- });
- }
-
/** {@inheritDoc} */
@Override public boolean onOwnerChanged(GridCacheEntryEx entry, GridCacheMvccCandidate owner) {
if (log.isDebugEnabled())
@@ -823,7 +808,7 @@ public final class GridDhtTxPrepareFuture extends GridCompoundFuture<IgniteInter
if (super.onDone(res, err.get())) {
// Don't forget to clean up.
- cctx.mvcc().removeFuture(this);
+ cctx.mvcc().removeMvccFuture(this);
return true;
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/1f103067/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridPartitionedGetFuture.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridPartitionedGetFuture.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridPartitionedGetFuture.java
index febe9ba..c3d9836 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridPartitionedGetFuture.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridPartitionedGetFuture.java
@@ -21,7 +21,6 @@ import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.LinkedHashMap;
-import java.util.List;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.atomic.AtomicReference;
@@ -57,10 +56,11 @@ import org.apache.ignite.internal.util.typedef.T2;
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.IgniteClosure;
import org.apache.ignite.lang.IgniteUuid;
import org.jetbrains.annotations.Nullable;
+import static org.apache.ignite.internal.processors.cache.distributed.dht.GridPartitionedSingleGetFuture.*;
+
/**
* Colocated get future.
*/
@@ -71,15 +71,15 @@ public class GridPartitionedGetFuture<K, V> extends CacheDistributedGetFutureAda
/** Logger reference. */
private static final AtomicReference<IgniteLogger> logRef = new AtomicReference<>();
+ /** Dummy version sent to older nodes for backward compatibility, */
+ private static final GridCacheVersion DUMMY_VER = new GridCacheVersion(0, 0, 0, 0);
+
/** Logger. */
private static IgniteLogger log;
/** Topology version. */
private AffinityTopologyVersion topVer;
- /** Version. */
- private GridCacheVersion ver;
-
/**
* @param cctx Context.
* @param keys Keys.
@@ -126,8 +126,6 @@ public class GridPartitionedGetFuture<K, V> extends CacheDistributedGetFutureAda
this.topVer = topVer;
- ver = cctx.versions().next();
-
if (log == null)
log = U.logger(cctx.kernalContext(), logRef, GridPartitionedGetFuture.class);
}
@@ -160,25 +158,6 @@ public class GridPartitionedGetFuture<K, V> extends CacheDistributedGetFutureAda
}
/** {@inheritDoc} */
- @Override public GridCacheVersion version() {
- return ver;
- }
-
- /** {@inheritDoc} */
- @SuppressWarnings("unchecked")
- @Override public Collection<? extends ClusterNode> nodes() {
- return
- F.viewReadOnly(futures(), new IgniteClosure<IgniteInternalFuture<Map<K, V>>, ClusterNode>() {
- @Nullable @Override public ClusterNode apply(IgniteInternalFuture<Map<K, V>> f) {
- if (isMini(f))
- return ((MiniFuture)f).node();
-
- return cctx.discovery().localNode();
- }
- });
- }
-
- /** {@inheritDoc} */
@Override public boolean onNodeLeft(UUID nodeId) {
boolean found = false;
@@ -219,7 +198,7 @@ public class GridPartitionedGetFuture<K, V> extends CacheDistributedGetFutureAda
if (super.onDone(res, err)) {
// Don't forget to clean up.
if (trackable)
- cctx.mvcc().removeFuture(this);
+ cctx.mvcc().removeFuture(futId);
cache().sendTtlUpdateRequest(expiryPlc);
@@ -274,9 +253,11 @@ public class GridPartitionedGetFuture<K, V> extends CacheDistributedGetFutureAda
add(new GridFinishedFuture<>(locVals));
if (hasRmtNodes) {
- trackable = true;
+ if (!trackable) {
+ trackable = true;
- cctx.mvcc().addFuture(this);
+ cctx.mvcc().addFuture(this, futId);
+ }
}
// Create mini futures.
@@ -343,7 +324,7 @@ public class GridPartitionedGetFuture<K, V> extends CacheDistributedGetFutureAda
cctx.cacheId(),
futId,
fut.futureId(),
- ver,
+ n.version().compareTo(SINGLE_GET_MSG_SINCE) >= 0 ? null : DUMMY_VER,
mappedKeys,
readThrough,
topVer,
@@ -390,7 +371,8 @@ public class GridPartitionedGetFuture<K, V> extends CacheDistributedGetFutureAda
boolean remote = false;
// Allow to get cached value from the local node.
- boolean allowLocRead = !forcePrimary || cctx.affinity().primary(cctx.localNode(), key, topVer);
+ boolean allowLocRead = (cctx.affinityNode() && !forcePrimary) ||
+ cctx.affinity().primary(cctx.localNode(), key, topVer);
while (true) {
GridCacheEntryEx entry;
@@ -521,28 +503,6 @@ public class GridPartitionedGetFuture<K, V> extends CacheDistributedGetFutureAda
}
/**
- * Finds affinity node to send get request to.
- *
- * @param key Key to get.
- * @param topVer Topology version.
- * @return Affinity node from which the key will be requested.
- */
- private ClusterNode affinityNode(KeyCacheObject key, AffinityTopologyVersion topVer) {
- if (!canRemap) {
- List<ClusterNode> nodes = cctx.affinity().nodes(key, topVer);
-
- for (ClusterNode node : nodes) {
- if (cctx.discovery().alive(node))
- return node;
- }
-
- return null;
- }
- else
- return cctx.affinity().primary(key, topVer);
- }
-
- /**
* @param infos Entry infos.
* @return Result map.
*/
@@ -557,7 +517,7 @@ public class GridPartitionedGetFuture<K, V> extends CacheDistributedGetFutureAda
if (needVer)
versionedResult(map, info.key(), info.value(), info.version());
- else
+ else {
cctx.addResult(map,
info.key(),
info.value(),
@@ -565,6 +525,7 @@ public class GridPartitionedGetFuture<K, V> extends CacheDistributedGetFutureAda
keepCacheObjects,
deserializePortable,
false);
+ }
}
return map;