Multithreading extensions

Question

I created a following class to manage multithreading without extra overhead, which exist when I use Parallel TPL class. It is also useful for systems without TPL.

Can it be enchanced?

    public static void Foreach<T>(this ICollection<T> source, Action<T> action)
    {
        var allDone = new ManualResetEventSlim(false);
        int completed = 0;
        foreach (var item in source)
        {
            ThreadPool.QueueUserWorkItem(state =>
            {
                var closure = (T)state;
                action(closure);
                if (Interlocked.Increment(ref completed) == source.Count)
                    allDone.Set();
            }, item);
        }
        allDone.Wait();
    }

    public static void Do(Action action1, Action action2)
    {
        var firstTask = QueueWaitableTask(action1);
        action2();
        firstTask.Wait();
    }

    public static ManualResetEventSlim QueueWaitableTask(Action action)
    {
        var result = new ManualResetEventSlim(false);
        ThreadPool.QueueUserWorkItem(mse =>
        {
            action();
            ((ManualResetEventSlim) mse).Set();
        }, result);
        return result;
    }

Answer 1

You should consider what happens when a user passes an action which can error:

new List<string> { "a", "b", "c" }.Foreach(_ => { throw new Exception(); });

In this case, your manual reset event will never be signalled and the code will never finish executing. Eeek!

As ManualResetEventSlim was introduced in .Net 4 you may as well use another class that was new at the same time CountdownEvent.

public static void Foreach<T>(this ICollection<T> source, Action<T> action)
{
    var allDone = new CountdownEvent(source.Count);
    foreach (var item in source)
    {
        ThreadPool.QueueUserWorkItem(state =>
        {
            try
            {
                action((T)state);
            }
            finally
            {
                allDone.Signal();
            }
        }, item);
    }
    allDone.Wait();
}

Having said that Foreach is a bad name, it's too similar to List.ForEach and doesn't even hint at the internal behaviour.

You should be checking that source != null and action != null.

Update: If you want to catch exceptions and rethrow them as an AggregateException, add them to a concurrent collection e.g. ConcurrentBag so you don't need to worry about synchronisation.