From 0212d5e04ad20ddeee78020d0b57f0b8fc8a2746 Mon Sep 17 00:00:00 2001 From: cremet Date: Thu, 27 Nov 2003 17:43:37 +0000 Subject: - Added a means of observing pi-calculus reduct... - Added a means of observing pi-calculus reductions in Pilib. - Made examples "pilib/twoPlaceBuffer" and "pilib/scheduler" use these examples. --- sources/examples/pilib/scheduler.scala | 76 +++++++++++++------------- sources/examples/pilib/twoPlaceBuffer.scala | 83 +++++++++++++---------------- 2 files changed, 75 insertions(+), 84 deletions(-) (limited to 'sources/examples') diff --git a/sources/examples/pilib/scheduler.scala b/sources/examples/pilib/scheduler.scala index ca377ddcf6..bc291d8f56 100644 --- a/sources/examples/pilib/scheduler.scala +++ b/sources/examples/pilib/scheduler.scala @@ -1,6 +1,6 @@ -object scheduler { +import scala.concurrent.pilib._; - import scala.concurrent.pilib._; +object scheduler { /** * Random number generator. @@ -13,60 +13,60 @@ object scheduler { * A cell of the scheduler whose attached agent is allowed to start. */ def A(a: Chan[unit], b: Chan[unit])(d: Chan[unit], c: Chan[unit]): unit = { - a.read; - C(a,b)(d,c) + ///- ... complete here ... + choice ( a * { x => C(a, b)(d, c) }) + ///+ } /** - * A cell of the scheduler in an intermediate state (incorrect) + * A cell of the scheduler in another intermediate state. */ -// def B(a: Chan[unit], b: Chan[unit])(d: Chan[unit], c: Chan[unit]): unit = { -// b.read; -// D(a,b)(d,c) -// } + def C(a: Chan[unit], b: Chan[unit])(d: Chan[unit], c: Chan[unit]): unit = { + ///- ... complete here ... + choice (c * { x => B(a, b)(d, c) }) + ///+ + } /** - * A cell of the scheduler in an intermediate state (correct). + * A cell of the scheduler whose attached agent is allowed to finish. */ - def B(a: Chan[unit], b: Chan[unit])(d: Chan[unit], c: Chan[unit]): unit = + def B(a: Chan[unit], b: Chan[unit])(d: Chan[unit], c: Chan[unit]): unit = { + ///- ... complete here ... + // choice (b * { x => D(a, b)(d, c) }) // incorrect naive solution choice ( - b * (x => D(a,b)(d,c)), - d(()) * ({ b.read; A(a,b)(d,c)}) - ); - - /** - * A cell of the scheduler in another intermediate state. - */ - def C(a: Chan[unit], b: Chan[unit])(d: Chan[unit], c: Chan[unit]): unit = { - c.read; - B(a,b)(d,c) - } + b * { x => choice ( d(()) * A(a, b)(d, c) ) }, // b.'d.A + d(()) * (choice (b * { x => A(a, b)(d, c) })) // 'd.b.A + ) + ///+ + } /** * A cell of the scheduler whose attached agent is not yet allowed to start. */ def D(a: Chan[unit], b: Chan[unit])(d: Chan[unit], c: Chan[unit]): unit = { - d.write(()); - A(a,b)(d,c) + ///- ... complete here ... + choice (d(()) * A(a, b)(d, c)) + ///+ } //***************** Agents ******************// def agent(i: Int)(a: Chan[unit], b: Chan[unit]): unit = { - Thread.sleep(random.nextInt(1000) + 1); - a.write(()); - System.out.println("Starting agent " + i); - Thread.sleep(random.nextInt(1000) + 1); - - // 10% chance that we sleep for a long while. - if(random.nextInt(10) == 0) { - System.out.println("Agent " + i + " sleeps"); - Thread.sleep(20000); + // 50% chance that we sleep forever + if (i == 0 && random.nextInt(10) < 5) { + a.attach(x => System.out.println("Start and sleeps ----> " + i)); + Thread.sleep(random.nextInt(1000)); + a.write(()); + } + else { + a.attach(x => System.out.println("Start ----> " + i)); + b.attach(x => System.out.println("Stop -> " + i)); + Thread.sleep(random.nextInt(1000)); + a.write(()); + Thread.sleep(random.nextInt(1000)); + b.write(()); + agent(i)(a, b) } - - b.write(()); - System.out.println("Ending agent " + i); - agent(i)(a, b) } //***************** Entry function ******************// @@ -77,7 +77,7 @@ object scheduler { def main(args: Array[String]): unit = { val agentNb = 5; - val agents = for(val i <- List.range(0, agentNb)) yield agent(i); + val agents = List.range(0, agentNb) map agent; scheduleAgents(agents); } diff --git a/sources/examples/pilib/twoPlaceBuffer.scala b/sources/examples/pilib/twoPlaceBuffer.scala index 15f29dee07..ad0ab60307 100644 --- a/sources/examples/pilib/twoPlaceBuffer.scala +++ b/sources/examples/pilib/twoPlaceBuffer.scala @@ -1,74 +1,65 @@ -/** Two-place buffer specification and implementation. */ -object twoPlaceBuffer { +import scala.concurrent.pilib._; - import scala.concurrent.pilib._; +/** Two-place buffer specification and implementation. */ +object twoPlaceBuffer with Executable { /** * Specification. */ - def Spec[a](put: Chan[a], get: Chan[a]): unit = { + def Spec[a](in: Chan[a], out: Chan[a]): unit = { - def B0: unit = { - val x = put.read; - B1(x) - } + def B0: unit = choice ( + in * (x => B1(x)) + ); def B1(x: a): unit = choice ( - get(x) * (B0), - put * (y => B2(x, y)) + out(x) * (B0), + in * (y => B2(x, y)) ); - def B2(x: a, y: a): unit = { - get.write(x); - B1(y) - }; + def B2(x: a, y: a): unit = choice ( + out(x) * (B1(y)) + ); B0 } /** - * Implementation. + * Implementation using two one-place buffers. */ - def Impl[a](put: Chan[a], get: Chan[a]): unit = { - - // An empty one-place buffer. - def B0(in: Chan[a], out: Chan[a]): unit = { - val x = in.read; - B1(in, out, x) + def Impl[a](in: Chan[a], out: Chan[a]): unit = { + ///- ... complete here ... + // one-place buffer + def OnePlaceBuffer[a](in: Chan[a], out: Chan[a]): unit = { + def B0: unit = choice ( in * (x => B1(x)) ); + def B1(x: a): unit = choice ( out(x) * (B0)); + B0 } - - // A full one-place buffer containing x. - def B1(in: Chan[a], out: Chan[a], x: a): unit = { - out.write(x); - B0(in, out) - }; - val hidden = new Chan[a]; - spawn < B0(put, hidden) | B0(hidden, get) > + spawn < OnePlaceBuffer(in, hidden) | OnePlaceBuffer(hidden, out) > + ///+ } val random = new java.util.Random(); - def Producer(n: Int, put: Chan[String]): unit = { - Thread.sleep(1 + random.nextInt(1000)); - val msg = "object " + n; - put.write(msg); - System.out.println("Producer gave " + msg); - Producer(n + 1, put) + def Producer(n: Int, in: Chan[String]): unit = { + Thread.sleep(random.nextInt(1000)); + val msg = "" + n; + choice (in(msg) * {}); + Producer(n + 1, in) } - def Consumer(get: Chan[String]): unit = { - Thread.sleep(1 + random.nextInt(1000)); - val msg = get.read; - System.out.println("Consummer took " + msg); - Consumer(get) + def Consumer(out: Chan[String]): unit = { + Thread.sleep(random.nextInt(1000)); + choice (out * { msg => () }); + Consumer(out) } - def main(args: Array[String]): unit = { - val put = new Chan[String]; - val get = new Chan[String]; - spawn < Producer(0, put) | Consumer(get) | Spec(put, get) > - //spawn < Producer(0, put) | Consumer(get) | Impl(put, get) > - } + val in = new Chan[String]; + in.attach(s => System.out.println("put " + s)); + val out = new Chan[String]; + out.attach(s => System.out.println("get " + s)); + //spawn < Producer(0, in) | Consumer(out) | Spec(in, out) > + spawn < Producer(0, in) | Consumer(out) | Impl(in, out) > } -- cgit v1.2.3