Multi Threading in Ruby

Question

I need to create 3 threads. Each thread will print on the screen a collor and sleep for x seconds. Thread A will print red; Thread B will print yellow; Thread C will print green;

All threads must wait until its their turn to print. The first thread to print must be Red, after printing, Red will tell Yellow that's its turn to print and so on. The threads must be able to print multiple times (user specific)

I'm stuck because calling @firstFlag.signal outside a Thread isn't working and the 3 threads aren't working on the right order How do I make the red Thread go first?

my code so far:

@lock = Mutex.new
@firstFlag = ConditionVariable.new
@secondFlag = ConditionVariable.new
@thirdFlag = ConditionVariable.new

print "Tell me n's vallue:"
@n = gets.to_i

@threads = Array.new

@threads << Thread.new() {
  t = Random.rand(1..3)
  n = 0
  @lock.synchronize {  
    for i in 0...@n do
      @firstFlag.wait(@lock, t)     
      puts "red : #{t}s"
      sleep(t)
      @secondFlag.signal      
    end
  }
}

@threads << Thread.new() {
  t = Random.rand(1..3)
  n = 0
  @lock.synchronize {  
    for i in 0...@n do         
      @secondFlag.wait(@lock, t)     
      puts "yellow : #{t}s"
      sleep(t)
      @thirdFlag.signal
    end
  }
}

@threads << Thread.new() {
  t = Random.rand(1..3)
  n = 0
  @lock.synchronize {  
    for i in 0...@n do      
      @thirdFlag.wait(@lock, t)     
      puts "green : #{t}s"
      sleep(t)
      @firstFlag.signal
    end
  }
}

@threads.each {|t| t.join}
@firstFlag.signal

Answer 1

I'd redesign this slightly. Think of your ConditionVariables as flags that a thread uses to say it's done for now, and name them accordingly:

@lock = Mutex.new
@thread_a_done = ConditionVariable.new
@thread_b_done = ConditionVariable.new
@thread_c_done = ConditionVariable.new

Now, thread A signals it's done by doing @thread_a_done.signal, and thread B can wait for that signal, etc. Thread A of course needs to wait until thread C is done, so we get this kind of structure:

 @threads << Thread.new() {
  t = Random.rand(1..3)
  @lock.synchronize {
    for i in 0...@n do
      @thread_c_done.wait(@lock)
      puts "A: red : #{t}s"
      sleep(t)
      @thread_a_done.signal
    end
  }
}

A problem here is that you need to make sure that thread A in the first iteration doesn't wait for a flag signal. After all, it's to go first, so it shouldn't wait for anyone else. So modify it to:

@thread_c_done.wait(@lock) unless i == 0 

Finally, once you have created your threads, kick them all off by invoking run, then join on each thread (so that your program doesn't exit before the last thread is done):

@threads.each(&:run)
@threads.each(&:join)

Oh btw I'd get rid of the timeouts in your wait as well. You have a hard requirement that they go in order. If you make the signal wait time out you screw that up - threads might still "jump the queue" so to speak.

EDIT as @casper remarked below, this still has a potential race condition: Thread A could call signal before thread B is waiting to receive it, in which case thread B will miss it and just wait indefinitely. A possible way to fix this is to use some form of a CountDownLatch - a shared object that all threads can wait on, which gets released as soon as all threads have signalled that they're ready. The ruby-concurrency gem has an implementation of this, and in fact might have other interesting things to use for more elegant multi-threaded programming.

Sticking with pure ruby though, you could possibly fix this by adding a second Mutex that guards shared access to a boolean flag to indicate the thread is ready.

Answer 2

Ok, thank you guys that answered. I've found a solution:

I've created a fourth thread. Because I found out that calling "@firstFlag.signal" outside a thread doesn't work, because ruby has a "main thread" that sleeps when you "run" other threads. So, "@firstFlag.signal" calling must be inside a thread so it can be on the same level of the CV.wait

I solved the issue using this:

@threads << Thread.new {
  sleep 1
  @firstFlag.signal
}

This fourth thread will wait for 1 sec before sending the first signal to red. This only sec seems to be enough for the others thread reach the wait point. And, I've removed the timeout, as you sugested.

//Edit//

I realized I don't need a fourth Thread, I could just make thread C do the first signal. I made thread C sleep for 1 sec to wait the other two threads enter in wait state, then it signals red to start and goes to wait too

@threads << Thread.new() {
  sleep 1
  @redFlag.signal

  t = Random.rand(1..3)
  n = 0
  @lock.synchronize {  
    for i in 0...@n do      
      @greenFlag.wait(@lock)     
      puts "verde : #{t}s"
      sleep(t)
      @redFlag.signal 
      n += 1
    end
  }
}

Answer 3

There are three bugs in your code:

First bug

Your wait calls use a timeout. This means your threads will become de-synchronized from your intended sequence, because the timeout will let each thread slip past your intended wait point.

Solution: change all your wait calls to NOT use a timeout:

@xxxxFlag.wait(@lock)

---

Second bug

You put your sequence trigger AFTER your Thread.join call in the end. Your join call will never return, and hence the last statement in your code will never be executed, and your thread sequence will never start.

Solution: change the order to signal the sequence start first, and then join the threads:

@firstFlag.signal
@threads.each {|t| t.join} 

---

Third bug

The problem with a wait/signal construction is that it does not buffer the signals. Therefore you have to ensure all threads are in their wait state before calling signal, otherwise you may encounter a race condition where a thread calls signal before another thread has called wait.

Solution: This a bit harder to solve, although it is possible to solve with Queue. But I propose a complete rethinking of your code instead. See below for the full solution.

---

Better solution

I think you need to rethink the whole construction, and instead of condition variables just use Queue for everything. Now the code becomes much less brittle, and because Queue itself is thread safe, you do not need any critical sections any more.

The advantage of Queue is that you can use it like a wait/signal construction, but it buffers the signals, which makes everything much simpler in this case.

Now we can rewrite the code:

redq    = Queue.new
yellowq = Queue.new
greenq  = Queue.new

Then each thread becomes like this:

@threads << Thread.new() {
  t = Random.rand(1..3)
  n = 0

  for i in 0...@n do
    redq.pop
    puts "red : #{t}s"
    sleep(t)
    yellowq.push(1)
  end
}

And finally to kick off the whole sequence:

redq.push(1)
@threads.each { |t| t.join }