Running external processes asynchronously in a windows service

Question

I am writing a program that moves csv files from a "queue" folder to a "processing" folder, then a third party process called import.exe is launched taking the csv file path as an argument. Import.exe is a long running task.

I need the program to continue running and checking the queue for new files. For this reason I’ve chosen a Windows Service application as it will be long running.

My problem is that I am overwhelmed with options and can't understand if I should approach this problem with background threads or parallel programming, or most likely a combination of both.

So far I have this code which is just running synchronously. You will quickly see that at the moment, I am just wildly firing processes without anyway to manage or check for completion. I have commented out process.WaitForExit () as obviously this is a blocking call.

public int maxConcurrentProcesses = 10;
protected override void OnStart(string[] args)
    {
        // Set up a timer to trigger every minute.
        System.Timers.Timer timer = new System.Timers.Timer(60000);            
        timer.Elapsed += new System.Timers.ElapsedEventHandler(this.OnTimer);
        timer.Start();
    }

private void OnTimer(object sender, System.Timers.ElapsedEventArgs args)
    {            
        // How many instances of import.exe are running?
        Process[] importProcesses = Process.GetProcessesByName("import");
        int countRunning = importProcesses.Count();

        // If there are less than maxConcurrentProcesses, create as many as needed to reach maxConcurrentProcesses
        if (countRunning < maxConcurrentProcesses)
        {
            int processesToStart = maxConcurrentProcesses - countRunning;
            for (int i = 0; i < processesToStart; i++)
            {
                FireOffImport();
            }
        }
    }

private void FireOffImport()
    {
        // Get the first file returned from the Queue folder
        string filePathSource = GetNextCSVInQueue();

        if (filePathSource != "")
        {
            // …
            // commandArguments = create our arguments here
            // …        
            // Move the file to processing folder here
            // … 

            // Give a new process the import tool location and arguments
            ProcessStartInfo startInfo = new ProcessStartInfo(importLocation + "\\import.exe", commandArguments);
            try
            {
                Process process = Process.Start(startInfo);
                // process.WaitForExit(20000);
                // If the process has exited, there will be 4 csv files created in the same directory as the file.                  
            }
            catch (Exception ex)
            {
               // Deal with exception here
            }
       }
    }

I also tried creating an array of Tasks, and running these asynchronously. But at the end I still had to call Task.WaitAll() before I could read the result. So even if one finishes early, it must wait for the longest running task.

I think I need to try loop through creating processes asynchronously perhaps using tasks, but I don’t understand how to do this as a background process, so that I can keep the service timer checking for number of processes if it needs to create more.

Answer 1

The first improvement of your code coming to mind would be to remove the timer and replace it with a System.IO.FileSystemWatcher and an event handler for the Created event. This way, your code needs not administrate which files were in the queue before and which new ones have arrived. Less code = less problems, usually.

Second, taking the word "Task" seriously hints to really performing a complete import task in 1 System.IO.Tasks.Task instance, including spawning the corresponding import process instance and waiting for it to exit when it is done.

If you wish to limit the amount of import processes running at any time, an alternative of the bookkeeping your code does, would be to replace the scheduler with a scheduler which limits the amount of tasks, allowed to run in parallel compared to the default scheduler. If each task is associated with 1 importer instance and maximum N tasks are allowed to run concurrently, you have maximum N instances of the importer process.

The code below shows (in form of a console application), how the above described could look like, minus the custom scheduler which is covered in the provided link.

using System.Threading.Tasks;

namespace ConsoleApplication4
{
    class Program
    {
        static string importerProcessName = "import.exe";
        static string RootFolder = @"E:\temp\A\";
        static string queuePath = System.IO.Path.Combine(RootFolder, "Queue" );
        static string processingPath = System.IO.Path.Combine(RootFolder, "Processing");
        static string donePath = System.IO.Path.Combine(RootFolder, "Done");
        static void Main(string[] args)
        {
            GrantFolders(); // Make sure we have all our folders ready for action...
            var watcher = new System.IO.FileSystemWatcher(queuePath, "*.txt");
            watcher.Created += watcher_Created;
            watcher.EnableRaisingEvents = true;
            System.Console.ReadLine();
        }
        static Task ProcessFile( string fileName )
        {
            Task task = new Task(() =>
            {
                System.Console.WriteLine("Processing: " + fileName);
                System.IO.File.Move(System.IO.Path.Combine(queuePath, fileName), System.IO.Path.Combine(processingPath, fileName));
                string commandLine = "-import " + System.IO.Path.Combine(processingPath, fileName);
                using (var importer = new System.Diagnostics.Process())
                {
                    importer.StartInfo = new System.Diagnostics.ProcessStartInfo(importerProcessName, commandLine);
                    importer.Start();
                    importer.WaitForExit(20000);
                    System.IO.File.Move(System.IO.Path.Combine(processingPath, fileName), System.IO.Path.Combine(donePath, fileName));
                    System.Console.WriteLine("Done with: " + fileName);
                }
            });
            return task;
        }
        static void watcher_Created(object sender, System.IO.FileSystemEventArgs e)
        {
            System.Console.WriteLine("Found in queue: " + e.Name);
            var task = ProcessFile(e.Name);
            task.Start();
        }

        private static void GrantFolders()
        {
            string[] paths = new string[] { queuePath, processingPath, donePath };
            foreach( var path in paths)
            {
                if(!System.IO.Directory.Exists(path))
                {
                    System.IO.Directory.CreateDirectory(path);
                }
            }
        }
    }
}