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

Improve DurabilityQueue: 

- All retries are appended to a delay queue so overlaps can be pruned
 - Quorum successes are not retried if there are superseding sync points covering the ranges
 - User-initiated requests are not purged unless the request has timed out or otherwise completed
 - Overlapping requests are queued up against the next to run
Alsp (C*):
 - Catch-up with quorums on restart
 - Manage an ordered set of keys in cache for faster range searches
Also (Accord):
 - Update copy of BTree and import IntervalBTree
 - Fix RedundantStatus WAS_OWNED_OVERRIDE_MASK
 - Add Catchup mechanism to reach parity with a quorum on restart

patch by Benedict; reviewed by Alex Petrov for CASSANDRA-21013
This commit is contained in:
Benedict Elliott Smith 2025-11-07 11:26:50 +00:00
parent 4b7ef7c8f3
commit f6fd49566f
14 changed files with 704 additions and 91 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
modules/accord

@ -1 +1 @@
Subproject commit a681b1bf7d9e8d1d797af2fd54bdd180a4863406
Subproject commit de03d14e4821b669fecad5dcdb2de2dba3a9d926

7
src/java/org/apache/cassandra/config/AccordSpec.java
View File

@ -178,6 +178,13 @@ public class AccordSpec
public TransactionalMode default_transactional_mode = TransactionalMode.off;
public boolean ephemeralReadEnabled = true;
public boolean state_cache_listener_jfr_enabled = true;
public DurationSpec.IntSecondsBound catchup_on_start_success_latency = new DurationSpec.IntSecondsBound(60);
public DurationSpec.IntSecondsBound catchup_on_start_fail_latency = new DurationSpec.IntSecondsBound(900);
public int catchup_on_start_max_attempts = 5;
public boolean catchup_on_start_exit_on_failure = true;
public boolean catchup_on_start = true;
public final JournalSpec journal = new JournalSpec();
public enum MixedTimeSourceHandling

49
src/java/org/apache/cassandra/service/accord/AccordCache.java
View File

@ -21,19 +21,17 @@ import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Comparator;
import java.util.IdentityHashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.TimeUnit;
import java.util.function.BiConsumer;
import java.util.function.BiFunction;
import java.util.function.Function;
import java.util.function.ToLongFunction;
import java.util.stream.Stream;
import javax.annotation.Nullable;
import com.google.common.annotations.VisibleForTesting;
@ -68,6 +66,7 @@ import org.apache.cassandra.service.accord.serializers.Version;
import org.apache.cassandra.utils.NoSpamLogger;
import org.apache.cassandra.utils.NoSpamLogger.NoSpamLogStatement;
import org.apache.cassandra.utils.ObjectSizes;
import org.apache.cassandra.utils.btree.BTree;
import static accord.utils.Invariants.illegalState;
import static accord.utils.Invariants.require;
@ -112,6 +111,7 @@ public class AccordCache implements CacheSize
long estimateShrunkHeapSize(Object shrunk);
boolean validate(AccordCommandStore commandStore, K key, V value);
S safeRef(AccordCacheEntry<K, V> node);
default Comparator<K> keyComparator() { return null; }
default AccordCacheEntry<K, V> newEntry(K key, AccordCache.Type<K, V, ?>.Instance owner)
{
@ -274,7 +274,7 @@ public class AccordCache implements CacheSize
entry.savingOrWaitingToSave().identity.onSuccess(onSuccess);
}
private void evict(AccordCacheEntry<?, ?> node, boolean updateUnreferenced)
private <K> void evict(AccordCacheEntry<K, ?> node, boolean updateUnreferenced)
{
if (logger.isTraceEnabled())
logger.trace("Evicting {}", node);
@ -297,7 +297,7 @@ public class AccordCache implements CacheSize
if (node.status() == LOADED && VALIDATE_LOAD_ON_EVICT)
owner.validateLoadEvicted(node);
AccordCacheEntry<?, ?> self = node.owner.cache.remove(node.key());
AccordCacheEntry<K, ?> self = node.owner.remove(node.key());
Invariants.require(self.references() == 0);
require(self == node, "Leaked node detected; was attempting to remove %s but cache had %s", node, self);
node.notifyListeners(Listener::onEvict);
@ -400,6 +400,8 @@ public class AccordCache implements CacheSize
// TODO (desired): don't need to store key separately as stored in node; ideally use a hash set that allows us to get the current entry
private final Map<K, AccordCacheEntry<K, V>> cache = new Object2ObjectHashMap<>();
private List<Listener<K, V>> listeners = null;
// TODO (expected): update this after releasing the lock
private OrderedKeys<K> orderedKeys;
public Instance(AccordCommandStore commandStore)
{
@ -435,7 +437,6 @@ public class AccordCache implements CacheSize
private AccordCacheEntry<K, V> acquire(K key, boolean onlyIfLoaded)
{
@SuppressWarnings("unchecked")
AccordCacheEntry<K, V> node = cache.get(key);
return node == null
? acquireAbsent(key, onlyIfLoaded)
@ -454,8 +455,10 @@ public class AccordCache implements CacheSize
node.increment();
Object prev = cache.put(key, node);
node.initSize(parent());
Invariants.require(prev == null, "%s not absent from cache: %s already present", key, node);
if (orderedKeys != null)
orderedKeys.add(key);
node.initSize(parent());
++size;
node.notifyListeners(Listener::onAdd);
maybeShrinkOrEvictSomeNodes();
@ -555,9 +558,12 @@ public class AccordCache implements CacheSize
maybeShrinkOrEvictSomeNodes();
}
public Stream<AccordCacheEntry<K, V>> stream()
AccordCacheEntry<K, ?> remove(K key)
{
return cache.values().stream();
AccordCacheEntry<K, ?> result = cache.remove(key);
if (orderedKeys != null && result != null)
orderedKeys.remove(key);
return result;
}
final Type<K, V, S> parent()
@ -565,6 +571,14 @@ public class AccordCache implements CacheSize
return Type.this;
}
public Iterable<K> keysBetween(K start, boolean startInclusive, K end, boolean endInclusive)
{
if (orderedKeys == null)
orderedKeys = new OrderedKeys<>(adapter.keyComparator(), cache.keySet());
return orderedKeys.between(start, startInclusive, end, endInclusive);
}
@Override
public Iterator<AccordCacheEntry<K, V>> iterator()
{
@ -602,11 +616,6 @@ public class AccordCache implements CacheSize
return cache.get(key);
}
public Set<K> keySet()
{
return cache.keySet();
}
@VisibleForTesting
public boolean isReferenced(K key)
{
@ -1039,6 +1048,12 @@ public class AccordCache implements CacheSize
{
return newNode.apply(key, owner);
}
@Override
public Comparator<K> keyComparator()
{
return Comparator.comparing(a -> ((Comparable) a));
}
}
static class SettableWrapper<K, V, S> extends FunctionalAdapter<K, V, S>
@ -1148,6 +1163,12 @@ public class AccordCache implements CacheSize
{
return new AccordSafeCommandsForKey(node);
}
@Override
public Comparator<RoutingKey> keyComparator()
{
return RoutingKey::compareAsRoutingKey;
}
}
public static class CommandAdapter implements Adapter<TxnId, Command, AccordSafeCommand>

13
src/java/org/apache/cassandra/service/accord/AccordCommandStore.java
View File

@ -55,6 +55,7 @@ import accord.local.RedundantBefore;
import accord.local.SafeCommandStore;
import accord.local.cfk.CommandsForKey;
import accord.primitives.PartialTxn;
import accord.primitives.Range;
import accord.primitives.Ranges;
import accord.primitives.RoutableKey;
import accord.primitives.Route;
@ -490,12 +491,16 @@ public class AccordCommandStore extends CommandStore
Ready ready = new Ready();
try (ExclusiveCaches caches = lockCaches())
{
for (AccordCacheEntry<RoutingKey, CommandsForKey> e : caches.commandsForKeys())
for (Range range : ranges)
{
if (ranges.contains(e.key()) && e.isModified())
for (RoutingKey k : caches.commandsForKeys().keysBetween(range.start(), range.startInclusive(), range.end(), range.endInclusive()))
{
ready.increment();
caches.global().saveWhenReadyExclusive(e, ready);
AccordCacheEntry<RoutingKey, CommandsForKey> e = caches.commandsForKeys().getUnsafe(k);
if (e.isModified())
{
ready.increment();
caches.global().saveWhenReadyExclusive(e, ready);
}
}
}
}

164
src/java/org/apache/cassandra/service/accord/AccordService.java
View File

@ -38,13 +38,16 @@ import javax.annotation.concurrent.GuardedBy;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.primitives.Ints;
import accord.local.Catchup;
import accord.topology.ActiveEpochs;
import accord.topology.EpochReady;
import accord.primitives.Txn;
import org.apache.cassandra.config.AccordSpec;
import org.apache.cassandra.db.SystemKeyspace;
import org.apache.cassandra.db.SystemKeyspace.BootstrapState;
import org.apache.cassandra.metrics.AccordReplicaMetrics;
import org.apache.cassandra.metrics.AccordSystemMetrics;
import org.apache.cassandra.service.accord.api.AccordViolationHandler;
import org.apache.cassandra.utils.Clock;
import org.apache.cassandra.utils.concurrent.AsyncFuture;
import org.apache.cassandra.utils.concurrent.ImmediateFuture;
import org.slf4j.Logger;
@ -142,6 +145,7 @@ import static org.apache.cassandra.config.DatabaseDescriptor.getAccordShardDurab
import static org.apache.cassandra.config.DatabaseDescriptor.getAccordShardDurabilityMaxSplits;
import static org.apache.cassandra.config.DatabaseDescriptor.getAccordShardDurabilityTargetSplits;
import static org.apache.cassandra.config.DatabaseDescriptor.getPartitioner;
import static org.apache.cassandra.db.SystemKeyspace.BootstrapState.COMPLETED;
import static org.apache.cassandra.journal.Params.ReplayMode.RESET;
import static org.apache.cassandra.metrics.ClientRequestsMetricsHolder.accordReadBookkeeping;
import static org.apache.cassandra.metrics.ClientRequestsMetricsHolder.accordWriteBookkeeping;
@ -288,30 +292,7 @@ public class AccordService implements IAccordService, Shutdownable
AccordService as = new AccordService(AccordTopology.tcmIdToAccord(tcmId));
unsafeInstance = as;
as.node.unsafeSetReplaying(true);
try
{
as.startup();
replayJournal(as);
}
finally
{
as.node.unsafeSetReplaying(false);
}
as.finishInitialization();
as.fastPathCoordinator.start();
ClusterMetadataService.instance().log().addListener(as.fastPathCoordinator);
as.node.durability().shards().reconfigure(Ints.checkedCast(getAccordShardDurabilityTargetSplits()),
Ints.checkedCast(getAccordShardDurabilityMaxSplits()),
Ints.checkedCast(getAccordShardDurabilityCycle(SECONDS)), SECONDS);
as.node.durability().global().setGlobalCycleTime(Ints.checkedCast(getAccordGlobalDurabilityCycle(SECONDS)), SECONDS);
as.state = State.STARTED;
// Only enable durability scheduling _after_ we have fully replayed journal
as.node.durability().start();
as.startup();
instance = as;
AccordReplicaMetrics.touch();
@ -326,7 +307,7 @@ public class AccordService implements IAccordService, Shutdownable
public static boolean replayJournal(AccordService as)
{
logger.info("Starting journal replay.");
long before = Clock.Global.nanoTime();
long start = nanoTime();
if (as.journalConfiguration().replayMode() == RESET)
AccordKeyspace.truncateCommandsForKey();
@ -337,8 +318,8 @@ public class AccordService implements IAccordService, Shutdownable
as.journal.unsafeSetStarted();
as.node.commandStores().forAllUnsafe(cs -> cs.unsafeProgressLog().start());
long after = Clock.Global.nanoTime();
logger.info("Finished journal replay. {}ms elapsed", NANOSECONDS.toMillis(after - before));
long end = nanoTime();
logger.info("Finished journal replay. {}s elapsed", String.format("%.2f", NANOSECONDS.toMillis(end - start)/1000.0));
return true;
}
@ -414,28 +395,116 @@ public class AccordService implements IAccordService, Shutdownable
{
if (state != State.INIT)
return;
journal.start(node);
node.load();
ClusterMetadata metadata = ClusterMetadata.current();
endpointMapper.updateMapping(metadata);
List<TopologyUpdate> images = journal.replayTopologies();
if (!images.isEmpty())
node.unsafeSetReplaying(true);
try
{
// Initialise command stores using latest topology from the log;
// if there are no local command stores, don't report any topologies and simply fetch the latest known in the cluster
// this avoids a registered (not joined) node learning of topologies, then later restarting with some intervening
// epochs having been garbage collected by the other nodes in the cluster
TopologyUpdate last = images.get(images.size() - 1);
if (!last.commandStores.isEmpty())
{
node.commandStores().initializeTopologyUnsafe(last);
journal.start(node);
node.load();
// Replay local epochs
for (TopologyUpdate image : images)
node.topology().reportTopology(image.global);
ClusterMetadata metadata = ClusterMetadata.current();
endpointMapper.updateMapping(metadata);
List<TopologyUpdate> images = journal.replayTopologies();
if (!images.isEmpty())
{
// Initialise command stores using latest topology from the log;
// if there are no local command stores, don't report any topologies and simply fetch the latest known in the cluster
// this avoids a registered (not joined) node learning of topologies, then later restarting with some intervening
// epochs having been garbage collected by the other nodes in the cluster
TopologyUpdate last = images.get(images.size() - 1);
if (!last.commandStores.isEmpty())
{
node.commandStores().initializeTopologyUnsafe(last);
// Replay local epochs
for (TopologyUpdate image : images)
node.topology().reportTopology(image.global);
}
}
replayJournal(this);
}
finally
{
node.unsafeSetReplaying(false);
}
finishInitialization();
catchup();
fastPathCoordinator.start();
ClusterMetadataService.instance().log().addListener(fastPathCoordinator);
node.durability().shards().reconfigure(Ints.checkedCast(getAccordShardDurabilityTargetSplits()),
Ints.checkedCast(getAccordShardDurabilityMaxSplits()),
Ints.checkedCast(getAccordShardDurabilityCycle(SECONDS)), SECONDS);
node.durability().global().setGlobalCycleTime(Ints.checkedCast(getAccordGlobalDurabilityCycle(SECONDS)), SECONDS);
// Only enable durability scheduling _after_ we have fully replayed journal
node.durability().start();
state = State.STARTED;
}
void catchup()
{
AccordSpec spec = DatabaseDescriptor.getAccord();
if (!spec.catchup_on_start)
{
logger.info("Not catching up with peers");
return;
}
BootstrapState bootstrapState = SystemKeyspace.getBootstrapState();
if (bootstrapState == COMPLETED)
{
long maxLatencyNanos = spec.catchup_on_start_fail_latency.toNanoseconds();
int attempts = 1;
while (true)
{
logger.info("Catching up with quorum...");
long start = nanoTime();
long failAt = start + maxLatencyNanos;
Future<Void> f = toFuture(Catchup.catchup(node));
if (!f.awaitUntilThrowUncheckedOnInterrupt(failAt))
{
if (spec.catchup_on_start_exit_on_failure)
{
logger.error("Catch up exceeded maximum latency of {}ns; shutting down", maxLatencyNanos);
throw new RuntimeException("Could not catch up with peers");
}
logger.error("Catch up exceeded maximum latency of {}ns; starting up", maxLatencyNanos);
break;
}
Throwable failed = f.cause();
if (failed != null)
throw new RuntimeException("Could not catch up with peers", failed);
long end = nanoTime();
double seconds = NANOSECONDS.toMillis(end - start)/1000.0;
logger.info("Finished catching up with all quorums. {}s elapsed.", String.format("%.2f", seconds));
// TODO (expected): make configurable
if (seconds <= spec.catchup_on_start_success_latency.toSeconds())
break;
if (++attempts > spec.catchup_on_start_max_attempts)
{
if (spec.catchup_on_start_exit_on_failure)
{
logger.error("Catch up was slow, aborting after {} attempts and shutting down", attempts);
throw new RuntimeException("Could not catch up with peers");
}
logger.info("Catch up was slow; continuing to startup after {} attempts.", attempts - 1);
break;
}
logger.info("Catch up was slow, so we may behind again; retrying");
}
}
else
{
logger.info("Not catching up with quorum, as bootstrap state is {}", bootstrapState);
}
}
@ -444,8 +513,7 @@ public class AccordService implements IAccordService, Shutdownable
* the latest epoch known to the node prior to restart. After that, we replay journal itself, and only after
* that we finish initializaiton and replay the rest of epochs.
*/
@VisibleForTesting
public void finishInitialization()
void finishInitialization()
{
endpointMapper.updateMapping(ClusterMetadata.current());
TopologyManager topology = node.topology();

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

@ -1001,9 +1001,9 @@ public abstract class AccordTask<R> extends SubmittableTask implements Function<
void startInternal(Caches caches)
{
for (RoutingKey key : caches.commandsForKeys().keySet())
for (Range range : ranges)
{
if (ranges.contains(key))
for (RoutingKey key : caches.commandsForKeys().keysBetween(range.start(), range.startInclusive(), range.end(), range.endInclusive()))
intersectingKeys.add((TokenKey) key);
}
caches.commandsForKeys().register(keyWatcher);

193
src/java/org/apache/cassandra/service/accord/OrderedKeys.java Normal file
View File

@ -0,0 +1,193 @@
/*
* 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;
import java.util.Arrays;
import java.util.Collection;
import java.util.Comparator;
import java.util.Iterator;
import org.apache.cassandra.utils.BulkIterator;
import org.apache.cassandra.utils.btree.BTree;
import org.apache.cassandra.utils.btree.UpdateFunction;
/**
* Maintains an ordered set of keys backed by a BTree, buffering additions and removals for batch updates.
*/
public class OrderedKeys<K> implements Iterable<K>
{
private final Comparator<K> comparator;
private Object[] map;
private final Object[] buffer = new Object[32];
private int bufferRemovals; // Bit field: 1 = removal, 0 = addition
private int bufferCount;
public OrderedKeys(Comparator<K> comparator)
{
this.comparator = comparator;
this.map = BTree.empty();
}
public OrderedKeys(Comparator<K> comparator, Collection<K> init)
{
this.comparator = comparator;
BTree.Builder<K> builder = BTree.builder(comparator);
for (K k : init)
builder.add(k);
this.map = builder.build();
}
public void add(K key, int removalBit)
{
if (bufferCount == buffer.length)
flush();
bufferRemovals |= removalBit << bufferCount;
buffer[bufferCount++] = key;
}
public void add(K key)
{
add(key, 0);
}
public void remove(K key)
{
add(key, 1);
}
/**
* Flushes all buffered additions and removals to the underlying BTree.
*/
public void flush()
{
if (bufferCount == 0)
return;
// Partition the buffer: additions first, then removals
int additionEnd = 0, removalStart = bufferCount;
while (additionEnd != removalStart)
{
if (0 == (bufferRemovals & (1 << additionEnd))) ++additionEnd;
else if (0 != (bufferRemovals & (1 << (removalStart - 1)))) --removalStart;
else
{
Object tmp = buffer[additionEnd];
buffer[additionEnd++] = buffer[--removalStart];
buffer[removalStart] = tmp;
}
}
// Sort both partitions
Arrays.sort(buffer, 0, removalStart, (Comparator) comparator);
Arrays.sort(buffer, removalStart, bufferCount, (Comparator) comparator);
int i = 0, j = removalStart;
// Merge additions and removals
additionEnd = 0;
int removalEnd = removalStart;
while (i < removalStart && j < bufferCount)
{
int c = comparator.compare((K) buffer[i], (K) buffer[j]);
if (c == 0)
{
// matched pairs can be ignored
++i; ++j;
}
else if (c < 0)
{
// Addition is smaller
if (additionEnd != i)
buffer[additionEnd] = buffer[i];
++additionEnd;
++i;
}
else
{
// Removal is smaller
if (removalEnd != j)
buffer[removalEnd] = buffer[j];
++removalEnd;
++j;
}
}
// Copy remaining additions
if (i != removalStart)
{
if (additionEnd == i) additionEnd = removalStart;
else
{
int count = removalStart - i;
System.arraycopy(buffer, i, buffer, additionEnd, count);
additionEnd += count;
}
}
// Copy remaining removals
if (j != bufferCount)
{
if (removalEnd == j) removalEnd = bufferCount;
else
{
int count = bufferCount - j;
System.arraycopy(buffer, j, buffer, removalEnd, count);
removalEnd += count;
}
}
// Apply additions to BTree
if (additionEnd != 0)
map = BTree.update(map, BTree.build(BulkIterator.of(buffer, 0), additionEnd, UpdateFunction.noOp()), comparator);
// Apply removals from BTree
if (removalEnd != removalStart)
map = BTree.subtract(map, BTree.build(BulkIterator.of(buffer, removalStart), removalEnd - removalStart, UpdateFunction.noOp()), comparator);
// Clear buffer
Arrays.fill(buffer, 0, bufferCount, null);
bufferRemovals = 0;
bufferCount = 0;
}
public Iterable<K> between(K start, boolean startInclusive, K end, boolean endInclusive)
{
return () -> {
flush();
return BTree.slice(map, comparator, start, startInclusive, end, endInclusive, BTree.Dir.ASC);
};
}
public Iterator<K> iterator()
{
flush();
return BTree.iterator(map);
}
public int size()
{
flush();
return BTree.size(map);
}
public int bufferSize()
{
return bufferCount;
}
}

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

@ -52,7 +52,7 @@ public class GetDurableBeforeSerializers
@Override
public void serialize(DurableBeforeReply msg, DataOutputPlus out) throws IOException
{
CommandStoreSerializers.durableBefore.serialize(msg.durableBeforeMap, out);
CommandStoreSerializers.durableBefore.serialize(msg.durableBefore, out);
}
@Override
@ -64,7 +64,7 @@ public class GetDurableBeforeSerializers
@Override
public long serializedSize(DurableBeforeReply msg)
{
return CommandStoreSerializers.durableBefore.serializedSize(msg.durableBeforeMap);
return CommandStoreSerializers.durableBefore.serializedSize(msg.durableBefore);
}
};
}

31
src/java/org/apache/cassandra/service/accord/serializers/SetDurableSerializers.java
View File

@ -22,8 +22,9 @@ import java.io.IOException;
import accord.messages.SetGloballyDurable;
import accord.messages.SetShardDurable;
import accord.primitives.Deps;
import accord.primitives.FullRoute;
import accord.primitives.SyncPoint;
import accord.primitives.MinimalSyncPoint;
import accord.primitives.RangeRoute;
import accord.primitives.Route;
import accord.primitives.Timestamp;
import accord.primitives.TxnId;
import org.apache.cassandra.io.UnversionedSerializer;
@ -38,7 +39,7 @@ public class SetDurableSerializers
@Override
public void serialize(SetShardDurable msg, DataOutputPlus out) throws IOException
{
syncPoint.serialize(msg.exclusiveSyncPoint, out);
syncPoint.serialize(msg.syncPoint, out);
CommandSerializers.outcomeDurability.serialize(msg.durability, out);
}
@ -52,7 +53,7 @@ public class SetDurableSerializers
@Override
public long serializedSize(SetShardDurable msg)
{
return syncPoint.serializedSize(msg.exclusiveSyncPoint)
return syncPoint.serializedSize(msg.syncPoint)
+ CommandSerializers.outcomeDurability.serializedSize(msg.durability);
}
};
@ -78,34 +79,34 @@ public class SetDurableSerializers
}
};
public static final UnversionedSerializer<SyncPoint> syncPoint = new UnversionedSerializer<>()
public static final UnversionedSerializer<MinimalSyncPoint> syncPoint = new UnversionedSerializer<>()
{
@Override
public void serialize(SyncPoint sp, DataOutputPlus out) throws IOException
public void serialize(MinimalSyncPoint sp, DataOutputPlus out) throws IOException
{
CommandSerializers.txnId.serialize(sp.syncId, out);
ExecuteAtSerializer.serialize(sp.syncId, sp.executeAt, out);
DepsSerializers.deps.serialize(sp.waitFor, out);
KeySerializers.fullRoute.serialize(sp.route, out);
DepsSerializers.deps.serialize(Deps.NONE, out);
KeySerializers.route.serialize(sp.route, out);
}
@Override
public SyncPoint deserialize(DataInputPlus in) throws IOException
public MinimalSyncPoint deserialize(DataInputPlus in) throws IOException
{
TxnId syncId = CommandSerializers.txnId.deserialize(in);
Timestamp executeAt = ExecuteAtSerializer.deserialize(syncId, in);
Deps waitFor = DepsSerializers.deps.deserialize(in);
FullRoute<?> route = KeySerializers.fullRoute.deserialize(in);
return SyncPoint.SerializationSupport.construct(syncId, executeAt, waitFor, route);
DepsSerializers.deps.deserialize(in);
Route<?> route = KeySerializers.route.deserialize(in);
return MinimalSyncPoint.SerializationSupport.construct(syncId, executeAt, (RangeRoute) route);
}
@Override
public long serializedSize(SyncPoint sp)
public long serializedSize(MinimalSyncPoint sp)
{
return CommandSerializers.txnId.serializedSize(sp.syncId)
+ ExecuteAtSerializer.serializedSize(sp.syncId, sp.executeAt)
+ DepsSerializers.deps.serializedSize(sp.waitFor)
+ KeySerializers.fullRoute.serializedSize(sp.route);
+ DepsSerializers.deps.serializedSize(Deps.NONE)
+ KeySerializers.route.serializedSize(sp.route);
}
};
}

27
src/java/org/apache/cassandra/utils/btree/BTree.java
View File

@ -75,7 +75,7 @@ public class BTree
* _DO NOT_ attempt to modify this field directly. Instead, use BRANCH_SHIFT above instead, as branch factor
* should _always_ be a power of 2.
*/
static final int BRANCH_FACTOR = 1 << BRANCH_SHIFT;
public static final int BRANCH_FACTOR = 1 << BRANCH_SHIFT;
public static final int MIN_KEYS = BRANCH_FACTOR / 2 - 1;
public static final int MAX_KEYS = BRANCH_FACTOR - 1;
public static final long STOP_SENTINEL_VALUE = Long.MAX_VALUE;
@ -538,6 +538,17 @@ public class BTree
}
}
/**
* Subtracts {@code subtract} from {@code update}.
*/
public static <Compare> Object[] subtract(Object[] toUpdate, Object[] subtract, Comparator<Compare> comparator)
{
try (Subtraction subtraction = Subtraction.get(comparator))
{
return subtraction.subtract(toUpdate, subtract);
}
}
public static void reverseInSitu(Object[] tree)
{
reverseInSitu(tree, height(tree), true);
@ -4240,6 +4251,20 @@ public class BTree
}
}
static class Subtraction<K, T extends K> extends AbstractSubtraction<K, T>
{
static final ThreadLocal<Subtraction> SHARED = new ThreadLocal<>();
static <K, T extends K> Subtraction<K, T> get(Comparator<K> comparator)
{
Subtraction subtraction = SHARED.get();
if (subtraction == null)
SHARED.set(subtraction = new Subtraction());
subtraction.comparator = comparator;
return subtraction;
}
}
private static abstract class AbstractTransformer<I, O> extends AbstractSeekingTransformer<I, O>
{
@Override

10
test/distributed/org/apache/cassandra/distributed/test/accord/AccordJournalIntegrationTest.java
View File

@ -47,6 +47,7 @@ public class AccordJournalIntegrationTest extends TestBaseImpl
{
try (WithProperties wp = new WithProperties().set(CassandraRelevantProperties.DTEST_ACCORD_JOURNAL_SANITY_CHECK_ENABLED, "true");
Cluster cluster = init(Cluster.build(1)
.withConfig(config -> config.set("accord.catchup_on_start", "false"))
.withoutVNodes()
.start()))
{
@ -95,6 +96,7 @@ public class AccordJournalIntegrationTest extends TestBaseImpl
{
try (Cluster cluster = Cluster.build(1)
.withoutVNodes()
.withConfig(config -> config.set("accord.catchup_on_start", "false"))
.start())
{
cluster.schemaChange("CREATE KEYSPACE " + KEYSPACE + " WITH replication = {'class': 'SimpleStrategy', 'replication_factor': 1};");
@ -122,7 +124,13 @@ public class AccordJournalIntegrationTest extends TestBaseImpl
@Test
public void restartWithEpochChanges() throws IOException
{
try (Cluster cluster = Cluster.build(3).withoutVNodes().withConfig(c -> c.with(GOSSIP).with(NETWORK)).start())
try (Cluster cluster = Cluster.build(3)
.withoutVNodes()
.withConfig(c -> {
c.with(GOSSIP).with(NETWORK);
c.set("accord.catchup_on_start", "false");
})
.start())
{
init(cluster);
final String TABLE = createTable(cluster);

5
test/distributed/org/apache/cassandra/distributed/test/accord/journal/AccordJournalReplayTest.java
View File

@ -95,6 +95,7 @@ public class AccordJournalReplayTest extends TestBaseImpl
.set("accord.shard_durability_target_splits", "1")
.set("accord.retry_syncpoint", "1s*attempts")
.set("accord.retry_durability", "1s*attempts")
.set("accord.catchup_on_start", "false")
.with(NETWORK, GOSSIP))
.start())
{
@ -125,8 +126,8 @@ public class AccordJournalReplayTest extends TestBaseImpl
Command command = Command.Executed.executed(txnId, SaveStatus.PreApplied, Status.Durability.NotDurable, StoreParticipants.execute(commandStore.unsafeGetRangesForEpoch(), route, txnId, txnId.epoch()), Ballot.ZERO, txnId, txn.intersecting(route, true), deps.intersecting(route), Ballot.ZERO, waitingOn, writes, ResultSerializers.APPLIED);
commandStore.journal.saveCommand(commandStore.id(), new Journal.CommandUpdate(null, command), () -> {});
SyncPoint<accord.primitives.Range> syncPoint = AccordService.getBlocking(CoordinateSyncPoint.exclusive(node, syncPointId, Ranges.of(key.asRange())));
AccordService.getBlocking(ExecuteSyncPoint.coordinate(node, syncPoint, 1).onQuorum());
SyncPoint syncPoint = AccordService.getBlocking(CoordinateSyncPoint.exclusive(node, syncPointId, Ranges.of(key.asRange())));
AccordService.getBlocking(ExecuteSyncPoint.coordinate(node, syncPoint, 1).onQuorumOrDone());
Keyspace.open("ks").getColumnFamilyStore("tbl").forceBlockingFlush(ColumnFamilyStore.FlushReason.UNIT_TESTS);
Keyspace.open("system_accord").getColumnFamilyStore("commands_for_key").forceBlockingFlush(ColumnFamilyStore.FlushReason.UNIT_TESTS);
spinUntilTrue(() -> {

4
test/distributed/org/apache/cassandra/distributed/test/accord/journal/JournalAccessRouteIndexOnStartupRaceTest.java
View File

@ -64,7 +64,9 @@ public class JournalAccessRouteIndexOnStartupRaceTest extends TestBaseImpl
@Test
public void test() throws IOException
{
try (Cluster cluster = Cluster.build(1).withInstanceInitializer(BBHelper::install).start())
try (Cluster cluster = Cluster.build(1)
.withConfig(config -> config.set("accord.catchup_on_start", "false"))
.withInstanceInitializer(BBHelper::install).start())
{
IInvokableInstance node = cluster.get(1);
node.nodetoolResult("disableautocompaction", ACCORD_KEYSPACE_NAME, JOURNAL).asserts().success();

282
test/unit/org/apache/cassandra/utils/OrderedKeysTest.java Normal file
View File

@ -0,0 +1,282 @@
/*
* 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.utils;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.TreeSet;
import org.junit.Assert;
import org.junit.Test;
import accord.utils.Invariants;
import org.apache.cassandra.service.accord.OrderedKeys;
import org.quicktheories.WithQuickTheories;
import org.quicktheories.core.Gen;
import org.quicktheories.generators.SourceDSL;
import org.quicktheories.impl.Constraint;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertTrue;
public class OrderedKeysTest implements WithQuickTheories
{
private Gen<List<Op>> operationsGen()
{
Gen<Integer> sizeGen = SourceDSL.integers().between(0, 200);
Gen<Integer> keyGen = SourceDSL.integers().between(0, 1000);
Gen<Boolean> opGen = SourceDSL.booleans().all();
return rng -> {
Set<Integer> additions = new HashSet<>();
int size = sizeGen.generate(rng);
List<Op> ops = new ArrayList<>(size);
for (int i = 0; i < size; i++)
{
while (true)
{
if (opGen.generate(rng))
{
int key = keyGen.generate(rng);
// Do not add same key twice
if (additions.contains(key))
continue;
ops.add(new AddOp(key));
additions.add(key);
}
else if (!additions.isEmpty())
{
int idx = 0;
if (additions.size() > 1)
idx = (int) rng.next(Constraint.between(0, additions.size() - 1));
Iterator<Integer> iter = additions.iterator();
Integer key = iter.next();
for (int j = 0; j < idx; j++)
key = iter.next();
ops.add(new RemoveOp(key));
additions.remove(key);
}
break;
}
}
return ops;
};
}
interface Op
{
void apply(OrderedKeys<Integer> orderedKeys, TreeSet<Integer> reference, Map<Integer, Object> cache);
}
static class AddOp implements Op
{
final Integer key;
AddOp(Integer key)
{
this.key = key;
}
@Override
public void apply(OrderedKeys<Integer> orderedKeys, TreeSet<Integer> model, Map<Integer, Object> cache)
{
orderedKeys.add(key);
cache.put(key, key.toString());
model.add(key);
}
@Override
public String toString()
{
return "Add(" + key + ")";
}
}
static class RemoveOp implements Op
{
final Integer key;
RemoveOp(Integer key)
{
this.key = key;
}
@Override
public void apply(OrderedKeys<Integer> orderedKeys, TreeSet<Integer> model, Map<Integer, Object> cache)
{
orderedKeys.remove(key);
Object removed = cache.remove(key);
Assert.assertEquals(removed, key.toString());
Invariants.require(model.remove(key));
}
@Override
public String toString()
{
return "Remove(" + key + ")";
}
}
@Test
public void simpleTest()
{
OrderedKeys<Integer> keys = new OrderedKeys<>(Integer::compareTo);
keys.add(1);
keys.add(5);
keys.add(3);
keys.flush();
assertEquals(3, keys.size());
Iterator<Integer> iter = keys.iterator();
assertEquals(Integer.valueOf(1), iter.next());
assertEquals(Integer.valueOf(3), iter.next());
assertEquals(Integer.valueOf(5), iter.next());
assertFalse(iter.hasNext());
// Remove a key
keys.remove(3);
keys.flush();
assertEquals(2, keys.size());
iter = keys.iterator();
assertEquals(Integer.valueOf(1), iter.next());
assertEquals(Integer.valueOf(5), iter.next());
assertFalse(iter.hasNext());
}
@Test
public void testDuplicates()
{
OrderedKeys<Integer> keys = new OrderedKeys<>(Integer::compareTo);
keys.add(1);
keys.remove(1);
keys.add(1);
keys.flush();
Iterator<Integer> iter = keys.iterator();
assertEquals((Integer) 1, iter.next());
assertFalse(iter.hasNext());
}
@Test
public void testBetween()
{
OrderedKeys<Integer> keys = new OrderedKeys<>(Integer::compareTo);
for (int i = 0; i < 10; i++)
keys.add(i);
keys.flush();
// incl
List<Integer> result = new ArrayList<>();
keys.between(3, true, 7, true).forEach(result::add);
assertEquals(List.of(3, 4, 5, 6, 7), result);
// excl
result.clear();
keys.between(3, false, 7, false).forEach(result::add);
assertEquals(List.of(4, 5, 6), result);
// incl/excl
result.clear();
keys.between(3, true, 7, false).forEach(result::add);
assertEquals(List.of(3, 4, 5, 6), result);
}
@Test
public void testBuffering()
{
OrderedKeys<Integer> keys = new OrderedKeys<>(Integer::compareTo);
for (int i = 0; i < 10; i++)
keys.add(i);
assertTrue("Buffer should have pending operations", keys.bufferSize() == 10);
// After flush, buffer should be empty
keys.flush();
assertEquals(0, keys.bufferSize());
assertEquals(10, keys.size());
}
@Test
public void testFlushOnOverflow()
{
OrderedKeys<Integer> keys = new OrderedKeys<>(Integer::compareTo);
// add more than 32 keys (default buffer size)
for (int i = 0; i < 50; i++)
keys.add(i);
assertTrue("Buffer should have flushed", keys.bufferSize() == (50 - 32));
}
@Test
public void randomOperationsMatchReference()
{
qt().withExamples(1000)
.withShrinkCycles(0)
.withFixedSeed(1)
.forAll(operationsGen())
.checkAssert((operations) -> {
Map<Integer, Object> cache = new HashMap<>();
OrderedKeys<Integer> orderedKeys = new OrderedKeys<>(Integer::compareTo);
TreeSet<Integer> model = new TreeSet<>();
for (int i = 0; i < operations.size(); i++)
{
Op op = operations.get(i);
op.apply(orderedKeys, model, cache);
if (Math.random() < 0.1)
{
orderedKeys.flush();
assertMatchesReference(operations.subList(0, i), orderedKeys, model, cache, "After: " + op);
}
}
orderedKeys.flush();
assertMatchesReference(operations, orderedKeys, model, cache, "Final state");
});
}
private void assertMatchesReference(List<Op> ops, OrderedKeys<Integer> orderedKeys, TreeSet<Integer> model, Map<Integer, Object> cache, String message)
{
List<Integer> actualList = new ArrayList<>();
orderedKeys.iterator().forEachRemaining(actualList::add);
List<Integer> expectedList = new ArrayList<>(model);
if (!expectedList.equals(actualList))
{
throw new AssertionError(String.format("%s\n" +
"Ops: %s\n" +
"Ordered Keys: %s\n" +
"Cache: %s\n" +
"Model: %s", message, ops, actualList, cache, model));
}
}
}