TimeoutManager: check whether a task has been finished within a given time

Question

I want to make a helper class, which raises an event whenever something is not finished in a certain time.

Example: my program sends out multiple network messages concurrently. After a random delay, each message should receive a single reply. My helper class should raise an event whenever no reply is received for a specific message.

The TimeoutManager is used like this:

//When sending out a message, tell the manager to start a timer
timeoutManager.Add("Message: ID=001")

//When receiving a reply, tell the manager to stop waiting for a timeout 
bool replyWasExpected = timeoutManager.MarkOldestItemAsFinished(x=>x.Contains("ID=001");

//When no reply is received in time, this event will be called.
timeoutManager.OnTimeout += (evt, args) =>{Console.WriteLine("TIMEOUT! NO REPLY!");};

I am looking at suggestions to solve this more elegantly, especially in regards to the data structures used.

The solution I implemented has some ugly hacks, detailed below. I am happy with the interface, but I dislike the code behind it. I am currently on net 4.0, so async/await is not available.

The whole code is below. The main points are:

TimeoutManager<T_Item> is a generic timeout-manager, and should deal with an arbitrary "item" or "work unit". I used strings or ints in the examples, but in practise, I would use some sort of Message object for each outgoing message.

It uses the BlockingCollection itemsWaitingForTimeout to remember each item,
and a single dedicated thread timeoutLoop() to wait for the timeout for all items.

Hack 1: I add each new item twice to the BlockingCollection. This way I can still mark an item as finished, after the dedicated thread has already removed it from the collection, but before the timeout occurs.

Hack 2: The dedicated thread uses Thread.Sleep() to wait for the timeout.

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

namespace McatXml
{

    public class TimeoutManager<T_Item>
    {
        public class TimeoutEventArgs : EventArgs
        {
            public T_Item Item { get; private set; }
            public TimeoutEventArgs(T_Item item) { this.Item = item; }
        }
        /// <summary> called whenever an item is not finished before the timeout </summary>
        public event EventHandler<TimeoutEventArgs> OnTimeout;


        private readonly TimeSpan timeout = TimeSpan.FromSeconds(30);
        private BlockingCollection<ItemWithTimeout> itemsWaitingForTimeout = new BlockingCollection<ItemWithTimeout>();

        /// <summary> private wrapper, to decorate an item with a timeout </summary>
        private class ItemWithTimeout
        {
            internal readonly T_Item Item;
            internal readonly Stopwatch Watch;
            internal volatile bool FinishedWaiting;
            internal ItemWithTimeout(T_Item item)
            {
                this.Item = item;
                this.Watch = Stopwatch.StartNew();
                this.FinishedWaiting = false;
            }
        }

        public TimeoutManager(TimeSpan timeout)
        {
            this.timeout = timeout;
            OnTimeout += (sender, args) => { };
            Task loop = new Task(this.timeoutLoop, TaskCreationOptions.LongRunning);
            loop.Start();
        }

        public void Add(T_Item item)
        {
            var itemExt = new ItemWithTimeout(item);
            itemsWaitingForTimeout.Add(itemExt);
            itemsWaitingForTimeout.Add(itemExt); // this is an ugly hack!
        }

        /// <summary> mark all items as finished, that fit the given condition  </summary>
        public bool MarkAllAsFinished(Func<T_Item, bool> isMatch = null)
        {
            return markAsFinished(stopAfterFirstHit: false, reverseOrder: false, isMatch: isMatch);
        }
        /// <summary> mark the most recent item as finished, that fits the given condition  </summary>
        public bool MarkNewestAsFinished(Func<T_Item, bool> isMatch = null)
        {
            return markAsFinished(stopAfterFirstHit: true, reverseOrder: true, isMatch: isMatch);
        }
        /// <summary> mark the oldest item as finished, that fits the given condition  </summary>
        public bool MarkOldestAsFinished(Func<T_Item, bool> isMatch = null)
        {
            return markAsFinished(stopAfterFirstHit: true, reverseOrder: false, isMatch: isMatch);
        }

        /// <summary> mark items as finished, that fit the given condition  </summary>
        private bool markAsFinished(bool stopAfterFirstHit, bool reverseOrder, Func<T_Item, bool> isMatch = null)
        {
            // get a snapshot of all currently waiting items
            var items = this.itemsWaitingForTimeout.ToArray();

            bool success = false;

            // start with the oldest or newest item?
            int startIdx = reverseOrder ? items.Length - 1 : 0;
            int inc = reverseOrder ? -1 : 1;

            for (int i = startIdx; i < items.Length && i >= 0; i += inc)
            {
                var item = items[i];
                if (item.FinishedWaiting) continue; // the item is already marked as finished
                if (isMatch == null || isMatch(item.Item))
                {
                    lock (item)
                    {
                        if (item.FinishedWaiting) continue; // another thread marked this as finished while we were waiting for the lock
                        item.FinishedWaiting = true;
                    }
                    success = true; // we found an item and marked it as finished
                    if (stopAfterFirstHit) break; // should we look for further items?
                }
            }
            return success; // did we find an item and mark it as finished?
        }

        /// <summary> for all items that are not finished, check whether their time is up  </summary>
        private void timeoutLoop()
        {
            foreach (var item in itemsWaitingForTimeout.GetConsumingEnumerable())
            {
                if (item.FinishedWaiting) continue; // item has already been finished

                while (!item.FinishedWaiting && item.Watch.Elapsed < this.timeout)
                {
                    // wait until the timeout has passed or the item is finished
                    Thread.Sleep(
                        TimeSpan.FromMilliseconds(Math.Max(1,
                                                           this.timeout.TotalMilliseconds -
                                                           item.Watch.ElapsedMilliseconds)));
                }

                if (item.FinishedWaiting) continue; // item has been finished while we were waiting
                lock (item)
                {
                    if (item.FinishedWaiting) continue; // item has been finished while we ackquired the lock

                    item.FinishedWaiting = true;
                }
                // item has not been finished in time!
                OnTimeout(this, new TimeoutEventArgs(item.Item));
            }
        }
    }
    static class Program
    {
        static void Main()
        {

            TimeoutManager<int> test = new TimeoutManager<int>(TimeSpan.FromSeconds(5));
            test.OnTimeout += (sender, args) => { Console.WriteLine("Timeout: {0}", args.Item); };

            // start 100 new timers, in random order
            Parallel.ForEach(Enumerable.Range(0, 100).OrderBy(i => Guid.NewGuid()), (i) =>
                {
                    test.Add(i);
                });

            // mark 90 tasks as finished 
            Parallel.ForEach(Enumerable.Range(0 , 90).OrderBy(i => Guid.NewGuid()), (i) =>
                {
                    if (!test.MarkAllAsFinished(x => x == i))
                        Console.WriteLine("could not mark as finished: {0}", i);
                });

            // after 5 seconds, 10 timeout events should be executed.
            Console.ReadLine();



        }
    }
}

Is there a simpler way to store my items and mark them as finished besides the BlockingCollection with double entries?

Did I overcomplicate this too much? Will my coworkers curse me If I use this in production?

I like the fact that the OnTimeout events will not be raised all at once, if 1000 timeouts happen at the same time. Nevertheless I am unsure whether the dedicated thread is justified.

Answer 1

Yes, I think you've overcomplicated this a bit. The BlockingCollection is not a great fit here, a simple list will do fine. Something like this perhaps:

using System;
using System.Collections.Generic;
using System.Linq;
using System.Runtime.Caching;
using System.Threading;
using System.Threading.Tasks;

public class SimpleTimeoutManager<T> : IDisposable
{
    private readonly List<ListEntry> _queue = new List<ListEntry>();
    private readonly TimeSpan _timeout;
    private readonly TimeSpan _timerInterval;
    private readonly CancellationTokenSource _cancellationTokenSource;
    private readonly CancellationToken _cancellationToken;

    public SimpleTimeoutManager(TimeSpan? timeout = null, TimeSpan? timerInterval = null)
    {
        _timeout = timeout ?? TimeSpan.FromSeconds(1);
        _timerInterval = timerInterval ?? TimeSpan.FromMilliseconds(100);
        _cancellationTokenSource = new CancellationTokenSource();
        _cancellationToken = _cancellationTokenSource.Token;
        new Task(Timer, TaskCreationOptions.LongRunning).Start();
    }

    public event EventHandler<TimeoutEventArgs> OnTimeout = delegate { };

    public void Dispose()
    {
        _cancellationTokenSource.Cancel();
    }

    public void Add(T item)
    {
        lock (_queue)
        {
            _queue.Add(new ListEntry(item, _timeout));
        }
    }

    public bool MarkAllAsFinished(Func<T, bool> isMatch = null)
    {
        isMatch = isMatch ?? (item => true);
        lock (_queue)
        {
            return _queue.RemoveAll(entry => isMatch(entry.Item)) > 0;
        }
    }

    public bool MarkNewestAsFinished(Func<T, bool> isMatch = null)
    {
        return MarkFirstMatchAsFinished(_queue.FindLastIndex, isMatch);
    }

    public bool MarkOldestAsFinished(Func<T, bool> isMatch = null)
    {
        return MarkFirstMatchAsFinished(_queue.FindIndex, isMatch);
    }

    private bool MarkFirstMatchAsFinished(Func<Predicate<ListEntry>, int> indexOfEntry, Func<T, bool> isMatch = null)
    {
        isMatch = isMatch ?? (item => true);
        lock (_queue)
        {
            var index = indexOfEntry(entry => isMatch(entry.Item));
            bool found = index >= 0;
            if (found)
            {
                _queue.RemoveAt(index);
            }

            return found;
        }
    }

    private void Timer()
    {
        while (!_cancellationToken.IsCancellationRequested)
        {
            DateTime wakeupAt = _queue.FirstOrDefault()?.Expires ?? DateTime.Now.Add(_timerInterval);
            var delay = wakeupAt - DateTime.Now;
            Thread.Sleep(delay);

            lock (_queue)
            {
                var expiredEntry = _queue.TakeWhile(entry => entry.HasExpired()).FirstOrDefault();
                if (expiredEntry != null)
                {
                    _queue.RemoveAt(0);
                    OnTimeout.Invoke(this, new TimeoutEventArgs(expiredEntry.Item));
                }
            }
        }
    }

    public class TimeoutEventArgs : EventArgs
    {
        public readonly T Item;

        public TimeoutEventArgs(T item)
        {
            Item = item;
        }
    }

    private class ListEntry
    {
        public readonly T Item;
        public readonly DateTime Expires;

        public ListEntry(T item, TimeSpan timeout)
        {
            Item = item;
            Expires = DateTime.Now.Add(timeout);
        }

        public bool HasExpired()
        {
            return Expires < DateTime.Now;
        }
    }
}