Complete FASTER sessions before releasing them during query processing

[davidmrdavid]

Should fix: #383

Context

Some customers are reporting large numbers of FASTER warnings of the following form:

Part00 query 52eaf82R1809 attempt 2024-04-18T09:25:04.4847863Z enumeration failed with exception System.IndexOutOfRangeException: Index was outside the bounds of the array.
at System.Collections.Generic.Dictionary2.TryInsert(TKey key, TValue value, InsertionBehavior behavior) at FASTER.core.FasterKV2.HandleOperationStatus[Input,Output,Context](FasterExecutionContext3 opCtx, PendingContext3& pendingContext, OperationStatus operationStatus, AsyncIOContext2& request) at DurableTask.Netherite.Faster.FasterKV.ReadOnQuerySessionAsync(String instanceId, CancellationToken cancellationToken) in /_/src/DurableTask.Netherite/StorageLayer/Faster/FasterKV.cs:line 59 at DurableTask.Netherite.Faster.FasterKV.QueryEnumeratedStates(EffectTracker effectTracker, PartitionQueryEvent queryEvent, IEnumerator1 enumerator, Int32 pageSize, TimeSpan timeBudget, DateTime attempt)+MoveNext() in /_/src/DurableTask.Netherite/StorageLayer/Faster/FasterKV.cs:line 1047

The stack trace here suggests this is occurring when processing DF client queries (like getting the status of an instanceId).

@TedHartMS suspects this error is occurring because two threads may be using the same FASTER session at the same time.

We originally thought this could not be the case as we have tight concurrency control over FASTER sessions in ReadOnQuerySessionAsync (notice the use of ConcurrentBag to obtain sessionIDs in a thread-safe way, and the semaphore):

durabletask-netherite/src/DurableTask.Netherite/StorageLayer/Faster/FasterKV.cs

Lines 50 to 71 in f8b5634

	async ValueTask<FasterKV<Key, Value>.ReadAsyncResult<EffectTracker, Output, object>> ReadOnQuerySessionAsync(string instanceId, CancellationToken cancellationToken)
	{
	await availableQuerySessions.WaitAsync();
	try
	{
	bool success = this.idleQuerySessions.TryTake(out var session);
	this.partition.Assert(success, "available sessions must be larger than or equal to semaphore count");
	try
	{
	var result = await this.querySessions[session].ReadAsync(TrackedObjectKey.Instance(instanceId), token: cancellationToken).ConfigureAwait(false);
	return result;
	}
	finally
	{
	this.idleQuerySessions.Add(session);
	}
	}
	finally
	{
	availableQuerySessions.Release();
	}
	}

However, Ted called out that we're not calling .Complete() on that session's result before exiting that method, which means that the FASTER session may be obtained through the ConcurrentBag before FASTER is done cleaning up it's previous execution data, leading to the exception reported above. Today, this .Complete() method is called later on, in the caller of the ReadOnQuerySessionAsync, here:

durabletask-netherite/src/DurableTask.Netherite/StorageLayer/Faster/FasterKV.cs

Lines 1055 to 1056 in f8b5634

	(Status status, Output output) = response.Complete();

This PR

This PR attempts to call refactor the code in ReadOnQuerySessionAsync and it's caller so that .Complete is called in ReadOnQuerySessionAsync itself, hopefully preventing the exception.

It also renames some variables in relevant methods to make them, in my opinion, easier to read. I also added several comments to help future maintainers understand the logic. This is a subjective change, and I'm open to feedback.

Request for reviewiers

I worry that there may be a reason why we weren't calling .Complete() in ReadOnQuerySessionAsync: most likely due to performance reasons? I'd appreciate a focus on possible performance regressions.

I also notice the caller of ReadOnQuerySessionAsync follows a "producer-consumer" pattern, where the producer issues FASTER read requests (through ReadOnQuerySessionAsync) and the consumer calls .Complete() on them. Now, the producer calls .Complete() and I worry I may have broken some underlying assumptions of that design. Please let me know if that seems true to you all.

[davidmrdavid]

here I'm just re-ordering some variables, to help with readability (it's a long method, I'm trying to group related things together).

[jviau]

This overall looks safe to me. You have moved the Complete call from just after channel read to immediately before channel write. There is no other code (of this repo) between those two points. But more importantly, you complete the FASTER session before the finally block runs, returning the session for another to take.

I can see how it was a race condition before

[davidmrdavid]

I'm going temporarily add a suffix to this PR's version to try and create a private release

[jviau]

I recommend just using preview. It is fine to publicly release a change like this as preview.

[sebastianburckhardt]

I worry that there may be a reason why we weren't calling .Complete() in ReadOnQuerySessionAsync:

I can't think of a good reason, perhaps performance was a concern, yes, but I don't think it really matters since from what I understand calling Complete() should always finish quickly.

I worry I may have broken some underlying assumptions of that design

looks fine to me.

[davidmrdavid]

Wonderful. If the diff looks good (it ended up looked weirder than anticipated after I indented everything in a try-finally block), then I'd love to run the perf tests with this change to see if we see any of the reported warnings still (either the dictionary conflict, or the semaphore ObjectDisposedException ). If we don't, then we're good. May need assistance with that, I don't have the scripts.

[davidmrdavid]

@sebastianburckhardt: can I get a review from you here? I'd like to get this in, if it looks good, on the upcoming release. We're hoping to release sometime next week

[sebastianburckhardt]

For consistency this await should use ConfigureAwait(false) also, I think.

And perhaps we could add a ReportProgress after this await? So we can detect cases where the await hangs or where the finally block does not get called (finally block in method with yield gets called only if caller correctly disposes enumerator).

[davidmrdavid]

Added the ConfigureAwait(false)

[davidmrdavid]

And added the ReportProgress on my last commit

[sebastianburckhardt]

Is there a specific reason you added a try/finally wrapper? I suppose it is for handling additional places where exceptions could be thrown that are not already caught?

[davidmrdavid]

That's precisely right. I just want to make sure there's no usages of the semaphore after it's been disposed and, to do that, I need to make sure the producer task is awaited before we exit this method (as after we exit, the semaphore will be disposed)