natas_hw
/
cassandra
mirror of https://github.com/apache/cassandra
0
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Journal diff serialization: validateFlags and WaitingOn size

This commit is contained in:
Benedict Elliott Smith 2024-10-10 13:39:12 +01:00 committed by David Capwell
parent 25b1bd74b8
commit b8f3b74587
5 changed files with 27 additions and 25 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
modules/accord

@ -1 +1 @@
Subproject commit 8f11c206ca178dffbe67a4b3d7e23288e9545d3a
Subproject commit 8d1204045f1e1d4f0c3e09d1e2fa47837e55f819

41
src/java/org/apache/cassandra/service/accord/SavedCommand.java
View File

@ -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)

4
src/java/org/apache/cassandra/service/accord/serializers/WaitingOnSerializer.java
View File

@ -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)

1
test/distributed/org/apache/cassandra/distributed/test/accord/AccordLoadTest.java
View File

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

4
test/unit/org/apache/cassandra/service/accord/serializers/WaitingOnSerializerTest.java
View File

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