#include "ccan/list/list.h"
static VALUE rb_cMutex, rb_cQueue, rb_cSizedQueue, rb_cConditionVariable;
static VALUE rb_eClosedQueueError;
struct sync_waiter {
rb_thread_t *th;
struct list_node node;
};
#define MUTEX_ALLOW_TRAP FL_USER1
static void
sync_wakeup(struct list_head *head, long max)
{
struct sync_waiter *cur = 0, *next;
list_for_each_safe(head, cur, next, node) {
list_del_init(&cur->node);
if (cur->th->status != THREAD_KILLED) {
rb_threadptr_interrupt(cur->th);
cur->th->status = THREAD_RUNNABLE;
if (--max == 0) return;
}
}
}
static void
wakeup_one(struct list_head *head)
{
sync_wakeup(head, 1);
}
static void
wakeup_all(struct list_head *head)
{
sync_wakeup(head, LONG_MAX);
}
typedef struct rb_mutex_struct {
rb_thread_t *th;
struct rb_mutex_struct *next_mutex;
struct list_head waitq;
} rb_mutex_t;
#if defined(HAVE_WORKING_FORK)
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);
#define mutex_mark NULL
static size_t
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++;
}
return n;
}
static void
mutex_free(void *ptr)
{
rb_mutex_t *mutex = ptr;
if (mutex->th) {
const char *err = rb_mutex_unlock_th(mutex, mutex->th);
if (err) rb_bug("%s", err);
}
ruby_xfree(ptr);
}
static size_t
mutex_memsize(const void *ptr)
{
return sizeof(rb_mutex_t);
}
static const rb_data_type_t mutex_data_type = {
"mutex",
{mutex_mark, mutex_free, mutex_memsize,},
0, 0, RUBY_TYPED_FREE_IMMEDIATELY
};
static rb_mutex_t *
mutex_ptr(VALUE obj)
{
rb_mutex_t *mutex;
TypedData_Get_Struct(obj, rb_mutex_t, &mutex_data_type, mutex);
return mutex;
}
VALUE
rb_obj_is_mutex(VALUE obj)
{
if (rb_typeddata_is_kind_of(obj, &mutex_data_type)) {
return Qtrue;
}
else {
return Qfalse;
}
}
static VALUE
mutex_alloc(VALUE klass)
{
VALUE obj;
rb_mutex_t *mutex;
obj = TypedData_Make_Struct(klass, rb_mutex_t, &mutex_data_type, mutex);
list_head_init(&mutex->waitq);
return obj;
}
static VALUE
mutex_initialize(VALUE self)
{
return self;
}
VALUE
rb_mutex_new(void)
{
return mutex_alloc(rb_cMutex);
}
VALUE
rb_mutex_locked_p(VALUE self)
{
rb_mutex_t *mutex = mutex_ptr(self);
return mutex->th ? Qtrue : Qfalse;
}
static void
mutex_locked(rb_thread_t *th, VALUE self)
{
rb_mutex_t *mutex = mutex_ptr(self);
if (th->keeping_mutexes) {
mutex->next_mutex = th->keeping_mutexes;
}
th->keeping_mutexes = mutex;
}
VALUE
rb_mutex_trylock(VALUE self)
{
rb_mutex_t *mutex = mutex_ptr(self);
VALUE locked = Qfalse;
if (mutex->th == 0) {
rb_thread_t *th = GET_THREAD();
mutex->th = th;
locked = Qtrue;
mutex_locked(th, self);
}
return locked;
}
static const rb_thread_t *patrol_thread = NULL;
static VALUE
do_mutex_lock(VALUE self, int interruptible_p)
{
rb_thread_t *th = GET_THREAD();
rb_mutex_t *mutex = mutex_ptr(self);
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");
}
if (rb_mutex_trylock(self) == Qfalse) {
struct sync_waiter w;
if (mutex->th == th) {
rb_raise(rb_eThreadError, "deadlock; recursive locking");
}
w.th = th;
while (mutex->th != th) {
enum rb_thread_status prev_status = th->status;
rb_hrtime_t *timeout = 0;
rb_hrtime_t rel = rb_msec2hrtime(100);
th->status = THREAD_STOPPED_FOREVER;
th->locking_mutex = self;
th->vm->sleeper++;
if ((vm_living_thread_num(th->vm) == th->vm->sleeper) &&
!patrol_thread) {
timeout = &rel;
patrol_thread = th;
}
list_add_tail(&mutex->waitq, &w.node);
native_sleep(th, timeout);
list_del(&w.node);
if (!mutex->th) {
mutex->th = th;
}
if (patrol_thread == th)
patrol_thread = NULL;
th->locking_mutex = Qfalse;
if (mutex->th && timeout && !RUBY_VM_INTERRUPTED(th->ec)) {
rb_check_deadlock(th->vm);
}
if (th->status == THREAD_STOPPED_FOREVER) {
th->status = prev_status;
}
th->vm->sleeper--;
if (mutex->th == th) mutex_locked(th, self);
if (interruptible_p) {
RUBY_VM_CHECK_INTS_BLOCKING(th->ec);
if (!mutex->th) {
mutex->th = th;
mutex_locked(th, self);
}
}
}
}
return self;
}
static VALUE
mutex_lock_uninterruptible(VALUE self)
{
return do_mutex_lock(self, 0);
}
VALUE
rb_mutex_lock(VALUE self)
{
return do_mutex_lock(self, 1);
}
VALUE
rb_mutex_owned_p(VALUE self)
{
VALUE owned = Qfalse;
rb_thread_t *th = GET_THREAD();
rb_mutex_t *mutex = mutex_ptr(self);
if (mutex->th == th)
owned = Qtrue;
return owned;
}
static const char *
rb_mutex_unlock_th(rb_mutex_t *mutex, rb_thread_t *th)
{
const char *err = NULL;
if (mutex->th == 0) {
err = "Attempt to unlock a mutex which is not locked";
}
else if (mutex->th != th) {
err = "Attempt to unlock a mutex which is locked by another thread";
}
else {
struct sync_waiter *cur = 0, *next;
rb_mutex_t **th_mutex = &th->keeping_mutexes;
mutex->th = 0;
list_for_each_safe(&mutex->waitq, cur, next, node) {
list_del_init(&cur->node);
switch (cur->th->status) {
case THREAD_RUNNABLE:
case THREAD_STOPPED_FOREVER:
rb_threadptr_interrupt(cur->th);
goto found;
case THREAD_STOPPED:
rb_bug("unexpected THREAD_STOPPED");
case THREAD_KILLED:
rb_bug("unexpected THREAD_KILLED");
continue;
}
}
found:
while (*th_mutex != mutex) {
th_mutex = &(*th_mutex)->next_mutex;
}
*th_mutex = mutex->next_mutex;
mutex->next_mutex = NULL;
}
return err;
}
VALUE
rb_mutex_unlock(VALUE self)
{
const char *err;
rb_mutex_t *mutex = mutex_ptr(self);
err = rb_mutex_unlock_th(mutex, GET_THREAD());
if (err) rb_raise(rb_eThreadError, "%s", err);
return self;
}
#if defined(HAVE_WORKING_FORK)
static void
rb_mutex_abandon_keeping_mutexes(rb_thread_t *th)
{
rb_mutex_abandon_all(th->keeping_mutexes);
th->keeping_mutexes = NULL;
}
static void
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);
th->locking_mutex = Qfalse;
}
}
static void
rb_mutex_abandon_all(rb_mutex_t *mutexes)
{
rb_mutex_t *mutex;
while (mutexes) {
mutex = mutexes;
mutexes = mutex->next_mutex;
mutex->th = 0;
mutex->next_mutex = 0;
list_head_init(&mutex->waitq);
}
}
#endif
static VALUE
rb_mutex_sleep_forever(VALUE time)
{
rb_thread_sleep_deadly_allow_spurious_wakeup();
return Qnil;
}
static VALUE
rb_mutex_wait_for(VALUE time)
{
rb_hrtime_t *rel = (rb_hrtime_t *)time;
sleep_hrtime(GET_THREAD(), *rel, 0);
return Qnil;
}
VALUE
rb_mutex_sleep(VALUE self, VALUE timeout)
{
time_t beg, end;
struct timeval t;
if (!NIL_P(timeout)) {
t = rb_time_interval(timeout);
}
rb_mutex_unlock(self);
beg = time(0);
if (NIL_P(timeout)) {
rb_ensure(rb_mutex_sleep_forever, Qnil, mutex_lock_uninterruptible, self);
}
else {
rb_hrtime_t rel = rb_timeval2hrtime(&t);
rb_ensure(rb_mutex_wait_for, (VALUE)&rel,
mutex_lock_uninterruptible, self);
}
RUBY_VM_CHECK_INTS_BLOCKING(GET_EC());
end = time(0) - beg;
return INT2FIX(end);
}
static VALUE
mutex_sleep(int argc, VALUE *argv, VALUE self)
{
VALUE timeout;
timeout = rb_check_arity(argc, 0, 1) ? argv[0] : Qnil;
return rb_mutex_sleep(self, timeout);
}
VALUE
rb_mutex_synchronize(VALUE mutex, VALUE (*func)(VALUE arg), VALUE arg)
{
rb_mutex_lock(mutex);
return rb_ensure(func, arg, rb_mutex_unlock, mutex);
}
static VALUE
rb_mutex_synchronize_m(VALUE self, VALUE args)
{
if (!rb_block_given_p()) {
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)
{
Check_TypedStruct(self, &mutex_data_type);
if (val)
FL_SET_RAW(self, MUTEX_ALLOW_TRAP);
else
FL_UNSET_RAW(self, MUTEX_ALLOW_TRAP);
}
#define queue_waitq(q) UNALIGNED_MEMBER_PTR(q, waitq)
PACKED_STRUCT_UNALIGNED(struct rb_queue {
struct list_head waitq;
rb_serial_t fork_gen;
const VALUE que;
int num_waiting;
});
#define szqueue_waitq(sq) UNALIGNED_MEMBER_PTR(sq, q.waitq)
#define szqueue_pushq(sq) UNALIGNED_MEMBER_PTR(sq, pushq)
PACKED_STRUCT_UNALIGNED(struct rb_szqueue {
struct rb_queue q;
int num_waiting_push;
struct list_head pushq;
long max;
});
static void
queue_mark(void *ptr)
{
struct rb_queue *q = ptr;
rb_gc_mark(q->que);
}
static size_t
queue_memsize(const void *ptr)
{
return sizeof(struct rb_queue);
}
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
};
static VALUE
queue_alloc(VALUE klass)
{
VALUE obj;
struct rb_queue *q;
obj = TypedData_Make_Struct(klass, struct rb_queue, &queue_data_type, q);
list_head_init(queue_waitq(q));
return obj;
}
static int
queue_fork_check(struct rb_queue *q)
{
rb_serial_t fork_gen = GET_VM()->fork_gen;
if (q->fork_gen == fork_gen) {
return 0;
}
q->fork_gen = fork_gen;
list_head_init(queue_waitq(q));
q->num_waiting = 0;
return 1;
}
static struct rb_queue *
queue_ptr(VALUE obj)
{
struct rb_queue *q;
TypedData_Get_Struct(obj, struct rb_queue, &queue_data_type, q);
queue_fork_check(q);
return q;
}
#define QUEUE_CLOSED FL_USER5
static void
szqueue_mark(void *ptr)
{
struct rb_szqueue *sq = ptr;
queue_mark(&sq->q);
}
static size_t
szqueue_memsize(const void *ptr)
{
return sizeof(struct rb_szqueue);
}
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
};
static VALUE
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));
return obj;
}
static struct rb_szqueue *
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)) {
list_head_init(szqueue_pushq(sq));
sq->num_waiting_push = 0;
}
return sq;
}
static VALUE
ary_buf_new(void)
{
return rb_ary_tmp_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)
{
return RARRAY_LEN(check_array(self, q->que));
}
static int
queue_closed_p(VALUE self)
{
return FL_TEST_RAW(self, QUEUE_CLOSED) != 0;
}
NORETURN(static void raise_closed_queue_error(VALUE self));
static void
raise_closed_queue_error(VALUE self)
{
rb_raise(rb_eClosedQueueError, "queue closed");
}
static VALUE
queue_closed_result(VALUE self, struct rb_queue *q)
{
assert(queue_length(self, q) == 0);
return Qnil;
}
static VALUE
rb_queue_initialize(VALUE self)
{
struct rb_queue *q = queue_ptr(self);
RB_OBJ_WRITE(self, &q->que, ary_buf_new());
list_head_init(queue_waitq(q));
return self;
}
static VALUE
queue_do_push(VALUE self, struct rb_queue *q, VALUE obj)
{
if (queue_closed_p(self)) {
raise_closed_queue_error(self);
}
rb_ary_push(check_array(self, q->que), obj);
wakeup_one(queue_waitq(q));
return self;
}
static VALUE
rb_queue_close(VALUE self)
{
struct rb_queue *q = queue_ptr(self);
if (!queue_closed_p(self)) {
FL_SET(self, QUEUE_CLOSED);
wakeup_all(queue_waitq(q));
}
return self;
}
static VALUE
rb_queue_closed_p(VALUE self)
{
return queue_closed_p(self) ? Qtrue : Qfalse;
}
static VALUE
rb_queue_push(VALUE self, VALUE obj)
{
return queue_do_push(self, queue_ptr(self), obj);
}
static VALUE
queue_sleep(VALUE arg)
{
rb_thread_sleep_deadly_allow_spurious_wakeup();
return Qnil;
}
struct queue_waiter {
struct sync_waiter w;
union {
struct rb_queue *q;
struct rb_szqueue *sq;
} as;
};
static VALUE
queue_sleep_done(VALUE p)
{
struct queue_waiter *qw = (struct queue_waiter *)p;
list_del(&qw->w.node);
qw->as.q->num_waiting--;
return Qfalse;
}
static VALUE
szqueue_sleep_done(VALUE p)
{
struct queue_waiter *qw = (struct queue_waiter *)p;
list_del(&qw->w.node);
qw->as.sq->num_waiting_push--;
return Qfalse;
}
static VALUE
queue_do_pop(VALUE self, struct rb_queue *q, int should_block)
{
check_array(self, q->que);
while (RARRAY_LEN(q->que) == 0) {
if (!should_block) {
rb_raise(rb_eThreadError, "queue empty");
}
else if (queue_closed_p(self)) {
return queue_closed_result(self, q);
}
else {
struct queue_waiter qw;
assert(RARRAY_LEN(q->que) == 0);
assert(queue_closed_p(self) == 0);
qw.w.th = GET_THREAD();
qw.as.q = q;
list_add_tail(&qw.as.q->waitq, &qw.w.node);
qw.as.q->num_waiting++;
rb_ensure(queue_sleep, self, queue_sleep_done, (VALUE)&qw);
}
}
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;
}
static VALUE
rb_queue_pop(int argc, VALUE *argv, VALUE self)
{
int should_block = queue_pop_should_block(argc, argv);
return queue_do_pop(self, queue_ptr(self), should_block);
}
static VALUE
rb_queue_empty_p(VALUE self)
{
return queue_length(self, queue_ptr(self)) == 0 ? Qtrue : Qfalse;
}
static VALUE
rb_queue_clear(VALUE self)
{
struct rb_queue *q = queue_ptr(self);
rb_ary_clear(check_array(self, q->que));
return self;
}
static VALUE
rb_queue_length(VALUE self)
{
return LONG2NUM(queue_length(self, queue_ptr(self)));
}
static VALUE
rb_queue_num_waiting(VALUE self)
{
struct rb_queue *q = queue_ptr(self);
return INT2NUM(q->num_waiting);
}
static VALUE
rb_szqueue_initialize(VALUE self, VALUE vmax)
{
long max;
struct rb_szqueue *sq = szqueue_ptr(self);
max = NUM2LONG(vmax);
if (max <= 0) {
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));
sq->max = max;
return self;
}
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;
}
static VALUE
rb_szqueue_max_get(VALUE self)
{
return LONG2NUM(szqueue_ptr(self)->max);
}
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 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;
}
static VALUE
rb_szqueue_push(int argc, VALUE *argv, VALUE self)
{
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) {
rb_raise(rb_eThreadError, "queue full");
}
else if (queue_closed_p(self)) {
goto closed;
}
else {
struct queue_waiter qw;
struct list_head *pushq = szqueue_pushq(sq);
qw.w.th = GET_THREAD();
qw.as.sq = sq;
list_add_tail(pushq, &qw.w.node);
sq->num_waiting_push++;
rb_ensure(queue_sleep, self, szqueue_sleep_done, (VALUE)&qw);
}
}
if (queue_closed_p(self)) {
closed:
raise_closed_queue_error(self);
}
return queue_do_push(self, &sq->q, argv[0]);
}
static VALUE
szqueue_do_pop(VALUE self, int should_block)
{
struct rb_szqueue *sq = szqueue_ptr(self);
VALUE retval = queue_do_pop(self, &sq->q, should_block);
if (queue_length(self, &sq->q) < sq->max) {
wakeup_one(szqueue_pushq(sq));
}
return retval;
}
static VALUE
rb_szqueue_pop(int argc, VALUE *argv, VALUE self)
{
int should_block = queue_pop_should_block(argc, argv);
return szqueue_do_pop(self, should_block);
}
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;
}
static VALUE
rb_szqueue_length(VALUE self)
{
struct rb_szqueue *sq = szqueue_ptr(self);
return LONG2NUM(queue_length(self, &sq->q));
}
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);
}
static VALUE
rb_szqueue_empty_p(VALUE self)
{
struct rb_szqueue *sq = szqueue_ptr(self);
return queue_length(self, &sq->q) == 0 ? Qtrue : Qfalse;
}
struct rb_condvar {
struct list_head waitq;
rb_serial_t fork_gen;
};
static size_t
condvar_memsize(const void *ptr)
{
return sizeof(struct rb_condvar);
}
static const rb_data_type_t cv_data_type = {
"condvar",
{0, RUBY_TYPED_DEFAULT_FREE, condvar_memsize,},
0, 0, RUBY_TYPED_FREE_IMMEDIATELY|RUBY_TYPED_WB_PROTECTED
};
static struct rb_condvar *
condvar_ptr(VALUE self)
{
struct rb_condvar *cv;
rb_serial_t fork_gen = GET_VM()->fork_gen;
TypedData_Get_Struct(self, struct rb_condvar, &cv_data_type, cv);
if (cv->fork_gen != fork_gen) {
cv->fork_gen = fork_gen;
list_head_init(&cv->waitq);
}
return cv;
}
static VALUE
condvar_alloc(VALUE klass)
{
struct rb_condvar *cv;
VALUE obj;
obj = TypedData_Make_Struct(klass, struct rb_condvar, &cv_data_type, cv);
list_head_init(&cv->waitq);
return obj;
}
static VALUE
rb_condvar_initialize(VALUE self)
{
struct rb_condvar *cv = condvar_ptr(self);
list_head_init(&cv->waitq);
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_funcallv(p->mutex, id_sleep, 1, &p->timeout);
}
static VALUE
delete_from_waitq(struct sync_waiter *w)
{
list_del(&w->node);
return Qnil;
}
static VALUE
rb_condvar_wait(int argc, VALUE *argv, VALUE self)
{
struct rb_condvar *cv = condvar_ptr(self);
struct sleep_call args;
struct sync_waiter w;
rb_scan_args(argc, argv, "11", &args.mutex, &args.timeout);
w.th = GET_THREAD();
list_add_tail(&cv->waitq, &w.node);
rb_ensure(do_sleep, (VALUE)&args, delete_from_waitq, (VALUE)&w);
return self;
}
static VALUE
rb_condvar_signal(VALUE self)
{
struct rb_condvar *cv = condvar_ptr(self);
wakeup_one(&cv->waitq);
return self;
}
static VALUE
rb_condvar_broadcast(VALUE self)
{
struct rb_condvar *cv = condvar_ptr(self);
wakeup_all(&cv->waitq);
return self;
}
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 char *name, VALUE super)
{
VALUE klass = rb_define_class_under(outer, name, super);
rb_define_const(rb_cObject, name, klass);
return klass;
}
static void
Init_thread_sync(void)
{
#undef rb_intern
#if 0
rb_cMutex = rb_define_class("Mutex", rb_cObject);
rb_cConditionVariable = rb_define_class("ConditionVariable", rb_cObject);
rb_cQueue = rb_define_class("Queue", rb_cObject);
rb_cSizedQueue = rb_define_class("SizedQueue", rb_cObject);
#endif
#define DEFINE_CLASS(name, super) \
rb_c##name = define_thread_class(rb_cThread, #name, rb_c##super)
DEFINE_CLASS(Mutex, Object);
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);
DEFINE_CLASS(Queue, Object);
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, 0);
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, "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_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);
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, "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");
DEFINE_CLASS(ConditionVariable, Object);
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");
}