C# multithreading console app

Question

I making simple multithreading signature program (it calculate SHA256 hash for parts of file). Please check my code and I would be grateful if you tell me about errors in my code

Program.cs

using System;
using System.IO;

namespace Signature
{
    class Program
    {
        public static TaskQueue Tasks;
        public static HashPrinter Printer;

        static void Main(string[] args)
        {
            if (args.Length < 2)
            {
                Console.WriteLine("You must specify two parameters. Example: Signature.exe [filename] [chunkSize]");
                return;
            }
            var filename = args[0];
            int chunk;
            if (!int.TryParse(args[1], out chunk))
            {
                Console.WriteLine("You shoul specify number as [chunkSize]");
                return;
            }
            try
            {
                Tasks = new TaskQueue(Environment.ProcessorCount);
                Printer = new HashPrinter(200000);
                var reader = new FileReader(filename, chunk);
                reader.Read();
            }
            catch (IndexOutOfRangeException ex)
            {
                PrintError(ex.Message, ex.StackTrace);
            }
            catch (FileNotFoundException ioException)
            {
                PrintError(ioException.Message, ioException.StackTrace);
            }
            catch (DirectoryNotFoundException ioException)
            {
                PrintError(ioException.Message, ioException.StackTrace);
            }
            catch (IOException ioException)
            {
                PrintError(ioException.Message,ioException.StackTrace);
            }
            catch (Exception err)
            {
                PrintError(err.Message,err.StackTrace);
            }
            Console.ReadLine();
        }

        private static void PrintError(string message,string trace)
        {
            Console.WriteLine(message);
            Console.WriteLine(trace);
        }
    }
}

TaskQueue.cs

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

namespace Signature
{
    internal class TaskQueue : IDisposable
    {
        private readonly object _locker = new object();
        private readonly Thread[] _threads;
        private readonly int _threadsCount;
        private Queue<KeyValuePair<long, byte[]>> _tasks = new Queue<KeyValuePair<long, byte[]>>();

        public TaskQueue(int threadsCount)
        {
            if (threadsCount <= 0)
                throw new IndexOutOfRangeException("Processor count should be positive");
            _threadsCount = threadsCount;
            _threads = new Thread[threadsCount];
            for (var i = 0; i < threadsCount; i++)
            {
                _threads[i] = new Thread(Consume) { IsBackground = false };
                _threads[i].Start();
            }
        }

        public void Dispose()
        {
            for (var i = 0; i < _threadsCount; i++)
            {
                AddTask(new KeyValuePair<long, byte[]>(-1, new byte[0]));
            }
            for (var i = 0; i < _threadsCount; i++)
            {
                _threads[i].Join();
            }
            Program.Printer.Dispose();
        }

        public void AddTask(KeyValuePair<long, byte[]> task)
        {
            lock (_locker)
            {
                _tasks.Enqueue(task);
                Monitor.Pulse(_locker);
            }
        }

        public string ByteArrayToString(byte[] array)
        {
            StringBuilder sb = new StringBuilder();
            int i;
            for (i = 0; i < array.Length; i++)
            {
                sb.Append($"{array[i]:X2}");
                if ((i % 4) == 3) sb.Append(" ");
            }
            return sb.ToString();
        }

        private void Consume()
        {
            while (true)
            {
                KeyValuePair<long, byte[]> task = new KeyValuePair<long, byte[]>();
                lock (_locker)
                {
                    while (!_tasks.Any())
                    {

                        Monitor.Wait(_locker);
                    }
                    task = _tasks.Dequeue();
                   }
                if (task.Key == -1)
                {
                    return;
                }
                Program.Printer.AddHashPart(new KeyValuePair<long, byte[]>(task.Key,ShaCreator.ComputeHash(task.Value)));
            }
        }
    }
}

FileReader.cs

using System;
using System.Collections.Generic;
using System.IO;

namespace Signature
{
    internal class FileReader
    {
        private readonly FileStream _inputFileStream;
        private readonly int _chunkSize;

        public FileReader(string filename, int chunkSize)
        {
            try
            {
                _inputFileStream = new FileStream(filename, FileMode.Open, FileAccess.Read);
            }
            catch (FileNotFoundException ioException)
            {
                throw new FileNotFoundException("File not found",ioException);
            }
            catch (DirectoryNotFoundException ioException)
            {
                throw new DirectoryNotFoundException("Directory not found",ioException);
            }
            catch (IOException ioException)
            {
                throw new IOException("IO error",ioException);
            }
            _chunkSize = chunkSize;
        }

        ~FileReader()
        {
            if (_inputFileStream != null)
            {
                _inputFileStream.Close();
            }
        }

        public void Read()
        {
            var buffer = new byte[_chunkSize];
            var bytesRead = 0;
            _inputFileStream.Seek(0, SeekOrigin.Begin);
            bytesRead = _inputFileStream.Read(buffer, 0, _chunkSize);
            long partNum = 1;
            Program.Tasks.AddTask(new KeyValuePair<long, byte[]>(partNum, buffer));
            while (bytesRead > 0)
            {
                partNum++;
                buffer = new byte[_chunkSize];
                bytesRead = _inputFileStream.Read(buffer, 0, _chunkSize);
                if (bytesRead != 0)
                {
                    if (bytesRead < _chunkSize)
                    {
                        Array.Resize(ref buffer,bytesRead);
                    }
                    Program.Tasks.AddTask(new KeyValuePair<long, byte[]>(partNum, buffer));
                }
            }
            Program.Tasks.Dispose();
        }
    }
}

HashPrinter.cs

using System;
using System.Collections.Generic;
using System.Diagnostics.Eventing.Reader;
using System.Text;
using System.Threading;

namespace Signature
{

    internal class HashPrinter : IDisposable
    {
        private readonly object _locker = new object();
        private readonly Thread _threadPrint;
        private readonly List<KeyValuePair<long, byte[]>> _hashList = new List<KeyValuePair<long, byte[]>>();
        private readonly int _maxListSize;
        private long _currentPrint = 1;

        public HashPrinter(int maxListSize)
        {
            if (maxListSize <= 0)
                throw new IndexOutOfRangeException("Max size should be positive");
            _maxListSize = maxListSize;
            _threadPrint = new Thread(Consume) { IsBackground = false };
            _threadPrint.Start();
        }

        public void Dispose()
        {
            AddHashPart(new KeyValuePair<long, byte[]>(-1, new byte[0]));
            _threadPrint.Join();

        }

        public void AddHashPart(KeyValuePair<long, byte[]> hashPart)
        {
            lock (_locker)
            {
                while (_hashList.Count >= _maxListSize)
                {
                    Monitor.Wait(_locker,10);
                }
                if (hashPart.Key == -1)
                {
                    Sort();
                }
                _hashList.Add(hashPart);
                Monitor.Pulse(_locker);
            }
        }

        private void Sort()
        {
            _hashList.Sort(Compare);
        }

        private int Compare(KeyValuePair<long, byte[]> a, KeyValuePair<long, byte[]> b)
        {
            return a.Key.CompareTo(b.Key);
        }

        public string ByteArrayToString(byte[] array)
        {
            StringBuilder sb = new StringBuilder();
            int i;
            for (i = 0; i < array.Length; i++)
            {
                sb.Append($"{array[i]:X2}");
                if ((i % 4) == 3) sb.Append(" ");
            }
            return sb.ToString();
        }

        private void Consume()
        {
            while (true)
            {
                KeyValuePair<long, byte[]> part;
                lock (_locker)
                {
                    while (_hashList.Count == 0)
                    {
                        Monitor.Wait(_locker);
                    }
                    part = _hashList[0];
                    if (part.Key == -1)
                    {
                        return;
                    }
                    if (part.Key == _currentPrint)
                    {
                        _currentPrint++;
                        Console.WriteLine("Part #{0}, Hash: {1}", part.Key, ByteArrayToString(part.Value));
                        //lock (_locker)
                        //{
                            _hashList.Remove(part);
                        //}
                    }
                    else
                    {
                        //lock (_locker)
                        Sort();// }
                    }
                }

            }
        }
    }
}

SHACreator.cs

using System.Security.Cryptography;

namespace Signature
{
    internal class ShaCreator
    {
        public static byte[] ComputeHash(byte[] dataBytes)
        {
            SHA256 mSha256 = SHA256Managed.Create();
            var hashValue = mSha256.ComputeHash(dataBytes);
            return hashValue;
        }
    }
}

Answer 1

Program

This is bad

public static TaskQueue Tasks;
public static HashPrinter Printer;  

I guess it is used in this way (a public static field) so it can be accessed from each other. A better way would be to inject (pass it e.g as a parameter to the constructor) into the classes which are using it.

If the excption's message and stacktrace is only printed/exported, then there is no need to distinguish between the types of the exception. This is just boilerplate code which isn't needed. The catch part should be reduced to

catch (Exception err)
{
    PrintError(err.Message,err.StackTrace);
}

TaskQueue ctor

if (threadsCount <= 0)
    throw new IndexOutOfRangeException("Processor count should be positive");  

From the documentation

The exception that is thrown when an attempt is made to access an element of an array or collection with an index that is outside its bounds.

It seems this won't fit here, will it ? The correct exception which should be thrown is either an ArgumentException or better an ArgumentOutOfRangeException. Both of this exceptions signal to the caller that a passed argument has a problem in some way, which is what you want to tell.

Btw, omitting braces {} although they might be optional can lead to serious bugs so I want to encourage you to always use them.

All of the class level variables which won't change are readonly which is good, but one lonely Queue<KeyValuePair<long, byte[]>> _tasks isn't. Is there a reason for this ?

Why is the ByteArrayToString() method public? In addition

    int i;
    for (i = 0; i < array.Length; i++)  

the int i could be in the loop declaration like so

for (int i = 0; i < array.Length; i++)   

its more idiomatic.

While we are at this method, I noticed that the same method is also in the HashPrinter class and therfor should be extracted to a separate class which can be static and used by both classes. You could also turn it into a extension method.

Instead of explicitly using the concrete type one should use var if the right hand side of an assignment makes the type clear.

E.g

private void Consume()
{
    while (true)
    {
        KeyValuePair<long, byte[]> task = new KeyValuePair<long, byte[]>();  

should be

private void Consume()
{
    while (true)
    {
        var task = new KeyValuePair<long, byte[]>();  

There is more to say, like Dispose() pattern, CancellationTokenSource for cancellation of a thread, consistentness of style, like sometimes C#6 features are used and sometimes not (Console.WriteLine("Part #{0}, Hash: {1}", part.Key, ByteArrayToString(part.Value));).

Answer 2

I want to add a few things. First, you shouldn't access managed objects in finalizers, because there is no guarantee that those objects or their dependencies are not garbage collected already. null checks are good and all, and nothing bad will probably happen from calling stream's Dispose, but why would you? Implement IDisposable on FileReader and call _inputFileStream.Close(); there. In case Dispose is not called, FileStream has its own finalizer to take care of it.

---

TaskQueue.Dispose impelmentation relies too much on queue being empty when it is called. In general case, depending on the queue size and complexity of the work current implementation might take minutes before Dispose returns which I think is unacceptable in most use cases. Instead you might want to terminate threads early by setting some bool flag and pulsing all threads. Or you can use WaitHandle. However if it is your intention to prevent any data loss, then I guess your implementation is alright. Just make sure to give this special data package a name:

 private static readonly KeyValuePair<long, byte[]> ExitMessage 
     = new KeyValuePair<long, byte[]>(-1, new byte[0]);

---

Instead of using generic KeyValuePair you can create your own small class to hold the data and give its properties meaningful names. Having Id instead of Key and Data instead of Value will improve readability. You can also add a property to represent the type of operation, instead of using -1 as some sort of termination code.

---

Generally you should not dispose objects which came from the outer scope. Disposable static instances are especially bad. It is not TaskQueue's job to manage Printer's lifetime, it did not create it, and it does not know if it is still in use. Printer.Dispose() call should be moved to outer scope, to the place where Printer was created. In your case, it is Program.Main function. You should also avoid static dependencies, they make your code harder to understand and almost impossible to re-use. If FileReader requires TaskQueue to function properly - make this requirement explicit by adding TaskQueue parameter to FileReader's constructor.

---

This

while (!_tasks.Any())
{

    Monitor.Wait(_locker);
}

looks fishy. Why is there while and not if? If your queue works correctly, if should do the job.

---

HashPrinter.Consume is inefficient. If _hashList.Count > 0 but part.Key != _currentPrint it will start to constantly re-sort _hashList while it is waiting for the next chunk, eating up single CPU core. Instead your working thread should be in waiting state until the next hashPart arrives.