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));
        }
    }
}

Answer 2

In the Parallel.Foreach, you read all the bytes of a file, then you order them using LINQ. This is not efficient. Try using Array.Sort. That can be 85% faster for a 25 Mb file.

Array.Sort 2230 ms
OrderBy 14650 ms

And because OnStop method waits the end of any iteration that is already started, it could stops your service a lot faster.

var fileBinary = File.ReadAllBytes(fileName);
Array.Sort(fileBinary);
var hash = sha.ComputeHash(fileBinary);

Answer 3

This is a working code. It stop instantly. Please note that the main idea come from : SylF.

But I cannot give a clear explanation why it happens... Update (after your comment below): You found the reason and it pretty nicely explain why you had that behavior. Thanks! I'm really happy to know.

Although the job is done in low priority threads, you should not notice any additional delay on a machine where the CPU is mostly free of works.

Sorry I mixed up your code sample to achieve some tests. But the main idea is to change the scheduler (which seems to be not recommended). But it is the only way I found.

Code:

using System;
using System.Collections.Generic;
using System.ComponentModel;
using System.Data;
using System.Diagnostics;
using System.IO;
using System.Linq;
using System.Security.Cryptography;
using System.ServiceProcess;
using System.Text;
using System.Threading;
using System.Threading.Tasks;

namespace StackOverflowQuestionWindowsService1
{
    public partial class Service1 : ServiceBase
    {
        private ManualResetEvent stopEvent = new ManualResetEvent(false);
        private Task mainTask;
        private StreamWriter writer = File.CreateText(@"C:\Temp\Log.txt");     //TAKE CARE - I do not append anymore  ********
        private CancellationTokenSource cancellationTokenSource = new CancellationTokenSource();
        private int count = 0;

        public Service1()
        {
            InitializeComponent();

            writer.AutoFlush = true;
        }

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

            Task.Run(()=>Run());
        }

        protected override void OnStop()
        {
            Log("OnStop with actual thread count: " + Process.GetCurrentProcess().Threads.Count);

            cancellationTokenSource.Cancel();
        }

        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()
        {
            Stopwatch stopWatchTotal = new Stopwatch();
            stopWatchTotal.Start();

            try
            {
                using (var sha = SHA256.Create())
                {
                    var parallelOptions = new ParallelOptions();
                    parallelOptions.MaxDegreeOfParallelism = -1;
                    parallelOptions.CancellationToken = cancellationTokenSource.Token;
                    parallelOptions.TaskScheduler = new PriorityScheduler(ThreadPriority.Lowest);

                    Parallel.ForEach(Directory.EnumerateFiles(Environment.SystemDirectory),
                        parallelOptions, (fileName, parallelLoopState) =>
                        {
                            // Thread.CurrentThread.Priority = ThreadPriority.Lowest;
                            Stopwatch stopWatch = new Stopwatch();
                            stopWatch.Start();

                            Interlocked.Increment(ref count);

                            if (parallelOptions.CancellationToken.IsCancellationRequested)
                            {
                                Log(String.Format($"{count}"));
                                return;
                            }

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

            stopWatchTotal.Stop();

            Log(FormatTicks(stopWatchTotal.ElapsedTicks));

            writer.Close();
            Process.GetCurrentProcess().Kill();
        }

        private string FormatTicks(long ticks)
        {
            return new TimeSpan(ticks).ToString();
        }
    }
}

Priority Scheduler: (Thanks to Roman Starkov at: StackOverflow which came from Bnaya Eshet at Microsoft)

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

namespace StackOverflowQuestionWindowsService1
{
    public class PriorityScheduler : TaskScheduler
    {
        public static PriorityScheduler AboveNormal = new PriorityScheduler(ThreadPriority.AboveNormal);
        public static PriorityScheduler BelowNormal = new PriorityScheduler(ThreadPriority.BelowNormal);
        public static PriorityScheduler Lowest = new PriorityScheduler(ThreadPriority.Lowest);

        private BlockingCollection<Task> _tasks = new BlockingCollection<Task>();
        private Thread[] _threads;
        private ThreadPriority _priority;
        private readonly int _maximumConcurrencyLevel = Math.Max(1, Environment.ProcessorCount);

        public PriorityScheduler(ThreadPriority priority)
        {
            _priority = priority;
        }

        public override int MaximumConcurrencyLevel
        {
            get { return _maximumConcurrencyLevel; }
        }

        protected override IEnumerable<Task> GetScheduledTasks()
        {
            return _tasks;
        }

        protected override void QueueTask(Task task)
        {
            _tasks.Add(task);

            if (_threads == null)
            {
                _threads = new Thread[_maximumConcurrencyLevel];
                for (int i = 0; i < _threads.Length; i++)
                {
                    int local = i;
                    _threads[i] = new Thread(() =>
                    {
                        foreach (Task t in _tasks.GetConsumingEnumerable())
                            base.TryExecuteTask(t);
                    });
                    _threads[i].Name = string.Format("PriorityScheduler: ", i);
                    _threads[i].Priority = _priority;
                    _threads[i].IsBackground = true;
                    _threads[i].Start();
                }
            }
        }

        protected override bool TryExecuteTaskInline(Task task, bool taskWasPreviouslyQueued)
        {
            return false; // we might not want to execute task that should schedule as high or low priority inline
        }
    }
}

Answer 4

In your parallelOptions object there is a property for a cancellation token source. You would set that property to a new CancellationTokenSource. Then in your parallel loop, you would put a call to parallelOptions.CancellationToken.ThrowIfCancellationRequested(). This would cause your threads to be terminated.

For a detailed example please see: How to: Cancel a Parallel.For or ForEach Loop

Edit: If you would like your service to stop quicker then you probably need to also cancel the execution of the ComputeHash function. Once your thread is in that call, it cannot be cancelled. So the solution is to do block transformations using the TransformBlock method in a loop. During that loop, you would need to check your CancellationToken, or in your case the manual reset event object. If you need guidance on how to do that, then please look at this answer: stop hashing operation using filestream. They have done an excellent job showing how to use block transformations using the MD5 algorithm, but it is directly portable to SHA256 algorithm.

Answer 5

The problem is that when you issue a stop command, your threads each need to see the stop command and then synchronize in order to actually stop. This means your stop will only be as fast as your slowest hash computation. What I would do if I were you, is to rewrite the hash computation part of your code so that it computes the hash iteratively instead of calling the built in function. This way, you can stop in the middle of a hash computation.

SHA256 has methods in it called TransformBlock and TransformFinalBlock to do this.

An example of some code I wrote for my own personal project is:

do
{
    if(SHOULD_STOP)
        STOP();
    oldBytesRead = bytesRead;
    oldBuffer = buffer;

    buffer = new byte[4096];
    bytesRead = stream.Read(buffer, 0, buffer.Length);

    totalBytesRead += bytesRead;

    if (bytesRead == 0)
    {
        hashAlgorithm.TransformFinalBlock(oldBuffer, 0, oldBytesRead);
    }
    else
    {
        hashAlgorithm.TransformBlock(oldBuffer, 0, oldBytesRead, oldBuffer, 0);
    }

    int progress = (int)((double)totalBytesRead * 100 / size);

} while (bytesRead != 0);

return BitConverter.ToString(hashAlgorithm.Hash).Replace("-", "").ToLowerInvariant();