From 57e29857cf79019af03f6a3dfe0bf6fd36e2fab2 Mon Sep 17 00:00:00 2001 From: Adam Ierymenko Date: Tue, 20 Oct 2015 15:27:53 -0700 Subject: Cluster work -- integrating with the rest of the code. --- node/Cluster.cpp | 193 +++++++++++++++++++++++++++++++++++++++----- node/Cluster.hpp | 118 ++++++++++++--------------- node/IncomingPacket.cpp | 14 +++- node/Node.cpp | 155 +++++++++++++++++++++++++++++++++-- node/Node.hpp | 14 ++++ node/Peer.cpp | 6 ++ node/RuntimeEnvironment.hpp | 24 ++++-- 7 files changed, 418 insertions(+), 106 deletions(-) (limited to 'node') diff --git a/node/Cluster.cpp b/node/Cluster.cpp index bfa39d22..5b76d1f0 100644 --- a/node/Cluster.cpp +++ b/node/Cluster.cpp @@ -31,10 +31,13 @@ #include #include #include +#include #include #include +#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 &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(msg) + 24,const_cast(dmsg.data()),dmsg.size()); } - if (dmsg.size() < 2) + if (dmsg.size() < 4) return; const uint16_t fromMemberId = dmsg.at(0); unsigned int ptr = 2; + if (fromMemberId == _id) + return; + const uint16_t toMemberId = dmsg.at(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(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(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;inode->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::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::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 newMemberIds; + for(std::vector::const_iterator mid(_memberIds.begin());mid!=_memberIds.end();++mid) { + if (*mid != memberId) + newMemberIds.push_back(*mid); + } + _memberIds = newMemberIds; +} + +bool Cluster::redirectPeer(const SharedPtr &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(&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 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::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::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 } } diff --git a/node/Cluster.hpp b/node/Cluster.hpp index 016730e3..f253e6f6 100644 --- a/node/Cluster.hpp +++ b/node/Cluster.hpp @@ -34,43 +34,32 @@ #include #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 &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,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 _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 zeroTierPhysicalEndpoints; Buffer 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)); } }; diff --git a/node/IncomingPacket.cpp b/node/IncomingPacket.cpp index 19747bbd..a4d45068 100644 --- a/node/IncomingPacket.cpp +++ b/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(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(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(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; } diff --git a/node/Node.cpp b/node/Node.cpp index 26d5513e..6eea3d3d 100644 --- a/node/Node.cpp +++ b/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 eps; + for(unsigned int i=0;icluster = 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(node)->addLocalInterfaceAddress(addr,metric,trust); + } catch ( ... ) { + return 0; + } +} + +void ZT_Node_clearLocalInterfaceAddresses(ZT_Node *node) +{ + try { + reinterpret_cast(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(node)->addLocalInterfaceAddress(addr,metric,trust); + return reinterpret_cast(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(node)->clearLocalInterfaceAddresses(); + return reinterpret_cast(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(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(node)->clusterHandleIncomingMessage(msg,len); } catch ( ... ) {} } diff --git a/node/Node.hpp b/node/Node.hpp index c7038ed4..b8bd4dc5 100644 --- a/node/Node.hpp +++ b/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 ------------------------------------------------------ diff --git a/node/Peer.cpp b/node/Peer.cpp index 6f566be4..0ba379c6 100644 --- a/node/Peer.cpp +++ b/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)) { diff --git a/node/RuntimeEnvironment.hpp b/node/RuntimeEnvironment.hpp index e5d1f446..2ec88f72 100644 --- a/node/RuntimeEnvironment.hpp +++ b/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 -- cgit v1.2.3 From eb79d4a2f34b34c49cd2d69efac22d9bc8ac27cb Mon Sep 17 00:00:00 2001 From: Adam Ierymenko Date: Tue, 20 Oct 2015 16:24:21 -0700 Subject: Wire up peer announcement in cluster. --- node/Cluster.cpp | 72 +++++++++++++++++++++++++++++++++++++++---------- node/Cluster.hpp | 10 +++++-- node/IncomingPacket.cpp | 1 - node/Peer.cpp | 7 ++++- node/Topology.cpp | 27 ++++++++++--------- 5 files changed, 87 insertions(+), 30 deletions(-) (limited to 'node') diff --git a/node/Cluster.cpp b/node/Cluster.cpp index 5b76d1f0..f1dc45b9 100644 --- a/node/Cluster.cpp +++ b/node/Cluster.cpp @@ -263,8 +263,6 @@ void Cluster::handleIncomingStateMessage(const void *msg,unsigned int len) remotePeerAddress.appendTo(rendezvousForDest); Buffer<2048> rendezvousForOtherEnd; - rendezvousForOtherEnd.addSize(2); // leave room for payload size - rendezvousForOtherEnd.append((uint8_t)STATE_MESSAGE_PROXY_SEND); remotePeerAddress.appendTo(rendezvousForOtherEnd); rendezvousForOtherEnd.append((uint8_t)Packet::VERB_RENDEZVOUS); const unsigned int rendezvousForOtherEndPayloadSizePtr = rendezvousForOtherEnd.size(); @@ -298,9 +296,8 @@ void Cluster::handleIncomingStateMessage(const void *msg,unsigned int len) } if (haveMatch) { + _send(fromMemberId,STATE_MESSAGE_PROXY_SEND,rendezvousForOtherEnd.data(),rendezvousForOtherEnd.size()); RR->sw->send(rendezvousForDest,true,0); - rendezvousForOtherEnd.setAt(0,(uint16_t)(rendezvousForOtherEnd.size() - 2)); - _send(fromMemberId,rendezvousForOtherEnd.data(),rendezvousForOtherEnd.size()); } } } @@ -331,14 +328,64 @@ void Cluster::handleIncomingStateMessage(const void *msg,unsigned int len) void Cluster::replicateHavePeer(const Identity &peerId) { + { // Use peer affinity table to track our own last announce time for peers + _PeerAffinity pa(peerId.address(),_id,RR->node->now()); + Mutex::Lock _l2(_peerAffinities_m); + std::vector<_PeerAffinity>::iterator i(std::lower_bound(_peerAffinities.begin(),_peerAffinities.end(),pa)); // O(log(n)) + if ((i != _peerAffinities.end())&&(i->key == pa.key)) { + if ((pa.timestamp - i->timestamp) >= ZT_CLUSTER_HAVE_PEER_ANNOUNCE_PERIOD) { + i->timestamp = pa.timestamp; + // continue to announcement + } else { + // we've already announced this peer recently, so skip + return; + } + } else { + _peerAffinities.push_back(pa); + std::sort(_peerAffinities.begin(),_peerAffinities.end()); // probably a more efficient way to insert but okay for now + // continue to announcement + } + } + + // announcement + Buffer<4096> buf; + peerId.serialize(buf,false); + { + Mutex::Lock _l(_memberIds_m); + for(std::vector::const_iterator mid(_memberIds.begin());mid!=_memberIds.end();++mid) { + Mutex::Lock _l2(_members[*mid].lock); + _send(*mid,STATE_MESSAGE_HAVE_PEER,buf.data(),buf.size()); + } + } } void Cluster::replicateMulticastLike(uint64_t nwid,const Address &peerAddress,const MulticastGroup &group) { + Buffer<4096> buf; + buf.append((uint64_t)nwid); + peerAddress.appendTo(buf); + group.mac().appendTo(buf); + buf.append((uint32_t)group.adi()); + { + Mutex::Lock _l(_memberIds_m); + for(std::vector::const_iterator mid(_memberIds.begin());mid!=_memberIds.end();++mid) { + Mutex::Lock _l2(_members[*mid].lock); + _send(*mid,STATE_MESSAGE_MULTICAST_LIKE,buf.data(),buf.size()); + } + } } void Cluster::replicateCertificateOfNetworkMembership(const CertificateOfMembership &com) { + Buffer<4096> buf; + com.serialize(buf); + { + Mutex::Lock _l(_memberIds_m); + for(std::vector::const_iterator mid(_memberIds.begin());mid!=_memberIds.end();++mid) { + Mutex::Lock _l2(_members[*mid].lock); + _send(*mid,STATE_MESSAGE_COM,buf.data(),buf.size()); + } + } } void Cluster::doPeriodicTasks() @@ -371,7 +418,7 @@ void Cluster::doPeriodicTasks() alive.append((uint8_t)_zeroTierPhysicalEndpoints.size()); for(std::vector::const_iterator pe(_zeroTierPhysicalEndpoints.begin());pe!=_zeroTierPhysicalEndpoints.end();++pe) pe->serialize(alive); - _send(*mid,alive.data(),alive.size()); + _send(*mid,STATE_MESSAGE_ALIVE,alive.data(),alive.size()); _members[*mid].lastAnnouncedAliveTo = now; } @@ -498,18 +545,15 @@ bool Cluster::redirectPeer(const SharedPtr &peer,const InetAddress &peerPh } } -void Cluster::_send(uint16_t memberId,const void *msg,unsigned int len) +void Cluster::_send(uint16_t memberId,StateMessageType type,const void *msg,unsigned int len) { _Member &m = _members[memberId]; // assumes m.lock is locked! - for(;;) { - if ((m.q.size() + len) > ZT_CLUSTER_MAX_MESSAGE_LENGTH) - _flush(memberId); - else { - m.q.append(msg,len); - break; - } - } + if ((m.q.size() + len + 3) > ZT_CLUSTER_MAX_MESSAGE_LENGTH) + _flush(memberId); + m.q.append((uint16_t)(len + 1)); + m.q.append((uint8_t)type); + m.q.append(msg,len); } void Cluster::_flush(uint16_t memberId) diff --git a/node/Cluster.hpp b/node/Cluster.hpp index f253e6f6..df061c60 100644 --- a/node/Cluster.hpp +++ b/node/Cluster.hpp @@ -42,12 +42,18 @@ #include "Buffer.hpp" #include "Mutex.hpp" #include "SharedPtr.hpp" +#include "Hashtable.hpp" /** * Timeout for cluster members being considered "alive" */ #define ZT_CLUSTER_TIMEOUT 30000 +/** + * How often should we announce that we have a peer? + */ +#define ZT_CLUSTER_HAVE_PEER_ANNOUNCE_PERIOD 60000 + /** * Desired period between doPeriodicTasks() in milliseconds */ @@ -238,7 +244,7 @@ public: bool redirectPeer(const SharedPtr &peer,const InetAddress &peerPhysicalAddress,bool offload); private: - void _send(uint16_t memberId,const void *msg,unsigned int len); + void _send(uint16_t memberId,StateMessageType type,const void *msg,unsigned int len); void _flush(uint16_t memberId); // These are initialized in the constructor and remain static @@ -292,7 +298,7 @@ private: std::vector _memberIds; Mutex _memberIds_m; - // Record tracking which members have which peers and how recently they claimed this + // Record tracking which members have which peers and how recently they claimed this -- also used to track our last claimed time struct _PeerAffinity { _PeerAffinity(const Address &a,uint16_t mid,uint64_t ts) : diff --git a/node/IncomingPacket.cpp b/node/IncomingPacket.cpp index a4d45068..7a47c0c6 100644 --- a/node/IncomingPacket.cpp +++ b/node/IncomingPacket.cpp @@ -272,7 +272,6 @@ bool IncomingPacket::_doHELLO(const RuntimeEnvironment *RR) TRACE("rejected HELLO from %s(%s): packet failed authentication",id.address().toString().c_str(),_remoteAddress.toString().c_str()); return true; } - peer = RR->topology->addPeer(newPeer); // Continue at // VALID diff --git a/node/Peer.cpp b/node/Peer.cpp index 0ba379c6..45e2fedd 100644 --- a/node/Peer.cpp +++ b/node/Peer.cpp @@ -34,6 +34,7 @@ #include "Network.hpp" #include "AntiRecursion.hpp" #include "SelfAwareness.hpp" +#include "Cluster.hpp" #include @@ -107,7 +108,6 @@ void Peer::received( // Learn paths if they've been confirmed via a HELLO or an ECHO RemotePath *slot = (RemotePath *)0; if (np < ZT_MAX_PEER_NETWORK_PATHS) { - // Add new path slot = &(_paths[np++]); } else { uint64_t slotLRmin = 0xffffffffffffffffULL; @@ -141,6 +141,11 @@ void Peer::received( } } } + +#ifdef ZT_ENABLE_CLUSTER + if ((pathIsConfirmed)&&(RR->cluster)) + RR->cluster->replicateHavePeer(_id); +#endif } if ((now - _lastAnnouncedTo) >= ((ZT_MULTICAST_LIKE_EXPIRE / 2) - 1000)) { diff --git a/node/Topology.cpp b/node/Topology.cpp index 56ca47c8..88c8856c 100644 --- a/node/Topology.cpp +++ b/node/Topology.cpp @@ -122,18 +122,22 @@ Topology::~Topology() SharedPtr Topology::addPeer(const SharedPtr &peer) { if (peer->address() == RR->identity.address()) { - TRACE("BUG: addNewPeer() caught and ignored attempt to add peer for self"); + TRACE("BUG: addPeer() caught and ignored attempt to add peer for self"); throw std::logic_error("cannot add peer for self"); } - const uint64_t now = RR->node->now(); - Mutex::Lock _l(_lock); - - SharedPtr &p = _peers.set(peer->address(),peer); - p->use(now); - saveIdentity(p->identity()); + SharedPtr np; + { + Mutex::Lock _l(_lock); + SharedPtr &hp = _peers[peer->address()]; + if (!hp) + hp = peer; + np = hp; + } + np->use(RR->node->now()); + saveIdentity(np->identity()); - return p; + return np; } SharedPtr Topology::getPeer(const Address &zta) @@ -143,13 +147,12 @@ SharedPtr Topology::getPeer(const Address &zta) return SharedPtr(); } - const uint64_t now = RR->node->now(); Mutex::Lock _l(_lock); SharedPtr &ap = _peers[zta]; if (ap) { - ap->use(now); + ap->use(RR->node->now()); return ap; } @@ -157,13 +160,13 @@ SharedPtr Topology::getPeer(const Address &zta) if (id) { try { ap = SharedPtr(new Peer(RR->identity,id)); - ap->use(now); + ap->use(RR->node->now()); return ap; } catch ( ... ) {} // invalid identity? } + // If we get here it means we read an invalid cache identity or had some other error _peers.erase(zta); - return SharedPtr(); } -- cgit v1.2.3 From 59e1444b274465f33ed95aab6a9214d25910f85d Mon Sep 17 00:00:00 2001 From: Adam Ierymenko Date: Tue, 20 Oct 2015 16:31:41 -0700 Subject: Finish wiring up Cluster, fix some issues with other recent changes. --- node/Cluster.cpp | 28 +++++++++++++++------------- node/IncomingPacket.cpp | 14 +++++++++++--- node/Peer.cpp | 5 +++++ 3 files changed, 31 insertions(+), 16 deletions(-) (limited to 'node') diff --git a/node/Cluster.cpp b/node/Cluster.cpp index f1dc45b9..8afd3deb 100644 --- a/node/Cluster.cpp +++ b/node/Cluster.cpp @@ -186,19 +186,21 @@ void Cluster::handleIncomingStateMessage(const void *msg,unsigned int len) try { Identity id; ptr += id.deserialize(dmsg,ptr); - RR->topology->saveIdentity(id); - - { // Add or update peer affinity entry - _PeerAffinity pa(id.address(),fromMemberId,RR->node->now()); - Mutex::Lock _l2(_peerAffinities_m); - std::vector<_PeerAffinity>::iterator i(std::lower_bound(_peerAffinities.begin(),_peerAffinities.end(),pa)); // O(log(n)) - if ((i != _peerAffinities.end())&&(i->key == pa.key)) { - i->timestamp = pa.timestamp; - } else { - _peerAffinities.push_back(pa); - std::sort(_peerAffinities.begin(),_peerAffinities.end()); // probably a more efficient way to insert but okay for now - } - } + if (id) { + RR->topology->saveIdentity(id); + + { // Add or update peer affinity entry + _PeerAffinity pa(id.address(),fromMemberId,RR->node->now()); + Mutex::Lock _l2(_peerAffinities_m); + std::vector<_PeerAffinity>::iterator i(std::lower_bound(_peerAffinities.begin(),_peerAffinities.end(),pa)); // O(log(n)) + if ((i != _peerAffinities.end())&&(i->key == pa.key)) { + i->timestamp = pa.timestamp; + } else { + _peerAffinities.push_back(pa); + std::sort(_peerAffinities.begin(),_peerAffinities.end()); // probably a more efficient way to insert but okay for now + } + } + } } catch ( ... ) { // ignore invalid identities } diff --git a/node/IncomingPacket.cpp b/node/IncomingPacket.cpp index 7a47c0c6..83e50a9e 100644 --- a/node/IncomingPacket.cpp +++ b/node/IncomingPacket.cpp @@ -43,6 +43,7 @@ #include "Salsa20.hpp" #include "SHA512.hpp" #include "World.hpp" +#include "Cluster.hpp" namespace ZeroTier { @@ -612,8 +613,15 @@ bool IncomingPacket::_doMULTICAST_LIKE(const RuntimeEnvironment *RR,const Shared const uint64_t now = RR->node->now(); // Iterate through 18-byte network,MAC,ADI tuples - for(unsigned int ptr=ZT_PACKET_IDX_PAYLOAD;ptrmc->add(now,at(ptr),MulticastGroup(MAC(field(ptr + 8,6),6),at(ptr + 14)),peer->address()); + for(unsigned int ptr=ZT_PACKET_IDX_PAYLOAD;ptr(ptr)); + const MulticastGroup group(MAC(field(ptr + 8,6),6),at(ptr + 14)); + RR->mc->add(now,nwid,group,peer->address()); +#ifdef ZT_ENABLE_CLUSTER + if (RR->cluster) + RR->cluster->replicateMulticastLike(nwid,peer->address(),group); +#endif + } peer->received(RR,_localAddress,_remoteAddress,hops(),packetId(),Packet::VERB_MULTICAST_LIKE,0,Packet::VERB_NOP); } catch ( ... ) { @@ -870,7 +878,7 @@ bool IncomingPacket::_doPUSH_DIRECT_PATHS(const RuntimeEnvironment *RR,const Sha } peer->setLastDirectPathPushReceived(now); - const RemotePath *currentBest = peer->getBestPath(); + const RemotePath *currentBest = peer->getBestPath(now); unsigned int count = at(ZT_PACKET_IDX_PAYLOAD); unsigned int ptr = ZT_PACKET_IDX_PAYLOAD + 2; diff --git a/node/Peer.cpp b/node/Peer.cpp index 45e2fedd..b01c0c4a 100644 --- a/node/Peer.cpp +++ b/node/Peer.cpp @@ -400,6 +400,11 @@ bool Peer::validateAndSetNetworkMembershipCertificate(const RuntimeEnvironment * _networkComs.set(nwid,_NetworkCom(RR->node->now(),com)); } +#ifdef ZT_ENABLE_CLUSTER + if (RR->cluster) + RR->cluster->replicateCertificateOfNetworkMembership(com); +#endif + return true; } -- cgit v1.2.3 From 2258e36a598e3d91a654ad4938b4fc6fbf05382a Mon Sep 17 00:00:00 2001 From: Adam Ierymenko Date: Tue, 20 Oct 2015 16:34:21 -0700 Subject: Move replication of COMs to avoid race condition. --- node/IncomingPacket.cpp | 16 ++++++++++++++++ node/Peer.cpp | 5 ----- 2 files changed, 16 insertions(+), 5 deletions(-) (limited to 'node') diff --git a/node/IncomingPacket.cpp b/node/IncomingPacket.cpp index 83e50a9e..c8526dfb 100644 --- a/node/IncomingPacket.cpp +++ b/node/IncomingPacket.cpp @@ -406,6 +406,10 @@ bool IncomingPacket::_doOK(const RuntimeEnvironment *RR,const SharedPtr &p CertificateOfMembership com; offset += com.deserialize(*this,ZT_PROTO_VERB_MULTICAST_FRAME__OK__IDX_COM_AND_GATHER_RESULTS); peer->validateAndSetNetworkMembershipCertificate(RR,nwid,com); +#ifdef ZT_ENABLE_CLUSTER + if (RR->cluster) + RR->cluster->replicateCertificateOfNetworkMembership(com); +#endif } if ((flags & 0x02) != 0) { @@ -533,6 +537,10 @@ bool IncomingPacket::_doEXT_FRAME(const RuntimeEnvironment *RR,const SharedPtr

validateAndSetNetworkMembershipCertificate(RR,network->id(),com); +#ifdef ZT_ENABLE_CLUSTER + if (RR->cluster) + RR->cluster->replicateCertificateOfNetworkMembership(com); +#endif } if (!network->isAllowed(peer)) { @@ -639,6 +647,10 @@ bool IncomingPacket::_doNETWORK_MEMBERSHIP_CERTIFICATE(const RuntimeEnvironment while (ptr < size()) { ptr += com.deserialize(*this,ptr); peer->validateAndSetNetworkMembershipCertificate(RR,com.networkId(),com); +#ifdef ZT_ENABLE_CLUSTER + if (RR->cluster) + RR->cluster->replicateCertificateOfNetworkMembership(com); +#endif } peer->received(RR,_localAddress,_remoteAddress,hops(),packetId(),Packet::VERB_NETWORK_MEMBERSHIP_CERTIFICATE,0,Packet::VERB_NOP); @@ -794,6 +806,10 @@ bool IncomingPacket::_doMULTICAST_FRAME(const RuntimeEnvironment *RR,const Share CertificateOfMembership com; offset += com.deserialize(*this,ZT_PROTO_VERB_MULTICAST_FRAME_IDX_COM); peer->validateAndSetNetworkMembershipCertificate(RR,nwid,com); +#ifdef ZT_ENABLE_CLUSTER + if (RR->cluster) + RR->cluster->replicateCertificateOfNetworkMembership(com); +#endif } // Check membership after we've read any included COM, since diff --git a/node/Peer.cpp b/node/Peer.cpp index b01c0c4a..45e2fedd 100644 --- a/node/Peer.cpp +++ b/node/Peer.cpp @@ -400,11 +400,6 @@ bool Peer::validateAndSetNetworkMembershipCertificate(const RuntimeEnvironment * _networkComs.set(nwid,_NetworkCom(RR->node->now(),com)); } -#ifdef ZT_ENABLE_CLUSTER - if (RR->cluster) - RR->cluster->replicateCertificateOfNetworkMembership(com); -#endif - return true; } -- cgit v1.2.3 From 35a12b94ea99e0236ff9455209161180d45e8eae Mon Sep 17 00:00:00 2001 From: Adam Ierymenko Date: Tue, 20 Oct 2015 16:48:49 -0700 Subject: Outfit Cluster with TRACE for debugging. --- node/Cluster.cpp | 28 ++++++++++++++++++++++++++-- 1 file changed, 26 insertions(+), 2 deletions(-) (limited to 'node') diff --git a/node/Cluster.cpp b/node/Cluster.cpp index 8afd3deb..41685a07 100644 --- a/node/Cluster.cpp +++ b/node/Cluster.cpp @@ -172,14 +172,28 @@ void Cluster::handleIncomingStateMessage(const void *msg,unsigned int len) ptr += 8; // skip local clock, not used m.load = dmsg.at(ptr); ptr += 8; ptr += 8; // skip flags, unused +#ifdef ZT_TRACE + std::string addrs; +#endif unsigned int physicalAddressCount = dmsg[ptr++]; for(unsigned int i=0;i 0) + addrs.push_back(','); + addrs.append(m.zeroTierPhysicalEndpoints.back().toString()); + } +#endif } m.lastReceivedAliveAnnouncement = RR->node->now(); +#ifdef ZT_TRACE + TRACE("[%u] I'm alive! send me peers at %s",(unsigned int)fromMemberId,addrs.c_str()); +#endif } break; case STATE_MESSAGE_HAVE_PEER: { @@ -200,6 +214,8 @@ void Cluster::handleIncomingStateMessage(const void *msg,unsigned int len) std::sort(_peerAffinities.begin(),_peerAffinities.end()); // probably a more efficient way to insert but okay for now } } + + TRACE("[%u] has %s",(unsigned int)fromMemberId,id.address().toString().c_str()); } } catch ( ... ) { // ignore invalid identities @@ -212,10 +228,15 @@ void Cluster::handleIncomingStateMessage(const void *msg,unsigned int len) const MAC mac(dmsg.field(ptr,6),6); ptr += 6; const uint32_t adi = dmsg.at(ptr); ptr += 4; RR->mc->add(RR->node->now(),nwid,MulticastGroup(mac,adi),address); + TRACE("[%u] %s likes %s/%u on %.16llu",(unsigned int)fromMemberId,address.toString().c_str(),mac.toString().c_str(),(unsigned int)adi,nwid); } break; case STATE_MESSAGE_COM: { - // TODO: not used yet + CertificateOfMembership com; + ptr += com.deserialize(dmsg,ptr); + if (com) { + TRACE("[%u] COM for %s on %.16llu rev %llu",(unsigned int)fromMemberId,com.issuedTo().toString().c_str(),com.networkId(),com.revision()); + } } break; case STATE_MESSAGE_RELAY: { @@ -228,6 +249,8 @@ void Cluster::handleIncomingStateMessage(const void *msg,unsigned int len) if (packetLen >= ZT_PROTO_MIN_FRAGMENT_LENGTH) { // ignore anything too short to contain a dest address const Address destinationAddress(reinterpret_cast(packet) + 8,ZT_ADDRESS_LENGTH); + TRACE("[%u] relay %u bytes to %s (%u remote paths included)",(unsigned int)fromMemberId,packetLen,destinationAddress.toString().c_str(),numRemotePeerPaths); + SharedPtr destinationPeer(RR->topology->getPeer(destinationAddress)); if (destinationPeer) { if ( @@ -313,6 +336,7 @@ void Cluster::handleIncomingStateMessage(const void *msg,unsigned int len) Packet outp(rcpt,RR->identity.address(),verb); outp.append(dmsg.field(ptr,len),len); RR->sw->send(outp,true,0); + TRACE("[%u] proxy send %s to %s length %u",(unsigned int)fromMemberId,Packet::verbString(verb),rcpt.toString().c_str(),len); } break; } } catch ( ... ) { -- cgit v1.2.3 From d6dee7bb5ccb4ae1fcf818afc246befc9133be72 Mon Sep 17 00:00:00 2001 From: Adam Ierymenko Date: Tue, 20 Oct 2015 17:22:53 -0700 Subject: Clustered handling of relaying. --- node/Cluster.cpp | 47 +++++++++++++++++++++++++++++++++++++++++++++++ node/Cluster.hpp | 11 +++++++++++ node/Switch.cpp | 12 +++++++++++- 3 files changed, 69 insertions(+), 1 deletion(-) (limited to 'node') diff --git a/node/Cluster.cpp b/node/Cluster.cpp index 41685a07..7f4a21b0 100644 --- a/node/Cluster.cpp +++ b/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 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 diff --git a/node/Cluster.hpp b/node/Cluster.hpp index df061c60..60157b15 100644 --- a/node/Cluster.hpp +++ b/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 * diff --git a/node/Switch.cpp b/node/Switch.cpp index 249a21d5..0edaa96d 100644 --- a/node/Switch.cpp +++ b/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 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()); -- cgit v1.2.3 From 6a7b47e5e19687b2dda19449d4f0758388065077 Mon Sep 17 00:00:00 2001 From: Adam Ierymenko Date: Tue, 20 Oct 2015 17:27:57 -0700 Subject: Forgot a return true. --- node/Cluster.cpp | 2 ++ 1 file changed, 2 insertions(+) (limited to 'node') diff --git a/node/Cluster.cpp b/node/Cluster.cpp index 7f4a21b0..abe970e0 100644 --- a/node/Cluster.cpp +++ b/node/Cluster.cpp @@ -394,6 +394,8 @@ bool Cluster::sendViaCluster(const Address &fromPeerAddress,const Address &toPee Mutex::Lock _l2(_members[canHasPeer].lock); _send(canHasPeer,STATE_MESSAGE_RELAY,buf.data(),buf.size()); } + + return true; } return false; -- cgit v1.2.3 From 978b056a0134edd9598a3edb219b31c61405c3b4 Mon Sep 17 00:00:00 2001 From: Adam Ierymenko Date: Tue, 20 Oct 2015 17:36:10 -0700 Subject: Wire in redirectPeer(), now about ready to test clustering! --- node/Cluster.cpp | 8 ++++---- node/Cluster.hpp | 4 ++-- node/Peer.cpp | 16 +++++++++------- 3 files changed, 15 insertions(+), 13 deletions(-) (limited to 'node') diff --git a/node/Cluster.cpp b/node/Cluster.cpp index abe970e0..d9514db5 100644 --- a/node/Cluster.cpp +++ b/node/Cluster.cpp @@ -549,7 +549,7 @@ void Cluster::removeMember(uint16_t memberId) _memberIds = newMemberIds; } -bool Cluster::redirectPeer(const SharedPtr &peer,const InetAddress &peerPhysicalAddress,bool offload) +bool Cluster::redirectPeer(const Address &peerAddress,const InetAddress &peerPhysicalAddress,bool offload) { if (!peerPhysicalAddress) // sanity check return false; @@ -585,7 +585,7 @@ bool Cluster::redirectPeer(const SharedPtr &peer,const InetAddress &peerPh } 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); + TRACE("peer %s is at [%d,%d,%d], distance to us is %f, sending to %u instead for better distance %f",peerAddress.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 @@ -593,7 +593,7 @@ bool Cluster::redirectPeer(const SharedPtr &peer,const InetAddress &peerPh // 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::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); + Packet outp(peerAddress,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()); @@ -611,7 +611,7 @@ bool Cluster::redirectPeer(const SharedPtr &peer,const InetAddress &peerPh 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); + TRACE("peer %s is at [%d,%d,%d], distance to us is %f and this seems to be the best",peerAddress.toString().c_str(),px,py,pz,currentDistance); return false; } } else { diff --git a/node/Cluster.hpp b/node/Cluster.hpp index 60157b15..2e60fd6b 100644 --- a/node/Cluster.hpp +++ b/node/Cluster.hpp @@ -247,12 +247,12 @@ public: /** * Redirect this peer to a better cluster member if needed * - * @param peer Peer to (possibly) redirect + * @param peerAddress 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,const InetAddress &peerPhysicalAddress,bool offload); + bool redirectPeer(const Address &peerAddress,const InetAddress &peerPhysicalAddress,bool offload); private: void _send(uint16_t memberId,StateMessageType type,const void *msg,unsigned int len); diff --git a/node/Peer.cpp b/node/Peer.cpp index 45e2fedd..fa7b3aa4 100644 --- a/node/Peer.cpp +++ b/node/Peer.cpp @@ -82,6 +82,7 @@ void Peer::received( { const uint64_t now = RR->node->now(); bool needMulticastGroupAnnounce = false; + bool pathIsConfirmed = false; { Mutex::Lock _l(_lock); @@ -89,8 +90,6 @@ void Peer::received( _lastReceive = now; if (!hops) { - bool pathIsConfirmed = false; - /* Learn new paths from direct (hops == 0) packets */ { unsigned int np = _numPaths; @@ -141,11 +140,6 @@ void Peer::received( } } } - -#ifdef ZT_ENABLE_CLUSTER - if ((pathIsConfirmed)&&(RR->cluster)) - RR->cluster->replicateHavePeer(_id); -#endif } if ((now - _lastAnnouncedTo) >= ((ZT_MULTICAST_LIKE_EXPIRE / 2) - 1000)) { @@ -159,6 +153,14 @@ void Peer::received( _lastMulticastFrame = now; } +#ifdef ZT_ENABLE_CLUSTER + if ((pathIsConfirmed)&&(RR->cluster)) { + // Either shuttle this peer off somewhere else or report to other members that we have it + if (!RR->cluster->redirectPeer(_id.address(),remoteAddr,false)) + RR->cluster->replicateHavePeer(_id); + } +#endif + if (needMulticastGroupAnnounce) { const std::vector< SharedPtr > networks(RR->node->allNetworks()); for(std::vector< SharedPtr >::const_iterator n(networks.begin());n!=networks.end();++n) -- cgit v1.2.3