Merge pull request #350 from JuliaParallel/kr/mempool-fix

Remove `poolset` kwarg that doesn't exist anymore

Author: jpsamaroo

Diff for: Project.toml

@@ -1,6 +1,6 @@
 name = "Dagger"
 uuid = "d58978e5-989f-55fb-8d15-ea34adc7bf54"
-version = "0.15.0"
+version = "0.15.1"
 
 [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"
 ContextVariablesX = "0.1"
-MemPool = "0.3.5"
+MemPool = "0.3.5, 0.4"
 Requires = "1"
 StatsBase = "0.28, 0.29, 0.30, 0.31, 0.32, 0.33"
 julia = "1.6"

Diff for: src/array/darray.jl

@@ -115,29 +115,11 @@ mutable struct DArray{T,N,F} <: ArrayOp{T, N}
     subdomains::AbstractArray{ArrayDomain{N}, N}
     chunks::AbstractArray{Union{Chunk,Thunk}, N}
     concat::F
-    freed::Threads.Atomic{UInt8}
     function DArray{T,N,F}(domain, subdomains, chunks, concat::Function) where {T, N,F}
-        new(domain, subdomains, chunks, concat, Threads.Atomic{UInt8}(0))
+        new(domain, subdomains, chunks, concat)
     end
 end
 
-function free_chunks(chunks)
-    @sync for c in chunks
-        if c isa Chunk{<:Any, DRef}
-            # increment refcount on the master node
-            @async free!(c.handle)
-        elseif c isa Thunk
-            free_chunks(c.inputs)
-        end
-    end
-end
-
-function free!(x::DArray)
-    freed = Bool(Threads.atomic_cas!(x.freed, UInt8(0), UInt8(1)))
-    !freed && @async Dagger.free_chunks(x.chunks)
-    nothing
-end
-
 # mainly for backwards-compatibility
 DArray{T, N}(domain, subdomains, chunks) where {T,N} = DArray(T, domain, subdomains, chunks)
 

Diff for: src/chunks.jl

@@ -62,21 +62,6 @@ shouldpersist(p::Chunk) = t.persist
 processor(c::Chunk) = c.processor
 affinity(c::Chunk) = affinity(c.handle)
 
-function unrelease(c::Chunk{<:Any,DRef})
-    # set spilltodisk = true if data is still around
-    try
-        destroyonevict(c.handle, false)
-        return c
-    catch err
-        if isa(err, KeyError)
-            return nothing
-        else
-            rethrow(err)
-        end
-    end
-end
-unrelease(c::Chunk) = c
-
 Base.:(==)(c1::Chunk, c2::Chunk) = c1.handle == c2.handle
 Base.hash(c::Chunk, x::UInt64) = hash(c.handle, x)
 
@@ -248,26 +233,11 @@ Base.map(f, s::Shard) = [Dagger.spawn(f, c) for c in values(s.chunks)]
 Create a chunk from sequential object `x` which resides on `proc`.
 """
 function tochunk(x::X, proc::P=OSProc(), scope::S=AnyScope(); persist=false, cache=false) where {X,P,S}
-    ref = poolset(x, destroyonevict=persist ? false : cache)
+    ref = poolset(x)
     Chunk{X,typeof(ref),P,S}(X, domain(x), ref, proc, scope, persist)
 end
 tochunk(x::Union{Chunk, Thunk}, proc=nothing, scope=nothing) = x
 
-# Check to see if the node is set to persist
-# if it is foce can override it
-function free!(s::Chunk{X,DRef,P,S}; force=true, cache=false) where {X,P,S}
-    if force || !s.persist
-        if cache
-            try
-                destroyonevict(s.handle, true) # keep around, but remove when evicted
-            catch err
-                isa(err, KeyError) || rethrow(err)
-            end
-        end
-    end
-end
-free!(x; force=true,cache=false) = x # catch-all for non-chunks
-
 function savechunk(data, dir, f)
     sz = open(joinpath(dir, f), "w") do io
         serialize(io, MemPool.MMWrap(data))

Diff for: src/compute.jl

@@ -1,4 +1,4 @@
-export stage, cached_stage, compute, debug_compute, free!, cleanup
+export stage, cached_stage, compute, debug_compute, cleanup
 
 ###### Scheduler #######
 

Diff for: src/sch/Sch.jl

@@ -7,7 +7,7 @@ import Random: randperm
 
 import ..Dagger
 import ..Dagger: Context, Processor, Thunk, WeakThunk, ThunkFuture, ThunkFailedException, Chunk, OSProc, AnyScope
-import ..Dagger: order, free!, dependents, noffspring, istask, inputs, unwrap_weak_checked, affinity, tochunk, timespan_start, timespan_finish, unrelease, procs, move, capacity, chunktype, processor, default_enabled, get_processors, get_parent, execute!, rmprocs!, addprocs!, thunk_processor, constrain, cputhreadtime
+import ..Dagger: order, dependents, noffspring, istask, inputs, unwrap_weak_checked, affinity, tochunk, timespan_start, timespan_finish, procs, move, capacity, chunktype, processor, default_enabled, get_processors, get_parent, execute!, rmprocs!, addprocs!, thunk_processor, constrain, cputhreadtime
 
 const OneToMany = Dict{Thunk, Set{Thunk}}
 
@@ -850,8 +850,7 @@ function fire_tasks!(ctx, thunks::Vector{<:Tuple}, (gproc, proc), state)
         state.running_on[thunk] = gproc
         if thunk.cache && thunk.cache_ref !== nothing
             # the result might be already cached
-            data = unrelease(thunk.cache_ref) # ask worker to keep the data around
-                                              # till this compute cycle frees it
+            data = thunk.cache_ref
             if data !== nothing
                 # cache hit
                 state.cache[thunk] = data