You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@cassandra.apache.org by bd...@apache.org on 2023/11/30 18:33:31 UTC
(cassandra-accord) branch trunk updated: Reduce command deps
This is an automated email from the ASF dual-hosted git repository.
bdeggleston pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/cassandra-accord.git
The following commit(s) were added to refs/heads/trunk by this push:
new d9ef5553 Reduce command deps
d9ef5553 is described below
commit d9ef555302f8774ed03325ba22d38ee0b80130a8
Author: Blake Eggleston <bl...@ultrablake.com>
AuthorDate: Mon Nov 27 14:18:04 2023 -0800
Reduce command deps
Patch by Blake Eggleston; Reviewed by Benedict Elliott Smith for CASSANDRA-18784
---
.../VisibleForImplementationTesting.java} | 16 +-
.../src/main/java/accord/coordinate/Barrier.java | 12 +-
.../java/accord/impl/AbstractSafeCommandStore.java | 177 +++++++--
.../main/java/accord/impl/CommandTimeseries.java | 329 +++++++++++++++-
.../src/main/java/accord/impl/CommandsForKey.java | 149 ++-----
.../accord/impl/CommandsForKeyGroupUpdater.java | 227 +++++++++++
.../java/accord/impl/CommandsForKeyUpdater.java | 117 ++++++
.../src/main/java/accord/impl/CommandsForKeys.java | 152 ++++++++
.../src/main/java/accord/impl/DepsBuilder.java | 176 +++++++++
...ndTimeseriesHolder.java => DomainCommands.java} | 9 +-
...TimeseriesHolder.java => DomainTimestamps.java} | 7 +-
.../java/accord/impl/InMemoryCommandStore.java | 431 +++++++++++++++++----
.../impl/InMemorySafeCommandsForKeyUpdate.java | 60 +++
...lder.java => InMemorySafeTimestampsForKey.java} | 39 +-
.../main/java/accord/impl/SafeCommandsForKey.java | 160 +-------
.../java/accord/impl/SafeTimestampsForKey.java | 87 +++++
.../main/java/accord/impl/TimestampsForKey.java | 150 +++++++
.../src/main/java/accord/local/Bootstrap.java | 2 +-
.../src/main/java/accord/local/CommandStore.java | 12 +-
.../src/main/java/accord/local/Commands.java | 2 +-
.../KeyHistory.java} | 20 +-
.../src/main/java/accord/local/PreLoadContext.java | 27 +-
.../main/java/accord/local/SafeCommandStore.java | 15 +-
.../src/main/java/accord/messages/Accept.java | 7 +
.../main/java/accord/messages/BeginRecovery.java | 34 +-
.../src/main/java/accord/messages/CheckStatus.java | 2 +-
.../src/main/java/accord/messages/GetDeps.java | 7 +
.../src/main/java/accord/messages/PreAccept.java | 46 +--
.../src/main/java/accord/primitives/Timestamp.java | 5 +
.../impl/CommandsForKeyGroupUpdaterTest.java | 180 +++++++++
.../src/test/java/accord/impl/list/ListWrite.java | 11 +-
.../test/java/accord/messages/PreAcceptTest.java | 14 +-
.../java/accord/primitives/DepsBuilderTest.java | 161 ++++++++
33 files changed, 2320 insertions(+), 523 deletions(-)
diff --git a/accord-core/src/main/java/accord/impl/CommandTimeseriesHolder.java b/accord-core/src/main/java/accord/api/VisibleForImplementationTesting.java
similarity index 73%
copy from accord-core/src/main/java/accord/impl/CommandTimeseriesHolder.java
copy to accord-core/src/main/java/accord/api/VisibleForImplementationTesting.java
index f0a88e75..01146825 100644
--- a/accord-core/src/main/java/accord/impl/CommandTimeseriesHolder.java
+++ b/accord-core/src/main/java/accord/api/VisibleForImplementationTesting.java
@@ -16,14 +16,16 @@
* limitations under the License.
*/
-package accord.impl;
+package accord.api;
-import accord.primitives.Timestamp;
+import java.lang.annotation.Documented;
+import java.lang.annotation.Inherited;
+import java.lang.annotation.Retention;
+import java.lang.annotation.RetentionPolicy;
-public interface CommandTimeseriesHolder
+@Documented
+@Inherited
+@Retention(RetentionPolicy.SOURCE)
+public @interface VisibleForImplementationTesting
{
- CommandTimeseries<?> byId();
- CommandTimeseries<?> byExecuteAt();
-
- Timestamp max();
}
diff --git a/accord-core/src/main/java/accord/coordinate/Barrier.java b/accord-core/src/main/java/accord/coordinate/Barrier.java
index 84885476..a1327f5a 100644
--- a/accord-core/src/main/java/accord/coordinate/Barrier.java
+++ b/accord-core/src/main/java/accord/coordinate/Barrier.java
@@ -22,6 +22,7 @@ import java.util.Objects;
import java.util.concurrent.ExecutionException;
import javax.annotation.Nonnull;
+import accord.local.*;
import com.google.common.annotations.VisibleForTesting;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -29,14 +30,8 @@ import org.slf4j.LoggerFactory;
import accord.api.BarrierType;
import accord.api.Key;
import accord.api.RoutingKey;
-import accord.local.Command;
-import accord.local.Node;
-import accord.local.PreLoadContext;
-import accord.local.SafeCommand;
-import accord.local.SafeCommandStore;
import accord.local.SafeCommandStore.TestDep;
import accord.local.SafeCommandStore.TestTimestamp;
-import accord.local.Status;
import accord.primitives.Routable.Domain;
import accord.primitives.Seekables;
import accord.primitives.Timestamp;
@@ -206,7 +201,7 @@ public class Barrier<S extends Seekables<?, ?>> extends AsyncResults.AbstractRes
ExistingTransactionCheck check = new ExistingTransactionCheck();
Key k = seekables.get(0).asKey();
node.commandStores().mapReduceConsume(
- contextFor(k),
+ contextFor(k, KeyHistory.ALL),
k.toUnseekable(),
minEpoch,
Long.MAX_VALUE,
@@ -256,6 +251,7 @@ public class Barrier<S extends Seekables<?, ?>> extends AsyncResults.AbstractRes
BarrierTxn found = safeStore.mapReduce(
seekables,
safeStore.ranges().allAfter(minEpoch),
+ KeyHistory.ALL,
// Barriers are trying to establish that committed transactions are applied before the barrier (or in this case just minEpoch)
// so all existing transaction types should ensure that at this point. An earlier txnid may have an executeAt that is after
// this barrier or the transaction we listen on and that is fine
@@ -266,7 +262,7 @@ public class Barrier<S extends Seekables<?, ?>> extends AsyncResults.AbstractRes
null,
Status.Committed,
Status.Applied,
- (p1, keyOrRange, txnId, executeAt, status, barrierTxn) -> {
+ (p1, keyOrRange, txnId, executeAt, status, deps, barrierTxn) -> {
if (keyOrRange.domain() == Domain.Key)
return new BarrierTxn(txnId, executeAt, status, keyOrRange.asKey());
return null;
diff --git a/accord-core/src/main/java/accord/impl/AbstractSafeCommandStore.java b/accord-core/src/main/java/accord/impl/AbstractSafeCommandStore.java
index aa053409..0965f46a 100644
--- a/accord-core/src/main/java/accord/impl/AbstractSafeCommandStore.java
+++ b/accord-core/src/main/java/accord/impl/AbstractSafeCommandStore.java
@@ -25,22 +25,17 @@ import java.util.Map;
import java.util.function.Consumer;
import java.util.function.Function;
+import accord.api.Key;
import accord.api.VisibleForImplementation;
import accord.impl.CommandTimeseries.CommandLoader;
-import accord.local.Command;
-import accord.local.CommonAttributes;
-import accord.local.PreLoadContext;
-import accord.local.RedundantBefore;
-import accord.local.SafeCommand;
-import accord.local.SafeCommandStore;
-import accord.primitives.Ranges;
-import accord.primitives.RoutableKey;
-import accord.primitives.Seekable;
-import accord.primitives.Seekables;
-import accord.primitives.TxnId;
+import accord.local.*;
+import accord.primitives.*;
import accord.utils.Invariants;
-public abstract class AbstractSafeCommandStore<CommandType extends SafeCommand, CommandsForKeyType extends SafeCommandsForKey> extends SafeCommandStore
+public abstract class AbstractSafeCommandStore<CommandType extends SafeCommand,
+ TimestampsForKeyType extends SafeTimestampsForKey,
+ CommandsForKeyType extends SafeCommandsForKey,
+ CommandsForKeyUpdateType extends SafeCommandsForKey.Update> extends SafeCommandStore
{
private static class PendingRegistration<T>
{
@@ -65,14 +60,6 @@ public abstract class AbstractSafeCommandStore<CommandType extends SafeCommand,
this.context = context;
}
- protected abstract CommandType getCommandInternal(TxnId txnId);
- protected abstract void addCommandInternal(CommandType command);
-
- protected abstract CommandsForKeyType getCommandsForKeyInternal(RoutableKey key);
- protected abstract void addCommandsForKeyInternal(CommandsForKeyType cfk);
-
- protected abstract CommandType getIfLoaded(TxnId txnId);
-
private static <K, V> V getIfLoaded(K key, Function<K, V> get, Consumer<V> add, Function<K, V> getIfLoaded)
{
V value = get.apply(key);
@@ -86,6 +73,27 @@ public abstract class AbstractSafeCommandStore<CommandType extends SafeCommand,
return value;
}
+ protected abstract CommandType getCommandInternal(TxnId txnId);
+ protected abstract void addCommandInternal(CommandType command);
+ protected abstract CommandType getIfLoaded(TxnId txnId);
+
+ protected abstract TimestampsForKeyType getTimestampsForKeyInternal(RoutableKey key);
+ protected abstract void addTimestampsForKeyInternal(TimestampsForKeyType cfk);
+ protected abstract TimestampsForKeyType getTimestampsForKeyIfLoaded(RoutableKey key);
+
+ protected abstract CommandLoader<?> cfkLoader(RoutableKey key);
+ protected abstract CommandsForKeyType getDepsCommandsForKeyInternal(RoutableKey key);
+ protected abstract void addDepsCommandsForKeyInternal(CommandsForKeyType cfk);
+ protected abstract CommandsForKeyType getDepsCommandsForKeyIfLoaded(RoutableKey key);
+
+ protected abstract CommandsForKeyType getAllCommandsForKeyInternal(RoutableKey key);
+ protected abstract void addAllCommandsForKeyInternal(CommandsForKeyType cfk);
+ protected abstract CommandsForKeyType getAllCommandsForKeyIfLoaded(RoutableKey key);
+
+ protected abstract CommandsForKeyUpdateType getCommandsForKeyUpdateInternal(RoutableKey key);
+ protected abstract CommandsForKeyUpdateType createCommandsForKeyUpdateInternal(RoutableKey key);
+ protected abstract void addCommandsForKeyUpdateInternal(CommandsForKeyUpdateType update);
+
@Override
protected CommandType getInternalIfLoadedAndInitialised(TxnId txnId)
{
@@ -106,11 +114,22 @@ public abstract class AbstractSafeCommandStore<CommandType extends SafeCommand,
return command;
}
- protected abstract CommandLoader<?> cfkLoader(RoutableKey key);
+ private CommandsForKeyType getCommandsIfLoaded(RoutableKey key, KeyHistory keyHistory)
+ {
+ switch (keyHistory)
+ {
+ case DEPS:
+ return getIfLoaded(key, this::getDepsCommandsForKeyInternal, this::addDepsCommandsForKeyInternal, this::getDepsCommandsForKeyIfLoaded);
+ case ALL:
+ return getIfLoaded(key, this::getAllCommandsForKeyInternal, this::addAllCommandsForKeyInternal, this::getAllCommandsForKeyIfLoaded);
+ default:
+ throw new IllegalArgumentException("CommandsForKey not available for " + keyHistory);
+ }
+ }
- public CommandsForKeyType ifLoadedAndInitialised(RoutableKey key)
+ private CommandsForKeyType commandsIfLoadedAndInitialised(RoutableKey key, KeyHistory keyHistory)
{
- CommandsForKeyType cfk = getIfLoaded(key, this::getCommandsForKeyInternal, this::addCommandsForKeyInternal, this::getIfLoaded);
+ CommandsForKeyType cfk = getCommandsIfLoaded(key, keyHistory);;
if (cfk == null)
return null;
if (cfk.isEmpty())
@@ -126,26 +145,124 @@ public abstract class AbstractSafeCommandStore<CommandType extends SafeCommand,
return cfk;
}
- public CommandsForKeyType commandsForKey(RoutableKey key)
+ protected CommandsForKeyType commandsForKey(RoutableKey key, KeyHistory keyHistory)
{
- CommandsForKeyType cfk = getCommandsForKeyInternal(key);
+ CommandsForKeyType cfk = getCommandsIfLoaded(key, keyHistory);
Invariants.checkState(cfk != null, "%s was not specified in PreLoadContext", key);
if (cfk.isEmpty())
cfk.initialize(cfkLoader(key));
return cfk;
}
- protected abstract CommandsForKeyType getIfLoaded(RoutableKey key);
-
@VisibleForImplementation
- public CommandsForKeyType maybeCommandsForKey(RoutableKey key)
+ private CommandsForKeyType maybeCommandsForKey(RoutableKey key, KeyHistory keyHistory)
{
- CommandsForKeyType cfk = getIfLoaded(key, this::getCommandsForKeyInternal, this::addCommandsForKeyInternal, this::getIfLoaded);
+ CommandsForKeyType cfk = getCommandsIfLoaded(key, keyHistory);
if (cfk == null || cfk.isEmpty())
return null;
return cfk;
}
+ public CommandsForKeyType depsCommandsIfLoadedAndInitialised(RoutableKey key)
+ {
+ return commandsIfLoadedAndInitialised(key, KeyHistory.DEPS);
+ }
+
+ public CommandsForKeyType depsCommandsForKey(RoutableKey key)
+ {
+ return commandsForKey(key, KeyHistory.DEPS);
+ }
+
+ @VisibleForImplementation
+ public CommandsForKeyType maybeDepsCommandsForKey(RoutableKey key)
+ {
+ return maybeCommandsForKey(key, KeyHistory.DEPS);
+ }
+
+ public CommandsForKeyType allCommandsIfLoadedAndInitialised(RoutableKey key)
+ {
+ return commandsIfLoadedAndInitialised(key, KeyHistory.ALL);
+ }
+
+ public CommandsForKeyType allCommandsForKey(RoutableKey key)
+ {
+ return commandsForKey(key, KeyHistory.ALL);
+ }
+
+ @VisibleForImplementation
+ public CommandsForKeyType maybeAllCommandsForKey(RoutableKey key)
+ {
+ return maybeCommandsForKey(key, KeyHistory.ALL);
+ }
+
+ public TimestampsForKeyType timestampsIfLoadedAndInitialised(RoutableKey key)
+ {
+ TimestampsForKeyType cfk = getIfLoaded(key, this::getTimestampsForKeyInternal, this::addTimestampsForKeyInternal, this::getTimestampsForKeyIfLoaded);
+ if (cfk == null)
+ return null;
+ if (cfk.isEmpty())
+ {
+ cfk.initialize();
+ }
+ return cfk;
+ }
+
+ public TimestampsForKeyType timestampsForKey(RoutableKey key)
+ {
+ TimestampsForKeyType tfk = getIfLoaded(key, this::getTimestampsForKeyInternal, this::addTimestampsForKeyInternal, this::getTimestampsForKeyIfLoaded);
+ Invariants.checkState(tfk != null, "%s was not specified in PreLoadContext", key);
+ if (tfk.isEmpty())
+ tfk.initialize();
+ return tfk;
+ }
+
+
+ @VisibleForImplementation
+ public TimestampsForKeyType maybeTimestampsForKey(RoutableKey key)
+ {
+ TimestampsForKeyType tfk = getIfLoaded(key, this::getTimestampsForKeyInternal, this::addTimestampsForKeyInternal, this::getTimestampsForKeyIfLoaded);
+ if (tfk == null || tfk.isEmpty())
+ return null;
+ return tfk;
+ }
+
+ public CommandsForKeyUpdateType getOrCreateCommandsForKeyUpdate(RoutableKey key)
+ {
+ CommandsForKeyUpdateType update = getIfLoaded(key, this::getCommandsForKeyUpdateInternal, this::addCommandsForKeyUpdateInternal, this::createCommandsForKeyUpdateInternal);
+ if (update == null)
+ {
+ update = createCommandsForKeyUpdateInternal(key);
+ addCommandsForKeyUpdateInternal(update);
+ }
+
+ if (update.isEmpty())
+ update.initialize();
+
+ return update;
+ }
+
+ @Override
+ public void removeCommandFromSeekableDeps(Seekable seekable, TxnId txnId, Timestamp executeAt, Status status)
+ {
+ // the cfk listener doesn't know if it can remove the given command from the deps set without loading
+ // the deps set, so we don't actually remove it until it becomes applied
+ if (!status.hasBeen(Status.Applied))
+ return;
+
+ switch (seekable.domain())
+ {
+ case Key:
+ Key key = seekable.asKey();
+ CommandsForKeyUpdater.Mutable<?> updater = getOrCreateCommandsForKeyUpdate(key).deps();
+ updater.commands().remove(txnId);
+ break;
+ case Range:
+ break;
+ default:
+ throw new IllegalArgumentException();
+ }
+ }
+
@Override
public boolean canExecuteWith(PreLoadContext context)
{
@@ -195,6 +312,7 @@ public abstract class AbstractSafeCommandStore<CommandType extends SafeCommand,
}
protected abstract void invalidateSafeState();
+ protected abstract void applyCommandForKeyUpdates();
public void postExecute()
{
@@ -205,6 +323,7 @@ public abstract class AbstractSafeCommandStore<CommandType extends SafeCommand,
completeRegistrations(attributeUpdates, pendingSeekableRegistrations, this::completeRegistration);
attributeUpdates.forEach(((txnId, attributes) -> get(txnId).updateAttributes(attributes)));
}
+ applyCommandForKeyUpdates();
}
public void complete()
diff --git a/accord-core/src/main/java/accord/impl/CommandTimeseries.java b/accord-core/src/main/java/accord/impl/CommandTimeseries.java
index abc319b8..4cc50464 100644
--- a/accord-core/src/main/java/accord/impl/CommandTimeseries.java
+++ b/accord-core/src/main/java/accord/impl/CommandTimeseries.java
@@ -18,14 +18,15 @@
package accord.impl;
-import java.util.List;
-import java.util.NavigableMap;
-import java.util.Objects;
-import java.util.TreeMap;
+import java.util.*;
+import java.util.function.BiConsumer;
+import java.util.function.Consumer;
import java.util.function.Predicate;
import java.util.stream.Stream;
import javax.annotation.Nullable;
+import com.google.common.collect.ImmutableMap;
+import com.google.common.collect.ImmutableSet;
import com.google.common.collect.ImmutableSortedMap;
import accord.api.Key;
@@ -46,12 +47,13 @@ import static accord.utils.Utils.ensureSortedMutable;
public class CommandTimeseries<D>
{
public enum TestTimestamp { BEFORE, AFTER }
+ public enum TimestampType { TXN_ID, EXECUTE_AT }
private final Seekable keyOrRange;
protected final CommandLoader<D> loader;
public final ImmutableSortedMap<Timestamp, D> commands;
- public CommandTimeseries(Update<D> builder)
+ public CommandTimeseries(Builder<D> builder)
{
this.keyOrRange = builder.keyOrRange;
this.loader = builder.loader;
@@ -101,12 +103,42 @@ public class CommandTimeseries<D>
public Timestamp maxTimestamp()
{
- return commands.isEmpty() ? Timestamp.NONE : commands.keySet().last();
+ if (commands.isEmpty())
+ return Timestamp.NONE;
+ Timestamp result = null;
+ for (D data : commands.values())
+ {
+ if (result == null)
+ {
+ result = Timestamp.max(loader.txnId(data), loader.executeAt(data));
+ }
+ else
+ {
+ result = Timestamp.max(result, loader.txnId(data));
+ result = Timestamp.max(result, loader.executeAt(data));
+ }
+ }
+ return result;
}
public Timestamp minTimestamp()
{
- return commands.isEmpty() ? Timestamp.NONE : commands.keySet().first();
+ if (commands.isEmpty())
+ return Timestamp.NONE;
+ Timestamp result = null;
+ for (D data : commands.values())
+ {
+ if (result == null)
+ {
+ result = Timestamp.min(loader.txnId(data), loader.executeAt(data));
+ }
+ else
+ {
+ result = Timestamp.min(result, loader.txnId(data));
+ result = Timestamp.min(result, loader.executeAt(data));
+ }
+ }
+ return result;
}
/**
@@ -117,14 +149,31 @@ public class CommandTimeseries<D>
* <p>
* TODO (expected, efficiency): TestDep should be asynchronous; data should not be kept memory-resident as only used for recovery
*/
- public <P1, T> T mapReduce(Kinds testKind, TestTimestamp testTimestamp, Timestamp timestamp,
+ public <P1, T> T mapReduce(Kinds testKind, TimestampType timestampType, TestTimestamp testTimestamp, Timestamp timestamp,
SafeCommandStore.TestDep testDep, @Nullable TxnId depId,
@Nullable Status minStatus, @Nullable Status maxStatus,
SafeCommandStore.CommandFunction<P1, T, T> map, P1 p1, T initialValue, Predicate<? super T> terminatePredicate)
{
+ Iterable<D> dataIterable;
+ Predicate<Timestamp> executeAtPredicate;
+ if (timestampType == TimestampType.TXN_ID)
+ {
+ dataIterable = testTimestamp == TestTimestamp.BEFORE ? commands.headMap(timestamp, false).values() : commands.tailMap(timestamp, false).values();
+ executeAtPredicate = ts -> true;
+ }
+ else
+ {
+ dataIterable = commands.values();
+ executeAtPredicate = testTimestamp == TestTimestamp.BEFORE ? ts -> ts.compareTo(timestamp) < 0 : ts -> ts.compareTo(timestamp) > 0;
+ }
- for (D data : (testTimestamp == TestTimestamp.BEFORE ? commands.headMap(timestamp, false) : commands.tailMap(timestamp, false)).values())
+
+ for (D data : dataIterable)
{
+ Timestamp executeAt = loader.executeAt(data);
+ if (!executeAtPredicate.test(executeAt))
+ continue;
+
TxnId txnId = loader.txnId(data);
if (!testKind.test(txnId.rw())) continue;
SaveStatus status = loader.saveStatus(data);
@@ -136,8 +185,7 @@ public class CommandTimeseries<D>
// If we don't have any dependencies, we treat a dependency filter as a mismatch
if (testDep != ANY_DEPS && (!status.known.deps.hasProposedOrDecidedDeps() || (deps.contains(depId) != (testDep == WITH))))
continue;
- Timestamp executeAt = loader.executeAt(data);
- initialValue = map.apply(p1, keyOrRange, txnId, executeAt, status.status, initialValue);
+ initialValue = map.apply(p1, keyOrRange, txnId, executeAt, status.status, () -> deps, initialValue);
if (terminatePredicate.test(initialValue))
break;
}
@@ -161,9 +209,9 @@ public class CommandTimeseries<D>
return commands.values().stream();
}
- Update<D> beginUpdate()
+ Builder<D> unbuild()
{
- return new Update<>(this);
+ return new Builder<>(this);
}
public CommandLoader<D> loader()
@@ -191,48 +239,48 @@ public class CommandTimeseries<D>
}
}
- public static class Update<D>
+ public static class Builder<D>
{
private final Seekable keyOrRange;
protected CommandLoader<D> loader;
protected NavigableMap<Timestamp, D> commands;
- public Update(Seekable keyOrRange, CommandLoader<D> loader)
+ public Builder(Seekable keyOrRange, CommandLoader<D> loader)
{
this.keyOrRange = keyOrRange;
this.loader = loader;
this.commands = new TreeMap<>();
}
- public Update(CommandTimeseries<D> timeseries)
+ public Builder(CommandTimeseries<D> timeseries)
{
this.keyOrRange = timeseries.keyOrRange;
this.loader = timeseries.loader;
this.commands = timeseries.commands;
}
- public Update<D> add(Timestamp timestamp, Command command)
+ public Builder<D> add(Timestamp timestamp, Command command)
{
commands = ensureSortedMutable(commands);
commands.put(timestamp, loader.saveForCFK(command));
return this;
}
- public Update<D> add(Timestamp timestamp, D value)
+ public Builder<D> add(Timestamp timestamp, D value)
{
commands = ensureSortedMutable(commands);
commands.put(timestamp, value);
return this;
}
- public Update<D> remove(Timestamp timestamp)
+ public Builder<D> remove(Timestamp timestamp)
{
commands = ensureSortedMutable(commands);
commands.remove(timestamp);
return this;
}
- public Update<D> removeBefore(Timestamp timestamp)
+ public Builder<D> removeBefore(Timestamp timestamp)
{
commands = ensureSortedMutable(commands);
commands.headMap(timestamp, false).clear();
@@ -244,4 +292,245 @@ public class CommandTimeseries<D>
return new CommandTimeseries<>(this);
}
}
+
+ public interface Update<T extends Timestamp, D>
+ {
+ interface Mutable<T extends Timestamp, D> extends Update<T, D>
+ {
+ void add(T ts, Command command);
+ void remove(T ts);
+ }
+
+ int numWrites();
+ int numDeletes();
+
+ boolean contains(T key);
+
+ default int numChanges()
+ {
+ return numWrites() + numDeletes();
+ }
+
+ default boolean isEmpty()
+ {
+ return numChanges() == 0;
+ }
+
+ void forEachWrite(BiConsumer<T, D> consumer);
+ void forEachDelete(Consumer<T> consumer);
+
+ default CommandTimeseries<D> apply(CommandTimeseries<D> current)
+ {
+ if (isEmpty())
+ return current;
+
+ CommandTimeseries.Builder<D> builder = current.unbuild();
+ forEachWrite(builder::add);
+ forEachDelete(builder::remove);
+ return builder.build();
+ }
+
+ }
+
+ public static class ImmutableUpdate<T extends Timestamp, D> implements Update<T, D>
+ {
+ private static final ImmutableUpdate<?, ?> EMPTY = new ImmutableUpdate<>(ImmutableMap.of(), ImmutableSet.of());
+
+ public final ImmutableMap<T, D> writes;
+ public final ImmutableSet<T> deletes;
+
+ public ImmutableUpdate(ImmutableMap<T, D> writes, ImmutableSet<T> deletes)
+ {
+ this.writes = writes;
+ this.deletes = deletes;
+ }
+
+ public boolean equals(Object o)
+ {
+ if (this == o) return true;
+ if (o == null || getClass() != o.getClass()) return false;
+ ImmutableUpdate<?, ?> immutable = (ImmutableUpdate<?, ?>) o;
+ return Objects.equals(writes, immutable.writes) && Objects.equals(deletes, immutable.deletes);
+ }
+
+ public int hashCode()
+ {
+ throw new UnsupportedOperationException();
+ }
+
+ public String toString()
+ {
+ return "Update.Immutable{" +
+ "writes=" + writes +
+ ", deletes=" + deletes +
+ '}';
+ }
+
+ public static <T extends Timestamp, D> ImmutableUpdate<T, D> empty()
+ {
+ return (ImmutableUpdate<T, D>) EMPTY;
+ }
+
+ @Override
+ public int numWrites()
+ {
+ return writes.size();
+ }
+
+ @Override
+ public void forEachWrite(BiConsumer<T, D> consumer)
+ {
+ writes.forEach(consumer);
+ }
+
+ @Override
+ public int numDeletes()
+ {
+ return deletes.size();
+ }
+
+ @Override
+ public void forEachDelete(Consumer<T> consumer)
+ {
+ deletes.forEach(consumer);
+ }
+
+ public boolean contains(T key)
+ {
+ return writes.containsKey(key) || deletes.contains(key);
+ }
+
+ public ImmutableUpdate<T, D> apply(Update<T, D> next)
+ {
+ if (next.isEmpty())
+ return this;
+
+ Map<T, D> writes = new HashMap<>(this.writes);
+ Set<T> deletes = new HashSet<>(this.deletes);
+
+ next.forEachDelete(k -> {
+ writes.remove(k);
+ deletes.add(k);
+ });
+
+ next.forEachWrite((k, v) -> {
+ writes.put(k , v);
+ deletes.remove(k);
+ });
+
+ return new ImmutableUpdate<>(ImmutableMap.copyOf(writes), ImmutableSet.copyOf(deletes));
+ }
+
+ public MutableUpdate<T, D> toMutable(CommandLoader<D> loader)
+ {
+ return new MutableUpdate<>(loader, new HashMap<>(writes), new HashSet<>(deletes));
+ }
+ }
+
+ public static class MutableUpdate<T extends Timestamp, D> implements Update.Mutable<T, D>
+ {
+ private final CommandLoader<D> loader;
+ private final Map<T, D> writes;
+ private final Set<T> deletes;
+
+ public MutableUpdate(CommandLoader<D> loader, Map<T, D> writes, Set<T> deletes)
+ {
+ this.loader = loader;
+ this.writes = writes;
+ this.deletes = deletes;
+ }
+
+ public MutableUpdate(CommandLoader<D> loader)
+ {
+ this(loader, new HashMap<>(), new HashSet<>());
+ }
+
+ @Override
+ public void add(T ts, Command command)
+ {
+ writes.put(ts, loader.saveForCFK(command));
+ deletes.remove(ts);
+ }
+
+ @Override
+ public void remove(T ts)
+ {
+ deletes.add(ts);
+ writes.remove(ts);
+ }
+
+ @Override
+ public int numWrites()
+ {
+ return writes.size();
+ }
+
+ @Override
+ public void forEachWrite(BiConsumer<T, D> consumer)
+ {
+ writes.forEach(consumer);
+ }
+
+ @Override
+ public int numDeletes()
+ {
+ return deletes.size();
+ }
+
+ @Override
+ public void forEachDelete(Consumer<T> consumer)
+ {
+ deletes.forEach(consumer);
+ }
+
+ public boolean contains(T key)
+ {
+ return writes.containsKey(key) || deletes.contains(key);
+ }
+
+ /**
+ * remove the given key from both the writes and deletes set
+ * @param key
+ */
+ void removeKeyUnsafe(T key)
+ {
+ writes.remove(key);
+ deletes.remove(key);
+ }
+
+ /**
+ * add the given write data directly, without using the command loader
+ * @param key
+ * @param value
+ */
+ void addWriteUnsafe(T key, D value)
+ {
+ writes.put(key, value);
+ deletes.remove(key);
+ }
+
+ /**
+ * if this update has a write or delete for the given key, remove
+ * it from this update and put it in the destination update
+ */
+ boolean transferKeyTo(T key, MutableUpdate<T, D> dst)
+ {
+ if (writes.containsKey(key))
+ {
+ dst.writes.put(key, this.writes.remove(key));
+ return true;
+ }
+ else if (deletes.remove(key))
+ {
+ dst.deletes.add(key);
+ return true;
+ }
+ return false;
+ }
+
+ public ImmutableUpdate<T, D> toImmutable()
+ {
+ return new ImmutableUpdate<>(ImmutableMap.copyOf(writes), ImmutableSet.copyOf(deletes));
+ }
+ }
}
diff --git a/accord-core/src/main/java/accord/impl/CommandsForKey.java b/accord-core/src/main/java/accord/impl/CommandsForKey.java
index ed238af3..59be3bdd 100644
--- a/accord-core/src/main/java/accord/impl/CommandsForKey.java
+++ b/accord-core/src/main/java/accord/impl/CommandsForKey.java
@@ -23,7 +23,6 @@ import accord.impl.CommandTimeseries.CommandLoader;
import com.google.common.collect.ImmutableSortedMap;
import java.util.Objects;
-import java.util.function.Consumer;
import accord.local.Command;
import accord.local.PreLoadContext;
@@ -33,13 +32,8 @@ import accord.primitives.Keys;
import accord.primitives.Timestamp;
import accord.primitives.TxnId;
-import static accord.local.Status.PreAccepted;
-import static accord.local.Status.PreCommitted;
-
-public class CommandsForKey implements CommandTimeseriesHolder
+public class CommandsForKey implements DomainCommands
{
- public static final long NO_LAST_EXECUTED_HLC = Long.MIN_VALUE;
-
public static class SerializerSupport
{
public static Listener listener(Key key)
@@ -47,13 +41,11 @@ public class CommandsForKey implements CommandTimeseriesHolder
return new Listener(key);
}
- public static <D> CommandsForKey create(Key key, Timestamp max,
- Timestamp lastExecutedTimestamp, long lastExecutedHlc, Timestamp lastWriteTimestamp,
+ public static <D> CommandsForKey create(Key key,
CommandLoader<D> loader,
- ImmutableSortedMap<Timestamp, D> byId,
- ImmutableSortedMap<Timestamp, D> byExecuteAt)
+ ImmutableSortedMap<Timestamp, D> commands)
{
- return new CommandsForKey(key, max, lastExecutedTimestamp, lastExecutedHlc, lastWriteTimestamp, loader, byId, byExecuteAt);
+ return new CommandsForKey(key, loader, commands);
}
}
@@ -95,8 +87,7 @@ public class CommandsForKey implements CommandTimeseriesHolder
@Override
public void onChange(SafeCommandStore safeStore, SafeCommand safeCommand)
{
- SafeCommandsForKey cfk = ((AbstractSafeCommandStore) safeStore).commandsForKey(listenerKey);
- cfk.listenerUpdate(safeCommand.current());
+ CommandsForKeys.listenerUpdate((AbstractSafeCommandStore<?,?,?,?>) safeStore, listenerKey, safeCommand.current());
}
@Override
@@ -107,53 +98,24 @@ public class CommandsForKey implements CommandTimeseriesHolder
}
private final Key key;
- private final Timestamp max;
- private final Timestamp lastExecutedTimestamp;
- // TODO (desired): we have leaked C* implementation details here
- private final long rawLastExecutedHlc;
- private final Timestamp lastWriteTimestamp;
- private final CommandTimeseries<?> byId;
- // TODO (expected): we probably do not need to separately maintain byExecuteAt - probably better to just filter byId
- private final CommandTimeseries<?> byExecuteAt;
+ private final CommandTimeseries<?> commands;
- <D> CommandsForKey(Key key, Timestamp max,
- Timestamp lastExecutedTimestamp,
- long rawLastExecutedHlc,
- Timestamp lastWriteTimestamp,
- CommandTimeseries<D> byId,
- CommandTimeseries<D> byExecuteAt)
+ <D> CommandsForKey(Key key,
+ CommandTimeseries<D> commands)
{
this.key = key;
- this.max = max;
- this.lastExecutedTimestamp = lastExecutedTimestamp;
- this.rawLastExecutedHlc = rawLastExecutedHlc;
- this.lastWriteTimestamp = lastWriteTimestamp;
- this.byId = byId;
- this.byExecuteAt = byExecuteAt;
+ this.commands = commands;
}
- <D> CommandsForKey(Key key, Timestamp max,
- Timestamp lastExecutedTimestamp,
- long rawLastExecutedHlc,
- Timestamp lastWriteTimestamp,
- CommandLoader<D> loader,
- ImmutableSortedMap<Timestamp, D> committedById,
- ImmutableSortedMap<Timestamp, D> committedByExecuteAt)
+ <D> CommandsForKey(Key key, CommandLoader<D> loader, ImmutableSortedMap<Timestamp, D> commands)
{
- this(key, max, lastExecutedTimestamp, rawLastExecutedHlc, lastWriteTimestamp,
- new CommandTimeseries<>(key, loader, committedById),
- new CommandTimeseries<>(key, loader, committedByExecuteAt));
+ this(key, new CommandTimeseries<>(key, loader, commands));
}
public <D> CommandsForKey(Key key, CommandLoader<D> loader)
{
this.key = key;
- this.max = Timestamp.NONE;
- this.lastExecutedTimestamp = Timestamp.NONE;
- this.rawLastExecutedHlc = 0;
- this.lastWriteTimestamp = Timestamp.NONE;
- this.byId = new CommandTimeseries<>(key, loader);
- this.byExecuteAt = new CommandTimeseries<>(key, loader);
+ this.commands = new CommandTimeseries<>(key, loader);
}
@Override
@@ -168,13 +130,7 @@ public class CommandsForKey implements CommandTimeseriesHolder
if (this == o) return true;
if (o == null || getClass() != o.getClass()) return false;
CommandsForKey that = (CommandsForKey) o;
- return rawLastExecutedHlc == that.rawLastExecutedHlc
- && key.equals(that.key)
- && Objects.equals(max, that.max)
- && Objects.equals(lastExecutedTimestamp, that.lastExecutedTimestamp)
- && Objects.equals(lastWriteTimestamp, that.lastWriteTimestamp)
- && byId.equals(that.byId)
- && byExecuteAt.equals(that.byExecuteAt);
+ return key.equals(that.key) && commands.equals(that.commands);
}
@Override
@@ -183,96 +139,35 @@ public class CommandsForKey implements CommandTimeseriesHolder
throw new UnsupportedOperationException();
}
- public final Command.DurableAndIdempotentListener asListener()
- {
- return new Listener(key());
- }
-
public Key key()
{
return key;
}
@Override
- public Timestamp max()
- {
- return max;
- }
-
- public Timestamp lastExecutedTimestamp()
- {
- return lastExecutedTimestamp;
- }
-
- public long lastExecutedHlc()
- {
- return rawLastExecutedHlc == NO_LAST_EXECUTED_HLC ? lastExecutedTimestamp.hlc() : rawLastExecutedHlc;
- }
-
- public long rawLastExecutedHlc()
- {
- return rawLastExecutedHlc;
- }
-
- public Timestamp lastWriteTimestamp()
+ public CommandTimeseries<?> commands()
{
- return lastWriteTimestamp;
+ return commands;
}
- @Override
- public CommandTimeseries<?> byId()
+ public <D> CommandsForKey update(CommandTimeseries.Update<TxnId, D> commands)
{
- return byId;
- }
+ if (commands.isEmpty())
+ return this;
- @Override
- public CommandTimeseries<?> byExecuteAt()
- {
- return byExecuteAt;
+ return new CommandsForKey(key, commands.apply((CommandTimeseries<D>) this.commands));
}
public boolean hasRedundant(TxnId redundantBefore)
{
- return byId.minTimestamp().compareTo(redundantBefore) < 0;
+ return commands.minTimestamp().compareTo(redundantBefore) < 0;
}
public CommandsForKey withoutRedundant(TxnId redundantBefore)
{
- Timestamp removeExecuteAt = byId.maxExecuteAtBefore(redundantBefore);
-
- return new CommandsForKey(key, max.compareTo(redundantBefore) < 0 ? Timestamp.NONE : max,
- lastExecutedTimestamp.compareTo(redundantBefore) < 0 ? Timestamp.NONE : lastExecutedTimestamp,
- rawLastExecutedHlc < redundantBefore.hlc() ? NO_LAST_EXECUTED_HLC : rawLastExecutedHlc,
- lastWriteTimestamp.compareTo(redundantBefore) < 0 ? Timestamp.NONE : lastWriteTimestamp,
- (CommandTimeseries)byId.beginUpdate().removeBefore(redundantBefore).build(),
- (CommandTimeseries)byExecuteAt.beginUpdate().removeBefore(removeExecuteAt).build()
- );
+ return new CommandsForKey(key, (CommandTimeseries) commands.unbuild().removeBefore(redundantBefore).build());
}
- public void forWitnessed(Timestamp minTs, Timestamp maxTs, Consumer<TxnId> consumer)
- {
- byId.between(minTs, maxTs, status -> status.hasBeen(PreAccepted)).forEach(consumer);
- byExecuteAt.between(minTs, maxTs, status -> status.hasBeen(PreCommitted)).forEach(consumer);
- }
-
- public void validateExecuteAtTime(Timestamp executeAt, boolean isForWriteTxn)
- {
- if (executeAt.compareTo(lastWriteTimestamp) < 0)
- throw new IllegalArgumentException(String.format("%s is less than the most recent write timestamp %s", executeAt, lastWriteTimestamp));
+ public interface Update {}
- int cmp = executeAt.compareTo(lastExecutedTimestamp);
- // execute can be in the past if it's for a read and after the most recent write
- if (cmp == 0 || (!isForWriteTxn && cmp < 0))
- return;
- if (cmp < 0)
- throw new IllegalArgumentException(String.format("%s is less than the most recent executed timestamp %s", executeAt, lastExecutedTimestamp));
- else
- throw new IllegalArgumentException(String.format("%s is greater than the most recent executed timestamp, cfk should be updated", executeAt, lastExecutedTimestamp));
- }
-
- public long hlcFor(Timestamp executeAt, boolean isForWriteTxn)
- {
- validateExecuteAtTime(executeAt, isForWriteTxn);
- return rawLastExecutedHlc == NO_LAST_EXECUTED_HLC ? lastExecutedTimestamp.hlc() : rawLastExecutedHlc;
- }
}
diff --git a/accord-core/src/main/java/accord/impl/CommandsForKeyGroupUpdater.java b/accord-core/src/main/java/accord/impl/CommandsForKeyGroupUpdater.java
new file mode 100644
index 00000000..63a3fe80
--- /dev/null
+++ b/accord-core/src/main/java/accord/impl/CommandsForKeyGroupUpdater.java
@@ -0,0 +1,227 @@
+/*
+ * 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 accord.impl;
+
+import accord.impl.CommandTimeseries.CommandLoader;
+import accord.primitives.Timestamp;
+
+import java.util.Objects;
+
+/**
+ * Contains updates for both commands for key sets. One for deps, one for all,
+ * and a common set affecting both
+ */
+public abstract class CommandsForKeyGroupUpdater<D>
+{
+ public static class Mutable<D> extends CommandsForKeyGroupUpdater<D>
+ {
+ private final CommandsForKeyUpdater.Mutable<D> deps;
+ private final CommandsForKeyUpdater.Mutable<D> all;
+ private final CommandsForKeyUpdater.Mutable<D> common;
+
+ public Mutable(CommandsForKeyUpdater.Mutable<D> deps, CommandsForKeyUpdater.Mutable<D> all, CommandsForKeyUpdater.Mutable<D> common)
+ {
+ this.deps = deps;
+ this.all = all;
+ this.common = common;
+ }
+
+ public Mutable(CommandLoader<D> loader)
+ {
+ this(new CommandsForKeyUpdater.Mutable<>(loader),
+ new CommandsForKeyUpdater.Mutable<>(loader),
+ new CommandsForKeyUpdater.Mutable<>(loader));
+ }
+
+ @Override
+ public CommandsForKeyUpdater.Mutable<D> deps()
+ {
+ return deps;
+ }
+
+ @Override
+ public CommandsForKeyUpdater.Mutable<D> all()
+ {
+ return all;
+ }
+
+ @Override
+ public CommandsForKeyUpdater.Mutable<D> common()
+ {
+ return common;
+ }
+
+ public Immutable<D> toImmutable()
+ {
+ return new Immutable<>(deps.toImmutable(), all.toImmutable(), common.toImmutable());
+ }
+ }
+
+ public static class Immutable<D> extends CommandsForKeyGroupUpdater<D>
+ {
+ private final CommandsForKeyUpdater.Immutable<D> deps;
+ private final CommandsForKeyUpdater.Immutable<D> all;
+ private final CommandsForKeyUpdater.Immutable<D> common;
+
+ public interface Factory<D, T extends Immutable<D>>
+ {
+ T create(CommandsForKeyUpdater.Immutable<D> deps, CommandsForKeyUpdater.Immutable<D> all, CommandsForKeyUpdater.Immutable<D> common);
+ }
+
+ public static <T> Factory<T, Immutable<T>> getFactory()
+ {
+ return Immutable::new;
+ }
+
+ public Immutable(CommandsForKeyUpdater.Immutable<D> deps, CommandsForKeyUpdater.Immutable<D> all, CommandsForKeyUpdater.Immutable<D> common)
+ {
+ this.deps = deps;
+ this.all = all;
+ this.common = common;
+ }
+
+ @Override
+ public String toString()
+ {
+ return "Immutable{" +
+ "deps=" + deps +
+ ", all=" + all +
+ ", common=" + common +
+ '}';
+ }
+
+ @Override
+ public boolean equals(Object o)
+ {
+ if (this == o) return true;
+ if (o == null || getClass() != o.getClass()) return false;
+ Immutable<?> immutable = (Immutable<?>) o;
+ return Objects.equals(deps, immutable.deps) && Objects.equals(all, immutable.all) && Objects.equals(common, immutable.common);
+ }
+
+ @Override
+ public int hashCode()
+ {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public CommandsForKeyUpdater.Immutable<D> deps()
+ {
+ return deps;
+ }
+
+ @Override
+ public CommandsForKeyUpdater.Immutable<D> all()
+ {
+ return all;
+ }
+
+ @Override
+ public CommandsForKeyUpdater.Immutable<D> common()
+ {
+ return common;
+ }
+
+ public Mutable<D> toMutable(CommandLoader<D> loader)
+ {
+ return new Mutable<>(deps.toMutable(loader), all.toMutable(loader), common.toMutable(loader));
+ }
+
+ private static <T extends Timestamp, D> void mergeCommon(CommandTimeseries.ImmutableUpdate<T, D> newCommon,
+ CommandTimeseries.MutableUpdate<T, D> mergedCommon,
+ CommandTimeseries.MutableUpdate<T, D> mergedDeps,
+ CommandTimeseries.MutableUpdate<T, D> mergedAll)
+ {
+ newCommon.forEachWrite((k, v) -> {
+ mergedCommon.addWriteUnsafe(k, v);
+ mergedDeps.removeKeyUnsafe(k);
+ mergedAll.removeKeyUnsafe(k);
+ });
+ newCommon.forEachDelete(k -> {
+ mergedCommon.remove(k);
+ mergedDeps.removeKeyUnsafe(k);
+ mergedAll.removeKeyUnsafe(k);
+ });
+ }
+
+ private static <T extends Timestamp, D> void mergeSpecific(CommandTimeseries.ImmutableUpdate<T, D> newSpecific,
+ CommandTimeseries.MutableUpdate<T, D> mergedSpecific, CommandTimeseries.MutableUpdate<T, D> mergedOther, CommandTimeseries.MutableUpdate<T, D> mergedCommon)
+ {
+ newSpecific.forEachWrite((k, v) -> {
+ mergedSpecific.addWriteUnsafe(k, v);
+ mergedCommon.transferKeyTo(k, mergedOther);
+ });
+
+ newSpecific.forEachDelete(k -> {
+ mergedSpecific.remove(k);
+ mergedCommon.transferKeyTo(k, mergedOther);
+ });
+ }
+
+ public static <D, T extends Immutable<D>> T merge(T current, T update, Factory<D, T> factory)
+ {
+ if (current == null || current.isEmpty())
+ return update;
+
+ if (update == null || update.isEmpty())
+ return current;
+
+ Mutable<D> merged = current.toMutable(null);
+
+ mergeCommon(update.common().commands(), merged.common.commands(), merged.deps.commands(), merged.all.commands());
+
+ mergeSpecific(update.deps().commands(), merged.deps.commands(), merged.all.commands(), merged.common.commands());
+ mergeSpecific(update.all().commands(), merged.all.commands(), merged.deps.commands(), merged.common.commands());
+
+ return factory.create(merged.deps.toImmutable(), merged.all.toImmutable(), merged.common.toImmutable());
+ }
+ }
+
+ public abstract CommandsForKeyUpdater<D> deps();
+
+ public abstract CommandsForKeyUpdater<D> all();
+
+ /**
+ * Update the deps and all CFK groups.
+ * In case conflicting updates have been applied to the common udpater and a specific one, the specific one should take precedence
+ */
+ public abstract CommandsForKeyUpdater<D> common();
+
+ public boolean isEmpty()
+ {
+ return deps().isEmpty() && all().isEmpty() && common().isEmpty();
+ }
+
+ public CommandsForKey applyToDeps(CommandsForKey current)
+ {
+ current = common().apply(current);
+ current = deps().apply(current);
+
+ return current;
+ }
+
+ public CommandsForKey applyToAll(CommandsForKey current)
+ {
+ current = common().apply(current);
+ current = all().apply(current);
+
+ return current;
+ }
+}
diff --git a/accord-core/src/main/java/accord/impl/CommandsForKeyUpdater.java b/accord-core/src/main/java/accord/impl/CommandsForKeyUpdater.java
new file mode 100644
index 00000000..c46c3275
--- /dev/null
+++ b/accord-core/src/main/java/accord/impl/CommandsForKeyUpdater.java
@@ -0,0 +1,117 @@
+/*
+ * 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 accord.impl;
+
+import accord.impl.CommandTimeseries.CommandLoader;
+import accord.primitives.TxnId;
+
+import java.util.Objects;
+
+/**
+ * Contains updates for a single commands for key dataset (ie: deps OR all)
+ * @param <D>
+ */
+public abstract class CommandsForKeyUpdater<D>
+{
+ public abstract CommandTimeseries.Update<TxnId, D> commands();
+
+ public static class Mutable<D> extends CommandsForKeyUpdater<D>
+ {
+ private final CommandTimeseries.MutableUpdate<TxnId, D> commands;
+
+ Mutable(CommandTimeseries.MutableUpdate<TxnId, D> commands)
+ {
+ this.commands = commands;
+ }
+
+ public Mutable(CommandLoader<D> loader)
+ {
+ this(new CommandTimeseries.MutableUpdate<>(loader));
+ }
+
+ @Override public CommandTimeseries.MutableUpdate<TxnId, D> commands() { return commands; }
+ public Immutable<D> toImmutable()
+ {
+ return new Immutable<>(commands.toImmutable());
+ }
+ }
+
+ public static class Immutable<D> extends CommandsForKeyUpdater<D>
+ {
+ private static final Immutable<Object> EMPTY = new Immutable<>(CommandTimeseries.ImmutableUpdate.empty());
+ private final CommandTimeseries.ImmutableUpdate<TxnId, D> commands;
+
+ public Immutable(CommandTimeseries.ImmutableUpdate<TxnId, D> commands)
+ {
+ this.commands = commands;
+ }
+
+ @Override
+ public String toString()
+ {
+ return "Immutable{" +
+ "commands=" + commands +
+ '}';
+ }
+
+ @Override
+ public boolean equals(Object o)
+ {
+ if (this == o) return true;
+ if (o == null || getClass() != o.getClass()) return false;
+ Immutable<?> immutable = (Immutable<?>) o;
+ return Objects.equals(commands, immutable.commands);
+ }
+
+ @Override
+ public int hashCode()
+ {
+ return Objects.hash(commands);
+ }
+
+ public static <D> Immutable<D> empty()
+ {
+ return (Immutable<D>) EMPTY;
+ }
+
+ @Override public CommandTimeseries.ImmutableUpdate<TxnId, D> commands() { return commands; }
+
+ public Mutable<D> toMutable(CommandLoader<D> loader)
+ {
+ return new Mutable<>(commands.toMutable(loader));
+ }
+ }
+
+ public int totalChanges()
+ {
+ return commands().numChanges();
+ }
+
+ public boolean isEmpty()
+ {
+ return commands().isEmpty();
+ }
+
+ public CommandsForKey apply(CommandsForKey current)
+ {
+ if (commands().isEmpty())
+ return current;
+ return current.update(commands());
+ }
+}
diff --git a/accord-core/src/main/java/accord/impl/CommandsForKeys.java b/accord-core/src/main/java/accord/impl/CommandsForKeys.java
new file mode 100644
index 00000000..a6f12c17
--- /dev/null
+++ b/accord-core/src/main/java/accord/impl/CommandsForKeys.java
@@ -0,0 +1,152 @@
+/*
+ * 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 accord.impl;
+
+import accord.api.VisibleForImplementation;
+import accord.local.Command;
+import accord.local.CommandStore;
+import accord.primitives.RoutableKey;
+import accord.primitives.Timestamp;
+import accord.primitives.TxnId;
+import com.google.common.annotations.VisibleForTesting;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import static accord.local.Command.NotDefined.uninitialised;
+
+public class CommandsForKeys
+{
+ private static final Logger logger = LoggerFactory.getLogger(CommandsForKeys.class);
+
+ private CommandsForKeys() {}
+
+ @VisibleForTesting
+ @VisibleForImplementation
+ public static TimestampsForKey updateMax(AbstractSafeCommandStore<?,?,?,?> safeStore, RoutableKey key, Timestamp timestamp)
+ {
+ SafeTimestampsForKey tfk = safeStore.timestampsForKey(key);
+ return tfk.updateMax(timestamp);
+ }
+
+ public static Command.DurableAndIdempotentListener registerCommand(SafeTimestampsForKey timestamps, SafeCommandsForKey.Update<?,?> update, Command command)
+ {
+
+ update.common().commands().add(command.txnId(), command);
+
+ timestamps.updateMax(command.executeAt());
+
+ return new CommandsForKey.Listener(timestamps.key());
+ }
+
+ public static Command.DurableAndIdempotentListener registerCommand(AbstractSafeCommandStore<?,?,?,?> safeStore, RoutableKey key, Command command)
+ {
+ return registerCommand(safeStore.timestampsForKey(key), safeStore.getOrCreateCommandsForKeyUpdate(key), command);
+ }
+
+ public static void registerNotWitnessed(AbstractSafeCommandStore<?,?,?,?> safeStore, RoutableKey key, TxnId txnId)
+ {
+ SafeTimestampsForKey tfk = safeStore.timestampsForKey(key);
+ SafeCommandsForKey cfk = safeStore.depsCommandsForKey(key);
+
+ // FIXME: should be the recovery commands
+ if (cfk.current().commands().commands.containsKey(txnId))
+ return;
+
+ tfk.updateMax(txnId);
+ SafeCommandsForKey.Update<?,?> update = safeStore.getOrCreateCommandsForKeyUpdate(key);
+
+ update.common().commands().add(txnId, uninitialised(txnId));
+ }
+
+ public static void listenerUpdate(AbstractSafeCommandStore<?,?,?,?> safeStore, RoutableKey listenerKey, Command command)
+ {
+ if (logger.isTraceEnabled())
+ logger.trace("[{}]: updating as listener in response to change on {} with status {} ({})",
+ listenerKey, command.txnId(), command.status(), command);
+
+ SafeTimestampsForKey tfk = safeStore.timestampsForKey(listenerKey);
+ SafeCommandsForKey.Update<?,?> update = safeStore.getOrCreateCommandsForKeyUpdate(listenerKey);
+
+ // add/remove the command on every listener update to avoid
+ // special denormalization handling in Cassandra
+ switch (command.status())
+ {
+ default: throw new AssertionError();
+ case PreAccepted:
+ case NotDefined:
+ case Accepted:
+ case AcceptedInvalidate:
+ case PreCommitted:
+ case Applied:
+ case PreApplied:
+ case Committed:
+ case ReadyToExecute:
+ update.common().commands().add(command.txnId(), command);
+ break;
+ case Invalidated:
+ update.common().commands().remove(command.txnId());
+ case Truncated:
+ break;
+ }
+
+ tfk.updateMax(command.executeAt());
+ }
+
+ public static TimestampsForKey updateLastExecutionTimestamps(CommandStore commandStore, SafeTimestampsForKey tfk, Timestamp executeAt, boolean isForWriteTxn)
+ {
+ TimestampsForKey current = tfk.current();
+
+ Timestamp lastWrite = current.lastWriteTimestamp();
+
+ if (executeAt.compareTo(lastWrite) < 0)
+ {
+ if (!commandStore.safeToReadAt(executeAt).contains(tfk.key().toUnseekable()))
+ return current;
+ throw new IllegalArgumentException(String.format("%s is less than the most recent write timestamp %s", executeAt, lastWrite));
+ }
+
+ Timestamp lastExecuted = current.lastExecutedTimestamp();
+ int cmp = executeAt.compareTo(lastExecuted);
+ // execute can be in the past if it's for a read and after the most recent write
+ if (cmp == 0 || (!isForWriteTxn && cmp < 0))
+ return current;
+
+ if (cmp < 0)
+ {
+ if (!commandStore.safeToReadAt(executeAt).contains(tfk.key().toUnseekable()))
+ return current;
+ throw new IllegalArgumentException(String.format("%s is less than the most recent executed timestamp %s", executeAt, lastExecuted));
+ }
+
+ long micros = executeAt.hlc();
+ long lastMicros = current.lastExecutedHlc();
+
+ Timestamp lastExecutedTimestamp = executeAt;
+ long lastExecutedHlc = micros > lastMicros ? Long.MIN_VALUE : lastMicros + 1;
+ Timestamp lastWriteTimestamp = isForWriteTxn ? executeAt : current.lastWriteTimestamp();
+
+ return tfk.updateLastExecutionTimestamps(lastExecutedTimestamp, lastExecutedHlc, lastWriteTimestamp);
+ }
+
+ @VisibleForImplementation
+ public static <D> TimestampsForKey updateLastExecutionTimestamps(AbstractSafeCommandStore<?,?,?,?> safeStore, RoutableKey key, Timestamp executeAt, boolean isForWriteTxn)
+ {
+ return updateLastExecutionTimestamps(safeStore.commandStore(), safeStore.timestampsForKey(key), executeAt, isForWriteTxn);
+ }
+}
diff --git a/accord-core/src/main/java/accord/impl/DepsBuilder.java b/accord-core/src/main/java/accord/impl/DepsBuilder.java
new file mode 100644
index 00000000..75e17d12
--- /dev/null
+++ b/accord-core/src/main/java/accord/impl/DepsBuilder.java
@@ -0,0 +1,176 @@
+/*
+ * 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 accord.impl;
+
+import accord.local.SafeCommandStore;
+import accord.local.Status;
+import accord.primitives.*;
+
+import java.util.*;
+import java.util.function.Consumer;
+
+public class DepsBuilder
+{
+ private static boolean CAN_PRUNE = Boolean.getBoolean("accord.can_prune_deps");
+ private final TxnId subject;
+ private final Map<Seekable, Builder> builders = new HashMap<>();
+
+ public static void setCanPruneUnsafe(boolean value)
+ {
+ CAN_PRUNE = value;
+ }
+
+ public static boolean getCanPrune()
+ {
+ return CAN_PRUNE;
+ }
+
+ public DepsBuilder(TxnId subject)
+ {
+ this.subject = subject;
+ }
+
+ private static class TxnInfo
+ {
+ private final TxnId txnId;
+ private final Status status;
+ private final Timestamp executeAt;
+ private final List<TxnId> depsIds;
+ private boolean required = false;
+
+ public TxnInfo(TxnId txnId, Status status, Timestamp executeAt, List<TxnId> depsIds)
+ {
+ this.txnId = txnId;
+ this.status = status;
+ this.executeAt = executeAt;
+ this.depsIds = depsIds;
+ }
+ }
+
+ /**
+ * To remove transitive dependencies, we filter out commands that exist in the deps list
+ * We need commands that are not superseded by other commands with later execution times
+ */
+ public static class Builder
+ {
+ private final Map<TxnId, TxnInfo> commands = new HashMap<>();
+ private final Map<TxnId, Set<TxnId>> supportingDeps = new HashMap<>();
+
+ public void add(TxnId txnId, Status status, Timestamp executeAt, List<TxnId> depsIds)
+ {
+ commands.put(txnId, new TxnInfo(txnId, status, executeAt, depsIds));
+ if (depsIds != null)
+ depsIds.forEach(id -> supportingDeps.computeIfAbsent(id, i -> new HashSet<>()).add(txnId));
+ }
+
+ private void markRequired(TxnInfo info)
+ {
+ if (info.required)
+ return;
+
+ info.required = true;
+
+ Set<TxnId> supportingIds = supportingDeps.get(info.txnId);
+ if (supportingIds == null || supportingIds.isEmpty())
+ return;
+
+ for (TxnId supportingId : supportingIds)
+ {
+ TxnInfo supporting = commands.get(supportingId);
+ if (supporting == null)
+ continue;
+
+ markRequired(supporting);
+ }
+ }
+
+ private boolean isRequired(TxnInfo info)
+ {
+ if (!CAN_PRUNE || info.required)
+ return true;
+
+ if (!info.status.hasBeen(Status.Applied))
+ return true;
+
+ Set<TxnId> supportingIds = supportingDeps.get(info.txnId);
+ if (supportingIds == null || supportingIds.isEmpty())
+ return true;
+
+ for (TxnId supportingId : supportingIds)
+ {
+ TxnInfo supporting = commands.get(supportingId);
+
+ // TODO: check for eviction of txnId and skip if it's evicted
+
+ if (!supporting.status.hasBeen(Status.Applied))
+ return true;
+
+ if (supporting.executeAt.compareTo(info.executeAt) < 0)
+ return true;
+
+ // FIXME: is this actually needed?
+ // let execution order strongly connected components
+ if (info.depsIds.contains(supportingId))
+ return true;
+ }
+
+ return false;
+ }
+
+ private void preprocess(TxnInfo info)
+ {
+ if (isRequired(info))
+ markRequired(info);
+ }
+
+ private void preprocess()
+ {
+ commands.values().forEach(this::preprocess);
+ }
+
+ public <T> void build(SafeCommandStore safeStore, Seekable seekable, Consumer<TxnId> consumer)
+ {
+ preprocess();
+ commands.forEach(((txnId, info) -> {
+ if (info.required)
+ consumer.accept(txnId);
+ else
+ safeStore.removeCommandFromSeekableDeps(seekable, txnId, info.executeAt, info.status);
+ }));
+ }
+ }
+
+ public void add(TxnId txnId, Seekable seekable, Status status, Timestamp executeAt, List<TxnId> depsIds)
+ {
+ // don't add self to deps
+ if (txnId.equals(subject))
+ return;
+
+ builders.computeIfAbsent(seekable, s -> new Builder()).add(txnId, status, executeAt, depsIds);
+ }
+
+ public PartialDeps buildPartialDeps(SafeCommandStore safeStore, Ranges ranges)
+ {
+ try (PartialDeps.Builder result = new PartialDeps.Builder(ranges);)
+ {
+ builders.forEach((seekable, builder) -> builder.build(safeStore, seekable, txnId -> result.add(seekable, txnId)));
+ return result.build();
+ }
+ }
+}
diff --git a/accord-core/src/main/java/accord/impl/CommandTimeseriesHolder.java b/accord-core/src/main/java/accord/impl/DomainCommands.java
similarity index 82%
copy from accord-core/src/main/java/accord/impl/CommandTimeseriesHolder.java
copy to accord-core/src/main/java/accord/impl/DomainCommands.java
index f0a88e75..ee1423f2 100644
--- a/accord-core/src/main/java/accord/impl/CommandTimeseriesHolder.java
+++ b/accord-core/src/main/java/accord/impl/DomainCommands.java
@@ -18,12 +18,9 @@
package accord.impl;
-import accord.primitives.Timestamp;
-
-public interface CommandTimeseriesHolder
+public interface DomainCommands
{
- CommandTimeseries<?> byId();
- CommandTimeseries<?> byExecuteAt();
+ enum Kind { COORDINATION, RECOVERY }
- Timestamp max();
+ CommandTimeseries<?> commands();
}
diff --git a/accord-core/src/main/java/accord/impl/CommandTimeseriesHolder.java b/accord-core/src/main/java/accord/impl/DomainTimestamps.java
similarity index 88%
copy from accord-core/src/main/java/accord/impl/CommandTimeseriesHolder.java
copy to accord-core/src/main/java/accord/impl/DomainTimestamps.java
index f0a88e75..2d614c89 100644
--- a/accord-core/src/main/java/accord/impl/CommandTimeseriesHolder.java
+++ b/accord-core/src/main/java/accord/impl/DomainTimestamps.java
@@ -20,10 +20,7 @@ package accord.impl;
import accord.primitives.Timestamp;
-public interface CommandTimeseriesHolder
+public interface DomainTimestamps
{
- CommandTimeseries<?> byId();
- CommandTimeseries<?> byExecuteAt();
-
Timestamp max();
-}
+}
\ No newline at end of file
diff --git a/accord-core/src/main/java/accord/impl/InMemoryCommandStore.java b/accord-core/src/main/java/accord/impl/InMemoryCommandStore.java
index 1a970c91..c0537d5a 100644
--- a/accord-core/src/main/java/accord/impl/InMemoryCommandStore.java
+++ b/accord-core/src/main/java/accord/impl/InMemoryCommandStore.java
@@ -18,7 +18,6 @@
package accord.impl;
-import java.util.AbstractMap;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
@@ -36,12 +35,13 @@ import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.function.BiConsumer;
-import java.util.function.BiFunction;
import java.util.function.Consumer;
import java.util.function.Function;
import java.util.function.Predicate;
import javax.annotation.Nullable;
+import accord.local.*;
+import accord.utils.TriFunction;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -50,17 +50,7 @@ import accord.api.DataStore;
import accord.api.Key;
import accord.api.ProgressLog;
import accord.impl.CommandTimeseries.CommandLoader;
-import accord.local.Command;
-import accord.local.CommandStore;
import accord.local.CommandStores.RangesForEpoch;
-import accord.local.CommonAttributes;
-import accord.local.Listeners;
-import accord.local.NodeTimeService;
-import accord.local.PreLoadContext;
-import accord.local.SafeCommand;
-import accord.local.SafeCommandStore;
-import accord.local.SaveStatus;
-import accord.local.Status;
import accord.primitives.AbstractKeys;
import accord.primitives.Deps;
import accord.primitives.PartialDeps;
@@ -87,6 +77,7 @@ import static accord.local.SaveStatus.Applying;
import static accord.local.SaveStatus.ReadyToExecute;
import static accord.local.Status.Committed;
import static accord.local.Status.Truncated;
+import static accord.local.Status.NotDefined;
import static accord.primitives.Routables.Slice.Minimal;
public abstract class InMemoryCommandStore extends CommandStore
@@ -96,7 +87,9 @@ public abstract class InMemoryCommandStore extends CommandStore
final NavigableMap<TxnId, GlobalCommand> commands = new TreeMap<>();
final NavigableMap<Timestamp, GlobalCommand> commandsByExecuteAt = new TreeMap<>();
- private final NavigableMap<RoutableKey, GlobalCommandsForKey> commandsForKey = new TreeMap<>();
+ private final NavigableMap<RoutableKey, GlobalTimestampsForKey> timestampsForKey = new TreeMap<>();
+ private final NavigableMap<RoutableKey, GlobalCommandsForKey> depsCommandsForKey = new TreeMap<>();
+ private final NavigableMap<RoutableKey, GlobalCommandsForKey> allCommandsForKey = new TreeMap<>();
// TODO (find library, efficiency): this is obviously super inefficient, need some range map
private final TreeMap<TxnId, RangeCommand> rangeCommands = new TreeMap<>();
@@ -131,7 +124,7 @@ public abstract class InMemoryCommandStore extends CommandStore
return rangeCommands;
}
- public GlobalCommand ifPresent(TxnId txnId)
+ public GlobalCommand commandIfPresent(TxnId txnId)
{
return commands.get(txnId);
}
@@ -226,19 +219,44 @@ public abstract class InMemoryCommandStore extends CommandStore
return commands.containsKey(txnId);
}
- public GlobalCommandsForKey ifPresent(Key key)
+ public GlobalCommandsForKey depsCommandsForKeyIfPresent(Key key)
{
- return commandsForKey.get(key);
+ return depsCommandsForKey.get(key);
}
- public GlobalCommandsForKey commandsForKey(Key key)
+ public GlobalCommandsForKey depsCommandsForKey(Key key)
{
- return commandsForKey.computeIfAbsent(key, GlobalCommandsForKey::new);
+ return depsCommandsForKey.computeIfAbsent(key, GlobalCommandsForKey::new);
}
- public boolean hasCommandsForKey(Key key)
+ public boolean hasDepsCommandsForKey(Key key)
{
- return commandsForKey.containsKey(key);
+ return depsCommandsForKey.containsKey(key);
+ }
+
+ public GlobalCommandsForKey allCommandsForKeyIfPresent(Key key)
+ {
+ return allCommandsForKey.get(key);
+ }
+
+ public GlobalCommandsForKey allCommandsForKey(Key key)
+ {
+ return allCommandsForKey.computeIfAbsent(key, GlobalCommandsForKey::new);
+ }
+
+ public boolean hasAllCommandsForKey(Key key)
+ {
+ return allCommandsForKey.containsKey(key);
+ }
+
+ public GlobalTimestampsForKey timestampsForKey(Key key)
+ {
+ return timestampsForKey.computeIfAbsent(key, GlobalTimestampsForKey::new);
+ }
+
+ public GlobalTimestampsForKey timestampsForKeyIfPresent(Key key)
+ {
+ return timestampsForKey.get(key);
}
public CommonAttributes register(InMemorySafeStore safeStore, Seekables<?, ?> keysOrRanges, Ranges slice, SafeCommand command, CommonAttributes attrs)
@@ -249,8 +267,7 @@ public abstract class InMemoryCommandStore extends CommandStore
case Key:
CommonAttributes.Mutable mutable = attrs.mutable();
forEach(keysOrRanges, slice, key -> {
- SafeCommandsForKey cfk = safeStore.commandsForKey(key);
- Command.DurableAndIdempotentListener listener = cfk.register(command.current()).asListener();
+ Command.DurableAndIdempotentListener listener = CommandsForKeys.registerCommand(safeStore, key, command.current());
mutable.addListener(listener);
});
return mutable;
@@ -270,8 +287,7 @@ public abstract class InMemoryCommandStore extends CommandStore
case Key:
CommonAttributes.Mutable mutable = attrs.mutable();
forEach(keyOrRange, slice, key -> {
- SafeCommandsForKey cfk = safeStore.commandsForKey(key);
- Command.DurableAndIdempotentListener listener = cfk.register(command.current()).asListener();
+ Command.DurableAndIdempotentListener listener = CommandsForKeys.registerCommand(safeStore, key, command.current());
mutable.addListener(listener);
});
return mutable;
@@ -283,7 +299,20 @@ public abstract class InMemoryCommandStore extends CommandStore
return attrs;
}
- private <O> O mapReduceForKey(InMemorySafeStore safeStore, Routables<?> keysOrRanges, Ranges slice, BiFunction<CommandsForKey, O, O> map, O accumulate, Predicate<? super O> terminate)
+ private NavigableMap<RoutableKey, GlobalCommandsForKey> globalCommandsMapFor(KeyHistory keyHistory)
+ {
+ switch (keyHistory)
+ {
+ case DEPS:
+ return depsCommandsForKey;
+ case ALL:
+ return allCommandsForKey;
+ default:
+ throw new IllegalStateException(keyHistory + " does not have a commands map");
+ }
+ }
+
+ private <O> O mapReduceForKey(InMemorySafeStore safeStore, Routables<?> keysOrRanges, Ranges slice, KeyHistory keyHistory, TriFunction<TimestampsForKey, CommandsForKey, O, O> map, O accumulate, Predicate<? super O> terminate)
{
switch (keysOrRanges.domain()) {
default:
@@ -293,10 +322,25 @@ public abstract class InMemoryCommandStore extends CommandStore
for (Key key : keys)
{
if (!slice.contains(key)) continue;
- SafeCommandsForKey forKey = safeStore.ifLoadedAndInitialised(key);
- if (forKey.current() == null)
+ SafeTimestampsForKey timestamps = safeStore.timestampsIfLoadedAndInitialised(key);
+ if (timestamps.current() == null)
+ continue;
+
+ CommandsForKey commands = null;
+ switch (keyHistory)
+ {
+ case DEPS:
+ commands = safeStore.depsCommandsIfLoadedAndInitialised(key).current();
+ break;
+ case ALL:
+ commands = safeStore.allCommandsIfLoadedAndInitialised(key).current();
+ break;
+ }
+
+ if (commands == null && keyHistory != KeyHistory.NONE)
continue;
- accumulate = map.apply(forKey.current(), accumulate);
+
+ accumulate = map.apply(timestamps.current(), commands, accumulate);
if (terminate.test(accumulate))
return accumulate;
}
@@ -306,11 +350,25 @@ public abstract class InMemoryCommandStore extends CommandStore
Ranges sliced = ranges.slice(slice, Minimal);
for (Range range : sliced)
{
- for (GlobalCommandsForKey forKey : commandsForKey.subMap(range.start(), range.startInclusive(), range.end(), range.endInclusive()).values())
+ NavigableMap<RoutableKey, GlobalCommandsForKey> commandsMap = keyHistory.isNone() ? null : globalCommandsMapFor(keyHistory);
+ for (Map.Entry<RoutableKey, GlobalTimestampsForKey> entry : timestampsForKey.subMap(range.start(), range.startInclusive(), range.end(), range.endInclusive()).entrySet())
{
- if (forKey.value() == null)
+ RoutableKey key = entry.getKey();
+ TimestampsForKey timestamps = entry.getValue().value();
+
+ GlobalCommandsForKey globalCommands = keyHistory.isNone() ? null : commandsMap.get(key);
+ if (timestamps == null)
+ {
+ Invariants.checkState(globalCommands == null || globalCommands.value() == null);
+ continue;
+ }
+
+ if (globalCommands == null)
continue;
- accumulate = map.apply(forKey.value(), accumulate);
+
+ CommandsForKey commands = keyHistory.isNone() ? null : Invariants.nonNull(globalCommands.value());
+
+ accumulate = map.apply(timestamps, commands, accumulate);
if (terminate.test(accumulate))
return accumulate;
}
@@ -331,7 +389,9 @@ public abstract class InMemoryCommandStore extends CommandStore
case Range:
Ranges ranges = (Ranges) keysOrRanges;
ranges.slice(slice).forEach(range -> {
- commandsForKey.subMap(range.start(), range.startInclusive(), range.end(), range.endInclusive())
+ depsCommandsForKey.subMap(range.start(), range.startInclusive(), range.end(), range.endInclusive())
+ .keySet().forEach(forEach);
+ allCommandsForKey.subMap(range.start(), range.startInclusive(), range.end(), range.endInclusive())
.keySet().forEach(forEach);
});
}
@@ -350,7 +410,9 @@ public abstract class InMemoryCommandStore extends CommandStore
case Range:
Range range = (Range) keyOrRange;
Ranges.of(range).slice(slice).forEach(r -> {
- commandsForKey.subMap(r.start(), r.startInclusive(), r.end(), r.endInclusive())
+ depsCommandsForKey.subMap(r.start(), r.startInclusive(), r.end(), r.endInclusive())
+ .keySet().forEach(forEach);
+ allCommandsForKey.subMap(r.start(), r.startInclusive(), r.end(), r.endInclusive())
.keySet().forEach(forEach);
});
}
@@ -382,22 +444,50 @@ public abstract class InMemoryCommandStore extends CommandStore
}
});
ranges.forEach(r -> {
- commandsForKey.subMap(r.start(), r.startInclusive(), r.end(), r.endInclusive()).values().forEach(forKey -> {
+ depsCommandsForKey.subMap(r.start(), r.startInclusive(), r.end(), r.endInclusive()).values().forEach(forKey -> {
+ if (!forKey.isEmpty())
+ forKey.value(forKey.value().withoutRedundant(syncId));
+ });
+ allCommandsForKey.subMap(r.start(), r.startInclusive(), r.end(), r.endInclusive()).values().forEach(forKey -> {
if (!forKey.isEmpty())
forKey.value(forKey.value().withoutRedundant(syncId));
});
});
}
- protected InMemorySafeStore createSafeStore(PreLoadContext context, RangesForEpoch ranges, Map<TxnId, InMemorySafeCommand> commands, Map<RoutableKey, InMemorySafeCommandsForKey> commandsForKeys)
+ protected InMemorySafeStore createSafeStore(PreLoadContext context, RangesForEpoch ranges,
+ Map<TxnId, InMemorySafeCommand> commands,
+ Map<RoutableKey, InMemorySafeTimestampsForKey> timestampsForKey,
+ Map<RoutableKey, InMemorySafeCommandsForKey> minimalCommandsForKeys,
+ Map<RoutableKey, InMemorySafeCommandsForKey> completeCommandsForKeys)
{
- return new InMemorySafeStore(this, ranges, context, commands, commandsForKeys);
+ return new InMemorySafeStore(this, ranges, context, commands, timestampsForKey, minimalCommandsForKeys, completeCommandsForKeys);
+ }
+
+ private void loadCommandsForKey(RoutableKey key,
+ KeyHistory keyHistory,
+ Map<RoutableKey, InMemorySafeTimestampsForKey> timestampsForKey,
+ Map<RoutableKey, InMemorySafeCommandsForKey> minimalCommands,
+ Map<RoutableKey, InMemorySafeCommandsForKey> completeCommands)
+ {
+ switch (keyHistory)
+ {
+ case ALL:
+ // loading recovery commands implicitly loads deps
+ completeCommands.put(key, allCommandsForKey((Key) key).createSafeReference());
+ case DEPS:
+ minimalCommands.put(key, depsCommandsForKey((Key) key).createSafeReference());
+ break;
+ }
+ timestampsForKey.put(key, timestampsForKey((Key) key).createSafeReference());
}
protected final InMemorySafeStore createSafeStore(PreLoadContext context, RangesForEpoch ranges)
{
Map<TxnId, InMemorySafeCommand> commands = new HashMap<>();
- Map<RoutableKey, InMemorySafeCommandsForKey> commandsForKeys = new HashMap<>();
+ Map<RoutableKey, InMemorySafeCommandsForKey> minimalCommands = new HashMap<>();
+ Map<RoutableKey, InMemorySafeCommandsForKey> completeCommands = new HashMap<>();
+ Map<RoutableKey, InMemorySafeTimestampsForKey> timestampsForKey = new HashMap<>();
context.forEachId(txnId -> commands.put(txnId, lazyReference(txnId)));
@@ -407,13 +497,22 @@ public abstract class InMemoryCommandStore extends CommandStore
{
case Key:
RoutableKey key = (RoutableKey) seekable;
- commandsForKeys.put(key, commandsForKey((Key) key).createSafeReference());
+ switch (context.keyHistory())
+ {
+ case ALL:
+ // loading recovery commands implicitly loads deps
+ completeCommands.put(key, allCommandsForKey((Key) key).createSafeReference());
+ case DEPS:
+ minimalCommands.put(key, depsCommandsForKey((Key) key).createSafeReference());
+ break;
+ }
+ timestampsForKey.put(key, timestampsForKey((Key) key).createSafeReference());
break;
case Range:
// load range cfks here
}
}
- return createSafeStore(context, ranges, commands, commandsForKeys);
+ return createSafeStore(context, ranges, commands, timestampsForKey, minimalCommands, completeCommands);
}
public SafeCommandStore beginOperation(PreLoadContext context)
@@ -526,7 +625,7 @@ public abstract class InMemoryCommandStore extends CommandStore
private Command loadForCFK(CFKEntry entry)
{
- GlobalCommand globalCommand = ifPresent(entry);
+ GlobalCommand globalCommand = commandIfPresent(entry);
if (globalCommand != null && globalCommand.value() != null)
return globalCommand.value();
if (entry.uninitialised)
@@ -663,19 +762,45 @@ public abstract class InMemoryCommandStore extends CommandStore
}
}
- public static class InMemorySafeStore extends AbstractSafeCommandStore<InMemorySafeCommand, InMemorySafeCommandsForKey>
+ public static class GlobalTimestampsForKey extends GlobalState<TimestampsForKey>
+ {
+ private final Key key;
+
+ public GlobalTimestampsForKey(RoutableKey key)
+ {
+ this.key = (Key) key;
+ }
+
+ public InMemorySafeTimestampsForKey createSafeReference()
+ {
+ return new InMemorySafeTimestampsForKey(key, this);
+ }
+ }
+
+ public static class InMemorySafeStore extends AbstractSafeCommandStore<InMemorySafeCommand, InMemorySafeTimestampsForKey, InMemorySafeCommandsForKey, InMemorySafeCommandsForKeyUpdate>
{
private final InMemoryCommandStore commandStore;
private final Map<TxnId, InMemorySafeCommand> commands;
- private final Map<RoutableKey, InMemorySafeCommandsForKey> commandsForKey;
+ private final Map<RoutableKey, InMemorySafeTimestampsForKey> timestampsForKey;
+ private final Map<RoutableKey, InMemorySafeCommandsForKey> depsCommandsForKey;
+ private final Map<RoutableKey, InMemorySafeCommandsForKey> allCommandsForKey;
+ private final Map<RoutableKey, InMemorySafeCommandsForKeyUpdate> updatesForKey = new HashMap<>();
private final RangesForEpoch ranges;
- public InMemorySafeStore(InMemoryCommandStore commandStore, RangesForEpoch ranges, PreLoadContext context, Map<TxnId, InMemorySafeCommand> commands, Map<RoutableKey, InMemorySafeCommandsForKey> commandsForKey)
+ public InMemorySafeStore(InMemoryCommandStore commandStore,
+ RangesForEpoch ranges,
+ PreLoadContext context,
+ Map<TxnId, InMemorySafeCommand> commands,
+ Map<RoutableKey, InMemorySafeTimestampsForKey> timestampsForKey,
+ Map<RoutableKey, InMemorySafeCommandsForKey> depsCommandsForKey,
+ Map<RoutableKey, InMemorySafeCommandsForKey> allCommandsForKey)
{
super(context);
this.commandStore = commandStore;
this.commands = commands;
- this.commandsForKey = commandsForKey;
+ this.depsCommandsForKey = depsCommandsForKey;
+ this.allCommandsForKey = allCommandsForKey;
+ this.timestampsForKey = timestampsForKey;
this.ranges = Invariants.nonNull(ranges);
}
@@ -692,41 +817,139 @@ public abstract class InMemoryCommandStore extends CommandStore
}
@Override
- protected InMemorySafeCommandsForKey getCommandsForKeyInternal(RoutableKey key)
+ protected InMemorySafeTimestampsForKey getTimestampsForKeyInternal(RoutableKey key)
{
- return commandsForKey.get(key);
+ return timestampsForKey.get(key);
}
@Override
- protected void addCommandsForKeyInternal(InMemorySafeCommandsForKey cfk)
+ protected void addTimestampsForKeyInternal(InMemorySafeTimestampsForKey tfk)
{
- commandsForKey.put(cfk.key(), cfk);
+ timestampsForKey.put(tfk.key(), tfk);
+ }
+
+ @Override
+ protected InMemorySafeTimestampsForKey getTimestampsForKeyIfLoaded(RoutableKey key)
+ {
+ if (!commandStore.canExposeUnloaded())
+ return null;
+ GlobalTimestampsForKey global = commandStore.timestampsForKeyIfPresent((Key) key);
+ return global != null ? global.createSafeReference() : null;
}
@Override
protected InMemorySafeCommand getIfLoaded(TxnId txnId)
{
- InMemorySafeCommand safeCommand = commands.get(txnId);
- if (safeCommand != null)
- return safeCommand;
if (!commandStore.canExposeUnloaded())
return null;
- GlobalCommand global = commandStore.ifPresent(txnId);
+ GlobalCommand global = commandStore.commandIfPresent(txnId);
return global != null ? global.createSafeReference() : null;
}
@Override
- protected InMemorySafeCommandsForKey getIfLoaded(RoutableKey key)
+ protected InMemorySafeCommandsForKey getDepsCommandsForKeyInternal(RoutableKey key)
+ {
+ return depsCommandsForKey.get(key);
+ }
+
+ @Override
+ protected void addDepsCommandsForKeyInternal(InMemorySafeCommandsForKey cfk)
+ {
+ depsCommandsForKey.put(cfk.key(), cfk);
+ }
+
+ @Override
+ protected InMemorySafeCommandsForKey getDepsCommandsForKeyIfLoaded(RoutableKey key)
+ {
+ if (!commandStore.canExposeUnloaded())
+ return null;
+ GlobalCommandsForKey global = commandStore.depsCommandsForKeyIfPresent((Key) key);
+ return global != null ? global.createSafeReference() : null;
+ }
+
+ @Override
+ protected InMemorySafeCommandsForKey getAllCommandsForKeyInternal(RoutableKey key)
+ {
+ return allCommandsForKey.get(key);
+ }
+
+ @Override
+ protected void addAllCommandsForKeyInternal(InMemorySafeCommandsForKey cfk)
+ {
+ allCommandsForKey.put(cfk.key(), cfk);
+ }
+
+ @Override
+ protected InMemorySafeCommandsForKey getAllCommandsForKeyIfLoaded(RoutableKey key)
{
- InMemorySafeCommandsForKey safeCommandsForKey = commandsForKey.get((Key) key);
- if (safeCommandsForKey != null)
- return safeCommandsForKey;
if (!commandStore.canExposeUnloaded())
return null;
- GlobalCommandsForKey global = commandStore.ifPresent((Key) key);
+ GlobalCommandsForKey global = commandStore.allCommandsForKeyIfPresent((Key) key);
return global != null ? global.createSafeReference() : null;
}
+ @Override
+ protected InMemorySafeCommandsForKeyUpdate getCommandsForKeyUpdateInternal(RoutableKey key)
+ {
+ return updatesForKey.get(key);
+ }
+
+ @Override
+ protected InMemorySafeCommandsForKeyUpdate createCommandsForKeyUpdateInternal(RoutableKey key)
+ {
+ return new InMemorySafeCommandsForKeyUpdate((Key) key, (CommandLoader<CFKEntry>) cfkLoader(key));
+ }
+
+ @Override
+ protected void addCommandsForKeyUpdateInternal(InMemorySafeCommandsForKeyUpdate update)
+ {
+ updatesForKey.put(update.key(), update);
+ }
+
+ private CommandsForKey applyUpdate(CommandsForKey cfk, InMemorySafeCommandsForKeyUpdate update, KeyHistory keyHistory)
+ {
+ switch (keyHistory)
+ {
+ case DEPS:
+ return update.applyToDeps(cfk);
+ case ALL:
+ return update.applyToAll(cfk);
+ default:
+ throw new IllegalArgumentException("Unhandled KeyHistory " + keyHistory);
+ }
+ }
+
+ private void applyCommandForKeyUpdate(InMemorySafeCommandsForKeyUpdate update, KeyHistory kind, Map<RoutableKey, InMemorySafeCommandsForKey> commandsMap, Function<Key, GlobalCommandsForKey> getter)
+ {
+ Key key = update.key();
+ InMemorySafeCommandsForKey safeCFK = commandsMap.get(key);
+ if (safeCFK != null)
+ {
+ if (safeCFK.isEmpty()) safeCFK.initialize(cfkLoader(key));
+ safeCFK.update(applyUpdate(safeCFK.current(), update, kind));
+ }
+ else
+ {
+ GlobalCommandsForKey globalCFK = getter.apply(key);
+ CommandsForKey cfk = globalCFK.value();
+ if (cfk == null)
+ cfk = new CommandsForKey(key, cfkLoader(key));
+ globalCFK.value(applyUpdate(cfk, update, kind));
+ }
+ }
+
+ private void applyCommandForKeyUpdate(InMemorySafeCommandsForKeyUpdate update)
+ {
+ applyCommandForKeyUpdate(update, KeyHistory.DEPS, depsCommandsForKey, commandStore::depsCommandsForKey);
+ applyCommandForKeyUpdate(update, KeyHistory.ALL, allCommandsForKey, commandStore::allCommandsForKey);
+ }
+
+ @Override
+ protected void applyCommandForKeyUpdates()
+ {
+ updatesForKey.values().forEach(this::applyCommandForKeyUpdate);
+ }
+
@Override
public CommandStore commandStore()
{
@@ -760,7 +983,7 @@ public abstract class InMemoryCommandStore extends CommandStore
@Override
public Timestamp maxConflict(Seekables<?, ?> keysOrRanges, Ranges slice)
{
- Timestamp timestamp = commandStore.mapReduceForKey(this, keysOrRanges, slice, (forKey, prev) -> Timestamp.nonNullOrMax(forKey.max(), prev), Timestamp.NONE, Objects::isNull);
+ Timestamp timestamp = commandStore.mapReduceForKey(this, keysOrRanges, slice, KeyHistory.NONE, (timestamps, commands, prev) -> Timestamp.nonNullOrMax(timestamps.max(), prev), Timestamp.NONE, Objects::isNull);
Seekables<?, ?> sliced = keysOrRanges.slice(slice, Minimal);
for (RangeCommand command : commandStore.rangeCommands.values())
{
@@ -784,7 +1007,6 @@ public abstract class InMemoryCommandStore extends CommandStore
RangesForEpoch rangesForEpoch = commandStore.rangesForEpoch;
Ranges allRanges = rangesForEpoch.all();
deps.keyDeps.keys().forEach(allRanges, key -> {
- SafeCommandsForKey cfk = commandsForKey(key);
deps.keyDeps.forEach(key, txnId -> {
// TODO (desired, efficiency): this can be made more efficient by batching by epoch
if (rangesForEpoch.coordinates(txnId).contains(key))
@@ -792,7 +1014,7 @@ public abstract class InMemoryCommandStore extends CommandStore
if (!rangesForEpoch.allBefore(txnId.epoch()).contains(key))
return;
- cfk.registerNotWitnessed(txnId);
+ CommandsForKeys.registerNotWitnessed(this, key, txnId);
});
});
@@ -819,22 +1041,46 @@ public abstract class InMemoryCommandStore extends CommandStore
return commandStore.time;
}
+ private static class TxnInfo
+ {
+ private final TxnId txnId;
+ private final Timestamp executeAt;
+ private final Status status;
+ private final List<TxnId> deps;
+
+ public TxnInfo(TxnId txnId, Timestamp executeAt, Status status, List<TxnId> deps)
+ {
+ this.txnId = txnId;
+ this.executeAt = executeAt;
+ this.status = status;
+ this.deps = deps;
+ }
+
+ public TxnInfo(Command command)
+ {
+ this.txnId = command.txnId();
+ this.executeAt = command.executeAt();
+ this.status = command.status();
+ PartialDeps deps = command.partialDeps();
+ this.deps = deps != null ? deps.txnIds() : Collections.emptyList();
+ }
+ }
// TODO (expected): instead of accepting a slice, accept the min/max epoch and let implementation handle it
@Override
- public <P1, T> T mapReduce(Seekables<?, ?> keysOrRanges, Ranges slice, Kinds testKind, TestTimestamp testTimestamp, Timestamp timestamp, TestDep testDep, @Nullable TxnId depId, @Nullable Status minStatus, @Nullable Status maxStatus, CommandFunction<P1, T, T> map, P1 p1, T accumulate, Predicate<? super T> terminate)
+ public <P1, T> T mapReduce(Seekables<?, ?> keysOrRanges, Ranges slice, KeyHistory keyHistory, Kinds testKind, TestTimestamp testTimestamp, Timestamp timestamp, TestDep testDep, @Nullable TxnId depId, @Nullable Status minStatus, @Nullable Status maxStatus, CommandFunction<P1, T, T> map, P1 p1, T accumulate, Predicate<? super T> terminate)
{
- accumulate = commandStore.mapReduceForKey(this, keysOrRanges, slice, (forKey, prev) -> {
- CommandTimeseries<?> timeseries;
+ accumulate = commandStore.mapReduceForKey(this, keysOrRanges, slice, keyHistory, (timestamps, commands, prev) -> {
+ CommandTimeseries.TimestampType timestampType;
switch (testTimestamp)
{
default: throw new AssertionError();
case STARTED_AFTER:
case STARTED_BEFORE:
- timeseries = forKey.byId();
+ timestampType = CommandTimeseries.TimestampType.TXN_ID;
break;
case EXECUTES_AFTER:
case MAY_EXECUTE_BEFORE:
- timeseries = forKey.byExecuteAt();
+ timestampType = CommandTimeseries.TimestampType.EXECUTE_AT;
}
CommandTimeseries.TestTimestamp remapTestTimestamp;
switch (testTimestamp)
@@ -848,7 +1094,7 @@ public abstract class InMemoryCommandStore extends CommandStore
case MAY_EXECUTE_BEFORE:
remapTestTimestamp = CommandTimeseries.TestTimestamp.BEFORE;
}
- return timeseries.mapReduce(testKind, remapTestTimestamp, timestamp, testDep, depId, minStatus, maxStatus, map, p1, prev, terminate);
+ return commands.commands().mapReduce(testKind, timestampType, remapTestTimestamp, timestamp, testDep, depId, minStatus, maxStatus, map, p1, prev, terminate);
}, accumulate, terminate);
if (terminate.test(accumulate))
@@ -856,7 +1102,7 @@ public abstract class InMemoryCommandStore extends CommandStore
// TODO (find lib, efficiency): this is super inefficient, need to store Command in something queryable
Seekables<?, ?> sliced = keysOrRanges.slice(slice, Minimal);
- Map<Range, List<Map.Entry<TxnId, TimestampAndStatus>>> collect = new TreeMap<>(Range::compare);
+ Map<Range, List<TxnInfo>> collect = new TreeMap<>(Range::compare);
commandStore.rangeCommands.forEach(((txnId, rangeCommand) -> {
Command command = rangeCommand.command.value();
// TODO (now): probably this isn't safe - want to ensure we take dependency on any relevant syncId
@@ -916,9 +1162,9 @@ public abstract class InMemoryCommandStore extends CommandStore
Routables.foldl(rangeCommand.ranges, sliced, (r, in, i) -> {
// TODO (easy, efficiency): pass command as a parameter to Fold
- List<Map.Entry<TxnId, TimestampAndStatus>> list = in.computeIfAbsent(r, ignore -> new ArrayList<>());
- if (list.isEmpty() || !list.get(list.size() - 1).getKey().equals(command.txnId()))
- list.add(new AbstractMap.SimpleImmutableEntry<>(command.txnId(), new TimestampAndStatus(command.executeAt(), command.status())));
+ List<TxnInfo> list = in.computeIfAbsent(r, ignore -> new ArrayList<>());
+ if (list.isEmpty() || !list.get(list.size() - 1).txnId.equals(command.txnId()))
+ list.add(new TxnInfo(command));
return in;
}, collect);
}));
@@ -947,20 +1193,33 @@ public abstract class InMemoryCommandStore extends CommandStore
Routables.foldl(ranges, sliced, (r, in, i) -> {
// TODO (easy, efficiency): pass command as a parameter to Fold
- List<Map.Entry<TxnId, TimestampAndStatus>> list = in.computeIfAbsent(r, ignore -> new ArrayList<>());
- if (list.isEmpty() || !list.get(list.size() - 1).getKey().equals(txnId))
- list.add(new AbstractMap.SimpleImmutableEntry<>(txnId, new TimestampAndStatus(txnId, Status.NotDefined)));
+ List<TxnInfo> list = in.computeIfAbsent(r, ignore -> new ArrayList<>());
+ if (list.isEmpty() || !list.get(list.size() - 1).txnId.equals(txnId))
+ {
+ GlobalCommand global = commandStore.commands.get(txnId);
+ if (global != null && global.value() != null)
+ {
+ Command command = global.value();
+ PartialDeps deps = command.partialDeps();
+ List<TxnId> depsIds = deps != null ? deps.txnIds() : Collections.emptyList();
+ list.add(new TxnInfo(txnId, txnId, command.status(), depsIds));
+ }
+ else
+ {
+ list.add(new TxnInfo(txnId, txnId, NotDefined, Collections.emptyList()));
+ }
+ }
return in;
}, collect);
}));
}
- for (Map.Entry<Range, List<Map.Entry<TxnId, TimestampAndStatus>>> e : collect.entrySet())
+ for (Map.Entry<Range, List<TxnInfo>> e : collect.entrySet())
{
- for (Map.Entry<TxnId, TimestampAndStatus> command : e.getValue())
+ for (TxnInfo command : e.getValue())
{
T initial = accumulate;
- accumulate = map.apply(p1, e.getKey(), command.getKey(), command.getValue().timestamp, command.getValue().status, initial);
+ accumulate = map.apply(p1, e.getKey(), command.txnId, command.executeAt, command.status, () -> command.deps, initial);
if (terminate.test(accumulate))
return accumulate;
}
@@ -1002,7 +1261,10 @@ public abstract class InMemoryCommandStore extends CommandStore
}
});
commands.values().forEach(SafeState::invalidate);
- commandsForKey.values().forEach(SafeState::invalidate);
+ timestampsForKey.values().forEach(SafeState::invalidate);
+ depsCommandsForKey.values().forEach(SafeState::invalidate);
+ allCommandsForKey.values().forEach(SafeState::invalidate);
+ updatesForKey.values().forEach(SafeState::invalidate);
}
@Override
@@ -1169,9 +1431,15 @@ public abstract class InMemoryCommandStore extends CommandStore
{
class DebugSafeStore extends InMemorySafeStore
{
- public DebugSafeStore(InMemoryCommandStore commandStore, RangesForEpoch ranges, PreLoadContext context, Map<TxnId, InMemorySafeCommand> commands, Map<RoutableKey, InMemorySafeCommandsForKey> commandsForKey)
+ public DebugSafeStore(InMemoryCommandStore commandStore,
+ RangesForEpoch ranges,
+ PreLoadContext context,
+ Map<TxnId, InMemorySafeCommand> commands,
+ Map<RoutableKey, InMemorySafeTimestampsForKey> timestampsForKey,
+ Map<RoutableKey, InMemorySafeCommandsForKey> minimalCommandsForKey,
+ Map<RoutableKey, InMemorySafeCommandsForKey> completeCommandsForKey)
{
- super(commandStore, ranges, context, commands, commandsForKey);
+ super(commandStore, ranges, context, commands, timestampsForKey, minimalCommandsForKey, completeCommandsForKey);
}
@Override
@@ -1209,11 +1477,10 @@ public abstract class InMemoryCommandStore extends CommandStore
}
@Override
- protected InMemorySafeStore createSafeStore(PreLoadContext context, RangesForEpoch ranges, Map<TxnId, InMemorySafeCommand> commands, Map<RoutableKey, InMemorySafeCommandsForKey> commandsForKeys)
+ protected InMemorySafeStore createSafeStore(PreLoadContext context, RangesForEpoch ranges, Map<TxnId, InMemorySafeCommand> commands, Map<RoutableKey, InMemorySafeTimestampsForKey> timestampsForKeyMap, Map<RoutableKey, InMemorySafeCommandsForKey> minimalCommandsForKeys, Map<RoutableKey, InMemorySafeCommandsForKey> completeCommandsForKeys)
{
- return new DebugSafeStore(this, ranges, context, commands, commandsForKeys);
+ return new DebugSafeStore(this, ranges, context, commands, timestampsForKeyMap, minimalCommandsForKeys, completeCommandsForKeys);
}
-
}
public static InMemoryCommandStore inMemory(CommandStore unsafeStore)
diff --git a/accord-core/src/main/java/accord/impl/InMemorySafeCommandsForKeyUpdate.java b/accord-core/src/main/java/accord/impl/InMemorySafeCommandsForKeyUpdate.java
new file mode 100644
index 00000000..2c7f9557
--- /dev/null
+++ b/accord-core/src/main/java/accord/impl/InMemorySafeCommandsForKeyUpdate.java
@@ -0,0 +1,60 @@
+/*
+ * 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 accord.impl;
+
+import accord.api.Key;
+import accord.impl.CommandTimeseries.CommandLoader;
+import accord.impl.InMemoryCommandStore.CFKEntry;
+import accord.utils.Invariants;
+
+public class InMemorySafeCommandsForKeyUpdate extends SafeCommandsForKey.Update<CommandsForKey.Update, CFKEntry>
+{
+ private static final CommandsForKey.Update VALUE = new CommandsForKey.Update() {};
+
+ private boolean invalidated = false;
+
+ public InMemorySafeCommandsForKeyUpdate(Key key, CommandLoader<CFKEntry> loader)
+ {
+ super(key, loader);
+ }
+
+ @Override
+ public void initialize()
+ {
+ Invariants.checkState(isEmpty());
+ }
+
+ @Override
+ public CommandsForKey.Update current()
+ {
+ return VALUE;
+ }
+
+ @Override
+ public void invalidate()
+ {
+ invalidated = true;
+ }
+
+ @Override
+ public boolean invalidated()
+ {
+ return invalidated;
+ }
+}
diff --git a/accord-core/src/main/java/accord/impl/CommandTimeseriesHolder.java b/accord-core/src/main/java/accord/impl/InMemorySafeTimestampsForKey.java
similarity index 52%
copy from accord-core/src/main/java/accord/impl/CommandTimeseriesHolder.java
copy to accord-core/src/main/java/accord/impl/InMemorySafeTimestampsForKey.java
index f0a88e75..9f96454d 100644
--- a/accord-core/src/main/java/accord/impl/CommandTimeseriesHolder.java
+++ b/accord-core/src/main/java/accord/impl/InMemorySafeTimestampsForKey.java
@@ -18,12 +18,41 @@
package accord.impl;
-import accord.primitives.Timestamp;
+import accord.api.Key;
+import accord.impl.InMemoryCommandStore.GlobalTimestampsForKey;
-public interface CommandTimeseriesHolder
+public class InMemorySafeTimestampsForKey extends SafeTimestampsForKey
{
- CommandTimeseries<?> byId();
- CommandTimeseries<?> byExecuteAt();
+ private boolean invalidated = false;
+ private final GlobalTimestampsForKey global;
- Timestamp max();
+ public InMemorySafeTimestampsForKey(Key key, GlobalTimestampsForKey global)
+ {
+ super(key);
+ this.global = global;
+ }
+
+ @Override
+ public TimestampsForKey current()
+ {
+ return global.value();
+ }
+
+ @Override
+ protected void set(TimestampsForKey update)
+ {
+ global.value(update);
+ }
+
+ @Override
+ public void invalidate()
+ {
+ invalidated = true;
+ }
+
+ @Override
+ public boolean invalidated()
+ {
+ return invalidated;
+ }
}
diff --git a/accord-core/src/main/java/accord/impl/SafeCommandsForKey.java b/accord-core/src/main/java/accord/impl/SafeCommandsForKey.java
index 1e5e41b0..df1a749f 100644
--- a/accord-core/src/main/java/accord/impl/SafeCommandsForKey.java
+++ b/accord-core/src/main/java/accord/impl/SafeCommandsForKey.java
@@ -19,23 +19,10 @@
package accord.impl;
import accord.api.Key;
-import accord.api.VisibleForImplementation;
import accord.impl.CommandTimeseries.CommandLoader;
-import accord.local.Command;
-import accord.local.SafeCommandStore;
-import accord.primitives.Timestamp;
-import accord.primitives.TxnId;
-import accord.utils.Invariants;
-import com.google.common.annotations.VisibleForTesting;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import static accord.local.Command.NotDefined.uninitialised;
public abstract class SafeCommandsForKey implements SafeState<CommandsForKey>
{
- private static final Logger logger = LoggerFactory.getLogger(SafeCommandsForKey.class);
-
private final Key key;
public SafeCommandsForKey(Key key)
@@ -50,7 +37,7 @@ public abstract class SafeCommandsForKey implements SafeState<CommandsForKey>
return key;
}
- private CommandsForKey update(CommandsForKey update)
+ CommandsForKey update(CommandsForKey update)
{
set(update);
return update;
@@ -61,149 +48,20 @@ public abstract class SafeCommandsForKey implements SafeState<CommandsForKey>
return update(new CommandsForKey(key, loader));
}
- @VisibleForTesting
- @VisibleForImplementation
- public static Timestamp updateMax(CommandsForKey cfk, Timestamp timestamp)
- {
- Invariants.checkArgument(cfk != null || timestamp != null);
- if (cfk == null)
- return timestamp;
- if (timestamp == null)
- return cfk.max();
- return Timestamp.max(cfk.max(), timestamp);
- }
-
- @VisibleForTesting
- @VisibleForImplementation
- public <D> CommandsForKey updateMax(Timestamp timestamp)
- {
- CommandsForKey current = current();
- return update(new CommandsForKey(current.key(),
- updateMax(current, timestamp),
- current.lastExecutedTimestamp(),
- current.rawLastExecutedHlc(),
- current.lastWriteTimestamp(),
- (CommandTimeseries<D>) current().byId(),
- (CommandTimeseries<D>) current().byExecuteAt()));
- }
-
- public <D> CommandsForKey register(Command command)
- {
- CommandsForKey current = current();
- CommandTimeseries.Update<D> byId = (CommandTimeseries.Update<D>) current().byId().beginUpdate();
- CommandTimeseries.Update<D> byExecuteAt = (CommandTimeseries.Update<D>) current().byExecuteAt().beginUpdate();
- return update(new CommandsForKey(current.key(),
- updateMax(current, command.executeAt()),
- current.lastExecutedTimestamp(),
- current.lastExecutedHlc(),
- current.lastWriteTimestamp(),
- byId.add(command.txnId(), command).build(),
- byExecuteAt.add(command.txnId(), command).build() ));
- }
-
- public <D> CommandsForKey registerNotWitnessed(TxnId txnId)
+ public static abstract class Update<U extends CommandsForKey.Update, D> extends CommandsForKeyGroupUpdater.Mutable<D> implements SafeState<U>
{
- CommandsForKey current = current();
- if (current.byId().commands.containsKey(txnId))
- return current;
+ private final Key key;
- CommandTimeseries.Update<D> byId = (CommandTimeseries.Update<D>) current().byId().beginUpdate();
- CommandTimeseries.Update<D> byExecuteAt = (CommandTimeseries.Update<D>) current().byExecuteAt().beginUpdate();
- return update(new CommandsForKey(current.key(),
- updateMax(current, txnId),
- current.lastExecutedTimestamp(),
- current.lastExecutedHlc(),
- current.lastWriteTimestamp(),
- byId.add(txnId, uninitialised(txnId)).build(),
- byExecuteAt.add(txnId, uninitialised(txnId)).build()));
- }
-
- public <D> CommandsForKey listenerUpdate(Command command)
- {
- if (logger.isTraceEnabled())
- logger.trace("[{}]: updating as listener in response to change on {} with status {} ({})",
- key(), command.txnId(), command.status(), command);
-
- CommandsForKey current = current();
- CommandTimeseries.Update<D> byId = (CommandTimeseries.Update<D>) current().byId().beginUpdate();
- CommandTimeseries.Update<D> byExecuteAt = (CommandTimeseries.Update<D>) current().byExecuteAt().beginUpdate();
-
- // add/remove the command on every listener update to avoid
- // special denormalization handling in Cassandra
- switch (command.status())
+ public Update(Key key, CommandLoader<D> loader)
{
- default: throw new AssertionError();
- case PreAccepted:
- case NotDefined:
- case Accepted:
- case AcceptedInvalidate:
- case PreCommitted:
- byId.add(command.txnId(), command);
- byExecuteAt.add(command.txnId(), command);
- break;
- case Applied:
- case PreApplied:
- case Committed:
- case ReadyToExecute:
- byId.add(command.txnId(), command);
- byExecuteAt.remove(command.txnId());
- byExecuteAt.add(command.executeAt(), command);
- break;
- case Invalidated:
- byId.remove(command.txnId());
- byExecuteAt.remove(command.txnId());
- case Truncated:
- break;
+ super(loader);
+ this.key = key;
}
- return update(new CommandsForKey(current.key(),
- updateMax(current, command.executeAt()),
- current.lastExecutedTimestamp(),
- current.lastExecutedHlc(),
- current.lastWriteTimestamp(),
- byId.build(),
- byExecuteAt.build()));
- }
-
- @VisibleForImplementation
- public <D> CommandsForKey updateLastExecutionTimestamps(SafeCommandStore safeStore, Timestamp executeAt, boolean isForWriteTxn)
- {
- CommandsForKey current = current();
-
- Timestamp lastWrite = current.lastWriteTimestamp();
-
- if (executeAt.compareTo(lastWrite) < 0)
+ public Key key()
{
- if (!safeStore.commandStore().safeToReadAt(executeAt).contains(key.toUnseekable()))
- return current;
- throw new IllegalArgumentException(String.format("%s is less than the most recent write timestamp %s", executeAt, lastWrite));
+ return key;
}
-
- Timestamp lastExecuted = current.lastExecutedTimestamp();
- int cmp = executeAt.compareTo(lastExecuted);
- // execute can be in the past if it's for a read and after the most recent write
- if (cmp == 0 || (!isForWriteTxn && cmp < 0))
- return current;
- if (cmp < 0)
- {
- if (!safeStore.commandStore().safeToReadAt(executeAt).contains(key.toUnseekable()))
- return current;
- throw new IllegalArgumentException(String.format("%s is less than the most recent executed timestamp %s", executeAt, lastExecuted));
- }
-
- long micros = executeAt.hlc();
- long lastMicros = current.lastExecutedHlc();
-
- Timestamp lastExecutedTimestamp = executeAt;
- long lastExecutedHlc = micros > lastMicros ? Long.MIN_VALUE : lastMicros + 1;
- Timestamp lastWriteTimestamp = isForWriteTxn ? executeAt : current.lastWriteTimestamp();
-
- return update(new CommandsForKey(current.key(),
- current.max(),
- lastExecutedTimestamp,
- lastExecutedHlc,
- lastWriteTimestamp,
- (CommandTimeseries<D>) current.byId(),
- (CommandTimeseries<D>) current.byExecuteAt()));
+ public abstract void initialize();
}
}
diff --git a/accord-core/src/main/java/accord/impl/SafeTimestampsForKey.java b/accord-core/src/main/java/accord/impl/SafeTimestampsForKey.java
new file mode 100644
index 00000000..ee5d58c1
--- /dev/null
+++ b/accord-core/src/main/java/accord/impl/SafeTimestampsForKey.java
@@ -0,0 +1,87 @@
+/*
+ * 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 accord.impl;
+
+import accord.api.Key;
+import accord.api.VisibleForImplementation;
+import accord.primitives.Timestamp;
+import accord.utils.Invariants;
+import com.google.common.annotations.VisibleForTesting;
+
+public abstract class SafeTimestampsForKey implements SafeState<TimestampsForKey>
+{
+ private final Key key;
+
+ public SafeTimestampsForKey(Key key)
+ {
+ this.key = key;
+ }
+
+ protected abstract void set(TimestampsForKey update);
+
+ public Key key()
+ {
+ return key;
+ }
+
+ private TimestampsForKey update(TimestampsForKey update)
+ {
+ set(update);
+ return update;
+ }
+
+ public TimestampsForKey initialize()
+ {
+ return update(new TimestampsForKey(key));
+ }
+
+ @VisibleForTesting
+ @VisibleForImplementation
+ public static Timestamp updateMax(TimestampsForKey tfk, Timestamp timestamp)
+ {
+ Invariants.checkArgument(tfk != null || timestamp != null);
+ if (tfk == null)
+ return timestamp;
+ if (timestamp == null)
+ return tfk.max();
+ return Timestamp.max(tfk.max(), timestamp);
+ }
+
+ @VisibleForTesting
+ @VisibleForImplementation
+ public <D> TimestampsForKey updateMax(Timestamp timestamp)
+ {
+ TimestampsForKey current = current();
+ return update(new TimestampsForKey(current.key(),
+ updateMax(current, timestamp),
+ current.lastExecutedTimestamp(),
+ current.rawLastExecutedHlc(),
+ current.lastWriteTimestamp()));
+ }
+
+ <D> TimestampsForKey updateLastExecutionTimestamps(Timestamp lastExecutedTimestamp, long lastExecutedHlc, Timestamp lastWriteTimestamp)
+ {
+ TimestampsForKey current = current();
+ return update(new TimestampsForKey(current.key(),
+ current.max(),
+ lastExecutedTimestamp,
+ lastExecutedHlc,
+ lastWriteTimestamp));
+ }
+}
diff --git a/accord-core/src/main/java/accord/impl/TimestampsForKey.java b/accord-core/src/main/java/accord/impl/TimestampsForKey.java
new file mode 100644
index 00000000..65fc094c
--- /dev/null
+++ b/accord-core/src/main/java/accord/impl/TimestampsForKey.java
@@ -0,0 +1,150 @@
+/*
+ * 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 accord.impl;
+
+import java.util.Objects;
+
+import accord.api.Key;
+import accord.primitives.Timestamp;
+import accord.primitives.TxnId;
+
+public class TimestampsForKey implements DomainTimestamps
+{
+ public static final long NO_LAST_EXECUTED_HLC = Long.MIN_VALUE;
+
+ public static class SerializerSupport
+ {
+ public static <D> TimestampsForKey create(Key key,
+ Timestamp max,
+ Timestamp lastExecutedTimestamp,
+ long lastExecutedHlc,
+ Timestamp lastWriteTimestamp)
+ {
+ return new TimestampsForKey(key, max, lastExecutedTimestamp, lastExecutedHlc, lastWriteTimestamp);
+ }
+ }
+
+ private final Key key;
+ private final Timestamp max;
+ private final Timestamp lastExecutedTimestamp;
+ // TODO (desired): we have leaked C* implementation details here
+ private final long rawLastExecutedHlc;
+ private final Timestamp lastWriteTimestamp;
+
+ public TimestampsForKey(Key key, Timestamp max, Timestamp lastExecutedTimestamp, long rawLastExecutedHlc, Timestamp lastWriteTimestamp)
+ {
+ this.key = key;
+ this.max = max;
+ this.lastExecutedTimestamp = lastExecutedTimestamp;
+ this.rawLastExecutedHlc = rawLastExecutedHlc;
+ this.lastWriteTimestamp = lastWriteTimestamp;
+ }
+
+ public TimestampsForKey(Key key)
+ {
+ this.key = key;
+ this.max = Timestamp.NONE;
+ this.lastExecutedTimestamp = Timestamp.NONE;
+ this.rawLastExecutedHlc = 0;
+ this.lastWriteTimestamp = Timestamp.NONE;
+ }
+
+ public boolean equals(Object o)
+ {
+ if (this == o) return true;
+ if (o == null || getClass() != o.getClass()) return false;
+ TimestampsForKey that = (TimestampsForKey) o;
+ return rawLastExecutedHlc == that.rawLastExecutedHlc && Objects.equals(key, that.key) && Objects.equals(max, that.max) && Objects.equals(lastExecutedTimestamp, that.lastExecutedTimestamp) && Objects.equals(lastWriteTimestamp, that.lastWriteTimestamp);
+ }
+
+ public int hashCode()
+ {
+ throw new UnsupportedOperationException();
+ }
+
+ public Key key()
+ {
+ return key;
+ }
+
+ public Timestamp max()
+ {
+ return max;
+ }
+
+ public Timestamp lastExecutedTimestamp()
+ {
+ return lastExecutedTimestamp;
+ }
+
+ public long lastExecutedHlc()
+ {
+ return rawLastExecutedHlc == NO_LAST_EXECUTED_HLC ? lastExecutedTimestamp.hlc() : rawLastExecutedHlc;
+ }
+
+ public long rawLastExecutedHlc()
+ {
+ return rawLastExecutedHlc;
+ }
+
+ public Timestamp lastWriteTimestamp()
+ {
+ return lastWriteTimestamp;
+ }
+
+ public TimestampsForKey withoutRedundant(TxnId redundantBefore)
+ {
+ return new TimestampsForKey(key, max.compareTo(redundantBefore) < 0 ? Timestamp.NONE : max,
+ lastExecutedTimestamp.compareTo(redundantBefore) < 0 ? Timestamp.NONE : lastExecutedTimestamp,
+ rawLastExecutedHlc < redundantBefore.hlc() ? NO_LAST_EXECUTED_HLC : rawLastExecutedHlc,
+ lastWriteTimestamp.compareTo(redundantBefore) < 0 ? Timestamp.NONE : lastWriteTimestamp);
+ }
+
+ public void validateExecuteAtTime(Timestamp executeAt, boolean isForWriteTxn)
+ {
+ if (executeAt.compareTo(lastWriteTimestamp) < 0)
+ throw new IllegalArgumentException(String.format("%s is less than the most recent write timestamp %s", executeAt, lastWriteTimestamp));
+
+ int cmp = executeAt.compareTo(lastExecutedTimestamp);
+ // execute can be in the past if it's for a read and after the most recent write
+ if (cmp == 0 || (!isForWriteTxn && cmp < 0))
+ return;
+ if (cmp < 0)
+ throw new IllegalArgumentException(String.format("%s is less than the most recent executed timestamp %s", executeAt, lastExecutedTimestamp));
+ else
+ throw new IllegalArgumentException(String.format("%s is greater than the most recent executed timestamp, cfk should be updated", executeAt, lastExecutedTimestamp));
+ }
+
+ public long hlcFor(Timestamp executeAt, boolean isForWriteTxn)
+ {
+ validateExecuteAtTime(executeAt, isForWriteTxn);
+ return rawLastExecutedHlc == NO_LAST_EXECUTED_HLC ? lastExecutedTimestamp.hlc() : rawLastExecutedHlc;
+ }
+
+ public String toString()
+ {
+ return "TimestampsForKey@" + System.identityHashCode(this) + '{' +
+ "key=" + key +
+ ", max=" + max +
+ ", lastExecutedTimestamp=" + lastExecutedTimestamp +
+ ", rawLastExecutedHlc=" + rawLastExecutedHlc +
+ ", lastWriteTimestamp=" + lastWriteTimestamp +
+ '}';
+ }
+}
diff --git a/accord-core/src/main/java/accord/local/Bootstrap.java b/accord-core/src/main/java/accord/local/Bootstrap.java
index 0652fd3c..f852f970 100644
--- a/accord-core/src/main/java/accord/local/Bootstrap.java
+++ b/accord-core/src/main/java/accord/local/Bootstrap.java
@@ -135,7 +135,7 @@ class Bootstrap
CoordinateSyncPoint.exclusive(node, globalSyncId, commitRanges)
// TODO (correcness) : PreLoadContext only works with Seekables, which doesn't allow mixing Keys and Ranges... But Deps has both Keys AND Ranges!
// ATM all known implementations store ranges in-memory, but this will not be true soon, so this will need to be addressed
- .flatMap(syncPoint -> node.withEpoch(epoch, () -> store.submit(contextFor(localSyncId, syncPoint.waitFor.keyDeps.keys()), safeStore1 -> {
+ .flatMap(syncPoint -> node.withEpoch(epoch, () -> store.submit(contextFor(localSyncId, syncPoint.waitFor.keyDeps.keys(), KeyHistory.DEPS), safeStore1 -> {
if (valid.isEmpty()) // we've lost ownership of the range
return AsyncResults.success(Ranges.EMPTY);
diff --git a/accord-core/src/main/java/accord/local/CommandStore.java b/accord-core/src/main/java/accord/local/CommandStore.java
index 3ba61142..737afe60 100644
--- a/accord-core/src/main/java/accord/local/CommandStore.java
+++ b/accord-core/src/main/java/accord/local/CommandStore.java
@@ -20,6 +20,7 @@ package accord.local;
import accord.api.ProgressLog;
import accord.api.DataStore;
+import accord.api.VisibleForImplementationTesting;
import accord.coordinate.CollectDeps;
import accord.local.Command.WaitingOn;
@@ -222,11 +223,11 @@ public abstract class CommandStore implements AgentExecutor
private static Timestamp maxApplied(SafeCommandStore safeStore, Seekables<?, ?> keysOrRanges, Ranges slice)
{
- return safeStore.mapReduce(keysOrRanges, slice, Kinds.Ws,
+ return safeStore.mapReduce(keysOrRanges, slice, KeyHistory.NONE, Kinds.Ws,
SafeCommandStore.TestTimestamp.STARTED_AFTER, Timestamp.NONE,
SafeCommandStore.TestDep.ANY_DEPS, null,
Status.PreApplied, Status.Truncated,
- (p1, key, txnId, executeAt, status, max) -> Timestamp.max(max, executeAt),
+ (p1, key, txnId, executeAt, status, deps, max) -> Timestamp.max(max, executeAt),
null, Timestamp.NONE, Timestamp.MAX::equals);
}
@@ -547,8 +548,11 @@ public abstract class CommandStore implements AgentExecutor
return redundantBefore;
}
- protected NavigableMap<TxnId, Ranges> bootstrapBeganAt() { return bootstrapBeganAt; }
- protected NavigableMap<Timestamp, Ranges> safeToRead() { return safeToRead; }
+ @VisibleForImplementationTesting
+ public NavigableMap<TxnId, Ranges> bootstrapBeganAt() { return bootstrapBeganAt; }
+
+ @VisibleForImplementationTesting
+ public NavigableMap<Timestamp, Ranges> safeToRead() { return safeToRead; }
public final boolean isRejectedIfNotPreAccepted(TxnId txnId, Unseekables<?> participants)
{
diff --git a/accord-core/src/main/java/accord/local/Commands.java b/accord-core/src/main/java/accord/local/Commands.java
index 88e925b7..bbcd148c 100644
--- a/accord-core/src/main/java/accord/local/Commands.java
+++ b/accord-core/src/main/java/accord/local/Commands.java
@@ -915,7 +915,7 @@ public class Commands
safeStore.commandStore().redundantBefore(), safeStore.commandStore().durableBefore());
}
- static Cleanup shouldCleanup(TxnId txnId, Status status, Durability durability, EpochSupplier toEpoch, Route<?> route, RedundantBefore redundantBefore, DurableBefore durableBefore)
+ public static Cleanup shouldCleanup(TxnId txnId, Status status, Durability durability, EpochSupplier toEpoch, Route<?> route, RedundantBefore redundantBefore, DurableBefore durableBefore)
{
return shouldCleanup(txnId, status, durability, toEpoch, route, redundantBefore, durableBefore, true);
}
diff --git a/accord-core/src/main/java/accord/impl/CommandTimeseriesHolder.java b/accord-core/src/main/java/accord/local/KeyHistory.java
similarity index 75%
rename from accord-core/src/main/java/accord/impl/CommandTimeseriesHolder.java
rename to accord-core/src/main/java/accord/local/KeyHistory.java
index f0a88e75..d5c9c3d4 100644
--- a/accord-core/src/main/java/accord/impl/CommandTimeseriesHolder.java
+++ b/accord-core/src/main/java/accord/local/KeyHistory.java
@@ -16,14 +16,20 @@
* limitations under the License.
*/
-package accord.impl;
+package accord.local;
-import accord.primitives.Timestamp;
-
-public interface CommandTimeseriesHolder
+/**
+ * For operations that need information on historical operations, this indicates the
+ * amount of data needed.
+ */
+public enum KeyHistory
{
- CommandTimeseries<?> byId();
- CommandTimeseries<?> byExecuteAt();
+ DEPS,
+ ALL,
+ NONE;
- Timestamp max();
+ public boolean isNone()
+ {
+ return this == NONE;
+ }
}
diff --git a/accord-core/src/main/java/accord/local/PreLoadContext.java b/accord-core/src/main/java/accord/local/PreLoadContext.java
index cacf81ae..0dea0598 100644
--- a/accord-core/src/main/java/accord/local/PreLoadContext.java
+++ b/accord-core/src/main/java/accord/local/PreLoadContext.java
@@ -42,6 +42,7 @@ import javax.annotation.Nullable;
*/
public interface PreLoadContext
{
+
@Nullable TxnId primaryTxnId();
/**
@@ -95,6 +96,8 @@ public interface PreLoadContext
*/
default Seekables<?, ?> keys() { return Keys.EMPTY; }
+ default KeyHistory keyHistory() { return KeyHistory.NONE; }
+
default boolean isSubsetOf(PreLoadContext superset)
{
if (superset.keys().domain() != keys().domain() || !superset.keys().containsAll(keys()))
@@ -125,7 +128,7 @@ public interface PreLoadContext
}
}
- static PreLoadContext contextFor(TxnId primary, Collection<TxnId> additional, Seekables<?, ?> keys)
+ static PreLoadContext contextFor(TxnId primary, Collection<TxnId> additional, Seekables<?, ?> keys, KeyHistory keyHistory)
{
return new PreLoadContext()
{
@@ -140,9 +143,17 @@ public interface PreLoadContext
@Override
public Seekables<?, ?> keys() { return keys; }
+
+ @Override
+ public KeyHistory keyHistory() { return keyHistory; }
};
}
+ static PreLoadContext contextFor(TxnId primary, Collection<TxnId> additional, Seekables<?, ?> keys)
+ {
+ return contextFor(primary, additional, keys, KeyHistory.NONE);
+ }
+
static PreLoadContext contextFor(TxnId primary, TxnId additional)
{
return contextFor(primary, Collections.singletonList(additional), Keys.EMPTY);
@@ -153,9 +164,14 @@ public interface PreLoadContext
return contextFor(primary, Collections.singletonList(additional), keys);
}
+ static PreLoadContext contextFor(TxnId txnId, Seekables<?, ?> keysOrRanges, KeyHistory keyHistory)
+ {
+ return contextFor(txnId, Collections.emptyList(), keysOrRanges, keyHistory);
+ }
+
static PreLoadContext contextFor(TxnId txnId, Seekables<?, ?> keysOrRanges)
{
- return contextFor(txnId, Collections.emptyList(), keysOrRanges);
+ return contextFor(txnId, keysOrRanges, KeyHistory.NONE);
}
static PreLoadContext contextFor(TxnId txnId)
@@ -168,9 +184,14 @@ public interface PreLoadContext
return contextFor(primary, additional, Keys.EMPTY);
}
+ static PreLoadContext contextFor(Key key, KeyHistory keyHistory)
+ {
+ return contextFor(null, Collections.emptyList(), Keys.of(key), keyHistory);
+ }
+
static PreLoadContext contextFor(Key key)
{
- return contextFor(null, Collections.emptyList(), Keys.of(key));
+ return contextFor(key, KeyHistory.NONE);
}
static PreLoadContext contextFor(Collection<TxnId> ids, Seekables<?, ?> keys)
diff --git a/accord-core/src/main/java/accord/local/SafeCommandStore.java b/accord-core/src/main/java/accord/local/SafeCommandStore.java
index c1eb61d7..765dc7c9 100644
--- a/accord-core/src/main/java/accord/local/SafeCommandStore.java
+++ b/accord-core/src/main/java/accord/local/SafeCommandStore.java
@@ -19,7 +19,9 @@
package accord.local;
import java.util.Iterator;
+import java.util.List;
import java.util.function.Predicate;
+import java.util.function.Supplier;
import accord.api.Agent;
import accord.api.DataStore;
@@ -36,6 +38,7 @@ import accord.primitives.Timestamp;
import accord.primitives.Txn.Kind.Kinds;
import accord.primitives.TxnId;
import accord.primitives.Unseekables;
+
import javax.annotation.Nullable;
import com.google.common.base.Predicates;
@@ -53,7 +56,7 @@ public abstract class SafeCommandStore
{
public interface CommandFunction<P1, I, O>
{
- O apply(P1 p1, Seekable keyOrRange, TxnId txnId, Timestamp executeAt, Status status, I in);
+ O apply(P1 p1, Seekable keyOrRange, TxnId txnId, Timestamp executeAt, Status status, Supplier<List<TxnId>> depsSupplier, I in);
}
public enum TestTimestamp
@@ -93,6 +96,8 @@ public abstract class SafeCommandStore
return maybeTruncate(safeCommand, command, txnId, null);
}
+ public abstract void removeCommandFromSeekableDeps(Seekable seekable, TxnId txnId, Timestamp executeAt, Status status);
+
// decidedExecuteAt == null if not yet PreCommitted
/**
@@ -158,17 +163,17 @@ public abstract class SafeCommandStore
/**
* Visits keys first and then ranges, both in ascending order.
- * Within each key or range visits TxnId in ascending order of queried timestamp.
+ * Within each key or range visits all unevicted txnids needed for the given scope in ascending order of queried timestamp.
*/
- public <P1, T> T mapReduce(Seekables<?, ?> keys, Ranges slice,
+ public <P1, T> T mapReduce(Seekables<?, ?> keys, Ranges slice, KeyHistory keyHistory,
Kinds testKind, TestTimestamp testTimestamp, Timestamp timestamp,
TestDep testDep, @Nullable TxnId depId, @Nullable Status minStatus, @Nullable Status maxStatus,
CommandFunction<P1, T, T> map, P1 p1, T initialValue)
{
- return mapReduce(keys, slice, testKind, testTimestamp, timestamp, testDep, depId, minStatus, maxStatus, map, p1, initialValue, Predicates.alwaysFalse());
+ return mapReduce(keys, slice, keyHistory, testKind, testTimestamp, timestamp, testDep, depId, minStatus, maxStatus, map, p1, initialValue, Predicates.alwaysFalse());
}
- public abstract <P1, T> T mapReduce(Seekables<?, ?> keys, Ranges slice,
+ public abstract <P1, T> T mapReduce(Seekables<?, ?> keys, Ranges slice, KeyHistory keyHistory,
Kinds testKind, TestTimestamp testTimestamp, Timestamp timestamp,
TestDep testDep, @Nullable TxnId depId, @Nullable Status minStatus, @Nullable Status maxStatus,
CommandFunction<P1, T, T> map, P1 p1, T initialValue, Predicate<? super T> terminate);
diff --git a/accord-core/src/main/java/accord/messages/Accept.java b/accord-core/src/main/java/accord/messages/Accept.java
index 5e069ab5..687372d0 100644
--- a/accord-core/src/main/java/accord/messages/Accept.java
+++ b/accord-core/src/main/java/accord/messages/Accept.java
@@ -24,6 +24,7 @@ import javax.annotation.Nullable;
import accord.api.RoutingKey;
import accord.local.Commands;
import accord.local.Commands.AcceptOutcome;
+import accord.local.KeyHistory;
import accord.local.Node.Id;
import accord.local.SafeCommand;
import accord.local.SafeCommandStore;
@@ -150,6 +151,12 @@ public class Accept extends TxnRequest.WithUnsynced<Accept.AcceptReply>
return keys;
}
+ @Override
+ public KeyHistory keyHistory()
+ {
+ return KeyHistory.DEPS;
+ }
+
@Override
public MessageType type()
{
diff --git a/accord-core/src/main/java/accord/messages/BeginRecovery.java b/accord-core/src/main/java/accord/messages/BeginRecovery.java
index 56adf49a..5341aea3 100644
--- a/accord-core/src/main/java/accord/messages/BeginRecovery.java
+++ b/accord-core/src/main/java/accord/messages/BeginRecovery.java
@@ -23,12 +23,8 @@ import javax.annotation.Nonnull;
import javax.annotation.Nullable;
import accord.api.Result;
-import accord.local.Command;
-import accord.local.Commands;
+import accord.local.*;
import accord.local.Node.Id;
-import accord.local.SafeCommand;
-import accord.local.SafeCommandStore;
-import accord.local.Status;
import accord.primitives.Ballot;
import accord.primitives.Deps;
import accord.primitives.FullRoute;
@@ -212,6 +208,12 @@ public class BeginRecovery extends TxnRequest<BeginRecovery.RecoverReply>
return partialTxn.keys();
}
+ @Override
+ public KeyHistory keyHistory()
+ {
+ return KeyHistory.ALL;
+ }
+
@Override
public MessageType type()
{
@@ -340,11 +342,11 @@ public class BeginRecovery extends TxnRequest<BeginRecovery.RecoverReply>
try (Deps.Builder builder = Deps.builder())
{
// any transaction that started
- safeStore.mapReduce(keys, ranges, startedBefore.rw().witnesses(), STARTED_BEFORE, startedBefore, WITHOUT, startedBefore, Accepted, PreCommitted,
- (startedBefore0, keyOrRange, txnId, executeAt, status, prev) -> {
- if (executeAt.compareTo(startedBefore0) > 0)
- prev.add(keyOrRange, txnId);
- return prev;
+ safeStore.mapReduce(keys, ranges, KeyHistory.ALL, startedBefore.rw().witnesses(), STARTED_BEFORE, startedBefore, WITHOUT, startedBefore, Accepted, PreCommitted,
+ (startedBefore0, keyOrRange, txnId, executeAt, status, deps, prev) -> {
+ if (executeAt.compareTo(startedBefore) > 0)
+ builder.add(keyOrRange, txnId);
+ return builder;
}, startedBefore, builder);
return builder.build();
}
@@ -354,8 +356,8 @@ public class BeginRecovery extends TxnRequest<BeginRecovery.RecoverReply>
{
try (Deps.Builder builder = Deps.builder())
{
- safeStore.mapReduce(keys, ranges, startedBefore.rw().witnesses(), STARTED_BEFORE, startedBefore, WITH, startedBefore, Committed, null,
- (p1, keyOrRange, txnId, executeAt, status, prev) -> prev.add(keyOrRange, txnId), null, (Deps.AbstractBuilder<Deps>)builder);
+ safeStore.mapReduce(keys, ranges, KeyHistory.ALL, startedBefore.rw().witnesses(), STARTED_BEFORE, startedBefore, WITH, startedBefore, Committed, null,
+ (p1, keyOrRange, txnId, executeAt, status, deps, prev) -> builder.add(keyOrRange, txnId), null, (Deps.AbstractBuilder<Deps>)builder);
return builder.build();
}
}
@@ -371,8 +373,8 @@ public class BeginRecovery extends TxnRequest<BeginRecovery.RecoverReply>
*
* TODO (required): consider carefully how _adding_ ranges to a CommandStore affects this
*/
- return safeStore.mapReduce(keys, ranges, startedAfter.rw().witnesses(), STARTED_AFTER, startedAfter, WITHOUT, startedAfter, Accepted, PreCommitted,
- (p1, keyOrRange, txnId, executeAt, status, prev) -> true, null, false, i -> i);
+ return safeStore.mapReduce(keys, ranges, KeyHistory.ALL, startedAfter.rw().witnesses(), STARTED_AFTER, startedAfter, WITHOUT, startedAfter, Accepted, PreCommitted,
+ (p1, keyOrRange, txnId, executeAt, status, deps, prev) -> true, null, false, i -> i);
}
private static boolean hasCommittedExecutesAfterWithoutWitnessing(SafeCommandStore safeStore, TxnId startedAfter, Ranges ranges, Seekables<?, ?> keys)
@@ -384,7 +386,7 @@ public class BeginRecovery extends TxnRequest<BeginRecovery.RecoverReply>
* witnessed us we are safe to propose the pre-accept timestamp regardless, whereas if any transaction
* has not witnessed us we can safely invalidate it.
*/
- return safeStore.mapReduce(keys, ranges, startedAfter.rw().witnesses(), EXECUTES_AFTER, startedAfter, WITHOUT, startedAfter, Committed, null,
- (p1, keyOrRange, txnId, executeAt, status, prev) -> true, null, false, i -> i);
+ return safeStore.mapReduce(keys, ranges, KeyHistory.ALL, startedAfter.rw().witnesses(), EXECUTES_AFTER, startedAfter, WITHOUT, startedAfter, Committed, null,
+ (p1, keyOrRange, txnId, executeAt, status, deps, prev) -> true, null, false, i -> i);
}
}
diff --git a/accord-core/src/main/java/accord/messages/CheckStatus.java b/accord-core/src/main/java/accord/messages/CheckStatus.java
index 5feec410..063756d5 100644
--- a/accord-core/src/main/java/accord/messages/CheckStatus.java
+++ b/accord-core/src/main/java/accord/messages/CheckStatus.java
@@ -312,7 +312,7 @@ public class CheckStatus extends AbstractEpochRequest<CheckStatus.CheckStatusRep
this.validForAll = FoundKnown.Nothing;
}
- private FoundKnownMap(boolean inclusiveEnds, RoutingKey[] starts, FoundKnown[] values)
+ public FoundKnownMap(boolean inclusiveEnds, RoutingKey[] starts, FoundKnown[] values)
{
this(inclusiveEnds, starts, values, FoundKnown.Nothing);
}
diff --git a/accord-core/src/main/java/accord/messages/GetDeps.java b/accord-core/src/main/java/accord/messages/GetDeps.java
index 28348ef2..16feb38f 100644
--- a/accord-core/src/main/java/accord/messages/GetDeps.java
+++ b/accord-core/src/main/java/accord/messages/GetDeps.java
@@ -20,6 +20,7 @@ package accord.messages;
import javax.annotation.Nonnull;
+import accord.local.KeyHistory;
import accord.local.Node.Id;
import accord.local.SafeCommandStore;
import accord.primitives.FullRoute;
@@ -115,6 +116,12 @@ public class GetDeps extends TxnRequest.WithUnsynced<PartialDeps>
return keys;
}
+ @Override
+ public KeyHistory keyHistory()
+ {
+ return KeyHistory.DEPS;
+ }
+
public static class GetDepsOk implements Reply
{
public final PartialDeps deps;
diff --git a/accord-core/src/main/java/accord/messages/PreAccept.java b/accord-core/src/main/java/accord/messages/PreAccept.java
index 45e9df5a..0b9b6c71 100644
--- a/accord-core/src/main/java/accord/messages/PreAccept.java
+++ b/accord-core/src/main/java/accord/messages/PreAccept.java
@@ -20,9 +20,9 @@ package accord.messages;
import java.util.List;
import java.util.Objects;
-import java.util.function.Function;
import javax.annotation.Nullable;
+import accord.primitives.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -31,18 +31,9 @@ import accord.local.Commands;
import accord.local.Node.Id;
import accord.local.SafeCommand;
import accord.local.SafeCommandStore;
+import accord.impl.DepsBuilder;
+import accord.local.*;
import accord.messages.TxnRequest.WithUnsynced;
-import accord.primitives.Deps;
-import accord.primitives.EpochSupplier;
-import accord.primitives.FullRoute;
-import accord.primitives.PartialDeps;
-import accord.primitives.PartialRoute;
-import accord.primitives.PartialTxn;
-import accord.primitives.Ranges;
-import accord.primitives.Seekables;
-import accord.primitives.Timestamp;
-import accord.primitives.Txn;
-import accord.primitives.TxnId;
import accord.topology.Shard;
import accord.topology.Topologies;
@@ -96,6 +87,12 @@ public class PreAccept extends WithUnsynced<PreAccept.PreAcceptReply> implements
return partialTxn.keys();
}
+ @Override
+ public KeyHistory keyHistory()
+ {
+ return KeyHistory.DEPS;
+ }
+
@Override
protected void process()
{
@@ -255,28 +252,17 @@ public class PreAccept extends WithUnsynced<PreAccept.PreAcceptReply> implements
static PartialDeps calculatePartialDeps(SafeCommandStore safeStore, TxnId txnId, Seekables<?, ?> keys, EpochSupplier minEpoch, Timestamp executeAt, Ranges ranges)
{
// TODO (expected): do not build covering ranges; no longer especially valuable given use of FullRoute
- return calculateDeps(safeStore, txnId, keys, minEpoch, executeAt, ranges, PartialDeps::builder);
- }
-
- private static <T extends Deps> T calculateDeps(SafeCommandStore safeStore, TxnId txnId, Seekables<?, ?> keys, EpochSupplier minEpoch, Timestamp executeAt, Ranges ranges, Function<Ranges, Deps.AbstractBuilder<T>> builderFactory)
- {
+ DepsBuilder builder = new DepsBuilder(txnId);
// could use MAY_EXECUTE_BEFORE to prune those we know execute later.
// NOTE: ExclusiveSyncPoint *relies* on STARTED_BEFORE to ensure it reports a dependency on *every* earlier TxnId that may execute after it.
// This is necessary for reporting to a bootstrapping replica which TxnId it must not prune from dependencies
// i.e. the source replica reports to the target replica those TxnId that STARTED_BEFORE and EXECUTES_AFTER.
- try (Deps.AbstractBuilder<T> depBuilder = builderFactory.apply(ranges); Deps.AbstractBuilder<T> redundantBuilder = builderFactory.apply(ranges))
- {
- // TODO (expected): this can be neater
- safeStore.mapReduce(keys, ranges, txnId.rw().witnesses(), STARTED_BEFORE, executeAt, ANY_DEPS, null, null, null,
- (txnId0, keyOrRange, testTxnId, testExecuteAt, testStatus, in) -> {
- if (!testTxnId.equals(txnId0))
- in.add(keyOrRange, testTxnId);
- return in;
- }, txnId, depBuilder);
- T deps = depBuilder.build();
- T redundant = safeStore.commandStore().redundantBefore().collectDeps(keys, redundantBuilder, minEpoch, executeAt).build();
- return (T)deps.with(redundant);
- }
+ safeStore.mapReduce(keys, ranges, KeyHistory.DEPS, txnId.rw().witnesses(), STARTED_BEFORE, executeAt, ANY_DEPS, null, null, null,
+ (p1, keyOrRange, testTxnId, testExecuteAt, status, deps, in) -> {
+ builder.add(testTxnId, keyOrRange, status, testExecuteAt, deps.get());
+ return in;
+ }, null, builder);
+ return builder.buildPartialDeps(safeStore, ranges);
}
/**
diff --git a/accord-core/src/main/java/accord/primitives/Timestamp.java b/accord-core/src/main/java/accord/primitives/Timestamp.java
index abf22e8c..ef890576 100644
--- a/accord-core/src/main/java/accord/primitives/Timestamp.java
+++ b/accord-core/src/main/java/accord/primitives/Timestamp.java
@@ -58,6 +58,11 @@ public class Timestamp implements Comparable<Timestamp>, EpochSupplier
return new Timestamp(epoch, hlc, flags, node);
}
+ public static Timestamp fromValues(long epoch, long hlc, int flags, int node)
+ {
+ return new Timestamp(epoch, hlc, flags, new Id(node));
+ }
+
public static Timestamp maxForEpoch(long epoch)
{
return new Timestamp(epochMsb(epoch) | 0x7fff, Long.MAX_VALUE, Id.MAX);
diff --git a/accord-core/src/test/java/accord/impl/CommandsForKeyGroupUpdaterTest.java b/accord-core/src/test/java/accord/impl/CommandsForKeyGroupUpdaterTest.java
new file mode 100644
index 00000000..164c282f
--- /dev/null
+++ b/accord-core/src/test/java/accord/impl/CommandsForKeyGroupUpdaterTest.java
@@ -0,0 +1,180 @@
+/*
+ * 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 accord.impl;
+
+import accord.impl.CommandsForKeyGroupUpdater.Immutable;
+import accord.local.Command;
+import accord.local.SaveStatus;
+import accord.primitives.Timestamp;
+import accord.primitives.TxnId;
+import com.google.common.collect.ImmutableSortedSet;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import java.util.List;
+
+public class CommandsForKeyGroupUpdaterTest
+{
+ private static class Sentinel { private static final Sentinel INSTANCE = new Sentinel(); }
+
+ private static final CommandTimeseries.CommandLoader<Sentinel> LOADER = new CommandTimeseries.CommandLoader<Sentinel>()
+ {
+ @Override public Sentinel saveForCFK(Command command) { return Sentinel.INSTANCE; }
+ @Override public TxnId txnId(Sentinel data) { throw new UnsupportedOperationException(); }
+ @Override public Timestamp executeAt(Sentinel data) { throw new UnsupportedOperationException(); }
+ @Override public SaveStatus saveStatus(Sentinel data) { throw new UnsupportedOperationException(); }
+ @Override public List<TxnId> depsIds(Sentinel data) { throw new UnsupportedOperationException(); }
+ };
+
+ private static final IntKey.Raw KEY = IntKey.key(1);
+
+ private static TxnId txnId(int i)
+ {
+ return TxnId.fromValues(0, i, 0, 1);
+ }
+
+ private static Timestamp ts(int i)
+ {
+ return Timestamp.fromValues(0, i, 0, 1);
+ }
+
+ private static CommandTimeseries<Sentinel> timeseries()
+ {
+ CommandTimeseries.Builder<Sentinel> builder = new CommandTimeseries.Builder<>(KEY, LOADER);
+ return builder.build();
+ }
+
+ private static CommandTimeseries<Sentinel> timeseries(Timestamp... timestamps)
+ {
+ CommandTimeseries.Builder<Sentinel> builder = new CommandTimeseries.Builder<>(KEY, LOADER);
+ for (Timestamp timestamp : timestamps)
+ builder.add(timestamp, Sentinel.INSTANCE);
+ return builder.build();
+ }
+
+ private static CommandTimeseries<Sentinel> timeseries(int... timestamps)
+ {
+ return timeseries(timestamps(timestamps));
+ }
+
+ private static CommandsForKey cfk(CommandTimeseries<Sentinel> commands)
+ {
+ return new CommandsForKey(KEY, commands);
+ }
+
+ private static CommandsForKey cfk(Timestamp[] byId)
+ {
+ return cfk(timeseries(byId));
+ }
+
+ private static CommandsForKey cfk()
+ {
+ return new CommandsForKey(KEY, timeseries());
+ }
+
+ private static Timestamp[] timestamps(int... values)
+ {
+ Timestamp[] timestamps = new Timestamp[values.length];
+ for (int i=0; i<values.length; i++)
+ timestamps[i] = ts(values[i]);
+ return timestamps;
+ }
+
+ private static void assertTimeseries(CommandTimeseries<?> timeseries, Timestamp... expected)
+ {
+ Assertions.assertEquals(ImmutableSortedSet.copyOf(expected), timeseries.commands.keySet());
+ }
+
+ private static void assertTimeseries(CommandTimeseries<?> timeseries, int... expected)
+ {
+ assertTimeseries(timeseries, timestamps(expected));
+ }
+
+ @Test
+ void commonTest()
+ {
+ CommandsForKeyGroupUpdater.Mutable<Sentinel> updater = new CommandsForKeyGroupUpdater.Mutable<>(LOADER);
+
+ updater.common().commands().add(txnId(1), null);
+
+ Assertions.assertEquals(cfk(timeseries(1)), updater.applyToDeps(cfk()));
+ Assertions.assertEquals(cfk(timeseries(1)), updater.applyToAll(cfk()));
+ }
+
+ @Test
+ void specificPrecedence()
+ {
+ CommandsForKeyGroupUpdater.Mutable<Sentinel> updater = new CommandsForKeyGroupUpdater.Mutable<>(LOADER);
+
+ updater.common().commands().add(txnId(2), null);
+ updater.common().commands().add(txnId(1), null);
+ updater.all().commands().remove(txnId(1));
+
+ Assertions.assertEquals(cfk(timeseries(1, 2)), updater.applyToDeps(cfk()));
+ Assertions.assertEquals(cfk(timeseries(2)), updater.applyToAll(cfk()));
+ }
+
+ /**
+ * A later specifc update should override a previous common update (making it specific to the opposite cfk)
+ */
+ @Test
+ void mergeOverridingSpecific()
+ {
+ CommandsForKeyGroupUpdater.Mutable<Sentinel> original = new CommandsForKeyGroupUpdater.Mutable<>(LOADER);
+
+ original.common().commands().add(txnId(1), null);
+ original.common().commands().add(txnId(2), null);
+
+ CommandsForKeyGroupUpdater.Mutable<Sentinel> update = new CommandsForKeyGroupUpdater.Mutable<>(LOADER);
+ update.deps().commands().remove(txnId(1));
+ update.deps().commands().add(txnId(3), null);
+
+ CommandsForKeyGroupUpdater.Mutable<Sentinel> expected = new CommandsForKeyGroupUpdater.Mutable<>(LOADER);
+
+ expected.common().commands().add(txnId(2), null);
+ expected.deps().commands().remove(txnId(1));
+ expected.deps().commands().add(txnId(3), null);
+ expected.all().commands().add(txnId(1), null);
+
+// Immutable<Sentinel> merged = original.toImmutable().mergeNewer(update.toImmutable());
+ Immutable<Sentinel> merged = Immutable.merge(original.toImmutable(), update.toImmutable(), Immutable.getFactory());
+ Assertions.assertEquals(expected.toImmutable(), merged);
+ }
+
+ @Test
+ void mergeOverridingCommon()
+ {
+ CommandsForKeyGroupUpdater.Mutable<Sentinel> original = new CommandsForKeyGroupUpdater.Mutable<>(LOADER);
+ original.deps().commands().add(txnId(1), null);
+ original.deps().commands().add(txnId(2), null);
+ original.all().commands().add(txnId(2), null);
+
+ CommandsForKeyGroupUpdater.Mutable<Sentinel> update = new CommandsForKeyGroupUpdater.Mutable<>(LOADER);
+ update.common().commands().remove(txnId(1));
+
+ CommandsForKeyGroupUpdater.Mutable<Sentinel> expected = new CommandsForKeyGroupUpdater.Mutable<>(LOADER);
+ expected.deps().commands().add(txnId(2), null);
+ expected.all().commands().add(txnId(2), null);
+ expected.common().commands().remove(txnId(1));
+
+ Immutable<Sentinel> merged = Immutable.merge(original.toImmutable(), update.toImmutable(), Immutable.getFactory());
+ Assertions.assertEquals(expected.toImmutable(), merged);
+ }
+}
+
diff --git a/accord-core/src/test/java/accord/impl/list/ListWrite.java b/accord-core/src/test/java/accord/impl/list/ListWrite.java
index 55a7fff3..7364f78b 100644
--- a/accord-core/src/test/java/accord/impl/list/ListWrite.java
+++ b/accord-core/src/test/java/accord/impl/list/ListWrite.java
@@ -23,20 +23,16 @@ import java.util.TreeMap;
import java.util.function.Function;
import java.util.stream.Collectors;
-import accord.primitives.PartialTxn;
+import accord.impl.*;
+import accord.primitives.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import accord.api.DataStore;
import accord.api.Key;
import accord.api.Write;
-import accord.impl.InMemoryCommandStore;
-import accord.impl.InMemorySafeCommandsForKey;
import accord.local.CommandStore;
import accord.local.SafeCommandStore;
-import accord.primitives.Seekable;
-import accord.primitives.Timestamp;
-import accord.primitives.Writes;
import accord.utils.Timestamped;
import accord.utils.async.AsyncChain;
import accord.utils.async.AsyncExecutor;
@@ -58,8 +54,7 @@ public class ListWrite extends TreeMap<Key, int[]> implements Write
ListStore s = (ListStore) store;
if (!containsKey(key))
return Writes.SUCCESS;
- InMemorySafeCommandsForKey cfk = ((InMemoryCommandStore.InMemorySafeStore) safeStore).commandsForKey((Key) key);
- cfk.updateLastExecutionTimestamps(safeStore, executeAt, true);
+ CommandsForKeys.updateLastExecutionTimestamps((AbstractSafeCommandStore<?, ?, ?, ?>) safeStore, (RoutableKey) key, executeAt, true);
return executor.apply(safeStore.commandStore()).submit(() -> {
int[] data = get(key);
diff --git a/accord-core/src/test/java/accord/messages/PreAcceptTest.java b/accord-core/src/test/java/accord/messages/PreAcceptTest.java
index 1afe638d..f872f7b6 100644
--- a/accord-core/src/test/java/accord/messages/PreAcceptTest.java
+++ b/accord-core/src/test/java/accord/messages/PreAcceptTest.java
@@ -102,7 +102,7 @@ public class PreAcceptTest
{
Raw key = IntKey.key(10);
CommandStore commandStore = node.unsafeForKey(key);
- Assertions.assertFalse(inMemory(commandStore).hasCommandsForKey(key));
+ Assertions.assertFalse(inMemory(commandStore).hasDepsCommandsForKey(key));
TxnId txnId = clock.idForNode(1, ID2);
Txn txn = writeTxn(Keys.of(key));
@@ -111,8 +111,8 @@ public class PreAcceptTest
preAccept.process(node, ID2, REPLY_CONTEXT);
commandStore.execute(PreLoadContext.contextFor(txnId, txn.keys()), safeStore -> {
- CommandsForKey cfk = ((AbstractSafeCommandStore) safeStore).commandsForKey(key).current();
- TxnId commandId = convert(cfk.byId(), CommandLoader::txnId).findFirst().get();
+ CommandsForKey cfk = ((AbstractSafeCommandStore) safeStore).depsCommandsForKey(key).current();
+ TxnId commandId = convert(cfk.commands(), CommandLoader::txnId).findFirst().get();
Command command = safeStore.ifInitialised(commandId).current();
Assertions.assertEquals(Status.PreAccepted, command.status());
});
@@ -139,7 +139,7 @@ public class PreAcceptTest
{
Raw key = IntKey.key(10);
CommandStore commandStore = node.unsafeForKey(key);
- Assertions.assertFalse(inMemory(commandStore).hasCommandsForKey(key));
+ Assertions.assertFalse(inMemory(commandStore).hasDepsCommandsForKey(key));
TxnId txnId = clock.idForNode(1, ID2);
Txn txn = writeTxn(Keys.of(key));
@@ -162,7 +162,7 @@ public class PreAcceptTest
{
Raw key = IntKey.key(10);
CommandStore commandStore = node.unsafeForKey(key);
- Assertions.assertFalse(inMemory(commandStore).hasCommandsForKey(key));
+ Assertions.assertFalse(inMemory(commandStore).hasDepsCommandsForKey(key));
TxnId txnId = clock.idForNode(1, ID2);
Txn txn = writeTxn(Keys.of(key));
@@ -280,8 +280,8 @@ public class PreAcceptTest
preAccept.process(node, ID2, REPLY_CONTEXT);
commandStore.execute(PreLoadContext.contextFor(txnId, txn.keys()), safeStore -> {
- CommandsForKey cfk = ((AbstractSafeCommandStore) safeStore).commandsForKey(key).current();
- TxnId commandId = convert(cfk.byId(), CommandLoader::txnId).findFirst().get();
+ CommandsForKey cfk = ((AbstractSafeCommandStore) safeStore).depsCommandsForKey(key).current();
+ TxnId commandId = convert(cfk.commands(), CommandLoader::txnId).findFirst().get();
Command command = safeStore.ifInitialised(commandId).current();
Assertions.assertEquals(Status.PreAccepted, command.status());
});
diff --git a/accord-core/src/test/java/accord/primitives/DepsBuilderTest.java b/accord-core/src/test/java/accord/primitives/DepsBuilderTest.java
new file mode 100644
index 00000000..cd56d222
--- /dev/null
+++ b/accord-core/src/test/java/accord/primitives/DepsBuilderTest.java
@@ -0,0 +1,161 @@
+/*
+ * 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 accord.primitives;
+
+import accord.api.Agent;
+import accord.api.DataStore;
+import accord.api.Key;
+import accord.api.ProgressLog;
+import accord.impl.DepsBuilder;
+import accord.impl.DepsBuilder.Builder;
+import accord.impl.IntKey;
+import accord.local.*;
+import com.google.common.collect.ImmutableList;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+
+import javax.annotation.Nullable;
+import java.util.*;
+import java.util.function.Predicate;
+
+public class DepsBuilderTest
+{
+ private static final Key KEY = IntKey.key(1);
+ private static class InstrumentedSafeStore extends SafeCommandStore
+ {
+ public static class Removed
+ {
+ public final Set<TxnId> commands = new HashSet<>();
+
+ void add(TxnId txnId, Timestamp executeAt)
+ {
+ commands.add(txnId);
+ }
+ }
+
+ Map<Seekable, Removed> removed = new HashMap<>();
+
+ @Override
+ public void removeCommandFromSeekableDeps(Seekable seekable, TxnId txnId, Timestamp executeAt, Status status)
+ {
+ removed.computeIfAbsent(seekable, s -> new Removed()).add(txnId, executeAt);
+ }
+
+ @Override protected SafeCommand getInternal(TxnId txnId) { throw new UnsupportedOperationException(); }
+ @Override protected SafeCommand getInternalIfLoadedAndInitialised(TxnId txnId) { throw new UnsupportedOperationException(); }
+ @Override public boolean canExecuteWith(PreLoadContext context) { throw new UnsupportedOperationException(); }
+ @Override public <P1, T> T mapReduce(Seekables<?, ?> keys, Ranges slice, KeyHistory keyHistory, Txn.Kind.Kinds testKind, TestTimestamp testTimestamp, Timestamp timestamp, TestDep testDep, @Nullable TxnId depId, @Nullable Status minStatus, @Nullable Status maxStatus, CommandFunction<P1, T, T> map, P1 p1, T initialValue, Predicate<? super T> terminate) { throw new UnsupportedOperationException(); }
+ @Override protected void register(Seekables<?, ?> keysOrRanges, Ranges slice, Command command) { throw new UnsupportedOperationException(); }
+ @Override protected void register(Seekable keyOrRange, Ranges slice, Command command) { throw new UnsupportedOperationException(); }
+ @Override public CommandStore commandStore() { throw new UnsupportedOperationException(); }
+ @Override public DataStore dataStore() { throw new UnsupportedOperationException(); }
+ @Override public Agent agent() { throw new UnsupportedOperationException(); }
+ @Override public ProgressLog progressLog() { throw new UnsupportedOperationException(); }
+ @Override public NodeTimeService time() { throw new UnsupportedOperationException(); }
+ @Override public CommandStores.RangesForEpoch ranges() { throw new UnsupportedOperationException(); }
+ @Override public Timestamp maxConflict(Seekables<?, ?> keys, Ranges slice) { throw new UnsupportedOperationException(); }
+ @Override public void registerHistoricalTransactions(Deps deps) { throw new UnsupportedOperationException(); }
+ @Override public void erase(SafeCommand safeCommand) { throw new UnsupportedOperationException(); }
+ }
+ private static final Node.Id NODE = new Node.Id(0);
+
+ private static TxnId txnId(int v)
+ {
+ return TxnId.fromValues(1, v, 0, 0);
+ }
+
+ private static List<TxnId> txnIds(int... vs)
+ {
+ ImmutableList.Builder<TxnId> builder = ImmutableList.builder();
+ Arrays.sort(vs);
+ for (int v : vs)
+ builder.add(txnId(v));
+ return builder.build();
+ }
+
+ private static Timestamp timestamp(int v)
+ {
+ return Timestamp.fromValues(1, v, NODE);
+ }
+
+ private static List<TxnId> build(SafeCommandStore safeStore, Seekable seekable, Builder builder)
+ {
+ List<TxnId> result = new ArrayList<>();
+ builder.build(safeStore, seekable, result::add);
+ result.sort(Comparator.naturalOrder());
+ return result;
+ }
+
+ private static List<TxnId> build(SafeCommandStore safeStore, Builder builder)
+ {
+ return build(safeStore, KEY, builder);
+ }
+
+ @BeforeAll
+ static void setUpClass()
+ {
+ DepsBuilder.setCanPruneUnsafe(true);
+ }
+
+ @AfterAll
+ static void tearDownClass()
+ {
+ DepsBuilder.setCanPruneUnsafe(false);
+ }
+
+ @Test
+ void basicTest()
+ {
+ InstrumentedSafeStore safeStore = new InstrumentedSafeStore();
+ Builder builder = new Builder();
+ builder.add(txnId(0), Status.Applied, timestamp(0), txnIds());
+ builder.add(txnId(1), Status.Applied, timestamp(1), txnIds(0));
+ Assertions.assertEquals(txnIds(1), build(safeStore, builder));
+ }
+
+ /**
+ * Committed deps should be included if they point to an uncommitted command
+ */
+ @Test
+ void uncommittedTest()
+ {
+ InstrumentedSafeStore safeStore = new InstrumentedSafeStore();
+ Builder builder = new Builder();
+ builder.add(txnId(0), Status.Accepted, timestamp(0), txnIds());
+ builder.add(txnId(1), Status.Applied, timestamp(1), txnIds(0));
+ builder.add(txnId(2), Status.Applied, timestamp(2), txnIds(1));
+ builder.add(txnId(3), Status.Applied, timestamp(3), txnIds(2));
+ Assertions.assertEquals(txnIds(0, 1, 2, 3), build(safeStore, builder));
+ }
+
+ /**
+ * Committed txns should be included if they point to a command with a later execution time
+ */
+ @Test
+ void executionOrderTest()
+ {
+ InstrumentedSafeStore safeStore = new InstrumentedSafeStore();
+ Builder builder = new Builder();
+ builder.add(txnId(0), Status.Applied, timestamp(2), txnIds());
+ builder.add(txnId(1), Status.Applied, timestamp(1), txnIds(0));
+ Assertions.assertEquals(txnIds(0, 1), build(safeStore, builder));
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@cassandra.apache.org
For additional commands, e-mail: commits-help@cassandra.apache.org