diff options
Diffstat (limited to 'thread_sync.c')
| -rw-r--r-- | thread_sync.c | 1175 |
1 files changed, 391 insertions, 784 deletions
diff --git a/thread_sync.c b/thread_sync.c index c0a0ca7103..e3916c97cb 100644 --- a/thread_sync.c +++ b/thread_sync.c @@ -2,12 +2,12 @@ #include "ccan/list/list.h" #include "builtin.h" -static VALUE rb_cMutex, rb_cQueue, rb_cSizedQueue, rb_cConditionVariable; -static VALUE rb_eClosedQueueError; +static VALUE rb_cMutex, rb_eClosedQueueError; /* Mutex */ typedef struct rb_mutex_struct { - rb_fiber_t *fiber; + rb_serial_t ec_serial; + rb_thread_t *th; // even if the fiber is collected, we might need access to the thread in mutex_free struct rb_mutex_struct *next_mutex; struct ccan_list_head waitq; /* protected by GVL */ } rb_mutex_t; @@ -80,33 +80,7 @@ static void rb_mutex_abandon_all(rb_mutex_t *mutexes); static void rb_mutex_abandon_keeping_mutexes(rb_thread_t *th); static void rb_mutex_abandon_locking_mutex(rb_thread_t *th); #endif -static const char* rb_mutex_unlock_th(rb_mutex_t *mutex, rb_thread_t *th, rb_fiber_t *fiber); - -/* - * Document-class: Thread::Mutex - * - * Thread::Mutex implements a simple semaphore that can be used to - * coordinate access to shared data from multiple concurrent threads. - * - * Example: - * - * semaphore = Thread::Mutex.new - * - * a = Thread.new { - * semaphore.synchronize { - * # access shared resource - * } - * } - * - * b = Thread.new { - * semaphore.synchronize { - * # access shared resource - * } - * } - * - */ - -#define mutex_mark ((void(*)(void*))0) +static const char* rb_mutex_unlock_th(rb_mutex_t *mutex, rb_thread_t *th, rb_serial_t ec_serial); static size_t rb_mutex_num_waiting(rb_mutex_t *mutex) @@ -123,13 +97,18 @@ rb_mutex_num_waiting(rb_mutex_t *mutex) rb_thread_t* rb_fiber_threadptr(const rb_fiber_t *fiber); +static bool +mutex_locked_p(rb_mutex_t *mutex) +{ + return mutex->ec_serial != 0; +} + static void 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 (mutex_locked_p(mutex)) { + const char *err = rb_mutex_unlock_th(mutex, mutex->th, 0); if (err) rb_bug("%s", err); } ruby_xfree(ptr); @@ -143,8 +122,8 @@ 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_WB_PROTECTED | RUBY_TYPED_FREE_IMMEDIATELY + {NULL, mutex_free, mutex_memsize,}, + 0, 0, RUBY_TYPED_FREE_IMMEDIATELY }; static rb_mutex_t * @@ -175,41 +154,24 @@ mutex_alloc(VALUE klass) return obj; } -/* - * call-seq: - * Thread::Mutex.new -> mutex - * - * Creates a new Mutex - */ -static VALUE -mutex_initialize(VALUE self) -{ - return self; -} - VALUE rb_mutex_new(void) { return mutex_alloc(rb_cMutex); } -/* - * call-seq: - * mutex.locked? -> true or false - * - * Returns +true+ if this lock is currently held by some thread. - */ VALUE rb_mutex_locked_p(VALUE self) { rb_mutex_t *mutex = mutex_ptr(self); - return RBOOL(mutex->fiber); + return RBOOL(mutex_locked_p(mutex)); } static void thread_mutex_insert(rb_thread_t *thread, rb_mutex_t *mutex) { + RUBY_ASSERT(!mutex->next_mutex); if (thread->keeping_mutexes) { mutex->next_mutex = thread->keeping_mutexes; } @@ -234,45 +196,50 @@ thread_mutex_remove(rb_thread_t *thread, rb_mutex_t *mutex) } static void -mutex_locked(rb_thread_t *th, VALUE self) +mutex_set_owner(rb_mutex_t *mutex, rb_thread_t *th, rb_serial_t ec_serial) { - rb_mutex_t *mutex = mutex_ptr(self); + mutex->th = th; + mutex->ec_serial = ec_serial; +} +static void +mutex_locked(rb_mutex_t *mutex, rb_thread_t *th, rb_serial_t ec_serial) +{ + mutex_set_owner(mutex, th, ec_serial); thread_mutex_insert(th, mutex); } -/* - * call-seq: - * mutex.try_lock -> true or false - * - * Attempts to obtain the lock and returns immediately. Returns +true+ if the - * lock was granted. - */ -VALUE -rb_mutex_trylock(VALUE self) +static inline bool +do_mutex_trylock(rb_mutex_t *mutex, rb_thread_t *th, rb_serial_t ec_serial) { - rb_mutex_t *mutex = mutex_ptr(self); - - if (mutex->fiber == 0) { + if (mutex->ec_serial == 0) { RUBY_DEBUG_LOG("%p ok", mutex); - rb_fiber_t *fiber = GET_EC()->fiber_ptr; - rb_thread_t *th = GET_THREAD(); - mutex->fiber = fiber; - - mutex_locked(th, self); - return Qtrue; + mutex_locked(mutex, th, ec_serial); + return true; } else { RUBY_DEBUG_LOG("%p ng", mutex); - return Qfalse; + return false; } } static VALUE -mutex_owned_p(rb_fiber_t *fiber, rb_mutex_t *mutex) +rb_mut_trylock(rb_execution_context_t *ec, VALUE self) +{ + return RBOOL(do_mutex_trylock(mutex_ptr(self), ec->thread_ptr, rb_ec_serial(ec))); +} + +VALUE +rb_mutex_trylock(VALUE self) { - return RBOOL(mutex->fiber == fiber); + return rb_mut_trylock(GET_EC(), self); +} + +static VALUE +mutex_owned_p(rb_serial_t ec_serial, rb_mutex_t *mutex) +{ + return RBOOL(mutex->ec_serial == ec_serial); } static VALUE @@ -292,13 +259,29 @@ delete_from_waitq(VALUE value) static inline rb_atomic_t threadptr_get_interrupts(rb_thread_t *th); +struct mutex_args { + VALUE self; + rb_mutex_t *mutex; + rb_execution_context_t *ec; +}; + +static inline void +mutex_args_init(struct mutex_args *args, VALUE mutex) +{ + args->self = mutex; + args->mutex = mutex_ptr(mutex); + args->ec = GET_EC(); +} + static VALUE -do_mutex_lock(VALUE self, int interruptible_p) +do_mutex_lock(struct mutex_args *args, int interruptible_p) { - rb_execution_context_t *ec = GET_EC(); + VALUE self = args->self; + rb_execution_context_t *ec = args->ec; rb_thread_t *th = ec->thread_ptr; rb_fiber_t *fiber = ec->fiber_ptr; - rb_mutex_t *mutex = mutex_ptr(self); + rb_serial_t ec_serial = rb_ec_serial(ec); + rb_mutex_t *mutex = args->mutex; rb_atomic_t saved_ints = 0; /* When running trap handler */ @@ -307,13 +290,13 @@ do_mutex_lock(VALUE self, int interruptible_p) rb_raise(rb_eThreadError, "can't be called from trap context"); } - if (rb_mutex_trylock(self) == Qfalse) { - if (mutex->fiber == fiber) { + if (!do_mutex_trylock(mutex, th, ec_serial)) { + if (mutex->ec_serial == ec_serial) { rb_raise(rb_eThreadError, "deadlock; recursive locking"); } - while (mutex->fiber != fiber) { - VM_ASSERT(mutex->fiber != NULL); + while (mutex->ec_serial != ec_serial) { + VM_ASSERT(mutex->ec_serial != 0); VALUE scheduler = rb_fiber_scheduler_current(); if (scheduler != Qnil) { @@ -327,12 +310,12 @@ do_mutex_lock(VALUE self, int interruptible_p) rb_ensure(call_rb_fiber_scheduler_block, self, delete_from_waitq, (VALUE)&sync_waiter); - if (!mutex->fiber) { - mutex->fiber = fiber; + if (!mutex->ec_serial) { + mutex_set_owner(mutex, th, ec_serial); } } else { - if (!th->vm->thread_ignore_deadlock && rb_fiber_threadptr(mutex->fiber) == th) { + if (!th->vm->thread_ignore_deadlock && mutex->th == th) { rb_raise(rb_eThreadError, "deadlock; lock already owned by another fiber belonging to the same thread"); } @@ -358,6 +341,7 @@ do_mutex_lock(VALUE self, int interruptible_p) rb_ractor_sleeper_threads_inc(th->ractor); rb_check_deadlock(th->ractor); + RUBY_ASSERT(!th->locking_mutex); th->locking_mutex = self; ccan_list_add_tail(&mutex->waitq, &sync_waiter.node); @@ -367,14 +351,13 @@ do_mutex_lock(VALUE self, int interruptible_p) ccan_list_del(&sync_waiter.node); // unlocked by another thread while sleeping - if (!mutex->fiber) { - mutex->fiber = fiber; + if (!mutex->ec_serial) { + mutex_set_owner(mutex, th, ec_serial); } 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); } @@ -382,10 +365,13 @@ do_mutex_lock(VALUE self, int interruptible_p) if (interruptible_p) { /* release mutex before checking for interrupts...as interrupt checking * code might call rb_raise() */ - if (mutex->fiber == fiber) mutex->fiber = 0; + if (mutex->ec_serial == ec_serial) { + mutex->th = NULL; + mutex->ec_serial = 0; + } RUBY_VM_CHECK_INTS_BLOCKING(th->ec); /* may release mutex */ - if (!mutex->fiber) { - mutex->fiber = fiber; + if (!mutex->ec_serial) { + mutex_set_owner(mutex, th, ec_serial); } } else { @@ -404,13 +390,13 @@ do_mutex_lock(VALUE self, int interruptible_p) } if (saved_ints) th->ec->interrupt_flag = saved_ints; - if (mutex->fiber == fiber) mutex_locked(th, self); + if (mutex->ec_serial == ec_serial) mutex_locked(mutex, th, ec_serial); } RUBY_DEBUG_LOG("%p locked", mutex); // assertion - if (mutex_owned_p(fiber, mutex) == Qfalse) rb_bug("do_mutex_lock: mutex is not owned."); + if (mutex_owned_p(ec_serial, mutex) == Qfalse) rb_bug("do_mutex_lock: mutex is not owned."); return self; } @@ -418,52 +404,57 @@ do_mutex_lock(VALUE self, int interruptible_p) static VALUE mutex_lock_uninterruptible(VALUE self) { - return do_mutex_lock(self, 0); + struct mutex_args args; + mutex_args_init(&args, self); + return do_mutex_lock(&args, 0); +} + +static VALUE +rb_mut_lock(rb_execution_context_t *ec, VALUE self) +{ + struct mutex_args args = { + .self = self, + .mutex = mutex_ptr(self), + .ec = ec, + }; + return do_mutex_lock(&args, 1); } -/* - * call-seq: - * mutex.lock -> self - * - * Attempts to grab the lock and waits if it isn't available. - * Raises +ThreadError+ if +mutex+ was locked by the current thread. - */ VALUE rb_mutex_lock(VALUE self) { - return do_mutex_lock(self, 1); + struct mutex_args args; + mutex_args_init(&args, self); + return do_mutex_lock(&args, 1); +} + +static VALUE +rb_mut_owned_p(rb_execution_context_t *ec, VALUE self) +{ + return mutex_owned_p(rb_ec_serial(ec), mutex_ptr(self)); } -/* - * call-seq: - * mutex.owned? -> true or false - * - * Returns +true+ if this lock is currently held by current thread. - */ VALUE rb_mutex_owned_p(VALUE self) { - rb_fiber_t *fiber = GET_EC()->fiber_ptr; - rb_mutex_t *mutex = mutex_ptr(self); - - return mutex_owned_p(fiber, mutex); + return rb_mut_owned_p(GET_EC(), self); } static const char * -rb_mutex_unlock_th(rb_mutex_t *mutex, rb_thread_t *th, rb_fiber_t *fiber) +rb_mutex_unlock_th(rb_mutex_t *mutex, rb_thread_t *th, rb_serial_t ec_serial) { RUBY_DEBUG_LOG("%p", mutex); - if (mutex->fiber == 0) { + if (mutex->ec_serial == 0) { return "Attempt to unlock a mutex which is not locked"; } - else if (mutex->fiber != fiber) { + else if (ec_serial && mutex->ec_serial != ec_serial) { return "Attempt to unlock a mutex which is locked by another thread/fiber"; } struct sync_waiter *cur = 0, *next; - mutex->fiber = 0; + mutex->ec_serial = 0; thread_mutex_remove(th, mutex); ccan_list_for_each_safe(&mutex->waitq, cur, next, node) { @@ -494,6 +485,24 @@ rb_mutex_unlock_th(rb_mutex_t *mutex, rb_thread_t *th, rb_fiber_t *fiber) return NULL; } +static void +do_mutex_unlock(struct mutex_args *args) +{ + const char *err; + rb_mutex_t *mutex = args->mutex; + rb_thread_t *th = rb_ec_thread_ptr(args->ec); + + err = rb_mutex_unlock_th(mutex, th, rb_ec_serial(args->ec)); + if (err) rb_raise(rb_eThreadError, "%s", err); +} + +static VALUE +do_mutex_unlock_safe(VALUE args) +{ + do_mutex_unlock((struct mutex_args *)args); + return Qnil; +} + /* * call-seq: * mutex.unlock -> self @@ -504,13 +513,21 @@ rb_mutex_unlock_th(rb_mutex_t *mutex, rb_thread_t *th, rb_fiber_t *fiber) VALUE rb_mutex_unlock(VALUE self) { - const char *err; - rb_mutex_t *mutex = mutex_ptr(self); - rb_thread_t *th = GET_THREAD(); - - err = rb_mutex_unlock_th(mutex, th, GET_EC()->fiber_ptr); - if (err) rb_raise(rb_eThreadError, "%s", err); + struct mutex_args args; + mutex_args_init(&args, self); + do_mutex_unlock(&args); + return self; +} +static VALUE +rb_mut_unlock(rb_execution_context_t *ec, VALUE self) +{ + struct mutex_args args = { + .self = self, + .mutex = mutex_ptr(self), + .ec = ec, + }; + do_mutex_unlock(&args); return self; } @@ -541,117 +558,99 @@ rb_mutex_abandon_all(rb_mutex_t *mutexes) while (mutexes) { mutex = mutexes; mutexes = mutex->next_mutex; - mutex->fiber = 0; + mutex->ec_serial = 0; mutex->next_mutex = 0; ccan_list_head_init(&mutex->waitq); } } #endif -static VALUE -rb_mutex_sleep_forever(VALUE self) -{ - rb_thread_sleep_deadly_allow_spurious_wakeup(self, Qnil, 0); - return Qnil; -} +struct rb_mutex_sleep_arguments { + VALUE self; + VALUE timeout; +}; static VALUE -rb_mutex_wait_for(VALUE time) -{ - rb_hrtime_t *rel = (rb_hrtime_t *)time; - /* permit spurious check */ - return RBOOL(sleep_hrtime(GET_THREAD(), *rel, 0)); -} - -VALUE -rb_mutex_sleep(VALUE self, VALUE timeout) +mutex_sleep_begin(VALUE _arguments) { - struct timeval t; + struct rb_mutex_sleep_arguments *arguments = (struct rb_mutex_sleep_arguments *)_arguments; + VALUE timeout = arguments->timeout; VALUE woken = Qtrue; - if (!NIL_P(timeout)) { - t = rb_time_interval(timeout); - } - - rb_mutex_unlock(self); - time_t beg = time(0); - VALUE scheduler = rb_fiber_scheduler_current(); if (scheduler != Qnil) { rb_fiber_scheduler_kernel_sleep(scheduler, timeout); - mutex_lock_uninterruptible(self); } else { if (NIL_P(timeout)) { - rb_ensure(rb_mutex_sleep_forever, self, mutex_lock_uninterruptible, self); + rb_thread_sleep_deadly_allow_spurious_wakeup(arguments->self, Qnil, 0); } else { - rb_hrtime_t rel = rb_timeval2hrtime(&t); - woken = rb_ensure(rb_mutex_wait_for, (VALUE)&rel, mutex_lock_uninterruptible, self); + struct timeval timeout_value = rb_time_interval(timeout); + rb_hrtime_t relative_timeout = rb_timeval2hrtime(&timeout_value); + /* permit spurious check */ + woken = RBOOL(sleep_hrtime(GET_THREAD(), relative_timeout, 0)); } } - RUBY_VM_CHECK_INTS_BLOCKING(GET_EC()); - if (!woken) return Qnil; - time_t end = time(0) - beg; - return TIMET2NUM(end); + return woken; } -/* - * call-seq: - * 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 - * the current thread. - * - * When the thread is next woken up, it will attempt to reacquire - * the lock. - * - * 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) +rb_mut_sleep(rb_execution_context_t *ec, VALUE self, VALUE timeout) { - VALUE timeout; + if (!NIL_P(timeout)) { + // Validate the argument: + rb_time_interval(timeout); + } + + rb_mut_unlock(ec, self); + time_t beg = time(0); - timeout = rb_check_arity(argc, 0, 1) ? argv[0] : Qnil; - return rb_mutex_sleep(self, timeout); + struct rb_mutex_sleep_arguments arguments = { + .self = self, + .timeout = timeout, + }; + + VALUE woken = rb_ec_ensure(ec, mutex_sleep_begin, (VALUE)&arguments, mutex_lock_uninterruptible, self); + + RUBY_VM_CHECK_INTS_BLOCKING(ec); + if (!woken) return Qnil; + time_t end = time(0) - beg; + return TIMET2NUM(end); } -/* - * call-seq: - * mutex.synchronize { ... } -> result of the block - * - * Obtains a lock, runs the block, and releases the lock when the block - * completes. See the example under Thread::Mutex. - */ +VALUE +rb_mutex_sleep(VALUE self, VALUE timeout) +{ + return rb_mut_sleep(GET_EC(), self, timeout); +} VALUE -rb_mutex_synchronize(VALUE mutex, VALUE (*func)(VALUE arg), VALUE arg) +rb_mutex_synchronize(VALUE self, VALUE (*func)(VALUE arg), VALUE arg) { - rb_mutex_lock(mutex); - return rb_ensure(func, arg, rb_mutex_unlock, mutex); + struct mutex_args args; + mutex_args_init(&args, self); + do_mutex_lock(&args, 1); + return rb_ec_ensure(args.ec, func, arg, do_mutex_unlock_safe, (VALUE)&args); } -/* - * call-seq: - * mutex.synchronize { ... } -> result of the block - * - * Obtains a lock, runs the block, and releases the lock when the block - * completes. See the example under Thread::Mutex. - */ static VALUE -rb_mutex_synchronize_m(VALUE self) +do_ec_yield(VALUE _ec) { - if (!rb_block_given_p()) { - rb_raise(rb_eThreadError, "must be called with a block"); - } + return rb_ec_yield((rb_execution_context_t *)_ec, Qundef); +} - return rb_mutex_synchronize(self, rb_yield, Qundef); +VALUE +rb_mut_synchronize(rb_execution_context_t *ec, VALUE self) +{ + struct mutex_args args = { + .self = self, + .mutex = mutex_ptr(self), + .ec = ec, + }; + do_mutex_lock(&args, 1); + return rb_ec_ensure(args.ec, do_ec_yield, (VALUE)ec, do_mutex_unlock_safe, (VALUE)&args); } void @@ -667,46 +666,61 @@ rb_mutex_allow_trap(VALUE self, int val) /* Queue */ -#define queue_waitq(q) UNALIGNED_MEMBER_PTR(q, 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; + long capa; + long len; + long offset; + VALUE *buffer; int num_waiting; -} RBIMPL_ATTR_PACKED_STRUCT_UNALIGNED_END(); +}; + +#define szqueue_waitq(sq) &sq->q.waitq +#define szqueue_pushq(sq) &sq->pushq -#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) -RBIMPL_ATTR_PACKED_STRUCT_UNALIGNED_BEGIN() struct rb_szqueue { struct rb_queue q; int num_waiting_push; struct ccan_list_head pushq; long max; -} RBIMPL_ATTR_PACKED_STRUCT_UNALIGNED_END(); +}; static void -queue_mark(void *ptr) +queue_mark_and_move(void *ptr) { struct rb_queue *q = ptr; - /* no need to mark threads in waitq, they are on stack */ - rb_gc_mark(q->que); + for (long index = 0; index < q->len; index++) { + rb_gc_mark_and_move(&q->buffer[((q->offset + index) % q->capa)]); + } +} + +static void +queue_free(void *ptr) +{ + struct rb_queue *q = ptr; + if (q->buffer) { + ruby_sized_xfree(q->buffer, q->capa * sizeof(VALUE)); + } } static size_t queue_memsize(const void *ptr) { - return sizeof(struct rb_queue); + const struct rb_queue *q = ptr; + return sizeof(struct rb_queue) + (q->capa * sizeof(VALUE)); } static const rb_data_type_t queue_data_type = { - "queue", - {queue_mark, RUBY_TYPED_DEFAULT_FREE, queue_memsize,}, - 0, 0, RUBY_TYPED_FREE_IMMEDIATELY|RUBY_TYPED_WB_PROTECTED + .wrap_struct_name = "Thread::Queue", + .function = { + .dmark = queue_mark_and_move, + .dfree = queue_free, + .dsize = queue_memsize, + .dcompact = queue_mark_and_move, + }, + .flags = RUBY_TYPED_FREE_IMMEDIATELY | RUBY_TYPED_WB_PROTECTED, }; static VALUE @@ -716,27 +730,27 @@ queue_alloc(VALUE klass) struct rb_queue *q; obj = TypedData_Make_Struct(klass, struct rb_queue, &queue_data_type, q); - ccan_list_head_init(queue_waitq(q)); + ccan_list_head_init(&q->waitq); return obj; } -static int +static inline bool queue_fork_check(struct rb_queue *q) { rb_serial_t fork_gen = GET_VM()->fork_gen; - if (q->fork_gen == fork_gen) { - return 0; + if (RB_LIKELY(q->fork_gen == fork_gen)) { + return false; } /* forked children can't reach into parent thread stacks */ q->fork_gen = fork_gen; - ccan_list_head_init(queue_waitq(q)); + ccan_list_head_init(&q->waitq); q->num_waiting = 0; - return 1; + return true; } -static struct rb_queue * -queue_ptr(VALUE obj) +static inline struct rb_queue * +raw_queue_ptr(VALUE obj) { struct rb_queue *q; @@ -746,6 +760,22 @@ queue_ptr(VALUE obj) return q; } +static inline void +check_queue(VALUE obj, struct rb_queue *q) +{ + if (RB_UNLIKELY(q->buffer == NULL)) { + rb_raise(rb_eTypeError, "%+"PRIsVALUE" not initialized", obj); + } +} + +static inline struct rb_queue * +queue_ptr(VALUE obj) +{ + struct rb_queue *q = raw_queue_ptr(obj); + check_queue(obj, q); + return q; +} + #define QUEUE_CLOSED FL_USER5 static rb_hrtime_t @@ -765,23 +795,37 @@ queue_timeout2hrtime(VALUE timeout) } static void -szqueue_mark(void *ptr) +szqueue_mark_and_move(void *ptr) { struct rb_szqueue *sq = ptr; - queue_mark(&sq->q); + queue_mark_and_move(&sq->q); +} + +static void +szqueue_free(void *ptr) +{ + struct rb_szqueue *sq = ptr; + queue_free(&sq->q); } static size_t szqueue_memsize(const void *ptr) { - return sizeof(struct rb_szqueue); + const struct rb_szqueue *sq = ptr; + return sizeof(struct rb_szqueue) + (sq->q.capa * sizeof(VALUE)); } static const rb_data_type_t szqueue_data_type = { - "sized_queue", - {szqueue_mark, RUBY_TYPED_DEFAULT_FREE, szqueue_memsize,}, - 0, 0, RUBY_TYPED_FREE_IMMEDIATELY|RUBY_TYPED_WB_PROTECTED + .wrap_struct_name = "Thread::SizedQueue", + .function = { + .dmark = szqueue_mark_and_move, + .dfree = szqueue_free, + .dsize = szqueue_memsize, + .dcompact = szqueue_mark_and_move, + }, + .parent = &queue_data_type, + .flags = RUBY_TYPED_FREE_IMMEDIATELY | RUBY_TYPED_WB_PROTECTED, }; static VALUE @@ -795,13 +839,13 @@ szqueue_alloc(VALUE klass) return obj; } -static struct rb_szqueue * -szqueue_ptr(VALUE obj) +static inline struct rb_szqueue * +raw_szqueue_ptr(VALUE obj) { struct rb_szqueue *sq; TypedData_Get_Struct(obj, struct rb_szqueue, &szqueue_data_type, sq); - if (queue_fork_check(&sq->q)) { + if (RB_UNLIKELY(queue_fork_check(&sq->q))) { ccan_list_head_init(szqueue_pushq(sq)); sq->num_waiting_push = 0; } @@ -809,28 +853,15 @@ szqueue_ptr(VALUE obj) return sq; } -static VALUE -ary_buf_new(void) -{ - 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); - } - return ary; -} - -static long -queue_length(VALUE self, struct rb_queue *q) +static inline struct rb_szqueue * +szqueue_ptr(VALUE obj) { - return RARRAY_LEN(check_array(self, q->que)); + struct rb_szqueue *sq = raw_szqueue_ptr(obj); + check_queue(obj, &sq->q); + return sq; } -static int +static inline bool queue_closed_p(VALUE self) { return FL_TEST_RAW(self, QUEUE_CLOSED) != 0; @@ -854,171 +885,95 @@ 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(q->len == 0); return Qnil; } -/* - * Document-class: Thread::Queue - * - * 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 (first in, first out) type of queue. - * In a FIFO queue, the first tasks added are the first retrieved. - * - * Example: - * - * queue = Thread::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 - * - */ +#define QUEUE_INITIAL_CAPA 8 -/* - * Document-method: Queue::new - * - * 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: - * - * 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 inline void +ring_buffer_init(struct rb_queue *q, long initial_capa) +{ + q->buffer = ALLOC_N(VALUE, initial_capa); + q->capa = initial_capa; +} -static VALUE -rb_queue_initialize(int argc, VALUE *argv, VALUE self) +static inline void +ring_buffer_expand(struct rb_queue *q) { - VALUE initial; - struct rb_queue *q = queue_ptr(self); - 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, initial); - } - return self; + RUBY_ASSERT(q->capa > 0); + VALUE *new_buffer = ALLOC_N(VALUE, q->capa * 2); + MEMCPY(new_buffer, q->buffer + q->offset, VALUE, q->capa - q->offset); + MEMCPY(new_buffer + (q->capa - q->offset), q->buffer, VALUE, q->offset); + VALUE *old_buffer = q->buffer; + q->buffer = new_buffer; + q->offset = 0; + ruby_sized_xfree(old_buffer, q->capa * sizeof(VALUE)); + q->capa *= 2; } -static VALUE -queue_do_push(VALUE self, struct rb_queue *q, VALUE obj) +static void +ring_buffer_push(VALUE self, struct rb_queue *q, VALUE obj) { - if (queue_closed_p(self)) { - raise_closed_queue_error(self); + if (RB_UNLIKELY(q->len >= q->capa)) { + ring_buffer_expand(q); } - rb_ary_push(check_array(self, q->que), obj); - wakeup_one(queue_waitq(q)); - return self; + RUBY_ASSERT(q->capa > q->len); + long index = (q->offset + q->len) % q->capa; + q->len++; + RB_OBJ_WRITE(self, &q->buffer[index], obj); } -/* - * Document-method: Thread::Queue#close - * call-seq: - * close - * - * Closes the queue. A closed queue cannot be re-opened. - * - * After the call to close completes, the following are true: - * - * - +closed?+ will return true - * - * - +close+ will be ignored. - * - * - calling enq/push/<< will raise a +ClosedQueueError+. - * - * - when +empty?+ is false, calling deq/pop/shift will return an object - * from the queue as usual. - * - when +empty?+ is true, deq(false) will not suspend the thread and will return nil. - * deq(true) will raise a +ThreadError+. - * - * ClosedQueueError is inherited from StopIteration, so that you can break loop block. - * - * Example: - * - * q = Thread::Queue.new - * Thread.new{ - * while e = q.deq # wait for nil to break loop - * # ... - * end - * } - * q.close - */ - static VALUE -rb_queue_close(VALUE self) +ring_buffer_shift(struct rb_queue *q) { - struct rb_queue *q = queue_ptr(self); - - if (!queue_closed_p(self)) { - FL_SET(self, QUEUE_CLOSED); - - wakeup_all(queue_waitq(q)); + if (!q->len) { + return Qnil; } - return self; + VALUE obj = q->buffer[q->offset]; + q->len--; + if (q->len == 0) { + q->offset = 0; + } + else { + q->offset = (q->offset + 1) % q->capa; + } + return obj; } -/* - * Document-method: Thread::Queue#closed? - * call-seq: closed? - * - * Returns +true+ if the queue is closed. - */ - static VALUE -rb_queue_closed_p(VALUE self) +queue_initialize(rb_execution_context_t *ec, VALUE self, VALUE initial) { - return RBOOL(queue_closed_p(self)); + struct rb_queue *q = raw_queue_ptr(self); + ccan_list_head_init(&q->waitq); + if (NIL_P(initial)) { + ring_buffer_init(q, QUEUE_INITIAL_CAPA); + } + else { + initial = rb_to_array(initial); + long len = RARRAY_LEN(initial); + long initial_capa = QUEUE_INITIAL_CAPA; + while (initial_capa < len) { + initial_capa *= 2; + } + ring_buffer_init(q, initial_capa); + MEMCPY(q->buffer, RARRAY_CONST_PTR(initial), VALUE, len); + q->len = len; + } + return self; } -/* - * Document-method: Thread::Queue#push - * call-seq: - * push(object) - * enq(object) - * <<(object) - * - * Pushes the given +object+ to the queue. - */ - static VALUE -rb_queue_push(VALUE self, VALUE obj) +queue_do_push(VALUE self, struct rb_queue *q, VALUE obj) { - return queue_do_push(self, queue_ptr(self), obj); + check_queue(self, q); + if (queue_closed_p(self)) { + raise_closed_queue_error(self); + } + ring_buffer_push(self, q, obj); + wakeup_one(&q->waitq); + return self; } static VALUE @@ -1059,12 +1014,11 @@ szqueue_sleep_done(VALUE p) return Qfalse; } -static VALUE -queue_do_pop(VALUE self, struct rb_queue *q, int should_block, VALUE timeout) +static inline VALUE +queue_do_pop(rb_execution_context_t *ec, VALUE self, struct rb_queue *q, VALUE non_block, VALUE timeout) { - check_array(self, q->que); - if (RARRAY_LEN(q->que) == 0) { - if (!should_block) { + if (q->len == 0) { + if (RTEST(non_block)) { rb_raise(rb_eThreadError, "queue empty"); } @@ -1074,22 +1028,20 @@ queue_do_pop(VALUE self, struct rb_queue *q, int should_block, VALUE timeout) } rb_hrtime_t end = queue_timeout2hrtime(timeout); - while (RARRAY_LEN(q->que) == 0) { + while (q->len == 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(q->len == 0); + RUBY_ASSERT(queue_closed_p(self) == 0); struct queue_waiter queue_waiter = { .w = {.self = self, .th = ec->thread_ptr, .fiber = nonblocking_fiber(ec->fiber_ptr)}, .as = {.q = q} }; - struct ccan_list_head *waitq = queue_waitq(q); + struct ccan_list_head *waitq = &q->waitq; ccan_list_add_tail(waitq, &queue_waiter.w.node); queue_waiter.as.q->num_waiting++; @@ -1106,116 +1058,32 @@ queue_do_pop(VALUE self, struct rb_queue *q, int should_block, VALUE timeout) } } - return rb_ary_shift(q->que); + return ring_buffer_shift(q); } static VALUE rb_queue_pop(rb_execution_context_t *ec, VALUE self, VALUE non_block, VALUE timeout) { - return queue_do_pop(self, queue_ptr(self), !RTEST(non_block), timeout); + return queue_do_pop(ec, self, queue_ptr(self), non_block, timeout); } -/* - * Document-method: Thread::Queue#empty? - * call-seq: empty? - * - * Returns +true+ if the queue is empty. - */ - -static VALUE -rb_queue_empty_p(VALUE self) -{ - return RBOOL(queue_length(self, queue_ptr(self)) == 0); -} - -/* - * Document-method: Thread::Queue#clear - * - * Removes all objects from the queue. - */ - -static VALUE -rb_queue_clear(VALUE self) -{ - struct rb_queue *q = queue_ptr(self); - - rb_ary_clear(check_array(self, q->que)); - return self; -} - -/* - * Document-method: Thread::Queue#length - * call-seq: - * length - * size - * - * Returns the length of the queue. - */ - -static VALUE -rb_queue_length(VALUE self) +static void +queue_clear(struct rb_queue *q) { - return LONG2NUM(queue_length(self, queue_ptr(self))); + q->len = 0; + q->offset = 0; } -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) +szqueue_initialize(rb_execution_context_t *ec, VALUE self, VALUE vmax) { - rb_raise(rb_eTypeError, "cannot freeze " "%+"PRIsVALUE, self); - UNREACHABLE_RETURN(self); -} - -/* - * Document-method: Thread::Queue#num_waiting - * - * Returns the number of threads waiting on the queue. - */ - -static VALUE -rb_queue_num_waiting(VALUE self) -{ - struct rb_queue *q = queue_ptr(self); - - return INT2NUM(q->num_waiting); -} - -/* - * Document-class: Thread::SizedQueue - * - * This class represents queues of specified size capacity. The push operation - * may be blocked if the capacity is full. - * - * See Thread::Queue for an example of how a Thread::SizedQueue works. - */ - -/* - * Document-method: SizedQueue::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; - struct rb_szqueue *sq = szqueue_ptr(self); + long max = NUM2LONG(vmax); + struct rb_szqueue *sq = raw_szqueue_ptr(self); - max = NUM2LONG(vmax); if (max <= 0) { rb_raise(rb_eArgError, "queue size must be positive"); } - - RB_OBJ_WRITE(self, szqueue_list(sq), ary_buf_new()); + ring_buffer_init(&sq->q, QUEUE_INITIAL_CAPA); ccan_list_head_init(szqueue_waitq(sq)); ccan_list_head_init(szqueue_pushq(sq)); sq->max = max; @@ -1223,74 +1091,12 @@ rb_szqueue_initialize(VALUE self, VALUE vmax) return self; } -/* - * Document-method: Thread::SizedQueue#close - * call-seq: - * close - * - * Similar to Thread::Queue#close. - * - * The difference is behavior with waiting enqueuing threads. - * - * If there are waiting enqueuing threads, they are interrupted by - * raising ClosedQueueError('queue closed'). - */ -static VALUE -rb_szqueue_close(VALUE self) -{ - if (!queue_closed_p(self)) { - struct rb_szqueue *sq = szqueue_ptr(self); - - FL_SET(self, QUEUE_CLOSED); - wakeup_all(szqueue_waitq(sq)); - wakeup_all(szqueue_pushq(sq)); - } - return self; -} - -/* - * Document-method: Thread::SizedQueue#max - * - * Returns the maximum size of the queue. - */ - -static VALUE -rb_szqueue_max_get(VALUE self) -{ - return LONG2NUM(szqueue_ptr(self)->max); -} - -/* - * Document-method: Thread::SizedQueue#max= - * call-seq: max=(number) - * - * Sets the maximum size of the queue to the given +number+. - */ - -static VALUE -rb_szqueue_max_set(VALUE self, VALUE vmax) -{ - long max = NUM2LONG(vmax); - long diff = 0; - struct rb_szqueue *sq = szqueue_ptr(self); - - if (max <= 0) { - rb_raise(rb_eArgError, "queue size must be positive"); - } - if (max > sq->max) { - diff = max - sq->max; - } - sq->max = max; - sync_wakeup(szqueue_pushq(sq), diff); - return vmax; -} - static VALUE rb_szqueue_push(rb_execution_context_t *ec, VALUE self, VALUE object, VALUE non_block, VALUE timeout) { struct rb_szqueue *sq = szqueue_ptr(self); - if (queue_length(self, &sq->q) >= sq->max) { + if (sq->q.len >= sq->max) { if (RTEST(non_block)) { rb_raise(rb_eThreadError, "queue full"); } @@ -1301,12 +1107,11 @@ rb_szqueue_push(rb_execution_context_t *ec, VALUE self, VALUE object, VALUE non_ } rb_hrtime_t end = queue_timeout2hrtime(timeout); - while (queue_length(self, &sq->q) >= sq->max) { + while (sq->q.len >= sq->max) { if (queue_closed_p(self)) { raise_closed_queue_error(self); } else { - rb_execution_context_t *ec = GET_EC(); struct queue_waiter queue_waiter = { .w = {.self = self, .th = ec->thread_ptr, .fiber = nonblocking_fiber(ec->fiber_ptr)}, .as = {.sq = sq} @@ -1333,85 +1138,17 @@ rb_szqueue_push(rb_execution_context_t *ec, VALUE self, VALUE object, VALUE non_ } static VALUE -szqueue_do_pop(VALUE self, int should_block, VALUE timeout) +rb_szqueue_pop(rb_execution_context_t *ec, VALUE self, VALUE non_block, VALUE timeout) { struct rb_szqueue *sq = szqueue_ptr(self); - VALUE retval = queue_do_pop(self, &sq->q, should_block, timeout); + VALUE retval = queue_do_pop(ec, self, &sq->q, non_block, timeout); - if (queue_length(self, &sq->q) < sq->max) { + if (sq->q.len < sq->max) { wakeup_one(szqueue_pushq(sq)); } return retval; } -static VALUE -rb_szqueue_pop(rb_execution_context_t *ec, VALUE self, VALUE non_block, VALUE timeout) -{ - return szqueue_do_pop(self, !RTEST(non_block), timeout); -} - -/* - * Document-method: Thread::SizedQueue#clear - * - * Removes all objects from the queue. - */ - -static VALUE -rb_szqueue_clear(VALUE self) -{ - struct rb_szqueue *sq = szqueue_ptr(self); - - rb_ary_clear(check_array(self, sq->q.que)); - wakeup_all(szqueue_pushq(sq)); - return self; -} - -/* - * Document-method: Thread::SizedQueue#length - * call-seq: - * length - * size - * - * Returns the length of the queue. - */ - -static VALUE -rb_szqueue_length(VALUE self) -{ - struct rb_szqueue *sq = szqueue_ptr(self); - - return LONG2NUM(queue_length(self, &sq->q)); -} - -/* - * Document-method: Thread::SizedQueue#num_waiting - * - * Returns the number of threads waiting on the queue. - */ - -static VALUE -rb_szqueue_num_waiting(VALUE self) -{ - struct rb_szqueue *sq = szqueue_ptr(self); - - return INT2NUM(sq->q.num_waiting + sq->num_waiting_push); -} - -/* - * Document-method: Thread::SizedQueue#empty? - * call-seq: empty? - * - * Returns +true+ if the queue is empty. - */ - -static VALUE -rb_szqueue_empty_p(VALUE self) -{ - struct rb_szqueue *sq = szqueue_ptr(self); - - return RBOOL(queue_length(self, &sq->q) == 0); -} - /* ConditionalVariable */ struct rb_condvar { @@ -1419,34 +1156,6 @@ struct rb_condvar { rb_serial_t fork_gen; }; -/* - * 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 - * resource becomes available. - * - * Example: - * - * mutex = Thread::Mutex.new - * resource = Thread::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 - * } - * } - */ - static size_t condvar_memsize(const void *ptr) { @@ -1488,21 +1197,8 @@ condvar_alloc(VALUE klass) return obj; } -/* - * Document-method: ConditionVariable::new - * - * Creates a new condition variable instance. - */ - -static VALUE -rb_condvar_initialize(VALUE self) -{ - struct rb_condvar *cv = condvar_ptr(self); - ccan_list_head_init(&cv->waitq); - return self; -} - struct sleep_call { + rb_execution_context_t *ec; VALUE mutex; VALUE timeout; }; @@ -1513,161 +1209,72 @@ static VALUE do_sleep(VALUE args) { struct sleep_call *p = (struct sleep_call *)args; - return rb_funcallv(p->mutex, id_sleep, 1, &p->timeout); + if (CLASS_OF(p->mutex) == rb_cMutex) { + return rb_mut_sleep(p->ec, p->mutex, p->timeout); + } + else { + return rb_funcallv(p->mutex, id_sleep, 1, &p->timeout); + } } -/* - * 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 -rb_condvar_wait(int argc, VALUE *argv, VALUE self) +rb_condvar_wait(rb_execution_context_t *ec, VALUE self, VALUE mutex, VALUE timeout) { - rb_execution_context_t *ec = GET_EC(); - struct rb_condvar *cv = condvar_ptr(self); - struct sleep_call args; - - rb_scan_args(argc, argv, "11", &args.mutex, &args.timeout); + struct sleep_call args = { + .ec = ec, + .mutex = mutex, + .timeout = timeout, + }; struct sync_waiter sync_waiter = { - .self = args.mutex, + .self = mutex, .th = ec->thread_ptr, .fiber = nonblocking_fiber(ec->fiber_ptr) }; ccan_list_add_tail(&cv->waitq, &sync_waiter.node); - return rb_ensure(do_sleep, (VALUE)&args, delete_from_waitq, (VALUE)&sync_waiter); + return rb_ec_ensure(ec, do_sleep, (VALUE)&args, delete_from_waitq, (VALUE)&sync_waiter); } -/* - * Document-method: Thread::ConditionVariable#signal - * - * Wakes up the first thread in line waiting for this lock. - */ - static VALUE -rb_condvar_signal(VALUE self) +rb_condvar_signal(rb_execution_context_t *ec, VALUE self) { struct rb_condvar *cv = condvar_ptr(self); wakeup_one(&cv->waitq); return self; } -/* - * Document-method: Thread::ConditionVariable#broadcast - * - * Wakes up all threads waiting for this lock. - */ - static VALUE -rb_condvar_broadcast(VALUE self) +rb_condvar_broadcast(rb_execution_context_t *ec, VALUE self) { struct rb_condvar *cv = condvar_ptr(self); wakeup_all(&cv->waitq); return self; } -NORETURN(static VALUE undumpable(VALUE obj)); -/* :nodoc: */ -static VALUE -undumpable(VALUE obj) -{ - rb_raise(rb_eTypeError, "can't dump %"PRIsVALUE, rb_obj_class(obj)); - UNREACHABLE_RETURN(Qnil); -} - -static VALUE -define_thread_class(VALUE outer, const ID name, VALUE super) -{ - VALUE klass = rb_define_class_id_under(outer, name, super); - rb_const_set(rb_cObject, name, klass); - return klass; -} - static void Init_thread_sync(void) { -#undef rb_intern -#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, rb_intern(#name), rb_c##super) - /* Mutex */ - DEFINE_CLASS(Mutex, Object); + rb_cMutex = rb_define_class_id_under(rb_cThread, rb_intern("Mutex"), rb_cObject); rb_define_alloc_func(rb_cMutex, mutex_alloc); - rb_define_method(rb_cMutex, "initialize", mutex_initialize, 0); - rb_define_method(rb_cMutex, "locked?", rb_mutex_locked_p, 0); - rb_define_method(rb_cMutex, "try_lock", rb_mutex_trylock, 0); - rb_define_method(rb_cMutex, "lock", rb_mutex_lock, 0); - rb_define_method(rb_cMutex, "unlock", rb_mutex_unlock, 0); - rb_define_method(rb_cMutex, "sleep", mutex_sleep, -1); - rb_define_method(rb_cMutex, "synchronize", rb_mutex_synchronize_m, 0); - rb_define_method(rb_cMutex, "owned?", rb_mutex_owned_p, 0); /* Queue */ - DEFINE_CLASS(Queue, Object); + VALUE rb_cQueue = rb_define_class_id_under_no_pin(rb_cThread, rb_intern("Queue"), rb_cObject); rb_define_alloc_func(rb_cQueue, queue_alloc); rb_eClosedQueueError = rb_define_class("ClosedQueueError", rb_eStopIteration); - rb_define_method(rb_cQueue, "initialize", rb_queue_initialize, -1); - rb_undef_method(rb_cQueue, "initialize_copy"); - rb_define_method(rb_cQueue, "marshal_dump", undumpable, 0); - 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, "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, "size", "length"); - - DEFINE_CLASS(SizedQueue, Queue); + VALUE rb_cSizedQueue = rb_define_class_id_under_no_pin(rb_cThread, rb_intern("SizedQueue"), rb_cQueue); rb_define_alloc_func(rb_cSizedQueue, szqueue_alloc); - rb_define_method(rb_cSizedQueue, "initialize", rb_szqueue_initialize, 1); - 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, "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_method(rb_cSizedQueue, "freeze", rb_queue_freeze, 0); - rb_define_alias(rb_cSizedQueue, "size", "length"); - /* CVar */ - DEFINE_CLASS(ConditionVariable, Object); + VALUE rb_cConditionVariable = rb_define_class_id_under_no_pin(rb_cThread, rb_intern("ConditionVariable"), rb_cObject); rb_define_alloc_func(rb_cConditionVariable, condvar_alloc); id_sleep = rb_intern("sleep"); - rb_define_method(rb_cConditionVariable, "initialize", rb_condvar_initialize, 0); - rb_undef_method(rb_cConditionVariable, "initialize_copy"); - rb_define_method(rb_cConditionVariable, "marshal_dump", undumpable, 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_provide("thread.rb"); } |