2
\$\begingroup\$

The code realizes producer/consumer problem with multiple producers and consumers. Have this code any potential deadlock or races?

//RandomDataProvider.cs

namespace MyNamespace.Core
{
    using System;

    /// <summary>
    /// Provides randomly generated data.
    /// </summary>
    public class RandomDataProvider
    {
        #region constuction

        /// <summary>
        /// Initializes new instance of the <see cref="RandomDataProvider"/> class.
        /// </summary>
        /// <param name="maxSleepInterval"></param>
        public RandomDataProvider(int maxSleepInterval)
        {
            _maxSleepInterval = maxSleepInterval;
        }

        #endregion // constuction

        #region implementation

        /// <summary>
        /// Gets the random sleeping interval.
        /// </summary>
        public int GetSleepInterval()
        {
            lock (_locker)
            {
                return _random.Next(_maxSleepInterval);
            }
        }

        /// <summary>
        /// Gets the random value.
        /// </summary>
        public int GetRandomValue()
        {
            lock (_locker)
            {
                return _random.Next(1, 100);
            }
        }

        #endregion // implementation

        #region representation

        private readonly Random _random = new Random();
        private readonly object _locker = new object();
        private readonly int _maxSleepInterval;

        #endregion // representation
    }
}

//Producer.cs

namespace MyNamespace.Core
{
    using System.Collections.Generic;
    using System.Threading;

    /// <summary>
    /// Reprecents <see cref="Producer"/> class.
    /// </summary>
    public sealed class Producer : RandomDataProvider
    {

        #region construction

        /// <summary>
        /// Initializes the new instance of <see cref="Producer"/> class.
        /// </summary>
        /// <param name="queue">Queue to which producers put items.</param>
        /// <param name="maxSleepInterval">The maximum sleep interval.</param>
        public Producer(Queue<int> queue, int maxSleepInterval)
            : base(maxSleepInterval)
        {
            _queue = queue;
        }

        #endregion // construction

        #region implementation

        /// <summary>
        /// Starts produce items and put to queue.
        /// </summary>
        /// <param name="token">Cancellation token.</param>
        public void StartProduce(CancellationToken token)
        {
            while (true)
            {
                // If canceled stop producing.
                if (token.IsCancellationRequested)
                {
                    return;
                }

                Thread.Sleep(GetSleepInterval());
                int item = GetRandomValue();

                lock (_queue)
                {
                    if (_isReachedMax)
                    {
                        // if producer reached max items, block and wait until count decreases 80.
                        while (_queue.Count > 80)
                        {
                            Monitor.Wait(_queue);
                        }

                        // now we can release thread
                        _isReachedMax = false;
                    }

                    _queue.Enqueue(item);

                    if (_queue.Count == 1)
                    {
                        // Pulse comsumres which waits on empty queue
                        Monitor.Pulse(_queue);
                    }
                    else if (_queue.Count >= 100)
                    {
                        // Set reached max count flag.
                        _isReachedMax = true;
                    }
                }
            }
        }

        #endregion // implementation


        #region representation

        private readonly Queue<int> _queue;
        private bool _isReachedMax;

        #endregion // representation

    }
}

//Consumer.cs

namespace MyNamespace.Core
{
    using System;
    using System.Collections.Generic;
    using System.IO;
    using System.Threading;

    /// <summary>
    /// Reprecents <see cref="Consumer"/> class.
    /// </summary>
    public sealed class Consumer : RandomDataProvider, IDisposable
    {

        /// <summary>
        /// Initializes the new instance of <see cref="Consumer"/> class.
        /// </summary>
        /// <param name="queue">The queue from which consumes items.</param>
        /// <param name="maxSleepInterval">The maximum sleep interval.</param>
        public Consumer(Queue<int> queue, int maxSleepInterval)
            :base(maxSleepInterval)
        {
            _queue = queue;
        }


        #region implementation

        #region Implementation of IDisposable

        /// <summary>
        /// Disposes the current instance.
        /// </summary>
        public void Dispose()
        {
            StreamWriter.Close();
        }

        #endregion

        /// <summary>
        /// Starts consume items from queue.
        /// </summary>
        /// <param name="token">Cancellation token.</param>
        public void StartConsume(CancellationToken token)
        {
            while (true)
            {
                Thread.Sleep(GetSleepInterval());

                int item;

                lock (_queue)
                {
                    // If no item in queue after cancel, return.
                    if (_queue.Count == 0 && token.IsCancellationRequested)
                    {
                        return;
                    }

                    // Wait when queue empty.
                    while (_queue.Count == 0)
                    {
                        Monitor.Wait(_queue);
                    }

                    item = _queue.Dequeue();

                    // if count decreases and reaches 80, then inform produsers which wait.
                    if (_queue.Count <= 80)
                    {
                        Monitor.PulseAll(_queue);
                    }
                }

                WriteToStream(item);
            }
        }

        #endregion // implementation

        #region operations

        /// <summary>
        /// Writes items to the stream.
        /// </summary>
        /// <param name="item">Item to write.</param>
        private void WriteToStream(int item)
        {
            lock (StreamWriter)
            {
                StreamWriter.Write("{0},", item);
            }
        }

        #endregion // operations

        #region representation

        private readonly Queue<int> _queue;
        private static readonly StreamWriter StreamWriter = new StreamWriter(FILE_PATH);

        #endregion // representation

        #region constants

        private const string FILE_PATH = "data.txt";

        #endregion // constants
    }
}

//Controller.cs

namespace Root
{
    using System;
    using System.Collections.Generic;
    using System.Threading;
    using System.Timers;

    using MyNamespace.Core;

    /// <summary>
    /// Provides functionality that controls <see cref="Producer"/> and <see cref="Consumer"/> concurency.
    /// </summary>
    public class Controller : IDisposable
    {

        #region construction

        /// <summary>
        /// Initializes the new instance of <see cref="Controller"/> class.
        /// </summary>
        /// <param name="queue">Shared Queue object.</param>
        public Controller(Queue<int> queue)
        {
            _queue = queue;

            _timer = new System.Timers.Timer(1000);
            _timer.Elapsed += ShowCount;
        }

        #endregion // construction


        #region implementation

        #region Implementation of IDisposable

        /// <summary>
        /// Disposes the current instance.
        /// </summary>
        public void Dispose()
        {
            foreach (var consumer in _consumers)
            {
                consumer.Dispose();
            }
        }

        #endregion

        /// <summary>
        /// Creates consumer/producer objects and ran them each in seperate thread. 
        /// </summary>
        public void Start()
        {
            _consumerThreads = new Thread[_consumerCount];
            _producerThreads = new Thread[_producerCount];
            _consumers = new Consumer[_consumerCount];
            _producers = new Producer[_producerCount];

            for (int i = 0; i < ProducerCount; ++i)
            {
                var producer = new Producer(_queue, _maxSleepInterval);
                _producers[i] = producer;
                _producerThreads[i] = new Thread(() => producer.StartProduce(_source.Token));
                _producerThreads[i].Start();
            }

            for (int i = 0; i < ConsumerCount; ++i)
            {
                var consumer = new Consumer(_queue, _maxSleepInterval);
                _consumers[i] = consumer;
                _consumerThreads[i] = new Thread(() => consumer.StartConsume(_source.Token));
                _consumerThreads[i].Start();
            }

            _timer.Start();
        }

        /// <summary>
        /// Cancels producing/consuming.
        /// </summary>
        public void Cancel()
        {
            _source.Cancel();
            _timer.Stop();

            for (int i = 0; i < ConsumerCount; ++i)
            {
                _consumerThreads[i].Join();
            }
        }

        #endregion // implementation

        #region properties

        /// <summary>
        /// Gets or sets consumers count.
        /// </summary>
        public int ConsumerCount
        {
            get { return _consumerCount; }
            set { _consumerCount = value; }
        }

        /// <summary>
        /// Gets or sets producers count.
        /// </summary>
        public int ProducerCount
        {
            get { return _producerCount; }
            set { _producerCount = value; }
        }

        /// <summary>
        /// Gets or sets maximum sleep interval.
        /// </summary>
        public int MaxSleepInterval
        {
            get { return _maxSleepInterval; }
            set { _maxSleepInterval = value; }
        }

        #endregion // properties

        #region operations

        /// <summary>
        /// Shows the queue size.
        /// </summary>
        /// <param name="sender">Sender object.</param>
        /// <param name="e">Event handler argumnets.</param>
        private void ShowCount(object sender, ElapsedEventArgs e)
        {
            int count;
            lock (_queue)
            {
                count = _queue.Count;
            }

            Console.WriteLine("Items in the queue: {0}", count);
        }

        #endregion // operations

        #region representation

        /// <summary>
        /// Shared between produsers and consumers queue object.
        /// </summary>
        private readonly Queue<int> _queue;

        /// <summary>
        /// Thread in which runs each consumer.
        /// </summary>
        private Thread[] _consumerThreads;

        /// <summary>
        /// Thread in which runs each producer.
        /// </summary>
        private Thread[] _producerThreads;

        /// <summary>
        /// Array of <see cref="Consumer"/> objects.
        /// </summary>
        private Consumer[] _consumers;

        /// <summary>
        /// Array of <see cref="Producer"/> objects.
        /// </summary>
        private Producer[] _producers;

        /// <summary>
        /// Timer object.
        /// </summary>
        private readonly System.Timers.Timer _timer;

        /// <summary>
        /// Provides cancellation token.
        /// </summary>
        private readonly CancellationTokenSource _source = new CancellationTokenSource();

        private int _consumerCount = 10;
        private int _producerCount = 10;
        private int _maxSleepInterval = 100;

        #endregion // representation
    }
}

At the end

//Program.cs

namespace Root
{
    using System;
    using System.Collections.Generic;

    class Program
    {
        private static void Main(string[] args)
        {
            int prdCnt;
            int conCnt;

            Console.Write("Enter produsers count[1-10]:");
            while (!int.TryParse(Console.ReadLine(), out prdCnt) || prdCnt < 1 || prdCnt > 10)
            {
                Console.WriteLine("Error");
                Console.Write("Enter produsers count[1-10]:");
            }

            Console.Write("Enter consumers count[1-10]:");
            while (!int.TryParse(Console.ReadLine(), out conCnt) || conCnt < 1 || conCnt > 10)
            {
                Console.WriteLine("Error");
                Console.Write("Enter consumers count[1-10]:");
            }

            Queue<int> queue = new Queue<int>();
            using (var controller = new Controller(queue))
            {
                controller.ConsumerCount = conCnt;
                controller.ProducerCount = prdCnt;
                controller.Start();

                Console.ReadLine();
                controller.Cancel();
            }
        }
    }
}
\$\endgroup\$
  • \$\begingroup\$ What .NET framework do you use? \$\endgroup\$ – almaz Feb 28 '13 at 18:50
  • \$\begingroup\$ @almaz .net 4.0 , 4.5 but using namespace System.Collections.Concurrent not allowed. \$\endgroup\$ – Hamlet Hakobyan Feb 28 '13 at 19:21
  • 3
    \$\begingroup\$ Is any particular reason why specific namespace is not allowed? I can't think of any case when CancellationToken would be allowed but not BlockingCollection... \$\endgroup\$ – almaz Feb 28 '13 at 20:18
2
\$\begingroup\$

The main issue in this code is that several classes rely on each other's proper behaviour in order to work correctly. I'm talking about sharing the instance of Queue<int> and requirement for proper locking on it in all places. See remarks to lock Statement for recommendations on proper usage.

Since you mentioned that you can't use the System.Collections.Concurrent namespace I suggest to create a class that mimics BlockingCollection functionality to avoid first issue. It should incapsulate the thread-safe manipulation with queue, so that both Consumer and Producer don't need to handle multithreading logic.

Also I don't see the reason for deriving Producer/Consumer from RandomDataProvider. They use functionality of RandomDataProvider, but not extend it, so classes should not derive from it and rather store the instance of RandomDataProvider in a field.

\$\endgroup\$

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.