Scheduler: Use a "scheduler" task for thread sleep (#57544)

A Julia thread runs Julia's scheduler in the context of the switching
task. If no task is found to switch to, the thread will sleep while
holding onto the (possibly completed) task, preventing the task from
being garbage collected. This recent [Discourse
post](https://discourse.julialang.org/t/weird-behaviour-of-gc-with-multithreaded-array-access/125433)
illustrates precisely this problem.

A solution to this would be for an idle Julia thread to switch to a
"scheduler" task, thereby freeing the old task.

This PR uses `OncePerThread` to create a "scheduler" task (that does
nothing but run `wait()` in a loop) and switches to that task when the
thread finds itself idle.

Other approaches considered and discarded in favor of this one:
#57465 and
#57543.

Author: kpamnany

base/task.jl

@@ -1145,6 +1145,16 @@ function throwto(t::Task, @nospecialize exc)
     return try_yieldto(identity)
 end
 
+@inline function wait_forever()
+    while true
+        wait()
+    end
+end
+
+const get_sched_task = OncePerThread{Task}() do
+    @task wait_forever()
+end
+
 function ensure_rescheduled(othertask::Task)
     ct = current_task()
     W = workqueue_for(Threads.threadid())
@@ -1181,25 +1191,39 @@ end
 
 checktaskempty = Partr.multiq_check_empty
 
-@noinline function poptask(W::StickyWorkqueue)
-    task = trypoptask(W)
-    if !(task isa Task)
-        task = ccall(:jl_task_get_next, Ref{Task}, (Any, Any, Any), trypoptask, W, checktaskempty)
-    end
-    set_next_task(task)
-    nothing
-end
-
 function wait()
     ct = current_task()
     # [task] user_time -yield-or-done-> wait_time
     record_running_time!(ct)
+    # let GC run
     GC.safepoint()
-    W = workqueue_for(Threads.threadid())
-    poptask(W)
-    result = try_yieldto(ensure_rescheduled)
+    # check for libuv events
     process_events()
-    # return when we come out of the queue
+
+    # get the next task to run
+    result = nothing
+    have_result = false
+    W = workqueue_for(Threads.threadid())
+    task = trypoptask(W)
+    if !(task isa Task)
+        # No tasks to run; switch to the scheduler task to run the
+        # thread sleep logic.
+        sched_task = get_sched_task()
+        if ct !== sched_task
+            result = yieldto(sched_task)
+            have_result = true
+        else
+            task = ccall(:jl_task_get_next, Ref{Task}, (Any, Any, Any),
+                         trypoptask, W, checktaskempty)
+        end
+    end
+    # We may have already switched tasks (via the scheduler task), so
+    # only switch if we haven't.
+    if !have_result
+        @assert task isa Task
+        set_next_task(task)
+        result = try_yieldto(ensure_rescheduled)
+    end
     return result
 end
 

stdlib/Sockets/test/runtests.jl

@@ -547,14 +547,12 @@ end
         fetch(r)
     end
 
-    let addr = Sockets.InetAddr(ip"127.0.0.1", 4444)
-        srv = listen(addr)
+    let addr = Sockets.InetAddr(ip"192.0.2.5", 4444)
         s = Sockets.TCPSocket()
         Sockets.connect!(s, addr)
         r = @async close(s)
         @test_throws Base._UVError("connect", Base.UV_ECANCELED) Sockets.wait_connected(s)
         fetch(r)
-        close(srv)
     end
 end
 

test/channels.jl

@@ -463,15 +463,14 @@ end
         cb = first(async.cond.waitq)
         @test isopen(async)
         ccall(:uv_async_send, Cvoid, (Ptr{Cvoid},), async)
-        ccall(:uv_async_send, Cvoid, (Ptr{Cvoid},), async)
-        @test isempty(Base.Workqueue)
         Base.process_events() # schedule event
         Sys.iswindows() && Base.process_events() # schedule event (windows?)
-        @test length(Base.Workqueue) == 1
         ccall(:uv_async_send, Cvoid, (Ptr{Cvoid},), async)
         @test tc[] == 0
         yield() # consume event
         @test tc[] == 1
+        ccall(:uv_async_send, Cvoid, (Ptr{Cvoid},), async)
+        Base.process_events()
         Sys.iswindows() && Base.process_events() # schedule event (windows?)
         yield() # consume event
         @test tc[] == 2