X-Git-Url: https://gerrit.o-ran-sc.org/r/gitweb?a=blobdiff_plain;f=src%2Frmr%2Fnng%2Fsrc%2Frtable_nng_static.c;h=7cd3e201bc6d7b75fd46e4f7259372a41628559f;hb=d533475b82551879def60d864b83739e99f6d5c4;hp=875cfe62eaf5f46978228fba2cb44a87c12f2687;hpb=750cbee6b224532fc10971c04df556ce4bceb832;p=ric-plt%2Flib%2Frmr.git diff --git a/src/rmr/nng/src/rtable_nng_static.c b/src/rmr/nng/src/rtable_nng_static.c index 875cfe6..7cd3e20 100644 --- a/src/rmr/nng/src/rtable_nng_static.c +++ b/src/rmr/nng/src/rtable_nng_static.c @@ -1,4 +1,4 @@ -// : vi ts=4 sw=4 noet : +// vim: ts=4 sw=4 noet : /* ================================================================================== Copyright (c) 2019 Nokia @@ -59,20 +59,31 @@ get things into a bad state if we allow a collision here. The lock grab only happens on the intial session setup. */ -//static int uta_link2( char* target, nng_socket* nn_sock, nng_dialer* dialer, pthread_mutex* gate ) { static int uta_link2( endpoint_t* ep ) { + static int flags = -1; + char* target; nng_socket* nn_sock; nng_dialer* dialer; char conn_info[NNG_MAXADDRLEN]; // string to give to nano to make the connection char* addr; int state = FALSE; + char* tok; if( ep == NULL ) { return FALSE; } - target = ep->addr; + if( flags < 0 ) { + tok = getenv( "RMR_ASYNC_CONN" ); + if( tok == NULL || *tok == '1' ) { + flags = NNG_FLAG_NONBLOCK; // start dialer asynch + } else { + flags = NO_FLAGS; + } + } + + target = ep->name; // always give name to transport so chaning dest IP does not break reconnect nn_sock = &ep->nn_sock; dialer = &ep->dialer; @@ -110,7 +121,7 @@ static int uta_link2( endpoint_t* ep ) { nng_dialer_setopt_ms( *dialer, NNG_OPT_RECONNMAXT, 2000 ); // cap backoff on retries to reasonable amount (2s) nng_dialer_setopt_ms( *dialer, NNG_OPT_RECONNMINT, 100 ); // start retry 100m after last failure with 2s cap - if( (state = nng_dialer_start( *dialer, NO_FLAGS )) != 0 ) { // can fail immediatly (unlike nanomsg) + if( (state = nng_dialer_start( *dialer, flags )) != 0 ) { // can fail immediatly (unlike nanomsg) pthread_mutex_unlock( &ep->gate ); fprintf( stderr, "[WRN] rmr: unable to create link to target: %s: %s\n", target, nng_strerror( state ) ); nng_close( *nn_sock ); @@ -206,7 +217,7 @@ extern endpoint_t* uta_add_ep( route_table_t* rt, rtable_ent_t* rte, char* ep_n the user pointer passed in and sets the return value to true (1). If the endpoint cannot be found false (0) is returned. */ -static int uta_epsock_byname( route_table_t* rt, char* ep_name, nng_socket* nn_sock ) { +static int uta_epsock_byname( route_table_t* rt, char* ep_name, nng_socket* nn_sock, endpoint_t** uepp ) { endpoint_t* ep; int state = FALSE; @@ -215,6 +226,9 @@ static int uta_epsock_byname( route_table_t* rt, char* ep_name, nng_socket* nn_s } ep = rmr_sym_get( rt->hash, ep_name, 1 ); + if( uepp != NULL ) { // caller needs endpoint too, give it back + *uepp = ep; + } if( ep == NULL ) { if( DEBUG ) fprintf( stderr, "[DBUG] get ep by name for %s not in hash!\n", ep_name ); if( ! ep_name || (ep = rt_ensure_ep( rt, ep_name)) == NULL ) { // create one if not in rt (support rts without entry in our table) @@ -244,7 +258,7 @@ static int uta_epsock_byname( route_table_t* rt, char* ep_name, nng_socket* nn_s /* Make a round robin selection within a round robin group for a route table entry. Returns the nanomsg socket if there is a rte for the message - type, and group is defined. Socket is returned via pointer in the parm + key, and group is defined. Socket is returned via pointer in the parm list (nn_sock). The group is the group number to select from. @@ -267,8 +281,7 @@ static int uta_epsock_byname( route_table_t* rt, char* ep_name, nng_socket* nn_s Both of these, in the grand scheme of things, is minor compared to the overhead of grabbing a lock on each call. */ -static int uta_epsock_rr( route_table_t *rt, uint64_t key, int group, int* more, nng_socket* nn_sock ) { - rtable_ent_t* rte; // matching rt entry +static int uta_epsock_rr( rtable_ent_t *rte, int group, int* more, nng_socket* nn_sock, endpoint_t** uepp ) { endpoint_t* ep; // seected end point int state = FALSE; // processing state int dummy; @@ -286,25 +299,19 @@ static int uta_epsock_rr( route_table_t *rt, uint64_t key, int group, int* more, return FALSE; } - if( rt == NULL ) { - *more = 0; - return FALSE; - } - - if( (rte = rmr_sym_pull( rt->hash, key )) == NULL ) { + if( rte == NULL ) { *more = 0; - //if( DEBUG ) fprintf( stderr, ">>>> rte not found for type = %lu\n", key ); return FALSE; } if( group < 0 || group >= rte->nrrgroups ) { - //if( DEBUG ) fprintf( stderr, ">>>> group out of range: key=%lu group=%d max=%d\n", key, group, rte->nrrgroups ); + //if( DEBUG ) fprintf( stderr, ">>>> group out of range: group=%d max=%d\n", group, rte->nrrgroups ); *more = 0; return FALSE; } if( (rrg = rte->rrgroups[group]) == NULL ) { - //if( DEBUG ) fprintf( stderr, ">>>> rrg not found for type key=%lu\n", key ); + //if( DEBUG ) fprintf( stderr, ">>>> rrg not found for group \n", group ); *more = 0; // groups are inserted contig, so nothing should be after a nil pointer return FALSE; } @@ -316,8 +323,7 @@ static int uta_epsock_rr( route_table_t *rt, uint64_t key, int group, int* more, //if( DEBUG ) fprintf( stderr, ">>>> nothing allocated for the rrg\n" ); return FALSE; - case 1: // exactly one, no rr to deal with and more is not possible even if fanout > 1 - //*nn_sock = rrg->epts[0]->nn_sock; + case 1: // exactly one, no rr to deal with ep = rrg->epts[0]; //if( DEBUG ) fprintf( stderr, ">>>> _rr returning socket with one choice in group \n" ); state = TRUE; @@ -332,6 +338,9 @@ static int uta_epsock_rr( route_table_t *rt, uint64_t key, int group, int* more, break; } + if( uepp != NULL ) { // caller needs refernce to endpoint too + *uepp = ep; + } if( state ) { // end point selected, open if not, get socket either way if( ! ep->open ) { // not connected if( ep->addr == NULL ) { // name didn't resolve before, try again @@ -353,4 +362,61 @@ static int uta_epsock_rr( route_table_t *rt, uint64_t key, int group, int* more, return state; } +/* + Finds the rtable entry which matches the key. Returns a nil pointer if + no entry is found. If try_alternate is set, then we will attempt + to find the entry with a key based only on the message type. +*/ +static inline rtable_ent_t* uta_get_rte( route_table_t *rt, int sid, int mtype, int try_alt ) { + uint64_t key; // key is sub id and mtype banged together + rtable_ent_t* rte; // the entry we found + + if( rt == NULL || rt->hash == NULL ) { + return NULL; + } + + key = build_rt_key( sid, mtype ); // first try with a 'full' key + if( ((rte = rmr_sym_pull( rt->hash, key )) != NULL) || ! try_alt ) { // found or not allowed to try the alternate, return what we have + return rte; + } + + if( sid != UNSET_SUBID ) { // not found, and allowed to try alternate; and the sub_id was set + key = build_rt_key( UNSET_SUBID, mtype ); // rebuild key + rte = rmr_sym_pull( rt->hash, key ); // see what we get with this + } + + return rte; +} + +/* + Return a string of count information. E.g.: + : + + Caller must free the string allocated if a buffer was not provided. + + Pointer returned is to a freshly allocated string, or the user buffer + for convenience. + + If the endpoint passed is a nil pointer, then we return a nil -- caller + must check! +*/ +static inline char* get_ep_counts( endpoint_t* ep, char* ubuf, int ubuf_len ) { + char* rs; // result string + + if( ep == NULL ) { + return NULL; + } + + if( ubuf != NULL ) { + rs = ubuf; + } else { + ubuf_len = 256; + rs = malloc( sizeof( char ) * ubuf_len ); + } + + snprintf( rs, ubuf_len, "%s %lld %lld %lld", ep->name, ep->scounts[EPSC_GOOD], ep->scounts[EPSC_FAIL], ep->scounts[EPSC_TRANS] ); + + return rs; +} + #endif