You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@cassandra.apache.org by be...@apache.org on 2024/10/10 12:39:20 UTC
(cassandra) branch cep-15-accord updated: Journal diff serialization: validateFlags and WaitingOn size
This is an automated email from the ASF dual-hosted git repository.
benedict pushed a commit to branch cep-15-accord
in repository https://gitbox.apache.org/repos/asf/cassandra.git
The following commit(s) were added to refs/heads/cep-15-accord by this push:
new 73008ac097 Journal diff serialization: validateFlags and WaitingOn size
73008ac097 is described below
commit 73008ac0970e6e3d0c6f37db7f590aecc8ef6505
Author: Benedict Elliott Smith <be...@apache.org>
AuthorDate: Thu Oct 10 13:39:12 2024 +0100
Journal diff serialization: validateFlags and WaitingOn size
---
modules/accord | 2 +-
.../cassandra/service/accord/SavedCommand.java | 41 +++++++++++-----------
.../accord/serializers/WaitingOnSerializer.java | 4 +--
.../distributed/test/accord/AccordLoadTest.java | 4 +--
.../serializers/WaitingOnSerializerTest.java | 4 +--
5 files changed, 27 insertions(+), 28 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 <OBJ, VAL> int collectFlags(OBJ lo, OBJ ro, Function<OBJ, VAL> convert, boolean allowClassMismatch, Fields field, int oldFlags)
+ private static <OBJ, VAL> int collectFlags(OBJ lo, OBJ ro, Function<OBJ, VAL> 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 85becbca11..4c249a1a28 100644
--- a/test/distributed/org/apache/cassandra/distributed/test/accord/AccordLoadTest.java
+++ b/test/distributed/org/apache/cassandra/distributed/test/accord/AccordLoadTest.java
@@ -30,11 +30,9 @@ import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
-import java.util.concurrent.Future;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
-import java.util.concurrent.atomic.AtomicReference;
import com.google.common.util.concurrent.RateLimiter;
import org.junit.BeforeClass;
@@ -45,7 +43,6 @@ import org.slf4j.LoggerFactory;
import org.apache.cassandra.config.CassandraRelevantProperties;
import org.apache.cassandra.distributed.api.ConsistencyLevel;
-import org.apache.cassandra.distributed.api.Feature;
import org.apache.cassandra.distributed.api.ICoordinator;
import org.apache.cassandra.distributed.api.IMessage;
import org.apache.cassandra.distributed.api.IMessageFilters;
@@ -70,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)));
});
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@cassandra.apache.org
For additional commands, e-mail: commits-help@cassandra.apache.org