@ -18,6 +18,7 @@
# include "config_parser.h"
# include "secure_channel.h"
# include "topo_node.h"
# include "topo_node_sqlite.h"
# include "topo_group.h"
# include "route_ping.h"
# include "route_connectivity.h"
@ -70,20 +71,22 @@ struct cm_exchange_pending {
uint8_t probes_done ;
} ;
struct CONN_MGR * conn_mgr_init ( struct UTUN_INSTANCE * instance ) {
if ( ! instance ) { DEBUG_ERROR ( DEBUG_CATEGORY_GENERAL , " NULL instance " ) ; return NULL ; }
struct CONN_MGR * conn_mgr_init ( struct TOPO_GROUP * group ) {
if ( ! group | | ! group - > instance ) { DEBUG_ERROR ( DEBUG_CATEGORY_GENERAL , " NULL group " ) ; return NULL ; }
struct CONN_MGR * mgr = u_calloc ( 1 , sizeof ( struct CONN_MGR ) ) ;
if ( ! mgr ) { DEBUG_ERROR ( DEBUG_CATEGORY_GENERAL , " alloc failed " ) ; return NULL ; }
mgr - > instance = instance ;
mgr - > instance = group - > instance ;
mgr - > group = group ;
mgr - > entry_capacity = 8 ;
mgr - > entries = u_calloc ( mgr - > entry_capacity , sizeof ( struct CONN_MGR_ENTRY ) ) ;
if ( ! mgr - > entries ) { DEBUG_ERROR ( DEBUG_CATEGORY_GENERAL , " entries alloc failed " ) ; u_free ( mgr ) ; return NULL ; }
etcp_router_bind ( instance , ETCP_RT_ID_CONN_MGR , conn_mgr_router_recv_handler ) ;
mgr - > bg_ping_timer = uasync_set_timeout ( instance - > ua , CONN_MGR_BG_PING_INTERVAL_TB , mgr , cm_bg_ping_timer_cb , " conn_mgr_bg_ping " ) ;
mgr - > candidate_ping_timer = uasync_set_timeout ( instance - > ua , CONN_MGR_CANDIDATE_PING_TB , mgr , cm_candidate_ping_timer_cb , " conn_mgr_cand_ping " ) ;
mgr - > direct_timeout_ms = CONN_MGR_CONNECT_DIRECT_TIMEOUT_MS ;
etcp_router_bind ( mgr - > instance , ETCP_RT_ID_CONN_MGR , conn_mgr_router_recv_handler ) ;
mgr - > bg_ping_timer = uasync_set_timeout ( mgr - > instance - > ua , CONN_MGR_BG_PING_INTERVAL_TB , mgr , cm_bg_ping_timer_cb , " conn_mgr_bg_ping " ) ;
mgr - > candidate_ping_timer = uasync_set_timeout ( mgr - > instance - > ua , CONN_MGR_CANDIDATE_PING_TB , mgr , cm_candidate_ping_timer_cb , " conn_mgr_cand_ping " ) ;
mgr - > bg_ping_cycle_start_tb = get_time_tb ( ) ;
mgr - > initialized = 1 ;
DEBUG_INFO ( DEBUG_CATEGORY_GENERAL , " conn_mgr: initialized, bg_ping started " ) ;
DEBUG_INFO ( DEBUG_CATEGORY_GENERAL , " conn_mgr: initialized for group %016llx , bg_ping started " , ( unsigned long long ) group - > group_id ) ;
return mgr ;
}
@ -157,7 +160,7 @@ static void cm_deliver_result(struct CONN_MGR_ENTRY* entry, int result) {
}
static void cm_update_nodeinfo ( struct CONN_MGR_ENTRY * entry ) {
struct TOPO_GROUP * group = topo_groups_get_default ( entry - > mgr - > instance - > topo_ groups ) ;
struct TOPO_GROUP * group = entry - > mgr - > group ;
if ( ! group ) return ;
struct TOPO_NODEQ * nq = topo_node_find_by_id ( group , entry - > node_id ) ;
if ( ! nq ) return ;
@ -167,18 +170,51 @@ static void cm_update_nodeinfo(struct CONN_MGR_ENTRY* entry) {
}
static void cm_clear_nodeinfo ( struct CONN_MGR * mgr , uint64_t node_id ) {
struct TOPO_NODEQ * nq = topo_node_find_by_id ( topo_groups_get_default ( mgr - > instance - > topo_ groups ) , node_id ) ;
struct TOPO_NODEQ * nq = topo_node_find_by_id ( mgr - > group , node_id ) ;
if ( ! nq ) return ;
nq - > conn_mgr_type = CONN_TYPE_NONE ;
nq - > conn_mgr_intermediariy_count = 0 ;
}
static void cm_cleanup_db_node ( struct CONN_MGR_ENTRY * entry ) {
if ( ! entry - > db_loaded ) return ;
struct TOPO_GROUP * group = entry - > mgr - > group ;
struct TOPO_NODEQ * nq = topo_node_find_by_id ( group , entry - > node_id ) ;
if ( ! nq ) return ;
if ( nq - > paths & & queue_entry_count ( nq - > paths ) > 0 ) {
DEBUG_INFO ( DEBUG_CATEGORY_GENERAL , " conn_mgr: db_node 0x%016llx got paths, keep in group " , ( unsigned long long ) entry - > node_id ) ;
return ;
}
DEBUG_INFO ( DEBUG_CATEGORY_GENERAL , " conn_mgr: removing db_node 0x%016llx from group (no paths) " , ( unsigned long long ) entry - > node_id ) ;
queue_remove_data ( group - > nodes , & nq - > ll ) ;
topo_node_registry_release ( group - > instance - > topo_groups , nq - > node ) ;
topo_node_free_lists ( group , nq ) ;
queue_entry_free ( & nq - > ll ) ;
}
int conn_mgr_connect_node ( struct CONN_MGR * mgr , uint64_t node_id , uint32_t idle_timeout_ms ,
conn_mgr_connect_callback_t cb , void * cb_arg ) {
if ( ! mgr ) return CONN_MGR_ERR_INTERNAL ;
struct TOPO_GROUP * group = topo_groups_get_default ( mgr - > instance - > topo_groups ) ;
struct TOPO_GROUP * group = mgr - > group ;
if ( ! group ) { DEBUG_ERROR ( DEBUG_CATEGORY_GENERAL , " BGP not initialized " ) ; return CONN_MGR_ERR_INTERNAL ; }
struct TOPO_NODEQ * target = topo_node_find_by_id ( group , node_id ) ;
int loaded_from_db = 0 ;
if ( ! target & & mgr - > instance - > topo_sqlite_db & & mgr - > instance - > topo_groups ) {
struct TOPO_NODE * ni = topo_node_sqlite_node_load ( mgr - > instance - > topo_sqlite_db , mgr - > instance - > topo_groups , node_id ) ;
if ( ni ) {
ni = topo_node_registry_acquire ( mgr - > instance - > topo_groups , ni ) ;
struct ll_entry * qe = queue_entry_new ( sizeof ( struct TOPO_NODEQ ) ) ;
if ( qe ) {
target = ( struct TOPO_NODEQ * ) qe ;
memset ( ( uint8_t * ) target + sizeof ( struct ll_entry ) , 0 , sizeof ( * target ) - sizeof ( struct ll_entry ) ) ;
target - > node = ni ; target - > hash_node_id = node_id ;
target - > conn_mgr_type = CONN_TYPE_NONE ;
queue_data_put_with_index ( group - > nodes , & target - > ll ) ;
loaded_from_db = 1 ;
DEBUG_INFO ( DEBUG_CATEGORY_GENERAL , " conn_mgr: node 0x%016llx loaded from DB " , ( unsigned long long ) node_id ) ;
} else { DEBUG_ERROR ( DEBUG_CATEGORY_GENERAL , " conn_mgr: failed to alloc TOPO_NODEQ for DB node 0x%016llx " , ( unsigned long long ) node_id ) ; }
}
}
if ( ! target ) { DEBUG_ERROR ( DEBUG_CATEGORY_GENERAL , " node 0x%016llx not found " , ( unsigned long long ) node_id ) ; return CONN_MGR_ERR_NOT_FOUND ; }
struct CONN_MGR_ENTRY * entry = cm_find_entry ( mgr , node_id ) ;
if ( entry & & entry - > state = = CONN_MGR_STATE_CONNECTED ) {
@ -221,6 +257,7 @@ int conn_mgr_connect_node(struct CONN_MGR* mgr, uint64_t node_id, uint32_t idle_
entry - > main . phase = 0 ;
entry - > main . timer = NULL ;
entry - > main . request_id = 0 ;
entry - > db_loaded = ( uint8_t ) loaded_from_db ;
int has_our_local = cm_has_local_addr ( group - > local_node ) ;
int has_target_local = cm_has_local_addr ( target ) ;
if ( has_our_local & & has_target_local ) { entry - > local_scan_state = CM_TRY_PENDING ; cm_start_local_scan ( entry ) ; }
@ -238,7 +275,7 @@ int conn_mgr_disconnect_node(struct CONN_MGR* mgr, uint64_t node_id) {
pkt . subcmd = CONN_MGR_SUBCMD_DISCONNECT ;
pkt . node_id = node_id ;
struct ll_entry * qe = queue_entry_new ( 0 ) ;
if ( qe ) { qe - > dgram = u_malloc ( sizeof ( pkt ) ) ; memcpy ( qe - > dgram , & pkt , sizeof ( pkt ) ) ; qe - > len = sizeof ( pkt ) ; etcp_route_send ( mgr - > instance , TOPO_GROUP_UTUN , node_id , qe , 1 ) ; }
if ( qe ) { qe - > dgram = u_malloc ( sizeof ( pkt ) ) ; memcpy ( qe - > dgram , & pkt , sizeof ( pkt ) ) ; qe - > len = sizeof ( pkt ) ; etcp_route_send ( mgr - > instance , mgr - > group - > group_id , node_id , qe , 1 ) ; }
cm_clear_nodeinfo ( mgr , node_id ) ;
cm_entry_destroy ( entry ) ;
entry - > state = CONN_MGR_STATE_DISCONNECTED ;
@ -258,6 +295,10 @@ int conn_mgr_set_idle_timeout(struct CONN_MGR* mgr, uint64_t node_id, uint32_t t
return CONN_MGR_OK ;
}
void conn_mgr_set_direct_timeout_ms ( struct CONN_MGR * mgr , uint32_t timeout_ms ) {
if ( mgr ) mgr - > direct_timeout_ms = timeout_ms ;
}
int conn_mgr_get_status ( struct CONN_MGR * mgr , uint64_t node_id , uint8_t * out_state , uint8_t * out_conn_type ) {
if ( ! mgr | | ! out_state | | ! out_conn_type ) return CONN_MGR_ERR_INTERNAL ;
struct CONN_MGR_ENTRY * entry = cm_find_entry ( mgr , node_id ) ;
@ -272,7 +313,7 @@ int conn_mgr_send(struct CONN_MGR* mgr, uint64_t node_id, struct ll_entry* entry
struct CONN_MGR_ENTRY * e = cm_find_entry ( mgr , node_id ) ;
if ( ! e | | e - > state ! = CONN_MGR_STATE_CONNECTED ) { queue_entry_free ( entry ) ; return - 1 ; }
e - > last_traffic_tb = get_time_tb ( ) ;
struct ETCP_ROUTER_CONN * rconn = etcp_router_conn_get ( mgr - > instance , TOPO_GROUP_UTUN , node_id , ETCP_RT_ID_CONN_MGR ) ;
struct ETCP_ROUTER_CONN * rconn = etcp_router_conn_get ( mgr - > instance , mgr - > group - > group_id , node_id , ETCP_RT_ID_CONN_MGR ) ;
if ( ! rconn ) { queue_entry_free ( entry ) ; return - 1 ; }
int ret = etcp_router_conn_send ( rconn , entry - > dgram , entry - > len ) ;
queue_entry_free ( entry ) ;
@ -401,7 +442,7 @@ struct cm_ping_ctx {
} ;
static void cm_start_local_scan ( struct CONN_MGR_ENTRY * entry ) {
struct TOPO_GROUP * group = topo_groups_get_default ( entry - > mgr - > instance - > topo_ groups ) ;
struct TOPO_GROUP * group = entry - > mgr - > group ;
struct TOPO_NODEQ * target = topo_node_find_by_id ( group , entry - > node_id ) ;
if ( ! target | | ! target - > node ) { entry - > local_scan_state = CM_TRY_FAILED ; return ; }
for ( const struct TOPO_ADDR4 * a = target - > node - > v4_addrs ; a ; a = a - > next ) {
@ -429,9 +470,39 @@ static void cm_start_local_scan(struct CONN_MGR_ENTRY* entry) {
static void cm_start_phase_direct ( struct CONN_MGR_ENTRY * entry ) {
if ( entry - > main_connect_state = = CM_TRY_OK ) return ;
if ( entry - > local_scan_state = = CM_TRY_OK ) return ;
struct TOPO_GROUP * group = topo_groups_get_default ( entry - > mgr - > instance - > topo_ groups ) ;
struct TOPO_GROUP * group = entry - > mgr - > group ;
struct TOPO_NODEQ * target = topo_node_find_by_id ( group , entry - > node_id ) ;
if ( ! target | | ! target - > node ) { cm_deliver_result ( entry , CONN_MGR_ERR_NOT_FOUND ) ; return ; }
if ( entry - > db_loaded ) {
for ( const struct TOPO_ADDR4 * a = target - > node - > v4_addrs ; a ; a = a - > next ) {
if ( a - > protocol ! = TOPO_PROTO_UDP ) continue ;
struct ETCP_SOCKET * s = entry - > mgr - > instance - > etcp_sockets ;
while ( s ) {
if ( s - > local_addr . ss_family = = AF_INET ) {
struct sockaddr_in sin ; memset ( & sin , 0 , sizeof ( sin ) ) ; sin . sin_family = AF_INET ;
memcpy ( & sin . sin_addr . s_addr , a - > addr , 4 ) ; sin . sin_port = htons ( a - > port ) ;
struct sockaddr_storage sa ; memset ( & sa , 0 , sizeof ( sa ) ) ; memcpy ( & sa , & sin , sizeof ( sin ) ) ;
struct ETCP_CONN * conn = etcp_connection_create ( entry - > mgr - > instance , NULL ) ;
if ( conn ) {
etcp_conn_add_init_cbk ( conn , cm_direct_init_cb , entry ) ;
sc_init_ctx ( & conn - > crypto_ctx , & entry - > mgr - > instance - > my_keys ) ;
sc_set_peer_public_key ( & conn - > crypto_ctx , target - > node - > public_key , 0 ) ;
if ( etcp_link_new ( conn , s , & sa , 0 ) ) {
entry - > main . timer = uasync_set_timeout ( entry - > mgr - > instance - > ua , ( int ) ( entry - > mgr - > direct_timeout_ms * 10 ) , entry , cm_direct_timeout_cb , " conn_mgr_db_direct " ) ;
return ;
}
}
}
s = s - > next ;
}
}
DEBUG_INFO ( DEBUG_CATEGORY_GENERAL , " conn_mgr: db_node 0x%016llx direct failed (no compatible addr/socket) " , ( unsigned long long ) entry - > node_id ) ;
cm_cleanup_db_node ( entry ) ;
cm_deliver_result ( entry , CONN_MGR_ERR_UNREACHABLE ) ;
return ;
}
uint8_t priority_order [ ] = { TOPO_ADDR_NAT , TOPO_ADDR_INTERFACE } ;
for ( int pri = 0 ; pri < 2 ; pri + + ) {
for ( const struct TOPO_ADDR4 * a = target - > node - > v4_addrs ; a ; a = a - > next ) {
@ -450,7 +521,7 @@ static void cm_start_phase_direct(struct CONN_MGR_ENTRY* entry) {
sc_init_ctx ( & conn - > crypto_ctx , & entry - > mgr - > instance - > my_keys ) ;
sc_set_peer_public_key ( & conn - > crypto_ctx , target - > node - > public_key , 0 ) ;
if ( etcp_link_new ( conn , s , & sa , 0 ) ) {
entry - > main . timer = uasync_set_timeout ( entry - > mgr - > instance - > ua , CONN_MGR_CONNECT_DIRECT_TIMEOUT_MS * 10 , entry , cm_direct_timeout_cb , " conn_mgr_direct " ) ;
entry - > main . timer = uasync_set_timeout ( entry - > mgr - > instance - > ua , ( int ) ( entry - > mgr - > direct_timeout_ms * 10 ) , entry , cm_direct_timeout_cb , " conn_mgr_direct " ) ;
return ;
}
}
@ -469,7 +540,13 @@ static void cm_start_phase_direct(struct CONN_MGR_ENTRY* entry) {
static void cm_direct_timeout_cb ( void * arg ) {
struct CONN_MGR_ENTRY * entry = ( struct CONN_MGR_ENTRY * ) arg ;
entry - > main . timer = NULL ;
struct TOPO_GROUP * group = topo_groups_get_default ( entry - > mgr - > instance - > topo_groups ) ;
if ( entry - > db_loaded ) {
DEBUG_INFO ( DEBUG_CATEGORY_GENERAL , " conn_mgr: db_node 0x%016llx direct timeout " , ( unsigned long long ) entry - > node_id ) ;
cm_cleanup_db_node ( entry ) ;
cm_deliver_result ( entry , CONN_MGR_ERR_UNREACHABLE ) ;
return ;
}
struct TOPO_GROUP * group = entry - > mgr - > group ;
struct TOPO_NODEQ * target = topo_node_find_by_id ( group , entry - > node_id ) ;
int has_our = cm_has_direct_ip ( group - > local_node ) ;
int has_target = target ? cm_has_direct_ip ( target ) : 0 ;
@ -518,7 +595,7 @@ static void cm_ping_cb(int success, uint16_t rtt, void* arg, uint64_t nonce,
ctx - > attempt + + ;
if ( ctx - > attempt < CONN_MGR_LOCAL_SCAN_ATTEMPTS & & ctx - > phase = = 0 ) {
etcp_send_ping_to_socket ( entry - > mgr - > instance , ctx - > sock ,
topo_node_find_by_id ( topo_groups_get_default ( entry - > mgr - > instance - > topo_ groups ) , entry - > node_id ) - > node - > public_key ,
topo_node_find_by_id ( entry - > mgr - > group , entry - > node_id ) - > node - > public_key ,
& ctx - > addr , CONN_MGR_LOCAL_SCAN_TIMEOUT_MS , cm_ping_cb , ctx , NULL , 0 , 0 ) ;
return ;
}
@ -529,7 +606,7 @@ static void cm_ping_cb(int success, uint16_t rtt, void* arg, uint64_t nonce,
}
static void cm_start_phase_reverse ( struct CONN_MGR_ENTRY * entry ) {
struct TOPO_GROUP * group = topo_groups_get_default ( entry - > mgr - > instance - > topo_ groups ) ;
struct TOPO_GROUP * group = entry - > mgr - > group ;
struct TOPO_NODEQ * local = group - > local_node ;
if ( ! local | | ! local - > node ) { cm_deliver_result ( entry , CONN_MGR_ERR_INTERNAL ) ; return ; }
uint32_t req_id = + + entry - > mgr - > next_request_id ;
@ -567,7 +644,7 @@ static void cm_start_phase_reverse(struct CONN_MGR_ENTRY* entry) {
if ( ! qe ) { u_free ( pkt ) ; cm_deliver_result ( entry , CONN_MGR_ERR_INTERNAL ) ; return ; }
qe - > dgram = pkt ;
qe - > len = ( uint16_t ) pkt_size ;
etcp_route_send ( entry - > mgr - > instance , TOPO_GROUP_UTUN , entry - > node_id , qe , 1 ) ;
etcp_route_send ( entry - > mgr - > instance , entry - > mgr - > group - > group_id , entry - > node_id , qe , 1 ) ;
struct cm_reverse_pending * rp = u_calloc ( 1 , sizeof ( struct cm_reverse_pending ) ) ;
if ( rp ) { rp - > request_id = req_id ; rp - > entry = entry ; rp - > next = entry - > mgr - > reverse_pending ;
@ -607,6 +684,7 @@ static int cm_start_phase_indirect(struct CONN_MGR_ENTRY* entry) {
if ( mgr - > best_candidate_count = = 0 ) {
DEBUG_ERROR ( DEBUG_CATEGORY_GENERAL , " conn_mgr: phase 3 (indirect) no candidates for 0x%016llx (direct+reverse failed) " ,
( unsigned long long ) entry - > node_id ) ;
cm_deliver_result ( entry , CONN_MGR_ERR_UNREACHABLE ) ;
return CONN_MGR_ERR_UNREACHABLE ;
}
uint32_t req_id = + + mgr - > next_request_id ;
@ -625,7 +703,7 @@ static int cm_start_phase_indirect(struct CONN_MGR_ENTRY* entry) {
qe - > dgram = u_malloc ( sizeof ( req ) ) ;
memcpy ( qe - > dgram , & req , sizeof ( req ) ) ;
qe - > len = ( uint16_t ) sizeof ( req ) ;
etcp_route_send ( entry - > mgr - > instance , TOPO_GROUP_UTUN , entry - > node_id , qe , 1 ) ;
etcp_route_send ( entry - > mgr - > instance , entry - > mgr - > group - > group_id , entry - > node_id , qe , 1 ) ;
entry - > main . timer = uasync_set_timeout ( mgr - > instance - > ua , CONN_MGR_INTERM_EXCHANGE_TIMEOUT_MS * 10 ,
entry , cm_exchange_timeout_cb , " conn_mgr_interm_exch " ) ;
{
@ -663,7 +741,7 @@ static void cm_compute_intermediaries(struct CONN_MGR_ENTRY* entry, struct CONN_
}
for ( uint8_t i = 0 ; i < resp - > my_count & & count < 8 ; i + + ) {
uint64_t id = resp - > my_candidates [ i ] . node_id ;
struct TOPO_NODEQ * nq = topo_node_find_by_id ( topo_groups_get_default ( entry - > mgr - > instance - > topo_ groups ) , id ) ;
struct TOPO_NODEQ * nq = topo_node_find_by_id ( entry - > mgr - > group , id ) ;
uint16_t our_rtt = nq ? cm_get_node_min_rtt ( nq ) : 0 ;
if ( our_rtt = = 0xFFFF ) our_rtt = 0 ;
all [ count ] . node_id = id ;
@ -689,11 +767,11 @@ static void cm_compute_intermediaries(struct CONN_MGR_ENTRY* entry, struct CONN_
for ( uint8_t i = 0 ; i < sel ; i + + ) {
pkt . selected [ i ] . node_id = all [ i ] . node_id ;
pkt . selected [ i ] . rtt = 0 ;
struct TOPO_NODEQ * nq = topo_node_find_by_id ( topo_groups_get_default ( entry - > mgr - > instance - > topo_ groups ) , all [ i ] . node_id ) ;
struct TOPO_NODEQ * nq = topo_node_find_by_id ( entry - > mgr - > group , all [ i ] . node_id ) ;
if ( nq ) pkt . selected [ i ] . rtt = cm_get_node_min_rtt ( nq ) ;
}
struct ll_entry * qe = queue_entry_new ( 0 ) ;
if ( qe ) { qe - > dgram = u_malloc ( sizeof ( pkt ) ) ; memcpy ( qe - > dgram , & pkt , sizeof ( pkt ) ) ; qe - > len = sizeof ( pkt ) ; etcp_route_send ( entry - > mgr - > instance , TOPO_GROUP_UTUN , entry - > node_id , qe , 1 ) ; }
if ( qe ) { qe - > dgram = u_malloc ( sizeof ( pkt ) ) ; memcpy ( qe - > dgram , & pkt , sizeof ( pkt ) ) ; qe - > len = sizeof ( pkt ) ; etcp_route_send ( entry - > mgr - > instance , entry - > mgr - > group - > group_id , entry - > node_id , qe , 1 ) ; }
entry - > conn_type = CONN_TYPE_INDIRECT ;
entry - > state = CONN_MGR_STATE_CONNECTED ;
@ -710,7 +788,7 @@ static void cm_exchange_probe_retry_cb(void* arg) {
uint8_t need_probe = 0 ;
uint64_t now_tb = get_time_tb ( ) ;
for ( uint8_t i = 0 ; i < ep - > cached_resp . my_count & & i < 4 ; i + + ) {
struct TOPO_NODEQ * nq = topo_node_find_by_id ( topo_groups_get_default ( mgr - > instance - > topo_ groups ) , ep - > cached_resp . my_candidates [ i ] . node_id ) ;
struct TOPO_NODEQ * nq = topo_node_find_by_id ( mgr - > group , ep - > cached_resp . my_candidates [ i ] . node_id ) ;
if ( ! nq | | ! cm_is_rtt_fresh ( nq , now_tb ) ) { need_probe + + ; break ; }
}
if ( need_probe = = 0 ) {
@ -740,7 +818,7 @@ static void cm_handle_interm_exchange_resp(struct CONN_MGR* mgr, const uint8_t*
uint8_t need_probe = 0 ;
uint64_t now_tb = get_time_tb ( ) ;
for ( uint8_t i = 0 ; i < ep - > cached_resp . my_count & & i < 4 ; i + + ) {
struct TOPO_NODEQ * nq = topo_node_find_by_id ( topo_groups_get_default ( mgr - > instance - > topo_ groups ) , ep - > cached_resp . my_candidates [ i ] . node_id ) ;
struct TOPO_NODEQ * nq = topo_node_find_by_id ( mgr - > group , ep - > cached_resp . my_candidates [ i ] . node_id ) ;
if ( ! nq | | ! cm_is_rtt_fresh ( nq , now_tb ) ) { need_probe + + ; route_connectivity_probe_node ( mgr - > instance , nq ) ; }
}
if ( need_probe = = 0 ) { ep - > probes_done = 1 ; cm_compute_intermediaries ( ep - > entry , & ep - > cached_resp ) ;
@ -829,7 +907,7 @@ static void cm_handle_direct_req(struct ETCP_CONN* conn, const uint8_t* data, si
memset ( & sa , 0 , sizeof ( sa ) ) ; memcpy ( & sa , & sin , sizeof ( sin ) ) ;
struct ETCP_CONN * new_conn = etcp_connection_create ( conn - > instance , NULL ) ;
if ( new_conn ) {
struct TOPO_NODEQ * nq = topo_node_find_by_id ( topo_groups_get_default ( conn - > instance - > topo_ groups ) , src_node_id ) ;
struct TOPO_NODEQ * nq = topo_node_find_by_id ( mgr - > group , src_node_id ) ;
if ( nq ) { sc_init_ctx ( & new_conn - > crypto_ctx , & conn - > instance - > my_keys ) ;
sc_set_peer_public_key ( & new_conn - > crypto_ctx , nq - > node - > public_key , 0 ) ; }
struct cm_reverse_pending * rp = u_calloc ( 1 , sizeof ( struct cm_reverse_pending ) ) ;
@ -851,7 +929,7 @@ static void cm_handle_interm_exchange_req(struct ETCP_CONN* conn,
uint64_t now_tb = get_time_tb ( ) ;
for ( uint8_t i = 0 ; i < req - > candidate_count & & i < 4 ; i + + ) {
uint64_t cand_id = req - > candidates [ i ] . node_id ;
struct TOPO_NODEQ * nq = topo_node_find_by_id ( topo_groups_get_default ( mgr - > instance - > topo_ groups ) , cand_id ) ;
struct TOPO_NODEQ * nq = topo_node_find_by_id ( mgr - > group , cand_id ) ;
if ( nq & & cm_is_rtt_fresh ( nq , now_tb ) ) continue ;
route_connectivity_probe_node ( mgr - > instance , nq ) ;
}
@ -866,7 +944,7 @@ static void cm_handle_interm_exchange_req(struct ETCP_CONN* conn,
resp . your_count = req - > candidate_count > 4 ? 4 : req - > candidate_count ;
for ( uint8_t i = 0 ; i < resp . your_count ; i + + ) {
resp . your_candidates [ i ] . node_id = req - > candidates [ i ] . node_id ;
struct TOPO_NODEQ * nq2 = topo_node_find_by_id ( topo_groups_get_default ( mgr - > instance - > topo_ groups ) , req - > candidates [ i ] . node_id ) ;
struct TOPO_NODEQ * nq2 = topo_node_find_by_id ( mgr - > group , req - > candidates [ i ] . node_id ) ;
uint16_t rtt = nq2 ? cm_get_node_min_rtt ( nq2 ) : 0 ;
resp . your_candidates [ i ] . rtt = rtt ;
}
@ -877,7 +955,7 @@ static void cm_handle_interm_exchange_req(struct ETCP_CONN* conn,
qe - > dgram = u_malloc ( resp_size ) ;
memcpy ( qe - > dgram , & resp , resp_size ) ;
qe - > len = ( uint16_t ) resp_size ;
etcp_route_send ( mgr - > instance , TOPO_GROUP_UTUN , conn - > peer_node_id , qe , 1 ) ;
etcp_route_send ( mgr - > instance , mgr - > group - > group_id , conn - > peer_node_id , qe , 1 ) ;
}
DEBUG_INFO ( DEBUG_CATEGORY_GENERAL , " conn_mgr: sent EXCHANGE_RESP to 0x%016llx my=%u your=%u " ,
( unsigned long long ) conn - > peer_node_id , resp . my_count , resp . your_count ) ;
@ -898,7 +976,7 @@ static void cm_bg_ping_noop_cb(int success, uint16_t rtt, void* arg, uint64_t no
static void cm_bg_ping_timer_cb ( void * arg ) {
struct CONN_MGR * mgr = ( struct CONN_MGR * ) arg ;
struct TOPO_GROUP * group = topo_groups_get_default ( mgr - > instance - > topo_ groups ) ;
struct TOPO_GROUP * group = mgr - > group ;
if ( ! group | | ! group - > nodes ) {
mgr - > bg_ping_timer = uasync_set_timeout ( mgr - > instance - > ua , CONN_MGR_BG_PING_INTERVAL_TB , mgr , cm_bg_ping_timer_cb , " conn_mgr_bg_ping " ) ;
return ;
@ -958,7 +1036,7 @@ static void cm_candidate_ping_timer_cb(void* arg) {
uint64_t now_tb = get_time_tb ( ) ;
for ( uint8_t i = 0 ; i < mgr - > best_candidate_count ; i + + ) {
uint64_t cand_id = mgr - > best_candidates [ i ] . node_id ;
struct TOPO_NODEQ * nq = topo_node_find_by_id ( topo_groups_get_default ( mgr - > instance - > topo_ groups ) , cand_id ) ;
struct TOPO_NODEQ * nq = topo_node_find_by_id ( mgr - > group , cand_id ) ;
if ( ! nq ) {
memmove ( & mgr - > best_candidates [ i ] , & mgr - > best_candidates [ i + 1 ] , ( mgr - > best_candidate_count - i - 1 ) * sizeof ( mgr - > best_candidates [ 0 ] ) ) ;
mgr - > best_candidate_count - - ; i - - ; continue ;
@ -969,7 +1047,7 @@ static void cm_candidate_ping_timer_cb(void* arg) {
memmove ( & mgr - > best_candidates [ i ] , & mgr - > best_candidates [ i + 1 ] , ( mgr - > best_candidate_count - i - 1 ) * sizeof ( mgr - > best_candidates [ 0 ] ) ) ;
mgr - > best_candidate_count - - ; i - - ; continue ;
}
struct ETCP_CONN * dconn = topo_group_find_conn_for_node ( topo_groups_get_default ( mgr - > instance - > topo_ groups ) , cand_id ) ;
struct ETCP_CONN * dconn = topo_group_find_conn_for_node ( mgr - > group , cand_id ) ;
int has_active = 0 ;
if ( dconn & & dconn - > links ) {
struct ETCP_LINK * l = dconn - > links ;