Handle multiple threads, one out one in, in a timed loop

Question

I need to process a large number of files overnight, with a defined start and end time to avoid disrupting users. I've been investigating but there are so many ways of handling threading now that I'm not sure which way to go. The files come into an Exchange inbox as attachments.

My current attempt, based on some examples from here and a bit of experimentation, is:

 while (DateTime.Now < dtEndTime.Value)
 {
            var finished = new CountdownEvent(1);
            for (int i = 0; i < numThreads; i++)
            {


                object state = offset;

                finished.AddCount();
                ThreadPool.QueueUserWorkItem(delegate
                {
                    try
                    {
                        StartProcessing(state);
                    }
                    finally
                    {
                        finished.Signal();
                    }
                });

                offset += numberOfFilesPerPoll;

            }
            finished.Signal();
            finished.Wait(); 


        }

It's running in a winforms app at the moment for ease, but the core processing is in a dll so I can spawn the class I need from a windows service, from a console running under a scheduler, however is easiest. I do have a Windows Service set up with a Timer object that kicks off the processing at a time set in the config file.

So my question is - in the above code, I initialise a bunch of threads (currently 10), then wait for them all to process. My ideal would be a static number of threads, where as one finishes I fire off another, and then when I get to the end time I just wait for all threads to complete. The reason for this is that the files I'm processing are variable sizes - some might take seconds to process and some might take hours, so I don't want the whole application to wait while one thread completes if I can have it ticking along in the background. (edit)As it stands, each thread instantiates a class and passes it an offset. The class then gets the next x emails from the inbox, starting at the offset (using the Exchange Web Services paging functionality). As each file is processed, it's moved to a separate folder. From some of the replies so far, I'm wondering if actually I should grab the e-mails in the outer loop, and spawn threads as needed. To cloud the issue, I currently have a backlog of e-mails that I'm trying to process through. Once the backlog has been cleared, it's likely that the nightly run will have a significantly lower load.

On average there are around 1000 files to process each night.

Update

I've rewritten large chunks of my code so that I can use the Parallel.Foreach and I've come up against an issue with thread safety. The calling code now looks like this:

public bool StartProcessing()
        {

            FindItemsResults<Item> emails = GetEmails();



            var source = new CancellationTokenSource(TimeSpan.FromHours(10));

            // Process files in parallel, with a maximum thread count.
            var opts = new ParallelOptions { MaxDegreeOfParallelism = 8, CancellationToken = source.Token };

            try
            {
                Parallel.ForEach(emails, opts, processAttachment);
            }

            catch (OperationCanceledException)
            {
                Console.WriteLine("Loop was cancelled.");
            }
            catch (Exception err)
            {
                WriteToLogFile(err.Message + "\r\n");
                WriteToLogFile(err.StackTrace + "r\n");
            }
            return true;
        }

So far so good (excuse temporary error handling). I have a new issue now with the fact that the properties of the "Item" object, which is an email, not being threadsafe. So for example when I start processing an e-mail, I move it to a "processing" folder so that another process can't grab it - but it turns out that several of the threads might be trying to process the same e-mail at a time. How do I guarantee that this doesn't happen? I know I need to add a lock, can I add this in the ForEach or should it be in the processAttachments method?

Answer 1

Use the TPL:

Parallel.ForEach( EnumerateFiles(),
                  new ParallelOptions { MaxDegreeOfParallelism = 10 },
                  file => ProcessFile( file ) );

Make EnumerateFiles stop enumerating when your end time is reached, trivially like this:

IEnumerable<string> EnumerateFiles()
{
    foreach (var file in Directory.EnumerateFiles( "*.txt" ))
        if (DateTime.Now < _endTime)
            yield return file;
        else
            yield break;
}

Answer 2

You can use a combination of Parallel.ForEach() along with a cancellation token source which will cancel the operation after a set time:

using System;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;

namespace Demo
{
    static class Program
    {
        static Random rng = new Random();

        static void Main()
        {
            // Simulate having a list of files.
            var fileList = Enumerable.Range(1, 100000).Select(i => i.ToString());

            // For demo purposes, cancel after a few seconds.
            var source = new CancellationTokenSource(TimeSpan.FromSeconds(10));

            // Process files in parallel, with a maximum thread count.
            var opts = new ParallelOptions {MaxDegreeOfParallelism = 8, CancellationToken = source .Token};

            try
            {
                Parallel.ForEach(fileList, opts, processFile);
            }

            catch (OperationCanceledException)
            {
                Console.WriteLine("Loop was cancelled.");
            }
        }

        static void processFile(string file)
        {
            Console.WriteLine("Processing file: " + file);

            // Simulate taking a varying amount of time per file.

            int delay;

            lock (rng)
            {
                delay = rng.Next(200, 2000);
            }

            Thread.Sleep(delay);

            Console.WriteLine("Processed file: " + file);
        }
    }
}

As an alternative to using a cancellation token, you can write a method that returns IEnumerable<string> which returns the list of filenames, and stop returning them when time is up, for example:

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

namespace Demo
{
    static class Program
    {
        static Random rng = new Random();

        static void Main()
        {
            // Process files in parallel, with a maximum thread count.
            var opts = new ParallelOptions {MaxDegreeOfParallelism = 8};
            Parallel.ForEach(fileList(), opts, processFile);
        }

        static IEnumerable<string> fileList()
        {
            // Simulate having a list of files.
            var fileList = Enumerable.Range(1, 100000).Select(x => x.ToString()).ToArray();

            // Simulate finishing after a few seconds.
            DateTime endTime = DateTime.Now + TimeSpan.FromSeconds(10);

            int i = 0;

            while (DateTime.Now <= endTime)
                yield return fileList[i++];
        }

        static void processFile(string file)
        {
            Console.WriteLine("Processing file: " + file);

            // Simulate taking a varying amount of time per file.

            int delay;

            lock (rng)
            {
                delay = rng.Next(200, 2000);
            }

            Thread.Sleep(delay);

            Console.WriteLine("Processed file: " + file);
        }
    }
}

Note that you don't need the try/catch with this approach.

Answer 3

You should consider using Microsoft's Reactive Framework. It lets you use LINQ queries to process multithreaded asynchronous processing in a very simple way.

Something like this:

var query =
    from file in filesToProcess.ToObservable()
    where DateTime.Now < stopTime
    from result in Observable.Start(() => StartProcessing(file))
    select new { file, result };

var subscription =
    query.Subscribe(x =>
    {
        /* handle result */
    });

Truly, that's all the code you need if StartProcessing is already defined.

Just NuGet "Rx-Main".

Oh, and to stop processing at any time just call subscription.Dispose().

Answer 4

This was a truly fascinating task, and it took me a while to get the code to a level that I was happy with it.

I ended up with a combination of the above.

The first thing worth noting is that I added the following lines to my web service call, as the operation timeout I was experiencing, and which I thought was because I'd exceeded some limit set on the endpoint, was actually due to a limit set by microsoft way back in .Net 2.0:

ServicePointManager.DefaultConnectionLimit = int.MaxValue;
ServicePointManager.Expect100Continue = false;

See here for more information:

What to set ServicePointManager.DefaultConnectionLimit to

As soon as I added those lines of code, my processing increased from 10/minute to around 100/minute.

But I still wasn't happy with the looping, and partitioning etc. My service moved onto a physical server to minimise CPU contention, and I wanted to allow the operating system to dictate how fast it ran, rather than my code throttling it.

After some research, this is what I ended up with - arguably not the most elegant code I've written, but it's extremely fast and reliable.

List<XElement> elements = new List<XElement>();
 while (XMLDoc.ReadToFollowing("ElementName"))
     {
   using (XmlReader r = XMLDoc.ReadSubtree())
      {
   r.Read();
   XElement node = XElement.Load(r);
//do some processing of the node here...
elements.Add(node);
}
}
//And now pass the list of elements through PLinQ to the actual web service call, allowing the OS/framework to handle the parallelism

int failCount=0; //the method call below sets this per request; we log and continue

failCount = elements.AsParallel()
                            .Sum(element => IntegrationClass.DoRequest(element.ToString()));

It ended up fiendishly simple and lightning fast.

I hope this helps someone else trying to do the same thing!