Merge pull request #120 from mschauer/asynch

Fix nv_ and add queue tests

Project.toml

@@ -1,7 +1,7 @@
 name = "ZigZagBoomerang"
 uuid = "36347407-b186-4a6a-8c98-4f4567861712"
 authors = ["Sebastiano Grazzi and Moritz Schauer"]
-version = "0.13.0"
+version = "0.13.1"
 
 [deps]
 DataStructures = "864edb3b-99cc-5e75-8d2d-829cb0a9cfe8"

src/morepriorityqueues.jl

@@ -58,7 +58,9 @@ struct PartialQueue{U,T,S,R}
     minima::R
     nregions::Int
 end
-nv_(G) = length(G)
+nv_(G::Vector) = length(G)
+nv_(G::AbstractGraph) = nv(G)
+
 div1(a,b) = (a-1)÷b + 1
 rkey(q, key) = div1(q.nregions*key, nv_(q.G))
 rkey((nregions, nv)::Tuple, key) = div1(nregions*key, nv)
@@ -85,25 +87,27 @@ function check(q::PartialQueue)
     end
 end
 
-function checkqueue(q::PartialQueue)
+function checkqueue(q::PartialQueue; fullcheck=false)
     minima = [Pair{Int64, Float64}[] for _ in 1:q.nregions]
     for i in 1:length(q.vals)
         q.ripes[i] == localmin(q, i) || error("Internal error")   
         q.ripes[i] && push!(minima[div1(q.nregions*i, nv_(q.G))], i=>q.vals[i])
     end
-    CHECKPQ && for i in 1:q.nregions
+    (CHECKPQ || fullcheck) && for i in 1:q.nregions
         if Set(first.(q.minima[i])) != Set(first.(minima[i]))
             println(setdiff(minima[i], q.minima[i]))
             println(setdiff(q.minima[i], minima[i]))
-
             error("corrupted")
         end
     end
+    return 
 end
 
 
 
 
+
+
 function collectmin(q::PartialQueue)
     all(isempty.(q.minima)) || error("Full queue")
     for i in findall(q.ripes)

test/priority.jl

@@ -1,5 +1,10 @@
 using DataStructures
 using ZigZagBoomerang: LinearQueue, PriorityQueues, rcat
+using ZigZagBoomerang, Graphs
+using ZigZagBoomerang: PartialQueue, dequeue!, saturate, rkey, div1
+using Graphs: Edge
+using Random
+using Base.Threads
 
 @testset "Queues" begin
     L = LinearQueue(0:2, [3.0, 1.0, 0.5])
@@ -31,4 +36,53 @@ using Test
     Q = PartialQueue(G, vals)
 
     Q[2] = 3
+
+
+
+
+    Random.seed!(1)
+
+    # number of regions
+    nregions = 2
+    # number of coordinates/keys
+    d = 10
+
+    # time surface
+    times = rand(d)
+
+    # undirected graph of neighbours
+    G = Graph(Edge.(([i=>i+1 for i in 1:d-1])))
+
+    # Make a Queue-structure keeping track of local minima
+    # (a coordinate is a local minima if it's time is smaller than all neighbours' (via the graph) times)
+    Q = PartialQueue(G, times, nregions)
+
+    # key 1:5 is in region 1, keys 6:10 in region 2
+    # we can work on different regions in parallel... see below
+    @test rkey(Q,5) == 1 
+    @test rkey(Q,6) == 2 
+
+
+    # dequeue! some local minima to work with
+    minima = dequeue!(Q) 
+
+    @test peek(Q) == [[],[]] # currently all local minima are removed from the queue to be worked on
+
+    # update/increment local time, (in parallel thanks to the threads makro)
+    @threads for r in 1:nregions
+        for (i,t) in minima[r]
+            println("work with $i, $t on $(Threads.threadid())")
+            Q[i] = t + rand()
+        end
+    end
+
+    # internal check
+    @test try 
+            ZigZagBoomerang.checkqueue(Q, fullcheck=true) == nothing
+          catch 
+            false 
+          end  
+
+    # By now we have a new set of local minimas we can handle in threads 
+    @test peek(Q) != [[],[]]
 end

test/runtests.jl

@@ -18,4 +18,5 @@ include("staticarrays.jl")
 include("sticky.jl")
 include("sticky_test.jl")
 
+include("priority.jl")
 #include("forwarddiff.jl")