Thread-safe queue mechanism

Question

Is it my queue mechanism thread safe? I just wonder if I need concurrent collections. Need I lock Enqueue method? Console displays queue count in incorrect order, Does it affect on maxQueueCount at Load method? Can I improve it in some way?

using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Threading.Tasks;
using System.Timers;

class Program
{
    public class Document : IItem
    {
        public Guid Id { get; set; }
    }

    static void Main()
    {
        var queueProvider = new Provider();
        var docs = new List<IItem>
        {
            new Document { Id = Guid.NewGuid() },
            new Document { Id = Guid.NewGuid() },
            new Document { Id = Guid.NewGuid() },
            new Document { Id = Guid.NewGuid() },
            new Document { Id = Guid.NewGuid() }
        };

        try
        {
            var tasks = new List<Task>();

            var task1 = Task.Factory.StartNew(() =>
            {
                var timer1 = new Timer(1000) { Interval = 1000 };
                timer1.Elapsed += (object sender, ElapsedEventArgs e) =>
                {
                    queueProvider.Load(docs, 1);
                };
                timer1.Enabled = true;
                timer1.Start();
            });
            tasks.Add(task1);

            var task2 = Task.Factory.StartNew(() =>
            {
                var timer1 = new Timer(1000) { Interval = 1000 };
                timer1.Elapsed += (object sender, ElapsedEventArgs e) =>
                {
                    queueProvider.Load(docs, 2);
                };
                timer1.Enabled = true;
                timer1.Start();
            });
            tasks.Add(task2);

            //Dequeue

            //var task3 = Task.Factory.StartNew(() =>
            //{
            //    var timer1 = new Timer(3000) { Interval = 1000 };
            //    timer1.Elapsed += (object sender, ElapsedEventArgs e) =>
            //    {
            //        queueProvider.Dequeue();
            //    };
            //    timer1.Enabled = true;
            //    timer1.Start();
            //});
            //tasks.Add(task3);

            Task.WaitAll(tasks.ToArray());
        }
        catch (Exception e)
        {
            Console.WriteLine(e);
        }

        Console.ReadKey();
    }
}


public interface IItem
{
    Guid Id { get; set; }
}

public interface IProvider
{
    void Enqueue(IItem feedingItem, int id);
}

public class Provider : IProvider
{
    private readonly ConcurrentQueue<IItem> queue;
    private readonly ConcurrentDictionary<Guid, DateTime> inputBuffor;

    private readonly object locker = new object();

    private int maxQueueCount = 3;

    public Provider()
    {
        queue = new ConcurrentQueue<IItem>();
        inputBuffor = new ConcurrentDictionary<Guid, DateTime>();
    }

    public IItem Dequeue()
    {
        queue.TryDequeue(out var item);

        Console.WriteLine("Dequeue: " + item.Id);

        return item;
    }

    public void Enqueue(IItem item, int id)
    {
        //lock (locker)
        //{
        if (inputBuffor.TryAdd(item.Id, DateTime.Now))
        {
            queue.Enqueue(item);

            Console.WriteLine("Enqueue: " + item.Id + "taskId: " + id);
            Console.WriteLine("Count: " + queue.Count + " Buffor: " + inputBuffor.Count);
        }
        else
        {
            Console.WriteLine("Not Enqueue: " + item.Id + "taskId: " + id);
        }
        //}
    }

    public void Load(IEnumerable<IItem> data, int id)
    {
        foreach (var item in data)
        {
            if (queue.Count < maxQueueCount)
                Enqueue(item, id);
        }
    }
}

Answer 1

So you want a queue with a maximum size, and you don't want the same item to be enqueued again.

There are a few problems with the code:

• Load is not thread-safe: it's not preventing other threads from enqueuing items after the queue-count check, so it's possible to add more items than allowed.
• Enqueue does not check the queue count at all, so it's even trivial to add more items than allowed!
• Load silently ignored items without letting the caller know. Even if it returned how many items were actually added, a caller still wouldn't know which ones. I don't think you need this method at all: a caller can easily loop over some items and call Enqueue directly.
• Enqueue doesn't tell whether it succeeded either. If a method can fail then it should communicate that to its callers (for example by naming it TryEnqueue instead, and by letting it return a boolean). It's also a good idea to document its behavior.
• Dequeue does not check the return value of queue.TryDequeue. That will cause a NullReferenceException on the Console.WriteLine line if the queue was empty. Again, documentation is a good idea: does it return null if the queue is empty, or should the caller expect an exception?
• Using two concurrent collections internally seems like overkill. They won't help you enforce the maximum count restriction either, so if that's important to you then you'll need to use some kind of locking. And if you do that, you might as well use non-concurrent collections internally, since the locks ensure that they're only accessed by one thread at a time.