mirror of
https://github.com/ruby/ruby.git
synced 2026-10-28 16:36:20 +08:00
Prevent starvation when acquiring mutex over and over (#15877)
Continually locking a mutex m can lead to starvation if all other threads are on the waitq of m. See https://bugs.ruby-lang.org/issues/21840 for more details. Solution: When a thread `T1` wakes up `T2` during mutex unlock but `T1` or any other thread successfully acquires it before `T2`, then we record the `running_time` of the thread during mutex acquisition. Then during unlock, if that thread's running_time is less than the saved running time, we set it back to the saved time. Fixes [Bug #21840]
This commit is contained in:
1 parent
3c634893e2
commit
994257ab06
2 files changed
+55
-2
Notes:
git
2026-01-26 19:35:05 +00:00
Merged-By: luke-gru <luke.gru@gmail.com>
No files matched your search
@@ -1664,4 +1664,37 @@ q.pop
|
||||
assert_operator elapsed, :>=, 0.1, "sub-millisecond sleeps should not return immediately"
|
||||
end;
|
||||
end
|
||||
|
||||
# [Bug #21840]
|
||||
def test_mutex_owner_doesnt_starve_waiters
|
||||
assert_ruby_status([], "#{<<~"begin;"}\n#{<<~'end;'}")
|
||||
begin;
|
||||
m = Mutex.new
|
||||
|
||||
fib = lambda { |n|
|
||||
return n if n <= 1
|
||||
fib(n - 1) + fib(n - 2)
|
||||
}
|
||||
|
||||
t1_running = false
|
||||
t1 = Thread.new do
|
||||
t1_running = true
|
||||
loop do
|
||||
fib(20)
|
||||
m.synchronize do
|
||||
File.open(__FILE__) { } # reset timeslice due to blocking operation
|
||||
end
|
||||
end
|
||||
end
|
||||
|
||||
loop until t1_running
|
||||
|
||||
3.times.map do
|
||||
Thread.new do
|
||||
m.synchronize do
|
||||
end
|
||||
end
|
||||
end.each(&:join)
|
||||
end;
|
||||
end
|
||||
end
|
||||
+22
-2
@@ -10,6 +10,8 @@ typedef struct rb_mutex_struct {
|
||||
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 */
|
||||
uint32_t saved_running_time_us;
|
||||
bool wait_waking; // Is there a thread waiting to be woken up by this mutex? Reset during every wakeup.
|
||||
} rb_mutex_t;
|
||||
|
||||
/* sync_waiter is always on-stack */
|
||||
@@ -212,8 +214,15 @@ mutex_locked(rb_mutex_t *mutex, rb_thread_t *th, rb_serial_t ec_serial)
|
||||
static inline bool
|
||||
do_mutex_trylock(rb_mutex_t *mutex, rb_thread_t *th, rb_serial_t ec_serial)
|
||||
{
|
||||
// NOTE: we can successfully lock a mutex even if there are other threads waiting on it. First one to it wins.
|
||||
if (mutex->ec_serial == 0) {
|
||||
RUBY_DEBUG_LOG("%p ok", mutex);
|
||||
if (mutex->wait_waking) {
|
||||
// If we acquired `mutex` without contention and before the thread that was popped off the waitq, we're going
|
||||
// to set our running_time back to what it was here during mutex unlock if it got reset during our critical
|
||||
// section. This is to prevent starvation of other threads waiting on the mutex.
|
||||
mutex->saved_running_time_us = th->running_time_us;
|
||||
}
|
||||
|
||||
mutex_locked(mutex, th, ec_serial);
|
||||
return true;
|
||||
@@ -350,7 +359,8 @@ do_mutex_lock(struct mutex_args *args, int interruptible_p)
|
||||
}
|
||||
ccan_list_del(&sync_waiter.node);
|
||||
|
||||
// unlocked by another thread while sleeping
|
||||
// If mutex->ec_serial != 0, the mutex was locked by another thread before we had the chance to acquire it.
|
||||
// We'll put ourselves on the waitq and sleep again.
|
||||
if (!mutex->ec_serial) {
|
||||
mutex_set_owner(mutex, th, ec_serial);
|
||||
}
|
||||
@@ -391,6 +401,7 @@ do_mutex_lock(struct mutex_args *args, int interruptible_p)
|
||||
|
||||
if (saved_ints) th->ec->interrupt_flag = saved_ints;
|
||||
if (mutex->ec_serial == ec_serial) mutex_locked(mutex, th, ec_serial);
|
||||
mutex->wait_waking = false;
|
||||
}
|
||||
|
||||
RUBY_DEBUG_LOG("%p locked", mutex);
|
||||
@@ -454,6 +465,15 @@ rb_mutex_unlock_th(rb_mutex_t *mutex, rb_thread_t *th, rb_serial_t ec_serial)
|
||||
|
||||
struct sync_waiter *cur = 0, *next;
|
||||
|
||||
|
||||
if (mutex->wait_waking) {
|
||||
uint32_t saved = mutex->saved_running_time_us;
|
||||
if (th->running_time_us < saved) {
|
||||
th->running_time_us = saved;
|
||||
}
|
||||
}
|
||||
|
||||
mutex->saved_running_time_us = 0;
|
||||
mutex->ec_serial = 0;
|
||||
thread_mutex_remove(th, mutex);
|
||||
|
||||
@@ -469,6 +489,7 @@ rb_mutex_unlock_th(rb_mutex_t *mutex, rb_thread_t *th, rb_serial_t ec_serial)
|
||||
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));
|
||||
mutex->wait_waking = true;
|
||||
rb_threadptr_interrupt(cur->th);
|
||||
return NULL;
|
||||
case THREAD_STOPPED: /* probably impossible */
|
||||
@@ -480,7 +501,6 @@ rb_mutex_unlock_th(rb_mutex_t *mutex, rb_thread_t *th, rb_serial_t ec_serial)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// We did not find any threads to wake up, so we can just return with no error:
|
||||
return NULL;
|
||||
}
|
||||
|
||||
Reference in new issue
Block a user