The following code should be mostly OK, but I'm trying to avoid any stylistic problems or find anything that I overlooked.

The code is an implementation of asynchronous leadership election on a one way ring.

Parts of the implementation are a bit unnatural, because it has some forced features.

main.cpp

#include <iostream>
#include "node.h"
#include <algorithm>
#include <stdexcept>
#include <unistd.h>
using namespace std;

int main()
{
// initialization
reset_sent_messages();
srand(time(NULL));

// prepare the array
const unsigned total_nodes = 1000;
vector< shared_ptr<Node> > nodes;
nodes.reserve(total_nodes);

// create nodes, put them into array
for (unsigned i = 0; i < total_nodes; ++i)
nodes.push_back(make_shared<Node>(i+1));

// shuffle the nodes randomly
random_shuffle(nodes.begin(),nodes.end());

// connect the nodes
for (unsigned i = 0; i < total_nodes-1; ++i)
nodes[i]->connect(nodes[i+1]);
nodes[total_nodes-1]->connect(nodes[0]);

// prepare the futures to store the elected leader information
for (auto i : nodes)

// run the threads with the main node logic
for (auto i : nodes)

/* NOTE:
* For demonstrational purposes we are using promise<->future for final
* synchronization. This is a very unnatural model. Normaly we wouldn't
* detach the threads and use join().
*
* Do note that if a thread isn't detached and the thread variable (returned
* by thread call) is destroyed, the program will be immediately terminated.
* This is due to the fact, that such thread would be effectively leaked.
* Not running in detached mode, but incapable of joining the spawn thread.
*/

// do the final synchronization (so that main doesn't end before the algorithm does)
for (unsigned i = 0; i < total_nodes; i++)
{
throw runtime_error("Node didn't correctly detect it's leader.");
}

// final reports
cout << "Asynchronous run finished." << endl;
cout << "Total number of sent messages was : " << sent_messages() << endl;

return 0;
}


node.h

#ifndef NODE_H
#define NODE_H

#include <queue>
#include <mutex>
#include <memory>
#include <future>
#include <atomic>

class Node
{

public:
Node(unsigned node_id);

/** \brief Put a new message into this nodes message buffer */
/** \brief Put a new message into this nodes message buffer */

/** \brief Connect this node to a next node in the circle */
void connect(std::shared_ptr<Node> next);

/** \brief Node logic */
void logic();

/** \brief Get the leader node id */

private:
/** \brief Try to pull one message from the buffer
*
* Non-blocking operation.
*/
bool checkout_message(std::pair<int,int>& message);

/** \brief Transmit one message to the connected node
*
* Blocking operation.
*/
void transmit_message(int node_id, int distance);

/** \brief Sub logic for processing the message
*
* - detection of transiting state
* - trasmition of appropriate messages
*/
void process_message(int n_id1, int n_id2);

/** \brief Sub logic for transiting nodes
*
* Simple re-trasmit routine.
*/
void transit_loop_step();

private:
std::queue<std::pair<int,int> > p_message_buffer;
std::mutex p_message_buffer_lock;

std::weak_ptr<Node> p_next;
std::mutex p_next_lock;

int p_id;
bool p_transit;
bool p_done;
};

extern std::atomic<unsigned> message_count;

/** \brief Return count of sent messages */
unsigned sent_messages();

/** \brief Reinitialize the count of sent messages back to zero */
void reset_sent_messages();

#endif // NODE_H


node.cpp

#include "node.h"
#include <iostream>
#include <stdexcept>
using namespace std;

atomic<unsigned> message_count;

unsigned sent_messages() { return message_count; }
void reset_sent_messages() { message_count = 0; }

static mutex cout_lock;
// helper routine to output full lines from the algorithm
#define atomic_log(x) do { cout_lock.lock(); cout << x; cout_lock.unlock(); } while(0)

Node::Node(unsigned node_id) : p_id(node_id), p_transit(false), p_done(false)
{
if (node_id == 0)
throw range_error("Node id has to be greater than 0.");
}

{
// simple blocking implementation
p_message_buffer_lock.lock();
p_message_buffer.push(message);
p_message_buffer_lock.unlock();
}

{
}

bool Node::checkout_message(std::pair<int,int>& message)
{
// try to lock the message buffer for this thread
// if it's locked, return false and wait a bit using yield
if (!p_message_buffer_lock.try_lock())
{
return false;
}

// if we don't have any messages, yield and return false
if (p_message_buffer.size() == 0)
{
p_message_buffer_lock.unlock();
return false;
}

// we have a message, retrieve it, and remove from buffer
message = p_message_buffer.front();
p_message_buffer.pop();

// don't forget to unlock
p_message_buffer_lock.unlock();
return true;
}

void Node::connect(shared_ptr<Node> next)
{
// simple blocking implementation
// only needed if connecting is done asynchronously
p_next_lock.lock();
p_next = next;
p_next_lock.unlock();
}

void Node::transmit_message(int node_id, int distance)
{
shared_ptr<Node> next;

// if the node isn't connected yet, do blocking wait
while(1)
{
p_next_lock.lock();
next = p_next.lock();
if (next) break;
p_next_lock.unlock();
}

p_next_lock.unlock();

++message_count;
}

void Node::process_message(int n_id1, int n_id2)
{
*
* - trasmit notification message telling other nodes
* - set self as leader
* - mark algorithm as done
*/
if (n_id1 == p_id || n_id2 == p_id)
{
atomic_log("Node [ " << p_id << " ] self-declared leader." << endl);
transmit_message(p_id,-1);
p_done = true;
}
/* detection of transit mode
*
* - mark the state
*/
else if (n_id1 > p_id || n_id2 > p_id)
{
p_transit = true;
atomic_log("Node [ " << p_id << " ] switched into transit mode." << endl);
}
/* if the node is not yet a leader, or in transit mode re-send the base messages */
else
{
transmit_message(p_id,1);
transmit_message(p_id,2);
}
}

void Node::transit_loop_step()
{

/* single step in transit mode
*
* - read a single message and re-transmit it to the next node
*/
pair<int,int> message;
if (checkout_message(message))
transmit_message(message.first,message.second);

if (message.second == -1)
{
atomic_log("Node [ " << p_id << " ] recognized node [ " << message.first << " ] as the leader node." << endl);
p_done = true;
}

}

void Node::logic()
{
transmit_message(p_id,1);
transmit_message(p_id,2);

int message1 = 0, message2 = 0;

while (true)
{
// if the algorithm is done, simply exit the thread
if (p_done)
return;

// if in transit mode, run a single step of transmit logic
if (p_transit)
{
transit_loop_step();
}
// if we have two messages, process them
else if (message1 != 0 && message2 != 0)
{
process_message(message1,message2);
message1 = 0; message2 = 0;
}
// if we don't have two messages, receive new messages
else
{
pair<int,int> message;
if (checkout_message(message))
{
if (message1 == 0)
{
message1 = message.first;
if (message.second == 2 && message.first > p_id)
transmit_message(message.first,1);
}
else
{
message2 = message.first;
if (message.second == 2 && message.first > p_id)
transmit_message(message.first,1);
}
}
}
}
}

{
}

• I'm really late here, but I thought I'd chime in and say it looks pretty good in general. I'd consider using a std::unique_lock though instead of manually calling lock() and unlock() on the mutex: eg, std::unique_lock<std::mutex>(p_message_buffer_lock);. This guarantees that even in the case of an exception, the mutex will be unlocked. – Yuushi Jun 14 '13 at 6:37

First, a general observation: you have a fair number of typos in your comments. Although the compiler doesn't check such things, I prefer to think of the comments as an integral part of the code, so even minor typos should be fixed.

• Random numbers

Right now, you're using srand() and random_shuffle. Although they were quite new when this question was written, at least if you were doing this today, you'd almost certainly want to use the "new" C++11 random number generators and std::shuffle instead.

• Locking

In a number of places, you have a pattern of locking, carrying out some action, then unlocking, such as:

void Node::receive_message(const pair<int,int> &message)
{
// simple blocking implementation
p_message_buffer_lock.lock();
p_message_buffer.push(message);
p_message_buffer_lock.unlock();
}


For such a case, I'd prefer to use std::lock_guard:

void Node::receive_message(const pair<int, int> &message) {
std::lock_guard guard(pmessage_buffer_lock);
p_message_buffer.push(message);
}


This makes the code a little shorter and simpler (not a big deal, but not a bad thing by any means), but much more importantly it adds quite a bit of exception safety. If (for example) p_message_buffer.push() were to throw an exception, your code wouldn't unlock the lock, but this will.

• Magic numbers

You have magic numbers sprinkled rather liberally throughout the code. For example:

transmit_message(p_id,1);
transmit_message(p_id,2);


Until or unless you've looked at transmit_message, it may not be obvious that these are distances. Even if it's only useful from a documentation viewpoint, I'd at least consider creating a type specifically to represent a transmission distance, and probably give it an explicit constructor, so these would look something like:

transmit_message(p_id, hops(1));
transmit_message(p_id, hops(2));


This seems to me to make the code considerably more self-explanatory, even hops ends up doing essentially nothing other than giving a name/context for the numbers.

• shared_ptr

At least as I read things, the ring of nodes doesn't look like a particularly good use-case for std::shared_ptr. In particular, shared_ptr is intended to represent shared ownership. Using shared_ptr basically asserts that each node in the ring owns its successor node, which doesn't seem very accurate. Equally bad, a ring (or cycle in general) is a case where shared_ptr (or reference counting in general) doesn't work correctly. Each item in the cycle has a reference to the next, so every item always has a non-zero reference count (so they never get cleaned up) even when none of them is accessible any more.