From: Robert Klemme <shortcutter@...>
Date: 2009-04-17T15:55:35+09:00
Subject: Re: Multi Threading

2009/4/17 Sriram Varahan <sriram.varahan@gmail.com>:
> Hello,
>
> #*******************STARTCODE
>
> start_time = Time.now
> $count = 0
> class Queue
>  def initialize *s # splat operator allows variable length argument
> list
>    @mutex = Mutex.new
>    @queue = []
>    s.each { |e| @queue.push e }
>  end
>
>  def enq v
>      @queue.push v
>  end
>
>  def deq
>    @mutex.synchronize {item = @queue.shift}
>  end
>
>  def empty?
>    @mutex.synchronize{@queue.length == 0}
>  end
>
>  def count
>   @mutex.synchronize do
>     $count += 1
>   end
>  end
> end
>
>
> #*****Test
>
> queue = Queue.new
> 500.times do |a|
>  queue.enq a
> end
> threads = []
>
>
> # Create 5 threads which fetch values from the Q.
>  5.times do
>    threads << Thread.new {
>    until queue.empty?
>      queue.count
>      puts "Thread ID: #{Thread.current}.Job  started"
>      puts "#{queue.deq}"
>      #sleep 0.0001
>      puts "Thread ID: #{Thread.current}.Job  complete"
>    end
>   }
>  end
>
>
> threads.each {|t| t.join }
> puts "Count"
> puts $count
> puts "timeTaken:"
> puts Time.now - start_time
>
> # *************CODE ENDS******************
>
>
> I have five threads which fetch values from a queue. The above code
> works perfectly well in case of a single thread. But the issue arises
> when there are more threads.
>
> In case of 5 threads the number of times the block is executed is 503
> where it should have been 500.
>
> I know the reason why?
> The "deq" and "empty?" methods are not synchronized.
> So when the final item is removed from the thread, other threads  access
> the empty? method before the @queue.length becomes 0.
>
> Hence the difference in count.
>
> If the sleep is activated this problem is solved.
>
> Any suggestion on how to get this working without the sleep?

There are several options. You could use MonitorMixin instead of Mutex
and include it in initialize

def initialize *s
  # @mutex == self so you do not need to change sync code
  @mutex = extend MonitorMixin
  @queue = s.dup
end

Then you can do external synchronization, e.g.

queue.synchronize do
  if queue.empty?
   # finish
  else
    elm = deq
  end
end

Much better though is this approach

require 'thread'

# use library class
queue = Queue.new

# _first_ start threads
# does not really matter but if filling
# the queue takes time work can
# begin immediately
threads = (1..5).map do
  label = Thread.current.to_s.freeze

  Thread.new queue do |q|
    until ( job = q.deq ) == :terminate
      puts "Thread ID: #{label}.Job  started"
      puts job
      puts "Thread ID: #{label}.Job  complete"
    end
  end
end

# fill queues
500.times do |a|
 queue.enq a
end

# "close" queues
threads.size.times { queue.enq :terminate }

# wait for termination
threads.each do |th|
  th.join
end

Cheers

robert

-- 
remember.guy do |as, often| as.you_can - without end
http://blog.rubybestpractices.com/