From b8f3b745870de4ffe0e5654ff58e5165cebb7ca4 Mon Sep 17 00:00:00 2001 From: Benedict Elliott Smith Date: Thu, 10 Oct 2024 13:39:12 +0100 Subject: [PATCH] Journal diff serialization: validateFlags and WaitingOn size --- modules/accord | 2 +- .../service/accord/SavedCommand.java | 41 ++++++++++--------- .../serializers/WaitingOnSerializer.java | 4 +- .../test/accord/AccordLoadTest.java | 1 + .../serializers/WaitingOnSerializerTest.java | 4 +- 5 files changed, 27 insertions(+), 25 deletions(-) diff --git a/modules/accord b/modules/accord index 8f11c206ca..8d1204045f 160000 --- a/modules/accord +++ b/modules/accord @@ -1 +1 @@ -Subproject commit 8f11c206ca178dffbe67a4b3d7e23288e9545d3a +Subproject commit 8d1204045f1e1d4f0c3e09d1e2fa47837e55f819 diff --git a/src/java/org/apache/cassandra/service/accord/SavedCommand.java b/src/java/org/apache/cassandra/service/accord/SavedCommand.java index 5a0ae5d8ef..56206fcdde 100644 --- a/src/java/org/apache/cassandra/service/accord/SavedCommand.java +++ b/src/java/org/apache/cassandra/service/accord/SavedCommand.java @@ -147,22 +147,22 @@ public class SavedCommand return new SavedCommand.DiffWriter(original, current); } - // TODO (required): this is very inefficient + // TODO (required): calculate flags once private static boolean anyFieldChanged(Command before, Command after) { - int flags = getFlags(before, after); - for (Fields field : Fields.values()) - { - if (getFieldChanged(field, flags)) - return true; - } + int flags = validateFlags(getFlags(before, after)); + return (flags >>> 16) != 0; + } - return false; - } + private static int validateFlags(int flags) + { + Invariants.checkState(0 == (~(flags >>> 16) & (flags & 0xffff))); + return flags; + } public static void serialize(Command before, Command after, DataOutputPlus out, int userVersion) throws IOException { - int flags = getFlags(before, after); + int flags = validateFlags(getFlags(before, after)); out.writeInt(flags); int iterable = toIterableSetFields(flags); @@ -207,7 +207,8 @@ public class SavedCommand case WAITING_ON: Command.WaitingOn waitingOn = getWaitingOn(after); long size = WaitingOnSerializer.serializedSize(waitingOn); - ByteBuffer serialized = WaitingOnSerializer.serialize(after.txnId(), waitingOn); + ByteBuffer serialized = WaitingOnSerializer.serialize(waitingOn); + Invariants.checkState(serialized.remaining() == size); out.writeInt((int) size); out.write(serialized); break; @@ -254,28 +255,28 @@ public class SavedCommand return null; } - private static int collectFlags(OBJ lo, OBJ ro, Function convert, boolean allowClassMismatch, Fields field, int oldFlags) + private static int collectFlags(OBJ lo, OBJ ro, Function convert, boolean allowClassMismatch, Fields field, int flags) { VAL l = null; VAL r = null; if (lo != null) l = convert.apply(lo); if (ro != null) r = convert.apply(ro); - if (r == null) - oldFlags = setFieldIsNull(field, oldFlags); - if (l == r) - return oldFlags; // no change + return flags; // no change + + if (r == null) + flags = setFieldIsNull(field, flags); if (l == null || r == null) - return setFieldChanged(field, oldFlags); + return setFieldChanged(field, flags); assert allowClassMismatch || l.getClass() == r.getClass() : String.format("%s != %s", l.getClass(), r.getClass()); if (l.equals(r)) - return oldFlags; // no change + return flags; // no change - return setFieldChanged(field, oldFlags); + return setFieldChanged(field, flags); } private static int setFieldChanged(Fields field, int oldFlags) @@ -593,7 +594,7 @@ public class SavedCommand public void serialize(DataOutputPlus out, int userVersion) throws IOException { - out.writeInt(flags); + out.writeInt(validateFlags(flags)); int iterable = toIterableSetFields(flags); while (iterable != 0) diff --git a/src/java/org/apache/cassandra/service/accord/serializers/WaitingOnSerializer.java b/src/java/org/apache/cassandra/service/accord/serializers/WaitingOnSerializer.java index 021f01ec75..7951135514 100644 --- a/src/java/org/apache/cassandra/service/accord/serializers/WaitingOnSerializer.java +++ b/src/java/org/apache/cassandra/service/accord/serializers/WaitingOnSerializer.java @@ -57,13 +57,13 @@ public class WaitingOnSerializer return (long) TypeSizes.LONG_SIZE * length; } - public static ByteBuffer serialize(TxnId txnId, WaitingOn waitingOn) throws IOException + public static ByteBuffer serialize(WaitingOn waitingOn) throws IOException { int keyCount = waitingOn.keys.size(); int txnIdCount = waitingOn.txnIdCount(); int waitingOnLength = (txnIdCount + keyCount + 63) / 64; int appliedOrInvalidatedLength = 0; - if (txnId.domain() == Routable.Domain.Range) + if (waitingOn.appliedOrInvalidated != null) appliedOrInvalidatedLength = (txnIdCount + 63) / 64; ByteBuffer out = ByteBuffer.allocate(TypeSizes.sizeofUnsignedVInt(keyCount) + TypeSizes.sizeofUnsignedVInt(txnIdCount) diff --git a/test/distributed/org/apache/cassandra/distributed/test/accord/AccordLoadTest.java b/test/distributed/org/apache/cassandra/distributed/test/accord/AccordLoadTest.java index 7cdf4ae1c5..4c249a1a28 100644 --- a/test/distributed/org/apache/cassandra/distributed/test/accord/AccordLoadTest.java +++ b/test/distributed/org/apache/cassandra/distributed/test/accord/AccordLoadTest.java @@ -67,6 +67,7 @@ public class AccordLoadTest extends AccordTestBase // AccordTestBase.setupCluster(builder -> builder, 3); AccordTestBase.setupCluster(builder -> builder.withConfig(config -> config .set("accord.schedule_durability_frequency", "5s") + .set("accord.ephemeral_read_enabled", "true") .set("accord.gc_delay", "5s")), 3); } diff --git a/test/unit/org/apache/cassandra/service/accord/serializers/WaitingOnSerializerTest.java b/test/unit/org/apache/cassandra/service/accord/serializers/WaitingOnSerializerTest.java index ce5aabba9f..1f869d22f9 100644 --- a/test/unit/org/apache/cassandra/service/accord/serializers/WaitingOnSerializerTest.java +++ b/test/unit/org/apache/cassandra/service/accord/serializers/WaitingOnSerializerTest.java @@ -55,12 +55,12 @@ public class WaitingOnSerializerTest TxnId txnId = TxnId.NONE; if (waitingOn.appliedOrInvalidated != null) txnId = new TxnId(txnId.epoch(), txnId.hlc(), txnId.kind(), Routable.Domain.Range, txnId.node); long expectedSize = WaitingOnSerializer.serializedSize(waitingOn); - ByteBuffer bb = WaitingOnSerializer.serialize(txnId, waitingOn); + ByteBuffer bb = WaitingOnSerializer.serialize(waitingOn); Assertions.assertThat(bb.remaining()).isEqualTo(expectedSize); Command.WaitingOn read = WaitingOnSerializer.deserialize(txnId, waitingOn.keys, waitingOn.directRangeDeps, waitingOn.directKeyDeps, bb); Assertions.assertThat(read) .isEqualTo(waitingOn) - .isEqualTo(WaitingOnSerializer.deserialize(txnId, waitingOn.keys, waitingOn.directRangeDeps, waitingOn.directKeyDeps, WaitingOnSerializer.serialize(txnId, waitingOn))); + .isEqualTo(WaitingOnSerializer.deserialize(txnId, waitingOn.keys, waitingOn.directRangeDeps, waitingOn.directKeyDeps, WaitingOnSerializer.serialize(waitingOn))); }); }