Multithreaded application in C#

Question

I have created a .net application for optimizing the pdf files.Actually I have to optimize many files and i have called a thread like this:

CheckForIllegalCrossThreadCalls = false;
thOptimize = new Thread(csCommon.pdfFilesCompressAndMove);
thOptimize.Start();

Also I have found the no. of processors and cores using this:

int processors=Environment.ProcessorCount
int coreCount = 0;
foreach (var item in new System.Management.ManagementObjectSearcher("Select * from  Win32_Processor").Get())
{
coreCount += int.Parse(item["NumberOfCores"].ToString());
}

I have found 4 processors and 2 cores in my machine.

Now my problem is that i want to use the functionpdfFilesCompressAndMove for all the processors i.e. i want to optimize multiple files at the same time.In other words,I want to keep busy all the processors in optimization.

Please guide me how is it possible?

Answer 1

What you want is a producer/consumer queue.

What happens here is that the producer creates work-items for the consumer to process. This works well when the producer can create the work for the consumer much faster than the consumer can process it. You then have one or more consumers processing this queue of work.

Here's a producer consumer class I use for this kind of thing:

public class ProducerConsumer<T>:IDisposable 
    {
        private  int _consumerThreads;
        private readonly Queue<T> _queue = new Queue<T>();
        private readonly object _queueLocker = new object();
        private readonly AutoResetEvent _queueWaitHandle = new AutoResetEvent(false);
        private readonly Action<T> _consumerAction;
        private readonly log4net.ILog _log4NetLogger = log4net.LogManager.GetLogger(System.Reflection.MethodBase.GetCurrentMethod().DeclaringType);
        private bool _isProcessing = true;

        public ProducerConsumer(Action<T> consumerAction,int consumerThreads,bool isStarted)
        {
            _consumerThreads = consumerThreads;

            if (consumerAction == null)
            {
                throw new ArgumentNullException("consumerAction");
            }
            _consumerAction = consumerAction;
            if (isStarted)
                Start();
            //just in case the config item is missing or is set to 0.  We don't want to have the queue build up
        }

        public ProducerConsumer(Action<T> consumerAction, int consumerThreads):this(consumerAction,consumerThreads,true)
        {


        }
        public void Dispose()
        {
            _isProcessing = false;
            lock(_queueLocker)
            {
                _queue.Clear();
            }
        }
        public void Start()
        {
            if (_consumerThreads == 0)
                _consumerThreads = 2;

            for (var loop = 0; loop < _consumerThreads; loop++)
                ThreadPool.QueueUserWorkItem(ConsumeItems);
        }

        public void Enqueue(T item)
        {
            lock (_queueLocker)
            {
                _queue.Enqueue(item);
                // After enqueuing the item, signal the consumer thread.            
                _queueWaitHandle.Set();
            }
        }

        private void ConsumeItems(object state)
        {
            while (_isProcessing)
            {
                try
                {
                    var nextItem = default(T);
                    bool doesItemExist;
                    lock (_queueLocker)
                    {
                        int queueCount = _queue.Count;
                        doesItemExist = queueCount > 0;
                        if (doesItemExist)
                        {
                            nextItem = _queue.Dequeue();
                        }
                        if (queueCount > 0 && queueCount % 50 == 0)
                            _log4NetLogger.Warn(String.Format("Queue is/has been growing.  Queue size now:{0}",
                                                              queueCount));
                    }
                    if (doesItemExist)
                    {
                        _consumerAction(nextItem);
                    }
                    else
                    {
                        _queueWaitHandle.WaitOne();
                    }
                }
                catch (Exception ex)
                {

                    _log4NetLogger.Error(ex);
                }

            }
        }
    }

It's a generic class, so T is the type of object you're giving to to process. You also provide it with an Action, which is the method that does the actual processing. This should allow you to process multiple PDF files at once in a clean way.