3
\$\begingroup\$

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;

    }
\$\endgroup\$
  • \$\begingroup\$ Can you post more of the code. E.g. what does getTargetNode look like? What type is calculator? A plain text description of what the code is supposed to be doing would also be helpful. \$\endgroup\$ – RobH Mar 27 '16 at 19:53
  • \$\begingroup\$ Updated the question...unsure how much code to post. Don't want to overwhelm... \$\endgroup\$ – Tums Mar 27 '16 at 20:28
  • \$\begingroup\$ Replacing string concatenation with a StringBuilder will reduce GC burden, if you have millions of strings. But, the more important question is, how slow is graph.getNode, actually? What problem are you trying to solve? Parallel.For only provides a (theoretical) constant-time improvement (i.e. 4x faster if you have 4 cores, at best). In big-O notation, this speed-up is completely irrelevant, especially if this is a programming assignment. If this is a real world app, otoh, you will typically concentrate on providing a progress bar and making the operation run in background. \$\endgroup\$ – Groo Mar 29 '16 at 8:23
  • \$\begingroup\$ All the answers were awesome. I marked Dudi as correct because his suggestions helped the most. Thank you for all your help. \$\endgroup\$ – Tums Mar 29 '16 at 13:56
2
\$\begingroup\$

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.

| improve this answer | |
\$\endgroup\$
  • \$\begingroup\$ The file reading benchmark is how I figured out how to parallel process. ;) Because the file is so large, I though I needed to do the work in the parallel loop. Do you have an example of how to use the ForEach overload you mentioned? And the custom partitioner? \$\endgroup\$ – Tums Mar 29 '16 at 6:07
2
\$\begingroup\$

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)
| improve this answer | |
\$\endgroup\$
  • \$\begingroup\$ Would this really be faster than parallel processing the file? I need to read an process millions of files. \$\endgroup\$ – Tums Mar 29 '16 at 6:00
  • \$\begingroup\$ @Tums it can be. actually I planed to offer it in myself but that's forgotten. anyway like I say you need to measure ;) about the answer, I don't agree with sections 1 and 2 but agree with section 3. If you also think that section 3 make more sense to you, I'm also consider to use Gene S offer. \$\endgroup\$ – Dudi Keleti Mar 29 '16 at 6:19
  • \$\begingroup\$ @Turns. Do you have millions of files, or millions or just lines in one file? Your code suggests the latter, whilst your previous comment the former. \$\endgroup\$ – JohnLBevan Mar 29 '16 at 6:53
  • \$\begingroup\$ 1. Parallel.For uses the default task scheduler, i.e. threads from the .NET thread pool. 2. Tasks in .NET are optimized to use the number of threads equal to the number of cores. Creating more threads than this will actually hurt performance. 3. This is the correct part: a) you won't need to block the reading thread, b) you won't have to allocate that huge string array. \$\endgroup\$ – Groo Mar 29 '16 at 7:48
  • 1
    \$\begingroup\$ Sorry @JohnLBevan I did mean millions of lines, not files. \$\endgroup\$ – Tums Mar 29 '16 at 13:08
1
\$\begingroup\$

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.

| improve this answer | |
\$\endgroup\$
  • \$\begingroup\$ The strongest path is a function that traverses through the graph. I was more interested in how to improve the code I posted though as I am doing something new for me.Do not catch the exceptions? Why not? \$\endgroup\$ – Tums Mar 29 '16 at 13:10
  • \$\begingroup\$ @Tums - I said don't throw generic System.Exception. Catching and handling exceptions is fine - sorry if that wasn't clear. \$\endgroup\$ – RobH Mar 29 '16 at 13:19
  • \$\begingroup\$ Got it....got confused for a second. Thanks. \$\endgroup\$ – Tums Mar 29 '16 at 13:21

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.