Add dedicated class based implementations for Append and Prepend. #600

quinmars · 2018-06-17T17:41:31Z

Here is my attempt to optimize the Append and Prepend operators. Please review it carefully, I'm not sure if I got the handling of the IDisposables right.

akarnokd · 2018-06-17T17:46:49Z

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

+
+        protected override void Run(_ sink) => sink.Run(_source);
+
+        internal sealed class _ : IdentitySink<TSource>


It would be great if we abandoned these underscore class names.

I agree with you, but that name is used in most - if not all - operators I have seen.

Definitely desirable, maybe in a later PR.

akarnokd · 2018-06-17T17:48:12Z

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

+            private static IDisposable ForwardValue(IScheduler scheduler, _ sink)
+            {
+                sink.ForwardOnNext(sink._value);
+                return scheduler.Schedule(sink, ForwardOnCompleted);


Is it necessary to schedule the OnCompleted signal separately?

Well, that was something I tried to ask in #567. The conclusion was to do it like ToObservable/StartWith do it. And ToObservable does schedule the OnCompleted separately.

I don't think we need the separate schedule here.

akarnokd · 2018-06-17T17:49:25Z

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

+            private static IDisposable ForwardOnCompleted(IScheduler _unused, _ sink)
+            {
+                sink.ForwardOnCompleted();
+                return BooleanDisposable.True;


This should not really be returned by operators and is supposed to be confined to Disposable methods. Return Disposable.Empty instead.

akarnokd · 2018-06-17T17:50:27Z

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

+            protected override void Dispose(bool val)
+            {
+                base.Dispose(val);
+                Disposable.TryDispose(ref _schedulerDisposable);


Usually we do this as if (disposing) { dispose custom resources } base.Dispose(disposing).

akarnokd · 2018-06-17T18:49:54Z

One thing to consider is that Append invites the following pattern:

IObservable<int> someSource = Observable.Return(0);
for (int i = 1; i < 1000; i++) {
   someSource = someSource.Append(i);
}

someSource.Subscribe(Console.WriteLine);

Does this still work with your modifications?

quinmars · 2018-06-17T19:04:27Z

It is tested for three consequtive Appends in Append_Many_Scheduler , but I haven't added any optimizations as it is done in CoreFX for Enumerable.Append, i.e., merging consequtive operators to one, instead of using the naive O(n²) implementation. I can't even say if it becomes quadratic for observables or not.

akarnokd · 2018-06-17T19:19:02Z

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

+
+        internal sealed class _ : IdentitySink<TSource>
+        {
+            private readonly IObservable<TSource> _source;


This field is unused.

akarnokd · 2018-06-17T19:25:53Z

I'm worried about the stacktrace depth I get:

akarnokd · 2018-06-17T20:07:31Z

TailRecursiveSink in Concat flattens out the deep structure so it is not a problem there. When this PR is merged, I can add the changes so that a chain of append is coalesced into one Concat(source, items).

quinmars · 2018-06-18T19:08:00Z

We could go the same route, as the corefx team has done it for the enumerable counter part [1], to optimize consecutive Appends and Prepends. That'd be:

Collect all appends and prepends in an AppendPrependObservable class
For the observing part: a sink that consist roughly of a IdentitySink that has a ToObservable.Sink starting part and a ToObservable.Sink ending part.

If we do that or anything else in that direction, we should propably skip the extra scheduling of the OnCompleted, at least between two Appends/Prepends.

I can do that, but that'd take some time.

[1] https://github.com/dotnet/corefx/blob/master/src/System.Linq/src/System/Linq/AppendPrepend.cs

akarnokd · 2018-06-18T19:12:31Z

It is certainly more interesting and challenging doing this type of "operator fusion". I leave it up to you if you want to tackle it.

quinmars · 2018-06-18T19:29:31Z

I'd give it a try than, if that's ok for @danielcweber as well.

danielcweber · 2018-06-18T20:46:33Z

Sure, go ahead.

…ization

danielcweber · 2018-06-21T06:47:42Z

Looks good. Is there still something missing to make the WIP go away?

quinmars · 2018-06-21T21:30:46Z

Not really. I first thought I put the optimizations, I'm working on, into this PR. But I think it is a good point to merge. I'll put the next commits in a new PR.

quinmars · 2018-06-21T23:50:32Z

@akarnokd now the stacktrace should be much smaller, hopefully. 😄

quinmars · 2018-06-21T23:53:28Z

@danielcweber, I changed my mind and put the changes into this PR. Let me know if I can help you during the review.

danielcweber · 2018-06-22T08:32:50Z

Unfortunately, I get 4 failing tests with NullReferenceExceptions locally:

Append_Many, Append_Many_Scheduler, Append_Many_Take, Append_Many_Take_Scheduler.

Result StackTrace:	
bei System.Reactive.Linq.ObservableImpl.AppendPrepend.AppendPrependMultiple`1._.LoopRec(State state, Action`1 recurse)
   bei System.Reactive.Concurrency.Scheduler.InvokeRec1State`1.InvokeFirst(TState state)
   bei System.Reactive.Concurrency.Scheduler.InvokeRec1[TState](IScheduler scheduler, ValueTuple`2 tuple)
   bei System.Reactive.Concurrency.Scheduler.<>c__48`1.<Schedule>b__48_0(IScheduler s, ValueTuple`2 p)
   bei System.Reactive.Concurrency.ImmediateScheduler.Schedule[TState](TState state, Func`3 action)
   bei System.Reactive.Concurrency.Scheduler.Schedule[TState](IScheduler scheduler, TState state, Action`2 action)
   bei System.Reactive.Linq.ObservableImpl.AppendPrepend.AppendPrependMultiple`1._.Schedule(TSource[] array, Action`1 continueWith)
   bei System.Reactive.Linq.ObservableImpl.AppendPrepend.AppendPrependMultiple`1._.Run()
   bei System.Reactive.Linq.ObservableImpl.AppendPrepend.AppendPrependMultiple`1.Run(_ sink)
   bei System.Reactive.Producer`2.<>c.<SubscribeRaw>b__1_0(ValueTuple`2 tuple)
   bei System.Reactive.Concurrency.Scheduler.<>c__74`1.<ScheduleAction>b__74_0(IScheduler _, ValueTuple`2 tuple)
   bei System.Reactive.Concurrency.CurrentThreadScheduler.Schedule[TState](TState state, TimeSpan dueTime, Func`3 action)
   bei System.Reactive.Concurrency.LocalScheduler.Schedule[TState](TState state, Func`3 action)
   bei System.Reactive.Concurrency.Scheduler.ScheduleAction[TState](IScheduler scheduler, TState state, Action`1 action)
   bei System.Reactive.Producer`2.SubscribeRaw(IObserver`1 observer, Boolean enableSafeguard)
   bei System.Reactive.Producer`2.Subscribe(IObserver`1 observer)
   bei Microsoft.Reactive.Testing.TestScheduler.<>c__DisplayClass4_0`1.<Start>b__1(IScheduler scheduler, Object state)
   bei System.Reactive.Concurrency.VirtualTimeScheduler`2.<>c__DisplayClass4_0`1.<ScheduleAbsolute>b__0(IScheduler scheduler, TState state1)
   bei System.Reactive.Concurrency.ScheduledItem`2.InvokeCore()
   bei System.Reactive.Concurrency.ScheduledItem`1.Invoke()
   bei System.Reactive.Concurrency.VirtualTimeSchedulerBase`2.Start()
   bei Microsoft.Reactive.Testing.TestScheduler.Start[T](Func`1 create, Int64 created, Int64 subscribed, Int64 disposed)
   bei Microsoft.Reactive.Testing.TestScheduler.Start[T](Func`1 create)
   bei ReactiveTests.Tests.AppendTest.Append_Many()
Result Message:	System.NullReferenceException : Der Objektverweis wurde nicht auf eine Objektinstanz festgelegt.

danielcweber · 2018-06-22T08:39:58Z

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

+                    }
+                }
+
+                private sealed class State


Please replace this class by the use of C#7 ValueTuples to have less allocations.

quinmars · 2018-06-22T18:45:47Z

Sorry, I was pretty sure that I did had run all tests the day before. I'll fix that.

quinmars · 2018-06-22T23:07:41Z

Man, in that commit was nearly everything wrong that could go wrong, sorry for that. I hope it's now ok. At least all tests pass. I changed the class to be a struct like you did in #644.

danielcweber · 2018-06-25T09:33:04Z

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

+                    if (_prepends != null)
+                    {
+                        var disposable = Schedule(_prepends, s => s.SetUpstream(s._source.SubscribeSafe(s)));
+                        Disposable.TrySetSerial(ref _schedulerDisposable, disposable);


That could indeed lead to trouble, in case the subscription (one line above) will complete synchronously, you will effetively be overwriting the scheduling from OnCompletd, thus cancelling that work. TrySetSingle should be fine.

I changed it to TrySetSingle. Should I than use two different backing fields for the two schedulers? Currently, they share one field variable.

No, the one field should be fine, it's just the TrySetSingle should only be on the first scheduling. Now you're using it on both, which is not what I meant. In fact, if you change both to SetSingle instead of TrySetSingle, you will probably notice one of the tests throwing. The second TrySetSingle will fail in the case that we have prepends and appends. So we're currently leaking the second scheduling.

The second time you schedule something, you want it to overwrite the previous scheduling, which is why you want to use TrySetSerial. The first time you schedule, however, you want to either set the disposable for the very first time or not set it at all. Because, if the scheduled work executes synchronously, the second scheduling will take place (successfully through TrySetSerial), and by the time Schedule returns, we definitely don't want to overwrite that second scheduling, which is why we want to use TrySetSingle.

akarnokd · 2018-06-26T09:32:43Z

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

+                    _scheduler = parent.Scheduler;
+
+                    if (parent._prepends != null)
+                        _prepends = parent._prepends.ToArray();


Could we avoid the creation of this array by simply walking the node structure?

Yes, we could. I haven't done it because we then have to duplicate the two loops. It's a tradeoff between code reuse and one allocation.

akarnokd · 2018-06-26T09:34:09Z

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

+                        _prepends = parent._prepends.ToArray();
+
+                    if (parent._appends != null)
+                        _appends = parent._appends.ToReverseArray();


This array creation could be avoided as well if the appends where in a doubly-linked node chain. In that case, it would be enough to traverse the chain once to find the very first value and then walking backwards.

Beware, by doubly linking the nodes, we can't pass around the original observable once we appended/prepended to it, or even append/prepend something to the original observable. The instances we get from appying operators should be, in a way, immutable. We would get subtle side effects here, if I got your idea correctly.

I was thinking about getting the immutable collections back in (#222), then it's just a stack and a queue.

The instances we get from appying operators should be, in a way, immutable.

Indeed, the _appends should be an immutable list. I was somehow thinking it being a copy-on-write list (i.e., ImmutableList) which requires way more allocation and copying. My mistake.

If we manage to mitigate the performance degradation that has been reported in #222, immutable collections could be helpful in a lot of places. It's used internally in the Roslyn compiler since it's highly concurrent.

The current implementation has a time complexity of O(n) and needs n + 1 allocations. I doubt the immutable collections offer a datatype that can beat that.

Adding a node to the lists as they are currently implemented is even in O(1). Yes, even immutable collections won't beat that. Maybe they'll find their way into the project eventually since they would be useful in the Subjects, but until that, this solution is perfectly fine.

Add dedicated class based implementations for Append and Prepend.

d4c92b7

akarnokd reviewed Jun 17, 2018

View reviewed changes

Merge remote-tracking branch 'origin/master' into appendprepend-optim…

8429619

…ization

quinmars changed the title ~~Add dedicated class based implementations for Append and Prepend.~~ [WIP] Add dedicated class based implementations for Append and Prepend. Jun 18, 2018

No extra scheduling for OnCompleted and ForwardSource

7e96c82

danielcweber approved these changes Jun 19, 2018

View reviewed changes

akarnokd approved these changes Jun 19, 2018

View reviewed changes

Add some tests to test the mixed usage of Prepend and Append

94deb3d

quinmars changed the title ~~[WIP] Add dedicated class based implementations for Append and Prepend.~~ Add dedicated class based implementations for Append and Prepend. Jun 21, 2018

quinmars added 3 commits June 21, 2018 23:30

Merge branch 'master' into appendprepend-optimization

b6267e5

Merge Append and Prepend class

43fd632

Add AppendPrependMultiple to optimize chaining of Append and Prepend

6264e6a

danielcweber reviewed Jun 22, 2018

View reviewed changes

quinmars added 2 commits June 23, 2018 00:28

Fix many problems with AppendPrepend

5616b8d

Use a struct instead of an class

6b09be3

Merge branch 'master' into appendprepend-optimization

9b63906

danielcweber approved these changes Jun 25, 2018

View reviewed changes

danielcweber reviewed Jun 25, 2018

View reviewed changes

Use TrySetSingle instead of TrySetSerial

ac32b6b

akarnokd reviewed Jun 26, 2018

View reviewed changes

quinmars and others added 2 commits June 26, 2018 22:29

Use TrySetSerial for the append scheduler disposable

ec1893b

Merge branch 'master' into appendprepend-optimization

f2d0576

danielcweber merged commit 738e3c9 into dotnet:master Jun 26, 2018

akarnokd mentioned this pull request Jul 24, 2018

4.1.0 release preparations #607

Closed


		protected override void Run(_ sink) => sink.Run(_source);

		internal sealed class _ : IdentitySink<TSource>

Add dedicated class based implementations for Append and Prepend. #600

Add dedicated class based implementations for Append and Prepend. #600

Conversation

quinmars commented Jun 17, 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

akarnokd commented Jun 17, 2018

quinmars commented Jun 17, 2018

Choose a reason for hiding this comment

akarnokd commented Jun 17, 2018

akarnokd commented Jun 17, 2018

quinmars commented Jun 18, 2018

akarnokd commented Jun 18, 2018

quinmars commented Jun 18, 2018

danielcweber commented Jun 18, 2018

danielcweber commented Jun 21, 2018

quinmars commented Jun 21, 2018

quinmars commented Jun 21, 2018

quinmars commented Jun 21, 2018

danielcweber commented Jun 22, 2018 • edited Loading

Choose a reason for hiding this comment

quinmars commented Jun 22, 2018

quinmars commented Jun 22, 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

danielcweber Jun 26, 2018 • edited Loading

Choose a reason for hiding this comment

danielcweber commented Jun 22, 2018 •

edited

Loading

danielcweber Jun 26, 2018 •

edited

Loading