Add Task Monitor

Author: johnmaxrin

Project.toml

@@ -12,6 +12,7 @@ MemPool = "f9f48841-c794-520a-933b-121f7ba6ed94"
 OnlineStats = "a15396b6-48d5-5d58-9928-6d29437db91e"
 PrecompileTools = "aea7be01-6a6a-4083-8856-8a6e6704d82a"
 Profile = "9abbd945-dff8-562f-b5e8-e1ebf5ef1b79"
+ProgressMeter = "92933f4c-e287-5a05-a399-4b506db050ca"
 Random = "9a3f8284-a2c9-5f02-9a11-845980a1fd5c"
 Requires = "ae029012-a4dd-5104-9daa-d747884805df"
 ScopedValues = "7e506255-f358-4e82-b7e4-beb19740aa63"
@@ -23,6 +24,17 @@ StatsBase = "2913bbd2-ae8a-5f71-8c99-4fb6c76f3a91"
 TimespanLogging = "a526e669-04d3-4846-9525-c66122c55f63"
 UUIDs = "cf7118a7-6976-5b1a-9a39-7adc72f591a4"
 
+[weakdeps]
+Colors = "5ae59095-9a9b-59fe-a467-6f913c188581"
+DataFrames = "a93c6f00-e57d-5684-b7b6-d8193f3e46c0"
+GraphViz = "f526b714-d49f-11e8-06ff-31ed36ee7ee0"
+Plots = "91a5bcdd-55d7-5caf-9e0b-520d859cae80"
+
+[extensions]
+GraphVizExt = "GraphViz"
+GraphVizSimpleExt = "Colors"
+PlotsExt = ["DataFrames", "Plots"]
+
 [compat]
 DataStructures = "0.18"
 Graphs = "1"
@@ -37,19 +49,8 @@ StatsBase = "0.28, 0.29, 0.30, 0.31, 0.32, 0.33, 0.34"
 TimespanLogging = "0.1"
 julia = "1.8"
 
-[extensions]
-GraphVizSimpleExt = "Colors"
-GraphVizExt = "GraphViz"
-PlotsExt = ["DataFrames", "Plots"]
-
 [extras]
 Colors = "5ae59095-9a9b-59fe-a467-6f913c188581"
 DataFrames = "a93c6f00-e57d-5684-b7b6-d8193f3e46c0"
 GraphViz = "f526b714-d49f-11e8-06ff-31ed36ee7ee0"
 Plots = "91a5bcdd-55d7-5caf-9e0b-520d859cae80"
-
-[weakdeps]
-Colors = "5ae59095-9a9b-59fe-a467-6f913c188581"
-DataFrames = "a93c6f00-e57d-5684-b7b6-d8193f3e46c0"
-GraphViz = "f526b714-d49f-11e8-06ff-31ed36ee7ee0"
-Plots = "91a5bcdd-55d7-5caf-9e0b-520d859cae80"

src/Dagger.jl

@@ -41,6 +41,7 @@ include("scopes.jl")
 include("utils/scopes.jl")
 include("eager_thunk.jl")
 include("queue.jl")
+include("task_monitor.jl")
 include("thunk.jl")
 include("submission.jl")
 include("chunks.jl")

src/queue.jl

@@ -14,8 +14,11 @@ enqueue!(::EagerTaskQueue, spec::Pair{EagerTaskSpec,EagerThunk}) =
 enqueue!(::EagerTaskQueue, specs::Vector{Pair{EagerTaskSpec,EagerThunk}}) =
     eager_launch!(specs)
 
-enqueue!(spec::Pair{EagerTaskSpec,EagerThunk}) =
-    enqueue!(get_options(:task_queue, EagerTaskQueue()), spec)
+function enqueue!(spec::Pair{EagerTaskSpec,EagerThunk}) 
+    println("Hi There")
+    enqueue!(MONITOR_QUEUE,spec)
+end
+
 enqueue!(specs::Vector{Pair{EagerTaskSpec,EagerThunk}}) =
     enqueue!(get_options(:task_queue, EagerTaskQueue()), specs)
 

src/task_monitor.jl

@@ -0,0 +1,36 @@
+using ProgressMeter
+
+struct MonitorTaskQueue <: AbstractTaskQueue
+    runningTasks::Vector{Pair{EagerTaskSpec,EagerThunk}}
+    finishedTasks::Vector{Pair{EagerTaskSpec,EagerThunk}}
+    MonitorTaskQueue() = new(Pair{EagerTaskSpec,EagerThunk}[], Pair{EagerTaskSpec,EagerThunk}[])
+end
+function enqueue!(queue::MonitorTaskQueue, spec::Pair{EagerTaskSpec,EagerThunk})
+    push!(queue.runningTasks, spec)
+    upper = get_options(:task_queue, EagerTaskQueue())
+    enqueue!(upper,spec)    
+end
+
+function enqueue!(queue::MonitorTaskQueue, specs::Vector{Pair{EagerTaskSpec,EagerThunk}})
+    append!(queue.runningTasks, specs)
+end
+
+const MONITOR_QUEUE = MonitorTaskQueue()
+
+function monitor()
+        errormonitor(Threads.@spawn begin
+            runningTasks = MONITOR_QUEUE.runningTasks
+            finishedTasks = MONITOR_QUEUE.finishedTasks
+            meter = Progress(length(runningTasks))
+            while !isempty(runningTasks)
+                for (i,task) in reverse(collect(enumerate(runningTasks)))
+                    if isready(task[2])
+                        next!(meter)
+                        push!(finishedTasks,task)
+                        deleteat!(runningTasks,i)
+                    end
+                end
+                sleep(0.1)
+            end
+        end)
+end

src/thunk.jl

@@ -309,7 +309,7 @@ function spawn(f, args...; kwargs...)
     args_kwargs = args_kwargs_to_pairs(args, kwargs)
 
     # Get task queue, and don't let it propagate
-    task_queue = get_options(:task_queue, EagerTaskQueue())
+    task_queue = get_options(:task_queue, MONITOR_QUEUE)
     options = NamedTuple(filter(opt->opt[1] != :task_queue, Base.pairs(options)))
     propagates = filter(prop->prop != :task_queue, propagates)
     options = merge(options, (;propagates))