Merge pull request #247 from JuliaParallel/jps/lazy-cache-errored-fix

Fix bug in lazy API state.errored check

Author: jpsamaroo

Diff for: Project.toml

@@ -1,6 +1,6 @@
 name = "Dagger"
 uuid = "d58978e5-989f-55fb-8d15-ea34adc7bf54"
-version = "0.11.9"
+version = "0.11.10"
 
 [deps]
 Colors = "5ae59095-9a9b-59fe-a467-6f913c188581"

Diff for: src/sch/Sch.jl

@@ -447,7 +447,7 @@ function compute_dag(ctx, d::Thunk; options=SchedulerOptions())
         @async cleanup_proc(state, p)
     end
     value = state.cache[d] # TODO: move(OSProc(), state.cache[d])
-    if state.errored[d]
+    if get(state.errored, d, false)
         throw(value)
     end
     if options.checkpoint !== nothing
@@ -747,7 +747,9 @@ function finish_task!(ctx, state, node, thunk_failed)
                     end
                     GC.@preserve inp begin
                         pop!(state.cache, inp)
-                        pop!(state.errored, inp)
+                        if haskey(state.errored, inp)
+                            pop!(state.errored, inp)
+                        end
                     end
                 elseif inp isa Chunk
                     push!(to_evict, inp)
@@ -788,7 +790,7 @@ function fire_tasks!(ctx, thunks::Vector{<:Tuple}, (gproc, proc), state)
             if data !== nothing
                 # cache hit
                 state.cache[thunk] = data
-                thunk_failed = state.errored[thunk]
+                thunk_failed = get(state.errored, thunk, false)
                 finish_task!(ctx, state, thunk, thunk_failed)
                 continue
             else

Diff for: test/cache.jl

@@ -1,14 +1,10 @@
-using Base.Test
-using Dagger
-using MemPool
-using Base.Test
-
 @testset "cache" begin
-    @everywhere gc(true)
-    # set available memory to 80KB on each worker
+    @everywhere GC.gc()
+    #= set available memory to 80KB on each worker
     @everywhere empty!(MemPool.datastore)
     @everywhere empty!(MemPool.lru_order)
     @everywhere MemPool.max_memsize[] = 8*10^4
+    =#
 
     thunks1 = map(delayed(_ -> rand(10^3), cache=true), workers())
     sum1 = delayed((x...)->sum([x...]))(map(delayed(sum), thunks1)...)
@@ -19,12 +15,9 @@ using Base.Test
     @test s1 == collect(sum1)
     @test -collect(sum1) == collect(sum2)
 
-    thunks1 = map(delayed(_ -> rand(10^4), cache=true), workers())
-    sum1 = delayed((x...)->sum([x...]))(map(delayed(sum), thunks1)...)
-    s1 = collect(sum1)
-    thunks2 = map(delayed(_ -> rand(10^4), cache=true), workers())
-    sum2 = delayed((x...)->sum([x...]))(map(delayed(sum), thunks2)...)
-    s2 = collect(sum2) # this should evict thunk1s from memory
-    @test s1 != collect(sum1)
-    @test s2 != collect(sum2)
+    # Issue #246
+    z = delayed(identity)(10)
+    zz = delayed(identity; cache=true)(10)
+    @test collect(z) == collect(z)
+    @test collect(zz) == collect(zz)
 end

Diff for: test/runtests.jl

@@ -30,11 +30,11 @@ include("ui.jl")
 include("checkpoint.jl")
 include("domain.jl")
 include("array.jl")
+include("cache.jl")
 try # TODO: Fault tolerance is sometimes unreliable
 include("fault-tolerance.jl")
 catch
 end
 println(stderr, "tests done. cleaning up...")
 Dagger.cleanup()
-#include("cache.jl")
 println(stderr, "all done.")