# Distributed (asynchronous) averaging

This code simulates a distributed averaging protocol. There is a network of nodes, and at the clicks of a Poisson process:

1. A random node wakes up

2. Transmits it's current average to all the nodes it is connected to

3. These nodes have a coroutine which waits to accept new values and compute the average.

from numpy.random import exponential, randint, uniform
import time
from itertools import product

class Node(object):
def __init__(self, number):
self.friends = []
self.count = 0
self.val = number
self.average = number
self.avg = self.averager()
self.avg.next()

def averager(self):
while True:
term = yield self.average
self.val += term
self.count += 1
self.average = self.val/self.count

class Network(object):
def __init__(self, num_nodes):
self.nodes = *num_nodes
for i in range(num_nodes):
self.nodes[i] = Node(randint(1, 10))
self.set_connections()

def set_connections(self):
for f, g in product(self.nodes, self.nodes):
p = uniform()
if p < 0.8:
f.friends.append(g)
g.friends.append(f)

def mainloop(self):
num_nodes = len(self.nodes)
while True:
next_time = exponential(0.1)
node = self.nodes[randint(num_nodes)]
for friend in node.friends:
friend.avg.send(node.average)
print friend.average
time.sleep(next_time)


So far I've only tested it as follows:

>>> n = Network(10)
>>> n.mainloop()


and then observed that the friend.average printed out all converge to the same value.

My main concern about this code is that a lot of the logic for waking up nodes is in the Network class, whilst I'd rather it was in the node class. Also, there's no exception/error handling, or proper testing.

The main purpose of this was to try and understand coroutines, which was a bit of a mind bender.

The way that the random graph is generated seems suspicious:

class Network(object):
def __init__(self, num_nodes):
self.nodes = *num_nodes
for i in range(num_nodes):
self.nodes[i] = Node(randint(1, 10))
self.set_connections()

def set_connections(self):
for f, g in product(self.nodes, self.nodes):
p = uniform()
if p < 0.8:
f.friends.append(g)
g.friends.append(f)


First of all, self.nodes should just be defined as

self.nodes = [Node(randint(1, 10)) for _ in range(num_nodes)]


The way that nodes changes type from a list of integers to a list of Node objects makes it feel slightly more dirty.

It looks like set_connections() is only meant to be called from the constructor, so it should be renamed _set_connections() to suggest that it is private. However, considering that __init__() would just consist of two lines (self.nodes = … and self._set_connections()), I would just lump everything into __init__().

product(self.nodes, self.nodes) seems wrong. It's possible to cause a Node to befriend itself. It's also possible to have two nodes list each other as friends twice. I suspect that what you want is combinations(self.nodes, 2).

def __init__(self, num_nodes):
self.nodes = [Node(randint(1, 10)) for _ in range(num_nodes)]
for f, g in combinations(self.nodes, 2):
if uniform() < 0.8:
f.friends.append(g)
g.friends.append(f)