scheduler: wake the sleeping event loop thread when IO work arrives (#63884)

Author: gbaraldi

doc/src/devdocs/scheduler-wakeup.md

@@ -118,6 +118,12 @@ that the loop is idle before parking: it takes the iolock with a trylock and
 releases it without running finalizers. If the iolock is busy, it retries instead
 of parking.
 
+The thread that armed a handle may not go to sleep for a long time, though, for
+example when the tasks of its pool keep yielding to each other. So when the check
+finds the loop idle, it also sets `jl_uv_loop_unwatched` under the iolock, and the
+scheduler clears it before it runs `uv_run`. A thread that leaves work in the loop
+while the flag is set clears it and wakes `io_loop_tid` from `jl_iolock_end`.
+
 [`IoLoopWake.tla`](https://github.com/JuliaLang/julia/blob/master/doc/src/devdocs/scheduler-wakeup/IoLoopWake.tla)
 models this handoff outside threaded regions, including iolock contention and
 finalizers that take the iolock before the loop runs and after its unlock. It

src/jl_uv.c

@@ -134,6 +134,7 @@ void jl_wake_signal_watchers(void)
 }
 
 jl_mutex_t jl_uv_mutex;
+int jl_uv_loop_unwatched = 0;
 
 void jl_init_uv(void)
 {
@@ -170,6 +171,12 @@ JL_DLLEXPORT void jl_iolock_begin(void) JL_CANSAFEPOINT
  */
 JL_DLLEXPORT void jl_iolock_end(void) JL_CANSAFEPOINT
 {
+    // if we leave work in the loop while nobody runs it, wake the event loop thread
+    // (before unlocking, which can throw)
+    if (jl_uv_mutex.count == 1 && jl_uv_loop_unwatched && uv_loop_alive(jl_io_loop)) {
+        jl_uv_loop_unwatched = 0;
+        jl_wake_io_loop_thread();
+    }
     JL_UV_UNLOCK();
 }
 
@@ -432,7 +439,7 @@ static void jl_proc_exit_cleanup_cb(uv_process_t *process, int64_t exit_status,
  */
 JL_DLLEXPORT void jl_close_uv(uv_handle_t *handle)
 {
-    JL_UV_LOCK();
+    jl_iolock_begin();
     if (handle->type == UV_PROCESS && ((uv_process_t*)handle)->pid != 0) {
         // take ownership of this handle,
         // so we can waitpid for the resource to exit and avoid leaving zombies
@@ -460,7 +467,7 @@ JL_DLLEXPORT void jl_close_uv(uv_handle_t *handle)
             uv_close(handle, &jl_uv_closeHandle);
         }
     }
-    JL_UV_UNLOCK();
+    jl_iolock_end();
 }
 
 /**
@@ -471,11 +478,11 @@ JL_DLLEXPORT void jl_close_uv(uv_handle_t *handle)
 JL_DLLEXPORT void jl_forceclose_uv(uv_handle_t *handle) JL_CANSAFEPOINT
 {
     if (!uv_is_closing(handle)) { // avoid double-closing the stream
-        JL_UV_LOCK();
+        jl_iolock_begin();
         if (!uv_is_closing(handle)) { // double-check
             uv_close(handle, &jl_uv_closeHandle);
         }
-        JL_UV_UNLOCK();
+        jl_iolock_end();
     }
 }
 
@@ -556,9 +563,9 @@ JL_DLLEXPORT int jl_spawn(char *name, char **argv,
         }
     }
     opts.exit_cb = cb;
-    JL_UV_LOCK();
+    jl_iolock_begin();
     int r = uv_spawn(loop, proc, &opts);
-    JL_UV_UNLOCK();
+    jl_iolock_end();
     return r;
 }
 
@@ -707,10 +714,10 @@ JL_DLLEXPORT int jl_uv_write(uv_stream_t *stream, const char *data, size_t n,
     uv_buf_t buf[1];
     buf[0].base = (char*)data;
     buf[0].len = n;
-    JL_UV_LOCK();
+    jl_iolock_begin();
     JL_SIGATOMIC_BEGIN();
     int err = uv_write(uvw, stream, buf, 1, writecb);
-    JL_UV_UNLOCK();
+    jl_iolock_end();
     JL_SIGATOMIC_END();
     return err;
 }
@@ -775,10 +782,10 @@ JL_DLLEXPORT void jl_uv_puts(uv_stream_t *stream, const char *str, size_t n)
         buf[0].base = data;
         buf[0].len = n;
         req->data = NULL;
-        JL_UV_LOCK();
+        jl_iolock_begin();
         JL_SIGATOMIC_BEGIN();
         int status = uv_write(req, stream, buf, 1, (uv_write_cb)jl_uv_writecb);
-        JL_UV_UNLOCK();
+        jl_iolock_end();
         JL_SIGATOMIC_END();
         if (status < 0) {
             jl_uv_writecb(req, status);
@@ -1261,9 +1268,9 @@ JL_DLLEXPORT int jl_queue_work(work_cb_t work_func, void *work_ctx,
     baton->notify_data1 = notify_data1;
     baton->notify_data2 = notify_data2;
 
-    JL_UV_LOCK();
+    jl_iolock_begin();
     uv_queue_work(jl_io_loop, &baton->req, jl_work_wrapper, jl_work_notifier);
-    JL_UV_UNLOCK();
+    jl_iolock_end();
 
     return 0;
 }

src/julia_internal.h

@@ -183,6 +183,10 @@ static uv_loop_t *const unused_uv_loop_arg = (uv_loop_t *)0xBAD10;
 
 extern jl_mutex_t jl_uv_mutex;
 extern _Atomic(int) jl_uv_n_waiters;
+// Set when a thread finds the event loop idle before it sleeps. Cleared before the
+// scheduler runs the loop, and by jl_iolock_end when it wakes the event loop thread.
+// Protected by jl_uv_mutex.
+extern int jl_uv_loop_unwatched;
 
 // Global data structures for accessing symbols and other globals
 #include "jl_internal_data.inc"
@@ -1390,6 +1394,7 @@ void jl_safepoint_exclude_gc_end(void) JL_NOTSAFEPOINT JL_NOTSAFEPOINT_LEAVE;
 void jl_wake_libuv(void) JL_NOTSAFEPOINT;
 void jl_init_signal_watchers(uv_async_cb cb);
 void jl_wake_signal_watchers(void) JL_NOTSAFEPOINT;
+void jl_wake_io_loop_thread(void) JL_NOTSAFEPOINT;
 void jl_send_abandon_signal(int16_t tid) JL_NOTSAFEPOINT;
 int jl_abandon_try_commit(jl_ptls_t ptls) JL_NOTSAFEPOINT;
 void JL_NORETURN jl_abandon_task_cb(void) JL_CANSAFEPOINT;

src/scheduler.c

@@ -326,6 +326,19 @@ JL_DLLEXPORT void jl_wakeup_thread_from_foreign(int16_t tid) JL_NOTSAFEPOINT
     }
 }
 
+// Called by a thread that is leaving work in the event loop while nobody runs it.
+// It may stay busy for a long time and never reach the check it makes before it
+// sleeps, so wake the event loop thread now.
+void jl_wake_io_loop_thread(void) JL_NOTSAFEPOINT
+{
+    jl_task_t *ct = jl_current_task;
+    int16_t tid = jl_atomic_load_relaxed(&io_loop_tid);
+    if (tid == jl_atomic_load_relaxed(&ct->tid))
+        return; // it runs uv_run itself before it sleeps
+    jl_fence(); // [^store_buffering_1]
+    wake_thread(tid);
+}
+
 // Round-robin start hint for jl_wakeup_threadpool, sharded across cache-line-padded
 // stripes so concurrent producers don't contend on a single counter.
 #define POOL_WAKE_HINT_STRIPES 64
@@ -569,6 +582,7 @@ JL_DLLEXPORT jl_task_t *jl_task_get_next(jl_value_t *trypoptask, jl_value_t *q,
                     if (enter_eventloop) {
                         uv_loop_t *loop = jl_global_event_loop();
                         loop->stop_flag = 0;
+                        jl_uv_loop_unwatched = 0;
                         JULIA_DEBUG_SLEEPWAKE( ptls->uv_run_enter = cycleclock() );
                         active = uv_run(loop, UV_RUN_ONCE);
                         JULIA_DEBUG_SLEEPWAKE( ptls->uv_run_leave = cycleclock() );
@@ -584,6 +598,8 @@ JL_DLLEXPORT jl_task_t *jl_task_get_next(jl_value_t *trypoptask, jl_value_t *q,
                         active = 1;
                         if (jl_mutex_trylock_nogc(&jl_uv_mutex)) {
                             active = uv_loop_alive(jl_global_event_loop());
+                            // whoever leaves work in the loop from now on wakes the event loop thread
+                            jl_uv_loop_unwatched = !active;
                             jl_mutex_unlock_nogc(&jl_uv_mutex);
                         }
                     }

test/threads.jl

@@ -801,6 +801,43 @@ end
     close(t)
 end
 
+@testset "event loop thread wakeup with busy workers" begin
+    # A timer or `@threadcall` started on a pool whose threads never sleep must still
+    # get serviced by the event loop thread (#63882).
+    script = """
+    # The ^C listener keeps the event loop alive, so the event loop thread never
+    # sleeps. Close it so the loop can go idle.
+    hooks = filter(f -> hasproperty(f, :listeners) && hasproperty(f, :cond) &&
+                        getproperty(f, :cond) isa Base.AsyncCondition, Base.atexit_hooks)
+    isdefined(Base, :start_sigint_listener) && length(hooks) != 1 && error("^C listener not found")
+    for f in hooks
+        ccall(:jl_set_sigint_cond, Cvoid, (Ptr{Cvoid},), C_NULL)
+        close(f.cond)
+    end
+    const done = Threads.Atomic{Bool}(false)
+    # keep the only default thread busy: it yields, but never sleeps
+    spinner = Threads.@spawn :default begin
+        t_end = time_ns() + 30_000_000_000
+        while !done[]
+            time_ns() > t_end && return false
+            yield()
+        end
+        return true
+    end
+    waiter = Threads.@spawn :default begin
+        for _ in 1:3
+            sleep(0.001)
+            @threadcall(:jl_getpagesize, Clong, ())
+        end
+        done[] = true
+    end
+    wait(waiter)
+    fetch(spinner) || exit(1)
+    """
+    cmd = `$(Base.julia_cmd()) --startup-file=no --threads=1,1 -e $script`
+    @test success(pipeline(cmd; stdout, stderr))
+end
+
 @testset "--timeout-for-safepoint-straggler command-line flag" begin
     program = "
         function main()