Perché questo sovraccarico Observable.Generate causa una perdita di memoria? [Utilizzo dell'orario <15 ms]

Question

Il seguente codice Rx.NET utilizzerà circa 500 MB di memoria dopo circa 10 secondi sulla mia macchina.

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(); 

Se uso il Observable.Generate di sovraccarico senza Func<int, TimeSpan> parametro mio utilizzo della memoria altipiani a 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(); 

Sembra essere solo un problema quando si utilizza SelectMany() o unione) metodi di estensione (.

Answer 1

Questo è un problema relativo all'utilità di pianificazione predefinita.

Con la versione TimeSpan lo scheduler è DefaultScheduler. Senza TimeSpan è CurrentThreadScheduler.

Quindi, per il generare in base al tempo è molto rapidamente cercando di pianificare tutte le operazioni e fondamentalmente si accumula una massiccia coda di eventi in attesa di essere eseguita. Quindi usa un carico di memoria.

Con il generare non basato sul tempo utilizza il thread corrente in modo che produca e consumi ogni valore generato in serie e quindi utilizzi pochissima memoria.

Oh, e questa non è una perdita di memoria. È solo la normale operazione se si tenta di pianificare un numero infinito di valori più velocemente di quanto possano essere consumati.

---

Ho decompilato il codice per capire quali scheduler sono stati utilizzati.

Ecco il decompilare non basato tempo:

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; 
    } 
} 

I metodi di cui sopra sono da Observable, QueryLanguage e SchedulerDefaults rispettivamente.