diff options
Diffstat (limited to 'thread_sync.c')
| -rw-r--r-- | thread_sync.c | 777 |
1 files changed, 404 insertions, 373 deletions
diff --git a/thread_sync.c b/thread_sync.c index e9eabb45ca..ae69cb4a6e 100644 --- a/thread_sync.c +++ b/thread_sync.c @@ -1,6 +1,6 @@ /* included by thread.c */ #include "ccan/list/list.h" -#include "coroutine/Stack.h" +#include "builtin.h" static VALUE rb_cMutex, rb_cQueue, rb_cSizedQueue, rb_cConditionVariable; static VALUE rb_eClosedQueueError; @@ -9,7 +9,7 @@ static VALUE rb_eClosedQueueError; typedef struct rb_mutex_struct { rb_fiber_t *fiber; struct rb_mutex_struct *next_mutex; - struct list_head waitq; /* protected by GVL */ + struct ccan_list_head waitq; /* protected by GVL */ } rb_mutex_t; /* sync_waiter is always on-stack */ @@ -17,25 +17,43 @@ struct sync_waiter { VALUE self; rb_thread_t *th; rb_fiber_t *fiber; - struct list_node node; + struct ccan_list_node node; +}; + +static inline rb_fiber_t* +nonblocking_fiber(rb_fiber_t *fiber) +{ + if (rb_fiberptr_blocking(fiber)) { + return NULL; + } + + return fiber; +} + +struct queue_sleep_arg { + VALUE self; + VALUE timeout; + rb_hrtime_t end; }; #define MUTEX_ALLOW_TRAP FL_USER1 static void -sync_wakeup(struct list_head *head, long max) +sync_wakeup(struct ccan_list_head *head, long max) { + RUBY_DEBUG_LOG("max:%ld", max); + struct sync_waiter *cur = 0, *next; - list_for_each_safe(head, cur, next, node) { - list_del_init(&cur->node); + ccan_list_for_each_safe(head, cur, next, node) { + ccan_list_del_init(&cur->node); if (cur->th->status != THREAD_KILLED) { - - if (cur->th->scheduler != Qnil && rb_fiberptr_blocking(cur->fiber) == 0) { + if (cur->th->scheduler != Qnil && cur->fiber) { rb_fiber_scheduler_unblock(cur->th->scheduler, cur->self, rb_fiberptr_self(cur->fiber)); } else { + RUBY_DEBUG_LOG("target_th:%u", rb_th_serial(cur->th)); rb_threadptr_interrupt(cur->th); cur->th->status = THREAD_RUNNABLE; } @@ -46,13 +64,13 @@ sync_wakeup(struct list_head *head, long max) } static void -wakeup_one(struct list_head *head) +wakeup_one(struct ccan_list_head *head) { sync_wakeup(head, 1); } static void -wakeup_all(struct list_head *head) +wakeup_all(struct ccan_list_head *head) { sync_wakeup(head, LONG_MAX); } @@ -65,14 +83,14 @@ static void rb_mutex_abandon_locking_mutex(rb_thread_t *th); static const char* rb_mutex_unlock_th(rb_mutex_t *mutex, rb_thread_t *th, rb_fiber_t *fiber); /* - * Document-class: Mutex + * Document-class: Thread::Mutex * - * Mutex implements a simple semaphore that can be used to coordinate access to - * shared data from multiple concurrent threads. + * Thread::Mutex implements a simple semaphore that can be used to + * coordinate access to shared data from multiple concurrent threads. * * Example: * - * semaphore = Mutex.new + * semaphore = Thread::Mutex.new * * a = Thread.new { * semaphore.synchronize { @@ -96,8 +114,8 @@ rb_mutex_num_waiting(rb_mutex_t *mutex) struct sync_waiter *w = 0; size_t n = 0; - list_for_each(&mutex->waitq, w, node) { - n++; + ccan_list_for_each(&mutex->waitq, w, node) { + n++; } return n; @@ -110,9 +128,9 @@ mutex_free(void *ptr) { rb_mutex_t *mutex = ptr; if (mutex->fiber) { - /* rb_warn("free locked mutex"); */ - const char *err = rb_mutex_unlock_th(mutex, rb_fiber_threadptr(mutex->fiber), mutex->fiber); - if (err) rb_bug("%s", err); + /* rb_warn("free locked mutex"); */ + const char *err = rb_mutex_unlock_th(mutex, rb_fiber_threadptr(mutex->fiber), mutex->fiber); + if (err) rb_bug("%s", err); } ruby_xfree(ptr); } @@ -126,7 +144,7 @@ mutex_memsize(const void *ptr) static const rb_data_type_t mutex_data_type = { "mutex", {mutex_mark, mutex_free, mutex_memsize,}, - 0, 0, RUBY_TYPED_FREE_IMMEDIATELY + 0, 0, RUBY_TYPED_WB_PROTECTED | RUBY_TYPED_FREE_IMMEDIATELY }; static rb_mutex_t * @@ -142,12 +160,7 @@ mutex_ptr(VALUE obj) VALUE rb_obj_is_mutex(VALUE obj) { - if (rb_typeddata_is_kind_of(obj, &mutex_data_type)) { - return Qtrue; - } - else { - return Qfalse; - } + return RBOOL(rb_typeddata_is_kind_of(obj, &mutex_data_type)); } static VALUE @@ -158,13 +171,13 @@ mutex_alloc(VALUE klass) obj = TypedData_Make_Struct(klass, rb_mutex_t, &mutex_data_type, mutex); - list_head_init(&mutex->waitq); + ccan_list_head_init(&mutex->waitq); return obj; } /* * call-seq: - * Mutex.new -> mutex + * Thread::Mutex.new -> mutex * * Creates a new Mutex */ @@ -191,11 +204,12 @@ rb_mutex_locked_p(VALUE self) { rb_mutex_t *mutex = mutex_ptr(self); - return mutex->fiber ? Qtrue : Qfalse; + return RBOOL(mutex->fiber); } static void -thread_mutex_insert(rb_thread_t *thread, rb_mutex_t *mutex) { +thread_mutex_insert(rb_thread_t *thread, rb_mutex_t *mutex) +{ if (thread->keeping_mutexes) { mutex->next_mutex = thread->keeping_mutexes; } @@ -204,7 +218,8 @@ thread_mutex_insert(rb_thread_t *thread, rb_mutex_t *mutex) { } static void -thread_mutex_remove(rb_thread_t *thread, rb_mutex_t *mutex) { +thread_mutex_remove(rb_thread_t *thread, rb_mutex_t *mutex) +{ rb_mutex_t **keeping_mutexes = &thread->keeping_mutexes; while (*keeping_mutexes && *keeping_mutexes != mutex) { @@ -239,50 +254,44 @@ rb_mutex_trylock(VALUE self) rb_mutex_t *mutex = mutex_ptr(self); if (mutex->fiber == 0) { - rb_fiber_t *fiber = GET_EC()->fiber_ptr; - rb_thread_t *th = GET_THREAD(); - mutex->fiber = fiber; - - mutex_locked(th, self); - return Qtrue; - } - - return Qfalse; -} + RUBY_DEBUG_LOG("%p ok", mutex); -/* - * At maximum, only one thread can use cond_timedwait and watch deadlock - * periodically. Multiple polling thread (i.e. concurrent deadlock check) - * introduces new race conditions. [Bug #6278] [ruby-core:44275] - */ -static const rb_thread_t *patrol_thread = NULL; + rb_fiber_t *fiber = GET_EC()->fiber_ptr; + rb_thread_t *th = GET_THREAD(); + mutex->fiber = fiber; -static VALUE -mutex_owned_p(rb_fiber_t *fiber, rb_mutex_t *mutex) -{ - if (mutex->fiber == fiber) { + mutex_locked(th, self); return Qtrue; } else { + RUBY_DEBUG_LOG("%p ng", mutex); return Qfalse; } } -static VALUE call_rb_fiber_scheduler_block(VALUE mutex) { - return rb_fiber_scheduler_block(rb_fiber_scheduler_current(), mutex, Qnil); +static VALUE +mutex_owned_p(rb_fiber_t *fiber, rb_mutex_t *mutex) +{ + return RBOOL(mutex->fiber == fiber); } static VALUE -delete_from_waitq(VALUE v) +call_rb_fiber_scheduler_block(VALUE mutex) { - struct sync_waiter *w = (void *)v; - list_del(&w->node); + return rb_fiber_scheduler_block(rb_fiber_scheduler_current(), mutex, Qnil); +} - COROUTINE_STACK_FREE(w); +static VALUE +delete_from_waitq(VALUE value) +{ + struct sync_waiter *sync_waiter = (void *)value; + ccan_list_del(&sync_waiter->node); return Qnil; } +static inline rb_atomic_t threadptr_get_interrupts(rb_thread_t *th); + static VALUE do_mutex_lock(VALUE self, int interruptible_p) { @@ -290,11 +299,12 @@ do_mutex_lock(VALUE self, int interruptible_p) rb_thread_t *th = ec->thread_ptr; rb_fiber_t *fiber = ec->fiber_ptr; rb_mutex_t *mutex = mutex_ptr(self); + rb_atomic_t saved_ints = 0; /* When running trap handler */ if (!FL_TEST_RAW(self, MUTEX_ALLOW_TRAP) && - th->ec->interrupt_mask & TRAP_INTERRUPT_MASK) { - rb_raise(rb_eThreadError, "can't be called from trap context"); + th->ec->interrupt_mask & TRAP_INTERRUPT_MASK) { + rb_raise(rb_eThreadError, "can't be called from trap context"); } if (rb_mutex_trylock(self) == Qfalse) { @@ -303,68 +313,70 @@ do_mutex_lock(VALUE self, int interruptible_p) } while (mutex->fiber != fiber) { + VM_ASSERT(mutex->fiber != NULL); + VALUE scheduler = rb_fiber_scheduler_current(); if (scheduler != Qnil) { - COROUTINE_STACK_LOCAL(struct sync_waiter, w); - w->self = self; - w->th = th; - w->fiber = fiber; + struct sync_waiter sync_waiter = { + .self = self, + .th = th, + .fiber = nonblocking_fiber(fiber) + }; - list_add_tail(&mutex->waitq, &w->node); + ccan_list_add_tail(&mutex->waitq, &sync_waiter.node); - rb_ensure(call_rb_fiber_scheduler_block, self, delete_from_waitq, (VALUE)w); + rb_ensure(call_rb_fiber_scheduler_block, self, delete_from_waitq, (VALUE)&sync_waiter); if (!mutex->fiber) { mutex->fiber = fiber; } } else { - enum rb_thread_status prev_status = th->status; - rb_hrtime_t *timeout = 0; - rb_hrtime_t rel = rb_msec2hrtime(100); + if (!th->vm->thread_ignore_deadlock && rb_fiber_threadptr(mutex->fiber) == th) { + rb_raise(rb_eThreadError, "deadlock; lock already owned by another fiber belonging to the same thread"); + } + struct sync_waiter sync_waiter = { + .self = self, + .th = th, + .fiber = nonblocking_fiber(fiber), + }; + + RUBY_DEBUG_LOG("%p wait", mutex); + + // similar code with `sleep_forever`, but + // sleep_forever(SLEEP_DEADLOCKABLE) raises an exception. + // Ensure clause is needed like but `rb_ensure` a bit slow. + // + // begin + // sleep_forever(th, SLEEP_DEADLOCKABLE); + // ensure + // ccan_list_del(&sync_waiter.node); + // end + enum rb_thread_status prev_status = th->status; th->status = THREAD_STOPPED_FOREVER; - th->locking_mutex = self; rb_ractor_sleeper_threads_inc(th->ractor); - /* - * Carefully! while some contended threads are in native_sleep(), - * ractor->sleeper is unstable value. we have to avoid both deadlock - * and busy loop. - */ - if ((rb_ractor_living_thread_num(th->ractor) == rb_ractor_sleeper_thread_num(th->ractor)) && - !patrol_thread) { - timeout = &rel; - patrol_thread = th; - } - - COROUTINE_STACK_LOCAL(struct sync_waiter, w); - w->self = self; - w->th = th; - w->fiber = fiber; - - list_add_tail(&mutex->waitq, &w->node); + rb_check_deadlock(th->ractor); - native_sleep(th, timeout); /* release GVL */ - - list_del(&w->node); + th->locking_mutex = self; - COROUTINE_STACK_FREE(w); + ccan_list_add_tail(&mutex->waitq, &sync_waiter.node); + { + native_sleep(th, NULL); + } + ccan_list_del(&sync_waiter.node); + // unlocked by another thread while sleeping if (!mutex->fiber) { mutex->fiber = fiber; } - if (patrol_thread == th) - patrol_thread = NULL; - - th->locking_mutex = Qfalse; - if (mutex->fiber && timeout && !RUBY_VM_INTERRUPTED(th->ec)) { - rb_check_deadlock(th->ractor); - } - if (th->status == THREAD_STOPPED_FOREVER) { - th->status = prev_status; - } rb_ractor_sleeper_threads_dec(th->ractor); + th->status = prev_status; + th->locking_mutex = Qfalse; + th->locking_mutex = Qfalse; + + RUBY_DEBUG_LOG("%p wakeup", mutex); } if (interruptible_p) { @@ -376,11 +388,27 @@ do_mutex_lock(VALUE self, int interruptible_p) mutex->fiber = fiber; } } + else { + // clear interrupt information + if (RUBY_VM_INTERRUPTED(th->ec)) { + // reset interrupts + if (saved_ints == 0) { + saved_ints = threadptr_get_interrupts(th); + } + else { + // ignore additional interrupts + threadptr_get_interrupts(th); + } + } + } } + if (saved_ints) th->ec->interrupt_flag = saved_ints; if (mutex->fiber == fiber) mutex_locked(th, self); } + RUBY_DEBUG_LOG("%p locked", mutex); + // assertion if (mutex_owned_p(fiber, mutex) == Qfalse) rb_bug("do_mutex_lock: mutex is not owned."); @@ -424,46 +452,46 @@ rb_mutex_owned_p(VALUE self) static const char * rb_mutex_unlock_th(rb_mutex_t *mutex, rb_thread_t *th, rb_fiber_t *fiber) { - const char *err = NULL; + RUBY_DEBUG_LOG("%p", mutex); if (mutex->fiber == 0) { - err = "Attempt to unlock a mutex which is not locked"; + return "Attempt to unlock a mutex which is not locked"; } else if (mutex->fiber != fiber) { - err = "Attempt to unlock a mutex which is locked by another thread/fiber"; + return "Attempt to unlock a mutex which is locked by another thread/fiber"; } - else { - struct sync_waiter *cur = 0, *next; - mutex->fiber = 0; - list_for_each_safe(&mutex->waitq, cur, next, node) { - list_del_init(&cur->node); + struct sync_waiter *cur = 0, *next; - if (cur->th->scheduler != Qnil && rb_fiberptr_blocking(cur->fiber) == 0) { - rb_fiber_scheduler_unblock(cur->th->scheduler, cur->self, rb_fiberptr_self(cur->fiber)); - goto found; - } - else { - switch (cur->th->status) { - case THREAD_RUNNABLE: /* from someone else calling Thread#run */ - case THREAD_STOPPED_FOREVER: /* likely (rb_mutex_lock) */ - rb_threadptr_interrupt(cur->th); - goto found; - case THREAD_STOPPED: /* probably impossible */ - rb_bug("unexpected THREAD_STOPPED"); - case THREAD_KILLED: - /* not sure about this, possible in exit GC? */ - rb_bug("unexpected THREAD_KILLED"); - continue; - } + mutex->fiber = 0; + thread_mutex_remove(th, mutex); + + ccan_list_for_each_safe(&mutex->waitq, cur, next, node) { + ccan_list_del_init(&cur->node); + + if (cur->th->scheduler != Qnil && cur->fiber) { + rb_fiber_scheduler_unblock(cur->th->scheduler, cur->self, rb_fiberptr_self(cur->fiber)); + return NULL; + } + else { + switch (cur->th->status) { + case THREAD_RUNNABLE: /* from someone else calling Thread#run */ + case THREAD_STOPPED_FOREVER: /* likely (rb_mutex_lock) */ + RUBY_DEBUG_LOG("wakeup th:%u", rb_th_serial(cur->th)); + rb_threadptr_interrupt(cur->th); + return NULL; + case THREAD_STOPPED: /* probably impossible */ + rb_bug("unexpected THREAD_STOPPED"); + case THREAD_KILLED: + /* not sure about this, possible in exit GC? */ + rb_bug("unexpected THREAD_KILLED"); + continue; } } - - found: - thread_mutex_remove(th, mutex); } - return err; + // We did not find any threads to wake up, so we can just return with no error: + return NULL; } /* @@ -500,7 +528,7 @@ rb_mutex_abandon_locking_mutex(rb_thread_t *th) if (th->locking_mutex) { rb_mutex_t *mutex = mutex_ptr(th->locking_mutex); - list_head_init(&mutex->waitq); + ccan_list_head_init(&mutex->waitq); th->locking_mutex = Qfalse; } } @@ -511,11 +539,11 @@ rb_mutex_abandon_all(rb_mutex_t *mutexes) rb_mutex_t *mutex; while (mutexes) { - mutex = mutexes; - mutexes = mutex->next_mutex; - mutex->fiber = 0; - mutex->next_mutex = 0; - list_head_init(&mutex->waitq); + mutex = mutexes; + mutexes = mutex->next_mutex; + mutex->fiber = 0; + mutex->next_mutex = 0; + ccan_list_head_init(&mutex->waitq); } } #endif @@ -523,7 +551,7 @@ rb_mutex_abandon_all(rb_mutex_t *mutexes) static VALUE rb_mutex_sleep_forever(VALUE self) { - rb_thread_sleep_deadly_allow_spurious_wakeup(self); + rb_thread_sleep_deadly_allow_spurious_wakeup(self, Qnil, 0); return Qnil; } @@ -532,14 +560,14 @@ rb_mutex_wait_for(VALUE time) { rb_hrtime_t *rel = (rb_hrtime_t *)time; /* permit spurious check */ - sleep_hrtime(GET_THREAD(), *rel, 0); - return Qnil; + return RBOOL(sleep_hrtime(GET_THREAD(), *rel, 0)); } VALUE rb_mutex_sleep(VALUE self, VALUE timeout) { struct timeval t; + VALUE woken = Qtrue; if (!NIL_P(timeout)) { t = rb_time_interval(timeout); @@ -559,18 +587,19 @@ rb_mutex_sleep(VALUE self, VALUE timeout) } else { rb_hrtime_t rel = rb_timeval2hrtime(&t); - rb_ensure(rb_mutex_wait_for, (VALUE)&rel, mutex_lock_uninterruptible, self); + woken = rb_ensure(rb_mutex_wait_for, (VALUE)&rel, mutex_lock_uninterruptible, self); } } RUBY_VM_CHECK_INTS_BLOCKING(GET_EC()); + if (!woken) return Qnil; time_t end = time(0) - beg; return TIMET2NUM(end); } /* * call-seq: - * mutex.sleep(timeout = nil) -> number + * mutex.sleep(timeout = nil) -> number or nil * * Releases the lock and sleeps +timeout+ seconds if it is given and * non-nil or forever. Raises +ThreadError+ if +mutex+ wasn't locked by @@ -581,6 +610,8 @@ rb_mutex_sleep(VALUE self, VALUE timeout) * * Note that this method can wakeup without explicit Thread#wakeup call. * For example, receiving signal and so on. + * + * Returns the slept time in seconds if woken up, or +nil+ if timed out. */ static VALUE mutex_sleep(int argc, VALUE *argv, VALUE self) @@ -596,7 +627,7 @@ mutex_sleep(int argc, VALUE *argv, VALUE self) * mutex.synchronize { ... } -> result of the block * * Obtains a lock, runs the block, and releases the lock when the block - * completes. See the example under +Mutex+. + * completes. See the example under Thread::Mutex. */ VALUE @@ -611,46 +642,51 @@ rb_mutex_synchronize(VALUE mutex, VALUE (*func)(VALUE arg), VALUE arg) * mutex.synchronize { ... } -> result of the block * * Obtains a lock, runs the block, and releases the lock when the block - * completes. See the example under +Mutex+. + * completes. See the example under Thread::Mutex. */ static VALUE rb_mutex_synchronize_m(VALUE self) { if (!rb_block_given_p()) { - rb_raise(rb_eThreadError, "must be called with a block"); + rb_raise(rb_eThreadError, "must be called with a block"); } return rb_mutex_synchronize(self, rb_yield, Qundef); } -void rb_mutex_allow_trap(VALUE self, int val) +void +rb_mutex_allow_trap(VALUE self, int val) { Check_TypedStruct(self, &mutex_data_type); if (val) - FL_SET_RAW(self, MUTEX_ALLOW_TRAP); + FL_SET_RAW(self, MUTEX_ALLOW_TRAP); else - FL_UNSET_RAW(self, MUTEX_ALLOW_TRAP); + FL_UNSET_RAW(self, MUTEX_ALLOW_TRAP); } /* Queue */ #define queue_waitq(q) UNALIGNED_MEMBER_PTR(q, waitq) -PACKED_STRUCT_UNALIGNED(struct rb_queue { - struct list_head waitq; +#define queue_list(q) UNALIGNED_MEMBER_PTR(q, que) +RBIMPL_ATTR_PACKED_STRUCT_UNALIGNED_BEGIN() +struct rb_queue { + struct ccan_list_head waitq; rb_serial_t fork_gen; const VALUE que; int num_waiting; -}); +} RBIMPL_ATTR_PACKED_STRUCT_UNALIGNED_END(); #define szqueue_waitq(sq) UNALIGNED_MEMBER_PTR(sq, q.waitq) +#define szqueue_list(sq) UNALIGNED_MEMBER_PTR(sq, q.que) #define szqueue_pushq(sq) UNALIGNED_MEMBER_PTR(sq, pushq) -PACKED_STRUCT_UNALIGNED(struct rb_szqueue { +RBIMPL_ATTR_PACKED_STRUCT_UNALIGNED_BEGIN() +struct rb_szqueue { struct rb_queue q; int num_waiting_push; - struct list_head pushq; + struct ccan_list_head pushq; long max; -}); +} RBIMPL_ATTR_PACKED_STRUCT_UNALIGNED_END(); static void queue_mark(void *ptr) @@ -680,7 +716,7 @@ queue_alloc(VALUE klass) struct rb_queue *q; obj = TypedData_Make_Struct(klass, struct rb_queue, &queue_data_type, q); - list_head_init(queue_waitq(q)); + ccan_list_head_init(queue_waitq(q)); return obj; } @@ -694,7 +730,7 @@ queue_fork_check(struct rb_queue *q) } /* forked children can't reach into parent thread stacks */ q->fork_gen = fork_gen; - list_head_init(queue_waitq(q)); + ccan_list_head_init(queue_waitq(q)); q->num_waiting = 0; return 1; } @@ -712,6 +748,22 @@ queue_ptr(VALUE obj) #define QUEUE_CLOSED FL_USER5 +static rb_hrtime_t +queue_timeout2hrtime(VALUE timeout) +{ + if (NIL_P(timeout)) { + return (rb_hrtime_t)0; + } + rb_hrtime_t rel = 0; + if (FIXNUM_P(timeout)) { + rel = rb_sec2hrtime(NUM2TIMET(timeout)); + } + else { + double2hrtime(&rel, rb_num2dbl(timeout)); + } + return rb_hrtime_add(rel, rb_hrtime_now()); +} + static void szqueue_mark(void *ptr) { @@ -737,9 +789,9 @@ szqueue_alloc(VALUE klass) { struct rb_szqueue *sq; VALUE obj = TypedData_Make_Struct(klass, struct rb_szqueue, - &szqueue_data_type, sq); - list_head_init(szqueue_waitq(sq)); - list_head_init(szqueue_pushq(sq)); + &szqueue_data_type, sq); + ccan_list_head_init(szqueue_waitq(sq)); + ccan_list_head_init(szqueue_pushq(sq)); return obj; } @@ -750,7 +802,7 @@ szqueue_ptr(VALUE obj) TypedData_Get_Struct(obj, struct rb_szqueue, &szqueue_data_type, sq); if (queue_fork_check(&sq->q)) { - list_head_init(szqueue_pushq(sq)); + ccan_list_head_init(szqueue_pushq(sq)); sq->num_waiting_push = 0; } @@ -760,14 +812,14 @@ szqueue_ptr(VALUE obj) static VALUE ary_buf_new(void) { - return rb_ary_tmp_new(1); + return rb_ary_hidden_new(1); } static VALUE check_array(VALUE obj, VALUE ary) { if (!RB_TYPE_P(ary, T_ARRAY)) { - rb_raise(rb_eTypeError, "%+"PRIsVALUE" not initialized", obj); + rb_raise(rb_eTypeError, "%+"PRIsVALUE" not initialized", obj); } return ary; } @@ -788,7 +840,7 @@ queue_closed_p(VALUE self) * Document-class: ClosedQueueError * * The exception class which will be raised when pushing into a closed - * Queue. See Queue#close and SizedQueue#close. + * Queue. See Thread::Queue#close and Thread::SizedQueue#close. */ NORETURN(static void raise_closed_queue_error(VALUE self)); @@ -802,38 +854,38 @@ raise_closed_queue_error(VALUE self) static VALUE queue_closed_result(VALUE self, struct rb_queue *q) { - assert(queue_length(self, q) == 0); + RUBY_ASSERT(queue_length(self, q) == 0); return Qnil; } /* - * Document-class: Queue + * Document-class: Thread::Queue * - * The Queue class implements multi-producer, multi-consumer queues. - * It is especially useful in threaded programming when information - * must be exchanged safely between multiple threads. The Queue class - * implements all the required locking semantics. + * The Thread::Queue class implements multi-producer, multi-consumer + * queues. It is especially useful in threaded programming when + * information must be exchanged safely between multiple threads. The + * Thread::Queue class implements all the required locking semantics. * - * The class implements FIFO type of queue. In a FIFO queue, the first - * tasks added are the first retrieved. + * The class implements FIFO (first in, first out) type of queue. + * In a FIFO queue, the first tasks added are the first retrieved. * * Example: * - * queue = Queue.new + * queue = Thread::Queue.new * * producer = Thread.new do * 5.times do |i| - * sleep rand(i) # simulate expense - * queue << i - * puts "#{i} produced" + * 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}" + * value = queue.pop + * sleep rand(i/2) # simulate expense + * puts "consumed #{value}" * end * end * @@ -844,14 +896,26 @@ queue_closed_result(VALUE self, struct rb_queue *q) /* * Document-method: Queue::new * - * Creates a new queue instance, optionally using the contents of an Enumerable + * call-seq: + * Thread::Queue.new -> empty_queue + * Thread::Queue.new(enumerable) -> queue + * + * Creates a new queue instance, optionally using the contents of an +enumerable+ * for its initial state. * - * Example: + * Example: * - * q = Queue.new - * q = Queue.new([a, b, c]) - * q = Queue.new(items) + * q = Thread::Queue.new + * #=> #<Thread::Queue:0x00007ff7501110d0> + * q.empty? + * #=> true + * + * q = Thread::Queue.new([1, 2, 3]) + * #=> #<Thread::Queue:0x00007ff7500ec500> + * q.empty? + * #=> false + * q.pop + * #=> 1 */ static VALUE @@ -859,11 +923,13 @@ rb_queue_initialize(int argc, VALUE *argv, VALUE self) { VALUE initial; struct rb_queue *q = queue_ptr(self); - RB_OBJ_WRITE(self, &q->que, ary_buf_new()); - list_head_init(queue_waitq(q)); - rb_scan_args(argc, argv, "01", &initial); + if ((argc = rb_scan_args(argc, argv, "01", &initial)) == 1) { + initial = rb_to_array(initial); + } + RB_OBJ_WRITE(self, queue_list(q), ary_buf_new()); + ccan_list_head_init(queue_waitq(q)); if (argc == 1) { - rb_ary_concat(q->que, rb_to_array(initial)); + rb_ary_concat(q->que, initial); } return self; } @@ -872,7 +938,7 @@ static VALUE queue_do_push(VALUE self, struct rb_queue *q, VALUE obj) { if (queue_closed_p(self)) { - raise_closed_queue_error(self); + raise_closed_queue_error(self); } rb_ary_push(check_array(self, q->que), obj); wakeup_one(queue_waitq(q)); @@ -880,7 +946,7 @@ queue_do_push(VALUE self, struct rb_queue *q, VALUE obj) } /* - * Document-method: Queue#close + * Document-method: Thread::Queue#close * call-seq: * close * @@ -903,7 +969,7 @@ queue_do_push(VALUE self, struct rb_queue *q, VALUE obj) * * Example: * - * q = Queue.new + * q = Thread::Queue.new * Thread.new{ * while e = q.deq # wait for nil to break loop * # ... @@ -918,16 +984,16 @@ rb_queue_close(VALUE self) struct rb_queue *q = queue_ptr(self); if (!queue_closed_p(self)) { - FL_SET(self, QUEUE_CLOSED); + FL_SET(self, QUEUE_CLOSED); - wakeup_all(queue_waitq(q)); + wakeup_all(queue_waitq(q)); } return self; } /* - * Document-method: Queue#closed? + * Document-method: Thread::Queue#closed? * call-seq: closed? * * Returns +true+ if the queue is closed. @@ -936,11 +1002,11 @@ rb_queue_close(VALUE self) static VALUE rb_queue_closed_p(VALUE self) { - return queue_closed_p(self) ? Qtrue : Qfalse; + return RBOOL(queue_closed_p(self)); } /* - * Document-method: Queue#push + * Document-method: Thread::Queue#push * call-seq: * push(object) * enq(object) @@ -956,17 +1022,18 @@ rb_queue_push(VALUE self, VALUE obj) } static VALUE -queue_sleep(VALUE self) +queue_sleep(VALUE _args) { - rb_thread_sleep_deadly_allow_spurious_wakeup(self); + struct queue_sleep_arg *args = (struct queue_sleep_arg *)_args; + rb_thread_sleep_deadly_allow_spurious_wakeup(args->self, args->timeout, args->end); return Qnil; } struct queue_waiter { struct sync_waiter w; union { - struct rb_queue *q; - struct rb_szqueue *sq; + struct rb_queue *q; + struct rb_szqueue *sq; } as; }; @@ -975,11 +1042,9 @@ queue_sleep_done(VALUE p) { struct queue_waiter *qw = (struct queue_waiter *)p; - list_del(&qw->w.node); + ccan_list_del(&qw->w.node); qw->as.q->num_waiting--; - COROUTINE_STACK_FREE(qw); - return Qfalse; } @@ -988,83 +1053,70 @@ szqueue_sleep_done(VALUE p) { struct queue_waiter *qw = (struct queue_waiter *)p; - list_del(&qw->w.node); + ccan_list_del(&qw->w.node); qw->as.sq->num_waiting_push--; - COROUTINE_STACK_FREE(qw); - return Qfalse; } static VALUE -queue_do_pop(VALUE self, struct rb_queue *q, int should_block) +queue_do_pop(VALUE self, struct rb_queue *q, int should_block, VALUE timeout) { check_array(self, q->que); - - while (RARRAY_LEN(q->que) == 0) { + if (RARRAY_LEN(q->que) == 0) { if (!should_block) { rb_raise(rb_eThreadError, "queue empty"); } - else if (queue_closed_p(self)) { + + if (RTEST(rb_equal(INT2FIX(0), timeout))) { + return Qnil; + } + } + + rb_hrtime_t end = queue_timeout2hrtime(timeout); + while (RARRAY_LEN(q->que) == 0) { + if (queue_closed_p(self)) { return queue_closed_result(self, q); } else { rb_execution_context_t *ec = GET_EC(); - assert(RARRAY_LEN(q->que) == 0); - assert(queue_closed_p(self) == 0); + RUBY_ASSERT(RARRAY_LEN(q->que) == 0); + RUBY_ASSERT(queue_closed_p(self) == 0); - COROUTINE_STACK_LOCAL(struct queue_waiter, qw); + struct queue_waiter queue_waiter = { + .w = {.self = self, .th = ec->thread_ptr, .fiber = nonblocking_fiber(ec->fiber_ptr)}, + .as = {.q = q} + }; - qw->w.self = self; - qw->w.th = ec->thread_ptr; - qw->w.fiber = ec->fiber_ptr; + struct ccan_list_head *waitq = queue_waitq(q); - qw->as.q = q; - list_add_tail(queue_waitq(qw->as.q), &qw->w.node); - qw->as.q->num_waiting++; + ccan_list_add_tail(waitq, &queue_waiter.w.node); + queue_waiter.as.q->num_waiting++; - rb_ensure(queue_sleep, self, queue_sleep_done, (VALUE)qw); + struct queue_sleep_arg queue_sleep_arg = { + .self = self, + .timeout = timeout, + .end = end + }; + + rb_ensure(queue_sleep, (VALUE)&queue_sleep_arg, queue_sleep_done, (VALUE)&queue_waiter); + if (!NIL_P(timeout) && (rb_hrtime_now() >= end)) + break; } } return rb_ary_shift(q->que); } -static int -queue_pop_should_block(int argc, const VALUE *argv) -{ - int should_block = 1; - rb_check_arity(argc, 0, 1); - if (argc > 0) { - should_block = !RTEST(argv[0]); - } - return should_block; -} - -/* - * Document-method: Queue#pop - * call-seq: - * pop(non_block=false) - * deq(non_block=false) - * shift(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 - * +ThreadError+ is raised. - */ - static VALUE -rb_queue_pop(int argc, VALUE *argv, VALUE self) +rb_queue_pop(rb_execution_context_t *ec, VALUE self, VALUE non_block, VALUE timeout) { - int should_block = queue_pop_should_block(argc, argv); - return queue_do_pop(self, queue_ptr(self), should_block); + return queue_do_pop(self, queue_ptr(self), !RTEST(non_block), timeout); } /* - * Document-method: Queue#empty? + * Document-method: Thread::Queue#empty? * call-seq: empty? * * Returns +true+ if the queue is empty. @@ -1073,11 +1125,11 @@ rb_queue_pop(int argc, VALUE *argv, VALUE self) static VALUE rb_queue_empty_p(VALUE self) { - return queue_length(self, queue_ptr(self)) == 0 ? Qtrue : Qfalse; + return RBOOL(queue_length(self, queue_ptr(self)) == 0); } /* - * Document-method: Queue#clear + * Document-method: Thread::Queue#clear * * Removes all objects from the queue. */ @@ -1092,7 +1144,7 @@ rb_queue_clear(VALUE self) } /* - * Document-method: Queue#length + * Document-method: Thread::Queue#length * call-seq: * length * size @@ -1106,8 +1158,24 @@ rb_queue_length(VALUE self) return LONG2NUM(queue_length(self, queue_ptr(self))); } +NORETURN(static VALUE rb_queue_freeze(VALUE self)); +/* + * call-seq: + * freeze + * + * The queue can't be frozen, so this method raises an exception: + * Thread::Queue.new.freeze # Raises TypeError (cannot freeze #<Thread::Queue:0x...>) + * + */ +static VALUE +rb_queue_freeze(VALUE self) +{ + rb_raise(rb_eTypeError, "cannot freeze " "%+"PRIsVALUE, self); + UNREACHABLE_RETURN(self); +} + /* - * Document-method: Queue#num_waiting + * Document-method: Thread::Queue#num_waiting * * Returns the number of threads waiting on the queue. */ @@ -1121,12 +1189,12 @@ rb_queue_num_waiting(VALUE self) } /* - * Document-class: SizedQueue + * Document-class: Thread::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. + * See Thread::Queue for an example of how a Thread::SizedQueue works. */ /* @@ -1144,23 +1212,23 @@ rb_szqueue_initialize(VALUE self, VALUE vmax) max = NUM2LONG(vmax); if (max <= 0) { - rb_raise(rb_eArgError, "queue size must be positive"); + rb_raise(rb_eArgError, "queue size must be positive"); } - RB_OBJ_WRITE(self, &sq->q.que, ary_buf_new()); - list_head_init(szqueue_waitq(sq)); - list_head_init(szqueue_pushq(sq)); + RB_OBJ_WRITE(self, szqueue_list(sq), ary_buf_new()); + ccan_list_head_init(szqueue_waitq(sq)); + ccan_list_head_init(szqueue_pushq(sq)); sq->max = max; return self; } /* - * Document-method: SizedQueue#close + * Document-method: Thread::SizedQueue#close * call-seq: * close * - * Similar to Queue#close. + * Similar to Thread::Queue#close. * * The difference is behavior with waiting enqueuing threads. * @@ -1171,17 +1239,17 @@ static VALUE rb_szqueue_close(VALUE self) { if (!queue_closed_p(self)) { - struct rb_szqueue *sq = szqueue_ptr(self); + struct rb_szqueue *sq = szqueue_ptr(self); - FL_SET(self, QUEUE_CLOSED); - wakeup_all(szqueue_waitq(sq)); - wakeup_all(szqueue_pushq(sq)); + FL_SET(self, QUEUE_CLOSED); + wakeup_all(szqueue_waitq(sq)); + wakeup_all(szqueue_pushq(sq)); } return self; } /* - * Document-method: SizedQueue#max + * Document-method: Thread::SizedQueue#max * * Returns the maximum size of the queue. */ @@ -1193,7 +1261,7 @@ rb_szqueue_max_get(VALUE self) } /* - * Document-method: SizedQueue#max= + * Document-method: Thread::SizedQueue#max= * call-seq: max=(number) * * Sets the maximum size of the queue to the given +number+. @@ -1207,114 +1275,83 @@ rb_szqueue_max_set(VALUE self, VALUE vmax) struct rb_szqueue *sq = szqueue_ptr(self); if (max <= 0) { - rb_raise(rb_eArgError, "queue size must be positive"); + rb_raise(rb_eArgError, "queue size must be positive"); } if (max > sq->max) { - diff = max - sq->max; + diff = max - sq->max; } sq->max = max; sync_wakeup(szqueue_pushq(sq), diff); return vmax; } -static int -szqueue_push_should_block(int argc, const VALUE *argv) -{ - int should_block = 1; - rb_check_arity(argc, 1, 2); - if (argc > 1) { - should_block = !RTEST(argv[1]); - } - return should_block; -} - -/* - * Document-method: SizedQueue#push - * call-seq: - * push(object, non_block=false) - * enq(object, non_block=false) - * <<(object) - * - * Pushes +object+ to the queue. - * - * If there is no space left in the queue, waits until space becomes - * available, unless +non_block+ is true. If +non_block+ is true, the - * thread isn't suspended, and +ThreadError+ is raised. - */ - static VALUE -rb_szqueue_push(int argc, VALUE *argv, VALUE self) +rb_szqueue_push(rb_execution_context_t *ec, VALUE self, VALUE object, VALUE non_block, VALUE timeout) { struct rb_szqueue *sq = szqueue_ptr(self); - int should_block = szqueue_push_should_block(argc, argv); - while (queue_length(self, &sq->q) >= sq->max) { - if (!should_block) { + if (queue_length(self, &sq->q) >= sq->max) { + if (RTEST(non_block)) { rb_raise(rb_eThreadError, "queue full"); } - else if (queue_closed_p(self)) { - break; + + if (RTEST(rb_equal(INT2FIX(0), timeout))) { + return Qnil; + } + } + + rb_hrtime_t end = queue_timeout2hrtime(timeout); + while (queue_length(self, &sq->q) >= sq->max) { + if (queue_closed_p(self)) { + raise_closed_queue_error(self); } else { rb_execution_context_t *ec = GET_EC(); - COROUTINE_STACK_LOCAL(struct queue_waiter, qw); - struct list_head *pushq = szqueue_pushq(sq); + struct queue_waiter queue_waiter = { + .w = {.self = self, .th = ec->thread_ptr, .fiber = nonblocking_fiber(ec->fiber_ptr)}, + .as = {.sq = sq} + }; - qw->w.self = self; - qw->w.th = ec->thread_ptr; - qw->w.fiber = ec->fiber_ptr; + struct ccan_list_head *pushq = szqueue_pushq(sq); - qw->as.sq = sq; - list_add_tail(pushq, &qw->w.node); + ccan_list_add_tail(pushq, &queue_waiter.w.node); sq->num_waiting_push++; - rb_ensure(queue_sleep, self, szqueue_sleep_done, (VALUE)qw); + struct queue_sleep_arg queue_sleep_arg = { + .self = self, + .timeout = timeout, + .end = end + }; + rb_ensure(queue_sleep, (VALUE)&queue_sleep_arg, szqueue_sleep_done, (VALUE)&queue_waiter); + if (!NIL_P(timeout) && rb_hrtime_now() >= end) { + return Qnil; + } } } - if (queue_closed_p(self)) { - raise_closed_queue_error(self); - } - - return queue_do_push(self, &sq->q, argv[0]); + return queue_do_push(self, &sq->q, object); } static VALUE -szqueue_do_pop(VALUE self, int should_block) +szqueue_do_pop(VALUE self, int should_block, VALUE timeout) { struct rb_szqueue *sq = szqueue_ptr(self); - VALUE retval = queue_do_pop(self, &sq->q, should_block); + VALUE retval = queue_do_pop(self, &sq->q, should_block, timeout); if (queue_length(self, &sq->q) < sq->max) { - wakeup_one(szqueue_pushq(sq)); + wakeup_one(szqueue_pushq(sq)); } return retval; } - -/* - * Document-method: SizedQueue#pop - * call-seq: - * pop(non_block=false) - * deq(non_block=false) - * shift(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 - * +ThreadError+ is raised. - */ - static VALUE -rb_szqueue_pop(int argc, VALUE *argv, VALUE self) +rb_szqueue_pop(rb_execution_context_t *ec, VALUE self, VALUE non_block, VALUE timeout) { - int should_block = queue_pop_should_block(argc, argv); - return szqueue_do_pop(self, should_block); + return szqueue_do_pop(self, !RTEST(non_block), timeout); } /* - * Document-method: SizedQueue#clear + * Document-method: Thread::SizedQueue#clear * * Removes all objects from the queue. */ @@ -1330,7 +1367,7 @@ rb_szqueue_clear(VALUE self) } /* - * Document-method: SizedQueue#length + * Document-method: Thread::SizedQueue#length * call-seq: * length * size @@ -1347,7 +1384,7 @@ rb_szqueue_length(VALUE self) } /* - * Document-method: SizedQueue#num_waiting + * Document-method: Thread::SizedQueue#num_waiting * * Returns the number of threads waiting on the queue. */ @@ -1361,7 +1398,7 @@ rb_szqueue_num_waiting(VALUE self) } /* - * Document-method: SizedQueue#empty? + * Document-method: Thread::SizedQueue#empty? * call-seq: empty? * * Returns +true+ if the queue is empty. @@ -1372,18 +1409,18 @@ rb_szqueue_empty_p(VALUE self) { struct rb_szqueue *sq = szqueue_ptr(self); - return queue_length(self, &sq->q) == 0 ? Qtrue : Qfalse; + return RBOOL(queue_length(self, &sq->q) == 0); } /* ConditionalVariable */ struct rb_condvar { - struct list_head waitq; + struct ccan_list_head waitq; rb_serial_t fork_gen; }; /* - * Document-class: ConditionVariable + * Document-class: Thread::ConditionVariable * * ConditionVariable objects augment class Mutex. Using condition variables, * it is possible to suspend while in the middle of a critical section until a @@ -1391,8 +1428,8 @@ struct rb_condvar { * * Example: * - * mutex = Mutex.new - * resource = ConditionVariable.new + * mutex = Thread::Mutex.new + * resource = Thread::ConditionVariable.new * * a = Thread.new { * mutex.synchronize { @@ -1433,7 +1470,7 @@ condvar_ptr(VALUE self) /* forked children can't reach into parent thread stacks */ if (cv->fork_gen != fork_gen) { cv->fork_gen = fork_gen; - list_head_init(&cv->waitq); + ccan_list_head_init(&cv->waitq); } return cv; @@ -1446,7 +1483,7 @@ condvar_alloc(VALUE klass) VALUE obj; obj = TypedData_Make_Struct(klass, struct rb_condvar, &cv_data_type, cv); - list_head_init(&cv->waitq); + ccan_list_head_init(&cv->waitq); return obj; } @@ -1461,7 +1498,7 @@ static VALUE rb_condvar_initialize(VALUE self) { struct rb_condvar *cv = condvar_ptr(self); - list_head_init(&cv->waitq); + ccan_list_head_init(&cv->waitq); return self; } @@ -1480,13 +1517,15 @@ do_sleep(VALUE args) } /* - * Document-method: ConditionVariable#wait + * Document-method: Thread::ConditionVariable#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. + * + * Returns the slept result on +mutex+. */ static VALUE @@ -1499,19 +1538,18 @@ rb_condvar_wait(int argc, VALUE *argv, VALUE self) rb_scan_args(argc, argv, "11", &args.mutex, &args.timeout); - COROUTINE_STACK_LOCAL(struct sync_waiter, w); - w->self = args.mutex; - w->th = ec->thread_ptr; - w->fiber = ec->fiber_ptr; + struct sync_waiter sync_waiter = { + .self = args.mutex, + .th = ec->thread_ptr, + .fiber = nonblocking_fiber(ec->fiber_ptr) + }; - list_add_tail(&cv->waitq, &w->node); - rb_ensure(do_sleep, (VALUE)&args, delete_from_waitq, (VALUE)w); - - return self; + ccan_list_add_tail(&cv->waitq, &sync_waiter.node); + return rb_ensure(do_sleep, (VALUE)&args, delete_from_waitq, (VALUE)&sync_waiter); } /* - * Document-method: ConditionVariable#signal + * Document-method: Thread::ConditionVariable#signal * * Wakes up the first thread in line waiting for this lock. */ @@ -1525,7 +1563,7 @@ rb_condvar_signal(VALUE self) } /* - * Document-method: ConditionVariable#broadcast + * Document-method: Thread::ConditionVariable#broadcast * * Wakes up all threads waiting for this lock. */ @@ -1548,10 +1586,10 @@ undumpable(VALUE obj) } static VALUE -define_thread_class(VALUE outer, const char *name, VALUE super) +define_thread_class(VALUE outer, const ID name, VALUE super) { - VALUE klass = rb_define_class_under(outer, name, super); - rb_define_const(rb_cObject, name, klass); + VALUE klass = rb_define_class_id_under(outer, name, super); + rb_const_set(rb_cObject, name, klass); return klass; } @@ -1559,15 +1597,15 @@ static void Init_thread_sync(void) { #undef rb_intern -#if 0 - rb_cMutex = rb_define_class("Mutex", rb_cObject); /* teach rdoc Mutex */ - rb_cConditionVariable = rb_define_class("ConditionVariable", rb_cObject); /* teach rdoc ConditionVariable */ - rb_cQueue = rb_define_class("Queue", rb_cObject); /* teach rdoc Queue */ - rb_cSizedQueue = rb_define_class("SizedQueue", rb_cObject); /* teach rdoc SizedQueue */ +#if defined(TEACH_RDOC) && TEACH_RDOC == 42 + rb_cMutex = rb_define_class_under(rb_cThread, "Mutex", rb_cObject); + rb_cConditionVariable = rb_define_class_under(rb_cThread, "ConditionVariable", rb_cObject); + rb_cQueue = rb_define_class_under(rb_cThread, "Queue", rb_cObject); + rb_cSizedQueue = rb_define_class_under(rb_cThread, "SizedQueue", rb_cObject); #endif #define DEFINE_CLASS(name, super) \ - rb_c##name = define_thread_class(rb_cThread, #name, rb_c##super) + rb_c##name = define_thread_class(rb_cThread, rb_intern(#name), rb_c##super) /* Mutex */ DEFINE_CLASS(Mutex, Object); @@ -1593,16 +1631,14 @@ Init_thread_sync(void) rb_define_method(rb_cQueue, "close", rb_queue_close, 0); rb_define_method(rb_cQueue, "closed?", rb_queue_closed_p, 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_define_method(rb_cQueue, "freeze", rb_queue_freeze, 0); rb_define_alias(rb_cQueue, "enq", "push"); rb_define_alias(rb_cQueue, "<<", "push"); - rb_define_alias(rb_cQueue, "deq", "pop"); - rb_define_alias(rb_cQueue, "shift", "pop"); rb_define_alias(rb_cQueue, "size", "length"); DEFINE_CLASS(SizedQueue, Queue); @@ -1612,17 +1648,10 @@ Init_thread_sync(void) rb_define_method(rb_cSizedQueue, "close", rb_szqueue_close, 0); 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, "empty?", rb_szqueue_empty_p, 0); rb_define_method(rb_cSizedQueue, "clear", rb_szqueue_clear, 0); rb_define_method(rb_cSizedQueue, "length", rb_szqueue_length, 0); rb_define_method(rb_cSizedQueue, "num_waiting", rb_szqueue_num_waiting, 0); - - rb_define_alias(rb_cSizedQueue, "enq", "push"); - rb_define_alias(rb_cSizedQueue, "<<", "push"); - rb_define_alias(rb_cSizedQueue, "deq", "pop"); - rb_define_alias(rb_cSizedQueue, "shift", "pop"); rb_define_alias(rb_cSizedQueue, "size", "length"); /* CVar */ @@ -1640,3 +1669,5 @@ Init_thread_sync(void) rb_provide("thread.rb"); } + +#include "thread_sync.rbinc" |