Agent is now generic

git-svn-id: file:///home/svn/mapi/trunk@463 8d5bb341-7cf1-0310-8cf6-ba355fef3186

agent/agent.c

@@ -12,7 +12,7 @@
 #include "mapi.h"
 #include "mapiipc.h"
 #include "parseconf.h"
-
+#include "mapilibhandler.h"
 
 #define MAXPENDING 500    /* Maximum outstanding connection requests */
 #define CONF_FILE "./mapi.conf:%s/.mapi.conf:/etc/mapi.conf"
@@ -102,6 +102,8 @@ int main(int argc, char **argv) {
 		die("Unexpected error on listen()");
 	}
 
+	is_agent();
+
 	while(1) {
 
 		clnt_len = sizeof clnt_addr;
@@ -133,9 +135,10 @@ void *handle_request(void *arg) {
 	int *active_flows = NULL;
 	int ac_fl_size=0;
 	mapi_function_info_t funct_info;
+	mapidflib_function_def_t *fdef;
+	int result_size;
 	mapi_flow_info_t flow_info;
 	struct timeval tv; /*used for timestamping results when produced */
-	struct timezone tz;
 	struct mapipkt *pkt;
 	struct in_addr client;
 	char* dev_addr;
@@ -232,39 +235,16 @@ void *handle_request(void *arg) {
 				break;
 			case READ_RESULT:
 				//fprintf(stdout,"READ_RESULT\n");
-
 				dbuf->cmd = READ_RESULT_ACK;
 				result = mapi_read_results(dbuf->fd,dbuf->fid);
-				gettimeofday(&tv, &tz);
+				gettimeofday(&tv, NULL);
 				dbuf->timestamp = tv.tv_sec*1000000 + tv.tv_usec;
 				if(result!=NULL){
 					mapi_get_function_info(dbuf->fd,dbuf->fid,&funct_info);
-					if(!strcmp(funct_info.name,"PKT_COUNTER") || !strcmp(funct_info.name,"BYTE_COUNTER")){
-						memcpy(dbuf->data, result, sizeof(unsigned long long));
-						dbuf->length = BASIC_SIZE + sizeof(unsigned long long);
-					}
-					else if(!strcmp(funct_info.name,"HASH")){
-						memcpy(dbuf->data, result, sizeof(unsigned long long));
-						dbuf->length = BASIC_SIZE + sizeof(unsigned long long);
-					}
-					else if(!strcmp(funct_info.name,"HASHSAMP")){
-						memcpy(dbuf->data, result, sizeof(struct sample));
-						dbuf->length = BASIC_SIZE + sizeof(struct sample);
-					}
-					else if(!strcmp(funct_info.name,"BUCKET")){
-						memcpy(dbuf->data, result, sizeof(struct bucket_data));
-						dbuf->length = BASIC_SIZE + sizeof(struct bucket_data);
-					}
-					else if(!strcmp(funct_info.name,"TO_FILE")){
-						memcpy(dbuf->data, result, sizeof(int));
-						dbuf->length = BASIC_SIZE + sizeof(int);
-					}
-                                        else if(!strcmp(funct_info.name,"TOP")){
-                                                memcpy(dbuf->data, result, 2*sizeof(unsigned int)*100+4);
-                                                dbuf->length = BASIC_SIZE + 2*sizeof(unsigned int)*100+4;
-                                        }
-
-					//to be continued...
+					fdef=mapilh_get_function_def(funct_info.name,funct_info.devtype);
+					result_size=fdef->shm_size;
+					memcpy(dbuf->data, result, result_size);
+					dbuf->length = BASIC_SIZE + result_size;
 				}
 				else{
 					fprintf(stdout,"mapi_read_results failed...\n");
@@ -275,7 +255,7 @@ void *handle_request(void *arg) {
 			case GET_NEXT_PKT:
 				//fprintf(stdout,"GET_NEXT_PKT\n");
 				pkt = (struct mapipkt *)mapi_get_next_pkt(dbuf->fd,dbuf->fid);
-				gettimeofday(&tv, &tz);
+				gettimeofday(&tv, NULL);
 				dbuf->timestamp = tv.tv_usec;
 				if(pkt!=NULL){
 					dbuf->cmd = GET_NEXT_PKT_ACK;
@@ -403,201 +383,13 @@ int die(char *msg) {
 }
 
 //calls the appropriate mapi_apply_function and returns the fid from mapid
-//
-//fully unoptimized:
-//1) find a way not to use strcmp
-//2) use generic int variables - less readable but uses less memory
 int getfid(struct dmapiipcbuf *dbuf){
 	int result;
-	//redundant declarations - but more readable code
-	char *p;
-	//for STR_SEARCH:
-	int offset, depth;
-	//for SAMPLE
-	int value, mode;
-	//for COOKING
-	int threshold, timeout;
-	//for HASHAMP
-	int range, keep;
-	//for TO_FILE
-	int format, count;
-	char *path;
-	//for BUCKET
-	unsigned long long ltimeout, pktcount;
-	int funct;
-	//for THRESHOLD
-	int type, boundary, divider;
-	unsigned long long lthreshold;
+	
 	char *function = (char *)dbuf->data;
-	char *data = (char *)(dbuf->data+strlen((char *)dbuf->data)+1);
+	char *data = (char *)(dbuf->data+strlen(function)+1);
 
-	if(!strcmp(function,"BPF_FILTER")){//Checked
-		fprintf(stdout,"%s, %s\n",function, data);
-		result = mapi_apply_function(dbuf->fd, function, data);
-		return(result);
-	}
-        else if(!strcmp(function,"TOP")){//Checked
-		int arg1, arg2, arg3, arg4, arg5;
-		p=data;
-		memcpy(&arg1,p,sizeof(int));
-		p+=sizeof(int);
-                memcpy(&arg2,p,sizeof(int));
-                p+=sizeof(int);
-                memcpy(&arg3,p,sizeof(int));
-                p+=sizeof(int);
-                memcpy(&arg4,p,sizeof(int));
-                p+=sizeof(int);
-                memcpy(&arg5,p,sizeof(int));
-                fprintf(stdout,"%s, %d %d %d %d %d\n",function, arg1, arg2, arg3, arg4, arg5);
-                result = mapi_apply_function(dbuf->fd, function, arg1, arg2, arg3, arg4, arg5);
-                return(result);
-        }
-        else if(!strcmp(function,"TRACK_FTP")){//Checked
-                fprintf(stdout,"TRACK_FTP\n");
-                result = mapi_apply_function(dbuf->fd, function);
-                return(result);
-        }
-        else if(!strcmp(function,"TRACK_GNUTELLA")){//Checked
-                fprintf(stdout,"TRACK_GNUTELLA\n");
-                result = mapi_apply_function(dbuf->fd, function);
-                return(result);
-        }
-        else if(!strcmp(function,"TRACK_DC")){//Checked
-                fprintf(stdout,"TRACK_DC\n");
-                result = mapi_apply_function(dbuf->fd, function);
-                return(result);
-        }
-        else if(!strcmp(function,"TRACK_TORRENT")){//Checked
-                fprintf(stdout,"TRACK_TORRENT\n");
-                result = mapi_apply_function(dbuf->fd, function);
-                return(result);
-        }
-        else if(!strcmp(function,"TRACK_EDONKEY")){//Checked
-                fprintf(stdout,"TRACK_EDONKEY\n");
-                result = mapi_apply_function(dbuf->fd, function);
-                return(result);
-        }
-        else if(!strcmp(function,"TRACK_IPOVERIP")){//Checked
-                fprintf(stdout,"TRACK_IPOVERIP\n");
-                result = mapi_apply_function(dbuf->fd, function);
-                return(result);
-		}
-	else if(!strcmp(function,"PKT_COUNTER")){//Checked
-		fprintf(stdout,"PKT_COUNTER\n");
-		result = mapi_apply_function(dbuf->fd, function);
-		return(result);
-	}
-	else if(!strcmp(function,"BYTE_COUNTER")){//Checked
-		fprintf(stdout,"BYTE_COUNTER\n");
-		return(mapi_apply_function(dbuf->fd, function));
-	}
-	else if(!strcmp(function,"STR_SEARCH")){//Checked
-		fprintf(stdout,"STR_SEARCH\n");
-		p = data + strlen(data) + 1;
-		memcpy(&offset,p,sizeof(int));
-		p += sizeof(int);
-		memcpy(&depth,p,sizeof(int));
-		result = mapi_apply_function(dbuf->fd, function, data, offset, depth);
-		return(result);
-	}
-	else if(!strcmp(function,"TO_BUFFER")){//Checked
-		fprintf(stdout,"TO_BUFFER\n");
-		return(mapi_apply_function(dbuf->fd, function));
-	}
-	else if(!strcmp(function,"SAMPLE")){//Checked
-		fprintf(stdout,"SAMPLE\n");
-		p = data;
-		memcpy(&value,p,sizeof(int));
-		p += sizeof(int);
-		memcpy(&mode,p,sizeof(int));
-		printf("value == %d, mode == %d\n",value,mode);
-		result = mapi_apply_function(dbuf->fd, function, value, mode);
-		return(result);
-	}
-	else if(!strcmp(function,"HASHSAMP")){
-		//XXX not tested
-		fprintf(stdout,"HASHSAMP\n");
-		p = data;
-		memcpy(&range,p,sizeof(int));
-		p += sizeof(int);
-		memcpy(&keep,p,sizeof(int));
-		printf("range == %d, keep == %d\n",range,keep);
-		result = mapi_apply_function(dbuf->fd, function, range, keep);
-		return(result);
-	}
-	else if(!strcmp(function,"TO_FILE")){
-		//XXX does not work the way we want
-		//if called here, the file will be created at the agent's machine
-		fprintf(stdout,"TO_FILE\n");
-		p = data;
-		memcpy(&format,p,sizeof(int));
-		p += sizeof(int);
-		path = (char *)malloc(strlen(p)+1);
-		memcpy(path,p,strlen(p)+1);
-		p += strlen(path)+1;
-		memcpy(&pktcount,p,sizeof(unsigned long long));
-		printf("format == %d, path == %s, pktcount == %llu\n",format,path,pktcount);
-		result = mapi_apply_function(dbuf->fd, function, format, path, pktcount);
-		free(path);
-		return(result);
-	}
-	else if(!strcmp(function,"ETHEREAL")){
-		//XXX this function is in the man page but isn't recognized by the mapi stub in this version
-		fprintf(stdout,"ETHEREAL\n");
-		return(mapi_apply_function(dbuf->fd, function, data));
-	}
-	else if(!strcmp(function,"HASH")){//Checked
-		fprintf(stdout,"HASH\n");
-		return(mapi_apply_function(dbuf->fd, function));
-	}
-	else if(!strcmp(function,"COOKING")){
-		//XXX not tested
-		fprintf(stdout,"COOKING\n");
-		p = data;
-		memcpy(&threshold,p,sizeof(int));
-		p += sizeof(int);
-		memcpy(&timeout,p,sizeof(int));
-		printf("threshold == %d, timeout == %d\n", threshold, timeout);
-		result = mapi_apply_function(dbuf->fd, function, threshold, timeout);
-		return(result);
-	}
-	else if(!strcmp(function,"BUCKET")){
-		//XXX not tested
-		fprintf(stdout,"BUCKET\n");
-		p = data;
-		memcpy(&ltimeout,p,sizeof(unsigned long long));
-		p += sizeof(unsigned long long);
-		memcpy(&funct,p,sizeof(int));
-		printf("timeout == %lld, function == %d\n", ltimeout, funct);
-		result = mapi_apply_function(dbuf->fd, function, ltimeout, function);
-		return(result);
-	}
-	else if(!strcmp(function,"THRESHOLD")){
-		//XXX not tested 
-		fprintf(stdout,"THRESHOLD\n");
-		p = data;
-		memcpy(&type,p,sizeof(int));
-		p += sizeof(int);
-		memcpy(&funct,p,sizeof(int));
-		p += sizeof(int);
-		memcpy(&lthreshold,p,sizeof(unsigned long long));//need to check type argument first
-		p += sizeof(unsigned long long);
-		memcpy(&boundary,p,sizeof(int));
-		p += sizeof(int);
-		memcpy(&timeout,p,sizeof(int));
-		p += sizeof(int);
-		memcpy(&divider,p,sizeof(int));
-		p += sizeof(int);
-		memcpy(&count,p,sizeof(int));
-		printf("type == %d, function == %d, ", type, funct); 
-		printf("boundary == %d, timeout == %d, divider == %d, count == %d", boundary, timeout, divider, count);
-		printf("threshold == %llu\n",lthreshold);
-		result = mapi_apply_function(dbuf->fd, function, threshold, timeout);
-		return(result);
-	}
-	else{
-		printf("Unknown function: %s\n", dbuf->data);
-	}
-	return(-1);
+	result = mapi_apply_function(dbuf->fd, function, data);
+	return (result);
 }
 

mapi.c

@@ -55,6 +55,8 @@ static int numflows=0;  // number of allocated (active) flows
 static int totalflows=0;  // number of flows so far (including closed flows)
 
 static int offline_devices;
+static int agent=0;
+
 
 /*
 //Structure used as a linked list to store information about
@@ -951,8 +953,8 @@ int mapi_apply_function(int fd, const char* funct, ...)
   char *argdescr_ptr;
   char *filename;
   char *temp;
-  unsigned int arg_size=0;//only used in dimapi - declared here to avoid multiple ifdefs later
-
+  unsigned int arg_size=0;	//only used in dimapi - declared here to avoid multiple ifdefs later
+  unsigned char* args;		//in case read from a buffer instead of va_list
 
 #ifdef DIMAPI
   unsigned char is_remote=0;
@@ -997,28 +999,48 @@ int mapi_apply_function(int fd, const char* funct, ...)
   va_start(vl,funct);
   pos = qbuf.data;  // point to start of arguments buffer
 
+  if (agent==1) {
+    args = va_arg(vl, unsigned char*);
+  }
+
   // parse function arguments
   if(strncmp(fdef->argdescr, "", 1)) { // there are some args
     argdescr_ptr = fdef->argdescr;
     while(strlen(argdescr_ptr) > 0){
       switch(*argdescr_ptr) {
       case 's':
-	temp=va_arg(vl, char*);
+	if (agent==0) temp=va_arg(vl, char*);
+	else {
+          temp=(char*)args;
+	  args+=strlen(temp)+1;
+	}
 	addarg(&pos, temp, STRING);
-	arg_size+=strlen(temp)+1;	
+	arg_size+=strlen(temp)+1;
 	break;
       case 'i':
-	tmp = va_arg(vl, int);
+	if (agent==0) tmp = va_arg(vl, int);
+	else {
+          memcpy(&tmp, args, sizeof(int));
+	  args+=sizeof(int);
+	}
 	addarg(&pos, &tmp, INT);
 	arg_size+=sizeof(int);
 	break;
       case 'c':
-	ctmp = va_arg(vl, int); //`char' is promoted to `int' when passed through `...'
+	if (agent==0) ctmp = va_arg(vl, int); //`char' is promoted to `int' when passed through `...'
+	else {
+          memcpy(&ctmp, args, sizeof(char));
+	  args+=sizeof(char);
+	}
 	addarg(&pos, &ctmp, CHAR);
 	arg_size+=sizeof(char);
 	break;
       case 'l':
-	ltmp = va_arg(vl, unsigned long long);
+	if (agent==0) ltmp = va_arg(vl, unsigned long long);
+	else {
+          memcpy(&ltmp, args, sizeof(unsigned long long));
+	  args+=sizeof(unsigned long long);
+	}
 	addarg(&pos, &ltmp, UNSIGNED_LONG_LONG);
 	arg_size+=sizeof(unsigned long long);
 	break;
@@ -1042,7 +1064,11 @@ int mapi_apply_function(int fd, const char* funct, ...)
 	*/
       case 'w':
 	//Open file for writing
-	filename=va_arg(vl, char*);
+	if (agent==0) filename=va_arg(vl, char*);
+	else {
+          filename=(char*)args;
+	  args+=strlen(filename)+1;
+	}
 #ifdef DIMAPI
         if(is_remote){//flow is remote
 	  addarg(&pos, filename, STRING);
@@ -1245,7 +1271,6 @@ void* mapi_read_results(int fd, int fid)
   struct dmapi_results* dres;
   unsigned int currhost = 0;
   struct timeval tv;
-  struct timezone tz;
   flist_node_t* fnode;
 #endif
 
@@ -1274,7 +1299,7 @@ void* mapi_read_results(int fd, int fid)
       hflow=(host_flow*)fnode->data;
       switch(hflow->dbuf->cmd) {
         case READ_RESULT_ACK:
-          gettimeofday(&tv, &tz);
+          gettimeofday(&tv, NULL);
           dres->res[currhost].t_received = tv.tv_sec*1000000 + tv.tv_usec;
 	  dres->res[currhost].t_produced = hflow->dbuf->timestamp;
 	  strncpy(dres->res[currhost].host_dev, hflow->rhost->hostname, HOST_DEV_SIZE);
@@ -2325,3 +2350,8 @@ int incr_totalflows() {
   return n;
 }
 
+void is_agent() {
+	agent=1;
+}
+
+

mapi.h

@@ -226,4 +226,8 @@ int mapi_get_function_info(int fd, int fid, mapi_function_info_t *info);
 //Get information about a function applied to a flow
 int mapi_get_next_function_info(int fd, int fid, mapi_function_info_t *info);
 
+#ifdef DIMAPI
+void is_agent();
+#endif
+
 #endif