chain_node.Tc

#include "libconfig.h++"
#include "logging.h"
#include <iostream>
#include <string>
#include <ctime>
#include <map>
#include <set>
#include <deque>
#include <sstream>
#include <ctime>
#include "sha.h"
#include "DiskStorage.h"
#include "tame_rpcserver.h"
#include "parseopt.h"
#include "ID_Value.h"
#include "Node.h"
#include "MemStorage.h"
#include "HttpStorage.h"
#include "Storage.h"
#include "connection_pool.Th"
#include "zookeeper.h"
#include "zoo_craq.Th"
#include <tclap/CmdLine.h>

using namespace CryptoPP;
using namespace std;

const unsigned int CHAIN_SIZE = 3;
log4cpp::Appender *app;
Storage * storage;

class rpc_server : public tame::server_t {
public:
  rpc_server (int fd, int v) : tame::server_t (fd, v) { tcp_nodelay(fd); }
  const rpc_program &get_prog () const { return chain_node_1; }
  void dispatch (svccb *sbp);
};

class rpc_server_factory : public tame::server_factory_t {
public:
  rpc_server_factory () : tame::server_factory_t () {}
  tame::server_t *alloc_server (int fd, int v) { return New rpc_server (fd, v); }
};

struct chain_meta {
	unsigned int chain_size;
	vector<string> data_centers;
};

struct key_meta {
	unsigned int committed;
	unsigned int max_pending;
	map<int, blob> pending_list;
	map<int, deque<svccb *> > write_reqs;
	bool is_head;
	bool is_tail;
	ID_Value chain_id;
};

typedef map<ID_Value, Node>::iterator ring_iter;
typedef map<ID_Value, key_meta>::iterator key_iter;

static void get_chain_info(ID_Value chain_id, ptr<callback<void, ptr<chain_meta> > > cb, CLOSURE);
static void process_query_obj_ver(svccb * sbp, CLOSURE);
static void process_tail_read(svccb * sbp, CLOSURE);
static void process_tail_read_ex(svccb * sbp, CLOSURE);
static void process_head_write(svccb * sbp, CLOSURE);
static void process_propagate(svccb * sbp, CLOSURE);
static void propagate(ID_Value chain_id, ID_Value id, bool send_committed, cbb cb, CLOSURE);
static void process_back_propagate(svccb * sbp, CLOSURE);
static void back_propagate(ID_Value chain_id, ID_Value id, bool send_committed, cbb cb, CLOSURE);
static void process_ack(svccb * sbp, CLOSURE);
static void process_add_chain(svccb * sbp, CLOSURE);
static void process_test_and_set(svccb * sbp, CLOSURE);
static void ack(ID_Value chain_id, ID_Value id, cbb cb, CLOSURE);
static void report_bad_node(Node n, CLOSURE);
static void node_added(Node node_changed, CLOSURE);
static void node_deleted(Node node_changed, CLOSURE);
static void ext_ring_ptr(chain_meta chain, string dc, ptr<callback<void, ptr<map<string, map<ID_Value, Node> >::iterator > > > cb, CLOSURE);
static void ext_ring_succ(chain_meta chain, ID_Value id, ptr<callback<void, ptr<Node> > > cb, CLOSURE);
static void ext_ring_pred(chain_meta chain, ID_Value id, ptr<callback<void, ptr<Node> > > cb, CLOSURE);
static void ext_ring_tail(chain_meta chain, ID_Value id, ptr<callback<void, ptr<Node> > > cb, CLOSURE);

ring_iter ring_succ(ID_Value id);
void ring_incr(ring_iter * it);
void ring_decr(ring_iter * it);

ring_iter ring_succ(ID_Value id);
void ring_incr(ring_iter * it);
void ring_decr(ring_iter * it);

unsigned int known_version;

map<ID_Value, Node> ring;
Node my_node;
string my_node_str;
bool ring_init = false;
bool init_interrupted = false;
ring_iter my_node_ptr;
ID_Value my_id;
string datacenter;

map<ID_Value, key_meta> key_meta_list;
map<ID_Value, chain_meta> chain_meta_list;
map<string, map<ID_Value, Node> > ext_rings;

bool update_running = false;

void dont_care(bool x) {}

void delay(long ms) {
  timespec ts;
  ts.tv_sec = ms / 1000;
  ts.tv_nsec = (ms % 1000) * 1000000;
  nanosleep(&ts, NULL);
}

str user_ip;
str get_ip_address() {
        vec<in_addr> addrs;
        if (!myipaddrs(&addrs))
        LOG_FATAL << "Cannot find my IP address.\n";

        in_addr *addr = addrs.base();
        in_addr *loopback = NULL;
        in_addr *touse = NULL;
        while (addr < addrs.lim ()) {
                if (ntohl (addr->s_addr) == INADDR_LOOPBACK) {
                        loopback = addr;
                }
                else {
                	touse = addr;
                	str check = inet_ntoa(*touse);
                	if(check == user_ip) {
                		break;
                	}
                }
                addr++;
        }
        if (addr >= addrs.lim () && (loopback == NULL))
                LOG_FATAL << "Cannot find my IP address.\n";

        if (touse == NULL) {
                LOG_WARN << "Using loopback address as my address.\n";
                touse = loopback;
        } else {
        	LOG_WARN << "Using IP address: " << inet_ntoa (*touse) << "\n";
        }
        str ids = inet_ntoa (*touse);
        return ids;
}

ID_Value get_sha1(string msg)
{
	byte buffer[SHA::DIGESTSIZE];
	SHA().CalculateDigest(buffer, (byte *)msg.c_str(), msg.length());
	ID_Value ret(buffer);
 	return ret;
}

tamed void get_chain_info(ID_Value chain_id, ptr<callback<void, ptr<chain_meta> > > cb) {
	tvars {
		ptr<chain_meta> ret;
		string * val;
		istringstream iss;
		string dc;
		map<ID_Value, chain_meta>::iterator it;
	}

	it = chain_meta_list.find(chain_id);
	if(it != chain_meta_list.end()) {
		ret = New refcounted<chain_meta>;
		*ret = it->second;
		TRIGGER(cb, ret);
		return;
	}

	twait{ czoo_get("/keys/" + chain_id.toString(), mkevent(val)); }
	if(val == NULL) {
		LOG_WARN << "Failed to get from zookeeper: ";
		LOG_WARN << chain_id.toString().c_str();
		ret = NULL;
		TRIGGER(cb, ret);
		return;
	}

	ret = New refcounted<chain_meta>;
	iss.str(*val);
	delete val;
	if(!(iss >> ret->chain_size)) {
		LOG_FATAL << "Got bad value back from zookeeper chain node!\n";
	}
	while(!iss.eof()) {
		iss >> dc;
		ret->data_centers.push_back(dc);
	}
	if(ret->data_centers.size() < 1) {
		LOG_FATAL << "Got no data centers back from zookeeper chain node!\n";
	}

	chain_meta_list[chain_id] = *ret;
	TRIGGER(cb, ret);

}

tamed void ext_ring_ptr(chain_meta chain, string dc, ptr<callback<void, ptr<map<string, map<ID_Value, Node> >::iterator> > > cb) {
	//TODO: Need to fix to deal with concurrent calls and watch value changes
	tvars {
		string dc_find;
		int i;
		ptr<map<string, map<ID_Value, Node> >::iterator> ret;
		map<string, map<ID_Value, Node> >::iterator it;
		map<ID_Value, Node>::iterator itt;
		vector<string> * node_list;
		vector<string *> node_vals;
		string find;
		string search;
		Node new_node;
		map<ID_Value, Node> new_ext_ring;
	}

	dc_find = dc;

	it = ext_rings.find(dc_find);
	if(it != ext_rings.end()) {
		ret = New refcounted<map<string, map<ID_Value, Node> >::iterator>;
		*ret = it;
		TRIGGER(cb, ret);
		return;
	}

	//TODO: Possible flash flood here
	twait { czoo_get_children("/nodes/" + dc_find, NULL, mkevent(node_list)); }
	if(node_list == NULL) {
		LOG_FATAL << "Error retrieving external node list!\n";
	}
	node_vals.resize((*node_list).size());
	twait {
		for(i=0; i<(*node_list).size(); i++) {
			find = (*node_list)[i];
			search = "/nodes/" + dc_find + "/" + find;
			czoo_get(search, mkevent(node_vals[i]));
		}
	}

	for(i=0; i<node_vals.size(); i++) {
		if(node_vals[i] == NULL) {
			LOG_FATAL << "Error occurred retrieving external node value!\n";
		}
		new_node.set_from_string(*node_vals[i]);
		new_ext_ring[new_node.getId()] = new_node;
		delete node_vals[i];
	}
	delete node_list;
	ext_rings[dc_find] = new_ext_ring;

	it = ext_rings.find(dc_find);
	ret = New refcounted<map<string, map<ID_Value, Node> >::iterator>;
	*ret = it;
	TRIGGER(cb, ret);

}

tamed void ext_ring_succ(chain_meta chain, ID_Value id, ptr<callback<void, ptr<Node> > > cb) {
	tvars {
		ptr<map<string, map<ID_Value, Node> >::iterator> val;
		map<ID_Value, Node>::iterator it;
		ptr<Node> ret;
		u_int i;
		string dc_find;
	}

	for(i=0; i<chain.data_centers.size(); i++) {
		if(chain.data_centers[i] == datacenter)
			break;
	}
	if(chain.data_centers.size() == 1 ||
			i>=chain.data_centers.size()-1) {
		TRIGGER(cb, NULL);
		return;
	} else {
		dc_find = chain.data_centers[i+1];
	}

	twait { ext_ring_ptr(chain, dc_find, mkevent(val)); }
	if(val == NULL) {
		TRIGGER(cb, NULL);
		return;
	}

	it = (*val)->second.lower_bound(id);
	if(it == (*val)->second.end())
		it = (*val)->second.begin();

	ret = New refcounted<Node>;
	*ret = it->second;
	TRIGGER(cb, ret);
}

tamed void ext_ring_pred(chain_meta chain, ID_Value id, ptr<callback<void, ptr<Node> > > cb) {
	tvars {
		ptr<map<string, map<ID_Value, Node> >::iterator> val;
		map<ID_Value, Node>::iterator it;
		ptr<Node> ret;
		u_int i;
		string dc_find;
	}

	for(i=0; i<chain.data_centers.size(); i++) {
		if(chain.data_centers[i] == datacenter)
			break;
	}
	if(chain.data_centers.size() == 1 ||
			i<=0) {
		TRIGGER(cb, NULL);
		return;
	} else {
		dc_find = chain.data_centers[i-1];
	}

	twait { ext_ring_ptr(chain, dc_find, mkevent(val)); }
	if(val == NULL) {
		TRIGGER(cb, NULL);
		return;
	}

	it = (*val)->second.lower_bound(id);
	if(it == (*val)->second.end())
		it = (*val)->second.begin();

	for(i=0; i<chain.chain_size-1; i++) {
		it++;
		if(it == (*val)->second.end())
			it = (*val)->second.begin();
	}

	ret = New refcounted<Node>;
	*ret = it->second;
	TRIGGER(cb, ret);
}

tamed void ext_ring_tail(chain_meta chain, ID_Value id, ptr<callback<void, ptr<Node> > > cb) {
	tvars {
		ptr<map<string, map<ID_Value, Node> >::iterator> val;
		map<ID_Value, Node>::iterator it;
		ptr<Node> ret;
		u_int i;
		string dc_find;
	}

	dc_find = chain.data_centers[chain.data_centers.size()-1];

	twait { ext_ring_ptr(chain, dc_find, mkevent(val)); }
	if(val == NULL) {
		TRIGGER(cb, NULL);
		return;
	}

	it = (*val)->second.lower_bound(id);
	if(it == (*val)->second.end())
		it = (*val)->second.begin();

	for(i=0; i<chain.chain_size-1; i++) {
		it++;
		if(it == (*val)->second.end())
			it = (*val)->second.begin();
	}

	ret = New refcounted<Node>;
	*ret = it->second;
	TRIGGER(cb, ret);
}

tamed void process_query_obj_ver(svccb * sbp) {
	tvars {
		rpc_hash parg;
		query_obj_ver_ret repl;
		ID_Value id;
		key_iter it;
	}

	parg = *(sbp->getarg<rpc_hash>());
	LOG_WARN << "Got QUERY_OBJ Request\n";

	id.set_from_rpc(parg);
	it = key_meta_list.find(id);
	if(it == key_meta_list.end()) {
		sbp->replyref(NULL);
		return;
	}

	repl.hist = (it->second).committed;
	repl.pend = (it->second).max_pending;
	sbp->replyref(repl);
}

tamed void process_tail_read(svccb * sbp) {
	tvars {
		rpc_hash parg;
		ptr<blob> repl;
		ID_Value id;
		key_iter it;
		ring_iter rit;
		int i;
		ptr<aclnt> cli;
		clnt_stat e;
		int fd;
		query_obj_ver_ret ret;
		map<int, blob>::iterator kit;
		blob to_rep;
		ID_Value chain_id;
		ptr<chain_meta> chain_info;
	}

	parg = *(sbp->getarg<rpc_hash>());
	LOG_WARN << "Got TAIL_READ Request\n";

	id.set_from_rpc(parg);

	it = key_meta_list.find(id);

	//CRAQ tail read

	if(it == key_meta_list.end() ) {
		sbp->replyref(NULL);
		return;
	}

	if(it->second.committed == it->second.max_pending) {
		LOG_WARN << "Clean READ " << id.toString().c_str() << "\n";
		twait { storage->get(id, mkevent(repl)); }
		sbp->replyref(*repl);
		return;
	} else {
		LOG_WARN << "Dirty READ " << id.toString().c_str() << "\n";

		//Find tail
		rit = ring_succ(id);
		for(i=0; i<CHAIN_SIZE-1; i++)
			ring_incr(&rit);

		twait { get_rpc_cli (rit->second.getIp().c_str(), rit->second.getPort(), &cli, &chain_node_1, mkevent(fd)); }

		if( fd<0 ) {
			report_bad_node(rit->second);
			sbp->replyref(NULL);
			return;
		}

		//Query tail
		twait {	cli->call(QUERY_OBJ_VER, &parg, &ret,  mkevent(e)); }

		if(e) {
			report_bad_node(rit->second);
			sbp->replyref(NULL);
			return;
		} else {
			//Refetch key
			it = key_meta_list.find(id);
			if(it == key_meta_list.end() ) {
				sbp->replyref(NULL);
				return;
			}

			//Got an ACK between call
			if(it->second.committed == it->second.max_pending) {
				LOG_WARN << "Clean READ " << id.toString().c_str() << "\n";
				twait { storage->get(id, mkevent(repl)); }
				sbp->replyref(*repl);
				return;
			}
			//See if we have the version the tail would return
			kit = it->second.pending_list.find(ret.hist);
			if(kit == it->second.pending_list.end()) {
				sbp->replyref(NULL);
				return;
			}
			//Return tail's committed version
			to_rep = kit->second;
			sbp->replyref(to_rep);
			return;
		}
	}

}

tamed void process_tail_read_ex(svccb * sbp) {
	tvars {
		tail_read_ex_arg parg;
		tail_read_ex_ret empty;
		ptr<blob> repl;
		ID_Value id;
		key_iter it;
		ring_iter rit;
		int i;
		ptr<aclnt> cli;
		clnt_stat e;
		int fd;
		query_obj_ver_ret ret;
		map<int, blob>::iterator kit;
		tail_read_ex_ret to_rep;
		ID_Value chain_id;
		ptr<chain_meta> chain_info;
		Node tail;
		ptr<Node> ext_tail;
		rpc_hash qry_id;
		timeval started;
		timeval cur_time;
		long sec_diff;
		long usec_diff;
	}

	gettimeofday(&cur_time, NULL);
	LOG_ALERT << "READ\t" << cur_time.tv_sec << "\t" << cur_time.tv_usec << "\n";

	parg = *(sbp->getarg<tail_read_ex_arg>());
	LOG_WARN << "Got TAIL_READ_EX Request\n";

	id.set_from_rpc(parg.id);

	it = key_meta_list.find(id);

	//CRAQ tail read

	if(it == key_meta_list.end() ) {
		sbp->replyref(empty);
		LOG_INFO << "iterator is at the end of the meta list";
		return;
	}

	if(!parg.dirty && it->second.committed == it->second.max_pending) {
		LOG_WARN << "Clean READ " << id.toString().c_str() << "\n";
		twait { storage->get(id, mkevent(repl)); }
		LOG_INFO << "after storage get";
		to_rep.data = *repl;
		LOG_INFO << "1";
		to_rep.dirty = false;
		LOG_INFO << "2";
		to_rep.ver = it->second.committed;
		LOG_INFO << "before replyref";
		sbp->replyref(to_rep);

		gettimeofday(&cur_time, NULL);
		LOG_ALERT << "READ_DONE\t" << cur_time.tv_sec << "\t" << cur_time.tv_usec << "\n";

		LOG_INFO << "after replyref";
		return;
	} else {
		LOG_WARN << "Dirty READ " << id.toString().c_str() << "\n";

		//lookup chain
		chain_id.set_from_rpc(parg.chain);
		twait{ get_chain_info(chain_id, mkevent(chain_info)); }
		if(chain_info == NULL) {
			LOG_FATAL << "Couldn't get chain info in read!\n";
			sbp->replyref(empty);
			return;
		}

		//Find tail
		if(chain_info->data_centers[chain_info->data_centers.size()-1] == datacenter) {
			rit = ring_succ(id);
			for(i=0; i<chain_info->chain_size-1; i++)
				ring_incr(&rit);
			tail = rit->second;
		} else {
			twait { ext_ring_tail(*chain_info, id, mkevent(ext_tail)); }
			if(ext_tail == NULL) {
				LOG_FATAL << "Error when trying to retrieve external tail!\n";
			}
			tail = *ext_tail;
		}

		LOG_INFO << "before get_rpc_cli call";
		twait { get_rpc_cli (tail.getIp().c_str(), tail.getPort(), &cli, &chain_node_1, mkevent(fd)); }
		LOG_INFO << "after get_rpc_cli call";

		if( fd<0 ) {
			report_bad_node(tail);
			sbp->replyref(empty);
			return;
		}

		//Query tail
		qry_id = id.get_rpc_id();
		//gettimeofday(&started, NULL);
		LOG_INFO << "before query_obj_ver call";
		twait {	cli->call(QUERY_OBJ_VER, &qry_id, &ret,  mkevent(e)); }
		LOG_INFO << "after query_obj_ver call";
		/*gettimeofday(&cur_time, NULL);
		sec_diff = cur_time.tv_sec - started.tv_sec;
		if(sec_diff == 0) {
			usec_diff = cur_time.tv_usec - started.tv_usec;
		} else {
			usec_diff = (1000000 - started.tv_usec) + cur_time.tv_usec;
			sec_diff--;
			sec_diff += (usec_diff / 1000000);
			usec_diff = usec_diff % 1000000;
		}
		LOG_WARN << sec_diff << "\t" << usec_diff << "\n";*/

		if(e) {
			LOG_INFO << "report bad node";
			report_bad_node(rit->second);
			sbp->replyref(empty);
			return;
		} else {
			//Refetch key
			it = key_meta_list.find(id);
			if(it == key_meta_list.end() ) {
				sbp->replyref(empty);
				return;
			}

			//Got an ACK between call
			if(it->second.committed == it->second.max_pending) {
				LOG_WARN << "Clean READ " << id.toString().c_str() << "\n";
				twait { storage->get(id, mkevent(repl)); }
				LOG_INFO << "after storage get 2";
				to_rep.data = *repl;
				LOG_INFO << "3";
				to_rep.dirty = true;
				LOG_INFO << "4";
				to_rep.ver = it->second.committed;
				LOG_INFO << "before replyref 2";
				sbp->replyref(to_rep);

				gettimeofday(&cur_time, NULL);
				LOG_ALERT << "READ_DONE\t" << cur_time.tv_sec << "\t" << cur_time.tv_usec << "\n";

				LOG_INFO << "after replyref 2";
				return;
			}
			//See if we have the version the tail would return
			kit = it->second.pending_list.find(ret.hist);
			if(kit == it->second.pending_list.end()) {
				kit = it->second.pending_list.find(it->second.max_pending); //really wrong
				if(kit == it->second.pending_list.end()) {
					sbp->replyref(empty);
					return;
				}
			}
			//Return tail's committed version
			to_rep.data = kit->second;
			to_rep.dirty = true;
			to_rep.ver = ret.hist;
			sbp->replyref(to_rep);

			gettimeofday(&cur_time, NULL);
			LOG_ALERT << "READ_DONE\t" << cur_time.tv_sec << "\t" << cur_time.tv_usec << "\n";

			return;
		}
	}

}

tamed void process_head_write(svccb * sbp) {
	tvars {
		head_write_arg parg;
		ID_Value id;
		ID_Value chain_id;
		key_iter it;
		ring_iter parent_ptr;
		key_meta wrt;
		bool ret_val;
		ptr<chain_meta> chain_info;
		timeval cur_time;
	}

	gettimeofday(&cur_time, NULL);
	LOG_ALERT << "WRITE\t" << cur_time.tv_sec << "\t" << cur_time.tv_usec << "\n";

	parg = *(sbp->getarg<head_write_arg>());
	LOG_WARN << "Got HEAD_WRITE Request\n";
	id.set_from_rpc(parg.id);
	chain_id.set_from_rpc(parg.chain);

	twait{ get_chain_info(chain_id, mkevent(chain_info)); }
	if(chain_info == NULL) {
		LOG_DEBUG << "Rejecting head_write because couldn't get chain info";
		sbp->replyref(false);
		return;
	}

	//Reject writes unless we can form a chain
	if(ring.size() < chain_info->chain_size) {
		LOG_DEBUG << "Rejecting head_write because chain size > num nodes";
		sbp->replyref(false);
		return;
	}

	//Reject if first data center is not us
	if(chain_info->data_centers[0] != datacenter) {
		LOG_DEBUG << "Rejecting head_write because we are not first datacenter";
		sbp->replyref(false);
		return;
	}

	it = key_meta_list.find(id);

	//If we're not the head, reject the request
	if(it != key_meta_list.end() && !(it->second).is_head ) {
		LOG_DEBUG << "Rejecting head_write because we are not the head 1";
		sbp->replyref(false);
		return;
	}

	parent_ptr = my_node_ptr;
	ring_decr(&parent_ptr);

	if(it == key_meta_list.end() && !id.betweenIncl(parent_ptr->first, my_id)) {
		//Reply false if we don't think we should be the head
		LOG_DEBUG << "Rejecting head_write because we are not the head 2";
		sbp->replyref(false);
		return;
	} else if(it == key_meta_list.end()) {
		//Create new key if this is the first
		wrt.committed = 0;
		wrt.max_pending = 1;
		if(chain_info->chain_size == 1) {
			wrt.is_tail = true;
		} else {
			wrt.is_tail = false;
		}
		wrt.is_head = true;
		wrt.pending_list[1] = parg.data;
		wrt.write_reqs[1].push_back(sbp);
		wrt.chain_id = chain_id;
		key_meta_list[id] = wrt;
	} else {
		//Update key if this is not the first
		wrt = it->second;
		wrt.max_pending++;
		wrt.pending_list[wrt.max_pending] = parg.data;
		wrt.write_reqs[wrt.max_pending].push_back(sbp);
		wrt.chain_id = chain_id;
		key_meta_list[id] = wrt;
	}

	twait { propagate(chain_id, id, false, mkevent(ret_val)); }
}

tamed void process_test_and_set(svccb * sbp) {
	tvars {
		test_and_set_arg parg;
		ID_Value id;
		ID_Value chain_id;
		key_iter it;
		ring_iter parent_ptr;
		key_meta wrt;
		bool ret_val;
		ptr<chain_meta> chain_info;
	}

	parg = *(sbp->getarg<test_and_set_arg>());
	LOG_WARN << "Got TEST_AND_SET Request\n";
	id.set_from_rpc(parg.id);
	chain_id.set_from_rpc(parg.chain);

	twait{ get_chain_info(chain_id, mkevent(chain_info)); }
	if(chain_info == NULL) {
		sbp->replyref(false);
		return;
	}

	//Reject writes unless we can form a chain
	if(ring.size() < chain_info->chain_size) {
		sbp->replyref(false);
		return;
	}

	//Reject if first data center is not us
	if(chain_info->data_centers[0] != datacenter) {
		sbp->replyref(false);
		return;
	}

	it = key_meta_list.find(id);

	//If key does not exist already or we're not the head, reject the request
	if(it == key_meta_list.end() || !(it->second).is_head ) {
		sbp->replyref(false);
		return;
	}

	//If requested version is not the latest committed version, just reject
	if(parg.ver != (it->second).committed) {
		sbp->replyref(false);
		return;
	}

	//Turn the test-and-set into a normal write and propagate
	wrt = it->second;
	wrt.max_pending++;
	wrt.pending_list[wrt.max_pending] = parg.data;
	wrt.write_reqs[wrt.max_pending].push_back(sbp);
	wrt.chain_id = chain_id;
	key_meta_list[id] = wrt;

	twait { propagate(chain_id, id, false, mkevent(ret_val)); }
}

tamed void process_propagate(svccb * sbp) {
	tvars {
		propagate_arg parg;
		ID_Value id;
		ID_Value chain_id;
		key_iter kit;
		ring_iter t;
		key_meta wrt;
		u_int i;
		bool in_succ;
		bool ret_val;
		bool set_succ;
		ptr<chain_meta> chain_info;
	}

	parg = *(sbp->getarg<propagate_arg>());
	LOG_WARN << "Got PROPAGATE Request\n";
	LOG_WARN << "Received Propagate key of size " << parg.data.size() << "\n";

	chain_id.set_from_rpc(parg.chain);

	twait{ get_chain_info(chain_id, mkevent(chain_info)); }
	if(chain_info == NULL) {
		LOG_FATAL << "Couldn't get chain info in propagate!\n";
		sbp->replyref(false);
		return;
	}

	in_succ = false;
	for(i=0; i<chain_info->data_centers.size(); i++) {
		if(chain_info->data_centers[i] == datacenter) {
			in_succ = true;
		}
	}
	//Return false if we don't think we should be storing a replica of this key
	if(!in_succ) {
		LOG_WARN << "Rejecting PROPOGATE since not in this datacenter";
		sbp->replyref(false);
		return;
	}

	id.set_from_rpc(parg.id);
	kit = key_meta_list.find(id);

	//Reply true if we already have a higher or equal version
	if(kit != key_meta_list.end() &&
		((kit->second.max_pending >= parg.ver && parg.committed == false) ||
		 (kit->second.committed >= parg.ver && parg.committed == true))) {
		 	LOG_WARN << "Already higher\n";
			sbp->replyref(true);
			return;
	}

	wrt.committed = 0;
	wrt.max_pending = 0;
	wrt.is_tail = false;
	wrt.is_head = false;
	wrt.chain_id = chain_id;

	t = ring_succ(id);
	in_succ = false;
	for(i=0; i<chain_info->chain_size; i++) {
		if(t == my_node_ptr) {
			in_succ = true;
			break;
		}
		ring_incr(&t);
	}
	if(i == chain_info->chain_size-1) {
        LOG_WARN << "I am the tail of this chain\n";
		wrt.is_tail = true;
    }
	if(i == 0)
		wrt.is_head = true;
	//Return false if we don't think we should be storing a replica of this key
	if(!in_succ) {
		LOG_WARN << "Not storing data since not in the chain\n";
		sbp->replyref(false);
		return;
	}

	if(kit != key_meta_list.end()) {
		wrt = kit->second;
	}

	//Update meta key
	if(parg.committed == true) {
		//TODO: set storage based on chain and key not just key!
		twait { storage->set(id, &parg.data, mkevent(set_succ)); }
		wrt.committed = parg.ver;
		if(wrt.max_pending < wrt.committed)
			wrt.max_pending = wrt.committed;
		wrt.pending_list[wrt.max_pending] = parg.data;
	} else {
		wrt.max_pending = parg.ver;
		wrt.pending_list[parg.ver] = parg.data;
	}
	if(wrt.is_tail &&
			chain_info->data_centers[chain_info->data_centers.size()-1] == datacenter) {
		wrt.committed = wrt.max_pending;
		twait { storage->set(id, &wrt.pending_list[wrt.max_pending], mkevent(set_succ)); }
		wrt.pending_list.clear();
	}
	key_meta_list[id] = wrt;

	if(wrt.is_tail &&
			chain_info->data_centers[chain_info->data_centers.size()-1] == datacenter) {
		sbp->replyref(true);
		LOG_WARN << "Storing this data since I'm the tail, replied.";
		twait { ack(chain_id, id, mkevent(ret_val)); }
	} else {
		sbp->replyref(true);
		twait { propagate(chain_id, id, parg.committed, mkevent(ret_val)); }
	}

}

tamed void process_back_propagate(svccb * sbp) {
	tvars {
		propagate_arg parg;
		ID_Value id;
		ID_Value chain_id;
		key_iter kit;
		ring_iter t;
		key_meta wrt;
		u_int i;
		bool in_succ;
		bool ret_val;
		bool set_succ;
		ptr<chain_meta> chain_info;
	}

	parg = *(sbp->getarg<propagate_arg>());
	LOG_WARN << "Got BACK_PROPAGATE Request\n";

	chain_id.set_from_rpc(parg.chain);

	twait{ get_chain_info(chain_id, mkevent(chain_info)); }
	if(chain_info == NULL) {
		LOG_FATAL << "Couldn't get chain info in back propagate!\n";
		sbp->replyref(false);
		return;
	}

	in_succ = false;
	for(i=0; i<chain_info->data_centers.size(); i++) {
		if(chain_info->data_centers[i] == datacenter) {
			in_succ = true;
		}
	}
	//Return false if we don't think we should be storing a replica of this key
	if(!in_succ) {
		sbp->replyref(false);
		return;
	}

	id.set_from_rpc(parg.id);

	kit = key_meta_list.find(id);
	//Reply true if we already have a higher or equal version
	if(kit != key_meta_list.end() &&
		((kit->second.max_pending >= parg.ver && parg.committed == false) ||
		 (kit->second.committed >= parg.ver && parg.committed == true))) {
			sbp->replyref(true);
			return;
	}

	wrt.committed = 0;
	wrt.max_pending = 0;
	wrt.is_tail = false;
	wrt.is_head = false;

	t = ring_succ(id);
	in_succ = false;
	for(i=0; i<chain_info->chain_size; i++) {
		if(t == my_node_ptr) {
			in_succ = true;
			break;
		}
		ring_incr(&t);
	}
	if(i == chain_info->chain_size-1)
		wrt.is_tail = true;
	if(i == 0)
		wrt.is_head = true;
	//Return false if we don't think we should be storing a replica of this key
	if(!in_succ) {
		sbp->replyref(false);
		return;
	}

	if(kit != key_meta_list.end()) {
		wrt = kit->second;
	}

	//Update meta key
	if(parg.committed == true) {
		twait { storage->set(id, &parg.data, mkevent(set_succ)); }
		wrt.committed = parg.ver;
		if(wrt.max_pending < wrt.committed)
			wrt.max_pending = wrt.committed;
		wrt.pending_list[wrt.max_pending] = parg.data;
	} else {
		wrt.max_pending = parg.ver;
		wrt.pending_list[parg.ver] = parg.data;
	}
	key_meta_list[id] = wrt;

	if(!wrt.is_head) {
		sbp->replyref(true);
		twait { back_propagate(chain_id, id, parg.committed, mkevent(ret_val)); }
	}

}

tamed void process_ack(svccb * sbp) {
	tvars {
		ack_arg parg;
		ID_Value id;
		ID_Value chain_id;
		key_iter kit;
		map<int, blob>::iterator pendit;
		map<int, deque<svccb *> >::iterator it;
		deque<svccb *>::iterator repls;
		bool ret_val;
		bool set_succ;
		timeval cur_time;
	}

	parg = *(sbp->getarg<ack_arg>());
	LOG_WARN << "Got ACK Request\n";

	chain_id.set_from_rpc(parg.chain);
	id.set_from_rpc(parg.id);

	kit = key_meta_list.find(id);

	//If we don't have this key, just reply false
	if(kit == key_meta_list.end()) {
		sbp->replyref(false);
		return;
	}

	//If we have higher or equal version committed, just reply true
	if(kit->second.committed >= parg.ver  ) {
		sbp->replyref(true);
		return;
	}

	//Try and find the acked version so we can commit and error if not found
	pendit = kit->second.pending_list.find(parg.ver);
	if(pendit == kit->second.pending_list.end()) {
		sbp->replyref(false);
		return;
	}

	twait { storage->set(id, &pendit->second, mkevent(set_succ)); }

	if(!set_succ) {
		sbp->replyref(false);
		return;
	}

	//Send replies for head writes that we just acked before erasing
	for(it = kit->second.write_reqs.begin(); it != kit->second.write_reqs.end(); ) {
		for(repls = it->second.begin(); repls != it->second.end(); repls++) {
			LOG_WARN << "Replying to write request\n";
			(*repls)->replyref(true);

			gettimeofday(&cur_time, NULL);
			LOG_ALERT << "WRITE_DONE\t" << cur_time.tv_sec << "\t" << cur_time.tv_usec << "\n";

		}
		it->second.clear();
		kit->second.write_reqs.erase(it++);
		if(it->first > parg.ver) break;
	}

	//Update committed version number
	kit->second.committed = parg.ver;

	LOG_WARN << "directly before pending list erase";

	//Erase all pending versions less than one just committed
	pendit = kit->second.pending_list.find(parg.ver);
	if(pendit != kit->second.pending_list.end())
		kit->second.pending_list.erase(kit->second.pending_list.begin(), pendit);

	LOG_WARN << "Updated key " << id.toString().c_str() << " to "
		 << kit->second.committed << "/" << kit->second.max_pending << "\n";

	sbp->replyref(true);
	//if(!kit->second.is_head) {
		twait { ack(chain_id, id, mkevent(ret_val)); }
	//}

}

tamed void process_add_chain(svccb * sbp) {
	tvars {
		add_chain_arg parg;
		ID_Value id;
		u_int chain_size;
		ostringstream ss;
		int i;
		string node_id;
		string node_val;
		int ret;
	}

	parg = *(sbp->getarg<add_chain_arg>());
	LOG_WARN << "Got ADD_CHAIN Request\n";

	id.set_from_rpc(parg.id);
	chain_size = parg.chain_size;

	if(chain_size < 1 || parg.data_centers.size() < 1) {
		sbp->replyref(ADD_CHAIN_FAILURE);
		return;
	}

	ss << chain_size;
	for(i=0; i<parg.data_centers.size(); i++) {
		ss << " " << parg.data_centers[i];
	}

	node_id = id.toString();
	node_val = ss.str();

	twait { czoo_create( "/keys/" + node_id, node_val, &ZOO_OPEN_ACL_UNSAFE, 0, mkevent(ret)); }
	if(ret == ZNODEEXISTS) {
		sbp->replyref(ADD_CHAIN_EXISTS);
		return;
	} else if(!(ret == ZOK)) {
		sbp->replyref(ADD_CHAIN_FAILURE);
		return;
	}

	sbp->replyref(ADD_CHAIN_SUCCESS);

}

tamed void propagate(ID_Value chain_id, ID_Value id, bool send_committed, cbb cb) {
	tvars {
		key_iter it;
		ring_iter succs;
		Node succ;
		map<int, blob>::iterator dt_it;
		propagate_arg arg;
		ptr<aclnt> cli;
		clnt_stat e;
		int fd;
		bool ret;
		bool rpc_ret;
		ptr<blob> get_result;
		u_int backoff;
		ptr<chain_meta> chain_info;
		ptr<Node> ext_succ;
	}

	twait{ get_chain_info(chain_id, mkevent(chain_info)); }
	if(chain_info == NULL) {
		LOG_FATAL << "Couldn't get chain info in propagate func!\n";
		TRIGGER(cb, false);
		return;
	}

	backoff = 0;
	rpc_ret = false;
	while(!rpc_ret) {
		it = key_meta_list.find(id);
		if(it == key_meta_list.end()) {
			TRIGGER(cb, false);
			return;
		}

		//if tail of last data center, done
		if(it->second.is_tail &&
				chain_info->data_centers[chain_info->data_centers.size()-1] == datacenter) {
			TRIGGER(cb, true);
			return;
		}
		//if just tail of this data center, we have to go to next one
		else if(it->second.is_tail) {
			twait { ext_ring_succ(*chain_info, id, mkevent(ext_succ)); }
			if(ext_succ == NULL) {
				LOG_FATAL << "Error when trying to retrieve external successor!\n";
			}
			succ = *ext_succ;
		} else {
			succs = my_node_ptr;
			ring_incr(&succs);
			succ = succs->second;
		}

		if(send_committed) {
			arg.id = id.get_rpc_id();
			arg.chain = chain_id.get_rpc_id();
			arg.ver = it->second.committed;
			twait { storage->get(id, mkevent(get_result)); }
			arg.data = *get_result;
			arg.committed = true;
		} else {
			dt_it = it->second.pending_list.find(it->second.max_pending);
			if(dt_it == it->second.pending_list.end()) {
				TRIGGER(cb, false);
				return;
			}
			arg.id = id.get_rpc_id();
			arg.chain = chain_id.get_rpc_id();
			arg.ver = it->second.max_pending;
			arg.data = dt_it->second;
			arg.committed = false;
		}

		LOG_WARN << "Propagating ID " << id.toString().c_str() << " to neighbor " << succ.toString().c_str() << "\n";
		twait { get_rpc_cli (succ.getIp().c_str(), succ.getPort(), &cli, &chain_node_1, mkevent(fd)); }

		if( fd<0 ) {
			report_bad_node(succ);
			backoff++;
			twait { delaycb (0, (u_int32_t) (.5 * 1000000000 * backoff), mkevent ()); }
			continue;
		}

		LOG_WARN << "Propagating key of size " << arg.data.size() << "\n";
		twait {	cli->call(PROPAGATE, &arg, &rpc_ret,  mkevent(e)); }
		if(e) {
			LOG_WARN << "Error propagating key\n";
			report_bad_node(succ);
			backoff++;
			twait { delaycb (0, (u_int32_t) (.5 * 1000000000 * backoff), mkevent ()); }
			continue;
		} else if(!rpc_ret) {
			LOG_WARN << "Bad return value from propogating key\n";
			backoff++;
			twait { delaycb (0, (u_int32_t) (.5 * 1000000000 * backoff), mkevent ()); }
			continue;
		}

	}

	TRIGGER(cb, true);
	return;

}

tamed void back_propagate(ID_Value chain_id, ID_Value id, bool send_committed, cbb cb) {
	tvars {
		key_iter it;
		ring_iter pred;
		map<int, blob>::iterator dt_it;
		propagate_arg arg;
		ptr<aclnt> cli;
		clnt_stat e;
		int fd;
		bool ret;
		bool rpc_ret;
		ptr<blob> get_result;
		const blob * st_val;
		u_int backoff;
	}

	backoff = 0;
	rpc_ret = false;
	while(!rpc_ret) {
		it = key_meta_list.find(id);
		if(it == key_meta_list.end()) {
			TRIGGER(cb, true);
			return;
		}

		if(it->second.is_head) {
			TRIGGER(cb, true);
			return;
		}

		pred = my_node_ptr;
		ring_decr(&pred);

		if(send_committed) {
			if(it->second.committed <= 0) {
				//There is no committed version, so die
				TRIGGER(cb, true);
				return;
			}
			arg.chain = chain_id.get_rpc_id();
			arg.id = id.get_rpc_id();
			arg.ver = it->second.committed;
			twait { storage->get(id, mkevent(get_result)); }
			st_val = get_result;
			if(!st_val) {
				LOG_FATAL << "Couldn't get value from storage " << id.toString().c_str() << "! Dying...\n";
			}
			arg.data = *st_val;
			arg.committed = true;
		} else {
			dt_it = it->second.pending_list.find(it->second.max_pending);
			if(dt_it == it->second.pending_list.end()) {
				TRIGGER(cb, false);
				return;
			}
			arg.chain = chain_id.get_rpc_id();
			arg.id = id.get_rpc_id();
			arg.ver = it->second.max_pending;
			arg.data = dt_it->second;
			arg.committed = false;
		}

		LOG_WARN << "Back Propagating ID " << id.toString().c_str() << " to neighbor " << pred->second.toString().c_str() << "\n";
		twait { get_rpc_cli (pred->second.getIp().c_str(), pred->second.getPort(), &cli, &chain_node_1, mkevent(fd)); }

		if( fd<0 ) {
			report_bad_node(pred->second);
			backoff++;
			twait { delaycb (0, (u_int32_t) (.5 * 1000000000 * backoff), mkevent ()); }
			continue;
		}

		twait {	cli->call(BACK_PROPAGATE, &arg, &rpc_ret,  mkevent(e)); }
		if(e || !rpc_ret) {
			report_bad_node(pred->second);
			backoff++;
			twait { delaycb (0, (u_int32_t) (.5 * 1000000000 * backoff), mkevent ()); }
			continue;
		}

	}

	TRIGGER(cb, true);
	return;

}

tamed void ack(ID_Value chain_id, ID_Value id, cbb cb) {
	tvars {
		key_iter it;
		ring_iter predi;
		Node pred;
		ack_arg arg;
		ptr<aclnt> cli;
		clnt_stat e;
		int fd;
		bool ret;
		bool rpc_ret;
		u_int backoff;
		ptr<chain_meta> chain_info;
		ptr<Node> ext_pred;
	}

	twait{ get_chain_info(chain_id, mkevent(chain_info)); }
	if(chain_info == NULL) {
		TRIGGER(cb, false);
		return;
	}

	backoff = 0;
	rpc_ret = false;
	while(!rpc_ret) {
		it = key_meta_list.find(id);
		if(it == key_meta_list.end()) {
			TRIGGER(cb, false);
			return;
		}

		//if head of first data center, done
		if(it->second.is_head &&
				chain_info->data_centers[0] == datacenter) {
			TRIGGER(cb, true);
			return;
		}
		//if just head of this data center, we have to go to prev one
		else if(it->second.is_head) {
			twait { ext_ring_pred(*chain_info, id, mkevent(ext_pred)); }
			if(ext_pred == NULL) {
				LOG_FATAL << "Error when trying to retrieve external predecessor!\n";
			}
			pred = *ext_pred;
		} else {
			predi = my_node_ptr;
			ring_decr(&predi);
			pred = predi->second;
		}

		arg.chain = chain_id.get_rpc_id();
		arg.id = id.get_rpc_id();
		arg.ver = it->second.committed;

		LOG_WARN << "ACKing ID " << id.toString().c_str() << " to neighbor " << pred.toString().c_str() << "\n";
		twait { get_rpc_cli (pred.getIp().c_str(), pred.getPort(), &cli, &chain_node_1, mkevent(fd)); }

		if( fd<0 ) {
			report_bad_node(pred);
			backoff++;
			twait { delaycb (0, 50 * 1000000 * backoff, mkevent ()); }
			continue;
		}

		twait {	cli->call(ACK, &arg, &rpc_ret,  mkevent(e)); }
		if(e) {
			report_bad_node(pred);
			backoff++;
			twait { delaycb (0, 50 * 1000000 * backoff, mkevent ()); }
			continue;
		} else if(!rpc_ret) {
			continue;
			backoff++;
			twait { delaycb (0, 50 * 1000000 * backoff, mkevent ()); }
		}

	}

	TRIGGER(cb, true);
	return;

}

void update_my_ptr() {
	my_node_ptr = ring.find(my_id);
	if(my_node_ptr == ring.end()) {
		LOG_FATAL << "Couldn't find my own ID in the node list! Killing myself...\n";
	}
}

void rpc_server::dispatch(svccb * sbp) {
	if(!sbp){}

	u_int p = sbp->proc();
	switch(p) {
		case TAIL_READ:
			process_tail_read(sbp);
			break;
		case TAIL_READ_EX:
			process_tail_read_ex(sbp);
			break;
 		case HEAD_WRITE:
 			process_head_write(sbp);
 			break;
 		case TEST_AND_SET:
 			process_test_and_set(sbp);
 			break;
 		case PROPAGATE:
 			process_propagate(sbp);
 			break;
 		case QUERY_OBJ_VER:
 			process_query_obj_ver(sbp);
 			break;
 		case ACK:
 			process_ack(sbp);
 			break;
 		case BACK_PROPAGATE:
 			process_back_propagate(sbp);
 			break;
 		case NO_OP:
 			sbp->replyref(true);
 			break;
 		case ADD_CHAIN:
 			process_add_chain(sbp);
 			break;
		default: {
			sbp->reject(PROC_UNAVAIL);
			break;
		}
	}
	LOG_INFO << "end of rpc_server::dispatch";
}

ring_iter ring_succ(ID_Value id) {
	ring_iter it = ring.lower_bound(id);
	if(it == ring.end())
		it = ring.begin();
	return it;
}

void ring_incr(ring_iter * it) {
	(*it)++;
	if( (*it)==ring.end() ) {
		(*it) = ring.begin();
	}
}

void ring_decr(ring_iter * it) {
	if( (*it)==ring.begin() ) {
		(*it) = ring.end();
	}
	(*it)--;
}

tamed void report_bad_node(Node n) {
	/*tvars {
		int fd;
		ptr<aclnt> cli;
		clnt_stat e;
		rpc_node arg;
	}*/

	/*LOG_WARN << "Reporting bad node " << n.toString().c_str() << "\n";
	invalidate_rpc_host(n.getIp().c_str(), n.getPort());

	twait { get_rpc_cli (manager_hostname, manager_port, &cli, &rpc_manager_1, mkevent(fd)); }

	if( fd<0 ) {
		LOG_FATAL << "Lost connection to manager! Dying...\n";
	} else {
		arg = n.get_rpc_node();
		twait {	cli->call(REPORT_BAD, &arg, NULL, mkevent(e)); }
		if(e) {
			LOG_FATAL << "Problem communicating with manager! Dying...\n";
		}
	}

	twait { get_updates(); }*/

}

tamed static void start_rpc_srv(int listen_port) {
	tvars {
		bool ret;
		rpc_server_factory fact;
	}
	twait {
		fact.run(listen_port, mkevent(ret));
	}
	LOG_WARN << "Exiting after RPC stopped. Return value was: " << ret << "\n";
	exit(ret);
}

tamed void node_added(Node node_changed) {
	tvars {
		ring_iter succ;
		ring_iter pred;
		ring_iter cs_head;
		ring_iter cs_head_pred;
		key_iter k;
		key_iter temp;
		bool ret;
		u_int i;
	}

	LOG_WARN << "Node added: " << node_changed.toString().c_str() << "\n";

	succ = my_node_ptr;
	ring_incr(&succ);
	pred = my_node_ptr;
	ring_decr(&pred);

	ring[node_changed.getId()] = node_changed;
	update_my_ptr();

	LOG_WARN << "Checking if node " << node_changed.getId().toString().c_str()
		 << " is between me " << my_id.toString().c_str()
		 << " and my succ " << succ->first.toString().c_str() << "\n";

	//Check if successor and propagate all keys
	if(node_changed.getId().between(my_id, succ->first)) {
		twait {
			for(k = key_meta_list.begin(); k != key_meta_list.end(); k++) {
				propagate(k->second.chain_id, k->first, false, mkevent(ret));
				propagate(k->second.chain_id, k->first, true, mkevent(ret));
			}
		}
		return;
	}

	cs_head = my_node_ptr;
	for(i=0; i<(CHAIN_SIZE-1); i++) {
		ring_decr(&cs_head);
	}
	cs_head_pred = cs_head;
	ring_decr(&cs_head_pred);

	LOG_WARN << "Checking if node " << node_changed.getId().toString().c_str()
		 << " is between pred " << pred->first.toString().c_str()
		 << " and me " << my_id.toString().c_str() << "\n";

	//Check if predecessor and back propagate all keys that i'm not the head for
	//   For keys that now belong to new guy, mark not head
	//	 For keys I was tail for and new guy is now tail, mark
	//	 For keys that I should become the tail, become it
	if(node_changed.getId().between(pred->first, my_id)) {
		twait {
			//Fire off back propagates for all keys that im not still head for
			for(k = key_meta_list.begin(); k != key_meta_list.end(); ) {
				//Only look at keys for which we are not STILL the head
				if(!k->first.between(node_changed.getId(), my_id)) {
					//We were the head, but now the new guy is
					if(k->second.is_head) {
						LOG_WARN << "No longer head for " << k->first.toString().c_str() << "\n";
						k->second.is_head = false;
					}

					back_propagate(k->second.chain_id, k->first, false, mkevent(ret));
					back_propagate(k->second.chain_id, k->first, true, mkevent(ret));

					if(k->second.is_tail) {
						//We are no longer tail so remove
						LOG_WARN << "Removing key " << k->first.toString().c_str() << "\n";
						key_meta_list.erase(k++);
					} else {
						//Check if we need to become the tail
						if(k->first.between(cs_head_pred->first, cs_head->first)) {
							LOG_WARN << "Becoming tail for " << k->first.toString().c_str() << "\n";
							k->second.is_tail = true;
						}
						k++;
					}

				} else {
					k++;
				}
			}
		}
		return;
	}

	LOG_WARN << "Checking if node " << node_changed.getId().toString().c_str()
		 << " is between first " << cs_head->first.toString().c_str()
		 << " and me " << my_id.toString().c_str() << "\n";

	//Check if in chain-size predecessor list and remove tail keys
	//    Since our predecessor is now the tail
	//	Also, we might have to become the new tail
	//    so any keys between predpred and pred i should be the tail for
	if(node_changed.getId().betweenIncl(cs_head->first, my_id)) {
		for(k = key_meta_list.begin(); k != key_meta_list.end(); ) {
			if(k->second.is_tail) {
				LOG_WARN << "Removing key " << k->first.toString().c_str() << "\n";
				key_meta_list.erase(k++);
			} else {
				if(k->first.between(cs_head_pred->first, cs_head->first)) {
					LOG_WARN << "Becoming tail for " << k->first.toString().c_str() << "\n";
					k->second.is_tail = true;
				}
				k++;
			}
		}
	}

}

tamed void node_deleted(Node node_changed) {

	tvars {
		ring_iter succ;
		ring_iter pred;
		ring_iter cs_head;
		ring_iter cs_head_pred;
		ring_iter it;
		key_iter k;
		key_iter temp;
		bool ret;
		u_int i;
		ptr<chain_meta> chain_info;
	}

	LOG_WARN << "Node deleted: " << node_changed.toString().c_str() << "\n";

	it = ring.find(node_changed.getId());
	if(it == ring.end()) {
		LOG_FATAL << "Deleting node that we didn't know about! Should never happen... dying!\n";
	}
	invalidate_rpc_host(it->second.getIp().c_str(), it->second.getPort());
	ring.erase(it);
	update_my_ptr();

	succ = my_node_ptr;
	ring_incr(&succ);
	pred = my_node_ptr;
	ring_decr(&pred);

	LOG_WARN << "Checking if node " << node_changed.getId().toString().c_str()
		 << " is between me " << my_id.toString().c_str()
		 << " and my succ " << succ->first.toString().c_str() << "\n";

	//Check if successor and propagate all keys
	if(node_changed.getId().between(my_id, succ->first)) {
		twait {
			for(k = key_meta_list.begin(); k != key_meta_list.end(); k++) {
				propagate(k->second.chain_id, k->first, false, mkevent(ret));
				propagate(k->second.chain_id, k->first, true, mkevent(ret));
			}
		}
		return;
	}

	LOG_WARN << "Checking if node " << node_changed.getId().toString().c_str()
		 << " is between pred " << pred->first.toString().c_str()
		 << " and me " << my_id.toString().c_str() << "\n";

	//Check if predecessor and back propagate all keys that i'm not the head for
	//	 For keys that I should become the head, become it
	//	 If I am tail, unmark tail and propagate
	//	 Can't search for keys that I should be the tail because I don't know about them
	//           and I will get a propagate from the dead dude's predecessor anyway
	if(node_changed.getId().between(pred->first, my_id)) {
		twait {
			for(k = key_meta_list.begin(); k != key_meta_list.end(); k++) {
				//First, find keys that I should be the head for
				if(k->first.between(pred->first, my_id)) {
					if(k->second.is_head == false) {
						LOG_WARN << "Becoming head for " << k->first.toString().c_str() << "\n";
						k->second.is_head = true;
					}
				//For all other keys, we should back propagate
				} else {
					if(k->second.is_tail) {
						LOG_WARN << "No longer tail for " << k->first.toString().c_str() << "\n";
						k->second.is_tail = false;
						propagate(k->second.chain_id, k->first, false, mkevent(ret));
						propagate(k->second.chain_id, k->first, true, mkevent(ret));
					}
					back_propagate(k->second.chain_id, k->first, false, mkevent(ret));
					back_propagate(k->second.chain_id, k->first, true, mkevent(ret));
				}
			}
		}
		return;
	}



	LOG_WARN << "Checking if node " << node_changed.getId().toString().c_str()
		 << " is between first " /*<< cs_head->first.toString().c_str()*/
		 << " and me " << my_id.toString().c_str() << "\n";

	//Check if in chain-size predecessor list and unmark tail and propagate
	for(k = key_meta_list.begin(); k != key_meta_list.end(); k++) {
		if(k->second.is_tail) {


			twait{ get_chain_info(k->second.chain_id, mkevent(chain_info)); }
			if(chain_info == NULL) {
				LOG_FATAL << "Couldn't get chain info in node_deleted\n";
			}

			cs_head = my_node_ptr;
			for(i=0; i<(chain_info->chain_size-1); i++) {
				ring_decr(&cs_head);
			}
			cs_head_pred = cs_head;
			ring_decr(&cs_head_pred);

			if(node_changed.getId().betweenIncl(cs_head->first, my_id)) {

				LOG_WARN << "No longer tail for " << k->first.toString().c_str() << "\n";
				k->second.is_tail = false;
				propagate(k->second.chain_id, k->first, false, wrap(dont_care));
				propagate(k->second.chain_id, k->first, true, wrap(dont_care));

			}

		}
	}

}

tamed static void
node_list_watcher(string path) {

	tvars {
		vector<string> * ret_node_list;
		set<string> new_list;
		int i, j;
		map<string, Node>::iterator old_it;
		set<string>::iterator new_it;
		set<string> to_add;
		set<string>::iterator it;
		vector<string *> add_rets;
		vector<string> add_ids;
		rendezvous_t<int> rv;
		string search;
		string * new_val;
		Node new_node;
	}

    if(!ring_init) {
        init_interrupted = true;
    	LOG_FATAL << "Updated node list while doing initial list. Re-building initial list.\n";
        return;
    }

	twait { czoo_get_children("/nodes/" + datacenter, &node_list_watcher, mkevent(ret_node_list)); }
	if(ret_node_list == NULL) {
		LOG_FATAL << "Error retrieving updated node list!\n";
	}
	for(i=0; i<ret_node_list->size(); i++) {
		new_list.insert( (*ret_node_list)[i] );
	}
	delete ret_node_list;

	old_it = zoo_nodes.begin();
	new_it = new_list.begin();

	while(old_it != zoo_nodes.end() || new_it != new_list.end()) {
		if(old_it == zoo_nodes.end()) {
			to_add.insert(*new_it);
			new_it++;
		} else if( new_it == new_list.end() ) {
			node_deleted(old_it->second);
			zoo_nodes.erase(old_it++);
		}
		else if( old_it->first == *new_it ) {
			old_it++;
			new_it++;
		} else if( old_it->first < *new_it ) {
			node_deleted(old_it->second);
			zoo_nodes.erase(old_it++);
		} else if( old_it->first > *new_it ) {
			to_add.insert(*new_it);
			new_it++;
		}
	}

	add_ids.resize(to_add.size());
	add_rets.resize(to_add.size());
	for( i=0, it = to_add.begin(); it != to_add.end(); i++, it++ ) {
		search = "/nodes/" + datacenter + "/" + (*it);
		add_ids[i] = *it;
        czoo_get(search, mkevent(rv, i, add_rets[i]));
	}
	for(i=0; i<add_rets.size(); i++) {
		twait(rv, j);
		if(add_rets[j] == NULL) {
			LOG_FATAL << "Failed to retrieve information about a node!\n";
		}
		new_node.set_from_string(*add_rets[j]);
		delete add_rets[j];
		zoo_nodes[add_ids[j]] = new_node;
		node_added(new_node);
	}

}

tamed static void
register_to_manager(int my_port, string zoo_list) {
	tvars {
		int fd,r;
		ptr<aclnt> cli;
		str ip;
		clnt_stat e, e2;
		ostringstream ss;
		unsigned int i;
		string my_ip;
		int ret;
		bool rc;
		vector<string> * node_list;
		string find;
		string search;
		string * found;
		vector<string *> node_vals;
		Node new_node;
	}

	ip = get_ip_address();
	srand ( time(NULL) );
	r = rand();
	ss << ip.cstr() << ":" << my_port << r;
	my_id = get_sha1(ss.str());
	LOG_WARN << "my id = " << my_id.toString().c_str() << "\n";

	my_ip = ip;
	my_node.setIp(my_ip);
	my_node.setId(my_id);
	my_node.setPort(my_port);

	ss.str("");
	ss << my_ip << " " << my_port << " " << my_id.toString();
	my_node_str = ss.str();

	twait { czoo_init( zoo_list.c_str(), mkevent(rc)); }
	if(!rc) {
		LOG_FATAL << "Couldn't connect to manager!\n";
	}

	twait { czoo_create( "/nodes", "", &ZOO_OPEN_ACL_UNSAFE, 0, mkevent(ret)); }
	if( !(ret == ZOK || ret == ZNODEEXISTS) ) {
		LOG_FATAL << "Error(" << ret << ") when trying to create root node!\n";
	}

	twait { czoo_create( "/keys", "", &ZOO_OPEN_ACL_UNSAFE, 0, mkevent(ret)); }
	if(!(ret == ZOK || ret == ZNODEEXISTS)) {
		LOG_FATAL << "Error(" << ret << ") when trying to create keys node!\n";
	}

	twait { czoo_create( "/nodes/" + datacenter, "", &ZOO_OPEN_ACL_UNSAFE, 0, mkevent(ret)); }
	if(!(ret == ZOK || ret == ZNODEEXISTS)) {
		LOG_FATAL << "Error(" << ret << ") when trying to create root data center node!\n";
	}

	twait { czoo_create( "/nodes/" + datacenter + "/node", my_node_str, &ZOO_OPEN_ACL_UNSAFE,
							ZOO_EPHEMERAL | ZOO_SEQUENCE, mkevent(ret)); }
	if(!(ret == ZOK)) {
		LOG_FATAL << "Error(" << ret << ") when trying to create my node!\n";
	}

	//LOG_WARN << "sleeping " << ret << "\n";
	//delay(5000);
	//LOG_WARN << "done sleeping\n";

        // Allows us to re-build list if nodes change as we're building it
	while(true) {
	    twait { czoo_get_children("/nodes/" + datacenter, &node_list_watcher, mkevent(node_list)); }
	    if(node_list == NULL) {
		    fatal << "Error retrieving initial node list!\n";
	    }
	    zoo_node_count = (*node_list).size();
	    node_vals.resize((*node_list).size());
	    //LOG_WARN << "size " << (*node_list).size() << "\n";
	    twait {
		    for(i=0; i<(*node_list).size(); i++) {
			    find = (*node_list)[i];
			    search = "/nodes/" + datacenter + "/" + find;
			    czoo_get(search, mkevent(node_vals[i]));
		    }
	    }
            if (init_interrupted) {
              init_interrupted = false;
              continue;
            }
	    for(i=0; i<node_vals.size(); i++) {
		    if(node_vals[i] == NULL) {
			    fatal << "Error occurred retrieving initial node value!\n";
		    }
		    new_node.set_from_string(*node_vals[i]);
		    ring[new_node.getId()] = new_node;
		    zoo_nodes[(*node_list)[i]] = new_node;
		    LOG_WARN << (*node_list)[i].c_str() << " - " << new_node.toString().c_str() << "\n";
		    delete node_vals[i];
	    }

	    delete node_list;
		if (!init_interrupted) break;
	}

	update_my_ptr();
	ring_init = true;

}

//set up log4cpp for logging purposes
static void log4cpp_init(string file, string priority) {

	//below is based on example code at http://developers.sun.com/solaris/articles/logging.html

	// instantiate an appender object that
    // will append to a log file
	app = new log4cpp::FileAppender("FileAppender", file);

    // instantiate a layout object
	// Two layouts come already available in log4cpp
	// unless you create your own.
	// BasicLayout includes a time stamp
	log4cpp::Layout *layout = new log4cpp::BasicLayout();

	// 3. attach the layout object to the
	// appender object
	app->setLayout(layout);

	// 5. Step 1
	// an Appender when added to a category becomes
	// an additional output destination unless
	// Additivity is set to false when it is false,
	// the appender added to the category replaces
	// all previously existing appenders
    LOG.setAdditivity(false);

	// 5. Step 2
    // this appender becomes the only one
	LOG.setAppender(app);

	// 6. Set up the priority for the category
	if (priority == "DEBUG") {
		LOG.setPriority(log4cpp::Priority::DEBUG);
	} else if (priority == "INFO") {
		LOG.setPriority(log4cpp::Priority::INFO);
	} else if (priority == "NOTICE") {
		LOG.setPriority(log4cpp::Priority::NOTICE);
	} else if (priority == "WARN") {
		LOG.setPriority(log4cpp::Priority::WARN);
	} else if (priority == "ERROR") {
		LOG.setPriority(log4cpp::Priority::ERROR);
	} else if (priority == "CRIT") {
		LOG.setPriority(log4cpp::Priority::CRIT);
	} else if (priority == "ALERT") {
		LOG.setPriority(log4cpp::Priority::ALERT);
	} else if (priority == "FATAL" || priority == "EMERG") {
		LOG.setPriority(log4cpp::Priority::FATAL);
	} else {
		LOG.setPriority(log4cpp::Priority::NOTSET);
	}

}

tamed static
void main2(int argc, char **argv) {
	int listen_port;
	string zookeeper_list;
	string log_file;
	string log_priority;
	string s_storage;
	int num_hex_chars;
	int lighttpd_port;
	str type;

	try
	{
		TCLAP::CmdLine cmd("chain_node starts up a single chain node in CRAQ", ' ', "0.2.1");
		//TCLAP::ValueArg<string> zooKeeperList("z", "zookeeper_list", "List of ZooKeeper nodes (ie '127.0.0.1:2000,10.0.0.1:2100')", true, "", "string", cmd);
		//TCLAP::ValueArg<int> listenPort("p", "listen_port", "Port to listen for connections", true, 0, "int", cmd);

		//for some reason, things break when the line below is commented out, so we'll leave it for now
		TCLAP::ValueArg<string> ipAddress("i", "ip_address", "IP Address to listen for connections", false, "", "string", cmd);
		//TCLAP::SwitchArg crSwitch("c", "chain_replication", "Sets node to CR instead of CRAQ", cmd, true);
		//TCLAP::ValueArg<string> dataCenter("d", "data_center", "Sets the data center name this node belongs to", true, "", "string", cmd);
		TCLAP::UnlabeledValueArg<string> configFile("config_file", "Configuration file from which to read initialization parameters", true, "", "string", cmd);
		cmd.parse(argc, argv);

		libconfig::Config cfg;
		cfg.readFile(configFile.getValue().c_str());

		string string_user_ip;

		cfg.lookupValue("node.zookeeper_list", zookeeper_list);
		//zookeeper_list = zooKeeperList.getValue();

		cfg.lookupValue("node.port", listen_port);
		//listen_port = listenPort.getValue();

		cfg.lookupValue("node.ip", string_user_ip);
		user_ip = string_user_ip.c_str();
		//user_ip = ipAddress.getValue().c_str();

		//datacenter = dataCenter.getValue();
		cfg.lookupValue("node.datacenter", datacenter);

		start_rpc_srv(listen_port);
		register_to_manager(listen_port, zookeeper_list);

		cfg.lookupValue("node.storage", s_storage);

		cfg.lookupValue("node.disk_folder_chars", num_hex_chars);

		cfg.lookupValue("node.lighttpd_port", lighttpd_port);

		//set up logging
		cfg.lookupValue("logging.file", log_file);
		cfg.lookupValue("logging.min_priority", log_priority);
		log4cpp_init(log_file, log_priority);
		LOG_DEBUG << "logging set up";

	}
	catch (TCLAP::ArgException &e)  // catch any exceptions
	{
		fatal << "error: " << e.error().c_str() << " for arg " << e.argId().c_str() << "\n";
	}
	catch (libconfig::ParseException &e) {
		fatal << "An error occured when parsing configuration file!\n"
					<< "Error: " << e.getError() << "\n"
					<< "File: " << e.getFile() << "\n"
					<< "Line Number: " << e.getLine() << "\n";
	}

	//TODO make number of hex characters configurable
	LOG_INFO << "storage type is " << s_storage;
	if (s_storage == "DISK") {
		LOG_INFO << "num_hex_chars is: " << num_hex_chars;
		storage = new DiskStorage(app, num_hex_chars);
	} else if (s_storage == "MEMORY") {
		storage = new MemStorage(app);
	} else if (s_storage == "HTTP") {
		storage = new HttpStorage(app, lighttpd_port);
	} else {
		LOG_ERROR << "unexpected storage parameter: " << s_storage << ", defaulting to memory storage";
		storage = new MemStorage(app);
	}

}

int main (int argc, char *argv[]) {
	main2(argc, argv);
	amain ();
}