diff --git a/CLAUDE.md b/CLAUDE.md index 9bc8b794..469ee712 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -474,12 +474,16 @@ a view's unit test. The MAUI invoker catches it. It treats the object as having 1. It uses the first registered invoker that claims the target. A generated binding also passes a fallback: an invoker to use when no registered one claims the target. With no invoker at all, the source comes back unchanged. -2. A notification runs inline when nothing is queued and `CheckAccess` passes. -3. Any other notification is queued. One callback empties the queue. It runs on `BindingSchedulers.MainThread` - when that is set, and through `Post` otherwise. +2. A notification runs inline when nothing is waiting and `CheckAccess` passes. +3. Any other notification waits. One drain delivers what waits. It runs on `BindingSchedulers.MainThread` when that + is set, and through `Post` otherwise. -A notification that arrives behind queued ones waits its turn, even on the owning thread. The view sees values in -the order the source produced them. +Only the latest value waits. A newer value replaces it, even one raised on the owning thread while a drain runs. +Completion and errors wait beside the value and are delivered after it. + +Keeping every value breaks two-way bindings. Writing a view raises the view's own change at once. If a newer value +is still waiting, that echo writes the older value back to the view model. The write raises another change, and +the two sides bounce forever. `ViewWriteSchedulingRuntimeTests` covers this for `Bind` and `BindTwoWay`. `MainThread` only carries writes from another thread to a claimed object. It never sees an on-thread write. It never sees a write to an unclaimed object. @@ -497,6 +501,8 @@ never sees a write to an unclaimed object. The generator declares a class when its platform type resolves in the compilation. It does not look at call sites. A call site can only name an invoker for a type that resolves. So every reference has a declaration. +An `Unsafe` binding only has the registered invokers. It routes writes only when the platform module is registered. + There is no WinUI invoker. No runtime package registers one. A generated one would route writes that the `Unsafe` twin does not. diff --git a/README.md b/README.md index 1a274168..8bf1ef31 100644 --- a/README.md +++ b/README.md @@ -616,10 +616,11 @@ right one. The binding asks on every write. -- A write on the owning thread runs straight away. Set a property on the UI thread, and the control has the new - value on the next line. -- A write from another thread waits for the owning thread. -- Writes keep their order. +- A write on the owning thread runs straight away when no earlier write is waiting. Set a property on the UI + thread, and the control has the new value on the next line. +- A write from another thread waits for the owning thread. A write on the owning thread also waits while an + earlier write is waiting. +- Only the latest value waits. A newer change replaces the waiting one, so a burst of changes becomes one write. Some objects have no owning thread. The binding writes to them straight away. @@ -629,7 +630,7 @@ Some objects have no owning thread. The binding writes to them straight away. - Any object that is not a WPF, WinForms or MAUI object, such as a plain view model. Every binding API does this: `BindOneWay`, `BindTwoWay`, `OneWayBind`, `Bind`, `BindTo`, and `BindCommand` when -it binds a new command to the control. Each `Unsafe` twin does the same. +it binds a new command to the control. Each `Unsafe` twin does the same through the registered invokers. ### Invokers @@ -640,6 +641,9 @@ An invoker you register is asked first. A generated binding knows its target's type when it compiles. For a WPF, WinForms or MAUI target, it carries that platform's invoker. So it routes writes even when the platform module is not registered. +An `Unsafe` binding only finds its target's type while the app runs. It uses the registered invokers alone. Register +the platform module when you use `Unsafe` bindings. + ### Choosing the thread yourself > [!TIP] @@ -751,7 +755,12 @@ message loop later. Where ReactiveUI does move a write, the order is the same. A write on the owning thread runs straight away. A write from another thread goes through the main-thread scheduler. Set `BindingSchedulers.MainThread` to -ReactiveUI's main-thread scheduler to match it exactly. +ReactiveUI's main-thread scheduler to use the same scheduler. + +A burst of changes from another thread is handled differently. ReactiveUI's one-way bindings and `BindTo` write +every value, on the thread that raised it. Its two-way `Bind` queues one signal per change and reads the current +value when each signal runs. Here every binding writes only the latest value, once. A binding's change stream +skips the values in between. ### A binding made through a type parameter is not generated diff --git a/src/ReactiveUI.Binding.Reactive/PublicAPI/net10.0/PublicAPI.txt b/src/ReactiveUI.Binding.Reactive/PublicAPI/net10.0/PublicAPI.txt index 2236e881..8d916186 100644 --- a/src/ReactiveUI.Binding.Reactive/PublicAPI/net10.0/PublicAPI.txt +++ b/src/ReactiveUI.Binding.Reactive/PublicAPI/net10.0/PublicAPI.txt @@ -376,7 +376,7 @@ namespace ReactiveUI.Binding.Reactive public override int GetAffinityForObjects() { } public override bool TryConvert(int from, object? conversionHint, out string? result) { } } - [System.Diagnostics.DebuggerDisplay("Handlers = {_handlers.Count}")] + [System.Diagnostics.DebuggerDisplay("Handlers = {_handlers.Length}")] public class Interaction : ReactiveUI.Binding.Reactive.IInteraction { public Interaction() { } diff --git a/src/ReactiveUI.Binding.Reactive/PublicAPI/net11.0/PublicAPI.txt b/src/ReactiveUI.Binding.Reactive/PublicAPI/net11.0/PublicAPI.txt index 2236e881..8d916186 100644 --- a/src/ReactiveUI.Binding.Reactive/PublicAPI/net11.0/PublicAPI.txt +++ b/src/ReactiveUI.Binding.Reactive/PublicAPI/net11.0/PublicAPI.txt @@ -376,7 +376,7 @@ namespace ReactiveUI.Binding.Reactive public override int GetAffinityForObjects() { } public override bool TryConvert(int from, object? conversionHint, out string? result) { } } - [System.Diagnostics.DebuggerDisplay("Handlers = {_handlers.Count}")] + [System.Diagnostics.DebuggerDisplay("Handlers = {_handlers.Length}")] public class Interaction : ReactiveUI.Binding.Reactive.IInteraction { public Interaction() { } diff --git a/src/ReactiveUI.Binding.Reactive/PublicAPI/net462/PublicAPI.txt b/src/ReactiveUI.Binding.Reactive/PublicAPI/net462/PublicAPI.txt index e9301bb6..d881b992 100644 --- a/src/ReactiveUI.Binding.Reactive/PublicAPI/net462/PublicAPI.txt +++ b/src/ReactiveUI.Binding.Reactive/PublicAPI/net462/PublicAPI.txt @@ -354,7 +354,7 @@ namespace ReactiveUI.Binding.Reactive public override int GetAffinityForObjects() { } public override bool TryConvert(int from, object? conversionHint, out string? result) { } } - [System.Diagnostics.DebuggerDisplay("Handlers = {_handlers.Count}")] + [System.Diagnostics.DebuggerDisplay("Handlers = {_handlers.Length}")] public class Interaction : ReactiveUI.Binding.Reactive.IInteraction { public Interaction() { } diff --git a/src/ReactiveUI.Binding.Reactive/PublicAPI/net47/PublicAPI.txt b/src/ReactiveUI.Binding.Reactive/PublicAPI/net47/PublicAPI.txt index e9301bb6..d881b992 100644 --- a/src/ReactiveUI.Binding.Reactive/PublicAPI/net47/PublicAPI.txt +++ b/src/ReactiveUI.Binding.Reactive/PublicAPI/net47/PublicAPI.txt @@ -354,7 +354,7 @@ namespace ReactiveUI.Binding.Reactive public override int GetAffinityForObjects() { } public override bool TryConvert(int from, object? conversionHint, out string? result) { } } - [System.Diagnostics.DebuggerDisplay("Handlers = {_handlers.Count}")] + [System.Diagnostics.DebuggerDisplay("Handlers = {_handlers.Length}")] public class Interaction : ReactiveUI.Binding.Reactive.IInteraction { public Interaction() { } diff --git a/src/ReactiveUI.Binding.Reactive/PublicAPI/net471/PublicAPI.txt b/src/ReactiveUI.Binding.Reactive/PublicAPI/net471/PublicAPI.txt index e9301bb6..d881b992 100644 --- a/src/ReactiveUI.Binding.Reactive/PublicAPI/net471/PublicAPI.txt +++ b/src/ReactiveUI.Binding.Reactive/PublicAPI/net471/PublicAPI.txt @@ -354,7 +354,7 @@ namespace ReactiveUI.Binding.Reactive public override int GetAffinityForObjects() { } public override bool TryConvert(int from, object? conversionHint, out string? result) { } } - [System.Diagnostics.DebuggerDisplay("Handlers = {_handlers.Count}")] + [System.Diagnostics.DebuggerDisplay("Handlers = {_handlers.Length}")] public class Interaction : ReactiveUI.Binding.Reactive.IInteraction { public Interaction() { } diff --git a/src/ReactiveUI.Binding.Reactive/PublicAPI/net472/PublicAPI.txt b/src/ReactiveUI.Binding.Reactive/PublicAPI/net472/PublicAPI.txt index e9301bb6..d881b992 100644 --- a/src/ReactiveUI.Binding.Reactive/PublicAPI/net472/PublicAPI.txt +++ b/src/ReactiveUI.Binding.Reactive/PublicAPI/net472/PublicAPI.txt @@ -354,7 +354,7 @@ namespace ReactiveUI.Binding.Reactive public override int GetAffinityForObjects() { } public override bool TryConvert(int from, object? conversionHint, out string? result) { } } - [System.Diagnostics.DebuggerDisplay("Handlers = {_handlers.Count}")] + [System.Diagnostics.DebuggerDisplay("Handlers = {_handlers.Length}")] public class Interaction : ReactiveUI.Binding.Reactive.IInteraction { public Interaction() { } diff --git a/src/ReactiveUI.Binding.Reactive/PublicAPI/net48/PublicAPI.txt b/src/ReactiveUI.Binding.Reactive/PublicAPI/net48/PublicAPI.txt index e9301bb6..d881b992 100644 --- a/src/ReactiveUI.Binding.Reactive/PublicAPI/net48/PublicAPI.txt +++ b/src/ReactiveUI.Binding.Reactive/PublicAPI/net48/PublicAPI.txt @@ -354,7 +354,7 @@ namespace ReactiveUI.Binding.Reactive public override int GetAffinityForObjects() { } public override bool TryConvert(int from, object? conversionHint, out string? result) { } } - [System.Diagnostics.DebuggerDisplay("Handlers = {_handlers.Count}")] + [System.Diagnostics.DebuggerDisplay("Handlers = {_handlers.Length}")] public class Interaction : ReactiveUI.Binding.Reactive.IInteraction { public Interaction() { } diff --git a/src/ReactiveUI.Binding.Reactive/PublicAPI/net481/PublicAPI.txt b/src/ReactiveUI.Binding.Reactive/PublicAPI/net481/PublicAPI.txt index e9301bb6..d881b992 100644 --- a/src/ReactiveUI.Binding.Reactive/PublicAPI/net481/PublicAPI.txt +++ b/src/ReactiveUI.Binding.Reactive/PublicAPI/net481/PublicAPI.txt @@ -354,7 +354,7 @@ namespace ReactiveUI.Binding.Reactive public override int GetAffinityForObjects() { } public override bool TryConvert(int from, object? conversionHint, out string? result) { } } - [System.Diagnostics.DebuggerDisplay("Handlers = {_handlers.Count}")] + [System.Diagnostics.DebuggerDisplay("Handlers = {_handlers.Length}")] public class Interaction : ReactiveUI.Binding.Reactive.IInteraction { public Interaction() { } diff --git a/src/ReactiveUI.Binding.Reactive/PublicAPI/net8.0/PublicAPI.txt b/src/ReactiveUI.Binding.Reactive/PublicAPI/net8.0/PublicAPI.txt index 2236e881..8d916186 100644 --- a/src/ReactiveUI.Binding.Reactive/PublicAPI/net8.0/PublicAPI.txt +++ b/src/ReactiveUI.Binding.Reactive/PublicAPI/net8.0/PublicAPI.txt @@ -376,7 +376,7 @@ namespace ReactiveUI.Binding.Reactive public override int GetAffinityForObjects() { } public override bool TryConvert(int from, object? conversionHint, out string? result) { } } - [System.Diagnostics.DebuggerDisplay("Handlers = {_handlers.Count}")] + [System.Diagnostics.DebuggerDisplay("Handlers = {_handlers.Length}")] public class Interaction : ReactiveUI.Binding.Reactive.IInteraction { public Interaction() { } diff --git a/src/ReactiveUI.Binding.Reactive/PublicAPI/net9.0/PublicAPI.txt b/src/ReactiveUI.Binding.Reactive/PublicAPI/net9.0/PublicAPI.txt index 2236e881..8d916186 100644 --- a/src/ReactiveUI.Binding.Reactive/PublicAPI/net9.0/PublicAPI.txt +++ b/src/ReactiveUI.Binding.Reactive/PublicAPI/net9.0/PublicAPI.txt @@ -376,7 +376,7 @@ namespace ReactiveUI.Binding.Reactive public override int GetAffinityForObjects() { } public override bool TryConvert(int from, object? conversionHint, out string? result) { } } - [System.Diagnostics.DebuggerDisplay("Handlers = {_handlers.Count}")] + [System.Diagnostics.DebuggerDisplay("Handlers = {_handlers.Length}")] public class Interaction : ReactiveUI.Binding.Reactive.IInteraction { public Interaction() { } diff --git a/src/ReactiveUI.Binding.Shared/Bindings/BindingHooks.cs b/src/ReactiveUI.Binding.Shared/Bindings/BindingHooks.cs index be6c57f8..40f05ddd 100644 --- a/src/ReactiveUI.Binding.Shared/Bindings/BindingHooks.cs +++ b/src/ReactiveUI.Binding.Shared/Bindings/BindingHooks.cs @@ -29,6 +29,10 @@ namespace ReactiveUI.Binding; public static class BindingHooks { /// Guards while it is (re)resolved. + /// + /// Only the first resolve after a refresh takes it. Holding it while the locator is read stops a resolve + /// that races from publishing the set that refresh dropped. + /// private static readonly Lock Gate = new(); /// The resolved hooks, or null while none has been resolved yet. @@ -87,7 +91,7 @@ public static bool ShouldBind( [MethodImpl(MethodImplOptions.AggressiveInlining)] private static IPropertyBindingHook[] Resolve() { - var resolved = _hooks; + var resolved = Volatile.Read(ref _hooks); if (resolved is not null) { return resolved; diff --git a/src/ReactiveUI.Binding.Shared/Builder/RxBindingBuilder.cs b/src/ReactiveUI.Binding.Shared/Builder/RxBindingBuilder.cs index 17d4fd30..ec176f5a 100644 --- a/src/ReactiveUI.Binding.Shared/Builder/RxBindingBuilder.cs +++ b/src/ReactiveUI.Binding.Shared/Builder/RxBindingBuilder.cs @@ -20,11 +20,8 @@ namespace ReactiveUI.Binding.Builder; /// public static class RxBindingBuilder { - /// Synchronization gate for initialization and reset operations. - private static readonly Lock _resetLock = new(); - - /// Tracks whether ReactiveUI.Binding has been initialized (0 = not initialized, 1 = initialized). - private static int _hasBeenInitialized; // 0 = false, 1 = true + /// Whether ReactiveUI.Binding has been initialized: 0 until runs, 1 after. + private static int _hasBeenInitialized; /// Creates a new using the current Splat locator. /// A new builder instance. @@ -35,18 +32,17 @@ public static ReactiveUIBindingBuilder CreateReactiveUIBindingBuilder() => /// Thrown if BuildApp() has not been called. public static void EnsureInitialized() { - lock (_resetLock) + if (Volatile.Read(ref _hasBeenInitialized) != 0) { - if (_hasBeenInitialized == 0) - { - throw new InvalidOperationException( - "ReactiveUI.Binding has not been initialized. You must initialize using the builder pattern.\n\n" - + "Example:\n" - + "RxBindingBuilder.CreateReactiveUIBindingBuilder()\n" - + " .WithCoreServices()\n" - + " .BuildApp();"); - } + return; } + + throw new InvalidOperationException( + "ReactiveUI.Binding has not been initialized. You must initialize using the builder pattern.\n\n" + + "Example:\n" + + "RxBindingBuilder.CreateReactiveUIBindingBuilder()\n" + + " .WithCoreServices()\n" + + " .BuildApp();"); } /// Resets the initialization state for testing purposes only. @@ -55,20 +51,12 @@ public static void EnsureInitialized() /// internal static void ResetForTesting() { - lock (_resetLock) - { - AppBuilder.ResetBuilderStateForTests(); - AppLocator.SetLocator(new ModernDependencyResolver()); - _hasBeenInitialized = 0; - } + AppBuilder.ResetBuilderStateForTests(); + AppLocator.SetLocator(new ModernDependencyResolver()); + Volatile.Write(ref _hasBeenInitialized, 0); } /// Marks ReactiveUI.Binding as initialized. Called by . - internal static void MarkAsInitialized() - { - lock (_resetLock) - { - _hasBeenInitialized = 1; - } - } + [MethodImpl(MethodImplOptions.AggressiveInlining)] + internal static void MarkAsInitialized() => Volatile.Write(ref _hasBeenInitialized, 1); } diff --git a/src/ReactiveUI.Binding.Shared/Helpers/CopyOnWriteArray.cs b/src/ReactiveUI.Binding.Shared/Helpers/CopyOnWriteArray.cs new file mode 100644 index 00000000..2867c374 --- /dev/null +++ b/src/ReactiveUI.Binding.Shared/Helpers/CopyOnWriteArray.cs @@ -0,0 +1,57 @@ +// 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. + +#if REACTIVE_SHIM +namespace ReactiveUI.Binding.Reactive.Helpers; +#else +namespace ReactiveUI.Binding.Helpers; +#endif + +/// Changes an array by publishing a changed copy, so a reader walks a stable set without a lock. +internal static class CopyOnWriteArray +{ + /// Publishes a copy of the array with added at the end. + /// The element type. + /// The field holding the array. + /// The item to add. + internal static void Add(ref T[] location, T item) + { + T[] current; + T[] updated; + + do + { + current = Volatile.Read(ref location); + updated = new T[current.Length + 1]; + Array.Copy(current, updated, current.Length); + updated[current.Length] = item; + } + while (!ReferenceEquals(Interlocked.CompareExchange(ref location, updated, current), current)); + } + + /// Publishes a copy of the array without the first occurrence of . + /// The element type. + /// The field holding the array. + /// The item to remove. An item that is not there leaves the array alone. + internal static void Remove(ref T[] location, T item) + { + T[] current; + T[] updated; + + do + { + current = Volatile.Read(ref location); + var index = Array.IndexOf(current, item); + if (index < 0) + { + return; + } + + updated = new T[current.Length - 1]; + Array.Copy(current, updated, index); + Array.Copy(current, index + 1, updated, index, current.Length - index - 1); + } + while (!ReferenceEquals(Interlocked.CompareExchange(ref location, updated, current), current)); + } +} diff --git a/src/ReactiveUI.Binding.Shared/Interactions/Interaction.cs b/src/ReactiveUI.Binding.Shared/Interactions/Interaction.cs index fc82830c..de907742 100644 --- a/src/ReactiveUI.Binding.Shared/Interactions/Interaction.cs +++ b/src/ReactiveUI.Binding.Shared/Interactions/Interaction.cs @@ -29,14 +29,14 @@ namespace ReactiveUI.Binding; /// if no handler handles the interaction. /// /// -[DebuggerDisplay("Handlers = {_handlers.Count}")] +[DebuggerDisplay("Handlers = {_handlers.Length}")] public class Interaction : IInteraction { - /// The list of registered interaction handlers, invoked in reverse order during . - private readonly List, Task>> _handlers = []; - - /// Synchronization gate for thread-safe handler registration and removal. - private readonly Lock _sync = new(); + /// + /// The registered handlers, invoked in reverse order during . The array is replaced + /// rather than changed, so a question already being handled walks the set it started with. + /// + private Func, Task>[] _handlers = []; /// public IDisposable RegisterHandler(Action> handler) @@ -80,7 +80,7 @@ Task ContentHandler(IInteractionContext context) public virtual async Task Handle(TInput input) { var context = GenerateContext(input); - var handlers = GetHandlers(); + var handlers = Volatile.Read(ref _handlers); for (var i = handlers.Length - 1; i >= 0; i--) { @@ -96,13 +96,7 @@ public virtual async Task Handle(TInput input) /// Gets all registered handlers by order of registration. /// All registered handlers. - protected Func, Task>[] GetHandlers() - { - lock (_sync) - { - return [.. _handlers]; - } - } + protected Func, Task>[] GetHandlers() => [.. Volatile.Read(ref _handlers)]; /// Gets an interaction context which is used to provide information about the interaction. /// The input that is being passed in. @@ -110,25 +104,17 @@ protected Func, Task>[] GetHandlers() protected virtual IOutputContext GenerateContext(TInput input) => new InteractionContext(input); - /// Adds a handler to the internal handler list under the synchronization gate. + /// Adds a handler to the registered set. /// The handler to add. - private void AddHandler(Func, Task> handler) - { - lock (_sync) - { - _handlers.Add(handler); - } - } + [MethodImpl(MethodImplOptions.AggressiveInlining)] + private void AddHandler(Func, Task> handler) => + CopyOnWriteArray.Add(ref _handlers, handler); - /// Removes a handler from the internal handler list under the synchronization gate. + /// Removes the first registration of a handler from the registered set. /// The handler to remove. - private void RemoveHandler(Func, Task> handler) - { - lock (_sync) - { - _ = _handlers.Remove(handler); - } - } + [MethodImpl(MethodImplOptions.AggressiveInlining)] + private void RemoveHandler(Func, Task> handler) => + CopyOnWriteArray.Remove(ref _handlers, handler); /// An observer that bridges an observable sequence to a , completing the task when the observable completes or faults. /// The element type of the observable sequence. diff --git a/src/ReactiveUI.Binding.Shared/Observables/AppliedChangeObservable.cs b/src/ReactiveUI.Binding.Shared/Observables/AppliedChangeObservable.cs index 628389bf..a1aafbb9 100644 --- a/src/ReactiveUI.Binding.Shared/Observables/AppliedChangeObservable.cs +++ b/src/ReactiveUI.Binding.Shared/Observables/AppliedChangeObservable.cs @@ -42,47 +42,14 @@ public IDisposable Subscribe(IObserver observer) { ArgumentExceptionHelper.ThrowIfNull(observer); - IObserver[] updated; - IObserver[] current; - - do - { - current = Volatile.Read(ref _observers); - updated = new IObserver[current.Length + 1]; - Array.Copy(current, updated, current.Length); - updated[current.Length] = observer; - } - while (!ReferenceEquals(Interlocked.CompareExchange(ref _observers, updated, current), current)); - + CopyOnWriteArray.Add(ref _observers, observer); return new Subscription(this, observer); } /// Drops one observer without disturbing a change already being delivered. - /// The observer to drop. - /// - /// An observer that is not there is left alone. A subscription drops its own place once and no other path - /// reaches here, so that is a guard against a future caller rather than something the current ones do. - /// - internal void Remove(IObserver observer) - { - IObserver[] current; - IObserver[] updated; - - do - { - current = Volatile.Read(ref _observers); - var index = Array.IndexOf(current, observer); - if (index < 0) - { - return; - } - - updated = new IObserver[current.Length - 1]; - Array.Copy(current, updated, index); - Array.Copy(current, index + 1, updated, index, current.Length - index - 1); - } - while (!ReferenceEquals(Interlocked.CompareExchange(ref _observers, updated, current), current)); - } + /// The observer to drop. An observer that is not there is left alone. + [MethodImpl(MethodImplOptions.AggressiveInlining)] + internal void Remove(IObserver observer) => CopyOnWriteArray.Remove(ref _observers, observer); /// Releases one observer's place in the change stream. /// The stream subscribed to. diff --git a/src/ReactiveUI.Binding.Shared/Observables/PluginPropertyObservable.cs b/src/ReactiveUI.Binding.Shared/Observables/PluginPropertyObservable.cs index 090e7b32..63d1ba3a 100644 --- a/src/ReactiveUI.Binding.Shared/Observables/PluginPropertyObservable.cs +++ b/src/ReactiveUI.Binding.Shared/Observables/PluginPropertyObservable.cs @@ -98,6 +98,8 @@ internal sealed class Subscription : IDisposable, IObserver /// Serializes the initial emit with notifications arriving on other threads, so the handler always /// sees a consistent and pair whatever the timing. + /// It is a lock rather than a hand-off to the thread already emitting, because a before-change + /// notification must read the value on the raising thread before that thread writes. /// private readonly Lock _gate = new(); diff --git a/src/ReactiveUI.Binding.Shared/Observables/PropertyChangingObservable.cs b/src/ReactiveUI.Binding.Shared/Observables/PropertyChangingObservable.cs index 44cd7da5..a923becc 100644 --- a/src/ReactiveUI.Binding.Shared/Observables/PropertyChangingObservable.cs +++ b/src/ReactiveUI.Binding.Shared/Observables/PropertyChangingObservable.cs @@ -64,7 +64,8 @@ internal sealed class Subscription : IDisposable /// /// Serializes the initial emit in the constructor with concurrent /// invocations on other threads, so a racing handler emit and the constructor's initial emit do - /// not interleave on the downstream observer. + /// not interleave on the downstream observer. It is a lock rather than a hand-off to the thread + /// already emitting, because the value must be read on the raising thread before that thread writes. /// private readonly Lock _gate = new(); diff --git a/src/ReactiveUI.Binding.Shared/Observables/PropertyObservable.cs b/src/ReactiveUI.Binding.Shared/Observables/PropertyObservable.cs index dda82427..443c63be 100644 --- a/src/ReactiveUI.Binding.Shared/Observables/PropertyObservable.cs +++ b/src/ReactiveUI.Binding.Shared/Observables/PropertyObservable.cs @@ -70,11 +70,10 @@ internal sealed class Subscription : IDisposable private readonly EqualityComparer _comparer; /// - /// Serializes the initial emit in the constructor with concurrent - /// invocations on other threads, so the handler always sees a consistent - /// / snapshot regardless of timing. + /// Counts the emits asked for and not yet served. The thread that raises it from zero serves every + /// emit asked for while it runs, so emits never overlap and no thread waits for another. /// - private readonly Lock _gate = new(); + private int _pendingEmits; /// The downstream observer. Set to on disposal. private IObserver? _observer; @@ -145,33 +144,48 @@ private void OnPropertyChanged(object? sender, PropertyChangedEventArgs e) EmitCurrent(); } - /// - /// Reads the current property value under and forwards it to the downstream - /// observer when the distinct-until-changed gate allows. Holding across the - /// read-decision-emit sequence ensures the constructor's initial emit and any concurrent - /// invocation cannot interleave on the downstream observer or - /// publish a duplicate when both see the same current value. - /// + /// Reads the current property value and forwards it downstream when the distinct gate allows. + /// + /// A call made while another emit runs returns at once. The running emit reads the property again + /// before it stops, so the value is never lost and never stale. A call from inside the downstream + /// observer is delivered after that observer returns. A throw clears the count, so the next change + /// still emits. + /// private void EmitCurrent() { - lock (_gate) + if (Interlocked.Increment(ref _pendingEmits) != 1) { - var observer = Volatile.Read(ref _observer); - if (observer is null) - { - return; - } - - var value = _parent._getter(_parent._source); + return; + } - if (_parent._distinctUntilChanged && _hasValue && _comparer.Equals(value!, _lastValue!)) + var unserved = 1; + try + { + do { - return; + var observer = Volatile.Read(ref _observer); + if (observer is null) + { + return; + } + + var value = _parent._getter(_parent._source); + + if (!_parent._distinctUntilChanged || !_hasValue || !_comparer.Equals(value!, _lastValue!)) + { + _lastValue = value; + _hasValue = true; + observer.OnNext(value!); + } + + unserved = Interlocked.Add(ref _pendingEmits, -unserved); } - - _lastValue = value; - _hasValue = true; - observer.OnNext(value!); + while (unserved != 0); + } + catch + { + Volatile.Write(ref _pendingEmits, 0); + throw; } } } diff --git a/src/ReactiveUI.Binding.Shared/Observables/ViewThreadObservable.cs b/src/ReactiveUI.Binding.Shared/Observables/ViewThreadObservable.cs index 1cadcb3b..067f0f76 100644 --- a/src/ReactiveUI.Binding.Shared/Observables/ViewThreadObservable.cs +++ b/src/ReactiveUI.Binding.Shared/Observables/ViewThreadObservable.cs @@ -2,8 +2,6 @@ // 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.Collections.Concurrent; - #if REACTIVE_SHIM namespace ReactiveUI.Binding.Reactive.Observables; #else @@ -27,36 +25,32 @@ public IDisposable Subscribe(IObserver observer) return sink; } - /// One notification waiting for the owning thread. - /// Which of the three notifications this is. - /// The value, for a next notification. - /// The error, for an error notification. - private readonly struct Notification(NotificationKind kind, T? value, Exception? error) - { - /// Gets which of the three notifications this is. - public NotificationKind Kind { get; } = kind; - - /// Gets the value, for a next notification. - public T? Value { get; } = value; - - /// Gets the error, for an error notification. - public Exception? Error { get; } = error; - } - - /// The subscription that writes inline on the owning thread and queues everything else. + /// The subscription that writes inline on the owning thread and holds the latest value for it otherwise. /// The observer applying the write. /// The object the write lands on. /// The invoker for the thread that owns . private sealed class Sink(IObserver observer, object target, IViewThreadInvoker invoker) : IObserver, IDisposable { - /// The notifications waiting for the owning thread, created on the first write that has to wait. - private ConcurrentQueue? _queue; + /// Guards the waiting notifications and the scheduled flag. + private readonly Lock _gate = new(); /// The upstream subscription, or null once disposed. private IDisposable? _upstream; - /// How many notifications are queued and not yet delivered. - private int _pending; + /// The latest value waiting for the owning thread. + private T? _value; + + /// Whether holds a value. + private bool _hasValue; + + /// The error waiting for the owning thread, or null. + private Exception? _error; + + /// Whether completion is waiting for the owning thread. + private bool _completed; + + /// Whether a drain is scheduled or running. + private bool _scheduled; /// Non-zero once the subscription is disposed. private int _disposed; @@ -69,37 +63,28 @@ private sealed class Sink(IObserver observer, object target, IViewThreadInvok /// public void OnNext(T value) { - if (TryDeliverInline()) + if (Admit(NotificationKind.Next, value, null)) { observer.OnNext(value); - return; } - - Enqueue(new(NotificationKind.Next, value, null)); } /// public void OnError(Exception error) { - if (TryDeliverInline()) + if (Admit(NotificationKind.Error, default, error)) { observer.OnError(error); - return; } - - Enqueue(new(NotificationKind.Error, default, error)); } /// public void OnCompleted() { - if (TryDeliverInline()) + if (Admit(NotificationKind.Completed, default, null)) { observer.OnCompleted(); - return; } - - Enqueue(new(NotificationKind.Completed, default, null)); } /// @@ -114,55 +99,72 @@ public void Dispose() Interlocked.Exchange(ref _upstream, null)!.Dispose(); } - /// Delivers the queued notifications in order, on whichever thread the invoker or main thread runs this. - private void Drain() + /// Decides whether a notification runs on the calling thread, and holds it for the owning thread otherwise. + /// Which notification arrived. + /// The value, for a next notification. + /// The error, for an error notification. + /// when the caller should deliver the notification now. + private bool Admit(NotificationKind kind, T? value, Exception? error) { - var queue = Volatile.Read(ref _queue)!; - - do + if (Volatile.Read(ref _disposed) != 0) { - // The pending count is only raised after an enqueue, so there is always a notification to take. - _ = queue.TryDequeue(out var notification); + return false; + } - if (Volatile.Read(ref _disposed) == 0) + bool startDrain; + lock (_gate) + { + if (!_scheduled && invoker.CheckAccess(target)) { - Deliver(notification); + return true; } - } - while (Interlocked.Decrement(ref _pending) != 0); - } - /// Determines whether a notification can be delivered on the calling thread now. - /// when disposed, when earlier notifications are still queued, or when the caller is not on the owning thread. - private bool TryDeliverInline() => - Volatile.Read(ref _disposed) == 0 - && Volatile.Read(ref _pending) == 0 - && invoker.CheckAccess(target); - - /// Queues a notification and, when nothing is draining, schedules a drain. - /// The notification to queue. - private void Enqueue(in Notification notification) - { - if (Volatile.Read(ref _disposed) != 0) - { - return; + Hold(kind, value, error); + startDrain = !_scheduled; + _scheduled = true; } - var queue = Volatile.Read(ref _queue); - if (queue is null) + if (startDrain) { - // The source delivers one notification at a time, so only this thread ever creates the queue. - queue = new(); - Volatile.Write(ref _queue, queue); + ScheduleDrain(); } - queue.Enqueue(notification); + return false; + } - if (Interlocked.Increment(ref _pending) != 1) + /// Holds a notification until the drain runs. + /// Which notification arrived. + /// The value, for a next notification. + /// The error, for an error notification. + private void Hold(NotificationKind kind, T? value, Exception? error) + { + switch (kind) { - return; + case NotificationKind.Next: + { + // A newer value replaces a waiting one, so an echo of an earlier write never writes an old value back. + _value = value; + _hasValue = true; + break; + } + + case NotificationKind.Error: + { + _error = error; + break; + } + + default: + { + _completed = true; + break; + } } + } + /// Schedules the drain on the host's main thread when one is set, and through the invoker otherwise. + private void ScheduleDrain() + { var mainThread = BindingSchedulers.MainThread; if (mainThread is null) { @@ -179,29 +181,62 @@ private void Enqueue(in Notification notification) }); } - /// Hands one notification to the observer. - /// The notification to deliver. - private void Deliver(in Notification notification) + /// Delivers what is waiting until nothing is left, including anything held while a write runs. + private void Drain() { - switch (notification.Kind) + while (true) { - case NotificationKind.Next: - { - observer.OnNext(notification.Value!); - break; - } + T? value; + bool hasValue; + Exception? error; + bool completed; - case NotificationKind.Error: + lock (_gate) { - observer.OnError(notification.Error!); - break; + if (!_hasValue && _error is null && !_completed) + { + _scheduled = false; + return; + } + + value = _value; + hasValue = _hasValue; + error = _error; + completed = _completed; + _value = default; + _hasValue = false; + _error = null; + _completed = false; } - default: + if (Volatile.Read(ref _disposed) != 0) { - observer.OnCompleted(); - break; + continue; } + + Deliver(value, hasValue, error, completed); + } + } + + /// Hands the waiting value, then any terminal notification, to the observer. + /// The waiting value. + /// Whether a value was waiting. + /// The waiting error, or null. + /// Whether completion was waiting. + private void Deliver(T? value, bool hasValue, Exception? error, bool completed) + { + if (hasValue) + { + observer.OnNext(value!); + } + + if (error is not null) + { + observer.OnError(error); + } + else if (completed) + { + observer.OnCompleted(); } } } diff --git a/src/ReactiveUI.Binding/PublicAPI/net10.0/PublicAPI.txt b/src/ReactiveUI.Binding/PublicAPI/net10.0/PublicAPI.txt index c75d652c..5eb6a9b3 100644 --- a/src/ReactiveUI.Binding/PublicAPI/net10.0/PublicAPI.txt +++ b/src/ReactiveUI.Binding/PublicAPI/net10.0/PublicAPI.txt @@ -375,7 +375,7 @@ namespace ReactiveUI.Binding public override int GetAffinityForObjects() { } public override bool TryConvert(int from, object? conversionHint, out string? result) { } } - [System.Diagnostics.DebuggerDisplay("Handlers = {_handlers.Count}")] + [System.Diagnostics.DebuggerDisplay("Handlers = {_handlers.Length}")] public class Interaction : ReactiveUI.Binding.IInteraction { public Interaction() { } diff --git a/src/ReactiveUI.Binding/PublicAPI/net11.0/PublicAPI.txt b/src/ReactiveUI.Binding/PublicAPI/net11.0/PublicAPI.txt index c75d652c..5eb6a9b3 100644 --- a/src/ReactiveUI.Binding/PublicAPI/net11.0/PublicAPI.txt +++ b/src/ReactiveUI.Binding/PublicAPI/net11.0/PublicAPI.txt @@ -375,7 +375,7 @@ namespace ReactiveUI.Binding public override int GetAffinityForObjects() { } public override bool TryConvert(int from, object? conversionHint, out string? result) { } } - [System.Diagnostics.DebuggerDisplay("Handlers = {_handlers.Count}")] + [System.Diagnostics.DebuggerDisplay("Handlers = {_handlers.Length}")] public class Interaction : ReactiveUI.Binding.IInteraction { public Interaction() { } diff --git a/src/ReactiveUI.Binding/PublicAPI/net462/PublicAPI.txt b/src/ReactiveUI.Binding/PublicAPI/net462/PublicAPI.txt index fc4b599d..10ce6f57 100644 --- a/src/ReactiveUI.Binding/PublicAPI/net462/PublicAPI.txt +++ b/src/ReactiveUI.Binding/PublicAPI/net462/PublicAPI.txt @@ -354,7 +354,7 @@ namespace ReactiveUI.Binding public override int GetAffinityForObjects() { } public override bool TryConvert(int from, object? conversionHint, out string? result) { } } - [System.Diagnostics.DebuggerDisplay("Handlers = {_handlers.Count}")] + [System.Diagnostics.DebuggerDisplay("Handlers = {_handlers.Length}")] public class Interaction : ReactiveUI.Binding.IInteraction { public Interaction() { } diff --git a/src/ReactiveUI.Binding/PublicAPI/net47/PublicAPI.txt b/src/ReactiveUI.Binding/PublicAPI/net47/PublicAPI.txt index fc4b599d..10ce6f57 100644 --- a/src/ReactiveUI.Binding/PublicAPI/net47/PublicAPI.txt +++ b/src/ReactiveUI.Binding/PublicAPI/net47/PublicAPI.txt @@ -354,7 +354,7 @@ namespace ReactiveUI.Binding public override int GetAffinityForObjects() { } public override bool TryConvert(int from, object? conversionHint, out string? result) { } } - [System.Diagnostics.DebuggerDisplay("Handlers = {_handlers.Count}")] + [System.Diagnostics.DebuggerDisplay("Handlers = {_handlers.Length}")] public class Interaction : ReactiveUI.Binding.IInteraction { public Interaction() { } diff --git a/src/ReactiveUI.Binding/PublicAPI/net471/PublicAPI.txt b/src/ReactiveUI.Binding/PublicAPI/net471/PublicAPI.txt index fc4b599d..10ce6f57 100644 --- a/src/ReactiveUI.Binding/PublicAPI/net471/PublicAPI.txt +++ b/src/ReactiveUI.Binding/PublicAPI/net471/PublicAPI.txt @@ -354,7 +354,7 @@ namespace ReactiveUI.Binding public override int GetAffinityForObjects() { } public override bool TryConvert(int from, object? conversionHint, out string? result) { } } - [System.Diagnostics.DebuggerDisplay("Handlers = {_handlers.Count}")] + [System.Diagnostics.DebuggerDisplay("Handlers = {_handlers.Length}")] public class Interaction : ReactiveUI.Binding.IInteraction { public Interaction() { } diff --git a/src/ReactiveUI.Binding/PublicAPI/net472/PublicAPI.txt b/src/ReactiveUI.Binding/PublicAPI/net472/PublicAPI.txt index fc4b599d..10ce6f57 100644 --- a/src/ReactiveUI.Binding/PublicAPI/net472/PublicAPI.txt +++ b/src/ReactiveUI.Binding/PublicAPI/net472/PublicAPI.txt @@ -354,7 +354,7 @@ namespace ReactiveUI.Binding public override int GetAffinityForObjects() { } public override bool TryConvert(int from, object? conversionHint, out string? result) { } } - [System.Diagnostics.DebuggerDisplay("Handlers = {_handlers.Count}")] + [System.Diagnostics.DebuggerDisplay("Handlers = {_handlers.Length}")] public class Interaction : ReactiveUI.Binding.IInteraction { public Interaction() { } diff --git a/src/ReactiveUI.Binding/PublicAPI/net48/PublicAPI.txt b/src/ReactiveUI.Binding/PublicAPI/net48/PublicAPI.txt index fc4b599d..10ce6f57 100644 --- a/src/ReactiveUI.Binding/PublicAPI/net48/PublicAPI.txt +++ b/src/ReactiveUI.Binding/PublicAPI/net48/PublicAPI.txt @@ -354,7 +354,7 @@ namespace ReactiveUI.Binding public override int GetAffinityForObjects() { } public override bool TryConvert(int from, object? conversionHint, out string? result) { } } - [System.Diagnostics.DebuggerDisplay("Handlers = {_handlers.Count}")] + [System.Diagnostics.DebuggerDisplay("Handlers = {_handlers.Length}")] public class Interaction : ReactiveUI.Binding.IInteraction { public Interaction() { } diff --git a/src/ReactiveUI.Binding/PublicAPI/net481/PublicAPI.txt b/src/ReactiveUI.Binding/PublicAPI/net481/PublicAPI.txt index fc4b599d..10ce6f57 100644 --- a/src/ReactiveUI.Binding/PublicAPI/net481/PublicAPI.txt +++ b/src/ReactiveUI.Binding/PublicAPI/net481/PublicAPI.txt @@ -354,7 +354,7 @@ namespace ReactiveUI.Binding public override int GetAffinityForObjects() { } public override bool TryConvert(int from, object? conversionHint, out string? result) { } } - [System.Diagnostics.DebuggerDisplay("Handlers = {_handlers.Count}")] + [System.Diagnostics.DebuggerDisplay("Handlers = {_handlers.Length}")] public class Interaction : ReactiveUI.Binding.IInteraction { public Interaction() { } diff --git a/src/ReactiveUI.Binding/PublicAPI/net8.0/PublicAPI.txt b/src/ReactiveUI.Binding/PublicAPI/net8.0/PublicAPI.txt index c75d652c..5eb6a9b3 100644 --- a/src/ReactiveUI.Binding/PublicAPI/net8.0/PublicAPI.txt +++ b/src/ReactiveUI.Binding/PublicAPI/net8.0/PublicAPI.txt @@ -375,7 +375,7 @@ namespace ReactiveUI.Binding public override int GetAffinityForObjects() { } public override bool TryConvert(int from, object? conversionHint, out string? result) { } } - [System.Diagnostics.DebuggerDisplay("Handlers = {_handlers.Count}")] + [System.Diagnostics.DebuggerDisplay("Handlers = {_handlers.Length}")] public class Interaction : ReactiveUI.Binding.IInteraction { public Interaction() { } diff --git a/src/ReactiveUI.Binding/PublicAPI/net9.0/PublicAPI.txt b/src/ReactiveUI.Binding/PublicAPI/net9.0/PublicAPI.txt index c75d652c..5eb6a9b3 100644 --- a/src/ReactiveUI.Binding/PublicAPI/net9.0/PublicAPI.txt +++ b/src/ReactiveUI.Binding/PublicAPI/net9.0/PublicAPI.txt @@ -375,7 +375,7 @@ namespace ReactiveUI.Binding public override int GetAffinityForObjects() { } public override bool TryConvert(int from, object? conversionHint, out string? result) { } } - [System.Diagnostics.DebuggerDisplay("Handlers = {_handlers.Count}")] + [System.Diagnostics.DebuggerDisplay("Handlers = {_handlers.Length}")] public class Interaction : ReactiveUI.Binding.IInteraction { public Interaction() { } diff --git a/src/tests/ReactiveUI.Binding.SourceGenerators.Tests/RuntimeExecution/ViewWriteSchedulingRuntimeTests.cs b/src/tests/ReactiveUI.Binding.SourceGenerators.Tests/RuntimeExecution/ViewWriteSchedulingRuntimeTests.cs index 66eb26e4..718156ce 100644 --- a/src/tests/ReactiveUI.Binding.SourceGenerators.Tests/RuntimeExecution/ViewWriteSchedulingRuntimeTests.cs +++ b/src/tests/ReactiveUI.Binding.SourceGenerators.Tests/RuntimeExecution/ViewWriteSchedulingRuntimeTests.cs @@ -21,6 +21,9 @@ public class ViewWriteSchedulingRuntimeTests /// What a scenario returns when the write went through the platform's own dispatcher. private const string Dispatched = "dispatched"; + /// What a burst scenario returns when both sides end on the latest value. + private const string Settled = "settled"; + /// Bindings onto stand-ins for WPF, WinForms and MAUI views, and onto a plain object. private const string SchedulingSource = """ using System; @@ -88,6 +91,40 @@ public class WpfView : System.Windows.Threading.DispatcherObject public string DisplayName { get; set; } = ""; } + public class TwoWayWpfView : System.Windows.Threading.DispatcherObject, INotifyPropertyChanged + { + private string _displayName = ""; + + public event PropertyChangedEventHandler PropertyChanged; + + public int Writes { get; private set; } + + public string DisplayName + { + get { return _displayName; } + set + { + Writes++; + if (Writes > 1000) + { + throw new InvalidOperationException("The binding kept writing."); + } + + _displayName = value; + var handler = PropertyChanged; + if (handler != null) + { + handler(this, new PropertyChangedEventArgs("DisplayName")); + } + } + } + } + + public class BoundWpfView : TwoWayWpfView, IViewFor + { + public object ViewModel { get; set; } + } + public class WinFormsView : System.Windows.Forms.Control { public string DisplayName { get; set; } = ""; @@ -262,6 +299,45 @@ public static string OneWayToAMauiViewWithNoDispatcher() return Describe(view.DisplayName, false, 0); } + public static string TwoWayBurstFromAnotherThread() + { + var viewModel = new MyViewModel(); + var view = new TwoWayWpfView(); + view.Dispatcher.Holds = true; + var binding = viewModel.BindTwoWay(view, x => x.Name, x => x.DisplayName); + + return RunBurst(viewModel, view); + } + + public static string ViewFirstBindBurstFromAnotherThread() + { + var viewModel = new MyViewModel(); + var view = new BoundWpfView(); + view.Dispatcher.Holds = true; + var binding = view.Bind(viewModel, x => x.Name, x => x.DisplayName); + + return RunBurst(viewModel, view); + } + + private static string RunBurst(MyViewModel viewModel, TwoWayWpfView view) + { + try + { + view.Dispatcher.Pump(view); + viewModel.Name = "A"; + viewModel.Name = "B"; + view.Dispatcher.Pump(view); + } + catch (Exception) + { + return "kept writing"; + } + + return viewModel.Name == "B" && view.DisplayName == "B" + ? "settled" + : "stale: " + viewModel.Name + " / " + view.DisplayName; + } + private static string Describe(string written, bool scheduled, int dispatched) { if (written != "changed") @@ -288,14 +364,40 @@ public enum DispatcherPriority public class Dispatcher { + private readonly System.Collections.Generic.Queue _held = new System.Collections.Generic.Queue(); + public int Posts { get; private set; } + public bool Holds { get; set; } + public object BeginInvoke(DispatcherPriority priority, Delegate method, object arg) { Posts++; + if (Holds) + { + _held.Enqueue(() => method.DynamicInvoke(arg)); + return null; + } + method.DynamicInvoke(arg); return null; } + + public void Pump(DispatcherObject owner) + { + owner.HasAccess = true; + try + { + while (_held.Count > 0) + { + _held.Dequeue()(); + } + } + finally + { + owner.HasAccess = false; + } + } } public class DispatcherObject @@ -418,6 +520,18 @@ public async Task BindOneWay_ToAMauiObjectFromAnotherThread_GoesThroughItsDispat public async Task BindOneWay_ToAMauiObjectWithNoDispatcher_WritesInline() => await Assert.That(await RunScenarioAsync("OneWayToAMauiViewWithNoDispatcher")).IsEqualTo(Inline); + /// Two changes from another thread before the view's thread runs leave a two-way binding on the latest value. + /// A task representing the asynchronous test operation. + [Test] + public async Task BindTwoWay_ABurstFromAnotherThread_SettlesOnTheLatestValue() => + await Assert.That(await RunScenarioAsync("TwoWayBurstFromAnotherThread")).IsEqualTo(Settled); + + /// Two changes from another thread before the view's thread runs leave a view-first binding on the latest value. + /// A task representing the asynchronous test operation. + [Test] + public async Task Bind_ABurstFromAnotherThread_SettlesOnTheLatestValue() => + await Assert.That(await RunScenarioAsync("ViewFirstBindBurstFromAnotherThread")).IsEqualTo(Settled); + /// Compiles the scenario, runs one of its entry points, and reports where the write was delivered. /// The static method on the scenario's Usage class to run. /// What the entry point reported. diff --git a/src/tests/ReactiveUI.Binding.Tests/Observables/PropertyObservableInitialEmitSerializationTests.cs b/src/tests/ReactiveUI.Binding.Tests/Observables/PropertyObservableInitialEmitSerializationTests.cs index 5b772a5f..5c4f95b2 100644 --- a/src/tests/ReactiveUI.Binding.Tests/Observables/PropertyObservableInitialEmitSerializationTests.cs +++ b/src/tests/ReactiveUI.Binding.Tests/Observables/PropertyObservableInitialEmitSerializationTests.cs @@ -28,12 +28,15 @@ public class PropertyObservableInitialEmitSerializationTests /// The property value a competing thread writes. private const string ReplacementName = "Bob"; + /// The property value written after a read has failed. + private const string ThirdName = "Carol"; + /// - /// How long to give a competing thread to complete its emit while the initial emit is still on the - /// stack. A serialized subscription blocks that thread for the whole window, so the wait always - /// expires; an unserialized one lets it through in microseconds. + /// How long the initial emit waits for a competing thread to finish its write. The subscription never + /// makes that thread wait, so the join ends as soon as the write does; the bound only turns a deadlock + /// into a failure. /// - private const int InterleaveWindowMilliseconds = 500; + private const int CompetitorTimeoutMilliseconds = 10_000; /// /// Subscriptions the unforced sweep builds. Sized from measurement: against unserialized code this @@ -75,8 +78,8 @@ public async Task Subscribe_PropertyChangedRaisedOnAnotherThreadWhileAttaching_E /// The same defect reached re-entrantly rather than across threads: the property read the /// constructor performs for its initial emit itself raises /// , so the handler runs part-way through - /// construction on the subscribing thread. This also pins that the serialization is re-entrant, - /// since a non-re-entrant gate would deadlock here rather than fail. + /// construction on the subscribing thread. This also pins that a notification raised on the emitting + /// thread is handed to the running emit, since waiting for that emit would deadlock here rather than fail. /// /// A representing the asynchronous unit test. [Test] @@ -111,20 +114,20 @@ public async Task Subscribe_PropertyChangedRaisedReentrantlyDuringInitialRead_Em } /// - /// Pins the change's central claim - that a competing handler runs either wholly before or wholly - /// after the initial emit, never inside it - by holding a competing thread against the initial emit - /// while it is on the stack and recording whether its emit overlaps. + /// A competing handler never runs inside the initial emit and never waits for it. A thread that + /// writes while the initial emit is on the stack finishes at once, and the running emit delivers its + /// value after the initial one. Waiting on the observer for such a thread is therefore not a deadlock. /// /// A representing the asynchronous unit test. [Test] - public async Task Subscribe_PropertyChangedRaisedOnAnotherThreadDuringInitialEmit_DoesNotOverlapTheInitialEmit() + public async Task Subscribe_PropertyChangedRaisedOnAnotherThreadDuringInitialEmit_NeitherOverlapsNorBlocksThatThread() { var source = new HookedViewModel { Name = InitialName }; using var competitorStarted = new ManualResetEventSlim(false); Thread? competitor = null; + var competitorFinished = false; - // Runs from inside the downstream call of the initial emit, which is the window the change keeps - // exclusive. The bounded join is what a blocked competing thread looks like from in here. + // Runs from inside the downstream call of the initial emit, which is the window no other emit may enter. var recorder = new EmissionRecorder { OnFirstValue = () => @@ -137,7 +140,7 @@ public async Task Subscribe_PropertyChangedRaisedOnAnotherThreadDuringInitialEmi competitor.Start(); competitorStarted.Wait(); - _ = competitor.Join(InterleaveWindowMilliseconds); + competitorFinished = competitor.Join(CompetitorTimeoutMilliseconds); }, }; @@ -152,11 +155,74 @@ public async Task Subscribe_PropertyChangedRaisedOnAnotherThreadDuringInitialEmi competitor!.Join(); await AssertNoErrors(recorder); + await Assert.That(competitorFinished).IsTrue(); await Assert.That(recorder.MaxConcurrentEmissions).IsEqualTo(1); await AssertSequence(recorder.Snapshot(), InitialName, ReplacementName); } } + /// An observer that changes the property it observes gets the change after its own call returns. + /// A representing the asynchronous unit test. + [Test] + public async Task OnNext_ObserverChangesTheObservedProperty_DeliversTheChangeAfterTheObserverReturns() + { + var source = new HookedViewModel { Name = InitialName }; + var recorder = new EmissionRecorder { OnFirstValue = () => source.Name = ReplacementName }; + + var observable = new PropertyObservable( + source, + nameof(HookedViewModel.Name), + static x => ((HookedViewModel)x).Name, + distinctUntilChanged: true); + + using (observable.Subscribe(recorder)) + { + await AssertNoErrors(recorder); + await Assert.That(recorder.MaxConcurrentEmissions).IsEqualTo(1); + await AssertSequence(recorder.Snapshot(), InitialName, ReplacementName); + } + } + + /// + /// A read that throws reaches the thread that raised the change, and the subscription keeps working: + /// the next change still emits. + /// + /// A representing the asynchronous unit test. + [Test] + public async Task OnPropertyChanged_ReadThrows_TheNextChangeStillEmits() + { + var source = new HookedViewModel { Name = InitialName }; + var recorder = new EmissionRecorder(); + var failNextRead = false; + + string? ReadOrFail(INotifyPropertyChanged instance) + { + if (!failNextRead) + { + return ((HookedViewModel)instance).Name; + } + + failNextRead = false; + throw new InvalidOperationException(nameof(ReadOrFail)); + } + + var observable = new PropertyObservable( + source, + nameof(HookedViewModel.Name), + ReadOrFail, + distinctUntilChanged: true); + + using (observable.Subscribe(recorder)) + { + failNextRead = true; + await Assert.That(() => source.Name = ReplacementName).ThrowsExactly(); + + source.Name = ThirdName; + + await AssertSequence(recorder.Snapshot(), InitialName, ThirdName); + } + } + /// /// The initial emit stays unconditional on an ordinary subscribe, including when the value equals /// the default for its type. The change applies the distinct-until-changed test to the initial emit diff --git a/src/tests/ReactiveUI.Binding.Tests/Observables/ViewThreadObservableTests.cs b/src/tests/ReactiveUI.Binding.Tests/Observables/ViewThreadObservableTests.cs index 15a11850..0ab815b5 100644 --- a/src/tests/ReactiveUI.Binding.Tests/Observables/ViewThreadObservableTests.cs +++ b/src/tests/ReactiveUI.Binding.Tests/Observables/ViewThreadObservableTests.cs @@ -71,10 +71,10 @@ public async Task OnNext_FromAnotherThread_WaitsForTheInvoker() } } - /// A burst from another thread posts one drain and keeps its order. + /// A burst from another thread posts one write, and that write carries only the latest value. /// A task representing the asynchronous test operation. [Test] - public async Task OnNext_ABurstFromAnotherThread_PostsOnceAndKeepsTheOrder() + public async Task OnNext_ABurstFromAnotherThread_WritesOnlyTheLatestValue() { var invoker = new StubViewThreadInvoker(); var (source, observer, subscription) = Subscribe(invoker); @@ -88,14 +88,14 @@ public async Task OnNext_ABurstFromAnotherThread_PostsOnceAndKeepsTheOrder() invoker.RunPosted(); await Assert.That(invoker.PostCount).IsEqualTo(1); - await Assert.That(string.Join(",", observer.Values)).IsEqualTo("first,second,third"); + await Assert.That(string.Join(",", observer.Values)).IsEqualTo(Third); } } - /// A value on the owning thread waits behind values still queued from another thread. + /// A value on the owning thread replaces a value still waiting from another thread. /// A task representing the asynchronous test operation. [Test] - public async Task OnNext_OnTheOwningThreadBehindQueuedValues_WaitsItsTurn() + public async Task OnNext_OnTheOwningThreadWhileAWriteWaits_ReplacesTheWaitingValue() { var invoker = new StubViewThreadInvoker(); var (source, observer, subscription) = Subscribe(invoker); @@ -111,7 +111,27 @@ public async Task OnNext_OnTheOwningThreadBehindQueuedValues_WaitsItsTurn() invoker.RunPosted(); await Assert.That(invoker.PostCount).IsEqualTo(1); - await Assert.That(string.Join(",", observer.Values)).IsEqualTo("first,second"); + await Assert.That(string.Join(",", observer.Values)).IsEqualTo(Second); + } + } + + /// A value that waits is written before the completion that follows it. + /// A task representing the asynchronous test operation. + [Test] + public async Task OnCompleted_AfterAWaitingValue_WritesTheValueFirst() + { + var invoker = new StubViewThreadInvoker(); + var completedAfter = string.Empty; + var source = new ManualObservable(); + var observer = new RecordingObserver(); + + using (new ViewThreadObservable(source, Target, invoker).Subscribe(new CompletionOrderObserver(observer, () => completedAfter = string.Join(",", observer.Values)))) + { + source.Observer?.OnNext(First); + source.Observer?.OnCompleted(); + invoker.RunPosted(); + + await Assert.That(completedAfter).IsEqualTo(First); } } @@ -302,6 +322,24 @@ private static (ManualObservable Source, RecordingObserver Obser return (source, observer, new ViewThreadObservable(source, Target, invoker).Subscribe(observer)); } + /// Forwards values to a recorder and reports what it held when completion arrived. + /// The recorder that receives each value. + /// Runs when completion arrives. + private sealed class CompletionOrderObserver(RecordingObserver inner, Action onCompleted) : IObserver + { + /// + [MethodImpl(MethodImplOptions.AggressiveInlining)] + public void OnCompleted() => onCompleted(); + + /// + [MethodImpl(MethodImplOptions.AggressiveInlining)] + public void OnError(Exception error) => inner.OnError(error); + + /// + [MethodImpl(MethodImplOptions.AggressiveInlining)] + public void OnNext(string value) => inner.OnNext(value); + } + /// A source that counts how often its subscription is disposed. private sealed class CountingObservable : IObservable {