Paradigms (Patterns) for Process Interaction in Distributed Programs

Andrews. ACM Computing Surveys, Vol. 23, No. 1, March, 1991

• There are four major types of processes in a distributed program.
• A filter is a data transformer; it receives streams of data values from its input channels, performs some transformation on the data, and puts streams of transformed data values into its output channels. Examples include Unix filters, parallel back substitution filters, and pipelined computer architecture.
• A client is a triggering process; a server is a reactive process. Clients make requests that trigger reactions from servers. A server continuously monitors its input channels for requests from its client, (typically) spins-off a new process to handle the request and goes back to monitoring its input channels. The parallel process then handles the construction of the response to the client's request. Examples include file servers, web servers, and dbms servers.
• A peer is one of a collection of identical processes that interact to provide a service or compute a result. They are programmed to communicate using a prescribed protocol, which they must both execute. Examples include data communications drivers, file transfer processes, and adaptive quadrature processes.
• Interaction Paradigms covered in this section
• one-way data flow through networks of filters
• requests and replies between clients and servers
• back-and-forth (heartbeat) interaction between neighboring processes
• probes and echoes in graphs
• broadcasts between processes in complete graphs
• token passing along edges in a graph
• coordination between decentralized server processes
• replicated workers sharing a bag of tasks
• coordinated peers in data communications protocols

Programming Notation

• A channel is an abstraction of a physical communication network.
• Degree of process synchronization
• synchronous message passing - caller waits till receiver takes message
• asynchronous - caller deposits message and proceeds (infinite buffer)
• buffered - caller waits only when buffer is full

• Asynchronous is used in these examples
• a channel is thus a queue of messages
• most flexible and what is typically provided in systems routines
• most natural for some applications

Filters: A Sorting Network

• Specification relates the input to the output.
• Example: sorting a list of n numbers into ascending order.
• In a Merge Network the goal is to merge repeatedly, and in parallel, two sorted lists into one sorted list.
• Each merge process receives from two ordered input streams and merges these into one ordered output stream.

• Java Code for Sorting Network

// Constructor for Sort Thread
Sort()
{
//Three channels must be passed to this thread constructor
//"in1" and "in2" are input channels
//"out" is output channel
}

public void run()
{
    v1 = in1.get();
    v2 = in2.get();
    while( (v1!=EOS) && (v2!=EOS) )
    {//find the smallest input value
    if(v1<=v2)
        {//output smallest value (from "in1") to output channel
out.put(v1);
        v1 = in1.get();
        }
    else
        {//output smallest value(from "in2") to output channel
out.put(v2);
        v2 = in2.get();
        }
    }

while( (v1!=EOS) && (v2==EOS) )
    {//channel "in2" is empty; keep processing items in "in1"
out.put(v1);
    v1 = in1.get();
    }

while( (v1==EOS) && (v2!=EOS) )
    {//channel "in1" is empty; keep processing items in "in2"
out.put(v2);
    v2 = in2.get();
    }
out.put(EOS);
}
}

Clients and Servers

• Centralized Servers: Active Monitors
• local variables record the state of the resource
• it services requests to access that resource

• Resource Allocation Monitor in Java
• Processes are controlled within the monitor
• Structured so that "signal" is the last thing in the monitor such that it can be programmed in a wide variety of monitors

// Resource allocation Monitor
class ResAllocMon{
int avail; int units;
    public ResAllocMon() {
        //if resources available, allocate them, else "wait"
        synchronized void acquire(int id) {
            while(avail==0) {
                try {
                    wait();
                } catch (InterruptedException e) {}
            }
            avail--;
            id = remove(units);
        }
        //Just release the resources and put them back into the pool
        synchronized void release(int id) {
            insert(id, units);
            if(avail==0)
                notify();
            avail++;
        }
    }

Resource Allocation Using a Central Process

• Simulating a monitor with a process
• Clients send messages and receive replies on a separate channel
• Each client has its own result channel
• Since the server cannot control client, the server must remember request and then service other requests.
• Duality of monitors and processes

monitor-based programs vs. message-based programs

permanent variables             local server variables

procedure identifiers             request channel and operation kinds

procedure call                     send request; receive reply

monitor entry                     receive request

procedure return                 send reply

wait statement                   save pending request

signal statement                 retrieve and process pending request

procedure bodies               arms of case statement on operation kind

• Most OSs provide monitor synchronization
• Most distributed systems are based on message-passing
• Java Code for message-based allocator
  
  

// Resource allocator and clients

class Allocator extends Thread{
int N = 10;
    Channel request; Channel reply[];
    //These two channels are used to accept requests from clients
    //and reply with "acquired resources"
int avail; int units; int[] pending; int[] index; int[] unitid;
    // Constructor of Allocator thread
    Allocator()
    {
    //initialize variables
    }

    public void run()
    {//This thread will run forever, or until system crashes or it is "stop"ed
        while(true)       {
            kind = request.get(index, unitid);
            if(kind==ACQUIRE) {
            //This is an ACQUIRE request from the client
                if(avail>0) {//request can be satisfied
                    avail--;
unitid[0] = remove(units);
                    reply[index[0]].put(unitid[0]);
                }
                else {//can't satisfy request, so put it in pending queue
                    insert(pending, index[0]);
                }
            }
            //This is a RELEASE request from the client
            else {//are there any requests for resources in the pending queue
                if(empty(pending)) {
                    avail++;
                    insert(units, unitid[0]);
                }
                else {//a thread is waiting for this RELEASEd resource
                    index[0] = remove(pending);
                    reply[index[0]].put(unitid[0]);
                }
            }
        }
    }
}

class Client{
int N = 10;
int unitid;
    Channel request;
    Channel reply[];

    // Constructor
Client(){
    //initial
    }
    public void run(){
request.put(i, ACQUIRE, 0);
        //wait for resource to be allocated
unitid = reply[i].get();

- - - use the resource
request.put(i, RELEASE, unitid);
    }
}

Disk Scheduling and Disk Access

• Requests for disk access

- "read(disk address)" and "write(disk address)"

- disk address is cylinder, track, and offset number

- objective in moving head disk is to minimize disk head movement

- Shortest Seek Time first is a typical scheduling algorithm

* queue incoming requests by their distance from current head position (some are lower and some are higher)

* seek to the closest request

• Separate Scheduler

• Scheduler as Intermediary

• Self-scheduling Disk Driver
• only two messages need to be exchanged

• Java Code for Self-scheduling Disk Driver
• requests are serviced to minimize distance between request and disk head

// Self_scheduling disk driver
class Schedule{
int N = 10;
    //channels from which to receive requests from clients in the request channel
    //and reply using a unique reply channel per   client
    Channel request; Channel reply[];
    //requests are queued in "lower" and "higher", respectively if their
    //disk addresses are lower or higher than the current head position
int lower[] = new int[100];
int higher[] = new int[100];
int headpos = 1, nsaved = 0;
int index[] = new int[1];
int cyl;

    // Constructor
Schedule(){
    //initial values
    }

    public void run()
    {
        while(true)
        {//any requests in channel or some pending
            while(!empty(request)||(nsaved==0)){
cyl = request.get(index, args);
                if(cyl<=headpos)
                    insert(lower,index[0],cyl,args);
                else
                    insert(higher,index[0],cyl,args);
nsaved++;
            }
            //if lower pending queue is empty schedule smallest one in "higher"
            //if higher pending queue is empty schedule largest one in "lower"
            if(size(lower)==0)
                remove(higher);
                else if(size(higher)==0)
                    remove(lower);
                    //take the closest request to the current head position by scanning both
                    //"lower" and "higher" pending queues
                        else if( (headpos-get(lower)) > (get(higher)-headpos) )
                            remove(higher);
                            else remove(lower);
                    //move head to next cylinder and access disk
headpos = cyl;
nsaved--;
                    // get result and send to client
                    reply[index[0]].put(results);
        }
    }
}

File Servers: Conversational Continuity

• Clients send "open", "read/write", "close" requests to file server
• Up to n files open at one time - one server process is allocated to each open file. The server allocator must allocate server processes to requests.
• Clients send request on "open" channel. Servers receive requests from "open" when they are free. Server passes own index to client and the client communicates with the server through a unique channel with that index.
• In summary, the client opens a conversation with a file server "i" and this conversation continues via "access[i]" until the client "closes" the file.
• Then the server goes back to looking for something to do in the "open" channel.

• Java Code for Clients and File Servers

// File servers and clients

class File{//channel objects for communication with clients
    Channel open;
    Channel access[];
    Channel open_reply[];
    Channel access_reply[];
int clientid;
        // Constructor
    File(){
    //initial
    }

    public void run(){
        while(true)
            {//listen for request from client on "open" channel
clientid = open.get(fname);
                // open data file; if successful then reply on the client-unique channel
open_reply[clientid].put(i);
file_open = true;
                while(file_open) {
                //only accept "reads", "writes", and "closes" if the file is open
                    k = access[i].get(args);
                        if(k==READ)
                    // process read request
                            else if(k==WRITE)
                    // process write request
                                else if(k==CLOSE){
                                // close file;
file_open = false;
                                }
                        //unique channel between server and client
access_reply[clientid].put(results);
                }
            }
        }
    }

class Client{
    Channel open;
    Channel access[];
    Channel open_reply[];
    Channel access_reply[];
    // Constructor
    Client(){
    //initial
    }

public void run(){
open.put("foo", j);
serverid = open_reply[j].get();
        //use the file and eventually close the file by executing
        access[serverid].put(arguments);
        results = access_reply[j].get();
        }
    }

Heartbeat Algorithms: Dispersing and Gathering State

• Node threads can only communicate with their neighbor threads, but we must determine a final state for a multi-threaded program.
• This is true for many scientific and engineering problems, including grid problems and network problems.
• Each thread exchanges information about its state with its neighbors until each thread knows the global state of (the relevant variables) of all threads.
• At each computational step, the information expands to reach all threads and then contracts to distribute the acknowledgement of receipt of information. That is why it is called a "heartbeat" algorithm.
• Example: Finding the topology of a network (which nodes are linked to which other nodes
• - A centralized solution would just keep an adjacency matrix "top", whose boolean cell at (i,j) is "true" if i and j are neighbors. Each node would send their "links" to the central thread.
• When the algorithm terminates the following predicate is true:
• TOPOLOGY:(forall p, 1<=p<=N, 1<=q<=N: top[p,q] iff links(p)[q])
• In a distributed solution, all threads must acquire knowledge of all links in the graph.
• A symmetric algorithm (where all threads use the same algorithm) is much easier (than an asymmetric algorithm) to understand, develop, and maintain.

• A distributed algorithm to compute the topology of a network
• Keep a boolean adjacency matrix called "top" where row "i" represents the links for node i. Thus, if (i,j) is true, then there is a link from node i to node j. Since this is a "connected" graph (there is a route to get from any node to any other node), every node has to be connected to at least one other node.
• Threads must exchange "top" enough times for the information to propagate across the links which compose the longest path between nodes. This is called the diameter D. The variable "R" in the program below keeps track of the number of rounds so far.

After R rounds the following predicate must hold:

ROUND: (forall: 1<=q<=N:
    (dist(p,q) <= R implies top[q,*] is filled in)

• The main loop of any "heartbeat" algorithm is:

Loop
    Send messages to neighbors;
    Receive message from neighbors;
Until termination condition.
    // Heartbeat algorithm for network topology(figure 10)

class Heartbeat{
int N = 10;
    Channel topology[]; //array of channels to exchange "top" with neighbors
    //If sender is done channel will return a "true" value.
int R = 0;
boolean done = false;
int[] sender = new int[1];
boolean qdone;
boolean[][] newtop = new boolean[[N][N];
boolean value=false, value1=true;

    // Constructor

Heartbeat()
    {
    //initial
    }

    public void run()
    {while(!done)
    {
    for(int q=0; q<N; q++)
    //if links[q] is true, then q is a neighbor of this thread
    if(links[q])
    //send your thread number, a false to indicate this is not your last message,
    // and the adjacency matrix to neighbor "q"
        topology[q].put(p, false, top);
        for(int q=0; q<N; q++)
            if(links[q])
            {
            //read "top" matrix from neighbors
            // if "qdone", this is your neighbor's last message
qdone = topology[p].get(sender, newtop);
          for(int y=0; y<N; y++)
              for(int z=0; z<N; z++)
                  top[y][z] = top[y][z] || newtop[y][z];
                  if(qdone)
                      active[sender[0]] = false;
            }
        // if every row of "top" has a "true" entry, then we have heard from all
        //nodes up to a distance of D from this thread; then we are done
            value1 = true;
            for(int y=0; y<N; y++) {
                value = false;
                for(int z=0; z<N; z++)
                //update your adjacency matrix
                value = value || top[y][z];
                value1 = value1 && value;
            }
            if(value1)
                done = true;
            R++;
            }

        //send last "top" (and a "true" to indicate this is your last message) to
        //neighbors and receive one more message from neighbors
        //to clear out channels
        for(int q=0; q<N; q++)
            if(active[q])
                topology[q].put(p,true,top);
                    for(int q=0; q<N; q++)
                        if(active[q])
qdone = topology[p].get(sender, newtop);
        }
}

Probe/echo Algorithms

• Traversal of trees and graphs in a distributed environment
• The probe/echo paradigm is the concurrent equivalent of a sequential "depth first search" algorithm
• Broadcast in a Network
• some initiator "i" wants to broadcast a message
• if you have a spanning tree, send the message along with the spanning tree to all neighbors; they forward it; each node will be notified; this costs n - 1 messages

• Can construct a spanning tree using parallel shortest path algorithm
• Example of a Spanning Tree

• Broadcast in a network of unknown topology

- send message to all neighbors; when a node receives the message for the first time, it passes the message to all neighbors, including the one from which it was received; after this first message, all others are ignored; this will cause 2(n-1) messages for a tree and 2n(n-1) for a complete graph.

• Network Topology Revisited (for general graphs with cycles)

- A node gathers the topology in two phases:

*probes are sent to neighbors

*echoes containing topology are returned to initiator

- initiator sends probe to neighbors

- when a node receives a probe, it sends a probe to all neighbors, except the one which sent it the probe; it waits for echoes from all neighbors; since there are cycles nodes can receive more than one probe from different routes; when this happens, it immediately sends an echo; when echoes are all reflected to the initiator, the topology is known; deadlock is avoided since every probe is echoed; this costs two messages across each link.

- This technique, sometimes called diffusing computations, can also be used to determine stable global states such as deadlock and program termination.

• Probe/echo for topology of a tree

// Probe/echo algorithm for topology of a tree

class Probe{
int source = 1;
int N = 10;
boolean[] links = new boolean[N];
boolean[][] localtop = new boolean[N][N];
boolean[][] newtop = new boolean[N][N];
int parent;
    Channel probe[]; //channels to send probes
    Channel echo[];//channels to receive echoes
    Channel finalecho; // one last echo to initiator

// Constructor
Probe()
    {
    //initialization
    }

    public void run(){
        parent = probe[p].get();//remember your parent's name
        for(int q=0; q<N; q++)
            if(links[q] && (q!=parent) )
                probe[q].put(p); //send probe to all neighbors
        for(int q=0; q<N; q++)
            if(links[q] && (q!=parent) )
                 echo[q].get(newtop); //wait for echoes to return from neighbors
                    for(int y=0; y<N; y++)
                        for(int z=0; z<N; z++) {
localtop[y][z] = localtop[y][z] || newtop[y][z];

                            //merge knowledge of topology
 }
if(p==source)
finalecho.put(localtop);//echo to initiator
                        else echo[parent].put(localtop);//echo to parent
       }
}

class Initiator{
int source = 1;
boolean[][] top = new boolean[N][N];
    Channel probe[];
    Channel finalecho;

// Constructor
Initiator()
    {
    //initial
    }

public void run(){
        probe[source].put(source);
finalecho.get(top);
        }
    }

• Probe/echo Program for Topology of a Graph

// Probe/echo algorithm for topology of a network

class Probe{
int source = 1;
int N = 10;
boolean[] links = new boolean[N];
boolean[][] localtop = new boolean[N][N];
boolean[][] newtop = new boolean[N][N];
int[] first = new int[1];
int[] sender = new int[1];
int need_echo;

    //need two types of channels - one with parms (type of message,sender,top)
    //and one with parms (sender,top) for final echo to initiator
    Channel1 probe_echo[];
    Channel2 finalecho;

    // Constructor
Probe(){
    //initial
    }

    public void run(){
        k = probe_echo[p].get(first,newtop);
        for(int q=0; q<N; q++) {
        if(links[q] && (q!=parent) ) need_echo++;
probe_echo[q].put(k,p,empty); }
            while(need_echo>0) {
                k = probe_echo[p].get(sender,newtop);
                if(k==PROBE)
                    //send echo immediately - this is redundant probe
probe_echo[sender[0]].put(ECHO,p,empty);
                else {
                    for(int y=0; y<N; y++)
                        for(int z=0; z<N; z++)
                        //merge local topology with received topology
localtop[y][z] = localtop[y][z] || newtop[y][z];
need_echo--; //until all echoes have been returned
                }
        }
        if(p==source)
finalecho.put(localtop);
        else probe_echo[first[0]].put(ECHO,p,localtop);
    }

}

class Initiator{
boolean[][] top = new boolean[N][N];
    Channel1 probe_echo[];
    Channel2 finalecho;

    // Constructor
Initiator(){
    //initial
    }

public void run(){
        probe[source].put(PROBE,source,empty);
finalecho.get(top);
    }
}

Broadcast Algorithms

• Message is sent to "all" other nodes; but unless we "order" these messages, some nodes can get them in different orders because they take different paths to get to the destination and have differing delays along the way.
• Logical Clocks and Ordering of Events in Distributed Systems
• communication actions are the significant events
• each process has internal ordering
• each send on a channel must precede ("happen before") the receipt of the message; time must pass for the message to be sent
• the "happened before" relation is transitive
• there is a total ordering between events which have a causal relationship
• but there is only a partial ordering among all the events in a distributed program.
• for example, if process A broadcasts a message to B and C, then C sends a message to B, B might get C's message before it gets A's message
• a global centralized clock is quite restrictive due to time delays
• a collection of perfectly synchronized local clocks is literally impossible
• build a collection of synchronized logical clocks
• each process has its own logical clock
• it ticks between any two events in one process
• each message sent has the local clock time (ts) inserted and the local clock is then incremented
• when a process receives a message with a timestamp (ts), it sets its own local clock (lc) to the maximum(lc, ts+1).
• thus all clocks are monotonically increasing
• thus all events have a global logical clock value
• processor ids can be used to resolve priorities of the same clock value

• Distributed Semaphores - synchronization of distributed processes
• semaphore invariant - The number of completed P operations is at most the number of completed V operations plus the semaphore's initial value.
• we need a way to count P and V operations, a way to delay P operations, and a way for the distributed processes to "share" the semaphore but yet maintain the semaphore invariant
• each time a process A executes a P or V operation, A first sends its intention to do so (along with the timestamp of its local clock value and its own (sender's) identity) to all those processes which can participate in the semaphore invariant, including itself; once process A has heard from all other processes that they are aware of A's request (along with their timestamps) and finds that it has the lowest clock value, it can then proceed.
• when a process receives a request for a P or V operation, it broadcasts this to all other processes and "ACK"nowledges to A that it has received the request, along with its own timestamp.
• each process must maintain a queue of messages, ordered by timestamp; when A's own request has the smallest timestamp and A has heard from every other process, it can proceed with the P or V operation; then it can delete the operation from its queue of messages.
• thus, each process maintains its own semaphore invariant:
• DSEM: nV = number of fully acknowledged V messages "and"
• nP = number of fully acknowledged P messages such that nV=>nP "and"
• mq is totally ordered by the time-stampe in P and V messages

Code Sketch for Distributed Semaphores Algorithm

type kind = enum(V,P,ACK)
chan sem[1:n] (sender: int, kind, timestamp: int)
chan go[1:n](timestamp: int)
User[i: 1..n]:: var lc: int:=0;  #logical clock
var ts: int;                      #timestamp in go message
    # execute a V operation
    broadcast sem(i,V,lc); lc := lc+1;  #time has passed
    ...
    #execute a P operation
    broadcast sem(i,P,lc); lc := lc+1; #time has passed
    receive go[i](tx); lc := max(lc,ts+1); lc := lc+1; #set to max of local clock or timestamp +1
    ...

Helper[i:1..n]:: var mq: queue of (int,kind,int)  #ordered by timestamp
var lc: int:=0;               #logical clock
var nV: int:= 0; nP:int :=0;    #semaphore counters
var sender: int; k: kind; ts: int; # values in messages
    do true -> {loop invariant DSEM}
      receive sem[i](sender,k,ts); lc := max(lc,ts+1); lc:=lc+1;
      if k=P or k=V ->
        insert(sender,k,ts) at appropriate place in mq
        broadcast sem(i,ACK,lc); lc:=lc+1;
      || k=ACK ->
          record that another ACK has been see
fa fully acknowledge V messages ->
              remove the message from mq; nV:= nV+1;
af
fa fully acknowledged P messages such that nV > nP ->
              remove the message from mq; nP:= nP+1;
              if sender =i -> send go[i](lc); lc :=lc+1 fi
af
      fi
   od

Token-passing Algorithms

• The Distributed Mutual Exclusion Problem. The possession of a token controls entry to critical sections of code. One solution is to order the processes in a ring of helper processes which pass the permission token around the ring. Each process has a helper process which, when it has the token, can grant permission to the critical section. Each time helper[i] gets the token, it checks to see if process[i] wants the critical region; if so, it holds the token until process[i] is done with the critical section; it then passes the token on to helper[i+1]. Thus each process maintains the invariant:
• DMUTEX: (forall a <=I <= n: p[i] is in its critical section means helper[i] has the token) and there is only one token
• This solution is fair as long as each process does not "hog" the token and there are no failures.
• An implementation of a token ring mutual exclusion algorithm

• Termination Detection in a Ring
• possible solutions include diffusing computations (probe/echo) and token-passing
• global termination conditions - every process is idle and no messages are in transit
• use a special token (message) which is not part of the regular computation; it travels the same ring as regular messages
• each time a process goes idle, it sends this token to its neighbor in the ring
• when a process receives the "idle" token, if it is idle, it forwards it; if not idle, it changes the state of the token to "active".
• if it returns to the initiating process and is still idle, then it has "flushed" regular messages out of the ring and the initiating process can assume the computation is terminated

• Rules for processes:

{RING: p[1]blue => (p[1]...p[1+token]blue and
ch[2]...ch[1+token mod n] empty)}

actions of p[1] when it first becomes idle
    - color[1]:=blue; token:=0; send ch[2](token)

actions of p[i:1..n] upon receiving a regular message:
    - color[i]:= red

actions of p[i:2..n] upon receiving the token
    - color[i] := blue; token:=token + 1; send ch[i mod n+1](token)

actions of p[1] upon receiving the token:
    - if color[1] = blue then announce termination and halt
    - else {color[1] := blue; token:= 0; send ch[2](token)}

• Termination Detection in a Graph

- assume a complete graph where every process is connected to every other process via a channel

- token must traverse every link (and therefore idle process) in the graph before we can determine that a program is terminated

- even though a process is blocked, waiting for a regular message, it will handle the token messages

- a process can receive either regular messages or tokens

- if a process sends a token around a cycle and it returns with a value of "nc" (the length of the cycle), then the program can terminate

- must find all cycles in the graph for the program to be terminated

- must maintain the GRAPH invariant:

{token has value T => (the last T channels token was received from were emtpy and all p[i] that passed it on were blue when they passed it) }

• Rules for processes

-actions of p[i: 1..n] upon receive a regular message

    * color[i] := red

- actions of p[i:1..n] upon receiving the token

    * if token = nc -> {announce termination and halt}

    * if color[i] = red -> color[i] := blue; token:=0

    * or color[i] = blue -> token := token + 1

    * set j to the channel corresponding to the next edge in cycle "c"

    * send ch[j](token)