{"id":55980,"date":"2009-08-04T17:55:00","date_gmt":"2009-08-04T17:55:00","guid":{"rendered":"https:\/\/blogs.msdn.microsoft.com\/pfxteam\/2009\/08\/04\/parallel-extensions-and-io\/"},"modified":"2009-08-04T17:55:00","modified_gmt":"2009-08-04T17:55:00","slug":"parallel-extensions-and-io","status":"publish","type":"post","link":"https:\/\/devblogs.microsoft.com\/dotnet\/parallel-extensions-and-io\/","title":{"rendered":"Parallel Extensions and I\/O"},"content":{"rendered":"

In this post, we\u2019ll investigate some ways that Parallel Extensions can be used to introduce parallelism and asynchrony to I\/O scenarios.<\/font><\/font><\/p>\n

Here\u2019s a simple scenario.  <\/span>I want to retrieve data from a number of web resources.<\/font><\/font><\/p>\n

static string[] Resources = new string[]<\/span><\/p>\n

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

    <\/span>“http:\/\/www.microsoft.com”, “http:\/\/www.msdn.com”,<\/span><\/p>\n

    <\/span>“http:\/\/www.msn.com”, “http:\/\/www.bing.com”<\/span><\/p>\n

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

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

Using the WebClient class, I might end up with the following.<\/span><\/font><\/font><\/p>\n

var data = new List<byte[]>();<\/span><\/p>\n

var wc = new WebClient();<\/span><\/p>\n

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

foreach (string resource in Resources)<\/span><\/p>\n

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

    <\/span>data.Add(wc.DownloadData(resource));<\/span><\/p>\n

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

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

\/\/ Use the data.<\/span><\/p>\n

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

However, these days, downloading data from the web usually utilizes only a small fraction of my available bandwidth.  <\/span>So there are potential performance gains here, and with TPL\u2019s parallel ForEach loop, they are easily had.<\/font><\/font><\/p>\n

var data = new ConcurrentBag<byte[]>();<\/span><\/p>\n

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

Parallel.ForEach(Resources, resource =><\/span><\/p>\n

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

    <\/span>data.Add((new WebClient()).DownloadData(resource));<\/span><\/p>\n

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

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

\/\/ Use the data.<\/span><\/p>\n

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

Note that WebClient instances do not support multiple pending asynchronous operations (and the class is not thread-safe), so I need a separate instance for each operation.  <\/span>Also, since the normal BCL collections (List<T>, etc.) are not thread-safe, I need something like ConcurrentBag<T> to store the results.  <\/span>Of course, storing all the data in a collection assumes the scenario requires that all retrieval operations complete before processing.  <\/span>If this was not the case, I could start processing each data chunk right after obtaining it right in the loop, exploiting more parallelism.  <\/span>However, for the purposes of this investigation, I wanted to determine the possible performance gains in the absence of CPU-intensive work.<\/font><\/font><\/p>\n

As it turns out, the above often yields linear speedup against sequential, with some variation due to the inconsistent nature of web site response times.  <\/span>And it was pretty straightforward.  <\/span>However, things would have been even easier had I started out with a \u201cLINQ\u201d frame of mind.  <\/span>First, I can convert my original sequential code to a LINQ query.  <\/span>Then, I can turn it into PLINQ using the AsParallel method and use WithDegreeOfParallelism to control the number of concurrent retrievals.<\/font><\/font><\/p>\n

var data =<\/span><\/p>\n

    <\/span>from resource in Resources<\/span><\/p>\n

        <\/span>.AsParallel()<\/span><\/p>\n

        <\/span>.WithDegreeOfParallelism(numConcurrentRetrievals)<\/span><\/p>\n

    <\/span>select (new WebClient()).DownloadData(resource);<\/span><\/p>\n

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

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

foreach (byte[] result in data) { }<\/span><\/p>\n

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

(As an aside, it\u2019s worth noting that WithDegreeOfParallelism causes PLINQ to use exactly<\/i> numConcurrentRetrivals Tasks.  <\/span>This differs from the MaxDegreeOfParallelism option that I could have used with my previous Parallel.ForEach code, because that option sets the maximum<\/i>; the actual number of threads still depends on the ThreadPool\u2019s thread-adjusting logic.)<\/font><\/font><\/p>\n

This code offers enhanced readability and makes storing the data easier.  <\/span>In addition, I can continue on the main thread, as PLINQ queries do not execute until the data they represent is accessed \u2013 that is, when MoveNext is called on the relevant enumerator.  <\/span>However, in this particular case, I don\u2019t want to delay my query\u2019s execution until I need the data; I actually want to execute my query while<\/i> continuing on the main thread.  <\/span>To do so, I can wrap my query in a Task and force its immediate execution using ToArray.<\/font><\/font><\/p>\n

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

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

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

        <\/span>from resource in Resources<\/span><\/p>\n

            <\/span>.AsParallel()<\/span><\/p>\n

            <\/span>.WithDegreeOfParallelism(numConcurrentRetrievals)<\/span><\/p>\n

        <\/span>select (new WebClient()).DownloadData(resource).ToArray();<\/span><\/p>\n

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

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

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

foreach (byte[] result in t.Result) { }<\/span><\/p>\n

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

\/\/ OR, use a continuation<\/span><\/p>\n

t.ContinueWith(dataTask =><\/span><\/p>\n

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

    <\/span>foreach (byte[] result in dataTask.Result) { }<\/span><\/p>\n

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

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

Now, I\u2019ve got asynchrony, and I still get similar speedup.  <\/span>However, there\u2019s still something about this code that is not ideal.  <\/span>The work (sending off download requests and blocking) requires almost no CPU, but it is being done by ThreadPool threads since I\u2019m using the default scheduler.  <\/span>Ideally, threads should only be used for CPU-bound work (when there\u2019s actually work to do).  <\/span>Of course, this probably won\u2019t matter much for most typical client applications, but in scenarios where resources are tight, it could be a serious issue.  <\/span>Therefore, it\u2019s worth investigating how we might reduce the number of blocked threads, perhaps by not using threads at all where possible.<\/font><\/font><\/p>\n

To achieve this, I\u2019ll be using ideas from a previous post: <\/font>Tasks and the Event-based Asynchronous Pattern<\/font><\/a>.  <\/span>That article explained how to create a Task<TResult> from any type that implements the EAP, and it presented an extension method for WebClient (available along with many others in the <\/font>ParallelExtensionsExtras<\/font><\/a>):<\/font><\/font><\/p>\n

public static Task<byte[]> DownloadDataTask(<\/span><\/p>\n

    <\/span>this WebClient webClient, Uri address);<\/span><\/p>\n

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

The key point is that this method produces a Task<TResult> by integrating WebClient\u2019s EAP implementation with a TaskCompletionSource<TResult>, and I can use it to rewrite my scenario.<\/font><\/font><\/p>\n

var tasks = new Queue<Task<byte[]>>();<\/span><\/p>\n

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

foreach (string resource in Resources)<\/span><\/p>\n

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

    <\/span>WebClient wc = new WebClient();<\/span><\/p>\n

    <\/span>tasks.Enqueue(wc.DownloadDataTask(new Uri(resource)));<\/span><\/p>\n

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

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

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

while (tasks.Count > 0)<\/span><\/p>\n

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

   <\/span> <\/span>byte[] result = tasks.Dequeue().Result;<\/span><\/p>\n

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

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

\/\/ OR, use a continuation<\/span><\/p>\n

Task<byte[]>.Factory.ContinueWhenAll(tasks.ToArray(), dataTasks =><\/span><\/p>\n

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

    <\/span>foreach (var dataTask in dataTasks) <\/span><\/p>\n

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

        <\/span>byte[] result = dataTask.Result;<\/span><\/p>\n

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

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

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

With this, I\u2019ve got a solution that uses parallelism for speed-up, is asynchronous, and does not burn more threads than necessary!<\/font><\/font><\/p>\n

To recap, in this post, we considered a typical I\/O scenario.  <\/span>First, we saw how easy it was to arrive at solutions that are better than the sequential one.  <\/span>Then, we delved deeper to discover a more complex solution (integrating EAP with Tasks) that offers even more benefits.<\/font><\/font><\/p>\n

PFX-and-IO.zip<\/a><\/p>\n","protected":false},"excerpt":{"rendered":"

In this post, we\u2019ll investigate some ways that Parallel Extensions can be used to introduce parallelism and asynchrony to I\/O scenarios. Here\u2019s a simple scenario.  I want to retrieve data from a number of web resources. static string[] Resources = new string[] {     “http:\/\/www.microsoft.com”, “http:\/\/www.msdn.com”,     “http:\/\/www.msn.com”, “http:\/\/www.bing.com” };   Using the WebClient class, […]<\/p>\n","protected":false},"author":485,"featured_media":58792,"comment_status":"open","ping_status":"closed","sticky":false,"template":"","format":"standard","meta":{"_acf_changed":false,"footnotes":""},"categories":[7908],"tags":[7907,7910,7912],"class_list":["post-55980","post","type-post","status-publish","format-standard","has-post-thumbnail","hentry","category-pfxteam","tag-net-4","tag-plinq","tag-task-parallel-library"],"acf":[],"blog_post_summary":"

In this post, we\u2019ll investigate some ways that Parallel Extensions can be used to introduce parallelism and asynchrony to I\/O scenarios. Here\u2019s a simple scenario.  I want to retrieve data from a number of web resources. static string[] Resources = new string[] {     “http:\/\/www.microsoft.com”, “http:\/\/www.msdn.com”,     “http:\/\/www.msn.com”, “http:\/\/www.bing.com” };   Using the WebClient class, […]<\/p>\n","_links":{"self":[{"href":"https:\/\/devblogs.microsoft.com\/dotnet\/wp-json\/wp\/v2\/posts\/55980","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\/485"}],"replies":[{"embeddable":true,"href":"https:\/\/devblogs.microsoft.com\/dotnet\/wp-json\/wp\/v2\/comments?post=55980"}],"version-history":[{"count":0,"href":"https:\/\/devblogs.microsoft.com\/dotnet\/wp-json\/wp\/v2\/posts\/55980\/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=55980"}],"wp:term":[{"taxonomy":"category","embeddable":true,"href":"https:\/\/devblogs.microsoft.com\/dotnet\/wp-json\/wp\/v2\/categories?post=55980"},{"taxonomy":"post_tag","embeddable":true,"href":"https:\/\/devblogs.microsoft.com\/dotnet\/wp-json\/wp\/v2\/tags?post=55980"}],"curies":[{"name":"wp","href":"https:\/\/api.w.org\/{rel}","templated":true}]}}