Performing parallel processing on a file

Question

I have some code that reads a file and then does some parallel processing of the data. There are millions of lines in the file and this section of the code is the bottleneck in my program. Any information on how to improve processing times or any other suggestions to improve the code (e.g. error processing, string manipulation, or anything to increase speed) is appreciated. I'm new to parallel processing in .NET.

// Read in the source and target file and start solving the strongest path problem.
        try
        {
            ConcurrentBag<string> strongestPaths = new ConcurrentBag<string>();
            String[] allFileLines = null;
            int maxSize = (int)Math.Floor((double)(Int32.MaxValue / 10000));

            // Allocate memory for the source and target file array
            allFileLines = new String[maxSize];

            using (StreamReader sr = File.OpenText(sourceTargetArg))
            {
                // Find the strongest path for each of the target nodes
                int x = 0;
                while (!sr.EndOfStream)
                {
                    allFileLines[x] = sr.ReadLine();
                    x++;

                    if (x == maxSize || sr.EndOfStream)
                    {                             
                        Parallel.For(0, allFileLines.Length, (i, loopState) => 
                        {
                            if (allFileLines[i] != null)
                            {
                                try
                                {
                                    Node targetNode = getTargetNode(graph, allFileLines[i]);

                                    // If the Target Node was not found, do not process this node, and continue
                                    if (targetNode != null)
                                    {
                                        var path = calculator.GetPath(targetNode);
                                        String targetPath = String.Empty;

                                        for (int j = 0; j < path.Count - 1; j++)
                                        {
                                            targetPath += path[j].Name + " " + path[j].getEdgeWeight(path[j + 1]) + " ";
                                        }
                                        targetPath = sourceNode.Name + " " + targetNode.Name + " " + path[path.Count - 1].PathWeight + ": " + targetPath + targetNode.Name;
                                        strongestPaths.Add(targetPath);
                                    }
                                }
                                catch (Exception e)
                                {
                                    // if this fails continue processing the rest of the target nodes, but warn the user
                                    Console.WriteLine(e.Message);
                                }
                            }
                            else
                            {
                                // Break from the parallel loops when current loops have compeleted
                                // This is to stop processing the the rest of the array
                                loopState.Break();
                            }
                        });

                        // Start processing the next chuck of data
                        x = 0;
                        Array.Clear(allFileLines, 0, allFileLines.Length);
                    }
                }
            }               
        }
        catch (OutOfMemoryException)
        {
            Console.WriteLine("Exception: Not enough memory to read in the source and target file.");
            Console.WriteLine();
        }
        catch (Exception e)
        {
            Console.WriteLine("Exception: " + e.Message);
            Console.WriteLine();
        }
        finally
        {
            // Release the memory in case
            if (allFileLines != null)
            {
                Array.Clear(allFileLines, 0, allFileLines.Length);
                allFileLines = null;
            }
            // Force garbage collection
            GC.Collect();
        }

EDIT: Added the function for getTargetNode. It is basically returning the node in my graph object. The 'calculator.GetPath' function gets the path from a target node to the source node to print out the path and weights. The paths of all the target nodes read into the source file are then written to an output file.

My graph is basically:

public class Graph
{
    internal ConcurrentDictionary<string, Node> Nodes { get; private set; }

with some functions. The program is for finding the fastest path between a source node and some target nodes on the graph. The slow part of the program above is just looping through the target nodes and displaying the path.

public static Node getTargetNode(Graph graph, string targetLine)
    {
        Node target = null;
        if (!String.IsNullOrEmpty(targetLine))
        {
            try
            {
                // Verify that the target node is a node in the graph
                target = graph.getNode(targetLine);
            }
            catch (KeyNotFoundException e)
            {
                throw new KeyNotFoundException("Invalid Input: The Target Node, " + targetLine.Trim() + ", in the Source and Target file is not a node in the graph. ", e);
            }
            catch (Exception e)
            {
                throw new Exception("Invalid Input: The Target Node, " + targetLine.Trim() + ", in the Source and Target file is invalid: " + e.Message, e);
            }
        }
        else
        {
            throw new Exception("Invalid Input: The Target Node in the Source and Target file is null or empty.");
        }
        return target;

    }

Answer 1

I saw that you anyway read till end of file, so I think (you need to measure it) that its will be faster if you'll read it all (line by line as you do) and then parallel the work once and avoid construct the parallelism for every chunk.

Also if you use Parallel.Foreach, you can avoid the null checking for allFileLines[i].

Consider to use a custom partitioner. You must measure it but take in mind that sometimes it better to have a large amount of data with less loops where a lot of loops with small chunk of data.

About the previous comment, if your inside work is short, a partitioner is your way to get a better performance.

Again, you need measure it but it might be faster if you collect the strongestPaths in local list (lock free) and then aggregate them into global list with lock when each work is complete.

For this you need to use this overload:

Parallel.ForEach<TSource, TLocal>(
    IEnumerable<TSource> source,
    Func<TLocal> localInit,
    Func<TSource, ParallelLoopState, TLocal, TLocal> body,
    Action<TLocal> localFinally);

ArrayClear in each loop work - is also can avoided if you use one big chunk. It's not a time consuming but still its need to go over thousands of items and set them to null.

ArrayClear in finally block - in principle, if you set the array to null, the GC will know that all his items are dead, so it redundant to do the clear. I don't now if you decide to do the clear after measuring it. if yes, ignore this comment.

About exceptions, it may be useless for you, but it worth to mention that you can aggregate them inside the loop and decide what to do after the loop is complete. Of course it cost in performance if a lot exception has occured (because the thread safety of the ConcurrentQueue).

    var exceptions = new ConcurrentQueue<Exception>();

    Parallel.ForEach(data, _ =>
    {
        try { throw new Exception(); }                   
        catch (Exception e) { exceptions.Enqueue(e); }
    });

    if (exceptions.Count > 0)
        // handle..

In anyway you need to measure every move because in this kind of work the speed depends on your loop work and in your current hardware.

For further reading look in this series

And you can find a file reading benchmark here

About the getTargetNode method, First, method names need to be PascalCasing Second, not so important but take a look on some changes I made:

public static Node GetTargetNode(Graph graph, string targetLine)
{
    if (string.IsNullOrEMpty(targetLine))
        throw new ArgumentNullException(nameof(targetLine));

    try
    {
        // Verify that the target node is a node in the graph
        return graph.GetNode(targetLine);
    }
    catch (KeyNotFoundException e)
    {
        throw new KeyNotFoundException("Invalid Input: The Target Node, " + targetLine.Trim() + ", in the Source and Target file is not a node in the graph. ", e);
    }
    catch (Exception e)
    {
        throw new Exception("Invalid Input: The Target Node, " + targetLine.Trim() + ", in the Source and Target file is invalid: " + e.Message, e);
    }
}

UPDATE

I'm adding example for partitioner and local finally usage

private static long ParallelPartitionerWithLocal(long from, long to)
{
    long result = 0;
    var partitioner = Partitioner.Create(from, to, 
               (to - from)/(Environment.ProcessorCount));

    Parallel.ForEach(partitioner, () => 0L /*local init*/, 
               (range, loopState, subTotal) /*body*/ =>
               {
                   for (var l = range.Item1; l < range.Item2; ++l)
                        subTotal += l;
                   return subTotal;
               }, 
               subTotal /*local finally*/ => Interlocked.Add(ref result, subTotal));

    return result;
}

Because result is global, we can't to change it inside the loop without lock, so we change a local var and in the end change the result just once.

The partitioner here is simple and based on the range and the numbers of cores.

Answer 2

You might consider using the Producer/Consumer pattern

Your producer will be responsible for reading the file and adding the lines into the BlockingCollection. Your consumer(s) would be the threads that read the lines from the BlockingCollection and process them. You could eliminate the "batching" of lines by just having the producer store one line at a time into the BlockingCollection.

The benefits of this approach:

1. I am not a Parallel.For guru but I suspect that it creates a set of new threads each time it is run, which is time consuming. Using the Producer/Consumer you have started all your threads before you start processing the file.
2. You can instantiate multiple consumers if the producer is reading in the content faster than the consumer(s) can process it.
3. Your code is better segregated so the reading of the file is separated from the processing of the file (Single-Responsibility Principle)

Answer 3

I'd strongly advise against mixing mechanics in the meaning of your code. It took me several reads to understand that your if (x == maxSize || sr.EndOfStream) is simply to achieve batching. You could make life easier by adding an extension method like this:

public static class EnumerableExtensions
{
    public static IEnumerable<IEnumerable<T>> Batch<T>(
        this IEnumerable<T> enumerable, 
        int batchSize)
    {
        if (batchSize < 1) 
        {
            throw new ArgumentOutOfRangeException();
        }
        var counter = 0;
        var batch = new List<T>();
        foreach (var item in enumerable)
        {
            batch.Add(item);
            if (++counter % batchSize == 0)
            {
                yield return batch;
                batch = new List<T>();
            }
        }
        if (batch.Count != 0)
        {
            yield return batch;
        }
    }
}

You can combine that with the File.ReadLines() method to get chunks of your file:

foreach (var batchOfLines in File.ReadLines(filePath).Batch(1000))
{
    // batchOfLines contains up to 1000 lines of your file.

    var paths = ComputeStrongestPaths(batchOfLines.ToArray());
}

The important thing here is that you use ReadLines and not ReadAllLines as the latter will load the whole file into memory.

I'd love to suggest improvements to the part where you compute the 'strongest path' but I don't know what a 'strong path' is and I haven't been able to find a definition with Google either.

One other thing I would suggest is that you don't throw new Exception( ... use a specific exception type or don't catch and wrap them at all.