Each peer now tracks the last time it announced multicast LIKEs independently and does so frequently enough to prevent expires. Also add a multicast debug facility for use on the testnet.

This commit is contained in:
Adam Ierymenko 2013-10-01 16:01:36 -04:00
parent 1a76455986
commit 3443b203e4
10 changed files with 106 additions and 41 deletions

View file

@ -250,19 +250,10 @@ error_no_ZT_ARCH_defined;
*/ */
#define ZT_MULTICAST_GLOBAL_MAX_DEPTH 500 #define ZT_MULTICAST_GLOBAL_MAX_DEPTH 500
/**
* Period between announcements of all multicast 'likes' in ms
*
* Announcement occurs when a multicast group is locally joined, but all
* memberships are periodically re-broadcast. If they're not they will
* expire.
*/
#define ZT_MULTICAST_LIKE_ANNOUNCE_ALL_PERIOD 120000
/** /**
* Expire time for multicast 'likes' in ms * Expire time for multicast 'likes' in ms
*/ */
#define ZT_MULTICAST_LIKE_EXPIRE ((ZT_MULTICAST_LIKE_ANNOUNCE_ALL_PERIOD * 2) + 1000) #define ZT_MULTICAST_LIKE_EXPIRE 120000
/** /**
* Time between polls of local taps for multicast membership changes * Time between polls of local taps for multicast membership changes

View file

@ -101,6 +101,9 @@ static inline std::string _mkDefaultHomePath()
Defaults::Defaults() Defaults::Defaults()
throw(std::runtime_error) : throw(std::runtime_error) :
#ifdef ZT_TRACE_MULTICAST
multicastTraceWatcher(ZT_TRACE_MULTICAST),
#endif
defaultHomePath(_mkDefaultHomePath()), defaultHomePath(_mkDefaultHomePath()),
supernodes(_mkSupernodeMap()) supernodes(_mkSupernodeMap())
{ {

View file

@ -33,6 +33,7 @@
#include <vector> #include <vector>
#include <map> #include <map>
#include "Constants.hpp"
#include "Identity.hpp" #include "Identity.hpp"
#include "InetAddress.hpp" #include "InetAddress.hpp"
@ -52,6 +53,13 @@ public:
throw(std::runtime_error); throw(std::runtime_error);
~Defaults() {} ~Defaults() {}
#ifdef ZT_TRACE_MULTICAST
/**
* Host to send UDP multicast trace messages to (human readable)
*/
const InetAddress multicastTraceWatcher;
#endif
/** /**
* Default home path for this platform * Default home path for this platform
*/ */

View file

@ -109,6 +109,12 @@ public:
this->fromString(ipSlashPort); this->fromString(ipSlashPort);
} }
InetAddress(const char *ipSlashPort)
throw()
{
this->fromString(std::string(ipSlashPort));
}
inline InetAddress &operator=(const InetAddress &a) inline InetAddress &operator=(const InetAddress &a)
throw() throw()
{ {

View file

@ -425,7 +425,6 @@ Node::ReasonForTermination Node::run()
uint64_t lastNetworkFingerprintCheck = 0; uint64_t lastNetworkFingerprintCheck = 0;
uint64_t networkConfigurationFingerprint = _r->sysEnv->getNetworkConfigurationFingerprint(); uint64_t networkConfigurationFingerprint = _r->sysEnv->getNetworkConfigurationFingerprint();
uint64_t lastMulticastCheck = 0; uint64_t lastMulticastCheck = 0;
uint64_t lastMulticastAnnounceAll = 0;
long lastDelayDelta = 0; long lastDelayDelta = 0;
while (impl->reasonForTermination == NODE_RUNNING) { while (impl->reasonForTermination == NODE_RUNNING) {
@ -468,27 +467,15 @@ Node::ReasonForTermination Node::run()
// those changes to peers. // those changes to peers.
if ((resynchronize)||((now - lastMulticastCheck) >= ZT_MULTICAST_LOCAL_POLL_PERIOD)) { if ((resynchronize)||((now - lastMulticastCheck) >= ZT_MULTICAST_LOCAL_POLL_PERIOD)) {
lastMulticastCheck = now; lastMulticastCheck = now;
bool announceAll = ((resynchronize)||((now - lastMulticastAnnounceAll) >= ZT_MULTICAST_LIKE_ANNOUNCE_ALL_PERIOD));
try { try {
std::map< SharedPtr<Network>,std::set<MulticastGroup> > toAnnounce; std::map< SharedPtr<Network>,std::set<MulticastGroup> > toAnnounce;
{
std::vector< SharedPtr<Network> > networks(_r->nc->networks()); std::vector< SharedPtr<Network> > networks(_r->nc->networks());
for(std::vector< SharedPtr<Network> >::const_iterator nw(networks.begin());nw!=networks.end();++nw) { for(std::vector< SharedPtr<Network> >::const_iterator nw(networks.begin());nw!=networks.end();++nw) {
if (((*nw)->updateMulticastGroups())||(announceAll)) if ((*nw)->updateMulticastGroups())
toAnnounce.insert(std::pair< SharedPtr<Network>,std::set<MulticastGroup> >(*nw,(*nw)->multicastGroups())); toAnnounce.insert(std::pair< SharedPtr<Network>,std::set<MulticastGroup> >(*nw,(*nw)->multicastGroups()));
} }
} if (toAnnounce.size())
if (toAnnounce.size()) {
_r->sw->announceMulticastGroups(toAnnounce); _r->sw->announceMulticastGroups(toAnnounce);
// Only update lastMulticastAnnounceAll if we've announced something. This keeps
// the announceAll condition true during startup when there are no multicast
// groups until there is at least one. Technically this shouldn't be required as
// updateMulticastGroups() should return true on any change, but why not?
if (announceAll)
lastMulticastAnnounceAll = now;
}
} catch (std::exception &exc) { } catch (std::exception &exc) {
LOG("unexpected exception announcing multicast groups: %s",exc.what()); LOG("unexpected exception announcing multicast groups: %s",exc.what());
} catch ( ... ) { } catch ( ... ) {

View file

@ -28,6 +28,7 @@
#include "../version.h" #include "../version.h"
#include "Constants.hpp" #include "Constants.hpp"
#include "Defaults.hpp"
#include "RuntimeEnvironment.hpp" #include "RuntimeEnvironment.hpp"
#include "Topology.hpp" #include "Topology.hpp"
#include "PacketDecoder.hpp" #include "PacketDecoder.hpp"
@ -36,6 +37,7 @@
#include "NodeConfig.hpp" #include "NodeConfig.hpp"
#include "Filter.hpp" #include "Filter.hpp"
#include "Service.hpp" #include "Service.hpp"
#include "Demarc.hpp"
namespace ZeroTier { namespace ZeroTier {
@ -461,7 +463,7 @@ bool PacketDecoder::_doMULTICAST_FRAME(const RuntimeEnvironment *_r,const Shared
return false; return false;
} }
uint16_t depth = at<uint16_t>(ZT_PROTO_VERB_MULTICAST_FRAME_IDX_PROPAGATION_DEPTH); unsigned int depth = at<uint16_t>(ZT_PROTO_VERB_MULTICAST_FRAME_IDX_PROPAGATION_DEPTH);
unsigned char *fifo = field(ZT_PROTO_VERB_MULTICAST_FRAME_IDX_PROPAGATION_FIFO,ZT_PROTO_VERB_MULTICAST_FRAME_LEN_PROPAGATION_FIFO); unsigned char *fifo = field(ZT_PROTO_VERB_MULTICAST_FRAME_IDX_PROPAGATION_FIFO,ZT_PROTO_VERB_MULTICAST_FRAME_LEN_PROPAGATION_FIFO);
unsigned char *bloom = field(ZT_PROTO_VERB_MULTICAST_FRAME_IDX_PROPAGATION_BLOOM,ZT_PROTO_VERB_MULTICAST_FRAME_LEN_PROPAGATION_BLOOM); unsigned char *bloom = field(ZT_PROTO_VERB_MULTICAST_FRAME_IDX_PROPAGATION_BLOOM,ZT_PROTO_VERB_MULTICAST_FRAME_LEN_PROPAGATION_BLOOM);
uint64_t nwid = at<uint64_t>(ZT_PROTO_VERB_MULTICAST_FRAME_IDX_NETWORK_ID); uint64_t nwid = at<uint64_t>(ZT_PROTO_VERB_MULTICAST_FRAME_IDX_NETWORK_ID);
@ -483,6 +485,12 @@ bool PacketDecoder::_doMULTICAST_FRAME(const RuntimeEnvironment *_r,const Shared
return true; return true;
} }
#ifdef ZT_TRACE_MULTICAST
char mct[256];
Utils::snprintf(mct,sizeof(mct),"%s <- %.16llx %.16llx %s via %s depth:%u len:%u",_r->identity.address().toString().c_str(),nwid,guid,origin.toString().c_str(),source().toString().c_str(),depth,frameLen);
_r->demarc->send(Demarc::ANY_PORT,ZT_DEFAULTS.multicastTraceWatcher,mct,strlen(mct),-1);
#endif
if (!dest.mac().isMulticast()) { if (!dest.mac().isMulticast()) {
TRACE("dropped MULTICAST_FRAME from %s(%s): %s is not a multicast/broadcast address",source().toString().c_str(),_remoteAddress.toString().c_str(),dest.mac().toString().c_str()); TRACE("dropped MULTICAST_FRAME from %s(%s): %s is not a multicast/broadcast address",source().toString().c_str(),_remoteAddress.toString().c_str(),dest.mac().toString().c_str());
return true; return true;
@ -589,7 +597,11 @@ bool PacketDecoder::_doMULTICAST_FRAME(const RuntimeEnvironment *_r,const Shared
// The rest of newFifo[] goes back into the packet // The rest of newFifo[] goes back into the packet
memcpy(fifo,newFifo + ZT_ADDRESS_LENGTH,ZT_PROTO_VERB_MULTICAST_FRAME_LEN_PROPAGATION_FIFO); memcpy(fifo,newFifo + ZT_ADDRESS_LENGTH,ZT_PROTO_VERB_MULTICAST_FRAME_LEN_PROPAGATION_FIFO);
//TRACE("forwarding MULTICAST_FRAME from %s(%s) to %s, original sender %s, current depth: %u",source().toString().c_str(),_remoteAddress.toString().c_str(),nextHop.toString().c_str(),origin.toString().c_str(),depth); #ifdef ZT_TRACE_MULTICAST
char mct[256];
Utils::snprintf(mct,sizeof(mct),"%s -> %.16llx %.16llx %s via %s",_r->identity.address().toString().c_str(),nwid,guid,origin.toString().c_str(),nextHop.toString().c_str());
_r->demarc->send(Demarc::ANY_PORT,ZT_DEFAULTS.multicastTraceWatcher,mct,strlen(mct),-1);
#endif
// Send to next hop, reusing this packet as scratch space // Send to next hop, reusing this packet as scratch space
newInitializationVector(); newInitializationVector();

View file

@ -26,6 +26,7 @@
*/ */
#include "Peer.hpp" #include "Peer.hpp"
#include "Switch.hpp"
namespace ZeroTier { namespace ZeroTier {
@ -66,6 +67,12 @@ void Peer::onReceive(const RuntimeEnvironment *_r,Demarc::Port localPort,const I
wp->localPort = localPort; wp->localPort = localPort;
if (!wp->fixed) if (!wp->fixed)
wp->addr = remoteAddr; wp->addr = remoteAddr;
if ((now - _lastAnnouncedTo) >= ((ZT_MULTICAST_LIKE_EXPIRE / 2) - 1000)) {
_lastAnnouncedTo = now;
_r->sw->announceMulticastGroups(SharedPtr<Peer>(this));
}
_dirty = true; _dirty = true;
} }

View file

@ -62,8 +62,8 @@
16 + \ 16 + \
1 \ 1 \
) * 2) + \ ) * 2) + \
sizeof(uint64_t) + \ (sizeof(uint64_t) * 3) + \
sizeof(uint64_t) \ (sizeof(uint16_t) * 3) \
) )
namespace ZeroTier { namespace ZeroTier {
@ -213,6 +213,15 @@ public:
return std::max(_lastUnicastFrame,_lastMulticastFrame); return std::max(_lastUnicastFrame,_lastMulticastFrame);
} }
/**
* @return Time we last announced state TO this peer, such as multicast LIKEs
*/
uint64_t lastAnnouncedTo() const
throw()
{
return _lastAnnouncedTo;
}
/** /**
* @return Lowest of measured latencies of all paths or 0 if unknown * @return Lowest of measured latencies of all paths or 0 if unknown
*/ */
@ -367,6 +376,10 @@ public:
_ipv6p.serialize(b); _ipv6p.serialize(b);
b.append(_lastUnicastFrame); b.append(_lastUnicastFrame);
b.append(_lastMulticastFrame); b.append(_lastMulticastFrame);
b.append(_lastAnnouncedTo);
b.append((uint16_t)_vMajor);
b.append((uint16_t)_vMinor);
b.append((uint16_t)_vRevision);
} }
template<unsigned int C> template<unsigned int C>
@ -384,10 +397,11 @@ public:
p += _ipv6p.deserialize(b,p); p += _ipv6p.deserialize(b,p);
_lastUnicastFrame = b.template at<uint64_t>(p); p += sizeof(uint64_t); _lastUnicastFrame = b.template at<uint64_t>(p); p += sizeof(uint64_t);
_lastMulticastFrame = b.template at<uint64_t>(p); p += sizeof(uint64_t); _lastMulticastFrame = b.template at<uint64_t>(p); p += sizeof(uint64_t);
_lastAnnouncedTo = b.template at<uint64_t>(p); p += sizeof(uint64_t);
_vMajor = b.template at<uint16_t>(p); p += sizeof(uint16_t);
_vMinor = b.template at<uint16_t>(p); p += sizeof(uint16_t);
_vRevision = b.template at<uint16_t>(p); p += sizeof(uint16_t);
_vMajor = 0;
_vMinor = 0;
_vRevision = 0;
_dirty = false; _dirty = false;
return (p - startAt); return (p - startAt);
@ -516,12 +530,12 @@ private:
uint64_t _lastUnicastFrame; uint64_t _lastUnicastFrame;
uint64_t _lastMulticastFrame; uint64_t _lastMulticastFrame;
uint64_t _lastAnnouncedTo;
// Fields below this line are not persisted with serialize()
unsigned int _vMajor,_vMinor,_vRevision; unsigned int _vMajor,_vMinor,_vRevision;
bool _dirty;
// Fields below this line are not persisted with serialize() ---------------
bool _dirty;
AtomicCounter __refCount; AtomicCounter __refCount;
}; };

View file

@ -400,6 +400,7 @@ void Switch::announceMulticastGroups(const std::map< SharedPtr<Network>,std::set
outp.reset((*p)->address(),_r->identity.address(),Packet::VERB_MULTICAST_LIKE); outp.reset((*p)->address(),_r->identity.address(),Packet::VERB_MULTICAST_LIKE);
} }
// network ID, MAC, ADI
outp.append((uint64_t)nwmgs->first->id()); outp.append((uint64_t)nwmgs->first->id());
outp.append(mg->mac().data,6); outp.append(mg->mac().data,6);
outp.append((uint32_t)mg->adi()); outp.append((uint32_t)mg->adi());
@ -412,6 +413,30 @@ void Switch::announceMulticastGroups(const std::map< SharedPtr<Network>,std::set
} }
} }
void Switch::announceMulticastGroups(const SharedPtr<Peer> &peer)
{
Packet outp(peer->address(),_r->identity.address(),Packet::VERB_MULTICAST_LIKE);
std::vector< SharedPtr<Network> > networks(_r->nc->networks());
for(std::vector< SharedPtr<Network> >::iterator n(networks.begin());n!=networks.end();++n) {
if (((*n)->isAllowed(peer->address()))||(_r->topology->isSupernode(peer->address()))) {
std::set<MulticastGroup> mgs((*n)->multicastGroups());
for(std::set<MulticastGroup>::iterator mg(mgs.begin());mg!=mgs.end();++mg) {
if ((outp.size() + 18) > ZT_UDP_DEFAULT_PAYLOAD_MTU) {
send(outp,true);
outp.reset(peer->address(),_r->identity.address(),Packet::VERB_MULTICAST_LIKE);
}
// network ID, MAC, ADI
outp.append((uint64_t)(*n)->id());
outp.append(mg->mac().data,6);
outp.append((uint32_t)mg->adi());
}
}
}
if (outp.size() > ZT_PROTO_MIN_PACKET_LENGTH)
send(outp,true);
}
void Switch::requestWhois(const Address &addr) void Switch::requestWhois(const Address &addr)
{ {
TRACE("requesting WHOIS for %s",addr.toString().c_str()); TRACE("requesting WHOIS for %s",addr.toString().c_str());

View file

@ -153,13 +153,25 @@ public:
/** /**
* Announce multicast group memberships * Announce multicast group memberships
* *
* This efficiently announces memberships, sending single packets with * This announces all the groups for all the networks in the supplied map to
* many LIKEs. * all peers with whom we have an active direct link. Only isAllowed() peers
* and supernodes get announcements for each given network.
* *
* @param allMemberships Memberships for a number of networks * @param allMemberships Memberships for a number of networks
*/ */
void announceMulticastGroups(const std::map< SharedPtr<Network>,std::set<MulticastGroup> > &allMemberships); void announceMulticastGroups(const std::map< SharedPtr<Network>,std::set<MulticastGroup> > &allMemberships);
/**
* Announce multicast group memberships
*
* This announces all current multicast memberships to a single peer. Only
* memberships for networks where the peer isAllowed() are included, unless
* the peer is a supernode.
*
* @param peer Peer to announce all memberships to
*/
void announceMulticastGroups(const SharedPtr<Peer> &peer);
/** /**
* Request WHOIS on a given address * Request WHOIS on a given address
* *