Add processes

Author: BenLauwens

src/SimJulia.jl

@@ -7,7 +7,7 @@ module SimJulia
 
   using DataStructures
 
-  import Base.run, Base.now, Base.isless, Base.show
+  import Base.run, Base.now, Base.isless, Base.show, Base.interrupt, Base.yield
   import Base.(&), Base.(|)
 
   export AbstractEvent
@@ -18,10 +18,13 @@ module SimJulia
   export succeed, fail, append_callback, @callback, remove_callback
   export Operator
   export (&), (|)
+  export Process, @process
+  export yield, interrupt
 
   include("base.jl")
   include("simulations.jl")
   include("events.jl")
   include("operators.jl")
-
+  include("tasks/base.jl")
+  include("processes.jl")
 end

src/processes.jl

@@ -0,0 +1,51 @@
+type Process <: AbstractProcess
+  bev :: BaseEvent
+  task :: Task
+  target :: AbstractEvent
+  resume :: Function
+  function Process(func::Function, env::Environment, args::Any...)
+    proc = new()
+    proc.bev = BaseEvent(env)
+    proc.task = @task func(env, args...)
+    proc.target = Timeout(env)
+    proc.resume = @callback execute(proc.target, proc)
+    return proc
+  end
+end
+
+macro process(expr)
+  expr.head != :call && error("Expression is not a function call!")
+  func = esc(expr.args[1])
+  args = [esc(expr.args[n]) for n in 2:length(expr.args)]
+  :(Process($(func), $(args...)))
+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)
+  ret = SimJulia.produce(nothing)
+  isa(ret, Exception) && throw(ret)
+  return ret
+end
+
+function execute(ev::AbstractEvent, proc::Process)
+  try
+    env = environment(ev)
+    set_active_process(env, proc)
+    ret = SimJulia.consume(proc.task, value(ev))
+    reset_active_process(env)
+    istaskdone(proc.task) && schedule(proc.bev, value=ret)
+  catch exc
+    rethrow(exc)
+  end
+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)
+  end
+end

src/tasks/base.jl

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

test/processes.jl

@@ -0,0 +1,70 @@
+using SimJulia
+
+function fibonnaci(sim::Simulation)
+  a = 0
+  b = 1
+  while true
+    println(a)
+    yield(Timeout(sim, 1))
+    a, b = b, a+b
+  end
+end
+
+function test_process(sim::Simulation, ev::AbstractEvent)
+  yield(ev)
+end
+
+function test_process_exception(sim::Simulation, ev::AbstractEvent)
+  try
+    value = yield(ev)
+    println("hi")
+  catch exc
+    println("$exc has been thrown")
+  end
+end
+
+function test_interrupter(sim::Simulation, proc::Process)
+  yield(Timeout(sim, 2))
+  interrupt(proc)
+end
+
+function test_interrupted(sim::Simulation)
+  try
+    yield(Timeout(sim, 10))
+  catch exc
+    if isa(exc, SimJulia.InterruptException)
+      println("$(active_process(sim)) interrupted")
+    end
+  end
+  yield(Timeout(sim, 10))
+  throw(TestException())
+end
+
+sim = Simulation()
+@process fibonnaci(sim)
+run(sim, 10)
+
+sim = Simulation()
+@process test_process(sim, succeed(Event(sim)))
+run(sim)
+
+sim = Simulation()
+@process test_process_exception(sim, Timeout(sim, 1, value=TestException()))
+try
+  run(sim)
+catch exc
+  println("$exc has been thrown")
+end
+
+sim = Simulation()
+@process test_process_exception(sim, Timeout(sim, value=TestException()))
+run(sim)
+
+sim = Simulation()
+proc = @process test_interrupted(sim)
+@process test_interrupter(sim, proc)
+try
+  run(sim)
+catch exc
+  println("$exc has been thrown")
+end

test/runtests.jl

@@ -4,6 +4,8 @@ for test_file in [
   "base.jl",
   "simulations.jl",
   "events.jl",
-  "operators.jl",]
+  "operators.jl",
+  "tasks/base.jl",
+  "processes.jl",]
   include(testpath(test_file))
 end

test/tasks/base.jl

@@ -0,0 +1,22 @@
+using SimJulia
+
+function fibonnaci(sim::Simulation)
+  a = 0.0
+  b = 1.0
+  for i = 1:10
+    println("Fibonnaci value equals ", a, " at time ", now(sim))
+    yield(Timeout(sim, 1.0))
+    a, b = b, a+b
+  end
+  a
+end
+
+function wait_for_process(sim::Simulation, process::Process)
+  val = yield(process)
+  println("Fibonnaci ended with ", val, " at time ", now(sim))
+end
+
+sim = Simulation()
+fib = @process fibonnaci(sim)
+wait = @process wait_for_process(sim, fib)
+run(sim)