nbio.c

实现非阻塞方式IO(NBIO)的java类

/* 
 * 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.
 * 
 * This has only been tested on Linux with glibc 2.1.1 as well as Solaris.
 * However this should be readily portable to other UNIX systems. 
 * It makes use of nonblocking I/O calls, the poll() system call, as 
 * well as (optionally) /dev/poll.
 */

#include <jni.h>
#include <stdio.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

/* System call support *******************************************************/

/* MDW: The below indirection of system calls is required because we want 
 * to circumvent the JDK's own system call wrappers - otherwise on green 
 * threads we don't get the effect of true nonblocking I/O.
 */

static int _nbio_syscalls_init = 0;

typedef struct _nbio_syscall_t {
  char *sym;
  int (*addr)();
} nbio_syscall_t;

nbio_syscall_t nbio_syscalls[] = {
#define SYSCALL_WRITE 0
  { "write", 0 },
#define SYSCALL_READ 1
  { "read", 0 },
#define SYSCALL_FCNTL 2
  { "fcntl", 0 },
#define SYSCALL_POLL 3
  { "poll", 0 },
#define SYSCALL_ACCEPT 4
  { "accept", 0 },
  { NULL, 0 }
};

static inline int nbio_real_fcntl(int fd, int cmd, long arg) {
  return (*nbio_syscalls[SYSCALL_FCNTL].addr)(fd, cmd, arg);
}

static inline int nbio_real_read(int fd, void *buf, size_t count) {
  return (*nbio_syscalls[SYSCALL_READ].addr)(fd, buf, count);
}

static inline int nbio_real_write(int fd, void *buf, size_t count) {
  return (*nbio_syscalls[SYSCALL_WRITE].addr)(fd, buf, count);
}

static inline int nbio_real_poll(struct pollfd *ufds, unsigned int nfds, int timeout) {
  return (*nbio_syscalls[SYSCALL_POLL].addr)(ufds, nfds, timeout);
}

static inline int nbio_real_accept(int fd, struct sockaddr *addr, int *addrlen) {
  return (*nbio_syscalls[SYSCALL_ACCEPT].addr)(fd, addr, addrlen);
}

static void nbio_init_syscalls(JNIEnv *env) {

  nbio_syscall_t *syscall = nbio_syscalls;

#ifdef linux
  void *dlMain = dlopen("/lib/libc.so.6", RTLD_LAZY);
  if (dlMain == NULL) {
    THROW_EXCEPTION(env, "java/lang/UnsatisfiedLinkError", "NBIO: Cannot open libc shared library -- contact [e-mail] for details");
    return;
  }
  while (syscall->sym != NULL) { 
    syscall->addr = dlsym(dlMain, syscall->sym);
    if (syscall->addr == NULL) {
      THROW_EXCEPTION(env, "java/lang/UnsatisfiedLinkError", "Cannot find system call -- contact [e-mail] for details");
      return;
    }
    syscall++;
  }
#endif /* linux */

#ifdef SOLARIS
  void *dlMain = dlopen("/lib/libc.so", RTLD_LAZY);
  void *dlSock = dlopen("/lib/libsocket.so.1", RTLD_LAZY);
  if (dlMain == NULL) {
    THROW_EXCEPTION(env, "java/lang/UnsatisfiedLinkError", "NBIO: Cannot open libc shared library -- contact [e-mail] for details");
    return;
  }
  if (dlSock == NULL) {
    THROW_EXCEPTION(env, "java/lang/UnsatisfiedLinkError", "NBIO: Cannot open libsocket shared library -- contact [e-mail] for details");
    return;
  }

  /* First try in libc, then in libsocket */
  while (syscall->sym != NULL) { 
    syscall->addr = dlsym(dlMain, syscall->sym);
    if (syscall->addr == NULL) {
      syscall->addr = dlsym(dlSock, syscall->sym);
      if (syscall->addr == NULL) {
	THROW_EXCEPTION(env, "java/lang/UnsatisfiedLinkError", "Cannot find system call -- contact [e-mail] for details");
	return;
      }
    }
    syscall++;
  }
#endif /* SOLARIS */

}

static void nbio_make_nonblocking(JNIEnv *env, int fd) {
  /* Set fd to nonblocking mode */
  if (nbio_real_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 (nbio_real_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:     ninja2_core_io_0005fcore_nbio_NonblockingSocketImpl
 * Method:    nbSocketCreate
 * Signature: (Z)V
 */
JNIEXPORT void JNICALL Java_ninja2_core_io_1core_nbio_NonblockingSocketImpl_nbSocketCreate(JNIEnv *env, jobject this, jboolean stream) {
  int fd;
  jclass cls;
  jfieldID fid;
  jobject fdobj;
  long enable = 1;

  if (!_nbio_syscalls_init) nbio_init_syscalls(env);

  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)
  setsockopt(fd, SOL_SOCKET, SO_REUSEADDR, (char *)&enable, sizeof(int));
  setsockopt(fd, SOL_SOCKET, SO_KEEPALIVE, (char *)&enable, sizeof(int));

  // Want this for all sockets
  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);

  cls = (*env)->GetObjectClass(env, this);
  fid = (*env)->GetFieldID(env, cls, "fd", "Lninja2/core/io_core/nbio/NBIOFileDescriptor;");
  fdobj = (*env)->GetObjectField(env, this, fid);
  if (fdobj == NULL) {
    THROW_EXCEPTION(env, "java/net/SocketException", "socket closed");
    return;
  }
  cls = (*env)->GetObjectClass(env, fdobj);
  fid = (*env)->GetFieldID(env, cls, "fd", "I");
  (*env)->SetIntField(env, fdobj, fid, fd);
  return;
}

/*
 * Class:     ninja2_core_io_0005fcore_nbio_NonblockingSocketImpl
 * Method:    nbSocketConnect
 * Signature: (Ljava/net/InetAddress;I)V
 */
JNIEXPORT void JNICALL Java_ninja2_core_io_1core_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;

  jclass cls = (*env)->GetObjectClass(env, this);
  jfieldID fid = (*env)->GetFieldID(env, cls, "fd", "Lninja2/core/io_core/nbio/NBIOFileDescriptor;");
  jobject fdobj = (*env)->GetObjectField(env, this, fid);
  if (fdobj == NULL) {
    THROW_EXCEPTION(env, "java/net/SocketException", "socket closed");
    return;
  }
  cls = (*env)->GetObjectClass(env, fdobj);
  fid = (*env)->GetFieldID(env, cls, "fd", "I");
  fd = (*env)->GetIntField(env, fdobj, fid);
  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;
  }
  cls = (*env)->GetObjectClass(env, address);
  fid = (*env)->GetFieldID(env, cls, "address", "I");
  inet_address = (*env)->GetIntField(env, address, fid);
  fid = (*env)->GetFieldID(env, cls, "family", "I");
  inet_family = (*env)->GetIntField(env, address, fid);

  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:

  cls = (*env)->GetObjectClass(env, this);
  fid = (*env)->GetFieldID(env, cls, "address", "Ljava/net/InetAddress;");
  (*env)->SetObjectField(env, this, fid, address);
  fid = (*env)->GetFieldID(env, cls, "port", "I");
  (*env)->SetIntField(env, this, fid, port);
  fid = (*env)->GetFieldID(env, cls, "localport", "I");
  localport = (*env)->GetIntField(env, this, fid);
  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, ntohs(him.sin_port));
  }

  return;
}


JNIEXPORT jboolean JNICALL Java_ninja2_core_io_1core_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;

  jclass cls = (*env)->GetObjectClass(env, this);
  jfieldID fid = (*env)->GetFieldID(env, cls, "fd", "Lninja2/core/io_core/nbio/NBIOFileDescriptor;");
  jobject fdobj = (*env)->GetObjectField(env, this, fid);
  if (fdobj == NULL) {
    THROW_EXCEPTION(env, "java/net/SocketException", "socket closed");
    return;
  }
  cls = (*env)->GetObjectClass(env, fdobj);
  fid = (*env)->GetFieldID(env, cls, "fd", "I");
  fd = (*env)->GetIntField(env, fdobj, fid);
  if (fd == -1) {
    THROW_EXCEPTION(env, "java/net/SocketException", "socket closed");
    return;
  }

  cls = (*env)->GetObjectClass(env, this);
  fid = (*env)->GetFieldID(env, cls, "address", "Ljava/net/InetAddress;");
  address = (*env)->GetObjectField(env, this, fid);
  fid = (*env)->GetFieldID(env, cls, "port", "I");
  port = (*env)->GetIntField(env, this, fid);

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

  cls = (*env)->GetObjectClass(env, address);
  fid = (*env)->GetFieldID(env, cls, "address", "I");
  inet_address = (*env)->GetIntField(env, address, fid);
  fid = (*env)->GetFieldID(env, cls, "family", "I");
  inet_family = (*env)->GetIntField(env, address, fid);

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

  return JNI_TRUE;
}

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

  jclass cls = (*env)->GetObjectClass(env, this);
  jfieldID fid = (*env)->GetFieldID(env, cls, "fd", "Lninja2/core/io_core/nbio/NBIOFileDescriptor;");
  jobject fdobj = (*env)->GetObjectField(env, this, fid);
  if (fdobj == NULL) {
    THROW_EXCEPTION(env, "java/net/SocketException", "socket closed");
    return;
  }
  cls = (*env)->GetObjectClass(env, fdobj);
  fid = (*env)->GetFieldID(env, cls, "fd", "I");
  fd = (*env)->GetIntField(env, fdobj, fid);
  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 {
    cls = (*env)->GetObjectClass(env, address);
    fid = (*env)->GetFieldID(env, cls, "address", "I");
    inet_address = (*env)->GetIntField(env, address, fid);
    fid = (*env)->GetFieldID(env, cls, "family", "I");
    inet_family = (*env)->GetIntField(env, address, fid);
  }