• Purposes
• increase availability of resources
• speed up finding a solution or improve performance

• Replicating Workers
• all workers receive task from a common workpool (bag)
• the workpool is represented by a channel
• all workers must be able to access this channel
• we parallelize a "divide-and-conquer" problem.

• Example: Adaptive Quadrature
• integration of a function using trapezoid rule
• a wildly varying function means using a fixed interval is inaccurate
• varying the width of the interval allows a better approximation
• in the example below, a width of W is good enough for the left-hand portion of the integration, but W/2 is better for the right-hand
• we will dynamically decide to reduce the interval when approximation is not close enough. At each interval, we calculate two subintervals and compare the sum of these to the parent value. If it is accurate enough we are done. If not, we will subdivide again.

• Implementation
• One channel will hold the workpool. Initially, an administrative process will put the interval [A,B] into the workpool. All server processes will initially wait for work in the workpool. When a server process gets work (an interval with its endpoints and estimate of its area), it calculates two subinterval trapezoids. If the sum of these two areas is within a small threshold of the overall area, these two areas are sent to the administrative process on another (result) channel. Otherwise, the server sends both intervals to the workpool.

Admin:: var l,r,fl,fr,a,b,area, total: real
//and other variables to record finished intervals

  fl := f(l); fr := f(r);
  area := (fl + fr) * (l+r)/2;

//put both left and right intervals into workpool

  send workpool(l,r,fl,fr,area);
  do //entire area not yet completed
    {receive result(a,b,area);
    total := total + area;
//record that the area between a and b has been calculated}
  od

worker[1:n] :: var a,b,m,fa,fb,fm:real;
var larea, rarea, tarea, diff: real;
//get adjacent intervals from the workpool

do true -> {receive workpool(a,b,fa,fb, tarea);
  m:=(a+b)/2; fm := f(m);

//compute difference between the two subintervals
// and the total area using trapezoids
  diff := tarea -(larea + rarea);
  if diff small then send result(a,b,tarea);
//if difference not small enough, put both left and right
//intervals back into pool
  else diff large then {send workpool(a,m,fa,fm,larea);
  send workpool(m,b,fm,fb,rarea);}
  fi
od
 
 

• Contention for access to the centralized workpool
• Decentralization of resources reduces contention
• Implement the workpool as multiple channels
• Balance the load among worker processes
• Creates problems in termination conditions
• Hide the difficulties within the workpool operations "getwork" and "putwork"
• A set of processes is assigned to each channel from which it can get work
• To distribute the load, each process puts work in the channels in a "round robin" fashion. It keep a channel pointer which is incremented each time it puts something in a channel (pointed to by channel pointer).

 

 
 
 
 
 
 
 
 
 
 
 
 
 

The termination condition now means that the workpool is empty only when all the channels are empty. All workers must also be idle, or some process which is still active could "putwork" into the workpool, even though all other processes are idle.
• Keep a count on how many workers in a group are idle. We must also keep a "mastercount" of all work groups which are idle. When all groups are idle, then the program can terminate. "mastercount" must be atomically updated.
• To keep the contention for "mastercount" low, we only update it when the workers become idle or active. That is, we keep a "count" for each workgroup and when it goes to "0" or is incremented from "0"; we only access "mastercount" on these two conditions.

 

Getwork(me, item)
Compute my channel number in the workpool;
Decrement counter for the channel;
if counter = -worker_group_size then
  begin (*my worker group is now idle*)
    increment mastercount;
    if mastercount = number_of_worker_groups then
      {send a termination flag to each worker process}
    for i:= 1 to number_of_worker_groups do
      for j:= 1 to worker_group_size do
        {put a termination flag into channel of worker group i}
  end;

read a work item from my channel into "item";
Putwork(me, item);
move my pointer to the next channel in workpool;
increment counter for the target channel;
if counter = - worker_group_size +1 then
  decrement mastercount; (*idle worker group now active*)
write "item" into the target channel;
 
 

Workpool in Java

// Workpool class

public Workpool(int a, int b, int[] c, int[] d, int[] e, Channel[] f)
{worker_group_size = a; num_worker_groups = b;
  nextchan = c; count = d;
  mastercount = e; channel = f;
}

// get an item from the workpool

public int getwork(int me) {
  int workcount, emptycount, mychan;
  int value;
  mychan = me/worker_group_size;
//decrease the # of workers in this group
  synchronized(count) {
  workcount = count[mychan]-1;
  count[mychan] = workcount;
  }
if(workcount == -worker_group_size) {//all workers in this group are idle
  synchronized(mastercount) {
  emptycount = mastercount + 1;
  mastercount = emptycount;
  }
// if all workers are done and the channels are empty, send termination message
if(emptycount == num_worker_groups)
  for(int w=0; w<num_worker_groups; w++)
    for(int j=0; j<worker_group_size; j++)
      channel[w].put(-1);
}
value = channel[mychan].get();
return(value);
}
 

// put an item in the workpool

public void putwork(int me, int item) {
  int workcount, emptycount, next;
  next = nextchan[me];
//increase the # of workers in this pool
  synchronized(count) {
  workcount = count[next]+1;
  count[next] = workcount;
  }
  if(workcount == -worker_group_size + 1) {//make pool active again
    synchronized(mastercount) {
      emptycount = mastercount - 1;
      mastercount = emptycount;
    }
  }
  channel[next].put(item);
// increment my "next" pointer to point to the next channel in "round robin" order
  nextchan[me] = (next + 1) % num_worker_groups;
}
 
 

• In a directed, weighted graph we want to find the shortest path between a given source vertex and every other vertex in the graph.

 

 
 
 
 
 
 
 
 
 
 
 

// This is a Java program which implements the shortest path algorithm using a
//replicated workers paradigm and multiple (shared memory) channels to reduce
//contention. The goal is to establish the shortest path from a node
//to every other node in a directed graph. The graph is represented as an adjacency
//matrix whose cell contents are the distance to each neighbor. Initially, the initial
//node of the graph is placed in a workpool. Each thread (worker) gets a work item
//from the workpool which represents a node x. It then calculates the distance to all
//of x's neighbors, using the procedure:
//for w=1..#neighbors do mindist[x] = min(mindist[x], newdist[x,w]).
//Since this is a shared memory implementation, all threads have access to shared
//variables, including an array "mindist", where mindist[i,j] represents the current
//minimum distance from node i to node j, the array "weight" which records the
//distance between neighboring nodes (this is the adjacency matrix), and an integer
//"mastercount" which keeps track of the number of currently active threads. In this
//program, the workpool is implemented using multiple channels to reduce
//contention. Multiple workers (worker_group_size) are assigned to a channel.
static final int num_worker_groups = 2;
static final int worker_group_size = 2;
static final int numworkers = num_worker_groups * worker_group_size;
static final int N = 5;
static int Infinity = 32000;
static int startvertex = 0;

public static void main(String[] args) {
  int mastercount = 0;
  int weight[][] = new int[N][N];
  int mindist[] = new int[N];
  boolean inflag[] = new boolean[N];
  int count[] = new int[num_worker_groups];
  int nextchan[] = new int[numworkers];
  Channel[] channel = new Channel[num_worker_groups];
  Worker[] worker_ = new Worker[numworkers];
  Workpool workpool = new Workpool(worker_group_size,num_worker_groups,nextchan,count,mastercount,channel);
// set initial values
  for(int j=0; j<N; j++) {
    mindist[j] = infinity;
    inflag[j] = false;
    for(int k=0; k<N; k++)
      weight[j][k] = Infinity;
  }
//create a channel per worker_group
  for(int j=0; j<num_worker_groups; j++)
    channel[j] = new Channel();
// put (initial) vertex number 0 in the queue
  mindist[startvertex] = 0;
  inflag[startvertex] = true;
  channel[0].put(startvertex);
  count[0]=1;
  for(int j=1; j<num_worker_groups; j++)
  count[j] = 0;
  mastercount = 0;
// initial weight values
  weight[0][1] = 4;
  weight[0][2] = 8;
  weight[1][2] = 3;
  weight[1][3] = 1;
  weight[2][4] = 5;
  weight[3][2] = 2;
  weight[3][4] = 10;
// Creating thread (workers)
  for (int i=0; i < worker_.length; i++) {
    worker_[i] =
      new Worker(i,weight,mindist,inflag,Infinity,nextchan,worker_group_size,N,workpool);
    worker_[i].start();
    System.out.println(i+" thread properly started");
  }
// wait for the thread end
  try {
    for (int i=0; i < integrator.length; i++) {
      worker_[i].join();
      System.out.println(i+" thread properly terminated");
    }
  } catch (InterruptedException e) {
    System.out.println("Interrupted");
}
// print the result
  for(int j=0; j<N; j++)
    System.out.println(mindist[j]);
}

// thread to compute the shortest path

class Worker extends Thread {
  int vertex, newdist,i,infinity,n,worker_group_size;
  int weight[][];
  int mindist[];
  int nextchan[];
  boolean inflag[], change_flag=false;
  Workpool workpool;

public Worker(int a,int[][] b,int[] c,boolean[] d,int e,int[] g,int l,int o,Workpool w) {
  i = a;
  weight = b;
  mindist = c;
  inflag = d;
  infinity = e;
  nextchan = g;
  worker_group_size = l;
  n = 0;
  workpool = w;
}
// shortest path algorithm worker thread
public void run ( ) {
//multiple workers taking from 1 channel
  nextchan[i] = i / worker_group_size;
// get a new vertex number to examine
  vertex = workpool.getwork(i);
  while ( vertex != -1) {
// vertex is removed from Work Pool
    inflag[vertex] = false;
// consider all outgoing edges of "vertex"
    for(int w=0;w<n;w++) {
      if(weight[vertex][w]<infinity) { //is this a neighbor?
// see if this is a shorter path to w
        newdist = mindist[vertex] + weight[vertex][w];
        synchronized(mindist) {
// mindist is shared and must be updated atomically
          if(newdist < mindist[w]) {
            mindist[w] = newdist;
            change_flag = true;
          }
        }
        if(change_flag) {
//if the value mindist of "w" changed, put it back in the pool
         change_flag = false;
         if(!inflag[w]) {
            inflag[w] = true;
            workpool.putwork(i,w);
        }
      }
    }
  } //End for all neighbors
  vertex = workpool.getwork(i);
} //End of main while statement
} //End of run
} // End of Worker thread
 
 

• Channels can be implemented in shared memory or across a network. Processes still "put" things in one end and "get" things out the other end.
• However, even if the nature of a channel (shared memory or distributed) is transparent to the processes, two problems must be addressed. First, termination detection must now be a distributed algorithm; and second, we must "map" the problem onto the distributed processors so that performance is acceptable.
• Good performance in mapping means that those data items which show some locality (referenced by the same processor close together in time) should reside on the same processor. For example, in the shortest path algorithm, we kept a "mindist" array which was a shared data item. But this would be inefficient to access a shared data item across multiple machines. So, we map all the information about a node in the graph into one process, so that access to mindist(i) for process i is local.
• Termination detection in shared memory is typically handled by a shared "count" (or multiple counts and a master count). Again, this would be inefficient in a distributed environment. So, for the shortest path algorithm we will investigate an acknowledgement protocol to detect a global termination condition.
• Shortest Path Algorithm in a Distributed Environment
• multiple channel distributed workpool
• items in the workpool consist of a node index i and a computed "trial" mindist[i]
• each process i will maintain mindist[i] (for node i) and have its own channel
• when trial distances to node i are computed by a process, this trialdistance is sent to process i through channel i in the workpool
• when process i receives from channel i, it checks to see if the trial mindist[i] in the item from the workpool is less than its current mindist; if so, it is made the new mindist[i]; else, it is discarded. Then process i calculates a new trialdistance to all neighbor nodes and passes those trialdistances to the appropriate processes.

• Termination Conditions
• the workpool is completely empty
• no work items are in transit
• all workers are waiting for work

• How do we guarantee these conditions in a distributed environment
• no shared memory to store entire state
• must exchange messages to establish global state

• A Termination Detection Algorithm
• each work item that is placed in the workpool must be acknowledged; this way, the sender knows some process handled the work item and is "idle"
• this builds an active worker (dependency) tree and any processes below must acknowledge the parent before the parent can go "idle"; that is, we will build a parent/worker hierarchy of workers; each time a worker puts an item in the workpool, it waits for an acknowledgement for all work items it has put in the workpool before it can go "idle"; the sending process becomes the parent and the process (which was idle and) which just received a message (and went active) becomes the worker; when the worker finishes this work item and goes "idle", it sends an acknowledgment to its parent; thus, we build a hierarchy of processes in the "idle" state; when initiator process (root node of the tree) receives an acknowledgement, then all processes below it in the tree are "idle" and the program is terminated
• if any process is active, then its parent is active and the program cannot terminate.

• Worker state diagram

 

• active worker tree
• W4, W7, and W18 must go "idle and acknowledge W3 before W3 goes "idle" and acknowledges W2, etc.
• all other workers are "idle" and have no relationship to one another

- all acknowledgements handled by the workpool methods

- Putwork Method

*Putwork(vertex, outputdistance)

Increment ackcount;
Build work item consisting of (me, outputdistance);
Write work item into workpool[vertex];

*Getwork Method

If (worker in active state, and work channel empty
    and all acknowledgments received) then

Begin
  Acknowledge current parent;
  Go into Idle State;
End
Repeat
Read new item from Workpool;
If (new item is acknowledgement) then decrement ack.count;
Until work item is received;

If (worker in active state) then
  Send acknowledgment to source of work item
Else begin
  Go into active state
  Set new parent from source of this work item;
End;
Return "distance" part of work item to calling worker;
 
 
 
 
 

Implementing the Worker
mindistance:= infinity;

Getwork(trialdistance)
while (not done)
  begin
  if trialdistance < mindistance then
    mindistance := trialdistance;
 for all my outgoing neighbors
   Putwork (neighbor vertex, mindistance + distance to neighbor)
Getwork(trialdistance);
end
answer := mindistance