BatchPool

The one-stop generic task batching and management library.
Contributions are welcome to add features, flexibility, performance test coverage...

Features

• Very low overhead, flexible, high test coverage, generic...
• Tasks are executed in the order in which they are received
• Supports Task, Func and Action
• BatchPoolContainer states: Enabled / Paused
• Dynamic batch size: update the size of the BatchPool
• Callbacks: supports Task, Func and Action
• Check the state of a task
• Task Cancellation
• batchPoolContainer Cancellation
• Adding tasks in batch
• Dynamic ordering
• Dynamic custom ordering
• Updatable dynamic ordering
• Updatable dynamic custom ordering
• Factory
• Waiting for tasks to finish
• BatchPoolContainerManager
• BatchPoolContainerManager with DI

Supports Task, Func and Action

// Set batchSize to configure the maximum number of tasks that can be run concurrentlyBatchPoolContainerbatchPoolContainer=newBatchPoolContainer(batchSize:5,isEnabled:true);// TaskTaskaTask=newTask(()=>Console.WriteLine("Hello"));BatchPoolTasktask=batchPoolContainer.Add(aTask);awaittask.WaitForTaskAsync();// FuncFunc<Task>aFunc=async()=>Console.WriteLine("Hello");BatchPoolTaskfunc=batchPoolContainer.Add(aFunc);awaitfunc.WaitForTaskAsync();// ActionActionanAction=()=>Console.WriteLine("Hello");BatchPoolTaskaction=batchPoolContainer.Add(anAction);awaitaction.WaitForTaskAsync();

BatchPoolContainer states: Enabled / Paused

// Set isEnabled to configure the state of the batchPoolContainer at initializationBatchPoolContainerbatchPoolContainer=newBatchPoolContainer(batchSize:5,isEnabled:false);TaskaTask=newTask(()=>Console.WriteLine("Hello"));BatchPoolTasktask=batchPoolContainer.Add(aTask);awaittask.WaitForTaskAsync();// Resume and forgetbatchPoolContainer.ResumeAndForget();// Or resume and wait for all task to finishawaitbatchPoolContainer.ResumeAndWaitForAllAsync();// Then pause again to prevent new pending tasks to runbatchPoolContainer.Pause();

Dynamic batch size: update the size of the BatchPool

BatchPoolContainerbatchPoolContainer=newBatchPoolContainer(batchSize:5,isEnabled:false);// Increase or reduce the capacity and wait for it to finish updating. (The batchPoolContainer will need to wait if a reduction is requested while it is currently processing)awaitbatchPoolContainer.UpdateCapacityAsync(10);// Perform the same operation in the backgroundbatchPoolContainer.UpdateCapacityAndForget(10);

Callbacks: supports Task, Func and Action

BatchPoolContainerbatchPoolContainer=newBatchPoolContainer(batchSize:5,isEnabled:true);TaskaTask=newTask(()=>Console.WriteLine("Hello"));// The callback will run as soon as the main task completesTaskaCallbackTask=newTask(()=>Console.WriteLine("Hello"));BatchPoolTasktask=batchPoolContainer.Add(aTask,aCallbackTask);awaittask.WaitForTaskAsync();

Check the state of a task

BatchPoolContainerbatchPoolContainer=newBatchPoolContainer(batchSize:5,isEnabled:true);TaskaTask=newTask(()=>Console.WriteLine("Hello"));BatchPoolTasktask=batchPoolContainer.Add(aTask);boolisCanceled=task.IsCanceled;boolisCompleted=task.IsCompleted;

Task Cancellation

BatchPoolContainerbatchPoolContainer=newBatchPoolContainer(batchSize:5,isEnabled:true);TaskaTask=newTask(()=>Console.WriteLine("Hello"));BatchPoolTasktask=batchPoolContainer.Add(aTask);// Attempt to cancelbooldidCancel=task.Cancel();// Attempt to cancel all pending tasks (pending = tasks that have not yet started processing due to the batch size, or the paused state of the BatchPool)batchPoolContainer.RemoveAndCancelPendingTasks();

batchPoolContainer Cancellation

CancellationTokenSourcecancellationTokenSource=newCancellationTokenSource();BatchPoolContainerbatchPoolContainer=newBatchPoolContainer(batchSize:5,isEnabled:true,cancellationToken:cancellationTokenSource.Token);// All pending tasks will be CanceledcancellationTokenSource.Cancel();

Adding tasks in batch

BatchPoolContainerbatchPoolContainer=newBatchPoolContainer(batchSize:5,isEnabled:true);TaskaTask1=newTask(()=>Console.WriteLine("Hello"));TaskaTask2=newTask(()=>Console.WriteLine("Hello"));List<Task>listOfTasks=newList<Task>(){aTask1,aTask2};ICollection<BatchPoolTask>tasks=batchPoolContainer.Add(listOfTasks);

Dynamic ordering

int order example:

intbatchSize=1;// The int type provided as the generic parameter will provide dynamic ordering using the default .NET comparer. Order exection will be ascending (smallest to largest).BatchPoolDynamicContainer<int>batchPool=newBatchPoolDynamicContainer<int>(batchSize,isEnabled:false);vartask1=newTask(()=>Console.WriteLine("Hello #1"));varbatchTask1=batchPool.Add(task1,priority:1);vartask2=newTask(()=>Console.WriteLine("Hello #2"));varbatchTask2=batchPool.Add(task2,2);vartask3=newTask(()=>Console.WriteLine("Hello #3"));varbatchTask3=batchPool.Add(task3,3);awaitbatchPool.ResumeAndWaitForAllAsync();

string order example:

BatchPoolDynamicContainer<string>batchPool=newBatchPoolDynamicContainer<string>(batchSize,isEnabled:false);vartask1=newTask(()=>Console.WriteLine("Hello #1"));varbatchTask1=batchPool.Add(task1,priority:"a");vartask2=newTask(()=>Console.WriteLine("Hello #2"));varbatchTask2=batchPool.Add(task2,"b");vartask3=newTask(()=>Console.WriteLine("Hello #3"));varbatchTask3=batchPool.Add(task3,"c");awaitbatchPool.ResumeAndWaitForAllAsync();

enum order example:

publicenumTestEnum{Critical=0,High=1,Medium=2,Low=3}intbatchSize=1;varbatchPool=newBatchPoolDynamicContainer<TestEnum>(batchSize,isEnabled:false);vartask1=newTask(()=>executionOrderTracker.Add(TestEnum.High.ToString()));varbatchTask1=batchPool.Add(task1,TestEnum.High);vartask2=newTask(()=>executionOrderTracker.Add(TestEnum.Medium.ToString()));varbatchTask2=batchPool.Add(task2,TestEnum.Medium);vartask3=newTask(()=>executionOrderTracker.Add(TestEnum.Low.ToString()));varbatchTask3=batchPool.Add(task3,TestEnum.Low);awaitbatchPool.ResumeAndWaitForAllAsync();

Dynamic custom ordering

Custom ordering simply by passing a new IComparer:

// Reverse the default orderpublicclassStringReverseComparer:IComparer<string>{publicintCompare(string?x,string?y){returny!.CompareTo(x!);}}BatchPoolDynamicContainer<string>batchPool=newBatchPoolDynamicContainer<string>(batchSize,newStringReverseComparer(),isEnabled:false);

Updatable dynamic ordering

intbatchSize=1;// Use BatchPoolUpdatableDynamicContainer instead of BatchPoolDynamicContainerBatchPoolUpdatableDynamicContainer<int>batchPool=newBatchPoolUpdatableDynamicContainer<int>(batchSize,isEnabled:false);vartask1=newTask(()=>Console.WriteLine("Hello #1"));varbatchTask1=batchPool.Add(task1,priority:1);vartask2=newTask(()=>Console.WriteLine("Hello #2"));varbatchTask2=batchPool.Add(task2,2);vartask3=newTask(()=>Console.WriteLine("Hello #3"));varbatchTask3=batchPool.Add(task3,3);// batchTask1 will now execute last instead of firstbatchPool.UpdatePriority(batchTask1,4);awaitbatchPool.ResumeAndWaitForAllAsync();

Updatable dynamic custom ordering

Custom ordering simply by passing a new IComparer:

// Reverse the default orderpublicclassStringReverseComparer:IComparer<string>{publicintCompare(string?x,string?y){returny!.CompareTo(x!);}}BatchPoolUpdatableDynamicContainer<string>batchPool=newBatchPoolUpdatableDynamicContainer<string>(batchSize,newStringReverseComparer(),isEnabled:false);

Technical Note: More checks are required for BatchPoolUpdatableDynamicContainer over the BatchPoolDynamicContainer. However, the performance impact should not be noticeable as they occur at O(1). UpdatePriority does not modify the existing data structure, but instead allows the container to check if a priority is current and valid or a new has been added.

Factory

The BatchPoolFactory simply provides a quick way to instantiate a new BatchPoolContainer:

// Default queueBatchPoolContainerbatchPool=BatchPoolFactory.GetQueueBatchPool(batchSize,isEnabled:false);// Dynamic orderingBatchPoolDynamicContainer<int>batchPool=BatchPoolFactory.GetDynamicallyOrderedBatchPool<int>(batchSize,isEnabled:false);// Dynamic ordering and update existing tasks priorityBatchPoolUpdatableDynamicContainer<int>batchPool=BatchPoolFactory.GetUpdatableDynamicallyOrderedBatchPool<int>(batchSize,isEnabled:false);

Waiting for tasks to finish

BatchPoolContainerbatchPoolContainer=newBatchPoolContainer(batchSize:5,isEnabled:true);TaskaTask1=newTask(()=>Console.WriteLine("Hello"));TaskaTask2=newTask(()=>Console.WriteLine("Hello"));List<Task>listOfTasks=newList<Task>(){aTask1,aTask2};List<BatchPoolTask>tasks=batchPoolContainer.Add(listOfTasks);// Wait for each task individuallyawaittasks[0].WaitForTaskAsync();awaittasks[1].WaitForTaskAsync();// Wait for all tasks to finishawaitbatchPoolContainer.WaitForAllAsync();// With timeoutInMillisecondsawaitbatchPoolContainer.WaitForAllAsync(timeoutInMilliseconds:100);// With timeoutawaitbatchPoolContainer.WaitForAllAsync(timeout:TimeSpan.FromMilliseconds(100));// With cancellationTokenCancellationTokenSourcecancellationTokenSource=newCancellationTokenSource();awaitbatchPoolContainer.WaitForAllAsync(cancellationToken:cancellationTokenSource.Token);// With cancellationToken and timeoutInMilliseconds/cancellationTokenawaitbatchPoolContainer.WaitForAllAsync(timeoutInMilliseconds:100,cancellationTokenSource.Token);awaitbatchPoolContainer.WaitForAllAsync(timeout:TimeSpan.FromMilliseconds(100),cancellationTokenSource.Token);

BatchPoolContainerManager

BatchPoolContainerManager<BatchPoolContainer>batchPoolContainerManager=newBatchPoolContainerManager<BatchPoolContainer>();// Create and register a BatchPoolBatchPoolContainerbatchPoolContainer=newBatchPoolContainer(batchSize:5,isEnabled:true);batchPoolContainer=batchPoolContainerManager.RegisterBatchPool("UniqueBatchPoolName",batchPoolContainer);// Retrieve the BatchPoolboolisFound=batchPoolContainerManager.TryGetBatchPool("UniqueBatchPoolName",outbatchPoolContainerretrievedBatchPool);TaskaTask=newTask(()=>Console.WriteLine("Hello"));BatchPoolTasktask=batchPoolContainer.Add(aTask);// Wait for all tasks in all BatchPools to finishawaitbatchPoolContainerManager.WaitForAllBatchPools();

BatchPoolContainerManager with DI

IHosthostBuilder=Host.CreateDefaultBuilder().ConfigureServices((_,services)=>services.AddSingleton<BatchPoolContainerManager<BatchPoolContainer>>()).Build();usingIServiceScopeserviceScope=hostBuilder.Services.CreateScope();IServiceProviderserviceProvider=serviceScope.ServiceProvider;BatchPoolContainerManager<BatchPoolContainer>batchPoolContainerManager=serviceProvider.GetRequiredService<BatchPoolContainerManager<BatchPoolContainer>>();