Tại sao quá tải Observable.Generate này gây ra rò rỉ bộ nhớ? [Sử dụng Timespan <15ms]

Question

Mã Rx.NET sau sẽ sử dụng hết khoảng 500 MB bộ nhớ sau khoảng 10 giây trên máy tính của tôi.

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

Nếu tôi sử dụng quá tải Observable.Generate mà không có một tham số Func<int, TimeSpan> cao nguyên sử dụng bộ nhớ của tôi lúc 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(); 

Dường như chỉ là vấn đề khi sử dụng phương pháp mở rộng SelectMany() hoặc Merge().

Answer 1

Đây là vấn đề mà trình lập lịch biểu mặc định được sử dụng.

Với phiên bản TimeSpan, trình lên lịch là DefaultScheduler. Không có TimeSpan, nó là CurrentThreadScheduler.

Vì vậy, đối với việc tạo thời gian, nó rất nhanh chóng cố gắng lên lịch cho tất cả các hoạt động và cơ bản xây dựng một hàng đợi lớn các sự kiện đang chờ để được thực hiện. Vì vậy nó sử dụng một tải bộ nhớ.

Với việc tạo dựa trên thời gian không, nó sử dụng chuỗi hiện tại để nó tạo và tiêu thụ từng giá trị được tạo trong chuỗi và do đó sử dụng rất ít bộ nhớ.

Ồ, và đây không phải là rò rỉ bộ nhớ. Nó chỉ là hoạt động bình thường nếu bạn cố gắng sắp xếp một số lượng vô hạn các giá trị nhanh hơn chúng có thể được tiêu thụ.

---

Tôi đã giải mã để tìm ra lịch trình nào được sử dụng.

Đây là dịch ngược không theo thời gian:

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

Các phương pháp trên là từ Observable, QueryLanguage, và SchedulerDefaults tương ứng.