Clustered handling of relaying.

node/Cluster.cpp

@@ -352,6 +352,53 @@ void Cluster::handleIncomingStateMessage(const void *msg,unsigned int len)
 	}
 }
 
+bool Cluster::sendViaCluster(const Address &fromPeerAddress,const Address &toPeerAddress,const void *data,unsigned int len)
+{
+	if (len > 16384) // sanity check
+		return false;
+
+	uint64_t mostRecentTimestamp = 0;
+	uint16_t canHasPeer = 0;
+
+	{	// Anyone got this peer?
+		Mutex::Lock _l2(_peerAffinities_m);
+		std::vector<_PeerAffinity>::iterator i(std::lower_bound(_peerAffinities.begin(),_peerAffinities.end(),_PeerAffinity(toPeerAddress,0,0))); // O(log(n))
+		while ((i != _peerAffinities.end())&&(i->address() == toPeerAddress)) {
+			uint16_t mid = i->clusterMemberId();
+			if ((mid != _id)&&(i->timestamp > mostRecentTimestamp)) {
+				mostRecentTimestamp = i->timestamp;
+				canHasPeer = mid;
+			}
+		}
+	}
+
+	const uint64_t now = RR->node->now();
+	if ((now - mostRecentTimestamp) < ZT_PEER_ACTIVITY_TIMEOUT) {
+		Buffer<16384> buf;
+
+		InetAddress v4,v6;
+		if (fromPeerAddress) {
+			SharedPtr<Peer> fromPeer(RR->topology->getPeer(fromPeerAddress));
+			if (fromPeer)
+				fromPeer->getBestActiveAddresses(now,v4,v6);
+		}
+		buf.append((uint8_t)( (v4) ? ((v6) ? 2 : 1) : ((v6) ? 1 : 0) ));
+		if (v4)
+			v4.serialize(buf);
+		if (v6)
+			v6.serialize(buf);
+		buf.append((uint16_t)len);
+		buf.append(data,len);
+
+		{
+			Mutex::Lock _l2(_members[canHasPeer].lock);
+			_send(canHasPeer,STATE_MESSAGE_RELAY,buf.data(),buf.size());
+		}
+	}
+
+	return false;
+}
+
 void Cluster::replicateHavePeer(const Identity &peerId)
 {
 	{	// Use peer affinity table to track our own last announce time for peers

node/Cluster.hpp

@@ -191,6 +191,17 @@ public:
 	 */
 	void handleIncomingStateMessage(const void *msg,unsigned int len);
 
+	/**
+	 * Send this packet via another node in this cluster if another node has this peer
+	 *
+	 * @param fromPeerAddress Source peer address (if known, should be NULL for fragments)
+	 * @param toPeerAddress Destination peer address
+	 * @param data Packet or packet fragment data
+	 * @param len Length of packet or fragment
+	 * @return True if this data was sent via another cluster member, false if none have this peer
+	 */
+	bool sendViaCluster(const Address &fromPeerAddress,const Address &toPeerAddress,const void *data,unsigned int len);
+
 	/**
 	 * Advertise to the cluster that we have this peer
 	 *

node/Switch.cpp

@@ -45,6 +45,7 @@
 #include "AntiRecursion.hpp"
 #include "SelfAwareness.hpp"
 #include "Packet.hpp"
+#include "Cluster.hpp"
 
 namespace ZeroTier {
 
@@ -567,6 +568,11 @@ void Switch::_handleRemotePacketFragment(const InetAddress &localAddr,const Inet
 			// It wouldn't hurt anything, just redundant and unnecessary.
 			SharedPtr<Peer> relayTo = RR->topology->getPeer(destination);
 			if ((!relayTo)||(!relayTo->send(RR,fragment.data(),fragment.size(),RR->node->now()))) {
+#ifdef ZT_ENABLE_CLUSTER
+				if ((RR->cluster)&&(RR->cluster->sendViaCluster(Address(),destination,fragment.data(),fragment.size())))
+					return; // sent by way of another member of this cluster
+#endif
+
 				// Don't know peer or no direct path -- so relay via root server
 				relayTo = RR->topology->getBestRoot();
 				if (relayTo)
@@ -642,7 +648,11 @@ void Switch::_handleRemotePacketHead(const InetAddress &localAddr,const InetAddr
 			if ((relayTo)&&((relayTo->send(RR,packet->data(),packet->size(),RR->node->now())))) {
 				unite(source,destination,false);
 			} else {
-				// Don't know peer or no direct path -- so relay via root server
+#ifdef ZT_ENABLE_CLUSTER
+				if ((RR->cluster)&&(RR->cluster->sendViaCluster(source,destination,packet->data(),packet->size())))
+					return; // sent by way of another member of this cluster
+#endif
+
 				relayTo = RR->topology->getBestRoot(&source,1,true);
 				if (relayTo)
 					relayTo->send(RR,packet->data(),packet->size(),RR->node->now());