Paradigms (Patterns) for Process Interaction in Distributed Programs

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

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) )
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)