You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@cassandra.apache.org by be...@apache.org on 2023/05/12 08:53:11 UTC
[cassandra-accord] 01/03: Separate durable and transient listeners
This is an automated email from the ASF dual-hosted git repository.
benedict pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/cassandra-accord.git
commit c58d841fbb26aa0f4ecaa782f2471003c4c58412
Author: Benedict Elliott Smith <be...@apache.org>
AuthorDate: Tue Apr 25 13:12:57 2023 +0100
Separate durable and transient listeners
patch by Benedict; reviewed by Ariel Weisberg for CASSANDRA-18524
---
.gitignore | 1 +
accord-core/build.gradle | 4 +-
.../main/java/accord/coordinate/Coordinate.java | 2 +
.../src/main/java/accord/impl/CommandsForKey.java | 8 +-
.../java/accord/impl/InMemoryCommandStore.java | 22 +++-
.../main/java/accord/impl/InMemorySafeCommand.java | 20 +++
.../src/main/java/accord/local/Command.java | 137 ++++++++++++---------
.../src/main/java/accord/local/CommandStores.java | 4 +-
.../src/main/java/accord/local/Commands.java | 4 +-
.../main/java/accord/local/CommonAttributes.java | 10 +-
.../src/main/java/accord/local/Listeners.java | 18 +--
.../src/main/java/accord/local/SafeCommand.java | 9 +-
.../main/java/accord/local/SafeCommandStore.java | 65 +++++++---
.../src/main/java/accord/messages/Defer.java | 8 +-
.../src/main/java/accord/messages/ReadData.java | 17 +--
.../main/java/accord/messages/WaitOnCommit.java | 8 +-
.../src/main/java/accord/primitives/Timestamp.java | 10 +-
.../src/test/java/accord/burn/BurnTest.java | 1 +
build.gradle | 1 +
19 files changed, 214 insertions(+), 135 deletions(-)
diff --git a/.gitignore b/.gitignore
index 05fa172a..bd3d7a7f 100644
--- a/.gitignore
+++ b/.gitignore
@@ -3,3 +3,4 @@
build/
accord-core/build/
accord-maelstrom/build/
+.rat-excludes.txt
diff --git a/accord-core/build.gradle b/accord-core/build.gradle
index bb315b9b..df6203b9 100644
--- a/accord-core/build.gradle
+++ b/accord-core/build.gradle
@@ -39,7 +39,7 @@ dependencies {
}
task burn(type: JavaExec) {
- classpath sourceSets.main.runtimeClasspath, sourceSets.test.output
+ classpath sourceSets.main.runtimeClasspath, sourceSets.test.runtimeClasspath
mainClass = 'accord.burn.BurnTest'
jvmArgs '-Dlogback.configurationFile=burn-logback.xml'
args = ['-c', '1']
@@ -103,7 +103,7 @@ task burnforkloop {
task burnloop {
doLast {
javaexec {
- classpath sourceSets.main.runtimeClasspath, sourceSets.test.output
+ classpath sourceSets.main.runtimeClasspath, sourceSets.test.runtimeClasspath
mainClass = 'accord.burn.BurnTest'
jvmArgs '-Dlogback.configurationFile=burn-logback.xml'
args = ['-c', project.hasProperty('burnTimes') ? project.getProperty('burnTimes') : '1000000']
diff --git a/accord-core/src/main/java/accord/coordinate/Coordinate.java b/accord-core/src/main/java/accord/coordinate/Coordinate.java
index f08dd399..6934cfb9 100644
--- a/accord-core/src/main/java/accord/coordinate/Coordinate.java
+++ b/accord-core/src/main/java/accord/coordinate/Coordinate.java
@@ -117,7 +117,9 @@ public class Coordinate extends AsyncResults.SettableResult<Result> implements C
if (!reply.isOk())
{
+ // TODO (expected): reads should not be recovered, so should not be preempted
// we've been preempted by a recovery coordinator; defer to it, and wait to hear any result
+ preAcceptIsDone = true;
tryFailure(new Preempted(txnId, route.homeKey()));
return;
}
diff --git a/accord-core/src/main/java/accord/impl/CommandsForKey.java b/accord-core/src/main/java/accord/impl/CommandsForKey.java
index 9c1d20ae..51ceeb31 100644
--- a/accord-core/src/main/java/accord/impl/CommandsForKey.java
+++ b/accord-core/src/main/java/accord/impl/CommandsForKey.java
@@ -41,9 +41,9 @@ public class CommandsForKey
{
public static class SerializerSupport
{
- public static CommandsForKey.Listener listener(Key key)
+ public static Listener listener(Key key)
{
- return new CommandsForKey.Listener(key);
+ return new Listener(key);
}
public static <D> CommandsForKey create(Key key, Timestamp max,
@@ -229,7 +229,7 @@ public class CommandsForKey
}
}
- public static class Listener implements CommandListener
+ public static class Listener implements Command.DurableAndIdempotentListener
{
protected final Key listenerKey;
@@ -354,7 +354,7 @@ public class CommandsForKey
throw new UnsupportedOperationException();
}
- public final CommandListener asListener()
+ public final Command.DurableAndIdempotentListener asListener()
{
return new Listener(key());
}
diff --git a/accord-core/src/main/java/accord/impl/InMemoryCommandStore.java b/accord-core/src/main/java/accord/impl/InMemoryCommandStore.java
index c7daf962..0b3fbe03 100644
--- a/accord-core/src/main/java/accord/impl/InMemoryCommandStore.java
+++ b/accord-core/src/main/java/accord/impl/InMemoryCommandStore.java
@@ -187,7 +187,7 @@ public abstract class InMemoryCommandStore extends CommandStore
CommonAttributes.Mutable mutable = attrs.mutable();
forEach(keysOrRanges, slice, key -> {
SafeCommandsForKey cfk = safeStore.commandsForKey(key);
- CommandListener listener = cfk.register(command.current()).asListener();
+ Command.DurableAndIdempotentListener listener = cfk.register(command.current()).asListener();
mutable.addListener(listener);
});
return mutable;
@@ -207,7 +207,7 @@ public abstract class InMemoryCommandStore extends CommandStore
CommonAttributes.Mutable mutable = attrs.mutable();
forEach(keyOrRange, slice, key -> {
SafeCommandsForKey cfk = safeStore.commandsForKey(key);
- CommandListener listener = cfk.register(command.current()).asListener();
+ Command.DurableAndIdempotentListener listener = cfk.register(command.current()).asListener();
mutable.addListener(listener);
});
return mutable;
@@ -475,6 +475,7 @@ public abstract class InMemoryCommandStore extends CommandStore
public static class GlobalCommand extends GlobalState<Command>
{
private final TxnId txnId;
+ private Listeners<Command.TransientListener> transientListeners = null;
public GlobalCommand(TxnId txnId)
{
@@ -485,6 +486,23 @@ public abstract class InMemoryCommandStore extends CommandStore
{
return new InMemorySafeCommand(txnId, this);
}
+
+ public void addListener(Command.TransientListener listener)
+ {
+ if (transientListeners == null) transientListeners = new Listeners<>();
+ else transientListeners.add(listener);
+ }
+
+ public void removeListener(Command.TransientListener listener)
+ {
+ if (transientListeners != null && transientListeners.remove(listener) && transientListeners.isEmpty())
+ transientListeners = null;
+ }
+
+ public Collection<Command.TransientListener> transientListeners()
+ {
+ return transientListeners == null ? Collections.emptySet() : transientListeners;
+ }
}
public static class GlobalCommandsForKey extends GlobalState<CommandsForKey>
diff --git a/accord-core/src/main/java/accord/impl/InMemorySafeCommand.java b/accord-core/src/main/java/accord/impl/InMemorySafeCommand.java
index 512f170a..fe4a3156 100644
--- a/accord-core/src/main/java/accord/impl/InMemorySafeCommand.java
+++ b/accord-core/src/main/java/accord/impl/InMemorySafeCommand.java
@@ -18,6 +18,8 @@
package accord.impl;
+import java.util.Collection;
+
import accord.impl.InMemoryCommandStore.GlobalCommand;
import accord.local.Command;
import accord.local.SafeCommand;
@@ -48,6 +50,24 @@ public class InMemorySafeCommand extends SafeCommand implements SafeState<Comman
global.value(update);
}
+ @Override
+ public void addListener(Command.TransientListener listener)
+ {
+ global.addListener(listener);
+ }
+
+ @Override
+ public void removeListener(Command.TransientListener listener)
+ {
+ global.removeListener(listener);
+ }
+
+ @Override
+ public Collection<Command.TransientListener> transientListeners()
+ {
+ return global.transientListeners();
+ }
+
@Override
public void invalidate()
{
diff --git a/accord-core/src/main/java/accord/local/Command.java b/accord-core/src/main/java/accord/local/Command.java
index 54aec102..59ab9330 100644
--- a/accord-core/src/main/java/accord/local/Command.java
+++ b/accord-core/src/main/java/accord/local/Command.java
@@ -40,6 +40,72 @@ import static java.lang.String.format;
public abstract class Command implements CommonAttributes
{
+ interface Listener
+ {
+ void onChange(SafeCommandStore safeStore, SafeCommand safeCommand);
+
+ /**
+ * Scope needed to run onChange
+ */
+ PreLoadContext listenerPreLoadContext(TxnId caller);
+ }
+
+ public interface DurableAndIdempotentListener extends Listener
+ {
+ }
+
+ public interface TransientListener extends Listener
+ {
+ }
+
+ public static class ProxyListener implements DurableAndIdempotentListener
+ {
+ protected final TxnId listenerId;
+
+ public ProxyListener(TxnId listenerId)
+ {
+ this.listenerId = listenerId;
+ }
+
+ @Override
+ public boolean equals(Object o)
+ {
+ if (this == o) return true;
+ if (o == null || getClass() != o.getClass()) return false;
+ ProxyListener that = (ProxyListener) o;
+ return listenerId.equals(that.listenerId);
+ }
+
+ @Override
+ public int hashCode()
+ {
+ return Objects.hash(listenerId);
+ }
+
+ @Override
+ public String toString()
+ {
+ return "ListenerProxy{" + listenerId + '}';
+ }
+
+ public TxnId txnId()
+ {
+ return listenerId;
+ }
+
+ @Override
+ public void onChange(SafeCommandStore safeStore, SafeCommand safeCommand)
+ {
+ Commands.listenerUpdate(safeStore, safeStore.command(listenerId), safeCommand);
+ }
+
+ @Override
+ public PreLoadContext listenerPreLoadContext(TxnId caller)
+ {
+ return PreLoadContext.contextFor(Utils.listOf(listenerId, caller), Keys.EMPTY);
+ }
+ }
+
static PreLoadContext contextForCommand(Command command)
{
Invariants.checkState(command.hasBeen(Status.PreAccepted) && command.partialTxn() != null);
@@ -116,54 +182,6 @@ public abstract class Command implements CommonAttributes
}
}
- public static class Listener implements CommandListener
- {
- protected final TxnId listenerId;
-
- public Listener(TxnId listenerId)
- {
- this.listenerId = listenerId;
- }
-
- @Override
- public boolean equals(Object o)
- {
- if (this == o) return true;
- if (o == null || getClass() != o.getClass()) return false;
- Listener that = (Listener) o;
- return listenerId.equals(that.listenerId);
- }
-
- @Override
- public int hashCode()
- {
- return Objects.hash(listenerId);
- }
-
- @Override
- public String toString()
- {
- return "ListenerProxy{" + listenerId + '}';
- }
-
- public TxnId txnId()
- {
- return listenerId;
- }
-
- @Override
- public void onChange(SafeCommandStore safeStore, SafeCommand safeCommand)
- {
- Commands.listenerUpdate(safeStore, safeStore.command(listenerId), safeCommand);
- }
-
- @Override
- public PreLoadContext listenerPreLoadContext(TxnId caller)
- {
- return PreLoadContext.contextFor(Utils.listOf(listenerId, caller), Keys.EMPTY);
- }
- }
-
private abstract static class AbstractCommand extends Command
{
private final TxnId txnId;
@@ -196,7 +214,7 @@ public abstract class Command implements CommonAttributes
this.progressKey = common.progressKey();
this.route = common.route();
this.promised = promised;
- this.listeners = common.listeners();
+ this.listeners = common.durableListeners();
}
@Override
@@ -212,7 +230,7 @@ public abstract class Command implements CommonAttributes
&& Objects.equals(progressKey, command.progressKey())
&& Objects.equals(route, command.route())
&& Objects.equals(promised, command.promised())
- && listeners.equals(command.listeners());
+ && listeners.equals(command.durableListeners());
}
@Override
@@ -258,7 +276,7 @@ public abstract class Command implements CommonAttributes
}
@Override
- public Listeners.Immutable listeners()
+ public Listeners.Immutable durableListeners()
{
if (listeners == null)
return Listeners.Immutable.EMPTY;
@@ -311,7 +329,7 @@ public abstract class Command implements CommonAttributes
public abstract TxnId txnId();
public abstract Ballot promised();
public abstract Status.Durability durability();
- public abstract Listeners.Immutable listeners();
+ public abstract Listeners.Immutable durableListeners();
public abstract SaveStatus saveStatus();
static boolean isSameClass(Command command, Class<? extends Command> klass)
@@ -407,9 +425,9 @@ public abstract class Command implements CommonAttributes
return status() == status;
}
- public final CommandListener asListener()
+ public final ProxyListener asListener()
{
- return new Listener(txnId());
+ return new ProxyListener(txnId());
}
public final boolean isWitnessed()
@@ -513,6 +531,13 @@ public abstract class Command implements CommonAttributes
return null;
}
+ @Override
+ public Ballot promised()
+ {
+ // we can be preacceptedInvalidated, so can be promised even if not witnessed
+ return super.promised;
+ }
+
@Override
public Ballot accepted()
{
@@ -922,13 +947,13 @@ public abstract class Command implements CommonAttributes
}
}
- static Command addListener(Command command, CommandListener listener)
+ static Command addListener(Command command, DurableAndIdempotentListener listener)
{
CommonAttributes attrs = command.mutable().addListener(listener);
return command.updateAttributes(attrs);
}
- static Command removeListener(Command command, CommandListener listener)
+ static Command removeListener(Command command, Listener listener)
{
CommonAttributes attrs = command.mutable().removeListener(listener);
return command.updateAttributes(attrs);
diff --git a/accord-core/src/main/java/accord/local/CommandStores.java b/accord-core/src/main/java/accord/local/CommandStores.java
index 943bc075..e6f3eed7 100644
--- a/accord-core/src/main/java/accord/local/CommandStores.java
+++ b/accord-core/src/main/java/accord/local/CommandStores.java
@@ -188,9 +188,9 @@ public abstract class CommandStores<S extends CommandStore>
return i;
}
- public boolean intersects(long epoch, AbstractKeys<?, ?> keys)
+ public boolean intersects(long epoch, Routables<?, ?> routables)
{
- return at(epoch).intersects(keys);
+ return at(epoch).intersects(routables);
}
public Ranges currentRanges()
diff --git a/accord-core/src/main/java/accord/local/Commands.java b/accord-core/src/main/java/accord/local/Commands.java
index 66546a40..49eb070d 100644
--- a/accord-core/src/main/java/accord/local/Commands.java
+++ b/accord-core/src/main/java/accord/local/Commands.java
@@ -317,7 +317,7 @@ public class Commands
case PreCommitted:
// we don't know when these dependencies will execute, and cannot execute until we do
- command = safeCommand.addListener(new Command.Listener(txnId));
+ command = safeCommand.addListener(new Command.ProxyListener(txnId));
update.addWaitingOnCommit(command.txnId());
break;
case Committed:
@@ -328,7 +328,7 @@ public class Commands
case ReadyToExecute:
case PreApplied:
case Applied:
- command = safeCommand.addListener(new Command.Listener(txnId));
+ command = safeCommand.addListener(new Command.ProxyListener(txnId));
insertPredecessor(txnId, executeAt, update, command);
case Invalidated:
break;
diff --git a/accord-core/src/main/java/accord/local/CommonAttributes.java b/accord-core/src/main/java/accord/local/CommonAttributes.java
index 033b526e..cd4cfb03 100644
--- a/accord-core/src/main/java/accord/local/CommonAttributes.java
+++ b/accord-core/src/main/java/accord/local/CommonAttributes.java
@@ -35,7 +35,7 @@ public interface CommonAttributes
Route<?> route();
PartialTxn partialTxn();
PartialDeps partialDeps();
- Listeners.Immutable listeners();
+ Listeners.Immutable durableListeners();
default Mutable mutable()
{
@@ -67,7 +67,7 @@ public interface CommonAttributes
this.route = attributes.route();
this.partialTxn = attributes.partialTxn();
this.partialDeps = attributes.partialDeps();
- this.listeners = attributes.listeners();
+ this.listeners = attributes.durableListeners();
}
@Override
@@ -162,7 +162,7 @@ public interface CommonAttributes
}
@Override
- public Listeners.Immutable listeners()
+ public Listeners.Immutable durableListeners()
{
if (listeners == null || listeners.isEmpty())
return Listeners.Immutable.EMPTY;
@@ -171,7 +171,7 @@ public interface CommonAttributes
return new Listeners.Immutable(listeners);
}
- public Mutable addListener(CommandListener listener)
+ public Mutable addListener(Command.DurableAndIdempotentListener listener)
{
if (listeners == null)
listeners = new Listeners();
@@ -181,7 +181,7 @@ public interface CommonAttributes
return this;
}
- public Mutable removeListener(CommandListener listener)
+ public Mutable removeListener(Command.Listener listener)
{
if (listener == null || listeners.isEmpty())
return this;
diff --git a/accord-core/src/main/java/accord/local/Listeners.java b/accord-core/src/main/java/accord/local/Listeners.java
index 45967fd2..c1e4a3a5 100644
--- a/accord-core/src/main/java/accord/local/Listeners.java
+++ b/accord-core/src/main/java/accord/local/Listeners.java
@@ -23,18 +23,18 @@ import accord.utils.DeterministicIdentitySet;
import java.util.Collection;
import java.util.function.Predicate;
-public class Listeners extends DeterministicIdentitySet<CommandListener>
+public class Listeners<L extends Command.Listener> extends DeterministicIdentitySet<L>
{
public Listeners()
{
}
- public Listeners(Listeners copy)
+ public Listeners(Listeners<L> copy)
{
super(copy);
}
- public static class Immutable extends Listeners
+ public static class Immutable extends Listeners<Command.DurableAndIdempotentListener>
{
public static final Immutable EMPTY = new Immutable();
@@ -43,18 +43,18 @@ public class Listeners extends DeterministicIdentitySet<CommandListener>
super();
}
- public Immutable(Listeners listeners)
+ public Immutable(Listeners<Command.DurableAndIdempotentListener> listeners)
{
super(listeners);
}
- Listeners mutable()
+ Listeners<Command.DurableAndIdempotentListener> mutable()
{
- return new Listeners(this);
+ return new Listeners<>(this);
}
@Override
- public boolean add(CommandListener item)
+ public boolean add(Command.DurableAndIdempotentListener item)
{
throw new UnsupportedOperationException("Cannot modify immutable set");
}
@@ -72,7 +72,7 @@ public class Listeners extends DeterministicIdentitySet<CommandListener>
}
@Override
- public boolean addAll(Collection<? extends CommandListener> c)
+ public boolean addAll(Collection<? extends Command.DurableAndIdempotentListener> c)
{
throw new UnsupportedOperationException("Cannot modify immutable set");
}
@@ -84,7 +84,7 @@ public class Listeners extends DeterministicIdentitySet<CommandListener>
}
@Override
- public boolean removeIf(Predicate<? super CommandListener> filter)
+ public boolean removeIf(Predicate<? super Command.DurableAndIdempotentListener> filter)
{
throw new UnsupportedOperationException("Cannot modify immutable set");
}
diff --git a/accord-core/src/main/java/accord/local/SafeCommand.java b/accord-core/src/main/java/accord/local/SafeCommand.java
index 5648860f..a25211b0 100644
--- a/accord-core/src/main/java/accord/local/SafeCommand.java
+++ b/accord-core/src/main/java/accord/local/SafeCommand.java
@@ -18,6 +18,8 @@
package accord.local;
+import java.util.Collection;
+
import accord.api.Result;
import accord.primitives.Ballot;
import accord.primitives.Timestamp;
@@ -37,6 +39,9 @@ public abstract class SafeCommand
public abstract Command current();
public abstract void invalidate();
public abstract boolean invalidated();
+ public abstract void addListener(Command.TransientListener listener);
+ public abstract void removeListener(Command.TransientListener listener);
+ public abstract Collection<Command.TransientListener> transientListeners();
public boolean isEmpty()
{
@@ -56,12 +61,12 @@ public abstract class SafeCommand
return update;
}
- public Command addListener(CommandListener listener)
+ public Command addListener(Command.DurableAndIdempotentListener listener)
{
return update(Command.addListener(current(), listener));
}
- public Command removeListener(CommandListener listener)
+ public Command removeListener(Command.DurableAndIdempotentListener listener)
{
return update(Command.removeListener(current(), listener));
}
diff --git a/accord-core/src/main/java/accord/local/SafeCommandStore.java b/accord-core/src/main/java/accord/local/SafeCommandStore.java
index 892cbfee..e73aff81 100644
--- a/accord-core/src/main/java/accord/local/SafeCommandStore.java
+++ b/accord-core/src/main/java/accord/local/SafeCommandStore.java
@@ -57,7 +57,7 @@ public interface SafeCommandStore
PreLoadContext context = PreLoadContext.contextFor(listOf(txnId, listenerId), Keys.EMPTY);
commandStore().execute(context, safeStore -> {
SafeCommand safeCommand = safeStore.command(txnId);
- CommandListener listener = new Command.Listener(listenerId);
+ Command.ProxyListener listener = new Command.ProxyListener(listenerId);
safeCommand.addListener(listener);
listener.onChange(safeStore, safeCommand);
}).begin(agent());
@@ -101,26 +101,53 @@ public interface SafeCommandStore
default void notifyListeners(SafeCommand safeCommand)
{
- TxnId txnId = safeCommand.txnId();
Command command = safeCommand.current();
- for (CommandListener listener : command.listeners())
+ for (Command.DurableAndIdempotentListener listener : command.durableListeners())
+ notifyListener(this, safeCommand, command, listener);
+
+ for (Command.TransientListener listener : safeCommand.transientListeners())
+ notifyListener(this, safeCommand, command, listener);
+ }
+
+ static void notifyListener(SafeCommandStore safeStore, SafeCommand safeCommand, Command command, Command.TransientListener listener)
+ {
+ if (!safeCommand.transientListeners().contains(listener))
+ return;
+
+ PreLoadContext context = listener.listenerPreLoadContext(command.txnId());
+ if (safeStore.canExecuteWith(context))
+ {
+ listener.onChange(safeStore, safeCommand);
+ }
+ else
+ {
+ TxnId txnId = command.txnId();
+ safeStore.commandStore()
+ .execute(context, safeStore2 -> {
+ SafeCommand safeCommand2 = safeStore2.command(txnId);
+ // listeners invocations may be triggered more than once asynchronously for different changes
+ // so one pending invocation may unregister the listener prior to the second invocation running
+ // so we check if the listener is still valid before running
+ if (safeCommand2.transientListeners().contains(listener))
+ listener.onChange(safeStore2, safeCommand2);
+ })
+ .begin(safeStore.agent());
+ }
+ }
+
+ static void notifyListener(SafeCommandStore safeStore, SafeCommand safeCommand, Command command, Command.DurableAndIdempotentListener listener)
+ {
+ PreLoadContext context = listener.listenerPreLoadContext(command.txnId());
+ if (safeStore.canExecuteWith(context))
+ {
+ listener.onChange(safeStore, safeCommand);
+ }
+ else
{
- if (!safeCommand.current().listeners().contains(listener))
- {
- // notifyListeners is done for every mutation, which can cause listeners to be different depending on
- // where you are in the stack frame...
- // To simplify listeners, double check that this wasn't changed before calling again
- continue;
- }
- PreLoadContext context = listener.listenerPreLoadContext(command.txnId());
- if (canExecuteWith(context))
- {
- listener.onChange(this, safeCommand);
- }
- else
- {
- commandStore().execute(context, safeStore -> listener.onChange(safeStore, safeStore.command(txnId))).begin(agent());
- }
+ TxnId txnId = command.txnId();
+ safeStore.commandStore()
+ .execute(context, safeStore2 -> listener.onChange(safeStore2, safeStore2.command(txnId)))
+ .begin(safeStore.agent());
}
}
}
diff --git a/accord-core/src/main/java/accord/messages/Defer.java b/accord-core/src/main/java/accord/messages/Defer.java
index 8ca0a487..9bc23be7 100644
--- a/accord-core/src/main/java/accord/messages/Defer.java
+++ b/accord-core/src/main/java/accord/messages/Defer.java
@@ -30,7 +30,7 @@ import static accord.messages.Defer.Ready.Expired;
import static accord.messages.Defer.Ready.No;
import static accord.messages.Defer.Ready.Yes;
-class Defer implements CommandListener
+class Defer implements Command.TransientListener
{
public enum Ready { No, Yes, Expired }
@@ -100,11 +100,5 @@ class Defer implements CommandListener
Invariants.checkState(caller.equals(request.txnId));
return request;
}
-
- @Override
- public boolean isTransient()
- {
- return true;
- }
}
diff --git a/accord-core/src/main/java/accord/messages/ReadData.java b/accord-core/src/main/java/accord/messages/ReadData.java
index 6350793c..dda52b7b 100644
--- a/accord-core/src/main/java/accord/messages/ReadData.java
+++ b/accord-core/src/main/java/accord/messages/ReadData.java
@@ -23,6 +23,7 @@ import java.util.Collections;
import java.util.HashSet;
import java.util.Set;
+import accord.local.Command.TransientListener;
import accord.primitives.*;
import accord.local.*;
@@ -43,7 +44,7 @@ import static accord.messages.TxnRequest.*;
import static accord.utils.MapReduceConsume.forEach;
// TODO (required, efficiency): dedup - can currently have infinite pending reads that will be executed independently
-public class ReadData extends AbstractEpochRequest<ReadData.ReadNack> implements CommandListener
+public class ReadData extends AbstractEpochRequest<ReadData.ReadNack> implements TransientListener
{
private static final Logger logger = LoggerFactory.getLogger(ReadData.class);
@@ -54,7 +55,7 @@ public class ReadData extends AbstractEpochRequest<ReadData.ReadNack> implements
return new ReadData(txnId, scope, executeAtEpoch, waitForEpoch);
}
}
- private class ObsoleteTracker implements CommandListener
+ private class ObsoleteTracker implements TransientListener
{
@Override
public void onChange(SafeCommandStore safeStore, SafeCommand safeCommand)
@@ -74,12 +75,6 @@ public class ReadData extends AbstractEpochRequest<ReadData.ReadNack> implements
{
return ReadData.this.listenerPreLoadContext(caller);
}
-
- @Override
- public boolean isTransient()
- {
- return true;
- }
}
private final ObsoleteTracker obsoleteTracker = new ObsoleteTracker();
@@ -131,12 +126,6 @@ public class ReadData extends AbstractEpochRequest<ReadData.ReadNack> implements
return PreLoadContext.contextFor(ids, keys());
}
- @Override
- public boolean isTransient()
- {
- return true;
- }
-
@Override
public synchronized void onChange(SafeCommandStore safeStore, SafeCommand safeCommand)
{
diff --git a/accord-core/src/main/java/accord/messages/WaitOnCommit.java b/accord-core/src/main/java/accord/messages/WaitOnCommit.java
index f27816fd..1dc2dc10 100644
--- a/accord-core/src/main/java/accord/messages/WaitOnCommit.java
+++ b/accord-core/src/main/java/accord/messages/WaitOnCommit.java
@@ -32,7 +32,7 @@ import static accord.local.Status.Committed;
import static accord.utils.Utils.listOf;
import accord.topology.Topology;
-public class WaitOnCommit implements Request, MapReduceConsume<SafeCommandStore, Void>, PreLoadContext, CommandListener
+public class WaitOnCommit implements Request, MapReduceConsume<SafeCommandStore, Void>, PreLoadContext, Command.TransientListener
{
private static final Logger logger = LoggerFactory.getLogger(WaitOnCommit.class);
@@ -146,12 +146,6 @@ public class WaitOnCommit implements Request, MapReduceConsume<SafeCommandStore,
node.reply(replyTo, replyContext, WaitOnCommitOk.INSTANCE);
}
- @Override
- public boolean isTransient()
- {
- return true;
- }
-
@Override
public Iterable<TxnId> txnIds()
{
diff --git a/accord-core/src/main/java/accord/primitives/Timestamp.java b/accord-core/src/main/java/accord/primitives/Timestamp.java
index 6ec20576..6477a567 100644
--- a/accord-core/src/main/java/accord/primitives/Timestamp.java
+++ b/accord-core/src/main/java/accord/primitives/Timestamp.java
@@ -42,7 +42,7 @@ public class Timestamp implements Comparable<Timestamp>
public static Timestamp maxForEpoch(long epoch)
{
- return new Timestamp(epochMsb(epoch) | 0xffff, Long.MAX_VALUE, Id.MAX);
+ return new Timestamp(epochMsb(epoch) | 0x7fff, Long.MAX_VALUE, Id.MAX);
}
public static Timestamp minForEpoch(long epoch)
@@ -50,6 +50,7 @@ public class Timestamp implements Comparable<Timestamp>
return new Timestamp(epochMsb(epoch), 0, Id.NONE);
}
+ // TODO (expected): we can only use 63 bits for HLC as we don't support negative timestamps, so can use 49 bits for epoch
public static final long MAX_EPOCH = (1L << 48) - 1;
private static final long HLC_INCR = 1L << 16;
private static final long MAX_FLAGS = HLC_INCR - 1;
@@ -61,6 +62,7 @@ public class Timestamp implements Comparable<Timestamp>
Timestamp(long epoch, long hlc, int flags, Id node)
{
+ Invariants.checkArgument(hlc >= 0);
Invariants.checkArgument(epoch <= MAX_EPOCH);
Invariants.checkArgument(flags <= MAX_FLAGS);
this.msb = epochMsb(epoch) | hlcMsb(hlc);
@@ -183,12 +185,12 @@ public class Timestamp implements Comparable<Timestamp>
private static long epoch(long msb)
{
- return msb >>> 16;
+ return msb >>> 15;
}
private static long epochMsb(long epoch)
{
- return epoch << 16;
+ return epoch << 15;
}
private static long hlcMsb(long hlc)
@@ -203,7 +205,7 @@ public class Timestamp implements Comparable<Timestamp>
private static long highHlc(long msb)
{
- return msb << 48;
+ return (msb & 0x7fff) << 48;
}
private static long lowHlc(long lsb)
diff --git a/accord-core/src/test/java/accord/burn/BurnTest.java b/accord-core/src/test/java/accord/burn/BurnTest.java
index c4feef07..4ba8b25d 100644
--- a/accord-core/src/test/java/accord/burn/BurnTest.java
+++ b/accord-core/src/test/java/accord/burn/BurnTest.java
@@ -18,6 +18,7 @@
package accord.burn;
+import java.util.ArrayDeque;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
diff --git a/build.gradle b/build.gradle
index 12f01335..26749ad3 100644
--- a/build.gradle
+++ b/build.gradle
@@ -31,4 +31,5 @@ rat {
// List of Gradle exclude directives, defaults to ['**/.gradle/**']
excludes.add("**/build/**")
excludes.add(".idea/**")
+ excludeFile.set(layout.projectDirectory.file(".rat-excludes.txt"))
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@cassandra.apache.org
For additional commands, e-mail: commits-help@cassandra.apache.org