Recurrent action on concurrent collection

Question

I have a bunch of concurrent threads carrying out operations which return alerts. The alerts have to be persisted to a DB, but this cannot be done concurrently as it may cause duplicate alerts being created. To avoid duplicate alerts I have written a class containing a ConcurrentBag<T> with a Timer that triggers Action<T> (in this case, sending my pending alerts to the DB) at a set interval.

I'm wondering if what I wrote it is written in a way that will not cause problems in the execution in the future. If alerts get lost due to an unexpected exception that would be problematic. I'd appreciate comments/criticisms/suggestions for improvement.

Here's the code:

class PeriodicConcurrentBag<T>
{
    public int Count
    {
        get { return this.items.Count(); }
    }

    public bool HasItems
    {
        get { return this.items.Any(); }
    }

    public bool IsStopped { get; private set; }

    private int timerPeriod;
    private Timer timer;
    private ConcurrentBag<T> items;
    private Action<IEnumerable<T>> actionOnItems;

    public PeriodicConcurrentBag(Action<IEnumerable<T>> actionOnItems, int periodInSeconds)
    {
        this.timerPeriod = periodInSeconds * 1000;
        this.actionOnItems = actionOnItems;

        this.items = new ConcurrentBag<T>();
    }

    public void Start()
    {
        this.IsStopped = false;
        this.timer = new Timer(this.InternalPeriodicAction, null, this.timerPeriod, Timeout.Infinite);
    }

    public void Stop()
    {
        this.IsStopped = true;
    }

    public void AddItem(T item)
    {
        this.items.Add(item);
    }

    public void AddItems(IEnumerable<T> items)
    {
        this.items.AddRange(items);
    }

    private void InternalPeriodicAction(object state)
    {
        if (!this.IsStopped)
        {
            var currentItems = Interlocked.Exchange(ref this.items, new ConcurrentBag<T>());
            this.actionOnItems(currentItems);

            if (!this.IsStopped)
                this.Start();
        }
    }
}

EDIT: The AddRange() method is an extension method I created for ConcurrentBag.

Answer 1

Your Stop method does not actually stop the timer. This is

1. really counter-intuitive
2. wastes CPU

---

There is also a racing condition. If someone calls Stop in-between those two lines:

           if (!this.IsStopped)
            this.Start();

the Stop call will be ignored. The moral of the story: just take the lock.

---

Timer implements IDisposable. So you should call Dispose when you no longer need it (in Stop method, for example.)

---

I would refactor actionOnItems into a public event.

---

int for time period is ambiguous and limiting. I can not use your class, if I want to send updates every 0.5 seconds, and I can easily make an error by assuming, that int represent milliseconds. You can eliminate those problems by replacing it with TimeSpan.

---

Edit: Also your AddRange method rises some red flags. I assume it is not an atomic operation, and you add items to the bag one by one. What will happen, if AddRange is called right before Interlocked.Exchange(ref this.items, new ConcurrentBag<T>()); ? Do you think the new items will go to the new bag or to the old one? How is your extension method synchronized with the rest of your code? You are walking on the really thin ice here. If you have a limited experience with multi-threading, I suggest you start simple. Remove ConcurrentBag, use regular list, but lock every single access point (both public AND private, methods AND properties). Once this is working, test if it meats your performance requirements. If it does, leave it at that. There is no point in fighting over nanoseconds if

1. You do not actually need those nanoseconds

and

2. You get an extremely bug-prone code as a result

Answer 2

In addition to NikitaB's answer just few small optimizations:

In Count property you're using LINQ Count() extension method. Count() try to cast to a known type and if possible return its Count or Length property but this is a concurrent bag and its thread-safe IEnumerable<T> implementation takes a snapshot which is then walked by Enumerable.Count() implementation to count how many items it has. It's expansive, change it to:

public int Count
{
    get { return this.items.Count; }
}

The same is true for HasItems property, in general I'm not happy with such small helpers because they add complexity to public interface adding very little value but if you want it change it to:

public bool HasItems
{
    get { return this . count != 0; }
}

Note that with C#6 syntax it may be simplified to:

public int Count => this.items.Count;
public bool HasItems => this.Items.Count != 0;

Fields timerPeriod and actionOnItems are initialized in constructor, mark them readonly:

private readonly int timerPeriod;
private readonly Action<IEnumerable<T>> actionOnItems;

And few design issues:

Standard collections have Add() and AddRange() methods, why do you want to use AddItem() and AddItems()? Stick to existing well-known naming convention and rename those methods.

You may want also to implement some basic interface to make integration easier, at least ICollection<T>.

periodInSeconds is handy but intervals are usually expressed in milliseconds, it may be less error-prone to stick with this unwritten convention.

actionOnItems is poor named, IMO. Action on items but when? When added? Periodically? I'd make it explicit.