briefly defer yielding the thread/gvl

Author: Jacob Lacouture

ractor.c

@@ -31,6 +31,9 @@ static VALUE rb_eRactorMovedError;
 static VALUE rb_eRactorClosedError;
 static VALUE rb_cRactorMovedObject;
 
+extern void rb_thread_sched_init(struct rb_thread_sched *, bool atfork);
+extern void rb_thread_sched_destroy(struct rb_thread_sched *);
+
 static void vm_ractor_blocking_cnt_inc(rb_vm_t *vm, rb_ractor_t *r, const char *file, int line);
 
 // Ractor locking
@@ -257,6 +260,8 @@ ractor_free(void *ptr)
         r->newobj_cache = NULL;
     }
 
+    rb_thread_sched_destroy(&r->threads.sched);
+
     ruby_xfree(r);
 }
 
@@ -2097,8 +2102,6 @@ rb_ractor_atfork(rb_vm_t *vm, rb_thread_t *th)
 }
 #endif
 
-void rb_thread_sched_init(struct rb_thread_sched *, bool atfork);
-
 void
 rb_ractor_living_threads_init(rb_ractor_t *r)
 {

test/-ext-/thread/test_instrumentation_api.rb

@@ -98,8 +98,11 @@ def test_io_release_gvl
     thread = nil
     full_timeline = record do
       thread = Thread.new do
-        w.write("Hello\n")
+        r.readline
       end
+      # Sleep causes the readline call to take long enough that its thread gets descheduled.
+      sleep(1)
+      w.puts("Hello")
       thread.join
     end
 

test/ruby/test_process.rb

@@ -1956,7 +1956,8 @@ def test_daemon_no_threads
           puts Dir.entries("/proc/self/task") - %W[. ..]
         end
         bug4920 = '[ruby-dev:43873]'
-        assert_include(1..2, data.size, bug4920)
+        # On pthread builds there will be an extra thread for the deferred-thread-wait worker
+        assert_include(1..3, data.size, bug4920)
         assert_not_include(data.map(&:to_i), pid)
       end
     else # darwin

thread.c

@@ -202,7 +202,7 @@ static inline void blocking_region_end(rb_thread_t *th, struct rb_blocking_regio
         /* Important that this is inlined into the macro, and not part of \
          * blocking_region_begin - see bug #20493 */ \
         RB_VM_SAVE_MACHINE_CONTEXT(th); \
-        thread_sched_to_waiting(TH_SCHED(th), th); \
+        thread_sched_blocking_region_enter(TH_SCHED(th), th); \
         exec; \
         blocking_region_end(th, &__region); \
     }; \
@@ -1519,9 +1519,9 @@ blocking_region_end(rb_thread_t *th, struct rb_blocking_region_buffer *region)
     /* entry to ubf_list impossible at this point, so unregister is safe: */
     unregister_ubf_list(th);
 
-    thread_sched_to_running(TH_SCHED(th), th);
-    rb_ractor_thread_switch(th->ractor, th);
+    thread_sched_blocking_region_exit(TH_SCHED(th), th);
 
+    rb_ractor_thread_switch(th->ractor, th);
     th->blocking_region_buffer = 0;
     rb_ractor_blocking_threads_dec(th->ractor, __FILE__, __LINE__);
     if (th->status == THREAD_STOPPED) {

thread_none.c

@@ -37,17 +37,25 @@ thread_sched_yield(struct rb_thread_sched *sched, rb_thread_t *th)
 {
 }
 
+static void
+thread_sched_blocking_region_enter(struct rb_thread_sched *sched, rb_thread_t *th)
+{
+}
+
+static void
+thread_sched_blocking_region_exit(struct rb_thread_sched *sched, rb_thread_t *th)
+{
+}
+
 void
 rb_thread_sched_init(struct rb_thread_sched *sched, bool atfork)
 {
 }
 
-#if 0
-static void
+void
 rb_thread_sched_destroy(struct rb_thread_sched *sched)
 {
 }
-#endif
 
 // Do nothing for mutex guard
 void

thread_pthread.c

@@ -75,6 +75,10 @@ static const void *const condattr_monotonic = NULL;
 // #define HAVE_SYS_EPOLL_H 0
 #endif
 
+#if !defined(SET_CURRENT_THREAD_NAME) && defined(__linux__) && defined(PR_SET_NAME)
+# define SET_CURRENT_THREAD_NAME(name) prctl(PR_SET_NAME, name)
+#endif
+
 #ifndef USE_MN_THREADS
   #if defined(__EMSCRIPTEN__) || defined(COROUTINE_PTHREAD_CONTEXT)
     // on __EMSCRIPTEN__ provides epoll* declarations, but no implementations.
@@ -692,10 +696,14 @@ rb_del_running_thread(rb_thread_t *th)
 static void
 thread_sched_set_running(struct rb_thread_sched *sched, rb_thread_t *th)
 {
+    ASSERT_thread_sched_locked(sched, NULL);
     RUBY_DEBUG_LOG("th:%u->th:%u", rb_th_serial(sched->running), rb_th_serial(th));
     VM_ASSERT(sched->running != th);
 
     sched->running = th;
+    // This cancels any deferred scheduling action.
+    sched->deferred_wait_th = NULL;
+    sched->deferred_wait_th_count += 1;
 }
 
 RBIMPL_ATTR_MAYBE_UNUSED()
@@ -1129,9 +1137,120 @@ thread_sched_yield(struct rb_thread_sched *sched, rb_thread_t *th)
     thread_sched_unlock(sched, th);
 }
 
+static void
+thread_sched_blocking_region_enter(struct rb_thread_sched *sched, rb_thread_t *th)
+{
+    thread_sched_lock(sched, th);
+    if (sched->is_running) {
+        VM_ASSERT(sched->running == th);
+
+        sched->deferred_wait_th = th;
+        sched->deferred_wait_th_count += 1;
+        rb_native_cond_signal(&sched->deferred_wait_cond);
+    } else {
+        VM_ASSERT(sched->running == NULL);
+        thread_sched_to_waiting_common(sched, th);
+    }
+    thread_sched_unlock(sched, th);
+}
+
+static void
+thread_sched_blocking_region_exit(struct rb_thread_sched *sched, rb_thread_t *th)
+{
+    thread_sched_lock(sched, th);
+
+    if (sched->running == th && th == sched->deferred_wait_th) {
+        // We never descheduled the thread. Cancel that request now.
+        sched->deferred_wait_th_count += 1;
+        sched->deferred_wait_th = NULL;
+    } else {
+        thread_sched_to_running_common(sched, th);
+    }
+
+    thread_sched_unlock(sched, th);
+}
+
+static void
+transfer_sched_lock(struct rb_thread_sched *sched, struct rb_thread_struct *current, struct rb_thread_struct *th)
+{
+    RUBY_DEBUG_LOG("Transferring sched ownership from:%u to th:%u", rb_th_serial(current), rb_th_serial(th));
+#if VM_CHECK_MODE
+    VM_ASSERT(sched->lock_owner == current);
+    sched->lock_owner = th;
+#endif
+}
+
+static void *
+deferred_wait_thread_worker(void *arg)
+{
+#ifdef SET_CURRENT_THREAD_NAME
+    SET_CURRENT_THREAD_NAME("rb_def_wait");
+#endif
+    struct rb_thread_sched *sched = (struct rb_thread_sched *) arg;
+    struct rb_thread_struct *lock_owner = sched->deferred_wait_th_dummy;
+
+    thread_sched_lock(sched, lock_owner);
+    for (;;) {
+        if (sched->stop) {
+            break;
+        }
+        // The cond-wait will drop the lock. We'll need to update lock_owner manually.
+        transfer_sched_lock(sched, lock_owner, NULL);
+        rb_native_cond_wait(&sched->deferred_wait_cond, &sched->lock_);
+        transfer_sched_lock(sched, NULL, lock_owner);
+        for (;;) {
+            if (!sched->deferred_wait_th) {
+                break;
+            }
+            struct rb_thread_struct *th = sched->deferred_wait_th;
+            int count = sched->deferred_wait_th_count;
+            thread_sched_unlock(sched, lock_owner);
+            usleep(50);
+
+            // th is not a stable reference here. Go back to our dummy thread.
+            lock_owner = sched->deferred_wait_th_dummy;
+            thread_sched_lock(sched, lock_owner);
+
+            if (count != sched->deferred_wait_th_count) {
+                continue;
+            }
+            VM_ASSERT(sched->is_running);
+            VM_ASSERT(th == sched->deferred_wait_th);
+            VM_ASSERT(sched->running == sched->deferred_wait_th);
+
+            // Before calling into the scheduler we need to transfer lock ownership (logically) from the worker
+            // thread to the target thread.
+            transfer_sched_lock(sched, lock_owner, th);
+            // We're now acting on behalf of the target thread.
+            lock_owner = th;
+            thread_sched_to_waiting_common(sched, th);
+            break;
+        }
+    }
+    thread_sched_unlock(sched, lock_owner);
+    return NULL;
+}
+
+static void
+start_deferred_wait_thread(struct rb_thread_sched *sched)
+{
+    pthread_attr_t attr;
+    int r;
+    r = pthread_attr_init(&attr);
+    if (r) {
+        rb_bug_errno("start_deferred_wait_thread - pthread_attr_init", r);
+    }
+    r = pthread_create(&sched->deferred_wait_pthread, &attr, deferred_wait_thread_worker, sched);
+    if (r) {
+        rb_bug_errno("start_deferred_wait_thread - pthread_create", r);
+    }
+    pthread_attr_destroy(&attr);
+}
+
 void
 rb_thread_sched_init(struct rb_thread_sched *sched, bool atfork)
 {
+    sched->stop = false;
     rb_native_mutex_initialize(&sched->lock_);
 
 #if VM_CHECK_MODE
@@ -1144,6 +1263,13 @@ rb_thread_sched_init(struct rb_thread_sched *sched, bool atfork)
 #if USE_MN_THREADS
     if (!atfork) sched->enable_mn_threads = true; // MN is enabled on Ractors
 #endif
+
+    rb_native_cond_initialize(&sched->deferred_wait_cond);
+    sched->deferred_wait_th = NULL;
+    sched->deferred_wait_th_count = 0;
+    sched->deferred_wait_th_dummy = (struct rb_thread_struct *) malloc(sizeof(struct rb_thread_struct));
+    sched->deferred_wait_th_dummy->serial = 100000;
+    start_deferred_wait_thread(sched);
 }
 
 static void
@@ -1504,21 +1630,30 @@ rb_ractor_sched_barrier_join(rb_vm_t *vm, rb_ractor_t *cr)
 // TODO
 
 static void clear_thread_cache_altstack(void);
+#endif
 
-static void
+void
 rb_thread_sched_destroy(struct rb_thread_sched *sched)
 {
+    // Stop the deferred wait worker
+    rb_native_mutex_lock(&sched->lock_);
+
+    sched->stop = true;
+    rb_native_cond_signal(&sched->deferred_wait_cond);
+    rb_native_mutex_unlock(&sched->lock_);
+
+    pthread_join(sched->deferred_wait_pthread, NULL);
+
     /*
-     * only called once at VM shutdown (not atfork), another thread
+     * only called once at VM/ractor shutdown (not atfork), another thread
      * may still grab vm->gvl.lock when calling gvl_release at
      * the end of thread_start_func_2
      */
-    if (0) {
-        rb_native_mutex_destroy(&sched->lock);
-    }
-    clear_thread_cache_altstack();
+    rb_native_mutex_destroy(&sched->lock_);
+    rb_native_cond_destroy(&sched->deferred_wait_cond);
+
+    //clear_thread_cache_altstack();
 }
-#endif
 
 #ifdef RB_THREAD_T_HAS_NATIVE_ID
 static int
@@ -1550,7 +1685,6 @@ thread_sched_atfork(struct rb_thread_sched *sched)
     vm->ractor.sched.running_cnt = 0;
 
     rb_native_mutex_initialize(&vm->ractor.sched.lock);
-    // rb_native_cond_destroy(&vm->ractor.sched.cond);
     rb_native_cond_initialize(&vm->ractor.sched.cond);
     rb_native_cond_initialize(&vm->ractor.sched.barrier_complete_cond);
     rb_native_cond_initialize(&vm->ractor.sched.barrier_release_cond);
@@ -2660,10 +2794,6 @@ setup_communication_pipe_internal(int pipes[2])
     set_nonblock(pipes[1]);
 }
 
-#if !defined(SET_CURRENT_THREAD_NAME) && defined(__linux__) && defined(PR_SET_NAME)
-# define SET_CURRENT_THREAD_NAME(name) prctl(PR_SET_NAME, name)
-#endif
-
 enum {
     THREAD_NAME_MAX =
 #if defined(__linux__)

thread_pthread.h

@@ -128,6 +128,14 @@ struct rb_thread_sched {
     int readyq_cnt;
     // ractor scheduling
     struct ccan_list_node grq_node;
+
+    // Deferred descheduler
+    bool stop;
+    pthread_t deferred_wait_pthread;
+    rb_nativethread_cond_t deferred_wait_cond;
+    int deferred_wait_th_count;
+    struct rb_thread_struct *deferred_wait_th;
+    struct rb_thread_struct *deferred_wait_th_dummy;
 };
 
 #ifdef RB_THREAD_LOCAL_SPECIFIER

thread_win32.c

@@ -148,22 +148,32 @@ thread_sched_yield(struct rb_thread_sched *sched, rb_thread_t *th)
     thread_sched_to_running(sched, th);
 }
 
+static void
+thread_sched_blocking_region_enter(struct rb_thread_sched *sched, rb_thread_t *th)
+{
+    thread_sched_to_waiting(sched, th);
+}
+
+static void
+thread_sched_blocking_region_exit(struct rb_thread_sched *sched, rb_thread_t *th)
+{
+    thread_sched_to_running(sched, th);
+}
+
 void
 rb_thread_sched_init(struct rb_thread_sched *sched, bool atfork)
 {
     if (GVL_DEBUG) fprintf(stderr, "sched init\n");
     sched->lock = w32_mutex_create();
 }
 
-#if 0
 // per-ractor
 void
 rb_thread_sched_destroy(struct rb_thread_sched *sched)
 {
-    if (GVL_DEBUG) fprintf(stderr, "sched destroy\n");
-    CloseHandle(sched->lock);
+//    if (GVL_DEBUG) fprintf(stderr, "sched destroy\n");
+//    CloseHandle(sched->lock);
 }
-#endif
 
 rb_thread_t *
 ruby_thread_from_native(void)