nbio.c

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

  return JNI_TRUE;
}

/*
 * Class:     seda_nbio_NonblockingSocketImpl
 * Method:    nbSocketBind
 * Signature: (Ljava/net/InetAddress;I)V
 */
JNIEXPORT void JNICALL Java_seda_nbio_NonblockingSocketImpl_nbSocketBind (JNIEnv *env, jobject this, jobject address, jint port) {
  int fd, inet_address, inet_family;
  struct sockaddr_in him;
  int ret;

  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;
  }
  
  /* A bind address of NULL represents AF_INET/INADDR_ANY */
  if (address == NULL) { 
    inet_address = INADDR_ANY;
    inet_family = AF_INET;
  } else {
    inet_address = (*env)->GetIntField(env, address, FID_java_net_InetAddress_address);
    inet_family = AF_INET;
  }

  memset((char *)&him,  0, sizeof(him));
  him.sin_port = htons((short)port);
  him.sin_addr.s_addr = (unsigned long)htonl(inet_address);
  him.sin_family = inet_family;

  if ((ret = bind(fd, (struct sockaddr *)&him, sizeof(him))) < 0) {
     // JRVB: some occurrences of errno - should have been myerrno
    int myerrno = errno;
    fprintf(stderr,"bind: returned %d, errno %d (%s)\n", ret, myerrno, strerror(myerrno));
    if (myerrno == EACCES) {
      THROW_EXCEPTION(env, "java/net/BindException", strerror(myerrno));
    } else {
      THROW_EXCEPTION(env, "java/net/SocketException", strerror(myerrno));
    }
    return;
  }

  (*env)->SetObjectField(env, this, FID_seda_nbio_NonblockingSocketImpl_address, address);
  (*env)->SetIntField(env, this, FID_seda_nbio_NonblockingSocketImpl_port, port);

  /* Set local port value */

  if (port == 0) {
    /* Set localport value -- may have been previously set by bind operation */
    int len = sizeof(him);
    if (getsockname(fd,(struct sockaddr *)&him, &len) == -1) {
      THROW_EXCEPTION(env, "java/net/SocketException", strerror(errno));
      return;
    }
    (*env)->SetIntField(env, this, FID_seda_nbio_NonblockingSocketImpl_localport, ntohs(him.sin_port));
  } else {
    (*env)->SetIntField(env, this, FID_seda_nbio_NonblockingSocketImpl_localport, port);
  }

  return;
}
  
/*
 * Class:     seda_nbio_NonblockingSocketImpl
 * Method:    nbSocketListen
 * Signature: (I)V
 */
JNIEXPORT void JNICALL Java_seda_nbio_NonblockingSocketImpl_nbSocketListen (JNIEnv *env, jobject this, jint count) {
  int fd;

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

  if (listen(fd, count) < 0) {
    THROW_EXCEPTION(env, "java/net/SocketException", strerror(errno));
    return;
  }

}

/*
 * Class:     seda_nbio_NonblockingSocketImpl
 * Method:    nbSocketAccept
 * Signature: (Lseda/nbio/NonblockingSocketImpl;)V
 */
JNIEXPORT jint JNICALL Java_seda_nbio_NonblockingSocketImpl_nbSocketAccept (JNIEnv *env, jobject this, jobject newsocket, jboolean block) {
  int fd, newfd;
  jclass cls;
  struct sockaddr_in him;
  int len, localport;
  jobject sptr_inetaddr, sptr_fdobj;

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

  if (newsocket == NULL) { 
    THROW_EXCEPTION(env, "java/lang/NullPointerException", "newsocket is NULL");
    return -1;
  }

  /* We expect that the 'fd' field of the newsocket has
   * been created outside of this method (but not initialized).
   */
  sptr_fdobj = (*env)->GetObjectField(env, newsocket, FID_seda_nbio_NonblockingSocketImpl_fd);

  if (sptr_fdobj == NULL) {
    THROW_EXCEPTION(env, "java/lang/NullPointerException", "newsocket uninitialized");
    return -1;
  }

  /* XXX MDW This is not threadsafe */
  if (!block) {
    nbio_make_nonblocking(env, fd);
  } else {
    nbio_make_blocking(env, fd);
  }

  DEBUG(fprintf(stderr,"NBIO: Doing accept() on fd=%d\n", fd));

#if defined(linux) || defined(__FreeBSD__)
  len = sizeof(him);
#endif 
#ifdef SOLARIS
  len = sizeof(struct sockaddr); 
#endif

  newfd = accept(fd, (struct sockaddr *)&him, &len);
  if (newfd < 0) {
    if (!block && errno == EWOULDBLOCK) return -1;
    else {
      THROW_EXCEPTION(env, "java/net/SocketException", strerror(errno));
      return -1;
    }
  }

  DEBUG(fprintf(stderr,"NBIO: accept() on fd=%d returned %d\n", fd, newfd));

  nbio_make_nonblocking(env, newfd);
  nbio_disable_nagle(env, newfd);

  (*env)->SetIntField(env, sptr_fdobj, FID_seda_nbio_NBIOFileDescriptor_fd, newfd);

  localport = (*env)->GetIntField(env, this, FID_seda_nbio_NonblockingSocketImpl_localport);

  /* Create empty InetAddress and initialize it */
  DEBUG(fprintf(stderr,"NBIO: accept() creating new InetAddress\n"));
  cls = (*env)->FindClass(env, "java/net/InetAddress");
  if (cls == NULL) {
    THROW_EXCEPTION(env, "java/net/SocketException", "Cannot find java.net.InetAddress class");
    return -1;
  }
  sptr_inetaddr = (*env)->AllocObject(env, cls);

  DEBUG(fprintf(stderr,"NBIO: accept() initializing new InetAddress\n"));

  (*env)->SetIntField(env, sptr_inetaddr, FID_java_net_InetAddress_address, ntohl(him.sin_addr.s_addr));

  // In JDK 1.4 the meaning of 'family' changed in InetAddress, so don't
  // set it here.
  //(*env)->SetIntField(env, sptr_inetaddr, FID_java_net_InetAddress_family, him.sin_family);

  (*env)->SetIntField(env, newsocket, FID_seda_nbio_NonblockingSocketImpl_port, ntohs(him.sin_port));
  (*env)->SetIntField(env, newsocket, FID_seda_nbio_NonblockingSocketImpl_localport, localport);
  (*env)->SetObjectField(env, newsocket, FID_seda_nbio_NonblockingSocketImpl_address, sptr_inetaddr);

  return 0;

}

/*
 * Class:     seda_nbio_NonblockingSocketImpl
 * Method:    nbSocketAvailable
 * Signature: ()I
 */
JNIEXPORT jint JNICALL Java_seda_nbio_NonblockingSocketImpl_nbSocketAvailable (JNIEnv *env, jobject this) {
  int fd;
  int bytes;

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

#if defined(linux) || defined(__FreeBSD__)
  if (ioctl(fd, FIONREAD, &bytes) < 0) {
#endif /* linux or __FreeBSD__ */
#ifdef SOLARIS
  if (ioctl(fd, I_NREAD, &bytes) < 0) {
#endif /* SOLARIS */
    THROW_EXCEPTION(env, "java/net/SocketException", strerror(errno));
    return -1;
  }

  DEBUG(fprintf(stderr,"NBIO: nbSocketAvailable called, %d bytes available\n", bytes));

  return bytes;

}

/*
 * Class:     seda_nbio_NonblockingSocketImpl
 * Method:    nbSocketClose
 * Signature: ()V
 */
JNIEXPORT void JNICALL Java_seda_nbio_NonblockingSocketImpl_nbSocketClose (JNIEnv * env, jobject this) {
  int fd; 

  jobject fdobj = (*env)->GetObjectField(env, this, FID_seda_nbio_NonblockingSocketImpl_fd);
  if (fdobj == NULL) {
    return;
  }
  fd = (*env)->GetIntField(env, fdobj, FID_seda_nbio_NBIOFileDescriptor_fd);
  if (fd == -1) {
    return;
  }

  close(fd);
  (*env)->SetIntField(env, fdobj, FID_seda_nbio_NBIOFileDescriptor_fd, -1);
}

/* NonblockingSocketInputStream *******************************************/

/*
 * Class:     seda_nbio_NonblockingSocketInputStream
 * Method:    nbSocketRead
 * Signature: ([BII)I
 */
JNIEXPORT jint JNICALL Java_seda_nbio_NonblockingSocketInputStream_nbSocketRead (JNIEnv *env, jobject this, jbyteArray b, jint off, jint len) {
  int fd;
  jobject fdobj;
  int datalen;
  char *data;
  int n;

  DEBUG(fprintf(stderr,"NBIO: nbSocketRead called, off=%d len=%d\n", off, len));

  fdobj = (*env)->GetObjectField(env, this, FID_seda_nbio_NonblockingSocketInputStream_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;
  }

  if (b == NULL) {
    THROW_EXCEPTION(env, "java/lang/NullPointerException", "null byte array passed to nbSocketRead");
    return -1;
  }
  datalen = (*env)->GetArrayLength(env, b);

  if (len < 0 || len + off > datalen) {
    THROW_EXCEPTION(env, "java/lang/ArrayIndexOutOfBoundsException", "len must be >= 0 and len + off <= array length");
  }

  data = (*env)->GetByteArrayElements(env, b, NULL);

  n = read(fd, data+off, len);
  
  DEBUG(fprintf(stderr,"NBIO: nbSocketRead: off is %d, len is %d, got %d, errno is %d\n", off, len, n, errno));

  if (n == 0) {
    // Socket was closed 
    (*env)->ReleaseByteArrayElements(env, b, data, JNI_ABORT);
    return -1; // EOF
  } else if (n < 0) {
    if (errno == EAGAIN) {
      (*env)->ReleaseByteArrayElements(env, b, data, JNI_ABORT);
      return 0; // No data returned
    } else {
      THROW_EXCEPTION(env, "java/net/SocketException", strerror(errno));
      (*env)->ReleaseByteArrayElements(env, b, data, JNI_ABORT);
      return -1; // error
    }
  }

  (*env)->ReleaseByteArrayElements(env, b, data, 0);
  return n;
}

/*
 * Class:     seda_nbio_NonblockingSocketOutputStream
 * Method:    nbSocketWrite
 * Signature: ([BII)I
 */
JNIEXPORT jint JNICALL Java_seda_nbio_NonblockingSocketOutputStream_nbSocketWrite (JNIEnv *env, jobject this, jbyteArray b, jint off, jint len) {
  int fd;
  jobject fdobj;
  int datalen;
  char *data;
  int n;

  DEBUG(fprintf(stderr,"NBIO: nbSocketWrite called, off=%d len=%d\n", off, len));

  fdobj = (*env)->GetObjectField(env, this, FID_seda_nbio_NonblockingSocketOutputStream_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;
  }

  if (b == NULL) {
    THROW_EXCEPTION(env, "java/lang/NullPointerException", "null byte array passed to nbSocketWrite");
    return -1;
  }
  datalen = (*env)->GetArrayLength(env, b);

  if (len < 0 || len + off > datalen) {
    THROW_EXCEPTION(env, "java/lang/ArrayIndexOutOfBoundsException", "len must be >= 0 and len + off <= array length");
  }

  data = (*env)->GetByteArrayElements(env, b, NULL);

  n = write(fd, data+off, len);
  
  DEBUG(fprintf(stderr,"NBIO: nbSocketWrite: off is %d, len is %d, got %d, errno is %d\n", off, len, n, errno));

  if (n < 0) {
    if ((errno == EAGAIN) || (errno == EINTR)) {
      (*env)->ReleaseByteArrayElements(env, b, data, JNI_ABORT);
      return 0; // No data returned
    } else {
      THROW_EXCEPTION(env, "java/net/SocketException", strerror(errno));
      (*env)->ReleaseByteArrayElements(env, b, data, JNI_ABORT);
      return -1; // error
    }
  }

  (*env)->ReleaseByteArrayElements(env, b, data, JNI_ABORT);
  return n;
}

/* UDP support code below */

/*
 * Class:     seda_nbio_NonblockingSocketImpl
 * Method:    nbReceive
 */
JNIEXPORT jint JNICALL Java_seda_nbio_NonblockingSocketImpl_nbReceive(JNIEnv *env, jobject this, jobject packet) {
  jbyteArray data;
  jbyte* recvdata;
  jint length, offset;
  struct sockaddr_in from;
  int fromlength;
  int fd, ret, sz;
  jclass cls; 
  jobject addrobj;
  jobject fdobj;

  DEBUG(fprintf(stderr,"NBIO: nbReceive called\n"));

  fdobj = (*env)->GetObjectField(env, this, FID_seda_nbio_NonblockingSocketImpl_fd);

  if (fdobj == NULL) {
    THROW_EXCEPTION(env, "java/net/SocketException", "socket closed");
    return -1;
  } 
  if(packet == NULL) {
    THROW_EXCEPTION(env, "java/lang/NullPointerException", "packet is null in nbReceive");
    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 in nbReceive is null");
    return -1;
  }