0

I am pasting 4 code snippets, where Parallelization is attempted using the following APIs:

Parallel.ForEach
Parallel.Invoke
Task.Run
Task.Factory.StartNew

Problem attempts to parallelize such that there are two for loops, inner and outer and for each execution of the outer loop everything in the inner loop should be executed in the Parallel.

I am able to get it working using Parallel.ForEach and all the others lead to exception as for the Inner counter, that value exceeds the array index, when in my view it should not be the case. Also index miss the target by one higher value, for example, when the size of the index is 500, it tries to access value 500, when ideally it should stop at 499. Am I missing something. Following are the code snippets, please have a look:

Parallel.ForEach (Working) 

using System;
using System.Collections.Generic;
using System.Threading;
using System.Threading.Tasks;

namespace TestThreads
{    
    class Elements
    {
        public int innerValue;
        public int outerValue;

        public Elements(int innerValue, int outerValue)
        {
            this.innerValue = innerValue;
            this.outerValue = outerValue;
        }
    }

    class Program
    {
        private static int threadingCounter = 0;
        private static int innerCounter = 500;
        private static int outerCounter = 1000;
        private static int[,] inOut;
        private static int hitCount = 0;
        private static int exceptionCount = 0;
        private static List<int> innerCount = new List<int>();



        static void Main(string[] args)
        {
            inOut = new int[innerCounter, outerCounter];
            Program myProgram = new Program();

            for (int iCount = 0; iCount < innerCounter; iCount++)
                innerCount.Add(iCount);

            try
            {
                for (int outer = 0; outer < outerCounter; outer++)
                {
                    Parallel.ForEach<int>(innerCount, inner =>
                    {
                        myProgram.ThreadCall(new Elements(inner,outer));
                    }
                    );

                    Console.WriteLine("Main Thread Released Post Wait ...");
                    Console.WriteLine("Hit Count :: " + hitCount);
                    Console.WriteLine("Exception Count :: " + exceptionCount);

                    if (threadingCounter != 0)
                        Console.WriteLine("Threading Counter post Inner Loop :: " + threadingCounter);
                }
            }
            catch (Exception ex)
            {
                Console.WriteLine("Exception :: " + ex.Message);
            }
            finally
            {                
                if (innerCount != null)
                    innerCount = null;

            }
        }


        public void ThreadCall(object state)
        {
            try
            {
                Interlocked.Increment(ref hitCount);

                Elements localElement = (Elements)state;
                int localInner = localElement.innerValue;
                int localOuter = localElement.outerValue;
                int finalValue = inOut[localInner, localOuter];
            }
            catch (Exception ex)
            {               
                Console.WriteLine("Exception :: " + ex.Message);
            }
        }
    }
}

Parallel.Invoke (Failing with array out of Index value, tries to fetch inner index - 500)

using System;
using System.Collections.Generic;
using System.Threading;
using System.Threading.Tasks;

namespace TestThreads
{
    /// <summary>
    /// 
    /// </summary>
    class Elements
    {
        public int innerValue;
        public int outerValue;

        public Elements(int innerValue, int outerValue)
        {
            this.innerValue = innerValue;
            this.outerValue = outerValue;
        }
    }

    /// <summary>
    /// 
    /// </summary>
    class Program
    {
        private static int innerCounter = 500;
        private static int outerCounter = 1000;
        private static int[,] inOut;
        private static int hitCount = 0;
        private static int exceptionCount = 0;

        private static List<Action> alSync = new List<Action>();

        /// <summary>
        /// 
        /// </summary>
        /// <param name="args"></param>
        static void Main(string[] args)
        {
            inOut = new int[innerCounter, outerCounter];
            Program myProgram = new Program();

            try
            {
                for (int outer = 0; outer < outerCounter; outer++)
                {

                    for (int inner = 0; inner < innerCounter; inner++)
                        alSync.Add(() => myProgram.ThreadCall(new Elements(inner, outer)));

                    Parallel.Invoke(alSync.ToArray());                            
                    alSync.Clear();

                    Console.WriteLine("Main Thread Released Post Wait ...");
                    Console.WriteLine("Hit Count :: " + hitCount);
                    Console.WriteLine("Exception Count :: " + exceptionCount);
                }

            }
            catch (Exception ex)
            {
                Console.WriteLine("Exception :: " + ex.Message);
            }
            finally
            {
                if (alSync != null)
                    alSync = null;
            }
        }

        /// <summary>
        /// 
        /// </summary>
        /// <param name="state"></param>
        public void ThreadCall(object state)
        {
            try
            {
                Interlocked.Increment(ref hitCount);
                Elements localElement = (Elements)state;
                int localInner = localElement.innerValue;
                int localOuter = localElement.outerValue;
                int finalValue = inOut[localInner, localOuter];
            }
            catch (Exception ex)
            {
                Interlocked.Increment(ref exceptionCount);
                Console.WriteLine("Exception :: " + ex.Message);
            }

        }
    }
}

Task.Run (Failing with array out of Index value, tries to fetch inner index - 500)

using System;
using System.Collections.Generic;
using System.Threading;
using System.Threading.Tasks;

namespace TestThreads
{
    /// <summary>
    /// 
    /// </summary>
    class Elements
    {
        public int innerValue;
        public int outerValue;

        public Elements(int innerValue, int outerValue)
        {
            this.innerValue = innerValue;
            this.outerValue = outerValue;
        }
    }

    /// <summary>
    /// 
    /// </summary>
    class Program
    {
        private static int innerCounter = 500;
        private static int outerCounter = 1000;
        private static int[,] inOut;
        private static int hitCount = 0;
        private static int exceptionCount = 0;

        private static List<Task> tlSync = new List<Task>();

        /// <summary>
        /// 
        /// </summary>
        /// <param name="args"></param>
        static void Main(string[] args)
        {
            inOut = new int[innerCounter, outerCounter];
            Program myProgram = new Program();

            try
            {
                for (int outer = 0; outer < outerCounter; outer++)
                {                  

                    for (int inner = 0; inner < innerCounter; inner++)
                        tlSync.Add(Task.Run(() => myProgram.ThreadCall(new Elements(inner, outer))));

                    Task.WaitAll(tlSync.ToArray());
                    tlSync.Clear();

                    Console.WriteLine("Main Thread Released Post Wait ...");
                    Console.WriteLine("Hit Count :: " + hitCount);
                    Console.WriteLine("Exception Count :: " + exceptionCount);
                }

            }
            catch (Exception ex)
            {
                Console.WriteLine("Exception :: " + ex.Message);
            }
            finally
            {               
                if (tlSync != null)
                    tlSync = null;
            }
        }

        /// <summary>
        /// 
        /// </summary>
        /// <param name="state"></param>
        public void ThreadCall(object state)
        {
            try
            {
                Interlocked.Increment(ref hitCount);
                Elements localElement = (Elements)state;
                int localInner = localElement.innerValue;
                int localOuter = localElement.outerValue;
                int finalValue = inOut[localInner, localOuter];
            }
            catch (Exception ex)
            {
                Interlocked.Increment(ref exceptionCount);
                Console.WriteLine("Exception :: " + ex.Message);
            }           

        }
    }
}

Task.Factory.StartNew (Failing with array out of Index value, tries to fetch inner index - 500)

using System;
using System.Collections.Generic;
using System.Threading;
using System.Threading.Tasks;

namespace TestThreads
{
    /// <summary>
    /// 
    /// </summary>
    class Elements
    {
        public int innerValue;
        public int outerValue;

        public Elements(int innerValue, int outerValue)
        {
            this.innerValue = innerValue;
            this.outerValue = outerValue;
        }
    }

    /// <summary>
    /// 
    /// </summary>
    class Program
    {
        private static int innerCounter = 500;
        private static int outerCounter = 1000;
        private static int[,] inOut;
        private static int hitCount = 0;
        private static int exceptionCount = 0;

        private static List<Task> tlSync = new List<Task>();

        /// <summary>
        /// 
        /// </summary>
        /// <param name="args"></param>
        static void Main(string[] args)
        {
            inOut = new int[innerCounter, outerCounter];
            Program myProgram = new Program();

            try
            {
                for (int outer = 0; outer < outerCounter; outer++)
                {

                    for (int inner = 0; inner < innerCounter; inner++)
                        tlSync.Add(Task.Factory.StartNew(() =>  
myProgram.ThreadCall(new Elements(inner, outer))));

                    Task.WaitAll(tlSync.ToArray());
                    tlSync.Clear();

                    Console.WriteLine("Main Thread Released Post Wait ...");
                    Console.WriteLine("Hit Count :: " + hitCount);
                    Console.WriteLine("Exception Count :: " + exceptionCount);
                }

            }
            catch (Exception ex)
            {
                Console.WriteLine("Exception :: " + ex.Message);
            }
            finally
            {
                if (tlSync != null)
                    tlSync = null;
            }
        }

        /// <summary>
        /// 
        /// </summary>
        /// <param name="state"></param>
        public void ThreadCall(object state)
        {
            try
            {
                Interlocked.Increment(ref hitCount);
                Elements localElement = (Elements)state;
                int localInner = localElement.innerValue;
                int localOuter = localElement.outerValue;
                int finalValue = inOut[localInner, localOuter];
            }
            catch (Exception ex)
            {
                Interlocked.Increment(ref exceptionCount);
                Console.WriteLine("Exception :: " + ex.Message);
            }

        }
    }
}
H H
  • 263,252
  • 30
  • 330
  • 514
Mrinal Kamboj
  • 11,300
  • 5
  • 40
  • 74

1 Answers1

0

Not sure what the use of tlSync is i don't see why you would want to push the tasks to a generic list.

If you want to do some stuff and update the ui you can maybe inspire from this function i used for a forms program i once wrote

var ui = TaskScheduler.FromCurrentSynchronizationContext();
someList
    .ForEach(f => Task.Factory.StartNew(() =>
    {
        // do stuff
    })
    .ContinueWith(task =>
    {
        if (task.IsCompleted)
        {
            // update UI
            // called after each task above completes
        }
    }, ui));

Or you might have a closure problem with inner/outer, and you could try changing this code

for (int inner = 0; inner < innerCounter; inner++)
    tlSync.Add(Task.Factory.StartNew(() => myProgram.ThreadCall(new Elements(inner, outer))));

To this

for (int inner = 0; inner < innerCounter; inner++) {
    var nInner = inner;
    var nOuter = outer;
    tlSync.Add(Task.Factory.StartNew(() => myProgram.ThreadCall(new Elements(nInner, nOuter))));    
}
Robert Hoffmann
  • 2,366
  • 19
  • 29
  • Tasks are pushed to a generic list, so that after each the inner loop parallel execution for every value of an Outer loop, same list can be passed on to WaitAll API of the Task as an array to wait for all the tasks to complete, before next sequence of the Outer loop gets executed. I am not looking for what you have suggested. My code is a running code, please suggest any modification that you may consider. – Mrinal Kamboj Jul 16 '13 at 22:44
  • Have you tried moving ``Task.WaitAll(tlSync.ToArray());`` out of the loops. Don't see why it would be needed inside the loop. – Robert Hoffmann Jul 16 '13 at 22:48
  • If that doesn't help try googling: c# task closures – Robert Hoffmann Jul 16 '13 at 22:53
  • That cannot be done, since I am waiting for all the inner loop tasks to complete, before outer loop is pushed further. This is the business requirement that I am working on, so cannot push it outside the last for loop – Mrinal Kamboj Jul 16 '13 at 22:53
  • By closures i mean that your task has a reference to inner, outer which is changing. inner, outer is not a individual variable but a reference to a moving variable. http://stackoverflow.com/questions/451779/how-to-tell-a-lambda-function-to-capture-a-copy-instead-of-a-reference-in-c – Robert Hoffmann Jul 16 '13 at 22:59
  • Thanks Robert, that's indeed the solution, it has helped me in understanding the crux of the issue. – Mrinal Kamboj Jul 17 '13 at 03:53