runtests: prefix with the testsuite and worker number

deps/checksums/Distributed-94777c0ef77990ee7e1a775653c8bb34237ae35a.tar.gz/md5

@@ -0,0 +1 @@
+0cc500af204a8ba34580f52596926d60

deps/checksums/Distributed-94777c0ef77990ee7e1a775653c8bb34237ae35a.tar.gz/sha512

@@ -0,0 +1 @@
+732173f2bdcfd9f1cebb69d1464ec3d6d6729b34268cbe622da86af7e8e09531b1ca7fee14ceeb3246ae03f2874c6716a896cbbf9132b3f660d68bd481d24c05

stdlib/Distributed.version

@@ -1,4 +1,4 @@
-DISTRIBUTED_BRANCH = master
-DISTRIBUTED_SHA1 = cd9219573d736b036077dff3cadddf369516d495
+DISTRIBUTED_BRANCH = ib/worker_output_customization
+DISTRIBUTED_SHA1 = 94777c0ef77990ee7e1a775653c8bb34237ae35a
 DISTRIBUTED_GIT_URL := https://github.com/JuliaLang/Distributed.jl
 DISTRIBUTED_TAR_URL = https://api.github.com/repos/JuliaLang/Distributed.jl/tarball/$1

test/embedding/embedding-test.jl

@@ -31,6 +31,6 @@ end
     @test lines[4] == "sqrt(2.0) in C: 1.414214e+00"
     @test lines[9] == "called bar"
     @test lines[10] == "calling new bar"
-    @test lines[11] == "      From worker 2:\tTaking over the world..."
+    @test lines[11] == "  From worker 2:\tTaking over the world..."
     @test "exception caught from C" in readlines(err)
 end

test/runtests.jl

@@ -14,6 +14,39 @@ include("buildkitetestjson.jl")
 const longrunning_delay = parse(Int, get(ENV, "JULIA_TEST_LONGRUNNING_DELAY", "45")) * 60 # minutes
 const longrunning_interval = parse(Int, get(ENV, "JULIA_TEST_LONGRUNNING_INTERVAL", "15")) * 60 # minutes
 
+# Helper to run code with prefixed output (uses Pipe + background reader)
+function with_output_prefix(f, prefix::String, io::IO, lock::ReentrantLock)
+    pipe = Pipe()
+    Base.link_pipe!(pipe; reader_supports_async=true, writer_supports_async=true)
+
+    reader_task = @async begin
+        try
+            while isopen(pipe) || bytesavailable(pipe) > 0
+                line = readline(pipe; keep=true)
+                isempty(line) && break
+                @lock lock begin
+                    printstyled(io, "  ", prefix, ": ", color=:light_black)
+                    print(io, line)
+                    endswith(line, '\n') || println(io)
+                end
+            end
+        catch e
+            e isa EOFError || rethrow()
+        end
+    end
+
+    try
+        redirect_stdout(pipe) do
+            redirect_stderr(pipe) do
+                f()
+            end
+        end
+    finally
+        close(pipe.in)
+        wait(reader_task)
+    end
+end
+
 (; tests, net_on, exit_on_error, use_revise, buildroot, seed) = choosetests(ARGS)
 tests = unique(tests)
 
@@ -123,6 +156,9 @@ cd(@__DIR__) do
     end
     skipped = 0
 
+    # Track which test is running on each worker (worker_id => test_name)
+    worker_current_test = Dict{Int, String}()
+
     @everywhere include("testdefs.jl")
 
     if use_revise
@@ -160,6 +196,21 @@ cd(@__DIR__) do
         stderr.lock = print_lock
     end
 
+    # Set up hook to display test name with worker output
+    Distributed.worker_output_hook[] = (ident, line) -> begin
+        wrkr_id = tryparse(Int, ident)
+        test_name = wrkr_id === nothing ? nothing : get(worker_current_test, wrkr_id, nothing)
+        @lock print_lock begin
+            if test_name !== nothing
+                printstyled("  ", test_name, " (", ident, "): ", color=:light_black)
+            else
+                printstyled("  From worker ", ident, ": ", color=:light_black)
+            end
+            println(line)
+        end
+        return true
+    end
+
     function print_testworker_stats(test, wrkr, resp)
         @nospecialize resp
         lock(print_lock)
@@ -268,6 +319,7 @@ cd(@__DIR__) do
                         test = popfirst!(tests)
                         running_tests[test] = now()
                         wrkr = p
+                        worker_current_test[wrkr] = test
 
                         # Create a timer for this test to report long-running status
                         test_timers[test] = Timer(longrunning_delay, interval=longrunning_interval) do timer
@@ -304,6 +356,7 @@ cd(@__DIR__) do
                                 Any[CapturedException(e, catch_backtrace())], time() - before
                             end
                         delete!(running_tests, test)
+                        delete!(worker_current_test, wrkr)
                         if haskey(test_timers, test)
                             close(test_timers[test])
                             delete!(test_timers, test)
@@ -361,8 +414,10 @@ cd(@__DIR__) do
             t == "SharedArrays" && (isolate = false)
             before = time()
             resp, duration = try
-                    r = @invokelatest runtests(t, test_path(t), isolate, seed=seed) # runtests is defined by the include above
-                    r, time() - before
+                    with_output_prefix("$t (1)", stdout, print_lock) do
+                        r = @invokelatest runtests(t, test_path(t), isolate, seed=seed) # runtests is defined by the include above
+                        r, time() - before
+                    end
                 catch e
                     isa(e, InterruptException) && rethrow()
                     Any[CapturedException(e, catch_backtrace())], time() - before
@@ -395,6 +450,7 @@ cd(@__DIR__) do
         if @isdefined test_timers
             foreach(close, values(test_timers))
         end
+        Distributed.worker_output_hook[] = nothing
     end
 
     #=