Adding Mutual exclusion logic for SAI index rebuilding and ZCS strea… - #5071
Adding Mutual exclusion logic for SAI index rebuilding and ZCS strea…#5071pranavshenoy wants to merge 1 commit into
Conversation
| reserved.forEach(s -> s.streamRebuildState().endRebuild()); | ||
| throw new RuntimeException(String.format( | ||
| "Cannot build SAI index on %s while entire-sstable (zero-copy) streaming is in progress.", | ||
| sstable.descriptor)); |
There was a problem hiding this comment.
nit: If we move forward with this, it might be nice to have a full listing of the SSTables that cannot be locked for rebuild. More of an operator concern than a correctness problem. It might also be good to cap it at a certain number, like 16 or 32 so we don't have a huge error message.
| { | ||
| if (ownsStreamRebuildStatus) | ||
| sstables.keySet().forEach(sstable -> sstable.streamRebuildState().endRebuild()); | ||
| } |
There was a problem hiding this comment.
Would it be possible to just release the SSTables one by one as they complete? That would narrow the window where ZCS wouldn't be possible...
| // components. An entire-sstable (zero-copy) stream reserves the same status when its outgoing file is | ||
| // constructed, so this ensures a rebuild cannot delete/rewrite SAI components underneath an in-flight | ||
| // stream (and vice versa the stream degrades to legacy). See CASSANDRA-21520. The returned builder owns | ||
| // these reservations and releases them when it finishes. |
There was a problem hiding this comment.
The big design question for this patch is whether or not we have to do this reservation for the entire set of SSTables. I think we do, and it's because there is no partial full index rebuild. If we have 100 SSTables, and 2 of them are being ZCS streamed, we can't just build 98 and wait until those are done with streaming, delaying the rebuild for an arbitrary window. Similarly, we can't lazily check SSTables as we delete and rebuild them, because ZCS streaming on a particular SSTable might start immediately before we attempt to rebuild it.
More tactically, we have group.dropIndexSSTables() below, which will actually delete column indexes on disk if there aren't queries in flight referencing them. Any attempt to change the design to per-SSTable locking would have to account for that.
|
Possible testing gaps worth looking at:
|
| // size no longer matches the size advertised in the manifest, and the production size check in | ||
| // ComponentContext.channel() must fail the stream rather than ship corrupt bytes. This check must be | ||
| // effective even with assertions disabled (it is a real exception, not an assert). CASSANDRA-21520. | ||
| node1.runOnInstance(IndexDeleteDuringEntireSSTableStreamingTest::truncateStreamedSaiComponent); |
There was a problem hiding this comment.
This is basically codifying the current behavior for us calling DROP INDEX? If we want to try to make that safe, we could try something like:
Index: src/java/org/apache/cassandra/io/sstable/format/SSTableStreamRebuildState.java
IDEA additional info:
Subsystem: com.intellij.openapi.diff.impl.patch.CharsetEP
<+>UTF-8
===================================================================
diff --git a/src/java/org/apache/cassandra/io/sstable/format/SSTableStreamRebuildState.java b/src/java/org/apache/cassandra/io/sstable/format/SSTableStreamRebuildState.java
--- a/src/java/org/apache/cassandra/io/sstable/format/SSTableStreamRebuildState.java (revision 5e7b788fa0098cc7fee821f4da8e4693d28b0907)
+++ b/src/java/org/apache/cassandra/io/sstable/format/SSTableStreamRebuildState.java (date 1788392191269)
@@ -18,6 +18,9 @@
package org.apache.cassandra.io.sstable.format;
+import java.util.ArrayList;
+import java.util.List;
+
import com.google.common.annotations.VisibleForTesting;
/**
@@ -41,6 +44,11 @@
private State state = State.NORMAL;
private int zcsStreamCount = 0;
+ // One-shot cleanup tasks (e.g. deleting a dropped/corrupt column index's components) that lost the race
+ // against an in-progress stream and are waiting for it to end. Drained back through
+ // deleteOrDeferUntilStreamingEnds() once the last stream finishes, so a task that loses the race again
+ // (e.g. to a rebuild that started in the meantime) simply re-queues instead of being lost.
+ private List<Runnable> pendingCleanup;
/**
* Attempt to begin an entire-sstable stream. Fails only if a rebuild is in progress.
@@ -60,11 +68,28 @@
/**
* Release one entire-sstable stream. Returns to {@code NORMAL} when the last stream ends. Defensive against
* over-release so cleanup paths cannot corrupt the state.
+ * <p>
+ * When the last stream ends, any cleanup deferred by {@link #deleteOrDeferUntilStreamingEnds} is retried.
*/
- public synchronized void endStreaming()
+ public void endStreaming()
{
- if (zcsStreamCount > 0 && --zcsStreamCount == 0)
- state = State.NORMAL;
+ List<Runnable> toRetry = null;
+ synchronized (this)
+ {
+ if (zcsStreamCount > 0 && --zcsStreamCount == 0)
+ {
+ state = State.NORMAL;
+ if (pendingCleanup != null)
+ {
+ toRetry = pendingCleanup;
+ pendingCleanup = null;
+ }
+ }
+ }
+ // Re-submit rather than run directly: something else (a new stream, a new rebuild) may have claimed
+ // NORMAL before we get here, in which case this simply re-queues instead of racing it.
+ if (toRetry != null)
+ toRetry.forEach(this::deleteOrDeferUntilStreamingEnds);
}
/**
@@ -89,6 +114,39 @@
state = State.NORMAL;
}
+ /**
+ * Runs a one-shot component cleanup (e.g. deleting a dropped or corrupt column index's files) with the same
+ * exclusivity a rebuild would use, so it cannot race a concurrent entire-sstable stream. There is no caller
+ * here to report failure to (this is invoked from arbitrary release() completions, e.g. a query finishing),
+ * so losing the race is not an error: {@code cleanup} is queued and retried once the active stream(s) end.
+ */
+ public void deleteOrDeferUntilStreamingEnds(Runnable cleanup)
+ {
+ boolean runNow;
+ synchronized (this)
+ {
+ runNow = tryBeginRebuild();
+ if (!runNow)
+ {
+ if (pendingCleanup == null)
+ pendingCleanup = new ArrayList<>();
+ pendingCleanup.add(cleanup);
+ }
+ }
+
+ if (runNow)
+ {
+ try
+ {
+ cleanup.run();
+ }
+ finally
+ {
+ endRebuild();
+ }
+ }
+ }
+
@VisibleForTesting
public synchronized State state()
{
Index: src/java/org/apache/cassandra/index/sai/StorageAttachedIndexGroup.java
IDEA additional info:
Subsystem: com.intellij.openapi.diff.impl.patch.CharsetEP
<+>UTF-8
===================================================================
diff --git a/src/java/org/apache/cassandra/index/sai/StorageAttachedIndexGroup.java b/src/java/org/apache/cassandra/index/sai/StorageAttachedIndexGroup.java
--- a/src/java/org/apache/cassandra/index/sai/StorageAttachedIndexGroup.java (revision 5e7b788fa0098cc7fee821f4da8e4693d28b0907)
+++ b/src/java/org/apache/cassandra/index/sai/StorageAttachedIndexGroup.java (date 1788392260481)
@@ -323,14 +323,16 @@
if (!results.right.isEmpty())
{
results.right.forEach(sstable -> {
- IndexDescriptor indexDescriptor = IndexDescriptor.create(sstable);
- indexDescriptor.deletePerSSTableIndexComponents();
- // Column indexes are invalid if their SSTable-level components are corrupted so delete
- // their associated index files and mark them non-queryable.
- indexes.forEach(index -> {
- indexDescriptor.deleteColumnIndex(index.termType(), index.identifier());
- index.makeIndexNonQueryable();
+ // Deleting these files must not race a concurrent entire-sstable stream of the same sstable
+ // (CASSANDRA-21520).
+ sstable.streamRebuildState().deleteOrDeferUntilStreamingEnds(() -> {
+ IndexDescriptor indexDescriptor = IndexDescriptor.create(sstable);
+ indexDescriptor.deletePerSSTableIndexComponents();
+ // Column indexes are invalid if their SSTable-level components are corrupted so delete
+ // their associated index files and mark them non-queryable.
+ indexes.forEach(index -> indexDescriptor.deleteColumnIndex(index.termType(), index.identifier()));
});
+ indexes.forEach(StorageAttachedIndex::makeIndexNonQueryable);
});
return indexes;
}
@@ -344,8 +346,10 @@
if (!invalid.isEmpty())
{
// Delete the index files and mark the index non-queryable, as its view may be compromised,
- // and incomplete, for our callers:
- invalid.forEach(context -> context.indexDescriptor.deleteColumnIndex(index.termType(), index.identifier()));
+ // and incomplete, for our callers. Deletion must not race a concurrent entire-sstable stream
+ // of the same sstable (CASSANDRA-21520).
+ invalid.forEach(context -> context.sstable.streamRebuildState()
+ .deleteOrDeferUntilStreamingEnds(() -> context.indexDescriptor.deleteColumnIndex(index.termType(), index.identifier())));
index.makeIndexNonQueryable();
incomplete.add(index);
}
Index: src/java/org/apache/cassandra/index/sai/disk/SSTableIndex.java
IDEA additional info:
Subsystem: com.intellij.openapi.diff.impl.patch.CharsetEP
<+>UTF-8
===================================================================
diff --git a/src/java/org/apache/cassandra/index/sai/disk/SSTableIndex.java b/src/java/org/apache/cassandra/index/sai/disk/SSTableIndex.java
--- a/src/java/org/apache/cassandra/index/sai/disk/SSTableIndex.java (revision 5e7b788fa0098cc7fee821f4da8e4693d28b0907)
+++ b/src/java/org/apache/cassandra/index/sai/disk/SSTableIndex.java (date 1788392207795)
@@ -232,10 +232,15 @@
/*
* When SSTable is removed, storage-attached index components will be automatically removed by LogTransaction.
* We only remove index components explicitly in case of index corruption or index rebuild.
+ *
+ * Deleting these files must not race a concurrent entire-sstable (zero-copy) stream of the same
+ * sstable, which reads them directly off disk (CASSANDRA-21520). getSSTable() is still valid here:
+ * sstableContext.close() above releases its own closeables, not the underlying SSTableReader.
*/
if (obsolete.get())
{
- sstableContext.indexDescriptor.deleteColumnIndex(indexTermType, indexIdentifier);
+ getSSTable().streamRebuildState()
+ .deleteOrDeferUntilStreamingEnds(() -> sstableContext.indexDescriptor.deleteColumnIndex(indexTermType, indexIdentifier));
}
}
}
2 participants
…ming
Thanks for sending a pull request! Here are some tips if you're new here:
Commit messages should follow the following format:
The Cassandra Jira