4
\$\begingroup\$

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");
    }
}
\$\endgroup\$
6
  • \$\begingroup\$ What if task to execute throws InvalidOperationException? \$\endgroup\$ May 28, 2017 at 14:14
  • \$\begingroup\$ Good question. This is a part I brought with me from the other article that I build upon. I'm inclided to just remove this part/catch. Can't really see why it should have a special status as is currently the case. \$\endgroup\$
    – 1iveowl
    May 28, 2017 at 18:26
  • \$\begingroup\$ I just updated the code above accordingly \$\endgroup\$
    – 1iveowl
    May 28, 2017 at 18:27
  • \$\begingroup\$ I think it's because Take() will throw if Dispose() is invoked after Count > 0 but before it. Not common but possible on the long run. Just wrap the Take() \$\endgroup\$ May 28, 2017 at 19:29
  • \$\begingroup\$ What do you mean by "Just wrap the Take()"? Do you mean using a lock to wrap it? \$\endgroup\$
    – 1iveowl
    Jun 1, 2017 at 20:01

2 Answers 2

2
\$\begingroup\$

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));
\$\endgroup\$
9
  • \$\begingroup\$ Does not address removing the queues, as it did before \$\endgroup\$ May 30, 2017 at 16:42
  • \$\begingroup\$ Thank you Bruno. How would you propose dealing with removing the queues that have completed? \$\endgroup\$
    – 1iveowl
    Jun 1, 2017 at 20:01
  • \$\begingroup\$ I like the Merge approach. \$\endgroup\$
    – 1iveowl
    Jun 1, 2017 at 20:02
  • \$\begingroup\$ @JasperHBojsen There's no way for you to know that no items will arrive. You can add an explicit method to delete it or you can use a timer and calculate the elapsed time since the last add. \$\endgroup\$ Jun 2, 2017 at 2:35
  • \$\begingroup\$ I tried implementing the the Merge and running it with the test. It didn't work. When subscribing to ObservableResults it completes instantly, which I suppose is no surprise as there is nothing yet in any queue. How do I keep it hot? The difference to the implement above is that the Subject keeps the Observable hot, right? \$\endgroup\$
    – 1iveowl
    Jun 2, 2017 at 17:35
1
\$\begingroup\$

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.

\$\endgroup\$
3
  • \$\begingroup\$ I thought this wouldn't work but now I understand why it works. So you create a queue and process the items on it (depending on operating system and thread behaviour you will have one or more items). When those items are processed the queue is deleted, upcoming items will create a new queue and the process repeats. I don't know if doing that is a good idea or not but my approach certainly consumes less memory and does less work (does not create and delete queues at all time). It also looks better arguably. But it's up to you I guess. \$\endgroup\$ Jun 3, 2017 at 11:53
  • \$\begingroup\$ See my newly section on charateristics of this solution. \$\endgroup\$
    – 1iveowl
    Jun 3, 2017 at 16:29
  • \$\begingroup\$ Upvoted since my solution still does not guarantee order. Your does. \$\endgroup\$ Jun 8, 2017 at 7:30

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge that you have read and understand our privacy policy and code of conduct.

Not the answer you're looking for? Browse other questions tagged or ask your own question.