Extended Thread Pool

• Project's site
• Source code on GitHub
• Download source code - 516.2 KB
• NuGet package

Introduction

Different kinds of Thread Pools are available: Smart Thread Pool, BlackHen.Threading, but nothing supports extensions. All Thread Pools are hard coded, and it's too difficult to work with any. I've tried to address these issues:

• A Thread Pool should be extensible and configurable
• A Thread Pool should be as simple as possible

So, I've created a Tiny Thread Pool. This Thread Pool is written using C# 4.0.

Tiny Thread Pool Features

• Really simple 
• Extensible queues
• Extensible Task Items 
• Limit on the maximum number of working threads
• Dynamic thread workers
• Task priority support 
• Extensible logging

Tiny Thread Pool Design

• ITaskItem represents a task
• ITaskQueue represents the task queue logic 
• ITaskQueueController represents the communication logic between the consumer and the producer (thread safe)
• WorkThread represents a thread worker
• TinyThreadPool controls work threads

Let's take a look at each class more deeply:

• The ITaskItem represents some work that should be done.

    public interface ITaskItem
    {
        void DoWork();
    } 

• TaskItemPriority - WorkThread priority can be specified for each task.
• The ITaskQueue is another simple interface that manages the task queue.

    /// <summary>
    /// Represent the queue.
    /// </summary>
    public interface ITaskQueue
    {
        /// <summary>
        /// Count of work item.
        /// </summary>
        int Count { get; }

        /// <summary>
        /// Dequeue the work item.
        /// </summary>
        /// <returns>The work item.</returns>
        IWorkItem Dequeue();

        /// <summary>
        /// Enqueue the work item.
        /// </summary>
        /// <param name="item">The work item.</param>
        void Enqueue(IWorkItem item);
    }  

• ITaskQueueController provides communication logic between the consumer and the producer.

    public interface ITaskQueueController : IDisposable
    {
        int ConsumersWaiting { get; }
        IWorkItem Dequeue();
        void Enqueue(IWorkItem item);
    } 

I've implemented two task queue controllers derived from ITaskQueueController:

• DefaultTaskQueueController
• BoundedTaskQueueController

Default Task Queue Controller

DefaultTaskQueueController is a thread-safe wrapper for the ITaskQueue:

    public sealed class DefaultTaskQueueController : TaskQueueController
    {
        public DefaultTaskQueueController(ITaskQueue taskQueue)
            : base(taskQueue)
        {
        }

        protected override IWorkItem DequeueCore()
        {
            lock (_locker)
            {
                while (_taskQueue.Count == 0 && !_isDisposed)
                {
                    _consumersWaiting++;
                    Monitor.Wait(_locker);
                    _consumersWaiting--;
                }
                if (_isDisposed)
                {
                    return null;
                }
                return _taskQueue.Dequeue();
            }
        }

        protected override void EnqueueCore(IWorkItem item)
        {
            lock (_locker)
            {
                _taskQueue.Enqueue(item);
                if (_consumersWaiting > 0)
                {
                    Monitor.PulseAll(_locker);
                }
            }
        }
    } 

Bounded Task Queue Controller

BoundedTaskQueueController (thread safe) - if the producer task or tasks create items at a rate faster than the consumer can process them, the system can run into unbounded memory usage. The BoundedTaskQueueController allows you to place a limit on the size that the queue may reach before the producer is forced to block.

    public sealed class BoundedTaskQueueController : TaskQueueController
    {
        private readonly int _maxTasksCount;
        private int _producersWaiting;

        public BoundedTaskQueueController(ITaskQueue taskQueue, int maxTasksCount)
            : base(taskQueue)
        {
            if (maxTasksCount < 1)
            {
                throw new ArgumentException("MaxTasksCount should be greater 0");
            }
            _maxTasksCount = maxTasksCount;
        }

        protected override IWorkItem DequeueCore()
        {
            IWorkItem taskItem;
            lock (_locker)
            {
                while (_taskQueue.Count == 0 && !_isDisposed)
                {
                    _consumersWaiting++;
                    Monitor.Wait(_locker);
                    _consumersWaiting--;
                }
                if (_isDisposed)
                {
                    return null;
                }
                taskItem = _taskQueue.Dequeue();
                if (_producersWaiting > 0)
                {
                    Monitor.PulseAll(_locker);
                }
            }
            return taskItem;
        }

        protected override void EnqueueCore(IWorkItem item)
        {
            lock (_locker)
            {
                while (_taskQueue.Count == (_maxTasksCount - 1) && !_isDisposed)
                {
                    _producersWaiting++;
                    Monitor.Wait(_locker);
                    _producersWaiting--;
                }
                _taskQueue.Enqueue(item);
                if (_consumersWaiting > 0)
                {
                    Monitor.PulseAll(_locker);
                }
            }
        }
    } 

Tiny Thread Pool

The TinyThreadPool manages the ITaskQueueController. The  TinyThreadPool is created thru Create method, for example 

var threadPool = TinyThreadPool.Create(x =>
{
    x.Name = "My ThreadPool";
    x.MinThreads = 2;
    x.MaxThreads = 10;
    x.MultiThreadingCapacity = MultiThreadingCapacity.Global;
});

var threadPool = TinyThreadPool.Default;

MultiThreadingCapacityType represent a threading capacity

    public enum MultiThreadingCapacityType
    {
        /// <summary>
        /// Represent all processors
        /// </summary>
        Global,
        /// <summary>
        /// Represent one processor
        /// </summary>
        PerProcessor
    }

The AddTask method is used to add a task into the task pipeline. If the maximum thread limit is not reached and ConsamersWaiting = 0, a new WorkThread will be created.  

You can add a task with a TaskItemPriority. Please note, DefaultTaskQueue does not validate TaskItemPriority use PriorityTaskQueue for priority tasks

        /// <summary>
        /// Add new task.
        /// </summary>
        /// <param name="taskItem">Represent task.</param>
        /// <param name="priority">Task priority.</param>
        /// <remarks>Default task priority is <see cref="TaskItemPriority.Normal" />.</remarks>
        public void AddTask(ITaskItem taskItem, TaskItemPriority priority = TaskItemPriority.Normal)
        {
            IWorkItem workItem = WorkItem.FromTaskItem(taskItem, priority);
            AddWorkItem(workItem);
        }

        /// <summary>
        /// Add new task as <see cref="Action" />.
        /// </summary>
        /// <param name="action">Represent task.</param>
        /// <param name="priority">Task priority.</param>
        /// <remarks>Default task priority is <see cref="TaskItemPriority.Normal" />.</remarks>
        public void AddTask(Action action, TaskItemPriority priority = TaskItemPriority.Normal)
        {
            IWorkItem workItem = WorkItem.FromAction(action, priority);
            AddWorkItem(workItem);
        } 

Work Thread

The WorkThread class executes a task item and provides logging.

        private void DoWork()
        {
            while (_isRun)
            {
                try
                {
                    IWorkItem workItem = _taskQueueController.Dequeue();
                    if (workItem == null)
                    {
                        continue;
                    }
                    ProcessItem(workItem);
                }
                catch (Exception ex)
                {
                    _log.Error(ex);
                }
            }
        } 

Thread Pool Extensibility 

In case you need a more powerful task queue, you need to implement the ITaskQueue, and don't worry about thread safety: you can create your own ITaskQueueController as well.  

Example

We create TinyThreadPool with custom settings. The SampleTask is derived by ITaskItem, please see  details below. 

internal class Program
{
    private static ITinyThreadPool _threadPool;

    private static void AddTasks()
    {
        for (int taskIndex = 0; taskIndex < 50; taskIndex++)
        {
            _threadPool.AddTask(new SampleTask(taskIndex));
        }
    }

    private static void Main()
    {
        // create default TinyThreadPool instance or thru method TinyThreadPool.Create
        // _threadPool = TinyThreadPool.Default;

        _threadPool = TinyThreadPool.Create(x =>
        {
            x.Name = "My ThreadPool";
            x.MinThreads = 2;
            x.MaxThreads = 10;
            x.MultiThreadingCapacity = MultiThreadingCapacity.Global;
        });

        AddTasks();
        Console.ReadKey();
    }

    private sealed class SampleTask : ITaskItem
    {
        private readonly int _taskIndex;

        public SampleTask(int taskIndex)
        {
            _taskIndex = taskIndex;
        }

        public void DoWork()
        {
            Thread.Sleep(100);
            Console.WriteLine("Task {0} has been finished", _taskIndex);
        }
    }
}

History

• 29 Jun 2008: Initial version.
• 02 Jul 2008
  • + TransactionalMsmqTaskItem, DefaultTaskQueue.
  • * TaskQueueController classes added disposable support.
  • * WorkThread updated stop logic.
  • * Sample projects: 
    • CoreDefaultMsmqSample uses Core.Threading.ThreadPools.TaskQueues.DefaultTaskQueue (see the App.config file for more details).
    • CoreDefaultSample uses CoreDefaultSample.TaskQueue (see the App.config file for more details).
• 10 Jul 2008
  • Added Mike.Strobel's suggestion.
  • + ActionTaskItem.
  • * ExtendedThreadPool added AddTask(Action action) and AddTask(Action action, ThreadPriority priority) methods.
  • Added more tests
• 02 Oct 2009
  • + PriorityTaskQueue.
  • + StatisticController.
  • * ExtendedThreadPool added support StatisticController, MultiThreadingCapacityType.
  • Added support Unity 1.2
  • Added CorePrioritySample. The project uses Core.Threading.ThreadPools.TaskQueues.PriorityTaskQueue (see the App.config file for more details).
  • Added more tests 
• 07 Apr 2013 
  •  ExtendedThreadPool  v2
  •  Unity has been removed
  •  ExtendedThreadPool creation has been simplified 
• 22 Apr 2015
  • ExtendedThreadPool -> TinyThreadPool
  • TinyThreadPool the new  creating API based on lambda
  • TinyThreadPool is a part of the Nelibur.Sword NuGet package