reactive/Rx.NET/Source/tests/Tests.System.Reactive/Stress/Linq/FromEvent.cs

// Licensed to the .NET Foundation under one or more agreements.
// The .NET Foundation licenses this file to you under the MIT License.
// See the LICENSE file in the project root for more information. 

#if STRESS
using System;
using System.Collections.Generic;
using System.Linq;
using System.Reactive.Disposables;
using System.Reactive.Linq;
using System.Reflection;
using System.Threading;

namespace ReactiveTests.Stress.Linq
{
    public class FromEvent
    {
        private static Lazy<Random> s_rand = new Lazy<Random>();

        /// <summary>
        /// Multiple threads are subscribing to a FromEventPattern sequence and disposing their subscriptions.
        /// While this is going on, one consumer does not want to be disturbed while receiving the sequence.
        /// 
        /// Runs a set of combinations of the RefCount_* tests.
        /// </summary>
        public static void RefCount_Mix()
        {
            Console.Title = MethodInfo.GetCurrentMethod().Name + " - 0% complete";

            for (int i = 1; i <= 100; i++)
            {
                var repeatCount = 10;

                foreach (var msgCount in new[] { 100, 1000, 10000, 100000 })
                {
                    // concurrency level {10, 20, ..., 100}
                    RefCount_ConcurrencyLevel_Linear(msgCount, repeatCount, 10, 100, 10);

                    // concurrency level {100, 200, ..., 1000}
                    RefCount_ConcurrencyLevel_Linear(msgCount, repeatCount, 100, 1000, 100);

                    // concurrency level {1, 2, 4, ..., 65536}
                    RefCount_ConcurrencyLevel_Exponential(msgCount, repeatCount, 1, 65536, 2);
                }

                foreach (var maxMsgCount in new[] { 10, 100, 1000, 10000, 100000 })
                {
                    foreach (var maxConcurrency in new[] { 10, 100, 1000, 10000, 100000 })
                    {
                        RefCount_Rand(repeatCount, maxMsgCount, maxConcurrency);
                    }
                }

                Console.Title = MethodInfo.GetCurrentMethod().Name + " - " + i + "% complete";
            }
        }

        /// <summary>
        /// Multiple threads are subscribing to a FromEventPattern sequence and disposing their subscriptions.
        /// While this is going on, one consumer does not want to be disturbed while receiving the sequence.
        /// Subscriptions are happening on the ThreadPool, possibly causing (expected) time gaps.
        /// 
        /// Runs a set of combinations of the RefCount_* tests.
        /// </summary>
        public static void RefCountWithPost_Mix()
        {
            Console.Title = MethodInfo.GetCurrentMethod().Name + " - 0% complete";

            for (int i = 1; i <= 100; i++)
            {
                var repeatCount = 10;

                foreach (var msgCount in new[] { 100, 1000, 10000, 100000 })
                {
                    // concurrency level {10, 20, ..., 100}
                    RefCountWithPost_ConcurrencyLevel_Linear(msgCount, repeatCount, 10, 100, 10);

                    // concurrency level {100, 200, ..., 1000}
                    RefCountWithPost_ConcurrencyLevel_Linear(msgCount, repeatCount, 100, 1000, 100);

                    // concurrency level {1, 2, 4, ..., 65536}
                    RefCountWithPost_ConcurrencyLevel_Exponential(msgCount, repeatCount, 1, 65536, 2);
                }

                foreach (var maxMsgCount in new[] { 10, 100, 1000, 10000, 100000 })
                {
                    foreach (var maxConcurrency in new[] { 10, 100, 1000, 10000, 100000 })
                    {
                        RefCountWithPost_Rand(repeatCount, maxMsgCount, maxConcurrency);
                    }
                }

                Console.Title = MethodInfo.GetCurrentMethod().Name + " - " + i + "% complete";
            }
        }

        /// <summary>
        /// Multiple threads are subscribing to a FromEventPattern sequence and disposing their subscriptions.
        /// While this is going on, one consumer does not want to be disturbed while receiving the sequence.
        /// 
        /// Uses random parameters for the number of messages and the level of concurrency.
        /// </summary>
        /// <param name="n">Number of iterations.</param>
        /// <param name="maxN">Maximum number of message.</param>
        /// <param name="maxM">Maximum level of concurrency.</param>
        public static void RefCount_Rand(int n, int maxN, int maxM)
        {
            RefCount_(RefCount_Rand_Params(n, maxN, maxM));
        }

        /// <summary>
        /// Multiple threads are subscribing to a FromEventPattern sequence and disposing their subscriptions.
        /// While this is going on, one consumer does not want to be disturbed while receiving the sequence.
        /// Subscriptions are happening on the ThreadPool, possibly causing (expected) time gaps.
        /// 
        /// Uses random parameters for the number of messages and the level of concurrency.
        /// </summary>
        /// <param name="n">Number of iterations.</param>
        /// <param name="maxN">Maximum number of message.</param>
        /// <param name="maxM">Maximum level of concurrency.</param>
        public static void RefCountWithPost_Rand(int n, int maxN, int maxM)
        {
            RefCountWithPost_(RefCount_Rand_Params(n, maxN, maxM));
        }

        private static IEnumerable<Tuple<int, int>> RefCount_Rand_Params(int n, int maxN, int maxM)
        {
            for (int i = 0; i < n; i++)
            {
                var N = s_rand.Value.Next(1, maxN);
                var M = s_rand.Value.Next(1, maxM);

                yield return new Tuple<int, int>(N, M);
            }
        }

        /// <summary>
        /// Multiple threads are subscribing to a FromEventPattern sequence and disposing their subscriptions.
        /// While this is going on, one consumer does not want to be disturbed while receiving the sequence.
        /// 
        /// Uses linear increments for the concurrency level.
        /// </summary>
        /// <param name="N">Number of messages.</param>
        /// <param name="n">Number of iterations.</param>
        /// <param name="min">Minimum level of concurrency.</param>
        /// <param name="max">Maximum level of concurrency.</param>
        /// <param name="step">Additive step size to increase level of concurrency.</param>
        public static void RefCount_ConcurrencyLevel_Linear(int N, int n, int min, int max, int step)
        {
            RefCount_(RefCount_ConcurrencyLevel_Linear_Params(N, n, min, max, step));
        }

        /// <summary>
        /// Multiple threads are subscribing to a FromEventPattern sequence and disposing their subscriptions.
        /// While this is going on, one consumer does not want to be disturbed while receiving the sequence.
        /// Subscriptions are happening on the ThreadPool, possibly causing (expected) time gaps.
        /// 
        /// Uses linear increments for the concurrency level.
        /// </summary>
        /// <param name="N">Number of messages.</param>
        /// <param name="n">Number of iterations.</param>
        /// <param name="min">Minimum level of concurrency.</param>
        /// <param name="max">Maximum level of concurrency.</param>
        /// <param name="step">Additive step size to increase level of concurrency.</param>
        public static void RefCountWithPost_ConcurrencyLevel_Linear(int N, int n, int min, int max, int step)
        {
            RefCountWithPost_(RefCount_ConcurrencyLevel_Linear_Params(N, n, min, max, step));
        }

        private static IEnumerable<Tuple<int, int>> RefCount_ConcurrencyLevel_Linear_Params(int N, int n, int min, int max, int step)
        {
            for (int i = 0; i < n; i++)
            {
                for (int M = min; M <= max; M += step)
                {
                    yield return new Tuple<int, int>(N, M);
                }
            }
        }

        /// <summary>
        /// Multiple threads are subscribing to a FromEventPattern sequence and disposing their subscriptions.
        /// While this is going on, one consumer does not want to be disturbed while receiving the sequence.
        /// 
        /// Uses exponential increments for the concurrency level.
        /// </summary>
        /// <param name="N">Number of messages.</param>
        /// <param name="n">Number of iterations.</param>
        /// <param name="min">Minimum level of concurrency.</param>
        /// <param name="max">Maximum level of concurrency.</param>
        /// <param name="step">Multiplicative step size to increase level of concurrency.</param>
        public static void RefCount_ConcurrencyLevel_Exponential(int N, int n, int min, int max, int step)
        {
            RefCount_(RefCount_ConcurrencyLevel_Exponential_Params(N, n, min, max, step));
        }

        /// <summary>
        /// Multiple threads are subscribing to a FromEventPattern sequence and disposing their subscriptions.
        /// While this is going on, one consumer does not want to be disturbed while receiving the sequence.
        /// Subscriptions are happening on the ThreadPool, possibly causing (expected) time gaps.
        /// 
        /// Uses exponential increments for the concurrency level.
        /// </summary>
        /// <param name="N">Number of messages.</param>
        /// <param name="n">Number of iterations.</param>
        /// <param name="min">Minimum level of concurrency.</param>
        /// <param name="max">Maximum level of concurrency.</param>
        /// <param name="step">Multiplicative step size to increase level of concurrency.</param>
        public static void RefCountWithPost_ConcurrencyLevel_Exponential(int N, int n, int min, int max, int step)
        {
            RefCountWithPost_(RefCount_ConcurrencyLevel_Exponential_Params(N, n, min, max, step));
        }

        private static IEnumerable<Tuple<int, int>> RefCount_ConcurrencyLevel_Exponential_Params(int N, int n, int min, int max, int step)
        {
            for (int i = 0; i < n; i++)
            {
                for (int M = min; M <= max; M *= step)
                {
                    yield return new Tuple<int, int>(N, M);
                }
            }
        }

        private static void RefCount_(IEnumerable<Tuple<int, int>> parameters)
        {
            foreach (var p in parameters)
            {
                var N = p.Item1;
                var M = p.Item2;

                Console.Write("N = {0}, M = {1} - ", N, M);

                var bar = new Bar();

                var foo = Observable.FromEventPattern<FooEventArgs>(h => { Console.Write("+"); bar.Foo += h; }, h => { bar.Foo -= h; Console.Write("-"); });

                var res = new List<int>();
                var n = 0;
                var e = new ManualResetEvent(false);

                var cd = new CountdownEvent(M * 2);
                for (int i = 0; i < M; i++)
                {
                    var f = new SingleAssignmentDisposable();

                    ThreadPool.QueueUserWorkItem(_ =>
                    {
                        f.Disposable = foo.Subscribe(__ => { Console.Write("!"); });
                        cd.Signal();
                    });

                    ThreadPool.QueueUserWorkItem(_ =>
                    {
                        f.Dispose();
                        cd.Signal();
                    });
                }

                Console.Write("{SB}");

                var d = foo.Subscribe(x =>
                {
                    //Console.Write("&");

                    if (++n == N)
                        e.Set();

                    res.Add(x.EventArgs.Qux);
                });

                Console.Write("{SE}");

                var t = new Thread(() =>
                {
                    Console.Write("{TB}");

                    for (int i = 0; i < N; i++)
                        bar.OnFoo(i);

                    Console.Write("{TE}");
                });

                t.Start();
                t.Join();

                cd.Wait();

                e.WaitOne();
                d.Dispose();

                if (!res.SequenceEqual(Enumerable.Range(0, N)))
                {
                    Console.WriteLine("Panic!");
                    break;
                }

                Console.WriteLine(".");
            }
        }

        private static void RefCountWithPost_(IEnumerable<Tuple<int, int>> parameters)
        {
            var worker = new Thread(() =>
            {
                SynchronizationContext.SetSynchronizationContext(new MySyncCtx());

                foreach (var p in parameters)
                {
                    var N = p.Item1;
                    var M = p.Item2;

                    Console.Write("N = {0}, M = {1} - ", N, M);

                    var bar = new Bar();

                    var foo = Observable.FromEventPattern<FooEventArgs>(h => { /*Console.Write("+");*/ bar.Foo += h; }, h => { bar.Foo -= h; /*Console.Write("-"); */});

                    var e = new ManualResetEvent(false);

                    var cd = new CountdownEvent(M * 2);
                    for (int i = 0; i < M; i++)
                    {
                        var f = new SingleAssignmentDisposable();

                        ThreadPool.QueueUserWorkItem(_ =>
                        {
                            f.Disposable = foo.Subscribe(__ => { /*Console.Write("!");*/ });
                            cd.Signal();
                        });

                        ThreadPool.QueueUserWorkItem(_ =>
                        {
                            f.Dispose();
                            cd.Signal();
                        });
                    }

                    var hasObserved = 0;

                    Console.Write("{SB}");

                    var d = foo.Subscribe(x =>
                    {
                        //
                        // [on BARTDE-M6500 with CPU and RAM pressure]
                        //
                        // Up to 8K concurrent observers, we typically don't see a time gap (expected worst-case behavior).
                        // The code below uses an event to check the desired behavior of eventually tuning in to the event stream.
                        //
                        Console.Write("&" + x.EventArgs.Qux);
                        e.Set();
                        Interlocked.Exchange(ref hasObserved, 1);
                    });

                    Console.Write("{SE}");

                    var t = new Thread(() =>
                    {
                        Console.Write("{TB}");

                        var i = 0;
                        while (Thread.VolatileRead(ref hasObserved) == 0)
                            bar.OnFoo(i++);

                        Console.Write("{TE}");
                    });

                    t.Start();
                    t.Join();

                    cd.Wait();

                    e.WaitOne();
                    d.Dispose();

                    Console.WriteLine(".");
                }
            });

            worker.Start();
            worker.Join();
        }

        class Bar
        {
            public event EventHandler<FooEventArgs> Foo;

            public void OnFoo(int x)
            {
                var foo = Foo;
                if (foo != null)
                    foo(this, new FooEventArgs { Qux = x });
            }
        }

        class FooEventArgs : EventArgs
        {
            public int Qux { get; set; }
        }

        class MySyncCtx : SynchronizationContext
        {
            public override void Post(SendOrPostCallback d, object state)
            {
                ThreadPool.QueueUserWorkItem(_ =>
                {
                    d(state);
                });
            }
        }
    }
}
#endif