mapiipc.c

#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <errno.h>
#include <sys/ipc.h>
#include <sys/msg.h>
#include <sys/socket.h>
#include <sys/types.h>
#include <sys/stat.h>
#include <sys/un.h>
#include <string.h>
#include <errno.h>

#include "mapiipc.h"
#include "debug.h"

#define HAVE_MSGHDR_MSG_CONTROL 1

#ifdef DIMAPI
#include <netdb.h>
#include <netinet/in.h>
#include <semaphore.h>
#include "flist.h"
#endif

// this file contains all the client-side IPC related functions

static int sock;
static int mapidaddr_len;
static struct sockaddr_un mapidaddr;
//static struct sockaddr_un fromaddr;
static char* mapidsocket;

int mapiipc_write(struct mapiipcbuf *qbuf)
//Sends an IPC message to mapid
{
  qbuf->uid=getuid();
  if(send(sock, qbuf, sizeof(struct mapiipcbuf), 0) == -1) {
    WARNING_CMD(printf("send: %s [%s:%d]\n",strerror(errno),__FILE__,__LINE__));
    //    exit(1);
    return -1;
  }
  return 0;
}

int mapiipc_read(struct mapiipcbuf *qbuf)
//Reads an IPC message. Blocking call
{
  if(recv(sock, qbuf, MAX_SEND_SIZE, 0) == -1){
    ERROR_CMD(printf("recv: %s [%s:%d]\n",strerror(errno),__FILE__,__LINE__));
    //exit(1);
    return -1;
  }
  return 0;
}

int mapiipc_client_init()
//Initializes IPC for mapi functions
{
  if ((sock = socket(AF_LOCAL, SOCK_STREAM, 0)) == -1) {
    ERROR_CMD(printf("socket: %s [%s:%d]\n",strerror(errno),__FILE__,__LINE__));
    //exit(1);
    return -1;
  }

  mapidsocket=malloc(sizeof(MAPIDSOCKHOME)-2+strlen(getenv("HOME")));
  sprintf(mapidsocket,MAPIDSOCKHOME,getenv("HOME")); 

  // Construct name of mapid's socket
  mapidaddr.sun_family = AF_LOCAL;
  strcpy(mapidaddr.sun_path, mapidsocket);
  mapidaddr_len = sizeof mapidaddr.sun_family + strlen(mapidaddr.sun_path);

  if (connect(sock, (struct sockaddr *)&mapidaddr, mapidaddr_len) < 0) 
  {
	//	ERROR_CMD(printf("connect: %s [%s:%d]\n",strerror(errno),__FILE__,__LINE__));
      free(mapidsocket);
      mapidsocket=strdup(MAPIDSOCKGLOBAL);
      strcpy(mapidaddr.sun_path, mapidsocket);
      mapidaddr_len = sizeof mapidaddr.sun_family + strlen(mapidaddr.sun_path);
      if (connect(sock, (struct sockaddr *)&mapidaddr, mapidaddr_len) < 0)
    	ERROR_CMD(printf("connect: %s [%s:%d]\n",strerror(errno),__FILE__,__LINE__));
      else {
	free(mapidsocket);
      	//exit(EXIT_FAILURE);
      	return -1;
      }
  }
  free(mapidsocket);
  return 0;
}

void mapiipc_client_close()
//Releases socket resources
{
  close(sock);
}

#ifdef DIMAPI
struct sockaddr_in remoteaddr;
flist_t *remote_flowlist=NULL;

int mapiipc_remote_write(struct dmapiipcbuf *dbuf, struct host *h)
//Sends an IPC message to mapid
{
//  qbuf->uid=getuid(); 
  if(send(h->sockfd, dbuf, dbuf->length, 0) == -1) {
    WARNING_CMD(printf("send: %s [%s:%d]\n",strerror(errno),__FILE__,__LINE__));
    //    exit(1);
    return -1;
  }
  return 0;
}

int mapiipc_remote_write_to_all(remote_flowdescr_t* rflow)
{
  host_flow* hflow;
  flist_node_t* fnode;

  pthread_mutex_lock(&rflow->mutex);
  rflow->pending_msgs=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) return -1;
    rflow->pending_msgs++;
  }
  pthread_mutex_unlock(&rflow->mutex);
  sem_wait(&rflow->fd_sem);
  return 0;
}

void *mapiipc_comm_thread(void *host) {
//Reads an IPC message. Blocking call
  //char buffer[DIMAPI_DATA_SIZE];	DELETE
  struct dmapiipcbuf* dbuf;
  remote_flowdescr_t* rflow;
  host_flow* hflow;
  int recv_bytes;
  //unsigned int dbuf_bytes=0;		DELETE
  int pending;
  int sockfd=((struct host *)host)->sockfd;

  /* Guarantees that thread resources are deallocated upon return */
  pthread_detach(pthread_self());

  //dbuf=(struct dmapiipcbuf*)&buffer[0];	DELETE
  dbuf = (struct dmapiipcbuf *)malloc(sizeof(struct dmapiipcbuf));  

  while (1) {

      if (host==NULL) break;
    /*if (dbuf_bytes==0 || (dbuf->length<DIMAPI_DATA_SIZE && dbuf_bytes<dbuf->length)) {
      if (host==NULL) return NULL;
      recv_bytes=recv(sockfd, (char*)&buffer+dbuf_bytes, DIMAPI_DATA_SIZE, 0);
      if (recv_bytes == -1) {
	//ERROR_CMD(printf("recv: %s [%s:%d]\n",strerror(errno),__FILE__,__LINE__));
	//exit(1);
	//printf("error in recv\n");
	continue ;
      }
      else if (recv_bytes == 0) {
	//printf("Socket closed\n");
	//exit(1);
	return NULL;
      }

      dbuf_bytes += recv_bytes; 
    }

    if (dbuf_bytes<dbuf->length) continue;*/

    recv_bytes=readn(sockfd, dbuf, BASIC_SIZE);

    if (recv_bytes == 0) { // the peer has gone
	//printf("Socket closed\n");
	break;
    }
    else if (recv_bytes == -1) {
	//ERROR_CMD(printf("recv: %s [%s:%d]\n",strerror(errno),__FILE__,__LINE__));
	continue;
    }
    
    if (dbuf->length > DIMAPI_DATA_SIZE) {
	fprintf(stderr,"Bad IPC message from agent\n");
	continue;
    }
    
    if (dbuf->length-BASIC_SIZE>0) {
	recv_bytes=readn(sockfd, (char*)dbuf+BASIC_SIZE, dbuf->length-BASIC_SIZE);
	    
	if (recv_bytes == 0) { // the peer has gone
		//printf("Socket closed\n");
		break;
	}
	else if (recv_bytes == -1) {
		//ERROR_CMD(printf("recv: %s [%s:%d]\n",strerror(errno),__FILE__,__LINE__));
		continue;
	}
    }

    hflow=(host_flow*)flist_get( ((struct host*)host)->flows, dbuf->fd );
    if (hflow!=NULL) {
      rflow=flist_get(remote_flowlist, hflow->scope_fd);
      if (dbuf->cmd==GET_NEXT_PKT_ACK) {
		memcpy(hflow->pkt, dbuf->data, dbuf->length-BASIC_SIZE);
		flist_append(rflow->pkt_list, 0, hflow);
		sem_post(&rflow->pkt_sem);
      }
      else {
	memcpy( hflow->dbuf, dbuf, dbuf->length );	//place data
	pthread_mutex_lock(&rflow->mutex);
	//if (rflow->pending_msgs>0) {
	--rflow->pending_msgs;
	pending=rflow->pending_msgs;
	pthread_mutex_unlock(&rflow->mutex);

	if ( pending==0 ) {
	  sem_post( &rflow->fd_sem );
	}
      	//}
	//else pthread_mutex_unlock(&rflow->mutex);
      }
    }
    else {
      fprintf(stderr,"Invalid IPC message, unknown fd %d\n",dbuf->fd);
      //exit(-1);
      //failure
      continue;
    }

    //dbuf_bytes=dbuf_bytes-dbuf->length;		DELETE
    //memcpy(buffer,buffer+dbuf->length,dbuf_bytes);	DELETE
  }

  free(dbuf);
  return NULL;
}


int mapiipc_remote_init(struct host *h)
//Initializes IPC for dmapi functions
{
  struct hostent* host=gethostbyname(h->hostname);
  struct timeval tv;

  if ((h->sockfd = socket(AF_INET, SOCK_STREAM, 0)) == -1) {
    ERROR_CMD(printf("socket: %s [%s:%d]\n",strerror(errno),__FILE__,__LINE__));
    //exit(-1);
    return -1;
  }

  /*
  tv.tv_sec=20;		//timeout 20 sec for recv
  tv.tv_usec=0;

  if (setsockopt(h->sockfd, SOL_SOCKET, SO_RCVTIMEO, &tv, sizeof(struct timeval)) == -1) {
    close(h->sockfd);
    printf("Unexpected error on setsockopt()");
    //exit(-1);
    return -1;
  }
  */

  tv.tv_sec=10;		//timeout 10 sec for send
  tv.tv_usec=0;

  if (setsockopt(h->sockfd, SOL_SOCKET, SO_SNDTIMEO, &tv, sizeof(struct timeval)) == -1) {
    close(h->sockfd);
    printf("Unexpected error on setsockopt()");
    //exit(-1);
    return -1;
  }

  if (host==NULL) {
    printf("Could not determine address for %s\n",h->hostname);
    //exit(1);
    return -1;
  }

  // Construct name of dmapid's socket
  remoteaddr.sin_family = AF_INET;
  remoteaddr.sin_addr = *((struct in_addr *)host->h_addr);
  remoteaddr.sin_port = htons(h->port);

  if (connect(h->sockfd, (struct sockaddr *)&remoteaddr, sizeof(remoteaddr)) < 0) {     
    ERROR_CMD(printf("connect failed: %s [%s:%d]\n",strerror(errno),__FILE__,__LINE__));
    //exit(EXIT_FAILURE);
    return -1;
  }
  return 0;

}

void mapiipc_remote_close(struct host *h)
//Releases socket resources
{
  shutdown(h->sockfd, SHUT_RDWR);
  close(h->sockfd);
}
#endif


// Helper functions for function arguments retrieval
int getargint(mapiFunctArg **pos){
	int i;
	i = *((int *)(*pos));
	//	printf("getint: %d\n", i);
	(*pos) += sizeof(int);
	return i;
}

char getargchar(mapiFunctArg **pos){
	char c;
	c = *((char *)(*pos));
	//printf("getchar: %c\n", c);
	(*pos) += sizeof(char);
	return c;
}

unsigned long long getargulonglong(mapiFunctArg **pos){
	unsigned long long l;
	l = *((unsigned long long *)(*pos));
	//printf("getulonglong: %lld\n", l);
	(*pos) += sizeof(unsigned long long);
	return l;
}

char * getargstr(mapiFunctArg **pos){
	char *s;
	s = (char*)*pos;
	//printf("getstr: %s\n", s);
	(*pos) += strlen(s)+1;
	return s;
}

void addarg(mapiFunctArg **pos, void *arg, int type)
// Helper function for mapi_apply_function()
// pos: current position in message's argument buffer.
// arg: argument to copy into buffer
// type: argument type
{
  switch(type){
    case INT:
      memcpy(*pos, arg, sizeof(int));
      //printf("add_arg: %d\n", *((int *)(*pos)));
      (*pos) += sizeof(int);
      break;
    case CHAR:
      memcpy(*pos, arg, sizeof(char));
      //printf("add_arg: %d\n", *((char *)(*pos)));
      (*pos) += sizeof(char);
      break;
    case UNSIGNED_LONG_LONG:
      memcpy(*pos, arg, sizeof(unsigned long long));
      //printf("add_arg: %llu\n", *((unsigned long long *)(*pos)));
      (*pos) += sizeof(unsigned long long);
      break;
    case STRING:
      memcpy(*pos, arg, strlen((char *)arg)+1);
      //printf("add_arg: %s\n", (char *)(*pos));
      (*pos) += strlen((char *)arg)+1;
      break;
    default:
      break;
  }
}

int mapiipc_send_fd(int sendfd)
{
  struct msghdr	msg;
  struct iovec	iov[1];
  char ptr[2];
  int ret;

#ifdef	HAVE_MSGHDR_MSG_CONTROL
  union {
    struct cmsghdr	cm;
    char control[CMSG_SPACE(sizeof(int))];
  } control_un;
  struct cmsghdr	*cmptr;
  
  msg.msg_control = control_un.control;
  msg.msg_controllen = sizeof(control_un.control);
  
  cmptr = CMSG_FIRSTHDR(&msg);
  cmptr->cmsg_len = CMSG_LEN(sizeof(int));
  cmptr->cmsg_level = SOL_SOCKET;
  cmptr->cmsg_type = SCM_RIGHTS;
  *((int *) CMSG_DATA(cmptr)) = sendfd;
#else
  msg.msg_accrights = (caddr_t) &sendfd;
  msg.msg_accrightslen = sizeof(int);
#endif
    
  iov[0].iov_base = ptr;
  iov[0].iov_len = 2;
  msg.msg_iov = iov;
  msg.msg_iovlen = 1;
  msg.msg_name = NULL;
  msg.msg_namelen = 0;

  ret=sendmsg(sock,&msg,0);
  return(ret);
}


int mapiipc_read_fd(int sock)
{
  struct msghdr	msg;
  struct iovec	iov[1];
  ssize_t n;
  int recvfd;
  char c[2];
  
#ifdef	HAVE_MSGHDR_MSG_CONTROL
  union {
    struct cmsghdr	cm;
    char				control[CMSG_SPACE(sizeof(int))];
  } control_un;
  struct cmsghdr	*cmptr;
  
  msg.msg_control = control_un.control;
  msg.msg_controllen = sizeof(control_un.control);
#else
  msg.msg_accrights = (caddr_t) &newfd;
  msg.msg_accrightslen = sizeof(int);
#endif
  
  iov[0].iov_base = &c;
  iov[0].iov_len = 2;
  msg.msg_iov = iov;
  msg.msg_iovlen = 1;
  msg.msg_name = NULL;
  msg.msg_namelen = 0;
  
  
  if ( (n = recvmsg(sock, &msg, 0)) <= 0)
    return(n);
  
#ifdef	HAVE_MSGHDR_MSG_CONTROL
  if ( (cmptr = CMSG_FIRSTHDR(&msg)) != NULL &&
       cmptr->cmsg_len == CMSG_LEN(sizeof(int))) {
    if (cmptr->cmsg_level != SOL_SOCKET) {
      // err_quit("control level != SOL_SOCKET");
      fprintf(stderr, "control level != SOL_SOCKET");
      return -1;
    }
    if (cmptr->cmsg_type != SCM_RIGHTS) {
      // err_quit("control type != SCM_RIGHTS");
      fprintf(stderr, "control type != SCM_RIGHTS");
      return -1;
    }
    recvfd = *((int *) CMSG_DATA(cmptr));
  } else
    recvfd = -1;		/* descriptor was not passed */
#else
  /* *INDENT-OFF* */
  if (msg.msg_accrightslen == sizeof(int))
    recvfd = newfd;
  else
    recvfd = -1;		/* descriptor was not passed */
  /* *INDENT-ON* */
#endif

  return recvfd;
}

/* Read "n" bytes from a socket. */
ssize_t readn(int fd, void *vptr, size_t n) {
        size_t nleft;
        ssize_t nread;
        char *ptr;

        ptr = vptr;
        nleft = n;
        while (nleft > 0) {
		errno=0;
                if ( (nread = read(fd, ptr, nleft)) < 0) {
                        if (errno == EINTR)
                                nread = 0;              /* and call read() again */
                        else
                                return(-1);
                } else if (nread == 0)
                        return 0;                          /* EOF */

                nleft -= nread;
                ptr   += nread;
        }
        return(n - nleft);              /* return >= 0 */
}