- ///////////////////////////////////////////////////////////
+///////////////////////////////////////////////////////////
// Matching predicates
///////////////////////////////////////////////////////////
PRED [LFTA_LEGAL, COST HIGH]str_exists_substr[string, string];
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
// support for SSL decryption
//#define SSL_ENABLED
+// support RMR streams
+#define RMR_ENABLED
+
#endif
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
////////////////////////////////////////////////////////
// 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;
}
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;i<sep->length;++i) buf1[i] = *(((char *)sep->offset)+i);
-buf1[i]='\0';
-for(i=0;i<str->length;++i) buf2[i] = *(((char *)str->offset)+i);
-buf2[i]='\0';
+//char buf1[MAXTUPLESZ-20], buf2[MAXTUPLESZ-20];
+//int i;
+//for(i=0;i<sep->length;++i) buf1[i] = *(((char *)sep->offset)+i);
+//buf1[i]='\0';
+//for(i=0;i<str->length;++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)
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;i<str->length;++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;i<res->length;++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.
#include <sys/socket.h>
#include <netinet/in.h>
+#include <errno.h>
+#include <string.h>
+
+#ifdef RMR_ENABLED
+#include <rmr/rmr.h>
+#include <rmr/RIC_message_types.h>
+#include <sys/epoll.h>
+#endif
#include "gsconfig.h"
#include "gstypes.h"
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;
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);
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;
break;
default:
usage:
- fprintf(stderr, "usage: %s [-r <bufsz>] [-p <port>] [-l <time_limit>] [-v] [-X] [-D] [-C <curl_dest>:<curl_port>] [-U <curl_url>] [-A <authentication_string>] [-V <ves_version>] <gshub-hostname>:<gshub-port> <gsinstance_name> query param1 param2...\n",
+ fprintf(stderr, "usage: %s [-r <bufsz>] [-p <port>] [-l <time_limit>] [-v] [-X] [-D] [-C <curl_dest>:<curl_port>] [-U <curl_url>] [-A <authentication_string>] [-V <ves_version>] [-R <rmr_port>] <gshub-hostname>:<gshub-port> <gsinstance_name> query param1 param2...\n",
*argv);
exit(1);
}
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");
"}}}\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){
Code Review / com / gs-lite.git / commitdiff