Exit ConcurrentQueue only if condition is met
How can I remove the next item ConcurrentQueue
only if some condition is met?
eg. if the next item to be removed completes a certain condition, remove it, otherwise leave it. Essentially a method 'DequeueIf'
or'TryDequeueIf'
Example:
var myQueue = new ConcurrentQueue<int>()
...
int item;
// only dequeue if the next item is 0
bool success = myQueue.TryDequeueIf(out item, x=>x==0)
Of course, you can call first TryPeek
, check the condition and then TryDequeue
, but this is no longer thread-safe.
I can wrap an integer TryPeek & TryDequeue
in a lock, but this kind defeats the purpose of using ConcurrentQueue; and means that all regular unconditional queues must also be blocked. I'm not sure if I'll have to block everything Enqueue
to save. If possible, I would like to avoid possible pitfalls in implementing my own locking strategy.
Is there a non-blocking solution using the .net4.0 class ConcurrentQueue
or one of the other concurrent classes?
source to share
This is not possible with built-in tools. What to do?
- Write your own regular queue. Just use a lock for each queue. If you have too much traffic in the queue, that will be very good. Illegal locks consume two related operations per cycle.
- Write a complex queue using blocking operations. By using a CAS redo loop, you can implement a predefined atomic capture operation. You can probably use the BCL source code as a starting point or inspiration.
- Give up the idea of having a line. Do it in a queue without a queue. If you are drawing the wrong element, just push it back into the queue.
source to share
As a complement to @usr's answer :
I can wrap a whole
TryPeek
andTryDequeue
in the lock, but this kind of hit the target useConcurrentQueue
Not entirely, or at least not completely, since one of its main goals is to coordinate the convergence between producer and consumer, and the use of blocking in this case will not block any producers, but only other consumers. But this means that you will need to use the same lock for all other error read operations.
So, if this is not acceptable, you will have to either
- Create your own concurrent queue, or a better data structure (heap?) Optimized for your specific usage pattern.
- Consider an alternative approach that best suits your needs: use the "push" or "event" model instead of the "pull" model. Using, for example, Rx and consumers watching with a filter
where
.
source to share
Most concurrentQueue implementations have a non-blocking implementation, allowing many producers and consumers to write to the queue at the same time without blocking each other (except for accidental restarting)
The basic requirement of the conditional dequeue is the implementation of the "pop if on top" function: "bool dequeueIf (T) // Deblocks T if T is still a head, returns false if it is not a head; this allows:
T t=null;
do{ // need to spin in case the head changes between the peek and dequeue attempt
done=-1; // assume failure
T t2=peek(),
if (t!=null && someComplexCondition(t)) {// then it what we want
if(!dequeueIf(T)) // check to see if the head is changed
{done=0;} // spin if the head is changed
else {done=1;t=t2;} // it meets the criteria, and we de-queued it.
}
} while (done==0);
// done = -1 == head does not meet criteria
// done = 1 == T contains poped item that met criteria
dequeueIf (T) needs to be done inside the queue.
In the case of .Net ConcurrentQueue, a modified version of Dequeue is required (from the result of T): Dequeue (ref T toRemove).
This goes to the newer version concurrentQueue.Segment.TryRemove (out T result): concurrentQueue.Segment.TryRemove (ref T toRemove)
This function should, before doing compareExchange for the de-queue, first compare the T in "local" with toRemove and only remove it if it is still on top (otherwise it can be assumed that other TryRemove threads have already knocked it out value from the queue, so you cannot return the value in question: toRemove must be null then).
Note; you cannot overload f (out T) and f (ref T), so returning a bool to indicate that you now "own" the T you requested seems like a reasonable solution, otherwise I would use the f (T toRemove) overload so that the original the function signature has not changed.
source to share