IGNITE-27606 Replace GridUnsafe with byte-array VarHandle views in PartitionUpdateCountersMessage

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 int partition(int idx) {
         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 long initialCounter(int idx) {
         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 long updatesCount(int idx) {
         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 long updatesCount(int idx) {
     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 Long nextCounter(int partId) {
     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} */

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));
+    }
+}

modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheTestSuite10.java

@@ -100,6 +100,7 @@
 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 static List<Class<?>> suite(Collection<Class> ignoredTests) {
         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);