runtests: prefix with the testsuite and worker number

deps/checksums/Distributed-14f81291c8a60ca4c6a1d78cf372ba45017b7e47.tar.gz/md5

@@ -0,0 +1 @@
+c41a5090f7bbfcee8810cb4edc3ec585

deps/checksums/Distributed-14f81291c8a60ca4c6a1d78cf372ba45017b7e47.tar.gz/sha512

@@ -0,0 +1 @@
+6d8f734d7068c38634bc79be32dd1eddc39e46361b0578028a4854feaffcbab22de1b36932e929582b0dc162c81345fdb6fe0aac90508f0f928b18a74de35e78

stdlib/Distributed.version

@@ -1,4 +1,4 @@
-DISTRIBUTED_BRANCH = master
-DISTRIBUTED_SHA1 = fde987fbb05c65bb6bd183396137383931107059
+DISTRIBUTED_BRANCH = ib/worker_output_customization
+DISTRIBUTED_SHA1 = 14f81291c8a60ca4c6a1d78cf372ba45017b7e47
 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

@@ -4,6 +4,7 @@ using Test
 using Distributed
 using Dates
 using Printf: @sprintf
+using StyledStrings: @styled_str
 using Base: Experimental
 using Base.ScopedValues
 
@@ -14,6 +15,34 @@ 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
 
+# Run `f()` with its stdout/stderr captured and re-emitted to `io`, each line prefixed
+# with a de-emphasized `prefix`. Used for the node-1 tests, whose output does not flow
+# through Distributed's worker-output redirection (and so its `worker_output_hook`).
+# `prefix` is styled with `StyledStrings`, so it is only colorized when `io` supports it.
+function with_output_prefix(f, prefix::AbstractString, io::IO, lock::ReentrantLock)
+    pipe = Pipe()
+    Base.link_pipe!(pipe; reader_supports_async=true, writer_supports_async=true)
+
+    reader_task = @async while !eof(pipe)
+        line = readline(pipe; keep=true)
+        @lock lock begin
+            print(io, styled"{bright_black: $prefix: }")
+            print(io, line)
+            endswith(line, '\n') || println(io)
+        end
+    end
+
+    try
+        redirect_stdio(; stdout=pipe, stderr=pipe) do
+            f()
+        end
+    finally
+        close(pipe.in)
+        wait(reader_task)
+        close(pipe)
+    end
+end
+
 (; tests, net_on, exit_on_error, use_revise, buildroot, seed) = choosetests(ARGS)
 tests = unique(tests)
 
@@ -161,6 +190,19 @@ cd(@__DIR__) do
         stderr.lock = print_lock
     end
 
+    # Track which test each worker is currently running (worker id => test name), so
+    # that worker output can be labeled with the test rather than just the worker id.
+    worker_current_test = Dict{Int, String}()
+
+    # Transform each line of worker output into a de-emphasized, test-labeled prefix.
+    # Distributed still owns the actual printing; see `Distributed.worker_output_hook`.
+    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)
+        label = test_name === nothing ? "From worker $ident" : "$test_name ($ident)"
+        return styled"{bright_black: $label: }$line"
+    end
+
     function print_testworker_stats(test, wrkr, resp)
         @nospecialize resp
         lock(print_lock)
@@ -272,6 +314,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
@@ -308,6 +351,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)
@@ -365,8 +409,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
@@ -399,6 +445,7 @@ cd(@__DIR__) do
         if @isdefined test_timers
             foreach(close, values(test_timers))
         end
+        Distributed.worker_output_hook[] = nothing
     end
 
     #=