update EntanglementHistory when skipping over a node, randomize tests,

use a counter to keep track of race conditions in messagebuffer and lower bounds for stdlibs compat
QuantumSavory · Dec 29, 2023 · 788576c · 788576c
1 parent b6cfaf8
commit 788576c
Show file tree

Hide file tree

Showing 13 changed files with 87 additions and 55 deletions.
diff --git a/.github/workflows/benchmark-comment.yml b/.github/workflows/benchmark-comment.yml
@@ -25,7 +25,7 @@ jobs:
       - uses: actions/checkout@v4
 
       # restore records from the artifacts
-      - uses: dawidd6/action-download-artifact@v2
+      - uses: dawidd6/action-download-artifact@v3
         with:
           workflow: benchmark.yml
           name: performance-tracking

diff --git a/.github/workflows/benchmark.yml b/.github/workflows/benchmark.yml
@@ -48,7 +48,7 @@ jobs:
         run: echo ${{ github.event.pull_request.number }} > ./pull-request-number.artifact
 
       # save as artifacts (performance tracking (comment) workflow will use it)
-      - uses: actions/upload-artifact@v3
+      - uses: actions/upload-artifact@v4
         with:
           name: performance-tracking
           path: ./*.artifact
diff --git a/LICENSE b/LICENSE
@@ -1,6 +1,6 @@
 MIT License
 
-Copyright (c) 2023 Abhishek Bhatt
+Copyright (c) 2023 Stefan Krastanov
 
 Permission is hereby granted, free of charge, to any person obtaining a copy
 of this software and associated documentation files (the "Software"), to deal

diff --git a/Project.toml b/Project.toml
@@ -20,6 +20,7 @@ QuantumInterface = "5717a53b-5d69-4fa3-b976-0bf2f97ca1e5"
 QuantumOptics = "6e0679c1-51ea-5a7c-ac74-d61b76210b0c"
 QuantumOpticsBase = "4f57444f-1401-5e15-980d-4471b28d5678"
 QuantumSymbolics = "efa7fd63-0460-4890-beb7-be1bbdfbaeae"
+Random = "9a3f8284-a2c9-5f02-9a11-845980a1fd5c"
 Reexport = "189a3867-3050-52da-a836-e630ba90ab69"
 ResumableFunctions = "c5292f4c-5179-55e1-98c5-05642aab7184"
 Statistics = "10745b16-79ce-11e8-11f9-7d13ad32a3b2"
@@ -38,16 +39,17 @@ Distributions = "0.25"
 DocStringExtensions = "0.9"
 Graphs = "1.7.3"
 IterTools = "1.4.0"
-LinearAlgebra = "1.9"
+LinearAlgebra = "1"
 Makie = "0.19, 0.20"
 NetworkLayout = "0.4.4"
 PrecompileTools = "1"
-Printf = "1.9"
+Printf = "1"
 QuantumClifford = "0.8"
 QuantumInterface = "0.3.3"
 QuantumOptics = "1.0.5"
 QuantumOpticsBase = "0.4.17"
 QuantumSymbolics = "0.2.5"
+Random = "1"
 Reexport = "1.2.2"
 ResumableFunctions = "0.6.1"
 Statistics = "1"

diff --git a/README.md b/README.md
@@ -14,7 +14,7 @@
         <td>Continuous integration</td>
         <td>
             <a href="https://github.com/QuantumSavory/QuantumSavory.jl/actions?query=workflow%3ACI+branch%3Amaster"><img src="https://github.com/QuantumSavory/QuantumSavory.jl/actions/workflows/ci.yml/badge.svg" alt="GitHub Workflow Status"></a>
-            <a href="https://buildkite.com/quantumsavory/ci-buildkite?branch=master"><img src="https://badge.buildkite.com/2713c9978db76235fbe98094b7cad9a05f04671fd442739c47.svg?branch=master" alt="Buildkite Workflow Status"></a>
+            <a href="https://buildkite.com/quantumsavory/quantumsavory"><img src="https://badge.buildkite.com/2713c9978db76235fbe98094b7cad9a05f04671fd442739c47.svg" alt="Buildkite Workflow Status"></a>
         </td>
     </tr><tr></tr>
     <tr>

diff --git a/src/ProtocolZoo/ProtocolZoo.jl b/src/ProtocolZoo/ProtocolZoo.jl
@@ -26,7 +26,7 @@ Process(prot::AbstractProtocol, args...; kwargs...) = Process((e,a...;k...)->pro
     remote_node::Int
     remote_slot::Int
 end
-Base.show(io::IO, tag::EntanglementCounterpart) = print(io, "Entangled to $(tag.remote_node)|$(tag.remote_slot)")
+Base.show(io::IO, tag::EntanglementCounterpart) = print(io, "Entangled to $(tag.remote_node).$(tag.remote_slot)")
 Tag(tag::EntanglementCounterpart) = Tag(EntanglementCounterpart, tag.remote_node, tag.remote_slot)
 
 @kwdef struct EntanglementHistory
@@ -36,7 +36,7 @@ Tag(tag::EntanglementCounterpart) = Tag(EntanglementCounterpart, tag.remote_node
     swap_remote_slot::Int
     swapped_local::Int
 end
-Base.show(io::IO, tag::EntanglementHistory) = print(io, "Was entangled to $(tag.remote_node)|$(tag.remote_slot), but swapped with |$(tag.swapped_local) which was entangled to $(tag.swap_remote_node)|$(tag.swap_remote_slot)")
+Base.show(io::IO, tag::EntanglementHistory) = print(io, "Was entangled to $(tag.remote_node).$(tag.remote_slot), but swapped with .$(tag.swapped_local) which was entangled to $(tag.swap_remote_node).$(tag.swap_remote_slot)")
 Tag(tag::EntanglementHistory) = Tag(EntanglementHistory, tag.remote_node, tag.remote_slot, tag.swap_remote_node, tag.swap_remote_slot, tag.swapped_local)
 
 @kwdef struct EntanglementUpdateX
@@ -47,7 +47,7 @@ Tag(tag::EntanglementHistory) = Tag(EntanglementHistory, tag.remote_node, tag.re
     new_remote_slot::Int
     correction::Int
 end
-Base.show(io::IO, tag::EntanglementUpdateX) = print(io, "Update slot |$(tag.past_remote_slot) which used to be entangled to $(tag.past_local_node)|$(tag.past_local_slot) to be entangled to $(tag.new_remote_node)|$(tag.new_remote_slot) and apply correction X$(tag.correction)")
+Base.show(io::IO, tag::EntanglementUpdateX) = print(io, "Update slot .$(tag.past_remote_slot) which used to be entangled to $(tag.past_local_node).$(tag.past_local_slot) to be entangled to $(tag.new_remote_node).$(tag.new_remote_slot) and apply correction X$(tag.correction)")
 Tag(tag::EntanglementUpdateX) = Tag(EntanglementUpdateX, tag.past_local_node, tag.past_local_slot, tag.past_remote_slot, tag.new_remote_node, tag.new_remote_slot, tag.correction)
 
 @kwdef struct EntanglementUpdateZ
@@ -58,7 +58,7 @@ Tag(tag::EntanglementUpdateX) = Tag(EntanglementUpdateX, tag.past_local_node, ta
     new_remote_slot::Int
     correction::Int
 end
-Base.show(io::IO, tag::EntanglementUpdateZ) = print(io, "Update slot |$(tag.past_remote_slot) which used to be entangled to $(tag.past_local_node)|$(tag.past_local_slot) to be entangled to $(tag.new_remote_node)|$(tag.new_remote_slot) and apply correction Z$(tag.correction)")
+Base.show(io::IO, tag::EntanglementUpdateZ) = print(io, "Update slot .$(tag.past_remote_slot) which used to be entangled to $(tag.past_local_node).$(tag.past_local_slot) to be entangled to $(tag.new_remote_node).$(tag.new_remote_slot) and apply correction Z$(tag.correction)")
 Tag(tag::EntanglementUpdateZ) = Tag(EntanglementUpdateZ, tag.past_local_node, tag.past_local_slot, tag.past_remote_slot, tag.new_remote_node, tag.new_remote_slot, tag.correction)
 
 """
@@ -93,6 +93,8 @@ $FIELDS
     retry_lock_time::LT = 0.1
     """how many rounds of this protocol to run (`-1` for infinite))"""
     rounds::Int = -1
+    """whether the protocol should find the first available free slots in the nodes to be entangled or check for free slots randomly from the available slots"""
+    randomize::Bool = false
 end
 
 """Convenience constructor for specifying `rate` of generation instead of success probability and time"""
@@ -110,10 +112,10 @@ end
     prot = _prot # weird workaround for no support for `struct A a::Int end; @resumable function (fa::A) return fa.a end`; see https://github.com/JuliaDynamics/ResumableFunctions.jl/issues/77
     rounds = prot.rounds
     while rounds != 0
-        a = findfreeslot(prot.net[prot.nodeA])
-        b = findfreeslot(prot.net[prot.nodeB])
+        a = findfreeslot(prot.net[prot.nodeA], randomize=prot.randomize)
+        b = findfreeslot(prot.net[prot.nodeB], randomize=prot.randomize)
         if isnothing(a) || isnothing(b)
-            isnothing(prot.retry_lock_time) && error("we do not yet support waiting on register to make qubits available") # TODO
+            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)
             continue
         end
@@ -175,7 +177,7 @@ end
         reg = prot.net[prot.node]
         qubit_pair = findswapablequbits(prot.net,prot.node)
         if isnothing(qubit_pair)
-            isnothing(prot.retry_lock_time) && error("we do not yet support waiting on register to make qubits available") # TODO
+            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)
             continue
         end
@@ -198,12 +200,12 @@ end
         # tag with EntanglementUpdateX past_local_node, past_local_slot_idx past_remote_slot_idx new_remote_node, new_remote_slot, correction
         msg1 = Tag(EntanglementUpdateX, prot.node, q1.idx, tag1[3], tag2[2], tag2[3], xmeas)
         put!(channel(prot.net, prot.node=>tag1[2]; permit_forward=true), msg1)
-        @debug "swapper @$(prot.node) send msg to $(tag1[2]): $msg1"
+        @debug "SwapperProt @$(prot.node): Send message to $(tag1[2]) | message=`$msg1`"
         # send from here to new entanglement counterpart:
         # tag with EntanglementUpdateZ past_local_node, past_local_slot_idx past_remote_slot_idx new_remote_node, new_remote_slot, correction
         msg2 = Tag(EntanglementUpdateZ, prot.node, q2.idx, tag2[3], tag1[2], tag1[3], zmeas)
         put!(channel(prot.net, prot.node=>tag2[2]; permit_forward=true), msg2)
-        @debug "swapper @$(prot.node) send msg to $(tag2[2]): $msg2"
+        @debug "SwapperProt @$(prot.node): Send message to $(tag2[2]) | message=`$msg2`"
         unlock(q1)
         unlock(q2)
         rounds==-1 || (rounds -= 1)
@@ -249,20 +251,23 @@ end
             workwasdone = false
             for (updatetagsymbol, updategate) in ((EntanglementUpdateX, X), (EntanglementUpdateZ, Z))
                 # look for EntanglementUpdate? past_remote_slot_idx local_slot_idx, new_remote_node, new_remote_slot_idx correction
-                msg = querypop!(mb, updatetagsymbol, ❓, ❓, ❓, ❓, ❓, ❓)
+                msg = querydelete!(mb, updatetagsymbol, ❓, ❓, ❓, ❓, ❓, ❓)
                 isnothing(msg) && continue
-                @debug "tracker @$(prot.node) received from $(msg.src) msg: $(msg.tag)"
+                @debug "EntanglementTracker @$(prot.node): Received from $(msg.src).$(msg.tag[3]) | message=`$(msg.tag)`"
                 workwasdone = true
                 (src, (_, pastremotenode, pastremoteslotid, localslotid, newremotenode, newremoteslotid, correction)) = msg
                 localslot = nodereg[localslotid]
                 # Check if the local slot is still present and believed to be entangled.
                 # We will need to perform a correction operation due to the swap,
                 # but there will be no message forwarding necessary.
-                counterpart = querypop!(localslot, EntanglementCounterpart, pastremotenode, pastremoteslotid)
+                counterpart = querydelete!(localslot, EntanglementCounterpart, pastremotenode, pastremoteslotid)
                 if !isnothing(counterpart)
-                    @debug "tracker @$(prot.node) counterpart requesting lock at $(now(prot.sim))"
+                    time_before_lock = now(prot.sim)
+                    @debug "EntanglementTracker @$(prot.node): EntanglementCounterpart requesting lock at $(now(prot.sim))"
                     @yield lock(localslot)
-                    @debug "tracker @$(prot.node) counterpart getting lock at $(now(prot.sim))"
+                    @debug "EntanglementTracker @$(prot.node): EntanglementCounterpart getting lock at $(now(prot.sim))"
+                    time_after_lock = now(prot.sim)
+                    time_before_lock != time_after_lock && @debug "EntanglementTracker @$(prot.node): Needed Δt=$(time_after_lock-time_before_lock) to get a lock"
                     if !isassigned(localslot)
                         unlock(localslot)
                         error("There was an error in the entanglement tracking protocol `EntanglementTracker`. We were attempting to forward a classical message from a node that performed a swap to the remote entangled node. However, on reception of that message it was found that the remote node has lost track of its part of the entangled state although it still keeps a `Tag` as a record of it being present.")
@@ -275,24 +280,26 @@ end
                 end
                 # If not, check if we have a record of the entanglement being swapped to a different remote node,
                 # and forward the message to that node.
-                history = querypop!(localslot, EntanglementHistory,
+                history = querydelete!(localslot, EntanglementHistory,
                                     pastremotenode, pastremoteslotid, # who we were entangled to (node, slot)
                                     ❓, ❓,                             # who we swapped with (node, slot)
                                     ❓)                                # which local slot used to be entangled with whom we swapped with
                 if !isnothing(history)
-                    @debug "tracker @$(prot.node) history: $(history) | msg: $msg"
+                    # @debug "tracker @$(prot.node) history: $(history) | msg: $msg"
                     _, _, _, whoweswappedwith_node, whoweswappedwith_slotidx, swappedlocal_slotidx = history
-                    msghist = Tag(updatetagsymbol, pastremotenode, swappedlocal_slotidx, whoweswappedwith_slotidx, newremotenode, newremoteslotid, correction)
+                    tag!(localslot, EntanglementHistory, newremotenode, newremoteslotid, whoweswappedwith_node, whoweswappedwith_slotidx, swappedlocal_slotidx)
+                    @debug "EntanglementTracker @$(prot.node): history=`$(history)` | message=`$msg` | Sending to $(whoweswappedwith_node).$(whoweswappedwith_slotidx)"
+                    msghist = Tag(updatetagsymbol, pastremotenode, pastremoteslotid, whoweswappedwith_slotidx, newremotenode, newremoteslotid, correction)
                     put!(channel(prot.net, prot.node=>whoweswappedwith_node; permit_forward=true), msghist)
                     #println("           history sends to $whoweswappedwith_node: ", msghist)
                     continue
                 end
                 error("`EntanglementTracker` on node $(prot.node) received a message $(msg) that it does not know how to handle (due to the absence of corresponding `EntanglementCounterpart` or `EntanglementHistory` tags). This is a bug in the protocol and should not happen -- please report an issue at QuantumSavory's repository.")
             end
         end
-        #println("tracker @$(prot.node) starting message wait at ", now(prot.sim), " with mb: ", mb.buffer)
+        @debug "EntanglementTracker @$(prot.node): Starting message wait at $(now(prot.sim)) with MessageBuffer containing: $(mb.buffer)"
         @yield wait(mb)
-        #println("tracker @$(prot.node) message wait ends at ", now(prot.sim))
+        @debug "EntanglementTracker @$(prot.node): Message wait ends at $(now(prot.sim))"
     end
 end
 

diff --git a/src/QuantumSavory.jl b/src/QuantumSavory.jl
@@ -4,6 +4,7 @@ using Reexport
 
 using IterTools
 using LinearAlgebra
+using Random: randperm
 using Graphs
 import ConcurrentSim
 using ConcurrentSim: Environment, Simulation, Store, DelayQueue, Resource,
@@ -47,7 +48,7 @@ export
     # uptotime.jl
     uptotime!, overwritetime!,
     # tags.jl and queries.jl
-    Tag, tag!, untag!, W, ❓, query, queryall, querypop!, findfreeslot,
+    Tag, tag!, untag!, W, ❓, query, queryall, querydelete!, findfreeslot,
     # quantumchannel.jl
     QuantumChannel,
     # backgrounds.jl

diff --git a/src/messagebuffer.jl b/src/messagebuffer.jl
@@ -4,6 +4,7 @@ struct MessageBuffer{T}
     node::Int
     buffer::Vector{NamedTuple{(:src,:tag), Tuple{Int,T}}}
     waiters::IdDict{Resource,Resource}
+    no_wait::Ref{Int} # keeps track of the situation when something is pushed in the buffer and no waiters are present. In that case, when the waiters are available after it they would get locked while the code that was supposed to unlock them has already run. So, we keep track the number of times this happens and put no lock on the waiters in this situation.
 end
 
 struct ChannelForwarder
@@ -15,6 +16,7 @@ end
 function Base.put!(cf::ChannelForwarder, tag::Tag)
     # shortest path calculated by Graphs.a_star
     nexthop = first(a_star(cf.net.graph, cf.src, cf.dst))
+    @debug "ChannelForwarder: Forwarding message from node $(nexthop.src) to node $(nexthop.dst) | message=$(tag)| end destination=$(cf.dst)"
     put!(channel(cf.net, cf.src=>nexthop.dst; permit_forward=false), tag_types.Forward(tag, cf.dst))
 end
 
@@ -23,10 +25,16 @@ end
         tag = @yield take!(ch)
         @cases tag begin
             Forward(innertag, enddestination) => begin # inefficient -- it recalculates the a_star at each hop TODO provide some caching mechanism
+                @debug "MessageBuffer @$(mb.node) at t=$(now(mb.sim)): Forwarding message to node $(enddestination) | message=`$(tag)`"
                 put!(channel(mb.net, mb.node=>enddestination; permit_forward=true), innertag)
             end
             _ => begin
                 #println("from $src storing in mb: $tag at $(now(sim))")
+                @debug "MessageBuffer @$(mb.node) at t=$(now(mb.sim)): Receiving from source $(src) | message=`$(tag)`"
+                length(mb.waiters) == 0 && @debug "MessageBuffer @$(mb.node) received a message, but there is no one waiting on that message buffer. The message was `$(tag)`."
+                if length(mb.waiters) == 0
+                    mb.no_wait[] += 1
+                end
                 push!(mb.buffer, (;src,tag));
                 for waiter in keys(mb.waiters)
                     unlock(waiter)
@@ -37,16 +45,21 @@ end
 end
 
 function MessageBuffer(net, node::Int, qs::Vector{NamedTuple{(:src,:channel), Tuple{Int, DelayQueue{T}}}}) where {T}
-    sim = qs[1].channel.store.env
+    sim = get_time_tracker(net)
     signal = IdDict{Resource,Resource}()
-    mb = MessageBuffer{T}(sim, net, node, Tuple{Int,T}[], signal)
+    no_wait = Ref{Int}(0)
+    mb = MessageBuffer{T}(sim, net, node, Tuple{Int,T}[], signal, no_wait)
     for (;src, channel) in qs
         @process take_loop_mb(sim, channel, src, mb)
     end
     mb
 end
 
-@resumable function wait_process(sim, mb::MessageBuffer)
+@resumable function wait_process(sim, mb)
+    if mb.no_wait[] != 0 # This happens only in the specific case when something is put in the buffer before there any waiters.
+        mb.no_wait[] -= 1
+        return
+    end
     waitresource = Resource(sim)
     lock(waitresource)
     mb.waiters[waitresource] = waitresource

diff --git a/src/queries.jl b/src/queries.jl
@@ -212,12 +212,12 @@ t=1.0: query returns nothing
 t=3.0: query returns SymbolIntInt(:second_tag, 123, 456)::Tag received from node 3
 ```
 """
-function querypop!(mb::MessageBuffer, args...)
+function querydelete!(mb::MessageBuffer, args...)
     r = query(mb, args...)
     return isnothing(r) ? nothing : popat!(mb.buffer, r.depth)
 end
 
-function querypop!(ref::RegRef, args...) # TODO there is a lot of code duplication here
+function querydelete!(ref::RegRef, args...) # TODO there is a lot of code duplication here
     r = query(ref, args...)
     return isnothing(r) ? nothing : popat!(ref.reg.tags[ref.idx], r.depth)
 end
@@ -317,7 +317,13 @@ julia> findfreeslot(reg) |> isnothing
 true
 ```
 """
-function findfreeslot(reg::Register)
+function findfreeslot(reg::Register; randomize=false)
+    if randomize
+        for i in randperm(length(reg.staterefs))
+            slot = reg[i]
+            islocked(slot) || isassigned(slot) || return slot
+        end
+    end
     for slot in reg
         islocked(slot) || isassigned(slot) || return slot
     end

diff --git a/test/runtests.jl b/test/runtests.jl
@@ -31,6 +31,7 @@ println("Starting tests with $(Threads.nthreads()) threads out of `Sys.CPU_THREA
 @doset "noninstant_and_backgrounds_qumode"
 @doset "messagebuffer"
 @doset "tags_and_queries"
+@doset "entanglement_tracker"
 
 @doset "circuitzoo_api"
 @doset "circuitzoo_purification"