Fix AwaitCompletion to yield results during source iteration #505

atifaziz · 2018-05-29T13:57:32Z

This PR addresses #502.

Changes:

A continuation is used to post task completions to the main loop so they can be reported and yielded while the source is still being iterated
A semaphore is used to gate concurrency (when limited)

The source iteration loop has been refactored considerably. It is more complex since it uses asynchronous notification and requires synchronisation of shared state. Previously, the naïve approach was to loop through the entire source, launching tasks and then wait in another loop to post notifications as they complete. That unfortunately led to the behaviour documented in #502.

Especially if we fail to notify under low memory conditions!

This removes the need for Lazy allocations

This reverts commit d973670.

atifaziz · 2018-05-29T18:26:16Z

Note that there are race conditions permitted, like some tasks could potentially post completion notices even after an error and termination of the main loop. It could cause some waste like keeping and adding to the notices collection but the impact should be very marginal. An optimisation could be added in the future by nulling out the notices collection so it's available for GC and notices are dropped if the collection is no longer (immaterial because no one would be listening on the other end of the line). In this PR, I am most interested in focusing on fixing the issue raised in #502 from a behavioural perspective (as long as the race conditions are harmless).

fsateler

Unfortunately we don't have tests for this operator... so we can't check this hasn't introduced regressions...

fsateler · 2018-05-29T16:13:32Z

MoreLinq/Experimental/Await.cs

+                        catch (OperationCanceledException e) when (e.CancellationToken == consumerCancellationTokenSource.Token)
+                        {
+                            var (error1, error2) = lastCriticalErrors;
+                            throw new Exception("One or more critical errors have occurred.",


What is the purpose of wrapping the AggregateException with an Exception?

The rationale behind this is that the AggregateException is an informational detail of the critical exception that's actually being raised. The exception raised here is due to a critical condition that made it impossible for the method to guarantee correct operation and report the original exception(s) (in the detail, as an aggregate).

I'm not following. This exception is telling me one or more errors occured. And this is precisely what AggregateException is for.

AggregateException is common with tasks and could confuse someone if AwaitCompletion throws it too, and if they have a catch for AggregateException. My intention was to make a clear distinction here. This is a single exception due to critical conditions, period. The aggregate within is for informational and diagnostic purposes only. Perhaps this is a futile attempt at semantics that exist just in my head?

fsateler · 2018-05-29T18:44:33Z

MoreLinq/Experimental/Await.cs

-                while (tasks.Count > 0)
+                var semaphore = maxConcurrency is int count
+                              ? new SemaphoreSlim(count, count)
+                              : null;


Finally the docs don't lie 😉

previously, using UnboundedConcurrency actually meant int.MaxValue concurrency. Now it really means unbounded.

Oh I see. Yeah. 😃

fsateler · 2018-05-29T18:52:49Z

MoreLinq/Experimental/Await.cs

+        static async Task StartAsync<T, TResult>(
+            this IEnumerator<T> enumerator,
+            Func<T, Task<TResult>> starter,
+            Action<T, Task<TResult>> onCompletion,


This is probably better named onNext

I can understand why you'd say that and you'll notice that I got side-tracked in d973670 by the observer pattern here. Later in da262bc, I reverted as allowing harmless race conditions meant I couldn't guarantee the pattern's semantics. I find onNext will mislead in the same direction once again. Tasks complete so completion here is in that sense rather than completion of the loop. I felt that the task as a parameter of the function would be sufficient to say what's completing in case there's any confusion. I guess that's not how you read it. I would prefer onTaskCompletion than onNext if you still feel strong about it.

OK, I see why you don't want onNext. onTaskCompletion would be a better name. The main issue I think is that onCompletion is too easy to confuse with completing the entire task instead of completing each iteration. But your proposed name works for me.

fsateler · 2018-05-29T19:18:14Z

MoreLinq/Experimental/Await.cs

-                            .ConfigureAwait(continueOnCapturedContext: false);
-
-                    if (completedTask == cancellationTaskSource.Task)
+                    if (semaphore != null)


I think a wrapper class would make the code simpler:

class SemaphoreSlimWrapper { private readonly SemaphoreSlim semaphore; public SemaphoreSlimWrapper(int? maxConcurrency) { semaphore = maxConcurrency is int count ? new SemaphoreSlim(count, count) : null; } public Task WaitAsync() => semaphore?.WaitAsync() ?? Task.CompletedTask; public void Release() => semaphore?.Release(); }

And then this if can be avoided.

Let me know what you make of it now with e956b46.

fsateler · 2018-05-29T19:23:26Z

MoreLinq/Experimental/Await.cs

+                        {
+                            await semaphore.WaitAsync(cancellationToken);
+                        }
+                        catch (OperationCanceledException e) when (e.CancellationToken == cancellationToken)


This catch block could be moved back up into the caller, and below make change the cancellation token check to cancellationToken.ThrowIfCancellationRequested(); . This should make this function simple, while not making the caller too complicated as it is already catching exceptions.

The catch in the caller is for exceptional issues in StartAsync, like invocation of starter throwing (task fails to start). The returns are normal conditions or exits that are accounted for.

fsateler · 2018-05-29T19:32:13Z

MoreLinq/Experimental/Await.cs

                        CancellationToken.None,
                        TaskCreationOptions.DenyChildAttach,
                        scheduler);

+                    // Remainde here is the main loop that waits for and


s/Remainde/Remainder/

atifaziz · 2018-05-29T20:18:00Z

Unfortunately we don't have tests for this operator... so we can't check this hasn't introduced regressions...

Aye, thus the experimental status. I just haven't got round to thinking about good tests although #439 is somewhat of a sprint/start at that.

This unties its initialization with ConcurrencyGate.

atifaziz · 2018-05-30T05:33:42Z

@fsateler Thanks for your review. We good to merge this?

atifaziz · 2018-06-03T09:40:30Z

@fsateler I'd like to keep the momentum towards releasing 3.0. I feel guilty merging this since you started a review but haven't approved the changes you requested, which I hope have addressed with follow-up commits and comments. If you're busy then just give me a sign that I should go ahead. I want to publish a release candidate and then let it simmer for a week before considering it golden. That should still leave some time to iron out any kinks should you spot any.

Also bear in mind that this is an experimental feature and while we do want to do our very best to get it right, I wouldn't want a minor issue like the choice of exception thrown (AggregateException embedded in an Exception versus not) under very critical circumstance to hold back the release. We can fork that into a separate issue to review if we want to give it some more thought. I don't think anyone will ever be catching that case to apply any sort of remedy except log and abort the request/program. The important thing is that we trip the main loop so a program doesn't enter a zombie state.

fsateler · 2018-06-03T14:39:56Z

Yes, go ahead. I think my concerns are addressed

atifaziz · 2018-06-04T06:11:31Z

Yes, go ahead.

Thanks, mark approved then?

fsateler

Looks Good

atifaziz added 20 commits May 28, 2018 15:04

Use continuations to report tasks to support slow sources

1ba1162

End notice unnecessary

a30df54

Refactor to observable

d973670

Move general helpers down

86c94f9

To-do notes about exceptional cases

4a4e08d

Especially if we fail to notify under low memory conditions!

Add arg validation to StartAsync

3111a90

Don't allocate semaphore if unbounded concurrency

688c2ee

Fold lazy start semantics into StartAsync

616241b

This removes the need for Lazy allocations

Consolidate pending completion actions

4514060

Revert "Refactor to observable"

da262bc

This reverts commit d973670.

Don't treat cancellation as completion; return early

03419d3

Filter token when catching OperationCanceledException

91f4896

Add back (missing) end notice break

98ddd41

Crititcal error handling

cf71a25

Minor reivew/clean-up of formatting and names

af05bd6

Move notification & error handling in main iterator

5f6a23b

Remove reundant cancellation for simplicity

5c759e6

Move EDI capture into critical section

69d5341

Capture original error with error notification failure

8dbdcbe

Lots of comments

57212f2

atifaziz requested a review from fsateler May 29, 2018 13:58

fsateler requested changes May 29, 2018

View reviewed changes

atifaziz added 6 commits May 29, 2018 22:19

Fix typo (s/Remainde/Remainder/)

bda7187

Model a concurrency gate around the semaphore

e956b46

Re-format conditional expression

ec6e365

Move CompletedTask singleton into own class

467ba7c

This unties its initialization with ConcurrencyGate.

Fix ConcurrencyGate.EnterAsync to handle cancellation

4a5cfd3

Rename task completion parameter for clarity

30478c0

fsateler approved these changes Jun 4, 2018

View reviewed changes

atifaziz merged commit 201dbb3 into morelinq:master Jun 4, 2018

atifaziz deleted the await-with-slow-source branch June 4, 2018 13:04

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Fix AwaitCompletion to yield results during source iteration #505

Fix AwaitCompletion to yield results during source iteration #505

atifaziz commented May 29, 2018 •

edited

Loading

atifaziz commented May 29, 2018

fsateler left a comment

fsateler May 29, 2018

atifaziz May 29, 2018

fsateler May 29, 2018

atifaziz May 29, 2018

fsateler May 29, 2018

atifaziz May 29, 2018

fsateler May 29, 2018

atifaziz May 29, 2018

fsateler May 29, 2018

atifaziz May 29, 2018 •

edited

Loading

fsateler May 29, 2018

fsateler May 29, 2018

atifaziz May 29, 2018 •

edited

Loading

fsateler May 29, 2018

atifaziz May 29, 2018

fsateler May 29, 2018

atifaziz commented May 29, 2018 •

edited

Loading

atifaziz commented May 30, 2018

atifaziz commented Jun 3, 2018

fsateler commented Jun 3, 2018

atifaziz commented Jun 4, 2018

fsateler left a comment

Fix AwaitCompletion to yield results during source iteration #505

Fix AwaitCompletion to yield results during source iteration #505

Conversation

atifaziz commented May 29, 2018 • edited Loading

atifaziz commented May 29, 2018

fsateler left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

atifaziz May 29, 2018 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

atifaziz May 29, 2018 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

atifaziz commented May 29, 2018 • edited Loading

atifaziz commented May 30, 2018

atifaziz commented Jun 3, 2018

fsateler commented Jun 3, 2018

atifaziz commented Jun 4, 2018

fsateler left a comment

Choose a reason for hiding this comment

atifaziz commented May 29, 2018 •

edited

Loading

atifaziz May 29, 2018 •

edited

Loading

atifaziz May 29, 2018 •

edited

Loading

atifaziz commented May 29, 2018 •

edited

Loading