Task Factory that processes non-similar items in parallel

Question

There are a bunch of items added to the database continuously which need to be processed. I want non similar items to be processed in parallel.

for e.g.:

Type A Items: Item 1, Item 2 ,Item 3

Type B Items: Item 4, Item 5 ,Item 6

Type C Items: Item 7, Item 8 ,Item 9

Item 1, Item 4 and Item 7 should be processed paralelly.

As more items of a type get added to the database, they will be selected and queued to be processed only after the previous items of this type are processed.

I think I can do this using a static Task Factory with a CustomTaskScheduler, which will start a new task only after the previous task of that type has finished? My question is how should my CustomTaskScheduler look?

class test
{
    private static void Main()
    {
        //List of items from the database
        var itemList = new List<Item>();
        itemList.Add(new Item(1, "A"));
        itemList.Add(new Item(2, "A"));
        itemList.Add(new Item(3, "A"));

        itemList.Add(new Item(4, "B"));
        itemList.Add(new Item(5, "B"));
        itemList.Add(new Item(6, "B"));

        itemList.Add(new Item(7, "C"));
        itemList.Add(new Item(8, "C"));
        itemList.Add(new Item(9, "C"));

        //This needs to be run on a timer picking up new items from the database every time
        new ProcessQueue().ProcessAllItems(itemList);

        Console.ReadLine();
    }
}

public class ProcessQueue
{
    private static CustomTaskScheduler customTaskScheduler = new CustomTaskScheduler(1);
    private static TaskFactory factory = new TaskFactory(customTaskScheduler);

    public void ProcessAllItems(List<Item> itemList)
    {
        var cts = new CancellationTokenSource();

        foreach (var item in itemList)
        {
            factory.StartNew(
                o =>
                executeTask(item.Id, item.ItemType),
                item.ItemType, //unique identifier for multiple threads
                cts.Token);
        }

    }

    public void executeTask(int id, string parentId)
    {
        Console.WriteLine("Item - {0} ItemType - {1} on thread {1}   ", id, parentId,
                          Thread.CurrentThread.ManagedThreadId);
        Thread.Sleep(5000);
    }
}

public class Item
{
    public Item(int id, string itemType)
    {
        Id = id;
        ItemType = itemType;
    }

    public int Id { get; set; }
    public string ItemType { get; set; }
}

Answer 1

How about this, adapted from How to: Create a Task Scheduler That Limits Concurrency. It uses Task.AsyncState, which should be your Item.ItemType given your example, to synchronise threads.

public class ParallelTypeScheduler : TaskScheduler {

    // Indicates whether the current thread is processing work items.
   [ThreadStatic]
   private static bool _currentThreadIsProcessingItems;

   // The list of tasks to be executed  
   private readonly ConcurrentDictionary<object, LinkedList<Task>> _tasks = new ConcurrentDictionary<object, LinkedList<Task>>; // protected by lock(_tasks) 

   // Indicates whether the scheduler is currently processing work items.  
   private readonly Dictionary<object, bool> _typesRunning = new Dictionary<object,bool>();

   // Queues a task to the scheduler.  
   protected sealed override void QueueTask(Task task)
   {
        LinkedList<Task> typesTasks = _tasks.GetOrAdd(task.AsyncState, new LinkedList<Task>());

        lock (typesTasks)
        {
            typesTasks.AddLast(task);

            if(!_typesRunning.ContainsKey(task.AsyncState) || _typesRunning[task.AsyncState] == false){
                _typesRunning[task.AsyncState] = true;
                NotifyThreadPoolOfPendingWork(task.AsyncState);
            }
        }
   }

   // Inform the ThreadPool that there's work to be executed for this scheduler.  
   private void NotifyThreadPoolOfPendingWork(object type)
   {
       ThreadPool.UnsafeQueueUserWorkItem(_ =>
       {
           // Note that the current thread is now processing work items. 
           // This is necessary to enable inlining of tasks into this thread.
           _currentThreadIsProcessingItems = true;
           try
           {
               LinkedList<Task> typedTasks;
               if (_tasks.TryGetValue(type, out typedTasks)) {

                   while (true) {

                       Task item;
                       lock (typedTasks)
                       {
                           // When there are no more items to be processed, 
                           // note that we're done processing, and get out. 
                           if (_tasks.Count == 0)
                           {
                               _typesRunning[type] = false;
                               break;
                           }

                           // Get the next item from the queue
                           item = typedTasks.First.Value;
                           typedTasks.RemoveFirst();
                       }

                       base.TryExecuteTask(item);

                   }

               }
           }
           // We're done processing items on the current thread 
           finally { _currentThreadIsProcessingItems = false; }
       }, null);
   }

   // Attempts to execute the specified task on the current thread.  
   protected sealed override bool TryExecuteTaskInline(Task task, bool taskWasPreviouslyQueued)
   {
       // If this thread isn't already processing a task, we don't support inlining 
       if (!_currentThreadIsProcessingItems) return false;

       // If the task was previously queued, remove it from the queue 
       if (taskWasPreviouslyQueued) {

          LinkedList<Task> typedTasks;
          // Try to run the task.  
          if (_tasks.TryGetValue(task.AsyncState, out typedTasks) && TryDequeue(typedTasks, task)) 
            return base.TryExecuteTask(task);
          else 
             return false; 

       } else  
          return base.TryExecuteTask(task);

   }

   // Attempt to remove a previously scheduled task from the scheduler.  
   protected sealed override bool TryDequeue(LinkedList<Task> typedTasks, Task task)
   {
       lock (typedTasks) return typedTasks.Remove(task);
   }

   // Gets an enumerable of the tasks currently scheduled on this scheduler.  
   protected sealed override IEnumerable<Task> GetScheduledTasks()
   {
       bool lockTaken = false;
       try
       {
           Monitor.TryEnter(_tasks, ref lockTaken);
           if (lockTaken) return _tasks.SelectMany(t => t.Value);
           else throw new NotSupportedException();
       }
       finally
       {
           if (lockTaken) Monitor.Exit(_tasks);
       }
   }

}

I'm not sure about how thread safety works with GetScheduledTasks though.