Threadsafe filtering queue

Question

I have implemented a thread safe filtering queue. The queue allows any objects, of the specified type to be added. A thread interested to take an object must specify which object it is interested in via Predicate<T>. A particularity of this implementation is that in my use case, the threads might not be allowed to remove the object from the queue, because others threads may be interested on that object as well.

public class FilterQueue<T>
{
    private readonly LinkedList<T> _values = new LinkedList<T>();
    private readonly object _hasWaiters = new object();
    private int _waiters;

    public void Add(T value)
    {
        lock (_values)
        {
            _values.AddLast(value);
            Monitor.PulseAll(_values);
        }
    }

    private LinkedListNode<T> FindNode(Predicate<T> pred)
    {
        var node = _values.First;
        while (node != null)
        {
            if (pred(node.Value))
            {
                return node;
            }
            node = node.Next;
        }
        return null;
    }

    public void WaitForWaiters()
    {
        lock (_values)
        {
            while (_waiters == 0)
            {
                SyncUtils.Wait(_values, _hasWaiters);
            }
        }
    }

    public void Clear()
    {
        lock (_values)
        {
            if (_waiters != 0)
            {
                throw new InvalidOperationException("There is still someone waiting for requests");
            }
            _values.Clear();
        }
    }

    public T Take(Predicate<T> hasMessage)
    {
        return Take(hasMessage, Timeout.InfiniteTimeSpan);
    }

    public T Take(Predicate<T> hasMessage, TimeSpan timeout, bool removeObject = false)
    {
        lock (_values)
        {
            var now = Environment.TickCount;
            int totalTimeout = (int)timeout.TotalMilliseconds;
            ++_waiters;
            SyncUtils.Pulse(_values, _hasWaiters);

            try
            {
                while (true)
                {
                    var node = FindNode(hasMessage);
                    if (node != null)
                    {
                        if (removeObject)
                        {
                            _values.Remove(node);
                        }
                        return node.Value;
                    }
                    Monitor.Wait(_values, totalTimeout);
                    if (SyncUtils.HasTimedOut(ref totalTimeout, now))
                    {
                        return default(T);
                    }
                }
            }
            catch (ThreadInterruptedException)
            {
                var node = FindNode(hasMessage);
                if (removeObject && node != null)
                {
                    _values.Remove(node);
                    Thread.CurrentThread.Interrupt();
                    return node.Value;
                }
                throw;
            }
            finally
            {
                --_waiters;
            }
        }
    }
}

And the utility methods used there, they are not for review either, they are here for completeness:

public static class SyncUtils
{
    private static void EnterUninterruptibly(Object lockObj, bool throwException = false, ThreadInterruptedException previous = null)
    {
        ThreadInterruptedException ex = previous;
        for (; ; )
        {
            try
            {
                Monitor.Enter(lockObj);
                break;
            }
            catch (ThreadInterruptedException e)
            {
                ex = e;
            }
        }
        if (throwException && ex != null)
        {
            throw ex;
        }
        if (ex != null)
        {
            Thread.CurrentThread.Interrupt(); // NOTE: dont't throw ThreadInterruptedException but DO keep interrupted status
        }
    }

    public static void Pulse(Object lockObj, Object condObj)
    {
        if (lockObj == condObj)
        {
            Monitor.Pulse(condObj);
        }
        else
        {
            EnterUninterruptibly(condObj); // NOTE: a Pulse should never throw ThreadInterruptedException
            Monitor.Pulse(condObj);
            Monitor.Exit(condObj);
        }
    }

    public static bool HasTimedOut(ref int timeout, int referenceTime)
    {
        if (timeout == Timeout.Infinite)
        {
            return false;
        }
        timeout = timeout - (Environment.TickCount - referenceTime);
        if (timeout <= 0)
        {
            timeout = 0;
            return true;
        }
        return false;
    }
}

Concurrent software should always be tested, for that reason I include a couple of tests (you don't have to review them):

public const int TakeNMessages = 10000;

static void Main(string[] args)
{
    foreach (var thread in TestWithWaitersFirst())
    {
        thread.Join();
    }
    Console.WriteLine("Completed " + "TestWithWaitersFirst");
    foreach (var thread in TestWithProducersFirst())
    {
        thread.Join();
    }
    Console.WriteLine("Completed " + "TestWithProducersFirst");
    Console.Read();
}

private static void WriteWithThreadId(string message)
{
    Console.WriteLine("Thread" + Thread.CurrentThread.ManagedThreadId + ": " + message);
}

private static IEnumerable<Thread> TestWithWaitersFirst()
{
    var queue = new FilterQueue<int?>();
    var takers = Enumerable.Range(0, 4).Select(i =>
    {
        return new Thread(() =>
        {
            for (int nTake = 0; nTake < TakeNMessages; ++nTake)
            {
                var value = TakeNMessages*i*10 + nTake;
                WriteWithThreadId("is waiting for " + value);
                queue.Take(n => n == value);
                WriteWithThreadId("received " + value);
            }
        });
    }).ToList();
    Thread.Sleep(200);
    var producers = Enumerable.Range(0, 4).Select(i =>
    {
        return new Thread(() =>
        {
            for (int nTake = 0; nTake < TakeNMessages; ++nTake)
            {
                var value = TakeNMessages * i * 10 + nTake;
                WriteWithThreadId("adding " + value);
                queue.Add(TakeNMessages * i * 10 + nTake);
            }
        });
    });
    var threads = takers.Concat(producers).ToArray();
    foreach (var thread in threads)
    {
        thread.Start();
    }
    return threads;
}

private static IEnumerable<Thread> TestWithProducersFirst()
{
    var queue = new FilterQueue<int?>();
    var producers = Enumerable.Range(0, 4).Select(i =>
    {
        return new Thread(() =>
        {
            for (int nTake = 0; nTake < TakeNMessages; ++nTake)
            {
                var value = TakeNMessages * i * 10 + nTake;
                WriteWithThreadId("adding " + value);
                queue.Add(TakeNMessages * i * 10 + nTake);
            }
        });
    });
    Thread.Sleep(200);
    var takers = Enumerable.Range(0, 4).Select(i =>
    {
        return new Thread(() =>
        {
            for (int nTake = 0; nTake < TakeNMessages; ++nTake)
            {
                var value = TakeNMessages * i * 10 + nTake;
                WriteWithThreadId("is waiting for " + value);
                queue.Take(n => n == value);
                WriteWithThreadId("received " + value);
            }
        });
    }).ToList();
    var threads = takers.Concat(producers).ToArray();
    foreach (var thread in threads)
    {
        thread.Start();
    }
    return threads;
}

Any comments are appreciated, specially regarding concurrent concerns that I may have missed.

Answer 1

I cannot tell you whether it's 100% thread safe and correct but I had some general comments on the design.

---

I wouldn't call this structure a Queue because it isn't one. There is no Push/Pop/Peek and it doesn't work like this. It is more an ObjectLocker then a queue.

---

Take

I think GetValueOrDefault would be a better name.

Secondly you might consider it to return Task<T> instead of just T as this method can take some time. With this you could await for the result.

If you chose to do this, then you also might consider another argument, the CancellationToken in case you decide to no longer wait.

---

SyncUtils

A parameter like throwException is a really bad choice. If you want to be able to switch between throwing and not throwing an exception then you should have two methods. One that is called EnterUninterruptibly that throws exceptions and the other one TryEnterUninterruptibly that doesn't throw and don't have to catch any exceptions.

You can perfectly solve it this way becasue the Monitor class already has such two methods: Enter and TryEnter that you can use for the new implementation.

---

DI

The Queue shouldn't rely on static SyncUtils. It would be better to pass it via DI as an instance that implements an IObjectSyncronizer or something similar.

Answer 2

While FilterQueue might seem to be useful the method Take leaves me a bit concerned. The reason why is that it optionally removes the item from the list. So if the method is always called with false the list will keep increasing.

This class probably resulted from badly designed code that always put all items on the list and then waiters would verify if that item on the list or not.

Instead of doing that, the waiters are the ones that should put items on the list. Then whenever an item arrived the waiters would be notified.

public class FilterList<T>
    where T : class
{
    private readonly LinkedList<T> _waitingMessages = new LinkedList<T>();

    public void WaitForWaiters()
    {
        lock (_waitingMessages)
        {
            while (_waitingMessages.Count == 0)
            {
                Monitor.Wait(_waitingMessages);
            }
        }
    }

    //An item arrived, notify waiters
    public void Notify(Predicate<T> pred)
    {
        lock (_waitingMessages)
        {
            var node = _waitingMessages.First;
            while (node != null)
            {
                var next = node.Next;
                if (pred(node.Value))
                {
                    _waitingMessages.Remove(node);
                    SyncUtils.Pulse(_waitingMessages, node);
                }
                node = next;
            }
        }
    }

    //The waiter adds a message to the list
    public object AddMessage(T expectedMessage)
    {
        lock (_waitingMessages)
        {
            var node = _waitingMessages.AddLast(expectedMessage);
            Monitor.PulseAll(_waitingMessages);
            return node;
        }
    }

    //The waiter waits for a notification
    public T WaitForNotification(object token, TimeSpan timeout)
    {
        var now = Environment.TickCount;
        int totalTimeout = (int)timeout.TotalMilliseconds;
        var node = token as LinkedListNode<T>;
        if (node == null)
        {
            throw new InvalidOperationException("token is not of the expected type");
        }

        lock (_waitingMessages)
        {
            if (node.Next == null && node.Previous == null)
            {
                return node.Value;
            }
            try
            {
                while (true)
                {
                    SyncUtils.Wait(_waitingMessages, node, totalTimeout);
                    if (node.Next == null && node.Previous == null)
                    {
                        return node.Value;
                    }
                    if (ObjectUtilities.HasTimedOut(ref totalTimeout, now))
                    {
                        _waitingMessages.Remove(node);
                        return null;
                    }
                }
            }
            catch (ThreadInterruptedException)
            {
                _waitingMessages.Remove(node);
                throw;
            }
        }
    }
}