Search code examples
c#multithreadingtask-parallel-library.net-4.5tpl-dataflow

Customizing ActionBlock<T>


I want to implement a prioritised ActionBlock<T>. So that i can Conditionally give priority to some TInput items by using a Predicate<T>.
I read Parallel Extensions Extras Samples and Guide to Implementing Custom TPL Dataflow Blocks.
But Still don`t figure out how can i implement this scenario.
---------------------------- EDIT ---------------------------
There are some tasks, which 5 of them can be run simultaneously. When user push the button, some (depends on predicate function) tasks should run with the most priority.
In fact i write this code

TaskScheduler taskSchedulerHighPriority;
ActionBlock<CustomObject> actionBlockLow;
ActionBlock<CustomObject> actionBlockHigh;
...
queuedTaskScheduler = new QueuedTaskScheduler(TaskScheduler.Default, 5);
taskSchedulerHigh = queuedTaskScheduler.ActivateNewQueue(0);
taskSchedulerLow = queuedTaskScheduler.ActivateNewQueue(1);
...
actionBlockHigh = new ActionBlock<CustomObject>(new Action<CustomObject>(method), new ExecutionDataflowBlockOptions { MaxDegreeOfParallelism = 5, SingleProducerConstrained = false, TaskScheduler = taskSchedulerHigh });
actionBlockLow = new ActionBlock<CustomObject>(new Action<CustomObject>(method), new ExecutionDataflowBlockOptions { MaxDegreeOfParallelism = 5, MaxMessagesPerTask = 1, TaskScheduler = taskSchedulerLow });
...     
if (predicate(customObject))
    actionBlockHigh.Post(customObject);
else
    actionBlockLow.Post(customObject);

But it seems priority does not take effected at all.
---------------------------- EDIT ------------------
I find the fact that when i use this line of code:

actionBlockHigh = new ActionBlock<AvlHistory>(new Action<AvlHistory>(SemaphoreAction), new ExecutionDataflowBlockOptions { TaskScheduler = taskSchedulerHigh });
actionBlockLow = new ActionBlock<AvlHistory>(new Action<AvlHistory>(SemaphoreAction), new ExecutionDataflowBlockOptions { TaskScheduler = taskSchedulerLow });

Cause application observe priorities of the Tasks correctly but only one task can be execute at a time, meanwhile using the first code block that is shown in flowing, cause application run 5 tasks simultaneously but in inappropriate priority order.

actionBlockHigh = new ActionBlock<AvlHistory>(new Action<AvlHistory>(SemaphoreAction), new ExecutionDataflowBlockOptions { MaxDegreeOfParallelism = 5, TaskScheduler = taskSchedulerHigh });
actionBlockLow = new ActionBlock<AvlHistory>(new Action<AvlHistory>(SemaphoreAction), new ExecutionDataflowBlockOptions { MaxDegreeOfParallelism = 5, TaskScheduler = taskSchedulerLow });

Update:
Tanks to svick, i should specify MaxMessagesPerTask for taskSchedulerLow.


Solution

  • Your question doesn't include many details, so the following is just a guess of what you might need.

    I think the simplest way to do this is to have two ActionBlocks, running on different priorities on QueuedTaskScheduler from ParallelExtensionsExtras. You would link to the high-priority one using a predicate and then to the low-priority one. Also, to make sure high-priority Tasks aren't waiting, set MaxMessagesPerTask of the low-priority block.

    In code, it would look something like:

    static ITargetBlock<T> CreatePrioritizedActionBlock<T>(
        Action<T> action, Predicate<T> isPrioritizedPredicate)
    {
        var buffer = new BufferBlock<T>();
    
        var scheduler = new QueuedTaskScheduler(1);
    
        var highPriorityScheduler = scheduler.ActivateNewQueue(0);
        var lowPriorityScheduler = scheduler.ActivateNewQueue(1);
    
        var highPriorityBlock = new ActionBlock<T>(
            action, new ExecutionDataflowBlockOptions
            {
                TaskScheduler = highPriorityScheduler
            });
        var lowPriorityBlock = new ActionBlock<T>(
            action, new ExecutionDataflowBlockOptions
            {
                TaskScheduler = lowPriorityScheduler,
                MaxMessagesPerTask = 1
            });
    
        buffer.LinkTo(highPriorityBlock, isPrioritizedPredicate);
        buffer.LinkTo(lowPriorityBlock);
    
        return buffer;
    }
    

    This is just a sketch of what you could do, for example, Completion of the returned block doesn't behave correctly.