Merge changes from trunk to mapi-2.0 branch

git-svn-id: file:///home/svn/mapi/branches/mapi-2.0@1381 8d5bb341-7cf1-0310-8cf6-ba355fef3186

applications/packetloss/packetloss.c

@@ -21,7 +21,7 @@
 #define WITH_NCURSES 0      // set to 1 if you want ncurses, 0 otherwise
 #define REFRESH_TIME 10     // 10 seconds base interval with which data will be fed into the RRD
 
-#define SHM_FLOWS_DEVICE_MAX 49999 // (DIMAPI_DATA_SIZE - sizeof(flows_stat) - sizeof(unsigned int)) / sizeof(struct flow_data))
+#define SHM_FLOWS_DEVICE_MAX 49998 // (DIMAPI_DATA_SIZE - sizeof(flows_stat) - cca24) / sizeof(struct flow_data))
 #define SHM_FLOWS_TOTAL_MAX 250000
 #define TIMESTAMP_TOLERANCE 430000000 // NTP 0.1
 #define MIN(x, y) ((x) < (y) ? (x) : (y)) 
@@ -337,7 +337,7 @@ int main(int argc, char *argv[])
 
 	unsigned int *results;      // number of results/flowdata = (unsigned int *) ...
 	struct flows_stat *flowsstats;
-	struct flow_data *flowdata; // result/flowdata            = (struct flow_data *)((char *) ... + sizeof(flows_stat) + sizeof(unsigned int));
+	struct flow_data *flowdata; // result/flowdata            = (struct flow_data *)((char *) ... ) + sizeof(flows_stat);
 
 	unsigned long long pkloss = 0;
 
@@ -710,7 +710,7 @@ int main(int argc, char *argv[])
 				}
 				results = (unsigned int *) host[h].results_dev[d].res; // number of results/flows at host h, dev d
 				flowsstats = (struct flows_stat *) host[h].results_dev[d].res;
-				flowdata = (struct flow_data *)((char *) host[h].results_dev[d].res + sizeof(flows_stat) + sizeof(unsigned int)); // results/flows array
+				flowdata = (struct flow_data *)((char *) host[h].results_dev[d].res + sizeof(flows_stat)); // results/flows array
 				fprintf(stdout, "packetloss: storing and matching results from %s:%d: %d\n", host[h].name, d, *results); fflush(stdout);
 
 				// store flowstats_dev
@@ -763,7 +763,7 @@ int main(int argc, char *argv[])
 				// }}}
 
 				results = (unsigned int *) host[h].results_dev[d].res; // number of results/flows
-				flowdata = (struct flow_data *) ((char *) host[h].results_dev[d].res + sizeof(flows_stat) + sizeof(unsigned int)); // results/flows array
+				flowdata = (struct flow_data *) ((char *) host[h].results_dev[d].res + sizeof(flows_stat)); // results/flows array
 
 				for (count = *results; count > 0; count--) {
 					if (LOGFILE) {

src/common/mapiipc.c

@@ -635,6 +635,7 @@ int mapiipc_remote_init(struct host *h)
   }
 
   if (host==NULL) {
+    close(h->sockfd);
     printf("ERROR: Could not determine address for host %s [%s:%d]\n", h->hostname, __FILE__, __LINE__);
     return -1;
   }
@@ -645,16 +646,19 @@ int mapiipc_remote_init(struct host *h)
   remoteaddr.sin_port = htons(h->port);
 
   if (connect(h->sockfd, (struct sockaddr *)&remoteaddr, sizeof(remoteaddr)) < 0) {
+    close(h->sockfd);
     printf("ERROR: connect failed (%s) [%s:%d]\n", strerror(errno), __FILE__, __LINE__);
     return -1;
   }
 
 #ifdef DIMAPISSL
 	if (SSL_set_fd(h->con, h->sockfd) == 0) {
+		close(h->sockfd);
 		ERR_print_errors_fp(stderr);
 		return -1;
 	}
 	if (SSL_connect(h->con) <= 0) {
+		close(h->sockfd);
 		ERR_print_errors_fp(stderr);
 		return -1;
 	}

src/drivers/mapidagdrv.c

@@ -1081,7 +1081,7 @@ mapidrv_stats (int devid, char **devtype, struct mapi_stat *stats)
 
   if (i!=NULL)
   {
-	stats->ps_recv=i->hwinfo.pkts;
+	stats->ps_recv=i->hwinfo.pkts + i->hwinfo.pkt_drop;
 	stats->ps_drop=i->hwinfo.pkt_drop;
 	stats->ps_ifdrop=0;
 	return 0;

src/extraflib/expiredflowshash.h

-#define EXFL_HASH_SIZE 262144 //4096
+#define EXFL_HASH_SIZE 262144 //4128
 
 #include "protocols.h"
 

src/extraflib/exprflow.c

@@ -35,7 +35,6 @@ pthread_t pthread;
 
 typedef struct shm {
 	unsigned int *size;
-	unsigned int *shm_flows;
 	flow_data *Table;
 	pthread_mutex_t *smmutex;
 }shm;
@@ -252,8 +251,8 @@ void poll_expired_flows(mapidflib_function_instance_t *instance) {
 
 	stats = (flows_stat *) instance->result.data;
 	shm_struct.size = (unsigned int *)instance->result.data;
-	shm_struct.Table = (flow_data *)((char *)instance->result.data+sizeof(flows_stat)+sizeof(unsigned int));
-	shm_struct.smmutex = (pthread_mutex_t *)(((char *)instance->result.data) + sizeof(flows_stat) + sizeof(unsigned int) + sizeof(struct flow_data)*data->shm_flows);
+	shm_struct.Table = (flow_data *)((char *)instance->result.data+sizeof(flows_stat));
+	shm_struct.smmutex = (pthread_mutex_t *)(((char *)instance->result.data) + sizeof(flows_stat) + sizeof(struct flow_data)*data->shm_flows);
 	
 	*(shm_struct.size) = 0;
 	while(data->run) {
@@ -383,12 +382,12 @@ static int exprflow_instance(mapidflib_function_instance_t *instance,
                              MAPI_UNUSED int fd,
                              MAPI_UNUSED mapidflib_flow_mod_t *flow_mod) {
 	mapiFunctArg* fargs;
-	int shm_flows; // max: (DIMAPI_DATA_SIZE - sizeof(flows_stat) - sizeof(unsigned int)) / sizeof(struct flow_data))
+	int shm_flows; // max: (DIMAPI_DATA_SIZE - sizeof(flows_stat) - cca24) / sizeof(struct flow_data))
 
 	fargs = instance->args;
 	shm_flows = getargint(&fargs);
 
-	instance->def->shm_size = sizeof(flows_stat) + sizeof(unsigned int) + sizeof(struct flow_data) * shm_flows + sizeof(unsigned int) + sizeof(pthread_mutex_t);
+	instance->def->shm_size = sizeof(flows_stat) + sizeof(struct flow_data) * shm_flows + sizeof(pthread_mutex_t);
 
 	return 0;
 }
@@ -424,23 +423,21 @@ static int exprflow_init(mapidflib_function_instance_t *instance, MAPI_UNUSED in
 	data->epoch=0;
 
 	// function arguments
-	data->shm_flows = getargint(&fargs); // max: (DIMAPI_DATA_SIZE - sizeof(flows_stat) - sizeof(unsigned int)) / sizeof(struct flow_data))
+	data->shm_flows = getargint(&fargs); // max: (DIMAPI_DATA_SIZE - sizeof(flows_stat) - cca24) / sizeof(struct flow_data))
 	data->expired_flows_list_size_max = getargint(&fargs);
 	data->packets_count_min = getargint(&fargs);
 
 	// Shared Memory Initialization
 	stats = (flows_stat *)instance->result.data;
 	shm_struct.size = (unsigned int *)instance->result.data;
-	shm_struct.shm_flows = (unsigned int*)((char *)instance->result.data+sizeof(flows_stat));
-	shm_struct.Table = (flow_data *)((char *)instance->result.data+sizeof(flows_stat)+sizeof(unsigned int));
-	shm_struct.smmutex = (pthread_mutex_t *)(((char *)instance->result.data) + sizeof(flows_stat) + sizeof(unsigned int) + sizeof(struct flow_data)*data->shm_flows);
+	shm_struct.Table = (flow_data *)((char *)instance->result.data+sizeof(flows_stat));
+	shm_struct.smmutex = (pthread_mutex_t *)(((char *)instance->result.data) + sizeof(flows_stat) + sizeof(struct flow_data) * data->shm_flows);
 
 	stats->received = 0;
 	stats->expired = 0;
 	*(shm_struct.size) = 0; //stats->sent = 0;
 	stats->ignored = 0;
 	stats->dropped = 0;
-	*(shm_struct.shm_flows) = data->shm_flows;
 
 	//mutex initialization
 	*(shm_struct.smmutex) = tmpmutex;
@@ -609,27 +606,22 @@ void print_flow(eflow_data *data) {
 }
 	
 static int exprflow_client_read_result(mapidflib_function_instance_t *instance,mapi_result_t *res) {
-	unsigned int size;
 	shm shm_struct;
-	eflow_data *data;
 	flows_stat *stats;
 
 	mapiFunctArg* fargs;
-	int shm_flows; // max: (DIMAPI_DATA_SIZE - sizeof(flows_stat) - sizeof(unsigned int)) / sizeof(struct flow_data))
+	int shm_flows; // max: (DIMAPI_DATA_SIZE - sizeof(flows_stat) - cca24) / sizeof(struct flow_data))
 
 	fargs = instance->args;
 	shm_flows = getargint(&fargs);
 
 	stats = (flows_stat *)instance->result.data;
 	shm_struct.size = (unsigned int *)instance->result.data;
-	shm_struct.shm_flows = (unsigned int *)((char *)instance->result.data+sizeof(flows_stat));
-	shm_struct.Table = (flow_data *)((char *)instance->result.data+sizeof(flows_stat)+sizeof(unsigned int));
-fprintf(stdout, "shm_flows: %u", *(shm_struct.shm_flows)); fflush(stdout);
-	shm_struct.smmutex = (pthread_mutex_t *)(((char *)instance->result.data) + sizeof(flows_stat) + sizeof(unsigned int) + sizeof(struct flow_data)*(*shm_struct.shm_flows));
+	shm_struct.Table = (flow_data *)((char *)instance->result.data+sizeof(flows_stat));
+	shm_struct.smmutex = (pthread_mutex_t *)(((char *)instance->result.data) + sizeof(flows_stat) + sizeof(struct flow_data) * shm_flows);
 
 	res->res = instance->internal_data;
-	res->size = sizeof(flows_stat) + sizeof(unsigned int) + sizeof(flow_data)*(*(shm_struct.size));
-	size = *shm_struct.size;
+	res->size = sizeof(flows_stat) + sizeof(flow_data) * (*(shm_struct.size));
 
 	pthread_mutex_lock( shm_struct.smmutex ); 
 	memcpy(instance->internal_data, instance->result.data, res->size);
@@ -640,25 +632,24 @@ fprintf(stdout, "shm_flows: %u", *(shm_struct.shm_flows)); fflush(stdout);
 	stats->dropped = 0;
 	pthread_mutex_unlock( shm_struct.smmutex ); 
 
-	data = (flow_data *)((char *)instance->internal_data+sizeof(flows_stat)+sizeof(unsigned int));
 	return(0);
 }
 
 static int exprflow_client_init( mapidflib_function_instance_t *instance, void *data) {
 
 	mapiFunctArg* fargs;
-	int shm_flows; // max: (DIMAPI_DATA_SIZE - sizeof(flows_stat) - sizeof(unsigned int)) / sizeof(struct flow_data))
+	int shm_flows; // max: (DIMAPI_DATA_SIZE - sizeof(flows_stat) - cca24) / sizeof(struct flow_data))
 
 	fargs = instance->args;
 	shm_flows = getargint(&fargs);
 
-	if(shm_flows > (DIMAPI_DATA_SIZE - sizeof(flows_stat) - sizeof(unsigned int)) / sizeof(struct flow_data)) {
-		printf("Cannot process %d flows. Maximum is %d.\n", shm_flows, (DIMAPI_DATA_SIZE - sizeof(flows_stat)) / sizeof(struct flow_data));
+	if(shm_flows > (DIMAPI_DATA_SIZE - sizeof(flows_stat) - 24) / sizeof(struct flow_data)) {
+		printf("Cannot process %d flows. Maximum is %d.\n", shm_flows, (DIMAPI_DATA_SIZE - sizeof(flows_stat) - 24) / sizeof(struct flow_data));
 		return(-1);
 	}
 
-	if((instance->internal_data = malloc(sizeof(flows_stat)+sizeof(unsigned int)+sizeof(struct flow_data)*shm_flows)) == NULL) {
-		printf("Malloc failed for size %d [%s:%d]\n", sizeof(flows_stat)+sizeof(unsigned int)+sizeof(struct flow_data)*shm_flows, __FILE__, __LINE__);
+	if((instance->internal_data = malloc(sizeof(flows_stat)+sizeof(struct flow_data)*shm_flows)) == NULL) {
+		printf("Malloc failed for size %d [%s:%d]\n", sizeof(flows_stat)+sizeof(struct flow_data)*shm_flows, __FILE__, __LINE__);
 		return(-1);
 	}
 	data = instance->internal_data;

src/lib/mapi.c

@@ -633,6 +633,12 @@ int mapi_create_flow(const char *dev)
 	  if (mapiipc_remote_init(h)<0) {
 	   	  local_err = MCOM_SOCKET_ERROR;
 		  printf("ERROR: Could not connect with host %s [%s:%d]\n", h->hostname, __FILE__, __LINE__);
+		  flist_destroy(h->flows);
+		  free(h->flows);
+		  flist_destroy(h->functions);
+		  free(h->functions);
+		  free(h->hostname);
+		  free(h);
 		  pthread_spin_unlock(&hostlist_lock);
 		  return -1;
 	  }
@@ -3912,6 +3918,7 @@ int mapi_is_sensor_down(int fd){
 
 	if(rflow != NULL && rflow->daemons_down == 1)
 		return 1;
+	else if (rflow!=NULL) return 0;
 #else
 	return 0;	// without reconnect always up ??
 #endif
@@ -3920,15 +3927,16 @@ int mapi_is_sensor_down(int fd){
 	flowdescr_t *flow = flist_get(flowlist, fd);
 	int check_mapid;
 
+	if (flow==NULL) return 0;
+
 	check_mapid = check_network_mapid();
 
-	if(check_mapid == 1 && flow != NULL)
+	if(check_mapid == 1) {  //up
 		flow->mapid_down = 0;
-
-	if(flow != NULL && flow->mapid_down == 1)
-		return 1;
-	else
 		return 0;
+	}
+	else 
+		return 1;
 #else
 	return 0;	// without reconnect always up ??
 #endif

src/mapicommd.c

@@ -485,6 +485,16 @@ void *handle_request(void *arg) {
 				}
 				break;
 			case READ_RESULT:
+#ifdef RECONNECT
+				if (mapi_is_sensor_down(dbuf->fd)==1) {
+					printf("SENSOR DOWN\n");
+					int local_err=MAPI_READ_RESULT_RECONNECTION;
+					dbuf->cmd = ERROR_ACK;
+					memcpy(dbuf->data, &local_err, sizeof(int));
+					dbuf->length = BASIC_SIZE + sizeof(int);
+					break;
+				}
+#endif
 				result = mapi_read_results(dbuf->fd,dbuf->fid);
 				if(result!=NULL && result->size < DIMAPI_DATA_SIZE){
 					dbuf->cmd = READ_RESULT_ACK;

tests/test_bucket.c

+#include <stdio.h>
+#include <unistd.h>
+#include <sys/time.h>
+#include <mapi.h>
+#include <mapi/bucket.h>
+
+#define LOOPS 100
+
+int main(int argc, char *argv[])
+{
+	int fd;
+	int fid, fid2;
+	mapi_results_t* res;
+	int err_no=0;
+	char error[512];
+	int loop=0;
+	struct bucket_data bucket;
+
+	if (argc != 2) {
+		fprintf(stderr, "\nWrong arguments\n");
+		return -1;
+	}
+
+	if ((fd = mapi_create_flow(argv[1])) < 0) {
+		fprintf(stderr, "Could not create flow using '%s'\n", argv[1]);
+		mapi_read_error( &err_no, error);
+		fprintf(stderr,"Errorcode :%d description: %s \n" ,err_no, error);
+		return -1;
+	}
+
+	if ((fid = mapi_apply_function(fd, "PKT_COUNTER")) == -1) {
+		fprintf(stderr, "Count not apply function PKT_COUNTER to flow %d\n", fd);
+		mapi_read_error( &err_no, error);
+		fprintf(stderr,"Errorcode :%d description: %s \n" ,err_no, error);
+		return -1;
+	}
+
+	if ((fid2 = mapi_apply_function(fd, "BUCKET",fd,fid,"1s",1)) == -1) {
+		fprintf(stderr, "Count not apply function PKT_COUNTER to flow %d\n", fd);
+		mapi_read_error( &err_no, error);
+		fprintf(stderr,"Errorcode :%d description: %s \n" ,err_no, error);
+		return -1;
+	}
+
+	if (mapi_connect(fd) < 0) {
+		fprintf(stderr, "Could not connect to flow %d\n", fd);
+		mapi_read_error( &err_no, error);
+		fprintf(stderr,"Errorcode :%d description: %s \n" ,err_no, error);
+		return -1;
+	}
+
+	while (loop++ <= LOOPS) {
+		res = mapi_read_results(fd, fid2);
+		bucket = *((struct bucket_data*)res->res);
+		printf("\npackets for one second interval: %llu\n", bucket.data);
+	}
+
+	if(mapi_close_flow(fd)<0){
+		fprintf(stderr,"Close flow failed\n");
+		mapi_read_error( &err_no, error);
+		fprintf(stderr,"Errorcode :%d description: %s \n" ,err_no, error);
+		return -1;
+	}
+
+	return 1;
+}
+