nioselectsource.java

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

/* 
 * Copyright (c) 2001 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]>
 * 
 */

package seda.sandStorm.lib.aSocket.nio;

import seda.sandStorm.api.*;
import seda.sandStorm.lib.aSocket.SelectSourceIF;

import java.nio.channels.*;
import java.net.*;
import java.io.*;
import java.util.*;

/**
 * A NIOSelectSource is an implementation of SourceIF which pulls events from
 * the operating system via the NIO Selector interface. This can be thought
 * of as a 'shim' which turns a Selector into a SourceIF. 
 *
 * <p>NIOSelectSource also "balances" the set of events returned on 
 * subsequent calls to dequeue, in order to avoid biasing the servicing of
 * underlying O/S events to a particular order. This feature can be 
 * disabled by creating a SelectSource with the boolean flag 'do_balance'.
 *
 * <p><b>Important note:</b> This class is not threadsafe with respect
 * to multiple threads calling dequeue() or blocking_dequeue() at once.
 * Clients must synchronize their access to this class. 
 *
 * @author Matt Welsh 
 */
public class NIOSelectSource implements SelectSourceIF {

  private static final boolean DEBUG = false;

  private Selector selector;
  private SelectionKey ready[];
  private int ready_offset, ready_size;

  private boolean do_balance;
  private final int BALANCER_SEQUENCE_SIZE = 10000;
  private int balancer_seq[];
  private int balancer_seq_off;
  private Object blocker;
  private String name = "(unknown)";

  // XXX MDW HACKING
  public Selector getSelector() {
    return selector;
  }

  /**
   * Create a new empty SelectSource. This SelectSource will perform
   * event balancing.
   */
  public NIOSelectSource() {
    this(true);
  }

  /**
   * Create a new empty SelectSource. 
   *
   * @param do_balance Indicates whether this SelectSource should perform
   * event balancing.
   */
  public NIOSelectSource(boolean do_balance) {
    blocker = new Object();
    try {
        selector = Selector.open();
    } catch (IOException e) {
        System.err.println("NIOSelectSource ("+name+"): error creating selector: " + e);
    }
    ready = null;
    ready_offset = ready_size = 0;
    this.do_balance = do_balance;

    if  (DEBUG) System.err.println("NIOSelectSource created, do_balance = "+do_balance);

    if (do_balance) initBalancer();
  }

  /**
   * Register a SelectItem with the SelectSource. The SelectItem should 
   * generally correspond to a Selectable along with a set of event flags
   * that we wish this SelectSource to test for. 
   *
   * <p>The user is allowed to modify the event flags in the SelectItem 
   * directly (say, to cause the SelectSource ignore a given SelectItem for 
   * the purposes of future calls to one of the dequeue methods). However,
   * modifying the event flags may not be synchronous with calls to dequeue -
   * generally because SelectSource maintains a cache of recently-received
   * events.
   */
  public Object register(Object nio_sc_obj, int ops) {
    if (DEBUG) System.err.println("NIOSelectSource ("+name+"): register " +nio_sc_obj +
        " : " + this);

    if (! (nio_sc_obj instanceof SelectableChannel)) {
        System.err.println(
           "register() called with non SelectableChannel argument.  " +
           "Should not happen!!"
        );
        return null;
    }

    SelectableChannel nio_sc = (SelectableChannel)nio_sc_obj;

    synchronized (blocker) {
      SelectionKey ret;
      try {
      	ret = nio_sc.register(selector, ops);
      } catch (ClosedChannelException cce) {
      	System.err.println("Closed Channel Exception: " + cce);
     	ret = null;
      }
      if (DEBUG) System.err.println("returning " + ret);
      if (DEBUG) System.err.println("numactive = " + numActive());
      blocker.notify();
      return ret;
    }
  }


  public void register(Object sel) {
      System.err.println(
        "Single argument register() called on NIOSelectSource.  " +
        "Should not happen!!"
      );
      return;
  }
  /**
   * Deregister a SelectItem with this SelectSource.
   * Note that after calling deregister, subsequent calls to dequeue
   * may in fact return this SelectItem as a result. This is because
   * the SelectQueue internally caches results.
   */
  public void deregister(Object selkey_obj) {
    if (DEBUG) System.err.println("NIOSelectSource ("+name+"): deregister "+selkey_obj);

    if (! (selkey_obj instanceof SelectionKey)) {
        System.err.println(
            "deregister() called on NIOSelectSource with non SelectionKey " +
            "argument.  Should not happen!!"
        );
        return;
    }

    synchronized (blocker) {
      SelectionKey selkey = (SelectionKey)selkey_obj;
      selkey.cancel();
      /* This must be done so that calls to close() actually close. */
      try {
	selector.selectNow();
      } catch (IOException ioe) {
	// Ignore
      }
      blocker.notify();
    }
  }

  /**
   * Must be called if the 'events' mask of any SelectItem registered
   * with this SelectSource changes. Pushes event mask changes down to
   * the underlying event-dispatch mechanism.
   */
  public void update() {
    // selset.update();
  }

  /**
   * Must be called if the 'events' mask of this SelectItem (which
   * must be registered with this SelectSource) changes. Pushes 
   * event mask changes down to the underlying event-dispatch mechanism.
   */
  public void update(Object sel) {
      System.err.println(
        "update() called on NIOSelectSource with argument.  " +
        "should not happen!!"
      );
      return;
  }


  /**
   * Return the number of SelectItems registered with the SelectSource.
   */
  public int numRegistered() {
    return selector.keys().size();
  }
  
  /**
   * Return the number of active SelectItems registered with the SelectSource.
   * An active SelectItem is one defined as having a non-zero events
   * interest mask.
   */
  public int numActive() {
    // does this mean number with a non-zero request mask, or number
    // in selectedKeys()
    Iterator key_iter = selector.keys().iterator();
    SelectionKey sk;
    int n_active = 0;
    while (key_iter.hasNext()) {
        sk = (SelectionKey)key_iter.next();
        if (sk.isValid() && sk.interestOps() != 0) n_active++;
    }
    return n_active;
  }

  /**
   * Return the number of elements waiting in the queue (that is,
   * which don't require a SelectSet poll operation to retrieve).
   */
  public int size() {
    synchronized (this) {
      return (ready_size - ready_offset);
    }
  }

  /** 
   * Dequeues the next element from the SelectSource without blocking.
   * Returns null if no entries available.
   */
  public QueueElementIF dequeue() {
    if (selector.keys().size() == 0) return null;

    if ((ready_size == 0) || (ready_offset == ready_size)) {
      doPoll(0);
    } 
    if (ready_size == 0) return null;
    return new NIOSelectorQueueElement(ready[ready_offset++]);
  }

  /** 
   * Dequeues all elements which are ready from the SelectSource.
   * Returns null if no entries available.
   */
  public QueueElementIF[] dequeue_all() {
    if (selector.keys().size() == 0) return null;

    if ((ready_size == 0) || (ready_offset == ready_size)) {
      doPoll(0);
    } 
    if (ready_size == 0) return null;
    NIOSelectorQueueElement ret[] =
        new NIOSelectorQueueElement[ready_size-ready_offset];