-/* ------------------------------------------------
- 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 "callbackregistries.h"
-#include "lapp.h"
-#include "gscpipc.h"
-#include "stdlib.h"
-#include "stdio.h"
-#include "string.h"
-#include <unistd.h>
-#include <signal.h>
-#include <sys/mman.h>
-#include "schemaparser.h"
-#include "errno.h"
-
-#include "gsconfig.h"
-#include "gstypes.h"
-
-struct ftacallbackalloc {
- gs_int32_t used;
- gs_sp_t name;
- alloc_fta fta_alloc_functionptr;
- gs_uint64_t prefilter;
-};
-
-static struct ftacallbackalloc * falloc =0;
-static gs_int32_t lalloc=0;
-
-struct ftacallbackstreamid {
- gs_int32_t refcnt;
- gs_uint32_t streamid;
- gs_uint32_t state;
- struct ringbuf * r;
-};
-
-static struct ftacallbackstreamid * fstreamid =0;
-static gs_int32_t lstreamid=0;
-static gs_uint32_t cstreamid;
-static gs_int32_t nstreamid;
-
-struct ftacallbackwakeup {
- gs_int32_t used;
- FTAID ftaid;
- struct ringbuf * r;
-};
-
-static struct ftacallbackwakeup * fwakeup =0;
-static gs_int32_t lwakeup=0;
-static gs_uint32_t swakeup;
-static gs_int32_t nwakeup;
-
-/* XXX memory of fta list has no owner struct */
-
-struct fta_list {
- struct fta_list * prev;
- struct fta_list * next;
- struct FTA * fta;
-};
-
-static struct fta_list * process=0;
-static struct fta_list * created=0;
-
-static struct fta_list * itteration=0;
-
-// Side queue datastructures
-
-struct sq {
- FTAID from;
- gs_int8_t buf[MAXMSGSZ];
- gs_int32_t length;
- struct sq * next;
-};
-
-static struct sq * sqtop=0;
-static struct sq * sqtail=0;
-
-
-/* HFTA internal print function*/
-
-gs_retval_t add_printfunction_to_stream( struct FTA * ftaid, gs_sp_t schema, 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_uint32_t parserversion;
- gs_int32_t schemaid;
- gs_uint32_t x;
- gs_int8_t temp[50000];
- if (ftaid->printfunc.in_use==1) {
- gslog(LOG_EMERG,"ERROR:Printfunction::only allow one print function per HFTA instance\n");
- return -1;
- }
- ftaid->printfunc.path=strdup(path);
- ftaid->printfunc.basename=strdup(basename);
- ftaid->printfunc.nexttime=0;
- ftaid->printfunc.split=(split%1000);
- ftaid->printfunc.itt=(split/1000)%1000;
- ftaid->printfunc.base=(split/1000000)%1000;
- ftaid->printfunc.delta=delta;
- ftaid->printfunc.in_use=1;
- if (ftaid->printfunc.split > MAXPRINTFILES) {
- gslog(LOG_EMERG,"ERROR:Printfunction SPLIT to large\n");
- return -1;
- }
- for (x=0;x<ftaid->printfunc.split;x++) ftaid->printfunc.fa[x]=0;
-
- if ((ftaid->printfunc.schemaid=ftaschema_parse_string(schema))<0) {
- gslog(LOG_EMERG,"ERROR:could not parse schema in HFTA print function");
- return -1;
- }
- if ((ftaid->printfunc.temporal_field=ftaschema_get_field_offset_by_name(
- ftaid->printfunc.schemaid,temporal_field))<0) {
- gslog(LOG_EMERG,"ERROR:could not get "
- "offset for timefield %s in HFTA print function\n",
- temporal_field);
- return -1;
- }
-
- if (ftaschema_get_field_type_by_name(
- ftaid->printfunc.schemaid,temporal_field)!=UINT_TYPE) {
- gslog(LOG_EMERG,"ERROR: illegal type for timefield "
- "%s in HFTA print function UINT expected\n",
- temporal_field);
- return -1;
- }
- if ((ftaid->printfunc.split_field=ftaschema_get_field_offset_by_name(
- ftaid->printfunc.schemaid,split_field))<0) {
- gslog(LOG_EMERG,"ERROR:could not get "
- "offset for splitfield %s in HFTA print function\n",
- split_field);
- return -1;
- }
-
- if (ftaschema_get_field_type_by_name(
- ftaid->printfunc.schemaid,split_field)!=UINT_TYPE) {
- gslog(LOG_EMERG,"ERROR: illegal type for splitfield"
- "%s in HFTA print function UINT expected\n",
- split_field);
- return -1;
- }
- parserversion=get_schemaparser_version();
- sprintf(temp,"GDAT\nVERSION:%u\nSCHEMALENGTH:%lu\n%s",parserversion,strlen(schema)+1,schema);
- ftaid->printfunc.header=strdup(temp);
- gslog(LOG_INFO,"Established print function for %s",basename);
- return 0;
-}
-
-gs_retval_t print_stream(struct FTA * self, gs_int32_t sz, void *tuple)
-{
- gs_int32_t problem;
- gs_uint32_t timeval;
- gs_uint32_t splitval;
- gs_uint32_t x;
- gs_uint32_t nsz;
- timeval=fta_unpack_uint(tuple,sz,self->printfunc.temporal_field,&problem);
- if (timeval==0) return 0; // ignore heartbeats till we see a real timestamp
- if (timeval>= self->printfunc.nexttime) {
- gs_int8_t oldname[1024];
- gs_int8_t newname[1024];
- if (self->printfunc.split==0) {
- if (self->printfunc.fa[0] != 0) {
- sprintf(oldname,"%s/%u%s.tmp",self->printfunc.path,self->printfunc.nexttime-self->printfunc.delta,self->printfunc.basename);
- sprintf(newname,"%s/%u%s",self->printfunc.path,self->printfunc.nexttime-self->printfunc.delta,self->printfunc.basename);
- fclose(self->printfunc.fa[0]);
- rename(oldname,newname);
- }
- if (self->printfunc.nexttime==0) {
- self->printfunc.nexttime=(timeval/self->printfunc.delta)*self->printfunc.delta;
- }
- sprintf(oldname,"%s/%u%s.tmp",self->printfunc.path,self->printfunc.nexttime,self->printfunc.basename);
- if ((self->printfunc.fa[0]=fopen(oldname,"w"))==0) {
- gslog(LOG_EMERG,"ERROR:Could not open file in HFTA print function\n");
- return -1;
- }
- if (setvbuf(self->printfunc.fa[0],0,_IOFBF,16000000)!=0) {
- gslog(LOG_EMERG,"ERROR:Could not setvbuf\n");
- }
- if (fwrite(self->printfunc.header,strlen(self->printfunc.header)+1,1,self->printfunc.fa[0])!=1) {
- gslog(LOG_EMERG,"ERROR:fwrite:xfgh1:%s:%u",self->printfunc.basename,errno);
- }
- gslog(LOG_INFO,"Opened file %s",oldname);
- } else {
- for(x=self->printfunc.base;x<self->printfunc.split;x=x+self->printfunc.itt) {
- if (self->printfunc.fa[x] != 0) {
- sprintf(oldname,"%s/%u_s%u%s.tmp",self->printfunc.path,self->printfunc.nexttime-self->printfunc.delta,x+1,self->printfunc.basename);
- sprintf(newname,"%s/%u_s%u%s",self->printfunc.path,self->printfunc.nexttime-self->printfunc.delta,x+1,self->printfunc.basename);
- fclose(self->printfunc.fa[x]);
- rename(oldname,newname);
- }
- if (self->printfunc.nexttime==0) {
- self->printfunc.nexttime=(timeval/self->printfunc.delta)*self->printfunc.delta;
- }
- sprintf(oldname,"%s/%u_s%u%s.tmp",self->printfunc.path,self->printfunc.nexttime,x+1,self->printfunc.basename);
- if ((self->printfunc.fa[x]=fopen(oldname,"w"))==0) {
- gslog(LOG_EMERG,"ERROR:Could not open file in HFTA print function\n");
- return -1;
- }
- if (setvbuf(self->printfunc.fa[x],0,_IOFBF,16000000)!=0) {
- gslog(LOG_EMERG,"ERROR:Could not setvbuf\n");
- }
- if (fwrite(self->printfunc.header,strlen(self->printfunc.header)+1,1,self->printfunc.fa[x])!=1) {
- gslog(LOG_EMERG,"ERROR:fwrite:xfgh2:%s:%u",self->printfunc.basename,errno);
- }
- gslog(LOG_INFO,"Opened file %s",oldname);
- }
- }
- self->printfunc.nexttime=self->printfunc.nexttime+self->printfunc.delta;
- }
- // don't write temporal tuples to file but use them to advance file name.
- if (ftaschema_is_temporal_tuple(self->printfunc.schemaid, tuple)) return 0;
- if (self->printfunc.split!=0) {
- splitval=fta_unpack_uint(tuple,sz,self->printfunc.split_field,&problem)%(self->printfunc.split);
- if (self->printfunc.fa[splitval]==0) {
- gslog(LOG_EMERG,"Inconsistent rangehash in print %u\n", splitval);
- exit(0);
- }
- } else {
- splitval=0;
- }
- nsz=htonl(sz);
- if (fwrite(&nsz,sizeof(gs_uint32_t),1,self->printfunc.fa[splitval])!=1) {
- gslog(LOG_EMERG,"Could not write to output in HFTA print\"%s\":%u.. EXITING\n",
- self->printfunc.basename,errno);
- exit(0);
- }
- if (fwrite(tuple,sz,1,self->printfunc.fa[splitval])!=1) {
- gslog(LOG_EMERG,"Could not write to output in HFTA print\"%s\"%u.. EXITING\n",
- self->printfunc.basename,errno);
- exit(0);
- }
- return 0;
-}
-
-/* registers an alloc function of an FTA and returns a unique index */
-gs_retval_t ftacallback_add_alloc(FTAname name, alloc_fta fta_alloc_functionptr,gs_uint64_t prefilter)
-{
- gs_int32_t x;
- gslog(LOG_INFO,"Register prefilter %llu for %s\n",prefilter,name);
- if (lalloc == 0) {
- if ((falloc = malloc(sizeof(struct ftacallbackalloc)*STARTSZ))==0) {
- gslog(LOG_EMERG,"ERROR:Out of memory ftacallback\n");
- return -1;
- }
- memset(falloc,0,sizeof(struct ftacallbackalloc)*STARTSZ);
- lalloc = STARTSZ;
- }
- for(x=0;(x<lalloc)&&(falloc[x].used!=0);x++);
- if (x == lalloc) {
- gs_int32_t y;
- lalloc = 2*lalloc;
- if ((falloc = realloc(falloc,lalloc*sizeof(struct ftacallbackalloc)))==0) {
- gslog(LOG_EMERG,"ERROR:Out of memory ftacallback\n");
- return -1;
- }
- for (y=x;y<lalloc;y++)
- falloc[y].used=0;
- }
- falloc[x].name=strdup(name);
- falloc[x].fta_alloc_functionptr=fta_alloc_functionptr;
- falloc[x].prefilter=prefilter;
- falloc[x].used=1;
- return x;
-}
-
-/* unregisters an alloc function of an FTA and makes the index available
- * for reuse
- */
-
-gs_retval_t ftacallback_rm_alloc(gs_uint32_t index)
-{
- falloc[index].used=0;
- free(falloc[index].name);
- return 0;
-}
-/* returns the prefilter for a given
- * index
- */
-
-gs_uint64_t ftacallback_get_prefilter(gs_int32_t index)
-{
- if ((index<lalloc) && (falloc[index].used!=0)) {
- return falloc[index].prefilter;
- }
- return 0;
-}
-
-/* returns the function pointer of the callback function for a given
- * index
- */
-
-alloc_fta ftacallback_get_alloc(gs_int32_t index)
-{
- if ((index<lalloc) && (falloc[index].used!=0)) {
- return falloc[index].fta_alloc_functionptr;
- }
- return 0;
-}
-
-
-
-
-
-/* associate ringbuffer with streamid (using refcounting) */
-gs_retval_t ftacallback_add_streamid(struct ringbuf * r, gs_uint32_t streamid) {
- gs_int32_t x;
- if (lstreamid == 0) {
- if ((fstreamid = malloc(sizeof(struct ftacallbackstreamid)*STARTSZ))==0) {
- gslog(LOG_EMERG,"ERROR:Out of memory ftacallback\n");
- return -1;
- }
- memset(fstreamid,0,sizeof(struct ftacallbackstreamid)*STARTSZ);
- lstreamid = STARTSZ;
- }
- /* first try to increment refcnt */
- for(x=0;(x<lstreamid)&&(
- (fstreamid[x].streamid!=streamid)
- ||(fstreamid[x].r!=r)
- ||(fstreamid[x].refcnt<=0)) ;x++);
- if (x>=lstreamid) {
- /* now try to find empty slot */
- for(x=0;(x<lstreamid)&&(fstreamid[x].refcnt!=0);x++);
- if (x >= lstreamid) {
- gs_int32_t y;
- lstreamid = 2*lstreamid;
- if ((fstreamid =
- realloc(fstreamid,sizeof(struct ftacallbackstreamid)*lstreamid))==0) {
- gslog(LOG_EMERG,"ERROR:Out of memory ftacallback\n");
- return -1;
- }
- for (y=x;y<lstreamid;y++) {
- fstreamid[y].refcnt=0;
- fstreamid[y].streamid=0;
- }
- }
- fstreamid[x].state=HFTA_RINGBUF_ATOMIC;
- fstreamid[x].streamid=streamid;
- fstreamid[x].r=r;
- }
- fstreamid[x].refcnt+=1;
- return 0;
-}
-
-
-/* unassosciate a ringbuffer from a streamid */
-
-gs_retval_t ftacallback_rm_streamid(struct ringbuf * r, gs_uint32_t streamid)
-{
- gs_int32_t x;
- for(x=0;x<lstreamid;x++) {
- if ((fstreamid[x].streamid == streamid)
- && (fstreamid[x].r == r)
- && (fstreamid[x].refcnt > 0))
- fstreamid[x].refcnt--;
- }
- return 0;
-}
-
-/* set the state for a given streamid and destination process */
-gs_retval_t ftacallback_state_streamid(gs_int32_t streamid,FTAID process, gs_int32_t state)
-{
- gs_int32_t x;
- for(x=0;x<lstreamid;x++) {
- if ((fstreamid[x].streamid == streamid)
- && (fstreamid[x].r->destid.ip == process.ip )
- && (fstreamid[x].r->destid.port == process.port )
- && (fstreamid[x].refcnt > 0))
- fstreamid[x].state=state;
- return 0;
- }
- return -1;
-}
-
-/* starts an itteration through all ringbuffers for a particular streamid */
-
-gs_retval_t ftacallback_start_streamid(gs_int32_t streamid)
-{
- cstreamid=streamid;
- nstreamid=0;
- return 0;
-}
-
-/* returns all the ringbuffer associated with the streamid passed in
- * ftacallback_start_streamid
- */
-struct ringbuf * ftacallback_next_streamid(gs_int32_t* state)
-{
- for(;(nstreamid<lstreamid)
- &&(fstreamid[nstreamid].streamid != cstreamid);
- nstreamid++);
- if (nstreamid<lstreamid) {
- nstreamid++;
- *state=fstreamid[nstreamid-1].state;
- return fstreamid[nstreamid-1].r;
- }
- return 0;
-}
-
-
-/* associate msgid with ringbuf */
-gs_retval_t ftacallback_add_wakeup(FTAID ftaid, struct ringbuf * r)
-{
- gs_int32_t x;
- if (lwakeup == 0) {
- if ((fwakeup = malloc(sizeof(struct ftacallbackwakeup)*STARTSZ))==0) {
- gslog(LOG_EMERG,"ERROR:Out of memory ftacallback\n");
- return -1;
- }
- memset(fwakeup,0,sizeof(struct ftacallbackwakeup)*STARTSZ);
- lwakeup = STARTSZ;
- }
- /* first try to find one for the same process */
- for(x=0;(x<lwakeup)&&(
- ((fwakeup[x].ftaid.ip!=ftaid.ip)
- || (fwakeup[x].ftaid.port!=ftaid.port))
- ||(fwakeup[x].used==0)) ;x++);
- if (x==lwakeup) {
- /* now try to find empty slot */
- for(x=0;(x<lwakeup)&&(fwakeup[x].used!=0);x++);
- if (x == lwakeup) {
- gs_int32_t y;
- lwakeup = 2*lwakeup;
- if ((fwakeup =
- realloc(fwakeup,sizeof(struct ftacallbackwakeup)*lwakeup))==0) {
- gslog(LOG_EMERG,"ERROR:Out of memory ftacallback\n");
- return -1;
- }
- for (y=x;y<lwakeup;y++)
- fwakeup[y].used=0;
- }
- }
- fwakeup[x].used=1;
- fwakeup[x].ftaid=ftaid;
- fwakeup[x].r=r;
- return x;
-}
-
-/* starts an itteration through all msgids associated with
- a streamid. This also uses data kept in the fstreamid
- registry
- */
-gs_retval_t ftacallback_start_wakeup(gs_uint32_t streamid)
-{
- swakeup=streamid;
- nwakeup=0;
- return 0;
-}
-
-/* returns all the msgid blocked on the streamid passed in
- * ftacallback_start_streamid and removes the msgid from
- * the wakeup list
- */
-FTAID * ftacallback_next_wakeup()
-{
- for(;(nwakeup<lstreamid)
- &&(fstreamid[nwakeup].streamid != swakeup);
- nwakeup++);
- if (nwakeup<lstreamid) {
- gs_int32_t x;
- for(x=0;x<lwakeup;x++)
- if ((fwakeup[x].r==fstreamid[nwakeup].r) &&
- (fwakeup[x].used==1)) {
- fwakeup[x].used=0;
- nwakeup++;
- return & fwakeup[x].ftaid;
- }
- }
- nwakeup ++;
- return (FTAID *) 0;
-}
-
-
-
-static gs_retval_t fta_list_add(struct fta_list ** root, struct FTA * after,
- struct FTA * fta) {
- struct fta_list * new;
- struct fta_list * tmp;
-
- if ((new=(struct fta_list *)malloc(sizeof(struct fta_list)))==0) {
- gslog(LOG_EMERG,"fta_list_add:: can't allocate memory\n");
- return -1;
- }
-
- new->fta=fta;
-
-
- if (after==0) {
- new->next=*root;
- new->prev=0;
- *root=new;
- if (new->next) {
- new->next->prev=new;
- }
- } else {
- tmp=*root;
- while((tmp)&&(tmp->fta!=after)) {
- tmp=tmp->next;
- }
- if (tmp==0) {
- gslog(LOG_EMERG,"fta_list_add:: can't find after fta\n");
- return -1;
- }
- new->next=tmp->next;
- new->prev=tmp;
- tmp->next=new;
- if (new->next) {
- new->next->prev=new;
- }
- }
- return 0;
-}
-
-static gs_retval_t fta_list_rm(struct fta_list ** root, struct FTA * fta) {
- struct fta_list * tmp;
- tmp=*root;
- while((tmp)&&(tmp->fta!=fta)) {
- tmp=tmp->next;
- }
- if (tmp==0) {
- gslog(LOG_EMERG,"fta_list_rm:: can't find fta\n");
- return -1;
- }
- if (tmp == (*root)) {
- *root=tmp->next;
- if (tmp->next) {
- tmp->next->prev=0;
- }
- } else {
- tmp->prev->next=tmp->next;
- if (tmp->next) {
- tmp->next->prev=tmp->prev;
- }
- }
-
- free(tmp);
- return 0;
-}
-
-
-static gs_retval_t fta_list_check(struct fta_list ** root, struct FTA * fta) {
- struct fta_list * tmp;
- tmp=*root;
- while((tmp)&&(tmp->fta!=fta)) {
- tmp=tmp->next;
- }
- if (tmp==0) {
- return -1;
- }
- return 0;
-}
-
-gs_retval_t ftaexec_insert(struct FTA * after, struct FTA * new)
-{
-
- if ((after!=0) && (fta_list_check(&process,after)<0)) {
- gslog(LOG_EMERG,"fta_insert:: ilegal adapter for after\n");
- return -1;
- }
-
- if (fta_list_check(&created,new)<0) {
- gslog(LOG_EMERG,"fta_insert:: ilegal adapter for new\n");
- return -1;
- }
-
- if (fta_list_check(&process,new)==0) {
- new->runrefcnt++;
- gslog(LOG_INFO,"fta_insert:: new already in process list reusing entry with streamid %d\n",new->ftaid.streamid);
- return 0;
- }
-
- if (fta_list_add(&process,after,new)<0) {
- gslog(LOG_EMERG,"fta_insert:: new can not be added to process list\n");
- return -1;
- }
-
- new->runrefcnt++;
-
- return 0;
-
-}
-
-gs_retval_t ftaexec_remove(struct FTA * id){
-
- if (fta_list_check(&process,id)<0) {
- gslog(LOG_EMERG,"fta_remove:: id not in process list\n");
- return -1;
- }
- id->runrefcnt--;
- if (id->runrefcnt<=0) {
- if (fta_list_rm(&process,id)<0) {
- gslog(LOG_EMERG,"fta_remove:: id could not be removed from process list\n");
- return -1;
- }
- }
- return 0;
-}
-
-
-struct FTA * ftaexec_alloc_instance(gs_uint32_t index, struct FTA * reuse,
- gs_uint32_t reusable,
- gs_int32_t command, gs_int32_t sz, void * data){
- struct FTA * f;
- FTAID ftaid;
- ftaid=gscpipc_getftaid();
- ftaid.index=index;
- ftaid.streamid=0;
- if (fta_list_check(&created,reuse)==0) {
- reuse->refcnt++;
- return reuse;
- }
- if (ftacallback_get_alloc(index)==0) return 0;
- f=ftacallback_get_alloc(index)(ftaid,reusable,command, sz, data);
- f->prefilter=ftacallback_get_prefilter(index);
- gslog(LOG_INFO,"Using prefilter %llu for fta %x\n",f->prefilter,f);
- if (fta_list_add(&created,0,f)<0) {
- gslog(LOG_EMERG,"fta_alloc_instance:: new fta can not be added to created list\n");
- return 0;
- }
- f->refcnt=1;
- return f;
-}
-
-
-
-gs_retval_t ftaexec_free_instance(struct FTA * id, gs_uint32_t recursive){
- id->refcnt --;
- if (id->refcnt==0) {
- if (fta_list_rm(&created,id)<0) {
- gslog(LOG_EMERG,"fta_free_instance:: fta could not be removed from created list\n");
- return -1;
- }
- /* just to make sure remove it form process list too */
- if (fta_list_check(&process,id)>=0) {
- fta_list_rm(&process,id);
- }
-
- id->free_fta(id,recursive);
- }
- return 0;
-}
-
-gs_retval_t ftaexec_control(struct FTA * id, gs_int32_t command, gs_int32_t sz, void * value){
- if (fta_list_check(&created,id)<0) {
- gslog(LOG_EMERG,"fta_control:: id not found in adapter's created list\n");
- return -1;
- }
- return id->control_fta(id,command,sz,value);
-}
-
-gs_retval_t ftaexec_process_control(gs_int32_t command, gs_int32_t sz, void * value){
- struct FTA * f;
- ftaexec_start();
- while((f=ftaexec_next())!=0) {
- f->control_fta(f,command,sz,value);
- }
- return 1;
-}
-
-
-/* Start itteration through list of active FTA */
-
-gs_retval_t ftaexec_start()
-{
- itteration=process;
- return 0;
-}
-
-/* get one FTA at a time */
-
-struct FTA * ftaexec_next()
-{
- struct FTA * fta=0;
- if (itteration) {
- fta=itteration->fta;
- itteration=itteration->next;
- }
- return fta;
-}
-
-
-/* adds a buffer to the end of the sidequeue*/
-gs_retval_t sidequeue_append(FTAID from, gs_sp_t buf, gs_int32_t length)
-{
- struct sq * s;
- if ((s=malloc(sizeof(struct sq)))==0) {
- gslog(LOG_EMERG,"Could not allocate memory for sidequeue");
- return -1;
- }
- s->from=from;
- memcpy(&s->buf[0],buf,MAXMSGSZ);
- s->length=length;
- s->next=0;
- if (sqtail) {
- sqtail->next=s;
- sqtail=s;
- } else {
- sqtop = s;
- sqtail = s;
- }
- return 0;
-}
-
-/* removes a buffer from the top of the sidequeue*/
-gs_retval_t sidequeue_pop(FTAID * from, gs_sp_t buf, gs_int32_t* length)
-{
- struct sq * s;
-
- if (sqtop) {
- *from=sqtop->from;
- memcpy(buf,&sqtop->buf[0],MAXMSGSZ);
- *length=sqtop->length;
- s=sqtop;
- sqtop=sqtop->next;
- if (sqtop==0) sqtail=0;
- free(s);
- return 0;
- }
- 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
+\r
+#include "callbackregistries.h"\r
+#include "lapp.h"\r
+#include "gscpipc.h"\r
+#include "stdlib.h"\r
+#include "stdio.h"\r
+#include "string.h"\r
+#include <unistd.h>\r
+#include <signal.h>\r
+#include <sys/mman.h>\r
+#include "schemaparser.h"\r
+#include "errno.h"\r
+\r
+#include "gsconfig.h"\r
+#include "gstypes.h"\r
+\r
+struct ftacallbackalloc {\r
+ gs_int32_t used;\r
+ gs_sp_t name;\r
+ alloc_fta fta_alloc_functionptr;\r
+ gs_uint64_t prefilter;\r
+};\r
+\r
+static struct ftacallbackalloc * falloc =0;\r
+static gs_int32_t lalloc=0;\r
+\r
+struct ftacallbackstreamid {\r
+ gs_int32_t refcnt;\r
+ gs_uint32_t streamid;\r
+ gs_uint32_t state;\r
+ struct ringbuf * r;\r
+};\r
+\r
+static struct ftacallbackstreamid * fstreamid =0;\r
+static gs_int32_t lstreamid=0;\r
+static gs_uint32_t cstreamid;\r
+static gs_int32_t nstreamid;\r
+\r
+struct ftacallbackwakeup {\r
+ gs_int32_t used;\r
+ FTAID ftaid;\r
+ struct ringbuf * r;\r
+};\r
+\r
+static struct ftacallbackwakeup * fwakeup =0;\r
+static gs_int32_t lwakeup=0;\r
+static gs_uint32_t swakeup;\r
+static gs_int32_t nwakeup;\r
+\r
+/* XXX memory of fta list has no owner struct */\r
+\r
+struct fta_list {\r
+ struct fta_list * prev;\r
+ struct fta_list * next;\r
+ struct FTA * fta;\r
+};\r
+\r
+static struct fta_list * process=0;\r
+static struct fta_list * created=0;\r
+\r
+static struct fta_list * itteration=0;\r
+\r
+// Side queue datastructures\r
+\r
+struct sq {\r
+ FTAID from;\r
+ gs_int8_t buf[MAXMSGSZ];\r
+ gs_int32_t length;\r
+ struct sq * next;\r
+};\r
+\r
+static struct sq * sqtop=0;\r
+static struct sq * sqtail=0;\r
+\r
+\r
+/* HFTA internal print function*/\r
+\r
+gs_retval_t add_printfunction_to_stream( struct FTA * ftaid, gs_sp_t schema, gs_sp_t path, gs_sp_t basename,\r
+ gs_sp_t temporal_field, gs_sp_t split_field, gs_uint32_t delta, gs_uint32_t split) {\r
+ gs_uint32_t parserversion;\r
+ gs_int32_t schemaid;\r
+ gs_uint32_t x;\r
+ gs_int8_t temp[50000];\r
+ if (ftaid->printfunc.in_use==1) {\r
+ gslog(LOG_EMERG,"ERROR:Printfunction::only allow one print function per HFTA instance\n");\r
+ return -1;\r
+ }\r
+ ftaid->printfunc.path=strdup(path);\r
+ ftaid->printfunc.basename=strdup(basename);\r
+ ftaid->printfunc.nexttime=0;\r
+ ftaid->printfunc.split=(split%1000);\r
+ ftaid->printfunc.itt=(split/1000)%1000;\r
+ ftaid->printfunc.base=(split/1000000)%1000;\r
+ ftaid->printfunc.delta=delta;\r
+ ftaid->printfunc.in_use=1;\r
+ if (ftaid->printfunc.split > MAXPRINTFILES) {\r
+ gslog(LOG_EMERG,"ERROR:Printfunction SPLIT to large\n");\r
+ return -1;\r
+ }\r
+ for (x=0;x<ftaid->printfunc.split;x++) ftaid->printfunc.fa[x]=0;\r
+ \r
+ if ((ftaid->printfunc.schemaid=ftaschema_parse_string(schema))<0) {\r
+ gslog(LOG_EMERG,"ERROR:could not parse schema in HFTA print function");\r
+ return -1;\r
+ }\r
+ if ((ftaid->printfunc.temporal_field=ftaschema_get_field_offset_by_name(\r
+ ftaid->printfunc.schemaid,temporal_field))<0) {\r
+ gslog(LOG_EMERG,"ERROR:could not get "\r
+ "offset for timefield %s in HFTA print function\n",\r
+ temporal_field);\r
+ return -1;\r
+ }\r
+ \r
+ if (ftaschema_get_field_type_by_name(\r
+ ftaid->printfunc.schemaid,temporal_field)!=UINT_TYPE) {\r
+ gslog(LOG_EMERG,"ERROR: illegal type for timefield "\r
+ "%s in HFTA print function UINT expected\n",\r
+ temporal_field);\r
+ return -1;\r
+ }\r
+ if ((ftaid->printfunc.split_field=ftaschema_get_field_offset_by_name(\r
+ ftaid->printfunc.schemaid,split_field))<0) {\r
+ gslog(LOG_EMERG,"ERROR:could not get "\r
+ "offset for splitfield %s in HFTA print function\n",\r
+ split_field);\r
+ return -1;\r
+ }\r
+ \r
+ if (ftaschema_get_field_type_by_name(\r
+ ftaid->printfunc.schemaid,split_field)!=UINT_TYPE) {\r
+ gslog(LOG_EMERG,"ERROR: illegal type for splitfield"\r
+ "%s in HFTA print function UINT expected\n",\r
+ split_field);\r
+ return -1;\r
+ }\r
+ parserversion=get_schemaparser_version();\r
+ sprintf(temp,"GDAT\nVERSION:%u\nSCHEMALENGTH:%lu\n%s",parserversion,strlen(schema)+1,schema);\r
+ ftaid->printfunc.header=strdup(temp);\r
+ gslog(LOG_INFO,"Established print function for %s",basename);\r
+ return 0;\r
+}\r
+\r
+gs_retval_t print_stream(struct FTA * self, gs_int32_t sz, void *tuple)\r
+{\r
+ gs_int32_t problem;\r
+ gs_uint32_t timeval;\r
+ gs_uint32_t splitval;\r
+ gs_uint32_t x;\r
+ gs_uint32_t nsz;\r
+ timeval=fta_unpack_uint(tuple,sz,self->printfunc.temporal_field,&problem);\r
+ if (timeval==0) return 0; // ignore heartbeats till we see a real timestamp\r
+ if (timeval>= self->printfunc.nexttime) {\r
+ gs_int8_t oldname[1024];\r
+ gs_int8_t newname[1024];\r
+ if (self->printfunc.split==0) {\r
+ if (self->printfunc.fa[0] != 0) {\r
+ sprintf(oldname,"%s/%u%s.tmp",self->printfunc.path,self->printfunc.nexttime-self->printfunc.delta,self->printfunc.basename);\r
+ sprintf(newname,"%s/%u%s",self->printfunc.path,self->printfunc.nexttime-self->printfunc.delta,self->printfunc.basename);\r
+ fclose(self->printfunc.fa[0]);\r
+ rename(oldname,newname);\r
+ }\r
+ if (self->printfunc.nexttime==0) {\r
+ self->printfunc.nexttime=(timeval/self->printfunc.delta)*self->printfunc.delta;\r
+ }\r
+ sprintf(oldname,"%s/%u%s.tmp",self->printfunc.path,self->printfunc.nexttime,self->printfunc.basename);\r
+ if ((self->printfunc.fa[0]=fopen(oldname,"w"))==0) {\r
+ gslog(LOG_EMERG,"ERROR:Could not open file in HFTA print function\n");\r
+ return -1;\r
+ }\r
+ if (setvbuf(self->printfunc.fa[0],0,_IOFBF,16000000)!=0) {\r
+ gslog(LOG_EMERG,"ERROR:Could not setvbuf\n");\r
+ }\r
+ if (fwrite(self->printfunc.header,strlen(self->printfunc.header)+1,1,self->printfunc.fa[0])!=1) {\r
+ gslog(LOG_EMERG,"ERROR:fwrite:xfgh1:%s:%u",self->printfunc.basename,errno);\r
+ }\r
+ gslog(LOG_INFO,"Opened file %s",oldname);\r
+ } else {\r
+ for(x=self->printfunc.base;x<self->printfunc.split;x=x+self->printfunc.itt) {\r
+ if (self->printfunc.fa[x] != 0) {\r
+ sprintf(oldname,"%s/%u_s%u%s.tmp",self->printfunc.path,self->printfunc.nexttime-self->printfunc.delta,x+1,self->printfunc.basename);\r
+ sprintf(newname,"%s/%u_s%u%s",self->printfunc.path,self->printfunc.nexttime-self->printfunc.delta,x+1,self->printfunc.basename);\r
+ fclose(self->printfunc.fa[x]);\r
+ rename(oldname,newname);\r
+ }\r
+ if (self->printfunc.nexttime==0) {\r
+ self->printfunc.nexttime=(timeval/self->printfunc.delta)*self->printfunc.delta;\r
+ }\r
+ sprintf(oldname,"%s/%u_s%u%s.tmp",self->printfunc.path,self->printfunc.nexttime,x+1,self->printfunc.basename);\r
+ if ((self->printfunc.fa[x]=fopen(oldname,"w"))==0) {\r
+ gslog(LOG_EMERG,"ERROR:Could not open file in HFTA print function\n");\r
+ return -1;\r
+ }\r
+ if (setvbuf(self->printfunc.fa[x],0,_IOFBF,16000000)!=0) {\r
+ gslog(LOG_EMERG,"ERROR:Could not setvbuf\n");\r
+ }\r
+ if (fwrite(self->printfunc.header,strlen(self->printfunc.header)+1,1,self->printfunc.fa[x])!=1) {\r
+ gslog(LOG_EMERG,"ERROR:fwrite:xfgh2:%s:%u",self->printfunc.basename,errno);\r
+ }\r
+ gslog(LOG_INFO,"Opened file %s",oldname);\r
+ }\r
+ }\r
+ self->printfunc.nexttime=self->printfunc.nexttime+self->printfunc.delta;\r
+ }\r
+ // don't write temporal tuples to file but use them to advance file name.\r
+ if (ftaschema_is_temporal_tuple(self->printfunc.schemaid, tuple)) return 0;\r
+ if (self->printfunc.split!=0) {\r
+ splitval=fta_unpack_uint(tuple,sz,self->printfunc.split_field,&problem)%(self->printfunc.split);\r
+ if (self->printfunc.fa[splitval]==0) {\r
+ gslog(LOG_EMERG,"Inconsistent rangehash in print %u\n", splitval);\r
+ exit(0);\r
+ }\r
+ } else {\r
+ splitval=0;\r
+ }\r
+ nsz=htonl(sz);\r
+ if (fwrite(&nsz,sizeof(gs_uint32_t),1,self->printfunc.fa[splitval])!=1) {\r
+ gslog(LOG_EMERG,"Could not write to output in HFTA print\"%s\":%u.. EXITING\n",\r
+ self->printfunc.basename,errno);\r
+ exit(0);\r
+ }\r
+ if (fwrite(tuple,sz,1,self->printfunc.fa[splitval])!=1) {\r
+ gslog(LOG_EMERG,"Could not write to output in HFTA print\"%s\"%u.. EXITING\n",\r
+ self->printfunc.basename,errno);\r
+ exit(0);\r
+ }\r
+ return 0;\r
+}\r
+\r
+/* registers an alloc function of an FTA and returns a unique index */\r
+gs_retval_t ftacallback_add_alloc(FTAname name, alloc_fta fta_alloc_functionptr,gs_uint64_t prefilter)\r
+{\r
+ gs_int32_t x;\r
+ gslog(LOG_INFO,"Register prefilter %llu for %s\n",prefilter,name);\r
+ if (lalloc == 0) {\r
+ if ((falloc = malloc(sizeof(struct ftacallbackalloc)*STARTSZ))==0) {\r
+ gslog(LOG_EMERG,"ERROR:Out of memory ftacallback\n");\r
+ return -1;\r
+ }\r
+ memset(falloc,0,sizeof(struct ftacallbackalloc)*STARTSZ);\r
+ lalloc = STARTSZ;\r
+ }\r
+ for(x=0;(x<lalloc)&&(falloc[x].used!=0);x++);\r
+ if (x == lalloc) {\r
+ gs_int32_t y;\r
+ lalloc = 2*lalloc;\r
+ if ((falloc = realloc(falloc,lalloc*sizeof(struct ftacallbackalloc)))==0) {\r
+ gslog(LOG_EMERG,"ERROR:Out of memory ftacallback\n");\r
+ return -1;\r
+ }\r
+ for (y=x;y<lalloc;y++)\r
+ falloc[y].used=0;\r
+ }\r
+ falloc[x].name=strdup(name);\r
+ falloc[x].fta_alloc_functionptr=fta_alloc_functionptr;\r
+ falloc[x].prefilter=prefilter;\r
+ falloc[x].used=1;\r
+ return x;\r
+}\r
+\r
+/* unregisters an alloc function of an FTA and makes the index available\r
+ * for reuse\r
+ */\r
+\r
+gs_retval_t ftacallback_rm_alloc(gs_uint32_t index)\r
+{\r
+ falloc[index].used=0;\r
+ free(falloc[index].name);\r
+ return 0;\r
+}\r
+/* returns the prefilter for a given\r
+ * index\r
+ */\r
+\r
+gs_uint64_t ftacallback_get_prefilter(gs_int32_t index)\r
+{\r
+ if ((index<lalloc) && (falloc[index].used!=0)) {\r
+ return falloc[index].prefilter;\r
+ }\r
+ return 0;\r
+}\r
+\r
+/* returns the function pointer of the callback function for a given\r
+ * index\r
+ */\r
+\r
+alloc_fta ftacallback_get_alloc(gs_int32_t index)\r
+{\r
+ if ((index<lalloc) && (falloc[index].used!=0)) {\r
+ return falloc[index].fta_alloc_functionptr;\r
+ }\r
+ return 0;\r
+}\r
+\r
+\r
+\r
+\r
+\r
+/* associate ringbuffer with streamid (using refcounting) */\r
+gs_retval_t ftacallback_add_streamid(struct ringbuf * r, gs_uint32_t streamid) {\r
+ gs_int32_t x;\r
+ if (lstreamid == 0) {\r
+ if ((fstreamid = malloc(sizeof(struct ftacallbackstreamid)*STARTSZ))==0) {\r
+ gslog(LOG_EMERG,"ERROR:Out of memory ftacallback\n");\r
+ return -1;\r
+ }\r
+ memset(fstreamid,0,sizeof(struct ftacallbackstreamid)*STARTSZ);\r
+ lstreamid = STARTSZ;\r
+ }\r
+ /* first try to increment refcnt */\r
+ for(x=0;(x<lstreamid)&&(\r
+ (fstreamid[x].streamid!=streamid)\r
+ ||(fstreamid[x].r!=r)\r
+ ||(fstreamid[x].refcnt<=0)) ;x++);\r
+ if (x>=lstreamid) {\r
+ /* now try to find empty slot */\r
+ for(x=0;(x<lstreamid)&&(fstreamid[x].refcnt!=0);x++);\r
+ if (x >= lstreamid) {\r
+ gs_int32_t y;\r
+ lstreamid = 2*lstreamid;\r
+ if ((fstreamid =\r
+ realloc(fstreamid,sizeof(struct ftacallbackstreamid)*lstreamid))==0) {\r
+ gslog(LOG_EMERG,"ERROR:Out of memory ftacallback\n");\r
+ return -1;\r
+ }\r
+ for (y=x;y<lstreamid;y++) {\r
+ fstreamid[y].refcnt=0;\r
+ fstreamid[y].streamid=0;\r
+ }\r
+ }\r
+ fstreamid[x].state=HFTA_RINGBUF_ATOMIC;\r
+ fstreamid[x].streamid=streamid;\r
+ fstreamid[x].r=r;\r
+ }\r
+ fstreamid[x].refcnt+=1;\r
+ return 0;\r
+}\r
+\r
+\r
+/* unassosciate a ringbuffer from a streamid */\r
+\r
+gs_retval_t ftacallback_rm_streamid(struct ringbuf * r, gs_uint32_t streamid)\r
+{\r
+ gs_int32_t x;\r
+ for(x=0;x<lstreamid;x++) {\r
+ if ((fstreamid[x].streamid == streamid)\r
+ && (fstreamid[x].r == r)\r
+ && (fstreamid[x].refcnt > 0))\r
+ fstreamid[x].refcnt--;\r
+ }\r
+ return 0;\r
+}\r
+\r
+/* set the state for a given streamid and destination process */\r
+gs_retval_t ftacallback_state_streamid(gs_int32_t streamid,FTAID process, gs_int32_t state)\r
+{\r
+ gs_int32_t x;\r
+ for(x=0;x<lstreamid;x++) {\r
+ if ((fstreamid[x].streamid == streamid)\r
+ && (fstreamid[x].r->destid.ip == process.ip )\r
+ && (fstreamid[x].r->destid.port == process.port )\r
+ && (fstreamid[x].refcnt > 0))\r
+ fstreamid[x].state=state;\r
+ return 0;\r
+ }\r
+ return -1;\r
+}\r
+\r
+/* starts an itteration through all ringbuffers for a particular streamid */\r
+\r
+gs_retval_t ftacallback_start_streamid(gs_int32_t streamid)\r
+{\r
+ cstreamid=streamid;\r
+ nstreamid=0;\r
+ return 0;\r
+}\r
+\r
+/* returns all the ringbuffer associated with the streamid passed in\r
+ * ftacallback_start_streamid\r
+ */\r
+struct ringbuf * ftacallback_next_streamid(gs_int32_t* state)\r
+{\r
+ for(;(nstreamid<lstreamid)\r
+ &&(fstreamid[nstreamid].streamid != cstreamid);\r
+ nstreamid++);\r
+ if (nstreamid<lstreamid) {\r
+ nstreamid++;\r
+ *state=fstreamid[nstreamid-1].state;\r
+ return fstreamid[nstreamid-1].r;\r
+ }\r
+ return 0;\r
+}\r
+\r
+\r
+/* associate msgid with ringbuf */\r
+gs_retval_t ftacallback_add_wakeup(FTAID ftaid, struct ringbuf * r)\r
+{\r
+ gs_int32_t x;\r
+ if (lwakeup == 0) {\r
+ if ((fwakeup = malloc(sizeof(struct ftacallbackwakeup)*STARTSZ))==0) {\r
+ gslog(LOG_EMERG,"ERROR:Out of memory ftacallback\n");\r
+ return -1;\r
+ }\r
+ memset(fwakeup,0,sizeof(struct ftacallbackwakeup)*STARTSZ);\r
+ lwakeup = STARTSZ;\r
+ }\r
+ /* first try to find one for the same process */\r
+ for(x=0;(x<lwakeup)&&(\r
+ ((fwakeup[x].ftaid.ip!=ftaid.ip)\r
+ || (fwakeup[x].ftaid.port!=ftaid.port))\r
+ ||(fwakeup[x].used==0)) ;x++);\r
+ if (x==lwakeup) {\r
+ /* now try to find empty slot */\r
+ for(x=0;(x<lwakeup)&&(fwakeup[x].used!=0);x++);\r
+ if (x == lwakeup) {\r
+ gs_int32_t y;\r
+ lwakeup = 2*lwakeup;\r
+ if ((fwakeup =\r
+ realloc(fwakeup,sizeof(struct ftacallbackwakeup)*lwakeup))==0) {\r
+ gslog(LOG_EMERG,"ERROR:Out of memory ftacallback\n");\r
+ return -1;\r
+ }\r
+ for (y=x;y<lwakeup;y++)\r
+ fwakeup[y].used=0;\r
+ }\r
+ }\r
+ fwakeup[x].used=1;\r
+ fwakeup[x].ftaid=ftaid;\r
+ fwakeup[x].r=r;\r
+ return x;\r
+}\r
+\r
+/* starts an itteration through all msgids associated with\r
+ a streamid. This also uses data kept in the fstreamid\r
+ registry\r
+ */\r
+gs_retval_t ftacallback_start_wakeup(gs_uint32_t streamid)\r
+{\r
+ swakeup=streamid;\r
+ nwakeup=0;\r
+ return 0;\r
+}\r
+\r
+/* returns all the msgid blocked on the streamid passed in\r
+ * ftacallback_start_streamid and removes the msgid from\r
+ * the wakeup list\r
+ */\r
+FTAID * ftacallback_next_wakeup()\r
+{\r
+ for(;(nwakeup<lstreamid)\r
+ &&(fstreamid[nwakeup].streamid != swakeup);\r
+ nwakeup++);\r
+ if (nwakeup<lstreamid) {\r
+ gs_int32_t x;\r
+ for(x=0;x<lwakeup;x++)\r
+ if ((fwakeup[x].r==fstreamid[nwakeup].r) &&\r
+ (fwakeup[x].used==1)) {\r
+ fwakeup[x].used=0;\r
+ nwakeup++;\r
+ return & fwakeup[x].ftaid;\r
+ }\r
+ }\r
+ nwakeup ++;\r
+ return (FTAID *) 0;\r
+}\r
+\r
+\r
+\r
+static gs_retval_t fta_list_add(struct fta_list ** root, struct FTA * after,\r
+ struct FTA * fta) {\r
+ struct fta_list * new;\r
+ struct fta_list * tmp;\r
+ \r
+ if ((new=(struct fta_list *)malloc(sizeof(struct fta_list)))==0) {\r
+ gslog(LOG_EMERG,"fta_list_add:: can't allocate memory\n");\r
+ return -1;\r
+ }\r
+ \r
+ new->fta=fta;\r
+ \r
+ \r
+ if (after==0) {\r
+ new->next=*root;\r
+ new->prev=0;\r
+ *root=new;\r
+ if (new->next) {\r
+ new->next->prev=new;\r
+ }\r
+ } else {\r
+ tmp=*root;\r
+ while((tmp)&&(tmp->fta!=after)) {\r
+ tmp=tmp->next;\r
+ }\r
+ if (tmp==0) {\r
+ gslog(LOG_EMERG,"fta_list_add:: can't find after fta\n");\r
+ return -1;\r
+ }\r
+ new->next=tmp->next;\r
+ new->prev=tmp;\r
+ tmp->next=new;\r
+ if (new->next) {\r
+ new->next->prev=new;\r
+ }\r
+ }\r
+ return 0;\r
+}\r
+\r
+static gs_retval_t fta_list_rm(struct fta_list ** root, struct FTA * fta) {\r
+ struct fta_list * tmp;\r
+ tmp=*root;\r
+ while((tmp)&&(tmp->fta!=fta)) {\r
+ tmp=tmp->next;\r
+ }\r
+ if (tmp==0) {\r
+ gslog(LOG_EMERG,"fta_list_rm:: can't find fta\n");\r
+ return -1;\r
+ }\r
+ if (tmp == (*root)) {\r
+ *root=tmp->next;\r
+ if (tmp->next) {\r
+ tmp->next->prev=0;\r
+ }\r
+ } else {\r
+ tmp->prev->next=tmp->next;\r
+ if (tmp->next) {\r
+ tmp->next->prev=tmp->prev;\r
+ }\r
+ }\r
+ \r
+ free(tmp);\r
+ return 0;\r
+}\r
+\r
+\r
+static gs_retval_t fta_list_check(struct fta_list ** root, struct FTA * fta) {\r
+ struct fta_list * tmp;\r
+ tmp=*root;\r
+ while((tmp)&&(tmp->fta!=fta)) {\r
+ tmp=tmp->next;\r
+ }\r
+ if (tmp==0) {\r
+ return -1;\r
+ }\r
+ return 0;\r
+}\r
+\r
+gs_retval_t ftaexec_insert(struct FTA * after, struct FTA * new)\r
+{\r
+ \r
+ if ((after!=0) && (fta_list_check(&process,after)<0)) {\r
+ gslog(LOG_EMERG,"fta_insert:: ilegal adapter for after\n");\r
+ return -1;\r
+ }\r
+ \r
+ if (fta_list_check(&created,new)<0) {\r
+ gslog(LOG_EMERG,"fta_insert:: ilegal adapter for new\n");\r
+ return -1;\r
+ }\r
+ \r
+ if (fta_list_check(&process,new)==0) {\r
+ new->runrefcnt++;\r
+ gslog(LOG_INFO,"fta_insert:: new already in process list reusing entry with streamid %d\n",new->ftaid.streamid);\r
+ return 0;\r
+ }\r
+ \r
+ if (fta_list_add(&process,after,new)<0) {\r
+ gslog(LOG_EMERG,"fta_insert:: new can not be added to process list\n");\r
+ return -1;\r
+ }\r
+ \r
+ new->runrefcnt++;\r
+ \r
+ return 0;\r
+ \r
+}\r
+\r
+gs_retval_t ftaexec_remove(struct FTA * id){\r
+ \r
+ if (fta_list_check(&process,id)<0) {\r
+ gslog(LOG_EMERG,"fta_remove:: id not in process list\n");\r
+ return -1;\r
+ }\r
+ id->runrefcnt--;\r
+ if (id->runrefcnt<=0) {\r
+ if (fta_list_rm(&process,id)<0) {\r
+ gslog(LOG_EMERG,"fta_remove:: id could not be removed from process list\n");\r
+ return -1;\r
+ }\r
+ }\r
+ return 0;\r
+}\r
+\r
+\r
+struct FTA * ftaexec_alloc_instance(gs_uint32_t index, struct FTA * reuse,\r
+ gs_uint32_t reusable,\r
+ gs_int32_t command, gs_int32_t sz, void * data){\r
+ struct FTA * f;\r
+ FTAID ftaid;\r
+ ftaid=gscpipc_getftaid();\r
+ ftaid.index=index;\r
+ ftaid.streamid=0;\r
+ if (fta_list_check(&created,reuse)==0) {\r
+ reuse->refcnt++;\r
+ return reuse;\r
+ }\r
+ if (ftacallback_get_alloc(index)==0) return 0;\r
+ f=ftacallback_get_alloc(index)(ftaid,reusable,command, sz, data);\r
+ f->prefilter=ftacallback_get_prefilter(index);\r
+ gslog(LOG_INFO,"Using prefilter %llu for fta %x\n",f->prefilter,f);\r
+ if (fta_list_add(&created,0,f)<0) {\r
+ gslog(LOG_EMERG,"fta_alloc_instance:: new fta can not be added to created list\n");\r
+ return 0;\r
+ }\r
+ f->refcnt=1;\r
+ return f;\r
+}\r
+\r
+\r
+\r
+gs_retval_t ftaexec_free_instance(struct FTA * id, gs_uint32_t recursive){\r
+ id->refcnt --;\r
+ if (id->refcnt==0) {\r
+ if (fta_list_rm(&created,id)<0) {\r
+ gslog(LOG_EMERG,"fta_free_instance:: fta could not be removed from created list\n");\r
+ return -1;\r
+ }\r
+ /* just to make sure remove it form process list too */\r
+ if (fta_list_check(&process,id)>=0) {\r
+ fta_list_rm(&process,id);\r
+ }\r
+ \r
+ id->free_fta(id,recursive);\r
+ }\r
+ return 0;\r
+}\r
+\r
+gs_retval_t ftaexec_control(struct FTA * id, gs_int32_t command, gs_int32_t sz, void * value){\r
+ if (fta_list_check(&created,id)<0) {\r
+ gslog(LOG_EMERG,"fta_control:: id not found in adapter's created list\n");\r
+ return -1;\r
+ }\r
+ return id->control_fta(id,command,sz,value);\r
+}\r
+\r
+gs_retval_t ftaexec_process_control(gs_int32_t command, gs_int32_t sz, void * value){\r
+ struct FTA * f;\r
+ ftaexec_start();\r
+ while((f=ftaexec_next())!=0) {\r
+ f->control_fta(f,command,sz,value);\r
+ }\r
+ return 1;\r
+}\r
+\r
+\r
+/* Start itteration through list of active FTA */\r
+\r
+gs_retval_t ftaexec_start()\r
+{\r
+ itteration=process;\r
+ return 0;\r
+}\r
+\r
+/* get one FTA at a time */\r
+\r
+struct FTA * ftaexec_next()\r
+{\r
+ struct FTA * fta=0;\r
+ if (itteration) {\r
+ fta=itteration->fta;\r
+ itteration=itteration->next;\r
+ }\r
+ return fta;\r
+}\r
+\r
+\r
+/* adds a buffer to the end of the sidequeue*/\r
+gs_retval_t sidequeue_append(FTAID from, gs_sp_t buf, gs_int32_t length)\r
+{\r
+ struct sq * s;\r
+ if ((s=malloc(sizeof(struct sq)))==0) {\r
+ gslog(LOG_EMERG,"Could not allocate memory for sidequeue");\r
+ return -1;\r
+ }\r
+ s->from=from;\r
+ memcpy(&s->buf[0],buf,MAXMSGSZ);\r
+ s->length=length;\r
+ s->next=0;\r
+ if (sqtail) {\r
+ sqtail->next=s;\r
+ sqtail=s;\r
+ } else {\r
+ sqtop = s;\r
+ sqtail = s;\r
+ }\r
+ return 0;\r
+}\r
+\r
+/* removes a buffer from the top of the sidequeue*/\r
+gs_retval_t sidequeue_pop(FTAID * from, gs_sp_t buf, gs_int32_t* length)\r
+{\r
+ struct sq * s;\r
+ \r
+ if (sqtop) {\r
+ *from=sqtop->from;\r
+ memcpy(buf,&sqtop->buf[0],MAXMSGSZ);\r
+ *length=sqtop->length;\r
+ s=sqtop;\r
+ sqtop=sqtop->next;\r
+ if (sqtop==0) sqtail=0;\r
+ free(s);\r
+ return 0;\r
+ }\r
+ return -1;\r
+}\r
Code Review / com / gs-lite.git / blobdiff