Async.Ix: Some fixes and proposed changes to operators #843

akarnokd · 2018-11-08T15:27:12Z

I did a more detailed run-through of the code and these are my proposed changes/fixes. See additional comments among the code changes below.

akarnokd · 2018-11-08T15:27:34Z

Ix.NET/Source/System.Interactive.Async.Tests/System/Linq/Operators/Amb.cs

+
+namespace Tests
+{
+    public class Amb : AsyncEnumerableExTests


There was no unit test for Amb at all.

akarnokd · 2018-11-08T15:28:21Z

Ix.NET/Source/System.Interactive.Async.Tests/System/Linq/Operators/Amb.cs

+
+    internal static class AsyncTestEx
+    {
+        internal static IAsyncEnumerable<TResult> Select<TSource, TResult>(this IAsyncEnumerable<TSource> source, Func<TSource, CancellationToken, Task<TResult>> mapper)


A Select overload exposing a cancellation token is missing from the standard API. I've added one here so I can test Amb propagating cancellation properly.

akarnokd · 2018-11-08T15:29:34Z

Ix.NET/Source/System.Interactive.Async.Tests/System/Linq/Operators/Using.cs

@@ -82,7 +82,22 @@ public void Using3()

            Assert.Equal(0, i);

-            AssertThrows<Exception>(() => xs.GetAsyncEnumerator(), ex_ => ex_ == ex);
+            var enumerator = xs.GetAsyncEnumerator();


DisposeAsync failures are somewhat peculiar as the chain may not follow through upon an intermediate crash.

akarnokd · 2018-11-08T15:31:37Z

Ix.NET/Source/System.Interactive.Async/AsyncIterator.cs

            {
-                enumerator.DisposeAsync(); // REVIEW: fire-and-forget?
-                throw;
+                // GetAsyncEnumerator should not throw but rather return a failing enumerator


Simply return a special IAsyncEnumerator that fails after the source DisposeAsync is called with the original exception.

What is the rationale here? Why can't a call to GetAsyncEnumerator throw?

My undestanding is that GetAsyncEnumerator on an arbitrary IAsyncEnumerable should be trusted, otherwise every such call should be defended with try-catches and also make sure the resulting cleanup does everything correctly and at the right time. Besides, the main point here is to avoid "fire-and-forget".

I'm not sure if that's a fair assumption, but it's a good question @stephentoub, any thoughts on what the API contract is for GetAsyncEnumerator?

It shouldn't be any different from IEnumerable<T>.GetEnumerator, and to my knowledge there's no guidance about GetEnumerator not throwing, though @MadsTorgersen can correct me if he knows otherwise. That said, it's super rare for GetEnumerator to throw.

Regardless, I agree with avoiding the fire-and-forget DisposeAsync call... having such a fire-and-forget is problematic because the caller doesn't know whether it can immediately use the resources referenced by the thing that should have been disposed.

Ix.NET/Source/System.Interactive.Async/AsyncIterator.cs

Ix.NET/Source/System.Linq.Async/System/Linq/Operators/Take.cs

akarnokd · 2018-11-08T15:45:09Z

Ix.NET/Source/System.Linq.Async/System/Linq/Operators/ToAsyncEnumerable.cs

+        /// all source items until they are requested via MoveNextAsync.
+        /// </summary>
+        /// <typeparam name="TSource">The element type of the source and result.</typeparam>
+        private sealed class ObservableToAsyncEnumerable<TSource> : IAsyncEnumerable<TSource>


Complete rewrite. Uses atomics instead of locks.

akarnokd · 2018-11-08T15:45:58Z

Ix.NET/Source/System.Linq.Async/System/Linq/Operators/ToAsyncEnumerable.cs

+
+                private void DisposeSource()
+                {
+                    var old = Interlocked.Exchange(ref _disposable, this);


Deferred disposing, the usual Rx stuff.

akarnokd · 2018-11-08T15:46:32Z

Ix.NET/Source/System.Linq.Async/System/Linq/Operators/ToObservable.cs

+                return enumerator;
+            }
+
+            private sealed class ToObservableEnumerator : IDisposable


Complete rewrite. Now features all the dance around possible reentrant MoveNextAsyncs and the need for an anytime dispose to wait out a MoveNextAsync.

akarnokd · 2018-11-08T15:48:59Z

Ix.NET/Source/System.Linq.Async/System/Threading/Tasks/TaskExt.cs

+        /// <summary>
+        ///  A constant TaskCompletionSource already completed with true.
+        /// </summary>
+        public static readonly TaskCompletionSource<bool> ResumeTrue;


This is part of the atomic resumption mechanism used in ToAsyncEnumerable and in many more coordinating operators for the future.

clairernovotny · 2018-11-08T15:59:28Z

@bartdesmet any thoughts?

clairernovotny · 2018-11-08T16:04:58Z

Ix.NET/Source/System.Interactive.Async/System/Linq/Operators/Amb.cs


-            public AmbAsyncIterator(IAsyncEnumerable<TSource> first, IAsyncEnumerable<TSource> second)
+            public AmbEnumerable(IAsyncEnumerable<TSource> source1, IAsyncEnumerable<TSource> source2)


Nit, but should the parameter names still be first, and second?

Can be renamed, but this is internal stuff so doesn't matter.

clairernovotny · 2018-11-08T16:07:00Z

Ix.NET/Source/System.Interactive.Async/System/Linq/Operators/Amb.cs

+                        _enumerator2 = _source2.GetAsyncEnumerator(_token2.Token);
+
+#pragma warning disable CS4014 // ContinueWith used
+                        _enumerator1.MoveNextAsync()


Can an await with a local function be used instead of a ContinueWith?

The problem with await is that only one thing can be awaited in a logical flow, so which one to await? The first or the second source? By awaiting one before the other, we give a guaranteed pereference to one of them thus makes Amb obsolete. Also await Task.WhenAny is no good either because if any of the source fails, await crashes and without an explicit winner, the operator can break.

@davidfowl what would you suggest here?

Also await Task.WhenAny is no good either because if any of the source fails, await crashes and without an explicit winner, the operator can break.

What do you mean by await crashes? Task.WhenAny will return a Task<Task<bool>> here, so awaiting it will give you back the task that completed, at which point you can choose to do with it whatever you want.

Right, I overlooked the definition of Task.WhenAny.

Updated to use Task.WhenAny, but I can't see it scale to the N-ary variant as the DisposeAsync() has to wait for all non-winners.

Ix.NET/Source/System.Interactive.Async/System/Linq/Operators/Amb.cs

clairernovotny · 2018-11-09T21:24:38Z

@bartdesmet Given that this is mostly your work in this branch, I'll defer merging until you're ready (or just merge whenever you're ready).

akarnokd · 2018-11-18T18:01:23Z

There were a lot of recent commits conflicting with this one. I'll have to do a re-review anyway so I'm closing this.

Async.Ix: Some fixes and proposed changes to operators

a104e7b

akarnokd commented Nov 8, 2018

View reviewed changes

clairernovotny requested a review from bartdesmet November 8, 2018 15:59

clairernovotny reviewed Nov 8, 2018

View reviewed changes

akarnokd closed this Nov 9, 2018

akarnokd reopened this Nov 9, 2018

Use Task.WhenAny

3b59d7d

akarnokd mentioned this pull request Nov 9, 2018

AsyncIx: More reviews/fixes/proposed changes #845

Merged

bartdesmet reviewed Nov 9, 2018

View reviewed changes

Ix.NET/Source/System.Interactive.Async/System/Linq/Operators/Amb.cs Outdated Show resolved Hide resolved

akarnokd added 3 commits November 9, 2018 19:55

Rethrow, .ConfigureAwait(false)

e6180d1

Restore Take(0)

95b5e30

Take restoration, take 2

98fda12

bartdesmet and others added 2 commits November 9, 2018 21:45

Merge branch 'IxAsyncCSharp8' into IxAsync8Fixes

ee83c55

Adjust ToObservable

8edf81f

akarnokd closed this Nov 18, 2018

akarnokd deleted the IxAsync8Fixes branch November 18, 2018 18:01

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Async.Ix: Some fixes and proposed changes to operators #843

Async.Ix: Some fixes and proposed changes to operators #843

akarnokd commented Nov 8, 2018

akarnokd Nov 8, 2018

akarnokd Nov 8, 2018

akarnokd Nov 8, 2018

akarnokd Nov 8, 2018

clairernovotny Nov 8, 2018

akarnokd Nov 8, 2018

clairernovotny Nov 8, 2018

stephentoub Nov 8, 2018

akarnokd Nov 8, 2018

akarnokd Nov 8, 2018

akarnokd Nov 8, 2018

akarnokd Nov 8, 2018

clairernovotny commented Nov 8, 2018

clairernovotny Nov 8, 2018

akarnokd Nov 8, 2018

clairernovotny Nov 8, 2018

akarnokd Nov 8, 2018

clairernovotny Nov 8, 2018

stephentoub Nov 8, 2018 •

edited

akarnokd Nov 8, 2018

akarnokd Nov 9, 2018

clairernovotny commented Nov 9, 2018

akarnokd commented Nov 18, 2018


		public AmbAsyncIterator(IAsyncEnumerable<TSource> first, IAsyncEnumerable<TSource> second)
		public AmbEnumerable(IAsyncEnumerable<TSource> source1, IAsyncEnumerable<TSource> source2)

Async.Ix: Some fixes and proposed changes to operators #843

Async.Ix: Some fixes and proposed changes to operators #843

Conversation

akarnokd commented Nov 8, 2018

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

Choose a reason for hiding this comment

Choose a reason for hiding this comment

clairernovotny commented Nov 8, 2018

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

stephentoub Nov 8, 2018 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

clairernovotny commented Nov 9, 2018

akarnokd commented Nov 18, 2018

stephentoub Nov 8, 2018 •

edited