X-Git-Url: https://gerrit.o-ran-sc.org/r/gitweb?a=blobdiff_plain;f=src%2Frmr%2Fcommon%2Fsrc%2Frt_generic_static.c;h=0de69c83a202a1132a2b06d98c8fe92395e4ae83;hb=a68562a02028434a87149d5996b291e83d33be51;hp=f93963c4a65584158dfd0ba5042b3da5ef6404da;hpb=84423e66ad5d4ed313548b3705ec925a934b0f2e;p=ric-plt%2Flib%2Frmr.git diff --git a/src/rmr/common/src/rt_generic_static.c b/src/rmr/common/src/rt_generic_static.c index f93963c..0de69c8 100644 --- a/src/rmr/common/src/rt_generic_static.c +++ b/src/rmr/common/src/rt_generic_static.c @@ -1,4 +1,4 @@ -// :vi sw=4 ts=4 noet: + // :vi sw=4 ts=4 noet2 /* ================================================================================== Copyright (c) 2019-2020 Nokia @@ -46,6 +46,8 @@ #include #include #include +#include +#include #include // needed for route manager messages @@ -323,13 +325,64 @@ static void send_rt_ack( uta_ctx_t* ctx, rmr_mbuf_t* smsg, char* table_id, int s } } +// ---- alarm generation -------------------------------------------------------------------------- + +/* + Given the user's context (not the thread private context) look to see if the application isn't + working fast enough and we're dropping messages. If the drop counter has changed since the last + peeked, and we have not raised an alarm, then we will alarm. If the counter hasn't changed, then we + set a timer and if the counter still hasn't changed when it expires we will clear the alarm. + + The private context is what we use to send so as not to interfere with the user flow. +*/ +static void alarm_if_drops( uta_ctx_t* uctx, uta_ctx_t* pctx ) { + static int alarm_raised = 0; + static int ok2clear = 0; // time that we can clear + static int lastd = 0; // the last counter value so we can compute delta + static int prob_id = 0; // problem ID we assume alarm manager handles dups between processes + + rmr_vlog( RMR_VL_DEBUG, "checking for drops... raised=%d 0k2clear=%d lastd=%d probid=%d\n", alarm_raised, ok2clear, lastd, prob_id ); + if( ! alarm_raised ) { + if( uctx->dcount - lastd == 0 ) { // not actively dropping, ok to do nothing + return; + } + + alarm_raised = 1; + uta_alarm( pctx, ALARM_DROPS | ALARM_RAISE, prob_id, "application running slow; RMR is dropping messages" ); + rmr_vlog( RMR_VL_INFO, "drop alarm raised" ); + } else { + if( uctx->dcount - lastd != 0 ) { // still dropping or dropping again; we've alarmed so nothing to do + lastd = uctx->dcount; + ok2clear = 0; // reset the timer + return; + } + + if( ok2clear == 0 ) { // first round where not dropping + ok2clear = time( NULL ) + 60; // we'll clear the alarm in 60s + } else { + if( time( NULL ) > ok2clear ) { // things still stable after expiry + rmr_vlog( RMR_VL_INFO, "drop alarm cleared\n" ); + alarm_raised = 0; + uta_alarm( pctx, ALARM_DROPS | ALARM_CLEAR, prob_id, "RMR message dropping has stopped" ); + prob_id++; + } + } + } +} + // ---- utility ----------------------------------------------------------------------------------- + +int isspace_with_fence(int c) { + _mm_lfence(); + return isspace( c ); +} + /* Little diddy to trim whitespace and trailing comments. Like shell, trailing comments must be at the start of a word (i.e. must be immediatly preceeded by whitespace). */ static char* clip( char* buf ) { - char* tok; + char* tok=NULL; while( *buf && isspace( *buf ) ) { // skip leading whitespace buf++; @@ -345,7 +398,7 @@ static char* clip( char* buf ) { } } - for( tok = buf + (strlen( buf ) - 1); tok > buf && isspace( *tok ); tok-- ); // trim trailing spaces too + for( tok = buf + (strlen( buf ) - 1); tok > buf && isspace_with_fence( *tok ); tok-- ); // trim trailing spaces too *(tok+1) = 0; return buf; @@ -530,7 +583,6 @@ static void build_entry( uta_ctx_t* ctx, char* ts_field, uint32_t subid, char* r uint64_t key = 0; // the symtab key will be mtype or sub_id+mtype char* tokens[128]; char* gtokens[64]; - int i; int ngtoks; // number of tokens in the group list int grp; // index into group list int cgidx; // contiguous group index (prevents the addition of a contiguous group without ep) @@ -555,6 +607,8 @@ static void build_entry( uta_ctx_t* ctx, char* ts_field, uint32_t subid, char* r rte->mtype = atoi( ts_field ); // capture mtype for debugging for( grp = 0, cgidx = 0; grp < ngtoks; grp++ ) { + int i; // avoid sonar grumbling by defining this here + if( (ntoks = uta_rmip_tokenise( gtokens[grp], ctx->ip_list, tokens, 64, ',' )) > 0 ) { // remove any references to our ip addrs for( i = 0; i < ntoks; i++ ) { if( strcmp( tokens[i], ctx->my_name ) != 0 ) { // don't add if it is us -- cannot send to ourself @@ -787,6 +841,71 @@ static void meid_parser( uta_ctx_t* ctx, uta_ctx_t* pctx, rmr_mbuf_t* mbuf, char } } +/* + This will close the current table snarf file (in *.inc) and open a new one. + The curent one is renamed. The final file name is determined by the setting of + RMR_SNARF_RT, and if not set then the variable RMR_SEED_RT is used and given + an additional extension of .snarf. If neither seed or snarf environment vars are + set then this does nothing. + + If this is called before the tmp snarf file is opened, then this just opens the file. +*/ +static void cycle_snarfed_rt( uta_ctx_t* ctx ) { + static int ok2warn = 0; // some warnings squelched on first call + + char* seed_fname; // the filename from env + char tfname[512]; // temp fname + char wfname[512]; // working buffer for filename + char* snarf_fname = NULL; // prevent overlay of the static table if snarf_rt not given + + if( ctx == NULL ) { + return; + } + + if( (snarf_fname = getenv( ENV_STASH_RT )) == NULL ) { // specific place to stash the rt not given + if( (seed_fname = getenv( ENV_SEED_RT )) != NULL ) { // no seed, we leave in the default file + memset( wfname, 0, sizeof( wfname ) ); + snprintf( wfname, sizeof( wfname ) - 1, "%s.stash", seed_fname ); + snarf_fname = wfname; + } + } + + if( snarf_fname == NULL ) { + rmr_vlog( RMR_VL_DEBUG, "cycle_snarf: no file to save in" ); + return; + } + + memset( tfname, 0, sizeof( tfname ) ); + snprintf( tfname, sizeof( tfname ) -1, "%s.inc", snarf_fname ); // must ensure tmp file is moveable + + if( ctx->snarf_rt_fd >= 0 ) { + char* msg= "### captured from route manager\n"; + write( ctx->snarf_rt_fd, msg, strlen( msg ) ); + if( close( ctx->snarf_rt_fd ) < 0 ) { + rmr_vlog( RMR_VL_WARN, "rmr_rtc: unable to close working rt snarf file: %s\n", strerror( errno ) ); + return; + } + + if( unlink( snarf_fname ) < 0 && ok2warn ) { // first time through this can fail and we ignore it + rmr_vlog( RMR_VL_WARN, "rmr_rtc: unable to unlink old static table: %s: %s\n", snarf_fname, strerror( errno ) ); + } + + if( rename( tfname, snarf_fname ) ) { + rmr_vlog( RMR_VL_WARN, "rmr_rtc: unable to move new route table to seed aname : %s -> %s: %s\n", tfname, snarf_fname, strerror( errno ) ); + } else { + rmr_vlog( RMR_VL_INFO, "latest route table info saved in: %s\n", snarf_fname ); + } + } + ok2warn = 1; + + ctx->snarf_rt_fd = open( tfname, O_WRONLY | O_CREAT | O_TRUNC, 0660 ); + if( ctx->snarf_rt_fd < 0 ) { + rmr_vlog( RMR_VL_WARN, "rmr_rtc: unable to open trt file: %s: %s\n", tfname, strerror( errno ) ); + } else { + if( DEBUG ) rmr_vlog( RMR_VL_DEBUG, "rmr_rtc: rt snarf file opened: %s\n", tfname ); + } +} + /* Parse a single record recevied from the route table generator, or read from a static route table file. Start records cause a new table to @@ -841,17 +960,22 @@ static void parse_rt_rec( uta_ctx_t* ctx, uta_ctx_t* pctx, char* buf, int vleve int grp; // group number rtable_ent_t const* rte; // route table entry added char* tokens[128]; - char* tok; // pointer into a token or string + char* tok=NULL; // pointer into a token or string char wbuf[1024]; if( ! buf ) { return; } + if( ctx && ctx->snarf_rt_fd >= 0 ) { // if snarfing table as it arrives, write this puppy + write( ctx->snarf_rt_fd, buf, strlen( buf ) ); + write( ctx->snarf_rt_fd, "\n", 1 ); + } + while( *buf && isspace( *buf ) ) { // skip leading whitespace buf++; } - for( tok = buf + (strlen( buf ) - 1); tok > buf && isspace( *tok ); tok-- ); // trim trailing spaces too + for( tok = buf + (strlen( buf ) - 1); tok > buf && isspace_with_fence( *tok ); tok-- ); // trim trailing spaces too *(tok+1) = 0; memset( tokens, 0, sizeof( tokens ) ); @@ -880,6 +1004,10 @@ static void parse_rt_rec( uta_ctx_t* ctx, uta_ctx_t* pctx, char* buf, int vleve case 'n': // newrt|{start|end} tokens[1] = clip( tokens[1] ); if( strcmp( tokens[1], "end" ) == 0 ) { // wrap up the table we were building + if( ctx && ctx->snarf_rt_fd >= 0 ) { + cycle_snarfed_rt( ctx ); // make it available and open a new one + } + if( ntoks >2 ) { if( ctx->new_rtable->updates != atoi( tokens[2] ) ) { // count they added didn't match what we received rmr_vlog( RMR_VL_ERR, "rmr_rtc: RT update had wrong number of records: received %d expected %s\n", @@ -901,6 +1029,7 @@ static void parse_rt_rec( uta_ctx_t* ctx, uta_ctx_t* pctx, char* buf, int vleve send_rt_ack( pctx, mbuf, ctx->table_id, RMR_OK, NULL ); ctx->rtable_ready = 1; // route based sends can now happen + ctx->flags |= CFL_FULLRT; // indicate we have seen a complete route table } else { if( DEBUG > 1 ) rmr_vlog_force( RMR_VL_DEBUG, "end of route table noticed, but one was not started!\n" ); ctx->new_rtable = NULL; @@ -962,11 +1091,19 @@ static void parse_rt_rec( uta_ctx_t* ctx, uta_ctx_t* pctx, char* buf, int vleve break; case 'u': // update current table, not a total replacement + if( ! (ctx->flags & CFL_FULLRT) ) { // we cannot update until we have a full table from route generator + rmr_vlog( RMR_VL_WARN, "route table update ignored: full table not previously recevied" ); + break; + } + tokens[1] = clip( tokens[1] ); if( strcmp( tokens[1], "end" ) == 0 ) { // wrap up the table we were building if( ctx->new_rtable == NULL ) { // update table not in progress break; } + if( ctx->snarf_rt_fd >= 0 ) { + cycle_snarfed_rt( ctx ); // make it available and open a new one + } if( ntoks >2 ) { if( ctx->new_rtable->updates != atoi( tokens[2] ) ) { // count they added didn't match what we received @@ -1038,8 +1175,13 @@ static void read_static_rt( uta_ctx_t* ctx, int vlevel ) { char* eor; // end of the record int rcount = 0; // record count for debug - if( (fname = getenv( ENV_SEED_RT )) == NULL ) { - return; + if( (fname = ctx->seed_rt_fname) == NULL ) { + if( (fname = getenv( ENV_SEED_RT )) == NULL ) { + return; + } + + ctx->seed_rt_fname = strdup( fname ); + fname = ctx->seed_rt_fname; } if( (fbuf = ensure_nlterm( uta_fib( fname ) ) ) == NULL ) { // read file into a single buffer (nil terminated string) @@ -1368,6 +1510,7 @@ static route_table_t* uta_rt_clone( uta_ctx_t* ctx, route_table_t* srt, route_ta */ static route_table_t* prep_new_rt( uta_ctx_t* ctx, int all ) { int counter = 0; + int ref_count; route_table_t* rt; if( ctx == NULL ) { @@ -1376,13 +1519,22 @@ static route_table_t* prep_new_rt( uta_ctx_t* ctx, int all ) { if( (rt = ctx->old_rtable) != NULL ) { ctx->old_rtable = NULL; - while( rt->ref_count > 0 ) { // wait for all who are using to stop + + pthread_mutex_lock( ctx->rtgate ); + ref_count = rt->ref_count; + pthread_mutex_unlock( ctx->rtgate ); + + while( ref_count > 0 ) { // wait for all who are using to stop if( counter++ > 1000 ) { rmr_vlog( RMR_VL_WARN, "rt_prep_newrt: internal mishap, ref count on table seems wedged" ); break; } usleep( 1000 ); // small sleep to yield the processer if that is needed + + pthread_mutex_lock( ctx->rtgate ); + ref_count = rt->ref_count; + pthread_mutex_unlock( ctx->rtgate ); } if( rt->hash != NULL ) { @@ -1395,6 +1547,7 @@ static route_table_t* prep_new_rt( uta_ctx_t* ctx, int all ) { rt = NULL; } + pthread_mutex_destroy(ctx->rtgate); rt = uta_rt_clone( ctx, ctx->rtable, rt, all ); // also sets the ephash pointer if( rt != NULL ) { // very small chance for nil, but not zero, so test rt->ref_count = 0; // take no chances; ensure it's 0! @@ -1545,4 +1698,5 @@ static inline void release_rt( uta_ctx_t* ctx, route_table_t* rt ) { } pthread_mutex_unlock( ctx->rtgate ); } + #endif