Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
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
10 changes: 10 additions & 0 deletions src/ReactiveUI.Primitives.Async.Core/Operators/ParityHelpers.cs
Original file line number Diff line number Diff line change
Expand Up @@ -354,6 +354,10 @@ public IObservableAsync<T> GetMin(params IObservableAsync<T>[] sources)
var allSources = new IObservableAsync<T>[sources.Length + 1];
allSources[0] = source;
sources.CopyTo(allSources, 1);
#if NET11_0_OR_GREATER
// The coordinator supplies its reusable array without copying it.
return new SyncLatestEnumerableSignal<T, T>(allSources, static values => ((ReadOnlySpan<T>)(T[])values).Min());
#else
return new SyncLatestEnumerableSignal<T, T>(allSources, static values =>
{
var min = values[0];
Expand All @@ -367,6 +371,7 @@ public IObservableAsync<T> GetMin(params IObservableAsync<T>[] sources)

return min;
});
#endif
}

/// <summary>Returns the maximum of the latest values from the supplied source sequences.</summary>
Expand All @@ -380,6 +385,10 @@ public IObservableAsync<T> GetMax(params IObservableAsync<T>[] sources)
var allSources = new IObservableAsync<T>[sources.Length + 1];
allSources[0] = source;
sources.CopyTo(allSources, 1);
#if NET11_0_OR_GREATER
// The coordinator supplies its reusable array without copying it.
return new SyncLatestEnumerableSignal<T, T>(allSources, static values => ((ReadOnlySpan<T>)(T[])values).Max());
#else
return new SyncLatestEnumerableSignal<T, T>(allSources, static values =>
{
var max = values[0];
Expand All @@ -393,6 +402,7 @@ public IObservableAsync<T> GetMax(params IObservableAsync<T>[] sources)

return max;
});
#endif
}
}

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -6,7 +6,6 @@
<TargetFrameworks>$(LibraryTargetFrameworks)</TargetFrameworks>
<ImplicitUsings>enable</ImplicitUsings>
<Nullable>enable</Nullable>
<LangVersion>preview</LangVersion>
<!-- Keep the default namespace so split-out engine types share the Async root. -->
<RootNamespace>ReactiveUI.Primitives.Async</RootNamespace>
<PackageDescription>Type-agnostic core of ReactiveUI.Primitives.Async, shared by the lean and Reactive leaves.</PackageDescription>
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -6,7 +6,6 @@
<TargetFrameworks>$(LibraryTargetFrameworks)</TargetFrameworks>
<ImplicitUsings>enable</ImplicitUsings>
<Nullable>enable</Nullable>
<LangVersion>preview</LangVersion>
<PackageDescription>ReactiveUI.Primitives.Async recompiled against System.Reactive's Unit and IScheduler for seamless interop with System.Reactive consumers.</PackageDescription>
</PropertyGroup>

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -2,7 +2,6 @@

<PropertyGroup>
<TargetFrameworks>$(LibraryTargetFrameworks)</TargetFrameworks>
<LangVersion>preview</LangVersion>
</PropertyGroup>

<ItemGroup Condition="'$(TargetFramework)' == 'net462'">
Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1,68 @@
// Copyright (c) 2019-2026 ReactiveUI Association Incorporated. All rights reserved.
// ReactiveUI Association Incorporated licenses this file to you under the MIT license.
// See the LICENSE file in the project root for full license information.

using System.Diagnostics.CodeAnalysis;
using BenchmarkDotNet.Attributes;

namespace ReactiveUI.Primitives.Benchmarks;

/// <summary>Compares the latest-value reduction used by asynchronous minimum operators.</summary>
[MemoryDiagnoser]
[System.Diagnostics.DebuggerDisplay("MinMaxReductionBenchmarks: Sources = {SourceCount}")]
public class MinMaxReductionBenchmarks
{
/// <summary>A stride coprime to every source count produces a repeatable permutation.</summary>
private const int ValueStride = 17;

/// <summary>Offsets the permutation so its first value need not be the minimum.</summary>
private const int ValueOffset = 42;

/// <summary>The coordinator exposes its reusable array through this interface.</summary>
[SuppressMessage(
"Performance",
"CA1859:Use concrete types when possible for improved performance",
Justification = "The baseline must measure the coordinator's IReadOnlyList<int> dispatch; an array field would change the measured code.")]
private IReadOnlyList<int> _values = [];

/// <summary>Gets or sets the number of latest values in the snapshot.</summary>
[Params(3, 8, 32, 128, 1024)]
public int SourceCount { get; set; }

/// <summary>Creates one snapshot outside the measured operations.</summary>
[GlobalSetup]
public void Setup()
{
var values = new int[SourceCount];
for (var i = 0; i < values.Length; i++)
{
values[i] = ((i * ValueStride) + ValueOffset) % values.Length;
}

_values = values;
}

/// <summary>Reduces the snapshot through the existing scalar comparison loop.</summary>
/// <returns>The minimum value.</returns>
[Benchmark(Baseline = true)]
public int ScalarMinimum()
{
var minimum = _values[0];
for (var i = 1; i < _values.Count; i++)
{
if (Comparer<int>.Default.Compare(_values[i], minimum) < 0)
{
minimum = _values[i];
}
}

return minimum;
}

#if NET11_0_OR_GREATER
/// <summary>Reduces the same snapshot with the .NET 11 span implementation.</summary>
/// <returns>The minimum value.</returns>
[Benchmark]
public int SpanMinimum() => ((ReadOnlySpan<int>)(int[])_values).Min();
#endif
}
Original file line number Diff line number Diff line change
Expand Up @@ -14,6 +14,102 @@ public sealed class SignalAsyncExtensionsTests
/// <summary>The second distinct value in an ordered sequence.</summary>
private const int SecondValue = 2;

/// <summary>Minimum and maximum reductions include values beyond vector boundaries.</summary>
/// <param name="sourceCount">The number of latest values to reduce.</param>
/// <returns>The test operation.</returns>
[Test]
[Arguments(3)]
[Arguments(8)]
[Arguments(17)]
[Arguments(65)]
public async Task GetMinMax_IntegerInputs_FindExtremesAcrossVectorBoundaries(int sourceCount)
{
var sources = new IObservableAsync<int>[sourceCount - 1];
for (var i = 0; i < sources.Length; i++)
{
sources[i] = SignalAsync.Return(i);
}

sources[^1] = SignalAsync.Return(int.MaxValue);
var first = SignalAsync.Return(int.MinValue);

await Assert.That(await first.GetMin(sources).FirstAsync()).IsEqualTo(int.MinValue);
await Assert.That(await first.GetMax(sources).FirstAsync()).IsEqualTo(int.MaxValue);
}

/// <summary>Each update reduces the current values after all sources have emitted.</summary>
/// <param name="cancellationToken">Cancels the test's signal notifications.</param>
/// <returns>The test operation.</returns>
[Test]
public async Task GetMinMax_Updates_UseCurrentValues(CancellationToken cancellationToken)
{
const int InitialMinimum = 3;
const int MiddleValue = 6;
const int InitialMaximum = 9;
const int UpdatedMaximum = 12;
await using var first = Signal.Create<int>();
await using var second = Signal.Create<int>();
await using var third = Signal.Create<int>();
List<int> minima = [];
List<int> maxima = [];
await using var minimum = await first.Values.GetMin(second.Values, third.Values).SubscribeAsync(minima.Add, cancellationToken);
await using var maximum = await first.Values.GetMax(second.Values, third.Values).SubscribeAsync(maxima.Add, cancellationToken);

await first.OnNextAsync(InitialMinimum, cancellationToken);
await second.OnNextAsync(MiddleValue, cancellationToken);
await Assert.That(minima).IsEmpty();
await Assert.That(maxima).IsEmpty();

await third.OnNextAsync(InitialMaximum, cancellationToken);
await first.OnNextAsync(UpdatedMaximum, cancellationToken);
await third.OnNextAsync(1, cancellationToken);

await Assert.That(minima).IsCollectionEqualTo([InitialMinimum, MiddleValue, 1]);
await Assert.That(maxima).IsCollectionEqualTo([InitialMaximum, UpdatedMaximum, UpdatedMaximum]);
}

/// <summary>Floating-point reductions retain the default comparer's NaN ordering.</summary>
/// <returns>The test operation.</returns>
[Test]
public async Task GetMinMax_FloatingPointInputs_PreserveNaNOrdering()
{
var source = SignalAsync.Return(double.NaN);
IObservableAsync<double>[] others = [SignalAsync.Return((double)SecondValue), SignalAsync.Return(1D)];

await Assert.That(double.IsNaN(await source.GetMin(others).FirstAsync())).IsTrue();
await Assert.That(await source.GetMax(others).FirstAsync()).IsEqualTo((double)SecondValue);
}

/// <summary>Equal floating-point values retain the first value's zero sign.</summary>
/// <returns>The test operation.</returns>
[Test]
public async Task GetMinMax_EqualZeroValues_PreserveFirstSign()
{
const double NegativeZero = -0D;
var source = SignalAsync.Return(NegativeZero);
var second = SignalAsync.Return(0D);

var minimum = await source.GetMin(second).FirstAsync();
var maximum = await source.GetMax(second).FirstAsync();

await Assert.That(BitConverter.DoubleToInt64Bits(minimum)).IsEqualTo(long.MinValue);
await Assert.That(BitConverter.DoubleToInt64Bits(maximum)).IsEqualTo(long.MinValue);
}

/// <summary>Non-numeric values continue to use their default comparison contract.</summary>
/// <returns>The test operation.</returns>
[Test]
public async Task GetMinMax_DateInputs_UseDefaultComparer()
{
var earliest = DateTime.UnixEpoch;
var latest = earliest.AddDays(1);
var source = SignalAsync.Return(latest);
var second = SignalAsync.Return(earliest);

await Assert.That(await source.GetMin(second).FirstAsync()).IsEqualTo(earliest);
await Assert.That(await source.GetMax(second).FirstAsync()).IsEqualTo(latest);
}

/// <summary>Chaining an enumerable preserves source order.</summary>
/// <returns>The test operation.</returns>
[Test]
Expand Down
Loading