com-library/libvcacom/vca_com.c
int vca_com_recv_msg(vca_com_t * com, 
		     vca_com_addr * src, 
		     char * msg,
		     unsigned long long * length,
		     unsigned int * c) {
  
  char sspace[4096];
  void * buffer = (void*) sspace;
  unsigned long long task_len = 0;
  vca_com_msg_hdr * hdr = NULL;
  int rc = 0;


  if(!com || !src || !msg || !c || length == 0) {
    return -1;
  }

  if (*length > 4096) {
    buffer = malloc(*length + sizeof(vca_com_msg_hdr));
  }

  switch(com->type) {
  case VCA_COM_MEM_SHARING:
     rc = vca_recv_task(com->com, (long*) &task_len, buffer, *c); 
     break;
  case VCA_COM_MEM_SHARING_HOST:
     rc = host_recv_task(com->com, (long*) &task_len, buffer, (int*) c);
     break;
  case VCA_COM_ZMQ_SOCKET:
    {
      vca_com_zmqs * zmqs = (vca_com_zmqs*)com->com;
      char tmpid[sizeof(zmqs->id)];
      do {
        rc = zmq_recv(zmqs->socket, tmpid, sizeof(tmpid), 0);
        if(rc < 0 || memcmp(zmqs->id, tmpid, zmqs->id_size))
  	  break;
        rc = zmq_recv(zmqs->socket, buffer, *length, 0);
        if(rc < 0)
	  break;
        task_len = (unsigned long long) rc; // set length of request
      } while (rc < 1); // to step over connection establishment
      if (rc > 0)
	rc = 0; // to emulate shared memory return semantics
      break;
    }
  }
 
  if(rc == 0) {
    // received a message
    hdr = vca_com_msg_get_hdr(buffer);
    vca_com_cpy_addr(&hdr->src, src);

    // check that task fits
    if(*length < task_len) {
	return -1;
    }
    
    *length = task_len - sizeof(vca_com_msg_hdr);
    memcpy(msg, vca_com_msg_get_msg(buffer), *length);
  }

  if(buffer != sspace) {
    free(buffer); // free if previously allocated
  }

  return rc;
}
// receive next message from src via com on channel c
receive next message from src via com on channel c
receive next message from src via com on channel c
int vca_com_recv_msg(vca_com_t * com, 
		     vca_com_addr * src, 
		     char * msg,
		     unsigned long long * length,
		     unsigned int * c) {
  char sspace[4096];
  void * buffer = (void*) sspace;
  unsigned long long task_len = 0;
  vca_com_msg_hdr * hdr = NULL;
  int rc = 0;
  if(!com || !src || !msg || !c || length == 0) {
    return -1;
  }
  if (*length > 4096) {
    buffer = malloc(*length + sizeof(vca_com_msg_hdr));
  }
  switch(com->type) {
  case VCA_COM_MEM_SHARING:
     rc = vca_recv_task(com->com, (long*) &task_len, buffer, *c); 
     break;
  case VCA_COM_MEM_SHARING_HOST:
     rc = host_recv_task(com->com, (long*) &task_len, buffer, (int*) c);
     break;
  case VCA_COM_ZMQ_SOCKET:
    {
      vca_com_zmqs * zmqs = (vca_com_zmqs*)com->com;
      char tmpid[sizeof(zmqs->id)];
      do {
        rc = zmq_recv(zmqs->socket, tmpid, sizeof(tmpid), 0);
        if(rc < 0 || memcmp(zmqs->id, tmpid, zmqs->id_size))
  	  break;
        rc = zmq_recv(zmqs->socket, buffer, *length, 0);
        if(rc < 0)
	  break;
        task_len = (unsigned long long) rc; 
      } while (rc < 1); 
      if (rc > 0)
	rc = 0; 
      break;
    }
  }
  if(rc == 0) {
    hdr = vca_com_msg_get_hdr(buffer);
    vca_com_cpy_addr(&hdr->src, src);
    if(*length < task_len) {
	return -1;
    }
    *length = task_len - sizeof(vca_com_msg_hdr);
    memcpy(msg, vca_com_msg_get_msg(buffer), *length);
  }
  if(buffer != sspace) {
    free(buffer); 
  }
  return rc;
}
int vca_com_recv_msg ( vca_com_t * com , vca_com_addr * src , char * msg , unsigned long long * length , unsigned int * c ) { char sspace [ 4096 ] ; void * buffer = ( void * ) sspace ; unsigned long long task_len = 0 ; vca_com_msg_hdr * hdr = NULL ; int rc = 0 ; if ( ! com || ! src || ! msg || ! c || length == 0 ) { return -1 ; } if ( * length > 4096 ) { buffer = malloc ( * length + sizeof ( vca_com_msg_hdr ) ) ; } switch ( com -> type ) { case VCA_COM_MEM_SHARING : rc = vca_recv_task ( com -> com , ( long * ) & task_len , buffer , * c ) ; break ; case VCA_COM_MEM_SHARING_HOST : rc = host_recv_task ( com -> com , ( long * ) & task_len , buffer , ( int * ) c ) ; break ; case VCA_COM_ZMQ_SOCKET : { vca_com_zmqs * zmqs = ( vca_com_zmqs * ) com -> com ; char tmpid [ sizeof ( zmqs -> id ) ] ; do { rc = zmq_recv ( zmqs -> socket , tmpid , sizeof ( tmpid ) , 0 ) ; if ( rc < 0 || memcmp ( zmqs -> id , tmpid , zmqs -> id_size ) ) break ; rc = zmq_recv ( zmqs -> socket , buffer , * length , 0 ) ; if ( rc < 0 ) break ; task_len = ( unsigned long long ) rc ; } while ( rc < 1 ) ; if ( rc > 0 ) rc = 0 ; break ; } } if ( rc == 0 ) { hdr = vca_com_msg_get_hdr ( buffer ) ; vca_com_cpy_addr ( & hdr -> src , src ) ; if ( * length < task_len ) { return -1 ; } * length = task_len - sizeof ( vca_com_msg_hdr ) ; memcpy ( msg , vca_com_msg_get_msg ( buffer ) , * length ) ; } if ( buffer != sspace ) { free ( buffer ) ; } return rc ; }
receive next message from src via com on channel c
receive next message from src via com on channel c
// set length of request // to step over connection establishment // to emulate shared memory return semantics // received a message // check that task fits // free if previously allocated