mondrian.html

Concurrent Programming in Java

The easiest stages to build are Linear, single-input, single-output
stages. A Painter is just:

<pre>
public class Painter extends SingleOutputPushStage implements PushStage {
  protected Color color_;
  public Painter(Color c) { super(); color_ = c;  }
  public synchronized void putA(Box p) {
    p.color(color_);
    next1_.putA(p);
  }
}
</pre>

Similarly for a Wrapper:

<pre>
public class Wrapper extends SingleOutputPushStage implements PushStage {
  protected int thickness_;
  public Wrapper(int thickness) { super(); thickness_ = thickness;  }
  public synchronized void putA(Box p) {
    next1_.putA(new WrappedBox(p, new Point(thickness_, thickness_)));
  }
}
</pre>

A Flipper may use the same strategy as well:

<pre>
public class Flipper extends SingleOutputPushStage implements PushStage {
  public Flipper() { super();  }
  public synchronized void putA(Box p) {
    if (p instanceof JoinedPair) {
      ((JoinedPair) p).flip();
      next1_.putA(p);
    }
  }
}
</pre>

However, there are some reasonable alternatives. Unlike the Painter
and Wrapper stages which apply to any kind of Box, Flippers only
pass through JoinedPairs. If a Flipper receives something other
than a Joined pair, it will just ``drop'' it. While this might
be OK, it would probably be a better idea to <em>explicity</em>
drop it by sending it to a stage specifically designed to
do so. In the spirit of Unix pipes and filters, we can call it:

<pre>
public class DevNull implements PushStage {
  public DevNull() { }
  public void putA(Box p) { }
}
</pre>

And then rework a version of Flipper to use it. (Left as an exercise!)


<h3>Dual Input Stages</h3>

The most basic kind of Dual Input Stage is a simple Merger,
that accepts messages on either channel, and relays them on
to a single successor:

<pre>
public class Merger extends SingleOutputPushStage implements DualInputPushStage {
  public Merger() { super(); }
  public synchronized void putA(Box p) { next1_.putA(p); }
  public synchronized void putB(Box p) { next1_.putA(p); }
}
</pre>

We have two kinds of Combiners, Horizontal and Vertical Joiners.
As did the representation classes, the stages share enough in
common to build a common superclass. As with most kinds of
Combiners, Joiner stages must block on one input until
they receive the other. This can be done via the usual
<a href="synchDesign.html" tppabs="http://www.foi.hr/~dpavlin/java/mirrors/g.oswego.edu/dl/pats/synchDesign.html"> guard mechanics</a>:

<pre>
public abstract class Joiner extends SingleOutputPushStage 
   implements DualInputPushStage {

  protected Box a_;
  protected Box b_;

  protected Joiner() {  super();  a_ = null;  b_ = null;  }

  protected abstract Box join(); // differs for Horizontal vs Vertical

  protected void output() {   // pass on a completed Pair if possible
    if (a_ != null && b_ != null) {
      Box out = join();
      a_ = null;
      b_ = null;
      next1_.putA(out);
      notifyAll();
    }
  }

  public synchronized void putA(Box p) {
    // block until a held a_ part is successfully joined with a b_
    while (a_ != null) {
      try { wait(); } catch (InterruptedException ex) { return; }
    }
    a_ = p;
    output();
  }

  public synchronized void putB(Box p) { // symmetrical to putA
    while (b_ != null) {
      try { wait(); } catch (InterruptedException ex) { return; }
    }
    b_ = p;
    output();
  }
}

public class HorizontalJoiner extends Joiner {
  public HorizontalJoiner() { super(); }
  public synchronized Box join() {
    return new HorizontallyJoinedPair(a_, b_);
  }
}


public class VerticalJoiner extends Joiner {
  public VerticalJoiner() { super(); }
  public synchronized Box join() {
    return new VerticallyJoinedPair(a_, b_);
  }
}

</pre>

<h3>Dual Output Stages</h3>

As with just about all Push-driven systems, our Output stages should
generate Threads to drive at least one of their outputs to maintain
liveness. This is true even for a simple kind of stage like an
Alternator that just outputs alternate inputs on alternate outputs:

<pre>
public class Alternator extends DualOutputPushStage implements PushStage {
  protected boolean  flip_;
  public Alternator() { super();  flip_ = false;  }
  public synchronized void putA(Box p) {
    if (!flip_) next1_.putA(p); 
    else   (new Thread(new PutBRunner(next2_, p))).start();
    flip_ = !flip_;
  }
}
</pre>

And similarly for a Cloner:

<pre>
public class Cloner extends DualOutputPushStage implements PushStage {
  public Cloner() { super(); }
  public synchronized void putA(Box p) {
    Box p2 = p.duplicate();
    (new Thread(new PutARunner(next1_, p))).start();
    next2_.putA(p2);
  }
}
</pre>

A Screener is a stage that directs all inputs obeying some
predicate to one channel, and all others to the other.
We can build a generic one by encapsulating the Predicate
to check as:

<pre>
public interface BoxPredicate {
  public boolean predicate(Box p);
}
</pre>

And implement it, for example with one that makes sure that a
Box fits within a given (symmetric, in this case) bound:

<pre>
public class MaxSizePredicate implements BoxPredicate {
  private int max_ ;
  public MaxSizePredicate(int max) { max_ = max;  }
  public boolean predicate(Box p) {
    Point sz = p.size();
    return sz.x <= max_ &&  sz.y <= max_;
  }
}
</pre>

The Screener itself is then just:

<pre>
publi class Screener extends DualOutputPushStage implements PushStage {
  BoxPredicate pred_;
  public Screener(BoxPredicate pred) { super(); pred_ = pred; }
  public synchronized void putA(Box p) {
    if (pred_.predicate(p)) 
      (new Thread(new PutARunner(next1_, p))).start();
    else 
      next2_.putA(p);
  }
}
</pre>

<h3>Sources</h3>

To make the results slightly more interesting, we'll seed
assembly lines with sources that randomly choose initial sizes:

<pre>
public class BasicBoxSource extends SingleOutputPushStage implements PushSource {
  protected Point size_;
  public BasicBoxSource(Point size) { size_ = size;  }
  protected synchronized Box produce() {
    return new BasicBox((int)(Math.random() * size_.x) + 1, 
                         (int)(Math.random() * size_.y) + 1);
  }
  public synchronized void start() { next1_.putA(produce()); }
}

</pre>

And to make them more fun to watch, an <a href="auton.html" tppabs="http://www.foi.hr/~dpavlin/java/mirrors/g.oswego.edu/dl/pats/auton.html"> autonomous loop</a>
version that can continually feed an assembly line:

<pre>
class AutonomousBasicBoxSource extends BasicBoxSource implements Runnable {
  protected int productionTime_;
  public AutonomousBasicBoxSource(Point size, int productionTime) {
    super(size);
    productionTime_ = productionTime;
  }
  
  public void run() {
    for (;;) {
      try {
        Thread.currentThread().sleep((int)(Math.random() * 2*productionTime_));
      }
      catch (InterruptedException ex) { return; }
      start();
    }
  }
}
</pre>

<h2>
Coordination
</h2>

Without having a scripting tool for these classes available, we have
to program Mondrian Assembly lines by manually creating instances of
desired stages, and linking them together. This is easy in principle,
but tedious and error-prone in practice because of the lack of
visual guidance about what stages are connection to what.

<p>
Here's one example written as an Applet, where we
treat the Applet itself as the sink of the entire flow. The
resulting Applet is the one running <a href="#secMondrian"> above</a>.

<pre>
public class AssemblyLine extends Applet implements PushStage {

  Box current_;  // the box currently being displayed

  // This assembly line has four sources:
  AutonomousBasicBoxSource source1;
  AutonomousBasicBoxSource source2;
  AutonomousBasicBoxSource source3;
  AutonomousBasicBoxSource source4;

  public AssemblyLine() {
    current_ = null;

    source1 = new AutonomousBasicBoxSource(new Point(50, 30), 1000);
    Painter painter1 = new Painter(Color.red);
    source1.attach1(painter1); 

    Alternator alternator1 = new Alternator();
    painter1.attach1(alternator1);

    HorizontalJoiner horizontalJoiner1 = new HorizontalJoiner();
    DualInputAdaptor adaptor1 = new DualInputAdaptor(horizontalJoiner1);

    HorizontalJoiner horizontalJoiner2 = new HorizontalJoiner();
    DualInputAdaptor adaptor2 = new DualInputAdaptor(horizontalJoiner2);

    alternator1.attach1(horizontalJoiner1);
    alternator1.attach2(horizontalJoiner2);

    source2 = new AutonomousBasicBoxSource(new Point(80, 60), 1000);
    Painter painter2 = new Painter(Color.green);
    source2.attach1(painter2);
    painter2.attach1(adaptor1);

    Merger merger1 = new Merger();
    DualInputAdaptor adaptor5 = new DualInputAdaptor(merger1);
    horizontalJoiner1.attach1(merger1);
    horizontalJoiner2.attach1(adaptor5);

    Wrapper wrapper1 = new Wrapper(6);
    merger1.attach1(wrapper1);

    Painter painter4 = new Painter(Color.orange);
    wrapper1.attach1(painter4);

    VerticalJoiner verticalJoiner1 = new VerticalJoiner();
    DualInputAdaptor adaptor3 = new DualInputAdaptor(verticalJoiner1);
    painter4.attach1(verticalJoiner1);

    Wrapper wrapper2 = new Wrapper(4);
    verticalJoiner1.attach1(wrapper2);

    Painter painter5 = new Painter(Color.yellow);
    wrapper2.attach1(painter5);

    Cloner cloner1 = new Cloner();
    painter5.attach1(cloner1);

    source3 = new AutonomousBasicBoxSource(new Point(70, 120), 1000);
    Painter painter3 = new Painter(Color.blue);
    source3.attach1(painter3);
    painter3.attach1(adaptor3);

    source4 = new AutonomousBasicBoxSource(new Point(40, 80), 1000);
    Painter painter6 = new Painter(Color.black);
    source4.attach1(painter6);
    painter6.attach1(adaptor2);


    HorizontalJoiner horizontalJoiner3 = new HorizontalJoiner();
    DualInputAdaptor adaptor4 = new DualInputAdaptor(horizontalJoiner3);
    cloner1.attach1(horizontalJoiner3);
    cloner1.attach2(adaptor4);

    Cloner cloner2 = new Cloner();
    horizontalJoiner3.attach1(cloner2);
    
    VerticalJoiner verticalJoiner2 = new VerticalJoiner();
    DualInputAdaptor adaptor6 = new DualInputAdaptor(verticalJoiner2);

    cloner2.attach1(verticalJoiner2);
    cloner2.attach2(adaptor6);

    Screener sizeScreener1 = new Screener(new MaxSizePredicate(200));
    DevNull discards = new DevNull();

    verticalJoiner2.attach1(sizeScreener1);
    sizeScreener1.attach1(this);
    sizeScreener1.attach2(discards);
    
  }

  public void start() {
    Thread t1 = new Thread(source1);
    Thread t2 = new Thread(source2);
    Thread t3 = new Thread(source3);
    Thread t4 = new Thread(source4);

    t1.setPriority(Thread.MIN_PRIORITY + 1);
    t2.setPriority(Thread.MIN_PRIORITY + 1);
    t3.setPriority(Thread.MIN_PRIORITY + 1);
    t4.setPriority(Thread.MIN_PRIORITY + 1);

    t1.start();
    t2.start();
    t3.start();
    t4.start();
  }

  public synchronized void putA(Box p) {
    // hold up until the previous display is finished
    while (current_ != null) {
      try { wait(); } catch (InterruptedException ex) { return; }
    }
    current_ = p;
    repaint();
  }

  public synchronized void paint(Graphics g) {
    if (current_ != null) {
      current_.show(g, new Point(0, 0));
      current_ = null;
      notifyAll();
    }
  }
  
}

</pre>

<p>

<p><a href="aopintro.html" tppabs="http://www.foi.hr/~dpavlin/java/mirrors/g.oswego.edu/dl/pats/aopintro.html">[Concurrent Programming in Java]</a>

<hr>
<address><A HREF="javascript:if(confirm('http://g.oswego.edu/dl  \n\nThis file was not retrieved by Teleport Pro, because it is addressed on a domain or path outside the boundaries set for its Starting Address.  \n\nDo you want to open it from the server?'))window.location='http://g.oswego.edu/dl'" tppabs="http://g.oswego.edu/dl">Doug Lea</A></address>
<!-- hhmts start -->
Last modified: Wed Feb 21 07:55:03 EST 1996
<!-- hhmts end -->
</body> </html>