mapi.c

#ifdef HAVE_CONFIG_H
#include <config.h>
#endif
#include <stdio.h>
#include <stdarg.h>
#include <stdlib.h>
#include <sys/file.h>
#include <fcntl.h>
#include <unistd.h>
#include <sys/types.h>
#include <sys/time.h>
#include <sys/stat.h>
#include <sys/mman.h>
#include <netinet/in.h>
#include <errno.h>
#include <pthread.h>
#include <string.h>
#include <sys/shm.h>
#include <sys/sem.h>
#include <unistd.h>
#include "mapi.h"
#include "mapiipc.h"
#include "mapilibhandler.h"
#include "flist.h"
#include "debug.h"
#include "parseconf.h"
#include "printfstring.h"
#include "mapi_errors.h"
#include "devgroupdb.h"

#ifdef WITH_ADMISSION_CONTROL
#include <regex.h>
#include <keynote.h>
#include <openssl/rsa.h>
#include "bytestream.h"
#endif

#ifdef WITH_ANONYMIZATION
#include "anonymization/anonymization.h"
#endif

#ifdef DIMAPI
#include <signal.h>
#include <semaphore.h>
#endif

#ifdef WITH_AUTHENTICATION
#define FROM_MAPI
#include "vod/vod.h"
#endif

static pthread_once_t mapi_is_initialized = PTHREAD_ONCE_INIT;
static pthread_once_t initialized = PTHREAD_ONCE_INIT;
static int minit=0; //Set to 1 when MAPI has been initialized

static pthread_spinlock_t mapi_lock;
static pthread_spinlock_t numflows_lock;  // for numflows and totalflows variables

static int local_err=0; /* occurence of a mapi.c error, translation of these errors */

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;

extern const errorstruct Errors[];

#ifdef DIMAPI
static int hostcmp(void *h1, void *h2);
static void delete_remote_flow(remote_flowdescr_t* rflow);
flist_t *hostlist=NULL;//list containing all remote hosts used so far
extern flist_t *remote_flowlist;
int dimapi_port;
extern sem_t stats_sem;

typedef struct function_data{

	int fid;			// real fid returned from mapicommd
	mapidflib_function_def_t* fdef;	// function definition
#ifdef RECONNECT
	host_flow *hflow;		// flow of a specified host
	char args[DIMAPI_DATA_SIZE];	// function's arguments
	int index;			// need for functions that have arguments that reference to a flow
#endif
} function_data;

static pthread_once_t dmapi_is_initialized = PTHREAD_ONCE_INIT;
static pthread_spinlock_t remote_ipc_lock;
static pthread_spinlock_t hostlist_lock;

static unsigned fdseed = 0;        // 'scope' flow descriptor seed (always increases)
static unsigned fidseed = 0;       // function descriptor seed (always increases)
static unsigned negfdseed=-1;       // generates temporary negative fd, for use before create_flow
#endif

typedef struct libinfo {
	char* name;
} libinfo_t;

typedef struct flowdescr {
  int fd;
  int file; 		// file descriptor for offline flows
  int fds[256];		// file descriptors
  int numfd;		// number of file descriptors
  char *devtype;
#ifdef RECONNECT
  char *device;
  int read_results_flag;
  int mapid_down;
#endif
  char *shm_base;
  pthread_spinlock_t *shm_spinlock;
  flist_t *flist;
  unsigned char is_connected;	// this should be 1 if the flow is connected 0 otherwise
  int format;			// this should be MFF_PCAP or MFF_DAG_ERF if the flow is offline, -1 otherwise  
} flowdescr_t;

#ifdef RECONNECT

typedef struct offline_device{
	char *path;			// specifies the name of the trace file to open
	char *previous_device;		// return value of mapi_create_offline_device (device name that should be used in mapi_start_offline_device)
	char *new_device;		// new device name (returned from mapid)
	int format;			// the format of the captured packets (MFF_PCAP or MFF_DAG_ERF)
	unsigned char is_started;	// this should be 1 if the device is started ( via mapi_start_offline_device() ), 0 otherwise
} offline_device;

static unsigned count_offline_devices = 0;	// used in offline_device_list (always increases)

extern flist_t *function_list;			// list which contains all applyed functions ( defined in mapiipc.c )
extern flist_t *offline_device_list;		// list which contains all offline devices ( defined in mapiipc.c )

#endif

extern flist_t *flowlist;	// defined in mapiipc.c

typedef struct functdescr {
  int fid;
  short result_init;
  mapidflib_function_def_t *def;
  mapidflib_function_t *funct;
  void *data;  
  mapi_results_t *result;
#ifdef RECONNECT
  flowdescr_t *flow;
  int numfd;
#endif
} functdescr_t;

typedef struct shm_result {
  void* ptr; //Pointer to shared data
  int size; //Size of shared data
} shm_result_t;

/*
 * Function declarations 
 */
static int
default_read_result_init(flowdescr_t *flow,functdescr_t* f,void* data);
int 
get_results_info(flowdescr_t *flow, functdescr_t *f, int user_fd, int user_fid);

//static int set_error(void* flow, int err_no , int is_remote);

static int send_fd(int *fds,int numfd);

//global var access functions
static int get_numflows();
static int incr_numflows();
static int decr_numflows();
static int get_totalflows();
static int incr_totalflows();

static void init()
//common initialization function for mapi and dimapi
{
#ifdef DIMAPI
  char* path=NULL, *libs=NULL, *str=NULL, *s=NULL;
  char* mapi_conf;
  
  mapi_conf = printf_string( CONFDIR"/"CONF_FILE );

  if (pc_load (mapi_conf))
    {
      conf_category_entry_t* empty_cat = pc_get_category("");
      const char *portstr;

      if( empty_cat == NULL ){
          printf("Configuration file has no empty category. Giving up\n");
          exit(1);
      }
      path = pc_get_param (empty_cat, "libpath");
      libs = pc_get_param (empty_cat, "libs");
      portstr = pc_get_param (empty_cat, "dimapi_port");
      if( portstr == NULL ){
          printf("ERROR: Configuration file has no entry for `dimapi_port'. Using default port %d\n", DEFAULT_DIMAPI_PORT);
	  dimapi_port = DEFAULT_DIMAPI_PORT;
      }
      else {
	      /* make sure that portstr is a valid number. */
	      dimapi_port = atoi( portstr );
	      if ( dimapi_port<=0 || dimapi_port>=65536 ) {
		      printf("ERROR: Invalid port given in configuration file. The default port %d is used\n", DEFAULT_DIMAPI_PORT);
		      dimapi_port = DEFAULT_DIMAPI_PORT;
	      }
      }
    }
  else 
    {
      printf("ERROR: Cannot load mapi.conf file. Giving up.\n");
      printf("Search path is: %s\n", mapi_conf);
      exit(1);
    }
#endif

  minit = 1;
  pthread_spin_init(&numflows_lock, PTHREAD_PROCESS_PRIVATE);
  
  flowlist = malloc(sizeof(flist_t));
  flist_init(flowlist);

#ifdef RECONNECT
  function_list = malloc(sizeof(flist_t));
  flist_init(function_list);
  
  offline_device_list = malloc(sizeof(flist_t));
  flist_init(offline_device_list);
#endif

#ifdef DIMAPI
  pthread_spin_init(&remote_ipc_lock, PTHREAD_PROCESS_PRIVATE);
  pthread_spin_init(&hostlist_lock, PTHREAD_PROCESS_PRIVATE);
  hostlist = malloc(sizeof(flist_t));
  remote_flowlist = malloc(sizeof(flist_t));
  flist_init(hostlist); 
  flist_init(remote_flowlist);

  str=libs;
  while((s=strchr(str,':'))!=NULL) {
    *s='\0';
    mapilh_load_library(path,str);
    str=s+1;
  }

  mapilh_load_library(path,str);
  free(mapi_conf);
  pc_close();
#endif
}

//Initializes MAPI - called only once by pthread_once()
static void mapi_init()
{
  struct mapiipcbuf qbuf;
  char libpath[4096],*str,*s;
  minit=1;
  if(mapiipc_client_init()==-1){
	 local_err = MCOM_INIT_SOCKET_ERROR;
	 minit=0;
	 fprintf(stderr, "\n--------------------------------------------------------\n");
	 fprintf(stderr, "WARNING: mapid may not be running at the given interface\n");
	 fprintf(stderr,"--------------------------------------------------------\n");
  }
  pthread_once(&initialized, (void*)init);
  pthread_spin_init(&mapi_lock, PTHREAD_PROCESS_PRIVATE);
  
  offline_devices = 0;
  
  //Get libpath from mapid
  qbuf.mtype=1;
  qbuf.cmd=GET_LIBPATH;
  qbuf.fd=getpid();
  qbuf.pid=getpid();
  pthread_spin_lock(&mapi_lock);
  
 if( mapiipc_write((struct mapiipcbuf*)&qbuf))
	   	local_err = MCOM_SOCKET_ERROR;
 if( mapiipc_read((struct mapiipcbuf*)&qbuf) )
	   	local_err = MCOM_SOCKET_ERROR;
 
  pthread_spin_unlock(&mapi_lock);
  switch(qbuf.cmd) {
      case GET_LIBPATH_ACK:
	  strncpy(libpath, (char *)qbuf.data, 4096);
	  break;
      default:
	   /* MAPI_ERROR_GETTING_LIBPATH */
	  return;
	  break;
  }
  printf("libpath=%s\n", libpath);
		  
  //get libs from mapid
  qbuf.mtype=1;
  qbuf.cmd=GET_LIBS;
  qbuf.fd=getpid();
  qbuf.pid=getpid();
  pthread_spin_lock(&mapi_lock);
  if(mapiipc_write((struct mapiipcbuf*)&qbuf))
	   	local_err = MCOM_SOCKET_ERROR;
 if( mapiipc_read((struct mapiipcbuf*)&qbuf))
	   	local_err = MCOM_SOCKET_ERROR;
  pthread_spin_unlock(&mapi_lock);
  switch(qbuf.cmd) {
      case GET_LIBS_ACK:
	  break;
      default:
 	  /* MAPI_ERROR_GETTING_LIBS */
	  return; 
	  break;
  }

  //Load function libraries
    str = (char *)qbuf.data;
    while((s=strchr(str,':'))!=NULL) {
      *s='\0';
      mapilh_load_library(libpath,str);
      str=s+1;
    }

    mapilh_load_library(libpath,str);
    return;
}

#ifdef DIMAPI
//Initializes DIMAPI - called only once by pthread_once()
static void dmapi_init()
{ 
  pthread_once(&initialized, (void*)init);
  return;
}
#endif

int mapi_connect(int fd)
//Connect to a mapi flow
//fd = flow descriptor
{
  struct mapiipcbuf qbuf;
  flowdescr_t* flow;

#ifdef DIMAPI
  remote_flowdescr_t* rflow;
  host_flow* hflow;
  flist_node_t* fnode;
#endif

  if (!minit) {
    printf("MAPI not initialized! [%s:%d]\n", __FILE__, __LINE__);
    local_err = MAPI_INIT_ERROR;
    return -1;
  }else if (fd<=0){
   printf("ERROR: Invalid flow descriptor (fd: %d) in mapi_connect\n", fd);
   return -1;	  
  }

#ifdef DIMAPI
  if ((rflow=flist_get(remote_flowlist,fd))!=NULL) {//flow is remote
    for (fnode=flist_head(rflow->host_flowlist); fnode!=NULL; fnode=flist_next(fnode)) {

      hflow=(host_flow*)fnode->data;
      hflow->dbuf->cmd=CONNECT;
      hflow->dbuf->fd=hflow->fd;
      hflow->dbuf->length=BASIC_SIZE;
    }
#ifdef WITH_AUTHENTICATION
    if(rflow->is_authenticated == 0)
    {
	   printf("ERROR: Flow with id %d is not authenticated\n", fd);
	   rflow->is_connected = 0;
	   return(-2);
    }
    
#endif
    if (mapiipc_remote_write_to_all(rflow)<0){ 
	local_err = MCOM_SOCKET_ERROR;
	return -1;
    }

    //wait results

    for (fnode=flist_head(rflow->host_flowlist); fnode!=NULL; fnode=flist_next(fnode)) {
      hflow=(host_flow*)fnode->data;

      switch(hflow->dbuf->cmd) {
        case CONNECT_ACK:
		rflow->is_connected=1;
          continue;
        case ERROR_ACK:
	  memcpy(&local_err, hflow->dbuf->data, sizeof(int));
          return -1;
        default:
	  local_err = MCOM_UNKNOWN_ERROR;
          return -1;
     }

    }
    return 0;
  }
#endif

  if ((flow=flist_get(flowlist,fd))==NULL) {
    printf("ERROR: Invalid flow %d [%s:%d]\n", fd, __FILE__, __LINE__);
    local_err = MAPI_INVALID_FLOW;
    return -1;
  }

  qbuf.mtype=1;
  qbuf.cmd=CONNECT;
#ifdef RECONNECT
  qbuf.fd = flow->fd;
  qbuf.user_fd = fd;
#else
  qbuf.fd=fd;
#endif
  qbuf.pid=getpid();
  pthread_spin_lock(&mapi_lock);
  if (mapiipc_write((struct mapiipcbuf*)&qbuf)<0) {
	  pthread_spin_unlock(&mapi_lock);	  
	  local_err = MCOM_SOCKET_ERROR;
	  return -1;
  }
  if (mapiipc_read((struct mapiipcbuf*)&qbuf)<0) {
	  pthread_spin_unlock(&mapi_lock);
	  local_err = MCOM_SOCKET_ERROR;
	  return -1;
  }
  pthread_spin_unlock(&mapi_lock);
  switch(qbuf.cmd)
    {
    case CONNECT_ACK:
	flow->is_connected=1;
      return 0;
    case ERROR_ACK:
      local_err =  qbuf.remote_errorcode;
      return -1;
    default:
      local_err= MCOM_UNKNOWN_ERROR;
      return -1;
    }  
}

#ifdef DIMAPI
static void delete_remote_flow(remote_flowdescr_t* rflow) 
{
  host_flow* hflow;
  flist_node_t* fnode, *fnode2, *fnode3;
  mapi_results_t* res;

  pthread_spin_lock(&remote_ipc_lock);
  flist_remove(remote_flowlist, rflow->fd);
  decr_numflows();
  pthread_spin_unlock(&remote_ipc_lock);

  sem_destroy(&rflow->fd_sem);
  sem_destroy(&rflow->pkt_sem);
  //pthread_mutex_destroy(&rflow->mutex);

  for (fnode=flist_head(rflow->host_flowlist); fnode!=NULL; fnode=fnode2 ) {
    fnode2=flist_next(fnode);
    hflow=(host_flow*)fnode->data;
    hflow->rhost->num_flows--;
    flist_remove(hflow->rhost->flows, hflow->fd);
    for (fnode3=flist_head(hflow->functions); fnode3!=NULL; fnode3=flist_next(fnode3)) {
      free( flist_remove(hflow->rhost->functions, ((function_data*)fnode3->data)->fid) );
    }
    if (hflow->rhost->num_flows==0) {
      pthread_cancel(*hflow->rhost->comm_thread);
      mapiipc_remote_close(hflow->rhost);	//close the socket
      flist_destroy(hflow->rhost->flows);
      free(hflow->rhost->flows);
      flist_destroy(hflow->rhost->functions);
      free(hflow->rhost->functions);
      free(hflow->rhost->hostname);
      flist_remove(hostlist, hflow->rhost->sockfd);
      free(hflow->rhost->comm_thread);
#ifdef RECONNECT
      sem_destroy(&hflow->rhost->connection);	// destroy semaphore
#endif
      free(hflow->rhost);
    }
    //we check if a host is using in other rflows and delete it -close the socket- if not
    flist_destroy(hflow->functions);
    free(hflow->functions);
    free(hflow->dev);
    free(hflow->dbuf);
    if (hflow->pkt!=NULL) free(hflow->pkt);
    flist_remove(rflow->host_flowlist,hflow->id);
    free(hflow);
  }
  flist_destroy(rflow->host_flowlist);
  free(rflow->host_flowlist);
  if (rflow->pkt_list!=NULL) {
	flist_destroy(rflow->pkt_list);
	free(rflow->pkt_list);
  }
  for (fnode=flist_head(rflow->function_res); fnode!=NULL; fnode=fnode2 ) { 
	fnode2=flist_next(fnode);
	res=(mapi_results_t*)fnode->data;
	free(res->res);
	free(res);
  }
  flist_destroy(rflow->function_res);
  free(rflow->function_res);
  if (rflow->pkt!=NULL) free(rflow->pkt);
#ifdef WITH_AUTHENTICATION
  if (rflow->username!=NULL) free(rflow->username);
  if (rflow->vo!=NULL) free(rflow->vo);
#ifdef RECONNECT
  if (rflow->password!=NULL) free(rflow->password);
#endif
#endif  
  free(rflow);
}
#endif

int mapi_create_flow(const char *dev)
//Create new flow
//dev=device that should be used
{
  struct mapiipcbuf qbuf;
  flowdescr_t *flow, *tmpflow;
  struct devgroupdb *devgroupdb;
  int devgroupid = 0;
  char *mapidsocket, *mapidsocketglobal;
/*  int local;*/

#ifdef DIMAPI
  remote_flowdescr_t *rflow;
  char *hostname=NULL, *s=NULL, *k=NULL;
  struct host *h=NULL;
  host_flow* hflow;
  char *devp;
  flist_node_t* fnode;
  unsigned int idgen=0;
#endif
#ifdef RECONNECT
  offline_device *device = NULL;
  flist_node_t *fnode_;
  int flag = 0;
#endif

  if(dev==NULL){
	 printf("ERROR: Wrong device name given (NULL) in mapi_create_flow\n");
	 local_err  = MAPI_DEVICE_INFO_ERR;
	 return -1;
  }

#ifndef DIMAPI
	devgroupdb = devgroupdb_open(3); // try local, global
	devgroupid = devgroupdb_getgroupidbydevice(devgroupdb, (char *) dev);

	if(!devgroupid) {
		mapidsocket = printf_string(MAPIDSOCKHOME, getenv("HOME"));
  	mapidsocketglobal = strdup(MAPIDSOCKGLOBAL);
  }
	else {
		mapidsocket = printf_string(MAPIDGSOCKHOME, getenv("HOME"), devgroupid);
		mapidsocketglobal = printf_string(MAPIDGSOCKGLOBAL, devgroupid);
	}

	mapiipc_set_socket_names(mapidsocket, mapidsocketglobal);

  pthread_once(&mapi_is_initialized, (void*)mapi_init);
#endif

  //check if flow is remote or not and call the appropriate init function
#ifdef DIMAPI
  if ( strchr(dev,':')==NULL) {
		devgroupdb = devgroupdb_open(3); // try local, global
		devgroupid = devgroupdb_getgroupidbydevice(devgroupdb, (char *) dev);

		if(!devgroupid) {
			mapidsocket = printf_string(MAPIDSOCKHOME, getenv("HOME"));
			mapidsocketglobal = strdup(MAPIDSOCKGLOBAL);
		}
		else {
			mapidsocket = printf_string(MAPIDGSOCKHOME, getenv("HOME"), devgroupid);
			mapidsocketglobal = printf_string(MAPIDGSOCKGLOBAL, devgroupid);
		}

		mapiipc_set_socket_names(mapidsocket, mapidsocketglobal);

  	pthread_once(&mapi_is_initialized, (void*)mapi_init);
	}
  else pthread_once(&dmapi_is_initialized, (void*)dmapi_init);

  if ((s = strchr(dev,':'))!=NULL) {
    devp=strdup(dev);
    rflow=(remote_flowdescr_t *)malloc(sizeof(remote_flowdescr_t));
    rflow->fd=++fdseed;
    sem_init(&rflow->fd_sem, 0, 0);
    sem_init(&rflow->pkt_sem, 0, 0);
    rflow->host_flowlist=(flist_t*)malloc(sizeof(flist_t));
    flist_init(rflow->host_flowlist);
    rflow->pkt_list=NULL;
    rflow->function_res=(flist_t*)malloc(sizeof(flist_t));
    rflow->is_connected = 0;
#ifdef WITH_AUTHENTICATION
    rflow->username=NULL;
    rflow->vo=NULL;
#ifdef RECONNECT
    rflow->password = NULL;
    rflow->daemons_down = 0;    
#endif
#endif
    rflow->pkt=NULL;
    flist_init(rflow->function_res);
    k=strtok(devp, ", ");

    while (k!=NULL) {
      if ((s = strchr(k,':'))!=NULL) {
        *s = '\0';
        hostname = k;
        k = s + 1;
        pthread_spin_lock(&hostlist_lock);	
        h = (struct host *)flist_search(hostlist, hostcmp, hostname);

        if(h==NULL){// Our host is a new one --> insert it in the hostlist
          h = (struct host *)malloc(sizeof(struct host));
	  h->hostname = strdup(hostname);
	  h->port = dimapi_port;
	  h->flows = (flist_t *)malloc(sizeof(flist_t));
	  flist_init(h->flows);
	  h->functions = (flist_t *)malloc(sizeof(flist_t));
	  flist_init(h->functions);
	  h->num_flows=0;
	  h->stats=NULL;
#ifdef RECONNECT
      	  sem_init(&h->connection, 0, 0);	// initialize semaphore
#endif
	  // Create the socket
	  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__);
		  pthread_spin_unlock(&hostlist_lock);
		  return -1;
	  }

	  h->comm_thread=(pthread_t *)malloc(sizeof(pthread_t));
	  pthread_create(h->comm_thread, NULL, *mapiipc_comm_thread, h);

          flist_append(hostlist, h->sockfd, h);
          pthread_spin_unlock(&hostlist_lock);
        }
        else{//host exists in the list
          pthread_spin_unlock(&hostlist_lock);
        }

        h->num_flows++;
	hflow=(host_flow*)malloc(sizeof(host_flow));
	hflow->scope_fd=rflow->fd;
	hflow->dev=strdup(k);
	flist_append(rflow->host_flowlist, ++idgen, hflow);
	hflow->id=idgen;
	flist_append(h->flows, --negfdseed, hflow);
	hflow->fd=negfdseed;
	hflow->dbuf=(struct dmapiipcbuf *)malloc(sizeof(struct dmapiipcbuf));
	hflow->pkt=NULL;
	hflow->rhost=h;
	hflow->functions=(flist_t *)malloc(sizeof(flist_t));
	flist_init(hflow->functions);

	hflow->dbuf->cmd=CREATE_FLOW;
	strncpy((char *)hflow->dbuf->data,k,DATA_SIZE);
	hflow->dbuf->length=BASIC_SIZE+strlen(k)+1;
      }
      else {
        //this is the case where the dev string contains both 'host:interface1' and 'interface2'
	//example: mapi_create_flow("139.91.70.98:eth0, 147.52.16.102:eth0, eth1");
	//user's intention is probably localhost:eth1
	//what should be done in this case?
      }
      k=strtok(NULL,", ");
    }

    free(devp);

    rflow->scope_size=flist_size(rflow->host_flowlist);
    pthread_spin_lock(&remote_ipc_lock);
    flist_append(remote_flowlist, rflow->fd, rflow);
    incr_numflows();
    incr_totalflows();
    pthread_spin_unlock(&remote_ipc_lock);

    if (mapiipc_remote_write_to_all(rflow)<0){
	    local_err = MCOM_SOCKET_ERROR;
	    return -1;
    }
    //sends to all hosts of rflow the proper dbuf, increment the pending_msgs makes sem_wait(rflow->fd_sem) and the comm_thread will get the results - the hflow->fd for every flow -

    //wait for results

    for (fnode=flist_head(rflow->host_flowlist); fnode!=NULL; fnode=flist_next(fnode)) {
      hflow=(host_flow*)fnode->data;
      if (hflow->dbuf->cmd == CREATE_FLOW_ACK) {
        hflow->fd=*((int*)hflow->dbuf->data);
	flist_remove(hflow->rhost->flows, hflow->dbuf->fd);
	flist_append(hflow->rhost->flows, hflow->fd, hflow);
      }
      else if (hflow->dbuf->cmd == ERROR_ACK) {
	memcpy(&local_err, hflow->dbuf->data, sizeof(int));
	printf("ERROR: Could not create flow in host %s [%s:%d]\n", hflow->rhost->hostname, __FILE__, __LINE__);
        delete_remote_flow(rflow);
	return -1;
      }
      else {
	local_err = MCOM_UNKNOWN_ERROR;
        delete_remote_flow(rflow);
	return -1;
      }
    }

    return rflow->fd;

  }
#endif

  pthread_spin_lock(&mapi_lock);
  
  if ((get_numflows() == 0) && (get_totalflows() > 0) && minit) { // socket has been closed, re-create it
    if(mapiipc_client_init()==-1){
	  local_err = MCOM_INIT_SOCKET_ERROR;
    }
    incr_numflows();
  }
  else 
    incr_numflows();
  
  pthread_spin_unlock(&mapi_lock);

#ifdef RECONNECT
  /* consider the following case :
     	- devicename = mapi_create_offline_device("./tracefile", MFF_PCAP);
	- {
		code ...
		--> reconnection
		code ...
	  }
	  fd = mapi_create_flow(devicename);

	  Now mapi_create_flow must send to mapid the new device name ...
  */

  pthread_spin_lock(&mapi_lock);

  for(fnode_ = flist_head(offline_device_list); fnode_ != NULL; fnode_ = flist_next(fnode_)){	// find specified device in offline_device_list

	  device = (offline_device *)fnode_->data;

	  if(!strcmp(device->previous_device, dev)){
		  flag = 1;					// FIXME
		  break;
	  }
  }
  if(flag)	strncpy((char *) qbuf.data, device->new_device, DATA_SIZE);	// the new device name
  else		strncpy((char *) qbuf.data, dev, DATA_SIZE);

  flag = 0;

  pthread_spin_unlock(&mapi_lock);
#else
  strncpy((char *)qbuf.data,dev,DATA_SIZE);
#endif

  qbuf.mtype=1;
  qbuf.cmd=CREATE_FLOW;
  qbuf.fd=getpid();
  qbuf.pid=getpid();
  pthread_spin_lock(&mapi_lock);

  if (mapiipc_write((struct mapiipcbuf*)&qbuf)<0) {
	  local_err = MCOM_SOCKET_ERROR;
	  pthread_spin_unlock(&mapi_lock);	  
   	  decr_numflows();
	  return -1;
  }
  if (mapiipc_read((struct mapiipcbuf*)&qbuf)<0) {
	  local_err = MCOM_SOCKET_ERROR;
	  pthread_spin_unlock(&mapi_lock);
	  decr_numflows();
	  return -1;
  }
  pthread_spin_unlock(&mapi_lock);
  switch(qbuf.cmd)
    {
    case CREATE_FLOW_ACK:
      tmpflow=flist_get(flowlist,qbuf.fd);
      if (tmpflow!=NULL) 
        {
	        printf("ERROR: Mapid gave us a fd (%d) which already exist in our lists, exiting [%s:%d]\n", qbuf.fd, __FILE__, __LINE__);
	        decr_numflows();
		return -1;
     	}
      flow=malloc(sizeof(flowdescr_t));
      if( flow == NULL ){
	  printf("ERROR: Out of memory [%s:%d]\n", __FILE__, __LINE__);
	  decr_numflows();
	  return -1;
      }
      flow->fd=qbuf.fd;
      flow->devtype=(char *)malloc(strlen((char *)qbuf.data)+1);
      flow->flist=malloc(sizeof(flist_t));
#ifdef RECONNECT
      flow->device = strdup(dev);
      flow->format = -1;	// in case of online flow, assigned to -1
      flow->read_results_flag = 0;
      flow->mapid_down = 0;
#endif
      flow->shm_base=NULL;
      flow->shm_spinlock=NULL;
      flow->file = -1;		// in case of online flow, assigned to -1
      flow->is_connected =0;
      flow->numfd = 0;		// initialize number of open file descriptors to zero
      flist_init(flow->flist);
      strcpy(flow->devtype,(char *)qbuf.data);
      pthread_spin_lock(&mapi_lock);
      flist_append(flowlist,qbuf.fd,flow);
      incr_totalflows();
#ifdef DIMAPI
      fdseed++;
#endif
      pthread_spin_unlock(&mapi_lock);
      return qbuf.fd;
      
    /* should probably have a separate error message for ERROR_ACK? */
    case ERROR_ACK:
      decr_numflows();
      local_err=qbuf.remote_errorcode;
      return -1;
    default:
      decr_numflows();
      local_err=MCOM_UNKNOWN_ERROR;
      return -1;
    }
}

int mapi_create_offline_flow(const char *dev, int format)
//Create new flow
//dev=device that should be used
{
  struct mapiipcbuf qbuf;
  flowdescr_t *flow=NULL;
  int file;
  struct devgroupdb *devgroupdb;
  int devgroupid = 0;
  char *mapidsocket, *mapidsocketglobal;

	devgroupdb = devgroupdb_open(3); // try local, global

	devgroupid = devgroupdb_getgroupid(devgroupdb); // get groupid of the latest instance of mapid

	if(!devgroupid) {
		mapidsocket = printf_string(MAPIDSOCKHOME, getenv("HOME"));
		mapidsocketglobal = strdup(MAPIDSOCKGLOBAL);
	}
	else {
		mapidsocket = printf_string(MAPIDGSOCKHOME, getenv("HOME"), devgroupid);
		mapidsocketglobal = printf_string(MAPIDGSOCKGLOBAL, devgroupid);
	}

	mapiipc_set_socket_names(mapidsocket, mapidsocketglobal);

  pthread_once(&mapi_is_initialized, (void*)mapi_init);

  //Check to see if file can be opened
  if (dev==NULL){
	  printf("ERROR: NULL device in mapi_create_offline_flow\n");
	  return -1;
  }
  else if ((file=open(dev,O_LARGEFILE))==-1) {
    local_err=MAPI_ERROR_FILE;
    return -1;
  }
  
  pthread_spin_lock(&mapi_lock);
  if ((get_numflows() == 0) && (get_totalflows() > 0) && minit) { // socket has been closed, re-create it
    if (mapiipc_client_init()<0) {
	    local_err = MCOM_INIT_SOCKET_ERROR;
	    pthread_spin_unlock(&mapi_lock);
	    return -1;
    }
    incr_numflows();
  }
  else
  	incr_numflows();

  pthread_spin_unlock(&mapi_lock);

  qbuf.mtype=1;
  qbuf.cmd=CREATE_OFFLINE_FLOW;
  qbuf.fd=getpid();
  qbuf.pid=getpid();
  qbuf.fid=format;
  strncpy((char *)qbuf.data,dev,DATA_SIZE);
  pthread_spin_lock(&mapi_lock);
  if (mapiipc_write((struct mapiipcbuf*)&qbuf)<0) {
	  pthread_spin_unlock(&mapi_lock);	 
	  local_err = MCOM_SOCKET_ERROR;
	  decr_numflows();
	  return -1;
  }
  if (mapiipc_read((struct mapiipcbuf*)&qbuf)<0) {
	  pthread_spin_unlock(&mapi_lock);
	  local_err = MCOM_SOCKET_ERROR;
	  decr_numflows();
	  return -1;
  }
  if(qbuf.cmd==SEND_FD) {
    if(mapiipc_send_fd(file)==-1) {
      local_err=MAPI_ERROR_SEND_FD;
      pthread_spin_unlock(&mapi_lock);
      decr_numflows();
      return -1;
    }
  } else {
    local_err=MAPI_ERROR_SEND_FD;
    pthread_spin_unlock(&mapi_lock);
    decr_numflows();
    return -1;
  }

  if (mapiipc_read((struct mapiipcbuf*)&qbuf)<0) {
	  pthread_spin_unlock(&mapi_lock);
	  local_err = MCOM_SOCKET_ERROR;
	  decr_numflows();
	  return -1;
  }
  pthread_spin_unlock(&mapi_lock);
  switch(qbuf.cmd)
    {
    case CREATE_OFFLINE_FLOW_ACK:
      flow=malloc(sizeof(flowdescr_t));
      flow->fd=qbuf.fd;
      flow->devtype=(char *)malloc(strlen((char *)qbuf.data)+1);
      flow->flist=malloc(sizeof(flist_t));
#ifdef RECONNECT
      flow->device = strdup(dev);
      flow->format = format;	// MFF_PCAP or MFF_DAG_ERF
#endif
      flow->shm_base=NULL;
      flow->shm_spinlock=NULL;
      flow->file = file;	// file descriptor of offline flow
      flow->is_connected=0;
      flow->numfd = 0;		// initialize number of open file descriptors to zero      
      flist_init(flow->flist);
      strcpy(flow->devtype,(char *)qbuf.data);
      pthread_spin_lock(&mapi_lock);
      flist_append(flowlist,qbuf.fd,flow);
      incr_totalflows();
#ifdef DIMAPI
      fdseed++;
#endif
      pthread_spin_unlock(&mapi_lock);
      return qbuf.fd;
    case ERROR_ACK:
      local_err=qbuf.remote_errorcode;
      decr_numflows();
      return -1;
    default:
      local_err=MCOM_UNKNOWN_ERROR;
      decr_numflows();
      return -1;
    }
}

char* mapi_create_offline_device(const char *path, int format)
//Create new flow
//dev=device that should be used
{
  struct mapiipcbuf qbuf;
  int file;
#ifdef RECONNECT
  offline_device *device = NULL;
#endif

  char *mapidsocket, *mapidsocketglobal;

	mapidsocket = printf_string(MAPIDSOCKHOME, getenv("HOME"));
 	mapidsocketglobal = strdup(MAPIDSOCKGLOBAL);

	mapiipc_set_socket_names(mapidsocket, mapidsocketglobal);

  pthread_once(&mapi_is_initialized, (void*)mapi_init);

  //Check to see if file can be opened
  
  if (path==NULL){
	  printf("ERROR: NULL path in mapi_create_offline_device\n");
	  return NULL;
  }
  else if ((file=open(path,O_LARGEFILE))==-1) {
    local_err=MAPI_ERROR_FILE;
    return NULL;
  }

  pthread_spin_lock(&mapi_lock);
  if ((get_numflows() == 0) && (get_totalflows() > 0) && minit) { // socket has been closed, re-create it
    if (mapiipc_client_init()<0) {
	    pthread_spin_unlock(&mapi_lock);
	    local_err = MCOM_INIT_SOCKET_ERROR;
	    return NULL;
    }
  }
  pthread_spin_unlock(&mapi_lock);

  qbuf.mtype=1;
  qbuf.cmd=CREATE_OFFLINE_DEVICE;
  qbuf.fd=getpid();
  qbuf.pid=getpid();
  qbuf.fid=format;
  strncpy((char *)qbuf.data,path,DATA_SIZE);
  pthread_spin_lock(&mapi_lock);
  if (mapiipc_write((struct mapiipcbuf*)&qbuf)<0) {
	  pthread_spin_unlock(&mapi_lock);	  
	  local_err = MCOM_SOCKET_ERROR;
	  return NULL;
  }
  if (mapiipc_read((struct mapiipcbuf*)&qbuf)<0) {
	  pthread_spin_unlock(&mapi_lock);
	  local_err = MCOM_SOCKET_ERROR;
	  return NULL;
  }
  if(qbuf.cmd==SEND_FD) {
    if(mapiipc_send_fd(file)==-1) {
      local_err=MAPI_ERROR_SEND_FD;
      pthread_spin_unlock(&mapi_lock);
      return NULL;      
    }
  } else {
    local_err=MAPI_ERROR_SEND_FD;
    pthread_spin_unlock(&mapi_lock);
    return NULL;
  }

  if (mapiipc_read((struct mapiipcbuf*)&qbuf)<0) {
	  pthread_spin_unlock(&mapi_lock);
	  local_err = MCOM_SOCKET_ERROR;
	  return NULL;
  }
  pthread_spin_unlock(&mapi_lock);
  switch(qbuf.cmd)
    {
    case CREATE_OFFLINE_DEVICE_ACK:

#ifdef RECONNECT
      device = (offline_device *)malloc(sizeof(offline_device));
      device->path = strdup(path);
      device->previous_device = strdup((char *) qbuf.data);	// previous device name
      device->new_device = strdup((char *) qbuf.data);		// new device name, after reconnection of client
      device->format = format;
      device->is_started = 0;
      pthread_spin_lock(&mapi_lock);
      flist_append(offline_device_list, count_offline_devices++, device);
      pthread_spin_unlock(&mapi_lock);
#endif
      offline_devices++;

      return strdup((char *)qbuf.data);
    case ERROR_ACK:
      local_err=qbuf.remote_errorcode;
      return NULL;
    default:
      local_err=MCOM_UNKNOWN_ERROR;
      return NULL;
    }
}

int mapi_start_offline_device(const char *dev)
//Create new flow
//dev=device that should be used
{
  struct mapiipcbuf qbuf;
#ifdef RECONNECT
  offline_device *device = NULL;
  flist_node_t *fnode;
#endif

  pthread_once(&mapi_is_initialized, (void*)mapi_init);

  if (dev==NULL){
	  printf("ERROR: NULL device in mapi_start_offline_device\n");
	  return -1;
  }
  pthread_spin_lock(&mapi_lock);
  if ((get_numflows() == 0)