--- /dev/null
+/* ------------------------------------------------
+ Copyright 2014 AT&T Intellectual Property
+ Licensed under the Apache License, Version 2.0 (the "License");
+ you may not use this file except in compliance with the License.
+ You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing, software
+ distributed under the License is distributed on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ See the License for the specific language governing permissions and
+ limitations under the License.
+ ------------------------------------------- */
+#include <gscpipc.h>
+#include <ipcencoding.h>
+#include <stdlib.h>
+#include <stdio.h>
+#include <lapp.h>
+#include <lappregistries.h>
+#include <string.h>
+#include <unistd.h>
+#include <signal.h>
+#include <sys/mman.h>
+#include "rdtsc.h"
+// If POLLING is defined applications poll every 100 msec instead of blocking
+#define POLLING
+
+struct processtate curprocess = {0,0,0,255,0};
+struct FTAID clearinghouseftaid = {0,0,0,0};
+
+/*
+ * sends the message passed in buf and waits for a result
+ * if a message returned is not a result it is put in the
+ * request queue. The resultsbuf has to be large enough
+ * for the largest result
+ */
+gs_retval_t ipc_call_and_wait(FTAID f, gs_sp_t msg, gs_sp_t result)
+{
+ struct hostcall * h = (struct hostcall *) msg;
+ gs_int8_t buf[MAXMSGSZ];
+ FTAID from;
+ gs_int32_t length;
+ gs_int32_t lowop;
+#ifdef PRINTMSG
+ fprintf(stderr, "HOST sending to %u.%u.%u.%u:%u of "
+ "type %u with length %u\n",
+ (f.ip>>24)&0xff,
+ (f.ip>>16)&0xff,
+ (f.ip>>8)&0xff,
+ (f.ip)&0xff,
+ f.port,h->callid,h->size);
+#endif
+ if (gscpipc_send(f,FTACALLBACK,msg,h->size,1)<0) {
+ gslog(LOG_EMERG,"ERROR:Could not send on message queue\n");
+ return -1;
+ }
+ h=(struct hostcall *) buf;
+ while (gscpipc_read(&from,&lowop,buf,&length,1)>0) {
+#ifdef PRINTMSG
+ fprintf(stderr, "HOST response from %u.%u.%u.%u:%u"
+ " of type %u with length %u\n",
+ (from.ip>>24)&0xff,
+ (from.ip>>16)&0xff,
+ (from.ip>>8)&0xff,
+ (from.ip)&0xff,
+ from.port,
+ h->callid,h->size);
+#endif
+ if ((lowop == FTACALLBACK) && (h->callid < RESULT_OPCODE_IGNORE)) {
+ h=(struct hostcall *) buf;
+ if (h->callid > RESULT_OPCODE_BASE) {
+ memcpy(result,buf,length);
+ return 0;
+ }
+ if (sidequeue_append(from,buf,length)<0) {
+ gslog(LOG_EMERG,"ERROR:: Could not add to sidequeue\n");
+ return -1;
+ }
+ }
+ }
+ gslog(LOG_EMERG, "ERROR::gscipc_read failed in ipc_call_and_wait\n");
+ return -1;
+}
+
+
+gs_retval_t hostlib_init(gs_int32_t type, gs_int32_t buffersize, gs_int32_t deviceid, gs_int32_t mapcnt, gs_sp_t map[])
+{
+ FILE * f;
+
+ if (curprocess.active != 0 ) {
+ return -1;
+ }
+
+ switch (type) {
+ case CLEARINGHOUSE:
+ if (gscpipc_init(1) < 0) {
+ gslog(LOG_EMERG,"ERROR:Could not initalize comm layer for "
+ "clearinghouse process\n");
+ return -1;
+ }
+ break;
+ case LFTA:
+#ifdef __linux__
+ mlockall(MCL_CURRENT|MCL_FUTURE);
+#endif
+ case APP:
+ case HFTA:
+ if (gscpipc_init(0) < 0) {
+ gslog(LOG_EMERG,"ERROR:Could not initalize comm layer for "
+ "non clearinghouse process\n");
+ return -1;
+ }
+ break;
+ default:
+ gslog(LOG_EMERG,"ERROR:Unknown process type\n");
+ return -1;
+ }
+
+ // if the buffersize is zero then allocating shared memory
+ // will fail. So only use it for the clearinghouse and LFTAs
+ if ((buffersize<(4*MAXTUPLESZ)) && (buffersize!=0)) {
+ gslog(LOG_EMERG,
+ "ERROR:buffersize in hostlib_init has to "
+ "be at least %u Bytes long\n",
+ 4*MAXTUPLESZ);
+ return -1;
+ }
+
+ curprocess.type=type;
+ curprocess.buffersize=buffersize;
+ curprocess.active = 1;
+ curprocess.deviceid=deviceid;
+ curprocess.mapcnt=mapcnt;
+ curprocess.map=map;
+ return 0;
+}
+
+void hostlib_free()
+{
+ if (curprocess.active != 1 ) {
+ return;
+ }
+ curprocess.active = 0;
+ gscpipc_free();
+}
+
+
+gs_retval_t fta_find(FTAname name, gs_uint32_t reuse, FTAID *ftaid,
+ gs_sp_t schema, gs_int32_t schemasz)
+{
+ gs_int8_t rb[MAXRES];
+ struct fta_find_arg a;
+ struct ftafind_result * sr = (struct ftafind_result *)rb;
+
+ a.h.callid = FTA_LOOKUP;
+ a.h.size = sizeof(struct fta_find_arg);
+ a.reuse=reuse;
+ if (strlen(name)>=(MAXFTANAME-1)) {
+ gslog(LOG_EMERG,"ERROR:FTA name (%s) to large\n",name);
+ return -1;
+ }
+ strcpy(a.name,name);
+ ipc_call_and_wait(clearinghouseftaid,(gs_sp_t )&a,rb);
+ if (sr->h.callid != FTAFIND_RESULT) {
+ gslog(LOG_EMERG,"ERROR:Wrong result code received in fta_find\n");
+ return -1;
+ }
+ if (sr->result >= 0) {
+ if (schema !=0) {
+ if (strlen(sr->schema) >= schemasz) {
+ gslog(LOG_EMERG,"Could not fit schema into schema buffer fta_find\n");
+ return -1;
+ } else {
+ strcpy(schema,sr->schema);
+ }
+ }
+ *ftaid=sr->f;
+ }
+ return sr->result;
+}
+
+gs_retval_t fta_alloc_instance(FTAID subscriber,
+ FTAID * ftaid, FTAname name, gs_sp_t schema,
+ gs_uint32_t reusable,
+ gs_int32_t command, gs_int32_t sz, void * data)
+{
+ gs_int8_t rb[MAXRES];
+ struct fta_alloc_instance_arg * a;
+ struct fta_result * fr = (struct fta_result *)rb;
+ struct ringbuf *r;
+
+ /* make sure we have the share memory required */
+ if ((r=gscpipc_createshm(*ftaid,curprocess.buffersize))==0) {
+ gslog(LOG_EMERG,"ERROR:could not allocate shared memory"
+ "for FTA %s\n",name);
+ return -1;
+ }
+
+ if (strlen(name)>=(MAXFTANAME-1)) {
+ gslog(LOG_EMERG,"ERROR:FTA name (%s) to large\n",name);
+ return -1;
+ }
+
+ if (strlen(schema)>=(MAXSCHEMASZ-1)) {
+ gslog(LOG_EMERG,"ERROR:FTA name (%s) to large\n",name);
+ return -1;
+ }
+
+ a = alloca(sizeof(struct fta_alloc_instance_arg) + sz);
+
+ a->h.callid = FTA_ALLOC_INSTANCE;
+ a->h.size = sizeof(struct fta_alloc_instance_arg)+ sz;
+ a->f=*ftaid;
+ a->subscriber=subscriber;
+ a->reusable=reusable;
+ a->command = command;
+ a->sz = sz;
+ memcpy(&a->data[0],data,sz);
+ strcpy(a->name,name);
+ strcpy(a->schema,schema);
+
+ ipc_call_and_wait(*ftaid,(gs_sp_t )a,rb);
+
+ if (fr->h.callid != FTA_RESULT) {
+ gslog(LOG_EMERG,"ERROR:Wrong result code received\n");
+ return -1;
+ }
+
+ *ftaid=fr->f;
+
+ if (fr->result==0) {
+ gslog(LOG_INFO,"Allocated fta instance %s with FTAID {ip=%u,port=%u,index=%u,streamid=%u}\n",name,ftaid->ip,ftaid->port,ftaid->index,ftaid->streamid);
+ return streamregistry_add(*ftaid,r);
+ }
+
+ return fr->result;
+}
+
+gs_retval_t fta_alloc_print_instance(FTAID subscriber,
+ FTAID * ftaid,
+ FTAname name, gs_sp_t schema, gs_uint32_t reusable,
+ gs_int32_t command, gs_int32_t sz, void * data,
+ gs_sp_t path,gs_sp_t basename,
+ gs_sp_t temporal_field, gs_sp_t split_field,
+ gs_uint32_t delta, gs_uint32_t split)
+{
+ gs_int8_t rb[MAXRES];
+ struct fta_alloc_instance_arg * a;
+ struct fta_result * fr = (struct fta_result *)rb;
+
+ if ((strlen(path)>=MAXPRINTSTRING-1)
+ || (strlen(basename)>=MAXPRINTSTRING-1)
+ || (strlen(temporal_field)>=MAXPRINTSTRING-1)) {
+ gslog(LOG_EMERG,"INTERNAL ERROR:fta_alloc_print_instance string"
+ " arguments to long\n");
+ return -1;
+ }
+ if (strlen(name)>=(MAXFTANAME-1)) {
+ gslog(LOG_EMERG,"ERROR:FTA name (%s) to large\n",name);
+ return -1;
+ }
+
+ if (strlen(schema)>=(MAXSCHEMASZ-1)) {
+ gslog(LOG_EMERG,"ERROR:FTA name (%s) to large\n",name);
+ return -1;
+ }
+
+ a = alloca(sizeof(struct fta_alloc_instance_arg) + sz);
+
+ a->h.callid = FTA_ALLOC_PRINT_INSTANCE;
+ a->h.size = sizeof(struct fta_alloc_instance_arg)+ sz;
+ a->f=*ftaid;
+ a->subscriber=subscriber;
+ a->reusable=reusable;
+ a->split=split;
+ strcpy(a->name,name);
+ strcpy(a->schema,schema);
+ a->command = command;
+ a->sz = sz;
+ strcpy(a->path,path);
+ strcpy(a->basename,basename);
+ strcpy(a->temporal_field,temporal_field);
+ strcpy(a->split_field,split_field);
+ a->delta=delta;
+ memcpy(&a->data[0],data,sz);
+
+ ipc_call_and_wait(*ftaid,(gs_sp_t )a,rb);
+
+ if (fr->h.callid != FTA_RESULT) {
+ gslog(LOG_EMERG,"ERROR:Wrong result code received\n");
+ return -1;
+ }
+
+ *ftaid=fr->f;
+
+ return fr->result;
+}
+
+gs_retval_t fta_free_instance(FTAID subscriber, FTAID ftaid, gs_uint32_t recursive)
+{
+ gs_int8_t rb[MAXRES];
+ struct fta_free_instance_arg a;
+ struct standard_result * sr = (struct standard_result *)rb;
+ struct ringbuf *r;
+
+ a.h.callid = FTA_FREE_INSTANCE;
+ a.h.size = sizeof(struct fta_free_instance_arg);
+ a.subscriber=subscriber;
+ a.f=ftaid;
+ a.recursive=recursive;
+ ipc_call_and_wait(ftaid,(gs_sp_t )&a,rb);
+ if (sr->h.callid != STANDARD_RESULT) {
+ gslog(LOG_EMERG,"ERROR:Wrong result code received\n");
+ return -1;
+ }
+
+ /* make sure we remove the mapping*/
+ streamregistry_remove(ftaid);
+
+ return sr->result;
+}
+
+gs_retval_t fta_control(FTAID subscriber,
+ FTAID ftaid, gs_int32_t command, gs_int32_t sz, void * value)
+{
+ gs_int8_t rb[MAXRES];
+ struct fta_control_arg * a;
+ struct standard_result * sr = (struct standard_result *)rb;
+
+ a = alloca(sizeof(struct fta_control_arg) + sz);
+
+ a->h.callid = FTA_CONTROL;
+ a->h.size = sizeof(struct fta_control_arg)+ sz;
+ a->subscriber=subscriber;
+ a->f=ftaid;
+ a->command = command;
+ a->sz = sz;
+ memcpy(&a->data[0],value,sz);
+
+ ipc_call_and_wait(ftaid,(gs_sp_t )a,rb);
+
+ if (sr->h.callid != STANDARD_RESULT) {
+ gslog(LOG_EMERG,"ERROR:Wrong result code received\n");
+ return -1;
+ }
+
+ return sr->result;
+}
+
+gs_retval_t fta_heartbeat(FTAID self,gs_uint64_t trace_id,
+ gs_uint32_t sz, fta_stat * trace){
+#ifdef CLEARINGHOUSE_HEARTBEAT
+ struct fta_heartbeat_arg * a;
+ a = alloca(sizeof(struct fta_heartbeat_arg) + (sz*sizeof(fta_stat)));
+ a->h.callid = FTA_HEARTBEAT;
+ a->h.size = sizeof(struct fta_heartbeat_arg)+(sz*sizeof(fta_stat));
+ a->sender=self;
+ a->trace_id=trace_id;
+ a->sz=sz;
+ if (sz!=0) {
+ memcpy(&a->data[0],trace,(sz*sizeof(fta_stat)));
+ }
+#ifdef PRINTMSG
+ fprintf(stderr, "HOST sending heartbeat to %u.%u.%u.%u:%u of "
+ "type %u with length %u\n",
+ (clearinghouseftaid.ip>>24)&0xff,
+ (clearinghouseftaid.ip>>16)&0xff,
+ (clearinghouseftaid.ip>>8)&0xff,
+ (clearinghouseftaid.ip)&0xff,
+ clearinghouseftaid.port,a->h.callid,a->h.size);
+#endif
+ if (gscpipc_send(clearinghouseftaid,FTACALLBACK,(gs_sp_t)a,a->h.size,1)<0) {
+ gslog(LOG_EMERG,"ERROR:Could not send on message queue\n");
+ return -1;
+ }
+#endif
+ return 0;
+}
+
+gs_retval_t fta_notify_producer_failure(FTAID self, FTAID producer){
+ struct fta_notify_producer_failure_arg a;
+ a.h.callid = FTA_PRODUCER_FAILURE;
+ a.h.size = sizeof(struct fta_notify_producer_failure_arg);
+ a.sender=self;
+ a.producer=producer;
+#ifdef PRINTMSG
+ fprintf(stderr, "HOST sending producer failure to %u.%u.%u.%u:%u of "
+ "type %u with length %u\n",
+ (clearinghouseftaid.ip>>24)&0xff,
+ (clearinghouseftaid.ip>>16)&0xff,
+ (clearinghouseftaid.ip>>8)&0xff,
+ (clearinghouseftaid.ip)&0xff,
+ clearinghouseftaid.port,a.h.callid,a.h.size);
+#endif
+ if (gscpipc_send(clearinghouseftaid,FTACALLBACK,(gs_sp_t)&a,a.h.size,1)<0) {
+ gslog(LOG_EMERG,"ERROR:Could not send on message queue\n");
+ return -1;
+ }
+ return 0;
+}
+
+gs_retval_t process_control(FTAID ftaid, gs_int32_t command, gs_int32_t sz, void * value)
+{
+ gs_int8_t rb[MAXRES];
+ struct process_control_arg * a;
+ struct standard_result * sr = (struct standard_result *)rb;
+
+
+ a = alloca(sizeof(struct process_control_arg) + sz);
+
+ a->h.callid = PROCESS_CONTROL;
+ a->h.size = sizeof(struct process_control_arg)+ sz;
+ a->command = command;
+ a->sz = sz;
+ memcpy(&a->data[0],value,sz);
+
+ ipc_call_and_wait(ftaid,(gs_sp_t )a,rb);
+
+ if (sr->h.callid != STANDARD_RESULT) {
+ gslog(LOG_EMERG,"ERROR:Wrong result code received\n");
+ return -1;
+ }
+
+ return sr->result;
+}
+
+
+static void timeouthandler ()
+{
+ struct timeout_result a;
+
+ a.h.callid=TIMEOUT;
+ a.h.size=sizeof(struct timeout_result);
+ if (gscpipc_send(gscpipc_getftaid(), FTACALLBACK, (gs_sp_t )&a,a.h.size,1)<0) {
+ gslog(LOG_EMERG,"ERROR:Could not send on message queue\n");
+ }
+}
+
+gs_retval_t gscp_get_buffer(FTAID * ftaid, gs_int32_t * size, void *tbuffer,
+ gs_int32_t tbuf_size, gs_int32_t timeout)
+{
+ struct ringbuf * r;
+ FTAID from;
+ gs_int32_t length;
+ gs_int8_t buf[MAXMSGSZ];
+ gs_int32_t lopp;
+ FTAID * f;
+ static gs_uint64_t s1=0;
+ static gs_uint64_t s2;
+ if (s1==0) {
+ s1=rdtsc();
+ }
+ s2=rdtsc();
+ cycles+=(s2-s1);
+start:
+#ifdef PRINTMSG
+ fprintf(stderr,"CHECK RINGBUFS\n");
+#endif
+#ifndef POLLING
+ /* use chance to cleanout message queue no reason
+ to keep anything else */
+ while (gscpipc_read(&from,&lopp,buf,&length,0)>0);
+#endif
+
+ streamregistry_getactiveringbuf_reset();
+ while ((r=streamregistry_getactiveringbuf())>0) {
+#ifdef PRINTMSG
+ fprintf(stderr,"Reading from ringpuffer %p [%p:%u]"
+ "(%u %u %u) \n",r,&r->start,r->end,r->reader,r->writer,
+ r->length);
+ if (UNREAD(r)) {
+ fprintf(stderr,"\t%u %u.%u.%u.%u:%u-%p %u\n",CURREAD(r)->next,
+ (CURREAD(r)->f.ip>>24)&0xff,
+ (CURREAD(r)->f.ip>>16)&0xff,
+ (CURREAD(r)->f.ip>>8)&0xff,
+ (CURREAD(r)->f.ip)&0xff,
+ CURREAD(r)->f.port,
+ CURREAD(r)->f.streamid,
+ CURREAD(r)->sz);
+ }
+
+#endif
+ if (UNREAD(r)) {
+ *ftaid=(CURREAD(r)->f);
+ *size=CURREAD(r)->sz;
+ memcpy(tbuffer,&(CURREAD(r)->data[0]),(tbuf_size>(*size))?(*size):tbuf_size);
+ intuple++;
+ inbytes+=CURREAD(r)->sz;
+ ADVANCEREAD(r);
+ s1=rdtsc();
+ return 0;
+ }
+ }
+ if (timeout == -1) {
+ *size=0;
+ s1=rdtsc();
+ return 1;
+ }
+ if (timeout !=0) {
+ signal(SIGALRM, timeouthandler);
+ alarm(timeout);
+ }
+
+#ifndef POLLING
+#ifdef PRINTMSG
+ fprintf(stderr,"START BLOCKCALLS\n");
+#endif
+ streamregistry_getactiveftaid_reset();
+ while ((f=streamregistry_getactiveftaid())!=0) {
+ struct gscp_get_buffer_arg a;
+ a.h.callid = GSCP_GET_BUFFER;
+ a.h.size = sizeof(struct gscp_get_buffer_arg);
+ a.timeout = timeout;
+#ifdef PRINTMSG
+ fprintf(stderr,"Waiting for %u.%u.%u.%u:%u\n",
+ (f->ip>>24)&0xff,
+ (f->ip>>16)&0xff,
+ (f->ip>>8)&0xff,
+ (f->ip)&0xff,
+ f->port
+ );
+#endif
+ if (gscpipc_send(*f,FTACALLBACK,(gs_sp_t )&a,a.h.size,1)<0) {
+ s1=rdtsc();
+ return -1;
+ }
+ }
+#ifdef PRINTMSG
+ fprintf(stderr,"BLOCK\n");
+#endif
+ while (gscpipc_read(&from,&lopp,buf,&length,1)>0) {
+#else // If we poll we return after 100 msec
+ sleepagain:
+ while (gscpipc_read(&from,&lopp,buf,&length,2)>0) {
+#endif
+ struct standard_result * sr = (struct standard_result *) buf;
+#ifdef PRINTMSG
+ fprintf(stderr,"Got return code %u\n",sr->h.callid);
+#endif
+ if (lopp==FTACALLBACK) {
+ if (timeout != 0) {
+ signal(SIGALRM, SIG_IGN);
+ }
+ if (sr->h.callid == WAKEUP) {
+ /* use chance to cleanout message queue no reason
+ to keep anything else */
+ while (gscpipc_read(&from,&lopp,buf,&length,0)>0);
+ goto start;
+ }
+ if (sr->h.callid == TIMEOUT) {
+ /* use chance to cleanout message queue no reason
+ to keep anything else */
+ while (gscpipc_read(&from,&lopp,buf,&length,0)>0);
+ *size=0;
+ s1=rdtsc();
+ return 1;
+ }
+ if (sidequeue_append(from,buf,length)<0) {
+ gslog(LOG_EMERG,"ERROR:: Could not add to sidequeue\n");
+ s1=rdtsc();
+ return -1;
+ }
+ }
+ }
+#ifdef POLLING
+ streamregistry_getactiveringbuf_reset();
+ while ((r=streamregistry_getactiveringbuf())>0) {
+#ifdef PRINTMSG
+ fprintf(stderr,"Reading from ringpuffer %p [%p:%u]"
+ "(%u %u %u) \n",r,&r->start,r->end,r->reader,r->writer,
+ r->length);
+ if (UNREAD(r)) {
+ fprintf(stderr,"\t%u %u.%u.%u.%u:%u-%p %u\n",CURREAD(r)->next,
+ (CURREAD(r)->f.ip>>24)&0xff,
+ (CURREAD(r)->f.ip>>16)&0xff,
+ (CURREAD(r)->f.ip>>8)&0xff,
+ (CURREAD(r)->f.ip)&0xff,
+ CURREAD(r)->f.port,
+ CURREAD(r)->f.streamid,
+ CURREAD(r)->sz);
+ }
+
+#endif
+ if (UNREAD(r)) {
+ *ftaid=(CURREAD(r)->f);
+ *size=CURREAD(r)->sz;
+ memcpy(tbuffer,&(CURREAD(r)->data[0]),(tbuf_size>(*size))?(*size):tbuf_size);
+ intuple++;
+ inbytes+=CURREAD(r)->sz;
+ ADVANCEREAD(r);
+ if (timeout != 0) {
+ signal(SIGALRM, SIG_IGN);
+ }
+ s1=rdtsc();
+ return 0;
+ }
+ }
+ goto sleepagain; // Try again
+#endif
+ gslog(LOG_EMERG,"Unexpected code reached in: gscp_get_buffer \n");
+ /* we should never get here */
+ s1=rdtsc();
+ return -1;
+ }
+
+
+
Code Review / com / gs-lite.git / blobdiff