Rework subscription in Producer such that all Sinks set the upstream disposable by themselves. #575

danielcweber · 2018-06-06T12:43:40Z

Sinks will no longer be passed an IDisposable on creation. Previously, the passed IDisposable would (through a SingleAssignmentDisposable) represent the upstream subscription (or whatever Producer.Run would return). Now, every Sink is responsible for setting the upstream by itself.

akarnokd

LGTM. There is a change to the csproj file which I'd think is due to VS eagerly adding stuff because you touched the generator files.

akarnokd · 2018-06-06T13:32:59Z

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


            stack = new Stack<IEnumerator<IObservable<TSource>>>();
            stack.Push(current);

            Drain();

-            return new RecursiveSinkDisposable(this);
+            SetUpstream(new RecursiveSinkDisposable(this));


Probably this allocation can now be eliminated by overriding Dispose(bool) and inlining that DisposeAll method there. Btw there are some operators that do not have an upstream so for them, this is not really needed anymore.

Not so trivial, the Sink would have to be protected against disposal before Run has run. Can do it in a follow up.

akarnokd · 2018-06-06T13:35:18Z

Rx.NET/Source/src/System.Reactive/Linq/Observable/Buffer.cs

                {
                    _list = new List<TSource>();

                    var d = parent._scheduler.SchedulePeriodic(parent._timeSpan, Tick);
                    var s = parent._source.SubscribeSafe(this);

-                    return StableCompositeDisposable.Create(d, s);
+                    SetUpstream(StableCompositeDisposable.Create(d, s));


With the new structure, we can eliminate most StableCompositeDisposable creation now.

Sounds good, in a follow-up PR then.

akarnokd · 2018-06-06T13:38:03Z

Rx.NET/Source/src/System.Reactive/Linq/Observable/CombineLatest.Generated.tt

@@ -98,7 +98,7 @@ for (var j = 1; j <= i; j++)
 }
 #>

-                return StableCompositeDisposable.Create(subscriptions);
+                Set(StableCompositeDisposable.Create(subscriptions));


Did you mean SetUpstream?

akarnokd · 2018-06-06T13:47:50Z

Rx.NET/Source/src/System.Reactive/Linq/Observable/Timeout.cs

                {
-
                    CreateTimer(0L);

                    Disposable.SetSingle(ref _mainDisposable, source.SubscribeSafe(this));


_mainDisposable could be _upstream now so this can be just SetUpstream and no need for a separate field.

Hmm...but _mainDisposable must be disposed without disposing the entire sink (L85).

Yeah, right, forgot you only added SetUpstream, and the field is still inaccessible from subclasses. Maybe worth considering so that a field can be saved if there's already one in the base Sink.

The upstream could have SerialDisposable-semantics, that way it should be possible to save the _otherDisposable field as well. I would not like to expose otherwise private fields.

SetUpstream would then call SetSerial.

Combining these woudl require different atomics thant SetSerial. I think SetUpstream can stay as SetSingle but _upstream could become protected so that other Disposable.TryX methods can be applied on it.

Why wouldn't it work for Timeout? It just replaces the main-upstream by the other-upstream. No combining.

The return of SubscribeSafe is racing with the rest of the operator; a similar bug was fixed in #560.

You're sure you don't refer to _timerDisposable? Maybe I'll work it into a subsequent PR we can then talk about, the idea seems nice and I think it can work. If it has races, you could show me what I'm missing.

akarnokd · 2018-06-06T13:50:53Z

Rx.NET/Source/src/System.Reactive/Linq/Observable/Using.cs

@@ -42,10 +42,12 @@ public IDisposable Run(Using<TSource, TResource> parent)
                }
                catch (Exception exception)
                {
-                    return StableCompositeDisposable.Create(Observable.Throw<TSource>(exception).SubscribeSafe(this), disposable);
+                    SetUpstream(StableCompositeDisposable.Create(Observable.Throw<TSource>(exception).SubscribeSafe(this), disposable));


This is not the fault of this PR but why create a sequence just to signal OnError(exception) practically immediately?

Maybe in earlier code there was a scheduler involved.

akarnokd · 2018-06-06T13:52:05Z

Rx.NET/Source/src/System.Reactive/Linq/Observable/Zip.Generated.tt

@@ -110,7 +110,7 @@ for (var j = 1; j <= i; j++)
 #>
                });

-                return StableCompositeDisposable.Create(disposables);
+                Set(StableCompositeDisposable.Create(disposables));


Did you mean SetUpstream?

akarnokd · 2018-06-06T13:52:54Z

Rx.NET/Source/src/System.Reactive/System.Reactive.csproj

@@ -56,5 +56,8 @@
    <None Update="Linq\Observable\Zip.Generated.tt" Generator="TextTemplatingFileGenerator" LastGenOutput="Zip.Generated.cs" />
    <Compile Update="Linq\Observable\Zip.Generated.cs" DesignTime="True" AutoGen="True" DependentUpon="Zip.Generated.tt" />   
  </ItemGroup>
+  <ItemGroup>
+    <Service Include="{508349b6-6b84-4df5-91f0-309beebad82d}" />


Will be reverted.

…disposable by themselves. This will save up to two allocations per subscription.

akarnokd approved these changes Jun 6, 2018

View reviewed changes

Rework subscription in Producer such that all Sinks set the upstream …

d0e9d67

…disposable by themselves. This will save up to two allocations per subscription.

danielcweber merged commit 09bf2c8 into dotnet:master Jun 6, 2018

danielcweber deleted the SinkOverhaul branch June 6, 2018 19:36

akarnokd mentioned this pull request Jun 18, 2018

4.1.0 release preparations #607

Closed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Rework subscription in Producer such that all Sinks set the upstream disposable by themselves. #575

Rework subscription in Producer such that all Sinks set the upstream disposable by themselves. #575

danielcweber commented Jun 6, 2018

akarnokd left a comment

akarnokd Jun 6, 2018

danielcweber Jun 6, 2018

akarnokd Jun 6, 2018

danielcweber Jun 6, 2018

akarnokd Jun 6, 2018

danielcweber Jun 6, 2018

akarnokd Jun 6, 2018

danielcweber Jun 6, 2018 •

edited

Loading

akarnokd Jun 6, 2018

danielcweber Jun 6, 2018

danielcweber Jun 6, 2018

akarnokd Jun 6, 2018

danielcweber Jun 6, 2018

akarnokd Jun 6, 2018

danielcweber Jun 6, 2018

akarnokd Jun 6, 2018

danielcweber Jun 6, 2018

akarnokd Jun 6, 2018

danielcweber Jun 6, 2018

akarnokd Jun 6, 2018

danielcweber Jun 6, 2018

Rework subscription in Producer such that all Sinks set the upstream disposable by themselves. #575

Rework subscription in Producer such that all Sinks set the upstream disposable by themselves. #575

Conversation

danielcweber commented Jun 6, 2018

akarnokd 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

danielcweber Jun 6, 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

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

danielcweber Jun 6, 2018 •

edited

Loading