waiting on register modification events

src/ProtocolZoo/ProtocolZoo.jl

@@ -208,9 +208,14 @@ 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.nodeA.stateindices.waiter)
+                @yield lock(prot.nodeB.stateindices.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 +402,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.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.nodeB.tag_waiter)
+                else
+                    @yield timeout(prot.sim, prot.period)
+                end
                 continue
             end
         end

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,58 @@
+#using QuantumSavory: AsymmetricSemaphore
 # TODO better constructors
 # TODO am I overusing Ref
+
+using ConcurrentSim
+using ResumableFunctions
+import Base: unlock, lock
+import Base: getindex, setindex!
+
+"""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
+
+"""Vector with a semaphore where processes can wait on until there's a change in the vector"""
+struct StateIndexVector
+    data::Vector{Int}
+    waiter::AsymmetricSemaphore
+end
+
+function StateIndexVector(data::Vector{Int})
+    env = ConcurrentSim.Simulation()
+    return StateIndexVector(data, AsymmetricSemaphore(env))
+end
+
+function getindex(vec::StateIndexVector, index::Int)
+    return vec.data[index]
+end
+
+function setindex!(vec::StateIndexVector, value::Int, index::Int)
+    vec.data[index] = value
+    unlock(vec.waiter)
+end
+
 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,15 +76,16 @@ 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)))
-Register(traits,reprs,bg) = Register(traits,reprs,bg,fill(nothing,length(traits)),zeros(Int,length(traits)),zeros(length(traits)))
+Register(traits,reprs,bg) = Register(traits,reprs,bg,fill(nothing,length(traits)),StateIndexVector(zeros(Int,length(traits))),zeros(length(traits)))
 Register(traits,bg::Base.AbstractVecOrTuple{<:Union{Nothing,<:AbstractBackground}}) = Register(traits,default_repr.(traits),bg)
 Register(traits,reprs::Base.AbstractVecOrTuple{<:AbstractRepresentation}) = Register(traits,reprs,fill(nothing,length(traits)))
 Register(traits) = Register(traits,default_repr.(traits))