Microsoft Parallel Extensions CTP released

In the past years, evolution of computing is turning to the grounds of high level concurrecy. Processors are becoming faster by being multi-core, and not by having higher frequencies. The time is coming so that even simple programs will be concurrent too. Parallel programs are harder to code, harder to test and can cause pseudo-random bugs that can drive programmers crazy. The best solution for this, as to many problems, is a dependable library and/or language support.

There are several types of problems that require specific solutions. Making a responsive UI, for instance. How many application have you seen that simply stay frozen while it's processing something? I've made some of them myself. And the programmer cannot be blamed, most projects don't have the time (and money) to invest on a responsive UI. The WPF framework solves this in a very neat way. Its interface engine is fully concurrent. But that's a subject for other post.

Yesterday, Microsoft released another CTP of its Parallel Extensions (PFX) library. You may get the details at their blog or at MSDN. In case you're not familiar with it, I'll give a short brief here. Basically, the library's comprehended by three parts:

- Task Parallel Library
Provides some imperative ways to express parallelism. The simpler one is the Invoke method. It receives a set of actions in the form of delegates and executes them concurrently, returning after all the actions have completed. If you have a loop in which iterations may run in parallel, you can use the For or the ForEach methods. When the problem requires a little more elaboration, there's the Task system. It's a set of classes that works somewhat like what the ThreadPool class does, but with much greater flexibility and control.

- PLINQ
It's a implementation of LINQ that allows anyone to easily make declarative parallel queries. All it's needed is to transform the IEnumerable<t> into a IParallelEnumerable<t> using the AsParallel() extention method. It's not optimized yet, but it certainly can boost many applications with almost no code change.

- Coordination Data Structures
It has a group of synchronization classes that enhance those that currently exists in the framework. Also, there are some thread-safe collection classes. Coincidently, they've created a ConcurrentQueue class that serves the same purpose as the one I've posted!

A detail that you can't avoid noticing in this framework is the extensive use of delegates. LINQ itself is based on delegates being passed to other functions. This type of construct is called a higher-order function. Another construct that is present in PFX is the continuation. Also, there are recurrent remarks in its documentation about side-effects and why they have to be avoided. All these characteristics are usually associated with functional langagues. No surprise some of the samples are in F#.

Functional programming has lived almost only at the academic world, but its future looks different. It holds a lot of interesting and useful concepts. The disturbing part is that very few programmers really know these concepts. Do you know any skilled/experienced/old programmer that can't understand OO programming? Well, can you really understand functional programming? I don't. At least not yet.

A C# BlockingQueue

Ok, things didn't go as I planned, so I'll stop promising and start delivering. This time I'll share with you a code I've used before.

The problem is quite simple and is not new: exchange messages between threads. One possible solution is message queuing, but it's too bulky to use inside a single process. The simplest way to make threads communicate is through shared variables, but it's obviously too simple.

So, what kind of feature I was looking for? To begin with, it needed to be fast and light. Secondly, it had to work in a fire-and-forget fashion, i.e., I wanted a thread to be able to send a message (or task, job, whatever) to another thread and go on. Obviously, in order to such a thing to function, it would have to use some sort of internal buffer. So, it looked like what I needed was a thread-safe FIFO queue.

All right, I could simply use a Queue<t> and wrap its calls around a lock statement. Or, even easier, use a non-generic Queue and transform if by calling the Queue.Synchronized() method. In both ways, we'd get a simply synchronized thread-safe queue.

Is that all? Of course not. Suppose we have a producer-consumer problem to solve. Usually, when the buffer is empty, the consumer is supposed to idly wait for an new item. How could we accomplish that using a synched queue? If we try to dequeue from an empty Queue, a exception is thrown. We could check the item count before dequeueing, but there would be two steps involved and it wouldn't be thread-safe even if the Queue instance is synched. Another possibility is catching the queue-is-empty exception, which is a very lousy trick. Anyway, we wouldn't be able to avoid writing synchronization code and this is not what I wanted.

So, considering that you are introduced to the problem, let me show the wish list that I gathered in the form of an interface:

public interface IBlockingQueue
{
/// <summary>Max number of item in the queue (-1 means there's no limit)</summary>
int MaxQueueDepth { get; }

/// <summary>Returns false when there's something wrong, e.g., queue full</summary>
bool TryEnqueue(object item);

/// <summary>Returns false when there's no message to read</summary>
bool TryDequeue(out object item);

/// <summary>Returns false when a timeout or an abort happens</summary>
bool WaitEnqueue(object item, int millisecondsTimeout, FuncGetVolatileFlag ExternalAbortSignaled);

/// <summary>Returns false when a timeout or an abort happens</summary>
bool WaitDequeue(out object item, int millisecondsTimeout, FuncGetVolatileFlag ExternalAbortSignaled);
}

public delegate bool FuncGetVolatileFlag();

TryDequeue() and TryEnqueue are two simply synchronized methods to access the queue. The other two (WaitDequeue() and WaitEnqueue()) are much more interesting. The first one will try to dequeue and, if the queue is empty, will wait for a new item. The latter will try to enqueue and, if the queue is full, will wait until there's room for the new item.

But that's still not enough. A common problem in a multithreaded application is when you need to close it and there are some threads still running. There's several possible approaches on how to close them. The safer choice might be to wait for every thread to end, but it might take too long. One alternative is to terminate everyone, but it's very risky. I think the best pick is to ask somehow the threads to stop and wait until they do so. With my blocking queue, you can choose between any of the 3.

Let me explain how. As you can see, I've put two addicional parameters. Both have the same purpose: cancel the action and return the code execution to the client code. Whilst the fisrt is obviously a timeout, one might wonder what that other parameter is. It's a delegate of a type that accepts functions that like this:bool func(). Its name is ExternalAbortSignaled and its type name is FuncGetVolatileFlag, so I hope it's not hard to recognize how it's used.

In order to eliminate any doubts, I'll describe it. The idea is that, while waiting, the ExternalAbortSignaled delegate will be called in short intervals. If it returns true at any time, the waiting ends and no item is enqueued/dequeued. Also the function returns false. In that way, it's possible to provide the queue an anonymous method that merely returns the value of a boolean flag. Whenever the flag is set to true, the wait function will return promptly and the application as a whole will be able to respond faster.

As you may know, java has a built-in BlockingQueue already. Sadly, java has direct support for delegates. So, I think it would be hard to code a functionality like an abort flag in an elegant way.

Now, if you want to know what is the best way to implement the waiting synchronization code, I'll let the explanation to Joseph, which is much more eloquent than me. It might be a good idea to take a look at my source code too. The technique that I've chosen is through Wait and Pulse.

Some ideas for future enhancements that I've thought of are:
- TryPeek() and WaitPeek()
- Transactions

Source code download link at google code

[Updated at 2008-04-28]