Memory leak on iOS when there's a chain of RxApp.MainThreadScheduler's

[PureWeen]

Do you want to request a feature or report a bug? Bug or possibly expected behavior if whoever is writing code is just doing a bad job at it

What is the current behavior? I've been having an issue on iOS where after a couple of hours of sitting there I start getting High Memory warnings from my app. I hooked the app up to Xamarin Profiler and noticed that the allocation of CoreFoundation.DispatchQueue (which is used by the NSRunloopScheduler) just increases forever. At first I was thinking this was a bug in Xamarin https://bugzilla.xamarin.com/show_bug.cgi?id=60496

Turns out I was just doing a bad job running my tests. But it at least made me understand why I would see leaks. Not sure specifically what the issue is with how RxUI uses it but something about having the schedulers chained is causing the action to not dispatch and/or release resources

If the current behavior is a bug, please provide the steps to reproduce and if possible a minimal demo of the problem

You can reproduce the issue here.

https://github.com/PureWeen/DispatchQueueMemoryLeak

        public ViewControllerViewModel()
        {
            _tick =
                this.WhenAnyValue(x => x.SomeRandomProperty)
                    .Select(_ => Observable.Interval(TimeSpan.FromSeconds(1)).Select(__ => DateTimeOffset.Now))
                    .Switch()
                    .ToProperty(this, x => x.Tick, scheduler: RxApp.MainThreadScheduler);

            //Having this here is what causes the memory leak
            this.WhenAnyValue(x => x.Tick)
                .ObserveOn(RxApp.MainThreadScheduler)
                .Subscribe(_ => SomeRandomProperty = $"{System.GC.GetTotalMemory(false)}");
        }

The TextBox on the screen is just taking a sample from System.GC.GetTotalMemory(false) every second

You can also run this application through the profiler on iOS and you'll see the allocations of CoreFoundation.DipatchQueue just increase forever

What is the expected behavior?

For the memory not to leak

What is the motivation / use case for changing the behavior? It's good when memory doesn't leak

Which versions of ReactiveUI, and which platform / OS are affected by this issue? Did this work in previous versions of ReativeUI? Please also test with the latest stable and development snapshot

The latest v8 Alpha build but I'm pretty sure this issue is also present on v7

Other information (e.g. stacktraces, related issues, suggestions how to fix)

[fschwiet]

I wonder if this could be related to #1990? That issue is based on a concern I have with how Mac's NSRunloopScheduler is implemented (which is what RxApp.MainThreadScheduler is) referencing documentation in Apple docs specifically talks about a thing not being removed from a run loop (maybe that is the leaked resource?)

If you have a repro, I have an alternative to the NSRunloopScheduler you could try. DispatchQueueSchedulerAdapter would be constructed with DispatchQueue.MainQueue and whatever scheduler you want to use to handling queueing things that have a delay (you could use an EventLoopScheduler). Then drop that in in place of RxApp.MainThreadScheduler.

 public class DispatchQueueSchedulerAdapter : IScheduler
    {
        private readonly IScheduler _scheduler;
        private readonly Action<Action> _dispatchQueue;

        public DispatchQueueSchedulerAdapter(IScheduler scheduler, Action<Action> dispatchQueue)
        {
            _scheduler = scheduler;
            _dispatchQueue = dispatchQueue;
        }

        public IDisposable Schedule<TState>(TState state, Func<IScheduler, TState, IDisposable> action)
        {
            return _scheduler.Schedule(state, (_1, _2) => RunInDispatcherAndWrapDispose(() => action(_1, _2)));
        }

        public IDisposable Schedule<TState>(TState state, TimeSpan dueTime, Func<IScheduler, TState, IDisposable> action)
        {
            return _scheduler.Schedule(state, dueTime, (_1, _2) => RunInDispatcherAndWrapDispose(() => action(_1, _2)));
        }

        public IDisposable Schedule<TState>(TState state, DateTimeOffset dueTime, Func<IScheduler, TState, IDisposable> action)
        {
            return _scheduler.Schedule(state, dueTime, (_1, _2) => RunInDispatcherAndWrapDispose(() => action(_1, _2)));
        }

        public DateTimeOffset Now => _scheduler.Now;

        private IDisposable RunInDispatcherAndWrapDispose(Func<IDisposable> action)
        {
            //  We're going to schedule the action to run in the CoreFoundation DispatchQueue.
            //  We'll return a disposable that prevents the action from running if it hasn't yet,
            //  or disposes the result of the action if it already has.

            var wasDisposed = false;
            IDisposable wrappedDisposable = null;

            _dispatchQueue(() =>
            {
                lock (this)
                {
                    if (wasDisposed)
                        return;
                }

                var immediateDisposable = action();

                lock (this)
                {
                    if (!wasDisposed)
                    {
                        wrappedDisposable = immediateDisposable;
                        immediateDisposable = null;
                    }
                }

                immediateDisposable?.Dispose();
            });

            return Disposable.Create(() =>
            {
                IDisposable toDispose = null;
                lock (this)
                {
                    wasDisposed = true;
                    toDispose = wrappedDisposable;
                    wrappedDisposable = null;
                }

                toDispose?.Dispose();
            });
        }

with tests:

    public class TestDispatchQueueSchedulerAdapter
    {
        private TestScheduler testScheduler;
        private FakeDispatchQueue fakeDispatchQueue;
        private IScheduler schedulerBeingTested;
        private int scheduledCallbackCalledCount;
        private string expectedState;
        private TrackedDispose scheduledCallbackDisposable;

        public TestDispatchQueueSchedulerAdapter()
        {
            testScheduler = new TestScheduler();

            scheduledCallbackCalledCount = 0;
            expectedState = "foo";
            scheduledCallbackDisposable = new TrackedDispose();

            fakeDispatchQueue = new FakeDispatchQueue();
            schedulerBeingTested = new DispatchQueueSchedulerAdapter(testScheduler, fakeDispatchQueue.DispatchAsync);
        }

        private IDisposable SetupSingleShotTimer()
        {
            return schedulerBeingTested.Schedule(expectedState, TimeSpan.FromSeconds(1), (sch, st) =>
            {
                st.Should().Be(expectedState);
                sch.Should().Be(testScheduler);
                scheduledCallbackCalledCount++;
                return scheduledCallbackDisposable;
            });
        }

        [Fact]
        public void When_schedule_is_cancelled_early_the_callback_will_not_run()
        {
            var timerDisposable = SetupSingleShotTimer();

            testScheduler.AdvanceBy(TimeSpan.FromSeconds(0.999).Ticks);

            fakeDispatchQueue.Queue.Count.Should().Be(0);
            scheduledCallbackCalledCount.Should().Be(0);

            timerDisposable.Dispose();

            fakeDispatchQueue.Queue.Count.Should().Be(0);
            scheduledCallbackCalledCount.Should().Be(0);

            testScheduler.AdvanceBy(TimeSpan.FromSeconds(100).Ticks);

            fakeDispatchQueue.Queue.Count.Should().Be(0);
            scheduledCallbackCalledCount.Should().Be(0);
        }

        [Fact]
        public void When_schedule_is_cancelled_after_timeout_but_before_dispatch_the_callback_will_not_run()
        {
            var timerDisposable = SetupSingleShotTimer();

            testScheduler.AdvanceBy(TimeSpan.FromSeconds(1).Ticks);

            fakeDispatchQueue.Queue.Count.Should().Be(1);
            scheduledCallbackCalledCount.Should().Be(0);
            scheduledCallbackDisposable.WasDisposed.Should().Be(false);

            timerDisposable.Dispose();
            fakeDispatchQueue.RunQueue();

            fakeDispatchQueue.Queue.Count.Should().Be(0);
            scheduledCallbackCalledCount.Should().Be(0);
            scheduledCallbackDisposable.WasDisposed.Should().Be(false);
        }

        [Fact]
        public void When_schedule_is_cancelled_after_timeout_and_dispatch_the_callback_runs_and_the_callback_disposable_is_disposed()
        {
            var timerDisposable = SetupSingleShotTimer();

            testScheduler.AdvanceBy(TimeSpan.FromSeconds(1).Ticks);
            fakeDispatchQueue.Queue.Count.Should().Be(1);
            fakeDispatchQueue.RunQueue();

            scheduledCallbackCalledCount.Should().Be(1);
            scheduledCallbackDisposable.WasDisposed.Should().Be(false);

            timerDisposable.Dispose();

            scheduledCallbackCalledCount.Should().Be(1);
            scheduledCallbackDisposable.WasDisposed.Should().Be(true);
        }
    }

    internal class TrackedDispose : IDisposable
    {
        private int _disposeCount;

            public bool WasDisposed
            {
                get
                {
                    if (_disposeCount == 0)
                        return false;
                    if (_disposeCount == 1)
                        return true;
                    
                    throw new Exception("Invalid dispose count: " + _disposeCount);
                }
            }

            public void Dispose()
        {
            _disposeCount++;
        }
    }

    internal class FakeDispatchQueue
    {
        public List<Action> Queue = new List<Action>();

        public void DispatchAsync(Action callback)
        {
            Queue.Add(callback);
        }

        public void RunQueue()
        {
            foreach (var callback in Queue)
                callback();

            Queue.Clear();
        }
    }

[fschwiet]

My comment described MacOS, but the same issue exists for iOS as NSRunloopScheduler is used on iOS as well.

[glennawatson]

I believe @RLittlesII is going to look at getting a PR for you

[fschwiet]

Just noting that we have a use for DispatchQueueSchedulerAdapter where we need a scheduler to run on a different DispatchQueue. So it'd be good to expose it in ReactiveX for others.

[glennawatson]

There are some changes I would make to the code above. Your can use the interlocked over a lock for thread safety. The dispose checks for 0 and 1 but not values above etc which the interlocked can get around. Etc for your knowledge @rlittlesii

[fschwiet]

Thanks for taking a look.

Your can use the interlocked over a lock for thread safety.

This sounds good as it would use less resources. If there is a behavioral bug though please let me know as I'm passing code using this to our test team.

The dispose checks for 0 and 1 but not values above

Are you referring to TrackedDispose? That's just something used for tests, and its asserting if the dispose count is >1 so the tests can warn if extra disposes are happening. As the test is written all that code should only run on the same thread (it's using TestScheduler, which I believe runs things in the same thread.)

[cabauman]

@RLittlesII @glennawatson do you know if a PR was ever made for this?