Skip to content

4.x: Improve blocking First & Last operators #590

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 7 commits into from
Jun 14, 2018
Merged

4.x: Improve blocking First & Last operators #590

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
bd709f2
4.x: Improve blocking First & Last operators
akarnokd Jun 14, 2018
b0a7357
Combine into an abstract base
akarnokd Jun 14, 2018
9c1a4ec
Push to base class.
akarnokd Jun 14, 2018
ba7a3bc
Dispose due to CountdownEvent
akarnokd Jun 14, 2018
625ecfd
Merge branch 'master' into FirstLastOptimize
danielcweber Jun 14, 2018
d142d0e
Remove unnecessary "this.", trigger VSTS
akarnokd Jun 14, 2018
71b48eb
Merge branch 'master' into FirstLastOptimize
danielcweber Jun 14, 2018
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
115 changes: 115 additions & 0 deletions Rx.NET/Source/src/System.Reactive/Linq/Observable/FirstLastBlocking.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,115 @@
// Licensed to the .NET Foundation under one or more agreements.
// The .NET Foundation licenses this file to you under the Apache 2.0 License.
// See the LICENSE file in the project root for more information.

using System;
using System.Collections.Generic;
using System.Reactive.Disposables;
using System.Text;
using System.Threading;

namespace System.Reactive.Linq.ObservableImpl
{

internal abstract class BaseBlocking<T> : CountdownEvent, IObserver<T>
Copy link
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

CountdownEvent vs. ManualResetEvent ?

Copy link
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

ManualResetEvent cannot be extended, CountdownEvent can, saving on an allocation.

{
protected IDisposable _upstream;

internal T _value;
internal bool _hasValue;
internal Exception _error;

int once;
Copy link
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

See above, ManualResetEvent would save this field.

Copy link
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Allocation vs an atomic operation tradeoff. If Signal wouldn't throw by default, we'd not need this.


internal BaseBlocking() : base(1) { }

internal void SetUpstream(IDisposable d)
{
Disposable.SetSingle(ref _upstream, d);
}

protected void Unblock()
{
if (Interlocked.CompareExchange(ref once, 1, 0) == 0)
{
Signal();
}
}

public abstract void OnCompleted();
public virtual void OnError(Exception error)
{
_value = default;
_error = error;
Unblock();
}
public abstract void OnNext(T value);

public new void Dispose()
{
base.Dispose();
if (!Disposable.GetIsDisposed(ref _upstream))
{
Disposable.TryDispose(ref _upstream);
}
}
}

internal sealed class FirstBlocking<T> : BaseBlocking<T>
{
internal FirstBlocking() : base() { }

public override void OnCompleted()
{
Unblock();
if (!Disposable.GetIsDisposed(ref _upstream))
{
Disposable.TryDispose(ref _upstream);
}
}

public override void OnError(Exception error)
{
base.OnError(error);
if (!Disposable.GetIsDisposed(ref _upstream))
{
Disposable.TryDispose(ref _upstream);
}
}

public override void OnNext(T value)
{
if (!_hasValue)
{
_value = value;
_hasValue = true;
Disposable.TryDispose(ref _upstream);
Unblock();
}
}
}

internal sealed class LastBlocking<T> : BaseBlocking<T>
{
internal LastBlocking() : base() { }

public override void OnCompleted()
{
Unblock();
Disposable.TryDispose(ref _upstream);
}

public override void OnError(Exception error)
Copy link
Collaborator

@danielcweber danielcweber Jun 14, 2018 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Save for the IsDisposed-check, the overriding code is almost identical. Enough reason to push it to the base-class ?

Copy link
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Last doesn't need that dispose check so why have it there. First needs it so why penalize everybody else?

Copy link
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

At least push some of the code into the base class.

{
base.OnError(error);
Disposable.TryDispose(ref _upstream);
}

public override void OnNext(T value)
{
_value = value;
_hasValue = true;
}

}
}
95 changes: 31 additions & 64 deletions Rx.NET/Source/src/System.Reactive/Linq/QueryLanguage.Blocking.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -67,45 +67,26 @@ public virtual TSource FirstOrDefault<TSource>(IObservable<TSource> source, Func

private static TSource FirstOrDefaultInternal<TSource>(IObservable<TSource> source, bool throwOnEmpty)
{
var value = default(TSource);
var seenValue = false;
var ex = default(Exception);

using (var evt = new WaitAndSetOnce())
using (var consumer = new FirstBlocking<TSource>())
{
//
// [OK] Use of unsafe Subscribe: fine to throw to our caller, behavior indistinguishable from going through the sink.
//
using (source.Subscribe/*Unsafe*/(new AnonymousObserver<TSource>(
v =>
{
if (!seenValue)
{
value = v;
}
seenValue = true;
evt.Set();
},
e =>
{
ex = e;
evt.Set();
},
() =>
{
evt.Set();
})))
using (var d = source.Subscribe(consumer))
{
evt.WaitOne();
}
}
consumer.SetUpstream(d);

ex.ThrowIfNotNull();
if (consumer.CurrentCount != 0)
{
consumer.Wait();
}
}

if (throwOnEmpty && !seenValue)
throw new InvalidOperationException(Strings_Linq.NO_ELEMENTS);
consumer._error.ThrowIfNotNull();

return value;
if (throwOnEmpty && !consumer._hasValue)
{
throw new InvalidOperationException(Strings_Linq.NO_ELEMENTS);
}
return consumer._value;
}
}

#endregion
Expand Down Expand Up @@ -182,41 +163,27 @@ public virtual TSource LastOrDefault<TSource>(IObservable<TSource> source, Func<

private static TSource LastOrDefaultInternal<TSource>(IObservable<TSource> source, bool throwOnEmpty)
{
var value = default(TSource);
var seenValue = false;
var ex = default(Exception);

using (var evt = new WaitAndSetOnce())
using (var consumer = new LastBlocking<TSource>())
{
//
// [OK] Use of unsafe Subscribe: fine to throw to our caller, behavior indistinguishable from going through the sink.
//
using (source.Subscribe/*Unsafe*/(new AnonymousObserver<TSource>(
v =>
{
seenValue = true;
value = v;
},
e =>
{
ex = e;
evt.Set();
},
() =>
{
evt.Set();
})))

using (var d = source.Subscribe(consumer))
{
evt.WaitOne();
}
}
consumer.SetUpstream(d);

ex.ThrowIfNotNull();
if (consumer.CurrentCount != 0)
{
consumer.Wait();
}
}

if (throwOnEmpty && !seenValue)
throw new InvalidOperationException(Strings_Linq.NO_ELEMENTS);
consumer._error.ThrowIfNotNull();

return value;
if (throwOnEmpty && !consumer._hasValue)
{
throw new InvalidOperationException(Strings_Linq.NO_ELEMENTS);
}
return consumer._value;
}
}

#endregion
Expand Down