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:16 UTC
(cassandra) branch cep-15-accord updated (6c5727cf0f -> 08fd33c4b2)
This is an automated email from the ASF dual-hosted git repository.
benedict pushed a change to branch cep-15-accord
in repository https://gitbox.apache.org/repos/asf/cassandra.git
discard 6c5727cf0f CommandsForKey Improvements incl Pruning
new 08fd33c4b2 CommandsForKey Improvements incl Pruning
This update added new revisions after undoing existing revisions.
That is to say, some revisions that were in the old version of the
branch are not in the new version. This situation occurs
when a user --force pushes a change and generates a repository
containing something like this:
* -- * -- B -- O -- O -- O (6c5727cf0f)
\
N -- N -- N refs/heads/cep-15-accord (08fd33c4b2)
You should already have received notification emails for all of the O
revisions, and so the following emails describe only the N revisions
from the common base, B.
Any revisions marked "omit" are not gone; other references still
refer to them. Any revisions marked "discard" are gone forever.
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
modules/accord | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@cassandra.apache.org
For additional commands, e-mail: commits-help@cassandra.apache.org
(cassandra) 01/01: CommandsForKey Improvements incl Pruning
Posted by be...@apache.org.
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