为什么这个 Observable.Generate 重载会导致内存泄漏?[使用时间跨度 <15ms] [英] Why does this Observable.Generate overload cause a memory leak? [Using Timespan < 15ms]

问题描述

以下 Rx.NET 代码将在我的机器上大约 10 秒后使用大约 500 MB 的内存.

The following Rx.NET code will use up about 500 MB of memory after about 10 seconds on my machine.

var stream =
    Observable.Range(0, 10000)
              .SelectMany(i => Observable.Generate(
                  0, 
                  j => true, 
                  j => j + 1, 
                  j => new { N = j },
                  j => TimeSpan.FromMilliseconds(1)));

stream.Subscribe();

如果我在没有 Func 参数的情况下使用 Observable.Generate 重载，我的内存使用量将达到 35 MB.

If I use the Observable.Generate overload without a Func<int, TimeSpan> parameter my memory usage plateaus at 35 MB.

var stream =
    Observable.Range(0, 10000)
              .SelectMany(i => Observable.Generate(
                  0,
                  j => true,
                  j => j + 1,
                  j => new { N = j }));
                  // j => TimeSpan.FromMilliseconds(1))); ** Removed! **

stream.Subscribe();

似乎只有在使用 SelectMany() 或 Merge() 扩展方法时才会出现问题.

It seems to only be a problem when using SelectMany() or Merge() extension methods.

推荐答案

这是使用哪个默认调度程序的问题.

This is an issue of which default scheduler is used.

对于 TimeSpan 版本，调度程序是 DefaultScheduler.没有 TimeSpan 它是 CurrentThreadScheduler.

With the TimeSpan version the scheduler is the DefaultScheduler. Without TimeSpan it is CurrentThreadScheduler.

因此，对于基于时间的生成，它会非常迅速地尝试安排所有操作，并且基本上会建立大量等待执行的事件队列.因此它使用了大量内存.

So, for the time-based generate it's very rapidly trying to schedule all of the operations and basically builds up a massive queue of events waiting to be executed. Thus it uses a load of memory.

对于非基于时间的生成，它使用当前线程，因此它将连续生成和消耗每个生成的值，因此使用很少的内存.

With the non-time-based generate it's using the current thread so it will produce and consume each generated value in series and thus use very little memory.

哦，这不是内存泄漏.如果您尝试以比它们可以消耗的速度更快地调度无限数量的值，这只是正常操作.

Oh, and this isn't a memory leak. It's just the normal operation if you try to schedule an infinite number of values faster than they can be consumed.

我反编译了代码以确定使用了哪些调度程序.

I decompiled the code to work out which schedulers were used.

这是非基于时间的反编译:

Here's the non-time-based decompile:

public static IObservable<TResult> Generate<TState, TResult>(TState initialState, Func<TState, bool> condition, Func<TState, TState> iterate, Func<TState, TResult> resultSelector)
{
    if (condition == null)
        throw new ArgumentNullException("condition");
    if (iterate == null)
        throw new ArgumentNullException("iterate");
    if (resultSelector == null)
        throw new ArgumentNullException("resultSelector");
    return Observable.s_impl.Generate<TState, TResult>(initialState, condition, iterate, resultSelector);
}

public virtual IObservable<TResult> Generate<TState, TResult>(TState initialState, Func<TState, bool> condition, Func<TState, TState> iterate, Func<TState, TResult> resultSelector)
{
    return (IObservable<TResult>)new Generate<TState, TResult>(initialState, condition, iterate, resultSelector, SchedulerDefaults.Iteration);
}

internal static IScheduler Iteration
{
    get
    {
        return (IScheduler)CurrentThreadScheduler.Instance;
    }
}

以上方法分别来自Observable、QueryLanguage和SchedulerDefaults.

The above methods are from Observable, QueryLanguage, and SchedulerDefaults respectively.

这篇关于为什么这个 Observable.Generate 重载会导致内存泄漏?[使用时间跨度 <15ms]的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！