RX - memory consumption and performance

Question

I've created a function Rate - to give an speed, as IObservable<double> of any IObservable. 

With the below code, I experience two undesired effects
a) memory consuption rapidly reaches the machine limit
b) The app is non-responsive.

If I change the speed of the generated events, or only subscribe to one rate counter, the responsiveness is better, but memory consumption is still troubling.

Is there a bug in my code, or is the issue with the RX?

Env:
Win 7
.Net 4 beta
64bit dual core

Code:

public static class RXX
{
    public static IObservable<double> Rate<_>(this IObservable<_> stream, int duration, int outputrate)
    {
        var SW = Stopwatch.StartNew();
        Func<int, _, int> accumulator;

        if (duration > 0)
            accumulator = (count, __) =>
            {
                if (SW.Elapsed.TotalMilliseconds > duration)
                {
                    count = 0;
                    SW = Stopwatch.StartNew();
                }
                return count + 1;
            };
        else
            accumulator = (count, __) => count + 1;

        var ratestream = stream.Scan(0, accumulator);
        return ((outputrate > 0) ? ratestream.Sample(outputrate) : ratestream)
            .Select(r => r / SW.Elapsed.TotalSeconds);
    }
}

class Program
{
    static void Main(string[] args)
    {
        var gen = Observable.Generate(
                                                0, //seed
                                                ii => true, //cond
                                                ii => // selector
                                                    new {
                                                        Prop1 = ii,
                                                        Prop2 = "RX Rate Play",
                                                    },
                                                ii =>0,  //delay
                                                ii => ++ii //iterator
                                                );

        gen.Rate(1000, 500).Subscribe(r => Console.WriteLine(string.Format("Rate (1sec)  : {0:0.00}", r)));
        gen.Rate(0, 500).Subscribe(r => Console.WriteLine(string.Format(   "Rate (total) : {0:0.00}", r)));
        Console.ReadKey();
    }
}

---

--Scott W.

 

Answer 1

I just investigated it and it is a performance bug in Rx.  I'll get a fix in for the next minor release of Rx.  Thanks for reporting this.

Answer 2

Fixed!