From: Robert Feldt <feldt@...>
Date: 2000-10-23T11:36:40+09:00
Subject: [ruby-talk:5765] Queue flawed!?

Hi,

Since no one has answered my Q on Queue I went ahead and asked the
interpreter. If the script below isn't flawed, running it seems to show
that Queue is flawed. I wrote a thread safe variant (TsQueue) and throwed
in a variant with arbitrary object and thread dispatching
(TsDispatchedQueue) while I was at it. I hope this can help eliminate bugs
in thread.rb.

If there are any flaws in the script so that my results aren't valid or 
if you get different results when running the script I'd
appreciate if you send me a note.

Here's the output when I run the script:
bash-2.04$ ruby -v queue_bug.rb
ruby 1.6.1 (2000-09-27) [i686-cygwin]
Queue (3 ThreadErrors, 12 errors) one of which is
  ERROR: expected 0 but got 1.
TsQueue OK. (0 errors, 0 ThreadErrors)
TsDispatchedQueue OK. (0 errors, 0 ThreadErrors)

Regards,

Robert

queue_bug.rb:
require 'thread'

def some(ary, &b)
  ary.each {|e| 
    if b.call(e)
      return true
    end
  }
  false
end

class TsQueue < Queue
  def initialize
    super
    @mutex = Mutex.new
    @dispatched_objects = {} # Maps thread ids to popped object
  end

  def push(obj)
    @mutex.lock
    begin
      @que.push obj
      if @waiting.length > 0
	t = @waiting.shift
	t.wakeup
	@dispatched_objects[t.id] = @que.shift
      end
    ensure
      @mutex.unlock
    end
  end

  def pop(non_block=false)
    @mutex.lock
    unlocked = false
    begin
      if @que.length > 0
	return @que.shift
      else
	if non_block
	  raise ThreadError, "queue empty"
	end
	@waiting.push Thread.current
	@mutex.unlock
	unlocked = true
	Thread.stop
	@mutex.lock
	obj = @dispatched_objects.delete(Thread.current.id)
	@mutex.unlock
	return obj
      end
    ensure
      @mutex.unlock unless unlocked
    end
  end
end

# Dispatchers map ([objs],get_prio) -> [obj, [objs]] were the first
# result are the dispatched object and  the second result are the 
# remaining objects. The proc 'get_prio' maps an object to its priority.
FIFO = proc{ |objs,gp| [objs[0], objs[1..-1]] }
LIFO = proc{ |objs,gp| [objs[-1], objs[0..-2]] }
HighestPriority = proc{ |objs,get_prio|
  # NOTE: The sort is not garantueed to preserve the ordering of elements
  # with the same priority. Define your own sorting dispatcher if you need
  # that!
  sorted_objs = objs.sort {|o1,o2| get_prio.call(o2) <=>
get_prio.call(o1)}
  FIFO.call(sorted_objs,get_prio)
}
LowestPriority = proc{ |objs,get_prio|
  # NOTE: The sort is not garantueed to preserve the ordering of elements
  # with the same priority. Define your own sorting dispatcher if you need
  # that!
  sorted_objs = objs.sort {|o1,o2| get_prio.call(o1) <=>
get_prio.call(o2)}
  FIFO.call(sorted_objs,get_prio)
}

class TsDispatchedQueue < TsQueue
  def initialize(objectdispatcher = FIFO, threaddispatcher = FIFO,
		 get_prio = proc{|e| e[0]})
    super()
    @get_prio, @odisp, @tdisp = get_prio, objectdispatcher,
threaddispatcher
  end

  def push(obj, prio=0)
    @mutex.lock
    begin
      @que.push [prio, obj]
      if @waiting.length > 0
	(p, t), @waiting = @tdisp.call(@waiting, @get_prio)
	t.wakeup
	(p, @dispatched_objects[t.id]), @que = @odisp.call(@que,
@get_prio)
      end
    ensure
      @mutex.unlock
    end
  end

  # Arg can be either:
  #   Numeric => priority of waiting thread, Blocking call
  #   true    => Non-blocking call
  #   other   => priority is 0, Blocking call
  def pop(arg=nil)
    @mutex.lock
    unlocked = false
    if true == arg
      non_blocking = true
    else
      non_blocking = false
      prio = (arg.kind_of?(Numeric) ? arg : 0)
    end
    begin
      if @que.length > 0
	(p, obj), @que = @odisp.call(@que, @get_prio)
	return obj
      else
	if non_blocking
	  raise ThreadError, "queue empty"
	end
	@waiting.push [prio, Thread.current]
	@mutex.unlock
	unlocked = true
	Thread.stop
	@mutex.lock
	obj = @dispatched_objects.delete(Thread.current.id)
	@mutex.unlock
	return obj
      end
    ensure
      @mutex.unlock unless unlocked
    end
  end    
end

[Queue, TsQueue, TsDispatchedQueue].each do |qclass|
  print "#{qclass.inspect} "
  num_tests = 0
  errors = []
  num_thread_errors = 0
  (2..10).each do |num_threads|
    10.times do
      num_tests += 1
      q = qclass.new
      threads = Array.new
      num_threads.times do |i|
	threads.push(Thread.new { q.pop })
      while(!threads.last.stop?)
	Thread.pass 
      end
      end
      begin
	num_threads.times {|i| q.push i}
	num_threads.times {|i|
	  if i != threads[i].value
	    errors.push "ERROR: expected #{i} but got #{threads[i].value}."
	  end
	}
      rescue ThreadError
      num_thread_errors += 1
      end
    end
  end
  if errors.length > 0 or num_thread_errors > 0
    print "(#{num_thread_errors} ThreadErrors, #{errors.length}
errors) one of which is\n  #{errors[0]}\n"
  else
    print "OK. (#{errors.length} errors, #{num_thread_errors}
ThreadErrors)\n"
  end
end