nbio.c

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

/* 
 * Copyright (c) 2000 by Matt Welsh and The Regents of the University of 
 * California. All rights reserved.
 *
 * Permission to use, copy, modify, and distribute this software and its
 * documentation for any purpose, without fee, and without written agreement is
 * hereby granted, provided that the above copyright notice and the following
 * two paragraphs appear in all copies of this software.
 * 
 * IN NO EVENT SHALL THE UNIVERSITY OF CALIFORNIA BE LIABLE TO ANY PARTY FOR
 * DIRECT, INDIRECT, SPECIAL, INCIDENTAL, OR CONSEQUENTIAL DAMAGES ARISING OUT
 * OF THE USE OF THIS SOFTWARE AND ITS DOCUMENTATION, EVEN IF THE UNIVERSITY OF
 * CALIFORNIA HAS BEEN ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 * 
 * THE UNIVERSITY OF CALIFORNIA SPECIFICALLY DISCLAIMS ANY WARRANTIES,
 * INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY
 * AND FITNESS FOR A PARTICULAR PURPOSE.  THE SOFTWARE PROVIDED HEREUNDER IS
 * ON AN "AS IS" BASIS, AND THE UNIVERSITY OF CALIFORNIA HAS NO OBLIGATION TO
 * PROVIDE MAINTENANCE, SUPPORT, UPDATES, ENHANCEMENTS, OR MODIFICATIONS.
 *
 * Author: Matt Welsh <[e-mail]>
 * 
 */

/*
 * This file implements the native method bindings for the nbio library.
 */

#include <jni.h>
#include <stdio.h>
#include <stdlib.h>

#include <errno.h>
#include <string.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <sys/ioctl.h>
#include <sys/poll.h>
#include <netinet/tcp.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <unistd.h>
#include <fcntl.h>
#include <dlfcn.h>
#include <sys/stat.h>

#ifdef SOLARIS
#include <stropts.h>
#include <sys/filio.h>
#endif

#ifdef HAS_DEVPOLL
#include <sys/devpoll.h>
#ifndef POLLREMOVE
#define POLLREMOVE 0x1000
#endif
#endif

#include "NonblockingSocket.h"
#include "NonblockingSocketImpl.h"
#include "NonblockingSocketInputStream.h"
#include "mdw-exceptions.h"
#include "mdw-btree.h"

#define DEBUG(_x) 

/* Constants *****************************************************************/

/* These need to stay consistent with the Java code - javah -jni does not
 * include constant definitions
 */
#define SELECTABLE_READ_READY 0x01
#define SELECTABLE_WRITE_READY 0x02
#define SELECTABLE_SELECT_ERROR 0x80

/* Java field/method IDs *****************************************************/

/* When we are first initialized we obtain all of the field/method IDs
 * that we need so that these don't have to be determined on the fly.
 */
static int _nbio_fids_init = 0;

static jfieldID FID_seda_nbio_NonblockingSocketInputStream_fd;
static jfieldID FID_seda_nbio_NonblockingSocketOutputStream_fd;
static jfieldID FID_seda_nbio_NonblockingSocketImpl_fd;
static jfieldID FID_seda_nbio_NonblockingSocketImpl_address;
static jfieldID FID_seda_nbio_NonblockingSocketImpl_port;
static jfieldID FID_seda_nbio_NonblockingSocketImpl_localport;
static jfieldID FID_seda_nbio_NBIOFileDescriptor_fd;
static jfieldID FID_java_net_InetAddress_address;
static jfieldID FID_java_net_InetAddress_family;
static jfieldID FID_java_net_DatagramPacket_buf;
static jfieldID FID_java_net_DatagramPacket_offset;
static jfieldID FID_java_net_DatagramPacket_length;
static jfieldID FID_java_net_DatagramPacket_address;
static jfieldID FID_java_net_DatagramPacket_port;
static jfieldID FID_seda_nbio_SelectItem_fd;
static jfieldID FID_seda_nbio_SelectItem_events;
static jfieldID FID_seda_nbio_SelectItem_revents;
static jfieldID FID_seda_nbio_SelectSetPollImpl_itemarr;
static jfieldID FID_seda_nbio_SelectSetDevPollImpl_itemarr;
static jfieldID FID_seda_nbio_SelectSetDevPollImpl_retevents;
static jfieldID FID_seda_nbio_SelectSetDevPollImpl_native_state;

static int nbio_init_fids(JNIEnv *env) {

  char _nbio_init_fids_err[512];
  jclass _nbio_init_fids_cls;

#define NBIO_GET_CLASS(__cname) { \
  _nbio_init_fids_cls = (*env)->FindClass(env, __cname); \
  if (_nbio_init_fids_cls == NULL) { \
    sprintf(_nbio_init_fids_err, "NBIO: Cannot resolve class %s in nbio_init_fids() -- this is a bug, please contact <[e-mail]", __cname); \
    THROW_EXCEPTION(env, "java/lang/UnsatisfiedLinkError", _nbio_init_fids_err); \
    return -1; \
  } \
  }

#define NBIO_GET_FIELD(__fname, __typename, __fid) \
  __fid = (*env)->GetFieldID(env, _nbio_init_fids_cls, __fname, __typename); \
  if (__fid == NULL) { \
    sprintf(_nbio_init_fids_err, "NBIO: Cannot resolve field %s (%s) in nbio_init_fids() -- this is a bug, please contact <[e-mail]", __fname, __typename); \
    THROW_EXCEPTION(env, "java/lang/UnsatisfiedLinkError", _nbio_init_fids_err); \
    return -1; \
  }

#define NBIO_GET_METHOD(__mname, __sig, __mid) \
  __mid = (*env)->GetMethodID(env, _nbio_init_fids_cls, __mname, __sig); \
  if (__mid == NULL) { \
    sprintf(_nbio_init_fids_err, "NBIO: Cannot resolve method %s (%s) in nbio_init_fids() -- this is a bug, please contact <[e-mail]", __mname, __sig); \
    THROW_EXCEPTION(env, "java/lang/UnsatisfiedLinkError", _nbio_init_fids_err); \
    return -1; \
  } 

  /* seda/nbio/NonblockingSocketInputStream */
  NBIO_GET_CLASS("seda/nbio/NonblockingSocketInputStream");
  NBIO_GET_FIELD("fd", "Lseda/nbio/NBIOFileDescriptor;", FID_seda_nbio_NonblockingSocketInputStream_fd);

  /* seda/nbio/NonblockingSocketOutputStream */
  NBIO_GET_CLASS("seda/nbio/NonblockingSocketOutputStream");
  NBIO_GET_FIELD("fd", "Lseda/nbio/NBIOFileDescriptor;", FID_seda_nbio_NonblockingSocketOutputStream_fd);

  /* seda/nbio/NonblockingSocketImpl */
  NBIO_GET_CLASS("seda/nbio/NonblockingSocketImpl");
  NBIO_GET_FIELD("fd", "Lseda/nbio/NBIOFileDescriptor;", FID_seda_nbio_NonblockingSocketImpl_fd);
  NBIO_GET_FIELD("address", "Ljava/net/InetAddress;", FID_seda_nbio_NonblockingSocketImpl_address);
  NBIO_GET_FIELD("port", "I", FID_seda_nbio_NonblockingSocketImpl_port);
  NBIO_GET_FIELD("localport", "I", FID_seda_nbio_NonblockingSocketImpl_localport);

  /* seda/nbio/NBIOFileDescriptor */
  NBIO_GET_CLASS("seda/nbio/NBIOFileDescriptor");
  NBIO_GET_FIELD("fd", "I", FID_seda_nbio_NBIOFileDescriptor_fd);

  /* java/net/InetAddress */
  NBIO_GET_CLASS("java/net/InetAddress");
  NBIO_GET_FIELD("address", "I", FID_java_net_InetAddress_address);
  NBIO_GET_FIELD("family", "I", FID_java_net_InetAddress_family);

  /* java/net/DatagramPacket */
  NBIO_GET_CLASS("java/net/DatagramPacket");
  NBIO_GET_FIELD("buf", "[B", FID_java_net_DatagramPacket_buf);
  NBIO_GET_FIELD("offset", "I", FID_java_net_DatagramPacket_offset);
  NBIO_GET_FIELD("length", "I", FID_java_net_DatagramPacket_length);
  NBIO_GET_FIELD("address", "Ljava/net/InetAddress;", FID_java_net_DatagramPacket_address);
  NBIO_GET_FIELD("port", "I", FID_java_net_DatagramPacket_port);

  /* seda/nbio/SelectItem */
  NBIO_GET_CLASS("seda/nbio/SelectItem");
  NBIO_GET_FIELD("fd", "Lseda/nbio/NBIOFileDescriptor;", FID_seda_nbio_SelectItem_fd);
  NBIO_GET_FIELD("events", "S", FID_seda_nbio_SelectItem_events);
  NBIO_GET_FIELD("revents", "S", FID_seda_nbio_SelectItem_revents);

  /* seda/nbio/SelectSetPollImpl */
  NBIO_GET_CLASS("seda/nbio/SelectSetPollImpl");
  NBIO_GET_FIELD("itemarr", "[Lseda/nbio/SelectItem;", FID_seda_nbio_SelectSetPollImpl_itemarr);

  /* seda/nbio/SelectSetDevPollImpl */
  NBIO_GET_CLASS("seda/nbio/SelectSetDevPollImpl");
  NBIO_GET_FIELD("itemarr", "[Lseda/nbio/SelectItem;", FID_seda_nbio_SelectSetDevPollImpl_itemarr);
  NBIO_GET_FIELD("retevents", "[Lseda/nbio/SelectItem;", FID_seda_nbio_SelectSetDevPollImpl_retevents);
  NBIO_GET_FIELD("native_state", "J", FID_seda_nbio_SelectSetDevPollImpl_native_state);

#undef NBIO_GET_CLASS
#undef NBIO_GET_FIELD

  _nbio_fids_init = 1;

  return 0;

}

static void nbio_make_nonblocking(JNIEnv *env, int fd) {
  /* Set fd to nonblocking mode */
  if (fcntl(fd, F_SETFL, O_NONBLOCK) < 0) {
    THROW_EXCEPTION(env, "java/net/SocketException", strerror(errno));
    return;
  }
  DEBUG(fprintf(stderr,"Set fd=%d to nonblocking mode\n", fd));
}

static void nbio_make_blocking(JNIEnv *env, int fd) {
  /* Set fd to blocking mode */
  if (fcntl(fd, F_SETFL, 0) < 0) {
    THROW_EXCEPTION(env, "java/net/SocketException", strerror(errno));
    return;
  }
  DEBUG(fprintf(stderr,"Set fd=%d to blocking mode\n", fd));
}

static void nbio_disable_nagle(JNIEnv *env, int fd) {
  int enable = 1;
  if (setsockopt(fd, IPPROTO_TCP, TCP_NODELAY, (char *)&enable, sizeof(int)) < 0) {
    THROW_EXCEPTION(env, "java/net/SocketException", strerror(errno));
    return;
  }
}

/* NonblockingSocketImpl *****************************************************/

/*
 * Class:     seda_nbio_NonblockingSocketImpl
 * Method:    nbSocketCreate
 * Signature: (Z)V
 */
JNIEXPORT void JNICALL Java_seda_nbio_NonblockingSocketImpl_nbSocketCreate(JNIEnv *env, jobject this, jboolean stream) {
  int fd;
  jobject fdobj;
  long enable = 1;

  if (!_nbio_fids_init) {
    if (nbio_init_fids(env) < 0) {
      return;
    }
  }

  fd = socket(AF_INET, (stream ? SOCK_STREAM: SOCK_DGRAM), 0);
  if (fd == -1) {
    THROW_EXCEPTION(env, "java/io/IOException", strerror(errno));
    return;
  }
 
  DEBUG(fprintf(stderr,"NBIO: Created socket, fd=%d\n", fd));

  // XXX MDW: Turn these on for all sockets
  // XXX MDW: (These are probably best to turn on only for servers)
  //
  // XXX JRVB: SO_REUSEADDR is also necessary for Multicast sockets to
  // XXX JRVB: work as expected, so if this is removed from here, it
  // XXX JRVB: needs to be specifically enabled for multicast sockets
  // XXX JRVB: elsewhere.
  setsockopt(fd, SOL_SOCKET, SO_REUSEADDR, (char *)&enable, sizeof(int));
  setsockopt(fd, SOL_SOCKET, SO_KEEPALIVE, (char *)&enable, sizeof(int));

  // Want this for all TCP sockets
  if (stream) nbio_disable_nagle(env, fd);

  // XXX Should also turn on SO_LINGER? Apache does for server sockets..
  // XXX Should also set SO_SNDBUF to increase send buffer size?

  nbio_make_nonblocking(env, fd);

  fdobj = (*env)->GetObjectField(env, this, FID_seda_nbio_NonblockingSocketImpl_fd);
  if (fdobj == NULL) {
    THROW_EXCEPTION(env, "java/net/SocketException", "socket closed");
    return;
  }
  (*env)->SetIntField(env, fdobj, FID_seda_nbio_NBIOFileDescriptor_fd, fd);
  DEBUG(fprintf(stderr,"NBIO: Returning from nbSocketCreate with fd=%d\n", fd));
  return;
}

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

  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 (address == NULL) {
    THROW_EXCEPTION(env, "java/lang/NullPointerException", "address is NULL");
    return;
  }
  inet_address = (*env)->GetIntField(env, address, FID_java_net_InetAddress_address);

  // In JDK 1.4 the meaning of the 'family' field changed in InetAddress,
  // so hardwire this to AF_INET (all we care about anyway).
  //inet_family = (*env)->GetIntField(env, address, FID_java_net_InetAddress_family);
  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;

again:
  if ((ret = connect(fd, (struct sockaddr *)&him, sizeof(him))) < 0) {
    myerrno = errno;
    DEBUG(fprintf(stderr,"NBIO: connect returned %d, errno %d\n", ret, myerrno));
    if (myerrno == EINPROGRESS) {
      /* This is ok - connection not yet done */
      goto connect_ok;
    } else if (myerrno == ECONNREFUSED) {
      THROW_EXCEPTION(env, "java/net/ConnectException", strerror(myerrno));
    } else if (myerrno == ETIMEDOUT || myerrno == EHOSTUNREACH) {
      THROW_EXCEPTION(env, "java/net/NoRouteToHostException", strerror(myerrno));
    } else if (myerrno == EINTR) {
      DEBUG(fprintf(stderr,"***** NBIO: connect: Interrupted, trying again\n"));
      goto again;
    } else {
      THROW_EXCEPTION(env, "java/net/SocketException", strerror(myerrno));
    }
    return;
  }

 connect_ok:

  (*env)->SetObjectField(env, this, FID_seda_nbio_NonblockingSocketImpl_address, address);
  (*env)->SetIntField(env, this, FID_seda_nbio_NonblockingSocketImpl_port, port);
  localport = (*env)->GetIntField(env, this, FID_seda_nbio_NonblockingSocketImpl_localport);
  if (localport == 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));
  }

  return;
}


JNIEXPORT jboolean JNICALL Java_seda_nbio_NonblockingSocketImpl_nbSocketConnectDone (JNIEnv *env, jobject this) {

  /* This is a bit strange. Although the man pages say you use
   * select() followed by getsockopt() to find out if the connection was
   * established, looks like you do select() and call connect() again...
   */

  int fd, inet_address, inet_family;
  jobject address; 
  jint port;
  struct sockaddr_in him;
  int ret;
  int myerrno;

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

  address = (*env)->GetObjectField(env, this, FID_seda_nbio_NonblockingSocketImpl_address);
  port = (*env)->GetIntField(env, this, FID_seda_nbio_NonblockingSocketImpl_port);

  if (address == NULL) {
    THROW_EXCEPTION(env, "java/lang/NullPointerException", "address is NULL");
    return JNI_FALSE;
  }

  inet_address = (*env)->GetIntField(env, address, FID_java_net_InetAddress_address);

  // In JDK 1.4 the meaning of the 'family' field changed in InetAddress,
  // so hardwire this to AF_INET (all we care about anyway).
  //inet_family = (*env)->GetIntField(env, address, FID_java_net_InetAddress_family);
  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;

  DEBUG(fprintf(stderr,"NBIO: connectDone: recalling connect on fd %d\n", fd));

again:
  if ((ret = connect(fd, (struct sockaddr *)&him, sizeof(him))) < 0) {
    myerrno = errno;
    DEBUG(fprintf(stderr,"NBIO: connectDone: connect got errorno %d\n", myerrno));
    if (myerrno == EINPROGRESS) return JNI_FALSE;
    else if (myerrno == EALREADY) return JNI_FALSE;
    else if (myerrno == EISCONN) return JNI_TRUE;
    else if (myerrno == EINTR) {
      DEBUG(fprintf(stderr,"NBIO: connectDone: connect returned EINTR, trying again"));
      goto again;
    } else {
      THROW_EXCEPTION(env, "java/net/SocketException", strerror(myerrno));
      return JNI_FALSE;
    }
  }