Consuming chunks from ConcurrentQueue

Question

I need to implement a queue of requests which can be populated from multiple threads. When this queue becomes larger than 1000 completed requests, these requests should be stored into a database.

public class RequestQueue
{
 private static BlockingCollection<VerificationRequest> _queue = new BlockingCollection<VerificationRequest>();
 private static ConcurrentQueue<VerificationRequest> _storageQueue = new ConcurrentQueue<VerificationRequest>();
 private static volatile bool isLoading = false;
 private static object _lock = new object();
 public static void Launch()
 {
 Task.Factory.StartNew(execute);
 }
 public static void Add(VerificationRequest request)
 {
 _queue.Add(request);
 }
 public static void AddRange(List<VerificationRequest> requests)
 {
 Parallel.ForEach(requests, new ParallelOptions() {MaxDegreeOfParallelism = 3},
 (request) => { _queue.Add(request); });
 }
 private static void execute()
 {
 Parallel.ForEach(_queue.GetConsumingEnumerable(), new ParallelOptions {MaxDegreeOfParallelism = 5}, EnqueueSaveRequest );
 }
 private static void EnqueueSaveRequest(VerificationRequest request)
 {
 _storageQueue.Enqueue( new RequestExecuter().ExecuteVerificationRequest( request ) );
 if (_storageQueue.Count > 1000 && !isLoading)
 {
 lock ( _lock )
 {
 if ( _storageQueue.Count > 1000 && !isLoading )
 {
 isLoading = true;
 var requestChunck = new List<VerificationRequest>();
 VerificationRequest req;
 for (var i = 0; i < 1000; i++)
 {
 if( _storageQueue.TryDequeue(out req))
 requestChunck.Add(req);
 }
 new VerificationRequestRepository().InsertRange(requestChunck);
 isLoading = false;
 }
 }
 } 
 }
}

Is there any way to implement this without lock and isLoading?

• 404: Why is everything static? A queue is an object, so create an instance of it and remove "static" from everywhere.
• madmanul: It supposed to be like singleton.
• 404: That seems like a terrible idea to me. If you want a singleton then create a single instance of the class and inject that instance where required. Otherwise you're limiting yourself to only one queue per application.
• madmanul: it should be only one queue per application
• 404: That may be the way you wish to use it in your application, but it doesn't make sense to design the queue itself that way.

Answer 1

Static class

As proposed by eurotrash I would not implement the class as static class because it has some disadvantages:

• It is only possible to use one class per application
• It is not possible to mock the class (e.g. for unit test / alternative implementations)
• It is not possible to use it with DI.
• It is not very common to have a statefull static classes - therefore it may confuse other developer

Naming

• isLoading should be renamed to _isLoading
• execute should be renamed to Execute or even better to some more descriptive name (e.g. StartConsumingTask)

Logic

Is there any way to implement this without lock and isLoading?

Currently your solution uses up to 5 tasks that process the items from the _queue and put them into the _storageQueue. If the _storageQueue has more than 1000 items, one of the 5 tasks writes them to the repository.

If the last part (write the items to the repository) is executed by a separate single task, the lock and the isLoading flag could be dropped.

The lock and the isLoding flag can be dropped if the last part (writing the items to the repository)

Tasks

• If you have long running tasks (like the one in the method Launch), it is better to start it with option TaskCreationOptions.LongRunning. Otherwise the thread for the task is taken from the threadpool and therefore blocks them for executing short-running procedures.

• _queue.Add (in method AddRange) is such a fast operation. I don't see a reason why to use Parallel.For here - just add the item in the current thread.

• Parallel.For in method execute makes sense if ExecuteVerificationRequest is realy time consuming - otherwise it adds only overhead and complexity.

Code with the suggestions above applied:

public class RequestQueue
{
 private BlockingCollection<VerificationRequest> _queue = new BlockingCollection<VerificationRequest>();
 private ConcurrentQueue<VerificationRequest> _storageQueue = new ConcurrentQueue<VerificationRequest>();
 private RequestQueue() { }
 public void Launch()
 {
 Task.Factory.StartNew(StartConsumingTask);
 Task.Factory.StartNew(StartPersistingTask);
 }
 public void Add(VerificationRequest request) => _queue.Add(request);
 public void AddRange(List<VerificationRequest> requests) => requests.ForEach(r => Add(r));
 private void StartConsumingTask()
 {
 Parallel.ForEach(_queue.GetConsumingEnumerable(), new ParallelOptions { MaxDegreeOfParallelism = 5 }, EnqueueSaveRequest);
 }
 private void EnqueueSaveRequest(VerificationRequest request)
 {
 _storageQueue.Enqueue(new RequestExecuter().ExecuteVerificationRequest(request));
 }
 private async Task StartPersistingTask()
 {
 while (true)
 {
 if (_storageQueue.Count > 1000)
 {
 var requestChunck = new List<VerificationRequest>();
 VerificationRequest req;
 for (var i = 0; i < 1000; i++)
 {
 if (_storageQueue.TryDequeue(out req))
 requestChunck.Add(req);
 }
 new VerificationRequestRepository().InsertRange(requestChunck);
 }
 else
 {
 await Task.Delay(100);
 }
 }
 }
 private static RequestQueue _instance = new RequestQueue();
 public static RequestQueue Instance => _instance; 
}

• t3chb0t: isLoading should be renamed to _isLoading - are you sure not to myIsLoading? you're definitelly cured ;-D
• JanDotNet: Because the other 3 variables also start with '_'. The actual prefix is not important - but it should ne consistent IMHO. ;)
• madmanul: isLoading prevents parallel consuming, but not parallel producing
• madmanul: "The lock and the isLoding flag can be dropped if the last part ..." - it cannot. Other thread will start consuming task too because it still can be more than 1000 jobs in queue
• JanDotNet: The part between isLoading = true and isLoading = false is not executed parallel, even if _storageQueue has 5000 items, right? So, my suggestion is, to run that code in a separate single task.