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 +++++++++++++++++++++++++++++++++++++++++++++++++------ 1 file changed, 172 insertions(+), 21 deletions(-) (limited to 'node/Cluster.cpp') 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 } } -- 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/Cluster.cpp') 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/Cluster.cpp') 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 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/Cluster.cpp') 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/Cluster.cpp') 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/Cluster.cpp') 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/Cluster.cpp') 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