From 91ec46d0cb04038b529286355cb34ac9daedbd25 Mon Sep 17 00:00:00 2001
From: Sokolov Yura 'funny-falcon <funny.falcon@gmail.com>
Date: Fri, 14 Sep 2012 12:49:05 +0400
Subject: [PATCH 1/5] test_thread: add failing test for condvar.wait timeout

Thread, which call ConditionVariable#wait with timeout, could
not remove itself from waiters until it acquires lock on mutex.
So that, ConditionVariable#signal will try to wakeup already
awoken thread instead of one still waiting for a signal.
While user should not rely on this timeouts when he calls
#signal, I still think it is better to fix it.
---
 test/ruby/test_thread.rb |   71 ++++++++++++++++++++++++++++++++++++++++++++++
 1 file changed, 71 insertions(+)

diff --git a/test/ruby/test_thread.rb b/test/ruby/test_thread.rb
index c8081b4..620aa5b 100644
--- a/test/ruby/test_thread.rb
+++ b/test/ruby/test_thread.rb
@@ -209,6 +209,77 @@ def test_condvar_nolock_3
     end.join
   end
 
+  class Serializer
+    def initialize
+      @mutex = Mutex.new
+      @condvar = ConditionVariable.new
+    end
+    def wait
+      @mutex.synchronize{
+        @condvar.wait(@mutex)
+      }
+    end
+    def signal
+      @mutex.synchronize{ @condvar.signal }
+    end
+  end
+
+  def test_condvar_wait_timeout
+    serialize = Serializer.new
+
+    mutex = Mutex.new
+    condvar = ConditionVariable.new
+    def condvar.waiters
+      @waiters
+    end
+    thread = Thread.new do
+      serialize.signal
+      mutex.synchronize do
+        condvar.wait(mutex, 0.001)
+      end
+    end
+    serialize.wait
+    mutex.synchronize do
+      sleep(0.01)
+      assert_not_includes(condvar.waiters, thread)
+    end
+  end
+
+  def test_condvar_wait_timeout_2
+    serialize = Serializer.new
+
+    mutex = Mutex.new
+    condvar = ConditionVariable.new
+
+    wait_timeout = Thread.new do
+      serialize.signal
+      mutex.synchronize do
+        condvar.wait(mutex, 0.001)
+      end
+    end
+    serialize.wait
+
+    wait_forever = Thread.new do
+      serialize.signal
+      mutex.synchronize do
+        condvar.wait(mutex)
+      end
+    end
+    serialize.wait
+
+    mutex.synchronize do
+      sleep(0.01)
+      condvar.signal
+    end
+    # If wait_timeout thread didn't remove himself
+    # from condvar sleepers, than signale tries to
+    # wakeup wait_timeout instead of wait_forever
+    # and fatal deadlock occures in a line below
+    wait_timeout.join
+    wait_forever.join
+
+  end
+
   def test_local_barrier
     dir = File.dirname(__FILE__)
     lbtest = File.join(dir, "lbtest.rb")
-- 
1.7.10


From 4929fe36158fc2978bb37bf0cb08db3c57a70870 Mon Sep 17 00:00:00 2001
From: Sokolov Yura 'funny-falcon <funny.falcon@gmail.com>
Date: Fri, 14 Sep 2012 13:53:10 +0400
Subject: [PATCH 2/5] * thread.c : add Mutex#sleep ability to accept block to
 run before lock

To fix case with ConditionVariable#wait timeout, simplest way is to
remove thread from sleepers before acquiring lock on mutex. To do
it, we should extend Mutex#sleep method to have ability accept block
and run it before trying to lock.
Perhaps, it is not the best way to do thing, but the simplest one.
---
 lib/thread.rb |   12 +++++-------
 thread.c      |   22 ++++++++++++++++++++--
 2 files changed, 25 insertions(+), 9 deletions(-)

diff --git a/lib/thread.rb b/lib/thread.rb
index 0a5d1f4..858a78d 100644
--- a/lib/thread.rb
+++ b/lib/thread.rb
@@ -63,13 +63,11 @@ def initialize
   # even if no other thread doesn't signal.
   #
   def wait(mutex, timeout=nil)
-    begin
-      # TODO: mutex should not be used
-      @waiters_mutex.synchronize do
-        @waiters.push(Thread.current)
-      end
-      mutex.sleep timeout
-    ensure
+    # TODO: mutex should not be used
+    @waiters_mutex.synchronize do
+      @waiters.push(Thread.current)
+    end
+    mutex.sleep timeout do
       @waiters_mutex.synchronize do
         @waiters.delete(Thread.current)
       end
diff --git a/thread.c b/thread.c
index 8320ed2..da53585 100644
--- a/thread.c
+++ b/thread.c
@@ -4056,20 +4056,38 @@ struct thgroup_list_params {
 }
 
 static VALUE
-rb_mutex_sleep_forever(VALUE time)
+wrap_thread_sleep_deadly()
 {
     rb_thread_sleep_deadly();
     return Qnil;
 }
 
 static VALUE
-rb_mutex_wait_for(VALUE time)
+rb_mutex_sleep_forever(VALUE time)
+{
+    if (rb_block_given_p()) {
+	return rb_ensure(wrap_thread_sleep_deadly, Qnil, rb_yield, Qnil);
+    }
+    return wrap_thread_sleep_deadly();
+}
+
+static VALUE
+wrap_rb_thread_wait_for(VALUE time)
 {
     const struct timeval *t = (struct timeval *)time;
     rb_thread_wait_for(*t);
     return Qnil;
 }
 
+static VALUE
+rb_mutex_wait_for(VALUE time)
+{
+    if (rb_block_given_p()) {
+	return rb_ensure(wrap_rb_thread_wait_for, time, rb_yield, Qnil);
+    }
+    return wrap_rb_thread_wait_for(time);
+}
+
 VALUE
 rb_mutex_sleep(VALUE self, VALUE timeout)
 {
-- 
1.7.10


From 13e9dbdc42f6a6269089ddefe05f6d2f75927037 Mon Sep 17 00:00:00 2001
From: Sokolov Yura <funny.falcon@gmail.com>
Date: Sun, 18 Mar 2012 16:22:59 +0400
Subject: [PATCH 3/5] Thread::RELY_ON_GVL

---
 lib/thread.rb |   33 +++++++++++++++++++++++++++++----
 thread.c      |    1 +
 2 files changed, 30 insertions(+), 4 deletions(-)

diff --git a/lib/thread.rb b/lib/thread.rb
index 858a78d..b37724e 100644
--- a/lib/thread.rb
+++ b/lib/thread.rb
@@ -20,6 +20,10 @@ class ThreadError < StandardError
   Thread.abort_on_exception = true
 end
 
+unless defined?(Thread::RELY_ON_GVL)
+  Thread::RELY_ON_GVL = false
+end
+
 #
 # ConditionVariable objects augment class Mutex. Using condition variables,
 # it is possible to suspend while in the middle of a critical section until a
@@ -63,7 +67,18 @@ def initialize
   # even if no other thread doesn't signal.
   #
   def wait(mutex, timeout=nil)
-    # TODO: mutex should not be used
+    # Rely on GVL for sychronizing @waiters.push
+    @waiters.push(Thread.current)
+    mutex.sleep timeout do
+      # We could not rely on GVL cause compare is called
+      @waiters_mutex.synchronize do
+        @waiters.delete(Thread.current)
+      end
+    end
+    self
+  end if Thread::RELY_ON_GVL
+
+  def wait(mutex, timeout=nil) # :nodoc: 
     @waiters_mutex.synchronize do
       @waiters.push(Thread.current)
     end
@@ -73,20 +88,30 @@ def wait(mutex, timeout=nil)
       end
     end
     self
-  end
+  end unless Thread::RELY_ON_GVL
 
   #
   # Wakes up the first thread in line waiting for this lock.
   #
   def signal
     begin
-      t = @waiters_mutex.synchronize {@waiters.shift}
+      t = @waiters.shift
       t.run if t
     rescue ThreadError
       retry
     end
     self
-  end
+  end if Thread::RELY_ON_GVL
+
+  def signal # :nodoc:
+    begin
+      t = @waiters_mutex.synchronize { @waiters.shift }
+      t.run if t
+    rescue ThreadError
+      retry
+    end
+    self
+  end unless Thread::RELY_ON_GVL
 
   #
   # Wakes up all threads waiting for this lock.
diff --git a/thread.c b/thread.c
index da53585..c394b53 100644
--- a/thread.c
+++ b/thread.c
@@ -4528,6 +4528,7 @@ struct exec_recursive_params {
     rb_define_singleton_method(rb_cThread, "list", rb_thread_list, 0);
     rb_define_singleton_method(rb_cThread, "abort_on_exception", rb_thread_s_abort_exc, 0);
     rb_define_singleton_method(rb_cThread, "abort_on_exception=", rb_thread_s_abort_exc_set, 1);
+    rb_define_const(rb_cThread, "RELY_ON_GVL", Qtrue);
 #if THREAD_DEBUG < 0
     rb_define_singleton_method(rb_cThread, "DEBUG", rb_thread_s_debug, 0);
     rb_define_singleton_method(rb_cThread, "DEBUG=", rb_thread_s_debug_set, 1);
-- 
1.7.10


From 2a02d772ce849633dc0650b78d3e0804a733f602 Mon Sep 17 00:00:00 2001
From: Sokolov Yura <funny.falcon@gmail.com>
Date: Sun, 18 Mar 2012 18:53:13 +0400
Subject: [PATCH 4/5] thread.rb: use hash with compare_by_identity instead of
 array for ordered storage

---
 lib/thread.rb |   64 ++++++++++++++++++++++++++-------------------------------
 1 file changed, 29 insertions(+), 35 deletions(-)

diff --git a/lib/thread.rb b/lib/thread.rb
index b37724e..7e4f203 100644
--- a/lib/thread.rb
+++ b/lib/thread.rb
@@ -56,7 +56,8 @@ class ConditionVariable
   # Creates a new ConditionVariable
   #
   def initialize
-    @waiters = []
+    @waiters = {}
+    @waiters.compare_by_identity
     @waiters_mutex = Mutex.new
   end
 
@@ -68,19 +69,17 @@ def initialize
   #
   def wait(mutex, timeout=nil)
     # Rely on GVL for sychronizing @waiters.push
-    @waiters.push(Thread.current)
+    @waiters[Thread.current] = true
     mutex.sleep timeout do
-      # We could not rely on GVL cause compare is called
-      @waiters_mutex.synchronize do
-        @waiters.delete(Thread.current)
-      end
+      # We could rely on GVL cause hash were set to compare_by_identity mode
+      @waiters.delete(Thread.current)
     end
     self
   end if Thread::RELY_ON_GVL
 
   def wait(mutex, timeout=nil) # :nodoc: 
     @waiters_mutex.synchronize do
-      @waiters.push(Thread.current)
+      @waiters[Thread.current] = true
     end
     mutex.sleep timeout do
       @waiters_mutex.synchronize do
@@ -95,7 +94,7 @@ def wait(mutex, timeout=nil) # :nodoc:
   #
   def signal
     begin
-      t = @waiters.shift
+      t, _ = @waiters.shift
       t.run if t
     rescue ThreadError
       retry
@@ -105,7 +104,7 @@ def signal
 
   def signal # :nodoc:
     begin
-      t = @waiters_mutex.synchronize { @waiters.shift }
+      t, _ = @waiters_mutex.synchronize { @waiters.shift }
       t.run if t
     rescue ThreadError
       retry
@@ -120,7 +119,7 @@ def broadcast
     # TODO: incomplete
     waiters0 = nil
     @waiters_mutex.synchronize do
-      waiters0 = @waiters.dup
+      waiters0 = @waiters.keys
       @waiters.clear
     end
     for t in waiters0
@@ -166,26 +165,29 @@ class Queue
   #
   def initialize
     @que = []
-    @waiting = []
+    @waiting = {}
+    @waiting.compare_by_identity
     @que.taint          # enable tainted communication
     @waiting.taint
     self.taint
     @mutex = Mutex.new
   end
 
+  def push_no_sync(obj) # :nodoc:
+    @que.push obj
+    begin
+      t, _ = @waiting.shift
+      t.wakeup if t
+    rescue ThreadError
+      retry
+    end
+  end
+  private :push_no_sync
   #
   # Pushes +obj+ to the queue.
   #
   def push(obj)
-    @mutex.synchronize{
-      @que.push obj
-      begin
-        t = @waiting.shift
-        t.wakeup if t
-      rescue ThreadError
-        retry
-      end
-    }
+    @mutex.synchronize{ push_no_sync(obj) }
   end
 
   #
@@ -209,9 +211,7 @@ def pop(non_block=false)
         while true
           if @que.empty?
             raise ThreadError, "queue empty" if non_block
-            # @waiting.include? check is necessary for avoiding a race against
-            # Thread.wakeup [Bug 5195]
-            @waiting.push Thread.current unless @waiting.include?(Thread.current)
+            @waiting[Thread.current] = true
             @mutex.sleep
           else
             return @que.shift
@@ -280,7 +280,8 @@ class SizedQueue < Queue
   def initialize(max)
     raise ArgumentError, "queue size must be positive" unless max > 0
     @max = max
-    @queue_wait = []
+    @queue_wait = {}
+    @queue_wait.compare_by_identity
     @queue_wait.taint           # enable tainted comunication
     super()
   end
@@ -309,7 +310,7 @@ def max=(max)
     if diff
       diff.times do
         begin
-          t = @queue_wait.shift
+          t, _ = @queue_wait.shift
           t.run if t
         rescue ThreadError
           retry
@@ -328,20 +329,13 @@ def push(obj)
       begin
         while true
           break if @que.length < @max
-          @queue_wait.push Thread.current unless @queue_wait.include?(Thread.current)
+          @queue_wait[Thread.current] = true
           @mutex.sleep
         end
       ensure
         @queue_wait.delete(Thread.current)
       end
-
-      @que.push obj
-      begin
-        t = @waiting.shift
-        t.wakeup if t
-      rescue ThreadError
-        retry
-      end
+      push_no_sync obj
     }
   end
 
@@ -363,7 +357,7 @@ def pop(*args)
     @mutex.synchronize {
       if @que.length < @max
         begin
-          t = @queue_wait.shift
+          t, _ = @queue_wait.shift
           t.wakeup if t
         rescue ThreadError
           retry
-- 
1.7.10


From 67e1b603225ffddb7c9b412805a5cdacfb2d36c5 Mon Sep 17 00:00:00 2001
From: Sokolov Yura <funny.falcon@gmail.com>
Date: Sun, 18 Mar 2012 19:36:09 +0400
Subject: [PATCH 5/5] thread.rb: cosmetic changes

One semantic difference:
 previously in SizedQueue#max= were used Thread#run outside of @mutex.synchronize
 now Thread#wakeup is called inside of @mutex.synchronize, as in SizedQueue#pop
I think, it is not big deal.
---
 lib/thread.rb |   76 ++++++++++++++++++++++++---------------------------------
 1 file changed, 32 insertions(+), 44 deletions(-)

diff --git a/lib/thread.rb b/lib/thread.rb
index 7e4f203..84e45be 100644
--- a/lib/thread.rb
+++ b/lib/thread.rb
@@ -200,27 +200,27 @@ def push(obj)
   #
   alias enq push
 
+  def pop_no_sync(non_block) # :nodoc:
+    if non_block
+      raise ThreadError, "queue empty" if @que.empty?
+    else
+      while @que.empty?
+        @waiting[Thread.current] = true
+        @mutex.sleep
+      end
+    end
+    @que.shift
+  ensure
+    @waiting.delete(Thread.current)
+  end
+  private :pop_no_sync
   #
   # Retrieves data from the queue.  If the queue is empty, the calling thread is
   # suspended until data is pushed onto the queue.  If +non_block+ is true, the
   # thread isn't suspended, and an exception is raised.
   #
   def pop(non_block=false)
-    @mutex.synchronize{
-      begin
-        while true
-          if @que.empty?
-            raise ThreadError, "queue empty" if non_block
-            @waiting[Thread.current] = true
-            @mutex.sleep
-          else
-            return @que.shift
-          end
-        end
-      ensure
-        @waiting.delete(Thread.current)
-      end
-    }
+    @mutex.synchronize{ pop_no_sync(non_block) }
   end
 
   #
@@ -293,30 +293,26 @@ def max
     @max
   end
 
+  def wakeup_queue_waiter # :nodoc:
+    t, _ = @queue_wait.shift
+    t.wakeup if t
+  rescue ThreadError
+    retry
+  end
+  private :wakeup_queue_waiter
+
   #
   # Sets the maximum size of the queue.
   #
   def max=(max)
     raise ArgumentError, "queue size must be positive" unless max > 0
-    diff = nil
     @mutex.synchronize {
-      if max <= @max
-        @max = max
-      else
-        diff = max - @max
-        @max = max
+      diff = max - @max
+      @max = max
+      if diff > 0
+        diff.times { wakeup_queue_waiter }
       end
     }
-    if diff
-      diff.times do
-        begin
-          t, _ = @queue_wait.shift
-          t.run if t
-        rescue ThreadError
-          retry
-        end
-      end
-    end
     max
   end
 
@@ -327,8 +323,7 @@ def max=(max)
   def push(obj)
     @mutex.synchronize{
       begin
-        while true
-          break if @que.length < @max
+        while @que.length >= @max
           @queue_wait[Thread.current] = true
           @mutex.sleep
         end
@@ -352,19 +347,12 @@ def push(obj)
   #
   # Retrieves data from the queue and runs a waiting thread, if any.
   #
-  def pop(*args)
-    retval = super
+  def pop(non_block=false)
     @mutex.synchronize {
-      if @que.length < @max
-        begin
-          t, _ = @queue_wait.shift
-          t.wakeup if t
-        rescue ThreadError
-          retry
-        end
-      end
+      retval = pop_no_sync(non_block)
+      wakeup_queue_waiter if @que.length < @max
+      retval
     }
-    retval
   end
 
   #
-- 
1.7.10

