From: vlad shkapenyuk Date: Fri, 19 Mar 2021 22:37:32 +0000 (-0400) Subject: Add new udafs and RMR support to gsprintconsole_ves X-Git-Url: https://gerrit.o-ran-sc.org/r/gitweb?p=com%2Fgs-lite.git;a=commitdiff_plain;h=HEAD Add new udafs and RMR support to gsprintconsole_ves Signed-off-by: vlad shkapenyuk Change-Id: I2b4be3b939719cdc119813e410752a8d71bfef2c --- diff --git a/cfg/external_fcns.def b/cfg/external_fcns.def index b506319..9a97a5c 100644 --- a/cfg/external_fcns.def +++ b/cfg/external_fcns.def @@ -1,4 +1,4 @@ - /////////////////////////////////////////////////////////// +/////////////////////////////////////////////////////////// // Matching predicates /////////////////////////////////////////////////////////// PRED [LFTA_LEGAL, COST HIGH]str_exists_substr[string, string]; @@ -750,6 +750,12 @@ uint FUN [LFTA_LEGAL, COST EXPENSIVE] string UDAF [HFTA_ONLY] CAT_aggr fstring8 (string, string); +//////////////////////////////////////////////////////////////// +// string aggregation via catenation, only collect strings different than the previous +////////////////////////////////////////////////////// + + string UDAF [HFTA_ONLY,RUNNING] CAT_aggr_diff fstring8 (string); + /////////////////////////////////////////////////////////// // integer array aggregation function // We are going to store 4 values in LFTA in fixed size buffer diff --git a/include/gsoptions.h b/include/gsoptions.h index d3c0267..af70414 100644 --- a/include/gsoptions.h +++ b/include/gsoptions.h @@ -29,5 +29,8 @@ // support for SSL decryption //#define SSL_ENABLED +// support RMR streams +#define RMR_ENABLED + #endif diff --git a/include/hfta/hfta_udaf.h b/include/hfta/hfta_udaf.h index 7bcf02b..8c72cd2 100644 --- a/include/hfta/hfta_udaf.h +++ b/include/hfta/hfta_udaf.h @@ -229,6 +229,16 @@ void CAT_aggr_HFTA_AGGR_UPDATE_(gs_sp_t s, vstring *sep, vstring *str); void CAT_aggr_HFTA_AGGR_OUTPUT_(vstring *res, gs_sp_t s); void CAT_aggr_HFTA_AGGR_DESTROY_(gs_sp_t s); +////////////////////////////////////////////// +// CAT_aggr, aggregate strings by catenation but only when the payload changes +////////////////////////////////////////////// +void CAT_aggr_diff_HFTA_AGGR_INIT_(gs_sp_t s); +void CAT_aggr_diff_HFTA_AGGR_REINIT_(gs_sp_t s); +void CAT_aggr_diff_HFTA_AGGR_UPDATE_(gs_sp_t s, vstring *str); +void CAT_aggr_diff_HFTA_AGGR_OUTPUT_(vstring *res, gs_sp_t s); +void CAT_aggr_diff_HFTA_AGGR_DESTROY_(gs_sp_t s); + + ///////////////////////////////////////////////////////// // time-averaged sum, from aperiodic reports //////////////////////////////////////////////////////// diff --git a/src/lib/gscphftaaux/hfta_udaf.cc b/src/lib/gscphftaaux/hfta_udaf.cc index 60ae750..a5ff5cb 100644 --- a/src/lib/gscphftaaux/hfta_udaf.cc +++ b/src/lib/gscphftaaux/hfta_udaf.cc @@ -667,7 +667,7 @@ void count_diff_hfta_HFTA_AGGR_DESTROY_(gs_sp_t scratch){ } // running_array_aggr aggregate void running_array_aggr_hfta_HFTA_AGGR_INIT_(vstring* scratch) { - scratch->offset = NULL; + scratch->offset = (gs_p_t)NULL; scratch->length = 0; } @@ -743,12 +743,12 @@ void CAT_aggr_HFTA_AGGR_REINIT_(gs_sp_t s){ v->val=""; } void CAT_aggr_HFTA_AGGR_UPDATE_(gs_sp_t s, vstring *sep, vstring *str){ -char buf1[MAXTUPLESZ-20], buf2[MAXTUPLESZ-20]; -int i; -for(i=0;ilength;++i) buf1[i] = *(((char *)sep->offset)+i); -buf1[i]='\0'; -for(i=0;ilength;++i) buf2[i] = *(((char *)str->offset)+i); -buf2[i]='\0'; +//char buf1[MAXTUPLESZ-20], buf2[MAXTUPLESZ-20]; +//int i; +//for(i=0;ilength;++i) buf1[i] = *(((char *)sep->offset)+i); +//buf1[i]='\0'; +//for(i=0;ilength;++i) buf2[i] = *(((char *)str->offset)+i); +//buf2[i]='\0'; CAT_aggr_scratch_ptr *p = (CAT_aggr_scratch_ptr *)s; CAT_aggr_scratch *v = p->ptr; if(v->val.size()>0) @@ -925,6 +925,77 @@ gs_int64_t extr_running_sum_max(vstring *v){ return vs->max; } + +// --------------------------------------------- +// aggr_diff : from a sequence of strings, collect +// the ones which are different than the previous. +// this includes the prior time period. +// the idea is to see the sequence of handovers + +struct CAT_aggr_diff_scratch{ + std::string val; + std::string prev_s; +// gs_int64_t prev_ts; // for now, just catenate strings +}; + +struct CAT_aggr_diff_scratch_ptr{ + CAT_aggr_diff_scratch *ptr; +}; + + + +void CAT_aggr_diff_HFTA_AGGR_INIT_(gs_sp_t s){ + CAT_aggr_diff_scratch_ptr *p = (CAT_aggr_diff_scratch_ptr *)s; + CAT_aggr_diff_scratch *v = new CAT_aggr_diff_scratch(); + v->prev_s = ""; + v->val = ""; + + p->ptr = v; +} +void CAT_aggr_diff_HFTA_AGGR_REINIT_(gs_sp_t s){ + CAT_aggr_diff_scratch_ptr *p = (CAT_aggr_diff_scratch_ptr *)s; + CAT_aggr_diff_scratch *v = p->ptr; + v->val=v->prev_s; +} +void CAT_aggr_diff_HFTA_AGGR_UPDATE_(gs_sp_t s, vstring *str){ + char str_buf[MAXTUPLESZ-20]; + int i; + for(i=0;ilength;++i) str_buf[i] = *(((char *)str->offset)+i); + str_buf[i]='\0'; + + CAT_aggr_diff_scratch_ptr *p = (CAT_aggr_diff_scratch_ptr *)s; + CAT_aggr_diff_scratch *v = p->ptr; + if(str_buf != v->prev_s){ + if(v->val.size()>0) + v->val += ':'; + v->val += str_buf; + v->prev_s = str_buf; + } +} + +void CAT_aggr_diff_HFTA_AGGR_OUTPUT_(vstring *res, gs_sp_t s){ + CAT_aggr_diff_scratch_ptr *p = (CAT_aggr_diff_scratch_ptr *)s; + CAT_aggr_diff_scratch *v = p->ptr; +//printf("output val=%s\n",v->val.c_str()); + res->offset = (gs_p_t)malloc(v->val.size()); + res->length = v->val.size(); + if(res->length>MAXTUPLESZ-20) + res->length=MAXTUPLESZ-20; +// v->val.copy((char *)(res->offset), 0, res->length); + const char *dat = v->val.c_str(); + memcpy((char *)(res->offset), dat, res->length); +// for(int i=0;ilength;++i) +// *(((char *)res->offset)+i) = dat[i]; + res->reserved = INTERNAL; +} +void CAT_aggr_diff_HFTA_AGGR_DESTROY_(gs_sp_t s){ + CAT_aggr_diff_scratch_ptr *p = (CAT_aggr_diff_scratch_ptr *)s; + CAT_aggr_diff_scratch *v = p->ptr; + delete v; +} + + + // --------------------------------------------- // Approximate count distinct. // Rely on the minhashing approach. diff --git a/src/tools/gsprintconsole_ves.c b/src/tools/gsprintconsole_ves.c index 5bb6205..f82be92 100644 --- a/src/tools/gsprintconsole_ves.c +++ b/src/tools/gsprintconsole_ves.c @@ -34,6 +34,14 @@ #include #include +#include +#include + +#ifdef RMR_ENABLED +#include +#include +#include +#endif #include "gsconfig.h" #include "gstypes.h" @@ -169,6 +177,12 @@ int main(int argc, char* argv[]) { endpoint dummyep; gs_uint32_t tip1,tip2,tip3,tip4; gs_sp_t instance_name; + gs_sp_t rmr_port = NULL; + +#ifdef RMR_ENABLED + // RMR-related parameters + gs_int32_t rmr_mtype = MC_REPORT; +#endif gs_sp_t curl_address = NULL; endpoint curl_endpoint; @@ -178,17 +192,29 @@ int main(int argc, char* argv[]) { gs_uint32_t ves_version=7; - gs_uint32_t tlimit = 0; // time limit in seconds time_t start_time, curr_time; gs_uint64_t post_success_cnt = 0ULL; gs_uint64_t post_failure_cnt = 0ULL; + +#ifdef RMR_ENABLED + void* mrc; //msg router context + struct epoll_event events[1]; // list of events to give to epoll + struct epoll_event epe; // event definition for event to listen to + gs_int32_t ep_fd = -1; // epoll's file des (given to epoll_wait) + gs_int32_t rcv_fd; // file des that NNG tickles -- give this to epoll to listen on + gs_int32_t nready; // number of events ready for receive + rmr_mbuf_t* rmr_sbuf; // send buffer + rmr_mbuf_t* rmr_rbuf; // received buffer + + gs_uint64_t rmr_post_success_cnt = 0ULL; + gs_uint64_t rmr_post_failure_cnt = 0ULL; +#endif gsopenlog(argv[0]); - - while ((ch = getopt(argc, argv, "l:p:r:vXDC:U:A:V:")) != -1) { + while ((ch = getopt(argc, argv, "l:p:r:vXDC:U:R:A:V:")) != -1) { switch (ch) { case 'r': bufsz=atoi(optarg); @@ -220,6 +246,9 @@ int main(int argc, char* argv[]) { curl_endpoint.ip=htonl(tip1<<24|tip2<<16|tip3<<8|tip4); curl_endpoint.port=htons(curl_endpoint.port); break; + case 'R': + rmr_port=strdup(optarg); + break; case 'U': curl_url = strdup(optarg); break; @@ -228,7 +257,7 @@ int main(int argc, char* argv[]) { break; default: usage: - fprintf(stderr, "usage: %s [-r ] [-p ] [-l ] [-v] [-X] [-D] [-C :] [-U ] [-A ] [-V ] : query param1 param2...\n", + fprintf(stderr, "usage: %s [-r ] [-p ] [-l ] [-v] [-X] [-D] [-C :] [-U ] [-A ] [-V ] [-R ] : query param1 param2...\n", *argv); exit(1); } @@ -274,7 +303,45 @@ int main(int argc, char* argv[]) { argv +=2; if (argc < 1) goto usage; - + + if (rmr_port) { +#ifdef RMR_ENABLED + /* initialize RMR library */ + if( (mrc = rmr_init( rmr_port, 1400, RMRFL_NONE )) == NULL ) { + fprintf(stderr, "%s::error:unable to initialise RMR\n", me); + exit( 1 ); + } + + rcv_fd = rmr_get_rcvfd( mrc ); // set up epoll things, start by getting the FD from MRr + if( rcv_fd < 0 ) { + fprintf(stderr, "%s::error:unable to set up polling fd\n", me); + exit( 1 ); + } + if( (ep_fd = epoll_create1( 0 )) < 0 ) { + fprintf(stderr, "%s::error:unable to create epoll fd: %d\n", me, errno); + exit( 1 ); + } + epe.events = EPOLLIN; + epe.data.fd = rcv_fd; + + if( epoll_ctl( ep_fd, EPOLL_CTL_ADD, rcv_fd, &epe ) != 0 ) { + fprintf(stderr, "%s::error:epoll_ctl status not 0 : %s\n", me, strerror(errno)); + exit( 1 ); + } + + rmr_sbuf = rmr_alloc_msg( mrc, MAXLINE ); // alloc first send buffer; subsequent buffers allcoated on send + rmr_rbuf = NULL; // don't need to alloc receive buffer + + while( ! rmr_ready( mrc ) ) { // must have a route table before we can send; wait til RMr say it has one + sleep( 10 ); + } + fprintf( stderr, "%s: RMR is ready\n", argv[0]); +#else + fprintf(stderr,"Runtime libraries built without RMR support. Rebuild with RMR_ENABLED defined in gsoptions.h\n"); + exit(0); +#endif + } + /* initialize host library and the sgroup */ if (verbose>=2) fprintf(stderr,"Initializing gscp\n"); @@ -626,8 +693,30 @@ int main(int argc, char* argv[]) { "}}}\n", measurement_interval ); } + +#ifdef RMR_ENABLED + if (rmr_port) { + rmr_sbuf->mtype = rmr_mtype; // fill in the message bits + rmr_sbuf->len = strlen(linebuf) + 1; // our receiver likely wants a nice acsii-z string + memcpy(rmr_sbuf->payload, linebuf, rmr_sbuf->len); + rmr_sbuf->state = 0; + rmr_sbuf = rmr_send_msg( mrc, rmr_sbuf); // send it (send returns an empty payload on success, or the original payload on fail/retry) + while( rmr_sbuf->state == RMR_ERR_RETRY ) { // soft failure (device busy?) retry + rmr_sbuf = rmr_send_msg( mrc, rmr_sbuf); // retry send until it's good (simple test; real programmes should do better) + } + if(rmr_sbuf->state != RMR_OK) { + gslog(LOG_WARNING, "rmr_send_msg() failure, strerror(errno) is %s", strerror(errno)); + rmr_post_failure_cnt++; + } else + rmr_post_success_cnt++; + if (((rmr_post_success_cnt+rmr_post_failure_cnt) % STAT_FREQUENCY) == 0) + gslog(LOG_WARNING, "%s: successful RMR posts - %llu, failed RMR posts - %llu", argv[0], rmr_post_success_cnt, rmr_post_failure_cnt); + } +#endif + if(curl_address==NULL){ - emit_line(); + if (!rmr_port) // if neither VES collector nor RMR is specified print to standard output + emit_line(); }else{ http_post_request_hdr(curl_endpoint, curl_url, linebuf, &http_code, curl_auth); if(http_code != 200 && http_code != 202){