Further testing, fixing and investigations

Author: Fijo

Assets/Src/Scripts/Unity/Concurrency/ThreadPoolOnlyScheduler.cs

@@ -0,0 +1,62 @@
+using System.Reactive.Concurrency;
+
+namespace System.Reactive.Unity.Concurrency {
+    /// <summary>
+    /// Represents an object that schedules units of work on the CLR thread pool. (This one will never fall back to the NewThreadScheduler unlike the ThreadPoolScheduler)
+    /// </summary>
+    /// <seealso cref="Instance">Singleton instance of this type exposed through this static property.</seealso>
+    public sealed class ThreadPoolOnlyScheduler : LocalScheduler, ISchedulerPeriodic {
+        private static readonly Lazy<ThreadPoolOnlyScheduler> _lazyDefault = new Lazy<ThreadPoolOnlyScheduler>(static () => new ThreadPoolOnlyScheduler());
+        private readonly ThreadPoolScheduler _innerScheduler;
+
+        public ThreadPoolOnlyScheduler() {
+            _innerScheduler = ThreadPoolScheduler.Instance;
+        }
+
+        /// <summary>
+        /// Gets the singleton instance of the thread pool only scheduler. (This one will never fall back to the NewThreadScheduler unlike the ThreadPoolScheduler)
+        /// </summary>
+        public static ThreadPoolOnlyScheduler Instance => _lazyDefault.Value;
+
+        /// <summary>
+        /// Schedules an action to be executed.
+        /// </summary>
+        /// <typeparam name="TState">The type of the state passed to the scheduled action.</typeparam>
+        /// <param name="state">State passed to the action to be executed.</param>
+        /// <param name="action">Action to be executed.</param>
+        /// <returns>The disposable object used to cancel the scheduled action (best effort).</returns>
+        /// <exception cref="ArgumentNullException"><paramref name="action"/> is null.</exception>
+        public override IDisposable Schedule<TState>(TState state, Func<IScheduler, TState, IDisposable> action) {
+            return _innerScheduler.Schedule(state, action);
+        }
+
+        /// <summary>
+        /// Schedules an action to be executed after dueTime.
+        /// </summary>
+        /// <typeparam name="TState">The type of the state passed to the scheduled action.</typeparam>
+        /// <param name="state">State passed to the action to be executed.</param>
+        /// <param name="action">Action to be executed.</param>
+        /// <param name="dueTime">Relative time after which to execute the action.</param>
+        /// <returns>The disposable object used to cancel the scheduled action (best effort).</returns>
+        /// <exception cref="ArgumentNullException"><paramref name="action"/> is null.</exception>
+        public override IDisposable Schedule<TState>(TState state, TimeSpan dueTime, Func<IScheduler, TState, IDisposable> action) {
+            return _innerScheduler.Schedule(state, dueTime, action);
+        }
+
+        /// <summary>
+        /// Schedules a periodic piece of work.
+        /// </summary>
+        /// <typeparam name="TState">The type of the state passed to the scheduled action.</typeparam>
+        /// <param name="state">Initial state passed to the action upon the first iteration.</param>
+        /// <param name="period">Period for running the work periodically.</param>
+        /// <param name="action">Action to be executed, potentially updating the state.</param>
+        /// <returns>The disposable object used to cancel the scheduled recurring action (best effort).</returns>
+        /// <exception cref="ArgumentNullException"><paramref name="action"/> is null.</exception>
+        /// <exception cref="ArgumentOutOfRangeException"><paramref name="period"/> is less than zero.</exception>
+        public IDisposable SchedulePeriodic<TState>(TState state, TimeSpan period, Func<TState, TState> action) {
+            return _innerScheduler.SchedulePeriodic(state, period, action);
+        }
+
+        public override IStopwatch StartStopwatch() => _innerScheduler.StartStopwatch();
+    }
+}

Assets/Src/Scripts/Unity/Concurrency/ThreadPoolOnlyScheduler.cs.meta

@@ -17,7 +17,8 @@ public UnityMainThreadScheduler() {
             _scheduleAction = new Action<object>(Schedule);
         }
 
-        /// Gets the singleton instance of the Windows Runtime thread pool scheduler.
+        /// <summary>
+        /// Gets the singleton instance of the Unity main thread scheduler.
         /// </summary>
         public static UnityMainThreadScheduler Instance => _lazyDefault.Value;
 

Assets/Src/Scripts/Unity/Concurrency/UnityMainThreadScheduler.cs

@@ -0,0 +1,13 @@
+using System.Reactive.Concurrency;
+using System.Reactive.PlatformServices;
+using System.Reactive.Unity.Concurrency;
+
+namespace System.Reactive.Unity.InternalUtil {
+    sealed class UnityEnlightenmentProvider : CurrentPlatformEnlightenmentProvider {
+        public override T GetService<T>(object[] args) {
+            if (typeof(T) == typeof(IScheduler) && args != null && ((string)args[0]) == "ThreadPool")
+                return (T)(object)ThreadPoolOnlyScheduler.Instance;
+            return base.GetService<T>(args);
+        }
+    }
+}

Assets/Src/Scripts/Unity/InternalUtil/UnityEnlightenmentProvider.cs

@@ -3,11 +3,13 @@
 using UnityEngine;
 using System.Reactive.Concurrency;
 using System.Reactive.Unity.Concurrency;
+using System.Reactive.PlatformServices;
 
 namespace System.Reactive.Unity {
     public static class ReactiveUnity {
         public static void SetupPatches() {
             InitSchedulerDefaults();
+            InitCurrentPlatformEnlightenmentProvider();
             AddUnityEqualityComparer<Vector2>();
             AddUnityEqualityComparer<Vector3>();
             AddUnityEqualityComparer<Vector4>();
@@ -28,11 +30,15 @@ private static void InitSchedulerDefaults() {
 #if WEB_GL
             SchedulerDefaults.AsyncConversions = UnityMainThreadScheduler.Instance;
 #else
-            // SchedulerDefaults.AsyncConversions = ThreadPoolScheduler.Instance;
-            SchedulerDefaults.AsyncConversions = DefaultScheduler.Instance;
+            // SchedulerDefaults.AsyncConversions = DefaultScheduler.Instance;
+            SchedulerDefaults.AsyncConversions = ThreadPoolOnlyScheduler.Instance;
 #endif
         }
 
+        private static void InitCurrentPlatformEnlightenmentProvider() {
+            PlatformEnlightenmentProvider.Current = new UnityEnlightenmentProvider();
+        }
+
         private static void AddUnityEqualityComparer<T>() {
             var comparer = UnityEqualityComparer.GetDefault<T>();
             ReactiveProperty<T>.defaultEqualityComparer = comparer;

Assets/Src/Scripts/Unity/InternalUtil/UnityEnlightenmentProvider.cs.meta

@@ -30,6 +30,13 @@ public class ReactiveTest
         /// </summary>
         public const long Disposed = 1000;
 
+        [OneTimeSetUp]
+        public void SetupOnce() {
+            System.Threading.ThreadPool.GetAvailableThreads(out var availableWorker, out var availableIoCompletion);
+            System.Threading.ThreadPool.GetMaxThreads(out var maxWorker, out var maxIoCompletion);
+            UnityEngine.Debug.LogWarningFormat("Worker: {0} / {1} ; IoCompletion: {2} / {3}", availableWorker, maxWorker, availableIoCompletion, maxIoCompletion);
+        }
+
         /// <summary>
         /// Factory method for an OnNext notification record at a given time with a given value.
         /// </summary>

Assets/Src/Scripts/Unity/ReactiveUnity.cs

@@ -272,9 +272,9 @@ public void EventLoop_ScheduleActionDue() {
             Assert.True(sw.ElapsedMilliseconds > 180, "due " + sw.ElapsedMilliseconds);
 
             static void Aot() {
-                CurrentThreadScheduler.Instance.Schedule<(Func<(BasicProducer<int>, SingleAssignmentDisposable, IObserver<int>), IDisposable>, (BasicProducer<int>, SingleAssignmentDisposable, IObserver<int>))>(
+                _ = CurrentThreadScheduler.Instance.Schedule<(Func<(BasicProducer<int>, SingleAssignmentDisposable, IObserver<int>), IDisposable>, (BasicProducer<int>, SingleAssignmentDisposable, IObserver<int>))>(
                     (null, (null, null, null)), TimeSpan.Zero, null);
-                Aot();
+                _ = nameof(Aot);
             }
         }
 
@@ -303,9 +303,9 @@ public void EventLoop_ScheduleActionDueNested() {
             Assert.True(sw.ElapsedMilliseconds > 380, "due " + sw.ElapsedMilliseconds);
 
             static void Aot() {
-                CurrentThreadScheduler.Instance.Schedule<(Action<(Producer<int, Where<int>.Predicate._>, Where<int>.Predicate._)>, (Producer<int, Where<int>.Predicate._>, Where<int>.Predicate._))>(
+                _ = CurrentThreadScheduler.Instance.Schedule<(Action<(Producer<int, Where<int>.Predicate._>, Where<int>.Predicate._)>, (Producer<int, Where<int>.Predicate._>, Where<int>.Predicate._))>(
                     (null, (null, null)), TimeSpan.Zero, null);
-                Aot();
+                _ = nameof(Aot);
             }
         }
 

Assets/Tests/Microsoft.Reactive.Testing/ReactiveTest.cs

@@ -836,8 +836,8 @@ public void BufferWithCount_Disposed()
             );
 
             static void Aot() {
-                new TestScheduler().ScheduleAbsolute<((Buffer<int>.Ferry._, int), Action<(Buffer<int>.Ferry._, int)>)>(((null, default), null), 0L, null);
-                Aot();
+                _ = new TestScheduler().ScheduleAbsolute<((Buffer<int>.Ferry._, int), Action<(Buffer<int>.Ferry._, int)>)>(((null, default), null), 0L, null);
+                _ = nameof(Aot);
             }
         }
 
@@ -1176,12 +1176,6 @@ public void BufferWithTime_Basic_Same_Periodic_Error()
                 new TimerRun(200, 480) { 300, 400 }
             );
 #endif
-
-            static void Aot() {
-                // TOOD invisible class!
-                // new TestScheduler().ScheduleAbsolute<((Action<Action<TimeSpan>>, SingleAssignmentDisposable, Scheduler.InvokeRec2State<Action<Action<TimeSpan>>>), Action<(Action<Action<TimeSpan>>, SingleAssignmentDisposable, Scheduler.InvokeRec2State<Action<Action<TimeSpan>>>)>)>(((null, null, null), null), 0L, null);
-                Aot();
-            }
         }
 
         [Test]
@@ -1333,8 +1327,8 @@ public void BufferWithTime_TickWhileOnCompleted()
                 });
 
             static void Aot() {
-                new TestScheduler().ScheduleAbsolute<((Window<int>.Ferry._, Subject<int>), Action<(Window<int>.Ferry._, Subject<int>)>)>(((null, null), null), 0L, null);
-                Aot();
+                _ = new TestScheduler().ScheduleAbsolute<((Window<int>.Ferry._, Subject<int>), Action<(Window<int>.Ferry._, Subject<int>)>)>(((null, null), null), 0L, null);
+                _ = nameof(Aot);
             }
         }
 

Assets/Tests/Tests.System.Reactive/Tests/Concurrency/EventLoopSchedulerTest.cs

@@ -11,11 +11,22 @@
 using ReactiveTests.Dummies;
 using NUnit.Framework;
 using UnityEngine.TestTools;
+using UniRx.Tests;
+using System.Reactive.Unity;
 
 namespace ReactiveTests.Tests
 {
     public class DelaySubscriptionTest : ReactiveTest
     {
+        [SetUp]
+        public void Init() {
+            TestUtil.SetSchedulerForImport();
+        }
+
+        [TearDown]
+        public void Dispose() {
+            ReactiveUnity.SetupPatches();
+        }
 
         [Test]
         public void DelaySubscription_ArgumentChecking()