nbio.c

The Staged Event-Driven Architecture (SEDA) is a new design for building scalable Internet services.

  offset = (*env)->GetIntField(env, packet, FID_java_net_DatagramPacket_offset);
  length = (*env)->GetIntField(env, packet, FID_java_net_DatagramPacket_length);
  sz = (*env)->GetArrayLength(env, data); 	

  DEBUG(fprintf(stderr, "NBIO: nbReceive: offset %d, len %d, sz %d\n", offset, length, sz));

  if( (length < 0) || (length > sz) ) {
    THROW_EXCEPTION(env, "java/lang/IllegalArgumentException", "length must be >= 0 and length <= array length");
    return -1;
  }
  if( (offset < 0) || (offset > length) ) {
    THROW_EXCEPTION(env, "java/lang/IllegalArgumentException", "offset must be >=0 and offset <= length");
  }

  fromlength = sizeof(struct sockaddr_in);

  recvdata = (*env)->GetByteArrayElements(env, data, NULL);
  if(recvdata == NULL) {
    THROW_EXCEPTION(env, "java/lang/NullPointerException", "can't access primitive array");
    return -1;
  }
  memset(recvdata, 0, length);
  
  if( (ret = recvfrom(fd, recvdata+offset, length, 0, (struct sockaddr*)&from, &fromlength)) <= 0) {
    // JRVB: added myerrno, to get correct reporting when DEBUG is on
    int myerrno = errno;
    DEBUG(fprintf(stderr,"NBIO: recvfrom returned %d\n", ret));
    // Set length to 0
    (*env)->SetIntField(env, packet, FID_java_net_DatagramPacket_length, 0);
    (*env)->ReleaseByteArrayElements(env, data, recvdata, JNI_ABORT);

    // These 2 cases indicate no data ready to be read
    if (ret == 0) return 0;
    if (myerrno == EAGAIN) return 0; 

    THROW_EXCEPTION(env, "java/net/SocketException", strerror(myerrno));
    return -1;
  }

  DEBUG(fprintf(stderr,"NBIO: recvfrom returned normally %d\n", ret));

  // Set the actual length
  (*env)->SetIntField(env, packet, FID_java_net_DatagramPacket_length, ret);

  // Release the copy of the primitive array
  (*env)->ReleaseByteArrayElements(env, data, recvdata, 0);

  // Set port in packet
  (*env)->SetIntField(env, packet, FID_java_net_DatagramPacket_port, ntohs(from.sin_port));

  // Create empty InetAddress and initialize it 
  DEBUG(fprintf(stderr,"NBIO: nbReceive() creating new InetAddress\n"));
  cls = (*env)->FindClass(env, "java/net/InetAddress");
  if (cls == NULL) {
    THROW_EXCEPTION(env, "java/lang/UnsatisfiedLinkError", "Cannot find java.net.InetAddress class");
    return -1;
  }
  addrobj = (*env)->AllocObject(env, cls);
  if(addrobj == NULL) {
    THROW_EXCEPTION(env, "java/lang/OutOfMemoryError", "Unable to allocate new InetAddress in nbReceive()");
    return -1;
  }

  // Set IP address in addrobj
  (*env)->SetIntField(env, addrobj, FID_java_net_InetAddress_address, ntohl(from.sin_addr.s_addr));
  // Set address in packet
  (*env)->SetObjectField(env, packet, FID_java_net_DatagramPacket_address, addrobj);

  return ret;
}

/*
 * Class:     seda_nbio_NonblockingSocketImpl
 * Method:    nbSendTo
 */
JNIEXPORT jint JNICALL Java_seda_nbio_NonblockingSocketImpl_nbSendTo(JNIEnv *env, jobject this, jobject packet) {

  jbyte* senddata;
  struct sockaddr_in from;
  int fd, ret, sz;
  jbyteArray data;
  jint offset, length, port;
  jobject addr;

  jobject fdobj = (*env)->GetObjectField(env, this, FID_seda_nbio_NonblockingSocketImpl_fd);
  if (fdobj == NULL) {
    THROW_EXCEPTION(env, "java/net/SocketException", "socket closed");
    return -1;
  } 
  fd = (*env)->GetIntField(env, fdobj, FID_seda_nbio_NBIOFileDescriptor_fd);
  if (fd == -1) {
    THROW_EXCEPTION(env, "java/net/SocketException", "socket closed");
    return -1;
  }  

  data = (*env)->GetObjectField(env, packet, FID_java_net_DatagramPacket_buf);
  if(data == NULL) {
    THROW_EXCEPTION(env, "java/lang/NullPointerException", "data buffer is null in nbSendTo");
    return -1;
  }

  offset = (*env)->GetIntField(env, packet, FID_java_net_DatagramPacket_offset);
  length = (*env)->GetIntField(env, packet, FID_java_net_DatagramPacket_length);
  sz = (*env)->GetArrayLength(env, data);

  if( (length < 0) || (length > sz) ) {
    THROW_EXCEPTION(env, "java/lang/IllegalArgumentException", "length must be >= 0 and length <= array length");
    return -1;
  }

  senddata = (*env)->GetByteArrayElements(env, data, NULL);
  if(senddata == NULL) {
    THROW_EXCEPTION(env, "java/lang/NullPointerException", "senddata in nbSendTo is null");
    return -1;
  }

  addr = (*env)->GetObjectField(env, packet, FID_java_net_DatagramPacket_address);
  // If address is null then we are connected and are calling send();
  // otherwise we need to call sendto().
  
  
  if(addr == NULL) {
   DEBUG(fprintf(stderr,"NBIO: send() called, size %d\n", length));
   if( (ret = send(fd, senddata+offset, length, 0)) < 0) {
      (*env)->ReleaseByteArrayElements(env, data, senddata, JNI_ABORT);
      if(errno == EAGAIN) {
	//return 0 indicates unable to send b/c it would block
	return 0;
      } else {
	THROW_EXCEPTION(env, "java/net/SocketException", strerror(errno));
	return -1;
      }
   }
  } else {
    // addr is not null
    memset(&from, 0, sizeof(struct sockaddr_in));
    from.sin_family = AF_INET;
    from.sin_addr.s_addr = htonl( (unsigned long)( (*env)->GetIntField(env, addr, FID_java_net_InetAddress_address) ) );

    port = (*env)->GetIntField(env, packet, FID_java_net_DatagramPacket_port);
    if( (port < 0) || (port > 0xffff) ) {
      THROW_EXCEPTION(env, "java/lang/InvalidArgumentException", "bad port in nbSendTo");
      (*env)->ReleaseByteArrayElements(env, data, senddata, JNI_ABORT);
      return -1;
    }
    from.sin_port = htons((unsigned short)port);

    DEBUG(fprintf(stderr,"NBIO: sendto() called, size %d\n", length));
    if( (ret = sendto(fd, senddata+offset, length, 0, (struct sockaddr*)&from, sizeof(struct sockaddr_in))) <0) {
      (*env)->ReleaseByteArrayElements(env, data, senddata, JNI_ABORT);
      if(errno == EAGAIN) {
	//this would block, return 0
	return 0;
      } else {
	THROW_EXCEPTION(env, "java/net/SocketException", strerror(errno));
	return -1;
      }
    }
  }
  (*env)->ReleaseByteArrayElements(env, data, senddata, JNI_ABORT);
  return ret;
}

/*************************************************************
 * Multicast support code below 
 *************************************************************/

/* helper function */
int mcast_get_fd(JNIEnv *env, jobject this) {
  int fd;
  jobject fdobj;

  fdobj = (*env)->GetObjectField(env, this, FID_seda_nbio_NonblockingSocketImpl_fd);
  if (fdobj == NULL) {
    THROW_EXCEPTION(env, "java/net/SocketException", "socket closed");
    return -1;
  }
  fd = (*env)->GetIntField(env, fdobj, FID_seda_nbio_NBIOFileDescriptor_fd);
  if (fd == -1) {
    THROW_EXCEPTION(env, "java/net/SocketException", "socket closed");
    return -1;
  }
  
  return fd;
}


/* helper function - do a setsockopt for a multicast option */
static void mcast_set_opt(JNIEnv *env, jobject this, jobject address, int opt, char *errmsg) {

  int fd = mcast_get_fd(env,this);
  int addr;
  struct ip_mreq mreq;

  // get the multicast address
  if (address == NULL) {
    THROW_EXCEPTION(env, "java/lang/NullPointerException", "group address is NULL");
    return;
  }
  addr = (*env)->GetIntField(env, address, FID_java_net_InetAddress_address);
  if(opt!=IP_MULTICAST_IF  &&  !IN_MULTICAST(addr)) {
    THROW_EXCEPTION(env, "java/lang/SocketException", "address is not a multicast address");
    return;
  }

  // set the socket option
  mreq.imr_multiaddr.s_addr=htonl(addr);
  mreq.imr_interface.s_addr=htonl(INADDR_ANY); // the value of this field seems to have no effect
  if (setsockopt(fd, IPPROTO_IP, opt, &mreq, sizeof(mreq)) != 0) {
    THROW_EXCEPTION(env, "java/net/SocketException", errmsg);
    return;
  }
}


/*
 * Join multicast group
 */
JNIEXPORT void JNICALL Java_seda_nbio_NonblockingSocketImpl_nbJoinGroup
   (JNIEnv *env, jobject this, jobject address) {
  mcast_set_opt(env, this, address, IP_ADD_MEMBERSHIP, "failed to join multicast group");
}

/*
 * Leave a multicast group
 */
JNIEXPORT void JNICALL Java_seda_nbio_NonblockingSocketImpl_nbLeaveGroup
  (JNIEnv *env, jobject this, jobject address) {
  mcast_set_opt(env, this, address, IP_DROP_MEMBERSHIP, "failed to leave multicast group");
}

/*
 * Set the interface used for this multicast socket
 */
JNIEXPORT void JNICALL Java_seda_nbio_NonblockingSocketImpl_nbSetInterface
  (JNIEnv *env, jobject this, jobject address) {
  mcast_set_opt(env, this, address, IP_MULTICAST_IF, "failed to set multicast interface");
}

/*
 * Get the TTL on multicast packets
 */
JNIEXPORT jint JNICALL Java_seda_nbio_NonblockingSocketImpl_nbGetTimeToLive
  (JNIEnv *env, jobject this) {
  int ttl;
  int len;
  int fd = mcast_get_fd(env,this);

  getsockopt(fd,IPPROTO_IP,IP_MULTICAST_TTL, &ttl,&len);
  return ttl;
}

/*
 * Set the TTL for multicast packets
 */
JNIEXPORT void JNICALL Java_seda_nbio_NonblockingSocketImpl_nbSetTimeToLive
  (JNIEnv *env, jobject this, jint ttl) {
  int fd = mcast_get_fd(env,this);

  setsockopt(fd,IPPROTO_IP,IP_MULTICAST_TTL, &ttl,sizeof(ttl));
}


/*
 * turn receiving your own packets on or off
 */
JNIEXPORT void JNICALL Java_seda_nbio_NonblockingSocketImpl_nbSeeLocalMessages
  (JNIEnv *env, jobject this, jboolean loop_state) {
  int fd = mcast_get_fd(env,this);
  int val = (loop_state ? 1 : 0);

  setsockopt(fd,IPPROTO_IP,IP_MULTICAST_LOOP, &val,sizeof(val));
}



/*
 * Class:     seda_nbio_NonblockingSocketImpl
 * Method:    nbDisconnect
 * Signature: ()V
 */

JNIEXPORT void JNICALL Java_seda_nbio_NonblockingSocketImpl_nbDisconnect
  (JNIEnv *env, jobject this) {
  int fd;
  struct sockaddr_in him;

  jobject fdobj = (*env)->GetObjectField(env, this, FID_seda_nbio_NonblockingSocketImpl_fd);
  if (fdobj == NULL) {
    THROW_EXCEPTION(env, "java/net/SocketException", "socket closed");
    return;
  }
  fd = (*env)->GetIntField(env, fdobj, FID_seda_nbio_NBIOFileDescriptor_fd);
  if (fd == -1) {
    THROW_EXCEPTION(env, "java/net/SocketException", "socket closed");
    return;
  }

#ifdef SOLARIS
  /* to disconnect on Solaris, connect on null address */
  if( connect(fd, (struct sockaddr*)NULL, 0) < 0) {
    THROW_EXCEPTION(env, "java/net/SocketException", strerror(errno));
  }
#endif // ifdef SOLARIS

#if defined(linux) || defined(__FreeBSD__)
  /* to disconnect on Linux or FreeBSD, connect to an addr with
   * family == AF_UNSPEC */
  memset((char *)&him,  0, sizeof(him));
  him.sin_family = AF_UNSPEC;
  if( connect(fd, (struct sockaddr *)&him, sizeof(struct sockaddr_in)) < 0) {
    THROW_EXCEPTION(env, "java/net/SocketException", strerror(errno));
  }
#endif

}

/* SelectSetDevPollImpl ******************************************************/

#ifndef HAS_DEVPOLL

JNIEXPORT jboolean JNICALL Java_seda_nbio_SelectSetDevPollImpl_supported(JNIEnv *env, jclass cls) {

  /* Safe setting when compiling on systems where /dev/poll not available */
  return JNI_FALSE;
}

#else  /* HAS_DEVPOLL */

/* Define to use B-Tree to map fd to selitemobj */
#undef USE_BTREE    
/* Max number of file descriptors if flat table used */
#define MAX_FDS 32768

typedef struct devpoll_impl_state {
  int devpoll_fd;
  int max_retevents;
  struct pollfd *retevents;
#ifdef USE_BTREE
  btree_node *tree;
#else
  jobject selitems[MAX_FDS];
#endif
} devpoll_impl_state;

JNIEXPORT jboolean JNICALL Java_seda_nbio_SelectSetDevPollImpl_supported(JNIEnv *env, jclass cls) {
  struct stat statbuf;

  /* If we were compiled on a system with /dev/poll, but running where
   * it's not available, return false
   */
  if (stat("/dev/poll", &statbuf) == 0) {
    return JNI_TRUE;
  } else {
    return JNI_FALSE;
  }
}

JNIEXPORT void JNICALL Java_seda_nbio_SelectSetDevPollImpl_init (JNIEnv * env, jobject this, jint max_retevents) {
  devpoll_impl_state *state;
  int i;

  DEBUG(fprintf(stderr,"SelectSetDevPollImpl.init(%d) called\n", max_retevents));

  if (!_nbio_fids_init) {
    if (nbio_init_fids(env) < 0) {
      return;
    }
  }
    
  // Allocate state
  state = (devpoll_impl_state *)malloc(sizeof(devpoll_impl_state));
  if (state == NULL) {
    THROW_EXCEPTION(env, "java/lang/OutOfMemoryError", "Cannot allocate devpoll_impl_state");
    return;
  }

  // Set state field
  (*env)->SetLongField(env, this, FID_seda_nbio_SelectSetDevPollImpl_native_state, (jlong)state);

  DEBUG(fprintf(stderr,"SelectSetDevPollImpl.init set state field\n"));

  state->max_retevents = max_retevents;
  state->devpoll_fd = open("/dev/poll", O_RDWR);
  if (state->devpoll_fd < 0) {
    THROW_EXCEPTION(env, "java/io/IOException", strerror(errno));
    free(state);
    return;
  }

#ifdef USE_BTREE
  state->tree = btree_newnode();
#else
  for (i = 0; i < MAX_FDS; i++) {
    state->selitems[i] = (jobject)NULL;
  }
#endif

  DEBUG(fprintf(stderr,"SelectSetDevPollImpl.init opened /dev/poll, fd %d\n", state->devpoll_fd));

  state->retevents = (struct pollfd *)malloc(max_retevents * sizeof(struct pollfd));
  if (state->retevents == NULL) {
    THROW_EXCEPTION(env, "java/lang/OutOfMemoryError", "Cannot allocate state->retevents");
    free(state);
    return;
  }

  return;
}