This is an automated email from the ASF dual-hosted git repository.
anton-vinogradov pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/ignite.git
The following commit(s) were added to refs/heads/master by this push:
new 152157d0040 IGNITE-27606 Replace GridUnsafe with byte-array VarHandle
views in PartitionUpdateCountersMessage (#13422)
152157d0040 is described below
commit 152157d0040fc695d7983acc1da62e3ac97b3a0e
Author: Anton Vinogradov <[email protected]>
AuthorDate: Mon Aug 3 20:31:46 2026 +0300
IGNITE-27606 Replace GridUnsafe with byte-array VarHandle views in
PartitionUpdateCountersMessage (#13422)
---
.../dht/PartitionUpdateCountersMessage.java | 37 ++++---
.../dht/PartitionUpdateCountersMessageTest.java | 118 +++++++++++++++++++++
.../ignite/testsuites/IgniteCacheTestSuite10.java | 2 +
3 files changed, 142 insertions(+), 15 deletions(-)
diff --git
a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/PartitionUpdateCountersMessage.java
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/PartitionUpdateCountersMessage.java
index d24ab7d7982..0f24739af71 100644
---
a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/PartitionUpdateCountersMessage.java
+++
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/PartitionUpdateCountersMessage.java
@@ -17,10 +17,12 @@
package org.apache.ignite.internal.processors.cache.distributed.dht;
+import java.lang.invoke.MethodHandles;
+import java.lang.invoke.VarHandle;
+import java.nio.ByteOrder;
import java.util.Arrays;
import java.util.Map;
import org.apache.ignite.internal.Order;
-import org.apache.ignite.internal.util.GridUnsafe;
import org.apache.ignite.internal.util.typedef.internal.U;
import org.apache.ignite.plugin.extensions.communication.Message;
import org.apache.ignite.plugin.extensions.communication.MessageFactory;
@@ -34,6 +36,16 @@ public class PartitionUpdateCountersMessage implements
Message {
/** */
private static final int ITEM_SIZE = 4 /* partition */ + 8 /* initial
counter */ + 8 /* updates count */;
+ /**
+ * Views over {@link #data}. The byte order is pinned instead of following
the host, so that the bytes a node puts
+ * on the wire do not depend on the architecture it runs on. Item fields
are not naturally aligned, which the plain
+ * {@code get}/{@code set} access modes used here allow.
+ */
+ private static final VarHandle INT_VIEW =
MethodHandles.byteArrayViewVarHandle(int[].class, ByteOrder.LITTLE_ENDIAN);
+
+ /** */
+ private static final VarHandle LONG_VIEW =
MethodHandles.byteArrayViewVarHandle(long[].class, ByteOrder.LITTLE_ENDIAN);
+
/** */
@Order(0)
int cacheId;
@@ -87,9 +99,7 @@ public class PartitionUpdateCountersMessage implements
Message {
if (idx >= size)
throw new ArrayIndexOutOfBoundsException();
- long off = GridUnsafe.BYTE_ARR_OFF + (long)idx * ITEM_SIZE;
-
- return GridUnsafe.getInt(data, off);
+ return (int)INT_VIEW.get(data, idx * ITEM_SIZE);
}
/**
@@ -100,9 +110,7 @@ public class PartitionUpdateCountersMessage implements
Message {
if (idx >= size)
throw new ArrayIndexOutOfBoundsException();
- long off = GridUnsafe.BYTE_ARR_OFF + (long)idx * ITEM_SIZE + 4;
-
- return GridUnsafe.getLong(data, off);
+ return (long)LONG_VIEW.get(data, idx * ITEM_SIZE + 4);
}
/**
@@ -113,9 +121,7 @@ public class PartitionUpdateCountersMessage implements
Message {
if (idx >= size)
throw new ArrayIndexOutOfBoundsException();
- long off = GridUnsafe.BYTE_ARR_OFF + (long)idx * ITEM_SIZE + 12;
-
- return GridUnsafe.getLong(data, off);
+ return (long)LONG_VIEW.get(data, idx * ITEM_SIZE + 12);
}
/**
@@ -128,11 +134,11 @@ public class PartitionUpdateCountersMessage implements
Message {
public void add(int part, long init, long updatesCnt) {
ensureSpace(size + 1);
- long off = GridUnsafe.BYTE_ARR_OFF + (long)size++ * ITEM_SIZE;
+ int off = size++ * ITEM_SIZE;
- GridUnsafe.putInt(data, off, part); off += 4;
- GridUnsafe.putLong(data, off, init); off += 8;
- GridUnsafe.putLong(data, off, updatesCnt);
+ INT_VIEW.set(data, off, part);
+ LONG_VIEW.set(data, off + 4, init);
+ LONG_VIEW.set(data, off + 12, updatesCnt);
}
/** Optimizes the memory used after adding counters with {@link #add(int,
long, long)}. */
@@ -170,8 +176,9 @@ public class PartitionUpdateCountersMessage implements
Message {
private void ensureSpace(int newSize) {
int req = newSize * ITEM_SIZE;
+ // Growth alone may fall short of the request: 1.33 of a one-item
array is still less than two items.
if (data.length < req)
- data = Arrays.copyOf(data, (int)(data.length * 1.33f));
+ data = Arrays.copyOf(data, Math.max(req, (int)(data.length *
1.33f)));
}
/** {@inheritDoc} */
diff --git
a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/dht/PartitionUpdateCountersMessageTest.java
b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/dht/PartitionUpdateCountersMessageTest.java
new file mode 100644
index 00000000000..15d07444a1e
--- /dev/null
+++
b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/dht/PartitionUpdateCountersMessageTest.java
@@ -0,0 +1,118 @@
+/*
+ * 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.Arrays;
+import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
+import org.junit.Test;
+
+/** Tests the packed counter storage of {@link
PartitionUpdateCountersMessage}. */
+public class PartitionUpdateCountersMessageTest extends GridCommonAbstractTest
{
+ /** */
+ private static final int CACHE_ID = 42;
+
+ /** */
+ @Test
+ public void testItemsAreReadBackAsWritten() {
+ PartitionUpdateCountersMessage msg = new
PartitionUpdateCountersMessage(CACHE_ID, 3);
+
+ msg.add(1, 100L, 5L);
+ msg.add(2, Long.MAX_VALUE, 1L);
+ msg.add(Integer.MAX_VALUE, 0L, Long.MAX_VALUE);
+
+ assertEquals(CACHE_ID, msg.cacheId());
+ assertEquals(3, msg.size());
+
+ assertEquals(1, msg.partition(0));
+ assertEquals(100L, msg.initialCounter(0));
+ assertEquals(5L, msg.updatesCount(0));
+
+ assertEquals(2, msg.partition(1));
+ assertEquals(Long.MAX_VALUE, msg.initialCounter(1));
+ assertEquals(1L, msg.updatesCount(1));
+
+ assertEquals(Integer.MAX_VALUE, msg.partition(2));
+ assertEquals(0L, msg.initialCounter(2));
+ assertEquals(Long.MAX_VALUE, msg.updatesCount(2));
+ }
+
+ /**
+ * Adding past the initial size must grow the storage. Growth by a factor
alone falls short of the request for a
+ * small array, and the write then lands outside it.
+ */
+ @Test
+ public void testAddPastInitialSize() {
+ PartitionUpdateCountersMessage msg = new
PartitionUpdateCountersMessage(CACHE_ID, 1);
+
+ for (int i = 0; i < 64; i++)
+ msg.add(i, i * 10L, i * 100L);
+
+ assertEquals(64, msg.size());
+
+ for (int i = 0; i < 64; i++) {
+ assertEquals(i, msg.partition(i));
+ assertEquals(i * 10L, msg.initialCounter(i));
+ assertEquals(i * 100L, msg.updatesCount(i));
+ }
+ }
+
+ /** The wire form must not depend on the byte order of the host, so the
layout is asserted byte by byte. */
+ @Test
+ public void testWireLayoutIsLittleEndian() {
+ PartitionUpdateCountersMessage msg = new
PartitionUpdateCountersMessage(CACHE_ID, 1);
+
+ msg.add(0x04030201, 0x0807060504030201L, 0x1817161514131211L);
+
+ byte[] expected = {
+ 0x01, 0x02, 0x03, 0x04,
+ 0x01, 0x02, 0x03, 0x04, 0x05, 0x06, 0x07, 0x08,
+ 0x11, 0x12, 0x13, 0x14, 0x15, 0x16, 0x17, 0x18
+ };
+
+ assertTrue("Unexpected wire layout", Arrays.equals(expected,
msg.data));
+ }
+
+ /** */
+ @Test
+ public void testFinishUpdatingTrimsSpareSpace() {
+ PartitionUpdateCountersMessage msg = new
PartitionUpdateCountersMessage(CACHE_ID, 8);
+
+ msg.add(1, 1L, 1L);
+
+ msg.finishUpdating();
+
+ assertEquals(20, msg.data.length);
+
+ assertEquals(1, msg.partition(0));
+ assertEquals(1L, msg.initialCounter(0));
+ assertEquals(1L, msg.updatesCount(0));
+ }
+
+ /** */
+ @Test
+ public void testNextCounterFollowsInitialCounter() {
+ PartitionUpdateCountersMessage msg = new
PartitionUpdateCountersMessage(CACHE_ID, 2);
+
+ msg.add(7, 30L, 2L);
+
+ assertEquals((Long)31L, msg.nextCounter(7));
+ assertEquals((Long)32L, msg.nextCounter(7));
+
+ assertNull(msg.nextCounter(8));
+ }
+}
diff --git
a/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheTestSuite10.java
b/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheTestSuite10.java
index d446b70b293..dd18bf7f1c4 100755
---
a/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheTestSuite10.java
+++
b/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheTestSuite10.java
@@ -100,6 +100,7 @@ import
org.apache.ignite.internal.processors.cache.distributed.dht.GridCachePart
import
org.apache.ignite.internal.processors.cache.distributed.dht.GridCachePartitionsStateValidatorSelfTest;
import
org.apache.ignite.internal.processors.cache.distributed.dht.GridCachePartitionsUpdateCountersAndSizeTest;
import
org.apache.ignite.internal.processors.cache.distributed.dht.IgniteCacheConcurrentPutGetRemoveTest;
+import
org.apache.ignite.internal.processors.cache.distributed.dht.PartitionUpdateCountersMessageTest;
import
org.apache.ignite.internal.processors.cache.distributed.near.GridCacheNearTxExceptionSelfTest;
import
org.apache.ignite.internal.processors.cache.distributed.near.GridCachePartitionedStorePutSelfTest;
import
org.apache.ignite.internal.processors.cache.distributed.replicated.GridCacheReplicatedTxExceptionSelfTest;
@@ -188,6 +189,7 @@ public class IgniteCacheTestSuite10 {
GridTestUtils.addTestIfNeeded(suite,
IgniteMessageFactoryImplTest.class, ignoredTests);
GridTestUtils.addTestIfNeeded(suite,
MessageDirectTypeIdConflictTest.class, ignoredTests);
GridTestUtils.addTestIfNeeded(suite,
IgniteCoreMessagesSerializationTest.class, ignoredTests);
+ GridTestUtils.addTestIfNeeded(suite,
PartitionUpdateCountersMessageTest.class, ignoredTests);
GridTestUtils.addTestIfNeeded(suite,
CommunicationMessageDelayTest.class, ignoredTests);
GridTestUtils.addTestIfNeeded(suite,
IgniteIncompleteCacheObjectSelfTest.class, ignoredTests);
