The concurrent version of the primes sieve algorithm operates by generating a stream of numbers. The multiples are removed by filter processes. The outline architecture of the program is depicted in Figure 11.1. It is essentially a process structure diagram from which we have omitted the details of action and process labels.

Figure 11.1. Primes Sieve process architecture.

The diagram describes the high-level structure of the program. The process GEN generates the stream of numbers from which multiples are filtered by the FILTER processes. To fully capture the architecture of the program, we must also consider how the application-specific processes interact. Interaction between these elements in the example is described by the PIPE processes. In the terminology of Software Architecture, these processes are termed connectors. Connectors encapsulate the interaction between the components of the architecture. Connectors and components are both modeled as processes. The distinction with respect to modeling is thus essentially methodological. We model the pipe connectors in the example as one-slot buffers as shown below:

const MAX = 9
     range NUM = 2..MAX
     set S = {[NUM],eos}
     PIPE = (put[x:S]->get[x]->PIPE).

The PIPE process buffers elements from the set S which consist of the numbers 2..MAX and the label eos, which is used to signal the end of a stream of numbers. This end of stream signal is required to correctly terminate the program.

To simplify modeling of the GEN and FILTER processes, we introduce an additional FSP construct – the conditional process. The construct can be used in the definition of both primitive and composite processes.

The definition of GEN using the conditional process construct is given below:

GEN       = GEN[2],
     GEN[x:NUM] = (out.put[x] ->
                    if x<MAX then
                    GEN[x+1]
                    else
                    (out.put.eos->end->GEN)
                  ).

The GEN process outputs the numbers 2 to MAX, followed by the signal eos. The action end is used to synchronize termination of the GEN process and the filters. After end occurs, the model re-initializes rather than terminates. This is done so that, if deadlock is detected during analysis, it will be an error and not because of correct termination. The FILTER process records the first value it gets and subsequently filters out multiples of that value from the numbers it receives and forwards to the next filter in the pipeline.

FILTER = (in.get[p:NUM]->prime[p]->FILTER[p]
              |in.get.eos->ENDFILTER
              ),
     FILTER[p:NUM] = (in.get[x:NUM] ->
                       if x%p!=0 then
                         (out.put[x]->FILTER[p])
                       else
                         FILTER[p]
                     |in.get.eos->ENDFILTER
                     ),
     ENDFILTER     = (out.put.eos->end->FILTER).

The composite process that conforms to the structure given in Figure 11.1 can now be defined as:

||PRIMES(N=4) =
        ( gen:GEN
        ||pipe[0..N-1]:PIPE
        ||filter[0..N-1]:FILTER
        )/{ pipe[0]/gen.out,
            pipe[i:0..N-1]/filter[i].in,
            pipe[i:1..N-1]/filter[i-1].out,
            end/{filter[0..N-1].end,gen.end}
          }@{filter[0..N-1].prime,end}.

Safety analysis of this model detects no deadlocks or errors. The minimized LTS for the model is depicted in Figure 11.2. This confirms that the model computes the primes between 2 and 9 and that the program terminates correctly.

Figure 11.2. Minimized PRIMES LTS.

In fact, the concurrent version of the primes sieve algorithm does not ensure that all the primes between 2 and MAX are computed: it computes the first N primes where N is the number of filters in the pipeline. This can easily be confirmed by changing MAX to 11 and re-computing the minimized LTS, which is the same as Figure 11.2. To compute all the primes in the range 2 to 11, five filters are required.

Unbuffered Pipes

We have modeled the filter pipeline using single-slot buffers as the pipes connecting filters. Would the behavior of the overall program change if we used unbuffered pipes? This question can easily be answered by constructing a model in which the PIPE processes are omitted and instead, filters communicate directly by shared actions. This model is listed below. The action pipe[i] relabels the out.put action of filter[i-1] and the in.get action of filter[i].

||PRIMESUNBUF(N=4) =
       (gen:GEN || filter[0..N-1]:FILTER)
         /{ pipe[0]/gen.out.put,
           pipe[i:0..N-1]/filter[i].in.get,
           pipe[i:1..N-1]/filter[i-1].out.put,
           end/{filter[0..N-1].end,gen.end}
         }@{filter[0..N-1].prime,end}.

The minimized LTS for the above model is depicted in Figure 11.3.

Figure 11.3. Minimized PRIMESUNBUF LTS.

The reader can see that the LTS of Figure 11.3 is identical to that of Figure 11.2. The behavior of the program with respect to generating primes and termination has not changed with the removal of buffers. Roscoe (1998) describes a program as being buffer tolerant if its required behavior does not change with the introduction of buffers. We have shown here that the primes sieve program is buffer tolerant to the introduction of a single buffer in each pipe. Buffer tolerance is an important property in the situation where we wish to distribute a concurrent program and, for example, locate filters on different machines. In this situation, buffering may be unavoidable if introduced by the communication system.

Abstracting from Application Detail

We showed above that the primes sieve program is tolerant to the introduction of a single buffer in each pipe. The usual problem of state space explosion arises if we try to analyze a model with more buffers. To overcome this problem, we can abstract from the detailed operation of the primes sieve program. Instead of modeling how primes are computed, we concentrate on how the components of the program interact, independently of the data values that they process. The abstract versions of components can be generated mechanically by relabeling the range of values NUM to be a single value. This is exactly the same technique that we used in section 10.2.2 to generate an abstract model of an asynchronous message port. The abstract versions of GEN, FILTER and PIPE are listed below.

||AGEN    = GEN/{out.put/out.put[NUM]}.

     ||AFILTER = FILTER/{out.put/out.put[NUM],
                         in.get /in.get.[NUM],
                         prime /prime[NUM]
                        }.

     ||APIPE   = PIPE/{put/put[NUM],get/get[NUM]}.

The LTS for the abstract version of a filter process is shown in Figure 11.4. In the detailed version of the filter, the decision to output a value depends on the computation as to whether the value is a multiple of the filter's prime. In the abstract version, this computation has been abstracted to the non-deterministic choice as to whether, in state (4), after an in.get action, the LTS moves to state (5) and does an out.put action or remains in state (4). This is a good example of how nondeterministic choice is used to abstract from computation. We could, of course, have written the abstract versions of the elements of the primes sieve program directly, rather than writing detailed versions and abstracting mechanically as we have done here. In fact, since FSP, as a design choice, has extremely limited facilities for describing and manipulating data, for complex programs abstraction is usually the best way to proceed.

To analyze the primes sieve program with multi-slot pipes, we can use a pipeline of APIPE processes defined recursively as follows:

||MPIPE(B=4) =
       if B==1 then
         APIPE
       else
         (APIPE/{mid/get} || MPIPE(B-1)/{mid/put})
       @{put,get}.

Figure 11.4. Minimized AFILTER LTS.

The abstract model for the primes program, with exactly the architecture of Figure 11.1, can now be defined as:

||APRIMES(N=4,B=3) =
        (gen:AGEN || PRIMEP(N)
        || pipe[0..N-1]:MPIPE(B)
        || filter[0..N-1]:AFILTER)
        /{ pipe[0]/gen.out,
           pipe[i:0..N-1]/filter[i].in,
           pipe[i:1..N-1]/filter[i-1].out,
           end/{filter[0..N-1].end,gen.end}
         }.

where PRIMEP is a safety property which we define in the following discussion.

progress END = {end}

property
PRIMEP(N=4)   = PRIMEP[0],
PRIMEP[i:0..N]= (when (i<N)
                filter[i].prime->PRIMEP[i+1]
                |end -> PRIMEP
                ).

Figure 11.5 is a screen shot of the Primes Sieve applet display. The implementation supports both a buffered and unbuffered implementation of the pipe connector. The figure depicts a run using unbuffered pipes. The box in the top left hand of the display depicts the latest number generated by the thread that implements GEN. The rest of the boxes, at the top, display the latest number received by a filter. The boxes below display the prime used by that filter to remove multiples.

The implementation follows in a straightforward way from the model developed in the previous section. The number generator and filter processes are implemented as threads. As mentioned above, we have provided two implementations for the pipe connector. The classes involved in the program and their inter-relationships are depicted in the class diagram of Figure 11.6. The display is handled by a single class, PrimesCanvas. The methods provided by this class, together with a description of their functions, are listed in Program 11.1.

The code for the Generator and Filter threads is listed in Programs 11.2 and 11.3. The implementation of these threads corresponds closely to the detailed models for GEN and FILTER developed in the previous section. Additional code has been added only to display the values generated and processed. To simplify the display and the pipe implementations, the end-of-stream signal is an integer value that does not occur in the set of generated numbers. The obvious values to use are 0, 1, −1 or MAX+1. In the applet class, Primes.EOS is defined to be −1.

Figure 11.5. Primes Sieve applet display.

Figure 11.6. Primes class diagram.

Instead of implementing buffered and unbuffered pipes from scratch, we have reused classes developed in earlier chapters. The synchronous message-passing class, Channel, from Chapter 10 is used to implement an unbuffered pipe and the bounded buffer class, BufferImpl, from Chapter 5 is used to implement buffered pipes. The Pipe interface and its implementations are listed in Program 11.4.

class PrimesCanvas extends Canvas {

  // display val in an upper box numbered index
  // boxes are numbered from the left
  synchronized void print(int index, int val){...}

  // display, val in a lower box numbered index
  // the lower box indexed by 0 is not displayed
  synchronized void prime(int index, int val){...}

  // clear all boxes
  synchronized void clear(){...}

  }

class Generator extends Thread {
  private PrimesCanvas display;
  private Pipe<Integer> out;
  static int MAX = 50;

  Generator(Pipe<Integer> c, PrimesCanvas d)
    {out=c; display = d;}

  public void run() {
    try {
      for (int i=2;i<=MAX;++i) {
        display.print(0,i);
        out.put(i);
        sleep(500);
      }
      display.print(0,Primes.EOS);
      out.put(Primes.EOS);
    } catch (InterruptedException e){}
   }

}

class Filter extends Thread {
  private PrimesCanvas display;
  private Pipe<Integer> in,out;
  private int index;

  Filter(Pipe<Integer> i, Pipe<Integer> o,
    int id, PrimesCanvas d)
    {in = i; out=o;display = d; index = id;}

  public void run() {
     int i,p;
      try {
        p = in.get();
        display.prime(index,p);
        if (p==Primes.EOS && out!=null) {
          out.put(p); return;
        }
        while(true) {
          i= in.get();
          display.print(index,i);
          sleep(1000);
          if (i==Primes.EOS) {
            if (out!=null) out.put(i); break;
          } else if (i%p!=0 && out!=null)
            out.put(i);
        }
      } catch (InterruptedException e){}
    }

  }

The structure of a generator thread and N filter threads connected by pipes is constructed by the go() method of the Primes applet class. The code is listed below:

public interface Pipe<T> {

  public void put(T o)
     throws InterruptedException; // put object into buffer

  public T get()
     throws InterruptedException; // get object from buffer
  }

  // Unbuffered pipe implementation
  public class PipeImplUnBuf<T> implements Pipe<T> {
     Channel<T> chan = new Channel<T>();

     public void put(T o)
      throws InterruptedException {
      chan.send(o);
    }

    public T get()
      throws InterruptedException {
      return chan.receive();
    }
  }

  // Buffered pipe implementation
  public class PipeImplBuf<T> implements Pipe<T> {
    Buffer<T> buf = new BufferImpl<T>(10);

    public void put(T o)
      throws InterruptedException {
      buf.put(o);
    }

    public T get()
      throws InterruptedException {
      return buf.get();
    }
  }

private void go(boolean buffered) {
         display.clear();

         //create channels
         ArrayList<Pipe<Integer> pipes =
           new ArrayList<Pipe<Integer>();

for (int i=0; i<N; ++i)
  if (buffered)
    pipes.add(new PipeImplBuf<Integer>());
  else
    pipes.add(new PipeImplUnBuf<Integer>());

  //create threads
  gen = new Generator(pipes.get(0),display);
  for (int i=0; i<N; ++i)
    filter[i] = new Filter(pipes.get(i),
           i<N-1?pipes.get(i+1):null,i+1,display);
    gen.start();
    for (int i=0; i<N; ++i) filter[i].start();
}

Linda is the collective name given by Carriero and Gelernter (1989a) to a set of primitive operations used to access a data structure called a tuple space. A tuple space is a shared associative memory consisting of a collection of tagged data records called tuples. Each data tuple in a tuple space has the form:

The tag is a literal string used to distinguish between tuples representing different classes of data. value_i are zero or more data values: integers, floats and so on.

There are three basic Linda operations for manipulating data tuples: out, in and rd. A process deposits a tuple in a tuple space using:

Execution of out completes when the expressions have been evaluated and the resulting tuple has been deposited in the tuple space. The operation is similar to an asynchronous message send except that the tuple is stored in an unordered tuple space rather than appended to the queue associated with a specific port. A process removes a tuple from the tuple space by executing:

Each field _i is either an expression or a formal parameter of the form ?var where var is a local variable in the executing process. The arguments to in are called a template; the process executing in blocks until the tuple space contains a tuple that matches the template and then removes it. A template matches a data tuple in the following circumstances: the tags are identical, the template and tuple have the same number of fields, the expressions in the template are equal to the corresponding values in the tuple, and the variables in the template have the same type as the corresponding values in the tuple. When the matching tuple is removed from the tuple space, the formal parameters in the template are assigned the corresponding values from the tuple. The in operation is similar to a message receive operation with the tag and values in the template serving to identify the port.

The third basic operation is rd, which functions in exactly the same way as in except that the tuple matching the template is not removed from the tuple space. The operation is used to examine the contents of a tuple space without modifying it. Linda also provides non-blocking versions of in and rd called inp and rdp whichreturntrueif amatchingtupleis found and return false otherwise.

Linda has a sixth operation called eval that creates an active or process tuple. The eval operation is similar to an out except that one of the arguments is a procedure that operates on the other arguments. A process is created to evaluate the procedure and the process tuple becomes a passive data tuple when the procedure terminates. This eval operation is not necessary when a system has some other mechanism for creating new processes. It is not used in the following examples.

Tuple Space Model

Our modeling approach requires that we construct finite state models. Consequently, we must model a tuple space with a finite set of tuple values. In addition, since a tuple space can contain more than one tuple with the same value, we must fix the numberof copies of each value that are allowed. We define this number to be the constant N and the allowed values to be the set Tuples.

const N = ...
set Tuples = {...}

The precise definition of N and Tuples depends on the context in which we use the tuple space model. Each tuple value is modeled by an FSP label of the form tag.val₁... val_n. We define a process to manage each tuple value and the tuple space is then modeled by the parallel composition of these processes:

const False = 0
const True = 1
range Bool = False..True

TUPLE(T='any) = TUPLE[0],
TUPLE[i:0..N]
  = (out[T]                   -> TUPLE[i+1]
    |when (i>0) in[T]         -> TUPLE[i-1]
    |when (i>0) inp[True][T]  -> TUPLE[i-1]
    |when (i==0)inp[False][T] -> TUPLE[i]
    |when (i>0) rd[T]         -> TUPLE[i]
    |rdp[i>0][T]              -> TUPLE[i]
    ).

||TUPLESPACE = forall [t:Tuples] TUPLE(t).

The LTS for TUPLE value any with N=2 is depicted in Figure 11.8. Exceeding the capacity by performing more than two out operations leads to an ERROR.

An example of a conditional operation on the tuple space would be:

inp[b:Bool][t:Tuples]

Figure 11.8. TUPLE LTS.

The value of the local variable t is only valid when b is true. Each TUPLE process has in its alphabet the operations on one specific tuple value. The alphabet of TUPLESPACE is defined by the set TupleAlpha:

set TupleAlpha
  = {{in,out,rd,rdp[Bool],inp[Bool]}.Tuples}

A process that shares access to the tuple space must include all the actions of this set in its alphabet.

public interface TupleSpace {

  // deposits data in tuple space
  public void out (String tag, Object data);

  // extracts object with tag from tuple space, blocks if not available
  public Object in (String tag)
                  throws InterruptedException;
  // reads object with tag from tuple space, blocks if not available
  public Object rd (String tag)

throws InterruptedException;
  // extracts object if available, return null if not available
  public Object inp (String tag);

  // reads object if available, return null if not available
  public Object rdp (String tag);

  }

We model a simple Supervisor–Worker system in which the supervisor initially outputs a set of tasks to the tuple space and then collects results. Each worker repetitively gets a task and computes the result. The algorithms for the supervisor and each worker process are sketched below:

Supervisor::
     forall tasks: out("task",...)
     forall results: in("result",...)
     out("stop")
   Worker::
     while not rdp("stop") do
        in("task",...)
        compute result
        out("result",...)

To terminate the program, the supervisor outputs a tuple with the tag "stop" when it has collected all the results it requires. Workers run until they read this tuple. The set of tuple values and the maximum number of copies of each value are defined for the model as:

const N      = 2
set   Tuples = {task,result,stop}

class TupleSpaceImpl implements TupleSpace {
  private Hashtable tuples = new Hashtable();

  public synchronized void out(String tag,Object data){
    Vector v = (Vector) tuples.get(tag);
    if (v == null) {
      v = new Vector();
      tuples.put(tag,v);
    }
    v.addElement(data);
    notifyAll();
  }

  private Object get(String tag, boolean remove) {
   Vector v = (Vector) tuples.get(tag);
   if (v == null) return null;
   if (v.size() == 0) return null;
   Object o = v.firstElement();
   if (remove) v.removeElementAt(0);
   return o;
 }

   public synchronized Object in (String tag)
                   throws InterruptedException {
     Object o;
     while ((o = get(tag,true)) == null) wait();
     return o;
   }

    public Object rd (String tag)
                    throws InterruptedException {
      Object o;
      while ((o = get(tag,false)) == null) wait();
      return o;
    }

    public synchronized Object inp (String tag) {
      return get(tag,true);
    }

    public synchronized Object rdp (String tag) {
      return get(tag,false);
    }
  }

The supervisor outputs N tasks to the tuple space, collects N results and then outputs the "stop" tuple and terminates.

SUPERVISOR   = TASK[1],
     TASK[i:1..N] =
       (out.task  ->
          if i<N then TASK[i+1] else RESULT[1]),
     RESULT[i:1..N] =
       (in.result ->
          if i<N then RESULT[i+1] else FINISH),
     FINISH =
       (out.stop  -> end -> STOP) + TupleAlpha.

The worker checks for the "stop" tuple before getting a task and outputting the result. The worker terminates when it reads "stop" successfully.

WORKER =
       (rdp[b:Bool].stop->
         if (!b) then
           (in.task -> out.result -> WORKER)
         else
           (end -> STOP)
       )+TupleAlpha.

The LTS for both SUPERVISOR and WORKER with N=2 is depicted in Figure 11.9.

Figure 11.9. SUPERVISOR and WORKER LTS.

In the primes sieve example, we arranged that the behavior was cyclic to avoid detecting a deadlock in the case of correct termination. An alternative way of avoiding this situation is to provide a process that can still engage in actions after the end action has occurred. We use this technique here and define an ATEND process that engages in the action ended after the correct termination action end occurs.

ATEND   = (end->ENDED),
     ENDED = (ended->ENDED).

A Supervisor–Worker model with two workers called redWork and blueWork, which conforms to the architecture of Figure 11.7, can now be defined by:

||SUPERVISOR_WORKER
         = ( supervisor:SUPERVISOR
           || {redWork,blueWork}:WORKER
           || {supervisor,redWork,blueWork}::TUPLESPACE
           || ATEND
           )/{end/{supervisor,redWork,blueWork}.end}.

Analysis

Safety analysis of this model using LTSA reveals the following deadlock:

Trace to DEADLOCK:
         supervisor.out.task
         supervisor.out.task
         redWork.rdp.0.stop            – rdp returns false
         redWork.in.task
         redWork.out.result
         supervisor.in.result
         redWork.rdp.0.stop            – rdp returns false
         redWork.in.task
         redWork.out.result
         supervisor.in.result
         redWork.rdp.0.stop            – rdp returns false
         supervisor.out.stop
         blueWork.rdp.1.stop           – rdp returns true

This trace is for an execution in which the red worker computes the results for the two tasks put into tuple space by the supervisor. This is quite legitimate behavior for a real system since workers can run at different speeds and take different amounts of time to start. The deadlock occurs because the supervisor only outputs the "stop" tuple after the red worker attempts to read it. When the red worker tries to read, the "stop" tuple has not yet been put into the tuple space and, consequently, the worker does not terminate but blocks waiting for another task. Since the supervisor has finished, no more tuples will be put into the tuple space and consequently, the worker will never terminate.

This deadlock, which can be repeated for different numbers of tasks and workers, indicates that the termination scheme we have adopted is incorrect. Although the supervisor completes the computation, workers may not terminate. It relies on a worker being able to input tuples until it reads the "stop" tuple. As the model demonstrates, this may not happen. This would be a difficult error to observe in an implementation since the program would produce the correct computational result. However, after an execution, worker processes would be blocked and consequently retain execution resources such as memory and system resources such as control blocks. Only after a number of executions might the user observe a system crash due to many hung processes. Nevertheless, this technique of using a "stop" tuple appears in an example Linda program in a standard textbook on concurrent programming!

A simple way of implementing termination correctly would be to make a worker wait for either inputting a "task" tuple or reading a "stop" tuple. Unfortunately, while this is easy to model, it cannot easily be implemented since Linda does not have an equivalent to the selective receive described in Chapter 10. Instead, we adopt a scheme in which the supervisor outputs a "task" tuple with a special stop value. When a worker inputs this value, it outputs it again and then terminates. Because a worker outputs the stop task before terminating, each worker will eventually input it and terminate. This termination technique appears in algorithms published by the designers of Linda (Carriero and Gelernter, 1989b). The revised algorithms for supervisor and worker are sketched below:

Supervisor::
        forall tasks:- out("task",...)
        forall results:- in("result",...)
        out("task",stop)
Worker::
         while true do
                  in("task",...)
                  if value is stop then out("task",stop); exit
                  compute result
                  out("result",...)

The tuple definitions and models for supervisor and worker now become:

set Tuples  = {task,task.stop,result}

SUPERVISOR   = TASK[1],
TASK[i:1..N] =
  out.task ->

if i<N then TASK[i+1] else RESULT[1]),
RESULT[i:1..N] =
  (in.result ->
      if i<N then RESULT[i+1] else FINISH),
FINISH =
   (out.task.stop -> end -> STOP)
   + TupleAlpha.

WORKER =
  (in.task -> out.result -> WORKER
  |in.task.stop -> out.task.stop -> end ->STOP
  ) + TupleAlpha.

The revised model does not deadlock and satisfies the progress property:

progress END = {ended}

A sample trace from this model, which again has the red worker computing both tasks, is shown in Figure 11.10.

Figure 11.10. Trace of Supervisor–Worker model.

In the first section of this chapter, the primes sieve application was modeled in some detail. We then abstracted from the application to investigate the concurrent properties of the Filter Pipeline architecture. In this section, we have modeled the Supervisor–Worker architecture directly without reference to an application. We were able to discover a problem with termination and provide a general solution that can be used in any application implemented within the framework of the architecture.

To illustrate the implementation and operation of Supervisor–Worker architectures, we develop a program that computes an approximate value of the area under a curve using the rectangle method. More precisely, the program computes an approximate value for the integral:

The rectangle method involves summing the areas of small rectangles that nearly fit under the curve as shown in Figure 11.11.

Figure 11.11. Rectangle method.

In the Supervisor–Worker implementation, the supervisor determines how many rectangles to compute and hands the task of computing the area of the rectangles to the workers. The demonstration program has four worker threads each with a different color attribute. When the supervisor inputs a result, it displays the rectangle corresponding to that result with the color of the worker. The display of a completed computation is depicted in Figure 11.12.

Each worker is made to run at a different speed by performing a delay before outputting the result to the tuple space. The value of this delay is chosen at random when the worker is created. Consequently, each run behaves differently. The display of Figure 11.12 depicts a run in which some workers compute more results than others. During a run, the number of the task that each worker thread is currently computing is displayed. The last task that the worker completed is displayed at the end of a run. The class diagram for the demonstration program is shown in Figure 11.13.

Figure 11.12. Supervisor–Worker applet.

Figure 11.13. Supervisor–Worker class diagram.

The displays for the supervisor and worker threads are handled, respectively, by the classes SupervisorCanvas and WorkerCanvas. The methods provided by these classes, together with a description of what they do, are listed in Program 11.7. The interface for the function f(x) together with three implementations are also included in Program 11.7.

class SupervisorCanvas extends Canvas {

  // display rectangle slice i with color c, add a to area field
  synchronized void setSlice(
             int i,double a,Color c) {...}

  // reset display to clear rectangle slices and draw curve for f
  synchronized void reset(Function f) {...}

}

  class WorkerCanvas extends Panel {
  // display current task number val
  synchronized void setTask(int val) {...}
}

  interface Function {
     double fn(double x);
}

  class OneMinusXsquared implements Function {
    public double fn (double x) {return 1-x*x;}
  }

  class OneMinusXcubed implements Function {
    public double fn (double x) {return 1-x*x*x;}
  }

  class XsquaredPlusPoint1 implements Function {
    public double fn (double x) {return x*x+0.1;}
  }

A task that is output to the tuple space by the supervisor thread is represented by a single integer value. This value identifies the rectangle for which the worker computes the area. A result requires a more complex data structure since, for display purposes, the result includes the rectangle number and the worker color attribute in addition to the computed area of the rectangle. The definition of the Result and the Supervisor classes is listed in Program 11.8.

The supervisor thread is a direct translation from the model. It outputs the set of rectangle tasks to the tuple space and then collects the results. Stop is encoded as a "task" tuple with the value −1, which falls outside the range of rectangle identifiers. The Worker thread class is listed in Program 11.9.

class Result {
  int task;
  Color worker;
  double area;
  Result(int s, double a, Color c)
    {task =s; worker=c; area=a;}
}

class Supervisor extends Thread {
  SupervisorCanvas display;
  TupleSpace bag;
  Integer stop = new Integer(-1);

  Supervisor(SupervisorCanvas d, TupleSpace b)
    { display = d; bag = b; }

  public void run () {
    try {
       // output tasks to tuplespace
       for (int i=0; i<SupervisorCanvas.Nslice; ++i)
          bag.out("task",new Integer(i));
       // collect results
       for (int i=0; i<display.Nslice; ++i) {
          Result r = (Result)bag.in("result");
          display.setSlice(r.task,r.area,r.worker);
        }
        // output stop tuple
        bag.out("task",stop);
      } catch (InterruptedException e){}
    }
  }

The choice in the worker model between a task tuple to compute and a stop task is implemented as a test on the value of the task. The worker thread terminates when it receives a negative task value. The worker thread is able to compute the area given only a single integer since this integer indicates which "slice" of the range of x from 0 to 1.0 for which it is to compute the rectangle. The worker is initialized with a function object.

The structure of supervisor, worker and tuple space is constructed by the go() method of the SupervisorWorker applet class. The code is listed below:

class Worker extends Thread {
  WorkerCanvas display;
  Function func;
  TupleSpace bag;
  int processingTime = (int)(6000*Math.random());

  Worker(WorkerCanvas d, TupleSpace b, Function f)
    { display = d; bag = b; func = f; }

  public void run () {
    double deltaX = 1.0/SupervisorCanvas.Nslice;
    try {
      while(true){
         // get new task from tuple space
         Integer task = (Integer)bag.in("task");
         int slice = task.intValue();
         if (slice <0) {   // stop if negative
             bag.out("task",task);
             break;
         }
         display.setTask(slice);
         sleep(processingTime);
         double area
           = deltaX*func.fn(deltaX*slice+deltaX/2);
         // output result to tuple space
         bag.out( "result",
            new Result(slice,area,display.worker));
       }
     } catch (InterruptedException e){}
   }
}

private void go(Function fn) {
  display.reset(fn);
  TupleSpace bag = new TupleSpaceImpl();
  redWork = new Worker(red,bag,fn);
  greenWork = new Worker(green,bag,fn);
  yellowWork = new Worker(yellow,bag,fn);
  blueWork = new Worker(blue,bag,fn);
  supervisor = new Supervisor(display,bag);
  redWork.start();
  greenWork.start();

yellowWork.start();
  blueWork.start();
  supervisor.start();
}

where display is an instance of SupervisorCanvas and red, green, yellow and blue are instances of WorkerCanvas.

The model is defined for a fixed set of listeners and a fixed set of event patterns:

set Listeners = {a,b,c,d}
set Pattern   = {pat1,pat2}

The event manager is modeled by a set of REGISTER processes, each of which controls the registration and event propagation for a single, particular listener.

REGISTER = IDLE,
     IDLE = (register[p:Pattern] -> MATCH[p]
            |announce[Pattern]   -> IDLE
            ),

     MATCH[p:Pattern] =
            (announce[a:Pattern] ->
                if (a==p) then
                   (event[a] -> MATCH[p]
                   |deregister -> IDLE)
                else
                   MATCH[p]
            |deregister -> IDLE
            ).

     ||EVENTMANAGER = (Listeners:REGISTER)
                    /{announce/Listeners.announce}.

The REGISTER process ensures that the event action for a listener only occurs if the listener has previously registered, has not yet unregistered, and if the event pattern matches the pattern with which the listener registered. Figure 11.15 depicts the LTS for a:REGISTER for listener a.

Figure 11.15. LTS for a:REGISTER.

The announcer is modeled as repeatedly announcing an event for one of the patterns defined by the Pattern set:

ANNOUNCER = (announce[Pattern] -> ANNOUNCER).

The listener initially registers for events of a particular pattern and then either performs local computation, modeled by the action compute, or receives an event. On receiving an event, the process either continues computing or deregisters and stops.

LISTENER(P='pattern) =
       (register[P] -> LISTENING),
     LISTENING =
       (compute  -> LISTENING
       |event[P] -> LISTENING
       |event[P] -> deregister -> STOP
       )+{register[Pattern]}.

ANNOUNCER_LISTENER describes a system with four listeners a, b, c, d, in which a and c register for events with pattern pat1 and b and d register for pat2.

||ANNOUNCER_LISTENER
                = ( a:LISTENER('pat1)
                  ||b:LISTENER('pat2)
                  ||c:LISTENER('pat1)
                  ||d:LISTENER('pat2)
                  ||EVENTMANAGER
                  ||ANNOUNCER
                  ||Listeners:SAFE).

Analysis

The safety property, SAFE, included in the composite process ANNOUNCER_ LISTENER, asserts that each listener only receives events while it is registered and only those events with the pattern for which it registered. The property is defined below:

property
  SAFE = (register[p:Pattern]  -> SAFE[p]),
  SAFE[p:Pattern]= (event[p]   -> SAFE[p]
                   |deregister -> SAFE
                   ).

Figure 11.16. ANNOUNCER_LISTENER trace.

Safety analysis reveals that the system has no deadlocks or safety violations. A sample execution trace is shown in Figure 11.16.

The important progress property for this system is that the announcer should be able to announce events independently of the state of listeners, i.e. whether or not listeners are registered and whether or not listeners have stopped. We can assert this using the following set of progress properties:

progress ANNOUNCE[p:Pattern] = {announce[p]}

Progress analysis using LTSA verifies that these properties hold for ANNOUNCER_ LISTENER.

To illustrate the use of the Announcer – Listener architecture, we implement the simple game depicted in Figure 11.17. The objective of the game is to hit all the moving colored blocks with the minimum number of mouse presses. A moving block is hit by pressing the mouse button when the mouse pointer is on top of the block. When a block is hit, it turns black and stops moving.

Each block is controlled by a separate thread that causes the block it controls to jump about the display at random. The threads also listen for mouse events that are announced by the display canvas in which the blocks move. These events are generated by the AWT and the program uses the AWT classes provided for event handling. The class diagram for the program is depicted in Figure 11.18.

Figure 11.17. EventDemo applet display.

class BoxCanvas extends Canvas {

  // clear all boxes
  synchronized void reset(){...}

  // draw colored box id at position x,y
  synchronized void moveBox(int id, int x, int y){...}

  // draw black box id at position x,y
  synchronized void blackBox(int id, int x, int y){...}
}

The display for the EventDemo applet is provided by the BoxCanvas class described in outline by Program 11.10.

The AWT interface MouseListener describes a set of methods that are invoked as a result of mouse actions. To avoid implementing all these methods, since we are only interested in the mouse-pressed event, we use the adapter class MouseAdapter which provides null implementations for all the methods in the MouseListener interface. MyClass extends MouseAdapter and provides an implementation for the mousePressed method. MyClass is an inner class of the BoxMover thread class; consequently, it can directly call private BoxMover methods. The code for BoxMover and MyListener is listed in Program 11.11.

Figure 11.18. EventDemo class diagram.

The run() method of BoxMover repeatedly computes a random position and displays a box at that position. After displaying the box, waitHit() is called. This uses a timed wait() and returns for two reasons: either the wait delay expires and the value false is returned or isHit() computes that a hit has occurred and calls notify() in which case true is returned. Because MyListener is registered as a listener for events generated by the display, whenever the display announces a mouse-pressed event, MyListener.mousePressed() is invoked. This in turn calls isHit() with the mouse coordinates contained in the MouseEvent parameter.

If we compare the run() method with the LISTENER process from the model, the difference in behavior is the addition of a timeout action which ensures that after a delay, if no mouse event occurs, a new box position is computed and displayed. A model specific to the BoxMover thread is described by:

BOXMOVER(P='pattern) =
       (register[P] -> LISTENING),
     LISTENING =
       (compute->       // compute and display position
         (timeout -> LISTENING    // no mouse event
         |event[P] -> timeout -> LISTENING // miss
         |event[P] -> deregister -> STOP    // hit
         )
       )+{register[Pattern]}.

import java.awt.event.*;
  class BoxMover extends Thread {
    private BoxCanvas display;
    private int id, delay, MaxX, MaxY, x, y;
    private boolean hit=false;
    private MyListener listener = new MyListener();

    BoxMover(BoxCanvas d, int id, int delay) {
      display = d; this.id = id; this.delay=delay;
      display.addMouseListener(listener); // register
      MaxX = d.getSize().width; MaxY = d.getSize().height;
    }

    private synchronized void isHit(int mx, int my) {
      hit = (mx>x && mx<x+display.BOXSIZE
          && my>y && my<y+display.BOXSIZE);
      if (hit) notify();
    }

    private synchronized boolean waitHit()
      throws InterruptedException {
      wait(delay);
      return hit;
    }

    public void run() {
      try {
        while (true) {
            x = (int)(MaxX*Math.random());
            y = (int)(MaxY*Math.random());
            display.moveBox(id,x,y);
            if (waitHit()) break;
        }
      } catch (InterruptedException e){}
      display.blackBox(id,x,y);
      display.removeMouseListener(listener); // deregister
    }

    class MyListener extends MouseAdapter {
      public void mousePressed(MouseEvent e) {
        isHit(e.getX(),e.getY());
      }
    }
  }

The reader should verify that the safety and progress properties for the ANNOUNCER_LISTENER still hold when BOXMOVER is substituted for LISTENER.