The Thread Pool in the .NET Framework

The Thread Pool in the .NET Framework

 

 

文/黃忠成

 

 

     Threads一直以來都是程式設計中最有趣也是風險最高的技巧之一,.NET Framework的出現大幅降低了Threads應用的難度及風險,

其所內建的Thread Pool機制則有效的管理Threads的數量及建立的時間,在.NET Framework中會為每個CPU核心建立一個Thread Pool,

以i7 四核8緒來說,.NET Framework會預先建立8個Thread Pool,但其可容納的Threads數量則依據版本不同而改變。

在2.0 SP1之前,每個Thread Pool預設最大可容納25個Threads,而2.0 SP1 之後則放大為每個核心 250個Threads,在4.0則改變為視乎系統狀態而定,

一般來說4.0 32-bit為1023,64 bit則為32768。

   要特別注意的是,在4.0中,當處於4核8緒的環境下,前8個Threads是立即建立並執行,超過的部分則以每個間隔不超過兩秒為單位逐步建立,這意思是說,

當你排入10個工作至Thread Pool中時,前8個會立即執行,而第9個則會間隔0.5秒後執行,接著再間隔0.5秒後執行第10個,你可以由下列的程式來得到驗證。

    


static void TestThreadPool()
 {
       for (int i = 0; i < 500; i++)
       {
           int i1 = i;
           ThreadPool.QueueUserWorkItem((state) =>
           {
                 Console.WriteLine("{0} on thread {1}", i1, Thread.CurrentThread.ManagedThreadId);
                 Thread.Sleep(500000);
            });
        }
}

static void Main(string[] args)
{
        TestThreadPool();
        Console.ReadKey();
}

執行後會發現相對於你核心數量的Threads會立即建立並執行,之後則以0.5秒的間隔建立其它的,這個間隔最大不會超過2秒,順帶一提,在TPL的行為也是一樣的。

 

Thread Pool的風險

 

    與一般的Threads建立一樣,Thread Pool依然存在當設計不當導致太多Threads同時存在而耗損資源的問題,只是相對於手動建立Threads來說少多了,當遭遇這類情況時,

可以透過設定其最大的Threads數量來限制(ThreadPool.SetMaxThreads)。這個技巧有個小陷阱,你不能設定比自己核心數少的數量,也就是說當4核8緒下,8以下的數字是不被接受的,

另外ThreadPool.SetMaxThreads是全域性的,有時當預期特定的行為可能會導致大量的Threads建立時,我們可以透過一些手段來為其提供獨立且受限於特定數量的Thread Pool(注意,

其仍然受限於ThreadPool MaxThreads的最大數量),最簡單的方式就是透過TPL。


static void TestSequrenceScheduler()
{
    ConcurrentExclusiveSchedulerPair exScheduler = new ConcurrentExclusiveSchedulerPair();
    TaskFactory factory = new TaskFactory(exScheduler.ExclusiveScheduler);
    for (int i = 0; i < 5; i++)
    {
       int i1 = i;
       factory.StartNew(() =>
       {
          Console.WriteLine("{0} on thread {1}", i1, Thread.CurrentThread.ManagedThreadId);
          Thread.Sleep(5000);
       });
    }
}

 

執行這段程式碼會發現同一時間只會有一個Thread執行,如果需要更多則可以利用下面的程式碼。

static void TestSequrenceScheduler2()
{
    ConcurrentExclusiveSchedulerPair exScheduler = new ConcurrentExclusiveSchedulerPair(TaskScheduler.Default, 2);
    TaskFactory factory = new TaskFactory(exScheduler.ConcurrentScheduler);
    for (int i = 0; i < 5; i++)
    {
       int i1 = i;
       factory.StartNew(() =>
       {
           Console.WriteLine("{0} on thread {1}", i1, Thread.CurrentThread.ManagedThreadId);
           Thread.Sleep(5000);
       });
    }
}

透過指定建立ConcurrentExclusiveSchedulerPair物件的第二個參數,可調節這個TaskFactory可同時執行的Threads數量,以本例而言是2個。

如果需要更細緻的來控制數量,那麼下面的程式碼可以完全接管TaskFactory中排程的部分。


using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading;
using System.Threading.Tasks;

namespace ConsoleApplication67
{
    /// <summary>
    /// Provides a task scheduler that ensures a maximum concurrency level while
    /// running on top of the ThreadPool.
    /// </summary>
    public class LimitedConcurrencyLevelTaskScheduler : TaskScheduler
    {
        /// <summary>Whether the current thread is processing work items.</summary>
        [ThreadStatic]
        private static bool _currentThreadIsProcessingItems;
        /// <summary>The list of tasks to be executed.</summary>
        private readonly LinkedList<Task> _tasks = new LinkedList<Task>(); // protected by lock(_tasks)
        /// <summary>The maximum concurrency level allowed by this scheduler.</summary>
        private readonly int _maxDegreeOfParallelism;
        /// <summary>Whether the scheduler is currently processing work items.</summary>
        private int _delegatesQueuedOrRunning = 0; // protected by lock(_tasks)

        /// <summary>
        /// Initializes an instance of the LimitedConcurrencyLevelTaskScheduler class with the
        /// specified degree of parallelism.
        /// </summary>
        /// <param name="maxDegreeOfParallelism">The maximum degree of parallelism provided by this scheduler.</param>
        public LimitedConcurrencyLevelTaskScheduler(int maxDegreeOfParallelism)
        {
            if (maxDegreeOfParallelism < 1) throw new ArgumentOutOfRangeException("maxDegreeOfParallelism");
            _maxDegreeOfParallelism = maxDegreeOfParallelism;
        }

        /// <summary>Queues a task to the scheduler.</summary>
        /// <param name="task">The task to be queued.</param>
        protected sealed override void QueueTask(Task task)
        {
            // Add the task to the list of tasks to be processed.  If there aren't enough
            // delegates currently queued or running to process tasks, schedule another.
            lock (_tasks)
            {
                _tasks.AddLast(task);
                if (_delegatesQueuedOrRunning < _maxDegreeOfParallelism)
                {
                    ++_delegatesQueuedOrRunning;
                    NotifyThreadPoolOfPendingWork();
                }
            }
        }

        /// <summary>
        /// Informs the ThreadPool that there's work to be executed for this scheduler.
        /// </summary>
        private void NotifyThreadPoolOfPendingWork()
        {
            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
                {
                    // Process all available items in the queue.
                    while (true)
                    {
                        Task item;
                        lock (_tasks)
                        {
                            // When there are no more items to be processed,
                            // note that we're done processing, and get out.
                            if (_tasks.Count == 0)
                            {
                                --_delegatesQueuedOrRunning;
                                break;
                            }

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

                        // Execute the task we pulled out of the queue
                        base.TryExecuteTask(item);
                    }
                }
                // We're done processing items on the current thread
                finally { _currentThreadIsProcessingItems = false; }
            }, null);
        }

        /// <summary>Attempts to execute the specified task on the current thread.</summary>
        /// <param name="task">The task to be executed.</param>
        /// <param name="taskWasPreviouslyQueued"></param>
        /// <returns>Whether the task could be executed on the current thread.</returns>
        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) TryDequeue(task);

            // Try to run the task.
            return base.TryExecuteTask(task);
        }

        /// <summary>Attempts to remove a previously scheduled task from the scheduler.</summary>
        /// <param name="task">The task to be removed.</param>
        /// <returns>Whether the task could be found and removed.</returns>
        protected sealed override bool TryDequeue(Task task)
        {
            lock (_tasks) return _tasks.Remove(task);
        }

        /// <summary>Gets the maximum concurrency level supported by this scheduler.</summary>
        public sealed override int MaximumConcurrencyLevel { get { return _maxDegreeOfParallelism; } }

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

使用方法很簡單。


static void TestLimitTaskScheduler()
{
   LimitedConcurrencyLevelTaskScheduler lcts = new LimitedConcurrencyLevelTaskScheduler(20);
   TaskFactory factory = new TaskFactory(lcts);
   for (int i = 0; i < 500; i++)
   {
      int i1 = i;
      factory.StartNew(() =>
      {
          Console.WriteLine("{0} on thread {1}", i1, Thread.CurrentThread.ManagedThreadId);
          Thread.Sleep(5000);
      });
   }
}