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

CEP-15 (Accord) Original and recover coordinators may hit a race condition with PreApply where reads and writes are interleaved, causing one of the coordinators to see the writes from the other 

patch by David Capwell; reviewed by Ariel Weisberg for CASSANDRA-18422
This commit is contained in:
David Capwell 2023-04-07 15:39:42 -07:00
parent 44824a6e45
commit 7ad2bf672c
4 changed files with 194 additions and 80 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
modules/accord

@ -1 +1 @@
Subproject commit bc81f81c75f93c73989a30bbc51b5c241a893c1a
Subproject commit 08aaab6e33d43406e0649146144e4df67648602a

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

@ -19,6 +19,7 @@
package org.apache.cassandra.service.accord;
import java.util.Map;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.TimeUnit;
@ -42,6 +43,7 @@ import accord.primitives.RoutableKey;
import accord.primitives.TxnId;
import accord.utils.Invariants;
import accord.utils.async.AsyncChain;
import accord.utils.async.AsyncChains;
import org.apache.cassandra.service.accord.async.AsyncOperation;
import org.apache.cassandra.utils.Clock;
import org.apache.cassandra.utils.concurrent.UncheckedInterruptedException;
@ -187,6 +189,12 @@ public class AccordCommandStore implements CommandStore
return AsyncOperation.create(this, loadCtx, function);
}
@Override
public <T> AsyncChain<T> submit(Callable<T> task)
{
return AsyncChains.ofCallable(executor, task);
}
public DataStore dataStore()
{
return dataStore;

251
test/simulator/test/org/apache/cassandra/simulator/paxos/HistoryValidatorTest.java
View File

@ -44,6 +44,7 @@ import com.carrotsearch.hppc.IntIntHashMap;
import com.carrotsearch.hppc.IntIntMap;
import com.carrotsearch.hppc.IntSet;
import org.apache.cassandra.config.CassandraRelevantProperties;
import com.carrotsearch.hppc.cursors.IntCursor;
import org.apache.cassandra.distributed.api.QueryResults;
import org.apache.cassandra.utils.Clock;
import org.assertj.core.api.AbstractThrowableAssert;
@ -282,6 +283,32 @@ public class HistoryValidatorTest
);
}
private static String trim(String log, int... keys)
{
// this is deaad code, but exists to help when new validation errors are detected
// the logic will shrink the history to only contain transactions that contain the set of keys
IntSet set = new IntHashSet();
IntStream.of(keys).forEach(set::add);
Parsed parsed = parse(log);
StringBuilder sb = new StringBuilder();
for (Witness w : parsed.witnesses)
{
boolean match = false;
for (IntCursor pk : w.pks())
{
if (set.contains(pk.value))
{
match = true;
break;
}
}
if (!match) continue;
sb.append(w).append("\n");
}
return sb.toString();
}
private void requiresMultiKeySupport()
{
Assume.assumeTrue("Validator " + factory.getClass() + " does not support multi-key", factory instanceof StrictSerializabilityValidator.Factory);
@ -357,79 +384,146 @@ public class HistoryValidatorTest
return new Event(EnumSet.of(Event.Type.WRITE), pk, null);
}
private void fromLog(String log)
private interface Operation
{
int pk();
void check(HistoryValidator.Checker check);
void appendString(StringBuilder sb);
}
private static class Read implements Operation
{
final int pk, id, count;
final int[] seq;
Read(int pk, int id, int count, int[] seq)
{
this.pk = pk;
this.id = id;
this.count = count;
this.seq = seq;
}
@Override
public int pk()
{
return pk;
}
@Override
public void check(HistoryValidator.Checker check)
{
check.read(pk, id, count, seq);
}
@Override
public void appendString(StringBuilder sb)
{
sb.append("read(pk=").append(pk).append(", id=").append(id).append(", count=").append(count).append(", seq=").append(Arrays.toString(seq)).append(")\n");
}
}
private static class Write implements Operation
{
final int pk, id;
final boolean success;
Write(int pk, int id, boolean success)
{
this.pk = pk;
this.id = id;
this.success = success;
}
@Override
public int pk()
{
return pk;
}
@Override
public void check(HistoryValidator.Checker check)
{
check.write(pk, id, success);
}
@Override
public void appendString(StringBuilder sb)
{
sb.append("write(pk=").append(pk).append(", id=").append(id).append(", success=").append(success).append(")\n");
}
}
private static class Witness
{
final int start, end;
final List<Operation> actions = new ArrayList<>();
Witness(int start, int end)
{
this.start = start;
this.end = end;
}
void read(int pk, int id, int count, int[] seq)
{
actions.add(new Read(pk, id, count, seq));
}
void write(int pk, int id, boolean success)
{
actions.add(new Write(pk, id, success));
}
void process(HistoryValidator validator)
{
try (HistoryValidator.Checker check = validator.witness(start, end))
{
for (Operation a : actions)
a.check(check);
}
}
IntSet pks()
{
IntSet pks = new IntHashSet();
for (Operation action : actions)
pks.add(action.pk());
return pks;
}
@Override
public String toString()
{
StringBuilder sb = new StringBuilder();
sb.append("Witness(start=").append(start).append(", end=").append(end).append(")\n");
for (Operation a : actions)
a.appendString(sb.append('\t'));
return sb.toString();
}
}
private static class Parsed
{
private final int[] keys;
private final List<Witness> witnesses;
private Parsed(int[] keys, List<Witness> witnesses)
{
this.keys = keys;
this.witnesses = witnesses;
}
}
private static Parsed parse(String log)
{
IntSet pks = new IntHashSet();
class Read
{
final int pk, id, count;
final int[] seq;
Read(int pk, int id, int count, int[] seq)
{
this.pk = pk;
this.id = id;
this.count = count;
this.seq = seq;
}
}
class Write
{
final int pk, id;
final boolean success;
Write(int pk, int id, boolean success)
{
this.pk = pk;
this.id = id;
this.success = success;
}
}
class Witness
{
final int start, end;
final List<Object> actions = new ArrayList<>();
Witness(int start, int end)
{
this.start = start;
this.end = end;
}
void read(int pk, int id, int count, int[] seq)
{
actions.add(new Read(pk, id, count, seq));
}
void write(int pk, int id, boolean success)
{
actions.add(new Write(pk, id, success));
}
void process(HistoryValidator validator)
{
try (HistoryValidator.Checker check = validator.witness(start, end))
{
for (Object a : actions)
{
if (a instanceof Read)
{
Read read = (Read) a;
check.read(read.pk, read.id, read.count, read.seq);
}
else
{
Write write = (Write) a;
check.write(write.pk, write.id, write.success);
}
}
}
}
}
List<Witness> witnesses = new ArrayList<>();
Witness current = null;
for (String line : log.split("\n"))
{
if (line.trim().isEmpty())
continue;
if (line.startsWith("Witness"))
{
if (current != null)
@ -469,9 +563,26 @@ public class HistoryValidatorTest
witnesses.add(current);
int[] keys = pks.toArray();
Arrays.sort(keys);
HistoryValidator validator = factory.create(keys);
for (Witness w : witnesses)
w.process(validator);
return new Parsed(keys, witnesses);
}
private void fromLog(String log)
{
Parsed parsed = parse(log);
HistoryValidator validator = factory.create(parsed.keys);
for (Witness w : parsed.witnesses)
{
try
{
w.process(validator);
}
catch (HistoryViolation e)
{
HistoryViolation hv = new HistoryViolation(e.primaryKey, "Violation detected for witnessed action " + w + "; " + e.getMessage() + ";\n" + log);
hv.setStackTrace(e.getStackTrace());
throw hv;
}
}
}
private static class Event

13
test/unit/org/apache/cassandra/service/accord/async/AsyncOperationTest.java
View File

@ -42,8 +42,8 @@ import org.slf4j.LoggerFactory;
import accord.api.RoutingKey;
import accord.impl.SafeCommandsForKey;
import accord.local.CheckedCommands;
import accord.local.Command;
import accord.local.Commands;
import accord.local.PreLoadContext;
import accord.local.SafeCommand;
import accord.local.SafeCommandStore;
@ -194,14 +194,9 @@ public class AsyncOperationTest
try
{
return getUninterruptibly(commandStore.submit(PreLoadContext.contextFor(Collections.singleton(txnId), partialTxn.keys()), safe -> {
Commands.AcceptOutcome result = Commands.preaccept(safe, txnId, partialTxn, route, null);
if (result != Commands.AcceptOutcome.Success) throw new IllegalStateException("Command mutation rejected: " + result);
result = Commands.accept(safe, txnId, Ballot.ZERO, partialRoute, partialTxn.keys(), null, executeAt, deps);
if (result != Commands.AcceptOutcome.Success) throw new IllegalStateException("Command mutation rejected: " + result);
Commands.CommitOutcome commit = Commands.commit(safe, txnId, route, null, partialTxn, executeAt, deps);
if (commit != Commands.CommitOutcome.Success) throw new IllegalStateException("Command mutation rejected: " + result);
CheckedCommands.preaccept(safe, txnId, partialTxn, route, null);
CheckedCommands.accept(safe, txnId, Ballot.ZERO, partialRoute, partialTxn.keys(), null, executeAt, deps);
CheckedCommands.commit(safe, txnId, route, null, partialTxn, executeAt, deps);
// clear cache
long cacheSize = commandStore.getCacheSize();