diff options
Diffstat (limited to 'sources')
-rw-r--r-- | sources/scala/concurrent/NameServer.scala | 25 | ||||
-rw-r--r-- | sources/scala/concurrent/Process.scala | 68 |
2 files changed, 93 insertions, 0 deletions
diff --git a/sources/scala/concurrent/NameServer.scala b/sources/scala/concurrent/NameServer.scala new file mode 100644 index 0000000000..21375d7e9d --- /dev/null +++ b/sources/scala/concurrent/NameServer.scala @@ -0,0 +1,25 @@ +package scala.concurrent; + +object NameServer { + val names=new scala.collection.mutable.HashMap[Symbol,Process]; + + def register(name:Symbol,proc:Process) = { + if (names.contains(name)) error("Name:" + name + " already registred"); + names += name -> proc; + } + + def unregister(name:Symbol) = { + if (names.contains(name)) + names -= name; + else + error("Name:" + name + " not registred"); + } + + def whereis(name:Symbol):Option[Process] = + names.get(name); + + def send(name:Symbol, msg:Actor#Message) = { + names(name).send(msg) + } + +} diff --git a/sources/scala/concurrent/Process.scala b/sources/scala/concurrent/Process.scala new file mode 100644 index 0000000000..0fd4b34da2 --- /dev/null +++ b/sources/scala/concurrent/Process.scala @@ -0,0 +1,68 @@ +package scala.concurrent; + +object Process { + def spawn(def body:unit):Process = { + val p = new Process(body); + p.start(); + p; + } + def spawn_link(def body:unit):Process = { + self.spawn_link(body); + } + + def send(p:Process,msg:Actor#Message) = + p.send(msg); + def receive[a](f: PartialFunction[Actor#Message, a]): a = + self.receive(f); + + def receiveWithin[a](msec: long)(f: PartialFunction[Actor#Message, a]):a = + self.receiveWithin(msec)(f); + + def self:Process = { + if (Thread.currentThread().isInstanceOf[Process]) + Thread.currentThread().asInstanceOf[Process] + else error("Self called outside a process"); + } + + def exit(p:Process,reason:AnyRef) = + p.exit(reason); +} + +class Process(def body:unit) extends Actor() { + private var exitReason:AnyRef = null; + private var links:List[Process] = Nil; + override def run() = { + try {body} + catch { + case _:java.lang.InterruptedException => + signal(exitReason); + case (exitSignal) => + signal(exitSignal); + } + } + + private def signal(s:Actor#Message) = { + links.foreach((p:Process) => p.send(Tuple3('EXIT,this,s))); + } + + def !(msg:Actor#Message) = + send(msg); + + def link(p:Process) = { + links = p::links; + } + + def spawn_link(def body:unit) = { + val p = new Process(body); + p.link(this); + p.start(); + p + } + + def self = this; + + def exit(reason:AnyRef):unit = { + exitReason = reason; + interrupt(); + } +} |