-/* ------------------------------------------------
- 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 "gsconfig.h"
-#include "gstypes.h"
-#include <gscpipc.h>
-#include <ipcencoding.h>
-#include <callbackregistries.h>
-#include <clearinghouseregistries.h>
-#include <lappregistries.h>
-#include <stdlib.h>
-#include <stdio.h>
-#include <lapp.h>
-#include <string.h>
-#include "rdtsc.h"
-
-gs_uint64_t shared_memory_full_warning =0;
-
-static gs_int32_t maxsnaplen = 0;
-
-struct FTAID clearinghouseftaid;
-
-gs_uint64_t intupledrop=0;
-gs_uint64_t outtupledrop=0;
-gs_uint64_t intuple=0;
-gs_uint64_t outtuple=0;
-gs_uint64_t inbytes=0;
-gs_uint64_t outbytes=0;
-gs_uint64_t cycles=0;
-
-
-
-/* following function is internal and defined in lappinterface.c */
-gs_retval_t ipc_call_and_wait(FTAID f, gs_sp_t msg, gs_sp_t result);
-
-gs_retval_t gscp_blocking_mode() {
-#ifdef BLOCKRINGBUFFER
- return 1;
-#else
- return 0;
-#endif
-}
-
-static void clock_signal_check() {
- struct FTA * fa;
- static gs_int32_t t=0;
- if (t==0) {
- t=time(0);
- } else {
- if (time(0)>t) {
- if (ftaexec_start()<0) {
- gslog(LOG_EMERG,"GSCPRTS::error::could not init check of "
- "FTA list\n");
- return;
- }
- while ((fa=ftaexec_next())!=0) {
- if (fa->clock_fta!=0) {
- fa->clock_fta(fa);
- }
- }
- t=time(0);
- if (t%GSLOGINTERVAL==0) gsstats();// log all the stats
- }
- }
-}
-
-static gs_retval_t send_standard_reply(FTAID f, gs_int32_t result) {
- struct standard_result r;
- r.h.callid=STANDARD_RESULT;
- r.h.size=sizeof(struct standard_result);
- r.result=result;
- if (gscpipc_send(f, FTACALLBACK, (gs_sp_t) &r,r.h.size,1)<0) {
- gslog(LOG_EMERG,"ERROR:Could not send on message queue\n");
- return -1;
- }
- return 0 ;
-}
-
-static gs_retval_t send_lookup_reply(FTAID f, gs_int32_t result,
- FTAID * ftaid,
- gs_sp_t* schema) {
- struct ftafind_result r;
- r.h.callid=FTAFIND_RESULT;
- r.h.size=sizeof(struct ftafind_result);
- r.result=result;
- r.f=*ftaid;
- if (result >=0) {
- if (strlen(*schema)>=(MAXSCHEMASZ-1)) {
- gslog(LOG_EMERG,"ERROR:FTA schema (%s) to large\n",(unsigned char *)schema);
- r.result=-1;
- } else {
- strcpy(r.schema,*schema);
- }
- }
- if (gscpipc_send(f, FTACALLBACK, (gs_sp_t) &r,r.h.size,1)<0) {
- gslog(LOG_EMERG,"ERROR:Could not send on message queue\n");
- return -1;
- }
- return 0 ;
-}
-
-
-static gs_retval_t send_fta_result(FTAID f,
- FTAID * ftaid, gs_int32_t result) {
- struct fta_result r;
- r.h.callid=FTA_RESULT;
- r.h.size=sizeof(struct fta_result);
- r.result=result;
- if (ftaid!=0) {
- r.f=*ftaid;
- }
- if (gscpipc_send(f, FTACALLBACK,(gs_sp_t)&r,r.h.size,1)<0) {
- gslog(LOG_EMERG,"ERROR:Could not send on message queue\n");
- return -1;
- }
- return 0 ;
-}
-
-
-// Is also used by the lfta rts enviroment on a post. So make it none
-// static.
-gs_retval_t send_wakeup(FTAID f)
-{
- struct wakeup_result a;
-
- a.h.callid=WAKEUP;
- a.h.size=sizeof(struct wakeup_result);
- if (gscpipc_send(f, FTACALLBACK, (gs_sp_t)&a,a.h.size,0)<0) {
- gslog(LOG_EMERG,"ERROR:Could not send on message queue\n");
- return -1;
- }
- return 0;
-}
-
-static gs_retval_t fta_register_instance(FTAID subscriber,
- FTAID f,gs_uint32_t reusable,
- FTAname name,
- gs_csp_t schema) {
- gs_int8_t rb[MAXRES];
- struct fta_register_arg a;
- struct standard_result * sr = (struct standard_result *)rb;
-
- if (curprocess.type != CLEARINGHOUSE) {
- a.h.callid = FTA_REGISTER;
- a.h.size = sizeof(struct fta_register_arg);
- 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 schema (%s) to large\n",schema);
- return -1;
- }
- strcpy(a.name,name);
- strcpy(a.schema,schema);
- a.f=f;
- a.subscriber=subscriber; /* consumer is the same as f for an FTA*/
- a.reusable=reusable;
- ipc_call_and_wait(clearinghouseftaid,(gs_sp_t )&a,rb);
- if (sr->h.callid != STANDARD_RESULT) {
- gslog(LOG_EMERG,"ERROR:Wrong result code received\n");
- return -1;
- }
- if (sr->result != 0) {
- return -1;
- }
- } else {
- if (ftalookup_register_fta(subscriber,f,name,reusable,schema)<0) {
- return -1;
- }
- }
- return 0;
-}
-
-static gs_retval_t fta_unregister_instance(FTAID subscriber,
- FTAID f) {
- gs_int8_t rb[MAXRES];
- struct fta_unregister_arg a;
- struct standard_result * sr = (struct standard_result *)rb;
-
- if (curprocess.type != CLEARINGHOUSE) {
- a.h.callid = FTA_UNREGISTER;
- a.h.size = sizeof(struct fta_register_arg);
- a.f=f;
- a.subscriber=subscriber; /* consumer is the same as f for an FTA*/
- ipc_call_and_wait(clearinghouseftaid,(gs_sp_t )&a,rb);
- if (sr->h.callid != STANDARD_RESULT) {
- gslog(LOG_EMERG,"ERROR:Wrong result code received\n");
- return -1;
- }
- return -1;
- } else {
- if (ftalookup_unregister_fta(f,subscriber)<0) {
- return -1;
- }
- }
- return 0;
-}
-
-gs_retval_t fta_start_service(gs_int32_t number)
-{
- gs_int8_t buf[MAXMSGSZ];
- FTAID from;
- gs_int32_t length;
- struct hostcall * h= (struct hostcall *) buf;
- gs_int32_t forever=0;
- gs_int32_t endtime=0;
- gs_int32_t block=1;
- gs_int32_t res;
- struct ringbuf * r;
- FTAID ftaid;
- FTAID * ftaidp;
- struct FTA * fta;
- gs_int32_t lopp;
- gs_int32_t preemptq;
- gs_int32_t endq;
- gs_uint64_t s1;
- gs_uint64_t s2;
-
- if (number == 0) {
- block=0;
- forever=1;
- }
- if (number < 0) {
- forever=1;
- }
- if (number > 0) {
- block=1;
- endtime=time(0)+number;
- }
-
- while((forever!=0)||
- (endtime==0)||(endtime>time(0))) {
- /* check if we need to give the FTAs there clock signal */
- if ((curprocess.type == LFTA)
- ||(curprocess.type == HFTA)) {
- clock_signal_check();
- }
-
-#ifdef POLLING
- poll:
-#endif
- preemptq=0;
- /* first empty out sidequeu then read from messagequeue */
- if (sidequeue_pop(&from,buf,&length)<0) {
- /* empty out the sidequeue before processing the shared
- memory */
- if (curprocess.type == HFTA) {
- /* process all the shared memory regions and register
- for callbacks */
- s1=rdtsc();
- endq=time(0)+2;
- streamregistry_getactiveringbuf_reset();
- while ((r=streamregistry_getactiveringbuf())>0) {
- while (UNREAD(r)) {
- struct FTA * fa;
- if (ftaexec_start()<0) {
- gslog(LOG_EMERG,"GSCPRTS::error::could not init check of "
- "FTA list\n");
- return -1;
- }
- while ((fa=ftaexec_next())!=0) {
- gs_int32_t x;
- for(x=0;x<fa->stream_subscribed_cnt;x++) {
- if ((fa->stream_subscribed[x].streamid
- ==CURREAD(r)->f.streamid)
- && (fa->stream_subscribed[x].ip
- ==CURREAD(r)->f.ip)
- && (fa->stream_subscribed[x].port
- ==CURREAD(r)->f.port)) {
- fa->accept_packet(fa,&(CURREAD(r)->f),
- &(CURREAD(r)->data[0]),
- CURREAD(r)->sz);
- }
- }
- }
- intuple++;
- inbytes+=CURREAD(r)->sz;
- ADVANCEREAD(r);
- if (endq <= time(0)) {
- preemptq=1;
- goto processmsg;
- }
- }
- }
- s2=rdtsc();
- cycles+=(s2-s1);
-#ifndef POLLING
- /* register wakeups all arround to make sure we don't sleep
- * for ever,
- */
- streamregistry_getactiveftaid_reset();
- while ((ftaidp=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 = 0;
- if (gscpipc_send(*ftaidp,FTACALLBACK,(gs_sp_t )&a,a.h.size,1)<0) {
- return -1;
- }
- }
-#endif
- }
- processmsg:
- do {
- /* even if we block we return every 100msec to be able to generate the clock signal to
- * the HFTAs
- */
- if ((res=gscpipc_read(&from,&lopp,buf,&length,((block==1)&&(preemptq==0))?2:0))<0) {
- gslog(LOG_EMERG,"GSCPRTS::error::reading from messagequeue\n");
- return -1;
- }
- /* check if we need to give the FTAs there clock signal */
- if ((curprocess.type == LFTA)
- ||(curprocess.type == HFTA)) {
- clock_signal_check();
- }
- if ((res==0) && (block==0)) {
- /* nonblocking and nothing to do so return */
- return 0;
- }
- if ((res==0) && (endtime!=0) && (endtime<time(0))) {
- /* timeout reached so return */
- return 0;
- }
-#ifdef POLLING
- if ((res==0)&&(curprocess.type == HFTA)) {
- goto poll;
- }
-#endif
- } while (res==0);
- if ((lopp)!=FTACALLBACK) {
- gslog(LOG_EMERG,"GSCPRTS::error::unknown lowlevel opp\n");
- return -1;
- }
- }
-
-#ifdef PRINTMSG
- gslog(LOG_EMERG, "HFTA message from %u of type %u of length %u\n",from,
- h->callid,h->size);
-#endif
- switch (h->callid) {
-
- case FTA_LOOKUP: {
- if (curprocess.type == CLEARINGHOUSE) {
- struct fta_find_arg * n;
- FTAID rf;
- gs_csp_t schema;
- n = (struct fta_find_arg *) buf;
- /* Note: Side effect of ftalookup_lookup_fta_index is to
- fill msgid and index */
- if (send_lookup_reply(from,
- ftalookup_lookup_fta_index(from,
- n->name,
- n->reuse,
- &rf,
- &schema),
- &rf,
- (gs_sp_t *)&schema)<0) {
- gslog(LOG_EMERG,"GSCPRTS::error::send standard reply faild\n");
- return -1;
- }
- } else {
- if (send_standard_reply(from,-1)<0) {
- gslog(LOG_EMERG,"GSCPRTS::error::send standard reply faild\n");
- return -1;
- }
- gslog(LOG_EMERG,"GSCPRTS::error: Non clearinghouse proccess got"
- "contacted for clearinghouse processing\n");
- }
- }
- break;
- case FTA_REGISTER: {
- if (curprocess.type == CLEARINGHOUSE) {
- struct fta_register_arg * n;
- n = (struct fta_register_arg *) buf;
- if (send_standard_reply(from,
- ftalookup_register_fta(n->subscriber,
- n->f,
- n->name,
- n->reusable,
- n->schema))<0) {
- gslog(LOG_EMERG,"GSCPRTS::error::send standard reply faild\n");
- return -1;
- }
- } else {
- if (send_standard_reply(from,-1)<0) {
- gslog(LOG_EMERG,"GSCPRTS::error::send standard reply faild\n");
- return -1;
- }
- gslog(LOG_EMERG,"GSCPRTS::error: Non clearinghouse proccess got"
- "contacted for clearinghouse processing\n");
- }
- }
- break;
- case FTA_UNREGISTER: {
- if (curprocess.type == CLEARINGHOUSE) {
- struct fta_unregister_arg * n;
- n = (struct fta_unregister_arg *) buf;
- if (send_standard_reply(from,
- ftalookup_unregister_fta(n->subscriber,
- n->f))<0) {
- gslog(LOG_EMERG,"GSCPRTS::error::send standard reply faild\n");
- return -1;
- }
- } else {
- if (send_standard_reply(from,-1)<0) {
- gslog(LOG_EMERG,"GSCPRTS::error::send standard reply faild\n");
- return -1;
- }
- gslog(LOG_EMERG,"GSCPRTS::error: Non clearinghouse proccess got"
- "contacted for clearinghouse processing\n");
- }
- }
- break;
- case FTA_ALLOC_INSTANCE:
- case FTA_ALLOC_PRINT_INSTANCE:
- if ((curprocess.type == LFTA)
- ||(curprocess.type == HFTA)) {
- struct ringbuf * r;
- struct fta_alloc_instance_arg * n;
- n = (struct fta_alloc_instance_arg *) buf;
- if ((fta=ftaexec_alloc_instance(n->f.index,
- (struct FTA *)n->f.streamid,
- n->reusable,
- n->command,
- n->sz,
- &(n->data[0])))==0) {
- gslog(LOG_EMERG,"GSCPRTS::warning::could not allocate"
- "FTA\n");
- if (send_fta_result(from,0,-1)<0) {
- gslog(LOG_EMERG,"GSCPRTS::error::send standard "
- "reply faild\n");
- return -1;
- }
- } else {
- /* shared memory is only required if data is beeing transfered */
- if ((h->callid!=FTA_ALLOC_PRINT_INSTANCE)&&
- ((r=gscpipc_getshm(from))==0)) {
- gslog(LOG_EMERG,"GSCPRTS::warning::could not get"
- "shared memory\n");
- ftaexec_remove(fta);
- if (send_fta_result(from,0,-1)<0) {
- gslog(LOG_EMERG,"GSCPRTS::error::send standard "
- "reply faild\n");
- return -1;
- }
- } else {
- /* no callback to register for print function */
- if ((h->callid!=FTA_ALLOC_PRINT_INSTANCE)&&(ftacallback_add_streamid(r,fta->ftaid.streamid)!=0)) {
- gslog(LOG_EMERG,"GSCPRTS::warning::could not add"
- "streamid to ringbuffer\n");
- ftaexec_free_instance(fta,1);
- if (send_fta_result(from,0,-1)<0) {
- gslog(LOG_EMERG,"GSCPRTS::error::send standard "
- "reply faild\n");
- return -1;
- }
- } else {
- if (ftaexec_insert(0,fta)<0) {
- gslog(LOG_EMERG,"GSCPRTS::warning::could not"
- "insert FTA\n");
- ftacallback_rm_streamid(r,fta->ftaid.streamid);
- ftaexec_free_instance(fta,1);
- if (send_fta_result(from,0,-1)<0) {
- gslog(LOG_EMERG,"GSCPRTS::error::send standard "
- "reply faild\n");
- return -1;
- }
- } else {
- if (fta_register_instance(n->subscriber,fta->ftaid,
- n->reusable,
- n->name,
- n->schema)!=0) {
- gslog(LOG_EMERG,"GSCPRTS::warning::could not register"
- " instance\n");
- ftaexec_remove(fta);
- ftacallback_rm_streamid(r,fta->ftaid.streamid);
- ftaexec_free_instance(fta,1);
- if (send_fta_result(from,0,-1)<0) {
- gslog(LOG_EMERG,"GSCPRTS::error::send standard "
- "reply faild\n");
- return -1;
- }
- } else {
- if (h->callid==FTA_ALLOC_PRINT_INSTANCE) {
- if (curprocess.type == LFTA) {
- gslog(LOG_EMERG,"GSCPRTS::error:: alloc print instance not "
- "implemented for LFTA.\n");
- ftaexec_remove(fta);
- ftaexec_free_instance(fta,1);
- if (send_fta_result(from,0,-1)<0) {
- gslog(LOG_EMERG,"GSCPRTS::error::send standard "
- "reply faild\n");
- return -1;
- }
- } else {
- if (add_printfunction_to_stream(fta, n->schema, n->path, n->basename,
- n->temporal_field, n->split_field, n->delta, n->split) < 0) {
- ftaexec_remove(fta);
- ftaexec_free_instance(fta,1);
- if (send_fta_result(from,0,-1)<0) {
- gslog(LOG_EMERG,"GSCPRTS::error::send standard "
- "reply faild\n");
- return -1;
- }
- }
- }
- }
- if (send_fta_result(from,&fta->ftaid,0)<0) {
- gslog(LOG_EMERG,"GSCPRTS::error::send standard "
- "reply faild\n");
- return -1;
- }
- }
- }
- }
- }
- }
- } else {
- if (send_fta_result(from,0,-1)<0) {
- gslog(LOG_EMERG,"GSCPRTS::error::send standard "
- "reply faild\n");
- return -1;
- }
- }
- break;
- case FTA_FREE_INSTANCE:{
- if ((curprocess.type == LFTA)
- ||(curprocess.type == HFTA)) {
- struct fta_free_instance_arg * n;
- n = (struct fta_free_instance_arg *) buf;
- if (((r=gscpipc_getshm(from))!=0)
- && ( ftaexec_remove((struct FTA *) n->f.streamid)==0)
- && ( ftacallback_rm_streamid(r,n->f.streamid)==0)
- && (ftaexec_free_instance((struct FTA *)n->f.streamid,
- n->recursive)
- ==0)
- && (fta_unregister_instance(n->subscriber,n->f)<0)) {
- if (send_standard_reply(from,0)<0) {
- gslog(LOG_EMERG,"GSCPRTS::error::send standard reply faild\n");
- return -1;
- }
- } else {
- if (send_standard_reply(from,-1)<0) {
- gslog(LOG_EMERG,"GSCPRTS::error::send standard reply faild\n");
- return -1;
- }
- }
- } else {
- if (send_standard_reply(from,-1)<0) {
- gslog(LOG_EMERG,"GSCPRTS::error::send standard reply faild\n");
- return -1;
- }
- }
- }
- break;
- case FTA_CONTROL:{
- if ((curprocess.type == LFTA)
- ||(curprocess.type == HFTA)) {
- struct fta_control_arg * n;
- n = (struct fta_control_arg *) buf;
- if (send_standard_reply(from,
- ftaexec_control((struct FTA *)
- n->f.streamid,
- n->command,
- n->sz,
- &(n->data[0])))<0) {
- gslog(LOG_EMERG,"GSCPRTS::error::send standard reply faild\n");
- return -1;
- }
- } else {
- if (send_standard_reply(from,-1)<0) {
- gslog(LOG_EMERG,"GSCPRTS::error::send standard reply faild\n");
- return -1;
- }
- gslog(LOG_EMERG,"GSCPRTS::error: Non clearinghouse or HFTA proccess got"
- "contacted for clearinghouse or HFTA processing\n");
- }
- }
- break;
- case FTA_PRODUCER_FAILURE: {
- if (curprocess.type == CLEARINGHOUSE) {
- struct fta_notify_producer_failure_arg * n;
- n = (struct fta_notify_producer_failure_arg *) buf;
- ftalookup_producer_failure(n->sender,n->producer);
- }
- }
- break;
- case FTA_HEARTBEAT: {
- if (curprocess.type == CLEARINGHOUSE) {
- struct fta_heartbeat_arg * n;
- n = (struct fta_heartbeat_arg *) buf;
- ftalookup_heartbeat(n->sender,n->trace_id,
- n->sz,&(n->data[0]));
- }
- }
- break;
- case GSCP_GET_BUFFER:{
- struct sgroup_get_buffer_arg * n;
- gs_int32_t res;
- struct ringbuf * r;
- n = (struct sgroup_get_buffer_arg *) buf;
-
- if ((r=gscpipc_getshm(from))==0) {
- gslog(LOG_EMERG,"GSCPRTS::error::proccess blocked without"
- "sharedmemory\n");
- } else {
- if (UNREAD(r)) {
- /* something arrived in the meantime so wakeup
- right away */
- if (send_wakeup(from)<0) {
- gslog(LOG_EMERG,"ERROR:Could not send wakeup\n");
- return -1;
- }
- } else {
-#ifndef POLLING
- if (ftacallback_add_wakeup(from,r)<0) {
- gslog(LOG_EMERG,"ERROR:Could not add wakeup\n");
- return -1;
- }
-#else
- gslog(LOG_EMERG,"Received wakeup request on polling systems\n");
-#endif
- }
- }
- }
- break;
- case PROCESS_CONTROL:{
- if ((curprocess.type == LFTA)
- ||(curprocess.type == HFTA)) {
- struct process_control_arg * n;
- n = (struct process_control_arg *) buf;
- if (send_standard_reply(from,
- ftaexec_process_control(n->command,
- n->sz,
- &(n->data[0])))<0) {
- gslog(LOG_EMERG,"GSCPRTS::error::send standard reply faild\n");
- return -1;
- }
- } else {
- if (send_standard_reply(from,-1)<0) {
- gslog(LOG_EMERG,"GSCPRTS::error::send standard reply faild\n");
- return -1;
- }
- gslog(LOG_EMERG,"GSCPRTS::error: Non clearinghouse or HFTA proccess got"
- "contacted for clearinghouse or HFTA processing\n");
- }
- }
- break;
- case WAKEUP:
- case TIMEOUT:
- break;
- default:
- gslog(LOG_EMERG,"GSCPRTS::error::illegal message queue type %u\n",h->callid);
- return -1;
- }
- /* use this occation to cleanup the messagequeue we can't afford
- a backlog in the real message queue since it is limited in
- size */
- while (gscpipc_read(&from,&lopp,buf,&length,0)>0) {
-#ifdef PRINTMSG
- gslog(LOG_EMERG, "request from %u of type %u with length %u\n",from,
- h->callid,h->size);
-#endif
- if ((lopp == FTACALLBACK) && (h->callid < RESULT_OPCODE_IGNORE)) {
- if (sidequeue_append(from,buf,length)<0) {
- gslog(LOG_EMERG,"ERROR:: Could not add to sidequeue\n");
- return -1;
- }
- }
-
- }
- }
- return 0;
-}
-
-static gs_retval_t map_match(gs_csp_t dev) {
- gs_int32_t x;
- for(x=0;x<curprocess.mapcnt;x++){
- if (strcmp(dev,curprocess.map[x])==0) {
- return 1;
- }
- }
- return 0;
-}
-
-gs_retval_t fta_max_snaplen()
-{
- return maxsnaplen;
-}
-
-FTAID fta_register(FTAname name,gs_uint32_t reusable, DEVname dev,
- alloc_fta fta_alloc_functionptr,
- gs_csp_t schema, gs_int32_t snaplen, gs_uint64_t prefilter)
-{
- gs_int8_t rb[MAXRES];
- struct fta_register_arg a;
- struct standard_result * sr = (struct standard_result *)rb;
- gs_int32_t index;
- FTAID res;
- FTAID reserr;
- res.ip=0;
- res.port=0;
- res.index=0;
- res.streamid=0;
- reserr=res;
-
- /* check if the device matches for the registration */
- if (((dev==0) && (curprocess.deviceid==0))
- || ((dev!=0)&&(map_match(dev)==1))) {
- if (dev!=0) {
- gslog(LOG_INFO,"Register %s on device %s\n",name,dev);
- } else {
- gslog(LOG_INFO,"Register %s on default device\n",name);
- }
- if ((index=ftacallback_add_alloc(name,fta_alloc_functionptr,prefilter))<0) {
- gslog(LOG_EMERG,"ERROR could not register callback\n");
- return res;
- }
-
- if (snaplen<0) {
- maxsnaplen=-1;
- } else {
- if (maxsnaplen!=-1) {
- maxsnaplen=(snaplen>maxsnaplen)?snaplen:maxsnaplen;
- }
- }
-
- res=gscpipc_getftaid();
- res.index=index;
- if (curprocess.type != CLEARINGHOUSE) {
- a.h.callid = FTA_REGISTER;
- a.h.size = sizeof(struct fta_register_arg);
- if (strlen(name)>=(MAXFTANAME-1)) {
- gslog(LOG_EMERG,"ERROR:FTA name (%s) to large\n",name);
- return reserr;
- }
- if (strlen(schema)>=(MAXSCHEMASZ-1)) {
- gslog(LOG_EMERG,"ERROR:FTA schema (%s) to large\n",schema);
- return reserr;
- }
- strcpy(a.name,name);
- strcpy(a.schema,schema);
- a.f=res;
- a.subscriber=res; /* consumer is the same as f for an FTA*/
- a.reusable=reusable;
- ipc_call_and_wait(clearinghouseftaid,(gs_sp_t )&a,rb);
- if (sr->h.callid != STANDARD_RESULT) {
- gslog(LOG_EMERG,"ERROR:Wrong result code received\n");
- return reserr;
- }
- if (sr->result != 0 ) {
- gslog(LOG_EMERG,"ERROR:Error in registration\n");
- return reserr;
- }
- } else {
- if (ftalookup_register_fta(gscpipc_getftaid(),
- res,name,reusable,schema)!=0) {
- return res;
- }
- }
- }
- return res;
-}
-
-gs_retval_t fta_unregister(FTAID ftaid)
-{
- gs_int8_t rb[MAXRES];
- struct fta_unregister_arg a;
- struct standard_result * sr = (struct standard_result *)rb;
- if (ftacallback_rm_alloc(ftaid.index)<0) {
- gslog(LOG_EMERG,"ERROR could not unregister callback\n");
- return -1;
- }
-
- if (curprocess.type != CLEARINGHOUSE) {
- a.h.callid = FTA_UNREGISTER;
- a.h.size = sizeof(struct fta_register_arg);
- a.f=ftaid;
- ipc_call_and_wait(clearinghouseftaid,(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;
- } else {
- return ftalookup_unregister_fta(gscpipc_getftaid(),ftaid);
- }
- return 0;
-}
-
-gs_retval_t hfta_post_tuple(struct FTA * self, gs_int32_t sz, void *tuple)
-{
- struct ringbuf * r;
- gs_uint32_t msgid;
- struct wakeup_result a;
- FTAID * f;
- gs_int32_t state;
-
- if (sz>MAXTUPLESZ) {
- gslog(LOG_EMERG,"Maximum tuple size is %u\n",MAXTUPLESZ);
- return -1;
- }
-
- if (self->printfunc.in_use==1) {
- return print_stream(self,sz,tuple);
- }
-
- if (ftacallback_start_streamid((gs_p_t )self)<0) {
- gslog(LOG_EMERG,"ERROR:Post for unkown streamid\n");
- return -1;
- }
- /* now make sure we have space to write in all atomic ringbuffer */
- while((r=ftacallback_next_streamid(&state))!=0) {
- if (state == HFTA_RINGBUF_ATOMIC) {
-#ifdef BLOCKRINGBUFFER
- while (!SPACETOWRITE(r)) {
- usleep(100);
- }
-#endif
- if (! SPACETOWRITE(r)) {
- /* atomic ring buffer and no space so post nothing */
- return -1;
- }
- }
- }
-
-
- if (ftacallback_start_streamid((gs_p_t )self)<0) {
- gslog(LOG_EMERG,"ERROR:Post for unkown streamid\n");
- return -1;
- }
-
- while((r=ftacallback_next_streamid(&state))!=0) {
- if (state != HFTA_RINGBUF_SUSPEND) {
- if (!SPACETOWRITE(r)) {
- //since memory is full we set a warning
- shared_memory_full_warning++;
- // give receiver a chance to clean up
- usleep(0);
- }
- if (SPACETOWRITE(r)) {
- CURWRITE(r)->f=self->ftaid;
- CURWRITE(r)->sz=sz;
- memcpy(&(CURWRITE(r)->data[0]),tuple,sz);
- outtuple++;
- outbytes=outbytes+CURWRITE(r)->sz;
- ADVANCEWRITE(r);
-#ifdef PRINTMSG
- gslog(LOG_EMERG,"Wrote in ringpuffer %p [%p:%u]"
- "(%u %u) \n",r,&r->start,r->end,r->reader,r->writer);
- gslog(LOG_EMERG,"\t%u %u\n",CURREAD(r)->next,
- CURREAD(r)->sz);
-#endif
- } else {
- outtupledrop++;
- }
- if (HOWFULL(r) > 500) {
- // buffer is at least half full
- shared_memory_full_warning++;
-#ifdef PRINTMSG
- gslog(LOG_EMERG,"\t\t buffer full\n");
-#endif
- }
- }
- }
-#ifndef POLLING
- if (ftacallback_start_wakeup((gs_p_t ) self)<0) {
- gslog(LOG_EMERG,"ERROR:Wakeup for unkown streamid\n");
- return -1;
- }
- a.h.callid=WAKEUP;
- a.h.size=sizeof(struct wakeup_result);
- while((f=ftacallback_next_wakeup())!=0) {
- if (send_wakeup(*f)<0) {
- gslog(LOG_EMERG,"ERROR:Could not send wakeup\n");
- return -1;
- }
- }
-#endif
- return 0;
-}
-
-gs_retval_t hfta_get_ringbuf_space(struct FTA * f, FTAID * r, gs_int32_t * space, gs_int32_t szr, gs_int32_t tuplesz)
-{
- gs_int32_t x=0;
- gs_int32_t state;
- struct ringbuf * ru;
-
- if (f->printfunc.in_use==1) {
- // XXX WHAT TO DO???
- gslog(LOG_INFO,"Checking space for printfunc");
- return 0;
- }
-
- if (ftacallback_start_streamid(f->ftaid.streamid)<0) {
- gslog(LOG_EMERG,"ERROR:Space check for unkown streamid in HFTA\n");
- return -1;
- }
-
- while ((ru=ftacallback_next_streamid(&state))!=0) {
- if (szr > x ) {
- r[x]=ru->destid;
- space[x]=TUPLEFIT(ru,tuplesz);
- }
- x++;
- }
- return x;
-}
-
-
-gs_retval_t hfta_set_ringbuf_type(struct FTA * f, FTAID process, gs_int32_t state)
-{
-
- if (ftacallback_state_streamid(f->ftaid.streamid,process,state)<0) {
- gslog(LOG_EMERG,"ERROR:state change for unkown streamid\n");
- return -1;
- }
- return 0;
-}
+/* ------------------------------------------------\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 "gsconfig.h"\r
+#include "gstypes.h"\r
+#include <gscpipc.h>\r
+#include <ipcencoding.h>\r
+#include <callbackregistries.h>\r
+#include <clearinghouseregistries.h>\r
+#include <lappregistries.h>\r
+#include <stdlib.h>\r
+#include <stdio.h>\r
+#include <lapp.h>\r
+#include <string.h>\r
+#include "rdtsc.h"\r
+\r
+gs_uint64_t shared_memory_full_warning =0;\r
+\r
+static gs_int32_t maxsnaplen = 0;\r
+\r
+struct FTAID clearinghouseftaid;\r
+\r
+gs_uint64_t intupledrop=0;\r
+gs_uint64_t outtupledrop=0;\r
+gs_uint64_t intuple=0;\r
+gs_uint64_t outtuple=0;\r
+gs_uint64_t inbytes=0;\r
+gs_uint64_t outbytes=0;\r
+gs_uint64_t cycles=0;\r
+\r
+\r
+\r
+/* following function is internal and defined in lappinterface.c */\r
+gs_retval_t ipc_call_and_wait(FTAID f, gs_sp_t msg, gs_sp_t result);\r
+\r
+gs_retval_t gscp_blocking_mode() {\r
+#ifdef BLOCKRINGBUFFER\r
+ return 1;\r
+#else\r
+ return 0;\r
+#endif\r
+}\r
+\r
+static void clock_signal_check() {\r
+ struct FTA * fa;\r
+ static gs_int32_t t=0;\r
+ if (t==0) {\r
+ t=time(0);\r
+ } else {\r
+ if (time(0)>t) {\r
+ if (ftaexec_start()<0) {\r
+ gslog(LOG_EMERG,"GSCPRTS::error::could not init check of "\r
+ "FTA list\n");\r
+ return;\r
+ }\r
+ while ((fa=ftaexec_next())!=0) {\r
+ if (fa->clock_fta!=0) {\r
+ fa->clock_fta(fa);\r
+ }\r
+ }\r
+ t=time(0);\r
+ if (t%GSLOGINTERVAL==0) gsstats();// log all the stats\r
+ }\r
+ }\r
+}\r
+\r
+static gs_retval_t send_standard_reply(FTAID f, gs_int32_t result) {\r
+ struct standard_result r;\r
+ r.h.callid=STANDARD_RESULT;\r
+ r.h.size=sizeof(struct standard_result);\r
+ r.result=result;\r
+ if (gscpipc_send(f, FTACALLBACK, (gs_sp_t) &r,r.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
+static gs_retval_t send_lookup_reply(FTAID f, gs_int32_t result,\r
+ FTAID * ftaid,\r
+ gs_sp_t* schema) {\r
+ struct ftafind_result r;\r
+ r.h.callid=FTAFIND_RESULT;\r
+ r.h.size=sizeof(struct ftafind_result);\r
+ r.result=result;\r
+ r.f=*ftaid;\r
+ if (result >=0) {\r
+ if (strlen(*schema)>=(MAXSCHEMASZ-1)) {\r
+ gslog(LOG_EMERG,"ERROR:FTA schema (%s) to large\n",(unsigned char *)schema);\r
+ r.result=-1;\r
+ } else {\r
+ strcpy(r.schema,*schema);\r
+ }\r
+ }\r
+ if (gscpipc_send(f, FTACALLBACK, (gs_sp_t) &r,r.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
+\r
+static gs_retval_t send_fta_result(FTAID f,\r
+ FTAID * ftaid, gs_int32_t result) {\r
+ struct fta_result r;\r
+ r.h.callid=FTA_RESULT;\r
+ r.h.size=sizeof(struct fta_result);\r
+ r.result=result;\r
+ if (ftaid!=0) {\r
+ r.f=*ftaid;\r
+ }\r
+ if (gscpipc_send(f, FTACALLBACK,(gs_sp_t)&r,r.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
+\r
+// Is also used by the lfta rts enviroment on a post. So make it none\r
+// static.\r
+gs_retval_t send_wakeup(FTAID f)\r
+{\r
+ struct wakeup_result a;\r
+\r
+ a.h.callid=WAKEUP;\r
+ a.h.size=sizeof(struct wakeup_result);\r
+ if (gscpipc_send(f, FTACALLBACK, (gs_sp_t)&a,a.h.size,0)<0) {\r
+ gslog(LOG_EMERG,"ERROR:Could not send on message queue\n");\r
+ return -1;\r
+ }\r
+ return 0;\r
+}\r
+\r
+static gs_retval_t fta_register_instance(FTAID subscriber,\r
+ FTAID f,gs_uint32_t reusable,\r
+ FTAname name,\r
+ gs_csp_t schema) {\r
+ gs_int8_t rb[MAXRES];\r
+ struct fta_register_arg a;\r
+ struct standard_result * sr = (struct standard_result *)rb;\r
+\r
+ if (curprocess.type != CLEARINGHOUSE) {\r
+ a.h.callid = FTA_REGISTER;\r
+ a.h.size = sizeof(struct fta_register_arg);\r
+ if (strlen(name)>=(MAXFTANAME-1)) {\r
+ gslog(LOG_EMERG,"ERROR:FTA name (%s) to large\n",name);\r
+ return -1;\r
+ }\r
+ if (strlen(schema)>=(MAXSCHEMASZ-1)) {\r
+ gslog(LOG_EMERG,"ERROR:FTA schema (%s) to large\n",schema);\r
+ return -1;\r
+ }\r
+ strcpy(a.name,name);\r
+ strcpy(a.schema,schema);\r
+ a.f=f;\r
+ a.subscriber=subscriber; /* consumer is the same as f for an FTA*/\r
+ a.reusable=reusable;\r
+ ipc_call_and_wait(clearinghouseftaid,(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
+ if (sr->result != 0) {\r
+ return -1;\r
+ }\r
+ } else {\r
+ if (ftalookup_register_fta(subscriber,f,name,reusable,schema)<0) {\r
+ return -1;\r
+ }\r
+ }\r
+ return 0;\r
+}\r
+\r
+static gs_retval_t fta_unregister_instance(FTAID subscriber,\r
+ FTAID f) {\r
+ gs_int8_t rb[MAXRES];\r
+ struct fta_unregister_arg a;\r
+ struct standard_result * sr = (struct standard_result *)rb;\r
+\r
+ if (curprocess.type != CLEARINGHOUSE) {\r
+ a.h.callid = FTA_UNREGISTER;\r
+ a.h.size = sizeof(struct fta_register_arg);\r
+ a.f=f;\r
+ a.subscriber=subscriber; /* consumer is the same as f for an FTA*/\r
+ ipc_call_and_wait(clearinghouseftaid,(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
+ return -1;\r
+ } else {\r
+ if (ftalookup_unregister_fta(f,subscriber)<0) {\r
+ return -1;\r
+ }\r
+ }\r
+ return 0;\r
+}\r
+\r
+gs_retval_t fta_start_service(gs_int32_t number)\r
+{\r
+ gs_int8_t buf[MAXMSGSZ];\r
+ FTAID from;\r
+ gs_int32_t length;\r
+ struct hostcall * h= (struct hostcall *) buf;\r
+ gs_int32_t forever=0;\r
+ gs_int32_t endtime=0;\r
+ gs_int32_t block=1;\r
+ gs_int32_t res;\r
+ struct ringbuf * r;\r
+ FTAID ftaid;\r
+ FTAID * ftaidp;\r
+ struct FTA * fta;\r
+ gs_int32_t lopp;\r
+ gs_int32_t preemptq;\r
+ gs_int32_t endq;\r
+ gs_uint64_t s1;\r
+ gs_uint64_t s2;\r
+\r
+ if (number == 0) {\r
+ block=0;\r
+ forever=1;\r
+ }\r
+ if (number < 0) {\r
+ forever=1;\r
+ }\r
+ if (number > 0) {\r
+ block=1;\r
+ endtime=time(0)+number;\r
+ }\r
+\r
+ while((forever!=0)||\r
+ (endtime==0)||(endtime>time(0))) {\r
+ /* check if we need to give the FTAs there clock signal */\r
+ if ((curprocess.type == LFTA)\r
+ ||(curprocess.type == HFTA)) {\r
+ clock_signal_check();\r
+ }\r
+\r
+#ifdef POLLING\r
+ poll:\r
+#endif\r
+ preemptq=0;\r
+ /* first empty out sidequeu then read from messagequeue */\r
+ if (sidequeue_pop(&from,buf,&length)<0) {\r
+ /* empty out the sidequeue before processing the shared\r
+ memory */\r
+ if (curprocess.type == HFTA) {\r
+ /* process all the shared memory regions and register\r
+ for callbacks */\r
+ s1=rdtsc();\r
+ endq=time(0)+2;\r
+ streamregistry_getactiveringbuf_reset();\r
+ while ((r=streamregistry_getactiveringbuf())>0) {\r
+ while (UNREAD(r)) {\r
+ struct FTA * fa;\r
+ if (ftaexec_start()<0) {\r
+ gslog(LOG_EMERG,"GSCPRTS::error::could not init check of "\r
+ "FTA list\n");\r
+ return -1;\r
+ }\r
+ while ((fa=ftaexec_next())!=0) {\r
+ gs_int32_t x;\r
+ for(x=0;x<fa->stream_subscribed_cnt;x++) {\r
+ if ((fa->stream_subscribed[x].streamid\r
+ ==CURREAD(r)->f.streamid)\r
+ && (fa->stream_subscribed[x].ip\r
+ ==CURREAD(r)->f.ip)\r
+ && (fa->stream_subscribed[x].port\r
+ ==CURREAD(r)->f.port)) {\r
+ fa->accept_packet(fa,&(CURREAD(r)->f),\r
+ &(CURREAD(r)->data[0]),\r
+ CURREAD(r)->sz);\r
+ }\r
+ }\r
+ }\r
+ intuple++;\r
+ inbytes+=CURREAD(r)->sz;\r
+ ADVANCEREAD(r);\r
+ if (endq <= time(0)) {\r
+ preemptq=1;\r
+ goto processmsg;\r
+ }\r
+ }\r
+ }\r
+ s2=rdtsc();\r
+ cycles+=(s2-s1);\r
+#ifndef POLLING\r
+ /* register wakeups all arround to make sure we don't sleep\r
+ * for ever,\r
+ */\r
+ streamregistry_getactiveftaid_reset();\r
+ while ((ftaidp=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 = 0;\r
+ if (gscpipc_send(*ftaidp,FTACALLBACK,(gs_sp_t )&a,a.h.size,1)<0) {\r
+ return -1;\r
+ }\r
+ }\r
+#endif\r
+ }\r
+ processmsg:\r
+ do {\r
+ /* even if we block we return every 100msec to be able to generate the clock signal to\r
+ * the HFTAs\r
+ */\r
+ if ((res=gscpipc_read(&from,&lopp,buf,&length,((block==1)&&(preemptq==0))?2:0))<0) {\r
+ gslog(LOG_EMERG,"GSCPRTS::error::reading from messagequeue\n");\r
+ return -1;\r
+ }\r
+ /* check if we need to give the FTAs there clock signal */\r
+ if ((curprocess.type == LFTA)\r
+ ||(curprocess.type == HFTA)) {\r
+ clock_signal_check();\r
+ }\r
+ if ((res==0) && (block==0)) {\r
+ /* nonblocking and nothing to do so return */\r
+ return 0;\r
+ }\r
+ if ((res==0) && (endtime!=0) && (endtime<time(0))) {\r
+ /* timeout reached so return */\r
+ return 0;\r
+ }\r
+#ifdef POLLING\r
+ if ((res==0)&&(curprocess.type == HFTA)) {\r
+ goto poll;\r
+ }\r
+#endif\r
+ } while (res==0);\r
+ if ((lopp)!=FTACALLBACK) {\r
+ gslog(LOG_EMERG,"GSCPRTS::error::unknown lowlevel opp\n");\r
+ return -1;\r
+ }\r
+ }\r
+\r
+#ifdef PRINTMSG\r
+ gslog(LOG_EMERG, "HFTA message from %u of type %u of length %u\n",from,\r
+ h->callid,h->size);\r
+#endif\r
+ switch (h->callid) {\r
+\r
+ case FTA_LOOKUP: {\r
+ if (curprocess.type == CLEARINGHOUSE) {\r
+ struct fta_find_arg * n;\r
+ FTAID rf;\r
+ gs_csp_t schema;\r
+ n = (struct fta_find_arg *) buf;\r
+ /* Note: Side effect of ftalookup_lookup_fta_index is to\r
+ fill msgid and index */\r
+ if (send_lookup_reply(from,\r
+ ftalookup_lookup_fta_index(from,\r
+ n->name,\r
+ n->reuse,\r
+ &rf,\r
+ &schema),\r
+ &rf,\r
+ (gs_sp_t *)&schema)<0) {\r
+ gslog(LOG_EMERG,"GSCPRTS::error::send standard reply faild\n");\r
+ return -1;\r
+ }\r
+ } else {\r
+ if (send_standard_reply(from,-1)<0) {\r
+ gslog(LOG_EMERG,"GSCPRTS::error::send standard reply faild\n");\r
+ return -1;\r
+ }\r
+ gslog(LOG_EMERG,"GSCPRTS::error: Non clearinghouse proccess got"\r
+ "contacted for clearinghouse processing\n");\r
+ }\r
+ }\r
+ break;\r
+ case FTA_REGISTER: {\r
+ if (curprocess.type == CLEARINGHOUSE) {\r
+ struct fta_register_arg * n;\r
+ n = (struct fta_register_arg *) buf;\r
+ if (send_standard_reply(from,\r
+ ftalookup_register_fta(n->subscriber,\r
+ n->f,\r
+ n->name,\r
+ n->reusable,\r
+ n->schema))<0) {\r
+ gslog(LOG_EMERG,"GSCPRTS::error::send standard reply faild\n");\r
+ return -1;\r
+ }\r
+ } else {\r
+ if (send_standard_reply(from,-1)<0) {\r
+ gslog(LOG_EMERG,"GSCPRTS::error::send standard reply faild\n");\r
+ return -1;\r
+ }\r
+ gslog(LOG_EMERG,"GSCPRTS::error: Non clearinghouse proccess got"\r
+ "contacted for clearinghouse processing\n");\r
+ }\r
+ }\r
+ break;\r
+ case FTA_UNREGISTER: {\r
+ if (curprocess.type == CLEARINGHOUSE) {\r
+ struct fta_unregister_arg * n;\r
+ n = (struct fta_unregister_arg *) buf;\r
+ if (send_standard_reply(from,\r
+ ftalookup_unregister_fta(n->subscriber,\r
+ n->f))<0) {\r
+ gslog(LOG_EMERG,"GSCPRTS::error::send standard reply faild\n");\r
+ return -1;\r
+ }\r
+ } else {\r
+ if (send_standard_reply(from,-1)<0) {\r
+ gslog(LOG_EMERG,"GSCPRTS::error::send standard reply faild\n");\r
+ return -1;\r
+ }\r
+ gslog(LOG_EMERG,"GSCPRTS::error: Non clearinghouse proccess got"\r
+ "contacted for clearinghouse processing\n");\r
+ }\r
+ }\r
+ break;\r
+ case FTA_ALLOC_INSTANCE:\r
+ case FTA_ALLOC_PRINT_INSTANCE:\r
+ if ((curprocess.type == LFTA)\r
+ ||(curprocess.type == HFTA)) {\r
+ struct ringbuf * r;\r
+ struct fta_alloc_instance_arg * n;\r
+ n = (struct fta_alloc_instance_arg *) buf;\r
+ if ((fta=ftaexec_alloc_instance(n->f.index,\r
+ (struct FTA *)n->f.streamid,\r
+ n->reusable,\r
+ n->command,\r
+ n->sz,\r
+ &(n->data[0])))==0) {\r
+ gslog(LOG_EMERG,"GSCPRTS::warning::could not allocate"\r
+ "FTA\n");\r
+ if (send_fta_result(from,0,-1)<0) {\r
+ gslog(LOG_EMERG,"GSCPRTS::error::send standard "\r
+ "reply faild\n");\r
+ return -1;\r
+ }\r
+ } else {\r
+ /* shared memory is only required if data is beeing transfered */\r
+ if ((h->callid!=FTA_ALLOC_PRINT_INSTANCE)&&\r
+ ((r=gscpipc_getshm(from))==0)) {\r
+ gslog(LOG_EMERG,"GSCPRTS::warning::could not get"\r
+ "shared memory\n");\r
+ ftaexec_remove(fta);\r
+ if (send_fta_result(from,0,-1)<0) {\r
+ gslog(LOG_EMERG,"GSCPRTS::error::send standard "\r
+ "reply faild\n");\r
+ return -1;\r
+ }\r
+ } else {\r
+ /* no callback to register for print function */\r
+ if ((h->callid!=FTA_ALLOC_PRINT_INSTANCE)&&(ftacallback_add_streamid(r,fta->ftaid.streamid)!=0)) {\r
+ gslog(LOG_EMERG,"GSCPRTS::warning::could not add"\r
+ "streamid to ringbuffer\n");\r
+ ftaexec_free_instance(fta,1);\r
+ if (send_fta_result(from,0,-1)<0) {\r
+ gslog(LOG_EMERG,"GSCPRTS::error::send standard "\r
+ "reply faild\n");\r
+ return -1;\r
+ }\r
+ } else {\r
+ if (ftaexec_insert(0,fta)<0) {\r
+ gslog(LOG_EMERG,"GSCPRTS::warning::could not"\r
+ "insert FTA\n");\r
+ ftacallback_rm_streamid(r,fta->ftaid.streamid);\r
+ ftaexec_free_instance(fta,1);\r
+ if (send_fta_result(from,0,-1)<0) {\r
+ gslog(LOG_EMERG,"GSCPRTS::error::send standard "\r
+ "reply faild\n");\r
+ return -1;\r
+ }\r
+ } else {\r
+ if (fta_register_instance(n->subscriber,fta->ftaid,\r
+ n->reusable,\r
+ n->name,\r
+ n->schema)!=0) {\r
+ gslog(LOG_EMERG,"GSCPRTS::warning::could not register"\r
+ " instance\n");\r
+ ftaexec_remove(fta);\r
+ ftacallback_rm_streamid(r,fta->ftaid.streamid);\r
+ ftaexec_free_instance(fta,1);\r
+ if (send_fta_result(from,0,-1)<0) {\r
+ gslog(LOG_EMERG,"GSCPRTS::error::send standard "\r
+ "reply faild\n");\r
+ return -1;\r
+ }\r
+ } else {\r
+ if (h->callid==FTA_ALLOC_PRINT_INSTANCE) {\r
+ if (curprocess.type == LFTA) {\r
+ gslog(LOG_EMERG,"GSCPRTS::error:: alloc print instance not "\r
+ "implemented for LFTA.\n");\r
+ ftaexec_remove(fta);\r
+ ftaexec_free_instance(fta,1);\r
+ if (send_fta_result(from,0,-1)<0) {\r
+ gslog(LOG_EMERG,"GSCPRTS::error::send standard "\r
+ "reply faild\n");\r
+ return -1;\r
+ }\r
+ } else {\r
+ if (add_printfunction_to_stream(fta, n->schema, n->path, n->basename,\r
+ n->temporal_field, n->split_field, n->delta, n->split) < 0) {\r
+ ftaexec_remove(fta);\r
+ ftaexec_free_instance(fta,1);\r
+ if (send_fta_result(from,0,-1)<0) {\r
+ gslog(LOG_EMERG,"GSCPRTS::error::send standard "\r
+ "reply faild\n");\r
+ return -1;\r
+ }\r
+ }\r
+ }\r
+ }\r
+ if (send_fta_result(from,&fta->ftaid,0)<0) {\r
+ gslog(LOG_EMERG,"GSCPRTS::error::send standard "\r
+ "reply faild\n");\r
+ return -1;\r
+ }\r
+ }\r
+ }\r
+ }\r
+ }\r
+ }\r
+ } else {\r
+ if (send_fta_result(from,0,-1)<0) {\r
+ gslog(LOG_EMERG,"GSCPRTS::error::send standard "\r
+ "reply faild\n");\r
+ return -1;\r
+ }\r
+ }\r
+ break;\r
+ case FTA_FREE_INSTANCE:{\r
+ if ((curprocess.type == LFTA)\r
+ ||(curprocess.type == HFTA)) {\r
+ struct fta_free_instance_arg * n;\r
+ n = (struct fta_free_instance_arg *) buf;\r
+ if (((r=gscpipc_getshm(from))!=0)\r
+ && ( ftaexec_remove((struct FTA *) n->f.streamid)==0)\r
+ && ( ftacallback_rm_streamid(r,n->f.streamid)==0)\r
+ && (ftaexec_free_instance((struct FTA *)n->f.streamid,\r
+ n->recursive)\r
+ ==0)\r
+ && (fta_unregister_instance(n->subscriber,n->f)<0)) {\r
+ if (send_standard_reply(from,0)<0) {\r
+ gslog(LOG_EMERG,"GSCPRTS::error::send standard reply faild\n");\r
+ return -1;\r
+ }\r
+ } else {\r
+ if (send_standard_reply(from,-1)<0) {\r
+ gslog(LOG_EMERG,"GSCPRTS::error::send standard reply faild\n");\r
+ return -1;\r
+ }\r
+ }\r
+ } else {\r
+ if (send_standard_reply(from,-1)<0) {\r
+ gslog(LOG_EMERG,"GSCPRTS::error::send standard reply faild\n");\r
+ return -1;\r
+ }\r
+ }\r
+ }\r
+ break;\r
+ case FTA_CONTROL:{\r
+ if ((curprocess.type == LFTA)\r
+ ||(curprocess.type == HFTA)) {\r
+ struct fta_control_arg * n;\r
+ n = (struct fta_control_arg *) buf;\r
+ if (send_standard_reply(from,\r
+ ftaexec_control((struct FTA *)\r
+ n->f.streamid,\r
+ n->command,\r
+ n->sz,\r
+ &(n->data[0])))<0) {\r
+ gslog(LOG_EMERG,"GSCPRTS::error::send standard reply faild\n");\r
+ return -1;\r
+ }\r
+ } else {\r
+ if (send_standard_reply(from,-1)<0) {\r
+ gslog(LOG_EMERG,"GSCPRTS::error::send standard reply faild\n");\r
+ return -1;\r
+ }\r
+ gslog(LOG_EMERG,"GSCPRTS::error: Non clearinghouse or HFTA proccess got"\r
+ "contacted for clearinghouse or HFTA processing\n");\r
+ }\r
+ }\r
+ break;\r
+ case FTA_PRODUCER_FAILURE: {\r
+ if (curprocess.type == CLEARINGHOUSE) {\r
+ struct fta_notify_producer_failure_arg * n;\r
+ n = (struct fta_notify_producer_failure_arg *) buf;\r
+ ftalookup_producer_failure(n->sender,n->producer);\r
+ }\r
+ }\r
+ break;\r
+ case FTA_HEARTBEAT: {\r
+ if (curprocess.type == CLEARINGHOUSE) {\r
+ struct fta_heartbeat_arg * n;\r
+ n = (struct fta_heartbeat_arg *) buf;\r
+ ftalookup_heartbeat(n->sender,n->trace_id,\r
+ n->sz,&(n->data[0]));\r
+ }\r
+ }\r
+ break;\r
+ case GSCP_GET_BUFFER:{\r
+ struct sgroup_get_buffer_arg * n;\r
+ gs_int32_t res;\r
+ struct ringbuf * r;\r
+ n = (struct sgroup_get_buffer_arg *) buf;\r
+\r
+ if ((r=gscpipc_getshm(from))==0) {\r
+ gslog(LOG_EMERG,"GSCPRTS::error::proccess blocked without"\r
+ "sharedmemory\n");\r
+ } else {\r
+ if (UNREAD(r)) {\r
+ /* something arrived in the meantime so wakeup\r
+ right away */\r
+ if (send_wakeup(from)<0) {\r
+ gslog(LOG_EMERG,"ERROR:Could not send wakeup\n");\r
+ return -1;\r
+ }\r
+ } else {\r
+#ifndef POLLING\r
+ if (ftacallback_add_wakeup(from,r)<0) {\r
+ gslog(LOG_EMERG,"ERROR:Could not add wakeup\n");\r
+ return -1;\r
+ }\r
+#else\r
+ gslog(LOG_EMERG,"Received wakeup request on polling systems\n");\r
+#endif\r
+ }\r
+ }\r
+ }\r
+ break;\r
+ case PROCESS_CONTROL:{\r
+ if ((curprocess.type == LFTA)\r
+ ||(curprocess.type == HFTA)) {\r
+ struct process_control_arg * n;\r
+ n = (struct process_control_arg *) buf;\r
+ if (send_standard_reply(from,\r
+ ftaexec_process_control(n->command,\r
+ n->sz,\r
+ &(n->data[0])))<0) {\r
+ gslog(LOG_EMERG,"GSCPRTS::error::send standard reply faild\n");\r
+ return -1;\r
+ }\r
+ } else {\r
+ if (send_standard_reply(from,-1)<0) {\r
+ gslog(LOG_EMERG,"GSCPRTS::error::send standard reply faild\n");\r
+ return -1;\r
+ }\r
+ gslog(LOG_EMERG,"GSCPRTS::error: Non clearinghouse or HFTA proccess got"\r
+ "contacted for clearinghouse or HFTA processing\n");\r
+ }\r
+ }\r
+ break;\r
+ case WAKEUP:\r
+ case TIMEOUT:\r
+ break;\r
+ default:\r
+ gslog(LOG_EMERG,"GSCPRTS::error::illegal message queue type %u\n",h->callid);\r
+ return -1;\r
+ }\r
+ /* use this occation to cleanup the messagequeue we can't afford\r
+ a backlog in the real message queue since it is limited in\r
+ size */\r
+ while (gscpipc_read(&from,&lopp,buf,&length,0)>0) {\r
+#ifdef PRINTMSG\r
+ gslog(LOG_EMERG, "request from %u of type %u with length %u\n",from,\r
+ h->callid,h->size);\r
+#endif\r
+ if ((lopp == FTACALLBACK) && (h->callid < RESULT_OPCODE_IGNORE)) {\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
+ }\r
+ }\r
+ return 0;\r
+}\r
+\r
+static gs_retval_t map_match(gs_csp_t dev) {\r
+ gs_int32_t x;\r
+ for(x=0;x<curprocess.mapcnt;x++){\r
+ if (strcmp(dev,curprocess.map[x])==0) {\r
+ return 1;\r
+ }\r
+ }\r
+ return 0;\r
+}\r
+\r
+gs_retval_t fta_max_snaplen()\r
+{\r
+ return maxsnaplen;\r
+}\r
+\r
+FTAID fta_register(FTAname name,gs_uint32_t reusable, DEVname dev,\r
+ alloc_fta fta_alloc_functionptr,\r
+ gs_csp_t schema, gs_int32_t snaplen, gs_uint64_t prefilter)\r
+{\r
+ gs_int8_t rb[MAXRES];\r
+ struct fta_register_arg a;\r
+ struct standard_result * sr = (struct standard_result *)rb;\r
+ gs_int32_t index;\r
+ FTAID res;\r
+ FTAID reserr;\r
+ res.ip=0;\r
+ res.port=0;\r
+ res.index=0;\r
+ res.streamid=0;\r
+ reserr=res;\r
+\r
+ /* check if the device matches for the registration */\r
+ if (((dev==0) && (curprocess.deviceid==0))\r
+ || ((dev!=0)&&(map_match(dev)==1))) {\r
+ if (dev!=0) {\r
+ gslog(LOG_INFO,"Register %s on device %s\n",name,dev);\r
+ } else {\r
+ gslog(LOG_INFO,"Register %s on default device\n",name);\r
+ }\r
+ if ((index=ftacallback_add_alloc(name,fta_alloc_functionptr,prefilter))<0) {\r
+ gslog(LOG_EMERG,"ERROR could not register callback\n");\r
+ return res;\r
+ }\r
+\r
+ if (snaplen<0) {\r
+ maxsnaplen=-1;\r
+ } else {\r
+ if (maxsnaplen!=-1) {\r
+ maxsnaplen=(snaplen>maxsnaplen)?snaplen:maxsnaplen;\r
+ }\r
+ }\r
+\r
+ res=gscpipc_getftaid();\r
+ res.index=index;\r
+ if (curprocess.type != CLEARINGHOUSE) {\r
+ a.h.callid = FTA_REGISTER;\r
+ a.h.size = sizeof(struct fta_register_arg);\r
+ if (strlen(name)>=(MAXFTANAME-1)) {\r
+ gslog(LOG_EMERG,"ERROR:FTA name (%s) to large\n",name);\r
+ return reserr;\r
+ }\r
+ if (strlen(schema)>=(MAXSCHEMASZ-1)) {\r
+ gslog(LOG_EMERG,"ERROR:FTA schema (%s) to large\n",schema);\r
+ return reserr;\r
+ }\r
+ strcpy(a.name,name);\r
+ strcpy(a.schema,schema);\r
+ a.f=res;\r
+ a.subscriber=res; /* consumer is the same as f for an FTA*/\r
+ a.reusable=reusable;\r
+ ipc_call_and_wait(clearinghouseftaid,(gs_sp_t )&a,rb);\r
+ if (sr->h.callid != STANDARD_RESULT) {\r
+ gslog(LOG_EMERG,"ERROR:Wrong result code received\n");\r
+ return reserr;\r
+ }\r
+ if (sr->result != 0 ) {\r
+ gslog(LOG_EMERG,"ERROR:Error in registration\n");\r
+ return reserr;\r
+ }\r
+ } else {\r
+ if (ftalookup_register_fta(gscpipc_getftaid(),\r
+ res,name,reusable,schema)!=0) {\r
+ return res;\r
+ }\r
+ }\r
+ }\r
+ return res;\r
+}\r
+\r
+gs_retval_t fta_unregister(FTAID ftaid)\r
+{\r
+ gs_int8_t rb[MAXRES];\r
+ struct fta_unregister_arg a;\r
+ struct standard_result * sr = (struct standard_result *)rb;\r
+ if (ftacallback_rm_alloc(ftaid.index)<0) {\r
+ gslog(LOG_EMERG,"ERROR could not unregister callback\n");\r
+ return -1;\r
+ }\r
+\r
+ if (curprocess.type != CLEARINGHOUSE) {\r
+ a.h.callid = FTA_UNREGISTER;\r
+ a.h.size = sizeof(struct fta_register_arg);\r
+ a.f=ftaid;\r
+ ipc_call_and_wait(clearinghouseftaid,(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
+ return sr->result;\r
+ } else {\r
+ return ftalookup_unregister_fta(gscpipc_getftaid(),ftaid);\r
+ }\r
+ return 0;\r
+}\r
+\r
+gs_retval_t hfta_post_tuple(struct FTA * self, gs_int32_t sz, void *tuple)\r
+{\r
+ struct ringbuf * r;\r
+ gs_uint32_t msgid;\r
+ struct wakeup_result a;\r
+ FTAID * f;\r
+ gs_int32_t state;\r
+\r
+ if (sz>MAXTUPLESZ) {\r
+ gslog(LOG_EMERG,"Maximum tuple size is %u\n",MAXTUPLESZ);\r
+ return -1;\r
+ }\r
+\r
+ if (self->printfunc.in_use==1) {\r
+ return print_stream(self,sz,tuple);\r
+ }\r
+\r
+ if (ftacallback_start_streamid((gs_p_t )self)<0) {\r
+ gslog(LOG_EMERG,"ERROR:Post for unkown streamid\n");\r
+ return -1;\r
+ }\r
+ /* now make sure we have space to write in all atomic ringbuffer */\r
+ while((r=ftacallback_next_streamid(&state))!=0) {\r
+ if (state == HFTA_RINGBUF_ATOMIC) {\r
+#ifdef BLOCKRINGBUFFER\r
+ while (!SPACETOWRITE(r)) {\r
+ usleep(100);\r
+ }\r
+#endif\r
+ if (! SPACETOWRITE(r)) {\r
+ /* atomic ring buffer and no space so post nothing */\r
+ return -1;\r
+ }\r
+ }\r
+ }\r
+\r
+\r
+ if (ftacallback_start_streamid((gs_p_t )self)<0) {\r
+ gslog(LOG_EMERG,"ERROR:Post for unkown streamid\n");\r
+ return -1;\r
+ }\r
+\r
+ while((r=ftacallback_next_streamid(&state))!=0) {\r
+ if (state != HFTA_RINGBUF_SUSPEND) {\r
+ if (!SPACETOWRITE(r)) {\r
+ //since memory is full we set a warning\r
+ shared_memory_full_warning++;\r
+ // give receiver a chance to clean up\r
+ usleep(0);\r
+ }\r
+ if (SPACETOWRITE(r)) {\r
+ CURWRITE(r)->f=self->ftaid;\r
+ CURWRITE(r)->sz=sz;\r
+ memcpy(&(CURWRITE(r)->data[0]),tuple,sz);\r
+ outtuple++;\r
+ outbytes=outbytes+CURWRITE(r)->sz;\r
+ ADVANCEWRITE(r);\r
+#ifdef PRINTMSG\r
+ gslog(LOG_EMERG,"Wrote in ringpuffer %p [%p:%u]"\r
+ "(%u %u) \n",r,&r->start,r->end,r->reader,r->writer);\r
+ gslog(LOG_EMERG,"\t%u %u\n",CURREAD(r)->next,\r
+ CURREAD(r)->sz);\r
+#endif\r
+ } else {\r
+ outtupledrop++;\r
+ }\r
+ if (HOWFULL(r) > 500) {\r
+ // buffer is at least half full\r
+ shared_memory_full_warning++;\r
+#ifdef PRINTMSG\r
+ gslog(LOG_EMERG,"\t\t buffer full\n");\r
+#endif\r
+ }\r
+ }\r
+ }\r
+#ifndef POLLING\r
+ if (ftacallback_start_wakeup((gs_p_t ) self)<0) {\r
+ gslog(LOG_EMERG,"ERROR:Wakeup for unkown streamid\n");\r
+ return -1;\r
+ }\r
+ a.h.callid=WAKEUP;\r
+ a.h.size=sizeof(struct wakeup_result);\r
+ while((f=ftacallback_next_wakeup())!=0) {\r
+ if (send_wakeup(*f)<0) {\r
+ gslog(LOG_EMERG,"ERROR:Could not send wakeup\n");\r
+ return -1;\r
+ }\r
+ }\r
+#endif\r
+ return 0;\r
+}\r
+\r
+gs_retval_t hfta_get_ringbuf_space(struct FTA * f, FTAID * r, gs_int32_t * space, gs_int32_t szr, gs_int32_t tuplesz)\r
+{\r
+ gs_int32_t x=0;\r
+ gs_int32_t state;\r
+ struct ringbuf * ru;\r
+\r
+ if (f->printfunc.in_use==1) {\r
+ // XXX WHAT TO DO???\r
+ gslog(LOG_INFO,"Checking space for printfunc");\r
+ return 0;\r
+ }\r
+\r
+ if (ftacallback_start_streamid(f->ftaid.streamid)<0) {\r
+ gslog(LOG_EMERG,"ERROR:Space check for unkown streamid in HFTA\n");\r
+ return -1;\r
+ }\r
+\r
+ while ((ru=ftacallback_next_streamid(&state))!=0) {\r
+ if (szr > x ) {\r
+ r[x]=ru->destid;\r
+ space[x]=TUPLEFIT(ru,tuplesz);\r
+ }\r
+ x++;\r
+ }\r
+ return x;\r
+}\r
+\r
+\r
+gs_retval_t hfta_set_ringbuf_type(struct FTA * f, FTAID process, gs_int32_t state)\r
+{\r
+\r
+ if (ftacallback_state_streamid(f->ftaid.streamid,process,state)<0) {\r
+ gslog(LOG_EMERG,"ERROR:state change for unkown streamid\n");\r
+ return -1;\r
+ }\r
+ return 0;\r
+}\r
Code Review / com / gs-lite.git / blobdiff