Both Coroutines as Processes are supported

Author: BenLauwens

src/SimJulia.jl

@@ -18,6 +18,7 @@ module SimJulia
   export Operator
   export (&), (|)
   export FiniteStateMachine, @stateful, @yield
+  export Coroutine, @Coroutine
   export Process, @Process
   export interrupt
   export Simulation, StopSimulation
@@ -28,12 +29,14 @@ module SimJulia
   include("base.jl")
   include("events.jl")
   include("operators.jl")
-  include("coroutine/utils.jl")
-  include("coroutine/transforms.jl")
-  include("coroutine/macro.jl")
-  include("process.jl")
   include("time.jl")
   include("simulation.jl")
+  include("coroutines/utils.jl")
+  include("coroutines/transforms.jl")
+  include("coroutines/macro.jl")
+  include("coroutines.jl")
+  include("processes/base.jl")
+  include("processes.jl")
   include("resources/base.jl")
   include("resources/containers.jl")
   include("resources/stores.jl")

src/coroutines.jl

@@ -0,0 +1,84 @@
+"""
+A `Coroutine` is an abstraction for an event yielding function, i.e. a process function.
+
+The process function can suspend its execution by yielding an instance of `AbstractEvent`. The `Environment` will take care of resuming the process function with the value of that event once it has happened. The exception of failed events is also thrown into the process function.
+
+A `Coroutine` is a subtype of `AbstractEvent`. It is triggered, once the process functions returns or raises an exception. The value of the process is the return value of the process function or the exception, respectively.
+
+**Signature**:
+
+Coroutine{E<:Environment} <: AbstractEvent{E}
+
+**Fields**:
+
+- bev :: BaseEvent{E}
+- task :: Task
+- target :: AbstractEvent{E}
+- resume :: Function
+
+**Constructor**:
+
+Coroutine{E<:Environment}(func::Function, env::E, args::Any...)
+"""
+type Coroutine{E<:Environment} <: AbstractProcess{E}
+  bev :: BaseEvent{E}
+  fsm :: FiniteStateMachine
+  target :: AbstractEvent{E}
+  resume :: Function
+  function Coroutine{E}(func::Function, env::E, args::Any...) where E<:Environment
+    proc = new()
+    proc.bev = BaseEvent(env)
+    proc.fsm = func(env, args...)
+    proc.target = Timeout(env)
+    proc.resume = append_callback(execute, proc.target, proc)
+    return proc
+  end
+end
+
+function Coroutine{E<:Environment}(func::Function, env::E, args::Any...)
+  Coroutine{E}(func, env, args...)
+end
+
+"""
+Creates a `Coroutine` with process function `func` having a required argument `env`, i.e. an instance of a subtype of `Environment`, and a variable number of arguments `args...`.
+
+**Signature**:
+
+@Coroutine func(env, args...)
+"""
+macro Coroutine(ex)
+  if ex.head == :call
+    func = esc(ex.args[1])
+    args = [esc(ex.args[n]) for n in 2:length(ex.args)]
+    return :(Coroutine($(func), $(args...)))
+  end
+end
+
+function execute{E<:Environment}(ev::AbstractEvent{E}, proc::Coroutine{E})
+  try
+    env = environment(ev)
+    set_active_process(env, proc)
+    target = proc.fsm(value(ev))
+    if iscoroutinedone(proc.fsm)
+      schedule(proc.bev, value=target)
+    else
+      if state(target) == triggered
+        proc.target = Timeout(env, value=value(target))
+      else
+        proc.target = target
+      end
+      proc.resume = append_callback(execute, proc.target, proc)
+    end
+    set_active_process(env)
+  catch exc
+    rethrow(exc)
+  end
+end
+
+function interrupt{E<:Environment}(proc::Coroutine{E}, cause::Any=nothing)
+  if !iscoroutinedone(proc.fsm)
+    remove_callback(proc.resume, proc.target)
+    proc.target = Timeout(environment(proc), priority=true, value=InterruptException(proc, cause))
+    proc.resume = append_callback(execute, proc.target, proc)
+  end
+end

src/coroutine/macro.jl src/coroutines/macro.jlsrc/coroutine/macro.jl renamed to src/coroutines/macro.jl

@@ -20,15 +20,15 @@ Process{E<:Environment} <: AbstractEvent{E}
 
 Process{E<:Environment}(func::Function, env::E, args::Any...)
 """
-type Process{E<:Environment} <: AbstractEvent{E}
+type Process{E<:Environment} <: AbstractProcess{E}
   bev :: BaseEvent{E}
-  fsm :: FiniteStateMachine
+  task :: Task
   target :: AbstractEvent{E}
   resume :: Function
   function Process{E}(func::Function, env::E, args::Any...) where E<:Environment
     proc = new()
     proc.bev = BaseEvent(env)
-    proc.fsm = func(env, args...)
+    proc.task = @task func(env, args...)
     proc.target = Timeout(env)
     proc.resume = append_callback(execute, proc.target, proc)
     return proc
@@ -54,34 +54,38 @@ macro Process(ex)
   end
 end
 
+function yield{E<:Environment}(target::AbstractEvent{E})
+  env = environment(target)
+  proc = active_process(env)
+  if state(target) == triggered
+    proc.target = Timeout(env, value=value(target))
+  else
+    proc.target = target
+  end
+  proc.resume = append_callback(execute, proc.target, proc)
+  ret = SimJulia.produce(nothing)
+  if isa(ret, Exception)
+    throw(ret)
+  end
+  return ret
+end
+
 function execute{E<:Environment}(ev::AbstractEvent{E}, proc::Process{E})
   try
     env = environment(ev)
     set_active_process(env, proc)
-    target = proc.fsm(value(ev))
-    if iscoroutinedone(proc.fsm)
-      schedule(proc.bev, value=target)
-    else
-      if state(target) == triggered
-        proc.target = Timeout(env, value=value(target))
-      else
-        proc.target = target
-      end
-      proc.resume = append_callback(execute, proc.target, proc)
-    end
+    ret = SimJulia.consume(proc.task, value(ev))
     set_active_process(env)
+    if istaskdone(proc.task)
+      schedule(proc.bev, value=ret)
+    end
   catch exc
     rethrow(exc)
   end
 end
 
-struct InterruptException{E<:Environment} <: Exception
-  by :: Process{E}
-  cause :: Any
-end
-
 function interrupt{E<:Environment}(proc::Process{E}, cause::Any=nothing)
-  if !iscoroutinedone(proc.fsm)
+  if !istaskdone(proc.fsm)
     remove_callback(proc.resume, proc.target)
     proc.target = Timeout(environment(proc), priority=true, value=InterruptException(proc, cause))
     proc.resume = append_callback(execute, proc.target, proc)

src/coroutine/transforms.jl src/coroutines/transforms.jlsrc/coroutine/transforms.jl renamed to src/coroutines/transforms.jl

@@ -0,0 +1,23 @@
+function produce(consumer::Task, values...)
+  ct = current_task()
+  ct.result = length(values)==1 ? values[1] : values
+  Base.schedule_and_wait(consumer)
+  return consumer.result
+end
+
+function produce(v)
+  ct = current_task()
+  consumer = ct.consumers
+  ct.consumers = nothing
+  Base.schedule_and_wait(consumer, v)
+  return consumer.result
+end
+produce(v...) = produce(v)
+
+function consume(producer::Task, values...)
+  istaskdone(producer) && return wait(producer)
+  ct = current_task()
+  ct.result = length(values)==1 ? values[1] : values
+  producer.consumers = ct
+  producer.state == :runnable ? Base.schedule_and_wait(producer) : wait()
+end

src/coroutine/utils.jl src/coroutines/utils.jlsrc/coroutine/utils.jl renamed to src/coroutines/utils.jl

@@ -52,6 +52,19 @@ macro Request(res, req, expr)
   end)
 end
 
+function Request{E<:Environment}(func::Function, res::Resource{E}; priority::Int=0)
+  req = Request(res, priority=priority)
+  try
+    func(req)
+  finally
+    if state(req) == triggered
+      yield(Release(res, priority=priority))
+    else
+      cancel(res, req)
+    end
+  end
+end
+
 function Get{N<:Number, E<:Environment}(con::Container{N, E}, amount::N; priority::Int=0) :: Get{E}
   get_ev = Get{E}(con.env)
   con.Get_queue[get_ev] = ContainerKey(priority, con.seid+=one(UInt), amount)

src/process.jl src/processes.jlsrc/process.jl renamed to src/processes.jl

@@ -8,6 +8,8 @@ function isless{T<:TimeType}(a::EventKey{T}, b::EventKey{T}) :: Bool
   (a.time < b.time) || (a.time == b.time && a.priority > b.priority) || (a.time == b.time && a.priority == b.priority && a.id < b.id)
 end
 
+abstract type AbstractProcess{E<:Environment} <: AbstractEvent{E} end
+
 """
 Execution environment for a simulation.
 
@@ -22,7 +24,7 @@ Simulation{T<:TimeType} <: Environment
 - heap :: PriorityQueue{BaseEvent{Simulation{T}}, EventKey{T}}
 - eid :: UInt
 - sid :: UInt
-- active_proc :: Nullable{Process}
+- active_proc :: Nullable{AbstractProcess}
 
 **Constructors**:
 
@@ -36,9 +38,9 @@ mutable struct Simulation{T<:TimeType} <: Environment
   heap :: DataStructures.PriorityQueue{BaseEvent{Simulation{T}}, EventKey{T}}
   eid :: UInt
   sid :: UInt
-  active_proc :: Nullable{Process{Simulation{T}}}
+  active_proc :: Nullable{AbstractProcess{Simulation{T}}}
   function Simulation{T}(initial_time::T) where T<:TimeType
-    new(initial_time, DataStructures.PriorityQueue(BaseEvent{Simulation{T}}, EventKey{T}), zero(UInt), zero(UInt), Nullable{Process{Simulation{T}}}())
+    new(initial_time, DataStructures.PriorityQueue(BaseEvent{Simulation{T}}, EventKey{T}), zero(UInt), zero(UInt), Nullable{AbstractProcess{Simulation{T}}}())
   end
 end
 
@@ -60,15 +62,15 @@ function now{T<:TimeType}(sim::Simulation{T}) :: T
   sim.time
 end
 
-function active_process{T<:TimeType}(sim::Simulation{T}) :: Process{Simulation{T}}
+function active_process{T<:TimeType}(sim::Simulation{T}) :: AbstractProcess{Simulation{T}}
   get(sim.active_proc)
 end
 
 function set_active_process{T<:TimeType}(sim::Simulation{T})
-  sim.active_proc = Nullable{Process{Simulation{T}}}()
+  sim.active_proc = Nullable{AbstractProcess{Simulation{T}}}()
 end
 
-function set_active_process{T<:TimeType}(sim::Simulation{T}, proc::Process{Simulation{T}})
+function set_active_process{T<:TimeType}(sim::Simulation{T}, proc::AbstractProcess{Simulation{T}})
   sim.active_proc = Nullable(proc)
 end
 
@@ -162,3 +164,8 @@ function schedule{T<:TimeType}(bev::BaseEvent{Simulation{T}}, delay::Number=0; p
   P = typeof(eps(bev.env.time))
   schedule(bev, P(delay), priority=priority, value=value)
 end
+
+struct InterruptException{E<:Environment} <: Exception
+  by :: AbstractProcess{E}
+  cause :: Any
+end

src/processes/base.jl

@@ -5,7 +5,8 @@ for test_file in [
   "test_events.jl",
   "test_operators.jl",
   "test_simulation.jl",
-  "test_process.jl",
+  "test_processes.jl",
+  "test_coroutines.jl",
   "test_containers.jl",
   "test_stores.jl",
   ]