mirror of https://github.com/apache/cassandra
Accord: Fix unit tests and improve burn test stability
patch by Benedict; reviewed by Alex Petrov for CASSANDRA-20113
This commit is contained in:
Benedict Elliott Smith
David Capwell
parent
f2ea674140
commit
ec7c7c9fce
|
|
@ -1 +1 @@
|
|||
Subproject commit a271897790aa3816c3dea2125b1e374b091bc090
|
||||
Subproject commit 61f4ebd21ff34b0807081d9df442236d8c334b52
|
||||
|
|
@ -38,6 +38,7 @@ import accord.api.DataStore;
|
|||
import accord.api.LocalListeners;
|
||||
import accord.api.ProgressLog;
|
||||
import accord.api.RoutingKey;
|
||||
import accord.impl.AbstractSafeCommandStore.CommandStoreCaches;
|
||||
import accord.impl.TimestampsForKey;
|
||||
import accord.local.Cleanup;
|
||||
import accord.local.Command;
|
||||
|
|
@ -119,7 +120,7 @@ public class AccordCommandStore extends CommandStore
|
|||
}
|
||||
}
|
||||
|
||||
public static final class ExclusiveCaches extends Caches implements AutoCloseable
|
||||
public static final class ExclusiveCaches extends Caches implements CommandStoreCaches<AccordSafeCommand, AccordSafeTimestampsForKey, AccordSafeCommandsForKey>
|
||||
{
|
||||
private final Lock lock;
|
||||
|
||||
|
|
@ -129,6 +130,25 @@ public class AccordCommandStore extends CommandStore
|
|||
this.lock = lock;
|
||||
}
|
||||
|
||||
|
||||
@Override
|
||||
public AccordSafeCommand acquireIfLoaded(TxnId txnId)
|
||||
{
|
||||
return commands().acquireIfLoaded(txnId);
|
||||
}
|
||||
|
||||
@Override
|
||||
public AccordSafeCommandsForKey acquireIfLoaded(RoutingKey key)
|
||||
{
|
||||
return commandsForKeys().acquireIfLoaded(key);
|
||||
}
|
||||
|
||||
@Override
|
||||
public AccordSafeTimestampsForKey acquireTfkIfLoaded(RoutingKey key)
|
||||
{
|
||||
return timestampsForKeys().acquireIfLoaded(key);
|
||||
}
|
||||
|
||||
@Override
|
||||
public void close()
|
||||
{
|
||||
|
|
|
|||
|
|
@ -561,7 +561,8 @@ public abstract class AccordExecutor implements CacheSize, AccordCacheEntry.OnLo
|
|||
private <K, V> void onLoadedExclusive(AccordCacheEntry<K, V> loaded, V value, Throwable fail, boolean isForRange)
|
||||
{
|
||||
--activeLoads;
|
||||
--activeRangeLoads;
|
||||
if (isForRange)
|
||||
--activeRangeLoads;
|
||||
|
||||
if (loaded.status() != EVICTED)
|
||||
{
|
||||
|
|
|
|||
|
|
@ -18,9 +18,7 @@
|
|||
|
||||
package org.apache.cassandra.service.accord;
|
||||
|
||||
import java.util.ArrayList;
|
||||
import java.util.Collections;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
import java.util.NavigableMap;
|
||||
import java.util.Set;
|
||||
|
|
@ -31,26 +29,18 @@ import com.google.common.annotations.VisibleForTesting;
|
|||
|
||||
import accord.api.Agent;
|
||||
import accord.api.DataStore;
|
||||
import accord.api.Key;
|
||||
import accord.api.ProgressLog;
|
||||
import accord.api.RoutingKey;
|
||||
import accord.impl.AbstractSafeCommandStore;
|
||||
import accord.impl.CommandsSummary;
|
||||
import accord.impl.SafeTimestampsForKey;
|
||||
import accord.local.CommandStores;
|
||||
import accord.local.CommandStores.RangesForEpoch;
|
||||
import accord.local.KeyHistory;
|
||||
import accord.local.NodeCommandStoreService;
|
||||
import accord.local.PreLoadContext;
|
||||
import accord.local.RedundantBefore;
|
||||
import accord.local.SafeCommandStore;
|
||||
import accord.local.cfk.CommandsForKey;
|
||||
import accord.primitives.AbstractKeys;
|
||||
import accord.primitives.AbstractUnseekableKeys;
|
||||
import accord.primitives.Participants;
|
||||
import accord.primitives.Ranges;
|
||||
import accord.primitives.Routable;
|
||||
import accord.primitives.Routables;
|
||||
import accord.primitives.RoutingKeys;
|
||||
import accord.primitives.Timestamp;
|
||||
import accord.primitives.Txn;
|
||||
import accord.primitives.TxnId;
|
||||
|
|
@ -58,14 +48,11 @@ import accord.primitives.Unseekables;
|
|||
import accord.utils.Invariants;
|
||||
import org.apache.cassandra.service.accord.AccordCommandStore.ExclusiveCaches;
|
||||
|
||||
import static accord.local.KeyHistory.TIMESTAMPS;
|
||||
import static accord.utils.Invariants.illegalArgument;
|
||||
import static accord.utils.Invariants.illegalState;
|
||||
|
||||
public class AccordSafeCommandStore extends SafeCommandStore
|
||||
public class AccordSafeCommandStore extends AbstractSafeCommandStore<AccordSafeCommand, AccordSafeTimestampsForKey, AccordSafeCommandsForKey, AccordCommandStore.ExclusiveCaches>
|
||||
{
|
||||
final AccordTask<?> task;
|
||||
final PreLoadContext context;
|
||||
private final @Nullable CommandsForRanges commandsForRanges;
|
||||
private final AccordCommandStore commandStore;
|
||||
private RangesForEpoch ranges;
|
||||
|
|
@ -75,7 +62,7 @@ public class AccordSafeCommandStore extends SafeCommandStore
|
|||
@Nullable CommandsForRanges commandsForRanges,
|
||||
AccordCommandStore commandStore)
|
||||
{
|
||||
this.context = task.preLoadContext();
|
||||
super(task.preLoadContext());
|
||||
this.task = task;
|
||||
this.commandsForRanges = commandsForRanges;
|
||||
this.commandStore = commandStore;
|
||||
|
|
@ -91,91 +78,6 @@ public class AccordSafeCommandStore extends SafeCommandStore
|
|||
return new AccordSafeCommandStore(operation, commandsForRanges, commandStore);
|
||||
}
|
||||
|
||||
@Override
|
||||
public PreLoadContext canExecute(PreLoadContext with)
|
||||
{
|
||||
if (with.isEmpty()) return with;
|
||||
if (with.keys().domain() == Routable.Domain.Range)
|
||||
return with.isSubsetOf(this.context) ? with : null;
|
||||
|
||||
if (!context().keyHistory().satisfies(with.keyHistory()))
|
||||
return null;
|
||||
|
||||
try (ExclusiveCaches caches = commandStore.tryLockCaches())
|
||||
{
|
||||
if (caches == null)
|
||||
return with.isSubsetOf(this.context) ? with : null;
|
||||
|
||||
for (TxnId txnId : with.txnIds())
|
||||
{
|
||||
if (null != getInternal(txnId))
|
||||
continue;
|
||||
|
||||
AccordSafeCommand safeCommand = caches.commands().acquireIfLoaded(txnId);
|
||||
if (safeCommand == null)
|
||||
return null;
|
||||
|
||||
add(safeCommand, caches);
|
||||
}
|
||||
|
||||
KeyHistory keyHistory = with.keyHistory();
|
||||
if (keyHistory == KeyHistory.NONE)
|
||||
return with;
|
||||
|
||||
List<RoutingKey> unavailable = null;
|
||||
AbstractUnseekableKeys keys = (AbstractUnseekableKeys) context.keys();
|
||||
if (keys.size() == 0)
|
||||
return with;
|
||||
|
||||
for (int i = 0 ; i < keys.size() ; ++i)
|
||||
{
|
||||
RoutingKey key = keys.get(i);
|
||||
if (keyHistory == TIMESTAMPS)
|
||||
{
|
||||
if (null != timestampsForKeyInternal(key))
|
||||
continue; // already in working set
|
||||
|
||||
AccordSafeTimestampsForKey safeTfk = caches.timestampsForKeys().acquireIfLoaded(key);
|
||||
if (safeTfk != null)
|
||||
{
|
||||
add(safeTfk, caches);
|
||||
continue;
|
||||
}
|
||||
}
|
||||
else
|
||||
{
|
||||
if (null != getInternal(key))
|
||||
continue; // already in working set
|
||||
|
||||
AccordSafeCommandsForKey safeCfk = caches.commandsForKeys().acquireIfLoaded(key);
|
||||
if (safeCfk != null)
|
||||
{
|
||||
add(safeCfk, caches);
|
||||
continue;
|
||||
}
|
||||
}
|
||||
if (unavailable == null)
|
||||
unavailable = new ArrayList<>();
|
||||
unavailable.add(key);
|
||||
}
|
||||
|
||||
if (unavailable == null)
|
||||
return with;
|
||||
|
||||
if (unavailable.size() == keys.size())
|
||||
return null;
|
||||
|
||||
Participants<RoutingKey> available = keys.without(RoutingKeys.ofSortedUnique(unavailable));
|
||||
return PreLoadContext.contextFor(context.primaryTxnId(), context.additionalTxnId(), available, keyHistory);
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public PreLoadContext context()
|
||||
{
|
||||
return context;
|
||||
}
|
||||
|
||||
@VisibleForTesting
|
||||
public Set<RoutingKey> commandsForKeysKeys()
|
||||
{
|
||||
|
|
@ -194,22 +96,12 @@ public class AccordSafeCommandStore extends SafeCommandStore
|
|||
}
|
||||
|
||||
@Override
|
||||
protected AccordSafeCommand ifLoadedAndInitialisedAndNotErasedInternal(TxnId txnId)
|
||||
protected ExclusiveCaches tryGetCaches()
|
||||
{
|
||||
try (ExclusiveCaches caches = commandStore.tryLockCaches())
|
||||
{
|
||||
if (caches == null)
|
||||
return null;
|
||||
|
||||
AccordSafeCommand command = caches.commands().acquireIfLoaded(txnId);
|
||||
if (command == null)
|
||||
return null;
|
||||
|
||||
return add(command, caches);
|
||||
}
|
||||
return commandStore.tryLockCaches();
|
||||
}
|
||||
|
||||
private AccordSafeCommand add(AccordSafeCommand safeCommand, ExclusiveCaches caches)
|
||||
protected AccordSafeCommand add(AccordSafeCommand safeCommand, ExclusiveCaches caches)
|
||||
{
|
||||
Object check = task.ensureCommands().putIfAbsent(safeCommand.txnId(), safeCommand);
|
||||
if (check == null)
|
||||
|
|
@ -224,7 +116,7 @@ public class AccordSafeCommandStore extends SafeCommandStore
|
|||
}
|
||||
}
|
||||
|
||||
private AccordSafeCommandsForKey add(AccordSafeCommandsForKey safeCfk, ExclusiveCaches caches)
|
||||
protected AccordSafeCommandsForKey add(AccordSafeCommandsForKey safeCfk, ExclusiveCaches caches)
|
||||
{
|
||||
Object check = task.ensureCommandsForKey().putIfAbsent(safeCfk.key(), safeCfk);
|
||||
if (check == null)
|
||||
|
|
@ -239,7 +131,7 @@ public class AccordSafeCommandStore extends SafeCommandStore
|
|||
}
|
||||
}
|
||||
|
||||
private AccordSafeTimestampsForKey add(AccordSafeTimestampsForKey safeTfk, ExclusiveCaches caches)
|
||||
protected AccordSafeTimestampsForKey add(AccordSafeTimestampsForKey safeTfk, ExclusiveCaches caches)
|
||||
{
|
||||
Object check = task.ensureTimestampsForKey().putIfAbsent(safeTfk.key(), safeTfk);
|
||||
if (check == null)
|
||||
|
|
@ -263,42 +155,7 @@ public class AccordSafeCommandStore extends SafeCommandStore
|
|||
return commandsForKey.get(key);
|
||||
}
|
||||
|
||||
@Override
|
||||
protected AccordSafeCommandsForKey ifLoadedInternal(RoutingKey key)
|
||||
{
|
||||
try (ExclusiveCaches caches = commandStore.tryLockCaches())
|
||||
{
|
||||
if (caches == null)
|
||||
return null;
|
||||
|
||||
AccordSafeCommandsForKey safeCfk = caches.commandsForKeys().acquireIfLoaded(key);
|
||||
if (safeCfk == null)
|
||||
return null;
|
||||
|
||||
Object check = task.ensureCommandsForKey().putIfAbsent(safeCfk.key(), safeCfk);
|
||||
if (check == null)
|
||||
{
|
||||
safeCfk.preExecute();
|
||||
return safeCfk;
|
||||
}
|
||||
else
|
||||
{
|
||||
caches.commandsForKeys().release(safeCfk, task);
|
||||
throw illegalState("Attempted to take a duplicate reference to CFK for %s", key);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public SafeTimestampsForKey timestampsForKey(RoutingKey key)
|
||||
{
|
||||
AccordSafeTimestampsForKey safeTfk = timestampsForKeyInternal(key);
|
||||
if (safeTfk == null)
|
||||
throw illegalArgument("%s not referenced in %s", key, context);
|
||||
return safeTfk;
|
||||
}
|
||||
|
||||
private AccordSafeTimestampsForKey timestampsForKeyInternal(RoutingKey key)
|
||||
protected AccordSafeTimestampsForKey timestampsForKeyInternal(RoutingKey key)
|
||||
{
|
||||
Map<RoutingKey, AccordSafeTimestampsForKey> timestampsForKey = task.timestampsForKey();
|
||||
if (timestampsForKey == null)
|
||||
|
|
|
|||
|
|
@ -1037,7 +1037,7 @@ public class AccordService implements IAccordService, Shutdownable
|
|||
{
|
||||
if (state.knows(blockedBy)) continue;
|
||||
// need to fetch the state
|
||||
if (safeStore.ifLoadedAndInitialisedAndNotErased(blockedBy) != null)
|
||||
if (safeStore.ifLoadedAndInitialised(blockedBy) != null)
|
||||
{
|
||||
AsyncChain<Void> chain = populate(state, safeStore, blockedBy);
|
||||
if (chain != null)
|
||||
|
|
@ -1077,7 +1077,7 @@ public class AccordService implements IAccordService, Shutdownable
|
|||
blocking = ((CommandsForKey.TxnInfo) blocking).plainTxnId();
|
||||
state.keys.put(pk, blocking);
|
||||
if (state.txns.containsKey(blocking)) return null;
|
||||
if (safeStore.ifLoadedAndInitialisedAndNotErased(blocking) != null) return populate(state, safeStore, blocking);
|
||||
if (safeStore.ifLoadedAndInitialised(blocking) != null) return populate(state, safeStore, blocking);
|
||||
return populate(state, safeStore.commandStore(), blocking);
|
||||
}
|
||||
|
||||
|
|
@ -1437,7 +1437,7 @@ public class AccordService implements IAccordService, Shutdownable
|
|||
{
|
||||
if (tracker.recordSuccess(from) == RequestStatus.Success)
|
||||
{
|
||||
node.configService().reportEpochRedundant(exclusiveSyncPoint.route.toRanges(), exclusiveSyncPoint.syncId.epoch());
|
||||
node.configService().reportEpochRedundant(exclusiveSyncPoint.route.toRanges(), exclusiveSyncPoint.syncId.epoch() - 1);
|
||||
trySuccess(exclusiveSyncPoint);
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -57,7 +57,7 @@ import org.apache.cassandra.index.accord.RoutesSearcher;
|
|||
import org.apache.cassandra.service.accord.api.AccordRoutingKey;
|
||||
|
||||
import static accord.local.SafeCommandStore.TestDep.ANY_DEPS;
|
||||
import static accord.local.SafeCommandStore.TestDep.WITH;
|
||||
import static accord.local.SafeCommandStore.TestDep.WITH_OR_INVALIDATED;
|
||||
import static accord.local.SafeCommandStore.TestStartedAt.STARTED_BEFORE;
|
||||
import static accord.local.SafeCommandStore.TestStatus.ANY_STATUS;
|
||||
import static accord.primitives.Routables.Slice.Minimal;
|
||||
|
|
@ -129,6 +129,9 @@ public class CommandsForRanges extends TreeMap<Timestamp, CommandsForRanges.Summ
|
|||
case IS_STABLE:
|
||||
if (!summary.saveStatus.hasBeen(Stable) || summary.saveStatus.hasBeen(Truncated))
|
||||
return;
|
||||
case IS_STABLE_OR_INVALIDATED:
|
||||
if (!summary.saveStatus.hasBeen(Stable) || summary.saveStatus.status == Truncated)
|
||||
return;
|
||||
}
|
||||
|
||||
if (testDep != ANY_DEPS)
|
||||
|
|
@ -152,7 +155,7 @@ public class CommandsForRanges extends TreeMap<Timestamp, CommandsForRanges.Summ
|
|||
// and so it is safe to execute, when in fact it is only a dependency on a different shard
|
||||
// (and that other shard, perhaps, does not know that it is a dependency - and so it is not durably known)
|
||||
// TODO (required): consider this some more
|
||||
if ((testDep == WITH) == !summary.hasAsDep)
|
||||
if ((testDep == WITH_OR_INVALIDATED) == (!summary.hasAsDep || summary.saveStatus == SaveStatus.Invalidated))
|
||||
return;
|
||||
|
||||
Invariants.checkState(testTxnId.equals(summary.findAsDep));
|
||||
|
|
|
|||
|
|
@ -33,7 +33,7 @@ import accord.local.Node;
|
|||
import accord.messages.Apply;
|
||||
import accord.primitives.Deps;
|
||||
import accord.primitives.FullRoute;
|
||||
import accord.primitives.Participants;
|
||||
import accord.primitives.Route;
|
||||
import accord.primitives.Timestamp;
|
||||
import accord.primitives.Txn;
|
||||
import accord.primitives.TxnId;
|
||||
|
|
@ -89,14 +89,13 @@ public class AccordInteropAdapter extends AbstractTxnAdapter
|
|||
}
|
||||
|
||||
@Override
|
||||
public void persist(Node node, Topologies all, FullRoute<?> route, Participants<?> sendTo, TxnId txnId, Txn txn, Timestamp executeAt, Deps deps, Writes writes, Result result, BiConsumer<? super Result, Throwable> callback)
|
||||
public void persist(Node node, Topologies all, FullRoute<?> fullRoute, Route<?> sendTo, TxnId txnId, Txn txn, Timestamp executeAt, Deps deps, Writes writes, Result result, BiConsumer<? super Result, Throwable> callback)
|
||||
{
|
||||
// TODO (required): we aren't using sendTo
|
||||
if (applyKind == Minimal && doInteropPersist(node, all, route, txnId, txn, executeAt, deps, writes, result, callback))
|
||||
if (applyKind == Minimal && doInteropPersist(node, all, sendTo, txnId, txn, executeAt, deps, writes, result, fullRoute, callback))
|
||||
return;
|
||||
|
||||
if (callback != null) callback.accept(result, null);
|
||||
new PersistTxn(node, all, txnId, route, txn, executeAt, deps, writes, result)
|
||||
new PersistTxn(node, all, txnId, sendTo, txn, executeAt, deps, writes, result, fullRoute)
|
||||
.start(Apply.FACTORY, applyKind, all, writes, result);
|
||||
}
|
||||
|
||||
|
|
@ -113,14 +112,14 @@ public class AccordInteropAdapter extends AbstractTxnAdapter
|
|||
return true;
|
||||
}
|
||||
|
||||
private static boolean doInteropPersist(Node node, Topologies all, FullRoute<?> route, TxnId txnId, Txn txn, Timestamp executeAt, Deps deps, Writes writes, Result result, BiConsumer<? super Result, Throwable> callback)
|
||||
private static boolean doInteropPersist(Node node, Topologies all, Route<?> sendTo, TxnId txnId, Txn txn, Timestamp executeAt, Deps deps, Writes writes, Result result, FullRoute<?> fullRoute, BiConsumer<? super Result, Throwable> callback)
|
||||
{
|
||||
Update update = txn.update();
|
||||
ConsistencyLevel consistencyLevel = update instanceof AccordUpdate ? ((AccordUpdate) update).cassandraCommitCL() : null;
|
||||
if (consistencyLevel == null || consistencyLevel == ConsistencyLevel.ANY || writes.isEmpty())
|
||||
return false;
|
||||
|
||||
new AccordInteropPersist(node, all, txnId, route, txn, executeAt, deps, writes, result, consistencyLevel, callback)
|
||||
new AccordInteropPersist(node, all, txnId, sendTo, txn, executeAt, deps, writes, result, fullRoute, consistencyLevel, callback)
|
||||
.start(AccordInteropApply.FACTORY, Minimal, all, writes, result);
|
||||
return true;
|
||||
}
|
||||
|
|
|
|||
|
|
@ -62,14 +62,14 @@ public class AccordInteropApply extends Apply implements LocalListeners.ComplexL
|
|||
public static final Apply.Factory FACTORY = new Apply.Factory()
|
||||
{
|
||||
@Override
|
||||
public Apply create(Kind kind, Id to, Topologies participates, TxnId txnId, FullRoute<?> route, Txn txn, Timestamp executeAt, Deps deps, Writes writes, Result result)
|
||||
public Apply create(Kind kind, Id to, Topologies participates, TxnId txnId, Route<?> route, Txn txn, Timestamp executeAt, Deps deps, Writes writes, Result result, FullRoute<?> fullRoute)
|
||||
{
|
||||
checkArgument(kind != Kind.Maximal, "Shouldn't need to send a maximal commit with interop support");
|
||||
ConsistencyLevel commitCL = txn.update() instanceof AccordUpdate ? ((AccordUpdate) txn.update()).cassandraCommitCL() : null;
|
||||
// Any asynchronous apply option should use the regular Apply that doesn't wait for writes to complete
|
||||
if (commitCL == null || commitCL == ConsistencyLevel.ANY)
|
||||
return Apply.FACTORY.create(kind, to, participates, txnId, route, txn, executeAt, deps, writes, result);
|
||||
return new AccordInteropApply(kind, to, participates, txnId, route, txn, executeAt, deps, writes, result);
|
||||
return Apply.FACTORY.create(kind, to, participates, txnId, route, txn, executeAt, deps, writes, result, fullRoute);
|
||||
return new AccordInteropApply(kind, to, participates, txnId, route, txn, executeAt, deps, writes, result, fullRoute);
|
||||
}
|
||||
};
|
||||
|
||||
|
|
@ -91,9 +91,9 @@ public class AccordInteropApply extends Apply implements LocalListeners.ComplexL
|
|||
super(kind, txnId, route, waitForEpoch, executeAt, deps, txn, fullRoute, writes, result);
|
||||
}
|
||||
|
||||
private AccordInteropApply(Kind kind, Id to, Topologies participates, TxnId txnId, FullRoute<?> route, Txn txn, Timestamp executeAt, Deps deps, Writes writes, Result result)
|
||||
private AccordInteropApply(Kind kind, Id to, Topologies participates, TxnId txnId, Route<?> route, Txn txn, Timestamp executeAt, Deps deps, Writes writes, Result result, FullRoute<?> fullRoute)
|
||||
{
|
||||
super(kind, to, participates, txnId, route, txn, executeAt, deps, writes, result);
|
||||
super(kind, to, participates, txnId, route, txn, executeAt, deps, writes, result, fullRoute);
|
||||
}
|
||||
|
||||
@Override
|
||||
|
|
|
|||
|
|
@ -30,6 +30,7 @@ import accord.local.Node;
|
|||
import accord.messages.Apply;
|
||||
import accord.primitives.Deps;
|
||||
import accord.primitives.FullRoute;
|
||||
import accord.primitives.Route;
|
||||
import accord.primitives.Timestamp;
|
||||
import accord.primitives.Txn;
|
||||
import accord.primitives.TxnId;
|
||||
|
|
@ -91,9 +92,9 @@ public class AccordInteropPersist extends Persist
|
|||
private final ConsistencyLevel consistencyLevel;
|
||||
private CallbackHolder callback;
|
||||
|
||||
public AccordInteropPersist(Node node, Topologies topologies, TxnId txnId, FullRoute<?> route, Txn txn, Timestamp executeAt, Deps deps, Writes writes, Result result, ConsistencyLevel consistencyLevel, BiConsumer<? super Result, Throwable> clientCallback)
|
||||
public AccordInteropPersist(Node node, Topologies topologies, TxnId txnId, Route<?> sendTo, Txn txn, Timestamp executeAt, Deps deps, Writes writes, Result result, FullRoute<?> fullRoute, ConsistencyLevel consistencyLevel, BiConsumer<? super Result, Throwable> clientCallback)
|
||||
{
|
||||
super(node, topologies, txnId, route, txn, executeAt, deps, writes, result);
|
||||
super(node, topologies, txnId, sendTo, txn, executeAt, deps, writes, result, fullRoute);
|
||||
Invariants.checkArgument(consistencyLevel == ConsistencyLevel.QUORUM || consistencyLevel == ConsistencyLevel.ALL || consistencyLevel == ConsistencyLevel.SERIAL || consistencyLevel == ConsistencyLevel.ONE);
|
||||
this.consistencyLevel = consistencyLevel;
|
||||
registerClientCallback(result, clientCallback);
|
||||
|
|
|
|||
|
|
@ -45,7 +45,8 @@ public class AwaitSerializer
|
|||
CommandSerializers.txnId.serialize(await.txnId, out, version);
|
||||
KeySerializers.participants.serialize(await.scope, out, version);
|
||||
out.writeByte(await.blockedUntil.ordinal());
|
||||
out.writeUnsignedVInt(await.awaitEpoch - await.txnId.epoch());
|
||||
out.writeUnsignedVInt(await.maxAwaitEpoch - await.txnId.epoch());
|
||||
out.writeUnsignedVInt(await.maxAwaitEpoch - await.minAwaitEpoch);
|
||||
out.writeUnsignedVInt32(await.callbackId + 1);
|
||||
Invariants.checkState(await.callbackId >= -1);
|
||||
}
|
||||
|
|
@ -56,10 +57,11 @@ public class AwaitSerializer
|
|||
TxnId txnId = CommandSerializers.txnId.deserialize(in, version);
|
||||
Participants<?> scope = KeySerializers.participants.deserialize(in, version);
|
||||
BlockedUntil blockedUntil = BlockedUntil.forOrdinal(in.readByte());
|
||||
long awaitEpoch = in.readUnsignedVInt() + txnId.epoch();
|
||||
long maxAwaitEpoch = in.readUnsignedVInt() + txnId.epoch();
|
||||
long minAwaitEpoch = maxAwaitEpoch - in.readUnsignedVInt();
|
||||
int callbackId = in.readUnsignedVInt32() - 1;
|
||||
Invariants.checkState(callbackId >= -1);
|
||||
return Await.SerializerSupport.create(txnId, scope, blockedUntil, awaitEpoch, callbackId);
|
||||
return Await.SerializerSupport.create(txnId, scope, blockedUntil, minAwaitEpoch, maxAwaitEpoch, callbackId);
|
||||
}
|
||||
|
||||
@Override
|
||||
|
|
@ -68,7 +70,8 @@ public class AwaitSerializer
|
|||
return CommandSerializers.txnId.serializedSize(await.txnId, version)
|
||||
+ KeySerializers.participants.serializedSize(await.scope, version)
|
||||
+ TypeSizes.BYTE_SIZE
|
||||
+ VIntCoding.computeUnsignedVIntSize(await.awaitEpoch - await.txnId.epoch())
|
||||
+ VIntCoding.computeUnsignedVIntSize(await.maxAwaitEpoch - await.txnId.epoch())
|
||||
+ VIntCoding.computeUnsignedVIntSize(await.maxAwaitEpoch - await.minAwaitEpoch)
|
||||
+ VIntCoding.computeUnsignedVIntSize(await.callbackId + 1);
|
||||
}
|
||||
};
|
||||
|
|
|
|||
|
|
@ -80,6 +80,7 @@ public class ReadDataSerializers
|
|||
{
|
||||
CommandSerializers.txnId.serialize(msg.txnId, out, version);
|
||||
KeySerializers.participants.serialize(msg.readScope, out, version);
|
||||
out.writeUnsignedVInt(msg.minEpoch());
|
||||
CommandSerializers.timestamp.serialize(msg.executeAt, out, version);
|
||||
KeySerializers.fullRoute.serialize(msg.route, out, version);
|
||||
CommandSerializers.partialTxn.serialize(msg.txn, out, version);
|
||||
|
|
@ -93,6 +94,7 @@ public class ReadDataSerializers
|
|||
return ApplyThenWaitUntilApplied.SerializerSupport.create(
|
||||
CommandSerializers.txnId.deserialize(in, version),
|
||||
KeySerializers.participants.deserialize(in, version),
|
||||
in.readUnsignedVInt(),
|
||||
CommandSerializers.timestamp.deserialize(in, version),
|
||||
KeySerializers.fullRoute.deserialize(in, version),
|
||||
CommandSerializers.partialTxn.deserialize(in, version),
|
||||
|
|
@ -106,6 +108,7 @@ public class ReadDataSerializers
|
|||
{
|
||||
return CommandSerializers.txnId.serializedSize(msg.txnId, version)
|
||||
+ KeySerializers.participants.serializedSize(msg.readScope, version)
|
||||
+ TypeSizes.sizeofUnsignedVInt(msg.minEpoch())
|
||||
+ CommandSerializers.timestamp.serializedSize(msg.executeAt, version)
|
||||
+ KeySerializers.fullRoute.serializedSize(msg.route, version)
|
||||
+ CommandSerializers.partialTxn.serializedSize(msg.txn, version)
|
||||
|
|
@ -279,7 +282,8 @@ public class ReadDataSerializers
|
|||
{
|
||||
CommandSerializers.txnId.serialize(waitUntilApplied.txnId, out, version);
|
||||
KeySerializers.participants.serialize(waitUntilApplied.readScope, out, version);
|
||||
out.writeUnsignedVInt(waitUntilApplied.executeAtEpoch);
|
||||
out.writeUnsignedVInt(waitUntilApplied.minEpoch());
|
||||
out.writeUnsignedVInt(waitUntilApplied.executeAtEpoch - waitUntilApplied.minEpoch());
|
||||
}
|
||||
|
||||
@Override
|
||||
|
|
@ -287,8 +291,9 @@ public class ReadDataSerializers
|
|||
{
|
||||
TxnId txnId = CommandSerializers.txnId.deserialize(in, version);
|
||||
Participants<?> readScope = KeySerializers.participants.deserialize(in, version);
|
||||
long executeAtEpoch = in.readUnsignedVInt();
|
||||
return WaitUntilApplied.SerializerSupport.create(txnId, readScope, executeAtEpoch);
|
||||
long minEpoch = in.readUnsignedVInt();
|
||||
long executeAtEpoch = minEpoch + in.readUnsignedVInt();
|
||||
return WaitUntilApplied.SerializerSupport.create(txnId, readScope, minEpoch, executeAtEpoch);
|
||||
}
|
||||
|
||||
@Override
|
||||
|
|
@ -296,7 +301,8 @@ public class ReadDataSerializers
|
|||
{
|
||||
return CommandSerializers.txnId.serializedSize(waitUntilApplied.txnId, version)
|
||||
+ KeySerializers.participants.serializedSize(waitUntilApplied.readScope, version)
|
||||
+ TypeSizes.sizeofUnsignedVInt(waitUntilApplied.executeAtEpoch);
|
||||
+ TypeSizes.sizeofUnsignedVInt(waitUntilApplied.minEpoch())
|
||||
+ TypeSizes.sizeofUnsignedVInt(waitUntilApplied.executeAtEpoch - waitUntilApplied.minEpoch());
|
||||
}
|
||||
};
|
||||
}
|
||||
|
|
|
|||
|
|
@ -207,8 +207,8 @@ public class CompactionAccordIteratorsTest
|
|||
testAccordCommandsPurger(redundantBefore(GT_TXN_ID), durableBefore(UNIVERSAL), expectAccordCommandsErase());
|
||||
// Universally durable but not locally; we're stale, but shouldn't erase
|
||||
testAccordCommandsPurger(redundantBefore(LT_TXN_ID), durableBefore(UNIVERSAL), expectAccordCommandsNoChange());
|
||||
// With redundantBefore at the txnId there should be no change because it is < not <=
|
||||
testAccordCommandsPurger(redundantBefore(TXN_ID), durableBefore(MAJORITY), expectAccordCommandsNoChange());
|
||||
// redundantBefore will be > txnId as must be exclusive sync point, so will truncate
|
||||
testAccordCommandsPurger(redundantBefore(TXN_ID), durableBefore(MAJORITY), expectAccordCommandsTruncated());
|
||||
testAccordCommandsPurger(redundantBefore(LT_TXN_ID), durableBefore(MAJORITY), expectAccordCommandsNoChange());
|
||||
// Durable at a majority can be truncated with minimal data preserved, it must be redundant for this to occur
|
||||
testAccordCommandsPurger(redundantBefore(GT_TXN_ID), durableBefore(MAJORITY), expectAccordCommandsTruncated());
|
||||
|
|
@ -249,10 +249,11 @@ public class CompactionAccordIteratorsTest
|
|||
testAccordCommandsForKeyPurger(null, expectedAccordCommandsForKeyNoChange());
|
||||
testAccordTimestampsForKeyPurger(redundantBefore(LT_TXN_ID), expectedAccordTimestampsForKeyNoChange());
|
||||
testAccordCommandsForKeyPurger(redundantBefore(LT_TXN_ID), expectedAccordCommandsForKeyNoChange());
|
||||
testAccordTimestampsForKeyPurger(redundantBefore(TXN_ID), expectedAccordTimestampsForKeyNoChange());
|
||||
testAccordCommandsForKeyPurger(redundantBefore(TXN_ID), expectedAccordCommandsForKeyNoChange());
|
||||
testAccordTimestampsForKeyPurger(redundantBefore(GT_TXN_ID), expectedAccordTimestampsForKeyEraseOne());
|
||||
testAccordCommandsForKeyPurger(redundantBefore(GT_TXN_ID), expectedAccordCommandsForKeyEraseOne());
|
||||
// will erase one more than expected as converted to ExclusiveSyncPoint id which is > base id
|
||||
testAccordTimestampsForKeyPurger(redundantBefore(TXN_ID), expectedAccordTimestampsForKeyEraseOne());
|
||||
testAccordCommandsForKeyPurger(redundantBefore(TXN_ID), expectedAccordCommandsForKeyEraseOne());
|
||||
testAccordTimestampsForKeyPurger(redundantBefore(GT_TXN_ID), expectedAccordTimestampsForKeyEraseAll());
|
||||
testAccordCommandsForKeyPurger(redundantBefore(GT_TXN_ID), expectedAccordCommandsForKeyEraseAll());
|
||||
testAccordTimestampsForKeyPurger(redundantBefore(GT_SECOND_TXN_ID), expectedAccordTimestampsForKeyEraseAll());
|
||||
testAccordCommandsForKeyPurger(redundantBefore(GT_SECOND_TXN_ID), expectedAccordCommandsForKeyEraseAll());
|
||||
}
|
||||
|
|
@ -399,7 +400,7 @@ public class CompactionAccordIteratorsTest
|
|||
private static RedundantBefore redundantBefore(TxnId txnId)
|
||||
{
|
||||
Ranges ranges = AccordTestUtils.fullRange(AccordTestUtils.keys(table, 42));
|
||||
txnId = txnId.as(Kind.Read, Range);
|
||||
txnId = txnId.as(Kind.ExclusiveSyncPoint, Range);
|
||||
return RedundantBefore.create(ranges, Long.MIN_VALUE, Long.MAX_VALUE, txnId, txnId, txnId, txnId, LT_TXN_ID.as(Range));
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -133,7 +133,7 @@ public class AccordTaskTest
|
|||
// TODO review: This change to `ifInitialized` was done in a lot of places and it doesn't preserve this property
|
||||
// I fixed this reference to point to `ifLoadedAndInitialised` and but didn't update other places
|
||||
Assert.assertNull(instance.ifInitialised(txnId));
|
||||
Assert.assertNull(instance.ifLoadedAndInitialisedAndNotErased(txnId));
|
||||
Assert.assertNull(instance.ifLoadedAndInitialised(txnId));
|
||||
}));
|
||||
|
||||
UntypedResultSet result = AccordKeyspace.loadCommandRow(commandStore, txnId);
|
||||
|
|
|
|||
|
|
@ -23,6 +23,7 @@ import java.util.ArrayList;
|
|||
import java.util.Arrays;
|
||||
import java.util.Collections;
|
||||
import java.util.Comparator;
|
||||
import java.util.EnumSet;
|
||||
import java.util.List;
|
||||
import java.util.Random;
|
||||
import java.util.Set;
|
||||
|
|
@ -202,6 +203,8 @@ public class CommandsForKeySerializerTest
|
|||
|
||||
case Erased:
|
||||
case ErasedOrVestigial:
|
||||
return Command.Truncated.erased(txnId, Status.Durability.UniversalOrInvalidated, StoreParticipants.empty(txnId));
|
||||
|
||||
case Invalidated:
|
||||
return Command.SerializerSupport.invalidated(txnId);
|
||||
}
|
||||
|
|
@ -431,7 +434,8 @@ public class CommandsForKeySerializerTest
|
|||
}
|
||||
}
|
||||
|
||||
Choices<SaveStatus> saveStatusChoices = Choices.uniform(SaveStatus.values());
|
||||
// TODO (expected): we currently don't explore TruncatedApply statuses because we don't transition through all phases and therefore don't adopt the Applied status
|
||||
Choices<SaveStatus> saveStatusChoices = Choices.uniform(EnumSet.complementOf(EnumSet.of(SaveStatus.TruncatedApply, SaveStatus.TruncatedApplyWithOutcome, SaveStatus.TruncatedApplyWithDeps)).toArray(SaveStatus[]::new));
|
||||
Supplier<SaveStatus> saveStatusSupplier = () -> {
|
||||
SaveStatus result = saveStatusChoices.choose(source);
|
||||
while (result == SaveStatus.TruncatedApplyWithDeps) // not a real save status
|
||||
|
|
@ -471,7 +475,8 @@ public class CommandsForKeySerializerTest
|
|||
while (commands.size() > 0)
|
||||
{
|
||||
int next = source.nextInt(commands.size());
|
||||
cfk = cfk.update(commands.get(next)).cfk();
|
||||
Command command = commands.get(next);
|
||||
cfk = cfk.update(command).cfk();
|
||||
commands.set(next, commands.get(commands.size() - 1));
|
||||
commands.remove(commands.size() - 1);
|
||||
}
|
||||
|
|
@ -486,7 +491,7 @@ public class CommandsForKeySerializerTest
|
|||
}
|
||||
TxnInfo info = cfk.get(i);
|
||||
InternalStatus expectStatus = InternalStatus.from(cmd.saveStatus);
|
||||
if (expectStatus == null) expectStatus = InternalStatus.TRANSITIVELY_KNOWN;
|
||||
if (expectStatus == null) expectStatus = InternalStatus.TRANSITIVE;
|
||||
if (expectStatus.hasExecuteAtOrDeps)
|
||||
Assert.assertEquals(cmd.executeAt, info.executeAt);
|
||||
Assert.assertEquals(expectStatus, info.status());
|
||||
|
|
@ -545,7 +550,7 @@ public class CommandsForKeySerializerTest
|
|||
{
|
||||
int idx = Arrays.binarySearch(ids, u.txnId);
|
||||
if (idx < 0)
|
||||
missing.add(TxnInfo.create(u.txnId, InternalStatus.TRANSITIVELY_KNOWN, true, u.txnId, Ballot.ZERO));
|
||||
missing.add(TxnInfo.create(u.txnId, InternalStatus.TRANSITIVE, true, u.txnId, Ballot.ZERO));
|
||||
}
|
||||
if (!missing.isEmpty())
|
||||
{
|
||||
|
|
|
|||
|
|
@ -443,7 +443,7 @@ public class AccordGenerators
|
|||
|
||||
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));
|
||||
return redundantBeforeEntry(Gens.bools().all(), range(partitioner), AccordGens.txnIds(Gens.pick(Txn.Kind.ExclusiveSyncPoint), ignore -> Routable.Domain.Range));
|
||||
}
|
||||
|
||||
public static Gen<RedundantBefore.Entry> redundantBeforeEntry(Gen<Boolean> emptyGen, Gen<Range> rangeGen, Gen<TxnId> txnIdGen)
|
||||
|
|
@ -451,11 +451,11 @@ public class AccordGenerators
|
|||
return rs -> {
|
||||
Range range = rangeGen.next(rs);
|
||||
TxnId locallyWitnessedOrInvalidatedBefore = emptyGen.next(rs) ? TxnId.NONE : txnIdGen.next(rs); // emptyable or range
|
||||
TxnId locallyAppliedOrInvalidatedBefore = emptyGen.next(rs) ? TxnId.NONE : txnIdGen.next(rs); // emptyable or range
|
||||
TxnId locallyDecidedAndAppliedOrInvalidatedBefore = locallyAppliedOrInvalidatedBefore;
|
||||
TxnId shardAppliedOrInvalidatedBefore = emptyGen.next(rs) ? TxnId.NONE : txnIdGen.next(rs); // emptyable or range
|
||||
TxnId shardOnlyAppliedOrInvalidatedBefore = shardAppliedOrInvalidatedBefore;
|
||||
TxnId gcBefore = shardAppliedOrInvalidatedBefore;
|
||||
TxnId locallyAppliedOrInvalidatedBefore = TxnId.nonNullOrMin(locallyWitnessedOrInvalidatedBefore, emptyGen.next(rs) ? TxnId.NONE : txnIdGen.next(rs)); // emptyable or range
|
||||
TxnId locallyDecidedAndAppliedOrInvalidatedBefore = TxnId.nonNullOrMin(locallyAppliedOrInvalidatedBefore, emptyGen.next(rs) ? TxnId.NONE : txnIdGen.next(rs)); // emptyable or range
|
||||
TxnId shardOnlyAppliedOrInvalidatedBefore = emptyGen.next(rs) ? TxnId.NONE : txnIdGen.next(rs); // emptyable or range
|
||||
TxnId shardAppliedOrInvalidatedBefore = TxnId.nonNullOrMin(locallyAppliedOrInvalidatedBefore, TxnId.nonNullOrMin(shardOnlyAppliedOrInvalidatedBefore, emptyGen.next(rs) ? TxnId.NONE : txnIdGen.next(rs))); // emptyable or range
|
||||
TxnId gcBefore = TxnId.nonNullOrMin(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
|
||||
|
||||
|
|
@ -469,7 +469,7 @@ public class AccordGenerators
|
|||
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);
|
||||
Gen<TxnId> txnIdGen = AccordGens.txnIds(Gens.pick(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);
|
||||
}
|
||||
|
|
@ -477,7 +477,7 @@ public class AccordGenerators
|
|||
public static Gen<DurableBefore> durableBeforeGen(IPartitioner partitioner)
|
||||
{
|
||||
Gen<Ranges> rangeGen = rangesArbitrary(partitioner);
|
||||
Gen<TxnId> txnIdGen = AccordGens.txnIds(Gens.pick(Txn.Kind.SyncPoint, Txn.Kind.ExclusiveSyncPoint), ignore -> Routable.Domain.Range);
|
||||
Gen<TxnId> txnIdGen = AccordGens.txnIds(Gens.pick(Txn.Kind.ExclusiveSyncPoint), ignore -> Routable.Domain.Range);
|
||||
|
||||
return (rs) -> {
|
||||
Ranges ranges = rangeGen.next(rs);
|
||||
|
|
|
|||
Loading…
Reference in New Issue