mapi.c

#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 "mapierror.h"

#include "mapi_errors.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[];

/*
//Structure used as a linked list to store information about
//registered functions
 struct functiondescr {
  char* function;
  int fd;
  int fid;
  void* ptr;
  int ptrsize;
  char* pktbuf;
  struct mapid_to_buffer *to_buffer;
  struct functiondescr *next;
  };

//Linked list of registered functions
static struct functiondescr *functions=NULL;
*/

#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;

typedef struct function_data {
	int fid;
	mapidflib_function_def_t* fdef;
} 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
  char *devtype;
  char *shm_base;
  pthread_spinlock_t *shm_spinlock;
  flist_t *flist;
/*  
  int error;
  char errstr[MAPI_ERRORSTR_LENGTH];
*/
  unsigned char is_connected; // This should be 1 if the flow is connected 0 otherwise
} flowdescr_t;

flist_t *flowlist=NULL;

typedef struct functdescr {
  int fid;
  short result_init;
  mapidflib_function_def_t *def;
  mapidflib_function_t *funct;
  void *data;  
  mapi_results_t *result;
} 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);

//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
 #define CONF_FILE "./mapi.conf:%s/.mapi.conf:/etc/mapi.conf"
  char* path=NULL, *libs=NULL, *str=NULL, *s=NULL;
  char* mapi_conf;
  mapi_conf = printf_string(CONF_FILE,getenv("HOME"));
  //init_error("/root/trunk/mapi_errors.dat");
  if (pc_load (mapi_conf))
    {
      conf_category_entry_t* empty_cat = pc_get_category("");
      const char *portstr;

      if( empty_cat == NULL ){
          fputs( "Configuration file has no empty category. Giving up.\n", stderr );
          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 ){
          fprintf(stderr,"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 ) {
		      fprintf(stderr, "Invalid port given in configuration file, the default port %d is used\n",DEFAULT_DIMAPI_PORT);
		      dimapi_port = DEFAULT_DIMAPI_PORT;
	      }
      }
    }
  else 
    {
      fputs("Error: cannot load mapi.conf file. Giving up.\n", stderr);
      fprintf( stderr, "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 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;
  }
  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;
  }
  DEBUG_CMD(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) {
    DEBUG_CMD(printf("Not initialized! [%s:%d]\n",__FILE__,__LINE__));
    return -1;
  }else if (fd<=0){
   //could not use set_error since the flow is not connected and may
   //cause seg fault
   fprintf(stderr,"Invalid Flow descriptor [%s:%d]\n",__FILE__,__LINE__); 
   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)
    {
	   fprintf(stderr, "Flow with id %d is not authenticated. Aborting. [%s:%d]\n", fd, __FILE__, __LINE__);
	   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:
	  local_err = MAPI_CONNECT; //is remote flow
	  //printf("Error! mapi_connect did not work!\n");
          return -1;
        default:
	  local_err = MAPI_CONNECT;
          //printf("Error! mapi_connect did not work!\n");
          return -1;
     }

    }
    return 0;
  }
#endif

  if ((flow=flist_get(flowlist,fd))==NULL) {
    DEBUG_CMD(printf("Invalid flow: %d [%s:%d]\n",fd,__FILE__,__LINE__));
    local_err = MAPI_INVALID_FLOW;
    return -1;
  } /*else if (flow->error!=0) {
    DEBUG_CMD(printf("Invalid flow: %d due to error #%d [%s:%d]\n",fd,flow->error,__FILE__,__LINE__));
    local_err = MAPI_INVALID_FLOW;
    return -1;
  }*/
 
  qbuf.mtype=1;
  qbuf.cmd=CONNECT;
  qbuf.fd=fd;
  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;

  pthread_spin_lock(&remote_ipc_lock);
  flist_remove(remote_flowlist, rflow->fd,FLIST_LEAVE_DATA);
  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, FLIST_LEAVE_DATA);
    if (hflow->rhost->num_flows==0) {
      mapiipc_remote_close(hflow->rhost);	//close the socket
      //pthread_kill(*hflow->rhost->comm_thread, 9);
      flist_destroy(hflow->rhost->flows, FLIST_LEAVE_DATA);
      free(hflow->rhost->flows);
      free(hflow->rhost->hostname);
      pthread_spin_lock(&hostlist_lock);
      flist_remove(hostlist, hflow->rhost->sockfd, FLIST_LEAVE_DATA);
      pthread_spin_unlock(&hostlist_lock);
      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, FLIST_FREE_DATA);
    free(hflow->functions);
    free(hflow->dev);
    free(hflow->dbuf);
    flist_remove(rflow->host_flowlist,hflow->id,FLIST_FREE_DATA);
  }
  if (rflow->pkt_list!=NULL) flist_destroy(rflow->pkt_list, FLIST_LEAVE_DATA);
  free(rflow->pkt);
  flist_destroy(rflow->function_res, FLIST_FREE_DATA);	
  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;

  if(dev==NULL){
	 fprintf(stderr, "Error wrong device name given \n\n");
	 local_err  = MAPI_DEVICE_INFO_ERR;
	 return -1;
  }
#ifdef DIMAPI
  remote_flowdescr_t *rflow;
  char *hostname=NULL, *s=NULL, *k=NULL;
  struct host *h=NULL;
  host_flow* hflow;
  char *devp=strdup(dev);
  flist_node_t* fnode;
  unsigned int idgen=0;
#endif
  //if(!minit)
    //  if((ret=mapi_init())!=0)
	//  return ret;
  
  //check if flow is remote or not and call the appropriate init function
#ifndef DIMAPI
  pthread_once(&mapi_is_initialized, (void*)mapi_init);
#endif

#ifdef DIMAPI
  if ( strchr(dev,':')==NULL) pthread_once(&mapi_is_initialized, (void*)mapi_init);
  else pthread_once(&dmapi_is_initialized, (void*)dmapi_init);

  if ((s = strchr(dev,':'))!=NULL) {
    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);
    pthread_mutex_init(&rflow->mutex, NULL);
    rflow->pending_msgs=0;
    rflow->host_flowlist=(flist_t*)malloc(sizeof(flist_t));
    flist_init(rflow->host_flowlist);
    rflow->pkt_list=NULL;
    rflow->pkt=(struct mapipkt*)malloc(sizeof(struct mapipkt)+PKT_LENGTH);
    rflow->function_res=(flist_t*)malloc(sizeof(flist_t));
    rflow->is_connected = 0;
    flist_init(rflow->function_res);
    k=strtok(devp, ", ");

    while (k!=NULL) {
      if ((s = strchr(k,':'))!=NULL) {
        *s = '\0';
        hostname = k;
        k = s + 1;
        //printf("host: %s - device: %s\n",hostname, k);
        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);
	  //printf("New host %s\n",hostname);
	  h->port = dimapi_port;
	  h->flows = (flist_t *)malloc(sizeof(flist_t));
	  flist_init(h->flows);
	  h->num_flows=0;
	  // Create the socket
	  if (mapiipc_remote_init(h)<0) {
	   	  local_err = MCOM_SOCKET_ERROR;
		  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);
	  //printf("New communication thread created\n");

          flist_append(hostlist, h->sockfd, h);
          pthread_spin_unlock(&hostlist_lock);
        }
        else{//host exists in the list
          //printf("%s host again\n",h->hostname);
          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);
	flist_append(h->flows, --negfdseed, hflow);
	hflow->fd=negfdseed;
	hflow->dbuf=(struct dmapiipcbuf *)malloc(sizeof(struct dmapiipcbuf));
	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 && *((int*)hflow->dbuf->data)!=-1) {
        hflow->fd=*((int*)hflow->dbuf->data);
	flist_remove(hflow->rhost->flows, hflow->dbuf->fd, FLIST_LEAVE_DATA);
	flist_append(hflow->rhost->flows, hflow->fd, hflow);
	printf("New host flow created with fd %d for rflow %d\n",hflow->fd,rflow->fd);
      }
      else {//error (fd==-1)
        delete_remote_flow(rflow);
	return -1;
      }
    }

    return rflow->fd;

  }
#endif

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

  qbuf.mtype=1;
  qbuf.cmd=CREATE_FLOW;
  qbuf.fd=getpid();
  qbuf.pid=getpid();
  strncpy((char *)qbuf.data,dev,DATA_SIZE);
  pthread_spin_lock(&mapi_lock);
  if (mapiipc_write((struct mapiipcbuf*)&qbuf)<0) {
	  local_err = MCOM_SOCKET_ERROR;
	  pthread_spin_unlock(&mapi_lock);	  
	  return -1;
  }
  if (mapiipc_read((struct mapiipcbuf*)&qbuf)<0) {
	  local_err = MCOM_SOCKET_ERROR;
	  pthread_spin_unlock(&mapi_lock);
	  return -1;
  }
  pthread_spin_unlock(&mapi_lock);
  switch(qbuf.cmd)
    {
    case CREATE_FLOW_ACK:
      tmpflow=flist_get(flowlist,qbuf.fd);
      if (tmpflow!=NULL) 
        {
	        ERROR_CMD(fprintf(stderr,"mapid gave us a fd which already exist in our lists (%d), exiting [%s:%d]\n",
	            qbuf.fd,__FILE__,__LINE__));
	        //exit(EXIT_FAILURE);
		return -1;
     	}
      flow=malloc(sizeof(flowdescr_t));
      if( flow == NULL ){
	  fputs( "Out of memory\n", stderr );
	  //exit( EXIT_FAILURE );
	  return -1;
      }
      flow->fd=qbuf.fd;
      flow->devtype=(char *)malloc(strlen((char *)qbuf.data)+1);
      flow->flist=malloc(sizeof(flist_t));
      flow->shm_base=NULL;
      flow->shm_spinlock=NULL;
      //flow->error=0;
      //flow->errstr[0]='\0';
      flow->is_connected =0;
      flist_init(flow->flist);
      strcpy(flow->devtype,(char *)qbuf.data);
      pthread_spin_lock(&mapi_lock);
      flist_append(flowlist,qbuf.fd,flow);
      incr_numflows();
      incr_totalflows();
      pthread_spin_unlock(&mapi_lock);
      return qbuf.fd;
      
    /* should probably have a separate error message for ERROR_ACK? */
    case ERROR_ACK:
      local_err=qbuf.remote_errorcode;
      return -1;
    default:
      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;

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

  //Check to see if file can be opened
  if (dev==NULL){
	  fprintf(stderr,"Error NULL device in mapi_create_offline_flow\n\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;
    }
  }
  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;
	  return -1;
  }
  if (mapiipc_read((struct mapiipcbuf*)&qbuf)<0) {
	  pthread_spin_unlock(&mapi_lock);
	  local_err = MCOM_SOCKET_ERROR;
	  return -1;
  }
  if(qbuf.cmd==SEND_FD) {
    if(mapiipc_send_fd(file)==-1) {
      local_err=MAPI_ERROR_SEND_FD;
      return -1;
    }
  } else {
    local_err=MAPI_ERROR_SEND_FD;
    pthread_spin_unlock(&mapi_lock);	  
    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 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));
      flow->shm_base=NULL;
      flow->shm_spinlock=NULL;
      //flow->error=0;
      //flow->errstr[0]='\0';
      flow->is_connected=0;
      flist_init(flow->flist);
      strcpy(flow->devtype,(char *)qbuf.data);
      pthread_spin_lock(&mapi_lock);
      flist_append(flowlist,qbuf.fd,flow);
      incr_numflows();
      incr_totalflows();
      pthread_spin_unlock(&mapi_lock);
      return qbuf.fd;
    case ERROR_ACK:
      local_err=qbuf.remote_errorcode;
      return -1;
    default:
      local_err=MCOM_UNKNOWN_ERROR;
      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;

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

  //Check to see if file can be opened
  
  if (path==NULL){
	  fprintf(stderr,"Error NULL path in mapi_create_offline_device\n\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
//    pthread_spin_lock(&mapi_lock);
    if (mapiipc_client_init()<0) {
	    pthread_spin_unlock(&mapi_lock);
	    local_err = MCOM_INIT_SOCKET_ERROR;
	    return NULL;
    }
//    pthread_spin_unlock(&mapi_lock);
  }
  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:
      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;

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

  if (dev==NULL){
	  fprintf(stderr,"Error NULL device in mapi_start_offline_device\n\n");
	  return -1;
  }
  pthread_spin_lock(&mapi_lock);
  if ((get_numflows() == 0) && (get_totalflows() > 0) && minit) { // socket has been closed, re-create it
    pthread_spin_lock(&mapi_lock);
    if (mapiipc_client_init()<0) {
	    pthread_spin_unlock(&mapi_lock);
	    local_err = MCOM_INIT_SOCKET_ERROR;
	    return -1;
    }
    pthread_spin_unlock(&mapi_lock);
  }
  pthread_spin_unlock(&mapi_lock);

  qbuf.mtype=1;
  qbuf.cmd=START_OFFLINE_DEVICE;
  qbuf.fd=getpid();
  qbuf.pid=getpid();
  strncpy((char *)qbuf.data,dev,DATA_SIZE);
  qbuf.fid=0;
  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 START_OFFLINE_DEVICE_ACK:
      return 0;
    case ERROR_ACK:
      local_err=qbuf.remote_errorcode;
      return -1;
    default:
      local_err=MCOM_UNKNOWN_ERROR;
      return -1;
    }
}

int mapi_delete_offline_device(char *dev)
//Create new flow
//dev=device that should be used
{
  struct mapiipcbuf qbuf;
  printf("closing down...\n");
  pthread_once(&mapi_is_initialized, (void*)mapi_init);

  if (dev==NULL){
	  fprintf(stderr,"Error NULL device in mapi_delete_offline_device\n\n");
	  return -1;
  }
/* socet is not closed
  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) return -1;
  }
  pthread_spin_unlock(&mapi_lock);
*/

  qbuf.mtype=1;
  qbuf.cmd=DELETE_OFFLINE_DEVICE;
  qbuf.fd=getpid();
  qbuf.pid=getpid();
  strncpy((char *)qbuf.data,dev,DATA_SIZE);
  free(dev);
  qbuf.fid=0;
  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 DELETE_OFFLINE_DEVICE_ACK:
      pthread_spin_lock(&mapi_lock);
  	  if((get_numflows() == 0) && --offline_devices == 0)
    	mapiipc_client_close();
  	  pthread_spin_unlock(&mapi_lock);
      return 0;
    case ERROR_ACK:
      local_err=qbuf.remote_errorcode;
      return -1;
    default:
      local_err=MCOM_UNKNOWN_ERROR;
      return -1;
    }
}



int mapi_close_flow(int fd) 
{
  //struct mapiipcbuf qbuf;
  functdescr_t *f=NULL;
  flowdescr_t *flow=NULL;

  if (fd<=0){
	  fprintf(stderr,"Error wrong fd in mapi_close_flow\n\n");
	  return -1;
  }
#ifdef DIMAPI
  remote_flowdescr_t *rflow=flist_get(remote_flowlist, fd);
  host_flow* hflow;  
  flist_node_t* fnode;

  if(rflow!=NULL){
    for (fnode=flist_head(rflow->host_flowlist); fnode!=NULL; fnode=flist_next(fnode)) {
      hflow=(host_flow*)fnode->data;
      hflow->dbuf->cmd=CLOSE_FLOW;
      hflow->dbuf->fd=hflow->fd;
      hflow->dbuf->length=BASIC_SIZE;
    }

    //if (mapiipc_remote_write_to_all(rflow)<0) return -1;
    pthread_mutex_lock(&rflow->mutex);
    rflow->pending_msgs=0;
    rflow->is_connected=0;
    for (fnode=flist_head(rflow->host_flowlist); fnode!=NULL; fnode=flist_next(fnode)) {
      hflow=(host_flow*)fnode->data;
      hflow->dbuf->fd=hflow->fd;
      if (mapiipc_remote_write(hflow->dbuf, hflow->rhost)<0) {
	  pthread_mutex_unlock(&rflow->mutex);
	  local_err = MCOM_SOCKET_ERROR;
	  return -1;
      }
      rflow->pending_msgs++;
    }
    pthread_mutex_unlock(&rflow->mutex);

    //no need to wait for results
    /*for (fnode=flist_head(rflow->host_flowlist); fnode!=NULL; fnode=flist_next(fnode)) {
      hflow=(host_flow*)fnode->data;
      switch(hflow->dbuf->cmd){
        case CLOSE_FLOW_ACK:
	      break;
        case ERROR_ACK:
              local_err=MCOM_ERROR_ACK;
              break;
        default:
	      local_err=MCOM_UNKNOWN_ERROR;
              break;
      }
    }*/

    delete_remote_flow(rflow);
    return 0;
  }
#endif

  if(flowlist && (flow = flist_get(flowlist, fd))==NULL){
    DEBUG_CMD(printf("Invalid flow: %d [%s:%d]\n",fd,__FILE__,__LINE__));
    local_err = MAPI_INVALID_FLOW;
    return -1;
  }
  else
  {
    pthread_spin_lock(&mapi_lock);
    //XXX MEMORY LEAK why leave data since the node is removed????
    flow = flist_remove(flowlist,fd,FLIST_LEAVE_DATA);
    pthread_spin_unlock(&mapi_lock);
  }
	
  //Delete functions applied first, before mapid is notified.
  pthread_spin_lock(&mapi_lock);
  while((f = flist_pop_first(flow->flist)) != NULL) {
    if(f->def->client_cleanup!=NULL && f->funct->instance->status==MAPIFUNC_INIT){
      f->def->client_cleanup(f->funct->instance);
    }
    if(f->result!=NULL)
	    free(f->result);
    free(f->funct->instance);
    free(f->funct);
    free(f->data);
    free(f);
  }
  pthread_spin_unlock(&mapi_lock);
  
 /* if (flow->error==0) {
  	qbuf.mtype=1;
   	qbuf.cmd=CLOSE_FLOW;
  	qbuf.fd=fd;
  	qbuf.fid=getpid();
  	qbuf.pid=getpid();
	pthread_spin_lock(&mapi_lock);
	if (mapiipc_write((struct mapiipcbuf*)&qbuf)<0) {
	  pthread_spin_unlock(&mapi_lock);	  
	  return -1;
	}
	if (mapiipc_read((struct mapiipcbuf*)&qbuf)<0) {
	  pthread_spin_unlock(&mapi_lock);
	  return -1;
	}
	pthread_spin_unlock(&mapi_lock);
  
  	switch(qbuf.cmd)
	  {
	  case CLOSE_FLOW_ACK:
		// if this is the last one, release socket resources
	      break;
	  case ERROR_ACK:
	      local_err=MCOM_ERROR_ACK;
	      break;
	  default:
	      local_err=MCOM_UNKNOWN_ERROR;
	      break;
	  }
  }*/

  //Detach shared mem
  if (flow->shm_base!=NULL) {
    if (shmdt(flow->shm_base)<0) {
      WARNING_CMD(printf("Warning: Could not detach shared mem (%s) [%s:%d]\n",strerror(errno),__FILE__,__LINE__));
    }
  }
  
  /* subtract, the flow should be closed either due to an error or
   * explicitly in this function. so numflows is really number of allocated
   * flows...
   */
  pthread_spin_lock(&mapi_lock);
  if((decr_numflows() == 0) && offline_devices == 0)
    mapiipc_client_close();
  pthread_spin_unlock(&mapi_lock);

  //Free flow resources
  free(flow->devtype);
  free(flow->flist);
  free(flow);  
  
  return 0;
}
//XXX Why send_fd returns 0 on error whereas every other function 
//return -1 on error?
static int send_fd(int *fds, int num) {
  int c;
  struct mapiipcbuf qbuf;

  for(c=0;c<num;c++) {
    if (mapiipc_read((struct mapiipcbuf*)&qbuf)<0){
	  local_err = MCOM_SOCKET_ERROR;
	  return 0; 
    }
    if(qbuf.cmd!=SEND_FD)
      return 0;
    if(mapiipc_send_fd(fds[c])==-1)	
	    return 0;
  }
  return 1;
}

int mapi_apply_function(int fd, const char* funct, ...) 
//Apply function to a mapi flow
//fd: flow descriptor
//funct: function to be added
{
  struct mapiipcbuf qbuf;
  int fds[256];
  int numfd=0;
  //struct functiondescr* f=malloc(sizeof(struct functiondescr));
  mapidflib_function_def_t *fdef;
  functdescr_t *f;
  mapiFunctArg *pos;
  int tmp;
  unsigned long long ltmp;
  char ctmp;
  //  char* stemp;
  //char* sbuff;
  va_list vl;
  flowdescr_t *flow;
  char *argdescr_ptr;
  char *filename;
  char *temp;
  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;
  remote_flowdescr_t* rflow;
  host_flow* hflow;
  function_data *fdata;
  flist_node_t* fnode;
#endif
 

  if (!minit) {
    DEBUG_CMD(printf("Not initialized! [%s:%d]\n",__FILE__,__LINE__));
    local_err = MAPI_INIT_ERROR;
    return -1;
  }
  else if(fd<=0){
    fprintf(stderr,"Error wrong fd in mapi_apply_function\n\n");
    local_err = MAPI_INVALID_FID_FUNCID;
    return -1;
  }
  if(funct==NULL){
    fprintf(stderr,"Error NULL function in mapi_apply_function\n\n");
    local_err = MFUNCT_COULD_NOT_APPLY_FUNCT;
    return -1;
  }
  
#ifdef DIMAPI
  if ((rflow=flist_get(remote_flowlist,fd))!=NULL) {

	  if(rflow->is_connected){
		printf("\nERROR can not apply function on an already connected flow\n");
    		//TODO when mapi_set_flow can support remote flows enable the folowing
		local_err = MAPI_FLOW_NOT_CONNECTED;
		return -1;
	}
    //we create a dummy flow descriptor just to get the function info we want from mapilh_get_function_def
    flow=(flowdescr_t *)malloc(sizeof(flowdescr_t));
    flow->devtype="1.3";
    is_remote = 1;//indicates that flow is remote
  }
  else 
#endif
  if ((flow=flist_get(flowlist,fd))==NULL) {
    DEBUG_CMD(printf("Invalid flow: %d [%s:%d]\n",fd,__FILE__,__LINE__));
    local_err = MAPI_INVALID_FLOW;
    return -1;
  } /*else if (flow->error!=0) {
    DEBUG_CMD(printf("Invalid flow: %d due to error #%d [%s:%d]\n",fd,flow->error,__FILE__,__LINE__));
    local_err = MAPI_INVALID_FLOW;
    return -1;
  }*/

  if(flow->is_connected){
	printf("\nERROR can not apply function on an already connected flow\n");
        local_err = MFUNCT_COULD_NOT_APPLY_FUNCT;
	return -1;
  }

  //Get information about function
  fdef=mapilh_get_function_def(funct,flow->devtype);

  if(fdef==NULL) {
    DEBUG_CMD(printf("Could not find/match function %s [%s:%d]\n",funct,__FILE__,__LINE__));
    local_err = MAPI_FUNCTION_NOT_FOUND;
    return -1;
  }

  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