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.