implement StackStore and QueueStore (#98)

Author: Krastanov

CHANGELOG.md

@@ -1,5 +1,9 @@
 # News
 
+## v1.3.0 - 2023-08-07
+
+- Implement ordered versions of `Store`, namely `QueueStore` and `StackStore`.
+
 ## v1.2.0 - 2023-08-06
 
 - Priorities can now be non-integer.

Project.toml

@@ -5,7 +5,7 @@ license = "MIT"
 desc = "A discrete event process oriented simulation framework."
 authors = ["Ben Lauwens and SimJulia and ConcurrentSim contributors"]
 repo = "https://github.com/JuliaDynamics/ConcurrentSim.jl.git"
-version = "1.2.0"
+version = "1.3.0"
 
 [deps]
 DataStructures = "864edb3b-99cc-5e75-8d2d-829cb0a9cfe8"

docs/src/guides/blockingandyielding.md

@@ -21,7 +21,7 @@ In particular, the `ConcurrentSim.jl` package uses the async "coroutines" model
 Without further ado, here is the typical API used with:
 
 - `ConcurrentSim.Resource` which is used to represent scarce resource that can be used by only up to a fixed number of tasks. If the limit is just one task (the default), this is very similar to `Base.ReentrantLock`. `Resource` is a special case of `Container` with an integer "resource counter".
-- `ConcurrentSim.Store` which is used to represent a FILO stack.
+- `ConcurrentSim.Store` which is used to represent an unordered heap. For the ordered versions, consider `QueueStore` or `StackStore`.
 
 ```@raw html
 <div style="width:120%;min-width:120%;">

src/ConcurrentSim.jl

@@ -17,7 +17,7 @@ module ConcurrentSim
   export @resumable, @yield
   export AbstractProcess, Simulation, run, now, active_process, StopSimulation
   export Process, @process, interrupt
-  export Container, Resource, Store, put!, get, cancel, request, tryrequest, release
+  export Container, Resource, Store, StackStore, QueueStore, put!, get, cancel, request, tryrequest, release
   export nowDatetime
 
   include("base.jl")
@@ -28,6 +28,7 @@ module ConcurrentSim
   include("resources/base.jl")
   include("resources/containers.jl")
   include("resources/stores.jl")
+  include("resources/ordered_stores.jl")
   include("utils/time.jl")
   include("deprecated_aliased.jl")
 end

src/resources/ordered_stores.jl

@@ -0,0 +1,103 @@
+"""
+    StackStore{N, T<:Number}
+
+A store in which items are stored in a FILO order.
+
+```jldoctest
+julia> sim = Simulation()
+       store = Store{Symbol}(sim)
+       stack = StackStore{Symbol}(sim)
+       items = [:a,:b,:a,:c];
+
+julia> [put!(store, item) for item in items];
+
+julia> [value(take!(store)) for _ in 1:length(items)]
+4-element Vector{Symbol}:
+ :a
+ :a
+ :b
+ :c
+
+julia> [put!(stack, item) for item in items];
+
+julia> [value(take!(stack)) for _ in 1:length(items)]
+4-element Vector{Symbol}:
+ :c
+ :a
+ :b
+ :a
+```
+
+See also: [`QueueStore`](@ref), [`Store`](@ref)
+"""
+const StackStore = Store{N, T, DataStructures.Stack{N}} where {N, T<:Number}
+StackStore{N}(env::Environment; capacity=typemax(UInt)) where {N} = StackStore{N, Int}(env; capacity)
+
+"""
+    QueueStore{N, T<:Number}
+
+A store in which items are stored in a FIFO order.
+
+```jldoctest
+julia> sim = Simulation()
+       store = Store{Symbol}(sim)
+       queue = QueueStore{Symbol}(sim)
+       items = [:a,:b,:a,:c];
+
+julia> [put!(store, item) for item in items];
+
+julia> [value(take!(store)) for _ in 1:length(items)]
+4-element Vector{Symbol}:
+ :a
+ :a
+ :b
+ :c
+
+julia> [put!(queue, item) for item in items];
+
+julia> [value(take!(queue)) for _ in 1:length(items)]
+4-element Vector{Symbol}:
+ :a
+ :b
+ :a
+ :c
+```
+
+See also: [`StackStore`](@ref), [`Store`](@ref)
+"""
+const QueueStore = Store{N, T, DataStructures.Queue{N}} where {N, T<:Number}
+QueueStore{N}(env::Environment; capacity=typemax(UInt)) where {N} = QueueStore{N, Int}(env; capacity)
+
+function do_put(sto::StackStore{N, T}, put_ev::Put, key::StorePutKey{N, T}) where {N, T<:Number}
+  if sto.load < sto.capacity
+    sto.load += one(UInt)
+    push!(sto.items, key.item)
+    schedule(put_ev)
+  end
+  false
+end
+
+function do_get(sto::StackStore{N, T}, get_ev::Get, key::StoreGetKey{T}) where {N, T<:Number}
+  key.filter !== get_any_item && error("Filtering not supported for `StackStore`. Use an unordered store instead, or submit a feature request for implementing filtering to our issue tracker.")
+  item = pop!(sto.items)
+  sto.load -= one(UInt)
+  schedule(get_ev; value=item)
+  true
+end
+
+function do_put(sto::QueueStore{N, T}, put_ev::Put, key::StorePutKey{N, T}) where {N, T<:Number}
+  if sto.load < sto.capacity
+    sto.load += one(UInt)
+    enqueue!(sto.items, key.item)
+    schedule(put_ev)
+  end
+  false
+end
+
+function do_get(sto::QueueStore{N, T}, get_ev::Get, key::StoreGetKey{T}) where {N, T<:Number}
+  key.filter !== get_any_item && error("Filtering not supported for `QueueStore`. Use an unordered store instead, or submit a feature request for implementing filtering to our issue tracker.")
+  item = dequeue!(sto.items)
+  sto.load -= one(UInt)
+  schedule(get_ev; value=item)
+  true
+end

src/resources/stores.jl

@@ -13,13 +13,15 @@ end
 """
     Store{N, T<:Number}(env::Environment; capacity::UInt=typemax(UInt))
 
-A store is a resource that can hold a number of items of type `N` in a FILO stack. It is similar to a `Base.Channel` with a finite capacity ([`put!`](@ref) blocks after reaching capacity).
+A store is a resource that can hold a number of items of type `N`. It is similar to a `Base.Channel` with a finite capacity ([`put!`](@ref) blocks after reaching capacity).
 The [`put!`](@ref) and [`take!`](@ref) functions are a convenient way to interact with such a "channel" in a way mostly compatible with other discrete event and concurrency frameworks.
 
 See [`Container`](@ref) for a more lock-like resource.
 
 Think of `Resource` and `Container` as locks and of `Store` as channels/stacks. They block only if empty (on taking) or full (on storing).
 
+`Store` does not guarantee any order of items. See [`StackStore`](@ref) and [`QueueStore`](@ref) for ordered variants.
+
 ```jldoctest
 julia> sim = Simulation(); store = Store{Int}(sim);
 
@@ -32,19 +34,23 @@ julia> value(take!(store))
 1
 ```
 """
-mutable struct Store{N, T<:Number} <: AbstractResource
+mutable struct Store{N, T<:Number, D} <: AbstractResource
   env :: Environment
   capacity :: UInt
   load :: UInt
-  items :: Dict{N, UInt}
+  items :: D
   seid :: UInt
   put_queue :: DataStructures.PriorityQueue{Put, StorePutKey{N, T}}
   get_queue :: DataStructures.PriorityQueue{Get, StoreGetKey{T}}
-  function Store{N, T}(env::Environment; capacity=typemax(UInt)) where {N, T<:Number}
-    new(env, UInt(capacity), zero(UInt), Dict{N, UInt}(), zero(UInt), DataStructures.PriorityQueue{Put, StorePutKey{N, T}}(), DataStructures.PriorityQueue{Get, StoreGetKey{T}}())
+  function Store{N, T, D}(env::Environment; capacity=typemax(UInt)) where {N, T<:Number, D}
+    new(env, UInt(capacity), zero(UInt), D(), zero(UInt), DataStructures.PriorityQueue{Put, StorePutKey{N, T}}(), DataStructures.PriorityQueue{Get, StoreGetKey{T}}())
   end
 end
 
+function Store{N, T}(env::Environment; capacity=typemax(UInt)) where {N, T<:Number}
+    Store{N, T, Dict{N, UInt}}(env; capacity=UInt(capacity))
+end
+
 function Store{N}(env::Environment; capacity=typemax(UInt)) where {N}
   Store{N, Int}(env; capacity=UInt(capacity))
 end
@@ -64,15 +70,15 @@ end
 
 get_any_item(::N) where N = true
 
-function get(sto::Store{N, T}, filter::Function=get_any_item; priority=zero(T)) where {N, T<:Number}
+function get(sto::Store{N, T, D}, filter::Function=get_any_item; priority=zero(T)) where {N, T<:Number, D}
   get_ev = Get(sto.env)
   sto.get_queue[get_ev] = StoreGetKey(sto.seid+=one(UInt), filter, T(priority))
   @callback trigger_put(get_ev, sto)
   trigger_get(get_ev, sto)
   get_ev
 end
 
-function do_put(sto::Store{N, T}, put_ev::Put, key::StorePutKey{N, T}) where {N, T<:Number}
+function do_put(sto::Store{N, T, Dict{N,UInt}}, put_ev::Put, key::StorePutKey{N, T}) where {N, T<:Number}
   if sto.load < sto.capacity
     sto.load += one(UInt)
     sto.items[key.item] = get(sto.items, key.item, zero(UInt)) + one(UInt)
@@ -81,7 +87,7 @@ function do_put(sto::Store{N, T}, put_ev::Put, key::StorePutKey{N, T}) where {N,
   false
 end
 
-function do_get(sto::Store{N, T}, get_ev::Get, key::StoreGetKey{T}) where {N, T<:Number}
+function do_get(sto::Store{N, T, Dict{N,UInt}}, get_ev::Get, key::StoreGetKey{T}) where {N, T<:Number}
   for (item, number) in sto.items
     if key.filter(item)
       sto.load -= one(UInt)