From: Joel VanderWerf <vjoel@...>
Date: 2008-07-11T04:03:16+09:00
Subject: Re: Thread-safe priority queue?

Joel VanderWerf wrote:
> Joel VanderWerf wrote:
>>
>> Looks like a race condition in that...
> 
> Proposed fix, using a condition var... still needs some eyeballing and 
> some tests:

That was not quite right either (because cond.signal only wakes the 
waiter, and doesn't schedule it). The following seems to complete 
without deadlocks or starvation.

require 'thread'
require 'rbtree'

class PriorityQueue
   def size
     @tree.size
   end

   def initialize(*)
     super
     @tree = MultiRBTree.new
     @mutex = Mutex.new
     @cond = ConditionVariable.new
   end

   # Push +obj+ with priority equal to +pri+ if given or, otherwise,
   # the result of sending #queue_priority to +obj+. Objects are
   # dequeued in priority order, and first-in-first-out among objects
   # with equal priorities.
   def push(obj, pri = obj.queue_priority)
     @mutex.synchronize do
       @tree.store(pri, obj)
       @cond.signal
     end
   end

   def pop(non_block=false)
     @mutex.synchronize do
       if (last=@tree.last)
         return @tree.delete(last[0]) # highest key, oldest first
       end

       if non_block
         raise ThreadError, "priority queue empty"
       end

       loop do
         @cond.wait(@mutex)
         if (last=@tree.last)
           return @tree.delete(last[0])
         end
       end
     end
   end
end


if __FILE__ == $0

   Thread.abort_on_exception = true

   pq = PriorityQueue.new

   n_items_per_thread = 1000
   n_writers = 10
   n_readers = 10

   writers = (0...n_writers).map do |i_thr|
     Thread.new do
       n_items_per_thread.times do |i|
         pri = rand(10)
         pq.push([pri, i, i_thr], pri)
         Thread.pass if rand(5) == 0
       end
     end
   end

   sleep 0.1 until pq.size > 100 # a little head start populating the tree

   results = Array.new(n_readers, 0)

   readers = (0...n_readers).map do |i|
     Thread.new do
       loop do
         pq.pop
         results[i] += 1
       end
     end
   end

   writers.each do |wr|
     wr.join
   end

   p results
   until pq.size == 0
     sleep 0.1
     p results
   end

   raise unless results.inject {|s,x|s+x} == n_items_per_thread * n_writers

end


-- 
       vjoel : Joel VanderWerf : path berkeley edu : 510 665 3407