IGNITE-26630 Separate the stripe key from the cache partition

modules/core/src/main/java/org/apache/ignite/internal/StripedMessage.java

@@ -0,0 +1,37 @@
+/*
+ * 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;
+
+import org.apache.ignite.plugin.extensions.communication.Message;
+
+/** Message that chooses the stripe it is processed in. */
+public interface StripedMessage extends Message {
+    /** Process in any stripe. */
+    public static final int ANY_STRIPE = -1;
+
+    /** Process in the pool itself, not in a stripe. */
+    public static final int NO_STRIPE = Integer.MIN_VALUE;
+
+    /**
+     * The value is an index, not a cache partition: messages sharing it are processed one after another
+     * by the same thread.
+     *
+     * @return Stripe index, {@link #ANY_STRIPE} or {@link #NO_STRIPE}.
+     */
+    public int stripeIdx();
+}

modules/core/src/main/java/org/apache/ignite/internal/managers/communication/GridIoManager.java

@@ -142,6 +142,8 @@
 import static org.apache.ignite.internal.GridTopic.TOPIC_COMM_SYSTEM;
 import static org.apache.ignite.internal.GridTopic.TOPIC_COMM_USER;
 import static org.apache.ignite.internal.GridTopic.TOPIC_IO_TEST;
+import static org.apache.ignite.internal.StripedMessage.ANY_STRIPE;
+import static org.apache.ignite.internal.StripedMessage.NO_STRIPE;
 import static org.apache.ignite.internal.managers.communication.GridIoPolicy.AFFINITY_POOL;
 import static org.apache.ignite.internal.managers.communication.GridIoPolicy.CALLER_THREAD;
 import static org.apache.ignite.internal.managers.communication.GridIoPolicy.DATA_STREAMER_POOL;
@@ -1384,21 +1386,21 @@ private void processRegularMessage(
             if (msg0.processFromNioThread())
                 c.run();
             else
-                ctx.pools().getStripedExecutorService().execute(-1, c);
+                ctx.pools().getStripedExecutorService().execute(ANY_STRIPE, c);
 
             return;
         }
 
-        final int part = msg.partition(); // Store partition to avoid possible recalculation.
+        final int stripeIdx = msg.stripeIdx(); // Store to avoid possible recalculation.
 
-        if (plc == GridIoPolicy.SYSTEM_POOL && part != GridIoMessage.STRIPE_DISABLED_PART) {
-            ctx.pools().getStripedExecutorService().execute(part, c);
+        if (plc == GridIoPolicy.SYSTEM_POOL && stripeIdx != NO_STRIPE) {
+            ctx.pools().getStripedExecutorService().execute(stripeIdx, c);
 
             return;
         }
 
-        if (plc == GridIoPolicy.DATA_STREAMER_POOL && part != GridIoMessage.STRIPE_DISABLED_PART) {
-            ctx.pools().getDataStreamerExecutorService().execute(part, c);
+        if (plc == GridIoPolicy.DATA_STREAMER_POOL && stripeIdx != NO_STRIPE) {
+            ctx.pools().getDataStreamerExecutorService().execute(stripeIdx, c);
 
             return;
         }

modules/core/src/main/java/org/apache/ignite/internal/managers/communication/GridIoMessage.java

@@ -21,8 +21,7 @@
 import org.apache.ignite.internal.GridTopicMessage;
 import org.apache.ignite.internal.NioField;
 import org.apache.ignite.internal.Order;
-import org.apache.ignite.internal.processors.cache.GridCacheMessage;
-import org.apache.ignite.internal.processors.datastreamer.DataStreamerRequest;
+import org.apache.ignite.internal.StripedMessage;
 import org.apache.ignite.internal.thread.context.OperationContextSnapshotMessage;
 import org.apache.ignite.internal.util.nio.GridNioServer.MessageWrapper;
 import org.apache.ignite.internal.util.tostring.GridToStringInclude;
@@ -33,10 +32,7 @@
 /**
  * Wrapper for all grid messages.
  */
-public class GridIoMessage implements Message, MessageWrapper {
-    /** */
-    public static final Integer STRIPE_DISABLED_PART = Integer.MIN_VALUE;
-
+public class GridIoMessage implements StripedMessage, MessageWrapper {
     /** Policy. */
     @Order(0)
     byte plc;
@@ -175,18 +171,9 @@ boolean isOrdered() {
         throw new AssertionError();
     }
 
-    /**
-     * Get single partition for this message (if applicable).
-     *
-     * @return Partition ID.
-     */
-    public int partition() {
-        if (msg instanceof GridCacheMessage)
-            return ((GridCacheMessage)msg).partition();
-        if (msg instanceof DataStreamerRequest)
-            return ((DataStreamerRequest)msg).partition();
-        else
-            return STRIPE_DISABLED_PART;
+    /** {@inheritDoc} */
+    @Override public int stripeIdx() {
+        return msg instanceof StripedMessage ? ((StripedMessage)msg).stripeIdx() : NO_STRIPE;
     }
 
     /**

modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheIoManager.java

@@ -445,7 +445,7 @@ private boolean processMissedHandler(UUID nodeId, GridCacheMessage cacheMsg) {
                     nearEvicted.add(req.nearKey(i));
 
                 GridDhtAtomicUpdateResponse dhtRes = new GridDhtAtomicUpdateResponse(req.cacheId(),
-                    req.partition(),
+                    req.stripeIdx(),
                     req.futureId());
 
                 dhtRes.nearEvicted(nearEvicted);
@@ -458,7 +458,7 @@ private boolean processMissedHandler(UUID nodeId, GridCacheMessage cacheMsg) {
 
                 if (req.nearNodeId() != null) {
                     GridDhtAtomicNearResponse nearRes = new GridDhtAtomicNearResponse(req.cacheId(),
-                        req.partition(),
+                        req.stripeIdx(),
                         req.nearFutureId(),
                         nodeId,
                         req.flags());
@@ -791,7 +791,7 @@ else if (msg instanceof GridDhtTxPrepareRequest) {
             GridDhtTxPrepareRequest req = (GridDhtTxPrepareRequest)msg;
 
             GridDhtTxPrepareResponse res = new GridDhtTxPrepareResponse(
-                req.partition(),
+                req.stripeIdx(),
                 req.version(),
                 req.futureId(),
                 req.miniId(),
@@ -806,7 +806,7 @@ else if (msg instanceof GridDhtAtomicUpdateRequest) {
 
             GridDhtAtomicUpdateResponse res = new GridDhtAtomicUpdateResponse(
                 req.cacheId(),
-                req.partition(),
+                req.stripeIdx(),
                 req.futureId());
 
             res.onError(req.classError());
@@ -815,7 +815,7 @@ else if (msg instanceof GridDhtAtomicUpdateRequest) {
 
             if (req.nearNodeId() != null) {
                 GridDhtAtomicNearResponse nearRes = new GridDhtAtomicNearResponse(req.cacheId(),
-                    req.partition(),
+                    req.stripeIdx(),
                     req.nearFutureId(),
                     nodeId,
                     req.flags());
@@ -832,7 +832,7 @@ else if (msg instanceof GridNearAtomicFullUpdateRequest) {
                 req.cacheId(),
                 nodeId,
                 req.futureId(),
-                req.partition(),
+                req.stripeIdx(),
                 false);
 
             res.error(req.classError());
@@ -901,7 +901,7 @@ else if (msg instanceof GridNearTxPrepareRequest) {
             GridNearTxPrepareRequest req = (GridNearTxPrepareRequest)msg;
 
             GridNearTxPrepareResponse res = new GridNearTxPrepareResponse(
-                req.partition(),
+                req.stripeIdx(),
                 req.version(),
                 req.futureId(),
                 req.miniId(),
@@ -980,7 +980,7 @@ else if (msg instanceof GridNearAtomicSingleUpdateRequest) {
                 req.cacheId(),
                 nodeId,
                 req.futureId(),
-                req.partition(),
+                req.stripeIdx(),
                 false);
 
             res.error(req.classError());
@@ -994,7 +994,7 @@ else if (msg instanceof GridNearAtomicSingleUpdateInvokeRequest) {
                 req.cacheId(),
                 nodeId,
                 req.futureId(),
-                req.partition(),
+                req.stripeIdx(),
                 false);
 
             res.error(req.classError());
@@ -1008,7 +1008,7 @@ else if (msg instanceof GridNearAtomicSingleUpdateFilterRequest) {
                 req.cacheId(),
                 nodeId,
                 req.futureId(),
-                req.partition(),
+                req.stripeIdx(),
                 false);
 
             res.error(req.classError());
@@ -1020,7 +1020,7 @@ else if (msg instanceof GridDhtAtomicSingleUpdateRequest) {
 
             GridDhtAtomicUpdateResponse res = new GridDhtAtomicUpdateResponse(
                 req.cacheId(),
-                req.partition(),
+                req.stripeIdx(),
                 req.futureId());
 
             res.onError(req.classError());
@@ -1029,7 +1029,7 @@ else if (msg instanceof GridDhtAtomicSingleUpdateRequest) {
 
             if (req.nearNodeId() != null) {
                 GridDhtAtomicNearResponse nearRes = new GridDhtAtomicNearResponse(req.cacheId(),
-                    req.partition(),
+                    req.stripeIdx(),
                     req.nearFutureId(),
                     nodeId,
                     req.flags());

modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheMessage.java

@@ -26,6 +26,7 @@
 import org.apache.ignite.IgniteLogger;
 import org.apache.ignite.internal.DeferredUnmarshalMessage;
 import org.apache.ignite.internal.Order;
+import org.apache.ignite.internal.StripedMessage;
 import org.apache.ignite.internal.managers.deployment.GridDeployment;
 import org.apache.ignite.internal.managers.deployment.GridDeploymentInfo;
 import org.apache.ignite.internal.managers.deployment.GridDeploymentInfoBean;
@@ -44,7 +45,7 @@
  *
  * @see DeployableMessage
  */
-public abstract class GridCacheMessage implements DeferredUnmarshalMessage {
+public abstract class GridCacheMessage implements DeferredUnmarshalMessage, StripedMessage {
     /** Maximum number of cache lookup indexes. */
     public static final int MAX_CACHE_MSG_LOOKUP_INDEX = 7;
 
@@ -123,11 +124,9 @@ public int lookupIndex() {
         return -1;
     }
 
-    /**
-     * @return Partition ID this message is targeted to or {@code -1} if it cannot be determined.
-     */
-    public int partition() {
-        return -1;
+    /** {@inheritDoc} */
+    @Override public int stripeIdx() {
+        return ANY_STRIPE;
     }
 
     /**

modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridDistributedLockRequest.java

@@ -366,8 +366,8 @@ public List<KeyCacheObject> keys() {
     }
 
     /** {@inheritDoc} */
-    @Override public int partition() {
-        return keys != null && !keys.isEmpty() ? keys.get(0).partition() : -1;
+    @Override public int stripeIdx() {
+        return keys != null && !keys.isEmpty() ? keys.get(0).partition() : ANY_STRIPE;
     }
 
     /**

modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridDistributedTxFinishResponse.java

@@ -64,7 +64,7 @@ public GridDistributedTxFinishResponse(int part, GridCacheVersion txId, IgniteUu
     }
 
     /** {@inheritDoc} */
-    @Override public final int partition() {
+    @Override public final int stripeIdx() {
         return part;
     }
 

modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridDistributedTxPrepareResponse.java

@@ -78,7 +78,7 @@ public GridDistributedTxPrepareResponse(int part, GridCacheVersion xid, @Nullabl
     }
 
     /** {@inheritDoc} */
-    @Override public int partition() {
+    @Override public int stripeIdx() {
         return part;
     }
 

modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridNearUnlockRequest.java

@@ -90,8 +90,8 @@ public void addKey(KeyCacheObject key) {
     }
 
     /** {@inheritDoc} */
-    @Override public int partition() {
-        return keys != null && !keys.isEmpty() ? keys.get(0).partition() : -1;
+    @Override public int stripeIdx() {
+        return keys != null && !keys.isEmpty() ? keys.get(0).partition() : ANY_STRIPE;
     }
 
     /** {@inheritDoc} */

modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxFinishRequest.java

@@ -200,7 +200,7 @@ public Collection<PartitionUpdateCountersMessage> updateCounters() {
 
 
     /** {@inheritDoc} */
-    @Override public int partition() {
+    @Override public int stripeIdx() {
         return U.safeAbs(version().hashCode());
     }
 

modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxPrepareRequest.java

@@ -320,7 +320,7 @@ public boolean skipCompletedVersion() {
     }
 
     /** {@inheritDoc} */
-    @Override public int partition() {
+    @Override public int stripeIdx() {
         return U.safeAbs(version().hashCode());
     }
 

modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicCache.java

@@ -1766,7 +1766,7 @@ private void onForceKeysError(final UUID nodeId,
         GridNearAtomicUpdateResponse res = new GridNearAtomicUpdateResponse(ctx.cacheId(),
             nodeId,
             req.futureId(),
-            req.partition(),
+            req.stripeIdx(),
             false);
 
         res.addFailedKeys(req.keys(), e);
@@ -1789,7 +1789,7 @@ private void updateAllAsyncInternal0(
         GridNearAtomicUpdateResponse res = new GridNearAtomicUpdateResponse(ctx.cacheId(),
             node.id(),
             req.futureId(),
-            req.partition(),
+            req.stripeIdx(),
             false);
 
         assert !req.returnValue() || (req.operation() == TRANSFORM || req.size() == 1);
@@ -3241,7 +3241,7 @@ private void processCheckUpdateRequest(UUID nodeId, GridNearAtomicCheckUpdateReq
         GridNearAtomicUpdateResponse res = new GridNearAtomicUpdateResponse(ctx.cacheId(),
             nodeId,
             checkReq.futureId(),
-            checkReq.partition(),
+            checkReq.stripeIdx(),
             false);
 
         GridCacheReturn ret = new GridCacheReturn(false, true);
@@ -3263,7 +3263,7 @@ private void processDhtAtomicUpdateRequest(UUID nodeId, GridDhtAtomicAbstractUpd
                 ", writeVer=" + req.writeVersion() + ", node=" + nodeId + ']');
         }
 
-        assert req.partition() >= 0 : req;
+        assert req.stripeIdx() >= 0 : req;
 
         GridCacheVersion ver = req.writeVersion();
 
@@ -3273,7 +3273,7 @@ private void processDhtAtomicUpdateRequest(UUID nodeId, GridDhtAtomicAbstractUpd
 
         if (req.nearNodeId() != null) {
             nearRes = new GridDhtAtomicNearResponse(ctx.cacheId(),
-                req.partition(),
+                req.stripeIdx(),
                 req.nearFutureId(),
                 nodeId,
                 req.flags());
@@ -3407,7 +3407,7 @@ else if (req.nearSize() > 0) {
 
             if (nearEvicted != null) {
                 dhtRes = new GridDhtAtomicUpdateResponse(ctx.cacheId(),
-                    req.partition(),
+                    req.stripeIdx(),
                     req.futureId());
 
                 dhtRes.nearEvicted(nearEvicted);
@@ -3441,7 +3441,7 @@ else if (req.nearSize() > 0) {
         if (dhtRes != null)
             sendDhtPrimaryResponse(nodeId, req, dhtRes);
         else
-            sendDeferredUpdateResponse(req.partition(), nodeId, req.futureId());
+            sendDeferredUpdateResponse(req.stripeIdx(), nodeId, req.futureId());
     }
 
     /**

modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicNearResponse.java

@@ -112,7 +112,7 @@ public byte flags() {
     }
 
     /** {@inheritDoc} */
-    @Override public int partition() {
+    @Override public int stripeIdx() {
         return partId;
     }
 

modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicSingleUpdateRequest.java

@@ -210,7 +210,7 @@ private void near(boolean near) {
     }
 
     /** {@inheritDoc} */
-    @Override public int partition() {
+    @Override public int stripeIdx() {
         int p = key.partition();
 
         assert p >= 0;