Auto implementing looping

Question

I don't know if the title makes sense, but in the app I am writing there are lots of (extension) methods. A simple example:

Objects:

Matter (Burn, Explode, Destroy, Freeze, Heat, Cool)
Atom (Attach, Detach)
<many more>

And a custom collection like:

ImmutableList<T>

And methods like these:

public static class Burner
{
    public static Matter Burn ( this Matter matter )
    {
        // matter is burning ...
    }
}

var matters = new ImmutableList<Matter>();
matters.Burn();

As you can see, Burn works on a single entity, but still appears on ImmutableList. That way I want to manage paralellization (burn in parallel) myself.

How do I do this the most performant way, or the cleanest, or the most maintainable way, or combined?

Firstly I would rather not define another extension method that takes ImmutableList inside each class (Burner, etc), because there are hundreds upon hundreds like these and they are probably gonna look the same. But I am open to ideas.

Also all code is mine, so I can change/add anything in any part of the code, not just the extension methods.

Answer 1

What's wrong with

matters.ForEach(Burner.Burn);

with your own implementation of ForEach?

Answer 2

create your own ForEachParallel extension then if you dont want to use PLinq or something

Answer 3

Here is a simple class that iterates in a parallel fashion.

Emre Aydinceren

Usage:

Parallel.ForEach(matters, matter=> matter.Burn() );

or

matters.ParallelForEach(matter=> matter.Burn());

/// <summary>
/// Provides concurrent processing on a sequence of elements
/// </summary>
public static class Parallel
{
    /// <summary>
    /// Number Of parallel tasks 
    /// </summary>
    public static int NumberOfParallelTasks;


    static Parallel()
    {
        NumberOfParallelTasks =  Environment.ProcessorCount < 65 ?  Environment.ProcessorCount : 64;  
    }

    /// <summary>
    /// Performs the specified action on each element of the sequence in seperate threads.
    /// </summary>
    /// <typeparam name="T">The type of the elements of source.</typeparam>
    /// <param name="source">A sequence that contains elements to perform action</param>
    /// <param name="action">The Action delegate to perform on each element of the IEnumerable.</param>
    public static void ForEach<T>( IEnumerable<T> source, Action<T> action )
    {
        if(source == null) return;

        //create a new stack for parallel task we want to run  , stack is very performant to add and read elements in sequence
        var stacks = new Stack<T>[NumberOfParallelTasks]; 

        //instantiate stacks
        for(var i = 0;i < NumberOfParallelTasks;i++)
        {
            stacks[i] = new Stack<T>();
        }

        var itemCount = 0;

        //spread items in source to all stacks while alternating between stacks
        foreach(var item in source)
        {
            stacks[itemCount++ % NumberOfParallelTasks].Push(item);
        }

        if(itemCount==0)return;

        //if we have less items than number of Parallel tasks we should only spun threads for active stacks
        var activeStackCount = itemCount < NumberOfParallelTasks ? itemCount : NumberOfParallelTasks;

        //events are used to notify thread pool completed
        var events = new ManualResetEvent[activeStackCount];

        for(var index = 0;index < activeStackCount;index++)
        {
            //assign index to a scope variable otherwise in multithreading index will not be consistant
            var listIndex = index;

            events[listIndex] = new ManualResetEvent(false); 

            ThreadPool.QueueUserWorkItem(delegate
            {
                //name the thread for debugging
                if(String.IsNullOrEmpty(Thread.CurrentThread.Name))
                {
                    Thread.CurrentThread.Name = String.Format("Parallel.ForEach Worker Thread No:{0}", listIndex);
                }

                try
                {   
                    //iterate through our stack 
                    var stack = stacks[listIndex];
                    foreach(var item in stack)
                    {
                        action(item); 
                    }   
                }
                finally
                {
                    //fire the event to signal WaitHandle that our thread is completed
                    events[listIndex].Set();
                }

            });
        }

        WaitAll(events);

    }

    private static void WaitAll(WaitHandle[] waitHandles)
    {
        if(Thread.CurrentThread.GetApartmentState() == ApartmentState.STA)
        {
            for(var index = 0;index < waitHandles.Length;index++) WaitHandle.WaitAny(waitHandles);
        }
        else
        {
            WaitHandle.WaitAll(waitHandles); 
        }
    }

    /// <summary>
    /// Performs the specified action on each element of the sequence in seperate threads.
    /// </summary>
    /// <typeparam name="T">The type of the elements of source.</typeparam>
    /// <param name="source">A sequence that contains elements to perform action</param>
    /// <param name="action">The Action delegate to perform on each element of the IEnumerable.</param>
    public static void  ParallelForEach<T>(this IEnumerable<T> source, Action<T> action)
    {
        ForEach(source, action);
    }

}

Answer 4

You may find this article to be an interesting read. It discusses how a parallel foreach might work, both doing it yourself and using the Parallel extensions CTP for .NET 3.5. With the CTP, you can do this (example taken from article above):

using System.Threading;

// A simple string collection
string[] numbers = { "One", "Two", "Three", "Four", "Five", "Six", "Seven",
  "Eight", "Nine", "Ten", "Eleven", "Twelve", "Thirteen", "Fourteen", "Fifteen"};

// equivalent to: foreach (string n in numbers)
Parallel.ForEach<string>(numbers, delegate(string n)
{
  Console.WriteLine("n={0}", n.ToString());
});

You should hesitate to use a CTP in production code, unless it's just for your own projects (in which case you should probably want to try CTPs).