Merge pull request #2262 from unoplatform/mergify/bp/release/stable/4…

….1/pr-2250

fix(mvux): Properties update might be deferred too long if dispatcher is under heavy stress (backport #2250)

src/Uno.Extensions.Core/Option.T.cs

@@ -166,6 +166,9 @@ public override string ToString()
 		{
 			OptionType.Undefined => $"Undefined<{typeof(T).Name}>",
 			OptionType.None => $"None<{typeof(T).Name}>",
-			_ => $"Some({(_value is IEnumerable enumerable ? string.Join(",", enumerable.Cast<object>()) : _value)})",
+			_ when _value is null => "Some(--null--)",
+			_ when _value is string str => $"Some({str})",
+			_ when _value is IEnumerable enumerable => $"Some({string.Join(",", enumerable.Cast<object>())})",
+			_ => $"Some({_value})",
 		};
 }

src/Uno.Extensions.Reactive.Tests/Presentation/Bindings/Given_BindableViewModelBase.cs

@@ -0,0 +1,70 @@
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Text;
+using System.Threading.Tasks;
+using FluentAssertions;
+using Microsoft.VisualStudio.TestTools.UnitTesting;
+using Uno.Extensions.Reactive.Testing;
+
+namespace Uno.Extensions.Reactive.Tests.Presentation.Bindings;
+
+[TestClass]
+public partial class Given_BindableViewModelBase : FeedUITests
+{
+	[TestMethod]
+	public async Task When_UpdateSourceMultipleTimeWhileUIThreadFreeze_Then_LastWin()
+	{
+		var sut = new BindableWhen_UpdateSourceMultipleTimeWhileUIThreadFreeze_Then_LastWin_Model();
+
+		var changes = 0;
+		var uiFrozen = new ManualResetEvent(false);
+		var bgCompleted = new ManualResetEvent(false);
+
+		// First complete init, including changes raised on the UI thread
+		await Dispatcher.ExecuteAsync(_ => sut.PropertyChanged += (snd, e) => changes++, CT);
+		await WaitFor(() => sut.Value == 42 && changes is 1);
+
+		// Freeze the UI thread
+		Dispatcher.TryEnqueue(() =>
+		{
+			uiFrozen.Set();
+			bgCompleted.WaitOne();
+		}).Should().BeTrue();
+		uiFrozen.WaitOne(1000).Should().BeTrue();
+
+		// Update the source multiple times from bg thread
+		await sut.Model.Value.SetAsync(43, CT);
+		await sut.Model.Value.SetAsync(44, CT);
+		await sut.Model.Value.SetAsync(45, CT);
+		await sut.Model.Value.SetAsync(46, CT);
+
+		// Release the UI thread so it can process the changes
+		bgCompleted.Set();
+		await Dispatcher.ExecuteAsync(_ => { }, CT); // Wait for the UI thread to run something
+
+		// Confirm the last change has been applied
+		await WaitFor(() => sut.Value == 46);
+		changes.Should().Be(2); // And only one property changed should have been raised for all changes
+	}
+
+	public partial class When_UpdateSourceMultipleTimeWhileUIThreadFreeze_Then_LastWin_Model
+	{
+		public IState<int> Value => State<int>.Value(this, () => 42);
+	}
+
+	private async Task WaitFor(Func<bool> predicate)
+	{
+		for (var i = 0; i < 100; i++)
+		{
+			if (predicate())
+			{
+				return;
+			}
+
+			await Task.Delay(1);
+		}
+
+		throw new TimeoutException();
+	}
+}

src/Uno.Extensions.Reactive.UI/Utils/Dispatching/DispatcherQueueProvider.cs

@@ -37,7 +37,7 @@ public Dispatcher(DispatcherQueue queue)
 
 		/// <inheritdoc />
 		public bool TryEnqueue(Action action)
-			=> _queue.TryEnqueue(() => action());
+			=> _queue.TryEnqueue(Unsafe.As<DispatcherQueueHandler>(action));
 
 		/// <inheritdoc />
 		public async ValueTask<TResult> ExecuteAsync<TResult>(AsyncFunc<TResult> action, CancellationToken ct)

src/Uno.Extensions.Reactive/Core/Internal/FeedSubscription.cs

@@ -35,7 +35,7 @@ public FeedSubscription(ISignal<Message<T>> feed, SourceContext rootContext)
 			isInitialSyncValuesSkippingAllowed: true);
 	}
 
-	string ISourceContextOwner.Name => $"Sub on '{_feed}' for ctx '{_context.Parent!.Owner.Name}'.";
+	string ISourceContextOwner.Name => $"Sub on '{LogHelper.GetIdentifier(_feed)}' for ctx '{_context.Parent!.Owner.Name}'.";
 
 	IDispatcher? ISourceContextOwner.Dispatcher => null;
 
@@ -64,7 +64,7 @@ public async IAsyncEnumerable<Message<T>> GetMessages(SourceContext subscriberCo
 			{
 				// We make sure that even if we are replaying a previous message, the changes collection contains all keys.
 				isFirstMessage = false;
-				yield return Message<T>.Initial.OverrideBy(msg);
+				yield return Message<T>.Initial.OverrideBy(msg);			
 			}
 			else
 			{

src/Uno.Extensions.Reactive/Presentation/Bindings/BindableViewModelBase.cs

@@ -1,7 +1,9 @@
 using System;
 using System.Collections.Generic;
 using System.ComponentModel;
+using System.Diagnostics;
 using System.Linq;
+using System.Reflection;
 using System.Runtime.CompilerServices;
 using System.Threading;
 using System.Threading.Tasks;
@@ -127,36 +129,48 @@ async void ViewModelToView(Action<TProperty> updated)
 		{
 			try
 			{
-				var initialValue = stateImpl.Current.Current.Data.SomeOrDefault(GetDefaultValueForBindings<TProperty>());
+				var defaultValue = GetDefaultValueForBindings<TProperty>();
+				var value = stateImpl.Current.Current.Data.SomeOrDefault(defaultValue);
 
 				// We run the update sync in setup, no matter the thread
-				updated(initialValue);
+				updated(value);
+
 				var ct = stateImpl.Context.Token;
 				var source = FeedUIHelper.GetSource(stateImpl, stateImpl.Context);
 				var dispatcher = await _dispatcher.GetFirstResolved(ct).ConfigureAwait(false);
-
-				dispatcher.TryEnqueue(async () =>
+				var updateScheduled = 0;
+
+				// Note: We use for each here to deduplicate updates in case of fast updates of the source.
+				//		 This also ensure to not wait to for the UI thread before fetching MoveNext the source.
+				_ = source
+					.ForEachAsync(OnMessage, ct)
+					.ContinueWith(
+						enumeration =>
+						{
+							this.Log().Error(
+								enumeration.Exception!,
+								$"Synchronization from ViewModel to View of '{propertyName}' failed."
+								+ "(This is a final error, changes made in the VM are no longer propagated to the View.)");
+						},
+						TaskContinuationOptions.OnlyOnFaulted);
+				void OnMessage(Message<TProperty> msg)
 				{
-					try
+					if (msg.Changes.Contains(MessageAxis.Data) && !(msg.Current.Get(BindingSource)?.Equals((this, propertyName)) ?? false))
 					{
-						// Note: No needs to use .WithCancellation() here as we are enumerating the stateImp which is going to be disposed anyway.
-						await foreach (var msg in source.WithCancellation(ct).ConfigureAwait(true))
+						value = msg.Current.Data.SomeOrDefault(defaultValue);
+						if (Interlocked.CompareExchange(ref updateScheduled, 1, 0) is 0)
 						{
-							if (msg.Changes.Contains(MessageAxis.Data) && !(msg.Current.Get(BindingSource)?.Equals((this, propertyName)) ?? false))
-							{
-								updated(msg.Current.Data.SomeOrDefault(GetDefaultValueForBindings<TProperty>()));
-								_propertyChanged.Raise(new PropertyChangedEventArgs(propertyName));
-							}
+							dispatcher.TryEnqueue(UpdateValue);
 						}
 					}
-					catch (Exception error)
-					{
-						this.Log().Error(
-							error,
-							$"Synchronization from ViewModel to View of '{propertyName}' failed."
-							+ "(This is a final error, changes made in the VM are no longer propagated to the View.)");
-					}
-				});
+				}
+
+				void UpdateValue()
+				{
+					updateScheduled = 0;
+					updated(value);
+					_propertyChanged.Raise(new PropertyChangedEventArgs(propertyName));
+				}
 			}
 			catch (Exception error)
 			{

src/Uno.Extensions.Reactive/Utils/Logging/LogHelper.cs

@@ -0,0 +1,29 @@
+using System;
+using System.Linq;
+using System.Threading;
+using Uno.Extensions.Reactive;
+
+namespace Uno.Extensions.Reactive.Logging;
+
+internal static class LogHelper
+{
+	public static string GetIdentifier(object? obj)
+		=> obj switch
+		{
+			null => "--null--",
+#if DEBUG
+			_ => $"{GetTypeName(obj)}-{obj.GetHashCode():X8}",
+#else
+			_ => obj.ToString()
+#endif
+		};
+
+	public static string GetTypeName(object obj)
+		=> obj.GetType() switch
+		{
+			{ IsGenericType: true } type => $"{type.Name}<{string.Join(", ", type.GenericTypeArguments.Select(GetTypeName))}>",
+			{ IsArray: true } type => $"{GetTypeName(type.GetElementType()!)}[]",
+			{ IsValueType: true } type => type.ToString(),
+			{ } type => type.Name
+		};
+}