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