Review producers and sinks #493

danielcweber · 2018-05-05T13:07:04Z

The main objective of this PR is to strengthen the concept of the Sink class. Before, it would wrap, yet
still internally expose an observer. Inheriting classes would access it directly and the pattern of disposing the Sink after calling OnCompleted/OnError on the wrapped observer would be on them and repeat throughout the code. This commit encapsulates the observer and exposes Forward*-methods, taking care of disposing at the right places. The result is a lot of saved duplicated code. Moreover, we find that almost all classes inheriting from Sink also implement an IObserver of some kind by themselves, so we establish a base class Sink<TSource, TTarget>. The concept of a Sink is now more obvious, i.e. it serves as the logic between the source-stream and the emitted target stream. At last, we introduce IdentitySink which will just relay events, again, this will save a lot of duplicated code.

akarnokd · 2018-05-05T14:04:16Z

Rx.NET/Source/src/System.Reactive/Internal/Sink.cs

    /// <remarks>Implementations of sinks are responsible to enforce the message grammar on the associated observer. Upon sending a terminal message, a pairing Dispose call should be made to trigger cancellation of related resources and to mute the outgoing observer.</remarks>
-    internal abstract class Sink<TSource> : IDisposable
+    internal abstract class Sink<TTarget, TSource> : IObserver<TSource>, IDisposable


I'd be more comfortable with the arguments ordered TSource, TTarget.

Yes, I was aimimg for that, it's only that way I could easily do a find/replace of the pattern Sink<...>, IObserver<...>. ReSharper may do that with one click maybe. I'll have a look.

Has been done.

akarnokd · 2018-05-05T14:06:13Z

Rx.NET/Source/src/System.Reactive/Internal/Sink.cs

+            ForwardOnCompleted();
+        }
+
+        protected virtual void Dispose(bool disposing)


I'm not that familiar with C# but saw this pattern mentioned several times. Is this Dispose(bool) called by the system on finalization?

It's not since Sink doesn't have a finalizer. It's just a pattern that IMHO should be followed if we want to have a clean, state-of-the-art codebase.

akarnokd · 2018-05-05T14:07:39Z

Rx.NET/Source/src/System.Reactive/Internal/Sink.cs


-        private sealed class _ : IObserver<TSource>
+        private sealed class _ : IObserver<TTarget>


I haven't looked around much. Why is this wrapper needed? I'd assume so that Sink.Dispose() is not accessible to whomever receives the _.

It's needed by the GetForwarder method and that's needed in one place only. I'd rather get rid of it but it's ok for now.

akarnokd · 2018-05-05T14:10:20Z

Rx.NET/Source/src/System.Reactive/Internal/TailRecursiveSink.cs

@@ -188,28 +187,21 @@ private bool TryGetEnumerator(IEnumerable<IObservable<TSource>> sources, out IEn
                // enumerating to find the next observable sequence. Therefore,
                // we feed those errors directly to the observer.
                //
-                _observer.OnError(exception);
-                base.Dispose();
+                ForwardOnError(exception);


I'd think the allocation L157 is not really necessary. It could be replaced by simply:

_subscription.Disposable = next.SubscribeSafe(this);

Not in the scope of this PR, at least not for now. I am sure there is a lot to gain in allocation efficiency though.

akarnokd · 2018-05-05T14:13:06Z

Rx.NET/Source/src/System.Reactive/Internal/TailRecursiveSink.cs

@@ -8,7 +8,7 @@

 namespace System.Reactive
 {
-    internal abstract class TailRecursiveSink<TSource> : Sink<TSource>, IObserver<TSource>
+    internal abstract class TailRecursiveSink<TSource> : Sink<TSource>


This class is odd to my eyes. It is supposed to implement a trampolining mechanism where only one inner source should be running at a time but don't want to get into an infinite recursion when trying to continue with the next one.

Given that it operates out of a synchronous/blocking IEnumerable, I can't see why the stacks are even necessary. Also it enables some transformations on top of those which I don't know why they are provided.

I will need to review that in depth.

akarnokd · 2018-05-08T07:37:02Z

Rx.NET/Source/src/System.Reactive/Disposables/AssignmentDisposable.cs

+        /// </summary>
+        protected IDisposable Get(ref IDisposable fieldRef)
+        {
+            var current = fieldRef;


Does ref preserve the volatileness of fieldRef? I'd call Volatile.Read(ref fieldRef) to be sure.

It's just a copy of what was there in SingleAssignmentDisposable. The old code produced warnings in this regard as well.

https://stackoverflow.com/a/425150/61158

akarnokd · 2018-05-08T07:38:07Z

Rx.NET/Source/src/System.Reactive/Disposables/AssignmentDisposable.cs

+            // We use a sentinel value to indicate we've been disposed. This sentinel never leaks
+            // to the outside world (see the Disposable property getter), so no-one can ever assign
+            // this value to us manually.
+            return fieldRef == BooleanDisposable.True;


Volatile.Read(ref fieldRef)?

You see both patterns throughout the code, sometimes with Volatile.Read, sometimes without. Won't the volatile field just do?

https://stackoverflow.com/a/425150/61158

The AssignmentDisposable shall be removed then. You may have to fetch and force-update your branch in some minutes, I'll do a rebase.

No need for removal, just use Volatile.Read as I suggested.

I might as well just copy-paste SingleAssignmentDisposable. Copy and paste may be justified at this level.

Yes, in Reactive4.NET I didn't specify fields as volatile and had a helper static class work with refs.

Awesome, that base helper static class does exactly that. Could you please review 3fc620d then.

Yes, much better.

akarnokd · 2018-05-08T07:40:29Z

Rx.NET/Source/src/System.Reactive/Disposables/AssignmentDisposable.cs

+
+        protected void Dispose(ref IDisposable fieldRef)
+        {
+            Interlocked.Exchange(ref fieldRef, BooleanDisposable.True)?.Dispose();


Since Rx is calling Dispose all over the place and even possibly multiple times, I'd bias this so that if fieldRef is already BooleanDisposable.True, don't exchange:

if (Volatile.Read(ref fieldRef) != BooleanDisposable.True) { Interlocked.Exchange(ref fieldRef, BooleanDisposable.True)?.Dispose(); }

Could be done.

akarnokd · 2018-05-08T16:52:05Z

The most non-trivial thing when writing an operator externally is this: Producer.cs#L52.

If you don't want to include Rx.NET, there is no way to know a Subscribe should trampoline. The Producers are internal so you can't externally depend on this feature either.

danielcweber · 2018-05-08T18:34:15Z

We'll get there, I'm sure.

akarnokd · 2018-05-08T18:51:24Z

Rx.NET/Source/src/System.Reactive/Internal/Sink.cs

+
+        protected virtual void Dispose(bool disposing)
+        {
+            _observer = NopObserver<TTarget>.Instance;


You don't really need full barrier volatile write here due to the full barrier of Exchange. A Volatile.Write(ref _observer, NopObserver<TTarget>.Instance) should suffice.

What about the other accesses to _observer? Should be Volatile.Read then, right?

_observer is declared volatile so you get it for free. I tend to use Volatile.Read to clearly indicate the barriers in the code.

akarnokd · 2018-05-08T18:52:42Z

Rx.NET/Source/src/System.Reactive/Internal/Sink.cs


-        public Sink(IObserver<TSource> observer, IDisposable cancel)
+        protected Sink(IObserver<TTarget> observer, IDisposable cancel)


Swap the assignment so that _observer = ... releases the plain _cancel field before the constructor returns.

what's the advantage ?

Correct publication of the IDisposable reference received with respect to the memory model.

What's the incorrect behaviour that could happen? (the memory model has always puzzled me)

Potential early publication of the referenced IDisposable before its state has become visible to other threads. Plain reference writes can be tricky in concurrent settings.

akarnokd · 2018-05-08T19:03:02Z

Rx.NET/Source/src/System.Reactive/Internal/Producer.cs

            if (CurrentThreadScheduler.IsScheduleRequired)
            {
-                var state = new State { sink = sink, inner = subscription.Inner };
+                var state = new State { sink = sink, inner = runDisposable };

                CurrentThreadScheduler.Instance.Schedule(state, Run);


I think this BasicProducer.Run could be replaced with a static delegate so that only one state object is allocated and not any delegate (I assume assigning a method to a delegate type actually instantiates some objects in C# right?):

private struct State { public BasicProducer<TSource> parent; public SingleAssignmentDisposable subscription; public IObserver<TSource> observer; } static readonly Func<IScheduler, State, IDisposable> RunRun = (_, x) => { x.subscription.Disposable = x.parent.Run(x.observer); return Disposable.Empty; }; //... if (CurrentThreadScheduler.IsScheduleRequired) { var state = new State { parent = this, subscription = subscription, observer = observer }; CurrentThreadScheduler.Instance.Schedule(state, RunRun); }

Yes. I'm actually puzzled that making Run static won't make the compiler cache the delegate...

Inlining Run as anonymous lambda that's not dependent on closure will do the trick. Interesting, I had thought that delegate caching will work on static methods as well.

I can implement all your suggestions, you may as well file a pull request to get the credit for the commit.

I'm working on TailRecursiveSink at the moment and I have these already changed in my version.

…still internally expose an observer. Inheriting classes would access it directly and the pattern of disposing the Sink after calling OnCompleted/OnError on the wrapped observer would be on them and repeat throughout the code. This commit encapsulates the observer and exposes Forward*-methods, taking care of disposing at the right places. The result is a lot of saved duplicated code. Moreover, we find that almost all classes inheriting from Sink also implement an IObserver of some kind by themselves, so we establish a base class Sink<TSource, TTarget>. The concept of a Sink is now more obvious, i.e. it serves as the logic between the source-stream and the emitted target stream. At last, we introduce IdentitySink which will just relay events, again, this will save a lot of duplicated code.

akarnokd reviewed May 5, 2018

View reviewed changes

akarnokd mentioned this pull request May 7, 2018

4.x: Example of a new internal structure for writing operators #494

Closed

akarnokd reviewed May 8, 2018

View reviewed changes

danielcweber changed the title ~~[WIP] Review producers and sinks~~ Review producers and sinks May 15, 2018

danielcweber mentioned this pull request May 17, 2018

4.x: Improve Amb #512

Merged

akarnokd approved these changes May 28, 2018

View reviewed changes

danielcweber merged commit f2ea212 into dotnet:master May 28, 2018

danielcweber deleted the ReviewProducersAndSinks branch May 28, 2018 16:27

akarnokd mentioned this pull request Jun 18, 2018

4.1.0 release preparations #607

Closed


		private sealed class _ : IObserver<TSource>
		private sealed class _ : IObserver<TTarget>


		public Sink(IObserver<TSource> observer, IDisposable cancel)
		protected Sink(IObserver<TTarget> observer, IDisposable cancel)

Review producers and sinks #493

Review producers and sinks #493

Uh oh!

Conversation

danielcweber commented May 5, 2018 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

akarnokd commented May 8, 2018 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

danielcweber commented May 8, 2018

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

danielcweber May 8, 2018 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

danielcweber commented May 5, 2018 •

edited

Loading

akarnokd commented May 8, 2018 •

edited

Loading

danielcweber May 8, 2018 •

edited

Loading