waiting on register modification events

src/ProtocolZoo/ProtocolZoo.jl

@@ -208,9 +208,13 @@ end
         b_ = findfreeslot(prot.net[prot.nodeB]; randomize=prot.randomize, margin=margin)
 
         if isnothing(a_) || isnothing(b_)
-            isnothing(prot.retry_lock_time) && error("We do not yet support waiting on register to make qubits available") # TODO
-            @debug "EntanglerProt between $(prot.nodeA) and $(prot.nodeB)|round $(round): Failed to find free slots. \nGot:\n1. \t $a_ \n2.\t $b_ \n retrying..."
-            @yield timeout(prot.sim, prot.retry_lock_time)
+            if isnothing(prot.retry_lock_time)
+                @debug "EntanglerProt between $(prot.nodeA) and $(prot.nodeB)|round $(round): Failed to find free slots. \nGot:\n1. \t $a_ \n2.\t $b_ \n waiting..."
+                @yield lock(prot.net[prot.nodeA].tag_waiter) | lock(prot.net[prot.nodeB].tag_waiter)
+            else
+                @debug "EntanglerProt between $(prot.nodeA) and $(prot.nodeB)|round $(round): Failed to find free slots. \nGot:\n1. \t $a_ \n2.\t $b_ \n retrying..."
+                @yield timeout(prot.sim, prot.retry_lock_time)
+            end
             continue
         end
         # we are now certain that a_ and b_ are not nothing. The compiler is not smart enough to figure this out
@@ -397,20 +401,27 @@ function EntanglementConsumer(net::RegisterNet, nodeA::Int, nodeB::Int; kwargs..
 end
 
 @resumable function (prot::EntanglementConsumer)()
-    if isnothing(prot.period)
-        error("In `EntanglementConsumer` we do not yet support waiting on register to make qubits available") # TODO
-    end
     while true
         query1 = query(prot.net[prot.nodeA], EntanglementCounterpart, prot.nodeB, ❓; locked=false, assigned=true) # TODO Need a `querydelete!` dispatch on `Register` rather than using `query` here followed by `untag!` below
         if isnothing(query1)
             @debug "EntanglementConsumer between $(prot.nodeA) and $(prot.nodeB): query on first node found no entanglement"
-            @yield timeout(prot.sim, prot.period)
+            if isnothing(prot.period)
+                @debug "Waiting on changes in $(prot.nodeA)"
+                @yield lock(prot.net[prot.nodeA].tag_waiter)
+            else
+                @yield timeout(prot.sim, prot.period)
+            end
             continue
         else
             query2 = query(prot.net[prot.nodeB], EntanglementCounterpart, prot.nodeA, query1.slot.idx; locked=false, assigned=true)
             if isnothing(query2) # in case EntanglementUpdate hasn't reached the second node yet, but the first node has the EntanglementCounterpart
                 @debug "EntanglementConsumer between $(prot.nodeA) and $(prot.nodeB): query on second node found no entanglement (yet...)"
-                @yield timeout(prot.sim, prot.period)
+                if isnothing(prot.period)
+                    @debug "Waiting on changes in $(prot.nodeB)"
+                    @yield lock(prot.net[prot.nodeB].tag_waiter)
+                else
+                    @yield timeout(prot.sim, prot.period)
+                end
                 continue
             end
         end
@@ -431,7 +442,9 @@ end
         push!(prot.log, (now(prot.sim), ob1, ob2))
         unlock(q1)
         unlock(q2)
-        @yield timeout(prot.sim, prot.period)
+        if !isnothing(prot.period)
+            @yield timeout(prot.sim, prot.period)
+        end
     end
 end
 
@@ -441,4 +454,4 @@ include("swapping.jl")
 include("switches.jl")
 using .Switches
 
-end # module
+end # module

src/ProtocolZoo/cutoff.jl

@@ -33,9 +33,6 @@ function CutoffProt(sim::Simulation, net::RegisterNet, node::Int; kwargs...)
 end
 
 @resumable function (prot::CutoffProt)()
-    if isnothing(prot.period)
-        error("In `CutoffProt` we do not yet support quing up and waiting on register") # TODO
-    end
     reg = prot.net[prot.node]
     while true
         for slot in reg # TODO these should be done in parallel, otherwise we will be waiting on each slot, greatly slowing down the cutoffs
@@ -72,6 +69,8 @@ end
 
             unlock(slot)
         end
-        @yield timeout(prot.sim, prot.period)
+        if !isnothing(prot.period)
+            @yield timeout(prot.sim, prot.period)
+        end
     end
 end

src/ProtocolZoo/swapping.jl

@@ -68,8 +68,11 @@ end
     while rounds != 0
         qubit_pair_ = findswapablequbits(prot.net, prot.node, prot.nodeL, prot.nodeH, prot.chooseL, prot.chooseH; agelimit=prot.agelimit)
         if isnothing(qubit_pair_)
-            isnothing(prot.retry_lock_time) && error("We do not yet support waiting on register to make qubits available") # TODO
-            @yield timeout(prot.sim, prot.retry_lock_time)
+            if isnothing(prot.retry_lock_time)
+                @yield lock(prot.net[prot.node].tag_waiter)
+            else
+                @yield timeout(prot.sim, prot.retry_lock_time)
+            end
             continue
         end
         # The compiler is not smart enough to figure out that qubit_pair_ is not nothing, so we need to tell it explicitly. A new variable name is needed due to @resumable.

src/QuantumSavory.jl

@@ -71,14 +71,15 @@ export
     # plots.jl
     registernetplot, registernetplot!, registernetplot_axis, resourceplot_axis
 
-
 #TODO you can not assume you can always in-place modify a state. Have all these functions work on stateref, not stateref[]
 # basically all ::QuantumOptics... should be turned into ::Ref{...}... but an abstract ref
 
 include("traits_and_defaults.jl")
 
 include("tags.jl")
 
+include("semaphore.jl")
+
 include("states_registers.jl")
 include("quantumchannel.jl")
 include("messagebuffer.jl")
@@ -116,4 +117,5 @@ include("ProtocolZoo/ProtocolZoo.jl")
 
 include("precompile.jl")
 
+
 end # module

src/queries.jl

@@ -20,6 +20,7 @@ function tag!(ref::RegRef, tag)
     id = guid()
     push!(ref.reg.guids, id)
     ref.reg.tag_info[id] = (;tag, slot=ref.idx, time=now(get_time_tracker(ref)))
+    unlock(ref.reg.tag_waiter)
     return id
 end
 
@@ -41,6 +42,7 @@ function untag!(ref::RegOrRegRef, id::Integer)
     isnothing(i) ? throw(QueryError("Attempted to delete a nonexistent tag id", untag!, id)) : deleteat!(reg.guids, i) # TODO make sure there is a clear error message
     to_be_deleted = reg.tag_info[id]
     delete!(reg.tag_info, id)
+    unlock(reg.tag_waiter)
     return to_be_deleted
 end
 

src/semaphore.jl

@@ -0,0 +1,29 @@
+using ConcurrentSim
+using ResumableFunctions
+import Base: unlock, lock
+
+"""Multiple processes can wait on this semaphore for a permission to run given by another process"""
+struct AsymmetricSemaphore
+    nbwaiters::Ref{Int}
+    lock::Resource
+end
+AsymmetricSemaphore(sim) = AsymmetricSemaphore(Ref(0), Resource(sim,1,level=1)) # start locked
+
+function Base.lock(s::AsymmetricSemaphore)
+    return @process _lock(s.lock.env, s)
+end
+
+@resumable function _lock(sim, s::AsymmetricSemaphore)
+    s.nbwaiters[] += 1
+    @yield lock(s.lock)
+    s.nbwaiters[] -= 1
+    if s.nbwaiters[] > 0
+        unlock(s.lock)
+    end
+end
+
+function unlock(s::AsymmetricSemaphore)
+    if s.nbwaiters[] > 0
+        unlock(s.lock)
+    end
+end

src/states_registers.jl

@@ -1,5 +1,7 @@
+#using QuantumSavory: AsymmetricSemaphore
 # TODO better constructors
 # TODO am I overusing Ref
+
 struct StateRef
     state::Base.RefValue{Any} # TODO it would be nice if this was not abstract but `uptotime!` converts between types... maybe make StateRef{T} state::RefValue{T} and a new function that swaps away the backpointers in the appropriate registers
     registers::Vector{Any} # TODO Should be Vector{Register}, but right now we occasionally set it to nothing to deal with padded storage
@@ -23,11 +25,12 @@ struct Register # TODO better type description
     tag_info::Dict{Int128, @NamedTuple{tag::Tag, slot::Int, time::Float64}}
     guids::Vector{Int128}
     netparent::Ref{Any}
+    tag_waiter::AsymmetricSemaphore
 end
 
 function Register(traits, reprs, bg, sr, si, at)
     env = ConcurrentSim.Simulation()
-    Register(traits, reprs, bg, sr, si, at, [ConcurrentSim.Resource(env) for _ in traits], Dict{Int128, Tuple{Tag, Int64, Float64}}(), [], nothing)
+    Register(traits, reprs, bg, sr, si, at, [ConcurrentSim.Resource(env) for _ in traits], Dict{Int128, Tuple{Tag, Int64, Float64}}(), [], nothing, AsymmetricSemaphore(env))
 end
 
 Register(traits,reprs,bg,sr,si) = Register(traits,reprs,bg,sr,si,zeros(length(traits)))

test/runtests.jl

@@ -46,6 +46,8 @@ println("Starting tests with $(Threads.nthreads()) threads out of `Sys.CPU_THREA
 
 @doset "stateszoo_api"
 
+@doset "semaphore"
+
 if get(ENV,"QUANTUMSAVORY_PLOT_TEST","")=="true"
     using Pkg
     Pkg.add("GLMakie")

test/test_protocolzoo_entanglement_consumer.jl

@@ -2,6 +2,7 @@ using QuantumSavory
 using QuantumSavory.ProtocolZoo: EntanglerProt, SwapperProt, EntanglementTracker, EntanglementConsumer
 using Graphs
 using ConcurrentSim
+using ResumableFunctions
 using Test
 
 if isinteractive()
@@ -44,3 +45,50 @@ for n in 3:30
     end
 
 end
+
+# test for period=nothing
+for n in 3:30
+    println(n)
+    regsize = 10
+    net = RegisterNet([Register(regsize) for j in 1:n])
+    sim = get_time_tracker(net)
+
+    @resumable function delayedProts(sim)
+        @yield timeout(sim, 10)
+        for e in edges(net)
+            eprot = EntanglerProt(sim, net, e.src, e.dst; rounds=-1, randomize=true, margin=5, hardmargin=3)
+            @process eprot()
+        end
+
+        for v in 2:n-1
+            sprot = SwapperProt(sim, net, v; nodeL = <(v), nodeH = >(v), chooseL = argmin, chooseH = argmax, rounds = -1)
+            @process sprot()
+        end
+
+        for v in vertices(net)
+            etracker = EntanglementTracker(sim, net, v)
+            @process etracker()
+        end
+    end
+    econ = EntanglementConsumer(sim, net, 1, n; period=nothing)
+    @process econ()
+    @process delayedProts(sim)
+
+    run(sim, 1000) 
+
+    @test econ.log[1][1] > 10 #the process should start after 10
+    for i in 1:length(econ.log)
+        @test econ.log[i][2] ≈ 1.0
+        @test econ.log[i][3] ≈ 1.0
+    end
+
+end
+
+
+
+
+@test econ.log[1][1] > 10 #the process should start after 10
+for i in 1:length(econ.log)
+    @test econ.log[i][2] ≈ 1.0
+    @test econ.log[i][3] ≈ 1.0
+end