Improve debugging utilities

Author: jpsamaroo

src/sch/Sch.jl

@@ -868,12 +868,12 @@ function do_task(to_proc, extra_util, thunk_id, f, data, send_result, persist, c
         if ((extra_util isa MaxUtilization) && (real_util[] > 0)) ||
            ((extra_util isa Real) && (extra_util + real_util[] > cap))
             # Fully subscribed, wait and re-check
-            @debug "($(myid())) $f Waiting for free $(typeof(to_proc)): $extra_util | $(real_util[])/$cap"
+            @debug "($(myid())) $f ($thunk_id) Waiting for free $(typeof(to_proc)): $extra_util | $(real_util[])/$cap"
             unlock(ACTIVE_TASKS_LOCK)
             wait(TASK_SYNC)
         else
             # Under-subscribed, calculate extra utilization and execute thunk
-            @debug "($(myid())) Using available $to_proc: $extra_util | $(real_util[])/$cap"
+            @debug "($(myid())) ($thunk_id) Using available $to_proc: $extra_util | $(real_util[])/$cap"
             extra_util = if extra_util isa MaxUtilization
                 count(c->typeof(c)===typeof(to_proc), children(from_proc))
             else
@@ -911,7 +911,7 @@ function do_task(to_proc, extra_util, thunk_id, f, data, send_result, persist, c
     lock(ACTIVE_TASKS_LOCK) do
         real_util[] -= extra_util
     end
-    @debug "($(myid())) Releasing $(typeof(to_proc)): $extra_util | $(real_util[])/$cap"
+    @debug "($(myid())) ($thunk_id) Releasing $(typeof(to_proc)): $extra_util | $(real_util[])/$cap"
     notify(TASK_SYNC)
     metadata = (
         pressure=real_util[],

src/sch/util.jl

@@ -61,13 +61,13 @@ function set_failed!(state, origin, thunk=origin)
 end
 
 "Internal utility, useful for debugging scheduler state."
-function print_sch_status(state, thunk)
+function print_sch_status(state, thunk; kwargs...)
     iob = IOBuffer()
-    print_sch_status(iob, state, thunk)
+    print_sch_status(iob, state, thunk; kwargs...)
     seek(iob, 0)
     write(stderr, iob)
 end
-function print_sch_status(io::IO, state, thunk; offset=0, limit=5)
+function print_sch_status(io::IO, state, thunk; offset=0, limit=5, max_inputs=3)
     function status_string(node)
         status = ""
         if node in state.errored
@@ -79,15 +79,25 @@ function print_sch_status(io::IO, state, thunk; offset=0, limit=5)
             status *= "R"
         elseif haskey(state.cache, node)
             status *= "C"
+        elseif node in state.finished
+            status *= "F"
         else
             status *= "?"
         end
         status
     end
-    offset == 0 && print(io, "($(status_string(thunk))) ")
+    if offset == 0
+        println(io, "Ready ($(length(state.ready))): $(join(map(t->t.id, state.ready), ','))")
+        println(io, "Running: ($(length(state.running))): $(join(map(t->t.id, collect(state.running)), ','))")
+        print(io, "($(status_string(thunk))) ")
+    end
     println(io, "$(thunk.id): $(thunk.f)")
-    for input in thunk.inputs
+    for (idx,input) in enumerate(thunk.inputs)
         input isa Thunk || continue
+        if idx > max_inputs
+            println(io, repeat(' ', offset+2), "…")
+            break
+        end
         status = status_string(input)
         if haskey(state.waiting, thunk) && input in state.waiting[thunk]
             status *= "W"
@@ -99,7 +109,7 @@ function print_sch_status(io::IO, state, thunk; offset=0, limit=5)
             status *= "d"
         end
         if haskey(state.futures, input)
-            status *= "f$(length(state.futures[input]))"
+            status *= "f($(length(state.futures[input])))"
         end
         print(io, repeat(' ', offset+2), "($status) ")
         if limit > 0