6.1. The Replicated-Worker Pattern

We are going to start with one of the most common application patterns, the replicated-worker pattern—a simple, yet powerful, application framework that is sometimes referred to as the “master-worker pattern.” Recall from our quick peek at the pattern in Chapter 3 that it's well suited to solving a computational problem that can be decomposed into a number of smaller, independent tasks, all of which turn out to be nearly identical. The replicated-worker pattern is often used for parallel computing, in which multiple machines take on the tasks to solve the problem faster.

Typically the replicated-worker pattern involves one master process along with any number of workers. The master takes a problem, divides it up into smaller tasks, and deposits them into a space. The workers spend their lives waiting for tasks, removing them from the space, computing them, and writing results back into the space. The master's job is to collect the task results and combine them into a meaningful overall solution.

Let's look at a simple pseudocode skeleton that applies to most replicated-worker applications. Here is the pseudocode for the master process:

public class Master {
  for (int i = 0; i < totalTasks; i++) {
     Task task = new Task(...);
     space.write(task, ...);
  }

  for (int i = 0; i < totalTasks; i++) {
     Result template = new Result(...);
     Result result = (Result)space.take(template, ...);
     //... combine results
  }
}

Here, the master iterates through all the tasks that need to be computed, creating a task entry for each and writing it into the space. The master then iterates again, this time removing as many result entries from the space as there were tasks and combining them into some meaningful form.

Now let's look at the pseudocode for each worker process:

public class Worker {
  for(;;) {
     Task template = new Task(...);
     Task task = (Task)space.take(template, ...);
     Result result = compute(task);
     space.write(result, ...);
  }
}

A worker repeatedly removes a task from the space, computes it, and writes the result of the computation back into the space, where it will be picked up by the master process. There are many subtle improvements we can make to this code (and we will do so in Chapter 11), but this basic structure represents almost all replicated-worker applications.

It's worth pointing out a couple important characteristics of the replicated-worker pattern. First, each worker process may compute many tasks, one after another. As soon as one task is complete, a worker can look for another task in the space and compute it. In this way, replicated-worker patterns naturally load balance: workers stay busy and compute tasks in relation to their availability and ability to do the work. While one worker is occupied with a big task, for instance, another worker might be picking up and completing several smaller tasks. Second, the class of applications that fit into the replicated-worker pattern scales naturally—we can add more workers running on more machines without rewriting our code, and the computation generally speeds up. In general, the more machines, the faster we can compute.

As an example of using the replicated-worker pattern, suppose that you want to ray trace an image (a popular technique for producing realistically rendered scenes). Typically you'd start with a model of the scene, and an image plane in front of the model that is divided into pixels. Rendering an image involves iterating through all the pixels in the plane and computing a color value for each. The task of computing the color value of each pixel involves tracing the rays of light that pass from a viewpoint (such as your eye) through the pixel in the image plane, and to the model. The computation is identical for each pixel, except that the parameters describing the pixel's position differ. Image ray tracing is a perfect candidate for the replicated-worker pattern, since it is made of up a number of independent tasks that are computationally identical.

In the case of ray tracing, the master process might generate tasks that describe collections of pixels to compute (for instance, scan lines of pixels running across an image). Each worker computes a few scan lines (a possibly lengthy process) and returns a result. The master collects the results and combines them to form an image. To produce computer animations, the master might iterate through lots of images and combine the results into an animation sequence.

6.1.1. Computing the Mandelbrot Set

In this section we are going to look at a concrete example that is in the same spirit as ray tracing but a little less complex (and still a lot of fun): computing a Mandelbrot set. Iterating through a particular formula results in values that can be mapped to colors and used to draw an image of the Mandelbrot set, as shown in Figure 6.1. It isn't necessary that you understand the mathematical details of computing a Mandelbrot image; the only thing you need to know is that given an (x, y) coordinate of the screen image, we can compute an integer value for that pixel and map it to a color. (See the sidebar below on “How to compute the Mandelbrot set” if you are interested in the mathematical details.)

As with ray tracing, when we compute and draw a Mandelbrot image, each pixel can be computed independently of all the rest. Unlike ray tracing, however, each pixel's color value can be computed fairly quickly, because the computation isn't very complex. In fact, because of the overhead of communicating task and result entries to and from the space, a replicated-worker version of a Mandelbrot computation may actually compute more slowly than a nonparallel version (refer to the sidebar below on “The Computation/Communication Ratio” for further explanation). But our aim here in building a Mandelbrot viewer isn't to build a fast parallel application, but to gain experience with the replicated-worker framework (and Mandelbrot computations are simpler than ray tracing).

In building a replicated-worker version of a Mandelbrot set viewer, we will create a master process that iterates through all the scan lines that need to be computed and writes out tasks (each representing multiple scan lines). The master then collects the results (in whatever order it finds them) and draws the results to the screen. We'll also create worker processes, each of which repeatedly removes a task from the space, computes the color values for the scan lines represented by the task, and writes the result back to the space for the master to pick up. Let's start building our Mandelbrot viewer by first defining entries for tasks and results.

6.1.2. Task and Result Entries

As we mentioned in Chapter 3, we often use a task bag and a result bag in replicated-worker computations. The task bag holds all the tasks a master process needs to have computed, while the result bag holds the results as they are completed. In most cases, bags are all we need to hold tasks and results. In cases where an ordering needs to be imposed on tasks, then we'll use a task queue, as we described in Section 5.6. For our Mandelbrot viewer, we don't care about the order in which the tasks are computed, so bags will suffice. The master picks up and draws whichever results it finds, in no particular order, until the entire image has been rendered.

First let's define a task entry that can be placed into the task bag. A task entry holds the coordinates of the region (representing a set of complex numbers) for which the application is collectively computing the Mandelbrot set, and also specifies the particular “slice” of that region that a worker must compute:

public class Task implements Entry {
  public Long jobId;      // id for Mandelbrot computation

  // region for which application is computing Mandelbrot
  public Double x1;       // starting x
  public Double y1;       // starting y
  public Double x2;       // ending x
  public Double y2;       // ending y

  // slice of the region this task computes
  public Integer start;   // starting scan line
  public Integer width;   // width of scan line in image
  public Integer height;  // number of scan lines in image
  public Integer lines;   // number of scan lines per task

  public Task() {
  }
}

You'll notice that we have a field called jobId, which provides a unique name for this particular Mandelbrot computation in the space. We'll assign the current date and time to this field in the form of a long integer value, which for our purposes should be unique. Next the task entry has x1, y1, x2, and y2 fields, specifying the region of complex numbers for which the Mandelbrot set is being collectively computed.

Next, the task entry has fields that are used to communicate to the worker how to compute a slice of that region. The start field indicates the first scan line to compute, width specifies how many pixels wide the scan lines are, height holds the number of scan lines in the overall region being computed by the application, and lines specifies the number of scan lines to compute.

Now let's look at the result entry, which holds the result of computing a task described by a task entry:

public class Result implements Entry {
  public Long jobId;      // id for Mandelbrot computation
  public Integer start;   // starting scan line of results
  public byte[][] points; // pixel values as byte integers

  public Result() {
  }

  public Result(Long jobId, Integer start, byte[][] points) {
    this.jobId = jobId;
    this.start = start;
    this.points = points;
  }
}

Each result contains a jobId field, which is the same as the jobId supplied in the task entry, a start field that indicates the starting scan line for the pixels in this result, and a points field that holds a set of pixel values.

Now that we've seen the task and result entries that will populate the task and result bags, let's take a look at the code for the master process.

6.1.3. The Master

Our master process is responsible for breaking the Mandelbrot computation into tasks, collecting the results, and also displaying the Mandelbrot image. In addition, the master process allows the user to interact with the image. A user can “zoom into” an area of the image by clicking on it and dragging a small rectangle. When the user releases the mouse button, the region bounded by the rectangle becomes the basis for a new Mandelbrot computation, and a new set of tasks is generated to be computed by the workers.

Minus some of the interface and graphics code, here is the skeleton of the Master class:

public class Master extends Applet
  implements Runnable, MouseListener, MouseMotionListener
{
  private JavaSpace space;
  private int xsize;      // dimensions of window
  private int ysize;
  private Long jobId;     // id for this computation
  private int tasks;      // number of tasks to generate
  private int lines = 0;  // # of scan lines per task
  private Thread master;

  // initial region for which Mandelbrot is being computed
  private double x1 = -2.25;
  private double x2 =  3.0;
  private double y1 = -1.8;
  private double y2 =  3.3;

  private boolean done = false;   // computation finished?
  private int progress;           // number of scan lines
  //... other variables go here

  public void init() {
    space = SpaceAccessor.getSpace();
    xsize = getSize().width;
    ysize = getSize().height;

    String taskStr = getParameter("tasks");
    if (taskStr == null) {
       tasks = 20;
    } else {
      tasks = Integer.parseInt(taskStr);
    }
    lines = ysize / tasks;  // scan lines per task

    //... code for offscreen graphics buffer goes here

    // set up listeners
    this.addMouseListener(this);
    this.addMouseMotionListener(this);

    // spawn thread to handle computations
    if (master == null) {
       master = new Thread(this);
       master.start();
    }
  }
  public void run() {
    generateTasks();
    collectResults();
  }

  //... generateTasks method goes here

  //... collectResults method goes here

  //... user interface and graphics methods go here
}

Note the x1, y1, x2, and y2 coordinates initialized in the variable declarations section; these values specify the starting region of the complex numbers for which the Mandelbrot set will be computed. In the init method, we perform setup for the Mandelbrot computation: We obtain a reference to a space object, the dimensions of the applet window in pixels, and the number of tasks to break the image into (obtained from the applet parameter tasks, if one exists, otherwise set to twenty). Given the image height and number of tasks, we compute the number of scan lines to be computed per task. Then we register the applet to be a listener for mouse events, which will occur when the user zooms into a region of the Mandelbrot image.

Finally, the master spawns a new thread that generates tasks and then collects results, as you see in the run method. Of course, some replicated-worker applications may have more advanced designs—such as allowing the task generation and the result collection to run concurrently, or performing multiple iterations of a computation. However, in this example we are trying to demonstrate a simple pattern.

Let's take a look at the details of the generateTasks method:

private void generateTasks() {
   Task task = new Task();

   long millis = System.currentTimeMillis();
   jobId = new Long(millis);
   task.jobId = jobId;

   task.x1 = new Double(x1);
   task.x2 = new Double(x2);
   task.y1 = new Double(y1);
   task.y2 = new Double(y2);
   task.width = new Integer(xsize);
   task.height = new Integer(ysize);
   task.lines = new Integer(lines);

   for (int i = 0; i < ysize; i += lines) {
     task.start = new Integer(i);
     try {
        System.out.println("Master writing task " +
          task.start + " for job " + task.jobId);
        space.write(task, null, Lease.FOREVER);
     } catch (Exception e) {
        e.printStackTrace();
     }
  }
}

This method creates a new Task entry and fills in its jobId field (the identifier for the overall Mandelbrot computation) with the current time in milliseconds. We then fill in the values that are identical for every task we will generate: the coordinates of the region the application is computing, the width and height of the entire image, and the number of scan lines per task. Finally, to generate tasks, we iterate through the scan lines in the image. Each task's start field is assigned the starting line for the slice of the image it will compute, and the task is written into the space.

When generateTasks is finished, the run method calls collectResults, which is defined as follows:

private void collectResults() {
   while (true) {
    while (done) {
     try {
         System.out.println(
           "Master done; waiting for new job.");
         Thread.sleep(500);
     } catch (InterruptedException e) {
        e.printStackTrace();
     }
    }

    Result template = new Result();
    template.jobId = jobId;
    Result result = null;
    try {
        result = (Result)
          space.take(template, null, Long.MAX_VALUE);
        System.out.println("Master collected result " +
          result.start + " for job " + result.jobId);
        display(result);
        progress = progress + lines;
    } catch (Exception e) {
       e.printStackTrace();
    }

    if (progress == ysize) {
       done = true;
       repaint();
    }
   }
}

The collectResults method consists of a while loop: each time through the loop we check to see if the current computation is done (signified by the done variable being equal to true). If it is done, then we loop until done becomes false (which happens when the user zooms into a new region), sleeping for half a second (500 milliseconds) on each iteration.

If the computation isn't done, then we must collect task results. To do this, we first create a Result template, leaving all fields as wildcards except for jobId. This template allows us to retrieve any results computed for the Mandelbrot, regardless of the order in which the tasks were generated or computed. Once we have a template, we take a result from the space and pass it to the display method, which draws the scan lines contained in it. We also update the progress variable; when progress is equal to the number of scan lines we are computing, we know that all results have been retrieved, and we set the done variable to true.

The generateTasks and collectResults methods cover the major points of how the master process works (leaving aside the details of the graphics display). The only other aspect we need to discuss is what happens when the user zooms into a finished Mandelbrot image. When the user clicks, drags, and releases the mouse, the mouseReleased method is called:

public void mouseReleased(MouseEvent e) {
  int x = e.getX();
  int y = e.getY();
  if (done) {
     x1 += ((float)x / (float)xsize) * x2;
     y1 += ((float)y / (float)ysize) * y2;
     x2 /= 2.0;
     y2 /= 2.0;
     x1 -= x2 / 2.0;
     y1 -= y2 / 2.0;
     done = false;
     drag = false;
     offg.setColor(Color.black);
     offg.fillRect(0, 0, xsize, ysize);
     progress = 0;
     repaint();
     generateTasks();
  }
}

Note that the mouseReleased method does nothing if done is false—the user can only start exploring a new region once the current Mandelbrot image has been fully computed. This method determines the new area of the image the user wants to explore, and it resets the master's x1, y1, x2, and y2 variables to that region. It's also important to point out that the method sets done to false (to indicate that a new Mandelbrot computation is underway) and progress for the new computation to zero. It then calls generateTasks, which creates a new set of task entries for the new Mandelbrot computation.

While generateTasks occurs in the master's main thread as a result of the mouseReleased event, our collectResults method is still running in the thread spawned by the master and will wake up within half a second to begin looking for new results.

6.1.4. The Worker

Now that we've seen how the master process generates tasks and collects results, let's look at the code for the worker processes that compute the tasks:

public class Worker {
  public static void main(String[] args) {
    JavaSpace space = SpaceAccessor.getSpace();

    Task template = new Task();
    Task task;

    for (;;) {
      byte[][] points;
    try {
       task = (Task)
         space.take(template, null, Long.MAX_VALUE);
       System.out.println("Worker got task "
         + task.start +
         " for job " + task.jobId);

       points = calculateMandelbrot(task);
       Result result = new Result(task.jobId,
         task.start, points);

       System.out.println(
         "Worker writing result for task " +
         result.start + " for job " + result.jobId);
       space.write(result, null, Lease.FOREVER);
    } catch (Exception e) {
       e.printStackTrace();
    }
   }
  }

  //... calculateMandelbrot method goes here
}

When we run a worker application, its main method is executed. In main, the worker obtains a handle to a space and then creates a Task template, leaving all fields as wildcards so that it can match any task in the bag. Then main enters a for loop that continually looks for tasks. Using the template, we take a task from the space and pass it to the calculateMandelbrot method, which calculates the pixel color values for the scan lines described in the task and returns them as a byte array. The calculateMandelbrot and zooming code were originally written by Robin Edwards.)

We construct a new Result entry that holds the task's job identifier, the starting scan line that was computed, and the array of bytes, and then write the result into the space. At that point, the worker begins the loop anew and looks for another task to compute.

At this point you should experiment with the Mandelbrot viewer. Start up any number of workers and then start the master process (of course, given the loosely coupled nature of our application, you could reverse the ordering). Watch the Mandelbrot image fill in, click to zoom into a region of interest, and watch as the workers compute the new area. Add or remove workers as you wish.