Dlaczego to obserwowalne. Generowanie przeciążenia powoduje wyciek pamięci? [Używanie Timespan <15ms]

Question

Następujący kod Rx.NET zużyje około 500 MB pamięci po około 10 sekundach na moim komputerze.

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

Jeśli używam przeciążenie Observable.Generate bez Func<int, TimeSpan> parametru moim wykorzystanie pamięci płaskowyże na 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(); 

Wydaje się, że jest to problem tylko przy użyciu metod rozszerzania SelectMany() lub Merge().

Answer 1

Jest to problem, którego domyślny harmonogram jest używany.

W wersji TimeSpan harmonogramem jest DefaultScheduler. Bez TimeSpan jest to CurrentThreadScheduler.

Tak więc, w przypadku generowania czasowego bardzo szybko próbuje zaplanować wszystkie operacje i zasadniczo buduje ogromną kolejkę zdarzeń oczekujących na wykonanie. W ten sposób wykorzystuje obciążenie pamięci.

W przypadku generowania nie w oparciu o czas wykorzystuje bieżący wątek, dzięki czemu generuje i zużywa każdą wygenerowaną wartość szeregowo, a tym samym zużywa bardzo mało pamięci.

Aha, i to nie jest przeciek pamięci. To zwykła operacja, jeśli próbujesz zaplanować nieskończoną liczbę wartości szybciej, niż można je spożyć.

---

Dekompilowałem kod, aby ustalić, które programy planujące zostały użyte.

Oto dekompilacji nie czas na bazie:

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

Powyższe metody są od Observable, QueryLanguage i SchedulerDefaults odpowiednio.