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/08/09 11:45:17 UTC
(cassandra) 01/01: CommandsForKey Improvements incl Pruning
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
commit 08fd33c4b28290151a9a18b1d771059b3b4ba52f
Author: Benedict Elliott Smith <be...@apache.org>
AuthorDate: Thu Jul 18 17:00:24 2024 +0100
CommandsForKey Improvements incl Pruning
CommandsForKey periodically self-prunes, so as to continue functioning well in-between garbage collections. Once we prune we are left with potentially incomplete information, and have to sometimes load per-command information from disk. But the payoff is ensuring CommandsForKey objects - which drive the majority of the state machine - are kept to a reasonable size.
patch by Benedict; reviewed by Blake Eggleston and David Capwell
---
modules/accord | 2 +-
.../config/CassandraRelevantProperties.java | 1 +
.../dht/IPartitionerDependentSerializer.java | 3 +-
src/java/org/apache/cassandra/io/util/Memory.java | 2 +-
.../org/apache/cassandra/net/MessagingService.java | 11 +
.../apache/cassandra/schema/SchemaKeyspace.java | 1 -
.../cassandra/service/accord/AccordCallback.java | 4 +-
.../service/accord/AccordCommandStore.java | 26 +-
.../cassandra/service/accord/AccordKeyspace.java | 9 +-
.../service/accord/AccordObjectSizes.java | 40 ++-
.../service/accord/AccordSafeCommand.java | 41 ++-
.../service/accord/AccordSafeCommandStore.java | 23 +-
.../service/accord/AccordSafeCommandsForKey.java | 14 +-
.../cassandra/service/accord/AccordSafeState.java | 2 +-
.../service/accord/AccordSafeTimestampsForKey.java | 7 -
.../cassandra/service/accord/AccordStateCache.java | 13 +-
.../service/accord/ImmutableAccordSafeState.java | 6 -
.../cassandra/service/accord/SavedCommand.java | 6 +-
.../cassandra/service/accord/api/AccordAgent.java | 13 +
.../service/accord/api/AccordRoutingKey.java | 4 +-
.../service/accord/async/AsyncLoader.java | 24 +-
.../service/accord/async/AsyncOperation.java | 38 +-
.../serializers/CommandStoreSerializers.java | 9 +-
.../serializers/CommandsForKeySerializer.java | 395 ++++++++++++++-------
.../service/accord/serializers/DepsSerializer.java | 137 ++++---
.../accord/serializers/WaitingOnSerializer.java | 133 ++-----
.../cassandra/service/accord/txn/TxnRead.java | 18 -
.../cassandra/service/accord/txn/TxnUpdate.java | 16 -
.../org/apache/cassandra/tcm/Transformation.java | 12 -
.../org/apache/cassandra/utils/MerkleTree.java | 2 +-
.../org/apache/cassandra/utils/concurrent/Ref.java | 44 ++-
.../apache/cassandra/utils/memory/BufferPool.java | 4 +-
.../test/accord/AccordIncrementalRepairTest.java | 2 +-
.../distributed/test/accord/AccordLoadTest.java | 10 +
.../test/log/FetchLogFromPeers2Test.java | 1 -
.../compaction/CompactionAccordIteratorsTest.java | 7 +-
test/unit/org/apache/cassandra/dht/TokenTest.java | 51 +++
.../apache/cassandra/io/IVersionedSerializers.java | 41 +++
.../service/accord/AccordCommandStoreTest.java | 12 +-
.../service/accord/AccordCommandTest.java | 2 +-
.../service/accord/AccordKeyspaceTest.java | 3 +-
.../service/accord/AccordStateCacheTest.java | 5 -
.../service/accord/async/AsyncLoaderTest.java | 14 +-
.../service/accord/async/AsyncOperationTest.java | 37 +-
.../serializers/CommandStoreSerializersTest.java | 72 ++++
.../serializers/CommandsForKeySerializerTest.java | 88 +++--
.../accord/serializers/DepsSerializerTest.java | 64 ++++
.../serializers/WaitingOnSerializerTest.java | 20 +-
.../apache/cassandra/utils/AccordGenerators.java | 65 +++-
.../cassandra/utils/CassandraGenerators.java | 11 +-
50 files changed, 1020 insertions(+), 545 deletions(-)
diff --git a/modules/accord b/modules/accord
index 4c870dc9b5..5f360e0b5b 160000
--- a/modules/accord
+++ b/modules/accord
@@ -1 +1 @@
-Subproject commit 4c870dc9b561a841ea9b923ff739953adcc00325
+Subproject commit 5f360e0b5b197156df0ef3d9985cd94d18ea1c92
diff --git a/src/java/org/apache/cassandra/config/CassandraRelevantProperties.java b/src/java/org/apache/cassandra/config/CassandraRelevantProperties.java
index 5d29984a0c..7a961491ee 100644
--- a/src/java/org/apache/cassandra/config/CassandraRelevantProperties.java
+++ b/src/java/org/apache/cassandra/config/CassandraRelevantProperties.java
@@ -561,6 +561,7 @@ public enum CassandraRelevantProperties
TEST_COMPRESSION("cassandra.test.compression"),
TEST_COMPRESSION_ALGO("cassandra.test.compression.algo", "lz4"),
TEST_DEBUG_REF_COUNT("cassandra.debugrefcount"),
+ TEST_DEBUG_REF_EVENTS("cassandra.debug.refevents"),
TEST_DRIVER_CONNECTION_TIMEOUT_MS("cassandra.test.driver.connection_timeout_ms", "5000"),
TEST_DRIVER_READ_TIMEOUT_MS("cassandra.test.driver.read_timeout_ms", "12000"),
TEST_ENCRYPTION("cassandra.test.encryption", "false"),
diff --git a/src/java/org/apache/cassandra/dht/IPartitionerDependentSerializer.java b/src/java/org/apache/cassandra/dht/IPartitionerDependentSerializer.java
index a70eb83771..8c612fb23b 100644
--- a/src/java/org/apache/cassandra/dht/IPartitionerDependentSerializer.java
+++ b/src/java/org/apache/cassandra/dht/IPartitionerDependentSerializer.java
@@ -19,6 +19,7 @@ package org.apache.cassandra.dht;
import java.io.IOException;
+import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.io.IVersionedSerializer;
import org.apache.cassandra.io.util.DataInputPlus;
@@ -43,6 +44,6 @@ public interface IPartitionerDependentSerializer<T> extends IVersionedSerializer
default T deserialize(DataInputPlus in, int version) throws IOException
{
- return deserialize(in, null, version);
+ return deserialize(in, DatabaseDescriptor.getPartitioner(), version);
}
}
diff --git a/src/java/org/apache/cassandra/io/util/Memory.java b/src/java/org/apache/cassandra/io/util/Memory.java
index 1d1fca2edf..6913caf4cf 100644
--- a/src/java/org/apache/cassandra/io/util/Memory.java
+++ b/src/java/org/apache/cassandra/io/util/Memory.java
@@ -81,7 +81,7 @@ public class Memory implements AutoCloseable, ReadableMemory
if (bytes < 0)
throw new IllegalArgumentException();
- if (Ref.DEBUG_ENABLED)
+ if (Ref.TRACE_ENABLED)
return new SafeMemory(bytes);
return new Memory(bytes);
diff --git a/src/java/org/apache/cassandra/net/MessagingService.java b/src/java/org/apache/cassandra/net/MessagingService.java
index 81853edfc9..c36fd71500 100644
--- a/src/java/org/apache/cassandra/net/MessagingService.java
+++ b/src/java/org/apache/cassandra/net/MessagingService.java
@@ -257,6 +257,17 @@ public class MessagingService extends MessagingServiceMBeanImpl implements Messa
return Collections.unmodifiableList(versions);
}
+
+ public List<Version> greaterThanOrEqual()
+ {
+ Version[] all = Version.values();
+ if (ordinal() == all.length - 1)
+ return Collections.singletonList(this);
+ List<Version> values = new ArrayList<>(all.length - ordinal());
+ for (int i = ordinal(); i < all.length; i++)
+ values.add(all[i]);
+ return values;
+ }
}
// Maintance Note:
// Try to keep Version enum in-sync for testing. By having the versions in the enum tests can get access without forcing this class
diff --git a/src/java/org/apache/cassandra/schema/SchemaKeyspace.java b/src/java/org/apache/cassandra/schema/SchemaKeyspace.java
index d4e4368acd..e293851014 100644
--- a/src/java/org/apache/cassandra/schema/SchemaKeyspace.java
+++ b/src/java/org/apache/cassandra/schema/SchemaKeyspace.java
@@ -27,7 +27,6 @@ import javax.annotation.concurrent.NotThreadSafe;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.collect.*;
-import com.google.common.collect.Maps;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
diff --git a/src/java/org/apache/cassandra/service/accord/AccordCallback.java b/src/java/org/apache/cassandra/service/accord/AccordCallback.java
index 955e034d11..6e658948a4 100644
--- a/src/java/org/apache/cassandra/service/accord/AccordCallback.java
+++ b/src/java/org/apache/cassandra/service/accord/AccordCallback.java
@@ -46,7 +46,7 @@ class AccordCallback<T extends Reply> extends SafeCallback<T> implements Request
@Override
public void onResponse(Message<T> msg)
{
- logger.debug("Received response {} from {}", msg.payload, msg.from());
+ logger.trace("Received response {} from {}", msg.payload, msg.from());
success(endpointMapper.mappedId(msg.from()), msg.payload);
}
@@ -60,7 +60,7 @@ class AccordCallback<T extends Reply> extends SafeCallback<T> implements Request
@Override
public void onFailure(InetAddressAndPort from, RequestFailure failure)
{
- logger.debug("Received failure {} from {} for {}", failure, from, this);
+ logger.trace("Received failure {} from {} for {}", failure, from, this);
// TODO (now): we should distinguish timeout failures with some placeholder Exception
failure(endpointMapper.mappedId(from), convertFailureMessage(failure));
}
diff --git a/src/java/org/apache/cassandra/service/accord/AccordCommandStore.java b/src/java/org/apache/cassandra/service/accord/AccordCommandStore.java
index 47417fef6e..929a1695d7 100644
--- a/src/java/org/apache/cassandra/service/accord/AccordCommandStore.java
+++ b/src/java/org/apache/cassandra/service/accord/AccordCommandStore.java
@@ -40,7 +40,7 @@ import org.slf4j.LoggerFactory;
import accord.api.Agent;
import accord.api.DataStore;
import accord.api.ProgressLog;
-import accord.local.CommandsForKey;
+import accord.local.cfk.CommandsForKey;
import accord.impl.TimestampsForKey;
import accord.local.Command;
import accord.local.CommandStore;
@@ -209,7 +209,7 @@ public class AccordCommandStore extends CommandStore implements CacheSize
commandCache =
stateCache.instance(TxnId.class,
AccordSafeCommand.class,
- AccordSafeCommand::new,
+ AccordSafeCommand.safeRefFactory(),
this::loadCommand,
this::appendToKeyspace,
this::validateCommand,
@@ -356,12 +356,18 @@ public class AccordCommandStore extends CommandStore implements CacheSize
boolean validateCommand(TxnId txnId, Command evicting)
{
+ if (!Invariants.isParanoid())
+ return true;
+
Command reloaded = loadCommand(txnId);
- return (evicting == null && reloaded == null) || (evicting != null && reloaded != null && reloaded.isEqualOrFuller(evicting));
+ return Objects.equals(evicting, reloaded);
}
boolean validateTimestampsForKey(RoutableKey key, TimestampsForKey evicting)
{
+ if (!Invariants.isParanoid())
+ return true;
+
TimestampsForKey reloaded = AccordKeyspace.unsafeLoadTimestampsForKey(this, (PartitionKey) key);
return Objects.equals(evicting, reloaded);
}
@@ -378,6 +384,9 @@ public class AccordCommandStore extends CommandStore implements CacheSize
boolean validateCommandsForKey(RoutableKey key, CommandsForKey evicting)
{
+ if (!Invariants.isParanoid())
+ return true;
+
CommandsForKey reloaded = AccordKeyspace.loadCommandsForKey(this, (PartitionKey) key);
return Objects.equals(evicting, reloaded);
}
@@ -494,7 +503,6 @@ public class AccordCommandStore extends CommandStore implements CacheSize
commandsForRanges.preExecute();
current = AccordSafeCommandStore.create(preLoadContext, commands, timestampsForKeys, commandsForKeys, commandsForRanges, this);
-
return current;
}
@@ -506,8 +514,14 @@ public class AccordCommandStore extends CommandStore implements CacheSize
public void completeOperation(AccordSafeCommandStore store)
{
Invariants.checkState(current == store);
- current.complete();
- current = null;
+ try
+ {
+ current.postExecute();
+ }
+ finally
+ {
+ current = null;
+ }
}
public void abortCurrentOperation()
diff --git a/src/java/org/apache/cassandra/service/accord/AccordKeyspace.java b/src/java/org/apache/cassandra/service/accord/AccordKeyspace.java
index 50d29c6e92..12b278c06d 100644
--- a/src/java/org/apache/cassandra/service/accord/AccordKeyspace.java
+++ b/src/java/org/apache/cassandra/service/accord/AccordKeyspace.java
@@ -48,7 +48,7 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import accord.api.Key;
-import accord.local.CommandsForKey;
+import accord.local.cfk.CommandsForKey;
import accord.impl.TimestampsForKey;
import accord.local.Command;
import accord.local.CommandStore;
@@ -550,9 +550,8 @@ public class AccordKeyspace
+ format("key %s, ", KEY_TUPLE)
+ "data blob, "
+ "PRIMARY KEY((store_id, key_token, key))"
- + ')')
- // TODO (expected): make this uncompressed, as not very compressable (except perhaps the primary key, but could switch to operating on tokens directly)
-// + " WITH compression = {'enabled':'false'};")
+ + ')'
+ + " WITH compression = {'class':'NoopCompressor'};")
.partitioner(FOR_KEYS_LOCAL_PARTITIONER)
.build();
}
@@ -624,7 +623,7 @@ public class AccordKeyspace
if (current == null)
return null;
- CommandsForKey updated = current.withRedundantBefore(redundantBefore);
+ CommandsForKey updated = current.withRedundantBeforeAtLeast(redundantBefore);
if (current == updated)
return row;
diff --git a/src/java/org/apache/cassandra/service/accord/AccordObjectSizes.java b/src/java/org/apache/cassandra/service/accord/AccordObjectSizes.java
index 160813d722..d3cd7a34fd 100644
--- a/src/java/org/apache/cassandra/service/accord/AccordObjectSizes.java
+++ b/src/java/org/apache/cassandra/service/accord/AccordObjectSizes.java
@@ -24,12 +24,12 @@ import java.util.function.ToLongFunction;
import accord.api.Key;
import accord.api.Result;
import accord.api.RoutingKey;
-import accord.local.CommandsForKey;
-import accord.local.CommandsForKey.TxnInfo;
+import accord.local.cfk.CommandsForKey;
+import accord.local.cfk.CommandsForKey.TxnInfo;
import accord.impl.TimestampsForKey;
import accord.local.Command;
import accord.local.Command.WaitingOn;
-import accord.local.CommandsForKey.TxnInfoWithMissing;
+import accord.local.cfk.CommandsForKey.TxnInfoExtra;
import accord.local.CommonAttributes;
import accord.local.Node;
import accord.local.SaveStatus;
@@ -69,6 +69,7 @@ import org.apache.cassandra.service.accord.txn.TxnResult;
import org.apache.cassandra.service.accord.txn.TxnWrite;
import org.apache.cassandra.utils.ObjectSizes;
+import static accord.local.cfk.CommandsForKey.NO_TXNIDS;
import static org.apache.cassandra.utils.ObjectSizes.measure;
public class AccordObjectSizes
@@ -205,7 +206,9 @@ public class AccordObjectSizes
return size;
}
- private static final long TIMESTAMP_SIZE = ObjectSizes.measureDeep(Timestamp.fromBits(0, 0, new Node.Id(0)));
+ // don't count Id size, as should normally be shared
+ private static final long TIMESTAMP_SIZE = ObjectSizes.measure(Timestamp.fromBits(0, 0, new Node.Id(0)));
+ private static final long BALLOT_SIZE = ObjectSizes.measure(Ballot.ZERO);
public static long timestamp()
{
@@ -217,6 +220,16 @@ public class AccordObjectSizes
return TIMESTAMP_SIZE;
}
+ public static long ballot()
+ {
+ return BALLOT_SIZE;
+ }
+
+ public static long ballot(Ballot ballot)
+ {
+ return ballot == Ballot.ZERO ? 0 : BALLOT_SIZE;
+ }
+
private static final long EMPTY_DEPS_SIZE = ObjectSizes.measureDeep(Deps.NONE);
public static long dependencies(Deps dependencies)
{
@@ -347,7 +360,7 @@ public class AccordObjectSizes
return size;
Command.Committed committed = command.asCommitted();
- size += WaitingOnSerializer.serializedSize(command.txnId(), committed.waitingOn);
+ size += WaitingOnSerializer.serializedSize(committed.waitingOn);
return size;
}
@@ -363,8 +376,8 @@ public class AccordObjectSizes
}
private static long EMPTY_CFK_SIZE = measure(new CommandsForKey(null));
- private static long EMPTY_INFO_SIZE = measure(TxnInfo.createMock(TxnId.NONE, null, null, null));
- private static long EMPTY_INFO_WITH_MISSING_ADDITIONAL_SIZE = measure(TxnInfo.createMock(TxnId.NONE, null, null, null)) - EMPTY_INFO_SIZE;
+ private static long EMPTY_INFO_SIZE = measure(TxnInfo.createMock(TxnId.NONE, null, null, NO_TXNIDS, Ballot.ZERO));
+ private static long EMPTY_INFO_EXTRA_ADDITIONAL_SIZE = EMPTY_INFO_SIZE - measure(TxnInfo.createMock(TxnId.NONE, null, null, null, null));
public static long commandsForKey(CommandsForKey cfk)
{
long size = EMPTY_CFK_SIZE;
@@ -374,13 +387,14 @@ public class AccordObjectSizes
for (int i = 0 ; i < cfk.size() ; ++i)
{
TxnInfo info = cfk.get(i);
- if (info.getClass() != TxnInfoWithMissing.class) continue;
- TxnInfoWithMissing infoWithMissing = (TxnInfoWithMissing) info;
- if (infoWithMissing.missing.length > 0)
+ if (info.getClass() != TxnInfoExtra.class) continue;
+ TxnInfoExtra infoExtra = (TxnInfoExtra) info;
+ if (infoExtra.missing.length > 0)
{
- size += EMPTY_INFO_WITH_MISSING_ADDITIONAL_SIZE;
- size += ObjectSizes.sizeOfReferenceArray(infoWithMissing.missing.length);
- size += infoWithMissing.missing.length * TIMESTAMP_SIZE;
+ size += EMPTY_INFO_EXTRA_ADDITIONAL_SIZE;
+ size += ObjectSizes.sizeOfReferenceArray(infoExtra.missing.length);
+ size += infoExtra.missing.length * TIMESTAMP_SIZE;
+ size += ballot(infoExtra.ballot);
}
}
return size;
diff --git a/src/java/org/apache/cassandra/service/accord/AccordSafeCommand.java b/src/java/org/apache/cassandra/service/accord/AccordSafeCommand.java
index 5c458aa45e..0cafdfec08 100644
--- a/src/java/org/apache/cassandra/service/accord/AccordSafeCommand.java
+++ b/src/java/org/apache/cassandra/service/accord/AccordSafeCommand.java
@@ -19,6 +19,7 @@
package org.apache.cassandra.service.accord;
import java.util.Objects;
+import java.util.function.Function;
import com.google.common.annotations.VisibleForTesting;
@@ -27,9 +28,37 @@ import accord.local.Command.TransientListener;
import accord.local.Listeners;
import accord.local.SafeCommand;
import accord.primitives.TxnId;
+import accord.utils.Invariants;
+import org.apache.cassandra.utils.concurrent.Ref;
+
+import static accord.utils.Invariants.Paranoia.LINEAR;
+import static accord.utils.Invariants.ParanoiaCostFactor.HIGH;
public class AccordSafeCommand extends SafeCommand implements AccordSafeState<TxnId, Command>
{
+ public static class DebugAccordSafeCommand extends AccordSafeCommand
+ {
+ final Ref<?> selfRef;
+ public DebugAccordSafeCommand(AccordCachingState<TxnId, Command> global)
+ {
+ super(global);
+ selfRef = new Ref<>(this, null);
+ selfRef.debug(global.key().toString());
+ }
+
+ @Override
+ public void invalidate()
+ {
+ super.invalidate();
+ selfRef.release();
+ }
+
+ public static void trace(AccordSafeCommand safeCommand, String message)
+ {
+ ((DebugAccordSafeCommand)safeCommand).selfRef.debug(message);
+ }
+ }
+
private boolean invalidated;
private final AccordCachingState<TxnId, Command> global;
private Command original;
@@ -111,13 +140,6 @@ public class AccordSafeCommand extends SafeCommand implements AccordSafeState<Tx
current = original;
}
- @Override
- public void postExecute()
- {
- checkNotInvalidated();
- global.set(current);
- }
-
@Override
public void invalidate()
{
@@ -150,4 +172,9 @@ public class AccordSafeCommand extends SafeCommand implements AccordSafeState<Tx
checkNotInvalidated();
return global.listeners();
}
+
+ public static Function<AccordCachingState<TxnId, Command>, AccordSafeCommand> safeRefFactory()
+ {
+ return Invariants.testParanoia(LINEAR, LINEAR, HIGH) ? DebugAccordSafeCommand::new : AccordSafeCommand::new;
+ }
}
diff --git a/src/java/org/apache/cassandra/service/accord/AccordSafeCommandStore.java b/src/java/org/apache/cassandra/service/accord/AccordSafeCommandStore.java
index f886971d08..ffdac078b0 100644
--- a/src/java/org/apache/cassandra/service/accord/AccordSafeCommandStore.java
+++ b/src/java/org/apache/cassandra/service/accord/AccordSafeCommandStore.java
@@ -28,7 +28,7 @@ import accord.api.DataStore;
import accord.api.Key;
import accord.api.ProgressLog;
import accord.impl.AbstractSafeCommandStore;
-import accord.local.CommandsForKey;
+import accord.local.cfk.CommandsForKey;
import accord.impl.CommandsSummary;
import accord.local.CommandStores.RangesForEpoch;
import accord.local.NodeTimeService;
@@ -298,27 +298,6 @@ public class AccordSafeCommandStore extends AbstractSafeCommandStore<AccordSafeC
}, accumulate);
}
- @Override
- protected void invalidateSafeState()
- {
- commands.values().forEach(AccordSafeCommand::invalidate);
- timestampsForKeys.values().forEach(AccordSafeTimestampsForKey::invalidate);
- commandsForKeys.values().forEach(AccordSafeCommandsForKey::invalidate);
- }
-
- public void postExecute(Map<TxnId, AccordSafeCommand> commands,
- Map<Key, AccordSafeTimestampsForKey> timestampsForKey,
- Map<Key, AccordSafeCommandsForKey> commandsForKeys,
- @Nullable AccordSafeCommandsForRanges commandsForRanges)
- {
- postExecute();
- commands.values().forEach(AccordSafeState::postExecute);
- timestampsForKey.values().forEach(AccordSafeState::postExecute);
- commandsForKeys.values().forEach(AccordSafeState::postExecute);
- if (commandsForRanges != null)
- commandsForRanges.postExecute();
- }
-
@Override
public String toString()
{
diff --git a/src/java/org/apache/cassandra/service/accord/AccordSafeCommandsForKey.java b/src/java/org/apache/cassandra/service/accord/AccordSafeCommandsForKey.java
index 808b4d4bc1..6f5e8f72d5 100644
--- a/src/java/org/apache/cassandra/service/accord/AccordSafeCommandsForKey.java
+++ b/src/java/org/apache/cassandra/service/accord/AccordSafeCommandsForKey.java
@@ -23,8 +23,8 @@ import java.util.Objects;
import com.google.common.annotations.VisibleForTesting;
import accord.api.Key;
-import accord.local.CommandsForKey;
-import accord.local.SafeCommandsForKey;
+import accord.local.cfk.CommandsForKey;
+import accord.local.cfk.SafeCommandsForKey;
public class AccordSafeCommandsForKey extends SafeCommandsForKey implements AccordSafeState<Key, CommandsForKey>
{
@@ -35,7 +35,7 @@ public class AccordSafeCommandsForKey extends SafeCommandsForKey implements Acco
public AccordSafeCommandsForKey(AccordCachingState<Key, CommandsForKey> global)
{
- super((Key) global.key());
+ super(global.key());
this.global = global;
this.original = null;
this.current = null;
@@ -117,14 +117,6 @@ public class AccordSafeCommandsForKey extends SafeCommandsForKey implements Acco
current = original;
}
- @Override
- public void postExecute()
- {
- checkNotInvalidated();
- if (current != original)
- global.set(current);
- }
-
@Override
public void invalidate()
{
diff --git a/src/java/org/apache/cassandra/service/accord/AccordSafeState.java b/src/java/org/apache/cassandra/service/accord/AccordSafeState.java
index 374968bcfb..d8e950d063 100644
--- a/src/java/org/apache/cassandra/service/accord/AccordSafeState.java
+++ b/src/java/org/apache/cassandra/service/accord/AccordSafeState.java
@@ -28,7 +28,7 @@ public interface AccordSafeState<K, V> extends SafeState<V>
void invalidate();
boolean invalidated();
void preExecute();
- void postExecute();
+
AccordCachingState<K, V> global();
default boolean hasUpdate()
diff --git a/src/java/org/apache/cassandra/service/accord/AccordSafeTimestampsForKey.java b/src/java/org/apache/cassandra/service/accord/AccordSafeTimestampsForKey.java
index a4c48c83e6..89baee84b9 100644
--- a/src/java/org/apache/cassandra/service/accord/AccordSafeTimestampsForKey.java
+++ b/src/java/org/apache/cassandra/service/accord/AccordSafeTimestampsForKey.java
@@ -105,13 +105,6 @@ public class AccordSafeTimestampsForKey extends SafeTimestampsForKey implements
current = original;
}
- @Override
- public void postExecute()
- {
- checkNotInvalidated();
- global.set(current);
- }
-
@Override
public void invalidate()
{
diff --git a/src/java/org/apache/cassandra/service/accord/AccordStateCache.java b/src/java/org/apache/cassandra/service/accord/AccordStateCache.java
index 085504f092..5fa5c6a941 100644
--- a/src/java/org/apache/cassandra/service/accord/AccordStateCache.java
+++ b/src/java/org/apache/cassandra/service/accord/AccordStateCache.java
@@ -219,6 +219,7 @@ public class AccordStateCache extends IntrusiveLinkedList<AccordCachingState<?,?
if (!node.hasListeners())
{
AccordCachingState<?, ?> self = instances.get(node.index).cache.remove(node.key());
+ Invariants.checkState(self.references == 0);
checkState(self == node, "Leaked node detected; was attempting to remove %s but cache had %s", node, self);
if (instance.listeners != null)
instance.listeners.forEach(l -> l.onEvict((AccordCachingState) node));
@@ -380,9 +381,8 @@ public class AccordStateCache extends IntrusiveLinkedList<AccordCachingState<?,?
@SuppressWarnings("unchecked")
AccordCachingState<K, V> node = (AccordCachingState<K, V>) cache.get(key);
if (node == null)
- {
return null;
- }
+
return safeRefFactory.apply(acquireExisting(node, false));
}
@@ -422,7 +422,8 @@ public class AccordStateCache extends IntrusiveLinkedList<AccordCachingState<?,?
node.load(loadExecutor, loadFunction);
node.references++;
- cache.put(key, node);
+ Object prev = cache.put(key, node);
+ Invariants.checkState(prev == null, "%s not absent from cache: %s already present", key, node);
if (listeners != null)
listeners.forEach(l -> l.onAdd(node));
maybeUpdateSize(node, heapEstimator);
@@ -467,13 +468,14 @@ public class AccordStateCache extends IntrusiveLinkedList<AccordCachingState<?,?
@SuppressWarnings("unchecked")
AccordCachingState<K, V> node = (AccordCachingState<K, V>) cache.get(key);
- checkState(node != null, "node is null for %s", key);
+ checkState(safeRef.global() != null, "safeRef node is null for %s", key);
+ checkState(safeRef.global() == node, "safeRef node not in map: %s != %s", safeRef.global(), node);
checkState(node.references > 0, "references (%d) are zero for %s (%s)", node.references, key, node);
- checkState(safeRef.global() == node);
checkState(!isInQueue(node));
if (safeRef.hasUpdate())
node.set(safeRef.current());
+ safeRef.invalidate();
maybeUpdateSize(node, heapEstimator);
@@ -688,6 +690,7 @@ public class AccordStateCache extends IntrusiveLinkedList<AccordCachingState<?,?
bytesCached = 0;
metrics.reset();;
instances.forEach(instance -> {
+ instance.cache.forEach((k, v) -> Invariants.checkState(v.references == 0));
instance.cache.clear();
instance.bytesCached = 0;
instance.instanceMetrics.reset();
diff --git a/src/java/org/apache/cassandra/service/accord/ImmutableAccordSafeState.java b/src/java/org/apache/cassandra/service/accord/ImmutableAccordSafeState.java
index 850f6f7e8d..8d9eb7c5ff 100644
--- a/src/java/org/apache/cassandra/service/accord/ImmutableAccordSafeState.java
+++ b/src/java/org/apache/cassandra/service/accord/ImmutableAccordSafeState.java
@@ -75,10 +75,4 @@ public abstract class ImmutableAccordSafeState<K, V> implements AccordSafeState<
{
checkNotInvalidated();
}
-
- @Override
- public void postExecute()
- {
- checkNotInvalidated();
- }
}
diff --git a/src/java/org/apache/cassandra/service/accord/SavedCommand.java b/src/java/org/apache/cassandra/service/accord/SavedCommand.java
index fb7034fcd1..282773c6d9 100644
--- a/src/java/org/apache/cassandra/service/accord/SavedCommand.java
+++ b/src/java/org/apache/cassandra/service/accord/SavedCommand.java
@@ -400,7 +400,7 @@ public class SavedCommand
if (diff.waitingOn != null)
{
size += Integer.BYTES;
- size += WaitingOnSerializer.serializedSize(diff.txnId, diff.waitingOn);
+ size += WaitingOnSerializer.serializedSize(diff.waitingOn);
}
if (diff.writes != null)
@@ -448,7 +448,7 @@ public class SavedCommand
if (diff.waitingOn != null)
{
- long size = WaitingOnSerializer.serializedSize(diff.txnId, diff.waitingOn);
+ long size = WaitingOnSerializer.serializedSize(diff.waitingOn);
ByteBuffer serialized = WaitingOnSerializer.serialize(diff.txnId, diff.waitingOn);
out.writeInt((int) size);
out.write(serialized);
@@ -556,7 +556,7 @@ public class SavedCommand
waitingOn = (localTxnId, deps) -> {
try
{
- return WaitingOnSerializer.deserialize(localTxnId, deps.keyDeps.keys(), deps.rangeDeps.txnIds(), buffer);
+ return WaitingOnSerializer.deserialize(localTxnId, deps.keyDeps.keys(), deps.rangeDeps, deps.directKeyDeps, buffer);
}
catch (IOException e)
{
diff --git a/src/java/org/apache/cassandra/service/accord/api/AccordAgent.java b/src/java/org/apache/cassandra/service/accord/api/AccordAgent.java
index c0fea38a37..899a00c4bf 100644
--- a/src/java/org/apache/cassandra/service/accord/api/AccordAgent.java
+++ b/src/java/org/apache/cassandra/service/accord/api/AccordAgent.java
@@ -119,6 +119,19 @@ public class AccordAgent implements Agent
return getReadRpcTimeout(MICROSECONDS);
}
+ // TODO (expected): we probably want additional configuration here so we can prune on shorter time horizons when we have a lot of transactions on a single key
+ @Override
+ public long cfkHlcPruneDelta()
+ {
+ return SECONDS.toMicros(10L);
+ }
+
+ @Override
+ public int cfkPruneInterval()
+ {
+ return 32;
+ }
+
@Override
public Txn emptyTxn(Kind kind, Seekables<?, ?> seekables)
{
diff --git a/src/java/org/apache/cassandra/service/accord/api/AccordRoutingKey.java b/src/java/org/apache/cassandra/service/accord/api/AccordRoutingKey.java
index e648e73800..3c8672ea93 100644
--- a/src/java/org/apache/cassandra/service/accord/api/AccordRoutingKey.java
+++ b/src/java/org/apache/cassandra/service/accord/api/AccordRoutingKey.java
@@ -43,7 +43,6 @@ import org.apache.cassandra.io.util.DataOutputBuffer;
import org.apache.cassandra.io.util.DataOutputPlus;
import org.apache.cassandra.schema.TableId;
import org.apache.cassandra.service.accord.TokenRange;
-import org.apache.cassandra.utils.ByteBufferUtil;
import org.apache.cassandra.utils.ObjectSizes;
import static org.apache.cassandra.config.DatabaseDescriptor.getPartitioner;
@@ -204,8 +203,7 @@ public abstract class AccordRoutingKey extends AccordRoutableKey implements Rout
static
{
- Token key = getPartitioner().decorateKey(ByteBufferUtil.EMPTY_BYTE_BUFFER).getToken();
- EMPTY_SIZE = ObjectSizes.measureDeep(new TokenKey(null, key));
+ EMPTY_SIZE = ObjectSizes.measure(new TokenKey(null, null));
}
final Token token;
diff --git a/src/java/org/apache/cassandra/service/accord/async/AsyncLoader.java b/src/java/org/apache/cassandra/service/accord/async/AsyncLoader.java
index 740cb3b5d1..87183603d6 100644
--- a/src/java/org/apache/cassandra/service/accord/async/AsyncLoader.java
+++ b/src/java/org/apache/cassandra/service/accord/async/AsyncLoader.java
@@ -18,7 +18,7 @@
package org.apache.cassandra.service.accord.async;
import accord.api.Key;
-import accord.local.CommandsForKey;
+import accord.local.cfk.CommandsForKey;
import accord.local.KeyHistory;
import accord.local.PreLoadContext;
import accord.primitives.*;
@@ -32,16 +32,20 @@ import com.google.common.collect.Iterables;
import org.apache.cassandra.service.accord.*;
import org.apache.cassandra.service.accord.api.AccordRoutingKey;
import org.apache.cassandra.service.accord.api.PartitionKey;
+import org.apache.cassandra.utils.NoSpamLogger;
+
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.*;
+import java.util.concurrent.TimeUnit;
import java.util.function.BiConsumer;
import java.util.stream.Collectors;
public class AsyncLoader
{
private static final Logger logger = LoggerFactory.getLogger(AsyncLoader.class);
+ private static final NoSpamLogger noSpamLogger = NoSpamLogger.getLogger(logger, 1L, TimeUnit.MINUTES);
enum State
{
@@ -83,7 +87,12 @@ public class AsyncLoader
List<AsyncChain<?>> listenChains)
{
S safeRef = cache.acquire(key);
- context.put(key, safeRef);
+ if (context.putIfAbsent(key, safeRef) != null)
+ {
+ noSpamLogger.warn("Context {} contained key {} more than once", context, key);
+ cache.release(safeRef);
+ return;
+ }
AccordCachingState.Status status = safeRef.globalStatus(); // globalStatus() completes
switch (status)
{
@@ -140,8 +149,7 @@ public class AsyncLoader
switch (keysOrRanges.domain())
{
case Key:
- // cast to Keys fails...
- Iterable<Key> keys = (Iterable<Key>) keysOrRanges;
+ AbstractKeys<Key> keys = (AbstractKeys<Key>) keysOrRanges;
keys.forEach(key -> referenceAndAssembleReadsForKey(key, context, chains));
break;
case Range:
@@ -233,9 +241,11 @@ public class AsyncLoader
{
case INITIALIZED:
state(State.SETUP);
+
case SETUP:
readResult = referenceAndDispatchReads(context);
state(State.LOADING);
+
case LOADING:
if (readResult != null)
{
@@ -252,13 +262,17 @@ public class AsyncLoader
}
}
state(State.FINISHED);
+
case FINISHED:
break;
+
default:
throw new IllegalStateException("Unexpected state: " + state);
}
- logger.trace("Exiting load for {} with state {}: {} {}", callback, state, txnIds, keysOrRanges);
+ if (logger.isTraceEnabled())
+ logger.trace("Exiting load for {} with state {}: {} {}", callback, state, txnIds, keysOrRanges);
+
return state == State.FINISHED;
}
}
diff --git a/src/java/org/apache/cassandra/service/accord/async/AsyncOperation.java b/src/java/org/apache/cassandra/service/accord/async/AsyncOperation.java
index 0e63881bf9..7a39b0393f 100644
--- a/src/java/org/apache/cassandra/service/accord/async/AsyncOperation.java
+++ b/src/java/org/apache/cassandra/service/accord/async/AsyncOperation.java
@@ -78,9 +78,13 @@ public abstract class AsyncOperation<R> extends AsyncChains.Head<R> implements R
void releaseResources(AccordCommandStore commandStore)
{
+ // TODO (expected): we should destructively iterate to avoid invoking second time in fail; or else read and set to null
commands.values().forEach(commandStore.commandCache()::release);
+ commands.clear();
timestampsForKey.values().forEach(commandStore.timestampsForKeyCache()::release);
+ timestampsForKey.clear();
commandsForKey.values().forEach(commandStore.commandsForKeyCache()::release);
+ commandsForKey.clear();
}
void revertChanges()
@@ -226,7 +230,8 @@ public abstract class AsyncOperation<R> extends AsyncChains.Head<R> implements R
finish(null, throwable);
}
- protected void runInternal()
+ // return true iff ready to run
+ protected boolean runInternal(boolean loadOnly)
{
switch (state)
{
@@ -235,8 +240,10 @@ public abstract class AsyncOperation<R> extends AsyncChains.Head<R> implements R
state(LOADING);
case LOADING:
if (!loader.load(context, this::onLoaded))
- return;
+ return false;
state(PREPARING);
+ if (loadOnly)
+ return true;
case PREPARING:
safeStore = commandStore.beginOperation(preLoadContext, context.commands, context.timestampsForKey, context.commandsForKey, context.commandsForRanges);
state(RUNNING);
@@ -262,14 +269,13 @@ public abstract class AsyncOperation<R> extends AsyncChains.Head<R> implements R
}
}
- safeStore.postExecute(context.commands, context.timestampsForKey, context.commandsForKey, context.commandsForRanges);
- context.releaseResources(commandStore);
commandStore.completeOperation(safeStore);
+ context.releaseResources(commandStore);
if (diffs != null)
{
state(COMPLETING);
this.commandStore.appendCommands(diffs, sanityCheck, () -> finish(result, null));
- return;
+ return false;
}
state(COMPLETING);
@@ -279,6 +285,8 @@ public abstract class AsyncOperation<R> extends AsyncChains.Head<R> implements R
case FAILED:
break;
}
+
+ return false;
}
@Override
@@ -292,7 +300,7 @@ public abstract class AsyncOperation<R> extends AsyncChains.Head<R> implements R
commandStore.setCurrentOperation(this);
try
{
- runInternal();
+ runInternal(false);
}
catch (Throwable t)
{
@@ -311,12 +319,28 @@ public abstract class AsyncOperation<R> extends AsyncChains.Head<R> implements R
}
}
+ private boolean preRun()
+ {
+ commandStore.checkInStoreThread();
+ try
+ {
+ return runInternal(true);
+ }
+ catch (Throwable t)
+ {
+ logger.error("Operation {} failed", this, t);
+ fail(t);
+ return false;
+ }
+ }
+
@Override
public void start(BiConsumer<? super R, Throwable> callback)
{
Invariants.checkState(this.callback == null);
this.callback = callback;
- commandStore.executor().execute(this);
+ if (!commandStore.inStore() || preRun())
+ commandStore.executor().execute(this);
}
static class ForFunction<R> extends AsyncOperation<R>
diff --git a/src/java/org/apache/cassandra/service/accord/serializers/CommandStoreSerializers.java b/src/java/org/apache/cassandra/service/accord/serializers/CommandStoreSerializers.java
index 1770ae75a5..5af4b53a36 100644
--- a/src/java/org/apache/cassandra/service/accord/serializers/CommandStoreSerializers.java
+++ b/src/java/org/apache/cassandra/service/accord/serializers/CommandStoreSerializers.java
@@ -129,7 +129,7 @@ public class CommandStoreSerializers
}
}), DurableBefore.Entry[]::new, DurableBefore.SerializerSupport::create);
- public static IVersionedSerializer<RedundantBefore> redundantBefore = new ReducingRangeMapSerializer<>(NullableSerializer.wrap(new IVersionedSerializer<RedundantBefore.Entry>()
+ public static final IVersionedSerializer<RedundantBefore.Entry> redundantBeforeEntry = new IVersionedSerializer<>()
{
@Override
public void serialize(RedundantBefore.Entry t, DataOutputPlus out, int version) throws IOException
@@ -152,10 +152,10 @@ public class CommandStoreSerializers
long startEpoch = in.readUnsignedVInt();
long endEpoch = in.readUnsignedVInt();
if (endEpoch == 0) endEpoch = Long.MAX_VALUE;
- else endEpoch = startEpoch + 1 + endEpoch;
- TxnId bootstrappedAt = CommandSerializers.txnId.deserialize(in, version);
+ else endEpoch = endEpoch - 1 + startEpoch;
TxnId locallyAppliedOrInvalidatedBefore = CommandSerializers.txnId.deserialize(in, version);
TxnId shardAppliedOrInvalidatedBefore = CommandSerializers.txnId.deserialize(in, version);
+ TxnId bootstrappedAt = CommandSerializers.txnId.deserialize(in, version);
Timestamp staleUntilAtLeast = CommandSerializers.nullableTimestamp.deserialize(in, version);
return new RedundantBefore.Entry(range, startEpoch, endEpoch, locallyAppliedOrInvalidatedBefore, shardAppliedOrInvalidatedBefore, bootstrappedAt, staleUntilAtLeast);
}
@@ -172,7 +172,8 @@ public class CommandStoreSerializers
size += CommandSerializers.nullableTimestamp.serializedSize(t.staleUntilAtLeast, version);
return size;
}
- }), RedundantBefore.Entry[]::new, RedundantBefore.SerializerSupport::create);
+ };
+ public static IVersionedSerializer<RedundantBefore> redundantBefore = new ReducingRangeMapSerializer<>(NullableSerializer.wrap(redundantBeforeEntry), RedundantBefore.Entry[]::new, RedundantBefore.SerializerSupport::create);
private static class TimestampToRangesSerializer<T extends Timestamp> implements IVersionedSerializer<NavigableMap<T, Ranges>>
{
diff --git a/src/java/org/apache/cassandra/service/accord/serializers/CommandsForKeySerializer.java b/src/java/org/apache/cassandra/service/accord/serializers/CommandsForKeySerializer.java
index a81b62b4a3..369fe57d8e 100644
--- a/src/java/org/apache/cassandra/service/accord/serializers/CommandsForKeySerializer.java
+++ b/src/java/org/apache/cassandra/service/accord/serializers/CommandsForKeySerializer.java
@@ -24,11 +24,13 @@ import java.util.Arrays;
import com.google.common.primitives.Ints;
import accord.api.Key;
-import accord.local.CommandsForKey;
-import accord.local.CommandsForKey.TxnInfo;
-import accord.local.CommandsForKey.InternalStatus;
-import accord.local.CommandsForKey.Unmanaged;
+import accord.local.cfk.CommandsForKey;
+import accord.local.cfk.CommandsForKey.TxnInfo;
+import accord.local.cfk.CommandsForKey.InternalStatus;
+import accord.local.cfk.CommandsForKey.TxnInfoExtra;
+import accord.local.cfk.CommandsForKey.Unmanaged;
import accord.local.Node;
+import accord.primitives.Ballot;
import accord.primitives.Routable.Domain;
import accord.primitives.Timestamp;
import accord.primitives.Txn;
@@ -38,9 +40,10 @@ import org.apache.cassandra.db.TypeSizes;
import org.apache.cassandra.db.marshal.ByteBufferAccessor;
import org.apache.cassandra.utils.vint.VIntCoding;
-import static accord.local.CommandsForKey.NO_PENDING_UNMANAGED;
-import static accord.primitives.Txn.Kind.ExclusiveSyncPoint;
+import static accord.local.cfk.CommandsForKey.NO_PENDING_UNMANAGED;
+import static accord.local.cfk.CommandsForKey.NO_TXNIDS;
import static accord.primitives.Txn.Kind.Read;
+import static accord.primitives.Txn.Kind.SyncPoint;
import static accord.primitives.Txn.Kind.Write;
import static accord.utils.ArrayBuffers.cachedInts;
import static accord.utils.ArrayBuffers.cachedTxnIds;
@@ -52,12 +55,15 @@ import static org.apache.cassandra.service.accord.serializers.CommandsForKeySeri
import static org.apache.cassandra.utils.ByteBufferUtil.readLeastSignificantBytes;
import static org.apache.cassandra.utils.ByteBufferUtil.writeLeastSignificantBytes;
import static org.apache.cassandra.utils.ByteBufferUtil.writeMostSignificantBytes;
+import static org.apache.cassandra.utils.vint.VIntCoding.decodeZigZag64;
+import static org.apache.cassandra.utils.vint.VIntCoding.encodeZigZag64;
public class CommandsForKeySerializer
{
private static final int HAS_MISSING_DEPS_HEADER_BIT = 0x1;
private static final int HAS_EXECUTE_AT_HEADER_BIT = 0x2;
- private static final int HAS_NON_STANDARD_FLAGS = 0x4;
+ private static final int HAS_BALLOT_HEADER_BIT = 0x4;
+ private static final int HAS_NON_STANDARD_FLAGS = 0x8;
/**
* We read/write a fixed number of intial bytes for each command, with an initial flexible number of flag bits
@@ -70,12 +76,13 @@ public class CommandsForKeySerializer
* two flag bytes:
* bit 0 is set if there are any missing ids;
* bit 1 is set if there are any executeAt specified
- * bit 2 is set if there are any queries present besides reads/writes
- * bits 3-4 number of header bytes to read for each command
- * bits 5-6: level 0 extra hlc bytes to read
- * bits 7-8: level 1 extra hlc bytes to read (+ 1 + level 0)
- * bits 9-10: level 2 extra hlc bytes to read (+ 1 + level 1)
- * bits 12-13: level 3 extra hlc bytes to read (+ 1 + level 2)
+ * bit 2 is set if there are any ballots specified
+ * bit 3 is set if there are any queries present besides reads/writes
+ * bits 4-5 number of header bytes to read for each command
+ * bits 8-9: level 0 extra hlc bytes to read
+ * bits 10-11: level 1 extra hlc bytes to read (+ 1 + level 0)
+ * bits 12-13: level 2 extra hlc bytes to read (+ 1 + level 1)
+ * bits 14-15: level 3 extra hlc bytes to read (+ 1 + level 2)
*
* In order, for each command, we consume:
* 3 bits for the InternalStatus of the command
@@ -104,6 +111,8 @@ public class CommandsForKeySerializer
// TODO (desired): determine timestamp resolution as a factor of 10
public static ByteBuffer toBytesWithoutKey(CommandsForKey cfk)
{
+ Invariants.checkArgument(!cfk.isLoadingPruned());
+
int commandCount = cfk.size();
if (commandCount == 0)
return ByteBuffer.allocate(1);
@@ -115,15 +124,15 @@ public class CommandsForKeySerializer
// whether we have any missing transactions to encode, any executeAt that are not equal to their TxnId
// and whether there are any non-standard flag bits to encode
boolean hasNonStandardFlags = false;
- int nodeIdCount = 0, missingIdCount = 0, executeAtCount = 0, bitsPerExecuteAtFlags = 0;
- int bitsPerExecuteAtEpochDelta = 0, bitsPerExecuteAtHlcDelta = 1; // to permit us to use full 64 bits and encode in 5 bits we force at least one hlc bit
+ int nodeIdCount = 0, missingIdCount = 0, executeAtCount = 0, ballotCount = 0;
+ int bitsPerExecuteAtEpoch = 0, bitsPerExecuteAtFlags = 0, bitsPerExecuteAtHlc = 1; // to permit us to use full 64 bits and encode in 5 bits we force at least one hlc bit
{
for (int i = 0 ; i < commandCount ; ++i)
{
- if (nodeIdCount + 1 >= nodeIds.length)
+ if (nodeIdCount + 2 >= nodeIds.length)
{
nodeIdCount = compact(nodeIds);
- if (nodeIdCount > nodeIds.length/2)
+ if (nodeIdCount > nodeIds.length/2 || nodeIdCount + 2 >= nodeIds.length)
nodeIds = cachedInts().resize(nodeIds, nodeIds.length, nodeIds.length * 2);
}
@@ -132,15 +141,27 @@ public class CommandsForKeySerializer
hasNonStandardFlags |= txnIdFlags(txn) != STANDARD;
nodeIds[nodeIdCount++] = txn.node.id;
- missingIdCount += txn.missing().length;
- if (txn.executeAt == txn)
- continue;
+ if (txn.executeAt != txn)
+ {
+ Invariants.checkState(txn.status.hasExecuteAtOrDeps);
+ nodeIds[nodeIdCount++] = txn.executeAt.node.id;
+ bitsPerExecuteAtEpoch = Math.max(bitsPerExecuteAtEpoch, numberOfBitsToRepresent(txn.executeAt.epoch() - txn.epoch()));
+ bitsPerExecuteAtHlc = Math.max(bitsPerExecuteAtHlc, numberOfBitsToRepresent(txn.executeAt.hlc() - txn.hlc()));
+ bitsPerExecuteAtFlags = Math.max(bitsPerExecuteAtFlags, numberOfBitsToRepresent(txn.executeAt.flags()));
+ executeAtCount += 1;
+ }
- nodeIds[nodeIdCount++] = txn.executeAt.node.id;
- bitsPerExecuteAtEpochDelta = Math.max(bitsPerExecuteAtEpochDelta, numberOfBitsToRepresent(txn.executeAt.epoch() - txn.epoch()));
- bitsPerExecuteAtHlcDelta = Math.max(bitsPerExecuteAtHlcDelta, numberOfBitsToRepresent(txn.executeAt.hlc() - txn.hlc()));
- bitsPerExecuteAtFlags = Math.max(bitsPerExecuteAtFlags, numberOfBitsToRepresent(txn.executeAt.flags()));
- executeAtCount += 1;
+ if (txn.getClass() == TxnInfoExtra.class)
+ {
+ TxnInfoExtra extra = (TxnInfoExtra) txn;
+ missingIdCount += extra.missing.length;
+ if (extra.ballot != Ballot.ZERO)
+ {
+ Invariants.checkArgument(txn.status.hasBallot);
+ nodeIds[nodeIdCount++] = extra.ballot.node.id;
+ ballotCount += 1;
+ }
+ }
}
nodeIdCount = compact(nodeIds);
Invariants.checkState(nodeIdCount > 0);
@@ -151,11 +172,14 @@ public class CommandsForKeySerializer
int bitsPerNodeId = numberOfBitsToRepresent(nodeIdCount);
int minHeaderBits = 7 + bitsPerNodeId + (hasNonStandardFlags ? 1 : 0);
int infoHeaderBits = (executeAtCount > 0 ? 1 : 0) + (missingIdCount > 0 ? 1 : 0);
+ int ballotHeaderBits = (ballotCount > 0 ? 1 : 0);
int maxHeaderBits = minHeaderBits;
int totalBytes = 0;
long prevEpoch = cfk.get(0).epoch();
long prevHlc = cfk.get(0).hlc();
+ int prunedBeforeIndex = cfk.prunedBefore().equals(TxnId.NONE) ? -1 : cfk.indexOf(cfk.prunedBefore());
+
int[] bytesHistogram = cachedInts().getInts(12);
Arrays.fill(bytesHistogram, 0);
for (int i = 0 ; i < commandCount ; ++i)
@@ -194,8 +218,10 @@ public class CommandsForKeySerializer
totalBytes += 2;
TxnInfo info = cfk.get(i);
- if (info.status.hasInfo)
+ if (info.status.hasExecuteAtOrDeps)
headerBits += infoHeaderBits;
+ if (info.status.hasBallot)
+ headerBits += ballotHeaderBits;
maxHeaderBits = Math.max(headerBits, maxHeaderBits);
int basicBytes = (headerBits + payloadBits + 7)/8;
bytesHistogram[basicBytes]++;
@@ -213,10 +239,11 @@ public class CommandsForKeySerializer
int flags = (missingIdCount > 0 ? HAS_MISSING_DEPS_HEADER_BIT : 0)
| (executeAtCount > 0 ? HAS_EXECUTE_AT_HEADER_BIT : 0)
+ | (ballotCount > 0 ? HAS_BALLOT_HEADER_BIT : 0)
| (hasNonStandardFlags ? HAS_NON_STANDARD_FLAGS : 0);
int headerBytes = (maxHeaderBits+7)/8;
- flags |= Invariants.checkArgument(headerBytes - 1, headerBytes <= 4) << 3;
+ flags |= Invariants.checkArgument(headerBytes - 1, headerBytes <= 4) << 4;
int hlcBytesLookup;
{ // 2bits per size, first value may be zero and remainder may be increments of 1-4;
@@ -234,7 +261,7 @@ public class CommandsForKeySerializer
while (l3-l2 > 4) ++l2;
while (l2-l1 > 4) ++l1;
hlcBytesLookup = setHlcBytes(l0, l1, l2, l3);
- flags |= (l0 | ((l1-(1+l0))<<2) | ((l2-(1+l1))<<4) | ((l3-(1+l2))<<6)) << 5;
+ flags |= (l0 | ((l1-(1+l0))<<2) | ((l2-(1+l1))<<4) | ((l3-(1+l2))<<6)) << 8;
}
int hlcFlagLookup = hlcBytesLookupToHlcFlagLookup(hlcBytesLookup);
@@ -252,21 +279,51 @@ public class CommandsForKeySerializer
prevEpoch = cfk.get(0).epoch();
prevHlc = cfk.get(0).hlc();
- // account for encoding redundantBefore
totalBytes += TypeSizes.sizeofUnsignedVInt(prevEpoch);
totalBytes += TypeSizes.sizeofUnsignedVInt(prevHlc);
+ totalBytes += TypeSizes.sizeofUnsignedVInt(prunedBeforeIndex + 1);
- if (missingIdCount + executeAtCount > 0)
+ int bitsPerBallotEpoch = 0, bitsPerBallotHlc = 1, bitsPerBallotFlags = 0;
+ if ((missingIdCount | executeAtCount | ballotCount) > 0)
{
+ if (ballotCount > 0)
+ {
+ Ballot prevBallot = null;
+ for (int i = 0 ; i < commandCount ; ++i)
+ {
+ TxnInfo txn = cfk.get(i);
+ if (txn.getClass() != TxnInfoExtra.class) continue;
+ if (!txn.status.hasBallot) continue;
+ TxnInfoExtra extra = (TxnInfoExtra) txn;
+ if (extra.ballot == Ballot.ZERO) continue;
+ if (prevBallot != null)
+ {
+ bitsPerBallotEpoch = Math.max(bitsPerBallotEpoch, numberOfBitsToRepresent(encodeZigZag64(extra.ballot.epoch() - prevBallot.epoch())));
+ bitsPerBallotHlc = Math.max(bitsPerBallotHlc, numberOfBitsToRepresent(encodeZigZag64(extra.ballot.hlc() - prevBallot.hlc())));
+ bitsPerBallotFlags = Math.max(bitsPerBallotFlags, numberOfBitsToRepresent(extra.ballot.flags()));
+ }
+ prevBallot = extra.ballot;
+ }
+ totalBytes += 2; // encode bit widths
+ }
+
+ if (executeAtCount > 0)
+ totalBytes += 2; // encode bit widths
+
// account for encoding missing id stream
int missingIdBits = 1 + numberOfBitsToRepresent(commandCount);
int executeAtBits = bitsPerNodeId
- + bitsPerExecuteAtEpochDelta
- + bitsPerExecuteAtHlcDelta
+ + bitsPerExecuteAtEpoch
+ + bitsPerExecuteAtHlc
+ bitsPerExecuteAtFlags;
- totalBytes += (missingIdBits * missingIdCount + executeAtBits * executeAtCount + 7)/8;
- if (executeAtCount > 0)
- totalBytes += 2;
+ int ballotBits = bitsPerNodeId
+ + bitsPerBallotEpoch
+ + bitsPerBallotHlc
+ + bitsPerBallotFlags;
+ totalBytes += (missingIdBits * missingIdCount
+ + executeAtBits * executeAtCount
+ + (ballotCount > 0 ? ballotBits * (ballotCount - 1) + bitsPerNodeId + 128 : 0)
+ + 7)/8;
}
// count unmanaged bytes
@@ -293,9 +350,11 @@ public class CommandsForKeySerializer
VIntCoding.writeUnsignedVInt(prevEpoch, out);
VIntCoding.writeUnsignedVInt(prevHlc, out);
+ VIntCoding.writeUnsignedVInt32(prunedBeforeIndex + 1, out);
int executeAtMask = executeAtCount > 0 ? 1 : 0;
int missingDepsMask = missingIdCount > 0 ? 1 : 0;
+ int ballotMask = ballotCount > 0 ? 1 : 0;
int flagsIncrement = hasNonStandardFlags ? 2 : 1;
// TODO (desired): check this loop compiles correctly to only branch on epoch case, for binarySearch and flushing
for (int i = 0 ; i < commandCount ; ++i)
@@ -307,15 +366,20 @@ public class CommandsForKeySerializer
long bits = status.ordinal();
int bitIndex = 3;
- int statusHasInfo = status.hasInfo ? 1 : 0;
- long hasExecuteAt = info.executeAt != null & info.executeAt != txnId ? 1 : 0;
+ int statusHasInfo = status.hasExecuteAtOrDeps ? 1 : 0;
+ int statusHasBallot = status.hasBallot ? 1 : 0;
+ long hasExecuteAt = info.executeAt != txnId ? 1 : 0;
bits |= hasExecuteAt << bitIndex;
bitIndex += statusHasInfo & executeAtMask;
- long hasMissingIds = info.missing() != CommandsForKey.NO_TXNIDS ? 1 : 0;
+ long hasMissingIds = info.getClass() == TxnInfoExtra.class && ((TxnInfoExtra)info).missing != CommandsForKey.NO_TXNIDS ? 1 : 0;
bits |= hasMissingIds << bitIndex;
bitIndex += statusHasInfo & missingDepsMask;
+ long hasBallot = info.getClass() == TxnInfoExtra.class && ((TxnInfoExtra)info).ballot != Ballot.ZERO ? 1 : 0;
+ bits |= hasBallot << bitIndex;
+ bitIndex += statusHasBallot & ballotMask;
+
long flagBits = txnIdFlagsBits(txnId);
boolean writeFullFlags = flagBits == RAW_BITS;
bits |= flagBits << bitIndex;
@@ -399,19 +463,23 @@ public class CommandsForKeySerializer
out.position(out.position() + offset);
}
- if ((executeAtCount | missingIdCount) > 0)
+ if ((executeAtCount | missingIdCount | ballotCount) > 0)
{
int bitsPerCommandId = numberOfBitsToRepresent(commandCount);
int bitsPerMissingId = 1 + bitsPerCommandId;
- int bitsPerExecuteAt = bitsPerExecuteAtEpochDelta + bitsPerExecuteAtHlcDelta + bitsPerExecuteAtFlags + bitsPerNodeId;
- Invariants.checkState(bitsPerExecuteAtEpochDelta < 64);
- Invariants.checkState(bitsPerExecuteAtHlcDelta <= 64);
+ int bitsPerExecuteAt = bitsPerExecuteAtEpoch + bitsPerExecuteAtHlc + bitsPerExecuteAtFlags + bitsPerNodeId;
+ int bitsPerBallot = bitsPerBallotEpoch + bitsPerBallotHlc + bitsPerBallotFlags + bitsPerNodeId;
+ Invariants.checkState(bitsPerExecuteAtEpoch < 64);
+ Invariants.checkState(bitsPerExecuteAtHlc <= 64);
Invariants.checkState(bitsPerExecuteAtFlags <= 16);
if (executeAtMask > 0) // we encode both 15 and 16 bits for flag length as 15 to fit in a short
- out.putShort((short) ((bitsPerExecuteAtEpochDelta << 10) | ((bitsPerExecuteAtHlcDelta-1) << 4) | (Math.min(15, bitsPerExecuteAtFlags))));
+ out.putShort((short) ((bitsPerExecuteAtEpoch << 10) | ((bitsPerExecuteAtHlc-1) << 4) | (Math.min(15, bitsPerExecuteAtFlags))));
+ if (ballotMask > 0) // we encode both 15 and 16 bits for flag length as 15 to fit in a short
+ out.putShort((short) ((bitsPerBallotEpoch << 10) | ((bitsPerBallotHlc-1) << 4) | (Math.min(15, bitsPerBallotFlags))));
long buffer = 0L;
int bufferCount = 0;
+ Ballot prevBallot = null;
for (int i = 0 ; i < commandCount ; ++i)
{
TxnInfo txn = cfk.get(i);
@@ -423,9 +491,9 @@ public class CommandsForKeySerializer
{
Invariants.checkState(executeAt.epoch() >= txn.epoch());
long executeAtBits = executeAt.epoch() - txn.epoch();
- int offset = bitsPerExecuteAtEpochDelta;
+ int offset = bitsPerExecuteAtEpoch;
executeAtBits |= (executeAt.hlc() - txn.hlc()) << offset ;
- offset += bitsPerExecuteAtHlcDelta;
+ offset += bitsPerExecuteAtHlc;
executeAtBits |= ((long)executeAt.flags()) << offset;
offset += bitsPerExecuteAtFlags;
executeAtBits |= ((long)nodeIdx) << offset;
@@ -434,10 +502,10 @@ public class CommandsForKeySerializer
}
else
{
- buffer = flushBits(buffer, bufferCount, executeAt.epoch() - txn.epoch(), bitsPerExecuteAtEpochDelta, out);
- bufferCount = (bufferCount + bitsPerExecuteAtEpochDelta) & 63;
- buffer = flushBits(buffer, bufferCount, executeAt.hlc() - txn.hlc(), bitsPerExecuteAtHlcDelta, out);
- bufferCount = (bufferCount + bitsPerExecuteAtHlcDelta) & 63;
+ buffer = flushBits(buffer, bufferCount, executeAt.epoch() - txn.epoch(), bitsPerExecuteAtEpoch, out);
+ bufferCount = (bufferCount + bitsPerExecuteAtEpoch) & 63;
+ buffer = flushBits(buffer, bufferCount, executeAt.hlc() - txn.hlc(), bitsPerExecuteAtHlc, out);
+ bufferCount = (bufferCount + bitsPerExecuteAtHlc) & 63;
buffer = flushBits(buffer, bufferCount, executeAt.flags(), bitsPerExecuteAtFlags, out);
bufferCount = (bufferCount + bitsPerExecuteAtFlags) & 63;
buffer = flushBits(buffer, bufferCount, nodeIdx, bitsPerNodeId, out);
@@ -445,20 +513,62 @@ public class CommandsForKeySerializer
}
}
- TxnId[] missing = txn.missing();
- if (missing.length > 0)
+ if (txn.getClass() == TxnInfoExtra.class)
{
- int j = 0;
- while (j < missing.length - 1)
+ TxnInfoExtra extra = (TxnInfoExtra) txn;
+
+ TxnId[] missing = extra.missing;
+ if (missing.length > 0)
{
- int missingId = cfk.indexOf(missing[j++]);
+ int j = 0;
+ while (j < missing.length - 1)
+ {
+ int missingId = cfk.indexOf(missing[j++]);
+ buffer = flushBits(buffer, bufferCount, missingId, bitsPerMissingId, out);
+ bufferCount = (bufferCount + bitsPerMissingId) & 63;
+ }
+ int missingId = cfk.indexOf(missing[missing.length - 1]);
+ missingId |= 1L << bitsPerCommandId;
buffer = flushBits(buffer, bufferCount, missingId, bitsPerMissingId, out);
bufferCount = (bufferCount + bitsPerMissingId) & 63;
}
- int missingId = cfk.indexOf(missing[missing.length - 1]);
- missingId |= 1L << bitsPerCommandId;
- buffer = flushBits(buffer, bufferCount, missingId, bitsPerMissingId, out);
- bufferCount = (bufferCount + bitsPerMissingId) & 63;
+
+ Ballot ballot = extra.ballot;
+ if (ballot != Ballot.ZERO)
+ {
+ int nodeIdx = Arrays.binarySearch(nodeIds, 0, nodeIdCount, ballot.node.id);
+ if (prevBallot == null)
+ {
+ buffer = flushBits(buffer, bufferCount, ballot.msb, 64, out);
+ buffer = flushBits(buffer, bufferCount, ballot.lsb, 64, out);
+ buffer = flushBits(buffer, bufferCount, nodeIdx, bitsPerNodeId, out);
+ bufferCount = (bufferCount + bitsPerNodeId) & 63;
+ }
+ else if (bitsPerBallot <= 64)
+ {
+ long ballotBits = encodeZigZag64(ballot.epoch() - prevBallot.epoch());
+ int offset = bitsPerBallotEpoch;
+ ballotBits |= encodeZigZag64(ballot.hlc() - prevBallot.hlc()) << offset ;
+ offset += bitsPerBallotHlc;
+ ballotBits |= ((long)ballot.flags()) << offset;
+ offset += bitsPerBallotFlags;
+ ballotBits |= ((long)nodeIdx) << offset;
+ buffer = flushBits(buffer, bufferCount, ballotBits, bitsPerBallot, out);
+ bufferCount = (bufferCount + bitsPerBallot) & 63;
+ }
+ else
+ {
+ buffer = flushBits(buffer, bufferCount, encodeZigZag64(ballot.epoch() - prevBallot.epoch()), bitsPerBallotEpoch, out);
+ bufferCount = (bufferCount + bitsPerBallotEpoch) & 63;
+ buffer = flushBits(buffer, bufferCount, encodeZigZag64(ballot.hlc() - prevBallot.hlc()), bitsPerBallotHlc, out);
+ bufferCount = (bufferCount + bitsPerBallotHlc) & 63;
+ buffer = flushBits(buffer, bufferCount, ballot.flags(), bitsPerBallotFlags, out);
+ bufferCount = (bufferCount + bitsPerBallotFlags) & 63;
+ buffer = flushBits(buffer, bufferCount, nodeIdx, bitsPerNodeId, out);
+ bufferCount = (bufferCount + bitsPerNodeId) & 63;
+ }
+ prevBallot = ballot;
+ }
}
}
@@ -502,6 +612,7 @@ public class CommandsForKeySerializer
return new CommandsForKey(key);
TxnId[] txnIds = cachedTxnIds().get(commandCount);
+ int[] decodeFlags = cachedInts().getInts(commandCount);
TxnInfo[] txns = new TxnInfo[commandCount];
int nodeIdCount = VIntCoding.readUnsignedVInt32(in);
int bitsPerNodeId = numberOfBitsToRepresent(nodeIdCount);
@@ -514,34 +625,43 @@ public class CommandsForKeySerializer
nodeIds[i] = new Node.Id(prev += VIntCoding.readUnsignedVInt32(in));
}
- int missingDepsMasks, executeAtMasks, txnIdFlagsMask;
+ int missingDepsMasks, executeAtMasks, ballotMasks, txnIdFlagsMask;
int headerByteCount, hlcBytesLookup;
{
int flags = in.getShort();
missingDepsMasks = 0 != (flags & HAS_MISSING_DEPS_HEADER_BIT) ? 1 : 0;
executeAtMasks = 0 != (flags & HAS_EXECUTE_AT_HEADER_BIT) ? 1 : 0;
+ ballotMasks = 0 != (flags & HAS_BALLOT_HEADER_BIT) ? 1 : 0;
txnIdFlagsMask = 0 != (flags & HAS_NON_STANDARD_FLAGS) ? 3 : 1;
- headerByteCount = 1 + ((flags >>> 3) & 0x3);
- hlcBytesLookup = setHlcByteDeltas((flags >>> 5) & 0x3, (flags >>> 7) & 0x3, (flags >>> 9) & 0x3, (flags >>> 11) & 0x3);
+ headerByteCount = 1 + ((flags >>> 4) & 0x3);
+ hlcBytesLookup = setHlcByteDeltas((flags >>> 8) & 0x3, (flags >>> 10) & 0x3, (flags >>> 12) & 0x3, (flags >>> 14) & 0x3);
}
long prevEpoch = VIntCoding.readUnsignedVInt(in);
long prevHlc = VIntCoding.readUnsignedVInt(in);
+ int prunedBeforeIndex = VIntCoding.readUnsignedVInt32(in) - 1;
+
for (int i = 0 ; i < commandCount ; ++i)
{
long header = readLeastSignificantBytes(headerByteCount, in);
header |= 1L << (8 * headerByteCount); // marker so we know where to shift-left most-significant bytes to
- InternalStatus status = InternalStatus.get((int) (header & 0x7));
+ int commandDecodeFlags = (int)(header & 0x7);
+ InternalStatus status = InternalStatus.get(commandDecodeFlags);
header >>>= 3;
+ commandDecodeFlags <<= 3;
- int executeAtInfoOffset, missingDepsInfoOffset;
{
- int infoMask = status.hasInfo ? 1 : 0;
- int executeAtMask = infoMask & executeAtMasks, missingDepsMask = infoMask & missingDepsMasks;
- executeAtInfoOffset = ((int)header & executeAtMask) << 1;
+ int infoMask = status.hasExecuteAtOrDeps ? 1 : 0;
+ int executeAtMask = infoMask & executeAtMasks;
+ int missingDepsMask = infoMask & missingDepsMasks;
+ commandDecodeFlags |= ((int)header & executeAtMask) << 1;
header >>>= executeAtMask;
- missingDepsInfoOffset = (int)header & missingDepsMask;
+ commandDecodeFlags |= (int)header & missingDepsMask;
header >>>= missingDepsMask;
+ int ballotMask = status.hasBallot ? ballotMasks : 0;
+ commandDecodeFlags |= ((int)header & ballotMask) << 2;
+ header >>>= ballotMask;
+ decodeFlags[i] = commandDecodeFlags;
}
Txn.Kind kind = TXN_ID_FLAG_BITS_KIND_LOOKUP[((int)header & txnIdFlagsMask)];
@@ -592,11 +712,8 @@ public class CommandsForKeySerializer
if (readEpochBytes > 0)
epoch += readEpochBytes == 1 ? (in.get() & 0xff) : in.getInt();
- TxnId txnId = kind != null ? new TxnId(epoch, hlc, kind, Domain.Key, node)
- : TxnId.fromValues(epoch, hlc, flags, node);
-
- txnIds[i] = txnId;
- txns[i] = DECODE_INFOS[(executeAtInfoOffset | missingDepsInfoOffset)*STATUS_COUNT + status.ordinal()];
+ txnIds[i] = kind != null ? new TxnId(epoch, hlc, kind, Domain.Key, node)
+ : TxnId.fromValues(epoch, hlc, flags, node);
prevEpoch = epoch;
prevHlc = hlc;
@@ -626,7 +743,7 @@ public class CommandsForKeySerializer
in.position(in.position() + offset);
}
- if (executeAtMasks + missingDepsMasks > 0)
+ if ((executeAtMasks | missingDepsMasks | ballotMasks) > 0)
{
TxnId[] missingIdBuffer = cachedTxnIds().get(8);
int missingIdCount = 0, maxIdBufferCount = 0;
@@ -634,25 +751,36 @@ public class CommandsForKeySerializer
int txnIdMask = (1 << bitsPerTxnId) - 1;
int bitsPerMissingId = bitsPerTxnId + 1;
- int decodeBits = executeAtMasks > 0 ? in.getShort() & 0xffff : 0;
- int bitsPerEpochDelta = decodeBits >>> 10;
- int bitsPerHlcDelta = 1 + ((decodeBits >>> 4) & 0x3f);
- int bitsPerFlags = decodeBits & 0xf;
- if (bitsPerFlags == 15) bitsPerFlags = 16;
- int bitsPerExecuteAt = bitsPerEpochDelta + bitsPerHlcDelta + bitsPerFlags + bitsPerNodeId;
-
- long epochDeltaMask = bitsPerEpochDelta == 0 ? 0 : (-1L >>> (64 - bitsPerEpochDelta));
- long hlcDeltaMask = (-1L >>> (64 - bitsPerHlcDelta));
- long flagsMask = bitsPerFlags == 0 ? 0 : (-1L >>> (64 - bitsPerFlags));
-
+ int decodeExecuteAtBits = executeAtMasks > 0 ? in.getShort() & 0xffff : 0;
+ int bitsPerExecuteAtEpoch = decodeExecuteAtBits >>> 10;
+ int bitsPerExecuteAtHlc = 1 + ((decodeExecuteAtBits >>> 4) & 0x3f);
+ int bitsPerExecuteAtFlags = decodeExecuteAtBits & 0xf;
+ if (bitsPerExecuteAtFlags == 15) bitsPerExecuteAtFlags = 16;
+ int bitsPerExecuteAt = bitsPerExecuteAtEpoch + bitsPerExecuteAtHlc + bitsPerExecuteAtFlags + bitsPerNodeId;
+
+ long executeAtEpochMask = bitsPerExecuteAtEpoch == 0 ? 0 : (-1L >>> (64 - bitsPerExecuteAtEpoch));
+ long executeAtHlcMask = (-1L >>> (64 - bitsPerExecuteAtHlc));
+ long executeAtFlagsMask = bitsPerExecuteAtFlags == 0 ? 0 : (-1L >>> (64 - bitsPerExecuteAtFlags));
+
+ int decodeBallotBits = ballotMasks > 0 ? in.getShort() & 0xffff : 0;
+ int bitsPerBallotEpoch = decodeBallotBits >>> 10;
+ int bitsPerBallotHlc = 1 + ((decodeBallotBits >>> 4) & 0x3f);
+ int bitsPerBallotFlags = decodeBallotBits & 0xf;
+ if (bitsPerBallotFlags == 15) bitsPerBallotFlags = 16;
+ int bitsPerBallot = bitsPerBallotEpoch + bitsPerBallotHlc + bitsPerBallotFlags + bitsPerNodeId;
+
+ long ballotEpochMask = bitsPerBallotEpoch == 0 ? 0 : (-1L >>> (64 - bitsPerBallotEpoch));
+ long ballotHlcMask = (-1L >>> (64 - bitsPerBallotHlc));
+ long ballotFlagsMask = bitsPerBallotFlags == 0 ? 0 : (-1L >>> (64 - bitsPerBallotFlags));
+
+ Ballot prevBallot = null;
final BitReader reader = new BitReader();
-
for (int i = 0 ; i < commandCount ; ++i)
{
TxnId txnId = txnIds[i];
- TxnInfo placeholder = txns[i];
- Timestamp executeAt;
- if (placeholder.executeAt == null)
+ int commandDecodeFlags = decodeFlags[i];
+ Timestamp executeAt = txnId;
+ if ((commandDecodeFlags & HAS_EXECUTE_AT_HEADER_BIT) != 0)
{
long epoch, hlc;
int flags;
@@ -660,30 +788,26 @@ public class CommandsForKeySerializer
if (bitsPerExecuteAt <= 64)
{
long executeAtBits = reader.read(bitsPerExecuteAt, in);
- epoch = txnId.epoch() + (executeAtBits & epochDeltaMask);
- executeAtBits >>>= bitsPerEpochDelta;
- hlc = txnId.hlc() + (executeAtBits & hlcDeltaMask);
- executeAtBits >>>= bitsPerHlcDelta;
- flags = (int)(executeAtBits & flagsMask);
- executeAtBits >>>= bitsPerFlags;
+ epoch = txnId.epoch() + (executeAtBits & executeAtEpochMask);
+ executeAtBits >>>= bitsPerExecuteAtEpoch;
+ hlc = txnId.hlc() + (executeAtBits & executeAtHlcMask);
+ executeAtBits >>>= bitsPerExecuteAtHlc;
+ flags = (int)(executeAtBits & executeAtFlagsMask);
+ executeAtBits >>>= bitsPerExecuteAtFlags;
id = nodeIds[(int)(executeAtBits & nodeIdMask)];
}
else
{
- epoch = txnId.epoch() + reader.read(bitsPerEpochDelta, in);
- hlc = txnId.hlc() + reader.read(bitsPerHlcDelta, in);
- flags = (int) reader.read(bitsPerFlags, in);
+ epoch = txnId.epoch() + reader.read(bitsPerExecuteAtEpoch, in);
+ hlc = txnId.hlc() + reader.read(bitsPerExecuteAtHlc, in);
+ flags = (int) reader.read(bitsPerExecuteAtFlags, in);
id = nodeIds[(int)(reader.read(bitsPerNodeId, in))];
}
executeAt = Timestamp.fromValues(epoch, hlc, flags, id);
}
- else
- {
- executeAt = txnId;
- }
- TxnId[] missing = placeholder.missing();
- if (missing == null)
+ TxnId[] missing = NO_TXNIDS;
+ if ((commandDecodeFlags & HAS_MISSING_DEPS_HEADER_BIT) != 0)
{
int prev = -1;
while (true)
@@ -704,7 +828,46 @@ public class CommandsForKeySerializer
missingIdCount = 0;
}
- txns[i] = TxnInfo.create(txnId, placeholder.status, executeAt, missing);
+ Ballot ballot = Ballot.ZERO;
+ if ((commandDecodeFlags & HAS_BALLOT_HEADER_BIT) != 0)
+ {
+ if (prevBallot == null)
+ {
+ long msb = reader.read(64, in);
+ long lsb = reader.read(64, in);
+ Node.Id id = nodeIds[(int)(reader.read(bitsPerNodeId, in))];
+ ballot = Ballot.fromBits(msb, lsb, id);
+ }
+ else
+ {
+ long epoch, hlc;
+ int flags;
+ Node.Id id;
+ if (bitsPerExecuteAt <= 64)
+ {
+ long ballotBits = reader.read(bitsPerBallot, in);
+ epoch = prevBallot.epoch() + decodeZigZag64(ballotBits & ballotEpochMask);
+ ballotBits >>>= bitsPerBallotEpoch;
+ hlc = prevBallot.hlc() + decodeZigZag64(ballotBits & ballotHlcMask);
+ ballotBits >>>= bitsPerBallotHlc;
+ flags = (int)(ballotBits & ballotFlagsMask);
+ ballotBits >>>= bitsPerBallotFlags;
+ id = nodeIds[(int)(ballotBits & nodeIdMask)];
+ }
+ else
+ {
+ epoch = prevBallot.epoch() + decodeZigZag64(reader.read(bitsPerBallotEpoch, in));
+ hlc = prevBallot.hlc() + decodeZigZag64(reader.read(bitsPerBallotHlc, in));
+ flags = (int) reader.read(bitsPerBallotFlags, in);
+ id = nodeIds[(int)(reader.read(bitsPerNodeId, in))];
+ }
+ ballot = Ballot.fromValues(epoch, hlc, flags, id);
+ }
+
+ prevBallot = ballot;
+ }
+
+ txns[i] = TxnInfo.create(txnId, InternalStatus.get(commandDecodeFlags >>> 3), executeAt, missing, ballot);
}
cachedTxnIds().forceDiscard(missingIdBuffer, maxIdBufferCount);
@@ -712,11 +875,11 @@ public class CommandsForKeySerializer
else
{
for (int i = 0 ; i < commandCount ; ++i)
- txns[i] = TxnInfo.create(txnIds[i], txns[i].status, txnIds[i]);
+ txns[i] = TxnInfo.create(txnIds[i], InternalStatus.get(decodeFlags[i] >>> 3), txnIds[i], Ballot.ZERO);
}
cachedTxnIds().forceDiscard(txnIds, commandCount);
- return CommandsForKey.SerializerSupport.create(key, txns, unmanageds);
+ return CommandsForKey.SerializerSupport.create(key, txns, unmanageds, prunedBeforeIndex == -1 ? TxnId.NONE : txns[prunedBeforeIndex]);
}
private static int getHlcBytes(int lookup, int index)
@@ -836,9 +999,9 @@ public class CommandsForKeySerializer
case Read:
case Write:
return STANDARD;
- case ExclusiveSyncPoint:
- return EXTENDED;
case SyncPoint:
+ return EXTENDED;
+ case ExclusiveSyncPoint:
case LocalOnly:
case EphemeralRead:
return RAW;
@@ -857,19 +1020,5 @@ public class CommandsForKeySerializer
}
}
- private static final Txn.Kind[] TXN_ID_FLAG_BITS_KIND_LOOKUP = new Txn.Kind[] { Read, Write, ExclusiveSyncPoint, null };
- private static final int STATUS_COUNT = InternalStatus.values().length;
- private static final TxnInfo[] DECODE_INFOS = new TxnInfo[4 * STATUS_COUNT];
- static
- {
- for (InternalStatus status : InternalStatus.values())
- {
- int ordinal = status.ordinal();
- DECODE_INFOS[ordinal] = TxnInfo.createMock(TxnId.NONE, status, TxnId.NONE, CommandsForKey.NO_TXNIDS);
- DECODE_INFOS[STATUS_COUNT+ordinal] = TxnInfo.createMock(TxnId.NONE, status, TxnId.NONE, null);
- DECODE_INFOS[2*STATUS_COUNT+ordinal] = TxnInfo.createMock(TxnId.NONE, status, null, CommandsForKey.NO_TXNIDS);
- DECODE_INFOS[3*STATUS_COUNT+ordinal] = TxnInfo.createMock(TxnId.NONE, status, null, null);
- }
- }
-
+ private static final Txn.Kind[] TXN_ID_FLAG_BITS_KIND_LOOKUP = new Txn.Kind[] { Read, Write, SyncPoint, null };
}
diff --git a/src/java/org/apache/cassandra/service/accord/serializers/DepsSerializer.java b/src/java/org/apache/cassandra/service/accord/serializers/DepsSerializer.java
index 0eba398ff9..dd78761cda 100644
--- a/src/java/org/apache/cassandra/service/accord/serializers/DepsSerializer.java
+++ b/src/java/org/apache/cassandra/service/accord/serializers/DepsSerializer.java
@@ -45,23 +45,23 @@ import static org.apache.cassandra.db.TypeSizes.sizeofUnsignedVInt;
public abstract class DepsSerializer<D extends Deps> extends IVersionedWithKeysSerializer.AbstractWithKeysSerializer implements IVersionedWithKeysSerializer<Seekables<?, ?>, D>
{
- public static final DepsSerializer<Deps> deps = new DepsSerializer<Deps>()
+ public static final DepsSerializer<Deps> deps = new DepsSerializer<>()
{
@Override
- Deps deserialize(KeyDeps keyDeps, RangeDeps rangeDeps, DataInputPlus in, int version)
+ Deps deserialize(KeyDeps keyDeps, RangeDeps rangeDeps, KeyDeps directKeyDeps, DataInputPlus in, int version)
{
- return new Deps(keyDeps, rangeDeps);
+ return new Deps(keyDeps, rangeDeps, directKeyDeps);
}
};
public static final IVersionedSerializer<Deps> nullableDeps = NullableSerializer.wrap(deps);
- public static final DepsSerializer<PartialDeps> partialDeps = new DepsSerializer<PartialDeps>()
+ public static final DepsSerializer<PartialDeps> partialDeps = new DepsSerializer<>()
{
@Override
- PartialDeps deserialize(KeyDeps keyDeps, RangeDeps rangeDeps, DataInputPlus in, int version) throws IOException
+ PartialDeps deserialize(KeyDeps keyDeps, RangeDeps rangeDeps, KeyDeps directKeyDeps, DataInputPlus in, int version) throws IOException
{
Ranges covering = KeySerializers.ranges.deserialize(in, version);
- return new PartialDeps(covering, keyDeps, rangeDeps);
+ return new PartialDeps(covering, keyDeps, rangeDeps, directKeyDeps);
}
@Override
@@ -72,9 +72,9 @@ public abstract class DepsSerializer<D extends Deps> extends IVersionedWithKeysS
}
@Override
- public void serialize(Seekables<?, ?> keys, PartialDeps partialDeps, DataOutputPlus out, int version) throws IOException
+ public void serialize(Seekables<?, ?> superset, PartialDeps partialDeps, DataOutputPlus out, int version) throws IOException
{
- super.serialize(keys, partialDeps, out, version);
+ super.serialize(superset, partialDeps, out, version);
KeySerializers.ranges.serialize(partialDeps.covering, out, version);
}
@@ -95,7 +95,7 @@ public abstract class DepsSerializer<D extends Deps> extends IVersionedWithKeysS
public static final IVersionedSerializer<PartialDeps> nullablePartialDeps = NullableSerializer.wrap(partialDeps);
- abstract D deserialize(KeyDeps keyDeps, RangeDeps rangeDeps, DataInputPlus in, int version) throws IOException;
+ abstract D deserialize(KeyDeps keyDeps, RangeDeps rangeDeps, KeyDeps directKeyDeps, DataInputPlus in, int version) throws IOException;
@Override
public void serialize(D deps, DataOutputPlus out, int version) throws IOException
@@ -105,9 +105,9 @@ public abstract class DepsSerializer<D extends Deps> extends IVersionedWithKeysS
}
@Override
- public void serialize(Seekables<?, ?> keys, D deps, DataOutputPlus out, int version) throws IOException
+ public void serialize(Seekables<?, ?> superset, D deps, DataOutputPlus out, int version) throws IOException
{
- if (keys.domain() == Key) serializeSubset(deps.keyDeps.keys(), keys, out);
+ if (superset.domain() == Key) serializeSubset(deps.keyDeps.keys(), superset, out);
else KeySerializers.keys.serialize(deps.keyDeps.keys(), out, version);
serializeWithoutKeys(deps, out, version);
}
@@ -148,21 +148,10 @@ public abstract class DepsSerializer<D extends Deps> extends IVersionedWithKeysS
private void serializeWithoutKeys(D deps, DataOutputPlus out, int version) throws IOException
{
- KeyDeps keyDeps = deps.keyDeps;
- {
- int txnIdCount = keyDeps.txnIdCount();
- out.writeUnsignedVInt32(txnIdCount);
- for (int i = 0; i < txnIdCount; i++)
- CommandSerializers.txnId.serialize(keyDeps.txnId(i), out, version);
+ serializeKeyDepsWithoutKeys(deps.keyDeps, out, version);
- int keysToTxnIdsCount = keysToTxnIdsCount(keyDeps);
- out.writeUnsignedVInt32(keysToTxnIdsCount);
- for (int i = 0; i < keysToTxnIdsCount; i++)
- out.writeUnsignedVInt32(keysToTxnIds(keyDeps, i));
- }
-
- RangeDeps rangeDeps = deps.rangeDeps;
{
+ RangeDeps rangeDeps = deps.rangeDeps;
int rangeCount = rangeDeps.rangeCount();
out.writeUnsignedVInt32(rangeCount);
for (int i = 0; i < rangeCount; i++)
@@ -178,24 +167,34 @@ public abstract class DepsSerializer<D extends Deps> extends IVersionedWithKeysS
for (int i = 0; i < rangesToTxnIdsCount; i++)
out.writeUnsignedVInt32(rangesToTxnIds(rangeDeps, i));
}
- }
- private D deserializeWithoutKeys(Keys keys, DataInputPlus in, int version) throws IOException
- {
- KeyDeps keyDeps;
{
- int txnIdCount = in.readUnsignedVInt32();
- TxnId[] txnIds = new TxnId[txnIdCount];
- for (int i = 0; i < txnIdCount; i++)
- txnIds[i] = CommandSerializers.txnId.deserialize(in, version);
-
- int keysToTxnIdsCount = in.readUnsignedVInt32();
- int[] keysToTxnIds = new int[keysToTxnIdsCount];
- for (int i = 0; i < keysToTxnIdsCount; i++)
- keysToTxnIds[i] = in.readUnsignedVInt32();
+ Keys keys = deps.directKeyDeps.keys();
+ boolean isSubset = isSubset(keys, deps.keyDeps.keys());
+ out.writeBoolean(isSubset);
+ if (isSubset) serializeSubset(keys, deps.keyDeps.keys(), out);
+ else KeySerializers.keys.serialize(keys, out, version);
- keyDeps = KeyDeps.SerializerSupport.create(keys, txnIds, keysToTxnIds);
+ serializeKeyDepsWithoutKeys(deps.directKeyDeps, out, version);
}
+ }
+
+ private void serializeKeyDepsWithoutKeys(KeyDeps keyDeps, DataOutputPlus out, int version) throws IOException
+ {
+ int txnIdCount = keyDeps.txnIdCount();
+ out.writeUnsignedVInt32(txnIdCount);
+ for (int i = 0; i < txnIdCount; i++)
+ CommandSerializers.txnId.serialize(keyDeps.txnId(i), out, version);
+
+ int keysToTxnIdsCount = keysToTxnIdsCount(keyDeps);
+ out.writeUnsignedVInt32(keysToTxnIdsCount);
+ for (int i = 0; i < keysToTxnIdsCount; i++)
+ out.writeUnsignedVInt32(keysToTxnIds(keyDeps, i));
+ }
+
+ private D deserializeWithoutKeys(Keys keys, DataInputPlus in, int version) throws IOException
+ {
+ KeyDeps keyDeps = deserializeKeyDeps(keys, in, version);
RangeDeps rangeDeps;
{
@@ -217,25 +216,34 @@ public abstract class DepsSerializer<D extends Deps> extends IVersionedWithKeysS
rangeDeps = RangeDeps.SerializerSupport.create(ranges, txnIds, rangesToTxnIds);
}
- return deserialize(keyDeps, rangeDeps, in, version);
+ KeyDeps directKeyDeps;
+ {
+ boolean isSubset = in.readBoolean();
+ Keys directKeys = isSubset ? (Keys)deserializeSubset(keys, in) : KeySerializers.keys.deserialize(in, version);
+ directKeyDeps = deserializeKeyDeps(directKeys, in, version);
+ }
+
+ return deserialize(keyDeps, rangeDeps, directKeyDeps, in, version);
}
- private long serializedSizeWithoutKeys(D deps, int version)
+ private static KeyDeps deserializeKeyDeps(Keys keys, DataInputPlus in, int version) throws IOException
{
- long size = 0L;
+ int txnIdCount = in.readUnsignedVInt32();
+ TxnId[] txnIds = new TxnId[txnIdCount];
+ for (int i = 0; i < txnIdCount; i++)
+ txnIds[i] = CommandSerializers.txnId.deserialize(in, version);
- KeyDeps keyDeps = deps.keyDeps;
- {
- int txnIdCount = keyDeps.txnIdCount();
- size += sizeofUnsignedVInt(txnIdCount);
- for (int i = 0; i < txnIdCount; i++)
- size += CommandSerializers.txnId.serializedSize(keyDeps.txnId(i), version);
+ int keysToTxnIdsCount = in.readUnsignedVInt32();
+ int[] keysToTxnIds = new int[keysToTxnIdsCount];
+ for (int i = 0; i < keysToTxnIdsCount; i++)
+ keysToTxnIds[i] = in.readUnsignedVInt32();
- int keysToTxnIdsCount = keysToTxnIdsCount(keyDeps);
- size += sizeofUnsignedVInt(keysToTxnIdsCount);
- for (int i = 0; i < keysToTxnIdsCount; i++)
- size += sizeofUnsignedVInt(keysToTxnIds(keyDeps, i));
- }
+ return KeyDeps.SerializerSupport.create(keys, txnIds, keysToTxnIds);
+ }
+
+ private long serializedSizeWithoutKeys(D deps, int version)
+ {
+ long size = serializedSizeOfKeyDepsWithoutKeys(deps.keyDeps, version);
RangeDeps rangeDeps = deps.rangeDeps;
{
@@ -255,7 +263,32 @@ public abstract class DepsSerializer<D extends Deps> extends IVersionedWithKeysS
size += sizeofUnsignedVInt(rangesToTxnIds(rangeDeps, i));
}
+ {
+ boolean isSubset = isSubset(deps.directKeyDeps.keys(), deps.keyDeps.keys());
+ size += 1;
+ size += isSubset ? serializedSubsetSize(deps.directKeyDeps.keys(), deps.keyDeps.keys()) : KeySerializers.keys.serializedSize(deps.directKeyDeps.keys(), version);
+ size += serializedSizeOfKeyDepsWithoutKeys(deps.directKeyDeps, version);
+ }
+ return size;
+ }
+
+ private static long serializedSizeOfKeyDepsWithoutKeys(KeyDeps keyDeps, int version)
+ {
+ int txnIdCount = keyDeps.txnIdCount();
+ long size = sizeofUnsignedVInt(txnIdCount);
+ for (int i = 0; i < txnIdCount; i++)
+ size += CommandSerializers.txnId.serializedSize(keyDeps.txnId(i), version);
+
+ int keysToTxnIdsCount = keysToTxnIdsCount(keyDeps);
+ size += sizeofUnsignedVInt(keysToTxnIdsCount);
+ for (int i = 0; i < keysToTxnIdsCount; i++)
+ size += sizeofUnsignedVInt(keysToTxnIds(keyDeps, i));
return size;
}
+ private static boolean isSubset(Keys test, Keys superset)
+ {
+ return test.foldl(superset, (k, p, v, i) -> v + 1, 0, 0, 0) == test.size();
+ }
+
}
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 6c22d28440..9bd2e0082a 100644
--- a/src/java/org/apache/cassandra/service/accord/serializers/WaitingOnSerializer.java
+++ b/src/java/org/apache/cassandra/service/accord/serializers/WaitingOnSerializer.java
@@ -21,78 +21,28 @@ package org.apache.cassandra.service.accord.serializers;
import java.io.IOException;
import java.nio.ByteBuffer;
-import accord.local.Command;
import accord.local.Command.WaitingOn;
+import accord.primitives.KeyDeps;
import accord.primitives.Keys;
+import accord.primitives.RangeDeps;
import accord.primitives.Routable;
-import accord.primitives.Timestamp;
import accord.primitives.TxnId;
import accord.utils.ImmutableBitSet;
import accord.utils.Invariants;
import accord.utils.SimpleBitSet;
-import accord.utils.SortedArrays.SortedArrayList;
import org.apache.cassandra.db.TypeSizes;
-import org.apache.cassandra.io.util.DataInputPlus;
-import org.apache.cassandra.io.util.DataOutputPlus;
+import org.apache.cassandra.utils.vint.VIntCoding;
public class WaitingOnSerializer
{
- public static void serialize(TxnId txnId, WaitingOn waitingOn, DataOutputPlus out) throws IOException
- {
- if (txnId.kind().awaitsOnlyDeps())
- {
- Timestamp executeAtLeast = waitingOn.executeAtLeast();
- out.writeBoolean(executeAtLeast != null);
- if (executeAtLeast != null)
- CommandSerializers.timestamp.serialize(executeAtLeast, out);
- }
- int keyCount = waitingOn.keys.size();
- int txnIdCount = waitingOn.txnIds.size();
- int waitingOnLength = (txnIdCount + keyCount + 63) / 64;
- serialize(waitingOnLength, waitingOn.waitingOn, out);
- if (txnId.domain() == Routable.Domain.Range)
- {
- int appliedOrInvalidatedLength = (txnIdCount + 63) / 64;
- serialize(appliedOrInvalidatedLength, waitingOn.appliedOrInvalidated, out);
- }
- }
-
- public static WaitingOn deserialize(TxnId txnId, Keys keys, SortedArrayList<TxnId> txnIds, DataInputPlus in) throws IOException
- {
- Timestamp executeAtLeast = null;
- if (txnId.kind().awaitsOnlyDeps())
- {
- if (in.readBoolean())
- executeAtLeast = CommandSerializers.timestamp.deserialize(in);
- }
- int waitingOnLength = (txnIds.size() + keys.size() + 63) / 64;
- ImmutableBitSet waitingOn = deserialize(waitingOnLength, in);
- ImmutableBitSet appliedOrInvalidated = null;
- if (txnId.domain() == Routable.Domain.Range)
- {
- int appliedOrInvalidatedLength = (txnIds.size() + 63) / 64;
- appliedOrInvalidated = deserialize(appliedOrInvalidatedLength, in);
- }
-
- WaitingOn result = new WaitingOn(keys, txnIds, waitingOn, appliedOrInvalidated);
- if (executeAtLeast != null)
- result = new Command.WaitingOnWithExecuteAt(result, executeAtLeast);
- return result;
- }
-
- public static long serializedSize(TxnId txnId, WaitingOn waitingOn)
+ public static long serializedSize(WaitingOn waitingOn)
{
int keyCount = waitingOn.keys.size();
- int txnIdCount = waitingOn.txnIds.size();
+ int txnIdCount = waitingOn.txnIdCount();
int waitingOnLength = (txnIdCount + keyCount + 63) / 64;
long size = serializedSize(waitingOnLength, waitingOn.waitingOn);
- if (txnId.kind().awaitsOnlyDeps())
- {
- Timestamp executeAtLeast = waitingOn.executeAtLeast();
- size += 1;
- if (executeAtLeast != null)
- size += CommandSerializers.timestamp.serializedSize();
- }
+ size += TypeSizes.sizeofUnsignedVInt(keyCount);
+ size += TypeSizes.sizeofUnsignedVInt(txnIdCount);
if (waitingOn.appliedOrInvalidated == null)
return size;
@@ -100,56 +50,26 @@ public class WaitingOnSerializer
return size + serializedSize(appliedOrInvalidatedLength, waitingOn.appliedOrInvalidated);
}
- private static void serialize(int length, SimpleBitSet write, DataOutputPlus out) throws IOException
- {
- long[] bits = SimpleBitSet.SerializationSupport.getArray(write);
- Invariants.checkState(length == bits.length);
- for (long v : bits)
- out.writeLong(v);
- }
-
- private static ImmutableBitSet deserialize(int length, DataInputPlus in) throws IOException
- {
- long[] bits = new long[length];
- for (int i = 0 ; i < length ; ++i)
- bits[i] = in.readLong();
- return ImmutableBitSet.SerializationSupport.construct(bits);
- }
-
public static long serializedSize(int length, SimpleBitSet write)
{
long[] bits = SimpleBitSet.SerializationSupport.getArray(write);
- Invariants.checkState(length == bits.length);
+ Invariants.checkState(length == bits.length, "Expected length %d != %d", length, bits.length);
return (long) TypeSizes.LONG_SIZE * length;
}
public static ByteBuffer serialize(TxnId txnId, WaitingOn waitingOn) throws IOException
{
int keyCount = waitingOn.keys.size();
- int txnIdCount = waitingOn.txnIds.size();
+ int txnIdCount = waitingOn.txnIdCount();
int waitingOnLength = (txnIdCount + keyCount + 63) / 64;
int appliedOrInvalidatedLength = 0;
if (txnId.domain() == Routable.Domain.Range)
appliedOrInvalidatedLength = (txnIdCount + 63) / 64;
- int size = TypeSizes.LONG_SIZE * (waitingOnLength + appliedOrInvalidatedLength);
- Timestamp executeAtLeast = null;
- if (txnId.kind().awaitsOnlyDeps())
- {
- executeAtLeast = waitingOn.executeAtLeast();
- size += 1;
- if (executeAtLeast != null)
- size += CommandSerializers.timestamp.serializedSize();
- }
-
- ByteBuffer out = ByteBuffer.allocate(size);
- if (txnId.kind().awaitsOnlyDeps())
- {
- out.put((byte)(executeAtLeast != null ? 1 : 0));
- if (executeAtLeast != null)
- CommandSerializers.timestamp.serialize(executeAtLeast, out);
- }
-
+ ByteBuffer out = ByteBuffer.allocate(TypeSizes.sizeofUnsignedVInt(keyCount) + TypeSizes.sizeofUnsignedVInt(txnIdCount)
+ + TypeSizes.LONG_SIZE * (waitingOnLength + appliedOrInvalidatedLength));
+ VIntCoding.writeUnsignedVInt32(keyCount, out);
+ VIntCoding.writeUnsignedVInt32(txnIdCount, out);
serialize(waitingOnLength, waitingOn.waitingOn, out);
if (appliedOrInvalidatedLength > 0)
serialize(appliedOrInvalidatedLength, waitingOn.appliedOrInvalidated, out);
@@ -164,33 +84,24 @@ public class WaitingOnSerializer
out.putLong(bits[i]);
}
- public static WaitingOn deserialize(TxnId txnId, Keys keys, SortedArrayList<TxnId> txnIds, ByteBuffer in) throws IOException
+ public static WaitingOn deserialize(TxnId txnId, Keys keys, RangeDeps directRangeDeps, KeyDeps directKeyDeps, ByteBuffer in) throws IOException
{
+ int txnIdCount = directRangeDeps.txnIdCount() + directKeyDeps.txnIdCount();
+ int waitingOnLength = (txnIdCount + keys.size() + 63) / 64;
int position = in.position();
- Timestamp executeAtLeast = null;
- if (txnId.kind().awaitsOnlyDeps())
- {
- if (in.get(position++) != 0)
- {
- executeAtLeast = CommandSerializers.timestamp.deserialize(in, position);
- position += CommandSerializers.timestamp.serializedSize();
- }
- }
-
- int waitingOnLength = (txnIds.size() + keys.size() + 63) / 64;
+ int a = VIntCoding.readUnsignedVInt32(in, position);
+ position += TypeSizes.sizeofUnsignedVInt(a);
+ int b = VIntCoding.readUnsignedVInt32(in, position);
+ position += TypeSizes.sizeofUnsignedVInt(a);
ImmutableBitSet waitingOn = deserialize(position, waitingOnLength, in);
ImmutableBitSet appliedOrInvalidated = null;
if (txnId.domain() == Routable.Domain.Range)
{
position += waitingOnLength*8;
- int appliedOrInvalidatedLength = (txnIds.size() + 63) / 64;
+ int appliedOrInvalidatedLength = (txnIdCount + 63) / 64;
appliedOrInvalidated = deserialize(position, appliedOrInvalidatedLength, in);
}
-
- WaitingOn result = new WaitingOn(keys, txnIds, waitingOn, appliedOrInvalidated);
- if (executeAtLeast != null)
- result = new Command.WaitingOnWithExecuteAt(result, executeAtLeast);
- return result;
+ return new WaitingOn(keys, directRangeDeps, directKeyDeps, waitingOn, appliedOrInvalidated);
}
private static ImmutableBitSet deserialize(int position, int length, ByteBuffer in)
diff --git a/src/java/org/apache/cassandra/service/accord/txn/TxnRead.java b/src/java/org/apache/cassandra/service/accord/txn/TxnRead.java
index 122336ad12..62dfe68e2c 100644
--- a/src/java/org/apache/cassandra/service/accord/txn/TxnRead.java
+++ b/src/java/org/apache/cassandra/service/accord/txn/TxnRead.java
@@ -35,7 +35,6 @@ import accord.primitives.Keys;
import accord.primitives.Ranges;
import accord.primitives.Seekable;
import accord.primitives.Timestamp;
-import accord.utils.SortedArrays;
import org.apache.cassandra.db.SinglePartitionReadCommand;
import accord.utils.async.AsyncChain;
import accord.utils.async.AsyncChains;
@@ -49,7 +48,6 @@ import org.apache.cassandra.service.accord.txn.TxnDataName.Kind;
import org.apache.cassandra.utils.ObjectSizes;
import org.apache.cassandra.utils.Simulate;
-import static accord.utils.SortedArrays.Search.CEIL;
import static com.google.common.base.Preconditions.checkArgument;
import static org.apache.cassandra.service.accord.AccordSerializers.consistencyLevelSerializer;
import static org.apache.cassandra.service.accord.IAccordService.SUPPORTED_READ_CONSISTENCY_LEVELS;
@@ -183,22 +181,6 @@ public class TxnRead extends AbstractKeySorted<TxnNamedRead> implements Read
return createTxnRead(reads, txnKeys.with((Keys)read.keys()), cassandraConsistencyLevel);
}
- @Override
- public boolean isEqualOrFuller(Read other)
- {
- TxnRead that = (TxnRead) other;
-
- int j = 0;
- for (int i = 0; i < that.items.length; ++i)
- {
- j = SortedArrays.exponentialSearch(this.items, j, this.items.length, that.items[i], this::compare, CEIL);
- if (j < 0 || !that.items[i].equals(this.items[j]))
- return false;
- }
-
- return this.txnKeys.containsAll(that.txnKeys);
- }
-
@Override
public AsyncChain<Data> read(Seekable key, SafeCommandStore safeStore, Timestamp executeAt, DataStore store)
{
diff --git a/src/java/org/apache/cassandra/service/accord/txn/TxnUpdate.java b/src/java/org/apache/cassandra/service/accord/txn/TxnUpdate.java
index cd2aa8a332..5a32065e97 100644
--- a/src/java/org/apache/cassandra/service/accord/txn/TxnUpdate.java
+++ b/src/java/org/apache/cassandra/service/accord/txn/TxnUpdate.java
@@ -125,22 +125,6 @@ public class TxnUpdate extends AccordUpdate
return Arrays.equals(fragments, txnUpdate.fragments) && Objects.equals(condition, txnUpdate.condition);
}
- @Override
- public boolean isEqualOrFuller(Update other)
- {
- TxnUpdate that = (TxnUpdate) other;
-
- int j = 0;
- for (int i = 0; i < that.keys.size(); ++i)
- {
- j = this.keys.findNext(j, that.keys.get(i), CEIL);
- if (j < 0 || !that.fragments[i].equals(this.fragments[j]))
- return false;
- }
-
- return this.condition.equals(that.condition);
- }
-
@Override
public int hashCode()
{
diff --git a/src/java/org/apache/cassandra/tcm/Transformation.java b/src/java/org/apache/cassandra/tcm/Transformation.java
index 12ba31f0c1..ea160482f3 100644
--- a/src/java/org/apache/cassandra/tcm/Transformation.java
+++ b/src/java/org/apache/cassandra/tcm/Transformation.java
@@ -39,18 +39,6 @@ import org.apache.cassandra.tcm.serialization.AsymmetricMetadataSerializer;
import org.apache.cassandra.tcm.serialization.VerboseMetadataSerializer;
import org.apache.cassandra.tcm.serialization.Version;
import org.apache.cassandra.tcm.transformations.*;
-import org.apache.cassandra.tcm.transformations.AlterSchema;
-import org.apache.cassandra.tcm.transformations.Assassinate;
-import org.apache.cassandra.tcm.transformations.BeginConsensusMigrationForTableAndRange;
-import org.apache.cassandra.tcm.transformations.CancelInProgressSequence;
-import org.apache.cassandra.tcm.transformations.CustomTransformation;
-import org.apache.cassandra.tcm.transformations.ForceSnapshot;
-import org.apache.cassandra.tcm.transformations.MaybeFinishConsensusMigrationForTableAndRange;
-import org.apache.cassandra.tcm.transformations.PrepareJoin;
-import org.apache.cassandra.tcm.transformations.PrepareLeave;
-import org.apache.cassandra.tcm.transformations.PrepareMove;
-import org.apache.cassandra.tcm.transformations.PrepareReplace;
-import org.apache.cassandra.tcm.transformations.Register;
import org.apache.cassandra.tcm.transformations.Startup;
import org.apache.cassandra.tcm.transformations.cms.AdvanceCMSReconfiguration;
import org.apache.cassandra.tcm.transformations.cms.FinishAddToCMS;
diff --git a/src/java/org/apache/cassandra/utils/MerkleTree.java b/src/java/org/apache/cassandra/utils/MerkleTree.java
index 2646057c04..cadc47ed27 100644
--- a/src/java/org/apache/cassandra/utils/MerkleTree.java
+++ b/src/java/org/apache/cassandra/utils/MerkleTree.java
@@ -749,7 +749,7 @@ public class MerkleTree
int size = offHeapBufferSize(innerNodeCount, partitioner);
logger.debug("Allocating direct buffer of size {} for an off-heap merkle tree", size);
ByteBuffer buffer = ByteBuffer.allocateDirect(size);
- if (Ref.DEBUG_ENABLED)
+ if (Ref.TRACE_ENABLED)
MemoryUtil.setAttachment(buffer, new Ref.DirectBufferRef<>(null, null));
return buffer;
}
diff --git a/src/java/org/apache/cassandra/utils/concurrent/Ref.java b/src/java/org/apache/cassandra/utils/concurrent/Ref.java
index e268f5fd73..911c1db881 100644
--- a/src/java/org/apache/cassandra/utils/concurrent/Ref.java
+++ b/src/java/org/apache/cassandra/utils/concurrent/Ref.java
@@ -61,6 +61,7 @@ import static java.util.Collections.emptyList;
import static org.apache.cassandra.concurrent.ExecutorFactory.Global.executorFactory;
import static org.apache.cassandra.concurrent.InfiniteLoopExecutor.SimulatorSafe.UNSAFE;
import static org.apache.cassandra.config.CassandraRelevantProperties.TEST_DEBUG_REF_COUNT;
+import static org.apache.cassandra.config.CassandraRelevantProperties.TEST_DEBUG_REF_EVENTS;
import static org.apache.cassandra.utils.Shared.Scope.SIMULATION;
import static org.apache.cassandra.utils.Throwables.maybeFail;
import static org.apache.cassandra.utils.Throwables.merge;
@@ -99,7 +100,8 @@ import static org.apache.cassandra.utils.Throwables.merge;
public final class Ref<T> implements RefCounted<T>
{
static final Logger logger = LoggerFactory.getLogger(Ref.class);
- public static final boolean DEBUG_ENABLED = TEST_DEBUG_REF_COUNT.getBoolean();
+ public static final boolean TRACE_ENABLED = TEST_DEBUG_REF_COUNT.getBoolean();
+ public static final boolean DEBUG_EVENTS_ENABLED = TEST_DEBUG_REF_EVENTS.getBoolean();
static OnLeak ON_LEAK;
@Shared(scope = SIMULATION)
@@ -170,10 +172,10 @@ public final class Ref<T> implements RefCounted<T>
public String printDebugInfo()
{
- if (DEBUG_ENABLED)
+ if (TRACE_ENABLED)
{
- state.debug.log(state.toString());
- return "Memory was freed by " + state.debug.deallocateThread;
+ ((Debug)state.debug).log(state.toString());
+ return "Memory was freed by " + ((Debug)state.debug).deallocateThread;
}
return "Memory was freed";
}
@@ -191,7 +193,7 @@ public final class Ref<T> implements RefCounted<T>
// ensures it is only released once, and that it is always released
static final class State extends PhantomReference<Ref>
{
- final Debug debug = DEBUG_ENABLED ? new Debug() : null;
+ final Object debug = TRACE_ENABLED ? new Debug() : DEBUG_EVENTS_ENABLED ? new ArrayList<>() : null;
final GlobalState globalState;
private volatile int released;
@@ -206,8 +208,8 @@ public final class Ref<T> implements RefCounted<T>
void assertNotReleased()
{
- if (DEBUG_ENABLED && released == 1)
- debug.log(toString());
+ if (TRACE_ENABLED && released == 1)
+ ((Debug)debug).log(toString());
assert released == 0;
}
@@ -216,8 +218,8 @@ public final class Ref<T> implements RefCounted<T>
if (releasedUpdater.getAndSet(this, 1) == 0)
{
accumulate = globalState.release(this, accumulate);
- if (DEBUG_ENABLED)
- debug.deallocate();
+ if (TRACE_ENABLED)
+ ((Debug)debug).deallocate();
}
return accumulate;
}
@@ -230,8 +232,8 @@ public final class Ref<T> implements RefCounted<T>
{
String id = this.toString();
logger.error("BAD RELEASE: attempted to release a reference ({}) that has already been released", id);
- if (DEBUG_ENABLED)
- debug.log(id);
+ if (TRACE_ENABLED)
+ ((Debug)debug).log(id);
throw new IllegalStateException("Attempted to release a reference that has already been released");
}
return;
@@ -240,16 +242,16 @@ public final class Ref<T> implements RefCounted<T>
if (leak)
{
String id = this.toString();
- logger.error("LEAK DETECTED: a reference ({}) to {} was not released before the reference was garbage collected", id, globalState);
- if (DEBUG_ENABLED)
- debug.log(id);
+ logger.error("LEAK DETECTED: a reference ({}) to {} was not released before the reference was garbage collected{}", id, globalState, (DEBUG_EVENTS_ENABLED ? "(debug: " + debug + ')' : ""));
+ if (TRACE_ENABLED)
+ ((Debug)debug).log(id);
OnLeak onLeak = ON_LEAK;
if (onLeak != null)
onLeak.onLeak(this);
}
- else if (DEBUG_ENABLED)
+ else if (TRACE_ENABLED)
{
- debug.deallocate();
+ ((Debug)debug).deallocate();
}
if (fail != null)
logger.error("Error when closing {}", globalState, fail);
@@ -299,6 +301,12 @@ public final class Ref<T> implements RefCounted<T>
}
}
+ public void debug(String event)
+ {
+ if (DEBUG_EVENTS_ENABLED)
+ ((List<String>)state.debug).add(event);
+ }
+
// the object that manages the actual cleaning up; this does not reference the target object
// so that we can detect when references are lost to the resource itself, and still cleanup afterwards
// the Tidy object MUST NOT contain any references to the object we are managing
@@ -383,10 +391,10 @@ public final class Ref<T> implements RefCounted<T>
private static final Set<GlobalState> globallyExtant = Collections.newSetFromMap(new ConcurrentHashMap<>());
static final ReferenceQueue<Object> referenceQueue = new ReferenceQueue<>();
private static final Shutdownable EXEC = executorFactory().infiniteLoop("Reference-Reaper", Ref::reapOneReference, UNSAFE);
- static final ScheduledExecutorService STRONG_LEAK_DETECTOR = !DEBUG_ENABLED ? null : executorFactory().scheduled("Strong-Reference-Leak-Detector");
+ static final ScheduledExecutorService STRONG_LEAK_DETECTOR = !TRACE_ENABLED ? null : executorFactory().scheduled("Strong-Reference-Leak-Detector");
static
{
- if (DEBUG_ENABLED)
+ if (TRACE_ENABLED)
{
STRONG_LEAK_DETECTOR.scheduleAtFixedRate(new Visitor(), 1, 15, TimeUnit.MINUTES);
STRONG_LEAK_DETECTOR.scheduleAtFixedRate(new StrongLeakDetector(), 2, 15, TimeUnit.MINUTES);
diff --git a/src/java/org/apache/cassandra/utils/memory/BufferPool.java b/src/java/org/apache/cassandra/utils/memory/BufferPool.java
index cddfc8fe61..e46b0e4d69 100644
--- a/src/java/org/apache/cassandra/utils/memory/BufferPool.java
+++ b/src/java/org/apache/cassandra/utils/memory/BufferPool.java
@@ -1330,7 +1330,7 @@ public class BufferPool
void setAttachment(ByteBuffer buffer)
{
- if (Ref.DEBUG_ENABLED)
+ if (Ref.TRACE_ENABLED)
MemoryUtil.setAttachment(buffer, new DirectBufferRef<>(this, null));
else
MemoryUtil.setAttachment(buffer, this);
@@ -1342,7 +1342,7 @@ public class BufferPool
if (attachment == null)
return false;
- if (Ref.DEBUG_ENABLED)
+ if (Ref.TRACE_ENABLED)
((DirectBufferRef<Chunk>) attachment).release();
return true;
diff --git a/test/distributed/org/apache/cassandra/distributed/test/accord/AccordIncrementalRepairTest.java b/test/distributed/org/apache/cassandra/distributed/test/accord/AccordIncrementalRepairTest.java
index 1e76473eeb..ac7ab99f26 100644
--- a/test/distributed/org/apache/cassandra/distributed/test/accord/AccordIncrementalRepairTest.java
+++ b/test/distributed/org/apache/cassandra/distributed/test/accord/AccordIncrementalRepairTest.java
@@ -36,7 +36,7 @@ import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import accord.local.CommandsForKey;
+import accord.local.cfk.CommandsForKey;
import accord.impl.SimpleProgressLog;
import accord.local.Node;
import accord.local.PreLoadContext;
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 8e663ab966..187f08a11c 100644
--- a/test/distributed/org/apache/cassandra/distributed/test/accord/AccordLoadTest.java
+++ b/test/distributed/org/apache/cassandra/distributed/test/accord/AccordLoadTest.java
@@ -41,6 +41,7 @@ import org.apache.cassandra.distributed.api.ConsistencyLevel;
import org.apache.cassandra.distributed.api.ICoordinator;
import org.apache.cassandra.distributed.api.IMessage;
import org.apache.cassandra.distributed.api.IMessageFilters;
+import org.apache.cassandra.distributed.shared.DistributedTestBase;
import org.apache.cassandra.net.Verb;
import org.apache.cassandra.utils.EstimatedHistogram;
@@ -172,4 +173,13 @@ public class AccordLoadTest extends AccordTestBase
{
return logger;
}
+
+ public static void main(String[] args) throws Throwable
+ {
+ DistributedTestBase.beforeClass();
+ AccordLoadTest.setUp();
+ AccordLoadTest test = new AccordLoadTest();
+ test.setup();
+ test.testLoad();
+ }
}
diff --git a/test/distributed/org/apache/cassandra/distributed/test/log/FetchLogFromPeers2Test.java b/test/distributed/org/apache/cassandra/distributed/test/log/FetchLogFromPeers2Test.java
index d3b549d20f..783583a4fb 100644
--- a/test/distributed/org/apache/cassandra/distributed/test/log/FetchLogFromPeers2Test.java
+++ b/test/distributed/org/apache/cassandra/distributed/test/log/FetchLogFromPeers2Test.java
@@ -26,7 +26,6 @@ import org.junit.Test;
import org.apache.cassandra.distributed.Cluster;
import org.apache.cassandra.distributed.api.ConsistencyLevel;
import org.apache.cassandra.distributed.test.TestBaseImpl;
-import org.apache.cassandra.distributed.test.log.FetchLogFromPeersTest.ClusterState;
import org.apache.cassandra.metrics.TCMMetrics;
import org.apache.cassandra.tcm.ClusterMetadata;
import org.apache.cassandra.tcm.ClusterMetadataService;
diff --git a/test/unit/org/apache/cassandra/db/compaction/CompactionAccordIteratorsTest.java b/test/unit/org/apache/cassandra/db/compaction/CompactionAccordIteratorsTest.java
index 4ac8d242e8..493283bdae 100644
--- a/test/unit/org/apache/cassandra/db/compaction/CompactionAccordIteratorsTest.java
+++ b/test/unit/org/apache/cassandra/db/compaction/CompactionAccordIteratorsTest.java
@@ -31,6 +31,8 @@ import java.util.stream.Collectors;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.Iterators;
+
+import accord.primitives.Routable;
import org.apache.cassandra.config.CassandraRelevantProperties;
import org.apache.cassandra.distributed.shared.WithProperties;
import org.apache.cassandra.service.accord.*;
@@ -42,7 +44,7 @@ import org.slf4j.LoggerFactory;
import accord.api.Key;
import accord.api.Result;
-import accord.local.CommandsForKey;
+import accord.local.cfk.CommandsForKey;
import accord.local.CheckedCommands;
import accord.local.Command;
import accord.local.CommandStore;
@@ -115,6 +117,7 @@ public class CompactionAccordIteratorsTest
private static final TxnId LT_TXN_ID = AccordTestUtils.txnId(EPOCH, HLC_START, NODE);
private static final TxnId TXN_ID = AccordTestUtils.txnId(EPOCH, LT_TXN_ID.hlc() + 1, NODE);
private static final TxnId SECOND_TXN_ID = AccordTestUtils.txnId(EPOCH, TXN_ID.hlc() + 1, NODE, Kind.Read);
+ private static final TxnId RANGE_TXN_ID = AccordTestUtils.txnId(EPOCH, TXN_ID.hlc() + 1, NODE, Kind.Read, Routable.Domain.Range);
private static final TxnId GT_TXN_ID = SECOND_TXN_ID;
// For CommandsForKey where we test with two commands
private static final TxnId[] TXN_IDS = new TxnId[]{ TXN_ID, SECOND_TXN_ID };
@@ -480,7 +483,7 @@ public class CompactionAccordIteratorsTest
CheckedCommands.apply(safe, txnId, route, null, txnId, partialDeps, partialTxn, result.left, result.right, appendDiffToKeyspace(commandStore));
}).beginAsResult());
getUninterruptibly(commandStore.execute(contextFor(txnId, txn.keys(), COMMANDS), safe -> {
- safe.get(txnId, txnId, route).addListener(new Command.ProxyListener(txnId)); // add a junk listener just to test it in compaction
+ safe.get(txnId, txnId, route).addListener(new Command.ProxyListener(RANGE_TXN_ID)); // add a junk listener just to test it in compaction
}).beginAsResult());
flush(commandStore);
// The apply chain is asychronous, so it is easiest to just spin until it is applied
diff --git a/test/unit/org/apache/cassandra/dht/TokenTest.java b/test/unit/org/apache/cassandra/dht/TokenTest.java
new file mode 100644
index 0000000000..234747c2cd
--- /dev/null
+++ b/test/unit/org/apache/cassandra/dht/TokenTest.java
@@ -0,0 +1,51 @@
+/*
+ * 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.cassandra.dht;
+
+import org.junit.Test;
+
+import org.apache.cassandra.config.DatabaseDescriptor;
+import org.apache.cassandra.io.IVersionedSerializers;
+import org.apache.cassandra.io.util.DataOutputBuffer;
+import org.apache.cassandra.net.MessagingService;
+import org.apache.cassandra.utils.AccordGenerators;
+import org.apache.cassandra.utils.CassandraGenerators;
+
+import static accord.utils.Property.qt;
+
+public class TokenTest
+{
+ static
+ {
+ DatabaseDescriptor.clientInitialization();
+ }
+
+ @Test
+ public void serde()
+ {
+ DataOutputBuffer output = new DataOutputBuffer();
+ qt().check(rs -> {
+ IPartitioner partitioner = AccordGenerators.partitioner().next(rs);
+ DatabaseDescriptor.setPartitionerUnsafe(partitioner);
+ Token token = AccordGenerators.fromQT(CassandraGenerators.token(partitioner)).next(rs);
+ for (MessagingService.Version version : MessagingService.Version.values())
+ IVersionedSerializers.testSerde(output, Token.compactSerializer, token, version.value);
+ });
+ }
+}
\ No newline at end of file
diff --git a/test/unit/org/apache/cassandra/io/IVersionedSerializers.java b/test/unit/org/apache/cassandra/io/IVersionedSerializers.java
new file mode 100644
index 0000000000..cbf62fb58c
--- /dev/null
+++ b/test/unit/org/apache/cassandra/io/IVersionedSerializers.java
@@ -0,0 +1,41 @@
+/*
+ * 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.cassandra.io;
+
+import java.io.IOException;
+
+import accord.utils.LazyToString;
+import accord.utils.ReflectionUtils;
+import org.apache.cassandra.io.util.DataInputBuffer;
+import org.apache.cassandra.io.util.DataOutputBuffer;
+import org.assertj.core.api.Assertions;
+
+public class IVersionedSerializers
+{
+ public static <T> void testSerde(DataOutputBuffer output, IVersionedSerializer<T> serializer, T input, int version) throws IOException
+ {
+ output.clear();
+ long expectedSize = serializer.serializedSize(input, version);
+ serializer.serialize(input, output, version);
+ Assertions.assertThat(output.getLength()).describedAs("The serialized size and bytes written do not match").isEqualTo(expectedSize);
+ DataInputBuffer in = new DataInputBuffer(output.unsafeGetBufferAndFlip(), false);
+ T read = serializer.deserialize(in, version);
+ Assertions.assertThat(read).describedAs("The deserialized output does not match the serialized input; difference %s", new LazyToString(() -> ReflectionUtils.recursiveEquals(read, input).toString())).isEqualTo(input);
+ }
+}
\ No newline at end of file
diff --git a/test/unit/org/apache/cassandra/service/accord/AccordCommandStoreTest.java b/test/unit/org/apache/cassandra/service/accord/AccordCommandStoreTest.java
index e56f634d48..bcc1f8bf11 100644
--- a/test/unit/org/apache/cassandra/service/accord/AccordCommandStoreTest.java
+++ b/test/unit/org/apache/cassandra/service/accord/AccordCommandStoreTest.java
@@ -34,7 +34,7 @@ import accord.api.Result;
import accord.impl.TimestampsForKey;
import accord.impl.TimestampsForKeys;
import accord.local.Command;
-import accord.local.CommandsForKey;
+import accord.local.cfk.CommandsForKey;
import accord.local.CommonAttributes;
import accord.local.SaveStatus;
import accord.primitives.Ballot;
@@ -133,7 +133,7 @@ public class AccordCommandStoreTest
attrs.partialDeps(dependencies);
SimpleBitSet waitingOnApply = new SimpleBitSet(3);
waitingOnApply.set(1);
- Command.WaitingOn waitingOn = new Command.WaitingOn(dependencies.keyDeps.keys(), dependencies.rangeDeps.txnIds(), new ImmutableBitSet(waitingOnApply), new ImmutableBitSet(2));
+ Command.WaitingOn waitingOn = new Command.WaitingOn(dependencies.keyDeps.keys(), dependencies.rangeDeps, dependencies.directKeyDeps, new ImmutableBitSet(waitingOnApply), new ImmutableBitSet(2));
attrs.addListener(new Command.ProxyListener(oldTxnId1));
Pair<Writes, Result> result = AccordTestUtils.processTxnResult(commandStore, txnId, txn, executeAt);
@@ -182,8 +182,8 @@ public class AccordCommandStoreTest
AccordSafeCommandsForKey cfk = new AccordSafeCommandsForKey(loaded(key, null));
cfk.initialize();
- cfk.set(cfk.current().update(null, command1));
- cfk.set(cfk.current().update(null, command2));
+ cfk.set(cfk.current().update(command1).cfk());
+ cfk.set(cfk.current().update(command2).cfk());
AccordKeyspace.getTimestampsForKeyMutation(commandStore, tfk, commandStore.nextSystemTimestampMicros()).apply();
logger.info("E: {}", tfk);
@@ -213,8 +213,8 @@ public class AccordCommandStoreTest
AccordSafeCommandsForKey cfk = new AccordSafeCommandsForKey(loaded(key, null));
cfk.initialize();
- cfk.set(cfk.current().update(null, command1));
- cfk.set(cfk.current().update(null, command2));
+ cfk.set(cfk.current().update(command1).cfk());
+ cfk.set(cfk.current().update(command2).cfk());
AccordKeyspace.getCommandsForKeyMutation(commandStore.id(), cfk.current(), commandStore.nextSystemTimestampMicros()).apply();
logger.info("E: {}", cfk);
diff --git a/test/unit/org/apache/cassandra/service/accord/AccordCommandTest.java b/test/unit/org/apache/cassandra/service/accord/AccordCommandTest.java
index d78bd63cbc..a9bdd7aa52 100644
--- a/test/unit/org/apache/cassandra/service/accord/AccordCommandTest.java
+++ b/test/unit/org/apache/cassandra/service/accord/AccordCommandTest.java
@@ -26,7 +26,7 @@ import org.junit.Test;
import accord.api.Key;
import accord.api.RoutingKey;
-import accord.local.CommandsForKey;
+import accord.local.cfk.CommandsForKey;
import accord.local.Command;
import accord.local.KeyHistory;
import accord.local.Node;
diff --git a/test/unit/org/apache/cassandra/service/accord/AccordKeyspaceTest.java b/test/unit/org/apache/cassandra/service/accord/AccordKeyspaceTest.java
index d58105339b..d6d2dab96b 100644
--- a/test/unit/org/apache/cassandra/service/accord/AccordKeyspaceTest.java
+++ b/test/unit/org/apache/cassandra/service/accord/AccordKeyspaceTest.java
@@ -108,8 +108,7 @@ public class AccordKeyspaceTest extends CQLTester.InMemory
PartialTxn partialTxn = txn.slice(scope, true);
RoutingKey routingKey = partialTxn.keys().get(0).asKey().toUnseekable();
FullRoute<?> route = partialTxn.keys().toRoute(routingKey);
- Deps deps = new Deps(KeyDeps.none((Keys) txn.keys()), RangeDeps.NONE);
- deps.slice(scope);
+ Deps deps = new Deps(KeyDeps.none((Keys) txn.keys()), RangeDeps.NONE, KeyDeps.NONE);
CommonAttributes.Mutable common = new CommonAttributes.Mutable(id);
common.partialTxn(partialTxn);
diff --git a/test/unit/org/apache/cassandra/service/accord/AccordStateCacheTest.java b/test/unit/org/apache/cassandra/service/accord/AccordStateCacheTest.java
index 5d632c0fb7..d82f143054 100644
--- a/test/unit/org/apache/cassandra/service/accord/AccordStateCacheTest.java
+++ b/test/unit/org/apache/cassandra/service/accord/AccordStateCacheTest.java
@@ -87,11 +87,6 @@ public class AccordStateCacheTest
original = global.get();
}
- @Override
- public void postExecute()
- {
- }
-
@Override
public Status globalStatus()
{
diff --git a/test/unit/org/apache/cassandra/service/accord/async/AsyncLoaderTest.java b/test/unit/org/apache/cassandra/service/accord/async/AsyncLoaderTest.java
index 33f0dc2458..6ef6a9cde8 100644
--- a/test/unit/org/apache/cassandra/service/accord/async/AsyncLoaderTest.java
+++ b/test/unit/org/apache/cassandra/service/accord/async/AsyncLoaderTest.java
@@ -32,7 +32,7 @@ import org.junit.BeforeClass;
import org.junit.Test;
import accord.api.Key;
-import accord.local.CommandsForKey;
+import accord.local.cfk.CommandsForKey;
import accord.impl.TimestampsForKey;
import accord.local.Command;
import accord.local.KeyHistory;
@@ -105,11 +105,13 @@ public class AsyncLoaderTest
commandCache.unsafeSetLoadFunction(id -> notDefined(id, txn));
AccordSafeCommand safeCommand = commandCache.acquire(txnId);
testLoad(executor, safeCommand, notDefined(txnId, txn));
+ AccordCachingState<TxnId, Command> safeCommandGlobal = safeCommand.global();
commandCache.release(safeCommand);
timestampsCache.unsafeSetLoadFunction(k -> new TimestampsForKey((PartitionKey) k));
AccordSafeTimestampsForKey safeTimestamps = timestampsCache.acquire(key);
testLoad(executor, safeTimestamps, new TimestampsForKey(key));
+ AccordCachingState<Key, TimestampsForKey> safeTimestampsGlobal = safeTimestamps.global();
timestampsCache.release(safeTimestamps);
AsyncLoader loader = new AsyncLoader(commandStore, singleton(txnId), Keys.of(key), TIMESTAMPS);
@@ -118,13 +120,13 @@ public class AsyncLoaderTest
commandStore.executeBlocking(() -> {
Context context = new Context();
boolean result = loader.load(context, (o, t) -> Assert.fail());
- Assert.assertEquals(safeCommand.global(), context.commands.get(txnId).global());
- Assert.assertEquals(safeTimestamps.global(), context.timestampsForKey.get(key).global());
+ Assert.assertEquals(safeCommandGlobal, context.commands.get(txnId).global());
+ Assert.assertEquals(safeTimestampsGlobal, context.timestampsForKey.get(key).global());
Assert.assertTrue(result);
});
- Assert.assertSame(safeCommand.global(), commandCache.getUnsafe(txnId));
- Assert.assertSame(safeTimestamps.global(), timestampsCache.getUnsafe(key));
+ Assert.assertSame(safeCommandGlobal, commandCache.getUnsafe(txnId));
+ Assert.assertSame(safeTimestampsGlobal, timestampsCache.getUnsafe(key));
}
/**
@@ -374,7 +376,7 @@ public class AsyncLoaderTest
@Test
public void inProgressCFKSaveTest()
{
- this.inProgressCFKSaveTest(COMMANDS, AccordCommandStore::commandsForKeyCache, context -> context.commandsForKey, CommandsForKey::new, (cfk, u) -> cfk.update(null, u));
+ this.inProgressCFKSaveTest(COMMANDS, AccordCommandStore::commandsForKeyCache, context -> context.commandsForKey, CommandsForKey::new, (cfk, u) -> cfk.update(u).cfk());
}
@Test
diff --git a/test/unit/org/apache/cassandra/service/accord/async/AsyncOperationTest.java b/test/unit/org/apache/cassandra/service/accord/async/AsyncOperationTest.java
index b50903d24b..ef6cd5b08d 100644
--- a/test/unit/org/apache/cassandra/service/accord/async/AsyncOperationTest.java
+++ b/test/unit/org/apache/cassandra/service/accord/async/AsyncOperationTest.java
@@ -32,7 +32,7 @@ import com.google.common.collect.Iterables;
import com.google.common.collect.Maps;
import org.apache.cassandra.concurrent.SimulatedExecutorFactory;
import org.apache.cassandra.concurrent.Stage;
-import org.apache.cassandra.config.CassandraRelevantProperties;
+
import org.junit.Assert;
import org.junit.Before;
import org.junit.BeforeClass;
@@ -41,7 +41,7 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import accord.api.RoutingKey;
-import accord.local.SafeCommandsForKey;
+import accord.local.cfk.SafeCommandsForKey;
import accord.local.CheckedCommands;
import accord.local.Command;
import accord.local.PreLoadContext;
@@ -103,11 +103,6 @@ public class AsyncOperationTest
private static final Logger logger = LoggerFactory.getLogger(AsyncOperationTest.class);
private static final AtomicLong clock = new AtomicLong(0);
- static
- {
- CassandraRelevantProperties.TEST_ACCORD_STORE_THREAD_CHECKS_ENABLED.setBoolean(false);
- }
-
@BeforeClass
public static void beforeClass() throws Throwable
{
@@ -360,16 +355,16 @@ public class AsyncOperationTest
.check((rs, ids) -> {
before(); // truncate tables
- createCommand(commandStore, rs, ids);
-
- Map<TxnId, Boolean> failed = selectFailedTxn(rs, ids);
+ assertNoReferences(commandStore, ids, keys);
+ createCommand(commandStore, rs, ids);
+ awaitDone(commandStore, ids, keys);
assertNoReferences(commandStore, ids, keys);
PreLoadContext ctx = contextFor(null, ids, keys, COMMANDS);
-
Consumer<SafeCommandStore> consumer = Mockito.mock(Consumer.class);
+ Map<TxnId, Boolean> failed = selectFailedTxn(rs, ids);
commandStore.commandCache().unsafeSetLoadFunction(txnId ->
{
logger.info("Attempting to load {}; expected to fail? {}", txnId, failed.get(txnId));
@@ -402,6 +397,9 @@ public class AsyncOperationTest
});
});
getUninterruptibly(o2);
+ awaitDone(commandStore, ids, keys);
+ assertNoReferences(commandStore, ids, keys);
+
});
}
@@ -419,8 +417,8 @@ public class AsyncOperationTest
logger.info("Test #{}", counter.incrementAndGet());
before(); // truncate tables
- createCommand(commandStore, rs, ids);
assertNoReferences(commandStore, ids, keys);
+ createCommand(commandStore, rs, ids);
PreLoadContext ctx = contextFor(null, ids, keys, COMMANDS);
@@ -445,11 +443,18 @@ public class AsyncOperationTest
// to simulate CommandsForKey not being found, use createCommittedAndPersist periodically as it does not update
switch (rs.nextInt(3))
{
- case 0: ids.forEach(id -> createStableAndPersist(commandStore, id)); break;
- case 1: ids.forEach(id -> createStableUsingFastLifeCycle(commandStore, id)); break;
- case 2: ids.forEach(id -> createStableUsingSlowLifeCycle(commandStore, id));
+ case 0:
+ logger.info("createStableAndPersist(): {}", ids);
+ ids.forEach(id -> createStableAndPersist(commandStore, id));
+ break;
+ case 1:
+ logger.info("createStableUsingFastLifeCycle(): {}", ids);
+ ids.forEach(id -> createStableUsingFastLifeCycle(commandStore, id));
+ break;
+ case 2:
+ logger.info("createStableUsingSlowLifeCycle(): {}", ids);
+ ids.forEach(id -> createStableUsingSlowLifeCycle(commandStore, id));
}
- commandStore.unsafeClearCache();
}
private static Map<TxnId, Boolean> selectFailedTxn(RandomSource rs, List<TxnId> ids)
diff --git a/test/unit/org/apache/cassandra/service/accord/serializers/CommandStoreSerializersTest.java b/test/unit/org/apache/cassandra/service/accord/serializers/CommandStoreSerializersTest.java
new file mode 100644
index 0000000000..738de4809e
--- /dev/null
+++ b/test/unit/org/apache/cassandra/service/accord/serializers/CommandStoreSerializersTest.java
@@ -0,0 +1,72 @@
+/*
+ * 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.cassandra.service.accord.serializers;
+
+import java.util.List;
+
+import org.junit.Test;
+
+import accord.local.RedundantBefore;
+import accord.utils.Gens;
+import org.apache.cassandra.config.DatabaseDescriptor;
+import org.apache.cassandra.dht.Murmur3Partitioner;
+import org.apache.cassandra.io.IVersionedSerializers;
+import org.apache.cassandra.io.util.DataOutputBuffer;
+import org.apache.cassandra.net.MessagingService.Version;
+import org.apache.cassandra.utils.AccordGenerators;
+
+import static accord.utils.Property.qt;
+import static org.apache.cassandra.net.MessagingService.Version.VERSION_51;
+
+public class CommandStoreSerializersTest
+{
+ static
+ {
+ DatabaseDescriptor.clientInitialization();
+ DatabaseDescriptor.setPartitionerUnsafe(Murmur3Partitioner.instance);
+ }
+
+ private static final List<Version> SUPPORTED_VERSIONS = VERSION_51.greaterThanOrEqual();
+
+ @Test
+ public void redundantBeforeEntry()
+ {
+ DataOutputBuffer buffer = new DataOutputBuffer();
+ qt().forAll(Gens.random(), AccordGenerators.partitioner()).check((rs, partitioner) -> {
+ DatabaseDescriptor.setPartitionerUnsafe(partitioner);
+ RedundantBefore.Entry entry = AccordGenerators.redundantBeforeEntry(partitioner).next(rs);
+ for (Version version : SUPPORTED_VERSIONS)
+ IVersionedSerializers.testSerde(buffer, CommandStoreSerializers.redundantBeforeEntry, entry, version.value);
+ });
+ }
+
+ @Test
+ public void redundantBefore()
+ {
+ DataOutputBuffer buffer = new DataOutputBuffer();
+ qt().forAll(Gens.random(), AccordGenerators.partitioner()).check((rs, partitioner) -> {
+ DatabaseDescriptor.setPartitionerUnsafe(partitioner);
+ // serializer doesn't support the empty set, so filter out
+ RedundantBefore redundantBefore = AccordGenerators.redundantBefore(partitioner).filter(r -> r.size() != 0).next(rs);
+ for (Version version : SUPPORTED_VERSIONS)
+ IVersionedSerializers.testSerde(buffer, CommandStoreSerializers.redundantBefore, redundantBefore, version.value);
+ });
+ }
+
+}
\ No newline at end of file
diff --git a/test/unit/org/apache/cassandra/service/accord/serializers/CommandsForKeySerializerTest.java b/test/unit/org/apache/cassandra/service/accord/serializers/CommandsForKeySerializerTest.java
index 202dc5cb78..33244f045b 100644
--- a/test/unit/org/apache/cassandra/service/accord/serializers/CommandsForKeySerializerTest.java
+++ b/test/unit/org/apache/cassandra/service/accord/serializers/CommandsForKeySerializerTest.java
@@ -39,11 +39,11 @@ import org.junit.BeforeClass;
import org.junit.Test;
import accord.api.Key;
-import accord.local.CommandsForKey;
-import accord.local.CommandsForKey.InternalStatus;
+import accord.local.cfk.CommandsForKey;
+import accord.local.cfk.CommandsForKey.InternalStatus;
import accord.local.Command;
-import accord.local.CommandsForKey.TxnInfo;
-import accord.local.CommandsForKey.Unmanaged;
+import accord.local.cfk.CommandsForKey.TxnInfo;
+import accord.local.cfk.CommandsForKey.Unmanaged;
import accord.local.CommonAttributes;
import accord.local.CommonAttributes.Mutable;
import accord.local.Listeners;
@@ -106,16 +106,18 @@ public class CommandsForKeySerializerTest
final SaveStatus saveStatus;
final PartialTxn txn;
final Timestamp executeAt;
+ final Ballot ballot;
final List<TxnId> deps = new ArrayList<>();
final List<TxnId> missing = new ArrayList<>();
boolean invisible;
- Cmd(TxnId txnId, PartialTxn txn, SaveStatus saveStatus, Timestamp executeAt)
+ Cmd(TxnId txnId, PartialTxn txn, SaveStatus saveStatus, Timestamp executeAt, Ballot ballot)
{
this.txnId = txnId;
this.saveStatus = saveStatus;
this.txn = txn;
this.executeAt = executeAt;
+ this.ballot = ballot;
}
CommonAttributes attributes()
@@ -132,7 +134,7 @@ public class CommandsForKeySerializerTest
{
for (TxnId id : deps)
builder.add((Key)txn.keys().get(0), id);
- mutable.partialDeps(new PartialDeps(AccordTestUtils.fullRange(txn), builder.build(), RangeDeps.NONE));
+ mutable.partialDeps(new PartialDeps(AccordTestUtils.fullRange(txn), builder.build(), RangeDeps.NONE, KeyDeps.NONE));
}
}
@@ -157,19 +159,19 @@ public class CommandsForKeySerializerTest
case PreCommittedWithDefinitionAndAcceptedDeps:
case PreCommittedWithAcceptedDeps:
case PreCommitted:
- return Command.SerializerSupport.accepted(attributes(), saveStatus, executeAt, Ballot.ZERO, Ballot.ZERO);
+ return Command.SerializerSupport.accepted(attributes(), saveStatus, executeAt, ballot, ballot);
case Committed:
- return Command.SerializerSupport.committed(attributes(), saveStatus, executeAt, Ballot.ZERO, Ballot.ZERO, null);
+ return Command.SerializerSupport.committed(attributes(), saveStatus, executeAt, ballot, ballot, null);
case Stable:
case ReadyToExecute:
- return Command.SerializerSupport.committed(attributes(), saveStatus, executeAt, Ballot.ZERO, Ballot.ZERO, Command.WaitingOn.EMPTY);
+ return Command.SerializerSupport.committed(attributes(), saveStatus, executeAt, ballot, ballot, Command.WaitingOn.EMPTY);
case PreApplied:
case Applying:
case Applied:
- return Command.SerializerSupport.executed(attributes(), saveStatus, executeAt, Ballot.ZERO, Ballot.ZERO, Command.WaitingOn.EMPTY, new Writes(txnId, executeAt, txn.keys(), new TxnWrite(Collections.emptyList(), true)), new TxnData());
+ return Command.SerializerSupport.executed(attributes(), saveStatus, executeAt, ballot, ballot, Command.WaitingOn.EMPTY, new Writes(txnId, executeAt, txn.keys(), new TxnWrite(Collections.emptyList(), true)), new TxnData());
case TruncatedApplyWithDeps:
case TruncatedApply:
@@ -219,7 +221,7 @@ public class CommandsForKeySerializerTest
}
}
- private static ObjectGraph generateObjectGraph(int txnIdCount, Supplier<TxnId> txnIdSupplier, Supplier<SaveStatus> saveStatusSupplier, Function<TxnId, PartialTxn> txnSupplier, Function<TxnId, Timestamp> timestampSupplier, IntSupplier missingCountSupplier, RandomSource source)
+ private static ObjectGraph generateObjectGraph(int txnIdCount, Supplier<TxnId> txnIdSupplier, Supplier<SaveStatus> saveStatusSupplier, Function<TxnId, PartialTxn> txnSupplier, Function<TxnId, Timestamp> timestampSupplier, Supplier<Ballot> ballotSupplier, IntSupplier missingCountSupplier, RandomSource source)
{
Cmd[] cmds = new Cmd[txnIdCount];
for (int i = 0 ; i < txnIdCount ; ++i)
@@ -227,10 +229,30 @@ public class CommandsForKeySerializerTest
TxnId txnId = txnIdSupplier.get();
SaveStatus saveStatus = saveStatusSupplier.get();
Timestamp executeAt = txnId;
- if (saveStatus.known.executeAt != ExecuteAtErased && saveStatus.known.executeAt != ExecuteAtUnknown)
+ if (!txnId.kind().awaitsOnlyDeps() && saveStatus.known.executeAt != ExecuteAtErased && saveStatus.known.executeAt != ExecuteAtUnknown)
executeAt = timestampSupplier.apply(txnId);
- cmds[i] = new Cmd(txnId, txnSupplier.apply(txnId), saveStatus, executeAt);
+ Ballot ballot;
+ switch (saveStatus.status)
+ {
+ default: throw new AssertionError();
+ case NotDefined:
+ case PreAccepted:
+ case Invalidated:
+ case Truncated:
+ ballot = Ballot.ZERO;
+ break;
+ case AcceptedInvalidate:
+ case Accepted:
+ case PreCommitted:
+ case Committed:
+ case Stable:
+ case PreApplied:
+ case Applied:
+ ballot = ballotSupplier.get();
+ }
+
+ cmds[i] = new Cmd(txnId, txnSupplier.apply(txnId), saveStatus, executeAt, ballot);
}
Arrays.sort(cmds, Comparator.comparing(o -> o.txnId));
for (int i = 0 ; i < txnIdCount ; ++i)
@@ -273,14 +295,14 @@ public class CommandsForKeySerializerTest
for (int j = 0 ; j < i ; ++j)
{
InternalStatus status = InternalStatus.from(cmds[j].saveStatus);
- if (status == null || !status.hasInfo) continue;
+ if (status == null || !status.hasExecuteAtOrDeps) continue;
if (cmds[j].txnId.kind().witnesses(cmds[i].txnId) && status.depsKnownBefore(cmds[j].txnId, cmds[j].executeAt).compareTo(cmds[i].txnId) > 0 && Collections.binarySearch(cmds[j].missing, cmds[i].txnId) < 0)
continue outer;
}
for (int j = i + 1 ; j < cmds.length ; ++j)
{
InternalStatus status = InternalStatus.from(cmds[j].saveStatus);
- if (status == null || !status.hasInfo) continue;
+ if (status == null || !status.hasExecuteAtOrDeps) continue;
if (cmds[j].txnId.kind().witnesses(cmds[i].txnId) && Collections.binarySearch(cmds[j].missing, cmds[i].txnId) < 0)
continue outer;
}
@@ -312,6 +334,11 @@ public class CommandsForKeySerializerTest
return min -> Timestamp.fromValues(epochSupplier.applyAsLong(min == null ? 1 : min.epoch()), hlcSupplier.applyAsLong(min == null ? 1 : min.hlc() + 1), flagSupplier.getAsInt(), idSupplier.get());
}
+ private static Supplier<Ballot> ballotSupplier(LongUnaryOperator epochSupplier, LongUnaryOperator hlcSupplier, IntSupplier flagSupplier, Supplier<Node.Id> idSupplier)
+ {
+ return () -> Ballot.fromValues(epochSupplier.applyAsLong(1), hlcSupplier.applyAsLong(1), flagSupplier.getAsInt(), idSupplier.get());
+ }
+
private static <T extends Timestamp> Function<Timestamp, T> timestampSupplier(Set<Timestamp> unique, Function<Timestamp, T> supplier)
{
return min -> {
@@ -329,7 +356,7 @@ public class CommandsForKeySerializerTest
@Test
public void serde()
{
- testOne(-6946067792202944553L);
+ testOne(-669467611022826851L);
Random random = new Random();
for (int i = 0 ; i < 10000 ; ++i)
{
@@ -364,8 +391,8 @@ public class CommandsForKeySerializerTest
float v = source.nextFloat();
if (v < 0.5) return Txn.Kind.Read;
if (v < 0.95) return Txn.Kind.Write;
- if (v < 0.99) return Txn.Kind.ExclusiveSyncPoint;
- return Txn.Kind.EphemeralRead; // not actually a valid value for CFK
+ if (v < 0.97) return Txn.Kind.SyncPoint;
+ return Txn.Kind.ExclusiveSyncPoint;
};
boolean permitMissing = source.decide(0.75f);
@@ -412,15 +439,21 @@ public class CommandsForKeySerializerTest
}
}
+ Supplier<Ballot> ballotSupplier;
+ {
+ Supplier<Ballot> delegate = ballotSupplier(epochSupplier, hlcSupplier, flagSupplier, idSupplier);
+ ballotSupplier = () -> source.decide(0.5f) ? Ballot.ZERO : delegate.get();
+ }
+
PartialTxn txn = createPartialTxn(0);
Key key = (Key) txn.keys().get(0);
- ObjectGraph graph = generateObjectGraph(source.nextInt(0, 100), () -> txnIdSupplier.apply(null), saveStatusSupplier, ignore -> txn, executeAtSupplier, missingCountSupplier, source);
+ ObjectGraph graph = generateObjectGraph(source.nextInt(0, 100), () -> txnIdSupplier.apply(null), saveStatusSupplier, ignore -> txn, executeAtSupplier, ballotSupplier, missingCountSupplier, source);
List<Command> commands = graph.toCommands();
CommandsForKey cfk = new CommandsForKey(key);
while (commands.size() > 0)
{
int next = source.nextInt(commands.size());
- cfk = cfk.update(null, commands.get(next));
+ cfk = cfk.update(commands.get(next)).cfk();
commands.set(next, commands.get(commands.size() - 1));
commands.remove(commands.size() - 1);
}
@@ -436,10 +469,12 @@ public class CommandsForKeySerializerTest
TxnInfo info = cfk.get(i);
InternalStatus expectStatus = InternalStatus.from(cmd.saveStatus);
if (expectStatus == null) expectStatus = InternalStatus.TRANSITIVELY_KNOWN;
- if (expectStatus.hasInfo)
+ if (expectStatus.hasExecuteAtOrDeps)
Assert.assertEquals(cmd.executeAt, info.executeAt);
Assert.assertEquals(expectStatus, info.status);
Assert.assertArrayEquals(cmd.missing.toArray(TxnId[]::new), info.missing());
+ if (expectStatus.hasBallot)
+ Assert.assertEquals(cmd.ballot, info.ballot());
++i;
}
@@ -471,7 +506,10 @@ public class CommandsForKeySerializerTest
Arrays.sort(ids, Comparator.naturalOrder());
TxnInfo[] info = new TxnInfo[ids.length];
for (int i = 0; i < info.length; i++)
- info[i] = TxnInfo.create(ids[i], rs.pick(InternalStatus.values()), ids[i], CommandsForKey.NO_TXNIDS);
+ {
+ InternalStatus status = rs.pick(InternalStatus.values());
+ info[i] = TxnInfo.create(ids[i], status, ids[i], CommandsForKey.NO_TXNIDS, Ballot.ZERO);
+ }
Gen<Unmanaged.Pending> pendingGen = Gens.enums().allMixedDistribution(Unmanaged.Pending.class).next(rs);
@@ -499,7 +537,7 @@ public class CommandsForKeySerializerTest
}
else unmanaged = CommandsForKey.NO_PENDING_UNMANAGED;
- CommandsForKey expected = CommandsForKey.SerializerSupport.create(pk, info, unmanaged);
+ CommandsForKey expected = CommandsForKey.SerializerSupport.create(pk, info, unmanaged, TxnId.NONE);
ByteBuffer buffer = CommandsForKeySerializer.toBytesWithoutKey(expected);
CommandsForKey roundTrip = CommandsForKeySerializer.fromBytes(pk, buffer);
@@ -515,8 +553,8 @@ public class CommandsForKeySerializerTest
PartitionKey pk = new PartitionKey(TableId.fromString("1b255f4d-ef25-40a6-0000-000000000009"), key);
TxnId txnId = TxnId.fromValues(11,34052499,2,1);
CommandsForKey expected = CommandsForKey.SerializerSupport.create(pk,
- new TxnInfo[] { TxnInfo.create(txnId, InternalStatus.PREACCEPTED, txnId, CommandsForKey.NO_TXNIDS) },
- CommandsForKey.NO_PENDING_UNMANAGED);
+ new TxnInfo[] { TxnInfo.create(txnId, InternalStatus.PREACCEPTED_OR_ACCEPTED_INVALIDATE, txnId, CommandsForKey.NO_TXNIDS, Ballot.ZERO) },
+ CommandsForKey.NO_PENDING_UNMANAGED, TxnId.NONE);
ByteBuffer buffer = CommandsForKeySerializer.toBytesWithoutKey(expected);
CommandsForKey roundTrip = CommandsForKeySerializer.fromBytes(pk, buffer);
diff --git a/test/unit/org/apache/cassandra/service/accord/serializers/DepsSerializerTest.java b/test/unit/org/apache/cassandra/service/accord/serializers/DepsSerializerTest.java
new file mode 100644
index 0000000000..4ee49b24b9
--- /dev/null
+++ b/test/unit/org/apache/cassandra/service/accord/serializers/DepsSerializerTest.java
@@ -0,0 +1,64 @@
+/*
+ * 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.cassandra.service.accord.serializers;
+
+import java.util.List;
+
+import org.junit.Test;
+
+import accord.primitives.Deps;
+import org.apache.cassandra.config.DatabaseDescriptor;
+import org.apache.cassandra.dht.IPartitioner;
+import org.apache.cassandra.dht.Murmur3Partitioner;
+import org.apache.cassandra.io.IVersionedSerializers;
+import org.apache.cassandra.io.util.DataOutputBuffer;
+import org.apache.cassandra.net.MessagingService;
+import org.apache.cassandra.schema.Schema;
+import org.apache.cassandra.schema.SchemaProvider;
+import org.apache.cassandra.utils.AccordGenerators;
+import org.mockito.Mockito;
+
+import static accord.utils.Property.qt;
+import static org.apache.cassandra.net.MessagingService.Version.VERSION_51;
+
+public class DepsSerializerTest
+{
+ static
+ {
+ DatabaseDescriptor.clientInitialization();
+ DatabaseDescriptor.setPartitionerUnsafe(Murmur3Partitioner.instance);
+ }
+
+ private static final List<MessagingService.Version> SUPPORTED_VERSIONS = VERSION_51.greaterThanOrEqual();
+
+ @Test
+ public void serde()
+ {
+ DataOutputBuffer buffer = new DataOutputBuffer();
+ qt().withSeed(-4368731546033726179L).check(rs -> {
+ IPartitioner partitioner = AccordGenerators.partitioner().next(rs);
+ Schema.instance = Mockito.mock(SchemaProvider.class);
+ DatabaseDescriptor.setPartitionerUnsafe(partitioner);
+ Mockito.when(Schema.instance.getExistingTablePartitioner(Mockito.any())).thenReturn(partitioner);
+ Deps deps = AccordGenerators.depsGen(partitioner).next(rs);
+ for (MessagingService.Version version : SUPPORTED_VERSIONS)
+ IVersionedSerializers.testSerde(buffer, DepsSerializer.deps, deps, version.value);
+ });
+ }
+}
\ No newline at end of file
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 07b626346a..d303626465 100644
--- a/test/unit/org/apache/cassandra/service/accord/serializers/WaitingOnSerializerTest.java
+++ b/test/unit/org/apache/cassandra/service/accord/serializers/WaitingOnSerializerTest.java
@@ -18,6 +18,8 @@
package org.apache.cassandra.service.accord.serializers;
+import java.nio.ByteBuffer;
+
import org.junit.BeforeClass;
import org.junit.Test;
@@ -32,8 +34,6 @@ import accord.utils.Utils;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.dht.LocalPartitioner;
import org.apache.cassandra.dht.Murmur3Partitioner;
-import org.apache.cassandra.io.util.DataInputBuffer;
-import org.apache.cassandra.io.util.DataOutputBuffer;
import org.apache.cassandra.utils.AccordGenerators;
import org.apache.cassandra.utils.CassandraGenerators;
import org.assertj.core.api.Assertions;
@@ -52,18 +52,16 @@ public class WaitingOnSerializerTest
@Test
public void serde()
{
- DataOutputBuffer buffer = new DataOutputBuffer();
qt().forAll(waitingOnGen()).check(waitingOn -> {
TxnId txnId = TxnId.NONE;
if (waitingOn.appliedOrInvalidated != null) txnId = new TxnId(txnId.epoch(), txnId.hlc(), txnId.kind(), Routable.Domain.Range, txnId.node);
- buffer.clear();
- long expectedSize = WaitingOnSerializer.serializedSize(txnId, waitingOn);
- WaitingOnSerializer.serialize(txnId, waitingOn, buffer);
- Assertions.assertThat(buffer.getLength()).isEqualTo(expectedSize);
- Command.WaitingOn read = WaitingOnSerializer.deserialize(txnId, waitingOn.keys, waitingOn.txnIds, new DataInputBuffer(buffer.unsafeGetBufferAndFlip(), false));
+ long expectedSize = WaitingOnSerializer.serializedSize(waitingOn);
+ ByteBuffer bb = WaitingOnSerializer.serialize(txnId, 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.txnIds, WaitingOnSerializer.serialize(txnId, waitingOn)));
+ .isEqualTo(WaitingOnSerializer.deserialize(txnId, waitingOn.keys, waitingOn.directRangeDeps, waitingOn.directKeyDeps, WaitingOnSerializer.serialize(txnId, waitingOn)));
});
}
@@ -78,7 +76,7 @@ public class WaitingOnSerializerTest
return rs -> {
Deps deps = depsGen.next(rs);
if (deps.isEmpty()) return Command.WaitingOn.EMPTY;
- int txnIdCount = deps.rangeDeps.txnIdCount();
+ int txnIdCount = deps.rangeDeps.txnIdCount() + deps.directKeyDeps.txnIdCount();
int keyCount = deps.keyDeps.keys().size();
int[] selected = Gens.arrays(Gens.ints().between(0, txnIdCount + keyCount - 1)).unique().ofSizeBetween(0, txnIdCount + keyCount).next(rs);
SimpleBitSet waitingOn = new SimpleBitSet(txnIdCount + keyCount, false);
@@ -99,7 +97,7 @@ public class WaitingOnSerializerTest
}
}
- return new Command.WaitingOn(deps.keyDeps.keys(), deps.rangeDeps.txnIds(), Utils.ensureImmutable(waitingOn), Utils.ensureImmutable(appliedOrInvalidated));
+ return new Command.WaitingOn(deps.keyDeps.keys(), deps.rangeDeps, deps.directKeyDeps, Utils.ensureImmutable(waitingOn), Utils.ensureImmutable(appliedOrInvalidated));
};
}
}
\ No newline at end of file
diff --git a/test/unit/org/apache/cassandra/utils/AccordGenerators.java b/test/unit/org/apache/cassandra/utils/AccordGenerators.java
index 670e029bd3..b4ae2fa07f 100644
--- a/test/unit/org/apache/cassandra/utils/AccordGenerators.java
+++ b/test/unit/org/apache/cassandra/utils/AccordGenerators.java
@@ -23,8 +23,11 @@ import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
+import java.util.function.BiFunction;
+import java.util.stream.Stream;
import accord.local.Command;
+import accord.local.RedundantBefore;
import accord.primitives.Deps;
import accord.primitives.FullRoute;
import accord.primitives.KeyDeps;
@@ -32,11 +35,14 @@ import accord.primitives.PartialTxn;
import accord.primitives.Range;
import accord.primitives.RangeDeps;
import accord.primitives.Ranges;
+import accord.primitives.Routable;
import accord.primitives.Timestamp;
+import accord.primitives.Txn;
import accord.primitives.TxnId;
import accord.utils.AccordGens;
import accord.utils.Gen;
import accord.utils.Gens;
+import accord.utils.RandomSource;
import org.apache.cassandra.db.DecoratedKey;
import org.apache.cassandra.dht.AccordSplitter;
import org.apache.cassandra.dht.IPartitioner;
@@ -48,7 +54,6 @@ import org.apache.cassandra.service.accord.api.AccordRoutingKey;
import org.apache.cassandra.service.accord.api.PartitionKey;
import org.quicktheories.impl.JavaRandom;
-import static accord.utils.AccordGens.txnIds;
import static org.apache.cassandra.service.accord.AccordTestUtils.TABLE_ID1;
import static org.apache.cassandra.service.accord.AccordTestUtils.createPartialTxn;
@@ -70,7 +75,7 @@ public class AccordGenerators
public static Gen<Command> commands()
{
- Gen<TxnId> ids = txnIds();
+ Gen<TxnId> ids = AccordGens.txnIds();
//TODO switch to Status once all types are supported
Gen<SupportedCommandTypes> supportedTypes = Gens.enums().all(SupportedCommandTypes.class);
//TODO goes against fuzz testing, and also limits to a very specific table existing...
@@ -197,6 +202,19 @@ public class AccordGenerators
return ranges(Gens.lists(fromQT(CassandraGenerators.TABLE_ID_GEN)).unique().ofSizeBetween(1, 10).map(l -> new HashSet<>(l)), ignore -> partitioner);
}
+ public static Gen<Ranges> rangesArbitrary(IPartitioner partitioner)
+ {
+ Gen<Range> rangeGen = range(partitioner);
+ Gen.IntGen sizeGen = Gens.ints().between(0, 10);
+ return rs -> {
+ int targetSize = sizeGen.nextInt(rs);
+ List<Range> ranges = new ArrayList<>(targetSize);
+ for (int i = 0; i < targetSize; i++)
+ ranges.add(rangeGen.next(rs));
+ return Ranges.of(ranges.toArray(Range[]::new));
+ };
+ }
+
public static Gen<KeyDeps> keyDepsGen()
{
return AccordGens.keyDeps(AccordGenerators.keys());
@@ -207,6 +225,16 @@ public class AccordGenerators
return AccordGens.keyDeps(AccordGenerators.keys(partitioner));
}
+ public static Gen<KeyDeps> directKeyDepsGen()
+ {
+ return AccordGens.directKeyDeps(AccordGenerators.keys());
+ }
+
+ public static Gen<KeyDeps> directKeyDepsGen(IPartitioner partitioner)
+ {
+ return AccordGens.directKeyDeps(AccordGenerators.keys(partitioner));
+ }
+
public static Gen<RangeDeps> rangeDepsGen()
{
return AccordGens.rangeDeps(AccordGenerators.range());
@@ -219,12 +247,41 @@ public class AccordGenerators
public static Gen<Deps> depsGen()
{
- return AccordGens.deps(keyDepsGen(), rangeDepsGen());
+ return AccordGens.deps(keyDepsGen(), rangeDepsGen(), directKeyDepsGen());
}
public static Gen<Deps> depsGen(IPartitioner partitioner)
{
- return AccordGens.deps(keyDepsGen(partitioner), rangeDepsGen(partitioner));
+ return AccordGens.deps(keyDepsGen(partitioner), rangeDepsGen(partitioner), directKeyDepsGen(partitioner));
+ }
+
+ public static Gen<RedundantBefore.Entry> redundantBeforeEntry(IPartitioner partitioner)
+ {
+ return redundantBeforeEntry(Gens.bools().all(), range(partitioner), AccordGens.txnIds(Gens.pick(Txn.Kind.SyncPoint, Txn.Kind.ExclusiveSyncPoint), ignore -> Routable.Domain.Range));
+ }
+
+ public static Gen<RedundantBefore.Entry> redundantBeforeEntry(Gen<Boolean> emptyGen, Gen<Range> rangeGen, Gen<TxnId> txnIdGen)
+ {
+ return rs -> {
+ Range range = rangeGen.next(rs);
+ TxnId locallyAppliedOrInvalidatedBefore = emptyGen.next(rs) ? TxnId.NONE : txnIdGen.next(rs); // emptyable or range
+ TxnId shardAppliedOrInvalidatedBefore = emptyGen.next(rs) ? TxnId.NONE : txnIdGen.next(rs); // emptyable or range
+ TxnId bootstrappedAt = txnIdGen.next(rs);
+ Timestamp staleUntilAtLeast = emptyGen.next(rs) ? null : txnIdGen.next(rs); // nullable
+
+ long maxEpoch = Stream.of(locallyAppliedOrInvalidatedBefore, shardAppliedOrInvalidatedBefore, bootstrappedAt, staleUntilAtLeast).filter(t -> t != null).mapToLong(Timestamp::epoch).max().getAsLong();
+ long startEpoch = rs.nextLong(maxEpoch);
+ long endEpoch = emptyGen.next(rs) ? Long.MAX_VALUE : 1 + rs.nextLong(startEpoch, Long.MAX_VALUE);
+ return new RedundantBefore.Entry(range, startEpoch, endEpoch, locallyAppliedOrInvalidatedBefore, shardAppliedOrInvalidatedBefore, bootstrappedAt, staleUntilAtLeast);
+ };
+ }
+
+ public static Gen<RedundantBefore> redundantBefore(IPartitioner partitioner)
+ {
+ Gen<Ranges> rangeGen = rangesArbitrary(partitioner);
+ Gen<TxnId> txnIdGen = AccordGens.txnIds(Gens.pick(Txn.Kind.SyncPoint, Txn.Kind.ExclusiveSyncPoint), ignore -> Routable.Domain.Range);
+ BiFunction<RandomSource, Range, RedundantBefore.Entry> entryGen = (rs, range) -> redundantBeforeEntry(Gens.bools().all(), i -> range, txnIdGen).next(rs);
+ return AccordGens.redundantBefore(rangeGen, entryGen);
}
public static <T> Gen<T> fromQT(org.quicktheories.core.Gen<T> qt)
diff --git a/test/unit/org/apache/cassandra/utils/CassandraGenerators.java b/test/unit/org/apache/cassandra/utils/CassandraGenerators.java
index 00c7c7286e..eb1b9abddd 100644
--- a/test/unit/org/apache/cassandra/utils/CassandraGenerators.java
+++ b/test/unit/org/apache/cassandra/utils/CassandraGenerators.java
@@ -823,6 +823,15 @@ public final class CassandraGenerators
public static Gen<DecoratedKey> decoratedKeys(Gen<IPartitioner> partitionerGen, Gen<ByteBuffer> keyGen)
{
- return rs -> partitionerGen.generate(rs).decorateKey(keyGen.generate(rs));
+ return rs -> {
+ IPartitioner partitioner = partitionerGen.generate(rs);
+ Gen<ByteBuffer> valueGen = keyGen;
+ if (partitioner instanceof LocalPartitioner)
+ {
+ LocalPartitioner lp = (LocalPartitioner) partitioner;
+ valueGen = AbstractTypeGenerators.getTypeSupport(lp.getTokenValidator()).bytesGen();
+ }
+ return partitioner.decorateKey(valueGen.generate(rs));
+ };
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@cassandra.apache.org
For additional commands, e-mail: commits-help@cassandra.apache.org