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 d24ab7d7982be..0f24739af71f3 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 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} */ 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 0000000000000..15d07444a1e32 --- /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 d446b70b293f9..dd18bf7f1c4db 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.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> suite(Collection 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);