Bnaya Eshet : Parallel, TDF

The 3rd day of the SDP

bnaya — Wed, 08 May 2013 04:17:58 GMT

The 3rd day of the SDP was over and I want to thanks all the attendants.
my yesterday session was about C# 5 async and await.

today I will have a full day tutorial on Rx and TPL Dataflow.
you can download the demo code and also the presentation for second and third days from here (the link is also having the demo code for today's sessions).

anyway, I also want to recommend 2of yesterday session.

if you were attend at the conference you should have an access to the sessions video page.
first I want to recommend Ofir Makmal's session about .NET 4.5 CLR improvements.

Ofir was talking about subjects like:

Multi-core JIT
Managed Profile Guided Optimization
Auto-Ngen
Background Server GC

my favorite subject was Multi-core JIT.

the following is a snippet for Multi-core JITing.

Code Snippet
 static void Main(string[] args)
{
    ProfileOptimization.SetProfileRoot(@"C:\MyApplicationFolder");
    ProfileOptimization.StartProfile("startup.profile");
}
 

my second recommendation is Manu Cohen-Yashar's session about .NET cryptography.
Manu's talk is great for anyone who want to use the .NET cryptography library.

Open House at Microsoft

bnaya — Mon, 14 Jan 2013 07:48:22 GMT

Open House at Microsoft

yesterday I was lecturing at Microsoft about VS 2012, .NET 4.5, async/await, Rx and TPL Dataflow.

there was 90 people attended and I hope that everybody has learn something new.

the code sample for this lecture available here.

Tpl Dataflow TOC

bnaya — Tue, 18 Dec 2012 14:03:46 GMT

Tpl Dataflow TOC

Part 1
Part 2: ITargetBlock
Part 3: ISourceBlock
Part 4: Block structure
Part 5: walkthrough

SDP 2012 - Day 2

bnaya — Mon, 19 Nov 2012 16:31:00 GMT

SDP 2012 - Day 2

the SDP 2012 conference has day 2 had completed.

I was speaking about async / await, Rx and TPL Dataflow.

you can find the materials for my lecture in here.

I will give the same lecture on Wednesday next week,
there is a chance that some place is still available,
so if you are interesting, you can check it with Sela Marketing.

My session at the SDP

bnaya — Tue, 13 Mar 2012 12:15:35 GMT

Sela conferences (SDP) is coming near (25-29/3/2012).

you can check the following link for all the SDP's tutorials and sessions http://www.sela.co.il/s/SDP2012/index.html.

I will present 2 full-day tutorials (Yaniv Rodenski will join me on the first one).

the first day Introduction to the Task Parallel Library is a full TPL 4 day which will give you a solid knowledge, HowTo and Consideration about building parallel application using the TPL library.
at the end of this day we will survey some improvement made in .NET 4.5 to the existing TPL Library.

the second day Parallel Programming: One Step Beyond assume that the audience are already familiar with TPL (or participate in the first day).
this day will focus on 3 subject:

The new async/await keywords (.NET 4.5 / C# 5)
TPL Dataflow (which is new level of abstraction above the TPL - also .NET 4.5 / C# 5)
Rx (Reactive Extension) which is a great Library for parallel composition over LINQ.

Tpl Dataflow walkthrough - Part 5

bnaya — Sat, 28 Jan 2012 17:04:31 GMT

Tpl Dataflow walkthrough - Part 5

this post is a complete walkthrough of a web crawler sample that was build purely by using Tpl Dataflow.

it was built on .NET 4.5 / C# 5 (on a virtual machine using VS 11).

I will analyze each part of this sample, both by discussing the Dataflow blocks and the patterns in used.

the sample code is available in here (it is a VS 11 project).

during the walkthrough you will see the following Tpl Dataflow blocks:

TransformBlock
TransformManyBlock
ActionBlock
BroadcastBlock

you will see how the aysnc / await signature of the Dataflow blocks is better for executing an IO bound operation (without freezing a worker ThreadPool thread).

I should also mention that this post is part of the Tpl Dataflow series which you better read before reading this one.

Disclamation: the web crawler sample is for educational purpose only (running web crawler application may be forbidden by the low of your country).

The sample topography:

Tpl Dataflow application is usually a collection of agents which is linked together in order to compose a complete solution. each agent is having its own responsibilities and concerns. the following diagram present the agent topography for this sample:

agents block type and responsibilities

Downloader: the responsibility of the downloader is to download the html of a web page. it is using a TransformBlock<Tin, Tout> which belong to the executer block family. the transform block is getting a url as the input message and it produce the page's html as it output.

the transform block is construct from:

input buffer (for url)
task (do the transformation)
output buffer (for the downloaded html)

the task is taking one message at a time from the input buffer, transform the message by a Func<Tin, Tout> delegate which it get as a constructor parameter and put the result in the output buffer, where it is available for other blocks to consume.

later we will see that our crawler transformation is actually taking Func<Tin, Task<Tout>> which is a better signature for IO bound operations (I will discuss it latter).

the transform block is a propagator block which mean that it exposed both as a target and a source block. it is implementing IPropagatorBlock<Tin, Tout>.

the following snippet show that IPropagatorBlock is simply an encapsulation of ITargetBlock and ISourceBlock.

Code Snippet
 public interface IPropagatorBlock<in TInput, out TOutput>
    : ITargetBlock<TInput>, 
      ISourceBlock<TOutput>, 
      IDataflowBlock
{
}
 

Start crawling

Code Snippet
 var downloader = new TransformBlock<string, string>(
    async (url) =>
    {
        // using IOCP the thread pool worker thread does return to the pool
        WebClient wc = new WebClient();
        string result = await wc.DownloadStringTaskAsync(url);
        return result;
    }, downloaderOptions);
 

as I was mentioning earlier the downloader contractor is getting Func<Tin, Task<Tout>>, therefore we can apply an async lambda expression (line 2). the code await for downloading (at line 6).

if you are not familiar with the async / await concept you can read this post or more posts in here.

anyway while awaiting for the download (DownloadStringTaskAsync) the block's task is actually return its worker thread to the ThreadPool and take advantage of the IOCP (IO Completion Port), this is an IO bound operation which mean that no CPU resources is needed while the network card fetching the data from the network.
it is important to understand that while the network card is handling the request the agent's task does not fetching another message from the buffer, the task will be interrupt when the data will be available.

analyzing the html

the crawler is using 2 agent for analyze the downloaded html:

link parser (which will look for links elements <a href="..."/>)
image parser (which will look for image elements <image src="..."/>)

both agent should be link to the downloader agent.
the problem is that linking both agent directly to the downloader agent will result with starvation of one of the agent.
unlike Rx the most blocks forward messages into the first linked target that accept the message, and ignore other linked targets. which mean that the message will be handle by a single agent at a time.

broadcast behavior can be achieved by using a BroadcastBlock<T> which is part of the pure buffer family.
the broadcast block is construct from:

input buffer
task
output buffer of single item.

the task is fetching a message from the input buffer and place it in the output buffer, from the output buffer the message submit to the linked block.

the broadcast block is getting a Func<T,T> delegate as a constructor parameter, the idea behind it is cloning (which will enable separation of the messages).
if you are passing a reference type message to multiple agents, without cloning, changes that made by one agent will be visible to all the other agents.

the broadcast block will use the cloning delegate before sending the message to the linked agents.
the cloning pattern will ensure that only single block is processing a message instance at a time, this will maintain the message ownership and avoid the needs of data synchronization for thread safety.

the crawler will use the following block definition for broadcasting:

Code Snippet
 var contentBroadcaster = new BroadcastBlock<string>(s => s);
 

in our case the html content is a string which is immutable, therefore no real cloning is needed.

the crawler will link the agents (blocks) to each other after the construction of all the relevant blocks, right now we are focusing on the agents themselves.

Link parser

the link parser is using the following regular expression in order to fetch all the links (<a href=..."/>) out from the html and extract the link's url.

Code Snippet
 private const string LINK_REGEX_HREF = 
    "\\shref=('|\\\")?(?<LINK>http\\://.*?(?=\\1)).*>";
private static readonly Regex _linkRegexHRef = 
    new Regex(LINK_REGEX_HREF);
 

unlike the downloader agent which get a single input (url) and produce a single output (html),
the link parser produce multiple outputs (links) per each input (html).
you can use the transform block and set the output type to array of links but the Tpl Dataflow is having a better block for this scenario.
because the processing of each link is independent of other links, it will be better if the transform output buffer will contain flatten links objects rather then a collection of link's array.

the crawler is using the TransformManyBlock<Tin,Tout>. this block is similar to the transform block with only one difference, the delegate at the constructor parameter is one of the following delegates:

Func<Tin, IEnumerable<Tout>>
Func<Tin, Task<IEnumerable<Tout>>>

the block task will extract the outputs results and put each of the extracted result, separately, in the output buffer.

this is the code for the link parser agent:

Code Snippet
 var linkParser = new TransformManyBlock<string, string>(
       (html) =>
       {
           var output = new List<string>();
           var links = _linkRegexHRef.Matches(html);
           foreach (Match item in links)
           {
               var value = item.Groups["LINK"].Value;
               output.Add(value);
           }
           return output;
       });
 

it is very straight forward, parse each html by using regex and return list of result which the block will extract into the output buffer.

Image parser

the image parser is quit similar to the link parser.
the only differences is that it using different regular expression which extract the image's url.

the regex part is:

Code Snippet
 private const string IMG_REGEX =
    "<\\s*img [^\\>]*src=('|\")?(?<IMG>http\\://.*?(?=\\1)).*>\\s*([^<]+|.*?)?\\s*</a>";
private static readonly Regex _imgRegex =
    new Regex(IMG_REGEX);
 

and the parser agent code is:

Code Snippet
 var imgParser = new TransformManyBlock<string, string>(
        (html) =>
        {
            var output = new List<string>();
            var images = _imgRegex.Matches(html);
            foreach (Match item in images)
            {
                var value = item.Groups["IMG"].Value;
                output.Add(value);
            }
            return output;
        });
 

writer agent

the last operational agent is the writer agent which will download the an image from a url and save it to the local disk.

the writer is using a simple action block, which is a simple executer block that have an input buffer and a task.

the task is fetching messages from the buffer and execute a delegate which is given as constructor parameter.
the delegate signature can be either Action<T> or Funk<T, Task>. the latter one is great for IO bound operation (from the same reasons discussed earlier when we was looking on the transform block signature).

because the writer is doing 2 IO bound operations:

download the image from the web
write the image to the file system

the crawler is using the Funk<T, Task> signature.
the writer code is:

Code Snippet
 var writer = new ActionBlock<string>(async url =>
{
    WebClient wc = new WebClient();
    // using IOCP the thread pool worker thread does return to the pool
    byte[] buffer = await wc.DownloadDataTaskAsync(url);
    string fileName = Path.GetFileName(url);
 
    string name = @"Images\" + fileName;
 
    using (Stream srm = File.OpenWrite(name))
    {
        await srm.WriteAsync(buffer, 0, buffer.Length);
    }
});
 

the first await at line 5, is awaiting until the task will be interrupt by the network card,
and the second await at line 12, will await until it will be interrupt by the file system controller.

you may have been notice that the second await is within a using block, you can read more about this topic at this post.

link it together

right now we are having most of our building blocks and it is time to define the data-flow by linking the block to each other.

the downloader should be link to the content broadcaster which in tern should be linked both to the image and link parser, the image parser should be linked to the writer and the link parser should be linked back to the downloader (so it can crawl farther).

but there is one last issue.
it happens that some web page is having links that is targeting an image. this lead us to more complex linking where the link parser should be linked both to the downloader and having conditional link to the writer for those url that is having an image suffix.
as we discuss earlier having a direct link from the link parser to both the downloader and the writer will results with starvation of one of those agents.
we do need a final broadcast block which will handle this distribution task.

Code Snippet
 var linkBroadcaster = new BroadcastBlock<string>(s => s);
 

the link parser will be linked to the broadcaster and the broadcaster will be liked to both the downloader and the writer.

we have spoke of the conditional link from the link parser and the writer, but it will be more effective if the link parser to the downloader will be link only those pages that are most likely having useful data like php, aspx, htm, ext...

Filtering linked messages

the following predicates will be use in order to filter linked messages:

Code Snippet
 StringComparison comparison = StringComparison.InvariantCultureIgnoreCase;
Predicate<string> linkFilter = link => 
    link.IndexOf(".aspx", comparison) != -1 ||
    link.IndexOf(".php", comparison) != -1 ||
    link.IndexOf(".htm", comparison) != -1 ||
    link.IndexOf(".html", comparison) != -1;
Predicate<string> imgFilter = url =>
    url.EndsWith(".jpg", comparison) ||
    url.EndsWith(".png", comparison) ||
    url.EndsWith(".gif", comparison);
 

the first predicate (line 2) will filter the downloader agent target and the second (line 7) will filter the link parser result which is targeting the writer agent.

compose the data-flow

finally we got to the agent composition.

Code Snippet
 IDisposable disposeAll = new CompositeDisposable(
    // from [downloader] to [contentBroadcaster]
    downloader.LinkTo(contentBroadcaster),
    // from [contentBroadcaster] to [imgParser]
    contentBroadcaster.LinkTo(imgParser),
    // from [contentBroadcaster] to [linkParserHRef]
    contentBroadcaster.LinkTo(linkParser),
    // from [linkParser] to [linkBroadcaster]
    linkParser.LinkTo(linkBroadcaster),
    // conditional link to from [linkBroadcaster] to [downloader]
    linkBroadcaster.LinkTo(downloader, linkFilter, true),
    // from [linkBroadcaster] to [writer]
    linkBroadcaster.LinkTo(writer, imgFilter, true),
    // from [imgParser] to [writer]
    imgParser.LinkTo(writer));
 

each LinkTo operation return a disposable instance which can be use to dispose the link when it no longer needed. the crawler compose all those disposable together into a single disposable called dispose All by using the CompositeDisposable which is part of the Rx library.

you can see the conditional LinkTo at line 11 and 13.
is is very important to set the last parameter of the LinkTo to true if you don't want to dispose the link when the filter doesn't match the criteria.

summary

this post was a walkthrough of a web crawler sample.
the complete sample, which is available in here (VS 11), is also having exception handling, agent termination after x amount of seconds, prevention of processing the same url twice and more. for simplicity the code within this post was a simplified version.

Tpl Dataflow (IDataflowBlock) - Part 5

bnaya — Tue, 20 Dec 2011 01:42:00 GMT

Tpl Dataflow (IDataflowBlock ) - Part 5

the previous post discus the concept ITargetBlock and ISourceBlock,
which is the TDF consumer/Producer contract.

you can find all the post in this series under the TDF tag.

this post focus on the IDataflowBlock contract which is the life-time management contract for all data-flow's blocks.

the IDataflowBlock define single property and 2 methods:

Code Snippet
 public interface IDataflowBlock
{
    Task Completion { get; }
 
    void Complete();
    void Fault(Exception exception);
}
 

ending the processing of Dataflow block is done either by calling Complete() or by Fault() in cases that the data flow should exit into fault state.

in case of Complete the block will finish the processing of all messages that already in it's inner buffer and decline (DecliningPermanently) any incoming messages.

Fault put the block into faulty state and it does not schedule any messages from its inner buffer.

the Completion property is using the TAP (task asynchrony pattern) concept as it API for monitoring the block state.

the completion property return a Task which enable either waiting (Wait) on, continuation (ContinueWith) or await.

the block will set the task completion when it will complete the processing of all the messages within it inner buffer (or when finishing current executing messages in case of faulty state).

because the block is using the task semantic for it's easy to handle block exceptions.

the following code demonstrate the management of ActionBlock lifetime.

Code Snippet
 var ab = new ActionBlock<int>(i =>
{
    Thread.Sleep(1000);
    Console.WriteLine(i);
});
 
ab.Completion.ContinueWith(t =>
{
    if (t.Status == TaskStatus.Faulted)
        Console.WriteLine("Failiur: {0}", t.Exception);
    else
        Console.WriteLine("Complete");
});
 
Console.WriteLine("Proccesing");
 
ab.Post(1);
ab.Post(2);
ab.Post(3);
 
if (Console.ReadKey().KeyChar == 'f')
    (ab as IDataflowBlock).Fault(new Exception("wrong tick"));
else
    ab.Complete();
 

at line 7 we have set a continuation callback which check the completion status and writing the completion information.

point of interest

as we discuss in the previous post the dataflow blocks does have internal task which is responsible for its execution (by default it is one task per block but it can be throttle to work with multiple tasks) this task does release when ever the block become idle (when the block will become active again it will construct new task).

it is important to understand that the completion task is a semantic task and not one of the worker tasks.

in case that you want to create you're own custom block you can use the TaskCompletionSource<T> in order to present the TAP (task asynchrony pattern) semantics.

you can learn more on the TAP concept in here.

Summary

every dataflow block implement the IDataflowBlock contract which enable to control and monitor the block life-time.

Tpl Dataflow - User Group

bnaya — Thu, 18 Aug 2011 03:02:13 GMT

Tpl Dataflow - User Group

Today I was lecturing at Microsoft User Group

about the new TPL Dataflow (TDF) library.

first I want to thanks all attendants.

I will post about more aspects of the Tpl Dataflow in future posts.

the presentation and code samples,

available for download from here and here.

download pdf.

download code samples.