diff --git a/CHANGES.txt b/CHANGES.txt index 21d4fac66d95..e9f75dfd625d 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -1,4 +1,5 @@ 7.0 + * Keep message ids unsigned so they do not inflate every message header once the id counter wraps (CASSANDRA-21575) * Add prepared statement cache stats to nodetool info (CASSANDRA-14366) * Don't increment client metrics on messaging service connection unpause (CASSANDRA-21491) * Add nodetool getreplicas (CASSANDRA-17665) diff --git a/src/java/org/apache/cassandra/net/ForwardingInfo.java b/src/java/org/apache/cassandra/net/ForwardingInfo.java index 0929f47e96f7..8a5c04e5b251 100644 --- a/src/java/org/apache/cassandra/net/ForwardingInfo.java +++ b/src/java/org/apache/cassandra/net/ForwardingInfo.java @@ -108,7 +108,7 @@ public ForwardingInfo deserialize(DataInputPlus in, int version) throws IOExcept for (int i = 0; i < count; i++) { targets.add(inetAddressAndPortSerializer.deserialize(in, version)); - ids[i] = in.readUnsignedVInt32(); + ids[i] = in.readUnsignedVInt(); } return new ForwardingInfo(targets, ids); diff --git a/src/java/org/apache/cassandra/net/Message.java b/src/java/org/apache/cassandra/net/Message.java index 14b510937fb5..f6f953c00bd1 100644 --- a/src/java/org/apache/cassandra/net/Message.java +++ b/src/java/org/apache/cassandra/net/Message.java @@ -483,13 +483,27 @@ private static long nextId() long id; do { - id = nextId.incrementAndGet(); + id = toUnsignedId(nextId.incrementAndGet()); } while (id == NO_ID); return id; } + /** + * Widens the id counter as unsigned. + *

+ * The counter is an {@code int} and wraps to {@link Integer#MIN_VALUE} once it passes + * {@link Integer#MAX_VALUE}. Widening it directly would sign-extend, and {@link Serializer} writes the id as an + * unsigned vint, so every negative id occupies the maximum width of 9 bytes instead of at most 5. Masking keeps + * ids in {@code [0, 2^32)}, which covers the same number of distinct values while staying cheap to encode. + */ + @VisibleForTesting + static long toUnsignedId(int counter) + { + return counter & 0xFFFFFFFFL; + } + /** * WARNING: this is inaccurate for messages from pre40 nodes, which can use 0 as an id (but will do so rarely) */ diff --git a/test/unit/org/apache/cassandra/net/ForwardingInfoTest.java b/test/unit/org/apache/cassandra/net/ForwardingInfoTest.java index cfbb945026b5..8aaea28f89fd 100644 --- a/test/unit/org/apache/cassandra/net/ForwardingInfoTest.java +++ b/test/unit/org/apache/cassandra/net/ForwardingInfoTest.java @@ -44,6 +44,42 @@ public void testSupportedVersions() throws Exception testVersion(version.value); } + /** + * Message ids are drawn from an unsigned 32-bit counter, so they span the whole [1, 2^32) range. They are written + * as 64-bit unsigned vints, and the reader has to use the matching width: reading them back as a 32-bit vint + * throws {@link org.apache.cassandra.utils.vint.VIntOutOfRangeException} for any id above Integer.MAX_VALUE. + */ + @Test + public void testLargeMessageIdsRoundTrip() throws Exception + { + InetAddressAndPort.initializeDefaultPort(65532); + List addresses = ImmutableList.of(InetAddressAndPort.getByName("127.0.0.1:7000"), + InetAddressAndPort.getByName("127.0.0.2:7000"), + InetAddressAndPort.getByName("127.0.0.3:7000"), + InetAddressAndPort.getByName("127.0.0.4:7000")); + + long[] ids = { 1L, Integer.MAX_VALUE, Integer.MAX_VALUE + 1L, 0xFFFFFFFFL }; + ForwardingInfo forwardingInfo = new ForwardingInfo(addresses, ids); + + for (MessagingService.Version version : MessagingService.Version.supportedVersions()) + { + ByteBuffer buffer; + try (DataOutputBuffer dob = new DataOutputBuffer()) + { + ForwardingInfo.serializer.serialize(forwardingInfo, dob, version.value); + buffer = dob.buffer(); + } + + assertEquals(buffer.remaining(), ForwardingInfo.serializer.serializedSize(forwardingInfo, version.value)); + + try (DataInputBuffer dib = new DataInputBuffer(buffer, false)) + { + ForwardingInfo deserialized = ForwardingInfo.serializer.deserialize(dib, version.value); + assertTrue(Arrays.equals(ids, deserialized.messageIds)); + } + } + } + private void testVersion(int version) throws Exception { InetAddressAndPort.initializeDefaultPort(65532); diff --git a/test/unit/org/apache/cassandra/net/MessageTest.java b/test/unit/org/apache/cassandra/net/MessageTest.java index 677199e4fcfc..8c4265510922 100644 --- a/test/unit/org/apache/cassandra/net/MessageTest.java +++ b/test/unit/org/apache/cassandra/net/MessageTest.java @@ -46,6 +46,7 @@ import org.apache.cassandra.utils.FBUtilities; import org.apache.cassandra.utils.FreeRunningClock; import org.apache.cassandra.utils.TimeUUID; +import org.apache.cassandra.utils.vint.VIntCoding; import static com.google.common.base.Throwables.getStackTraceAsString; import static org.apache.cassandra.exceptions.RemoteExceptionTest.normalizeThrowable; @@ -368,4 +369,55 @@ public void testCreationTime() long localTimeNanos = localClock.now(); assertTrue( Message.Serializer.calculateCreationTimeNanos(remoteCreatedAt, localClock.translate(), localTimeNanos) > 0); } + + /** + * The id counter is an int, so it wraps to Integer.MIN_VALUE after Integer.MAX_VALUE. Ids must stay within + * [0, 2^32) across that wrap: the serializer writes them as unsigned vints, so a sign-extended negative id would + * occupy the maximum width of 9 bytes rather than at most 5. + */ + @Test + public void testIdsRemainUnsignedAcrossCounterWrap() + { + for (int counter : new int[]{ 1, Integer.MAX_VALUE - 1, Integer.MAX_VALUE, Integer.MIN_VALUE, Integer.MIN_VALUE + 1, -1 }) + { + long id = Message.toUnsignedId(counter); + assertTrue("id must not be negative, got " + id + " for counter " + counter, id >= 0); + assertTrue("id must fit in 32 unsigned bits, got " + id, id <= 0xFFFFFFFFL); + assertTrue("id must encode within 5 bytes, got " + VIntCoding.computeUnsignedVIntSize(id), + VIntCoding.computeUnsignedVIntSize(id) <= 5); + } + + // the mapping stays injective, so ids remain as distinct as the counter itself + assertEquals(Integer.MAX_VALUE + 1L, Message.toUnsignedId(Integer.MIN_VALUE)); + assertEquals(0xFFFFFFFFL, Message.toUnsignedId(-1)); + } + + /** + * Ids now span the whole unsigned 32-bit range, so the message header must round-trip the upper half too. + */ + @Test + public void testLargeIdRoundTrips() throws IOException + { + for (Version version : Version.supportedVersions()) + { + for (long id : new long[]{ 1L, Integer.MAX_VALUE, Integer.MAX_VALUE + 1L, 0xFFFFFFFFL }) + { + Message msg = Message.builder(Verb._TEST_1, noPayload) + .withId(id) + .from(FBUtilities.getBroadcastAddressAndPort()) + .build(); + + try (DataOutputBuffer out = new DataOutputBuffer()) + { + serializer.serialize(msg, out, version.value); + assertEquals(msg.serializedSize(version.value), out.getLength()); + + try (DataInputBuffer in = new DataInputBuffer(out.buffer(), false)) + { + assertEquals(id, serializer.deserialize(in, msg.from(), version.value).id()); + } + } + } + } + } }