Joel VanderWerf
7/10/2008 7:03:00 PM
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