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
.
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 ActionBlock
s, 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 Task
s 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.
If you love us? You can donate to us via Paypal or buy me a coffee so we can maintain and grow! Thank you!
Donate Us With