Lecture Notes for Concurrent Programming

31 July 2002 - KWIC Indexing, Pipe-and-Filter Style

KWIC Indexing

• KWIC Indexing

• Filter Templates

  • Abstract Filter Methods

• The Termination Interface

  • Pipe-and-Filter Termination

  • Pipe-and-Filter Termination Class

  • Using Pipe-and-Filter Termination

• Stopping Threads

• Putting It All Together

KWIC Indexing

• Create KWIC entries from a description.

  class KWICIndexEntry
  
    static KWICIndexEntry []
    entries(String description)
  
    private 
    KWICIndexEntry(String words[], int index_word)
  
    boolean 
    less(KWICIndexEntry kie)
  
    public String 
    toString()
  

  See the complete code.

• This is a (quasi) factory pattern.

• The return object is mutable.

  • Rely on the one-reference rule.

• No buffers, no threads.

Filters

• Each filter does the same thing.

  • Take an object from the input buffer.

  • Process the object

  • Put the object to the output buffer.

• This smells like a template pattern.

  public void 
  run()
  
    while true
  
      Object o = input_object()
      if (o == null) break
  
      if !isValidObject(o)
        throw new Error()
  
      o = process_input(o)
      output_object(o)
  
    end_loop()
  

The Filter Template

abstract class FilterRunnableTemplate
implements Runnable

  abstract protected boolean isValidObject(Object o)
  abstract protected void process_input(Object o)

  protected buffer inb, outb

  protected void end_loop()
    try outb.put(null)
    catch (InterruptedException ie) { }

  FilterRunnable(buffer ib, buffer ob)
    inb = ib
    outb = ob

  protected Object input_object()
    try return inb.get()
    catch (InterruptedException ie)
      throw new Error()

  protected void output_object(Object o)
    try outb.put(o)
    catch (InterruptedException ie)
      throw new Error()

  public void run()
    // As before.

See the complete code.

Example Abstract Filter Methods

• The rotate filter process-input method.

  protected void process_input(Object o)
    KWICIndexEntry entries[] = 
      KWICIndexEntry.entries((String) o)
    if entries != null
      try outb.put(o)
      catch (InterruptedException ie) return
  

  See the complete code.

• The sort filter end-loop method.

  protected void end_loop()
    try
      KWICIndexEntry db[] = 
        new KWICIndexEntry[descriptions.size()]
      descriptions.toArray(db)
      outb.put(db)
      outb.put(null)
    catch (Exception ie)
      throw new RuntimeException()
  

  See the complete code.

• The output filter output method.

  protected void
  output_object(Object o)
    KWICIndexEntry [] ieg = (KWICIndexEntry []) o
    for (int i = 0; i < ieg.length; i++)
      System.out.println(ieg[i])
  

  See the complete code.

The Termination Interface

• A termination mechanism.

• There are many different ways to terminate.

  • Abstract them in a termination interface.

  interface TerminationInterface 
    void normal(int id)
    void abnormal(int id)
  

• Each thread has a unique id.

• Each thread can terminate normally or abnormally.

• This is a general interface, not KWIC specific.

  • Not pipe-and-filter specific either.

Pipe-and-Filter Termination.

• n threads in sequence.

• Each thread can terminate normally or abnormally.

• Pipe-and-filter termination properties.

  • Normal termination proceeds from input to output.

  • Any filter may terminate abnormally.

  • A filter normally terminating out of sequence is an abnormal termination.

• The termination invariant.

  

The Pipe-and-Filter Termination Condition

• Keep a status array for thread termination.

  class PAFTermination 
  implements TerminationInterface
  
    final static int
      unterminated = 0
      normal       = 1
      abnormal     = 2
  
    final private int status[];
  

• The termination condition.

  int check()
  
    int i = 0
    while (i < status.length) && (status[i] == normal)
       i++
  
    if i == status.length
      return normal
  
    while (i < status.length) && (status[i] == unterminated) 
      i++
  
    return i == status.length ? unterminated : abnormal
  

Pipe-and-Filter Termination Class

• The parent and child methods.

  synchronized int disposition()
  
    // Wait until all threads have terminated normally, 
    // or a thread terminates abnormally.
  
    while true
      try wait() 
      catch (InterruptedException ie)
        return abnormal
      final int i = check()
      if (i != unterminated) return i
  
  
  synchronized void abnormal(int id)
  
    // Process id terminates abnormally.
  
    status[id] = abnormal
    notify()
  
  
  synchronized void normal(int id)
  
    // Process id terminates normally.
  
    status[id] = normal
    notify()
  

  See the complete code.

Using Pipe-and-Filter Termination

• The parent thread creates a termination instance ti.

• Each child thread gets reference to ti.

• Each child calls ti.normal() or ti.abnormal().

• The parent sits in ti.disposition().

  

Stopping Threads

• Termination is execution behavior, not computation behavior.

  • It is based on computation behavior, though.

  • The basis is abstracted by the termination interface.

• Don't embed termination in the filter runnable.

  • Runnables do computation, not execution.

• Threads do execution; extend threads

  class StoppingThread extends Thread
  
    private final TerminationInterface halter;
    private final int id;
  
    StoppingThread
      (int id, TerminationInterface halter, Runnable r)
      super(r)
      this.halter = halter
      this.id = id
  
    public void run()
      boolean normal = true
      try super.run()
      catch (Exception e)
        normal = false
      if normal halter.normal(id);
      else halter.abnormal(id);
  

Putting It All Together

class kwic {

  public static void main(String[] args)

    buffer b1, b2, b3

    StoppingThread threads[] = new StoppingThread[4]
    PAFTermination halter = new PAFTermination(4)

    InputFilter if = new InputFilter("input", b1)
    RotateFilter rf = new RotateFilter("rotate", b1, b2)
    SortFilter sf = new SortFilter("sort", b2, b3)
    OutputFilter of = new OutputFilter("output", b3)

    threads[0] = new StoppingThread(0, halter, if)
    threads[1] = new StoppingThread(1, halter, rf)
    threads[2] = new StoppingThread(2, halter, sf)
    threads[3] = new StoppingThread(3, halter, of)

    for (int i = 0; i < threads.length; i++)
      threads[i].start()

    if halter.disposition() != PAFTermination.normal
      System.out.println("Abnormal termination.")
      for (int i = 0; i < threads.length; i++)
	if threads[i].isAlive()
	  threads[i].interrupt()

Inventory

• What have we built?

• A KWIC indexer, but also:

  1. A KWIC index entry generator.

  2. A filter template.

  3. A termination interface.

  4. A pipe-and-filter terminator.

  5. A stopping thread.

• These are all independent of the original problem.

• This is a good start on concurrency toolkit.

Points to Remember

• Ruthlessly separate function.

  • Keep separate concepts separate: computation, coordination, termination.

• Use design to weave the functions together as appropriate.

  • Distributed computation, centralized termination.

• Overload threads for new execution behavior, not new computations.

  • The stopping thread, maybe the pipe-and-filter thread, but not the KWIC thread.

• Extract components and generalize them.

  • But don't over-generalize; listen to your code and the problem.

• Gather components into a concurrency toolkit.

  • It's a lot of effort; amortize it over a lot of use.

This page last modified on 9 August 2002.