Bnaya Eshet : Task

Async / Await for .NET 4, Silverlight and Windows Phone

bnaya — Tue, 07 May 2013 09:25:00 GMT

Async / Await for .NET 4, Silverlight and Windows Phone

if you have to target .NET 4, Silverlight and Windows Phone and
still want to use the async / await pattern.

the BCL team provide you with a new NuGet package named Microsoft.Bcl.Async.

this package was announced as stable a few week ago.
so you can check it out if you're having VS 2012 but should target one of the above platforms.

be aware that this package won't work with 2010.

EF 6: Async

bnaya — Tue, 19 Mar 2013 15:02:51 GMT

EF 6: Async

this post is the first in a series about what's new in EF 6.

great improvements are about to come with Entity Framework 6.
it is a major release and the first one since EF become an open source.
each post in the series will be dedicate to a single feature.

this post will focus on a new EF a-sync features.

the first question that should be asked is, why do we need parallel data access?
moreover why do we need a dedicate parallel data access API, rather then using the TPL Task.Run(…) in order to introduce parallelism?

the answer is related both to thread safety and ThreadPool optimization.

a dedicate data access API for parallel programming, ensured a thread safe access, which may not be a trivial task.

furthermore, when the database is executing a query the .NET thread is idle (waiting for the execution result).
if this idle thread has taken from ThreadPool (either directly or through high level API like TPL Task.Run(…)) it could lead to a ThreadPool starvation.

assuming that you have a 2 second query, Task.Run(…) will hold the ThreadPool's thread for 2 second even though it idle while the database is executing the query.

what you really want is an API that operate over IOCP (IO completion port). IOCP API will release the ThreadPool's thread back to the pool and re-claim a thread at the completion time.

at the API level you can find a-sync operations everywhere.
I will present some of the main API.

the following snippets, will use the following entities (code first):

Code Snippet
 public class Author
{
    public Author()
    {
        Blogs = new List<Blog>();
    }
    public int Id { get; set; }
    public string Name { get; set; }
    public IList<Blog> Blogs { get; set; }
}
 
public class Blog
{
    public int Id { get; set; }
    public string Title { get; set; }
    public string Content { get; set; }
    public int AuthorId { get; set; }
 
}
 

and the following context:

Code Snippet
 public class Context : DbContext
{
    public DbSet<Author> Authors { get; set; }
    public DbSet<Blog> Blogs { get; set; }
}
 

a-sync save

Code Snippet
 using (var context = new Context())
{
    context.Authors.Add(author);
    int affected = await context.SaveChangesAsync();
}
 

notice the await at line 4.

a-sync read

Code Snippet
 using (var context = new Context())
{
    Author[] authors = await context.Authors
        .Include(a => a.Blogs)
        .ToArrayAsync();
    return authors;
}
 

line 5, is where the magic happens.

execute command

Code Snippet
 using (var context = new Context())
{
    int affected = await 
        context.Database.ExecuteSqlCommandAsync("DELETE FROM BLOGS");
}
 

summary

EF 6 is having a great parallel API, which respect IOCP and compatible with the new async / await syntax.

Parallel and The C# Memory Model

bnaya — Wed, 06 Mar 2013 06:46:13 GMT

Parallel and The C# Memory Model

Parallel programming can be tricky, both compiler and CPU's optimization can lead into a twilight zone's debugging.

lets take the following code snippet snippet:

Code Snippet
 class Program
{
    static void Main(string[] args)
    {
        Console.WriteLine("Start");
        var u = new Util();
        u.Exec();
        Console.ReadKey();
    }
}
 
public class Util
{
    private bool _stop = true;
    public void Exec()
    {
        Task t = Task.Run(() =>
            {
                bool b = true;
                while (_stop)
                {
                    b = !b;
                }
                Console.WriteLine("Complete {0}", b);
            });
        Thread.Sleep(30);
        _stop = false;
    }
}
 

can you predict the outcome?
will it ever reach the completion at line 24?

you can download the snippet from here.

now download the snippet and try to execute it in the following order:

Compiled in Debug mode and double click on the exe file (without an attached debugger).
Compiled in Release mode and double click on the exe file (without an attached debugger).
Compiled in Release mode and run it F5 (with an attached debugger).

so why does is execute predictably in Debug mode, while behaving quit strange in Release mode?
even stranger why does it execute predictably in Release mode when a debugger attached?

what happening is a matter of a single thread optimization that can occurs at the compiler (JIT) or CPU level which is taking assumption which is not acceptable for parallel execution.

it can be fixed up using different API like Thread.MemoryBarrier and other.

instead of explaining this out, I will offer you a reading of part 1 and part 2 of a great article on those matters.

Summary

the optimization world is still rely on a single thread assumption (with some special instruction for parallel execution).
while the computing world becoming more parallel each day, this single thread assumption priority may have to be change in the future.

in order to avoid pitfalls, you should be aware o it and use a parallel dedicated APIs.

Immutable Collections

bnaya — Sun, 17 Feb 2013 19:45:12 GMT

Immutable Collections

Immutability is a pattern which is suit well parallel programming,
but you have to be aware of a potential memory pressure risk when it's not implemented right or used wisely.

this post will cover a new BCL library (still in its preview stage) which is targeting immutable collections.

.NET is already having Concurrent implementation for Queue, Stack, Bug and Dictionary, which is thread-safe, but other type of collection like List is missing.
another type of collection are read only collection, but those type of collection are not truly thread-safe because it is just a wrapper upon other collection which can be mutate by it's owner.
so iterating on a read only collection doesn't guarantee thread-safety because the collection can be mutate during the iteration by it's original owner.
currently we do have one option to create an immutable collection by using the ToArray() extension method, but as I mentioned before it can lead to a memory pressure while using a large collection.

the BCL team was targeting this issue and are working on a new kind of collections which are immutable.
those collection is having a cool concept which guarantee that an immutable sequence will not changed, but it can be part of another immutable sequence. this mean that you can take an immutable collection and add an item. this will keep the immutable part as is and you will get a new collection that include your addition and the original collection, but if someone is using the original collection he won't be affected by your mutation.
as you can see the BCL immutable collection can be mutate but the mutation will return a new collection and leave the original collection untouched.

Start using the immutable collection

you can start using it though NuGet.
on the current preview version it includes:

ImmutableStack<T>
ImmutableQueue<T>
ImmutableList<T>
ImmutableHashSet<T>
ImmutableSortedSet<T>
ImmutableDictionary<K, V>
ImmutableSortedDictionary<K, V>

Fluent API

I will speak about 3 APIs:

Add
AddRange
Builder

Add

Code Snippet
 [Test]
public void List_AddItem_Test()
{
    var col = ImmutableList<int>.Empty;
    var col1 = col.Add(1);
    var col2 = col1.Add(2);
 
    Assert.AreEqual(0, col.Count);
    Assert.AreEqual(1, col1.Count);
    Assert.AreEqual(2, col2.Count);
}
 

whenever you want an empty instance of immutable collection you should use the static Empty property (there is no public constructor available).

you should always remember that Adding an item will not affect the original collection therefore you must handle the return value (is is the same concept as string.Replace).

finally you can notice at lines 8-10 that the above code was actually construct 3 different immutable collections.

AddRange

both from performance and usability it is better using the AddRange API when you intend to add multiple items.

Code Snippet
 [Test]
public void List_AddRange_Test()
{
    var col = ImmutableList<int>.Empty;
    var col1 = col.AddRange(Enumerable.Range(0, 1000));
 
    Assert.AreEqual(0, col.Count);
    Assert.AreEqual(1000, col1.Count);
}
 

Builder

if you want to efficiently add multiple items and AddRange doesn't do the job for you, you can use a Builder (the concept is quit similar to StringBuilder).
you should remember that unlike the immutable collection a builder is not a thread-safe structure, therefore you shouldn't share its state between threads.
when you complete the mutation, you can extract an immutable collection from the builder.

Code Snippet
 [Test]
public void List_Builder_Test()
{
    var col = Enumerable.Range(0, 1000).ToImmutableList();
    var builder = col.ToBuilder();
    for (int i = 0; i < 100; i++)
    {
        builder.Remove(i * 10);
    }
    var col1 = builder.ToImmutable();
    Assert.AreEqual(1000, col.Count);
    Assert.AreEqual(900, col1.Count);
}
 

line 5: getting a builder from immutable collection.
line 10: getting an immutable back from the builder (after the mutation).

Migration

if you are using a read only collection though the IReadOnlyCollection, IReadOnlyList or IReadOnlyDictionary, migration will be easy because the immutable collections are implementing those interfaces.

Optimization

we already spoke about the memory pressure optimization, but what is the characterize of those collection in compare with the well known BCL mutable collection?
as the BCL team argue that immutable collections have been heavily tuned for maximum performance and minimum GC pressure.

the following table was published by the BCL team.
it compare the performance behavior of the BCL immutable and immutable collection (and remember that this is only a preview version which mean that further optimization can be applied).

you can read more on this at this post.

summary

Immutability is a great pattern for parallelism.
the BCL team are giving us a robust and well design immutable collection.
the potential memory pressure has reduced by the internal data-structure design.
so it is definitely something to wait for its release.

you can read more about it at here.
and if you want more background and comparison to the read only collection, you can check out this post.

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.

Async and AggregateException

bnaya — Mon, 14 Jan 2013 06:19:42 GMT

Async and AggregateException

this post is a complementary to Eran Stiller's post.

I was reading Eran Stiller's post about exception handling using async methods and I want to add a few side notes.

1) the exception handling behavior decisions is well documented in this post (by the TPL team), it decided after they had consider different options.

2) I was suggesting that compiler will check whether the a catch of AggregateException is implemented and if so to avoid the unwrapping behavior.
in this case the it can be assumed that programmer do expect multiple exceptions.
unfortunately, for the current version, I was speaking with them after the release of .NET 4.5.
I hope that we will see this mitigation in future release.

3) for the current version you can use the following work around if you do want to catch aggregate exception:

Code Snippet
 await Task.WhenAll(t1, t2)
    .ContinueWith(t => { if (t.Exception != null) throw t.Exception; });
 

by re-throwing the exception the unwrapped exception will be AggregateException.
it may be wise to wrap this pattern in an extension method because it can be a very repetitive one.

Proper Disclosure

Eran is one of my colleague and definitely one of the more talent Architect I was happened to work with.

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

Will it crash my process

bnaya — Mon, 10 Dec 2012 08:54:52 GMT

Will it crash my process

here is a short question,
it was taken from a real life bug that occurs at one of my customers.

will the following code crash the process?

Code Snippet
 List<object> items = new List<object>();
 
try
{
    Parallel.For(0, 10000000, t =>
    {
        items.Add("1");
    });
 
}
catch (AggregateException ex)
{
}
 

it is a non thread safe code that should throw an IndexOutOfRangeException.

but it wouldn't crash the process,
but what about the following code?

Code Snippet
 List<object> items = new List<object>();
 
try
{
    Parallel.For(0, 10000000, t =>
    {
        items.Add("1");
    });
 
}
catch (AggregateException ex)
{
}
 
items.Add("?");
 

despite the fact that the last addition is seem totally safe,
it will occasion crash the process.

so why does it happens?

the parallel code part was putting the list itself in a corrupted state,
which mean that further access will result in an exception.

the problem is that further access may be locate in a different code file, ant this will make this bug a very illusive one.

Async - Handling multiple Exceptions

bnaya — Mon, 10 Dec 2012 06:30:25 GMT

Async - Handling multiple Exceptions

the SDP conference was ended a few week ago and I finally find a time to write some comments.

It was a very successful conference, the feedback and evaluations, of most sessions scored higher than 4.5, some well-over it. For example, one workshop had a perfect 5 / 5 score, and two other workshops scored 4.92 / 5 and 4.9 / 5. The highest score for a breakout session was 9.39 / 10, which is the highest score we’ve seen to date. my score was 4.90 / 5.

but this post isn't about scores, it is about fixing a sample that I was showing at the SDP about exception handling within the async / await context.

when you do use the great async / await syntax you may run into a problem while trying to handle multiple exception which was thrown from async operations (it can happens in cases like parent child relationship or when using WhenAll API).

by design the catch closure will handle the first exception and ignore all the other, you can read about it in here.

the following code will catch a single NullReferenceException or ArgumentException exception (the AggregateException will be ignored):

Code Snippet
     try
    {
        await Task.Factory.StartNew(() =>
            {
                Task.Factory.StartNew(() => 
                    { throw new NullReferenceException(); },
                        TaskCreationOptions.AttachedToParent);
                Task.Factory.StartNew(() => 
                    { throw new ArgumentException(); },
                        TaskCreationOptions.AttachedToParent);
            });
    }
    catch (AggregateException ex)
    {
        // this catch will never be target
    }
    catch (Exception ex)
    {
        Console.WriteLine("## {0} ##", ex.GetType().Name);
    }
}
 

if you do want to catch the AggregateException you can use the following work around :

Code Snippet
 try
{
    Task t = Task.Factory.StartNew(() =>
        {
            Task.Factory.StartNew(() => 
                { throw new NullReferenceException(); },
                    TaskCreationOptions.AttachedToParent);
            Task.Factory.StartNew(() => 
                { throw new ArgumentException(); },
                    TaskCreationOptions.AttachedToParent);
        });
    await t.ContinueWith(tsk => { if (tsk.Exception != null) throw tsk.Exception; });
}
catch (AggregateException ex)
{
    foreach (var exc in ex.Flatten().InnerExceptions)
    {
        Console.WriteLine(exc.GetType().Name);
    }
}
 

the difference is at line 12, I'm using a continuation that re-throw the AggregateException.

because this is a fairly common practice you may better wrap it within an extension method like the following one:

Code Snippet
 public static Task ThrowMultiple(this Task t)
{
    return t.ContinueWith(tsk => { if (tsk.Exception != null) throw tsk.Exception; });
}
 
public static Task<T> ThrowMultiple<T>(this Task<T> t)
{
    return t.ContinueWith(tsk => 
    {
        if (tsk.Exception != null)
            throw tsk.Exception;
        return tsk.Result;
    });
}
 

you can use it as follow:

Code Snippet
 try
{
    await Task.Factory.StartNew(() =>
        {
            Task.Factory.StartNew(() =>
                { throw new NullReferenceException(); },
                    TaskCreationOptions.AttachedToParent);
            Task.Factory.StartNew(() =>
                { throw new ArgumentException(); },
                    TaskCreationOptions.AttachedToParent);
        }).ThrowMultiple();
}
catch (AggregateException ex)
{
    foreach (var exc in ex.Flatten().InnerExceptions)
    {
        Console.WriteLine(exc.GetType().Name);
    }
}
 

Summary

catching a single exception within an async method is a design decision,
but you may want to override this decision in some scenarios.

I hope that in a future release of .NET framework Microsoft consider to throw an AggregateException when the compiler will detect that the developer was adding an AggregateException catch block.

in the meantime you can use the extension methods.

you should not forget to catch a non AggregateException if you have any code before the await (this code will run synchronously).

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.

SDP 2012 - Day 1

bnaya — Sun, 18 Nov 2012 21:21:40 GMT

SDP 2012 - Day 1

the SDP 2012 conference has started today.

I was speaking about what's new in .NET 4.5 parallelism.

you can find the materials for my lecture in here.

tomorrow I'm giving a full day tutorial about async / await, Rx and TPL Dataflow.

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.

Parallel.ForEach behavior

bnaya — Mon, 27 Feb 2012 16:48:57 GMT

Parallel.ForEach behavior

this post is a direct continuation for the previous post about "Real-life story: Blocking Collection".

(Real-life story: Blocking Collection).ContinueWith (t => this post);
or
await (Real-life story: Blocking Collection);
this post;

my colleague Bram Veldhoen has suggest to demonstrate the behavior of the Parallel.ForEach thread's hunger in more pure fashion which doesn't include BlockingCollection<T> or any other high level Enumerable.

the following code demonstrate the issue by using a slow Enumerable:

Code Snippet
 private static int _getInt = 0;
public static IEnumerable<int> GetInts()
{
    Thread.Sleep(30 * 5000);
    yield return Interlocked.Increment(ref _getInt);
}
 
private static Timer _tmr;
private static void Main()
{
 
    _tmr = new Timer(state =>
        {
            var p = Process.GetCurrentProcess();
            int workerThreads, ioThreads;
            ThreadPool.GetAvailableThreads(out workerThreads, out ioThreads);
            Console.WriteLine("Thread Count = {0}, Available ThreadPool Threads = {1}",
                p.Threads.Count, workerThreads);
        }, null, 0, 1000);
    Parallel.ForEach(GetInts(), i => { });
    Console.ReadKey();
}
 

when running this code we can see the following output:

we can see that thread count trend is constantly raising while the available ThreadPool's worker threads is declining.

you can read the previous post for solution to this issue.

Technorati Tags: SELA,Parallel,Task,Thread,collection
Windows Live Tags: SELA,Parallel,Task,Thread,collection
WordPress Tags: SELA,Parallel,Task,Thread,collection
Blogger Labels: SELA,Parallel,Task,Thread,collection

Real-life story: Blocking Collection

bnaya — Mon, 27 Feb 2012 00:00:00 GMT

Real-life story: Blocking Collection

this post will discuss a real-life story which uncover none trivial (yet logical) behavior which related to Parallel.ForEach and BlockingCollection<T>.

I will explain why it happens and what how can we handle it right.

it all start when Guy Eden from ITG has found that the following code seem to leak memory:

Code Snippet
 private static void Main()
{
    var bc = new BlockingCollection<int>();
    Task.Factory.StartNew(() => 
        Parallel.ForEach(bc.GetConsumingEnumerable(), i => { }));
    Console.ReadKey();
}
 

it was leaking even those the blocking collection were empty (nothing was adding item to the collection).

while monitoring this code we can see a steep memory curve constantly raising up to the sky.

I found it very surprising because it seem quit reasonable scenario and BlockingCollection<T> is a prime member of the new TPL concurrent collection family.

so I have decided to ask Stephen Toub about this behavior and as always Stephen has enlighten me about what is really happening in this scenario.

the following is a snippet from Stephen's response:

"The Parallel.ForEach uses the ThreadPool. ForEach doesn’t use a fixed number of threads, but instead will use whatever threads the pool will make available to it. And the ThreadPool has a starvation mechanism, which will introduce additional threads if all threads in the pool are blocked and not making forward progress."

there are couple of options to cope with this issue:

the first is fairly simple which is to use the MaxDegreeOfParallelism as shown in the following snippet:

Code Snippet
 var bc = new BlockingCollection<int>();
Task.Factory.StartNew(() =>
    {
        ParallelOptions options = new ParallelOptions
            {
                MaxDegreeOfParallelism = 30
            };
        Parallel.ForEach(bc.GetConsumingEnumerable(), options, i => { });
    });
 

but this solution is suffering from inefficient use of threads because we are obviously targeting a long running scenario where it is better not to consume the thread from the ThreadPool.

the other option is to use a custom scheduler (actually this is one of the few scenario where custom scheduler seem right).

the following snippet show a simplify version of the custom scheduler:

Code Snippet
 public sealed class LongRunningTaskScheduler:TaskScheduler, IDisposable
{
    private readonly ConcurrentQueue<Task> _requests = new ConcurrentQueue<Task>();
    private readonly CancellationTokenSource _cancellation = new CancellationTokenSource();
    private readonly ManualResetEventSlim _sync = new ManualResetEventSlim(false);
    private int _counter = 0;
 
    public LongRunningTaskScheduler(int maxDegreeOfParallism)
    {
        for (int i = 0; i < maxDegreeOfParallism; i++)
        {
            Task.Factory.StartNew(Execute, TaskCreationOptions.LongRunning);
        }
    }
    private void Execute()
    { 
        var cancelToken = _cancellation.Token;
        while (!cancelToken.IsCancellationRequested)
        {
            _sync.Wait(cancelToken);
            Task t;
            while(_requests.TryDequeue(out t))
                base.TryExecuteTask(t);
            _sync.Reset();
        }
    }
    protected override void QueueTask(Task task)
    {
        _requests.Enqueue(task);
        _sync.Set();
        int c = Interlocked.Increment(ref _counter);
        Console.Write("{0}, ", c);
    }
    protected override bool TryExecuteTaskInline(Task task, bool taskWasPreviouslyQueued)
    {
        return false;
    }
    protected override IEnumerable<Task> GetScheduledTasks()
    {
        return _requests.ToArray();
    }
    public void Dispose()
    {
        _cancellation.Cancel();
    }
}
 

the scheduler is getting maxDegreeOfParallelism as ctor parameter (better name may be poolSize) and construct a pool of long running tasks.

the tasks will be synchronized using ManualResetEventSlim (which accept cancellation token as parameter to the Wait API AutomaticResetEvent does not have a slim version and does not accept cancellation token) (see lines 4,5,20).

the scheduler will execute the tasks on a non ThreadPool thread, therefore won't interfere the ThreadPool heuristics.

the following snippet show how to use this scheduler:

Code Snippet
 var bc = new BlockingCollection<int>();
Task.Factory.StartNew(() =>
    {
        ParallelOptions options = new ParallelOptions
        {
            TaskScheduler = new LongRunningTaskScheduler(30),
        };
        Parallel.ForEach(bc.GetConsumingEnumerable(), options, i => { Console.WriteLine("."); });
    }, TaskCreationOptions.LongRunning);
 

Summary

you should be aware of the Parallel.ForEach behavior which assume short running actions and try to use whatever threads the pool will make available to it.

in cases that you want to use it for potentially long actions, you better use scheduler that will take off the load to a non thread pool threads or at least use the MaxDegreeOfParallelism.

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.