c# - 使用 TimeSpan 选择器的 Observable.Generate 似乎会泄漏内存 [当使用 TimeSpan > 15ms 时]

Question

我正在研究 Observable.Generate 的使用，以使用 msdn 网站上的示例作为起点来创建按时间间隔采样的结果序列。

以下没有 TimeSpan 选择器的代码不会出现内存泄漏:

IObservable<string> obs = Observable.Generate(initialState: 1,
                                              condition: x => x < 1000,
                                              iterate: x => x + 1,
                                              resultSelector: x => x.ToString());
obs.Subscribe(x => Console.WriteLine(x));

但是，以下带有 TimeSpan 选择器的代码会出现内存泄漏:

TimeSpan timeSpan = TimeSpan.FromSeconds(1);
IObservable<string> obs = Observable.Generate(initialState: 1,
                                              condition: x => x < 1000,
                                              iterate: x => x + 1,
                                              resultSelector: x => x.ToString(),
                                              timeSelector: x => timeSpan);
obs.Subscribe(x => Console.WriteLine(x));

例如，这个玩具应用程序将使用 VS 2015 Community 附带的 Memory Profiler 快速显示内存泄漏:

using System;
using System.Reactive.Linq;

namespace Sample
{
    public class Program
    {
        static void Main()
        {
            IObservable<string> obs = Observable.Generate(1, x => x < 1000*1000, x => x + 1, x => x.ToString(), x => TimeSpan.FromMilliseconds(500));
            obs.Subscribe(x => { /*Do nothing but simply run the observable*/ });
            Console.ReadLine();
        }
    }
}

内存泄漏是不断增长的集合:

System.Reactive.Disposables StableCompositeDisposable.Binary
System.Reactive.Disposables SingleAssignmentDisposable

我是否错误地使用了这个 API？我应该期望内存增长还是 Reactive 的错误？

Answer 1

这对我来说确实像是一个错误 - 或者至少是 DefaultScheduler 的“递归”调度实现中的困惑/不受欢迎的行为(它不是真正的递归，我说的是将调度程序本身传递给计划操作的过载这样您就可以安排继续)。

您看到的一次性用品是通过调用 DefaultScheduler.Schedule 方法创建的(此处为第 71 行:https://github.com/Reactive-Extensions/Rx.NET/blob/master/Rx.NET/Source/System.Reactive.Core/Reactive/Concurrency/DefaultScheduler.cs)。

此处发现此问题的其他尝试失败的原因有几个。首先，一次性用品最终会被处理掉——但只有在生成 OnCompletes 时才会被处理掉。或 OnErrors ，此时 System.Reactive.AnonymousSafeObserver<T>当您订阅它时由 Generate 返回，它是否已清理。

其次，如果您使用短的 TimeSpan (记住 .NET 定时器的最小分辨率无论如何是 15 毫秒)然后 Rx 将优化定时器的使用并调用 QueueUserWorkItem没有使用计时器，因此永远不会创建这些一次性用品。

如果深入研究 Generate 的实现 (https://github.com/Reactive-Extensions/Rx.NET/blob/master/Rx.NET/Source/System.Reactive.Linq/Reactive/Linq/Observable/Generate.cs)，您会发现它通过了 IDisposable由对 Schedule 的初始调用返回，将其返回给观察者，观察者会一直坚持到错误/完成。这防止了整个递归调用链的收集 - 并且意味着如果您确实需要取消，或者当清理发生时，只有在那时才会处理每个计划的操作的一次性。

您可以在下面直接使用 DefaultScheduler 的代码中看到相同的效果 - 对 cancel 的引用最后一行足以导致泄漏。确保使用发布版本，否则编译器将保留取消直到方法结束。

// ensure you are using a release build of this code
ManualResetEvent mre = new ManualResetEvent();
IDisposable cancel;
int maxCount = 20;

TimeSpan timeSpan = TimeSpan.FromSeconds(1);

Func<IScheduler, int, IDisposable> recurse = null;
recurse = (self, state) =>
{
    Console.WriteLine(state);

    if (state == maxCount)
    {
        mre.Set();
        return Disposable.Empty;
    }

    return self.Schedule(state + 1, timeSpan, recurse);
};

cancel = Scheduler.Default.Schedule(1, timeSpan, recurse);

mre.WaitOne();

// uncomment the following line, and you'll get the same leak
// leave it commented, and cancel reference is GC'd early and there's no leak
// if(cancel == null) Console.WriteLine("Hang on to cancel");

我使用 Jetbrains dotMemory API 进行内存转储以得出结论 - 我已经去除了上面那些 API 调用的代码，但如果您有该产品，这里有完整的要点，您将能够看到非常清楚地取消注释最后一行的影响:https://gist.github.com/james-world/f20377ea610fb8fc0ee811d27f7a837c或者，您可以使用 MS 分析器 API - 我目前还没有将其分页到我的大脑工作集中!