Distributed Aggregation Algorithm

Distributed aggreation algorithms belong to a special class of distributed algorithms for mathematical operations. At each node in the distributed system, one or more values are stored, the nodes can communicate with each other, and are allowed to perform certain mathematical operations. The distributed algorithm now has to determine the exact sequence of operations that each node has to follow in order to achieve a global goal (for instance find the max, min, gcd or average value for all nodes).

Simple Operations

Invariants involving one node

It is always important to identify invariants in distributed algorithms, especially to prove safety and liveness properties in order to verify the distributed algorithm. Invariants remain constant and unchanged during all operations at the node. For simple mathematical operations such as maximum (Max), minimum (Min), and greatest common divisor (GCD) it is easy to find invariants.

If one node with value z receives a message from another node with value e, the overall value of Max, Min and GCD for all nodes is invariant (or constant) if the value at one node is replaced by the Max, Min, or GCD of both nodes, that is if the following operations are executed:

If node has value z and receives message from node with value e:
GCD: if (e < z) { z = ((z-1)%e)+1); }
MAX: if (z < e) { z = Max(z,e); }
MIN: if (e < z) { z = Min(z,e); }

If an invariant has been found, and a corresponding operation has been identified that satisfies both safety (the goal is never violated) and liveness (there is progress towards the goal) properties it is therefore quite easy to construct a distributed algorithm, see for instance the distributed gcd algorithm:

On receiving a message from a neighbor:
Receive (int e)
{ 
  if (Condition) 
  {
    z = InvariantOp(z,e)
    Send(z, all Neighbors)
  }
}

Invariants involving two nodes

The situation for operations such as sum, average, variance and standard deviation is a bit more complicated. If a value of one node is replaced by the sum or average of two nodes, the overall value of the sum or average obviously changes and is no longer invariant. Nevertheless, an invariant to calculate the average is possible, if we always consider a pair of nodes. If the values at both nodes are replaced by the average value, the overall average for all nodes is invariant.

If a pair of node communicates, and one node has value z and the node has value e:
AVG: z = e = Avg(z,e)

Therefore the reciple to construct a distributed algorithm in this case is: select two neighbor nodes, exchange values and perform operations which live the desired value invariant.

At each node execute the following:
OnTimer()
{
     NeighborNode = PickRandomNeighbor();
     Send(z, NeighborNode)
     WaitForResponse(MeighborNode);
}

On receiving a message from a neighbor:
Receive (int e)
{ 
  if (Condition) 
  {
    Send(z, NeighborNode)
    z = InvariantOp(z,e)
  }
}

Complex Operations

Complex mathematical operations such variance or standard deviation are of course as the name says more complicated to implement as a distributed algorithm. They require the storage of more than one parameter, register or value at each node.

To calcute the variance of n numbers distributed at an equal number of nodes, one can use the following distributed algorithm, according to the equation Var(x) = E((x-E(x))^2)

Store four values/parameters at each node: x, E(x), (x-E(x))^2, E((x-E(x))^2)
 1.Param: local value (remains unchanged)
 2.Param: average of 2.Param and local value
          (copy every time from 1. Param if 1. Param has changed)
 3.Param: (Value of 1.Param - Value of 2.Param)^2
          (calculate every time if 2. Param has changed)
 4.Param: average of 4.Param
          (copy every time from 3. Param if 3. Param has changed)
At each step
-Select pair of nodes
-Exchange value of 2 parameters (parameters 2 and 4)
-Calculate values of parameters 2-4 at both nodes, replace
 for average values (parameters 2 and 4) the previous values at 
 both nodes with the average of both nodes, calculate
 3. parameter if 2. parameter has changed

Alternative, there is the following faster version which requires one more parameter at each node. It works according to the equation Var(x) = E(x^2)-(E(x))^2

Store five values/parameters at each node: x, E(x), x^2, E(x^2), E(x^2)-(E(x))^2
 1.Param: local value (remains unchanged)
 2.Param: average of 2.Param and local value
          (copy every time from 1.Param if 1.Param has changed)
 3.Param: square of local value
          (calculate every time if 1.Param has changed)
 4.Param: average of 4.Param and square
          (copy every time from 3.Param if 3.Param has changed)
 5.Param: 4.Param-(2.Param)^2
          (copy every time from 3. Param if 3. Param has changed)
At each step
-Select pair of nodes
-Exchange value of 2 parameters (parameters 2 and 4)
-Calculate values of params 2-5 at both nodes, replace
 for average values (parameters 2 and 4) the previous values at 
 both nodes with the average of both nodes, calculate
 the others

To calculate the standard deviation, one has to add another parameter at each node, since the standard deviation can be defined as the square root of the variance.

Literature

Mark Jelasity, Alberto Montresor, and Ozalp Babaoglu, Gossip-Based Aggregation in Large Dynamic Networks, ACM Transactions on Computer Systems, Vol. 23, No. 3, August (2005) 219–252