Merge pull request #221 from JuliaParallel/jps/net-min

Optimize thunk placement for input affinity

Author: jpsamaroo

Diff for: Project.toml

@@ -1,6 +1,6 @@
 name = "Dagger"
 uuid = "d58978e5-989f-55fb-8d15-ea34adc7bf54"
-version = "0.13.4"
+version = "0.13.5"
 
 [deps]
 Colors = "5ae59095-9a9b-59fe-a467-6f913c188581"
@@ -21,7 +21,7 @@ UUIDs = "cf7118a7-6976-5b1a-9a39-7adc72f591a4"
 
 [compat]
 Colors = "0.10, 0.11, 0.12"
-MemPool = "0.3.4"
+MemPool = "0.3.5"
 Requires = "1"
 StatsBase = "0.28, 0.29, 0.30, 0.31, 0.32, 0.33"
 TableOperations = "1"

Diff for: benchmarks/benchmark.jl

@@ -204,17 +204,17 @@ function nmf_suite(ctx; dagger, accel)
                     @info "Starting $_nw worker Dagger NNMF (scale by $_scale)"
                     if $accel == "cuda"
                         # FIXME: Allocate with CUDA.rand if possible
-                        $X[] = Dagger.mapchunks(CUDA.cu, compute(rand(Blocks($nrow, $bsz), Float32, $nrow, $ncol); options=$opts))
-                        $W[] = Dagger.mapchunks(CUDA.cu, compute(rand(Blocks($nrow, $bsz), Float32, $nrow, $nfeatures); options=$opts))
-                        $H[] = Dagger.mapchunks(CUDA.cu, compute(rand(Blocks($nrow, $bsz), Float32, $nfeatures, $ncol); options=$opts))
+                        $X[] = Dagger.mapchunks(CUDA.cu, compute(rand(Blocks($bsz, $bsz), Float32, $nrow, $ncol); options=$opts))
+                        $W[] = Dagger.mapchunks(CUDA.cu, compute(rand(Blocks($bsz, $bsz), Float32, $nrow, $nfeatures); options=$opts))
+                        $H[] = Dagger.mapchunks(CUDA.cu, compute(rand(Blocks($bsz, $bsz), Float32, $nfeatures, $ncol); options=$opts))
                     elseif $accel == "amdgpu"
-                        $X[] = Dagger.mapchunks(ROCArray, compute(rand(Blocks($nrow, $ncol÷$p), Float32, $nrow, $ncol); options=$opts))
-                        $W[] = Dagger.mapchunks(ROCArray, compute(rand(Blocks($nrow, $ncol÷$p), Float32, $nrow, $nfeatures); options=$opts))
-                        $H[] = Dagger.mapchunks(ROCArray, compute(rand(Blocks($nrow, $ncol÷$p), Float32, $nfeatures, $ncol); options=$opts))
+                        $X[] = Dagger.mapchunks(ROCArray, compute(rand(Blocks($bsz, $bsz), Float32, $nrow, $ncol); options=$opts))
+                        $W[] = Dagger.mapchunks(ROCArray, compute(rand(Blocks($bsz, $bsz), Float32, $nrow, $nfeatures); options=$opts))
+                        $H[] = Dagger.mapchunks(ROCArray, compute(rand(Blocks($bsz, $bsz), Float32, $nfeatures, $ncol); options=$opts))
                     elseif $accel == "cpu"
-                        $X[] = compute(rand(Blocks($nrow, $bsz), Float32, $nrow, $ncol); options=$opts)
-                        $W[] = compute(rand(Blocks($nrow, $bsz), Float32, $nrow, $nfeatures); options=$opts)
-                        $H[] = compute(rand(Blocks($nrow, $bsz), Float32, $nfeatures, $ncol); options=$opts)
+                        $X[] = compute(rand(Blocks($bsz, $bsz), Float32, $nrow, $ncol); options=$opts)
+                        $W[] = compute(rand(Blocks($bsz, $bsz), Float32, $nrow, $nfeatures); options=$opts)
+                        $H[] = compute(rand(Blocks($bsz, $bsz), Float32, $nfeatures, $ncol); options=$opts)
                     end
                 end teardown=begin
                     if render != "" && !live
@@ -280,7 +280,7 @@ function main()
         push!(d3r, LinePlot(:core, :loadavg, "CPU Load Average", "Average Running Threads"))
         push!(d3r, LinePlot(:core, :bytes, "Allocated Bytes", "Bytes"))
         push!(d3r, LinePlot(:core, :mem, "Available Memory", "% Free"))
-        push!(d3r, GraphPlot(:core, :id, :timeline, :profile, "DAG"))
+        #push!(d3r, GraphPlot(:core, :id, :timeline, :profile, "DAG"))
         push!(lw.creation_handlers, d3r)
         push!(lw.deletion_handlers, d3r)
         ml.aggregators[:logwindow] = lw

Diff for: src/Dagger.jl

@@ -51,6 +51,8 @@ include("array/sort.jl")
 
 # Other
 include("ui/graph.jl")
+include("ui/gantt-common.jl")
+include("ui/gantt-text.jl")
 
 include("table/dtable.jl")
 include("table/operations.jl")
@@ -63,15 +65,15 @@ function __init__()
 
     @require Luxor="ae8d54c2-7ccd-5906-9d76-62fc9837b5bc" begin
         # Gantt chart renderer
-        include("ui/gantt.jl")
+        include("ui/gantt-luxor.jl")
     end
     @require Mux="a975b10e-0019-58db-a62f-e48ff68538c9" begin
         # Gantt chart HTTP server
-        include("ui/gantt-server.jl")
+        include("ui/gantt-mux.jl")
     end
     @require ProfileSVG="132c30aa-f267-4189-9183-c8a63c7e05e6" begin
         # Profile renderer
-        include("ui/profile.jl")
+        include("ui/profile-profilesvg.jl")
     end
     @require FFMPEG="c87230d0-a227-11e9-1b43-d7ebe4e7570a" begin
         @require FileIO="5789e2e9-d7fb-5bc7-8068-2c6fae9b9549" begin

Diff for: src/chunks.jl

@@ -53,17 +53,13 @@ mutable struct Chunk{T, H, P<:Processor, S<:AbstractScope}
     processor::P
     scope::S
     persist::Bool
-    function (::Type{Chunk{T,H,P,S}})(::Type{T}, domain, handle, processor, scope, persist) where {T,H,P,S}
-        c = new{T,H,P,S}(T, domain, handle, processor, scope, persist)
-        finalizer(x -> @async(myid() == 1 && free!(x)), c)
-        c
-    end
 end
 
 domain(c::Chunk) = c.domain
 chunktype(c::Chunk) = c.chunktype
 persist!(t::Chunk) = (t.persist=true; t)
 shouldpersist(p::Chunk) = t.persist
+processor(c::Chunk) = c.processor
 affinity(c::Chunk) = affinity(c.handle)
 
 function unrelease(c::Chunk{<:Any,DRef})
@@ -117,7 +113,7 @@ move(to_proc::Processor, x) =
     move(OSProc(), to_proc, x)
 
 ### ChunkIO
-affinity(r::DRef) = Pair{OSProc, UInt64}[OSProc(r.owner) => r.size]
+affinity(r::DRef) = OSProc(r.owner)=>r.size
 function affinity(r::FileRef)
     if haskey(MemPool.who_has_read, r.file)
         Pair{OSProc, UInt64}[OSProc(dref.owner) => r.size for dref in MemPool.who_has_read[r.file]]

Diff for: src/processor.jl

@@ -101,7 +101,8 @@ struct OSProc <: Processor
     end
 end
 const OSPROC_CACHE = Dict{Int,Vector{Processor}}()
-children(proc::OSProc) = OSPROC_CACHE[proc.pid]
+get_parent(proc::OSProc) = proc
+children(proc::OSProc) = get(OSPROC_CACHE, proc.pid, Processor[])
 function get_proc_hierarchy()
     children = Processor[]
     for name in keys(PROCESSOR_CALLBACKS)
@@ -158,11 +159,17 @@ iscompatible_arg(proc::ThreadProc, opts, x) = true
 @static if VERSION >= v"1.3.0-DEV.573"
     function execute!(proc::ThreadProc, f, args...)
         tls = get_tls()
-        task = Threads.@spawn begin
+        task = Task() do
             set_tls!(tls)
             prof_task_put!(tls.sch_handle.thunk_id.id)
             f(args...)
         end
+        ret = ccall(:jl_set_task_tid, Cint, (Any, Cint), task, proc.tid-1)
+        if ret == 0
+            error("jl_set_task_tid == 0")
+        end
+        @assert Threads.threadid(task) == proc.tid
+        schedule(task)
         try
             fetch(task)
         catch err