Cluster work -- integrating with the rest of the code.

include/ZeroTierOne.h

@@ -128,6 +128,16 @@ extern "C" {
  */
 #define ZT_CIRCUIT_TEST_MAX_HOP_BREADTH 256
 
+/**
+ * Maximum number of cluster members (and max member ID plus one)
+ */
+#define ZT_CLUSTER_MAX_MEMBERS 256
+
+/**
+ * Maximum allowed cluster message length in bytes
+ */
+#define ZT_CLUSTER_MAX_MESSAGE_LENGTH 65535
+
 /**
  * A null/empty sockaddr (all zero) to signify an unspecified socket address
  */
@@ -174,7 +184,17 @@ enum ZT_ResultCode
 	/**
 	 * Network ID not valid
 	 */
-	ZT_RESULT_ERROR_NETWORK_NOT_FOUND = 1000
+	ZT_RESULT_ERROR_NETWORK_NOT_FOUND = 1000,
+
+	/**
+	 * The requested operation is not supported on this version or build
+	 */
+	ZT_RESULT_ERROR_UNSUPPORTED_OPERATION = 1001,
+
+	/**
+	 * The requestion operation was given a bad parameter or was called in an invalid state
+	 */
+	ZT_RESULT_ERROR_BAD_PARAMETER = 1002
 };
 
 /**
@@ -1320,6 +1340,105 @@ enum ZT_ResultCode ZT_Node_circuitTestBegin(ZT_Node *node,ZT_CircuitTest *test,v
  */
 void ZT_Node_circuitTestEnd(ZT_Node *node,ZT_CircuitTest *test);
 
+/**
+ * Initialize cluster operation
+ *
+ * This initializes the internal structures and state for cluster operation.
+ * It takes two function pointers. The first is to a function that can be
+ * used to send data to cluster peers (mechanism is not defined by Node),
+ * and the second is to a function that can be used to get the location of
+ * a physical address in X,Y,Z coordinate space (e.g. as cartesian coordinates
+ * projected from the center of the Earth).
+ *
+ * Send function takes an arbitrary pointer followed by the cluster member ID
+ * to send data to, a pointer to the data, and the length of the data. The
+ * maximum message length is ZT_CLUSTER_MAX_MESSAGE_LENGTH (65535). Messages
+ * must be delivered whole and may be dropped or transposed, though high
+ * failure rates are undesirable and can cause problems. Validity checking or
+ * CRC is also not required since the Node validates the authenticity of
+ * cluster messages using cryptogrphic methods and will silently drop invalid
+ * messages.
+ *
+ * Address to location function is optional and if NULL geo-handoff is not
+ * enabled (in this case x, y, and z in clusterInit are also unused). It
+ * takes an arbitrary pointer followed by a physical address and three result
+ * parameters for x, y, and z. It returns zero on failure or nonzero if these
+ * three coordinates have been set. Coordinate space is arbitrary and can be
+ * e.g. coordinates on Earth relative to Earth's center. These can be obtained
+ * from latitutde and longitude with versions of the Haversine formula.
+ *
+ * See: http://stackoverflow.com/questions/1185408/converting-from-longitude-latitude-to-cartesian-coordinates
+ *
+ * Neither the send nor the address to location function should block. If the
+ * address to location function does not have a location for an address, it
+ * should return zero and then look up the address for future use since it
+ * will be called again in (typically) 1-3 minutes.
+ *
+ * Note that both functions can be called from any thread from which the
+ * various Node functions are called, and so must be thread safe if multiple
+ * threads are being used.
+ *
+ * @param node Node instance
+ * @param myId My cluster member ID (less than or equal to ZT_CLUSTER_MAX_MEMBERS)
+ * @param zeroTierPhysicalEndpoints Preferred physical address(es) for ZeroTier clients to contact this cluster member (for peer redirect)
+ * @param numZeroTierPhysicalEndpoints Number of physical endpoints in zeroTierPhysicalEndpoints[] (max allowed: 255)
+ * @param x My cluster member's X location
+ * @param y My cluster member's Y location
+ * @param z My cluster member's Z location
+ * @param sendFunction Function to be called to send data to other cluster members
+ * @param sendFunctionArg First argument to sendFunction()
+ * @param addressToLocationFunction Function to be called to get the location of a physical address or NULL to disable geo-handoff
+ * @param addressToLocationFunctionArg First argument to addressToLocationFunction()
+ * @return OK or UNSUPPORTED_OPERATION if this Node was not built with cluster support
+ */
+enum ZT_ResultCode ZT_Node_clusterInit(
+	ZT_Node *node,
+	unsigned int myId,
+	const struct sockaddr_storage *zeroTierPhysicalEndpoints,
+	unsigned int numZeroTierPhysicalEndpoints,
+	int x,
+	int y,
+	int z,
+	void (*sendFunction)(void *,unsigned int,const void *,unsigned int),
+	void *sendFunctionArg,
+	int (*addressToLocationFunction)(void *,const struct sockaddr_storage *,int *,int *,int *),
+	void *addressToLocationFunctionArg);
+
+/**
+ * Add a member to this cluster
+ *
+ * Calling this without having called clusterInit() will do nothing.
+ *
+ * @param node Node instance
+ * @param memberId Member ID (must be less than or equal to ZT_CLUSTER_MAX_MEMBERS)
+ * @return OK or error if clustering is disabled, ID invalid, etc.
+ */
+enum ZT_ResultCode ZT_Node_clusterAddMember(ZT_Node *node,unsigned int memberId);
+
+/**
+ * Remove a member from this cluster
+ *
+ * Calling this without having called clusterInit() will do nothing.
+ *
+ * @param node Node instance
+ * @param memberId Member ID to remove (nothing happens if not present)
+ */
+void ZT_Node_clusterRemoveMember(ZT_Node *node,unsigned int memberId);
+
+/**
+ * Handle an incoming cluster state message
+ *
+ * The message itself contains cluster member IDs, and invalid or badly
+ * addressed messages will be silently discarded.
+ *
+ * Calling this without having called clusterInit() will do nothing.
+ *
+ * @param node Node instance
+ * @param msg Cluster message
+ * @param len Length of cluster message
+ */
+void ZT_Node_clusterHandleIncomingMessage(ZT_Node *node,const void *msg,unsigned int len);
+
 /**
  * Get ZeroTier One version
  *

make-mac.mk

@@ -6,7 +6,7 @@ ifeq ($(origin CXX),default)
 endif
 
 INCLUDES=
-DEFS=
+DEFS=-DZT_ENABLE_CLUSTER
 LIBS=
 ARCH_FLAGS=-arch x86_64
 

node/Cluster.cpp

@@ -31,10 +31,13 @@
 #include <stdio.h>
 #include <stdlib.h>
 #include <string.h>
+#include <math.h>
 
 #include <algorithm>
 #include <utility>
 
+#include "../version.h"
+
 #include "Cluster.hpp"
 #include "RuntimeEnvironment.hpp"
 #include "MulticastGroup.hpp"
@@ -42,22 +45,44 @@
 #include "Salsa20.hpp"
 #include "Poly1305.hpp"
 #include "Packet.hpp"
+#include "Identity.hpp"
 #include "Peer.hpp"
 #include "Switch.hpp"
 #include "Node.hpp"
 
 namespace ZeroTier {
 
-Cluster::Cluster(const RuntimeEnvironment *renv,uint16_t id,DistanceAlgorithm da,int32_t x,int32_t y,int32_t z,void (*sendFunction)(void *,uint16_t,const void *,unsigned int),void *arg) :
+static inline double _dist3d(int x1,int y1,int z1,int x2,int y2,int z2)
+	throw()
+{
+	double dx = ((double)x2 - (double)x1);
+	double dy = ((double)y2 - (double)y1);
+	double dz = ((double)z2 - (double)z1);
+	return sqrt((dx * dx) + (dy * dy) + (dz * dz));
+}
+
+Cluster::Cluster(
+	const RuntimeEnvironment *renv,
+	uint16_t id,
+	const std::vector<InetAddress> &zeroTierPhysicalEndpoints,
+	int32_t x,
+	int32_t y,
+	int32_t z,
+	void (*sendFunction)(void *,unsigned int,const void *,unsigned int),
+	void *sendFunctionArg,
+	int (*addressToLocationFunction)(void *,const struct sockaddr_storage *,int *,int *,int *),
+	void *addressToLocationFunctionArg) :
 	RR(renv),
 	_sendFunction(sendFunction),
-	_arg(arg),
+	_sendFunctionArg(sendFunctionArg),
+	_addressToLocationFunction(addressToLocationFunction),
+	_addressToLocationFunctionArg(addressToLocationFunctionArg),
 	_x(x),
 	_y(y),
 	_z(z),
-	_da(da),
 	_id(id),
-	_members(new _Member[65536])
+	_zeroTierPhysicalEndpoints(zeroTierPhysicalEndpoints),
+	_members(new _Member[ZT_CLUSTER_MAX_MEMBERS])
 {
 	uint16_t stmp[ZT_SHA512_DIGEST_LEN / sizeof(uint16_t)];
 
@@ -114,16 +139,20 @@ void Cluster::handleIncomingStateMessage(const void *msg,unsigned int len)
 		s20.decrypt12(reinterpret_cast<const char *>(msg) + 24,const_cast<void *>(dmsg.data()),dmsg.size());
 	}
 
-	if (dmsg.size() < 2)
+	if (dmsg.size() < 4)
 		return;
 	const uint16_t fromMemberId = dmsg.at<uint16_t>(0);
 	unsigned int ptr = 2;
+	if (fromMemberId == _id)
+		return;
+	const uint16_t toMemberId = dmsg.at<uint16_t>(ptr);
+	ptr += 2;
+	if (toMemberId != _id)
+		return;
 
 	_Member &m = _members[fromMemberId];
 	Mutex::Lock mlck(m.lock);
 
-	m.lastReceivedFrom = RR->node->now();
-
 	try {
 		while (ptr < dmsg.size()) {
 			const unsigned int mlen = dmsg.at<uint16_t>(ptr); ptr += 2;
@@ -143,11 +172,13 @@ void Cluster::handleIncomingStateMessage(const void *msg,unsigned int len)
 						ptr += 8; // skip local clock, not used
 						m.load = dmsg.at<uint64_t>(ptr); ptr += 8;
 						ptr += 8; // skip flags, unused
-						m.physicalAddressCount = dmsg[ptr++];
-						if (m.physicalAddressCount > ZT_CLUSTER_MEMBER_MAX_PHYSICAL_ADDRS)
-							m.physicalAddressCount = ZT_CLUSTER_MEMBER_MAX_PHYSICAL_ADDRS;
-						for(unsigned int i=0;i<m.physicalAddressCount;++i)
-							ptr += m.physicalAddresses[i].deserialize(dmsg,ptr);
+						unsigned int physicalAddressCount = dmsg[ptr++];
+						for(unsigned int i=0;i<physicalAddressCount;++i) {
+							m.zeroTierPhysicalEndpoints.push_back(InetAddress());
+							ptr += m.zeroTierPhysicalEndpoints.back().deserialize(dmsg,ptr);
+							if (!(m.zeroTierPhysicalEndpoints.back()))
+								m.zeroTierPhysicalEndpoints.pop_back();
+						}
 						m.lastReceivedAliveAnnouncement = RR->node->now();
 					}	break;
 
@@ -298,7 +329,7 @@ void Cluster::handleIncomingStateMessage(const void *msg,unsigned int len)
 	}
 }
 
-void Cluster::replicateHavePeer(const Address &peerAddress)
+void Cluster::replicateHavePeer(const Identity &peerId)
 {
 }
 
@@ -312,23 +343,59 @@ void Cluster::replicateCertificateOfNetworkMembership(const CertificateOfMembers
 
 void Cluster::doPeriodicTasks()
 {
-	// Go ahead and flush whenever possible right now
+	const uint64_t now = RR->node->now();
+
 	{
 		Mutex::Lock _l(_memberIds_m);
 		for(std::vector<uint16_t>::const_iterator mid(_memberIds.begin());mid!=_memberIds.end();++mid) {
 			Mutex::Lock _l2(_members[*mid].lock);
-			_flush(*mid);
+
+			if ((now - _members[*mid].lastAnnouncedAliveTo) >= ((ZT_CLUSTER_TIMEOUT / 2) - 1000)) {
+				Buffer<2048> alive;
+				alive.append((uint16_t)ZEROTIER_ONE_VERSION_MAJOR);
+				alive.append((uint16_t)ZEROTIER_ONE_VERSION_MINOR);
+				alive.append((uint16_t)ZEROTIER_ONE_VERSION_REVISION);
+				alive.append((uint8_t)ZT_PROTO_VERSION);
+				if (_addressToLocationFunction) {
+					alive.append((int32_t)_x);
+					alive.append((int32_t)_y);
+					alive.append((int32_t)_z);
+				} else {
+					alive.append((int32_t)0);
+					alive.append((int32_t)0);
+					alive.append((int32_t)0);
+				}
+				alive.append((uint64_t)now);
+				alive.append((uint64_t)0); // TODO: compute and send load average
+				alive.append((uint64_t)0); // unused/reserved flags
+				alive.append((uint8_t)_zeroTierPhysicalEndpoints.size());
+				for(std::vector<InetAddress>::const_iterator pe(_zeroTierPhysicalEndpoints.begin());pe!=_zeroTierPhysicalEndpoints.end();++pe)
+					pe->serialize(alive);
+				_send(*mid,alive.data(),alive.size());
+				_members[*mid].lastAnnouncedAliveTo = now;
+			}
+
+			_flush(*mid); // does nothing if nothing to flush
 		}
 	}
 }
 
 void Cluster::addMember(uint16_t memberId)
 {
+	if (memberId >= ZT_CLUSTER_MAX_MEMBERS)
+		return;
+
 	Mutex::Lock _l2(_members[memberId].lock);
 
-	Mutex::Lock _l(_memberIds_m);
-	_memberIds.push_back(memberId);
-	std::sort(_memberIds.begin(),_memberIds.end());
+	{
+		Mutex::Lock _l(_memberIds_m);
+		if (std::find(_memberIds.begin(),_memberIds.end(),memberId) != _memberIds.end())
+			return;
+		_memberIds.push_back(memberId);
+		std::sort(_memberIds.begin(),_memberIds.end());
+	}
+
+	_members[memberId].clear();
 
 	// Generate this member's message key from the master and its ID
 	uint16_t stmp[ZT_SHA512_DIGEST_LEN / sizeof(uint16_t)];
@@ -346,6 +413,89 @@ void Cluster::addMember(uint16_t memberId)
 	_members[memberId].q.append(iv,16);
 	_members[memberId].q.addSize(8); // room for MAC
 	_members[memberId].q.append((uint16_t)_id);
+	_members[memberId].q.append((uint16_t)memberId);
+}
+
+void Cluster::removeMember(uint16_t memberId)
+{
+	Mutex::Lock _l(_memberIds_m);
+	std::vector<uint16_t> newMemberIds;
+	for(std::vector<uint16_t>::const_iterator mid(_memberIds.begin());mid!=_memberIds.end();++mid) {
+		if (*mid != memberId)
+			newMemberIds.push_back(*mid);
+	}
+	_memberIds = newMemberIds;
+}
+
+bool Cluster::redirectPeer(const SharedPtr<Peer> &peer,const InetAddress &peerPhysicalAddress,bool offload)
+{
+	if (!peerPhysicalAddress) // sanity check
+		return false;
+	if (_addressToLocationFunction) {
+		// Pick based on location if it can be determined
+		int px = 0,py = 0,pz = 0;
+		if (_addressToLocationFunction(_addressToLocationFunctionArg,reinterpret_cast<const struct sockaddr_storage *>(&peerPhysicalAddress),&px,&py,&pz) == 0) {
+			// No geo-info so no change
+			return false;
+		}
+
+		// Find member closest to this peer
+		const uint64_t now = RR->node->now();
+		std::vector<InetAddress> best; // initial "best" is for peer to stay put
+		const double currentDistance = _dist3d(_x,_y,_z,px,py,pz);
+		double bestDistance = (offload ? 2147483648.0 : currentDistance);
+		unsigned int bestMember = _id;
+		{
+			Mutex::Lock _l(_memberIds_m);
+			for(std::vector<uint16_t>::const_iterator mid(_memberIds.begin());mid!=_memberIds.end();++mid) {
+				_Member &m = _members[*mid];
+				Mutex::Lock _ml(m.lock);
+
+				// Consider member if it's alive and has sent us a location and one or more physical endpoints to send peers to
+				if ( ((now - m.lastReceivedAliveAnnouncement) < ZT_CLUSTER_TIMEOUT) && ((m.x != 0)||(m.y != 0)||(m.z != 0)) && (m.zeroTierPhysicalEndpoints.size() > 0) ) {
+					double mdist = _dist3d(m.x,m.y,m.z,px,py,pz);
+					if (mdist < bestDistance) {
+						bestMember = *mid;
+						best = m.zeroTierPhysicalEndpoints;
+					}
+				}
+			}
+		}
+
+		if (best.size() > 0) {
+			TRACE("peer %s is at [%d,%d,%d], distance to us is %f, sending to %u instead for better distance %f",peer->address().toString().c_str(),px,py,pz,currentDistance,bestMember,bestDistance);
+
+			/* if (peer->remoteVersionProtocol() >= 5) {
+				// If it's a newer peer send VERB_PUSH_DIRECT_PATHS which is more idiomatic
+			} else { */
+				// Otherwise send VERB_RENDEZVOUS for ourselves, which will trick peers into trying other endpoints for us even if they're too old for PUSH_DIRECT_PATHS
+				for(std::vector<InetAddress>::const_iterator a(best.begin());a!=best.end();++a) {
+					if ((a->ss_family == AF_INET)||(a->ss_family == AF_INET6)) {
+						Packet outp(peer->address(),RR->identity.address(),Packet::VERB_RENDEZVOUS);
+						outp.append((uint8_t)0); // no flags
+						RR->identity.address().appendTo(outp); // HACK: rendezvous with ourselves! with really old peers this will only work if I'm a root server!
+						outp.append((uint16_t)a->port());
+						if (a->ss_family == AF_INET) {
+							outp.append((uint8_t)4);
+							outp.append(a->rawIpData(),4);
+						} else {
+							outp.append((uint8_t)16);
+							outp.append(a->rawIpData(),16);
+						}
+						RR->sw->send(outp,true,0);
+					}
+				}
+			//}
+
+			return true;
+		} else {
+			TRACE("peer %s is at [%d,%d,%d], distance to us is %f and this seems to be the best",peer->address().toString().c_str(),px,py,pz,currentDistance);
+			return false;
+		}
+	} else {
+		// TODO: pick based on load if no location info?
+		return false;
+	}
 }
 
 void Cluster::_send(uint16_t memberId,const void *msg,unsigned int len)
@@ -366,7 +516,7 @@ void Cluster::_flush(uint16_t memberId)
 {
 	_Member &m = _members[memberId];
 	// assumes m.lock is locked!
-	if (m.q.size() > 26) { // 16-byte IV + 8-byte MAC + 2-byte cluster member ID (latter two bytes are inside crypto envelope)
+	if (m.q.size() > (24 + 2 + 2)) { // 16-byte IV + 8-byte MAC + 2 byte from-member-ID + 2 byte to-member-ID
 		// Create key from member's key and IV
 		char keytmp[32];
 		memcpy(keytmp,m.key,32);
@@ -389,7 +539,7 @@ void Cluster::_flush(uint16_t memberId)
 		memcpy(m.q.field(16,8),mac,8);
 
 		// Send!
-		_sendFunction(_arg,memberId,m.q.data(),m.q.size());
+		_sendFunction(_sendFunctionArg,memberId,m.q.data(),m.q.size());
 
 		// Prepare for more
 		m.q.clear();
@@ -397,7 +547,8 @@ void Cluster::_flush(uint16_t memberId)
 		Utils::getSecureRandom(iv,16);
 		m.q.append(iv,16);
 		m.q.addSize(8); // room for MAC
-		m.q.append((uint16_t)_id);
+		m.q.append((uint16_t)_id); // from member ID
+		m.q.append((uint16_t)memberId); // to member ID
 	}
 }
 

node/Cluster.hpp

@@ -34,43 +34,32 @@
 #include <algorithm>
 
 #include "Constants.hpp"
+#include "../include/ZeroTierOne.h"
 #include "Address.hpp"
 #include "InetAddress.hpp"
 #include "SHA512.hpp"
 #include "Utils.hpp"
 #include "Buffer.hpp"
 #include "Mutex.hpp"
+#include "SharedPtr.hpp"
 
 /**
  * Timeout for cluster members being considered "alive"
  */
-#define ZT_CLUSTER_TIMEOUT ZT_PEER_ACTIVITY_TIMEOUT
+#define ZT_CLUSTER_TIMEOUT 30000
 
 /**
- * Maximum cluster message length in bytes
- *
- * Cluster nodes speak via TCP, with data encapsulated into individually
- * encrypted and authenticated messages. The maximum message size is
- * 65535 (0xffff) since the TCP stream uses 16-bit message size headers
- * (and this is a reasonable chunk size anyway).
- */
-#define ZT_CLUSTER_MAX_MESSAGE_LENGTH 65535
-
-/**
- * Maximum number of physical addresses we will cache for a cluster member
- */
-#define ZT_CLUSTER_MEMBER_MAX_PHYSICAL_ADDRS 8
-
-/**
- * How frequently should doPeriodicTasks() be ideally called? (ms)
+ * Desired period between doPeriodicTasks() in milliseconds
  */
-#define ZT_CLUSTER_PERIODIC_TASK_DEADLINE 10
+#define ZT_CLUSTER_PERIODIC_TASK_PERIOD 50
 
 namespace ZeroTier {
 
 class RuntimeEnvironment;
 class CertificateOfMembership;
 class MulticastGroup;
+class Peer;
+class Identity;
 
 /**
  * Multi-homing cluster state replication and packet relaying
@@ -95,22 +84,6 @@ class MulticastGroup;
 class Cluster
 {
 public:
-	/**
-	 * Which distance algorithm is this cluster using?
-	 */
-	enum DistanceAlgorithm
-	{
-		/**
-		 * Simple linear distance in three dimensions
-		 */
-		DISTANCE_SIMPLE = 0,
-
-		/**
-		 * Haversine formula using X,Y as lat,long and ignoring Z
-		 */
-		DISTANCE_HAVERSINE = 1
-	};
-
 	/**
 	 * State message types
 	 */
@@ -184,25 +157,18 @@ public:
 
 	/**
 	 * Construct a new cluster
-	 *
-	 * @param renv Runtime environment
-	 * @param id This member's ID in the cluster
-	 * @param da Distance algorithm this cluster uses to compute distance and hand off peers
-	 * @param x My X
-	 * @param y My Y
-	 * @param z My Z
-	 * @param sendFunction Function to call to send messages to other cluster members
-	 * @param arg First argument to sendFunction
 	 */
 	Cluster(
 		const RuntimeEnvironment *renv,
 		uint16_t id,
-		DistanceAlgorithm da,
+		const std::vector<InetAddress> &zeroTierPhysicalEndpoints,
 		int32_t x,
 		int32_t y,
 		int32_t z,
-		void (*sendFunction)(void *,uint16_t,const void *,unsigned int),
-		void *arg);
+		void (*sendFunction)(void *,unsigned int,const void *,unsigned int),
+		void *sendFunctionArg,
+		int (*addressToLocationFunction)(void *,const struct sockaddr_storage *,int *,int *,int *),
+		void *addressToLocationFunctionArg);
 
 	~Cluster();
 
@@ -222,9 +188,9 @@ public:
 	/**
 	 * Advertise to the cluster that we have this peer
 	 *
-	 * @param peerAddress Peer address that we have
+	 * @param peerId Identity of peer that we have
 	 */
-	void replicateHavePeer(const Address &peerAddress);
+	void replicateHavePeer(const Identity &peerId);
 
 	/**
 	 * Advertise a multicast LIKE to the cluster
@@ -243,7 +209,7 @@ public:
 	void replicateCertificateOfNetworkMembership(const CertificateOfMembership &com);
 
 	/**
-	 * Call every ~ZT_CLUSTER_PERIODIC_TASK_DEADLINE milliseconds.
+	 * Call every ~ZT_CLUSTER_PERIODIC_TASK_PERIOD milliseconds.
 	 */
 	void doPeriodicTasks();
 
@@ -254,6 +220,23 @@ public:
 	 */
 	void addMember(uint16_t memberId);
 
+	/**
+	 * Remove a member ID from this cluster
+	 *
+	 * @param memberId Member ID to remove
+	 */
+	void removeMember(uint16_t memberId);
+
+	/**
+	 * Redirect this peer to a better cluster member if needed
+	 *
+	 * @param peer Peer to (possibly) redirect
+	 * @param peerPhysicalAddress Physical address of peer's current best path (where packet was most recently received or getBestPath()->address())
+	 * @param offload Always redirect if possible -- can be used to offload peers during shutdown
+	 * @return True if peer was redirected
+	 */
+	bool redirectPeer(const SharedPtr<Peer> &peer,const InetAddress &peerPhysicalAddress,bool offload);
+
 private:
 	void _send(uint16_t memberId,const void *msg,unsigned int len);
 	void _flush(uint16_t memberId);
@@ -262,44 +245,45 @@ private:
 	uint16_t _masterSecret[ZT_SHA512_DIGEST_LEN / sizeof(uint16_t)];
 	unsigned char _key[ZT_PEER_SECRET_KEY_LENGTH];
 	const RuntimeEnvironment *RR;
-	void (*_sendFunction)(void *,uint16_t,const void *,unsigned int);
-	void *_arg;
+	void (*_sendFunction)(void *,unsigned int,const void *,unsigned int);
+	void *_sendFunctionArg;
+	int (*_addressToLocationFunction)(void *,const struct sockaddr_storage *,int *,int *,int *);
+	void *_addressToLocationFunctionArg;
 	const int32_t _x;
 	const int32_t _y;
 	const int32_t _z;
-	const DistanceAlgorithm _da;
 	const uint16_t _id;
+	const std::vector<InetAddress> _zeroTierPhysicalEndpoints;
 
 	struct _Member
 	{
 		unsigned char key[ZT_PEER_SECRET_KEY_LENGTH];
 
-		uint64_t lastReceivedFrom;
 		uint64_t lastReceivedAliveAnnouncement;
-		uint64_t lastSentTo;
 		uint64_t lastAnnouncedAliveTo;
 
 		uint64_t load;
 		int32_t x,y,z;
 
-		InetAddress physicalAddresses[ZT_CLUSTER_MEMBER_MAX_PHYSICAL_ADDRS];
-		unsigned int physicalAddressCount;
+		std::vector<InetAddress> zeroTierPhysicalEndpoints;
 
 		Buffer<ZT_CLUSTER_MAX_MESSAGE_LENGTH> q;
 
 		Mutex lock;
 
-		_Member() :
-			lastReceivedFrom(0),
-			lastReceivedAliveAnnouncement(0),
-			lastSentTo(0),
-			lastAnnouncedAliveTo(0),
-			load(0),
-			x(0),
-			y(0),
-			z(0),
-			physicalAddressCount(0) {}
-
+		inline void clear()
+		{
+			lastReceivedAliveAnnouncement = 0;
+			lastAnnouncedAliveTo = 0;
+			load = 0;
+			x = 0;
+			y = 0;
+			z = 0;
+			zeroTierPhysicalEndpoints.clear();
+			q.clear();
+		}
+
+		_Member() { this->clear(); }
 		~_Member() { Utils::burn(key,sizeof(key)); }
 	};
 

node/IncomingPacket.cpp

@@ -871,6 +871,8 @@ bool IncomingPacket::_doPUSH_DIRECT_PATHS(const RuntimeEnvironment *RR,const Sha
 		}
 		peer->setLastDirectPathPushReceived(now);
 
+		const RemotePath *currentBest = peer->getBestPath();
+
 		unsigned int count = at<uint16_t>(ZT_PACKET_IDX_PAYLOAD);
 		unsigned int ptr = ZT_PACKET_IDX_PAYLOAD + 2;
 		unsigned int v4Count = 0,v6Count = 0;
@@ -889,16 +891,20 @@ bool IncomingPacket::_doPUSH_DIRECT_PATHS(const RuntimeEnvironment *RR,const Sha
 					InetAddress a(field(ptr,4),4,at<uint16_t>(ptr + 4));
 					if ( ((flags & 0x01) == 0) && (Path::isAddressValidForPath(a)) ) {
 						TRACE("attempting to contact %s at pushed direct path %s",peer->address().toString().c_str(),a.toString().c_str());
-						if (v4Count++ < ZT_PUSH_DIRECT_PATHS_MAX_ENDPOINTS_PER_TYPE)
-							peer->attemptToContactAt(RR,_localAddress,a,RR->node->now());
+						if (v4Count++ < ZT_PUSH_DIRECT_PATHS_MAX_ENDPOINTS_PER_TYPE) {
+							if ((!currentBest)||(currentBest->address() != a))
+								peer->attemptToContactAt(RR,_localAddress,a,RR->node->now());
+						}
 					}
 				}	break;
 				case 6: {
 					InetAddress a(field(ptr,16),16,at<uint16_t>(ptr + 16));
 					if ( ((flags & 0x01) == 0) && (Path::isAddressValidForPath(a)) ) {
 						TRACE("attempting to contact %s at pushed direct path %s",peer->address().toString().c_str(),a.toString().c_str());
-						if (v6Count++ < ZT_PUSH_DIRECT_PATHS_MAX_ENDPOINTS_PER_TYPE)
-							peer->attemptToContactAt(RR,_localAddress,a,RR->node->now());
+						if (v6Count++ < ZT_PUSH_DIRECT_PATHS_MAX_ENDPOINTS_PER_TYPE) {
+							if ((!currentBest)||(currentBest->address() != a))
+								peer->attemptToContactAt(RR,_localAddress,a,RR->node->now());
+						}
 					}
 				}	break;
 			}

node/Node.cpp

@@ -46,6 +46,7 @@
 #include "Address.hpp"
 #include "Identity.hpp"
 #include "SelfAwareness.hpp"
+#include "Cluster.hpp"
 
 const struct sockaddr_storage ZT_SOCKADDR_NULL = {0};
 
@@ -135,6 +136,9 @@ Node::~Node()
 	delete RR->antiRec;
 	delete RR->mc;
 	delete RR->sw;
+#ifdef ZT_ENABLE_CLUSTER
+	delete RR->cluster;
+#endif
 }
 
 ZT_ResultCode Node::processWirePacket(
@@ -329,7 +333,18 @@ ZT_ResultCode Node::processBackgroundTasks(uint64_t now,volatile uint64_t *nextB
 	}
 
 	try {
-		*nextBackgroundTaskDeadline = now + (uint64_t)std::max(std::min(timeUntilNextPingCheck,RR->sw->doTimerTasks(now)),(unsigned long)ZT_CORE_TIMER_TASK_GRANULARITY);
+#ifdef ZT_ENABLE_CLUSTER
+		// If clustering is enabled we have to call cluster->doPeriodicTasks() very often, so we override normal timer deadline behavior
+		if (RR->cluster) {
+			RR->sw->doTimerTasks(now);
+			RR->cluster->doPeriodicTasks();
+			*nextBackgroundTaskDeadline = now + ZT_CLUSTER_PERIODIC_TASK_PERIOD; // this is really short so just tick at this rate
+		} else {
+#endif
+			*nextBackgroundTaskDeadline = now + (uint64_t)std::max(std::min(timeUntilNextPingCheck,RR->sw->doTimerTasks(now)),(unsigned long)ZT_CORE_TIMER_TASK_GRANULARITY);
+#ifdef ZT_ENABLE_CLUSTER
+		}
+#endif
 	} catch ( ... ) {
 		return ZT_RESULT_FATAL_ERROR_INTERNAL;
 	}
@@ -554,6 +569,62 @@ void Node::circuitTestEnd(ZT_CircuitTest *test)
 	}
 }
 
+ZT_ResultCode Node::clusterInit(
+	unsigned int myId,
+	const struct sockaddr_storage *zeroTierPhysicalEndpoints,
+	unsigned int numZeroTierPhysicalEndpoints,
+	int x,
+	int y,
+	int z,
+	void (*sendFunction)(void *,unsigned int,const void *,unsigned int),
+	void *sendFunctionArg,
+	int (*addressToLocationFunction)(void *,const struct sockaddr_storage *,int *,int *,int *),
+	void *addressToLocationFunctionArg)
+{
+#ifdef ZT_ENABLE_CLUSTER
+	if (RR->cluster)
+		return ZT_RESULT_ERROR_BAD_PARAMETER;
+
+	std::vector<InetAddress> eps;
+	for(unsigned int i=0;i<numZeroTierPhysicalEndpoints;++i)
+		eps.push_back(InetAddress(zeroTierPhysicalEndpoints[i]));
+	std::sort(eps.begin(),eps.end());
+	RR->cluster = new Cluster(RR,myId,eps,x,y,z,sendFunction,sendFunctionArg,addressToLocationFunction,addressToLocationFunctionArg);
+
+	return ZT_RESULT_OK;
+#else
+	return ZT_RESULT_ERROR_UNSUPPORTED_OPERATION;
+#endif
+}
+
+ZT_ResultCode Node::clusterAddMember(unsigned int memberId)
+{
+#ifdef ZT_ENABLE_CLUSTER
+	if (!RR->cluster)
+		return ZT_RESULT_ERROR_BAD_PARAMETER;
+	RR->cluster->addMember((uint16_t)memberId);
+	return ZT_RESULT_OK;
+#else
+	return ZT_RESULT_ERROR_UNSUPPORTED_OPERATION;
+#endif
+}
+
+void Node::clusterRemoveMember(unsigned int memberId)
+{
+#ifdef ZT_ENABLE_CLUSTER
+	if (RR->cluster)
+		RR->cluster->removeMember((uint16_t)memberId);
+#endif
+}
+
+void Node::clusterHandleIncomingMessage(const void *msg,unsigned int len)
+{
+#ifdef ZT_ENABLE_CLUSTER
+	if (RR->cluster)
+		RR->cluster->handleIncomingStateMessage(msg,len);
+#endif
+}
+
 /****************************************************************************/
 /* Node methods used only within node/                                      */
 /****************************************************************************/
@@ -806,6 +877,22 @@ void ZT_Node_freeQueryResult(ZT_Node *node,void *qr)
 	} catch ( ... ) {}
 }
 
+int ZT_Node_addLocalInterfaceAddress(ZT_Node *node,const struct sockaddr_storage *addr,int metric, enum ZT_LocalInterfaceAddressTrust trust)
+{
+	try {
+		return reinterpret_cast<ZeroTier::Node *>(node)->addLocalInterfaceAddress(addr,metric,trust);
+	} catch ( ... ) {
+		return 0;
+	}
+}
+
+void ZT_Node_clearLocalInterfaceAddresses(ZT_Node *node)
+{
+	try {
+		reinterpret_cast<ZeroTier::Node *>(node)->clearLocalInterfaceAddresses();
+	} catch ( ... ) {}
+}
+
 void ZT_Node_setNetconfMaster(ZT_Node *node,void *networkControllerInstance)
 {
 	try {
@@ -829,19 +916,75 @@ void ZT_Node_circuitTestEnd(ZT_Node *node,ZT_CircuitTest *test)
 	} catch ( ... ) {}
 }
 
-int ZT_Node_addLocalInterfaceAddress(ZT_Node *node,const struct sockaddr_storage *addr,int metric, enum ZT_LocalInterfaceAddressTrust trust)
+enum ZT_ResultCode ZT_Node_clusterInit(
+	ZT_Node *node,
+	unsigned int myId,
+	const struct sockaddr_storage *zeroTierPhysicalEndpoints,
+	unsigned int numZeroTierPhysicalEndpoints,
+	int x,
+	int y,
+	int z,
+	void (*sendFunction)(void *,unsigned int,const void *,unsigned int),
+	void *sendFunctionArg,
+	int (*addressToLocationFunction)(void *,const struct sockaddr_storage *,int *,int *,int *),
+	void *addressToLocationFunctionArg)
 {
 	try {
-		return reinterpret_cast<ZeroTier::Node *>(node)->addLocalInterfaceAddress(addr,metric,trust);
+		return reinterpret_cast<ZeroTier::Node *>(node)->clusterInit(myId,zeroTierPhysicalEndpoints,numZeroTierPhysicalEndpoints,x,y,z,sendFunction,sendFunctionArg,addressToLocationFunction,addressToLocationFunctionArg);
 	} catch ( ... ) {
-		return 0;
+		return ZT_RESULT_FATAL_ERROR_INTERNAL;
 	}
 }
 
-void ZT_Node_clearLocalInterfaceAddresses(ZT_Node *node)
+/**
+ * Add a member to this cluster
+ *
+ * Calling this without having called clusterInit() will do nothing.
+ *
+ * @param node Node instance
+ * @param memberId Member ID (must be less than or equal to ZT_CLUSTER_MAX_MEMBERS)
+ * @return OK or error if clustering is disabled, ID invalid, etc.
+ */
+enum ZT_ResultCode ZT_Node_clusterAddMember(ZT_Node *node,unsigned int memberId)
 {
 	try {
-		reinterpret_cast<ZeroTier::Node *>(node)->clearLocalInterfaceAddresses();
+		return reinterpret_cast<ZeroTier::Node *>(node)->clusterAddMember(memberId);
+	} catch ( ... ) {
+		return ZT_RESULT_FATAL_ERROR_INTERNAL;
+	}
+}
+
+/**
+ * Remove a member from this cluster
+ *
+ * Calling this without having called clusterInit() will do nothing.
+ *
+ * @param node Node instance
+ * @param memberId Member ID to remove (nothing happens if not present)
+ */
+void ZT_Node_clusterRemoveMember(ZT_Node *node,unsigned int memberId)
+{
+	try {
+		reinterpret_cast<ZeroTier::Node *>(node)->clusterRemoveMember(memberId);
+	} catch ( ... ) {}
+}
+
+/**
+ * Handle an incoming cluster state message
+ *
+ * The message itself contains cluster member IDs, and invalid or badly
+ * addressed messages will be silently discarded.
+ *
+ * Calling this without having called clusterInit() will do nothing.
+ *
+ * @param node Node instance
+ * @param msg Cluster message
+ * @param len Length of cluster message
+ */
+void ZT_Node_clusterHandleIncomingMessage(ZT_Node *node,const void *msg,unsigned int len)
+{
+	try {
+		reinterpret_cast<ZeroTier::Node *>(node)->clusterHandleIncomingMessage(msg,len);
 	} catch ( ... ) {}
 }
 

node/Node.hpp

@@ -110,6 +110,20 @@ public:
 	void setNetconfMaster(void *networkControllerInstance);
 	ZT_ResultCode circuitTestBegin(ZT_CircuitTest *test,void (*reportCallback)(ZT_Node *,ZT_CircuitTest *,const ZT_CircuitTestReport *));
 	void circuitTestEnd(ZT_CircuitTest *test);
+	ZT_ResultCode clusterInit(
+		unsigned int myId,
+		const struct sockaddr_storage *zeroTierPhysicalEndpoints,
+		unsigned int numZeroTierPhysicalEndpoints,
+		int x,
+		int y,
+		int z,
+		void (*sendFunction)(void *,unsigned int,const void *,unsigned int),
+		void *sendFunctionArg,
+		int (*addressToLocationFunction)(void *,const struct sockaddr_storage *,int *,int *,int *),
+		void *addressToLocationFunctionArg);
+	ZT_ResultCode clusterAddMember(unsigned int memberId);
+	void clusterRemoveMember(unsigned int memberId);
+	void clusterHandleIncomingMessage(const void *msg,unsigned int len);
 
 	// Internal functions ------------------------------------------------------
 

node/Peer.cpp

@@ -213,6 +213,12 @@ bool Peer::doPingAndKeepalive(const RuntimeEnvironment *RR,uint64_t now,int inet
 
 void Peer::pushDirectPaths(const RuntimeEnvironment *RR,RemotePath *path,uint64_t now,bool force)
 {
+#ifdef ZT_ENABLE_CLUSTER
+	// Cluster mode disables normal PUSH_DIRECT_PATHS in favor of cluster-based peer redirection
+	if (RR->cluster)
+		return;
+#endif
+
 	Mutex::Lock _l(_lock);
 
 	if (((now - _lastDirectPathPushSent) >= ZT_DIRECT_PATH_PUSH_INTERVAL)||(force)) {

node/RuntimeEnvironment.hpp

@@ -43,6 +43,7 @@ class Multicaster;
 class AntiRecursion;
 class NetworkController;
 class SelfAwareness;
+class Cluster;
 
 /**
  * Holds global state for an instance of ZeroTier::Node
@@ -51,14 +52,17 @@ class RuntimeEnvironment
 {
 public:
 	RuntimeEnvironment(Node *n) :
-		node(n),
-		identity(),
-		localNetworkController((NetworkController *)0),
-		sw((Switch *)0),
-		mc((Multicaster *)0),
-		antiRec((AntiRecursion *)0),
-		topology((Topology *)0),
-		sa((SelfAwareness *)0)
+		node(n)
+		,identity()
+		,localNetworkController((NetworkController *)0)
+		,sw((Switch *)0)
+		,mc((Multicaster *)0)
+		,antiRec((AntiRecursion *)0)
+		,topology((Topology *)0)
+		,sa((SelfAwareness *)0)
+#ifdef ZT_ENABLE_CLUSTER
+		,cluster((Cluster *)0)
+#endif
 	{
 	}
 
@@ -86,6 +90,10 @@ public:
 	AntiRecursion *antiRec;
 	Topology *topology;
 	SelfAwareness *sa;
+
+#ifdef ZT_ENABLE_CLUSTER
+	Cluster *cluster;
+#endif
 };
 
 } // namespace ZeroTier