Bnaya Eshet : IDataflowBlock

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 (ISourceBlock) - Part 3

bnaya — Sat, 10 Dec 2011 15:35:42 GMT

Tpl Dataflow (ISourceBlock) - Part 3

the previous post discus the concept ITargetBlock
which is the TDF consumer contract.

this post will focus on the source block which is the producer contract.

as mention in previous post, sources and targets engage in a protocol for transferring messages between them.

Source Block:

the source block main responsibility is to produce (or manipulate) data which will be consume by the target.

as we learn in previous post the target may consume the data either directly (push) or indirectly (pull).

a target can be attached to source using the ISourceBlock<T> LinkTo method :

LinkTo:

link to provide the ability to link one or more targets to a source, this is how we can build a Dataflow network.
when target is linked to sources the sources automatically propagate any data they contain to targets (asynchronously upon the source distribution strategy).

Detaching linked target from a source:

the LinkTo method returns a IDisposable which can be use to detach the target from the source (at nay time).

Dataflow network:

actually we can dynamically change the Dataflow network at runtime while attaching and detaching blocks.

Pull:

When source push data to a target, the protocol employed may allow the target to simply accept and take ownership on the offered data, or it may require the target to communicate back to the source (pull).

see the previous post for more details.

Race condition:

when a target is pulling data out of one or more sources it may have to ensure that the data does not simultaneously consumed by other targets.
it is getting even more complicate when a target try to atomically consume messages from multiple sources (which mean either succeeding to consume all messages on none).

Two-phase commit:

by using the two-phase commit protocol the target can atomically consume multiple messages (from one or more sources).

the ISourceBlock<T> define the following methods in order to enable two-phase commit protocol:

ConsumeMeassage (commit)
ReserveMessage (prepare)
ReleaseMessage (roll-back)

Code Snippet
 interface ISourceBlock<out TOutput> : IDataflowBlock
{
    IDisposable LinkTo(
        ITargetBlock<TOutput> target, 
        bool unlinkAfterOne);
 
    TOutput ConsumeMessage(
        DataflowMessageHeader messageHeader, 
        ITargetBlock<TOutput> target, 
        out bool messageConsumed);
    bool ReserveMessage(
        DataflowMessageHeader messageHeader, 
        ITargetBlock<TOutput> target);
    void ReleaseReservation(
        DataflowMessageHeader messageHeader, 
        ITargetBlock<TOutput> target);
}
 

Two-phase commit in action:

when target want to atomically consume messages from one or more sources:

it try to reserve the messages on each source (the message header has the message identifier).
if it failed to reserve one of the messages, it should release other reservation using ReleaseResarvation (roll back).
if all reservation succeed it can safely consume the messages, using ConsumeMessage (commit).

Consume Message:

even without two-phase commit, the ConsumeMessage design to handle race condition.
when ever we call the consume message, the source block assign the consume message's messageConsumed out parameter to indicate whether the consume operation has succeed. this way the source can ensure that the message will be consume by a single target and avoid race conditions.

Summary

using the TDF source and target contract we can apply push and pulling strategy while avoiding race condition.

in the following posts I will speak about the blocks house keeping contract and propagating.

SDP Agenda - TPL and Beyond

bnaya — Thu, 01 Dec 2011 01:13:46 GMT

SDP Agenda - TPL and Beyond

the SDP conference is taking place next week
and we have close the final agenda.

the first day (Introduction to TPL) will include:

TPL introduction
What's new in TPL 4.5

the second day (Parallel Programming: One Step Beyond) will include:

Async (new syntax for .NET 4.5)
Rx - Reactive Extension
TPL Dataflow

Rx - User Group session

bnaya — Tue, 22 Nov 2011 13:28:36 GMT

Rx - User Group session

this is just a reminder,
tomorrow (Wednesday 2011-11-22)
I'm giving Rx (Reactive Extension) session at Microsoft Israel user group.

if you want to hear about Rx you can register at here.

on the other hand if you want a full day tutorial on TPL 4.5 you can resister to Sela Developer Practice (this one is not for free).

Yaniv Rodenski and I will give 2 days tutorials about TPL and Async programing in .NET 4.5.

the first day is a walk through TPL 4 and the new async and await keywords of .NET 4.5.

and the second day will focus on what's new in TPL 4.5, TPL Dataflow and Rx (Reactive Extension).

the agenda still may slightly shift but you should be aware that the second day assume deep understanding of TPL 4, therefore it is very recommended to have both the first and second day.

if you can only attend one day and you don't have real solid understanding and practice in TPL 4, I recommend the first day over the second.

del.icio.us Tags: TPL,Rx,Dataflow,Sela,IObservable,IObserver

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.

Tpl Dataflow - Part 1

bnaya — Tue, 02 Aug 2011 18:57:56 GMT

Tpl Dataflow - Part 1

this post is the first of post series which will focus

on the new Tpl Dataflow library (TDF).

TDF goal is to address high-throughput / low-latency

flow scenario of complex computing / IO intensive / immense traffic.

The library is using a few basic buffering and message base pattern in order to

enable basic block which can be compose together into full scale scenario.

each building block construct as agent which have internal buffer and execution management.

one of the most simple block is the ActionBlock:

each time item is send into the ActionBlock it will be buffer in

the internal buffer, than the Block Task will take

one item at a-time and will process it.

by default the ActionBlock schedule only single Task at a-time,

but latter on this series we will see that this default can be change.

actually the Block release Its Task whenever the buffer become empty.

Code Snippet
 var ab = new ActionBlock<int>(i => Console.Write(i));
Parallel.For(0, 1000, i => ab.Post(i));
 

as we can see the ActionBlock syntax can be very simple.

it define an action (in the above case Console.Write) and it can fed with data, ab.Post(i).

Summary

ActionBlock is one of the simplest block.

in the future post we will survey other blocks, TDF interfaces, blocks overloads and concepts,

we will see how to chain blocks in a row, and we will

discuss some of the architecture decisions.

Point Of Interest:

My lecture on this topic at Microsoft User group has change to 2011-08-17.

you can register using this link.