{"id":55929,"date":"2010-04-07T10:00:00","date_gmt":"2010-04-07T10:00:00","guid":{"rendered":"https:\/\/blogs.msdn.microsoft.com\/pfxteam\/2010\/04\/07\/parallelextensionsextras-tour-5-stataskscheduler\/"},"modified":"2010-04-07T10:00:00","modified_gmt":"2010-04-07T10:00:00","slug":"parallelextensionsextras-tour-5-stataskscheduler","status":"publish","type":"post","link":"https:\/\/devblogs.microsoft.com\/dotnet\/parallelextensionsextras-tour-5-stataskscheduler\/","title":{"rendered":"ParallelExtensionsExtras Tour – #5 – StaTaskScheduler"},"content":{"rendered":"

(The full set of ParallelExtensionsExtras Tour posts is available <\/em>here<\/em><\/font><\/a>.)<\/em>  <\/font><\/font><\/p>\n

The Task Parallel Library (TPL) supports a wide array of semantics for scheduling tasks, even though it only includes two in the box (one using the ThreadPool, and one using SynchronizationContext, which exists primarily to run tasks on UI threads).  <\/span>This support comes from the extensibility of the abstract TaskScheduler class, which may be derived from to implement custom scheduling algorithms.  <\/span><\/font><\/font>ParallelExtensionsExtras<\/font><\/a> includes a multitude of such derived types, and in our tour through ParallelExtensionsExtras, we’ll walk through a few of them… today, we’ll start with StaTaskScheduler.<\/p>\n

<\/font><\/font><\/p>\n

In an ideal world, there would be no legacy code; all code would be up-to-date on best practices and would be using the latest and greatest technologies.  <\/span>In the real world, that’s almost never possible, and we need to be able to accommodate designs of the past.  <\/span>Single-threaded apartments (STA) from COM represent one such technology we need to be able to integrate with (rather than describing STA here, for more information see <\/font>http:\/\/msdn.microsoft.com\/en-us\/library\/ms680112(VS.85).aspx)<\/font><\/a>.  <\/span>If a components is STA, it must only be used from an STA thread.  <\/span>However, all of the ThreadPool’s threads are multi-threaded apartment (MTA) threads, and by default Tasks from TPL run on these threads.  <\/span>The good news is that TPL’s implementation is able to run on either MTA or STA threads, and takes into account relevant differences around underlying APIs like WaitHandle.WaitAll (which only supports MTA threads when the method is provided multiple wait handles).  <\/span>Given that, we can take advantage of the TaskScheduler API to implement an StaTaskScheduler which runs tasks on STA threads.  <\/span>This then allows you to use Tasks with STA components.  <\/span>StaTaskScheduler is implemented in the StaTaskScheduler.cs file in ParallelExtensionsExtras.<\/p>\n

<\/font><\/font><\/p>\n

Our scheduler will derive from TaskScheduler and will also implement IDisposable in order to enable shutdown of the STA threads being used by the scheduler.  <\/span>It will contain two fields: a List<Thread> for storing the threads created and used by the scheduler, and a BlockingCollection<Task> for storing all tasks scheduled to the scheduler, enabling the scheduler’s threads to consume and process them.<\/p>\n

<\/font><\/font><\/p>\n

public sealed class StaTaskScheduler : TaskScheduler, IDisposable<\/p>\n

<\/font><\/p>\n

{<\/p>\n

<\/font><\/p>\n

    <\/span>private readonly List<Thread> _threads;<\/p>\n

<\/font><\/p>\n

    <\/span>private BlockingCollection<Task> _tasks;<\/font><\/p>\n

    …<\/font><\/p>\n

}<\/p>\n

<\/font><\/p>\n

\n

 <\/font><\/p>\n<\/p>\n

When created, the scheduler will instantiate the BlockingCollection<Task> and spin up numberOfThreads threads, where numberOfThreads is provided to the constructor of the type.  <\/span>Each of the threads is explicitly marked with ApartmentState.STA.  <\/span>The thread procedure for each thread is a simple dispatch loop that continually takes from the BlockingCollection<Task> <\/span>(blocking if the collection is empty) and executes the discovered Task.<\/p>\n

<\/font><\/font><\/p>\n

public StaTaskScheduler(int numberOfThreads)<\/p>\n

<\/font><\/p>\n

{<\/p>\n

<\/font><\/p>\n

    <\/span>if (numberOfThreads < 1) <\/p>\n

<\/font><\/p>\n

        <\/span>throw new ArgumentOutOfRangeException(“concurrencyLevel”);<\/p>\n

<\/font><\/p>\n

\n

 <\/font><\/p>\n<\/p>\n

    <\/span>_tasks = new BlockingCollection<Task>();<\/p>\n

<\/font><\/p>\n

\n

 <\/font><\/p>\n<\/p>\n

    <\/span>_threads = Enumerable.Range(0, numberOfThreads).Select(i =><\/p>\n

<\/font><\/p>\n

               <\/span>{<\/p>\n

<\/font><\/p>\n

                   <\/span>var thread = new Thread(() =><\/p>\n

<\/font><\/p>\n

                   <\/span>{<\/p>\n

<\/font><\/p>\n

                       <\/span>foreach (var t in <\/font><\/p>\n

                           _tasks.GetConsumingEnumerable())<\/p>\n

<\/font><\/p>\n

                       <\/span>{<\/p>\n

<\/font><\/p>\n

                           <\/span>TryExecuteTask(t);<\/p>\n

<\/font><\/p>\n

                       <\/span>}<\/p>\n

<\/font><\/p>\n

                   <\/span>});<\/p>\n

<\/font><\/p>\n

                   <\/span>thread.IsBackground = true;<\/p>\n

<\/font><\/p>\n

                   <\/span>thread.SetApartmentState(ApartmentState.STA);<\/p>\n

<\/font><\/p>\n

                   <\/span>return thread;<\/p>\n

<\/font><\/p>\n

               <\/span>}).ToList();<\/p>\n

<\/font><\/p>\n

\n

 <\/font><\/p>\n<\/p>\n

    <\/span>_threads.ForEach(t => t.Start());<\/p>\n

<\/font><\/p>\n

}<\/p>\n

<\/font><\/p>\n

\n

 <\/font><\/p>\n<\/p>\n

That’s the most complicated part of the implementation.  <\/span>The QueueTask method on StaTaskScheduler just takes the Task provided by the TPL infrastructure and stores it into the BlockingCollection<Task> (this will be called, for example, during Task.Factory.StartNew):<\/font><\/font><\/p>\n

protected override void QueueTask(Task task)<\/p>\n

<\/font><\/p>\n

{<\/p>\n

<\/font><\/p>\n

    <\/span>_tasks.Add(task);<\/p>\n

<\/font><\/p>\n

}<\/p>\n

<\/font><\/p>\n

\n

 <\/font><\/p>\n<\/p>\n

The GetScheduledTasks method, which is used by the debugging infrastructure in Visual Studio to feed the Parallel Tasks and Parallel Stacks window, returns a copy of the contents of the BlockingCollection<Task>:<\/p>\n

<\/font><\/font><\/p>\n

protected override IEnumerable<Task> GetScheduledTasks()<\/p>\n

<\/font><\/p>\n

{<\/p>\n

<\/font><\/p>\n

    <\/span>return _tasks.ToArray();<\/p>\n

<\/font><\/p>\n

}<\/p>\n

<\/font><\/p>\n

\n

 <\/font><\/p>\n<\/p>\n

TryExecuteTaskInline is implemented to allow task inlining (running the Task on the current thread, which might happen in a call to Task.Wait, Task.WaitAll, Task<TResult>.Result, or Task.RunSynchronously), but only if the current thread is an STA thread:<\/p>\n

<\/font><\/font><\/p>\n

protected override bool TryExecuteTaskInline(<\/p>\n

<\/font><\/p>\n

    <\/span>Task task, bool taskWasPreviouslyQueued)<\/p>\n

<\/font><\/p>\n

{<\/p>\n

<\/font><\/p>\n

    <\/span>return<\/p>\n

<\/font><\/p>\n

        <\/span>Thread.CurrentThread.GetApartmentState() == <\/font><\/p>\n

            ApartmentState.STA &&<\/p>\n

<\/font><\/p>\n

        <\/span>TryExecuteTask(task);<\/p>\n

<\/font><\/p>\n

}<\/p>\n

<\/font><\/p>\n

\n

 <\/font><\/p>\n<\/p>\n

MaximumConcurrencyLevel returns the number of threads utilized by the scheduler:<\/p>\n

<\/font><\/font><\/p>\n

public override int MaximumConcurrencyLevel<\/p>\n

<\/font><\/p>\n

{<\/p>\n

<\/font><\/p>\n

    <\/span>get { return _threads.Count; }<\/p>\n

<\/font><\/p>\n

}<\/p>\n

<\/font><\/p>\n

\n

 <\/font><\/p>\n<\/p>\n

And finally, our Dispose implementation marks the BlockingCollection<Task> as complete for adding (so that as soon as the collection is empty, the threads will exit), waits for all threads to exit, and then cleans up the BlockingCollection<Task>.<\/p>\n

<\/font><\/font><\/p>\n

public void Dispose()<\/p>\n

<\/font><\/p>\n

{<\/p>\n

<\/font><\/p>\n

    <\/span>if (_tasks != null)<\/p>\n

<\/font><\/p>\n

    <\/span>{<\/p>\n

<\/font><\/p>\n

        <\/span>_tasks.CompleteAdding();<\/p>\n

<\/font><\/p>\n

\n

 <\/font><\/p>\n<\/p>\n

        <\/span>foreach (var thread in _threads) thread.Join();<\/p>\n

<\/font><\/p>\n

\n

 <\/font><\/p>\n<\/p>\n

        <\/span>_tasks.Dispose();<\/p>\n

<\/font><\/p>\n

        <\/span>_tasks = null;<\/p>\n

<\/font><\/p>\n

    <\/span>}<\/p>\n

<\/font><\/p>\n

}<\/p>\n

<\/font><\/p>\n

\n

 <\/font><\/p>\n<\/p>\n

With this in place, we can now use our StaTaskScheduler.  <\/span>Commonly, an StaTaskScheduler will be instantiated with only one thread, so that the same thread is used to process all tasks queued to the scheduler:<\/p>\n

<\/font><\/font><\/p>\n

var sta = new StaTaskScheduler(numberOfThreads:1);<\/p>\n

<\/font><\/p>\n

<\/p>\n

…<\/p>\n

<\/font><\/p>\n

<\/p>\n

for(int i=0; i<10; i++) <\/p>\n

<\/font><\/p>\n

<\/p>\n

{<\/p>\n

<\/font><\/p>\n

<\/p>\n

<\/p>\n

    \/\/ All 10 tasks will be executed sequentially <\/p>\n

<\/font><\/p>\n

<\/p>\n

    \/\/ due to using only one thread.<\/p>\n

<\/font><\/p>\n

    Task.Factory.StartNew(() =><\/p>\n

<\/font><\/p>\n

    {<\/p>\n

<\/font><\/p>\n

        … <\/span>\/\/ run code here that requires an STA thread<\/p>\n

<\/font><\/p>\n

    }, CancellationToken.None, TaskCreationOptions.None, sta);<\/font><\/p>\n

}<\/p>\n

<\/font><\/p>\n

\n

 <\/font><\/p>\n<\/p>\n

Another approach, depending on the use case, is to use multiple threads, but to dedicate one STA instance to each thread.  <\/span>For example, let’s say you have an STA object CoolObject that provides a method DoStuff.  <\/span>You want to be able to have multiple tasks take advantage of DoStuff concurrently, but that can’t be done on the same object instance, and unfortunately CoolObject is relatively expensive to instantiate.  <\/span>A solution then is to store one CoolObject instance in thread-local storage for each of the scheduler’s threads, and each task can then access that thread-local copy:<\/p>\n

<\/font><\/font><\/p>\n

var sta = new StaTaskScheduler(numberOfThreads:4);<\/p>\n

<\/font><\/p>\n

var coolObject = new ThreadLocal<CoolObject>(() => new CoolObject());<\/p>\n

<\/font><\/p>\n

…<\/p>\n

<\/font><\/p>\n

for(int i=0; i<N; i++)<\/p>\n

<\/font><\/p>\n

{<\/p>\n

<\/font><\/p>\n

    <\/span>Task.Factory.StartNew(() =><\/p>\n

<\/font><\/p>\n

   <\/span> <\/span>{<\/p>\n

<\/font><\/p>\n

        <\/span>…<\/p>\n

<\/font><\/p>\n

        <\/span>coolObject.Value.DoStuff();<\/p>\n

<\/font><\/p>\n

        <\/span>…<\/p>\n

<\/font><\/p>\n

    <\/span>}, CancellationToken.None, TaskCreationOptions.None, sta);<\/p>\n

<\/font><\/p>\n

}<\/p>\n

<\/font><\/p>\n

\n

 <\/font><\/p>\n<\/p>\n

In general, it’s desirable to avoid STA objects, but with legacy code that’s a luxury we often can’t afford, and it’s great to be able to still use TPL with such objects.<\/span><\/p>\n","protected":false},"excerpt":{"rendered":"

(The full set of ParallelExtensionsExtras Tour posts is available here.)   The Task Parallel Library (TPL) supports a wide array of semantics for scheduling tasks, even though it only includes two in the box (one using the ThreadPool, and one using SynchronizationContext, which exists primarily to run tasks on UI threads).  This support comes from the extensibility […]<\/p>\n","protected":false},"author":360,"featured_media":58792,"comment_status":"open","ping_status":"closed","sticky":false,"template":"","format":"standard","meta":{"_acf_changed":false,"footnotes":""},"categories":[7908],"tags":[7907,7911,7909,7924,7912],"class_list":["post-55929","post","type-post","status-publish","format-standard","has-post-thumbnail","hentry","category-pfxteam","tag-net-4","tag-code-samples","tag-parallel-extensions","tag-parallelextensionsextras","tag-task-parallel-library"],"acf":[],"blog_post_summary":"

(The full set of ParallelExtensionsExtras Tour posts is available here.)   The Task Parallel Library (TPL) supports a wide array of semantics for scheduling tasks, even though it only includes two in the box (one using the ThreadPool, and one using SynchronizationContext, which exists primarily to run tasks on UI threads).  This support comes from the extensibility […]<\/p>\n","_links":{"self":[{"href":"https:\/\/devblogs.microsoft.com\/dotnet\/wp-json\/wp\/v2\/posts\/55929","targetHints":{"allow":["GET"]}}],"collection":[{"href":"https:\/\/devblogs.microsoft.com\/dotnet\/wp-json\/wp\/v2\/posts"}],"about":[{"href":"https:\/\/devblogs.microsoft.com\/dotnet\/wp-json\/wp\/v2\/types\/post"}],"author":[{"embeddable":true,"href":"https:\/\/devblogs.microsoft.com\/dotnet\/wp-json\/wp\/v2\/users\/360"}],"replies":[{"embeddable":true,"href":"https:\/\/devblogs.microsoft.com\/dotnet\/wp-json\/wp\/v2\/comments?post=55929"}],"version-history":[{"count":0,"href":"https:\/\/devblogs.microsoft.com\/dotnet\/wp-json\/wp\/v2\/posts\/55929\/revisions"}],"wp:featuredmedia":[{"embeddable":true,"href":"https:\/\/devblogs.microsoft.com\/dotnet\/wp-json\/wp\/v2\/media\/58792"}],"wp:attachment":[{"href":"https:\/\/devblogs.microsoft.com\/dotnet\/wp-json\/wp\/v2\/media?parent=55929"}],"wp:term":[{"taxonomy":"category","embeddable":true,"href":"https:\/\/devblogs.microsoft.com\/dotnet\/wp-json\/wp\/v2\/categories?post=55929"},{"taxonomy":"post_tag","embeddable":true,"href":"https:\/\/devblogs.microsoft.com\/dotnet\/wp-json\/wp\/v2\/tags?post=55929"}],"curies":[{"name":"wp","href":"https:\/\/api.w.org\/{rel}","templated":true}]}}