2#include "ccan/list/list.h"
5static VALUE rb_cMutex, rb_cQueue, rb_cSizedQueue, rb_cConditionVariable;
6static VALUE rb_eClosedQueueError;
12 struct ccan_list_head waitq;
20 struct ccan_list_node node;
26 if (rb_fiberptr_blocking(fiber)) {
39#define MUTEX_ALLOW_TRAP FL_USER1
42sync_wakeup(
struct ccan_list_head *head,
long max)
44 RUBY_DEBUG_LOG(
"max:%ld", max);
48 ccan_list_for_each_safe(head, cur, next, node) {
49 ccan_list_del_init(&cur->node);
51 if (cur->th->status != THREAD_KILLED) {
52 if (cur->th->scheduler !=
Qnil && cur->fiber) {
56 RUBY_DEBUG_LOG(
"target_th:%u", rb_th_serial(cur->th));
57 rb_threadptr_interrupt(cur->th);
58 cur->th->status = THREAD_RUNNABLE;
61 if (--max == 0)
return;
67wakeup_one(
struct ccan_list_head *head)
73wakeup_all(
struct ccan_list_head *head)
75 sync_wakeup(head, LONG_MAX);
78#if defined(HAVE_WORKING_FORK)
79static void rb_mutex_abandon_all(
rb_mutex_t *mutexes);
80static void rb_mutex_abandon_keeping_mutexes(
rb_thread_t *th);
81static void rb_mutex_abandon_locking_mutex(
rb_thread_t *th);
109#define mutex_mark ((void(*)(void*))0)
117 ccan_list_for_each(&mutex->waitq, w, node) {
132 const char *err = rb_mutex_unlock_th(mutex, rb_fiber_threadptr(mutex->fiber), mutex->fiber);
133 if (err) rb_bug(
"%s", err);
139mutex_memsize(
const void *ptr)
146 {mutex_mark, mutex_free, mutex_memsize,},
147 0, 0, RUBY_TYPED_WB_PROTECTED | RUBY_TYPED_FREE_IMMEDIATELY
161rb_obj_is_mutex(
VALUE obj)
167mutex_alloc(
VALUE klass)
174 ccan_list_head_init(&mutex->waitq);
185mutex_initialize(
VALUE self)
193 return mutex_alloc(rb_cMutex);
207 return RBOOL(mutex->fiber);
213 if (thread->keeping_mutexes) {
214 mutex->next_mutex = thread->keeping_mutexes;
217 thread->keeping_mutexes = mutex;
223 rb_mutex_t **keeping_mutexes = &thread->keeping_mutexes;
225 while (*keeping_mutexes && *keeping_mutexes != mutex) {
227 keeping_mutexes = &(*keeping_mutexes)->next_mutex;
230 if (*keeping_mutexes) {
231 *keeping_mutexes = mutex->next_mutex;
232 mutex->next_mutex = NULL;
241 thread_mutex_insert(th, mutex);
256 if (mutex->fiber == 0) {
257 RUBY_DEBUG_LOG(
"%p ok", mutex);
261 mutex->fiber = fiber;
263 mutex_locked(th, self);
267 RUBY_DEBUG_LOG(
"%p ng", mutex);
275 return RBOOL(mutex->fiber == fiber);
279call_rb_fiber_scheduler_block(
VALUE mutex)
285delete_from_waitq(
VALUE value)
296do_mutex_lock(
VALUE self,
int interruptible_p)
306 th->ec->interrupt_mask & TRAP_INTERRUPT_MASK) {
311 if (mutex->fiber == fiber) {
315 while (mutex->fiber != fiber) {
316 VM_ASSERT(mutex->fiber != NULL);
319 if (scheduler !=
Qnil) {
323 .fiber = nonblocking_fiber(fiber)
326 ccan_list_add_tail(&mutex->waitq, &
sync_waiter.node);
331 mutex->fiber = fiber;
335 if (!th->vm->thread_ignore_deadlock && rb_fiber_threadptr(mutex->fiber) == th) {
336 rb_raise(
rb_eThreadError,
"deadlock; lock already owned by another fiber belonging to the same thread");
342 .fiber = nonblocking_fiber(fiber),
345 RUBY_DEBUG_LOG(
"%p wait", mutex);
356 enum rb_thread_status prev_status = th->status;
357 th->status = THREAD_STOPPED_FOREVER;
358 rb_ractor_sleeper_threads_inc(th->ractor);
359 rb_check_deadlock(th->ractor);
361 th->locking_mutex = self;
363 ccan_list_add_tail(&mutex->waitq, &
sync_waiter.node);
365 native_sleep(th, NULL);
371 mutex->fiber = fiber;
374 rb_ractor_sleeper_threads_dec(th->ractor);
375 th->status = prev_status;
376 th->locking_mutex =
Qfalse;
378 RUBY_DEBUG_LOG(
"%p wakeup", mutex);
381 if (interruptible_p) {
384 if (mutex->fiber == fiber) mutex->fiber = 0;
385 RUBY_VM_CHECK_INTS_BLOCKING(th->ec);
387 mutex->fiber = fiber;
392 if (RUBY_VM_INTERRUPTED(th->ec)) {
394 if (saved_ints == 0) {
395 saved_ints = threadptr_get_interrupts(th);
399 threadptr_get_interrupts(th);
405 if (saved_ints) th->ec->interrupt_flag = saved_ints;
406 if (mutex->fiber == fiber) mutex_locked(th, self);
409 RUBY_DEBUG_LOG(
"%p locked", mutex);
412 if (mutex_owned_p(fiber, mutex) ==
Qfalse) rb_bug(
"do_mutex_lock: mutex is not owned.");
418mutex_lock_uninterruptible(
VALUE self)
420 return do_mutex_lock(self, 0);
433 return do_mutex_lock(self, 1);
443rb_mutex_owned_p(
VALUE self)
448 return mutex_owned_p(fiber, mutex);
454 RUBY_DEBUG_LOG(
"%p", mutex);
456 if (mutex->fiber == 0) {
457 return "Attempt to unlock a mutex which is not locked";
459 else if (mutex->fiber != fiber) {
460 return "Attempt to unlock a mutex which is locked by another thread/fiber";
466 thread_mutex_remove(th, mutex);
468 ccan_list_for_each_safe(&mutex->waitq, cur, next, node) {
469 ccan_list_del_init(&cur->node);
471 if (cur->th->scheduler !=
Qnil && cur->fiber) {
476 switch (cur->th->status) {
477 case THREAD_RUNNABLE:
478 case THREAD_STOPPED_FOREVER:
479 RUBY_DEBUG_LOG(
"wakeup th:%u", rb_th_serial(cur->th));
480 rb_threadptr_interrupt(cur->th);
483 rb_bug(
"unexpected THREAD_STOPPED");
486 rb_bug(
"unexpected THREAD_KILLED");
510 err = rb_mutex_unlock_th(mutex, th, GET_EC()->fiber_ptr);
516#if defined(HAVE_WORKING_FORK)
520 rb_mutex_abandon_all(th->keeping_mutexes);
521 th->keeping_mutexes = NULL;
527 if (th->locking_mutex) {
528 rb_mutex_t *mutex = mutex_ptr(th->locking_mutex);
530 ccan_list_head_init(&mutex->waitq);
531 th->locking_mutex =
Qfalse;
542 mutexes = mutex->next_mutex;
544 mutex->next_mutex = 0;
545 ccan_list_head_init(&mutex->waitq);
556mutex_sleep_begin(
VALUE _arguments)
559 VALUE timeout = arguments->timeout;
563 if (scheduler !=
Qnil) {
567 if (
NIL_P(timeout)) {
568 rb_thread_sleep_deadly_allow_spurious_wakeup(arguments->self,
Qnil, 0);
572 rb_hrtime_t relative_timeout = rb_timeval2hrtime(&timeout_value);
574 woken = RBOOL(sleep_hrtime(GET_THREAD(), relative_timeout, 0));
584 if (!
NIL_P(timeout)) {
590 time_t beg = time(0);
597 VALUE woken =
rb_ensure(mutex_sleep_begin, (
VALUE)&arguments, mutex_lock_uninterruptible, self);
599 RUBY_VM_CHECK_INTS_BLOCKING(GET_EC());
600 if (!woken)
return Qnil;
601 time_t end = time(0) - beg;
602 return TIMET2NUM(end);
653rb_mutex_synchronize_m(
VALUE self)
663rb_mutex_allow_trap(
VALUE self,
int val)
675#define queue_waitq(q) UNALIGNED_MEMBER_PTR(q, waitq)
676#define queue_list(q) UNALIGNED_MEMBER_PTR(q, que)
677RBIMPL_ATTR_PACKED_STRUCT_UNALIGNED_BEGIN()
679 struct ccan_list_head waitq;
680 rb_serial_t fork_gen;
683} RBIMPL_ATTR_PACKED_STRUCT_UNALIGNED_END();
685#define szqueue_waitq(sq) UNALIGNED_MEMBER_PTR(sq, q.waitq)
686#define szqueue_list(sq) UNALIGNED_MEMBER_PTR(sq, q.que)
687#define szqueue_pushq(sq) UNALIGNED_MEMBER_PTR(sq, pushq)
688RBIMPL_ATTR_PACKED_STRUCT_UNALIGNED_BEGIN()
691 int num_waiting_push;
692 struct ccan_list_head pushq;
694} RBIMPL_ATTR_PACKED_STRUCT_UNALIGNED_END();
706queue_memsize(
const void *ptr)
714 0, 0, RUBY_TYPED_FREE_IMMEDIATELY|RUBY_TYPED_WB_PROTECTED
718queue_alloc(
VALUE klass)
724 ccan_list_head_init(queue_waitq(q));
731 rb_serial_t fork_gen = GET_VM()->fork_gen;
733 if (q->fork_gen == fork_gen) {
737 q->fork_gen = fork_gen;
738 ccan_list_head_init(queue_waitq(q));
754#define QUEUE_CLOSED FL_USER5
757queue_timeout2hrtime(
VALUE timeout)
759 if (
NIL_P(timeout)) {
760 return (rb_hrtime_t)0;
764 rel = rb_sec2hrtime(NUM2TIMET(timeout));
769 return rb_hrtime_add(rel, rb_hrtime_now());
773szqueue_mark(
void *ptr)
781szqueue_memsize(
const void *ptr)
789 0, 0, RUBY_TYPED_FREE_IMMEDIATELY|RUBY_TYPED_WB_PROTECTED
793szqueue_alloc(
VALUE klass)
797 &szqueue_data_type, sq);
798 ccan_list_head_init(szqueue_waitq(sq));
799 ccan_list_head_init(szqueue_pushq(sq));
804szqueue_ptr(
VALUE obj)
809 if (queue_fork_check(&sq->q)) {
810 ccan_list_head_init(szqueue_pushq(sq));
811 sq->num_waiting_push = 0;
827 rb_raise(
rb_eTypeError,
"%+"PRIsVALUE
" not initialized", obj);
839queue_closed_p(
VALUE self)
851NORETURN(
static void raise_closed_queue_error(
VALUE self));
854raise_closed_queue_error(
VALUE self)
856 rb_raise(rb_eClosedQueueError,
"queue closed");
927rb_queue_initialize(
int argc,
VALUE *argv,
VALUE self)
930 struct rb_queue *q = queue_ptr(self);
931 if ((argc =
rb_scan_args(argc, argv,
"01", &initial)) == 1) {
932 initial = rb_to_array(initial);
935 ccan_list_head_init(queue_waitq(q));
945 if (queue_closed_p(self)) {
946 raise_closed_queue_error(self);
949 wakeup_one(queue_waitq(q));
987rb_queue_close(
VALUE self)
989 struct rb_queue *q = queue_ptr(self);
991 if (!queue_closed_p(self)) {
992 FL_SET(self, QUEUE_CLOSED);
994 wakeup_all(queue_waitq(q));
1008rb_queue_closed_p(
VALUE self)
1010 return RBOOL(queue_closed_p(self));
1026 return queue_do_push(self, queue_ptr(self), obj);
1030queue_sleep(
VALUE _args)
1033 rb_thread_sleep_deadly_allow_spurious_wakeup(args->self, args->timeout, args->end);
1046queue_sleep_done(
VALUE p)
1050 ccan_list_del(&qw->w.node);
1051 qw->as.q->num_waiting--;
1057szqueue_sleep_done(
VALUE p)
1061 ccan_list_del(&qw->w.node);
1062 qw->as.sq->num_waiting_push--;
1070 check_array(self, q->que);
1072 if (!should_block) {
1081 rb_hrtime_t end = queue_timeout2hrtime(timeout);
1083 if (queue_closed_p(self)) {
1084 return queue_closed_result(self, q);
1093 .w = {.self = self, .th = ec->thread_ptr, .fiber = nonblocking_fiber(ec->fiber_ptr)},
1097 struct ccan_list_head *waitq = queue_waitq(q);
1109 if (!
NIL_P(timeout) && (rb_hrtime_now() >= end))
1120 return queue_do_pop(self, queue_ptr(self), !
RTEST(non_block), timeout);
1131rb_queue_empty_p(
VALUE self)
1133 return RBOOL(queue_length(self, queue_ptr(self)) == 0);
1143rb_queue_clear(
VALUE self)
1145 struct rb_queue *q = queue_ptr(self);
1161rb_queue_length(
VALUE self)
1163 return LONG2NUM(queue_length(self, queue_ptr(self)));
1166NORETURN(
static VALUE rb_queue_freeze(
VALUE self));
1176rb_queue_freeze(
VALUE self)
1178 rb_raise(
rb_eTypeError,
"cannot freeze " "%+"PRIsVALUE, self);
1189rb_queue_num_waiting(
VALUE self)
1191 struct rb_queue *q = queue_ptr(self);
1193 return INT2NUM(q->num_waiting);
1220 rb_raise(rb_eArgError,
"queue size must be positive");
1224 ccan_list_head_init(szqueue_waitq(sq));
1225 ccan_list_head_init(szqueue_pushq(sq));
1244rb_szqueue_close(
VALUE self)
1246 if (!queue_closed_p(self)) {
1249 FL_SET(self, QUEUE_CLOSED);
1250 wakeup_all(szqueue_waitq(sq));
1251 wakeup_all(szqueue_pushq(sq));
1263rb_szqueue_max_get(
VALUE self)
1265 return LONG2NUM(szqueue_ptr(self)->max);
1283 rb_raise(rb_eArgError,
"queue size must be positive");
1285 if (max > sq->max) {
1286 diff = max - sq->max;
1289 sync_wakeup(szqueue_pushq(sq), diff);
1298 if (queue_length(self, &sq->q) >= sq->max) {
1299 if (
RTEST(non_block)) {
1308 rb_hrtime_t end = queue_timeout2hrtime(timeout);
1309 while (queue_length(self, &sq->q) >= sq->max) {
1310 if (queue_closed_p(self)) {
1311 raise_closed_queue_error(self);
1316 .w = {.self = self, .th = ec->thread_ptr, .fiber = nonblocking_fiber(ec->fiber_ptr)},
1320 struct ccan_list_head *pushq = szqueue_pushq(sq);
1323 sq->num_waiting_push++;
1331 if (!
NIL_P(timeout) && rb_hrtime_now() >= end) {
1337 return queue_do_push(self, &sq->q,
object);
1341szqueue_do_pop(
VALUE self,
int should_block,
VALUE timeout)
1344 VALUE retval = queue_do_pop(self, &sq->q, should_block, timeout);
1346 if (queue_length(self, &sq->q) < sq->max) {
1347 wakeup_one(szqueue_pushq(sq));
1355 return szqueue_do_pop(self, !
RTEST(non_block), timeout);
1365rb_szqueue_clear(
VALUE self)
1370 wakeup_all(szqueue_pushq(sq));
1384rb_szqueue_length(
VALUE self)
1388 return LONG2NUM(queue_length(self, &sq->q));
1398rb_szqueue_num_waiting(
VALUE self)
1402 return INT2NUM(sq->q.num_waiting + sq->num_waiting_push);
1413rb_szqueue_empty_p(
VALUE self)
1417 return RBOOL(queue_length(self, &sq->q) == 0);
1423 struct ccan_list_head waitq;
1424 rb_serial_t fork_gen;
1510condvar_memsize(
const void *ptr)
1518 0, 0, RUBY_TYPED_FREE_IMMEDIATELY|RUBY_TYPED_WB_PROTECTED
1522condvar_ptr(
VALUE self)
1525 rb_serial_t fork_gen = GET_VM()->fork_gen;
1530 if (cv->fork_gen != fork_gen) {
1531 cv->fork_gen = fork_gen;
1532 ccan_list_head_init(&cv->waitq);
1539condvar_alloc(
VALUE klass)
1545 ccan_list_head_init(&cv->waitq);
1557rb_condvar_initialize(
VALUE self)
1560 ccan_list_head_init(&cv->waitq);
1575 return rb_funcallv(p->mutex, id_sleep, 1, &p->timeout);
1593rb_condvar_wait(
int argc,
VALUE *argv,
VALUE self)
1600 rb_scan_args(argc, argv,
"11", &args.mutex, &args.timeout);
1604 .th = ec->thread_ptr,
1605 .fiber = nonblocking_fiber(ec->fiber_ptr)
1608 ccan_list_add_tail(&cv->waitq, &
sync_waiter.node);
1619rb_condvar_signal(
VALUE self)
1622 wakeup_one(&cv->waitq);
1633rb_condvar_broadcast(
VALUE self)
1636 wakeup_all(&cv->waitq);
1643undumpable(
VALUE obj)
1650define_thread_class(
VALUE outer,
const ID name,
VALUE super)
1658Init_thread_sync(
void)
1661#if defined(TEACH_RDOC) && TEACH_RDOC == 42
1668#define DEFINE_CLASS(name, super) \
1669 rb_c##name = define_thread_class(rb_cThread, rb_intern(#name), rb_c##super)
1672 DEFINE_CLASS(Mutex, Object);
1684 DEFINE_CLASS(Queue, Object);
1705 DEFINE_CLASS(SizedQueue, Queue);
1715 rb_define_method(rb_cSizedQueue,
"num_waiting", rb_szqueue_num_waiting, 0);
1719 DEFINE_CLASS(ConditionVariable, Object);
1722 id_sleep = rb_intern(
"sleep");
1724 rb_define_method(rb_cConditionVariable,
"initialize", rb_condvar_initialize, 0);
1729 rb_define_method(rb_cConditionVariable,
"broadcast", rb_condvar_broadcast, 0);
1734#include "thread_sync.rbinc"
#define RUBY_ASSERT(...)
Asserts that the given expression is truthy if and only if RUBY_DEBUG is truthy.
std::atomic< unsigned > rb_atomic_t
Type that is eligible for atomic operations.
#define rb_define_method(klass, mid, func, arity)
Defines klass#mid.
VALUE rb_define_class(const char *name, VALUE super)
Defines a top-level class.
VALUE rb_define_class_under(VALUE outer, const char *name, VALUE super)
Defines a class under the namespace of outer.
VALUE rb_define_class_id_under(VALUE outer, ID id, VALUE super)
Identical to rb_define_class_under(), except it takes the name in ID instead of C's string.
void rb_define_alias(VALUE klass, const char *name1, const char *name2)
Defines an alias of a method.
void rb_undef_method(VALUE klass, const char *name)
Defines an undef of a method.
int rb_scan_args(int argc, const VALUE *argv, const char *fmt,...)
Retrieves argument from argc and argv to given VALUE references according to the format string.
int rb_block_given_p(void)
Determines if the current method is given a block.
#define FL_UNSET_RAW
Old name of RB_FL_UNSET_RAW.
#define Qundef
Old name of RUBY_Qundef.
#define INT2FIX
Old name of RB_INT2FIX.
#define UNREACHABLE_RETURN
Old name of RBIMPL_UNREACHABLE_RETURN.
#define FL_TEST_RAW
Old name of RB_FL_TEST_RAW.
#define FL_SET
Old name of RB_FL_SET.
#define LONG2NUM
Old name of RB_LONG2NUM.
#define Qtrue
Old name of RUBY_Qtrue.
#define INT2NUM
Old name of RB_INT2NUM.
#define Qnil
Old name of RUBY_Qnil.
#define Qfalse
Old name of RUBY_Qfalse.
#define T_ARRAY
Old name of RUBY_T_ARRAY.
#define NIL_P
Old name of RB_NIL_P.
#define Check_TypedStruct(v, t)
Old name of rb_check_typeddata.
#define NUM2LONG
Old name of RB_NUM2LONG.
#define FIXNUM_P
Old name of RB_FIXNUM_P.
#define FL_SET_RAW
Old name of RB_FL_SET_RAW.
int rb_typeddata_is_kind_of(VALUE obj, const rb_data_type_t *data_type)
Checks if the given object is of given kind.
VALUE rb_eTypeError
TypeError exception.
VALUE rb_eStopIteration
StopIteration exception.
VALUE rb_ensure(VALUE(*b_proc)(VALUE), VALUE data1, VALUE(*e_proc)(VALUE), VALUE data2)
An equivalent to ensure clause.
VALUE rb_eThreadError
ThreadError exception.
VALUE rb_obj_class(VALUE obj)
Queries the class of an object.
VALUE rb_cThread
Thread class.
double rb_num2dbl(VALUE num)
Converts an instance of rb_cNumeric into C's double.
VALUE rb_equal(VALUE lhs, VALUE rhs)
This function is an optimised version of calling #==.
#define RB_OBJ_WRITE(old, slot, young)
Declaration of a "back" pointer.
Defines RBIMPL_HAS_BUILTIN.
VALUE rb_ary_concat(VALUE lhs, VALUE rhs)
Destructively appends the contents of latter into the end of former.
VALUE rb_ary_shift(VALUE ary)
Destructively deletes an element from the beginning of the passed array and returns what was deleted.
VALUE rb_ary_hidden_new(long capa)
Allocates a hidden (no class) empty array.
VALUE rb_ary_clear(VALUE ary)
Destructively removes everything form an array.
VALUE rb_ary_push(VALUE ary, VALUE elem)
Special case of rb_ary_cat() that it adds only one element.
static int rb_check_arity(int argc, int min, int max)
Ensures that the passed integer is in the passed range.
void rb_provide(const char *feature)
Declares that the given feature is already provided by someone else.
VALUE rb_mutex_new(void)
Creates a mutex.
VALUE rb_mutex_trylock(VALUE mutex)
Attempts to lock the mutex, without waiting for other threads to unlock it.
VALUE rb_mutex_locked_p(VALUE mutex)
Queries if there are any threads that holds the lock.
VALUE rb_mutex_synchronize(VALUE mutex, VALUE(*func)(VALUE arg), VALUE arg)
Obtains the lock, runs the passed function, and releases the lock when it completes.
VALUE rb_mutex_sleep(VALUE self, VALUE timeout)
Releases the lock held in the mutex and waits for the period of time; reacquires the lock on wakeup.
VALUE rb_mutex_unlock(VALUE mutex)
Releases the mutex.
VALUE rb_mutex_lock(VALUE mutex)
Attempts to lock the mutex.
struct timeval rb_time_interval(VALUE num)
Creates a "time interval".
void rb_const_set(VALUE space, ID name, VALUE val)
Names a constant.
void rb_define_alloc_func(VALUE klass, rb_alloc_func_t func)
Sets the allocator function of a class.
VALUE rb_yield(VALUE val)
Yields the block.
#define RARRAY_LEN
Just another name of rb_array_len.
#define RUBY_TYPED_DEFAULT_FREE
This is a value you can set to rb_data_type_struct::dfree.
#define TypedData_Get_Struct(obj, type, data_type, sval)
Obtains a C struct from inside of a wrapper Ruby object.
#define TypedData_Make_Struct(klass, type, data_type, sval)
Identical to TypedData_Wrap_Struct, except it allocates a new data region internally instead of takin...
VALUE rb_fiber_scheduler_current(void)
Identical to rb_fiber_scheduler_get(), except it also returns RUBY_Qnil in case of a blocking fiber.
VALUE rb_fiber_scheduler_block(VALUE scheduler, VALUE blocker, VALUE timeout)
Non-blocking wait for the passed "blocker", which is for instance Thread.join or Mutex....
VALUE rb_fiber_scheduler_kernel_sleep(VALUE scheduler, VALUE duration)
Non-blocking sleep.
VALUE rb_fiber_scheduler_unblock(VALUE scheduler, VALUE blocker, VALUE fiber)
Wakes up a fiber previously blocked using rb_fiber_scheduler_block().
#define RTEST
This is an old name of RB_TEST.
This is the struct that holds necessary info for a struct.
uintptr_t ID
Type that represents a Ruby identifier such as a variable name.
uintptr_t VALUE
Type that represents a Ruby object.
static bool RB_TYPE_P(VALUE obj, enum ruby_value_type t)
Queries if the given object is of given type.