Generic parallel task queue returning an observable sequence of return value

Question

This is a continuation of Parallel Task Queue that runs in sequence.

Using Reactive Extensions I want to create a generic class a that process a task and executes it on a specific queue returning a TReturnValue on an observable stream.

To specify which queue to be used to process the task, a queue id is specified.

If two tasks are running on different queues (i.e. with different queue ids) then the two tasks should be run in parallel. Tasks on the same queue should be run in sequence.

All tasks no matter which queue they run on, should return TReturnValue on the same one and only observable stream.

A ConcurrentDirectory of BlockingCollections using the queue id as key is used to keep track of each queue.

To prevent memory leaks, the BlockingCollection and Key is removed from the Directory when the queue is empty (i.e. all work has completed on the queue). Should new task emerge with the same queue id, a new BlockingCollection will be added again and put in the dictionary with the queue id as key.

A Subject is used to create the observer for the TReturnValue's. I've read that Subjects should be used with care, and herein lies my key question:

Is the following the right approach, specifically in terms of using a Subject? Will I run into any threading/blocking issues with this approach? Any other considerations?

Interface:

public interface IParallelTaskQueueRx<TReturnValue> : IDisposable
{
    IObservable<TReturnValue> ObservableResults { get; }

    void ProcessTaskOnSpecificQueue(
        Func<Task<TReturnValue>> myTask,
        string queueId);
}

Implementation:

public class ParallelTaskQueueRx<TReturnValue> : IParallelTaskQueueRx<TReturnValue>
{
    private readonly ISubject<TReturnValue> _subjectReturnValue = new Subject<TReturnValue>();
    private IObserver<TReturnValue> ObserverReturnValue => _subjectReturnValue.AsObserver();

    private readonly IDictionary<string, BlockingCollection<Func<Task<TReturnValue>>>> _queueDirectory =
        new ConcurrentDictionary<string, BlockingCollection<Func<Task<TReturnValue>>>>();

    public IObservable<TReturnValue> ObservableResults => _subjectReturnValue.AsObservable();

    public void ProcessTaskOnSpecificQueue(
        Func<Task<TReturnValue>> myTask, 
        string queueId)
    {
        if (!_queueDirectory.ContainsKey(queueId))
        {
            Debug.WriteLine($"Creating Queue: {queueId}");

            _queueDirectory.Add(queueId, new BlockingCollection<Func<Task<TReturnValue>>>());

            _queueDirectory[queueId].Add(myTask);

            ProcessQueue(queueId);
        }
        else
        {
            Debug.WriteLine($"Adding value to queue: {queueId}");
            _queueDirectory[queueId].Add(myTask);
        }
    }

    private void ProcessQueue(string queueId)
    {
        Task.Run(async () =>
        {
            while (_queueDirectory[queueId].Count > 0)
            {
                try
                {
                    var task = _queueDirectory[queueId].Take();
                    var result = await task();

                    lock (ObserverReturnValue)
                    {
                        ObserverReturnValue.OnNext(result);
                    }
                }
                catch (Exception ex)
                {
                    lock (ObserverReturnValue)
                    {
                        ObserverReturnValue.OnError(ex);
                    }
                }
            }

            _queueDirectory[queueId].CompleteAdding();

            lock (_queueDirectory)
            {
                _queueDirectory.Remove(queueId);
            }

            Debug.WriteLine($"Removing Queue: {queueId}");

        });
    }

    public void Dispose()
    {
        _queueDirectory.Clear();

        ObserverReturnValue.OnCompleted();
    }
}

Simple test:

class Program
{
    static void Main(string[] args)
    {
        var parallelTaskQueue = new ParallelTaskQueueRx<string>();

        var disp = parallelTaskQueue.ObservableResults.Subscribe(
            System.Console.WriteLine,
            ex => {System.Console.WriteLine($"Error: {ex.Message}");},
            () => {System.Console.WriteLine("Compleded");});

        Start(parallelTaskQueue);

        System.Console.ReadKey();
    }

    private static async void Start(ParallelTaskQueueRx<string> parallelTaskQueue)
    {
        parallelTaskQueue.ProcessTaskOnSpecificQueue(async () =>
            {
                await Task.Delay(TimeSpan.FromSeconds(1));
                return "Queue1: #1 (1 sec delay)";
            },"Queue1");

        parallelTaskQueue.ProcessTaskOnSpecificQueue(async () =>
            {
                await Task.Delay(TimeSpan.FromMilliseconds(250));
                return "Queue2: #1 (0,25 sec delay)";
            },"Queue2");

        System.Console.WriteLine("----Waiting 0,25 sec----");
        await Task.Delay(TimeSpan.FromMilliseconds(250));

        parallelTaskQueue.ProcessTaskOnSpecificQueue(async () =>
            {
                await Task.Delay(TimeSpan.FromMilliseconds(250));
                return "Queue1: #2 (0,25 sec delay)";
            },"Queue1");

        parallelTaskQueue.ProcessTaskOnSpecificQueue(async () =>
            {
                await Task.Delay(TimeSpan.FromMilliseconds(1250));
                return "Queue2 2 (1,25 sec delay)";
            },"Queue2");

        parallelTaskQueue.ProcessTaskOnSpecificQueue(() => Task.FromResult("Queue2 #3 (no delay)"),"Queue2");

        parallelTaskQueue.ProcessTaskOnSpecificQueue(() => Task.FromResult("Queue2 #4 (no delay)"),"Queue2");

        System.Console.WriteLine("----Waiting 5 sec----");
        await Task.Delay(TimeSpan.FromSeconds(5));

        parallelTaskQueue.ProcessTaskOnSpecificQueue(async () =>
            {
                await Task.Delay(TimeSpan.FromSeconds(1));
                return "Queue1 #3 (1 sec delay)";
            },"Queue1");

        parallelTaskQueue.ProcessTaskOnSpecificQueue(async () =>
            {
                await Task.Delay(TimeSpan.FromMilliseconds(250));
                return "Queue2 #5 (0,25 sec delay)";
            },"Queue2");

        System.Console.WriteLine("----Waiting 0,25 sec----");
        await Task.Delay(TimeSpan.FromMilliseconds(250));

        parallelTaskQueue.ProcessTaskOnSpecificQueue(async () =>
            {
                await Task.Delay(TimeSpan.FromMilliseconds(250));
                return "Queue1 #4 (0,25 sec delay)";
            },"Queue1");
    }
}

Answer 1

A lock statement inside a try construct should generate a compile error, so programmers don't keep falling on that trap. Read the answer of Eric Lippert to learn what is wrong with that, he is able to explain it better than I am.

_queueDirectory is a concurrent dictionary, however you are locking it as well.

Is CompleteAdding guaranteed to be thread safe? If not you better look into it.

Here is a revised version of ProcessQueue

private void ProcessQueue(string queueId)
{
    Task.Run(async () =>
    {
        while (_queueDirectory[queueId].Count > 0)
        {
            var task = _queueDirectory[queueId].Take();
            var result = default(TReturnValue);
            try{
                result = await task();
            }catch(Exception ex){
                lock (ObserverReturnValue)
                {
                    ObserverReturnValue.OnError(ex);
                    continue;
                }
            }
            lock(ObserverReturnValue){
                ObserverReturnValue.OnNext(result);
            }
        }

        _queueDirectory[queueId].CompleteAdding();
        _queueDirectory.Remove(queueId);

        Debug.WriteLine($"Removing Queue: {queueId}");
    });
}

---

Your solution seems to have some existential problems mainly on ProcessTaskOnSpecificQueue. Let me describe what you are currently doing:

• If the queue is empty create a new queue with a task
• If not add an item to the queue
• If the queue was empty start a task that processes all existing elements in this point in time, ignoring all upcoming items

This seems to be pretty boring (It wasn't specified if this was a requirement or not but my feeling is that it should have been). The only way you could possibly address this would be to have an infinite loop on ProcessQueue.

However I would suggest taking a different approach that consists of having a dictionary of Observable (one per each queue). Each queue would merge the elements into it, but different queues would be able to run in parallel (As specified).

I don't know if this particular implementation follows that specification but could at least work as a guideline I believe:

private readonly IDictionary<string, Observale<TReturnValue>> _publishers =
    new ConcurrentDictionary<string, Observale<TReturnValue>>();

public IObservable<TReturnValue> ObservableResults {
    return _publishers.Values.ToObservable().SelectMany(o => o);
}

public void ProcessTaskOnSpecificQueue(
    Func<Task<TReturnValue>> myTask, 
    string queueId)
{
    _var observable = myTask().ToObservable();
    if(!_publishers.ContainsKey(queueId)){
        _publishers[queueId] = observable;
    }else{
        _publishers[queueId] = _publishers[queueId].Merge(observable);
    }
}

---

As per comments this doesn't have the desired effects either. The best thing to do that occurred to me would be then to use an ObservableCollection. This would be a reliable to get hold of all added items. We might need to do something different regarding completing the work of a queue serially though.

Putting the items into an observable queue would leak memory like the OP suggested. Hence we can go for a solution where we simply have an event. We can also create an Observable from that event. So there's little change from my last previous solution (if you include the suggestion I made about using the observable collection add event to get an observable).

public class DataEventArgs<T> : EventArgs
{
    public DataEventArgs(T data) {
        Data = data;
    }
    public T Data { get; private set; }
}

public class ParallelTaskQueueRx<TReturnValue>
{
    public event EventHandler<DataEventArgs<TReturnValue>> ItemProcessed;

    public IObservable<TReturnValue> ObservableResults
    {
        get
        {
            return Observable.FromEventPattern<DataEventArgs<TReturnValue>>(
                h=> ItemProcessed += h,
                h => ItemProcessed -= h
            ).Select(e => e.EventArgs.Data);
        }
    }

    private readonly IDictionary<string, SemaphoreSlim> _queueDirectory =
        new ConcurrentDictionary<string, SemaphoreSlim>();

    public void ProcessTaskOnSpecificQueue(
        Func<Task<TReturnValue>> myTask,
        string queueId)
    {
        if (!_queueDirectory.ContainsKey(queueId))
        {
            var semaphore = new SemaphoreSlim(1, 1);
            _queueDirectory.Add(queueId, semaphore);
        }
        ThreadPool.QueueUserWorkItem(async ctx => {
            _queueDirectory[queueId].Wait();
            var result = await myTask();
            _queueDirectory[queueId].Release();
             ItemProcessed?.Invoke(this, new DataEventArgs<TReturnValue>(result));
        });
    }
}

I tested this one and it seemed to work just fine. Usage

var parallellqueue = new ParallelTaskQueueRx<string>();
parallellqueue.ObservableResults.ForEachAsync(valuue => Console.WriteLine(value));

Answer 2

Based on the great input received above, I've settled on the slightly modified solution below. The solution is still using a Subject to keep the Observable hot.

The code is also on GitHub and have been released as a NuGet.

In terms of using a Subject this article was helpful, although a bit long and complex to read: To Use Subject Or Not To Use Subject?

public class ParallelTaskQueueRx<TReturnValue> : IParallelTaskQueueRx<TReturnValue>
{
    private readonly ISubject<TReturnValue> _subjectReturnValue = new Subject<TReturnValue>();
    private IObserver<TReturnValue> ObserverReturnValue => _subjectReturnValue.AsObserver();

    private readonly IDictionary<string, BlockingCollection<Func<Task<TReturnValue>>>> _queueDirectory =
        new ConcurrentDictionary<string, BlockingCollection<Func<Task<TReturnValue>>>>();

    public IObservable<TReturnValue> ObservableResults => _subjectReturnValue.AsObservable();

    public void ProcessTaskOnSpecificQueue(
        Func<Task<TReturnValue>> myTask, 
        string queueId)
    {

        if (!_queueDirectory.ContainsKey(queueId))
        {
            Debug.WriteLine($"Creating Queue: {queueId}");

            _queueDirectory.Add(queueId, new BlockingCollection<Func<Task<TReturnValue>>>());

            _queueDirectory[queueId].Add(myTask);

            ProcessQueue(queueId);
        }
        else
        {
            Debug.WriteLine($"Adding value to queue: {queueId}");
            _queueDirectory[queueId].Add(myTask);
        }
    }

    private void ProcessQueue(string queueId)
    {
        Task.Run(async () =>
        {
            while (_queueDirectory[queueId].Count > 0)
            {
                var task = _queueDirectory[queueId].Take();
                TReturnValue result;

                try
                {
                    result = await task();
                }
                catch (Exception ex)
                {
                    lock (ObserverReturnValue)
                    {
                        ObserverReturnValue.OnError(ex);
                        continue;
                    }
                }

                lock (ObserverReturnValue)
                {
                    ObserverReturnValue.OnNext(result);
                }
            }

            _queueDirectory[queueId].CompleteAdding();
            _queueDirectory.Remove(queueId);

            Debug.WriteLine($"Removing Queue: {queueId}");

        });
    }

    public void Dispose()
    {
        _queueDirectory.Clear();
        ObserverReturnValue.OnCompleted();
    }
}

Some charateristcs of this implementations are, that if a new task arrives at a specific queue before the previous task has completed, then this new task is added to the queue and processed next.

The queue is not deleted before all tasks on it have been processed.

So, in situations where there are very few tasks coming in and the tasks are quickyly completed, a new queue is created with every new task. This adds a bit of overhead.

On the other hand, if the number of tasks coming in are high and/or they take long time to process, then the queue is maintained and only created with the first task.

These characteristics make this solution well suited for managing constant high-load and/or burst scenarios.