You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by sb...@apache.org on 2017/03/02 15:34:28 UTC
[2/2] ignite git commit: ignite-4705
ignite-4705
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/30bfae0e
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/30bfae0e
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/30bfae0e
Branch: refs/heads/ignite-4705-2
Commit: 30bfae0e6d5e8ae5098771d6754e4fa24becbd13
Parents: 19c340c
Author: sboikov <sb...@gridgain.com>
Authored: Thu Mar 2 14:47:21 2017 +0300
Committer: sboikov <sb...@gridgain.com>
Committed: Thu Mar 2 18:34:19 2017 +0300
----------------------------------------------------------------------
.../communication/GridIoMessageFactory.java | 8 +-
.../processors/cache/GridCacheIoManager.java | 4 +-
.../processors/cache/GridCacheMvccManager.java | 2 +-
.../GridDhtAtomicAbstractUpdateFuture.java | 219 +++----
.../GridDhtAtomicAbstractUpdateRequest.java | 54 +-
.../dht/atomic/GridDhtAtomicCache.java | 269 ++-------
.../dht/atomic/GridDhtAtomicNearResponse.java | 115 +---
.../atomic/GridDhtAtomicSingleUpdateFuture.java | 16 +-
.../GridDhtAtomicSingleUpdateRequest.java | 4 -
.../dht/atomic/GridDhtAtomicUpdateFuture.java | 14 +-
.../dht/atomic/GridDhtAtomicUpdateRequest.java | 4 -
.../dht/atomic/GridDhtAtomicUpdateResponse.java | 35 +-
...idNearAtomicAbstractSingleUpdateRequest.java | 475 +--------------
.../GridNearAtomicAbstractUpdateFuture.java | 107 ++--
.../GridNearAtomicAbstractUpdateRequest.java | 285 +++++++--
.../atomic/GridNearAtomicFullUpdateRequest.java | 594 +------------------
.../atomic/GridNearAtomicMappingResponse.java | 244 --------
...GridNearAtomicSingleUpdateFilterRequest.java | 11 +-
.../GridNearAtomicSingleUpdateFuture.java | 67 +--
...GridNearAtomicSingleUpdateInvokeRequest.java | 18 +-
.../GridNearAtomicSingleUpdateRequest.java | 47 +-
.../dht/atomic/GridNearAtomicUpdateFuture.java | 72 +--
.../atomic/GridNearAtomicUpdateResponse.java | 32 -
.../distributed/near/GridNearAtomicCache.java | 10 +-
...niteCacheClientNodeChangingTopologyTest.java | 3 -
25 files changed, 568 insertions(+), 2141 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ignite/blob/30bfae0e/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 908d1d6..fa59291 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
@@ -72,7 +72,6 @@ import org.apache.ignite.internal.processors.cache.distributed.dht.atomic.GridDh
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.GridNearAtomicFullUpdateRequest;
-import org.apache.ignite.internal.processors.cache.distributed.dht.atomic.GridNearAtomicMappingResponse;
import org.apache.ignite.internal.processors.cache.distributed.dht.atomic.GridNearAtomicSingleUpdateFilterRequest;
import org.apache.ignite.internal.processors.cache.distributed.dht.atomic.GridNearAtomicSingleUpdateInvokeRequest;
import org.apache.ignite.internal.processors.cache.distributed.dht.atomic.GridNearAtomicSingleUpdateRequest;
@@ -121,11 +120,11 @@ import org.apache.ignite.internal.processors.datastreamer.DataStreamerEntry;
import org.apache.ignite.internal.processors.datastreamer.DataStreamerRequest;
import org.apache.ignite.internal.processors.datastreamer.DataStreamerResponse;
import org.apache.ignite.internal.processors.hadoop.HadoopJobId;
+import org.apache.ignite.internal.processors.hadoop.shuffle.HadoopDirectShuffleMessage;
import org.apache.ignite.internal.processors.hadoop.shuffle.HadoopShuffleAck;
import org.apache.ignite.internal.processors.hadoop.shuffle.HadoopShuffleFinishRequest;
import org.apache.ignite.internal.processors.hadoop.shuffle.HadoopShuffleFinishResponse;
import org.apache.ignite.internal.processors.hadoop.shuffle.HadoopShuffleMessage;
-import org.apache.ignite.internal.processors.hadoop.shuffle.HadoopDirectShuffleMessage;
import org.apache.ignite.internal.processors.igfs.IgfsAckMessage;
import org.apache.ignite.internal.processors.igfs.IgfsBlockKey;
import org.apache.ignite.internal.processors.igfs.IgfsBlocksMessage;
@@ -176,11 +175,6 @@ public class GridIoMessageFactory implements MessageFactory {
Message msg = null;
switch (type) {
- case -47:
- msg = new GridNearAtomicMappingResponse();
-
- break;
-
case -46:
msg = new UpdateErrors();
http://git-wip-us.apache.org/repos/asf/ignite/blob/30bfae0e/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 c46b01a..d1a6753 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
@@ -494,9 +494,7 @@ public class GridCacheIoManager extends GridCacheSharedManagerAdapter {
* @return Atomic future ID if applicable for message.
*/
@Nullable private GridCacheVersion atomicWriteVersion(GridCacheMessage cacheMsg) {
- if (cacheMsg instanceof GridNearAtomicAbstractUpdateRequest)
- return ((GridNearAtomicAbstractUpdateRequest)cacheMsg).updateVersion();
- else if (cacheMsg instanceof GridDhtAtomicAbstractUpdateRequest)
+ if (cacheMsg instanceof GridDhtAtomicAbstractUpdateRequest)
return ((GridDhtAtomicAbstractUpdateRequest)cacheMsg).writeVersion();
return null;
http://git-wip-us.apache.org/repos/asf/ignite/blob/30bfae0e/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 f3d3fb6..7b87769 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
@@ -136,7 +136,7 @@ public class GridCacheMvccManager extends GridCacheSharedManagerAdapter {
private volatile boolean stopping;
/** */
- private final AtomicLong atomicFutId = new AtomicLong();
+ private final AtomicLong atomicFutId = new AtomicLong(U.currentTimeMillis());
/** Lock callback. */
@GridToStringExclude
http://git-wip-us.apache.org/repos/asf/ignite/blob/30bfae0e/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicAbstractUpdateFuture.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicAbstractUpdateFuture.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicAbstractUpdateFuture.java
index 094d643..96f3611 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicAbstractUpdateFuture.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicAbstractUpdateFuture.java
@@ -19,7 +19,6 @@ package org.apache.ignite.internal.processors.cache.distributed.dht.atomic;
import java.util.ArrayList;
import java.util.Collection;
-import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.UUID;
@@ -52,7 +51,6 @@ import org.jetbrains.annotations.NotNull;
import org.jetbrains.annotations.Nullable;
import static org.apache.ignite.cache.CacheWriteSynchronizationMode.FULL_SYNC;
-import static org.apache.ignite.cache.CacheWriteSynchronizationMode.PRIMARY_SYNC;
/**
* DHT atomic cache backup update future.
@@ -78,7 +76,7 @@ public abstract class GridDhtAtomicAbstractUpdateFuture extends GridFutureAdapte
protected final GridCacheContext cctx;
/** Future version. */
- protected final long futId;
+ protected final Long futId;
/** Update request. */
final GridNearAtomicAbstractUpdateRequest updateReq;
@@ -94,25 +92,31 @@ public abstract class GridDhtAtomicAbstractUpdateFuture extends GridFutureAdapte
private volatile int resCnt;
/** */
- private boolean repliedToNear;
+ private final GridNearAtomicUpdateResponse updateRes;
/** */
- private boolean affMapping = true;
+ private final GridDhtAtomicCache.UpdateReplyClosure completionCb;
/**
* @param cctx Cache context.
* @param writeVer Write version.
* @param updateReq Update request.
+ * @param updateRes Response.
+ * @param completionCb Callback to invoke to send response to near node.
*/
protected GridDhtAtomicAbstractUpdateFuture(
GridCacheContext cctx,
GridCacheVersion writeVer,
- GridNearAtomicAbstractUpdateRequest updateReq
+ GridNearAtomicAbstractUpdateRequest updateReq,
+ GridNearAtomicUpdateResponse updateRes,
+ GridDhtAtomicCache.UpdateReplyClosure completionCb
) {
this.cctx = cctx;
this.updateReq = updateReq;
this.writeVer = writeVer;
+ this.updateRes = updateRes;
+ this.completionCb = completionCb;
futId = cctx.mvcc().atomicFutureId();
@@ -145,7 +149,6 @@ public abstract class GridDhtAtomicAbstractUpdateFuture extends GridFutureAdapte
}
/**
- * @param nearNodeId Near node ID.
* @param entry Entry to map.
* @param val Value to write.
* @param entryProcessor Entry processor.
@@ -159,7 +162,6 @@ public abstract class GridDhtAtomicAbstractUpdateFuture extends GridFutureAdapte
@SuppressWarnings("ForLoopReplaceableByForEach")
final void addWriteEntry(
AffinityAssignment affAssignment,
- UUID nearNodeId,
GridDhtCacheEntry entry,
@Nullable CacheObject val,
EntryProcessor<Object, Object, Object> entryProcessor,
@@ -175,9 +177,7 @@ public abstract class GridDhtAtomicAbstractUpdateFuture extends GridFutureAdapte
List<ClusterNode> dhtNodes = cctx.dht().topology().nodes(entry.partition(), affAssignment, affNodes);
- if (dhtNodes != null)
- affMapping = false;
- else
+ if (dhtNodes == null)
dhtNodes = affNodes;
if (log.isDebugEnabled())
@@ -198,7 +198,6 @@ public abstract class GridDhtAtomicAbstractUpdateFuture extends GridFutureAdapte
if (updateReq == null) {
updateReq = createRequest(
node.id(),
- nearNodeId,
futId,
writeVer,
syncMode,
@@ -236,7 +235,6 @@ public abstract class GridDhtAtomicAbstractUpdateFuture extends GridFutureAdapte
protected abstract void addNearKey(KeyCacheObject key, Collection<UUID> readers);
/**
- * @param nearNodeId Near node ID.
* @param readers Entry readers.
* @param entry Entry.
* @param val Value.
@@ -245,15 +243,12 @@ public abstract class GridDhtAtomicAbstractUpdateFuture extends GridFutureAdapte
* @param expireTime Expire time for near cache update (optional).
*/
final void addNearWriteEntries(
- UUID nearNodeId,
Collection<UUID> readers,
GridDhtCacheEntry entry,
@Nullable CacheObject val,
EntryProcessor<Object, Object, Object> entryProcessor,
long ttl,
long expireTime) {
- affMapping = false;
-
CacheWriteSynchronizationMode syncMode = updateReq.writeSynchronizationMode();
addNearKey(entry.key(), readers);
@@ -272,7 +267,6 @@ public abstract class GridDhtAtomicAbstractUpdateFuture extends GridFutureAdapte
updateReq = createRequest(
node.id(),
- nearNodeId,
futId,
writeVer,
syncMode,
@@ -332,14 +326,11 @@ public abstract class GridDhtAtomicAbstractUpdateFuture extends GridFutureAdapte
GridDhtAtomicAbstractUpdateRequest req = mappings != null ? mappings.get(nodeId) : null;
- boolean needReplyToNear = false;
+// boolean needReplyToNear = false;
if (req != null) {
synchronized (this) {
if (req.onResponse()) {
- if (errors != null || (nodeErr && !repliedToNear))
- needReplyToNear = repliedToNear = true;
-
resCnt0 = resCnt;
resCnt0 += 1;
@@ -349,53 +340,50 @@ public abstract class GridDhtAtomicAbstractUpdateFuture extends GridFutureAdapte
else
return false;
}
-
- if (needReplyToNear) {
- assert !F.isEmpty(mappings);
-
- List<UUID> dhtNodes = new ArrayList<>(mappings.size());
-
- dhtNodes.addAll(mappings.keySet());
-
- GridDhtAtomicNearResponse res = new GridDhtAtomicNearResponse(cctx.cacheId(),
- req.partition(),
- req.nearFutureId(),
- cctx.localNodeId(),
- dhtNodes,
- req.flags());
-
- res.errors(errors);
-
- res.failedNodeId(nodeId);
-
- try {
- cctx.io().send(req.nearNodeId(), res, cctx.ioPolicy());
-
- if (msgLog.isDebugEnabled()) {
- msgLog.debug("DTH update fut, sent response on DHT node fail " +
- "[futId=" + futId +
- ", writeVer=" + writeVer +
- ", node=" + req.nearNodeId() +
- ", failedNode=" + nodeId + ']');
- }
- }
- catch (ClusterTopologyCheckedException ignored) {
- if (msgLog.isDebugEnabled()) {
- msgLog.debug("DTH update fut, failed to notify near node on DHT node fail, near node left " +
- "[futId=" + futId +
- ", writeVer=" + writeVer +
- ", node=" + req.nearNodeId() +
- ", failedNode=" + nodeId + ']');
- }
- }
- catch (IgniteCheckedException ignored) {
- U.error(msgLog, "DTH update fut, failed to notify near node on DHT node fail " +
- "[futId=" + futId +
- ", writeVer=" + writeVer +
- ", node=" + req.nearNodeId() +
- ", failedNode=" + nodeId + ']');
- }
- }
+//
+// if (needReplyToNear) {
+// assert !F.isEmpty(mappings);
+//
+// List<UUID> dhtNodes = new ArrayList<>(mappings.size());
+//
+// dhtNodes.addAll(mappings.keySet());
+//
+// GridDhtAtomicNearResponse res = new GridDhtAtomicNearResponse(cctx.cacheId(),
+// req.partition(),
+// req.nearFutureId(),
+// cctx.localNodeId(),
+// req.flags());
+//
+// res.errors(errors);
+//
+// try {
+// cctx.io().send(req.nearNodeId(), res, cctx.ioPolicy());
+//
+// if (msgLog.isDebugEnabled()) {
+// msgLog.debug("DTH update fut, sent response on DHT node fail " +
+// "[futId=" + futId +
+// ", writeVer=" + writeVer +
+// ", node=" + req.nearNodeId() +
+// ", failedNode=" + nodeId + ']');
+// }
+// }
+// catch (ClusterTopologyCheckedException ignored) {
+// if (msgLog.isDebugEnabled()) {
+// msgLog.debug("DTH update fut, failed to notify near node on DHT node fail, near node left " +
+// "[futId=" + futId +
+// ", writeVer=" + writeVer +
+// ", node=" + req.nearNodeId() +
+// ", failedNode=" + nodeId + ']');
+// }
+// }
+// catch (IgniteCheckedException ignored) {
+// U.error(msgLog, "DTH update fut, failed to notify near node on DHT node fail " +
+// "[futId=" + futId +
+// ", writeVer=" + writeVer +
+// ", node=" + req.nearNodeId() +
+// ", failedNode=" + nodeId + ']');
+// }
+// }
if (resCnt0 == mappings.size())
onDone();
@@ -406,97 +394,49 @@ public abstract class GridDhtAtomicAbstractUpdateFuture extends GridFutureAdapte
return false;
}
- final void init(GridNearAtomicUpdateResponse updateRes, GridCacheReturn ret) {
- repliedToNear = updateReq.writeSynchronizationMode() == PRIMARY_SYNC ||
- ret.hasValue() ||
- updateRes.nearVersion() != null ||
- updateRes.nodeId().equals(cctx.localNodeId());
- }
-
/**
* Sends requests to remote nodes.
*
- * @param updateRes Response.
- * @param completionCb Callback to invoke to send response to near node.
* @param ret Cache operation return value.
*/
- final void map(GridNearAtomicUpdateResponse updateRes,
- GridDhtAtomicCache.UpdateReplyClosure completionCb,
- GridCacheReturn ret) {
- boolean fullSync = updateReq.writeSynchronizationMode() == FULL_SYNC;
-
- boolean needReplyToNear = repliedToNear;
+ final void map(ClusterNode nearNode, GridCacheReturn ret) {
+ if (F.isEmpty(mappings)) {
+ completionCb.apply(updateReq, updateRes);
- List<UUID> dhtNodes = null;
+ onDone();
- boolean affMapping = false;
+ return;
+ }
- if (fullSync) {
- if (!F.isEmpty(mappings)) {
- if (updateReq.size() == 1 && this.affMapping)
- affMapping = true;
- else {
- dhtNodes = new ArrayList<>(mappings.size());
+ boolean fullSync = updateReq.writeSynchronizationMode() == FULL_SYNC;
- dhtNodes.addAll(mappings.keySet());
- }
- }
- else
- dhtNodes = Collections.emptyList();
- }
- else
- dhtNodes = Collections.emptyList();
+ if (updateReq.dhtReplyToNear()) {
+ assert fullSync;
- updateRes.mapping(dhtNodes);
+ boolean needReplyToNear = ret.hasValue() || updateRes.nearVersion() != null;
- if (!F.isEmpty(mappings)) {
- sendDhtRequests(fullSync && !needReplyToNear, dhtNodes, affMapping, ret);
+ sendDhtRequests(true, nearNode, ret);
if (needReplyToNear)
completionCb.apply(updateReq, updateRes);
- else {
- if (fullSync && GridDhtAtomicCache.IGNITE_ATOMIC_SND_MAPPING_TO_NEAR) {
- GridNearAtomicMappingResponse mappingRes = new GridNearAtomicMappingResponse(
- cctx.cacheId(),
- updateReq.partition(),
- updateReq.futureId(),
- dhtNodes,
- affMapping);
-
- try {
- cctx.io().send(updateRes.nodeId(), mappingRes, cctx.ioPolicy());
- }
- catch (IgniteCheckedException e) {
- U.error(msgLog, "Failed to send mapping response [futId=" + futId +
- ", writeVer=" + writeVer +
- ", node=" + updateRes.nodeId() + ']');
- }
- }
- }
}
else {
- completionCb.apply(updateReq, updateRes);
+ sendDhtRequests(false, nearNode, ret);
- onDone();
+ if (!fullSync)
+ completionCb.apply(updateReq, updateRes);
}
}
/**
* @param nearReplyInfo {@code True} if need add information for near node response.
- * @param dhtNodes DHT nodes.
* @param ret Return value.
*/
- private void sendDhtRequests(boolean nearReplyInfo, List<UUID> dhtNodes, boolean affMapping, GridCacheReturn ret) {
+ private void sendDhtRequests(boolean nearReplyInfo, ClusterNode nearNode, GridCacheReturn ret) {
for (GridDhtAtomicAbstractUpdateRequest req : mappings.values()) {
try {
if (nearReplyInfo) {
- if (affMapping) {
- assert dhtNodes == null;
-
- req.affinityMapping(true);
- }
- else
- req.dhtNodes(dhtNodes);
+ req.nearReplyInfo(nearNode.id(), updateReq.futureId());
if (!ret.hasValue())
req.setResult(ret.success());
@@ -549,8 +489,13 @@ public abstract class GridDhtAtomicAbstractUpdateFuture extends GridFutureAdapte
}
/**
+ * @param updateRes Response.
+ * @param err Error.
+ */
+ protected abstract void addFailedKeys(GridNearAtomicUpdateResponse updateRes, Throwable err);
+
+ /**
* @param nodeId Node ID.
- * @param nearNodeId Near node ID.
* @param futId Future ID.
* @param writeVer Update version.
* @param syncMode Write synchronization mode.
@@ -562,7 +507,6 @@ public abstract class GridDhtAtomicAbstractUpdateFuture extends GridFutureAdapte
*/
protected abstract GridDhtAtomicAbstractUpdateRequest createRequest(
UUID nodeId,
- UUID nearNodeId,
long futId,
GridCacheVersion writeVer,
CacheWriteSynchronizationMode syncMode,
@@ -584,6 +528,13 @@ public abstract class GridDhtAtomicAbstractUpdateFuture extends GridFutureAdapte
clsr.apply(suc);
}
+ if (updateReq.writeSynchronizationMode() == FULL_SYNC && !updateReq.dhtReplyToNear()) {
+ if (!suc)
+ addFailedKeys(updateRes, err);
+
+ completionCb.apply(updateReq, updateRes);
+ }
+
return true;
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/30bfae0e/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicAbstractUpdateRequest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicAbstractUpdateRequest.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicAbstractUpdateRequest.java
index 896c163..e04e381 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicAbstractUpdateRequest.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicAbstractUpdateRequest.java
@@ -58,12 +58,6 @@ public abstract class GridDhtAtomicAbstractUpdateRequest extends GridCacheMessag
/** */
static final int DHT_ATOMIC_RESULT_SUCCESS_MASK = 0x10;
- /** */
- static final int DHT_ATOMIC_PRIMARY_DHT_FAIL_RESPONSE = 0x20;
-
- /** */
- static final int DHT_ATOMIC_AFF_MAPPING_FLAG_MASK = 0x40;
-
/** Message index. */
public static final int CACHE_MSG_IDX = nextIndexId();
@@ -102,10 +96,6 @@ public abstract class GridDhtAtomicAbstractUpdateRequest extends GridCacheMessag
/** Additional flags. */
protected byte flags;
- /** */
- @GridDirectCollection(UUID.class)
- private List<UUID> dhtNodes;
-
/**
* Empty constructor required by {@link Externalizable}.
*/
@@ -118,14 +108,10 @@ public abstract class GridDhtAtomicAbstractUpdateRequest extends GridCacheMessag
*
* @param cacheId Cache ID.
* @param nodeId Node ID.
- * @param nearNodeId Near node ID.
- * @param nearFutId Future ID on near node.
*/
protected GridDhtAtomicAbstractUpdateRequest(int cacheId,
UUID nodeId,
long futId,
- UUID nearNodeId,
- long nearFutId,
GridCacheVersion writeVer,
CacheWriteSynchronizationMode syncMode,
@NotNull AffinityTopologyVersion topVer,
@@ -138,8 +124,6 @@ public abstract class GridDhtAtomicAbstractUpdateRequest extends GridCacheMessag
this.cacheId = cacheId;
this.nodeId = nodeId;
this.futId = futId;
- this.nearNodeId = nearNodeId;
- this.nearFutId = nearFutId;
this.writeVer = writeVer;
this.syncMode = syncMode;
this.topVer = topVer;
@@ -153,11 +137,11 @@ public abstract class GridDhtAtomicAbstractUpdateRequest extends GridCacheMessag
setFlag(true, DHT_ATOMIC_KEEP_BINARY_FLAG_MASK);
}
- /**
- * @param affMapping
- */
- public void affinityMapping(boolean affMapping) {
- setFlag(affMapping, DHT_ATOMIC_AFF_MAPPING_FLAG_MASK);
+ void nearReplyInfo(UUID nearNodeId, long nearFutId) {
+ assert nearNodeId != null;
+
+ this.nearNodeId = nearNodeId;
+ this.nearFutId = nearFutId;
}
/**
@@ -176,20 +160,6 @@ public abstract class GridDhtAtomicAbstractUpdateRequest extends GridCacheMessag
return nearNodeId;
}
- /**
- * @param dhtNodes DHT nodes.
- */
- void dhtNodes(List<UUID> dhtNodes) {
- this.dhtNodes = dhtNodes;
- }
-
- /**
- * @return DHT nodes.
- */
- List<UUID> dhtNodes() {
- return dhtNodes;
- }
-
/** {@inheritDoc} */
@Override public int lookupIndex() {
return CACHE_MSG_IDX;
@@ -478,12 +448,6 @@ public abstract class GridDhtAtomicAbstractUpdateRequest extends GridCacheMessag
}
switch (writer.state()) {
- case 3:
- if (!writer.writeCollection("dhtNodes", dhtNodes, MessageCollectionItemType.UUID))
- return false;
-
- writer.incrementState();
-
case 4:
if (!writer.writeByte("flags", flags))
return false;
@@ -554,14 +518,6 @@ public abstract class GridDhtAtomicAbstractUpdateRequest extends GridCacheMessag
return false;
switch (reader.state()) {
- case 3:
- dhtNodes = reader.readCollection("dhtNodes", MessageCollectionItemType.UUID);
-
- if (!reader.isLastRead())
- return false;
-
- reader.incrementState();
-
case 4:
flags = reader.readByte("flags");
http://git-wip-us.apache.org/repos/asf/ignite/blob/30bfae0e/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 a58f1ca..f5d06dd 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
@@ -34,7 +34,6 @@ import javax.cache.processor.EntryProcessorResult;
import org.apache.ignite.IgniteCheckedException;
import org.apache.ignite.IgniteException;
import org.apache.ignite.IgniteLogger;
-import org.apache.ignite.IgniteSystemProperties;
import org.apache.ignite.cluster.ClusterNode;
import org.apache.ignite.internal.IgniteInternalFuture;
import org.apache.ignite.internal.NodeStoppingException;
@@ -93,14 +92,12 @@ import org.apache.ignite.internal.util.typedef.CI2;
import org.apache.ignite.internal.util.typedef.CO;
import org.apache.ignite.internal.util.typedef.CX1;
import org.apache.ignite.internal.util.typedef.F;
-import org.apache.ignite.internal.util.typedef.P1;
import org.apache.ignite.internal.util.typedef.internal.A;
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.IgniteBiTuple;
import org.apache.ignite.lang.IgniteClosure;
-import org.apache.ignite.lang.IgniteInClosure;
import org.apache.ignite.lang.IgniteOutClosure;
import org.apache.ignite.lang.IgniteUuid;
import org.apache.ignite.plugin.security.SecurityPermission;
@@ -138,14 +135,6 @@ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K, V> {
Integer.getInteger(IGNITE_ATOMIC_DEFERRED_ACK_TIMEOUT, 500);
/** */
- static final boolean IGNITE_ATOMIC_SND_MAPPING_TO_NEAR =
- IgniteSystemProperties.getBoolean("IGNITE_ATOMIC_SND_MAPPING_TO_NEAR", false);
-
- /** */
- private static final boolean IGNITE_ATOMIC_DHT_REPLY_AFTER_ACK =
- IgniteSystemProperties.getBoolean("IGNITE_ATOMIC_DHT_REPLY_AFTER_ACK", true);
-
- /** */
private final ThreadLocal<Map<UUID, GridDhtAtomicDeferredUpdateResponse>> defRes =
new ThreadLocal<Map<UUID, GridDhtAtomicDeferredUpdateResponse>>() {
@Override protected Map<UUID, GridDhtAtomicDeferredUpdateResponse> initialValue() {
@@ -244,12 +233,6 @@ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K, V> {
@Override public void start() throws IgniteCheckedException {
super.start();
- // TODO IGNITE-4705.
- log.info("Atomic cache start [name=" + name() +
- ", mode=" + configuration().getWriteSynchronizationMode() +
- ", IGNITE_ATOMIC_SND_MAPPING_TO_NEAR=" + IGNITE_ATOMIC_SND_MAPPING_TO_NEAR +
- ", IGNITE_ATOMIC_DHT_REPLY_AFTER_ACK=" + IGNITE_ATOMIC_DHT_REPLY_AFTER_ACK + ']');
-
CacheMetricsImpl m = new CacheMetricsImpl(ctx);
if (ctx.dht().near() != null)
@@ -395,19 +378,6 @@ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K, V> {
}
});
- ctx.io().addHandler(ctx.cacheId(),
- GridNearAtomicMappingResponse.class,
- new CI2<UUID, GridNearAtomicMappingResponse>() {
- @Override public void apply(UUID uuid, GridNearAtomicMappingResponse msg) {
- processDhtAtomicNearMappingResponse(uuid, msg);
- }
-
- @Override public String toString() {
- return "GridNearAtomicMappingResponse handler " +
- "[msgIdx=" + GridNearAtomicMappingResponse.CACHE_MSG_IDX + ']';
- }
- });
-
if (near == null) {
ctx.io().addHandler(
ctx.cacheId(),
@@ -1813,7 +1783,7 @@ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K, V> {
boolean sndPrevVal = !top.rebalanceFinished(req.topologyVersion());
- dhtFut = createDhtFuture(ver, req);
+ dhtFut = createDhtFuture(ver, req, res, completionCb);
expiry = expiryPolicy(req.expiry());
@@ -1867,11 +1837,8 @@ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K, V> {
res.returnValue(retVal);
- if (dhtFut != null) {
- dhtFut.init(res, res.returnValue());
-
+ if (dhtFut != null)
ctx.mvcc().addAtomicFuture(dhtFut.id(), dhtFut);
- }
}
else {
// Should remap all keys.
@@ -1904,8 +1871,6 @@ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K, V> {
}
}
catch (GridDhtInvalidPartitionException ignore) {
- assert !req.fastMap() || req.clientRequest() : req;
-
if (log.isDebugEnabled())
log.debug("Caught invalid partition exception for cache entry (will remap update request): " + req);
@@ -1936,7 +1901,7 @@ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K, V> {
else {
// If there are backups, map backup update future.
if (dhtFut != null) {
- dhtFut.map(res, completionCb, res.returnValue());
+ dhtFut.map(node, res.returnValue());
// Otherwise, complete the call.
}
else
@@ -2469,7 +2434,7 @@ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K, V> {
dhtFut);
if (dhtFut == null && !F.isEmpty(filteredReaders)) {
- dhtFut = createDhtFuture(ver, req);
+ dhtFut = createDhtFuture(ver, req, null, null); // TODO IGNITE-4705.
readersOnly = true;
}
@@ -2488,7 +2453,6 @@ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K, V> {
if (!readersOnly) {
dhtFut.addWriteEntry(
affAssignment,
- nearNode.id(),
entry,
updRes.newValue(),
entryProcessor,
@@ -2502,7 +2466,6 @@ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K, V> {
if (!F.isEmpty(filteredReaders))
dhtFut.addNearWriteEntries(
- nearNode.id(),
filteredReaders,
entry,
updRes.newValue(),
@@ -2644,17 +2607,8 @@ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K, V> {
GridCacheOperation op;
if (putMap != null) {
- // If fast mapping, filter primary keys for write to store.
- Map<KeyCacheObject, CacheObject> storeMap = req.fastMap() ?
- F.view(putMap, new P1<CacheObject>() {
- @Override public boolean apply(CacheObject key) {
- return ctx.affinity().primaryByKey(ctx.localNode(), key, req.topologyVersion());
- }
- }) :
- putMap;
-
try {
- ctx.store().putAll(null, F.viewReadOnly(storeMap, new C1<CacheObject, IgniteBiTuple<CacheObject, GridCacheVersion>>() {
+ ctx.store().putAll(null, F.viewReadOnly(putMap, new C1<CacheObject, IgniteBiTuple<CacheObject, GridCacheVersion>>() {
@Override public IgniteBiTuple<CacheObject, GridCacheVersion> apply(CacheObject v) {
return F.t(v, ver);
}
@@ -2667,17 +2621,8 @@ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K, V> {
op = UPDATE;
}
else {
- // If fast mapping, filter primary keys for write to store.
- Collection<KeyCacheObject> storeKeys = req.fastMap() ?
- F.view(rmvKeys, new P1<Object>() {
- @Override public boolean apply(Object key) {
- return ctx.affinity().primaryByKey(ctx.localNode(), key, req.topologyVersion());
- }
- }) :
- rmvKeys;
-
try {
- ctx.store().removeAll(null, storeKeys);
+ ctx.store().removeAll(null, rmvKeys);
}
catch (CacheStorePartialUpdateException e) {
storeErr = e;
@@ -2712,10 +2657,6 @@ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K, V> {
assert writeVal != null || op == DELETE : "null write value found.";
- boolean primary = !req.fastMap() || ctx.affinity().primaryByPartition(ctx.localNode(),
- entry.partition(),
- req.topologyVersion());
-
Collection<UUID> readers = null;
Collection<UUID> filteredReaders = null;
@@ -2738,11 +2679,11 @@ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K, V> {
expiry,
/*event*/true,
/*metrics*/true,
- primary,
+ /*primary*/true,
/*verCheck*/false,
topVer,
null,
- replicate ? primary ? DR_PRIMARY : DR_BACKUP : DR_NONE,
+ replicate ? DR_PRIMARY : DR_NONE,
CU.TTL_NOT_CHANGED,
CU.EXPIRE_TIME_CALCULATE,
null,
@@ -2777,7 +2718,7 @@ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K, V> {
batchRes.addDeleted(entry, updRes, entries);
if (dhtFut == null && !F.isEmpty(filteredReaders)) {
- dhtFut = createDhtFuture(ver, req);
+ dhtFut = createDhtFuture(ver, req, null, null); // TODO IGNITE-4705.
batchRes.readersOnly(true);
}
@@ -2789,7 +2730,6 @@ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K, V> {
if (!batchRes.readersOnly()) {
dhtFut.addWriteEntry(
affAssignment,
- nearNode.id(),
entry,
writeVal,
entryProcessor,
@@ -2803,7 +2743,6 @@ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K, V> {
if (!F.isEmpty(filteredReaders))
dhtFut.addNearWriteEntries(
- nearNode.id(),
filteredReaders,
entry,
writeVal,
@@ -2813,30 +2752,26 @@ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K, V> {
}
if (hasNear) {
- if (primary) {
- if (!ctx.affinity().partitionBelongs(nearNode, entry.partition(), topVer)) {
- int idx = firstEntryIdx + i;
-
- if (req.operation() == TRANSFORM) {
- res.addNearValue(idx,
- writeVal,
- updRes.newTtl(),
- CU.EXPIRE_TIME_CALCULATE);
- }
- else
- res.addNearTtl(idx, updRes.newTtl(), CU.EXPIRE_TIME_CALCULATE);
-
- if (writeVal != null || entry.hasValue()) {
- IgniteInternalFuture<Boolean> f = entry.addReader(nearNode.id(), req.messageId(), topVer);
+ if (!ctx.affinity().partitionBelongs(nearNode, entry.partition(), topVer)) {
+ int idx = firstEntryIdx + i;
- assert f == null : f;
- }
+ if (req.operation() == TRANSFORM) {
+ res.addNearValue(idx,
+ writeVal,
+ updRes.newTtl(),
+ CU.EXPIRE_TIME_CALCULATE);
}
- else if (readers.contains(nearNode.id())) // Reader became primary or backup.
- entry.removeReader(nearNode.id(), req.messageId());
else
- res.addSkippedIndex(firstEntryIdx + i);
+ res.addNearTtl(idx, updRes.newTtl(), CU.EXPIRE_TIME_CALCULATE);
+
+ if (writeVal != null || entry.hasValue()) {
+ IgniteInternalFuture<Boolean> f = entry.addReader(nearNode.id(), req.messageId(), topVer);
+
+ assert f == null : f;
+ }
}
+ else if (readers.contains(nearNode.id())) // Reader became primary or backup.
+ entry.removeReader(nearNode.id(), req.messageId());
else
res.addSkippedIndex(firstEntryIdx + i);
}
@@ -3083,12 +3018,14 @@ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K, V> {
*/
@Nullable private GridDhtAtomicAbstractUpdateFuture createDhtFuture(
GridCacheVersion writeVer,
- GridNearAtomicAbstractUpdateRequest updateReq
+ GridNearAtomicAbstractUpdateRequest updateReq,
+ GridNearAtomicUpdateResponse updateRes,
+ GridDhtAtomicCache.UpdateReplyClosure completionCb
) {
if (updateReq.size() == 1)
- return new GridDhtAtomicSingleUpdateFuture(ctx, writeVer, updateReq);
+ return new GridDhtAtomicSingleUpdateFuture(ctx, writeVer, updateReq, updateRes, completionCb);
else
- return new GridDhtAtomicUpdateFuture(ctx, writeVer, updateReq);
+ return new GridDhtAtomicUpdateFuture(ctx, writeVer, updateReq, updateRes, completionCb);
}
/**
@@ -3098,7 +3035,6 @@ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K, V> {
private void processNearAtomicUpdateRequest(UUID nodeId, GridNearAtomicAbstractUpdateRequest req) {
if (msgLog.isDebugEnabled()) {
msgLog.debug("Received near atomic update request [futId=" + req.futureId() +
- ", writeVer=" + req.updateVersion() +
", node=" + nodeId + ']');
}
@@ -3143,13 +3079,22 @@ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K, V> {
GridCacheVersion ver = req.writeVersion();
- GridDhtAtomicNearResponse nearRes = ctx.config().getWriteSynchronizationMode() == FULL_SYNC ?
- new GridDhtAtomicNearResponse(ctx.cacheId(),
+ GridDhtAtomicUpdateResponse dhtRes = null;
+ GridDhtAtomicNearResponse nearRes = null;
+
+ if (req.nearNodeId() != null) {
+ nearRes = new GridDhtAtomicNearResponse(ctx.cacheId(),
req.partition(),
req.nearFutureId(),
nodeId,
- req.dhtNodes(),
- req.flags()) : null;
+ req.flags());
+ }
+ else if (req.writeSynchronizationMode() == FULL_SYNC) {
+ dhtRes = new GridDhtAtomicUpdateResponse(ctx.cacheId(),
+ req.partition(),
+ req.futureId(),
+ ctx.deploymentEnabled());
+ }
boolean replicate = ctx.isDrEnabled();
@@ -3233,43 +3178,45 @@ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K, V> {
// Ignore.
}
catch (IgniteCheckedException e) {
+ IgniteCheckedException err =
+ new IgniteCheckedException("Failed to update key on backup node: " + key, e);
+
if (nearRes != null)
- nearRes.addFailedKey(key, new IgniteCheckedException("Failed to update key on backup node: " + key, e));
+ nearRes.addFailedKey(key, err);
+ else if (dhtRes != null)
+ dhtRes.addFailedKey(key, err);
U.error(log, "Failed to update key on backup node: " + key, e);
}
}
- GridDhtAtomicUpdateResponse dhtRes = null;
-
if (isNearEnabled(cacheCfg)) {
List<KeyCacheObject> nearEvicted =
((GridNearAtomicCache<K, V>)near()).processDhtAtomicUpdateRequest(nodeId, req, nearRes);
if (nearEvicted != null) {
- dhtRes = new GridDhtAtomicUpdateResponse(ctx.cacheId(),
- req.partition(),
- req.futureId(),
- ctx.deploymentEnabled());
+ if (dhtRes == null) {
+ dhtRes = new GridDhtAtomicUpdateResponse(ctx.cacheId(),
+ req.partition(),
+ req.futureId(),
+ ctx.deploymentEnabled());
+ }
dhtRes.nearEvicted(nearEvicted);
}
}
if (nearRes != null) {
- if (IGNITE_ATOMIC_DHT_REPLY_AFTER_ACK)
- sendDhtNearResponse(nodeId, req, nearRes);
- else {
- sendDhtNearResponse(null, req, nearRes);
+ sendDhtNearResponse(req, nearRes);
+ sendDeferredUpdateResponse(req.partition(), nodeId, req.futureId());
+ }
+ else {
+ if (dhtRes != null)
+ sendDhtPrimaryResponse(nodeId, req, dhtRes);
+ else
sendDeferredUpdateResponse(req.partition(), nodeId, req.futureId());
- }
}
- else
- sendDeferredUpdateResponse(req.partition(), nodeId, req.futureId());
-
- if (dhtRes != null)
- sendDhtPrimaryResponse(nodeId, req, dhtRes);
}
/**
@@ -3306,44 +3253,6 @@ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K, V> {
/**
*
*/
- private class DeferredResponseClosure implements IgniteInClosure<IgniteException>, Runnable {
- /** */
- private static final long serialVersionUID = 0L;
-
- /** */
- private final int part;
-
- /** */
- private final UUID primaryId;
-
- /** */
- private final long futId;
-
- /**
- * @param part Partition ID.
- * @param primaryId Primary ID.
- * @param futId Future ID.
- */
- DeferredResponseClosure(int part, UUID primaryId, long futId) {
- this.part = part;
- this.primaryId = primaryId;
- this.futId = futId;
- }
-
- /** {@inheritDoc} */
- @Override public void run() {
- sendDeferredUpdateResponse(part, primaryId, futId);
- }
-
- /** {@inheritDoc} */
- @Override public void apply(IgniteException e) {
- ctx.kernalContext().getStripedExecutorService().execute(part, this);
- }
- }
-
- /**
- *
- */
private class DeferredUpdateTimeout implements GridTimeoutObject, Runnable {
/** */
private final int part;
@@ -3483,33 +3392,12 @@ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K, V> {
}
/**
- * @param primaryId Primary node ID.
* @param req Request.
* @param nearRes Response to send.
*/
- private void sendDhtNearResponse(final UUID primaryId,
- final GridDhtAtomicAbstractUpdateRequest req,
- GridDhtAtomicNearResponse nearRes) {
- DeferredResponseClosure c = primaryId != null ?
- new DeferredResponseClosure(req.partition(), primaryId, req.futureId()) : null;
-
+ private void sendDhtNearResponse(final GridDhtAtomicAbstractUpdateRequest req, GridDhtAtomicNearResponse nearRes) {
try {
- ClusterNode node = ctx.discovery().node(req.nearNodeId());
-
- if (node == null)
- throw new ClusterTopologyCheckedException("Node left: " + req.nearNodeId());
-
- if (c != null) {
- ctx.gridIO().send(node,
- TOPIC_CACHE,
- nearRes,
- ctx.ioPolicy(),
- c);
-
- c = null;
- }
- else
- ctx.gridIO().send(node, TOPIC_CACHE, nearRes, ctx.ioPolicy());
+ ctx.gridIO().send(req.nearNodeId(), TOPIC_CACHE, nearRes, ctx.ioPolicy());
if (msgLog.isDebugEnabled()) {
msgLog.debug("Sent DHT near response [futId=" + req.futureId() +
@@ -3531,35 +3419,6 @@ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K, V> {
", node=" + req.nearNodeId() +
", res=" + nearRes + ']', e);
}
- finally {
- if (c != null)
- c.apply(null);
- }
- }
-
- /**
- * @param nodeId Node ID.
- * @param res Response.
- */
- private void processDhtAtomicNearMappingResponse(UUID nodeId, GridNearAtomicMappingResponse res) {
- GridNearAtomicAbstractUpdateFuture updateFut =
- (GridNearAtomicAbstractUpdateFuture)ctx.mvcc().atomicFuture(res.futureId());
-
- if (updateFut != null) {
- if (msgLog.isDebugEnabled()) {
- msgLog.debug("Received near mapping response [futId=" + res.futureId() +
- ", node=" + nodeId + ']');
- }
-
- updateFut.onMappingReceived(nodeId, res);
- }
- else {
- if (msgLog.isDebugEnabled()) {
- msgLog.debug("Failed to find future for near mapping response [futId=" + res.futureId() +
- ", node=" + nodeId +
- ", res=" + res + ']');
- }
- }
}
/**
http://git-wip-us.apache.org/repos/asf/ignite/blob/30bfae0e/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicNearResponse.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicNearResponse.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicNearResponse.java
index b397f0f..595e41a 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicNearResponse.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicNearResponse.java
@@ -18,25 +18,19 @@
package org.apache.ignite.internal.processors.cache.distributed.dht.atomic;
import java.nio.ByteBuffer;
-import java.util.List;
import java.util.UUID;
import org.apache.ignite.IgniteCheckedException;
-import org.apache.ignite.internal.GridDirectCollection;
import org.apache.ignite.internal.processors.cache.GridCacheMessage;
import org.apache.ignite.internal.processors.cache.GridCacheReturn;
import org.apache.ignite.internal.processors.cache.GridCacheSharedContext;
import org.apache.ignite.internal.processors.cache.KeyCacheObject;
import org.apache.ignite.internal.util.tostring.GridToStringExclude;
-import org.apache.ignite.internal.util.tostring.GridToStringInclude;
import org.apache.ignite.internal.util.typedef.internal.S;
-import org.apache.ignite.plugin.extensions.communication.MessageCollectionItemType;
import org.apache.ignite.plugin.extensions.communication.MessageReader;
import org.apache.ignite.plugin.extensions.communication.MessageWriter;
import org.jetbrains.annotations.Nullable;
-import static org.apache.ignite.internal.processors.cache.distributed.dht.atomic.GridDhtAtomicAbstractUpdateRequest.DHT_ATOMIC_AFF_MAPPING_FLAG_MASK;
import static org.apache.ignite.internal.processors.cache.distributed.dht.atomic.GridDhtAtomicAbstractUpdateRequest.DHT_ATOMIC_HAS_RESULT_MASK;
-import static org.apache.ignite.internal.processors.cache.distributed.dht.atomic.GridDhtAtomicAbstractUpdateRequest.DHT_ATOMIC_PRIMARY_DHT_FAIL_RESPONSE;
import static org.apache.ignite.internal.processors.cache.distributed.dht.atomic.GridDhtAtomicAbstractUpdateRequest.DHT_ATOMIC_RESULT_SUCCESS_MASK;
/**
@@ -59,19 +53,11 @@ public class GridDhtAtomicNearResponse extends GridCacheMessage {
private UUID primaryId;
/** */
- @GridDirectCollection(UUID.class)
- @GridToStringInclude
- private List<UUID> mapping;
-
- /** */
@GridToStringExclude
private byte flags;
/** */
- private UpdateErrors errors;
-
- /** */
- private UUID failedNodeId;
+ private UpdateErrors errs;
/**
*
@@ -85,14 +71,12 @@ public class GridDhtAtomicNearResponse extends GridCacheMessage {
* @param partId Partition.
* @param futId Future ID.
* @param primaryId Primary node ID.
- * @param mapping Update mapping.
* @param flags Flags.
*/
public GridDhtAtomicNearResponse(int cacheId,
int partId,
long futId,
UUID primaryId,
- List<UUID> mapping,
byte flags)
{
assert primaryId != null;
@@ -101,54 +85,21 @@ public class GridDhtAtomicNearResponse extends GridCacheMessage {
this.partId = partId;
this.futId = futId;
this.primaryId = primaryId;
- this.mapping = mapping;
this.flags = flags;
}
/**
- * @return {@code True} if update mapping matches affinity function result.
- */
- boolean affinityMapping() {
- return isFlag(DHT_ATOMIC_AFF_MAPPING_FLAG_MASK);
- }
-
- /**
* @return Errors.
*/
@Nullable UpdateErrors errors() {
- return errors;
+ return errs;
}
/**
* @param errors Errors.
*/
void errors(UpdateErrors errors) {
- this.errors = errors;
- }
-
- /**
- * @return Failed node ID.
- */
- UUID failedNodeId() {
- return failedNodeId;
- }
-
- /**
- * @param failedNodeId Failed node ID (used when primary notifies near node).
- */
- void failedNodeId(UUID failedNodeId) {
- assert failedNodeId != null;
-
- this.failedNodeId = failedNodeId;
-
- setFlag(true, DHT_ATOMIC_PRIMARY_DHT_FAIL_RESPONSE);
- }
-
- /**
- * @return {@code True} if message is sent from primary when DHT node fails.
- */
- boolean primaryDhtFailureResponse() {
- return isFlag(DHT_ATOMIC_PRIMARY_DHT_FAIL_RESPONSE);
+ this.errs = errors;
}
/**
@@ -168,10 +119,10 @@ public class GridDhtAtomicNearResponse extends GridCacheMessage {
* @param e Error.
*/
public void addFailedKey(KeyCacheObject key, Throwable e) {
- if (errors == null)
- errors = new UpdateErrors();
+ if (errs == null)
+ errs = new UpdateErrors();
- errors.addFailedKey(key, e);
+ errs.addFailedKey(key, e);
}
/**
@@ -191,13 +142,6 @@ public class GridDhtAtomicNearResponse extends GridCacheMessage {
}
/**
- * @return Update mapping.
- */
- public List<UUID> mapping() {
- return mapping;
- }
-
- /**
* @param flag Set or clear.
* @param mask Mask.
*/
@@ -246,16 +190,16 @@ public class GridDhtAtomicNearResponse extends GridCacheMessage {
@Override public void prepareMarshal(GridCacheSharedContext ctx) throws IgniteCheckedException {
super.prepareMarshal(ctx);
- if (errors != null)
- errors.prepareMarshal(this, ctx.cacheContext(cacheId));
+ if (errs != null)
+ errs.prepareMarshal(this, ctx.cacheContext(cacheId));
}
/** {@inheritDoc} */
@Override public void finishUnmarshal(GridCacheSharedContext ctx, ClassLoader ldr) throws IgniteCheckedException {
super.finishUnmarshal(ctx, ldr);
- if (errors != null)
- errors.finishUnmarshal(this, ctx.cacheContext(cacheId), ldr);
+ if (errs != null)
+ errs.finishUnmarshal(this, ctx.cacheContext(cacheId), ldr);
}
/** {@inheritDoc} */
@@ -274,13 +218,7 @@ public class GridDhtAtomicNearResponse extends GridCacheMessage {
switch (writer.state()) {
case 3:
- if (!writer.writeMessage("errors", errors))
- return false;
-
- writer.incrementState();
-
- case 4:
- if (!writer.writeUuid("failedNodeId", failedNodeId))
+ if (!writer.writeMessage("errs", errs))
return false;
writer.incrementState();
@@ -297,12 +235,6 @@ public class GridDhtAtomicNearResponse extends GridCacheMessage {
writer.incrementState();
- case 7:
- if (!writer.writeCollection("mapping", mapping, MessageCollectionItemType.UUID))
- return false;
-
- writer.incrementState();
-
case 8:
if (!writer.writeInt("partId", partId))
return false;
@@ -332,15 +264,7 @@ public class GridDhtAtomicNearResponse extends GridCacheMessage {
switch (reader.state()) {
case 3:
- errors = reader.readMessage("errors");
-
- if (!reader.isLastRead())
- return false;
-
- reader.incrementState();
-
- case 4:
- failedNodeId = reader.readUuid("failedNodeId");
+ errs = reader.readMessage("errs");
if (!reader.isLastRead())
return false;
@@ -363,14 +287,6 @@ public class GridDhtAtomicNearResponse extends GridCacheMessage {
reader.incrementState();
- case 7:
- mapping = reader.readCollection("mapping", MessageCollectionItemType.UUID);
-
- if (!reader.isLastRead())
- return false;
-
- reader.incrementState();
-
case 8:
partId = reader.readInt("partId");
@@ -394,10 +310,9 @@ public class GridDhtAtomicNearResponse extends GridCacheMessage {
/** {@inheritDoc} */
@Override public String toString() {
- return S.toString(GridDhtAtomicNearResponse.class, this, "flags",
+ return S.toString(GridDhtAtomicNearResponse.class, this,
+ "flags",
"res=" + isFlag(DHT_ATOMIC_HAS_RESULT_MASK) +
- "|resOk=" + isFlag(DHT_ATOMIC_RESULT_SUCCESS_MASK) +
- "|affMap=" + isFlag(DHT_ATOMIC_AFF_MAPPING_FLAG_MASK) +
- "|dhtFail=" + isFlag(DHT_ATOMIC_PRIMARY_DHT_FAIL_RESPONSE));
+ "|resOk=" + isFlag(DHT_ATOMIC_RESULT_SUCCESS_MASK));
}
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/30bfae0e/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicSingleUpdateFuture.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicSingleUpdateFuture.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicSingleUpdateFuture.java
index 2cc370f..e393322 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicSingleUpdateFuture.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicSingleUpdateFuture.java
@@ -47,9 +47,11 @@ class GridDhtAtomicSingleUpdateFuture extends GridDhtAtomicAbstractUpdateFuture
GridDhtAtomicSingleUpdateFuture(
GridCacheContext cctx,
GridCacheVersion writeVer,
- GridNearAtomicAbstractUpdateRequest updateReq
+ GridNearAtomicAbstractUpdateRequest updateReq,
+ GridNearAtomicUpdateResponse updateRes,
+ GridDhtAtomicCache.UpdateReplyClosure completionCb
) {
- super(cctx, writeVer, updateReq);
+ super(cctx, writeVer, updateReq, updateRes, completionCb);
}
/** {@inheritDoc} */
@@ -67,7 +69,6 @@ class GridDhtAtomicSingleUpdateFuture extends GridDhtAtomicAbstractUpdateFuture
/** {@inheritDoc} */
@Override protected GridDhtAtomicAbstractUpdateRequest createRequest(
UUID nodeId,
- UUID nearNodeId,
long futId,
GridCacheVersion writeVer,
CacheWriteSynchronizationMode syncMode,
@@ -81,8 +82,6 @@ class GridDhtAtomicSingleUpdateFuture extends GridDhtAtomicAbstractUpdateFuture
cctx.cacheId(),
nodeId,
futId,
- nearNodeId,
- updateReq.futureId(),
writeVer,
syncMode,
topVer,
@@ -97,8 +96,6 @@ class GridDhtAtomicSingleUpdateFuture extends GridDhtAtomicAbstractUpdateFuture
cctx.cacheId(),
nodeId,
futId,
- nearNodeId,
- updateReq.futureId(),
writeVer,
syncMode,
topVer,
@@ -112,6 +109,11 @@ class GridDhtAtomicSingleUpdateFuture extends GridDhtAtomicAbstractUpdateFuture
}
}
+ /** {@inheritDoc} */
+ @Override protected void addFailedKeys(GridNearAtomicUpdateResponse updateRes, Throwable err) {
+
+ }
+
/**
* @param ttl TTL.
* @param conflictExpireTime Conflict expire time.
http://git-wip-us.apache.org/repos/asf/ignite/blob/30bfae0e/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicSingleUpdateRequest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicSingleUpdateRequest.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicSingleUpdateRequest.java
index fa7c445..3d3ce04 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicSingleUpdateRequest.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicSingleUpdateRequest.java
@@ -86,8 +86,6 @@ public class GridDhtAtomicSingleUpdateRequest extends GridDhtAtomicAbstractUpdat
int cacheId,
UUID nodeId,
long futId,
- UUID nearNodeId,
- long nearFutId,
GridCacheVersion writeVer,
CacheWriteSynchronizationMode syncMode,
@NotNull AffinityTopologyVersion topVer,
@@ -100,8 +98,6 @@ public class GridDhtAtomicSingleUpdateRequest extends GridDhtAtomicAbstractUpdat
super(cacheId,
nodeId,
futId,
- nearNodeId,
- nearFutId,
writeVer,
syncMode,
topVer,
http://git-wip-us.apache.org/repos/asf/ignite/blob/30bfae0e/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 1c12193..ed57cf0 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
@@ -46,9 +46,11 @@ class GridDhtAtomicUpdateFuture extends GridDhtAtomicAbstractUpdateFuture {
GridDhtAtomicUpdateFuture(
GridCacheContext cctx,
GridCacheVersion writeVer,
- GridNearAtomicAbstractUpdateRequest updateReq
+ GridNearAtomicAbstractUpdateRequest updateReq,
+ GridNearAtomicUpdateResponse updateRes,
+ GridDhtAtomicCache.UpdateReplyClosure completionCb
) {
- super(cctx, writeVer, updateReq);
+ super(cctx, writeVer, updateReq, updateRes, completionCb);
mappings = U.newHashMap(updateReq.size());
}
@@ -66,7 +68,6 @@ class GridDhtAtomicUpdateFuture extends GridDhtAtomicAbstractUpdateFuture {
/** {@inheritDoc} */
@Override protected GridDhtAtomicAbstractUpdateRequest createRequest(
UUID nodeId,
- UUID nearNodeId,
long futId,
GridCacheVersion writeVer,
CacheWriteSynchronizationMode syncMode,
@@ -79,8 +80,6 @@ class GridDhtAtomicUpdateFuture extends GridDhtAtomicAbstractUpdateFuture {
cctx.cacheId(),
nodeId,
futId,
- nearNodeId,
- updateReq.futureId(),
writeVer,
syncMode,
topVer,
@@ -94,6 +93,11 @@ class GridDhtAtomicUpdateFuture extends GridDhtAtomicAbstractUpdateFuture {
}
/** {@inheritDoc} */
+ @Override protected void addFailedKeys(GridNearAtomicUpdateResponse updateRes, Throwable err) {
+
+ }
+
+ /** {@inheritDoc} */
@Override public String toString() {
return S.toString(GridDhtAtomicUpdateFuture.class, this, "super", super.toString());
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/30bfae0e/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 ef42af8..029ea42 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
@@ -147,8 +147,6 @@ public class GridDhtAtomicUpdateRequest extends GridDhtAtomicAbstractUpdateReque
int cacheId,
UUID nodeId,
long futId,
- UUID nearNodeId,
- long nearFutId,
GridCacheVersion writeVer,
CacheWriteSynchronizationMode syncMode,
@NotNull AffinityTopologyVersion topVer,
@@ -163,8 +161,6 @@ public class GridDhtAtomicUpdateRequest extends GridDhtAtomicAbstractUpdateReque
super(cacheId,
nodeId,
futId,
- nearNodeId,
- nearFutId,
writeVer,
syncMode,
topVer,
http://git-wip-us.apache.org/repos/asf/ignite/blob/30bfae0e/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicUpdateResponse.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicUpdateResponse.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicUpdateResponse.java
index b1a46d5..10806b1 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicUpdateResponse.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicUpdateResponse.java
@@ -53,7 +53,7 @@ public class GridDhtAtomicUpdateResponse extends GridCacheMessage implements Gri
private long futId;
/** */
- private UpdateErrors errors;
+ private UpdateErrors errs;
/** Evicted readers. */
@GridToStringInclude
@@ -84,10 +84,21 @@ public class GridDhtAtomicUpdateResponse extends GridCacheMessage implements Gri
}
/**
+ * @param key Key.
+ * @param e Error.
+ */
+ public void addFailedKey(KeyCacheObject key, Throwable e) {
+ if (errs == null)
+ errs = new UpdateErrors();
+
+ errs.addFailedKey(key, e);
+ }
+
+ /**
* @return Errors.
*/
@Nullable UpdateErrors errors() {
- return errors;
+ return errs;
}
/** {@inheritDoc} */
@@ -108,15 +119,15 @@ public class GridDhtAtomicUpdateResponse extends GridCacheMessage implements Gri
* @param err Error.
*/
public void onError(IgniteCheckedException err){
- if (errors == null)
- errors = new UpdateErrors();
+ if (errs == null)
+ errs = new UpdateErrors();
- errors.onError(err);
+ errs.onError(err);
}
/** {@inheritDoc} */
@Override public IgniteCheckedException error() {
- return errors != null ? errors.error() : null;
+ return errs != null ? errs.error() : null;
}
/**
@@ -158,8 +169,8 @@ public class GridDhtAtomicUpdateResponse extends GridCacheMessage implements Gri
prepareMarshalCacheObjects(nearEvicted, cctx);
- if (errors != null)
- errors.prepareMarshal(this, cctx);
+ if (errs != null)
+ errs.prepareMarshal(this, cctx);
}
/** {@inheritDoc} */
@@ -170,8 +181,8 @@ public class GridDhtAtomicUpdateResponse extends GridCacheMessage implements Gri
finishUnmarshalCacheObjects(nearEvicted, cctx, ldr);
- if (errors != null)
- errors.finishUnmarshal(this, cctx, ldr);
+ if (errs != null)
+ errs.finishUnmarshal(this, cctx, ldr);
}
/** {@inheritDoc} */
@@ -200,7 +211,7 @@ public class GridDhtAtomicUpdateResponse extends GridCacheMessage implements Gri
switch (writer.state()) {
case 3:
- if (!writer.writeMessage("errors", errors))
+ if (!writer.writeMessage("errors", errs))
return false;
writer.incrementState();
@@ -240,7 +251,7 @@ public class GridDhtAtomicUpdateResponse extends GridCacheMessage implements Gri
switch (reader.state()) {
case 3:
- errors = reader.readMessage("errors");
+ errs = reader.readMessage("errors");
if (!reader.isLastRead())
return false;
http://git-wip-us.apache.org/repos/asf/ignite/blob/30bfae0e/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridNearAtomicAbstractSingleUpdateRequest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridNearAtomicAbstractSingleUpdateRequest.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridNearAtomicAbstractSingleUpdateRequest.java
index 2a17813..3a9055e 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridNearAtomicAbstractSingleUpdateRequest.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridNearAtomicAbstractSingleUpdateRequest.java
@@ -18,19 +18,13 @@
package org.apache.ignite.internal.processors.cache.distributed.dht.atomic;
import java.io.Externalizable;
-import java.nio.ByteBuffer;
import java.util.UUID;
import javax.cache.expiry.ExpiryPolicy;
-import org.apache.ignite.IgniteLogger;
import org.apache.ignite.cache.CacheWriteSynchronizationMode;
import org.apache.ignite.internal.GridDirectTransient;
import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
import org.apache.ignite.internal.processors.cache.CacheEntryPredicate;
import org.apache.ignite.internal.processors.cache.GridCacheOperation;
-import org.apache.ignite.internal.processors.cache.GridCacheSharedContext;
-import org.apache.ignite.internal.processors.cache.version.GridCacheVersion;
-import org.apache.ignite.plugin.extensions.communication.MessageReader;
-import org.apache.ignite.plugin.extensions.communication.MessageWriter;
import org.jetbrains.annotations.NotNull;
import org.jetbrains.annotations.Nullable;
@@ -44,59 +38,6 @@ public abstract class GridNearAtomicAbstractSingleUpdateRequest extends GridNear
/** */
private static final CacheEntryPredicate[] NO_FILTER = new CacheEntryPredicate[0];
- /** Fast map flag mask. */
- private static final int FAST_MAP_FLAG_MASK = 0x1;
-
- /** Flag indicating whether request contains primary keys. */
- private static final int HAS_PRIMARY_FLAG_MASK = 0x2;
-
- /** Topology locked flag. Set if atomic update is performed inside TX or explicit lock. */
- private static final int TOP_LOCKED_FLAG_MASK = 0x4;
-
- /** Skip write-through to a persistent storage. */
- private static final int SKIP_STORE_FLAG_MASK = 0x8;
-
- /** */
- private static final int CLIENT_REQ_FLAG_MASK = 0x10;
-
- /** Keep binary flag. */
- private static final int KEEP_BINARY_FLAG_MASK = 0x20;
-
- /** Return value flag. */
- private static final int RET_VAL_FLAG_MASK = 0x40;
-
- /** Target node ID. */
- @GridDirectTransient
- protected UUID nodeId;
-
- /** Future version. */
- protected long futId;
-
- /** Update version. Set to non-null if fastMap is {@code true}. */
- private GridCacheVersion updateVer;
-
- /** Topology version. */
- protected AffinityTopologyVersion topVer;
-
- /** Write synchronization mode. */
- protected CacheWriteSynchronizationMode syncMode;
-
- /** Update operation. */
- protected GridCacheOperation op;
-
- /** Subject ID. */
- protected UUID subjId;
-
- /** Task name hash. */
- protected int taskNameHash;
-
- /** */
- @GridDirectTransient
- private GridNearAtomicUpdateResponse res;
-
- /** Compressed boolean flags. */
- protected byte flags;
-
/**
* Empty constructor required by {@link Externalizable}.
*/
@@ -110,8 +51,6 @@ public abstract class GridNearAtomicAbstractSingleUpdateRequest extends GridNear
* @param cacheId Cache ID.
* @param nodeId Node ID.
* @param futId Future ID.
- * @param fastMap Fast map scheme flag.
- * @param updateVer Update version set if fast map is performed.
* @param topVer Topology version.
* @param topLocked Topology locked flag.
* @param syncMode Synchronization mode.
@@ -121,15 +60,12 @@ public abstract class GridNearAtomicAbstractSingleUpdateRequest extends GridNear
* @param taskNameHash Task name hash code.
* @param skipStore Skip write-through to a persistent storage.
* @param keepBinary Keep binary flag.
- * @param clientReq Client node request flag.
* @param addDepInfo Deployment info flag.
*/
protected GridNearAtomicAbstractSingleUpdateRequest(
int cacheId,
UUID nodeId,
long futId,
- boolean fastMap,
- @Nullable GridCacheVersion updateVer,
@NotNull AffinityTopologyVersion topVer,
boolean topLocked,
CacheWriteSynchronizationMode syncMode,
@@ -137,89 +73,25 @@ public abstract class GridNearAtomicAbstractSingleUpdateRequest extends GridNear
boolean retval,
@Nullable UUID subjId,
int taskNameHash,
+ boolean stableTop,
boolean skipStore,
boolean keepBinary,
- boolean clientReq,
boolean addDepInfo
) {
- this.cacheId = cacheId;
- this.nodeId = nodeId;
- this.futId = futId;
- this.updateVer = updateVer;
- this.topVer = topVer;
- this.syncMode = syncMode;
- this.op = op;
- this.subjId = subjId;
- this.taskNameHash = taskNameHash;
- this.addDepInfo = addDepInfo;
-
- fastMap(fastMap);
- topologyLocked(topLocked);
- returnValue(retval);
- skipStore(skipStore);
- keepBinary(keepBinary);
- clientRequest(clientReq);
- }
-
- /** {@inheritDoc} */
- @Override public int lookupIndex() {
- return CACHE_MSG_IDX;
- }
-
- /**
- * @return Mapped node ID.
- */
- @Override public UUID nodeId() {
- return nodeId;
- }
-
- /**
- * @param nodeId Node ID.
- */
- @Override public void nodeId(UUID nodeId) {
- this.nodeId = nodeId;
- }
-
- /**
- * @return Subject ID.
- */
- @Override public UUID subjectId() {
- return subjId;
- }
-
- /**
- * @return Task name hash.
- */
- @Override public int taskNameHash() {
- return taskNameHash;
- }
-
- /**
- * @return Future version.
- */
- @Override public long futureId() {
- return futId;
- }
-
- /**
- * @return Update version for fast-map request.
- */
- @Override public GridCacheVersion updateVersion() {
- return updateVer;
- }
-
- /**
- * @return Topology version.
- */
- @Override public AffinityTopologyVersion topologyVersion() {
- return topVer;
- }
-
- /**
- * @return Cache write synchronization mode.
- */
- @Override public CacheWriteSynchronizationMode writeSynchronizationMode() {
- return syncMode;
+ super(cacheId,
+ nodeId,
+ futId,
+ topVer,
+ topLocked,
+ syncMode,
+ op,
+ retval,
+ subjId,
+ taskNameHash,
+ stableTop,
+ skipStore,
+ keepBinary,
+ addDepInfo);
}
/**
@@ -230,331 +102,14 @@ public abstract class GridNearAtomicAbstractSingleUpdateRequest extends GridNear
}
/**
- * @return Update operation.
- */
- @Override public GridCacheOperation operation() {
- return op;
- }
-
- /**
* @return Optional arguments for entry processor.
*/
@Override @Nullable public Object[] invokeArguments() {
return null;
}
- /**
- * @param res Response.
- * @return {@code True} if current response was {@code null}.
- */
- @Override public boolean onResponse(GridNearAtomicUpdateResponse res) {
- if (this.res == null) {
- this.res = res;
-
- return true;
- }
-
- return false;
- }
-
- /**
- * @return Response.
- */
- @Override @Nullable public GridNearAtomicUpdateResponse response() {
- return res;
- }
-
- /** {@inheritDoc} */
- @Override public boolean addDeploymentInfo() {
- return addDepInfo;
- }
-
- /** {@inheritDoc} */
- @Override public IgniteLogger messageLogger(GridCacheSharedContext ctx) {
- return ctx.atomicMessageLogger();
- }
-
- /**
- * @return Flag indicating whether this is fast-map udpate.
- */
- @Override public boolean fastMap() {
- return isFlag(FAST_MAP_FLAG_MASK);
- }
-
- /**
- * Sets fastMap flag value.
- */
- public void fastMap(boolean val) {
- setFlag(val, FAST_MAP_FLAG_MASK);
- }
-
- /**
- * @return Topology locked flag.
- */
- @Override public boolean topologyLocked() {
- return isFlag(TOP_LOCKED_FLAG_MASK);
- }
-
- /**
- * Sets topologyLocked flag value.
- */
- public void topologyLocked(boolean val) {
- setFlag(val, TOP_LOCKED_FLAG_MASK);
- }
-
- /**
- * @return {@code True} if request sent from client node.
- */
- @Override public boolean clientRequest() {
- return isFlag(CLIENT_REQ_FLAG_MASK);
- }
-
- /**
- * Sets clientRequest flag value.
- */
- public void clientRequest(boolean val) {
- setFlag(val, CLIENT_REQ_FLAG_MASK);
- }
-
- /**
- * @return Return value flag.
- */
- @Override public boolean returnValue() {
- return isFlag(RET_VAL_FLAG_MASK);
- }
-
- /**
- * Sets returnValue flag value.
- */
- public void returnValue(boolean val) {
- setFlag(val, RET_VAL_FLAG_MASK);
- }
-
- /**
- * @return Skip write-through to a persistent storage.
- */
- @Override public boolean skipStore() {
- return isFlag(SKIP_STORE_FLAG_MASK);
- }
-
- /**
- * Sets skipStore flag value.
- */
- public void skipStore(boolean val) {
- setFlag(val, SKIP_STORE_FLAG_MASK);
- }
-
- /**
- * @return Keep binary flag.
- */
- @Override public boolean keepBinary() {
- return isFlag(KEEP_BINARY_FLAG_MASK);
- }
-
- /**
- * Sets keepBinary flag value.
- */
- public void keepBinary(boolean val) {
- setFlag(val, KEEP_BINARY_FLAG_MASK);
- }
-
- /**
- * @return Flag indicating whether this request contains primary keys.
- */
- @Override public boolean hasPrimary() {
- return isFlag(HAS_PRIMARY_FLAG_MASK);
- }
-
- /**
- * Sets hasPrimary flag value.
- */
- public void hasPrimary(boolean val) {
- setFlag(val, HAS_PRIMARY_FLAG_MASK);
- }
-
/** {@inheritDoc} */
@Nullable @Override public CacheEntryPredicate[] filter() {
return NO_FILTER;
}
-
- /**
- * Sets flag mask.
- *
- * @param flag Set or clear.
- * @param mask Mask.
- */
- private void setFlag(boolean flag, int mask) {
- flags = flag ? (byte)(flags | mask) : (byte)(flags & ~mask);
- }
-
- /**
- * Reads flag mask.
- *
- * @param mask Mask to read.
- * @return Flag value.
- */
- private boolean isFlag(int mask) {
- return (flags & mask) != 0;
- }
-
- /** {@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.writeByte("flags", flags))
- return false;
-
- writer.incrementState();
-
- case 4:
- if (!writer.writeLong("futId", futId))
- return false;
-
- writer.incrementState();
-
- case 5:
- if (!writer.writeByte("op", op != null ? (byte)op.ordinal() : -1))
- return false;
-
- writer.incrementState();
-
- case 6:
- if (!writer.writeUuid("subjId", subjId))
- return false;
-
- writer.incrementState();
-
- case 7:
- if (!writer.writeByte("syncMode", syncMode != null ? (byte)syncMode.ordinal() : -1))
- 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();
-
- case 10:
- if (!writer.writeMessage("updateVer", updateVer))
- 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:
- flags = reader.readByte("flags");
-
- if (!reader.isLastRead())
- return false;
-
- reader.incrementState();
-
- case 4:
- futId = reader.readLong("futId");
-
- if (!reader.isLastRead())
- return false;
-
- reader.incrementState();
-
- case 5:
- byte opOrd;
-
- opOrd = reader.readByte("op");
-
- if (!reader.isLastRead())
- return false;
-
- op = GridCacheOperation.fromOrdinal(opOrd);
-
- reader.incrementState();
-
- case 6:
- subjId = reader.readUuid("subjId");
-
- if (!reader.isLastRead())
- return false;
-
- reader.incrementState();
-
- case 7:
- byte syncModeOrd;
-
- syncModeOrd = reader.readByte("syncMode");
-
- if (!reader.isLastRead())
- return false;
-
- syncMode = CacheWriteSynchronizationMode.fromOrdinal(syncModeOrd);
-
- 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();
-
- case 10:
- updateVer = reader.readMessage("updateVer");
-
- if (!reader.isLastRead())
- return false;
-
- reader.incrementState();
-
- }
-
- return reader.afterMessageRead(GridNearAtomicAbstractSingleUpdateRequest.class);
- }
-
- /** {@inheritDoc} */
- @Override public byte fieldsCount() {
- return 11;
- }
}