Using task parallel library in IEnumerable implementation to achieve speed improvement

Question

Following code is simplified version of the code that I am trying to optimize.

void Main()
{
    var words = new List<string> {"abcd", "wxyz", "1234"};

    foreach (var character in SplitItOut(words))
    {
        Console.WriteLine (character);
    }
}

public IEnumerable<char> SplitItOut(IEnumerable<string> words)
{
    foreach (string word in words)
    {
        var characters = GetCharacters(word);

        foreach (char c in characters)
        {
            yield return c;
        }
    }
}

char[] GetCharacters(string word)
{
    Thread.Sleep(5000);
    return word.ToCharArray();
}

I cannot change the signature of method SplitItOut.The GetCharacters method is expensive to call but is thread safe. The input to SplitItOut method can contain 100,000+ entries and a single call to GetCharacters() method can take around 200ms. It can also throw exceptions which I can ignore. Order of the results do not matter.

In my first attempt I came up with following implementation using TPL which speeds up the things quite a bit, but is blocking till I am done processing all the words.

public IEnumerable<char> SplitItOut(IEnumerable<string> words)
{
    Task<char[][]> tasks = Task<char[][]>.Factory.StartNew(() =>
    {
        ConcurrentBag<char[]> taskResults = new ConcurrentBag<char[]>();

        Parallel.ForEach(words,
            word => 
            {
                taskResults.Add(GetCharacters(word));
            });

        return taskResults.ToArray();
    });

    foreach (var wordResult in tasks.Result)
    {
        foreach (var c in wordResult)
        {
            yield return c;
        }
    }
}

I am looking for any better implementation for method SplitItOut() than this. Lower processing time is my priority here.

Answer 1

I put your code through the profiler built into Visual Studio, and it looks like the overhead of the Task was hurting you. I refactored it slightly to remove the Task, and it improved the performance a bit. Without your actual algorithm and dataset, it's hard to tell exactly what the issue is or where the performance can be improved. If you have VS Premium or Ultimate, there are built-in profiling tools that will help you out a lot. You can also grab the trial of ANTS.

One thing to bear in mind: Don't try to prematurely optimize. If your code is performing acceptably, don't add stuff to possibly make it faster at the expense of readability and maintainability. If it's not performing to an acceptable level, profile it before you start messing with it.

In any case, here's my refactoring of your algorithm:

    public static IEnumerable<char> SplitItOut(IEnumerable<string> words)
    {
        var taskResults = new ConcurrentBag<char[]>();

        Parallel.ForEach(words, word => taskResults.Add(GetCharacters(word)));

        return taskResults.SelectMany(wordResult => wordResult);
    }

Answer 2

If I'm reading your question correctly, you're not looking to just speed up the parallel processing that creates the chars from the words - you would like your enumerable to produce each one as soon as it's ready. With the implementation you currently have (and the other answers I currently see), the SplitItOut will wait until all of the words have been sent to GetCharacters, and all results returned before producing the first one.

In cases like this, I like to think of things as splitting my process into producers and a consumer. Your producer thread(s) will take the available words and call GetCharacters, then dump the results somewhere. The consumer will yield up characters to the caller of SplitItOut as soon as they are ready. Really, the consumer is the caller of SplitItOut.

We can make use of the BlockingCollection as both a way to yield up the characters, and as the "somewhere" to put the results. We can use the ConcurrentBag as a place to put the words that have yet to be split:

static void Main()
        {
            var words = new List<string> { "abcd", "wxyz", "1234"};

            foreach (var character in SplitItOut(words))
            {
                Console.WriteLine(character);
            }
        }


        static char[] GetCharacters(string word)
        {
            Thread.Sleep(5000);
            return word.ToCharArray();
        }

No changes to your main or GetCharacters - since these represent your constraints (can't change caller, can't change expensive operation)

        public static IEnumerable<char> SplitItOut(IEnumerable<string> words)
        {
            var source = new ConcurrentBag<string>(words);
            var chars = new BlockingCollection<char>();

            var tasks = new[]
                   {
                       Task.Factory.StartNew(() => CharProducer(source, chars)),
                       Task.Factory.StartNew(() => CharProducer(source, chars)),
                       //add more, tweak away, or use a factory to create tasks.
                       //measure before you simply add more!
                   };

            Task.Factory.ContinueWhenAll(tasks, t => chars.CompleteAdding());

            return chars.GetConsumingEnumerable();
        }

Here, we change the SplitItOut method to do four things:

1. Initialize a concurrentbag with all of the words we wish to split. (side note: If you want to enumerate over words on demand, you can start a new task to push them in rather than doing it in the constructor)
2. Start up our char "producer" Tasks. You can start a set number, use a factory, whatever. I suggest not going task-crazy before you measure.
3. Signal the BlockingCollection that we are done when all tasks have completed.
4. "Consume" all of the produced chars (we make it easy on ourselves and just return an IEnumerable<char> rather than foreach and yield, but you could do it the long way if you wish)

All that's missing is our producer implementation. I've expanded out all the linq shortcuts to make it clear, but it's super simple:

        private static void CharProducer(ConcurrentBag<string> words, BlockingCollection<char> output)
        {
            while(!words.IsEmpty)
            {
                string word;
                if(words.TryTake(out word))
                {
                    foreach (var c in GetCharacters(word))
                    {
                        output.Add(c);
                    }
                }
            }
        }

This simply

1. Takes a word out of the ConcurrentBag (unless it's empty - if it is, task is done!)
2. Calls the expensive method
3. Puts the output in the BlockingCollection