Parallel extension enchancement

Question

As suggested in comments here I created a new topic for next version. I edited a bit original code, added exception processing and optimization, that first N = 1/core count tasks are processed on main thread.

public static void Foreach<T>(ICollection<T> source, Action<T> action)
{
    var toBeProcessedOnMainThread = (int) Math.Ceiling((double)source.Count / Environment.ProcessorCount);
    var countdown = new CountdownEvent(source.Count - toBeProcessedOnMainThread);
    ConcurrentBag<Exception> exceptionsBag = null;
    var onMainThread = source.Take(toBeProcessedOnMainThread);
    foreach (var item in source.Skip(toBeProcessedOnMainThread))
    {
        ThreadPool.QueueUserWorkItem(state =>
        {
            var closure = (T)state;
            try
            {
                action(closure);
            }
            catch (Exception ex)
            {
                HandleException(ref exceptionsBag, countdown, ex);
            }
            finally
            {
                countdown.Signal();
            }
        }, item);
    }
    foreach (T value in onMainThread)
    {
        try
        {
            action(value);
        }
        catch (Exception ex)
        {
            HandleException(ref exceptionsBag, countdown, ex);
        }
    }
    countdown.Wait();
    if (exceptionsBag != null)
        throw new AggregateException(exceptionsBag);
}

private static void HandleException(ref ConcurrentBag<Exception> exceptions, object syncRoot, Exception ex)
{
    if (exceptions == null)
        lock (syncRoot)
            if (exceptions == null)
                exceptions = new ConcurrentBag<Exception>();

    exceptions.Add(ex);
}

I do lock countdown because I do not want to introduce another local variable, and it locks only once and only if exception occurs. Normally it won't be called an thus won't be locked.

Answer 1

RubberDuck is right. It looks like you are reinventing the wheel. Unless you are doing this whole thing for educational purposes, you should rather use TPL. ForEach method does exactly what you are trying to do and most likely does a better job at splitting processing power between the tasks:

Parallel.ForEach(source, action);

Regarding your implementation: CountdownEvent implements IDisposable, so you should dispose it after you are done using it. Or wrap it into a using block.

Also this code is the kind of micro-optimization you should avoid:

if (exceptions == null)
    lock (syncRoot)
        if (exceptions == null)
            exceptions = new ConcurrentBag<Exception>();

Is new ConcurrentBag call a bottleneack of this method? Does it take anywhere near as much resources as calling an Action method? "No? Then don't introduce additional concurrency!" :) Just call ConcurrentBag<Exception> exceptionsBag = new ConcurrentBag<Exception>(); and forget about synchronization. Or at least follow Eric's advice in above link and use Lazy<T>.