Code review

Author: BenLauwens

src/SimJulia.jl

@@ -12,24 +12,24 @@ module SimJulia
 
   export AbstractEvent
   export value, state, environment
-  export Simulation
-  export run, now, active_process
   export Event, Timeout
-  export succeed, fail, append_callback, @callback, remove_callback
+  export succeed, fail, @callback, remove_callback
   export Operator
   export (&), (|)
+  export Simulation
+  export run, now, active_process
   export Process, @process
   export yield, interrupt
-  export FiniteStateMachine, @stateful, @yield, iscoroutinedone
+  export FiniteStateMachine, @stateful, @yield
   export Coroutine, @coroutine
   export Container, Resource, Store
   export Put, Get, Request, Release, cancel, capacity, request, @request
 
   include("base.jl")
-  include("simulations.jl")
   include("events.jl")
   include("operators.jl")
   include("tasks/base.jl")
+  include("simulations.jl")
   include("processes.jl")
   include("finitestatemachines/utils.jl")
   include("finitestatemachines/transforms.jl")

src/base.jl

@@ -56,3 +56,37 @@ end
 function remove_callback(cb::Function, ev::AbstractEvent)
   DataStructures.dequeue!(ev.bev.callbacks, cb)
 end
+
+function schedule(ev::AbstractEvent, delay::Number=zero(Float64); priority::Int8=zero(Int8), value::Any=nothing)
+  env = environment(ev)
+  bev = ev.bev
+  bev.value = value
+  env.heap[bev] = EventKey(env.time + delay, priority, env.sid+=one(UInt))
+  bev.state = scheduled
+end
+
+struct StopSimulation <: Exception
+  value :: Any
+  function StopSimulation(value::Any=nothing)
+    new(value)
+  end
+end
+
+function stop_simulation(ev::AbstractEvent)
+  throw(StopSimulation(value(ev)))
+end
+
+function run(env::Environment, until::AbstractEvent)
+  @callback stop_simulation(until)
+  try
+    while true
+      step(env)
+    end
+  catch exc
+    if isa(exc, StopSimulation)
+      return exc.value
+    else
+      rethrow(exc)
+    end
+  end
+end

src/coroutines.jl

@@ -27,10 +27,10 @@ function execute(ev::AbstractEvent, proc::Coroutine)
     target = proc.fsm(value(ev))
     reset_active_process(env)
     if iscoroutinedone(proc.fsm)
-      schedule(proc.bev, value=target)
+      schedule(proc; value=target)
     else
-      proc.target = state(target) == triggered ? Timeout(env, value=value(target)) : target
-      proc.resume = append_callback(execute, proc.target, proc)
+      proc.target = state(target) == triggered ? Timeout(env; value=value(target)) : target
+      proc.resume = @callback execute(proc.target, proc)
     end
   catch exc
     rethrow(exc)
@@ -40,7 +40,7 @@ end
 function interrupt(proc::Coroutine, cause::Any=nothing)
   if !iscoroutinedone(proc.fsm)
     remove_callback(proc.resume, proc.target)
-    proc.target = Timeout(environment(proc), priority=typemax(Int8), value=InterruptException(proc, cause))
-    proc.resume = append_callback(execute, proc.target, proc)
+    proc.target = Timeout(environment(proc); priority=typemax(Int8), value=InterruptException(proc, cause))
+    proc.resume = @callback execute(proc.target, proc)
   end
 end

src/events.jl

@@ -8,23 +8,23 @@ end
 function succeed(ev::Event; priority::Int8=zero(Int8), value::Any=nothing) :: Event
   sta = state(ev)
   (sta == scheduled || sta == triggered) && throw(EventNotIdle(ev))
-  schedule(ev.bev, priority=priority, value=value)
+  schedule(ev; priority=priority, value=value)
   ev
 end
 
 function fail(ev::Event, exc::Exception; priority::Int8=zero(Int8)) :: Event
-  succeed(ev, priority=priority, value=exc)
+  succeed(ev; priority=priority, value=exc)
 end
 
 struct Timeout <: AbstractEvent
   bev :: BaseEvent
   function Timeout(env::Environment, delay::Number=0; priority::Int8=zero(Int8), value::Any=nothing)
     ev = new(BaseEvent(env))
-    schedule(ev.bev, delay, priority=priority, value=value)
+    schedule(ev, delay; priority=priority, value=value)
     ev
   end
 end
 
-function run(sim::Simulation, until::Number=typemax(Float64))
-  run(sim, Timeout(sim, until-now(sim)))
+function run(env::Environment, until::Number=typemax(Float64))
+  run(env, Timeout(env, until-now(env)))
 end

src/operators.jl

@@ -15,7 +15,7 @@ struct Operator <: AbstractEvent
     event_state_values = Dict{AbstractEvent, StateValue}()
     for ev in tuple(fev, events...)
       event_state_values[ev] = StateValue(state(ev))
-      append_callback(check, ev, op, event_state_values)
+      @callback check(ev, op, event_state_values)
     end
     op
   end
@@ -25,34 +25,34 @@ function check(ev::AbstractEvent, op::Operator, event_state_values::Dict{Abstrac
   val = value(ev)
   if state(op) == idle
     if isa(val, Exception)
-      schedule(op.bev, value=val)
+      schedule(op; value=val)
     else
       event_state_values[ev] = StateValue(state(ev), val)
       if op.eval(collect(values(event_state_values)))
-        schedule(op.bev, value=event_state_values)
+        schedule(op; value=event_state_values)
       end
     end
   elseif state(op) == scheduled
     if isa(val, Exception)
-      schedule(op.bev, priority=typemax(Int8), value=val)
+      schedule(op; priority=typemax(Int8), value=val)
     else
       event_state_values[ev] = StateValue(state(ev), val)
     end
   end
 end
 
 function eval_and(state_values::Vector{StateValue})
-  return all(map((sv)->sv.state == triggered, state_values))
+  all(map((sv)->sv.state == triggered, state_values))
 end
 
 function eval_or(state_values::Vector{StateValue})
-  return any(map((sv)->sv.state == triggered, state_values))
+  any(map((sv)->sv.state == triggered, state_values))
 end
 
 function (&)(ev1::AbstractEvent, ev2::AbstractEvent)
-  return Operator(eval_and, ev1, ev2)
+  Operator(eval_and, ev1, ev2)
 end
 
 function (|)(ev1::AbstractEvent, ev2::AbstractEvent)
-  return Operator(eval_or, ev1, ev2)
+  Operator(eval_or, ev1, ev2)
 end

src/processes.jl

@@ -23,8 +23,8 @@ end
 function yield(target::AbstractEvent)
   env = environment(target)
   proc = active_process(env)
-  proc.target = state(target) == triggered ? Timeout(env, value=value(target)) : target
-  proc.resume = append_callback(execute, proc.target, proc)
+  proc.target = state(target) == triggered ? Timeout(env; value=value(target)) : target
+  proc.resume = @callback execute(proc.target, proc)
   ret = SimJulia.produce(nothing)
   isa(ret, Exception) && throw(ret)
   return ret
@@ -36,7 +36,7 @@ function execute(ev::AbstractEvent, proc::Process)
     set_active_process(env, proc)
     ret = SimJulia.consume(proc.task, value(ev))
     reset_active_process(env)
-    istaskdone(proc.task) && schedule(proc.bev, value=ret)
+    istaskdone(proc.task) && schedule(proc; value=ret)
   catch exc
     rethrow(exc)
   end
@@ -45,7 +45,7 @@ end
 function interrupt(proc::Process, cause::Any=nothing)
   if !istaskdone(proc.task)
     remove_callback(proc.resume, proc.target)
-    proc.target = Timeout(environment(proc), priority=typemax(Int8), value=InterruptException(proc, cause))
-    proc.resume = append_callback(execute, proc.target, proc)
+    proc.target = Timeout(environment(proc); priority=typemax(Int8), value=InterruptException(proc, cause))
+    proc.resume = @callback execute(proc.target, proc)
   end
 end

src/resources/containers.jl

@@ -29,14 +29,14 @@ end
 function Put{N<:Number}(con::Container{N}, amount::N; priority::Int=0) :: Put
   put_ev = Put(con.env)
   con.Put_queue[put_ev] = ContainerKey(priority, con.seid+=one(UInt), amount)
-  append_callback(trigger_get, put_ev, con)
+  @callback trigger_get(put_ev, con)
   trigger_put(put_ev, con)
-  return put_ev
+  put_ev
 end
 
 const Request = Put
 
-Request(res::Resource; priority::Int=0) = Put(res, 1, priority=priority)
+Request(res::Resource; priority::Int=0) = Put(res, 1; priority=priority)
 
 macro request(res, req, expr)
   esc(quote
@@ -51,12 +51,12 @@ macro request(res, req, expr)
 end
 
 function request(func::Function, res::Resource; priority::Int=0)
-  req = Request(res, priority=priority)
+  req = Request(res; priority=priority)
   try
     func(req)
   finally
     if state(req) == triggered
-      yield(Release(res, priority=priority))
+      yield(Release(res; priority=priority))
     else
       cancel(res, req)
     end
@@ -66,29 +66,25 @@ end
 function Get{N<:Number}(con::Container{N}, amount::N; priority::Int=0) :: Get
   get_ev = Get(con.env)
   con.Get_queue[get_ev] = ContainerKey(priority, con.seid+=one(UInt), amount)
-  append_callback(trigger_put, get_ev, con)
+  @callback trigger_put(get_ev, con)
   trigger_get(get_ev, con)
-  return get_ev
+  get_ev
 end
 
 const Release = Get
 
-Release(res::Resource; priority::Int=0) = Get(res, 1, priority=priority)
+Release(res::Resource; priority::Int=0) = Get(res, 1; priority=priority)
 
 function do_put{N<:Number}(con::Container{N}, put_ev::Put, key::ContainerKey{N}) :: Bool
-  if con.level + key.amount <= con.capacity
-    schedule(put_ev.bev)
-    con.level += key.amount
-    return true
-  end
-  return false
+  con.level + key.amount > con.capacity && return false
+  schedule(put_ev)
+  con.level += key.amount
+  true
 end
 
 function do_get{N<:Number}(con::Container{N}, get_ev::Get, key::ContainerKey{N}) :: Bool
-  if con.level - key.amount >= zero(N)
-    schedule(get_ev.bev)
-    con.level -= key.amount
-    return true
-  end
-  return false
+  con.level - key.amount < zero(N) && return false
+  schedule(get_ev)
+  con.level -= key.amount
+  true
 end

src/resources/stores.jl

@@ -29,7 +29,7 @@ end
 function Put{T}(sto::Store{T}, item::T; priority::Int=0) :: Put
   put_ev = Put(sto.env)
   sto.Put_queue[put_ev] = StorePutKey(priority, sto.seid+=one(UInt), item)
-  append_callback(trigger_get, put_ev, sto)
+  @callback trigger_get(put_ev, sto)
   trigger_put(put_ev, sto)
   return put_ev
 end
@@ -41,15 +41,15 @@ end
 function Get{T}(sto::Store{T}, filter::Function=get_any_item; priority::Int=0) :: Get
   get_ev = Get(sto.env)
   sto.Get_queue[get_ev] = StoreGetKey(priority, sto.seid+=one(UInt), filter)
-  append_callback(trigger_put, get_ev, sto)
+  @callback trigger_put(get_ev, sto)
   trigger_get(get_ev, sto)
   return get_ev
 end
 
 function do_put{T}(sto::Store{T}, put_ev::Put, key::StorePutKey{T}) :: Bool
   if length(sto.items) < sto.capacity
     push!(sto.items, key.item)
-    schedule(put_ev.bev)
+    schedule(put_ev)
   end
   return false
 end
@@ -58,7 +58,7 @@ function do_get{T}(sto::Store{T}, get_ev::Get, key::StoreGetKey) :: Bool
   for item in sto.items
     if key.filter(item)
       delete!(sto.items, item)
-      schedule(get_ev.bev, value=item)
+      schedule(get_ev; value=item)
       break
     end
   end

src/simulations.jl

@@ -1,5 +1,12 @@
 abstract type AbstractProcess <: AbstractEvent end
 
+struct InterruptException <: Exception
+  by :: AbstractProcess
+  cause :: Any
+end
+
+struct EmptySchedule <: Exception end
+
 struct EventKey
   time :: Float64
   priority :: Int8
@@ -21,35 +28,6 @@ mutable struct Simulation <: Environment
   end
 end
 
-function now(sim::Simulation)
-  sim.time
-end
-
-function active_process(sim::Simulation) :: AbstractProcess
-  get(sim.active_proc)
-end
-
-function reset_active_process(sim::Simulation)
-  sim.active_proc = Nullable{AbstractProcess}()
-end
-
-function set_active_process(sim::Simulation, proc::AbstractProcess)
-  sim.active_proc = Nullable(proc)
-end
-
-struct StopSimulation <: Exception
-  value :: Any
-  function StopSimulation(value::Any=nothing)
-    new(value)
-  end
-end
-
-function stop_simulation(ev::AbstractEvent)
-  throw(StopSimulation(value(ev)))
-end
-
-struct EmptySchedule <: Exception end
-
 function step(sim::Simulation)
   isempty(sim.heap) && throw(EmptySchedule())
   (bev, key) = DataStructures.peek(sim.heap)
@@ -61,28 +39,18 @@ function step(sim::Simulation)
   end
 end
 
-function run(sim::Simulation, until::AbstractEvent)
-  append_callback(stop_simulation, until)
-  try
-    while true
-      step(sim)
-    end
-  catch exc
-    if isa(exc, StopSimulation)
-      return exc.value
-    else
-      rethrow(exc)
-    end
-  end
+function now(sim::Simulation)
+  sim.time
 end
 
-function schedule(bev::BaseEvent, delay::Number=zero(Float64); priority::Int8=zero(Int8), value::Any=nothing)
-  bev.value = value
-  bev.env.heap[bev] = EventKey(bev.env.time + delay, priority, bev.env.sid+=one(UInt))
-  bev.state = scheduled
+function active_process(sim::Simulation) :: AbstractProcess
+  get(sim.active_proc)
 end
 
-struct InterruptException <: Exception
-  by :: AbstractProcess
-  cause :: Any
+function reset_active_process(sim::Simulation)
+  sim.active_proc = Nullable{AbstractProcess}()
+end
+
+function set_active_process(sim::Simulation, proc::AbstractProcess)
+  sim.active_proc = Nullable(proc)
 end