Rest of GitHub issue #140 implementation.

This commit is contained in:
Adam Ierymenko 2015-06-01 17:50:44 -07:00
parent 318c2f025b
commit 960ceb4791
4 changed files with 44 additions and 31 deletions

View file

@ -264,7 +264,7 @@ void Network::requestConfiguration()
outp.append((uint64_t)_config->revision()); outp.append((uint64_t)_config->revision());
else outp.append((uint64_t)0); else outp.append((uint64_t)0);
} }
RR->sw->send(outp,true); RR->sw->send(outp,true,_id);
} }
void Network::addMembershipCertificate(const CertificateOfMembership &cert,bool forceAccept) void Network::addMembershipCertificate(const CertificateOfMembership &cert,bool forceAccept)

View file

@ -108,14 +108,14 @@ void OutboundMulticast::sendOnly(const RuntimeEnvironment *RR,const Address &toA
_packetWithCom.newInitializationVector(); _packetWithCom.newInitializationVector();
_packetWithCom.setDestination(toAddr); _packetWithCom.setDestination(toAddr);
//TRACE(">>MC %.16llx -> %s (with COM)",(unsigned long long)this,toAddr.toString().c_str()); //TRACE(">>MC %.16llx -> %s (with COM)",(unsigned long long)this,toAddr.toString().c_str());
RR->sw->send(_packetWithCom,true); RR->sw->send(_packetWithCom,true,_nwid);
return; return;
} }
} }
//TRACE(">>MC %.16llx -> %s (without COM)",(unsigned long long)this,toAddr.toString().c_str()); //TRACE(">>MC %.16llx -> %s (without COM)",(unsigned long long)this,toAddr.toString().c_str());
_packetNoCom.newInitializationVector(); _packetNoCom.newInitializationVector();
_packetNoCom.setDestination(toAddr); _packetNoCom.setDestination(toAddr);
RR->sw->send(_packetNoCom,true); RR->sw->send(_packetNoCom,true,_nwid);
} }
} // namespace ZeroTier } // namespace ZeroTier

View file

@ -114,8 +114,6 @@ void Switch::onLocalEthernet(const SharedPtr<Network> &network,const MAC &from,c
if (to.isMulticast()) { if (to.isMulticast()) {
// Destination is a multicast address (including broadcast) // Destination is a multicast address (including broadcast)
const uint64_t now = RR->node->now();
MulticastGroup mg(to,0); MulticastGroup mg(to,0);
if (to.isBroadcast()) { if (to.isBroadcast()) {
@ -145,7 +143,7 @@ void Switch::onLocalEthernet(const SharedPtr<Network> &network,const MAC &from,c
* multicast addresses on bridge interfaces and subscribing each slave. * multicast addresses on bridge interfaces and subscribing each slave.
* But in that case this does no harm, as the sets are just merged. */ * But in that case this does no harm, as the sets are just merged. */
if (fromBridged) if (fromBridged)
network->learnBridgedMulticastGroup(mg,now); network->learnBridgedMulticastGroup(mg,RR->node->now());
// Check multicast/broadcast bandwidth quotas and reject if quota exceeded // Check multicast/broadcast bandwidth quotas and reject if quota exceeded
if (!network->updateAndCheckMulticastBalance(mg,len)) { if (!network->updateAndCheckMulticastBalance(mg,len)) {
@ -158,7 +156,7 @@ void Switch::onLocalEthernet(const SharedPtr<Network> &network,const MAC &from,c
RR->mc->send( RR->mc->send(
((!nconf->isPublic())&&(nconf->com())) ? &(nconf->com()) : (const CertificateOfMembership *)0, ((!nconf->isPublic())&&(nconf->com())) ? &(nconf->com()) : (const CertificateOfMembership *)0,
nconf->multicastLimit(), nconf->multicastLimit(),
now, RR->node->now(),
network->id(), network->id(),
nconf->activeBridges(), nconf->activeBridges(),
mg, mg,
@ -180,7 +178,7 @@ void Switch::onLocalEthernet(const SharedPtr<Network> &network,const MAC &from,c
// bundle this with EXT_FRAME instead of sending two packets. // bundle this with EXT_FRAME instead of sending two packets.
Packet outp(toZT,RR->identity.address(),Packet::VERB_NETWORK_MEMBERSHIP_CERTIFICATE); Packet outp(toZT,RR->identity.address(),Packet::VERB_NETWORK_MEMBERSHIP_CERTIFICATE);
nconf->com().serialize(outp); nconf->com().serialize(outp);
send(outp,true); send(outp,true,network->id());
} }
if (fromBridged) { if (fromBridged) {
@ -193,7 +191,7 @@ void Switch::onLocalEthernet(const SharedPtr<Network> &network,const MAC &from,c
outp.append((uint16_t)etherType); outp.append((uint16_t)etherType);
outp.append(data,len); outp.append(data,len);
outp.compress(); outp.compress();
send(outp,true); send(outp,true,network->id());
} else { } else {
// FRAME is a shorter version that can be used when there's no bridging and no COM // FRAME is a shorter version that can be used when there's no bridging and no COM
Packet outp(toZT,RR->identity.address(),Packet::VERB_FRAME); Packet outp(toZT,RR->identity.address(),Packet::VERB_FRAME);
@ -201,7 +199,7 @@ void Switch::onLocalEthernet(const SharedPtr<Network> &network,const MAC &from,c
outp.append((uint16_t)etherType); outp.append((uint16_t)etherType);
outp.append(data,len); outp.append(data,len);
outp.compress(); outp.compress();
send(outp,true); send(outp,true,network->id());
} }
//TRACE("%.16llx: UNICAST: %s -> %s etherType==%s(%.4x) vlanId==%u len==%u fromBridged==%d",network->id(),from.toString().c_str(),to.toString().c_str(),etherTypeName(etherType),etherType,vlanId,len,(int)fromBridged); //TRACE("%.16llx: UNICAST: %s -> %s etherType==%s(%.4x) vlanId==%u len==%u fromBridged==%d",network->id(),from.toString().c_str(),to.toString().c_str(),etherTypeName(etherType),etherType,vlanId,len,(int)fromBridged);
@ -259,21 +257,21 @@ void Switch::onLocalEthernet(const SharedPtr<Network> &network,const MAC &from,c
outp.append((uint16_t)etherType); outp.append((uint16_t)etherType);
outp.append(data,len); outp.append(data,len);
outp.compress(); outp.compress();
send(outp,true); send(outp,true,network->id());
} }
} }
} }
void Switch::send(const Packet &packet,bool encrypt) void Switch::send(const Packet &packet,bool encrypt,uint64_t nwid)
{ {
if (packet.destination() == RR->identity.address()) { if (packet.destination() == RR->identity.address()) {
TRACE("BUG: caught attempt to send() to self, ignored"); TRACE("BUG: caught attempt to send() to self, ignored");
return; return;
} }
if (!_trySend(packet,encrypt)) { if (!_trySend(packet,encrypt,nwid)) {
Mutex::Lock _l(_txQueue_m); Mutex::Lock _l(_txQueue_m);
_txQueue.insert(std::pair< Address,TXQueueEntry >(packet.destination(),TXQueueEntry(RR->node->now(),packet,encrypt))); _txQueue.insert(std::pair< Address,TXQueueEntry >(packet.destination(),TXQueueEntry(RR->node->now(),packet,encrypt,nwid)));
} }
} }
@ -423,7 +421,7 @@ void Switch::doAnythingWaitingForPeer(const SharedPtr<Peer> &peer)
Mutex::Lock _l(_txQueue_m); Mutex::Lock _l(_txQueue_m);
std::pair< std::multimap< Address,TXQueueEntry >::iterator,std::multimap< Address,TXQueueEntry >::iterator > waitingTxQueueItems(_txQueue.equal_range(peer->address())); std::pair< std::multimap< Address,TXQueueEntry >::iterator,std::multimap< Address,TXQueueEntry >::iterator > waitingTxQueueItems(_txQueue.equal_range(peer->address()));
for(std::multimap< Address,TXQueueEntry >::iterator txi(waitingTxQueueItems.first);txi!=waitingTxQueueItems.second;) { for(std::multimap< Address,TXQueueEntry >::iterator txi(waitingTxQueueItems.first);txi!=waitingTxQueueItems.second;) {
if (_trySend(txi->second.packet,txi->second.encrypt)) if (_trySend(txi->second.packet,txi->second.encrypt,txi->second.nwid))
_txQueue.erase(txi++); _txQueue.erase(txi++);
else ++txi; else ++txi;
} }
@ -505,7 +503,7 @@ unsigned long Switch::doTimerTasks(uint64_t now)
{ // Time out TX queue packets that never got WHOIS lookups or other info. { // Time out TX queue packets that never got WHOIS lookups or other info.
Mutex::Lock _l(_txQueue_m); Mutex::Lock _l(_txQueue_m);
for(std::multimap< Address,TXQueueEntry >::iterator i(_txQueue.begin());i!=_txQueue.end();) { for(std::multimap< Address,TXQueueEntry >::iterator i(_txQueue.begin());i!=_txQueue.end();) {
if (_trySend(i->second.packet,i->second.encrypt)) if (_trySend(i->second.packet,i->second.encrypt,i->second.nwid))
_txQueue.erase(i++); _txQueue.erase(i++);
else if ((now - i->second.creationTime) > ZT_TRANSMIT_QUEUE_TIMEOUT) { else if ((now - i->second.creationTime) > ZT_TRANSMIT_QUEUE_TIMEOUT) {
TRACE("TX %s -> %s timed out",i->second.packet.source().toString().c_str(),i->second.packet.destination().toString().c_str()); TRACE("TX %s -> %s timed out",i->second.packet.source().toString().c_str(),i->second.packet.destination().toString().c_str());
@ -725,7 +723,7 @@ Address Switch::_sendWhoisRequest(const Address &addr,const Address *peersAlread
return Address(); return Address();
} }
bool Switch::_trySend(const Packet &packet,bool encrypt) bool Switch::_trySend(const Packet &packet,bool encrypt,uint64_t nwid)
{ {
SharedPtr<Peer> peer(RR->topology->getPeer(packet.destination())); SharedPtr<Peer> peer(RR->topology->getPeer(packet.destination()));
@ -734,8 +732,27 @@ bool Switch::_trySend(const Packet &packet,bool encrypt)
Path *viaPath = peer->getBestPath(now); Path *viaPath = peer->getBestPath(now);
if (!viaPath) { if (!viaPath) {
SharedPtr<Peer> sn(RR->topology->getBestSupernode()); SharedPtr<Peer> relay;
if (!(sn)||(!(viaPath = sn->getBestPath(now))))
if (nwid) {
SharedPtr<Network> network(RR->node->network(nwid));
if (network) {
SharedPtr<NetworkConfig> nconf(network->config2());
if (nconf) {
unsigned int latency = ~((unsigned int)0);
for(std::vector< std::pair<Address,InetAddress> >::const_iterator r(nconf->relays().begin());r!=nconf->relays().end();++r) {
SharedPtr<Peer> rp(RR->topology->getPeer(r->first));
if ((rp->hasActiveDirectPath(now))&&(rp->latency() <= latency))
rp.swap(relay);
}
}
}
}
if (!relay)
relay = RR->topology->getBestSupernode();
if (!(relay)||(!(viaPath = relay->getBestPath(now))))
return false; return false;
} }

View file

@ -111,8 +111,9 @@ public:
* *
* @param packet Packet to send * @param packet Packet to send
* @param encrypt Encrypt packet payload? (always true except for HELLO) * @param encrypt Encrypt packet payload? (always true except for HELLO)
* @param nwid Network ID or 0 if message is not related to a specific network
*/ */
void send(const Packet &packet,bool encrypt); void send(const Packet &packet,bool encrypt,uint64_t nwid);
/** /**
* Send RENDEZVOUS to two peers to permit them to directly connect * Send RENDEZVOUS to two peers to permit them to directly connect
@ -183,15 +184,8 @@ private:
void _handleRemotePacketFragment(const InetAddress &fromAddr,const void *data,unsigned int len); void _handleRemotePacketFragment(const InetAddress &fromAddr,const void *data,unsigned int len);
void _handleRemotePacketHead(const InetAddress &fromAddr,const void *data,unsigned int len); void _handleRemotePacketHead(const InetAddress &fromAddr,const void *data,unsigned int len);
void _handleBeacon(const InetAddress &fromAddr,const Buffer<ZT_PROTO_BEACON_LENGTH> &data); void _handleBeacon(const InetAddress &fromAddr,const Buffer<ZT_PROTO_BEACON_LENGTH> &data);
Address _sendWhoisRequest(const Address &addr,const Address *peersAlreadyConsulted,unsigned int numPeersAlreadyConsulted);
Address _sendWhoisRequest( bool _trySend(const Packet &packet,bool encrypt,uint64_t nwid);
const Address &addr,
const Address *peersAlreadyConsulted,
unsigned int numPeersAlreadyConsulted);
bool _trySend(
const Packet &packet,
bool encrypt);
const RuntimeEnvironment *const RR; const RuntimeEnvironment *const RR;
volatile uint64_t _lastBeacon; volatile uint64_t _lastBeacon;
@ -226,13 +220,15 @@ private:
struct TXQueueEntry struct TXQueueEntry
{ {
TXQueueEntry() {} TXQueueEntry() {}
TXQueueEntry(uint64_t ct,const Packet &p,bool enc) : TXQueueEntry(uint64_t ct,const Packet &p,bool enc,uint64_t nw) :
creationTime(ct), creationTime(ct),
nwid(nw),
packet(p), packet(p),
encrypt(enc) {} encrypt(enc) {}
uint64_t creationTime; uint64_t creationTime;
Packet packet; // unencrypted/untagged for TX queue uint64_t nwid;
Packet packet; // unencrypted/unMAC'd packet -- this is done at send time
bool encrypt; bool encrypt;
}; };
std::multimap< Address,TXQueueEntry > _txQueue; std::multimap< Address,TXQueueEntry > _txQueue;