* common.mk: use RUNRUBY instead of MINIRUBY because MINIRUBY can't

require extension libraries. The patch is from nobu
  (Nobuyoshi Nakada).

* ext/thread/extconf.rb: for build ext/thread/thread.c.

* include/ruby/intern.h: ditto.

* thread.c: ditto.

* lib/thread.rb: removed and replaced by ext/thread/thread.c.

* ext/thread/thread.c: Queue, SizedQueue and ConditionVariable
  implementations in C. This patch is based on patches from panaggio
  (Ricardo Panaggio) and funny_falcon (Yura Sokolov) and  ko1
  (Koichi Sasada). [ruby-core:31513] [Feature #3620]

* test/thread/test_queue.rb (test_queue_thread_raise): add a test for
  ensuring that killed thread should be removed from waiting threads.
  It is based on a code by ko1 (Koichi Sasada). [ruby-core:45950]

git-svn-id: svn+ssh://ci.ruby-lang.org/ruby/trunk@42862 b2dd03c8-39d4-4d8f-98ff-823fe69b080e

ChangeLog

@@ -1,3 +1,26 @@
+Fri Sep  6 23:53:31 2013  Masaki Matsushita  <glass.saga@gmail.com>
+
+	* common.mk: use RUNRUBY instead of MINIRUBY because MINIRUBY can't
+	  require extension libraries. The patch is from nobu
+	  (Nobuyoshi Nakada).
+
+	* ext/thread/extconf.rb: for build ext/thread/thread.c.
+
+	* include/ruby/intern.h: ditto.
+
+	* thread.c: ditto.
+
+	* lib/thread.rb: removed and replaced by ext/thread/thread.c.
+
+	* ext/thread/thread.c: Queue, SizedQueue and ConditionVariable
+	  implementations in C. This patch is based on patches from panaggio
+	  (Ricardo Panaggio) and funny_falcon (Yura Sokolov) and  ko1
+	  (Koichi Sasada). [ruby-core:31513] [Feature #3620]
+
+	* test/thread/test_queue.rb (test_queue_thread_raise): add a test for
+	  ensuring that killed thread should be removed from waiting threads.
+	  It is based on a code by ko1 (Koichi Sasada). [ruby-core:45950]
+
 Fri Sep  6 22:47:12 2013  Tanaka Akira  <akr@fsij.org>
 
 	* configure.in: Define ac_cv_func_clock_getres to yes for mingw*.

common.mk

@@ -123,7 +123,7 @@ SCRIPT_ARGS   =	--dest-dir="$(DESTDIR)" \
 		--make-flags="$(MAKEFLAGS)"
 EXTMK_ARGS    =	$(SCRIPT_ARGS) --extension $(EXTS) --extstatic $(EXTSTATIC) \
 		--make-flags="V=$(V) MINIRUBY='$(MINIRUBY)'" --
-INSTRUBY      =	$(SUDO) $(MINIRUBY) $(srcdir)/tool/rbinstall.rb
+INSTRUBY      =	$(SUDO) $(RUNRUBY) -r./$(arch)-fake $(srcdir)/tool/rbinstall.rb
 INSTRUBY_ARGS =	$(SCRIPT_ARGS) \
 		--data-mode=$(INSTALL_DATA_MODE) \
 		--prog-mode=$(INSTALL_PROG_MODE) \
@@ -449,7 +449,7 @@ post-no-install-doc::
 
 CLEAR_INSTALLED_LIST = clear-installed-list
 
-install-prereq: $(CLEAR_INSTALLED_LIST) PHONY
+install-prereq: $(CLEAR_INSTALLED_LIST) yes-fake PHONY
 
 clear-installed-list: PHONY
 	@> $(INSTALLED_LIST) set MAKE="$(MAKE)"

ext/thread/extconf.rb

@@ -0,0 +1,3 @@
+require 'mkmf'
+
+create_makefile('thread')

ext/thread/thread.c

@@ -0,0 +1,566 @@
+#include <ruby.h>
+
+enum {
+    CONDVAR_WAITERS = 0
+};
+
+enum {
+    QUEUE_QUE       = 0,
+    QUEUE_WAITERS   = 1,
+    SZQUEUE_WAITERS = 2,
+    SZQUEUE_MAX     = 3
+};
+
+#define GET_CONDVAR_WAITERS(cv) RSTRUCT_GET((cv), CONDVAR_WAITERS)
+
+#define GET_QUEUE_QUE(q)        RSTRUCT_GET((q), QUEUE_QUE)
+#define GET_QUEUE_WAITERS(q)    RSTRUCT_GET((q), QUEUE_WAITERS)
+#define GET_SZQUEUE_WAITERS(q)  RSTRUCT_GET((q), SZQUEUE_WAITERS)
+#define GET_SZQUEUE_MAX(q)      RSTRUCT_GET((q), SZQUEUE_MAX)
+#define GET_SZQUEUE_ULONGMAX(q) NUM2ULONG(GET_SZQUEUE_MAX(q))
+
+static VALUE
+ary_buf_new(void)
+{
+    return rb_ary_tmp_new(1);
+}
+
+static void
+wakeup_first_thread(VALUE list)
+{
+    VALUE thread;
+
+    while (!NIL_P(thread = rb_ary_shift(list))) {
+	if (RTEST(rb_thread_wakeup_alive(thread))) break;
+    }
+}
+
+static void
+wakeup_all_threads(VALUE list)
+{
+    VALUE thread;
+    long i;
+
+    for (i=0; i<RARRAY_LEN(list); i++) {
+	thread = RARRAY_AREF(list, i);
+	rb_thread_wakeup_alive(thread);
+    }
+    rb_ary_clear(list);
+}
+
+/*
+ *  Document-class: ConditionVariable
+ *
+ *  ConditionVariable objects augment class Mutex. Using condition variables,
+ *  it is possible to suspend while in the middle of a critical section until a
+ *  resource becomes available.
+ *
+ *  Example:
+ *
+ *    require 'thread'
+ *
+ *    mutex = Mutex.new
+ *    resource = ConditionVariable.new
+ *
+ *    a = Thread.new {
+ *	 mutex.synchronize {
+ *	   # Thread 'a' now needs the resource
+ *	   resource.wait(mutex)
+ *	   # 'a' can now have the resource
+ *	 }
+ *    }
+ *
+ *    b = Thread.new {
+ *	 mutex.synchronize {
+ *	   # Thread 'b' has finished using the resource
+ *	   resource.signal
+ *	 }
+ *    }
+ */
+
+/*
+ * Document-method: new
+ * call-seq: new
+ *
+ * Creates a new condvar.
+ */
+
+static VALUE
+rb_condvar_initialize(VALUE self)
+{
+    RSTRUCT_SET(self, CONDVAR_WAITERS, ary_buf_new());
+    return self;
+}
+
+struct sleep_call {
+    VALUE mutex;
+    VALUE timeout;
+};
+
+static ID id_sleep;
+
+static VALUE
+do_sleep(VALUE args)
+{
+    struct sleep_call *p = (struct sleep_call *)args;
+    return rb_funcall2(p->mutex, id_sleep, 1, &p->timeout);
+}
+
+static VALUE
+delete_current_thread(VALUE ary)
+{
+    return rb_ary_delete(ary, rb_thread_current());
+}
+
+/*
+ * Document-method: wait
+ * call-seq: wait(mutex, timeout=nil)
+ *
+ * Releases the lock held in +mutex+ and waits; reacquires the lock on wakeup.
+ *
+ * If +timeout+ is given, this method returns after +timeout+ seconds passed,
+ * even if no other thread doesn't signal.
+ */
+
+static VALUE
+rb_condvar_wait(int argc, VALUE *argv, VALUE self)
+{
+    VALUE waiters = GET_CONDVAR_WAITERS(self);
+    VALUE mutex, timeout;
+    struct sleep_call args;
+
+    rb_scan_args(argc, argv, "11", &mutex, &timeout);
+
+    args.mutex   = mutex;
+    args.timeout = timeout;
+    rb_ary_push(waiters, rb_thread_current());
+    rb_ensure(do_sleep, (VALUE)&args, delete_current_thread, waiters);
+
+    return self;
+}
+
+/*
+ * Document-method: signal
+ * call-seq: signal
+ *
+ * Wakes up the first thread in line waiting for this lock.
+ */
+
+static VALUE
+rb_condvar_signal(VALUE self)
+{
+    wakeup_first_thread(GET_CONDVAR_WAITERS(self));
+    return self;
+}
+
+/*
+ * Document-method: broadcast
+ * call-seq: broadcast
+ *
+ * Wakes up all threads waiting for this lock.
+ */
+
+static VALUE
+rb_condvar_broadcast(VALUE self)
+{
+    wakeup_all_threads(GET_CONDVAR_WAITERS(self));
+    return self;
+}
+
+/*
+ *  Document-class: Queue
+ *
+ *  This class provides a way to synchronize communication between threads.
+ *
+ *  Example:
+ *
+ *    require 'thread'
+ *    queue = Queue.new
+ *
+ *  producer = Thread.new do
+ *    5.times do |i|
+ *	 sleep rand(i) # simulate expense
+ *	 queue << i
+ *	 puts "#{i} produced"
+ *    end
+ *  end
+ *
+ *  consumer = Thread.new do
+ *    5.times do |i|
+ *	 value = queue.pop
+ *	 sleep rand(i/2) # simulate expense
+ *	 puts "consumed #{value}"
+ *    end
+ *  end
+ *
+ */
+
+/*
+ * Document-method: new
+ * call-seq: new
+ *
+ * Creates a new queue.
+ */
+
+static VALUE
+rb_queue_initialize(VALUE self)
+{
+    RSTRUCT_SET(self, QUEUE_QUE, ary_buf_new());
+    RSTRUCT_SET(self, QUEUE_WAITERS, ary_buf_new());
+    return self;
+}
+
+static VALUE
+queue_do_push(VALUE self, VALUE obj)
+{
+    rb_ary_push(GET_QUEUE_QUE(self), obj);
+    wakeup_first_thread(GET_QUEUE_WAITERS(self));
+    return self;
+}
+
+/*
+ * Document-method: push
+ * call-seq: push(obj)
+ *
+ * Pushes +obj+ to the queue.
+ */
+
+static VALUE
+rb_queue_push(VALUE self, VALUE obj)
+{
+    return queue_do_push(self, obj);
+}
+
+static unsigned long
+queue_length(VALUE self)
+{
+    return RARRAY_LEN(GET_QUEUE_QUE(self));
+}
+
+static unsigned long
+queue_num_waiting(VALUE self)
+{
+    return RARRAY_LEN(GET_QUEUE_WAITERS(self));
+}
+
+struct waiting_delete {
+    VALUE waiting;
+    VALUE th;
+};
+
+static VALUE
+queue_delete_from_waiting(struct waiting_delete *p)
+{
+    rb_ary_delete(p->waiting, p->th);
+    return Qnil;
+}
+
+static VALUE
+queue_do_pop(VALUE self, VALUE should_block)
+{
+    struct waiting_delete args;
+    args.waiting = GET_QUEUE_WAITERS(self);
+    args.th	 = rb_thread_current();
+
+    while (queue_length(self) == 0) {
+	if (!(int)should_block) {
+	    rb_raise(rb_eThreadError, "queue empty");
+	}
+	rb_ary_push(args.waiting, args.th);
+	rb_ensure((VALUE (*)())rb_thread_sleep_deadly, (VALUE)0, queue_delete_from_waiting, (VALUE)&args);
+    }
+
+    return rb_ary_shift(GET_QUEUE_QUE(self));
+}
+
+static VALUE
+queue_pop_should_block(int argc, VALUE *argv)
+{
+    VALUE should_block = Qtrue;
+    switch (argc) {
+      case 0:
+	break;
+      case 1:
+	should_block = RTEST(argv[0]) ? Qtrue : Qfalse;
+	break;
+      default:
+	rb_raise(rb_eArgError, "wrong number of arguments (%d for 1)", argc);
+    }
+    return should_block;
+}
+
+/*
+ * Document-method: pop
+ * call_seq: pop(non_block=false)
+ *
+ * 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.
+ */
+
+static VALUE
+rb_queue_pop(int argc, VALUE *argv, VALUE self)
+{
+    VALUE should_block = queue_pop_should_block(argc, argv);
+    return queue_do_pop(self, should_block);
+}
+
+/*
+ * Document-method: empty?
+ * call-seq: empty?
+ *
+ * Returns +true+ if the queue is empty.
+ */
+
+static VALUE
+rb_queue_empty_p(VALUE self)
+{
+    return queue_length(self) == 0 ? Qtrue : Qfalse;
+}
+
+/*
+ * Document-method: clear
+ * call-seq: clear
+ *
+ * Removes all objects from the queue.
+ */
+
+static VALUE
+rb_queue_clear(VALUE self)
+{
+    rb_ary_clear(GET_QUEUE_QUE(self));
+    return self;
+}
+
+/*
+ * Document-method: length
+ * call-seq: length
+ *
+ * Returns the length of the queue.
+ */
+
+static VALUE
+rb_queue_length(VALUE self)
+{
+    unsigned long len = queue_length(self);
+    return ULONG2NUM(len);
+}
+
+/*
+ * Document-method: num_waiting
+ * call-seq: num_waiting
+ *
+ * Returns the number of threads waiting on the queue.
+ */
+
+static VALUE
+rb_queue_num_waiting(VALUE self)
+{
+    unsigned long len = queue_num_waiting(self);
+    return ULONG2NUM(len);
+}
+
+/*
+ *  Document-class: SizedQueue
+ *
+ * This class represents queues of specified size capacity.  The push operation
+ * may be blocked if the capacity is full.
+ *
+ * See Queue for an example of how a SizedQueue works.
+ */
+
+/*
+ * Document-method: new
+ * call-seq: new(max)
+ *
+ * Creates a fixed-length queue with a maximum size of +max+.
+ */
+
+static VALUE
+rb_szqueue_initialize(VALUE self, VALUE vmax)
+{
+    long max;
+
+    max = NUM2LONG(vmax);
+    if (max <= 0) {
+	rb_raise(rb_eArgError, "queue size must be positive");
+    }
+
+    RSTRUCT_SET(self, QUEUE_QUE, ary_buf_new());
+    RSTRUCT_SET(self, QUEUE_WAITERS, ary_buf_new());
+    RSTRUCT_SET(self, SZQUEUE_WAITERS, ary_buf_new());
+    RSTRUCT_SET(self, SZQUEUE_MAX, vmax);
+
+    return self;
+}
+
+/*
+ * Document-method: max
+ * call-seq: max
+ *
+ * Returns the maximum size of the queue.
+ */
+
+static VALUE
+rb_szqueue_max_get(VALUE self)
+{
+    return GET_SZQUEUE_MAX(self);
+}
+
+/*
+ * Document-method: max=
+ * call-seq: max=(n)
+ *
+ * Sets the maximum size of the queue.
+ */
+
+static VALUE
+rb_szqueue_max_set(VALUE self, VALUE vmax)
+{
+    long max = NUM2LONG(vmax), diff = 0;
+    VALUE t;
+
+    if (max <= 0) {
+	rb_raise(rb_eArgError, "queue size must be positive");
+    }
+    if ((unsigned long)max > GET_SZQUEUE_ULONGMAX(self)) {
+	diff = max - GET_SZQUEUE_ULONGMAX(self);
+    }
+    RSTRUCT_SET(self, SZQUEUE_MAX, vmax);
+    while (diff > 0 && !NIL_P(t = rb_ary_shift(GET_QUEUE_QUE(self)))) {
+	rb_thread_wakeup_alive(t);
+    }
+    return vmax;
+}
+
+/*
+ * Document-method: push
+ * call-seq: push(obj)
+ *
+ * Pushes +obj+ to the queue.  If there is no space left in the queue, waits
+ * until space becomes available.
+ */
+
+static VALUE
+rb_szqueue_push(VALUE self, VALUE obj)
+{
+    struct waiting_delete args;
+    args.waiting = GET_QUEUE_WAITERS(self);
+    args.th      = rb_thread_current();
+
+    while (queue_length(self) >= GET_SZQUEUE_ULONGMAX(self)) {
+	rb_ary_push(args.waiting, args.th);
+	rb_ensure((VALUE (*)())rb_thread_sleep_deadly, (VALUE)0, queue_delete_from_waiting, (VALUE)&args);
+    }
+    return queue_do_push(self, obj);
+}
+
+static VALUE
+szqueue_do_pop(VALUE self, VALUE should_block)
+{
+    VALUE retval = queue_do_pop(self, should_block);
+
+    if (queue_length(self) < GET_SZQUEUE_ULONGMAX(self)) {
+	wakeup_first_thread(GET_SZQUEUE_WAITERS(self));
+    }
+
+    return retval;
+}
+
+/*
+ * Document-method: pop
+ * call_seq: pop(non_block=false)
+ *
+ * Returns the number of threads waiting on the queue.
+ */
+
+static VALUE
+rb_szqueue_pop(int argc, VALUE *argv, VALUE self)
+{
+    VALUE should_block = queue_pop_should_block(argc, argv);
+    return szqueue_do_pop(self, should_block);
+}
+
+/*
+ * Document-method: pop
+ * call_seq: pop(non_block=false)
+ *
+ * Returns the number of threads waiting on the queue.
+ */
+
+static VALUE
+rb_szqueue_num_waiting(VALUE self)
+{
+    long len = queue_num_waiting(self);
+    len += RARRAY_LEN(GET_SZQUEUE_WAITERS(self));
+    return ULONG2NUM(len);
+}
+
+#ifndef UNDER_THREAD
+#define UNDER_THREAD 1
+#endif
+
+void
+Init_thread(void)
+{
+#if UNDER_THREAD
+#define ALIAS_GLOBCAL_CONST(name) do {	              \
+	ID id = rb_intern_const(#name);	              \
+	if (!rb_const_defined_at(rb_cObject, id)) {   \
+	    rb_const_set(rb_cObject, id, rb_c##name); \
+	}                                             \
+    } while (0)
+#else
+#define ALIAS_GLOBCAL_CONST(name) do { /* nothing */ } while (0)
+#endif
+
+    VALUE rb_cConditionVariable = rb_struct_define_without_accessor_under(
+	UNDER_THREAD ? rb_cThread : 0,
+	"ConditionVariable", rb_cObject, rb_struct_alloc_noinit,
+	"waiters", NULL);
+    VALUE rb_cQueue = rb_struct_define_without_accessor_under(
+	UNDER_THREAD ? rb_cThread : 0,
+	"Queue", rb_cObject, rb_struct_alloc_noinit,
+	"que", "waiters", NULL);
+    VALUE rb_cSizedQueue = rb_struct_define_without_accessor_under(
+	UNDER_THREAD ? rb_cThread : 0,
+	"SizedQueue", rb_cQueue, rb_struct_alloc_noinit,
+	"que", "waiters", "queue_waiters", "size", NULL);
+
+    id_sleep = rb_intern("sleep");
+
+    rb_define_method(rb_cConditionVariable, "initialize", rb_condvar_initialize, 0);
+    rb_define_method(rb_cConditionVariable, "wait", rb_condvar_wait, -1);
+    rb_define_method(rb_cConditionVariable, "signal", rb_condvar_signal, 0);
+    rb_define_method(rb_cConditionVariable, "broadcast", rb_condvar_broadcast, 0);
+
+    rb_define_method(rb_cQueue, "initialize", rb_queue_initialize, 0);
+    rb_define_method(rb_cQueue, "push", rb_queue_push, 1);
+    rb_define_method(rb_cQueue, "pop", rb_queue_pop, -1);
+    rb_define_method(rb_cQueue, "empty?", rb_queue_empty_p, 0);
+    rb_define_method(rb_cQueue, "clear", rb_queue_clear, 0);
+    rb_define_method(rb_cQueue, "length", rb_queue_length, 0);
+    rb_define_method(rb_cQueue, "num_waiting", rb_queue_num_waiting, 0);
+
+    rb_alias(rb_cQueue, rb_intern("enq"), rb_intern("push"));
+    rb_alias(rb_cQueue, rb_intern("<<"), rb_intern("push"));
+    rb_alias(rb_cQueue, rb_intern("deq"), rb_intern("pop"));
+    rb_alias(rb_cQueue, rb_intern("shift"), rb_intern("pop"));
+    rb_alias(rb_cQueue, rb_intern("size"), rb_intern("length"));
+
+    rb_define_method(rb_cSizedQueue, "initialize", rb_szqueue_initialize, 1);
+    rb_define_method(rb_cSizedQueue, "max", rb_szqueue_max_get, 0);
+    rb_define_method(rb_cSizedQueue, "max=", rb_szqueue_max_set, 1);
+    rb_define_method(rb_cSizedQueue, "push", rb_szqueue_push, 1);
+    rb_define_method(rb_cSizedQueue, "pop", rb_szqueue_pop, -1);
+    rb_define_method(rb_cSizedQueue, "num_waiting", rb_szqueue_num_waiting, 0);
+    rb_alias(rb_cSizedQueue, rb_intern("enq"), rb_intern("push"));
+    rb_alias(rb_cSizedQueue, rb_intern("<<"), rb_intern("push"));
+    rb_alias(rb_cSizedQueue, rb_intern("deq"), rb_intern("pop"));
+    rb_alias(rb_cSizedQueue, rb_intern("shift"), rb_intern("pop"));
+
+    rb_provide("thread.rb");
+    ALIAS_GLOBCAL_CONST(ConditionVariable);
+    ALIAS_GLOBCAL_CONST(Queue);
+    ALIAS_GLOBCAL_CONST(SizedQueue);
+}

include/ruby/intern.h

@@ -428,6 +428,7 @@ int rb_thread_alone(void);
 DEPRECATED(void rb_thread_polling(void));
 void rb_thread_sleep(int);
 void rb_thread_sleep_forever(void);
+void rb_thread_sleep_deadly(void);
 VALUE rb_thread_stop(void);
 VALUE rb_thread_wakeup(VALUE);
 VALUE rb_thread_wakeup_alive(VALUE);

lib/thread.rb

@@ -1,366 +0,0 @@
-#
-#               thread.rb - thread support classes
-#                       by Yukihiro Matsumoto <matz@netlab.co.jp>
-#
-# Copyright (C) 2001  Yukihiro Matsumoto
-# Copyright (C) 2000  Network Applied Communication Laboratory, Inc.
-# Copyright (C) 2000  Information-technology Promotion Agency, Japan
-#
-
-unless defined? Thread
-  raise "Thread not available for this ruby interpreter"
-end
-
-unless defined? ThreadError
-  class ThreadError < StandardError
-  end
-end
-
-if $DEBUG
-  Thread.abort_on_exception = true
-end
-
-#
-# ConditionVariable objects augment class Mutex. Using condition variables,
-# it is possible to suspend while in the middle of a critical section until a
-# resource becomes available.
-#
-# Example:
-#
-#   require 'thread'
-#
-#   mutex = Mutex.new
-#   resource = ConditionVariable.new
-#
-#   a = Thread.new {
-#     mutex.synchronize {
-#       # Thread 'a' now needs the resource
-#       resource.wait(mutex)
-#       # 'a' can now have the resource
-#     }
-#   }
-#
-#   b = Thread.new {
-#     mutex.synchronize {
-#       # Thread 'b' has finished using the resource
-#       resource.signal
-#     }
-#   }
-#
-class ConditionVariable
-  #
-  # Creates a new ConditionVariable
-  #
-  def initialize
-    @waiters = {}
-    @waiters_mutex = Mutex.new
-  end
-
-  #
-  # Releases the lock held in +mutex+ and waits; reacquires the lock on wakeup.
-  #
-  # If +timeout+ is given, this method returns after +timeout+ seconds passed,
-  # even if no other thread has signaled.
-  #
-  def wait(mutex, timeout=nil)
-    Thread.handle_interrupt(StandardError => :never) do
-      begin
-        Thread.handle_interrupt(StandardError => :on_blocking) do
-          @waiters_mutex.synchronize do
-            @waiters[Thread.current] = true
-          end
-          mutex.sleep timeout
-        end
-      ensure
-        @waiters_mutex.synchronize do
-          @waiters.delete(Thread.current)
-        end
-      end
-    end
-    self
-  end
-
-  #
-  # Wakes up the first thread in line waiting for this lock.
-  #
-  def signal
-    Thread.handle_interrupt(StandardError => :on_blocking) do
-      begin
-        t, _ = @waiters_mutex.synchronize { @waiters.shift }
-        t.run if t
-      rescue ThreadError
-        retry # t was already dead?
-      end
-    end
-    self
-  end
-
-  #
-  # Wakes up all threads waiting for this lock.
-  #
-  def broadcast
-    Thread.handle_interrupt(StandardError => :on_blocking) do
-      threads = nil
-      @waiters_mutex.synchronize do
-        threads = @waiters.keys
-        @waiters.clear
-      end
-      for t in threads
-        begin
-          t.run
-        rescue ThreadError
-        end
-      end
-    end
-    self
-  end
-end
-
-#
-# This class provides a way to synchronize communication between threads.
-#
-# Example:
-#
-#   require 'thread'
-#
-#   queue = Queue.new
-#
-#   producer = Thread.new do
-#     5.times do |i|
-#       sleep rand(i) # simulate expense
-#       queue << i
-#       puts "#{i} produced"
-#     end
-#   end
-#
-#   consumer = Thread.new do
-#     5.times do |i|
-#       value = queue.pop
-#       sleep rand(i/2) # simulate expense
-#       puts "consumed #{value}"
-#     end
-#   end
-#
-#   consumer.join
-#
-class Queue
-  #
-  # Creates a new queue.
-  #
-  def initialize
-    @que = []
-    @que.taint          # enable tainted communication
-    @num_waiting = 0
-    self.taint
-    @mutex = Mutex.new
-    @cond = ConditionVariable.new
-  end
-
-  #
-  # Pushes +obj+ to the queue.
-  #
-  def push(obj)
-    Thread.handle_interrupt(StandardError => :on_blocking) do
-      @mutex.synchronize do
-        @que.push obj
-        @cond.signal
-      end
-      self
-    end
-  end
-
-  #
-  # Alias of push
-  #
-  alias << push
-
-  #
-  # Alias of push
-  #
-  alias enq push
-
-  #
-  # 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)
-    Thread.handle_interrupt(StandardError => :on_blocking) do
-      @mutex.synchronize do
-        while true
-          if @que.empty?
-            if non_block
-              raise ThreadError, "queue empty"
-            else
-              begin
-                @num_waiting += 1
-                @cond.wait @mutex
-              ensure
-                @num_waiting -= 1
-              end
-            end
-          else
-            return @que.shift
-          end
-        end
-      end
-    end
-  end
-
-  #
-  # Alias of pop
-  #
-  alias shift pop
-
-  #
-  # Alias of pop
-  #
-  alias deq pop
-
-  #
-  # Returns +true+ if the queue is empty.
-  #
-  def empty?
-    @que.empty?
-  end
-
-  #
-  # Removes all objects from the queue.
-  #
-  def clear
-    @que.clear
-    self
-  end
-
-  #
-  # Returns the length of the queue.
-  #
-  def length
-    @que.length
-  end
-
-  #
-  # Alias of length.
-  #
-  alias size length
-
-  #
-  # Returns the number of threads waiting on the queue.
-  #
-  def num_waiting
-    @num_waiting
-  end
-end
-
-#
-# This class represents queues of specified size capacity.  The push operation
-# may be blocked if the capacity is full.
-#
-# See Queue for an example of how a SizedQueue works.
-#
-class SizedQueue < Queue
-  #
-  # Creates a fixed-length queue with a maximum size of +max+.
-  #
-  def initialize(max)
-    raise ArgumentError, "queue size must be positive" unless max > 0
-    @max = max
-    @enque_cond = ConditionVariable.new
-    @num_enqueue_waiting = 0
-    super()
-  end
-
-  #
-  # Returns the maximum size of the queue.
-  #
-  def max
-    @max
-  end
-
-  #
-  # Sets the maximum size of the queue.
-  #
-  def max=(max)
-    raise ArgumentError, "queue size must be positive" unless max > 0
-
-    @mutex.synchronize do
-      if max <= @max
-        @max = max
-      else
-        diff = max - @max
-        @max = max
-        diff.times do
-          @enque_cond.signal
-        end
-      end
-    end
-    max
-  end
-
-  #
-  # Pushes +obj+ to the queue.  If there is no space left in the queue, waits
-  # until space becomes available.
-  #
-  def push(obj)
-    Thread.handle_interrupt(RuntimeError => :on_blocking) do
-      @mutex.synchronize do
-        while true
-          break if @que.length < @max
-          @num_enqueue_waiting += 1
-          begin
-            @enque_cond.wait @mutex
-          ensure
-            @num_enqueue_waiting -= 1
-          end
-        end
-
-        @que.push obj
-        @cond.signal
-      end
-      self
-    end
-  end
-
-  #
-  # Alias of push
-  #
-  alias << push
-
-  #
-  # Alias of push
-  #
-  alias enq push
-
-  #
-  # Retrieves data from the queue and runs a waiting thread, if any.
-  #
-  def pop(*args)
-    retval = super
-    @mutex.synchronize do
-      if @que.length < @max
-        @enque_cond.signal
-      end
-    end
-    retval
-  end
-
-  #
-  # Alias of pop
-  #
-  alias shift pop
-
-  #
-  # Alias of pop
-  #
-  alias deq pop
-
-  #
-  # Returns the number of threads waiting on the queue.
-  #
-  def num_waiting
-    @num_waiting + @num_enqueue_waiting
-  end
-end
-
-# Documentation comments:
-#  - How do you make RDoc inherit documentation from superclass?

test/thread/test_queue.rb

@@ -1,6 +1,7 @@
 require 'test/unit'
 require 'thread'
 require 'tmpdir'
+require 'timeout'
 require_relative '../ruby/envutil'
 
 class TestQueue < Test::Unit::TestCase
@@ -133,4 +134,25 @@ class TestQueue < Test::Unit::TestCase
     assert_same q, retval
   end
 
+  def test_queue_thread_raise
+    q = Queue.new
+    th1 = Thread.new do
+      begin
+        q.pop
+      rescue RuntimeError
+        sleep
+      end
+    end
+    th2 = Thread.new do
+      sleep 0.1
+      q.pop
+    end
+    sleep 0.1
+    th1.raise
+    sleep 0.1
+    q << :s
+    assert_nothing_raised(TimeoutError) do
+      timeout(1) { th2.join }
+    end
+  end
 end

thread.c

@@ -1058,7 +1058,7 @@ rb_thread_sleep_forever(void)
     sleep_forever(GET_THREAD(), 0, 1);
 }
 
-static void
+void
 rb_thread_sleep_deadly(void)
 {
     thread_debug("rb_thread_sleep_deadly\n");