Using Rx to limit to X concurrent Tasks

Question

I'm very new to Rx, and I'm trying to use it to have up to X number of concurrent subscribing tasks. The data source is actually from a database, so I'm having to poll the database. I realise that the idea behind Rx is that it's push and not pull - so the poll doesn't really fit that well, but conceptually the data going into the database is a stream of events that I want to subscribe to and do something with.

The main issue that I have is that the LimitedConcurrencyLevelTaskScheduler doesn't seem to be successfully limiting to the specified number of tasks. It runs a lot more concurrently than the 8 I've specified.

I'm also unsure which is the better way out of the two below solutions (or possibly, both are wrong?!).

Here's one method I've tried which uses the Observable.Timer ...

public static void Main()
{
    var taskFactory = new TaskFactory (new LimitedConcurrencyLevelTaskScheduler (8));
    var scheduler = new TaskPoolScheduler (taskFactory);

    Observable.Timer (TimeSpan.FromMilliseconds (10), scheduler)
        .SelectMany (x => Observable.FromAsync (GetItemsSource))
        .Repeat ()
        .ObserveOn (scheduler)
        .Subscribe (x => Observable.FromAsync(y => DoSomethingAsync (x.ToList())));

    Console.ReadKey ();
}

private static async Task> GetItemsSource()
{
    return await _myRepo.GetMoreAsync(10);
}

private static async Task DoSomethingAsync(IEnumerable items)
{
    // Do something with the data
}

I've also tried doing it this way instead ...

public static void Main()
{
    GetItemsSource()
        .ObserveOn(scheduler)
        .Select (async x => await DoSomethingAsync(x))
        .Subscribe();

    Console.ReadKey ();
}

public static IObservable GetItemsSource()
{
    return Observable.Create(
        async obs =>
        {
            while (true)
            {
                var item = (await _myRepo.GetMoreAsync(1)).FirstOrDefault();

                if(item != null)
                {
                    obs.OnNext(item);
                }

                await Task.Delay(TimeSpan.FromMilliseconds(10))
            }
        });
}

private static async Task DoSomethingAsync(IEnumerable items)
{
    // Do something with the data
}

Obviously very simple examples, without error handling or cancellation support.

Both seem to work, but neither limit to 8 concurrent tasks.

As I say, I'm very new to Rx, and am probably missing lots of fundamental things. I certainly plan on doing a lot of reading to fully understand Rx as it seems very powerful, but for now I want to get something working quickly.

UPDATE

Following on from Enigmativity's answer and comments, here's some code which logs out the concurrent counts ...

void Main()
{
    var taskFactory = new TaskFactory(new LimitedConcurrencyLevelTaskScheduler(8));
    var scheduler = new TaskPoolScheduler(taskFactory);

    using (
        (
            from n in Observable.Interval(TimeSpan.FromMilliseconds(10), scheduler)
            from g in Observable.FromAsync(GetItemsSource, scheduler)
            from u in Observable.FromAsync(() => DoSomethingAsync(g), scheduler)
            select u)
        .ObserveOn(scheduler)
        .Subscribe())
    {
        Console.ReadLine();
    }
}

private static volatile int _numIn = 0;
private static volatile int _numOut = 0;

public static async Task> GetItemsSource()
{
    try
    {
        _numIn++;

        $"Concurrent tasks (in): {_numIn}".Dump();

        // Simulate async API call
        await Task.Delay(TimeSpan.FromMilliseconds(10));

        return new List { Guid.NewGuid() };
    }
    finally
    {
        _numIn--;
    }
}

private static async Task DoSomethingAsync(IEnumerable deliveryIds)
{
    try
    {
        _numOut++;

        // Simulate async calls required to process the event
        await Task.Delay(TimeSpan.FromMilliseconds(1000));

        $"Concurrent tasks (out): {_numOut}".Dump();
    }
    finally
    {
        _numOut--;
    }
}

This shows around 64 concurrent tasks running.

Update 2

It does look like this is because the subscriber is async. If I test using a non-async subscriber, it works fine. Unfortunately I need to have an async subscriber, as it needs to call other async methods.

It looks like I can do something similar by doing this ...

GetItemsSource2()
    .Select(x => Observable.FromAsync(() => DoSomethingAsync(x)))
    .Merge(64)
    .Subscribe();

So using Merge instead of the LimitedConcurrencyLevelTaskScheduler.

Using Rx to limit to X concurrent Tasks

UPDATE

Update 2

Answers (1)

Related Questions