Interrupt on windows! 😯

.github/workflows/test.yml

@@ -17,7 +17,7 @@ on:
 jobs:
     test:
         runs-on: ${{ matrix.os }}
-        timeout-minutes: 20
+        timeout-minutes: 25
 
         strategy:
             # Without setting this, a failing test cancels all others

README.md

@@ -63,6 +63,9 @@ In Distributed, only "process 1 can add or remove workers". Malt.jl does not hav
 ### Process isolation
 Malt.jl worker processes **do not inherit** `ENV` variables, command-line arguments or the Pkg environment from their host.
 
+### Interrupt on Windows
+Malt.jl supports **interrupting a worker process on Windows**, not just on UNIX.
+
 ### Heterogenous computing
 Malt.jl does not have API like `@everywhere` or `Distributed.procs`: Malt is **not the right tool for homogenous computing**.
 

src/Malt.jl

@@ -103,7 +103,11 @@ mutable struct Worker <: AbstractWorker
     function Worker(; env=String[], exeflags=[])
         # Spawn process
         cmd = _get_worker_cmd(; env, exeflags)
-        proc = open(cmd, "w+")
+        proc = open(Cmd(
+            cmd; 
+            detach=true,
+            windows_hide=true,
+        ), "w+")
 
         # Keep internal list
         __iNtErNaL_get_running_procs()
@@ -612,18 +616,20 @@ Send an interrupt signal to the worker process. This will interrupt the
 latest request (`remote_call*` or `remote_eval*`) that was sent to the worker.
 """
 function interrupt(w::Worker)
-    if Sys.iswindows()
-        # TODO: not yet implemented
-        @warn "Malt.interrupt is not yet supported on Windows"
-        # _assert_is_running(w)
-        # _send_msg(w, MsgType.from_host_fake_interrupt, (), false)
-        nothing
+    if !isrunning(w)
+        @warn "Tried to interrupt a worker that has already shut down." summary(w)
     else
-        Base.kill(w.proc, Base.SIGINT)
+        if Sys.iswindows()
+            ccall((:GenerateConsoleCtrlEvent,"Kernel32"), Bool, (UInt32, UInt32), UInt32(1), UInt32(getpid(w.proc)))
+        else
+            Base.kill(w.proc, Base.SIGINT)
+        end
     end
+    nothing
 end
 function interrupt(w::InProcessWorker)
-    schedule(w.latest_request_task, InterruptException(); error=true)
+    istaskdone(w.latest_request_task) || schedule(w.latest_request_task, InterruptException(); error=true)
+    nothing
 end
 
 

test/interrupt.jl

@@ -0,0 +1,146 @@
+win = Sys.iswindows()
+
+@testset "Interrupt: $W" for W in (m.DistributedStdlibWorker, m.InProcessWorker, m.Worker)
+# @testset "Interrupt: $W" for W in (m.Worker,)
+
+    no_interrupt_possible = (Sys.iswindows() && W === m.DistributedStdlibWorker) || W === m.InProcessWorker
+
+
+    w = W()
+
+    @test m.isrunning(w)
+    @test m.remote_call_fetch(&, w, true, true)
+
+
+    ex1 = quote
+        local x = 0.0
+        for i = 1:4000
+            k = [sqrt(abs(sin(cos(tan(x))))) ^ (1 / i) for z in 1:i]
+            x += sum(k)
+        end
+        x
+    end |> Base.remove_linenums!
+
+    ex2 = quote
+        local x = 0.0
+        for i in 1:20_000_000
+            x += sqrt(abs(sin(cos(tan(x)))))^(1/i)
+        end
+        x
+    end |> Base.remove_linenums!
+
+    ex3 = :(sleep(3)) |> Base.remove_linenums!
+
+    # expressions in this list can be interrupted with a single Ctrl+C
+    # open a terminal and try this.
+    # (some expressions like `while true end` need multiple Ctrl+C in short succession to force throw SIGINT)
+    exs = no_interrupt_possible ? [ex1, ex3] : [
+        ex1,
+        ex3,
+        ex1, # second time because interrupts should be reliable
+        (
+            VERSION > v"1.10.0-0" ? [ex2, ex2] : []
+        )...,
+    ]
+
+
+
+    @testset "single interrupt $ex" for ex in exs
+
+        f() = m.remote_eval(w, ex)
+
+        t1 = @elapsed wait(f())
+        t2 = @elapsed wait(f())
+
+        t3 = @elapsed begin
+            t = f()
+            @test !istaskdone(t)
+            sleep(.1)
+            m.interrupt(w)
+            r = try
+                fetch(t)
+            catch e
+                e
+            end
+            no_interrupt_possible || @test r isa TaskFailedException
+        end
+
+        t4 = @elapsed begin
+            t = f()
+            @test !istaskdone(t)
+            sleep(.1)
+            m.interrupt(w)
+            r = try
+                fetch(t)
+            catch e
+                e
+            end
+            no_interrupt_possible || @test r isa TaskFailedException
+        end
+
+        @info "test run" ex t1 t2 t3 t4
+        no_interrupt_possible || @test t4 < min(t1,t2) * 0.8
+
+        # still running and responsive
+        @test m.isrunning(w)
+        @test m.remote_call_fetch(&, w, true, true)
+
+    end
+
+
+    if !no_interrupt_possible
+        @testset "hard interrupt" begin
+
+            function hard_interrupt(w)
+                finish_task = m.remote_call(&, w, true, true)
+
+                done() = !m.isrunning(w) || istaskdone(finish_task)
+
+                while !done()
+                    for _ in 1:5
+                        print(" 🔥 ")
+                        m.interrupt(w)
+                        sleep(0.18)
+                        if done()
+                            break
+                        end
+                    end
+                    sleep(1.5)
+                end
+            end
+
+
+            t = m.remote_eval(w, :(while true end))
+
+            @test !istaskdone(t)
+            @test m.isrunning(w)
+
+            hard_interrupt(w)
+
+
+            @info "xx" istaskdone(t) m.isrunning(w)
+
+            @test try
+                fetch(t)
+            catch e
+                e
+            end isa TaskFailedException
+
+            # hello
+            @test true
+
+            if Sys.iswindows() && VERSION < v"1.10.0-beta3"
+                # fixed by https://github.com/JuliaLang/julia/pull/51307 which will probably land in v1.10.0-beta3
+                @test_broken m.isrunning(w)
+            else
+                # still running and responsive
+                @test m.isrunning(w)
+                @test m.remote_call_fetch(&, w, true, true)
+            end
+        end
+    end
+
+
+    m.stop(w)
+    @test !m.isrunning(w)
+end

test/runtests.jl

@@ -4,6 +4,8 @@ using Test
 
 v() = @assert isempty(m.__iNtErNaL_get_running_procs())
 
+v()
+include("interrupt.jl")
 v()
 include("basic.jl")
 v()