nbio.c

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

JNIEXPORT void JNICALL Java_seda_nbio_SelectSetDevPollImpl_register (JNIEnv * env, jobject this, jobject selitemobj) {
  devpoll_impl_state *state;
  jobject fdobj;
  short events, realevents;
  struct pollfd pfd;
  jobject selitem;

  DEBUG(fprintf(stderr,"SelectSetDevPollImpl.register called\n"));

  // Get state
  state = (devpoll_impl_state *)(((*env)->GetLongField(env, this, FID_seda_nbio_SelectSetDevPollImpl_native_state)) & 0xffffffff);

  DEBUG(fprintf(stderr,"SelectSetDevPollImpl.register got state, devpoll_fd %d\n", state->devpoll_fd));

  // Get fd
  fdobj = (*env)->GetObjectField(env, selitemobj, FID_seda_nbio_SelectItem_fd);
  pfd.fd = (*env)->GetIntField(env, fdobj, FID_seda_nbio_NBIOFileDescriptor_fd);
  // Get events
  events = (*env)->GetShortField(env, selitemobj, FID_seda_nbio_SelectItem_events);
  realevents = 0;
  if (events != 0) {
    if (events & SELECTABLE_READ_READY) {
      realevents |= (POLLIN | POLLPRI);
    }
    if (events & SELECTABLE_WRITE_READY) {
      realevents |= POLLOUT;
    }
  }
  pfd.events = realevents;
  DEBUG(fprintf(stderr,"nbio: events was 0x%lx, pfd.events now 0x%lx\n", events, realevents));

  DEBUG(fprintf(stderr,"SelectSetDevPollImpl.register adding (fd=%d,events=0x%x)\n", pfd.fd, pfd.events));

  // Register
  if (write(state->devpoll_fd, &pfd, sizeof(struct pollfd)) != sizeof(struct pollfd)) {
    THROW_EXCEPTION(env, "java/io/IOException", strerror(errno));
    return;
  }

#ifdef USE_BTREE
  data = btree_search(state->tree, pfd.fd);
  if (data == NULL) {
    // Insert into B-tree
    state->tree = btree_insert(state->tree, pfd.fd, (*env)->NewGlobalRef(env, selitemobj));
  }
#else
  if ((pfd.fd < 0) || (pfd.fd >= MAX_FDS)) {
    fprintf(stderr,"Warning: pfd.fd is %d, out of range 0...%d.\n",pfd.fd,MAX_FDS);
    fprintf(stderr,"  You need to recompile nbio.c with a larger MAX_FDS value.\n");
    fprintf(stderr,"  This is part of the NBIO package.\n");
    return;
  }
  selitem = state->selitems[pfd.fd];
  if (selitem == (jobject)NULL) {
    state->selitems[pfd.fd] = (*env)->NewGlobalRef(env, selitemobj);
  }
#endif

  return;
}

JNIEXPORT void JNICALL Java_seda_nbio_SelectSetDevPollImpl_deregister (JNIEnv * env, jobject this, jobject selitemobj) {
  devpoll_impl_state *state;
  jobject fdobj, selitem;
  struct pollfd pfd;

  DEBUG(fprintf(stderr,"SelectSetDevPollImpl.deregister called\n"));

  // Get state
  state = (devpoll_impl_state *)(((*env)->GetLongField(env, this, FID_seda_nbio_SelectSetDevPollImpl_native_state)) & 0xffffffff);

  DEBUG(fprintf(stderr,"SelectSetDevPollImpl.deregister got state, devpoll_fd %d\n", state->devpoll_fd));

  // Get fd
  fdobj = (*env)->GetObjectField(env, selitemobj, FID_seda_nbio_SelectItem_fd);
  pfd.fd = (*env)->GetIntField(env, fdobj, FID_seda_nbio_NBIOFileDescriptor_fd);
  pfd.events = POLLREMOVE;

  DEBUG(fprintf(stderr,"SelectSetDevPollImpl.deregister removing (fd=%d,events=0x%x)\n", pfd.fd, pfd.events));

  // Deeegister
  if (write(state->devpoll_fd, &pfd, sizeof(struct pollfd)) != sizeof(struct pollfd)) {
    THROW_EXCEPTION(env, "java/io/IOException", strerror(errno));
    return;
  }

#ifdef USE_BTREE
  // XXX NEED TO DELETE NODE FROM BTREE
#else
  if ((pfd.fd < 0) || (pfd.fd >= MAX_FDS)) {
    fprintf(stderr,"Warning: pfd.fd is %d, out of range 0...%d.\n",pfd.fd,MAX_FDS);
    fprintf(stderr,"  You need to recompile nbio.c with a larger MAX_FDS value.\n");
    return;
  }
  selitem = state->selitems[pfd.fd];
  if (selitem != (jobject)NULL) {
    // selitem can be NULL if deregister is not synchronized 
    // (although it should be)
    (*env)->DeleteGlobalRef(env, selitem);
    state->selitems[pfd.fd] = (jobject)NULL;
  }
#endif

  return;
}

JNIEXPORT jint JNICALL Java_seda_nbio_SelectSetDevPollImpl_doSelect (JNIEnv * env, jobject this, jint timeout, jint num_fds) {
  devpoll_impl_state *state;
  jobject selitemobj;
  jobjectArray itemarr, retitemarr;
  struct dvpoll dopoll;
  int itemarrlen, retitemarrlen, ret, i, retfd, count;
  struct pollfd *pfd;
  short realevents;

  DEBUG(fprintf(stderr,"SelectSetDevPollImpl.doSelect called\n"));

  // Get state
  state = (devpoll_impl_state *)(((*env)->GetLongField(env, this, FID_seda_nbio_SelectSetDevPollImpl_native_state)) & 0xffffffff);

  DEBUG(fprintf(stderr,"SelectSetDevPollImpl.doSelect got state, devpoll_fd=%d\n",state->devpoll_fd));

  // Get itemarray
  itemarr = (jobjectArray)((*env)->GetObjectField(env, this, FID_seda_nbio_SelectSetDevPollImpl_itemarr));
  if (itemarr == NULL) {
    // This can happen if we have an empty SelectSet 
    return 0;
  }
  itemarrlen = (*env)->GetArrayLength(env, itemarr);
  if (itemarrlen <= 0) {
    THROW_EXCEPTION(env, "java/lang/ArrayIndexOutOfBoundsException", "SelectItem[] array has size <= 0");
    return 0;
  }

  DEBUG(fprintf(stderr,"SelectSetDevPollImpl.doSelect got itemarr, len %d\n",itemarrlen));

  // Get retitemarr
  retitemarr = (jobjectArray)((*env)->GetObjectField(env, this, FID_seda_nbio_SelectSetDevPollImpl_retevents));
  retitemarrlen = (*env)->GetArrayLength(env, retitemarr);
  if (retitemarrlen <= 0) {
    THROW_EXCEPTION(env, "java/lang/ArrayIndexOutOfBoundsException", "SelectItem[] ret array has size <= 0");
    return 0;
  }

  DEBUG(fprintf(stderr,"SelectSetDevPollImpl.doSelect got retitemarr, length %d\n", retitemarrlen));

  // Fill in dopoll
  dopoll.dp_timeout = timeout;
  dopoll.dp_nfds = (num_fds > state->max_retevents) ? (state->max_retevents) : (num_fds);
  dopoll.dp_fds = state->retevents;

  DEBUG(fprintf(stderr,"SelectSetDevPollImpl.doSelect (devpollfd %d) doing DP_POLL\n", state->devpoll_fd));

  ret = ioctl(state->devpoll_fd, DP_POLL, &dopoll);

  DEBUG(fprintf(stderr,"SelectSetDevPollImpl.doSelect (devpollfd %d) DP_POLL returned %d\n", state->devpoll_fd, ret));

  if (ret == 0) {
    return 0;
  }
  if (ret < 0) {
    int myerrno = errno;
    // Don't throw an exception if we were interrupted
    if (myerrno != EINTR) {
      THROW_EXCEPTION(env, "java/io/IOException", strerror(myerrno));
    } 
    return 0;
  }

  // Need to synchronize in case register/deregister called while
  // we assign SelectItems to retitemarr
  (*env)->MonitorEnter(env, this); 

  count = 0;
  for (i = 0; i < ret; i++) {
    pfd = &(state->retevents[i]);

    DEBUG(fprintf(stderr,"SelectSetDevPollImpl.doSelect ret[%d] fd %d revents 0x%x\n", i, pfd->fd, pfd->revents));
    retfd = pfd->fd;

#ifdef USE_BTREE
    data = btree_search(state->tree, retfd);
    if (data == NULL) {
      // This can be caused by a socket closing (and being reregistered
      // from the SelectSet) asynchronously with respect to a call to
      // doSelect(). In this case just skip over it
      continue;
    }
    selitemobj = (jobject)data;
#else 
    if ((retfd < 0) || (retfd >= MAX_FDS)) {
      fprintf(stderr,"Warning: retfd is %d, out of range 0...%d.\n",retfd,MAX_FDS);
      fprintf(stderr,"  You need to recompile nbio.c with a larger MAX_FDS value.\n");
      THROW_EXCEPTION(env, "java/lang/ArrayIndexOutOfBoundsException", "retfd out of range");
      (*env)->MonitorExit(env, this); 
      return 0;
    }
    selitemobj = state->selitems[retfd];
    if (selitemobj == (jobject)NULL) {
      // This can be caused by a socket closing (and being reregistered
      // from the SelectSet) asynchronously with respect to a call to
      // doSelect(). In this case just skip over it
      continue;
    }
#endif
    DEBUG(fprintf(stderr,"Got selitemobj 0x%lx\n", (unsigned long)selitemobj));

    realevents = 0;
    if (pfd->revents & (POLLIN | POLLPRI)) {
      realevents |= SELECTABLE_READ_READY;
    }
    if (pfd->revents & POLLOUT) {
      realevents |= SELECTABLE_WRITE_READY;
    }
    if (pfd->revents & (POLLERR | POLLHUP | POLLNVAL)) {
      realevents |= SELECTABLE_SELECT_ERROR;
    }

    (*env)->SetShortField(env, selitemobj, FID_seda_nbio_SelectItem_revents, realevents);
    DEBUG(fprintf(stderr,"Set revents\n"));

    if (count >= retitemarrlen) {
      fprintf(stderr,"WARNING: *** NBIO devPollImpl.doSelect(): count %d max_ret %d -- this is a bug, please contact [e-mail]\n", count, retitemarrlen);
    }
    (*env)->SetObjectArrayElement(env, retitemarr, count, selitemobj);

    if ((*env)->ExceptionOccurred(env)) {
      // Possible that the SelectItem was freed before we managed to do
      // the SetShortField/SetObjectArrayElement? Should not happen
      // if we are synchronized with deregister.
      fprintf(stderr,"WARNING: *** NBIO devPollImpl.doSelect() got exception: this is a bug - please contact [e-mail]\n");
      (*env)->ExceptionDescribe(env);
      // Clear it and skip this item
      (*env)->ExceptionClear(env);
      continue;
    }

    DEBUG(fprintf(stderr,"Set retitemarr[%d]\n", count));
    count++;
  }

  (*env)->MonitorExit(env, this); 
  return count;
}

#endif /* HAS_DEVPOLL */

/* SelectSetPollImpl *******************************************************/

JNIEXPORT jint JNICALL Java_seda_nbio_SelectSetPollImpl_doSelect(JNIEnv *env, jobject this, jint timeout) {
  jobjectArray itemarr;
  jobject selitemobj, fdobj;
  struct pollfd *ufds;
  int *ufds_map;
  int itemarrlen;
  int i, n;
  int ret;
  int num_ufds = 0;
  short events, realevents;

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

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

  itemarr = (jobjectArray)(*env)->GetObjectField(env, this, FID_seda_nbio_SelectSetPollImpl_itemarr);
  if (itemarr == NULL) {
    // This can happen if we have an empty SelectSet 
    return 0;
  }

  DEBUG(fprintf(stderr,"NBIO: doSelect: got itemarr\n"));

  itemarrlen = (*env)->GetArrayLength(env, itemarr);
  if (itemarrlen <= 0) {
    THROW_EXCEPTION(env, "java/lang/ArrayIndexOutOfBoundsException", "SelectItem[] array has size <= 0");
    return 0;
  }

  DEBUG(fprintf(stderr,"NBIO: doSelect: itemarrlen is %d\n", itemarrlen));

  DEBUG(fprintf(stderr,"NBIO: doSelect: got SelectItem class\n"));
  
  // Only allocate a ufd if events != 0
  for (i = 0; i < itemarrlen; i++) {
    selitemobj = (*env)->GetObjectArrayElement(env, itemarr, i);
    if (selitemobj == NULL) {
      fprintf(stderr,"NBIO: WARNING: itemarr[%d] is NULL! (itemarrlen=%d)\n", i, itemarrlen);
      THROW_EXCEPTION(env, "java/lang/NullPointerException", "SelectItem element is null");
      return 0;
    }
    events = (*env)->GetShortField(env, selitemobj, FID_seda_nbio_SelectItem_events);
    if (events != 0) num_ufds++;
  }

  if (num_ufds == 0) return 0;

  ufds = (struct pollfd *)malloc(sizeof(struct pollfd) * num_ufds);
  if (ufds == NULL) {
    THROW_EXCEPTION(env, "java/lang/OutOfMemoryError", "cannot allocate pollfd array");
    return 0;
  }
  DEBUG(fprintf(stderr,"NBIO: doSelect: allocated %d ufds\n", num_ufds));
  ufds_map = (int *)malloc(sizeof(int) * num_ufds);
  if (ufds_map == NULL) {
    THROW_EXCEPTION(env, "java/lang/OutOfMemoryError", "cannot allocate ufds_map");
    free(ufds);
    return 0;
  }
    
  n = 0;
  for (i = 0; i < itemarrlen; i++) {
    selitemobj = (*env)->GetObjectArrayElement(env, itemarr, i);
    if (selitemobj == NULL) {
      THROW_EXCEPTION(env, "java/lang/NullPointerException", "SelectItem element is null");
      free(ufds);
      free(ufds_map);
      return 0;
    }

    realevents = 0;
    events = (*env)->GetShortField(env, selitemobj, FID_seda_nbio_SelectItem_events);
    if (events != 0) {
      if (events & SELECTABLE_READ_READY) {
        realevents |= (POLLIN | POLLPRI);
      }
      if (events & SELECTABLE_WRITE_READY) {
        realevents |= POLLOUT;
      }
      ufds[n].events = realevents;
      DEBUG(fprintf(stderr,"NBIO: doSelect: ufds[%d].events is 0x%x\n", n, ufds[n].events));
      ufds[n].revents = 0;
      fdobj = (*env)->GetObjectField(env, selitemobj, FID_seda_nbio_SelectItem_fd);
      ufds[n].fd = (*env)->GetIntField(env, fdobj, FID_seda_nbio_NBIOFileDescriptor_fd);
      DEBUG(fprintf(stderr,"NBIO: doSelect: ufds[%d].fd is %d\n", n, ufds[n].fd));
      ufds_map[n] = i;
      n++;
    }
  }

  /* XXX MDW: poll() is interruptible. Under Linux, a signal (say, from 
   * the GC) might interrupt poll. For now I don't deal with this
   * properly - I just go ahead and return early from doSelect() if the 
   * call was interrupted.
   */
  DEBUG(fprintf(stderr,"NBIO: Doing poll, %d fds, timeout %d\n", n, timeout));

  ret = poll(ufds, num_ufds, timeout);

  DEBUG(fprintf(stderr,"NBIO: doSelect: did poll, timeout %d, ret is %d, errno is %d\n", timeout, ret, errno));

  if (ret == 0) {
    free(ufds);
    free(ufds_map);
    return 0;
  }
  if (ret < 0) {
    int myerrno = errno;
    // Don't throw an exception if we were interrupted
    if (myerrno != EINTR) {
      THROW_EXCEPTION(env, "java/io/IOException", strerror(myerrno));
    } 
    free(ufds);
    free(ufds_map);
    return 0;
  }

  for (n = 0; n < num_ufds; n++) {
    DEBUG(fprintf(stderr,"NBIO: doSelect: ufds[%d].revents is 0x%x\n", n, ufds[n].revents));

    if (ufds[n].revents != 0) {

      i = ufds_map[n];
      selitemobj = (*env)->GetObjectArrayElement(env, itemarr, i);
      if (selitemobj == NULL) {
        THROW_EXCEPTION(env, "java/lang/NullPointerException", "SelectItem element is null");
        free(ufds);
        free(ufds_map);
        return 0;
      }

      realevents = 0;
      if (ufds[n].revents & (POLLIN | POLLPRI)) {
        realevents |= SELECTABLE_READ_READY;
      }
      if (ufds[n].revents & POLLOUT) {
        realevents |= SELECTABLE_WRITE_READY;
      }
      if (ufds[n].revents & (POLLERR | POLLHUP | POLLNVAL)) {
        realevents |= SELECTABLE_SELECT_ERROR;
      }

      DEBUG(fprintf(stderr,"NBIO: doSelect: setting itemarr[%d].revents to 0x%x\n", i, realevents));
      (*env)->SetShortField(env, selitemobj, FID_seda_nbio_SelectItem_revents, realevents);
    }
  }

  free(ufds);
  free(ufds_map);

  DEBUG(fprintf(stderr,"NBIO: doSelect: returning %d\n", ret));
  return ret;

}