.NET 6. Parallel processing, refill queue when less than MaxDegreesOfParallelism threads are running

TravelingFox 0
Viewed 17

I'd like to process some data in parallel using a BackgroundService.

There may be more than MaxDegreesOfParallelism items to process in the database, but as the processing can take several minutes per row, I'd like to only load as many items into the local processing queue as can actually be processed, so I avoid having potentially outdated data.

For example, if there are 20 items to be processed in the database and MaxDegreesOfParallelism (the number of concurrent processing jobs allowed) is 10, I only want to load 10 items. After a few minutes, the first 3 items may be done processing. At this point, I want to load the next 3 items from the database. Of course I could load all 20 items when the program starts, but when "processing slots" become available after a few minutes, those 20 items may no longer reflect the current state of the database.

I guess I could use a ConcurrentQueue<T> for this - but how would I know when "processing slots" become available? It wouldn't have to be immediate, it would be OK if the "data loading" service/thread could check every minute or so how many "processing slots" are available and load the corresponding number of items to be added to the processing queue.

Or maybe I should just use Task.WhenAny, so that I could load additional items as soon as the first item has been processed? But then I'd still need to know how many "slots" are available. (But I could use the database to track that, hmmm...)

multithreading async-await parallel-processing .net-6.0 .net
1 Answers

I assume you're using TPL dataflow. You can use ActionBlock with BoundedCapacity to limit the number of items that get queued.

ActionBlock will then await on SendAsync call. SendAsync will be blocked when the buffer fills up (preventing additional records from being fetched, until a record completes processing, allowing SendAsync to push the next record.)

Action<int> fn = record => {
Thread.Sleep(1000); // record processing task goes here.
Console.WriteLine(record);
};

var opts = new ExecutionDataflowBlockOptions { BoundedCapacity = 5, MaxDegreeOfParallelism = 2 };
// Sets the block's buffer size to one message

var actionBlock = new ActionBlock<int>(fn, opts);

while (true /* add your exit condition here */) {
    var record = GetFromDB(index++);
    await actionBlock.SendAsync(record);  
    // After first 5 records, the bounded capacity will prevent additional records from being fetched until one of the existing records gets processed.
}
userCV3z0
Related