2014-09-19 01:28:14 +00:00
/*
2015-02-17 21:11:34 +00:00
* ZeroTier One - Network Virtualization Everywhere
* Copyright ( C ) 2011 - 2015 ZeroTier , Inc .
2014-09-19 01:28:14 +00:00
*
* This program is free software : you can redistribute it and / or modify
* it under the terms of the GNU General Public License as published by
* the Free Software Foundation , either version 3 of the License , or
* ( at your option ) any later version .
*
* This program is distributed in the hope that it will be useful ,
* but WITHOUT ANY WARRANTY ; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE . See the
* GNU General Public License for more details .
*
* You should have received a copy of the GNU General Public License
* along with this program . If not , see < http : //www.gnu.org/licenses/>.
*
* - -
*
* ZeroTier may be used and distributed under the terms of the GPLv3 , which
* are available at : http : //www.gnu.org/licenses/gpl-3.0.html
*
* If you would like to embed ZeroTier into a commercial application or
* redistribute it in a modified binary form , please contact ZeroTier Networks
* LLC . Start here : http : //www.zerotier.com/
*/
# include <algorithm>
# include "Constants.hpp"
2014-10-10 00:58:31 +00:00
# include "RuntimeEnvironment.hpp"
2014-09-25 22:57:43 +00:00
# include "SharedPtr.hpp"
2014-09-24 21:02:16 +00:00
# include "Multicaster.hpp"
2014-09-19 01:28:14 +00:00
# include "Topology.hpp"
2014-09-25 22:08:29 +00:00
# include "Switch.hpp"
# include "Packet.hpp"
2014-09-25 22:57:43 +00:00
# include "Peer.hpp"
2014-09-30 23:28:25 +00:00
# include "CMWC4096.hpp"
2014-10-04 20:46:29 +00:00
# include "C25519.hpp"
2014-09-26 21:18:25 +00:00
# include "CertificateOfMembership.hpp"
2014-09-19 01:28:14 +00:00
namespace ZeroTier {
2014-10-01 21:05:25 +00:00
Multicaster : : Multicaster ( const RuntimeEnvironment * renv ) :
RR ( renv )
2014-09-19 01:28:14 +00:00
{
}
2014-09-24 21:02:16 +00:00
Multicaster : : ~ Multicaster ( )
2014-09-19 01:28:14 +00:00
{
}
2014-11-21 18:50:27 +00:00
void Multicaster : : addMultiple ( uint64_t now , uint64_t nwid , const MulticastGroup & mg , const void * addresses , unsigned int count , unsigned int totalKnown )
2014-10-29 22:26:32 +00:00
{
const unsigned char * p = ( const unsigned char * ) addresses ;
const unsigned char * e = p + ( 5 * count ) ;
Mutex : : Lock _l ( _groups_m ) ;
MulticastGroupStatus & gs = _groups [ std : : pair < uint64_t , MulticastGroup > ( nwid , mg ) ] ;
while ( p ! = e ) {
2014-11-21 18:50:27 +00:00
_add ( now , nwid , mg , gs , Address ( p , 5 ) ) ;
2014-10-29 22:26:32 +00:00
p + = 5 ;
}
}
2014-10-05 17:34:25 +00:00
unsigned int Multicaster : : gather ( const Address & queryingPeer , uint64_t nwid , const MulticastGroup & mg , Packet & appendTo , unsigned int limit ) const
2014-09-30 23:28:25 +00:00
{
unsigned char * p ;
2014-10-10 01:32:05 +00:00
unsigned int added = 0 , i , k , rptr , totalKnown = 0 ;
2014-10-28 22:33:10 +00:00
uint64_t a , picked [ ( ZT_PROTO_MAX_PACKET_LENGTH / 5 ) + 2 ] ;
2014-09-30 23:28:25 +00:00
2014-10-10 01:32:05 +00:00
if ( ! limit )
2014-09-30 23:28:25 +00:00
return 0 ;
2014-10-29 23:24:19 +00:00
else if ( limit > 0xffff )
2014-10-10 01:32:05 +00:00
limit = 0xffff ;
2014-10-29 23:24:19 +00:00
const unsigned int totalAt = appendTo . size ( ) ;
appendTo . addSize ( 4 ) ; // sizeof(uint32_t)
const unsigned int addedAt = appendTo . size ( ) ;
appendTo . addSize ( 2 ) ; // sizeof(uint16_t)
2014-10-10 01:32:05 +00:00
{ // Return myself if I am a member of this group
2015-04-06 22:14:54 +00:00
SharedPtr < Network > network ( RR - > node - > network ( nwid ) ) ;
2015-04-07 02:34:36 +00:00
if ( ( network ) & & ( network - > subscribedToMulticastGroup ( mg , true ) ) ) {
2014-10-10 01:32:05 +00:00
RR - > identity . address ( ) . appendTo ( appendTo ) ;
+ + totalKnown ;
+ + added ;
}
2014-09-30 23:28:25 +00:00
}
2014-10-10 01:32:05 +00:00
Mutex : : Lock _l ( _groups_m ) ;
2014-09-30 23:28:25 +00:00
2014-10-10 01:32:05 +00:00
std : : map < std : : pair < uint64_t , MulticastGroup > , MulticastGroupStatus > : : const_iterator gs ( _groups . find ( std : : pair < uint64_t , MulticastGroup > ( nwid , mg ) ) ) ;
if ( ( gs ! = _groups . end ( ) ) & & ( ! gs - > second . members . empty ( ) ) ) {
2014-10-19 19:56:39 +00:00
totalKnown + = ( unsigned int ) gs - > second . members . size ( ) ;
2014-10-10 01:32:05 +00:00
// Members are returned in random order so that repeated gather queries
// will return different subsets of a large multicast group.
k = 0 ;
2014-10-29 23:24:19 +00:00
while ( ( added < limit ) & & ( k < gs - > second . members . size ( ) ) & & ( ( appendTo . size ( ) + ZT_ADDRESS_LENGTH ) < = ZT_UDP_DEFAULT_PAYLOAD_MTU ) ) {
2014-10-10 01:32:05 +00:00
rptr = ( unsigned int ) RR - > prng - > next32 ( ) ;
2014-10-28 22:33:10 +00:00
2014-09-30 23:28:25 +00:00
restart_member_scan :
2014-10-10 01:32:05 +00:00
a = gs - > second . members [ rptr % ( unsigned int ) gs - > second . members . size ( ) ] . address . toInt ( ) ;
for ( i = 0 ; i < k ; + + i ) {
if ( picked [ i ] = = a ) {
+ + rptr ;
goto restart_member_scan ;
}
}
picked [ k + + ] = a ;
if ( queryingPeer . toInt ( ) ! = a ) { // do not return the peer that is making the request as a result
p = ( unsigned char * ) appendTo . appendField ( ZT_ADDRESS_LENGTH ) ;
* ( p + + ) = ( unsigned char ) ( ( a > > 32 ) & 0xff ) ;
* ( p + + ) = ( unsigned char ) ( ( a > > 24 ) & 0xff ) ;
* ( p + + ) = ( unsigned char ) ( ( a > > 16 ) & 0xff ) ;
* ( p + + ) = ( unsigned char ) ( ( a > > 8 ) & 0xff ) ;
* p = ( unsigned char ) ( a & 0xff ) ;
+ + added ;
2014-09-30 23:28:25 +00:00
}
2014-10-04 01:42:41 +00:00
}
2014-09-30 23:28:25 +00:00
}
2014-10-10 01:32:05 +00:00
appendTo . setAt ( totalAt , ( uint32_t ) totalKnown ) ;
appendTo . setAt ( addedAt , ( uint16_t ) added ) ;
2014-10-10 00:58:31 +00:00
//TRACE("..MC Multicaster::gather() attached %u of %u peers for %.16llx/%s (2)",n,(unsigned int)(gs->second.members.size() - skipped),nwid,mg.toString().c_str());
2014-09-30 23:28:25 +00:00
2014-10-10 01:32:05 +00:00
return added ;
2014-09-30 23:28:25 +00:00
}
2014-10-11 22:49:31 +00:00
std : : vector < Address > Multicaster : : getMembers ( uint64_t nwid , const MulticastGroup & mg , unsigned int limit ) const
2014-10-05 17:34:25 +00:00
{
std : : vector < Address > ls ;
Mutex : : Lock _l ( _groups_m ) ;
std : : map < std : : pair < uint64_t , MulticastGroup > , MulticastGroupStatus > : : const_iterator gs ( _groups . find ( std : : pair < uint64_t , MulticastGroup > ( nwid , mg ) ) ) ;
if ( gs = = _groups . end ( ) )
return ls ;
2014-10-11 22:49:31 +00:00
for ( std : : vector < MulticastGroupMember > : : const_reverse_iterator m ( gs - > second . members . rbegin ( ) ) ; m ! = gs - > second . members . rend ( ) ; + + m ) {
ls . push_back ( m - > address ) ;
if ( ls . size ( ) > = limit )
break ;
2014-10-05 17:34:25 +00:00
}
return ls ;
}
2014-09-30 23:28:25 +00:00
void Multicaster : : send (
const CertificateOfMembership * com ,
unsigned int limit ,
uint64_t now ,
uint64_t nwid ,
2014-10-04 20:15:02 +00:00
const std : : vector < Address > & alwaysSendTo ,
2014-09-30 23:28:25 +00:00
const MulticastGroup & mg ,
const MAC & src ,
unsigned int etherType ,
const void * data ,
unsigned int len )
2014-09-22 20:18:24 +00:00
{
2014-11-21 18:50:27 +00:00
unsigned long idxbuf [ 8194 ] ;
unsigned long * indexes = idxbuf ;
2014-09-23 17:26:30 +00:00
Mutex : : Lock _l ( _groups_m ) ;
2014-09-30 23:28:25 +00:00
MulticastGroupStatus & gs = _groups [ std : : pair < uint64_t , MulticastGroup > ( nwid , mg ) ] ;
2014-09-22 20:18:24 +00:00
2014-11-21 18:50:27 +00:00
if ( ! gs . members . empty ( ) ) {
2014-11-26 21:14:18 +00:00
// Allocate a memory buffer if group is monstrous
if ( gs . members . size ( ) > ( sizeof ( idxbuf ) / sizeof ( unsigned long ) ) )
2014-11-21 18:50:27 +00:00
indexes = new unsigned long [ gs . members . size ( ) ] ;
// Generate a random permutation of member indexes
for ( unsigned long i = 0 ; i < gs . members . size ( ) ; + + i )
indexes [ i ] = i ;
2014-12-03 00:50:53 +00:00
for ( unsigned long i = ( unsigned long ) gs . members . size ( ) - 1 ; i > 0 ; - - i ) {
2014-11-21 18:50:27 +00:00
unsigned long j = RR - > prng - > next32 ( ) % ( i + 1 ) ;
unsigned long tmp = indexes [ j ] ;
indexes [ j ] = indexes [ i ] ;
indexes [ i ] = tmp ;
}
2014-11-21 19:27:53 +00:00
}
2014-10-01 21:05:25 +00:00
2014-11-21 19:27:53 +00:00
if ( gs . members . size ( ) > = limit ) {
2014-11-26 21:14:18 +00:00
// Skip queue if we already have enough members to complete the send operation
2014-11-21 19:27:53 +00:00
OutboundMulticast out ;
out . init (
RR ,
now ,
nwid ,
com ,
limit ,
2014-11-23 22:00:27 +00:00
1 , // we'll still gather a little from peers to keep multicast list fresh
2014-11-21 19:27:53 +00:00
src ,
mg ,
etherType ,
data ,
len ) ;
unsigned int count = 0 ;
for ( std : : vector < Address > : : const_iterator ast ( alwaysSendTo . begin ( ) ) ; ast ! = alwaysSendTo . end ( ) ; + + ast ) {
out . sendOnly ( RR , * ast ) ;
if ( + + count > = limit )
break ;
}
2014-10-04 05:30:10 +00:00
2014-11-21 19:27:53 +00:00
unsigned long idx = 0 ;
2014-11-26 21:14:18 +00:00
while ( ( count < limit ) & & ( idx < gs . members . size ( ) ) ) {
Address ma ( gs . members [ indexes [ idx + + ] ] . address ) ;
if ( std : : find ( alwaysSendTo . begin ( ) , alwaysSendTo . end ( ) , ma ) = = alwaysSendTo . end ( ) ) {
out . sendOnly ( RR , ma ) ;
2014-11-21 19:27:53 +00:00
+ + count ;
}
}
} else {
unsigned int gatherLimit = ( limit - ( unsigned int ) gs . members . size ( ) ) + 1 ;
if ( ( now - gs . lastExplicitGather ) > = ZT_MULTICAST_EXPLICIT_GATHER_DELAY ) {
gs . lastExplicitGather = now ;
SharedPtr < Peer > sn ( RR - > topology - > getBestSupernode ( ) ) ;
if ( sn ) {
2014-11-26 21:14:18 +00:00
TRACE ( " >>MC upstream GATHER up to %u for group %.16llx/%s " , gatherLimit , nwid , mg . toString ( ) . c_str ( ) ) ;
2014-11-21 19:27:53 +00:00
Packet outp ( sn - > address ( ) , RR - > identity . address ( ) , Packet : : VERB_MULTICAST_GATHER ) ;
outp . append ( nwid ) ;
outp . append ( ( uint8_t ) 0 ) ;
mg . mac ( ) . appendTo ( outp ) ;
outp . append ( ( uint32_t ) mg . adi ( ) ) ;
2014-11-26 21:14:18 +00:00
outp . append ( ( uint32_t ) gatherLimit ) ;
2014-11-21 19:27:53 +00:00
outp . armor ( sn - > key ( ) , true ) ;
sn - > send ( RR , outp . data ( ) , outp . size ( ) , now ) ;
}
2014-11-26 21:14:18 +00:00
gatherLimit = 0 ;
2014-11-21 19:27:53 +00:00
}
2014-10-28 22:33:10 +00:00
2014-11-21 19:27:53 +00:00
gs . txQueue . push_back ( OutboundMulticast ( ) ) ;
OutboundMulticast & out = gs . txQueue . back ( ) ;
out . init (
RR ,
now ,
nwid ,
com ,
limit ,
gatherLimit ,
src ,
mg ,
etherType ,
data ,
len ) ;
unsigned int count = 0 ;
for ( std : : vector < Address > : : const_iterator ast ( alwaysSendTo . begin ( ) ) ; ast ! = alwaysSendTo . end ( ) ; + + ast ) {
out . sendAndLog ( RR , * ast ) ;
if ( + + count > = limit )
break ;
}
2014-10-04 05:30:10 +00:00
2014-11-21 19:27:53 +00:00
unsigned long idx = 0 ;
while ( ( count < limit ) & & ( idx < gs . members . size ( ) ) ) {
2014-11-26 21:14:18 +00:00
Address ma ( gs . members [ indexes [ idx + + ] ] . address ) ;
if ( std : : find ( alwaysSendTo . begin ( ) , alwaysSendTo . end ( ) , ma ) = = alwaysSendTo . end ( ) ) {
out . sendAndLog ( RR , ma ) ;
2014-11-21 19:27:53 +00:00
+ + count ;
}
}
2014-09-22 20:18:24 +00:00
}
2014-10-04 20:46:29 +00:00
2014-11-26 21:14:18 +00:00
// Free allocated memory buffer if any
2014-11-21 19:27:53 +00:00
if ( indexes ! = idxbuf )
delete [ ] indexes ;
2014-11-25 20:46:51 +00:00
# ifdef ZT_SUPPORT_LEGACY_MULTICAST
2014-11-26 21:30:00 +00:00
// This sends a P5 multicast up to our supernode, who then
// redistributes it manually down to all <1.0.0 peers for
// legacy support. These peers don't support the new multicast
// frame type, so even if they receive it they will ignore it.
2014-10-04 20:46:29 +00:00
{
SharedPtr < Peer > sn ( RR - > topology - > getBestSupernode ( ) ) ;
if ( sn ) {
uint32_t rn = RR - > prng - > next32 ( ) ;
Packet outp ( sn - > address ( ) , RR - > identity . address ( ) , Packet : : VERB_P5_MULTICAST_FRAME ) ;
outp . append ( ( uint16_t ) 0xffff ) ; // do not forward
outp . append ( ( unsigned char ) 0 , 320 + 1024 ) ; // empty queue and bloom filter
2014-10-14 19:37:35 +00:00
outp . append ( ( unsigned char ) ( ( com ) ? ZT_PROTO_VERB_P5_MULTICAST_FRAME_FLAGS_HAS_MEMBERSHIP_CERTIFICATE : 0 ) ) ;
2014-10-04 20:46:29 +00:00
outp . append ( ( uint64_t ) nwid ) ;
outp . append ( ( uint16_t ) 0 ) ;
outp . append ( ( unsigned char ) 0 ) ;
outp . append ( ( unsigned char ) 0 ) ;
RR - > identity . address ( ) . appendTo ( outp ) ;
outp . append ( ( const void * ) & rn , 3 ) ; // random multicast ID
2014-10-11 22:49:31 +00:00
if ( src )
src . appendTo ( outp ) ;
else MAC ( RR - > identity . address ( ) , nwid ) . appendTo ( outp ) ;
2014-10-04 20:46:29 +00:00
mg . mac ( ) . appendTo ( outp ) ;
outp . append ( ( uint32_t ) mg . adi ( ) ) ;
outp . append ( ( uint16_t ) etherType ) ;
outp . append ( ( uint16_t ) len ) ;
outp . append ( data , len ) ;
2014-10-14 19:37:35 +00:00
unsigned int signedPortionLen = outp . size ( ) - ZT_PROTO_VERB_P5_MULTICAST_FRAME_IDX__START_OF_SIGNED_PORTION ;
2014-10-04 20:46:29 +00:00
2014-10-14 19:37:35 +00:00
C25519 : : Signature sig ( RR - > identity . sign ( outp . field ( ZT_PROTO_VERB_P5_MULTICAST_FRAME_IDX__START_OF_SIGNED_PORTION , signedPortionLen ) , signedPortionLen ) ) ;
2014-10-04 20:46:29 +00:00
outp . append ( ( uint16_t ) sig . size ( ) ) ;
2014-10-19 19:56:39 +00:00
outp . append ( sig . data , ( unsigned int ) sig . size ( ) ) ;
2014-10-04 20:46:29 +00:00
if ( com ) com - > serialize ( outp ) ;
outp . compress ( ) ;
outp . armor ( sn - > key ( ) , true ) ;
sn - > send ( RR , outp . data ( ) , outp . size ( ) , now ) ;
}
}
2014-11-26 21:14:18 +00:00
# endif // ZT_SUPPORT_LEGACY_MULTICAST
2014-09-22 20:18:24 +00:00
}
2014-10-01 21:05:25 +00:00
void Multicaster : : clean ( uint64_t now )
2014-09-22 20:18:24 +00:00
{
2014-09-23 17:26:30 +00:00
Mutex : : Lock _l ( _groups_m ) ;
2014-09-30 23:28:25 +00:00
for ( std : : map < std : : pair < uint64_t , MulticastGroup > , MulticastGroupStatus > : : iterator mm ( _groups . begin ( ) ) ; mm ! = _groups . end ( ) ; ) {
2015-06-18 22:24:17 +00:00
for ( std : : vector < OutboundMulticast > : : iterator tx ( mm - > second . txQueue . begin ( ) ) ; tx ! = mm - > second . txQueue . end ( ) ; ) {
if ( ( tx - > expired ( now ) ) | | ( tx - > atLimit ( ) ) ) {
// erase element (replace by last)
* tx = mm - > second . txQueue . back ( ) ;
mm - > second . txQueue . pop_back ( ) ;
} else + + tx ;
2014-09-25 22:08:29 +00:00
}
2014-11-24 18:05:16 +00:00
unsigned long count = 0 ;
{
std : : vector < MulticastGroupMember > : : iterator reader ( mm - > second . members . begin ( ) ) ;
std : : vector < MulticastGroupMember > : : iterator writer ( reader ) ;
while ( reader ! = mm - > second . members . end ( ) ) {
if ( ( now - reader - > timestamp ) < ZT_MULTICAST_LIKE_EXPIRE ) {
* writer = * reader ;
+ + writer ;
+ + count ;
}
+ + reader ;
2014-09-19 01:28:14 +00:00
}
}
if ( count ) {
2014-11-21 18:50:27 +00:00
mm - > second . members . resize ( count ) ;
2014-09-19 01:28:14 +00:00
+ + mm ;
2014-09-25 22:08:29 +00:00
} else if ( mm - > second . txQueue . empty ( ) ) {
_groups . erase ( mm + + ) ;
2014-10-29 00:25:34 +00:00
} else {
mm - > second . members . clear ( ) ;
+ + mm ;
}
2014-09-25 22:08:29 +00:00
}
}
2014-11-21 18:50:27 +00:00
void Multicaster : : _add ( uint64_t now , uint64_t nwid , const MulticastGroup & mg , MulticastGroupStatus & gs , const Address & member )
2014-09-25 22:08:29 +00:00
{
// assumes _groups_m is locked
2014-09-26 05:08:52 +00:00
2014-10-04 01:42:41 +00:00
// Do not add self -- even if someone else returns it
if ( member = = RR - > identity . address ( ) )
return ;
2014-09-25 22:57:43 +00:00
for ( std : : vector < MulticastGroupMember > : : iterator m ( gs . members . begin ( ) ) ; m ! = gs . members . end ( ) ; + + m ) {
if ( m - > address = = member ) {
m - > timestamp = now ;
return ;
}
2014-09-19 01:28:14 +00:00
}
2014-09-26 05:08:52 +00:00
2014-11-21 18:50:27 +00:00
gs . members . push_back ( MulticastGroupMember ( member , now ) ) ;
2014-10-01 21:05:25 +00:00
2014-10-10 01:32:05 +00:00
//TRACE("..MC %s joined multicast group %.16llx/%s via %s",member.toString().c_str(),nwid,mg.toString().c_str(),((learnedFrom) ? learnedFrom.toString().c_str() : "(direct)"));
2014-10-10 00:58:31 +00:00
2015-06-18 22:24:17 +00:00
for ( std : : vector < OutboundMulticast > : : iterator tx ( gs . txQueue . begin ( ) ) ; tx ! = gs . txQueue . end ( ) ; ) {
2014-11-26 21:39:57 +00:00
if ( tx - > atLimit ( ) ) {
gs . txQueue . erase ( tx + + ) ;
} else {
tx - > sendIfNew ( RR , member ) ;
2015-06-18 22:24:17 +00:00
if ( tx - > atLimit ( ) ) {
// erase element (replace by last)
* tx = gs . txQueue . back ( ) ;
gs . txQueue . pop_back ( ) ;
} else + + tx ;
2014-10-05 17:34:25 +00:00
}
2014-10-01 21:05:25 +00:00
}
2014-09-19 01:28:14 +00:00
}
} // namespace ZeroTier