belliottsmith commented on code in PR #4247:
URL: https://github.com/apache/cassandra/pull/4247#discussion_r2205845729
##########
src/java/org/apache/cassandra/service/accord/AccordJournal.java:
##########
@@ -492,54 +504,202 @@ public void forEach(Consumer<JournalKey> consumer)
@Override
public void replay(CommandStores commandStores)
{
- final Semaphore concurrency =
Semaphore.newSemaphore(FBUtilities.getAvailableProcessors());
+ // TODO (expected): make the parallelisms configurable
+ final Semaphore commandParallelism =
Semaphore.newSemaphore(getAvailableProcessors());
+ final int commandStoreParallelism = Math.max(Math.max(1,
Math.min(getAvailableProcessors(), 4)), getAvailableProcessors() / 4);
final AtomicBoolean abort = new AtomicBoolean();
+ // TODO (expected): balance work submission by AccordExecutor
+ final IntArrayList activeCommandStoreIds = new IntArrayList();
+ final ReplayQueue pendingCommandStores = new
ReplayQueue(commandStores.all());
- try (CloseableIterator<Journal.KeyRefs<JournalKey>> iter =
journalTable.keyIterator())
+ class ReplayStream implements Closeable
{
- JournalKey prev = null;
- while (iter.hasNext())
+ final CommandStore commandStore;
+ final Loader loader;
+ final CloseableIterator<Journal.KeyRefs<JournalKey>> iter;
+ JournalKey prev;
+
+ public ReplayStream(CommandStore commandStore)
{
- if (abort.get())
- break;
+ this.commandStore = commandStore;
+ this.loader = commandStore.loader();
+ this.iter = journalTable.keyIterator(new
JournalKey(TxnId.NONE, COMMAND_DIFF, commandStore.id()), new
JournalKey(TxnId.MAX.withoutNonIdentityFlags(), COMMAND_DIFF,
commandStore.id()));
+ }
+ boolean replay()
+ {
JournalKey key;
long[] segments;
+ while (true)
{
+ if (!iter.hasNext())
+ {
+ logger.info("Completed replay of {}", commandStore);
+ return false;
+ }
+
Journal.KeyRefs<JournalKey> ref = iter.next();
- key = ref.key();
- if (key.type != JournalKey.Type.COMMAND_DIFF)
+ if (ref.key().type != COMMAND_DIFF)
continue;
+ key = ref.key();
segments = journalTable.shouldIndex(key) ?
ref.copyOfSegments() : null;
+ break;
}
- CommandStore commandStore =
commandStores.forId(key.commandStoreId);
- AccordCommandStoreLoader loader = (AccordCommandStoreLoader)
commandStore.loader();
TxnId txnId = key.id;
Invariants.require(prev == null ||
key.commandStoreId != prev.commandStoreId ||
key.id.compareTo(prev.id) != 0,
"duplicate key detected %s == %s", key,
prev);
prev = key;
-
- concurrency.acquireThrowUncheckedOnInterrupt(1);
+ commandParallelism.acquireThrowUncheckedOnInterrupt(1);
loader.load(txnId)
.map(route -> {
- if (segments != null)
+ if (segments != null && route != null)
{
for (long segment : segments)
- journalTable.safeNotify(index ->
index.update(segment, key.commandStoreId, txnId, route));
+ journalTable.safeNotify(index ->
index.update(segment, key.commandStoreId, txnId, (Route)route));
}
return null;
}).begin((success, fail) -> {
- concurrency.release(1);
+ commandParallelism.release(1);
if (fail != null && !journal.handleError("Could not
replay command " + txnId, fail))
abort.set(true);
});
+
+ return true;
+ }
+
+ @Override
+ public void close()
+ {
+ iter.close();
+ }
+ }
+
+ final Int2ObjectHashMap<ReplayStream> replayStreams = new
Int2ObjectHashMap<>();
+ try
+ {
+ int cur = 0;
+ while (true)
+ {
+ if (abort.get())
+ break;
+
+ if (cur == activeCommandStoreIds.size())
+ {
+ if (activeCommandStoreIds.size() < commandStoreParallelism
&& !pendingCommandStores.isEmpty())
+ {
+ CommandStore next = pendingCommandStores.next();
+ int id = next.id();
+ activeCommandStoreIds.add(id);
+ replayStreams.put(id, new ReplayStream(next));
Review Comment:
I actually don't think there's a need for this, as we shouldn't write new
data for command stores we haven't started replaying yet, and if the existing
files get compacted that also seems fine so long as we witness the newer
ones....?
But, we can probably encapsulate a supplier that takes its own references if
you like, shouldn't be too hard.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
