Stopping Parallel.ForEach in Windows Service with below normal priority

Question

I have a Parallel.ForEach code in my Windows Service. If ParallelOptions.MaxDegreeOfParallelism is set to -1 I'm using the most of my CPU's. However stopping the service lasts for half a minute. Some internal controller thread that should receive the signal that the service should be stopped is starved out of processor time. I set the process priority to below normal, but that could be irrelevant info here.

What can I do to shorten the time of stopping the service even when all threads are busy?

I was toying with the idea to temporarily lower the priority of the threads from the thread pool, since I don't have any async code, but Internet says that's a bad idea, so asking here for a "proper" way.

The threads (both OS and .NET) are in all cases different between OnStart and OnStop. Also, if stopping is very prolonged then the OS thread in which OnStop will sometimes eventually be called is a new thread, not showing earlier in the log.

To build this code create new Windows service project, add ProjectInstaller class from designer, change Account to LocalService, and install once with InstallUtil. Make sure LocalService can write to C:\Temp.

public partial class Service1 : ServiceBase
{
    private ManualResetEvent stopEvent = new ManualResetEvent(false);
    private Task mainTask;
    private StreamWriter writer = File.AppendText(@"C:\Temp\Log.txt");

    public Service1()
    {
        InitializeComponent();

        writer.AutoFlush = true;
    }

    protected override void OnStart(string[] args)
    {
        Log("--------------");
        Log("OnStart");

        mainTask = Task.Run(new Action(Run));
    }

    protected override void OnStop()
    {
        Log("OnStop");
        stopEvent.Set();

        mainTask.Wait();
        Log("--------------");
    }

    private void Log(string line)
    {
        writer.WriteLine(String.Format("{0:yyyy-MM-dd HH:mm:ss.fff}: [{1,2}] {2}",
            DateTime.Now, Thread.CurrentThread.ManagedThreadId, line));
    }

    private void Run()
    {
        try
        {
            using (var sha = SHA256.Create())
            {
                var parallelOptions = new ParallelOptions();
                parallelOptions.MaxDegreeOfParallelism = -1;

                Parallel.ForEach(Directory.EnumerateFiles(Environment.SystemDirectory),
                    parallelOptions, (fileName, parallelLoopState) =>
                {
                    if (stopEvent.WaitOne(0))
                    {
                        Log("Stop requested");
                        parallelLoopState.Stop();
                        return;
                    }

                    try
                    {
                        var hash = sha.ComputeHash(File.ReadAllBytes(fileName).OrderBy(x => x).ToArray());
                        Log(String.Format("file={0}, sillyhash={1}", fileName, Convert.ToBase64String(hash)));
                    }
                    catch (Exception ex)
                    {
                        Log(String.Format("file={0}, exception={1}", fileName, ex.Message));
                    }
                });
            }
        }
        catch (Exception ex)
        {
            Log(String.Format("exception={0}", ex.Message));
        }
    }
}

Answer 1

This code will stop the service within a second or two while the threads that are already computing will end only after they are done with their actual work. OnStop method receives the signal right away as you can see in Services. But, the TaskManager shows that the process associated with the service will stop only after the consuming threads all finish.

This uses a BlockingCollection of strings (paths) that a separate thread is filling. And there are a number of threads with low priority that will consume the strings.

public partial class Service1 : ServiceBase
{
    private StreamWriter writer = File.AppendText(@"C:\temp\Log.txt");

    const int nbTreads = 30;
    BlockingCollection<string> dataItems;
    bool stopCompute = false;
    List<Thread> threads = new List<Thread>();
    Thread threadProd;
    private object aLock = new object();

    public Service1()
    {
        InitializeComponent();

        dataItems = new BlockingCollection<string>(nbTreads);

        writer.AutoFlush = true;
    }


    protected override void OnStart(string[] args)
    {
        Log("--------------");
        Log("OnStart");
        threadProd = new Thread(new ThreadStart(ProduireNomFichier));
        threadProd.Start();
        Thread.Sleep(1000); // fill the collection a little
        for (int i = 0; i < nbTreads; i++)
        {
            Thread threadRun = new Thread(() => Run());
            threadRun.Priority = ThreadPriority.Lowest;
            threadRun.Start();
            threads.Add(threadRun);
        }
    }

    private void ProduireNomFichier()
    {
        foreach (string nomFichier in Directory.EnumerateFiles(Environment.SystemDirectory))
        {
            dataItems.Add(nomFichier);
        }
    }

    protected override void OnStop()
    {
        lock (aLock)
        {
            stopCompute = true;
        }
        Log("OnStop");
        Log("--------------");
        threadProd.Abort();
    }

    private void Log(string line)
    {
        writer.WriteLine(String.Format("{0:yyyy-MM-dd HH:mm:ss.fff}: [{1,2}] {2}",
            DateTime.Now, Thread.CurrentThread.ManagedThreadId, line));
    }

    private void Run()
    {
        try
        {
            using (var sha = SHA256.Create())
            {
                while (dataItems.TryTake(out string fileName))
                {
                    lock (aLock)
                    {
                        if (stopCompute) return;
                    }
                    try
                    {
                        var hash = sha.ComputeHash(File.ReadAllBytes(fileName).OrderBy(x => x).ToArray());
                        Log(String.Format("file={0}, sillyhash={1}", fileName, Convert.ToBase64String(hash)));
                    }
                    catch (Exception ex)
                    {
                        Log(String.Format("file={0}, exception={1}", fileName, ex.Message));
                    }
                }
            }
        }
        catch (Exception ex)
        {
            Log(String.Format("exception={0}", ex.Message));
        }
    }
}