[Distributed] kill(::LocalManager, ...) should actually call kill()

When dealing with a local process, if we want to remove a process, we
can try a little harder than simply calling `remote_do(exit, id)`.  We
can actually `kill()` the process by sending `SIGTERM`, then `SIGKILL`.
Because we use `Distributed` to run our Base test workers, this can
provide a more certain method of closing our workers at the end of test
sets, as well as a better way of killing processes such that they dump
core in the event that they do get stuck.

stdlib/Distributed/src/managers.jl

@@ -725,3 +725,26 @@ function kill(manager::SSHManager, pid::Int, config::WorkerConfig)
     cancel_ssh_tunnel(config)
     nothing
 end
+
+function kill(manager::LocalManager, pid::Int, config::WorkerConfig; exit_timeout = 15, term_timeout = 15)
+    # First, try sending `exit()` to the remote over the usual control channels
+    remote_do(exit, pid)
+
+    timer_task = @async begin
+        sleep(exit_timeout)
+
+        # Check to see if our child exited, and if not, send an actual kill signal
+        if !process_exited(config.process)
+            @warn("Failed to gracefully kill worker $(pid), sending SIGTERM")
+            kill(config.process, Base.SIGTERM)
+
+            sleep(term_timeout)
+            if !process_exited(config.process)
+                @warn("Worker $(pid) ignored SIGTERM, sending SIGKILL")
+                kill(config.process, Base.SIGKILL)
+            end
+        end
+    end
+    errormonitor(timer_task)
+    return nothing
+end

stdlib/Distributed/test/distributed_exec.jl

@@ -1856,6 +1856,26 @@ end end
 
 include("splitrange.jl")
 
+# Clear all workers for timeout tests (issue #45785)
+rmprocs(workers())
+begin
+    # First, assert that we get no messages when we close a cooperative worker
+    w = only(addprocs(1))
+    @test_nowarn begin
+        wait(rmprocs([w]))
+    end
+
+    # Next, ensure we get a log message when a worker does not cleanly exit
+    w = only(addprocs(1))
+    @test_logs (:warn, r"sending SIGTERM") begin
+        remote_do(w) do
+            # Cause the 'exit()' message that `rmprocs()` sends to do nothing
+            Core.eval(Base, :(exit() = nothing))
+        end
+        wait(rmprocs([w]))
+    end
+end
+
 # Run topology tests last after removing all workers, since a given
 # cluster at any time only supports a single topology.
 rmprocs(workers())