Add dequeueAll to Queue #386
Conversation
| @@ -76,6 +76,12 @@ abstract class Queue[T] { | |||
|
|
|||
| private val count = new AtomicInteger(0) | |||
|
|
|||
| def dequeueAll(fn: T => Boolean): Seq[T] = { | |||
| val (result, putBack) = toSeq.partition(fn) | |||
There was a problem hiding this comment.
Instead of partitioning and putting back, will using a version of trimTo that takes in a predicate work better?
That version could trim all elements that satisfy the predicate.
Or am I missing any queue semantics here?
There was a problem hiding this comment.
The issue is that for the items that fail the predicate need to be put back into the queue. This cannot be done better than we have here if we only have poll and add.
The better solution is to atomically swap the tail out with an empty queue, filter the existing items and push those failing the filter onto a new queue, then move into a phase of draining the tail. When the tail is drained, atomically swap the refilled queue with old location. I don't think this is needed for the case we are considering because there is only one thread processing input tuples and that is the same thread that will cass this dequeueAll. So, there is no (or little) concern with out-of-order issues.
There was a problem hiding this comment.
Got it. Thanks for the explanation.
applyCumulative method on Aggregator
Useful to clear out finished futures if we enqueue them.