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 Subject
s 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");
}
}