Skip to content

Fixes around CU copies and delete deadlocks #2082

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
kevin-montrose merged 3 commits into from
Aug 24, 2026
Merged

Fixes around CU copies and delete deadlocks #2082

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
391ec3b
with recovery cleanup, blocking during delete request is unnecessary …
kevin-montrose Aug 24, 2026
6f985ba
update vector-sets.md
kevin-montrose Aug 24, 2026
991de99
Merge branch 'main' into users/kmontrose/vectorSessionFunctionSizingF…
kevin-montrose Aug 24, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
40 changes: 5 additions & 35 deletions libs/server/Resp/Vector/VectorManager.Cleanup.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -87,7 +87,7 @@ public bool Reader<TSourceLogRecord>(in TSourceLogRecord logRecord, RecordMetada
}

private readonly VectorSetCleanupWorkChannel<object> cleanupTaskChannel;
private readonly VectorSetCleanupWorkChannel<(ulong Context, TaskCompletionSource MarkCompleted)> requestCleanupTaskChannel;
private readonly VectorSetCleanupWorkChannel<ulong> requestCleanupTaskChannel;
private readonly VectorSetCleanupWorkChannel<object> requestDropTaskChannel;
private readonly VectorSetCleanupWorkSet<(ulong Context, nint IndexPtr)> requestedDrops;
private readonly ConcurrentDictionary<ulong, byte[]> potentiallyDeleted;
Expand Down Expand Up @@ -203,8 +203,6 @@ private async Task RunRequestCleanupTaskAsync()
//
// The fact that we're in an OnDispose means Reset() isn't running.

var completions = new List<TaskCompletionSource>();

try
{
// TODO: this doesn't work with non-RESP impls... which maybe we don't care about?
Expand All @@ -220,14 +218,9 @@ private async Task RunRequestCleanupTaskAsync()
lock (this)
{
// Read all pending requests so we can do one update
while (requestCleanupTaskChannel.TryRead(out var t))
while (requestCleanupTaskChannel.TryRead(out var context))
Comment thread
kevin-montrose marked this conversation as resolved.
{
if (t.MarkCompleted != null)
{
completions.Add(t.MarkCompleted);
}

var (contextIndex, contextValue) = ContextMetadata.DecomposeContext(t.Context);
var (contextIndex, contextValue) = ContextMetadata.DecomposeContext(context);
if (!contextMetadatas[contextIndex].IsCleaningUp(contextIndex != 0, contextValue))
{
contextMetadatas[contextIndex].MarkCleaningUp(contextIndex != 0, contextValue);
Expand All @@ -246,35 +239,12 @@ private async Task RunRequestCleanupTaskAsync()

ExceptionInjectionHelper.TriggerException(ExceptionInjectionType.VectorSet_Interrupt_Delete_3);

foreach (var completion in completions)
{
try
{
_ = completion.TrySetResult();
}
catch (Exception innerE)
{
logger?.LogError(innerE, "While completing Vector Set cleanup request");
}
}

// Pump the cleanup task once we're done
_ = cleanupTaskChannel.TryPublish();
}
catch (Exception e)
{
foreach (var completion in completions)
{
try
{
_ = completion.TrySetException(e);
}
catch (Exception innerE)
{
// Best effort
logger?.LogError(innerE, "While cancelling Vector Set cleanup requests");
}
}
logger?.LogError(e, "During request cleanup task");
}
Comment thread
kevin-montrose marked this conversation as resolved.
finally
{
Expand Down Expand Up @@ -536,7 +506,7 @@ static void QueueCleanups(VectorManager self)
if (needsDelete)
{
// No need to wait for marking, since the record is already "deleted"
if (!self.requestCleanupTaskChannel.TryPublish((context, null)))
if (!self.requestCleanupTaskChannel.TryPublish(context))
{
self.logger?.LogWarning("Could not request delete of abandoned Vector Set {key}", SpanByte.ToShortString(key));
}
Expand Down
11 changes: 2 additions & 9 deletions libs/server/Resp/Vector/VectorManager.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -667,7 +667,7 @@ internal bool TrySetAttribute(ReadOnlySpan<byte> indexValue, ReadOnlySpan<byte>
/// <summary>
/// Request deletion of a Vector Set given the VALUE of the index key.
/// </summary>
internal void RequestDeletion(Span<byte> value)
internal void RequestDeletion(ReadOnlySpan<byte> value)
{
if (value.Length != IndexSize)
{
Expand All @@ -687,16 +687,11 @@ internal void RequestDeletion(Span<byte> value)
return;
}

var tcs = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);

if (!requestCleanupTaskChannel.TryPublish((context, tcs)))
if (!requestCleanupTaskChannel.TryPublish(context))
{
throw new GarnetException("Could not submit request for Vector Set cleanup, aborting delete");
}

// Wait until the context is _marked_ for cleanup, but not the actual cleanup
AsyncUtils.BlockingWait(tcs.Task);

// Tell DiskANN to clean itself up
DropIndex(value);
}
Expand All @@ -709,8 +704,6 @@ internal void RequestDeletion(Span<byte> value)
/// There's subtlety here because the DiskANN index might be in use (on the current or other threads)
/// and we can't allow the index to be recreated until any requested drops are processed.
/// </summary>
/// <param name="key"></param>
/// <param name="value"></param>
internal void RequestDropInMemoryIndex(ReadOnlySpan<byte> key, ReadOnlySpan<byte> value)
{
if (value.Length != IndexSize)
Expand Down
2 changes: 1 addition & 1 deletion libs/server/Storage/Functions/VectorStore/VectorSessionFunctions.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -369,7 +369,7 @@ public readonly bool CopyUpdater<TSourceLogRecord>(in TSourceLogRecord srcLogRec
Debug.Assert(input.WriteDesiredSize <= newValueAligned.Length, "Insufficient space for copy update, this should never happen");
Debug.Assert(input.WriteDesiredSize <= oldValueAligned.Length, "Insufficient space for copy update, this should never happen");

oldValueAligned.CopyTo(newValueAligned);
oldValueAligned[..input.WriteDesiredSize].CopyTo(newValueAligned[..input.WriteDesiredSize]);
Comment thread
kevin-montrose marked this conversation as resolved.

unsafe
{
Expand Down
1 change: 0 additions & 1 deletion website/docs/dev/vector-sets.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -135,7 +135,6 @@ Deletion of Vector Sets is detected in the `GarnetTriggers.OnDispose` callback,
This takes place in four steps:
1. The Vector Set context is marked for deletion from `GarnetTriggers.OnDispose`
* A background task does this, as we do not have a usable storage session in the `GarnetTriggers` callback
* We _block_ on that background task, if an error occurs an exception is raised and the delete aborted
2. `GarnetTriggers.OnDispose` returns, deleting the index key
3. A background cleanup task scans the Tsavorite log for element keys, [see Cleanup](#cleanup) for more detail
4. The Vector Set context is marked available
Comment thread
kevin-montrose marked this conversation as resolved.
Expand Down
Loading