selectormanager.java

pastry的java实现的2.0b版

/*************************************************************************

"FreePastry" Peer-to-Peer Application Development Substrate 

Copyright 2002, Rice University. All rights reserved.

Redistribution and use in source and binary forms, with or without
modification, are permitted provided that the following conditions are
met:

- Redistributions of source code must retain the above copyright
notice, this list of conditions and the following disclaimer.

- Redistributions in binary form must reproduce the above copyright
notice, this list of conditions and the following disclaimer in the
documentation and/or other materials provided with the distribution.

- Neither  the name  of Rice  University (RICE) nor  the names  of its
contributors may be  used to endorse or promote  products derived from
this software without specific prior written permission.

This software is provided by RICE and the contributors on an "as is"
basis, without any representations or warranties of any kind, express
or implied including, but not limited to, representations or
warranties of non-infringement, merchantability or fitness for a
particular purpose. In no event shall RICE or contributors be liable
for any direct, indirect, incidental, special, exemplary, or
consequential damages (including, but not limited to, procurement of
substitute goods or services; loss of use, data, or profits; or
business interruption) however caused and on any theory of liability,
whether in contract, strict liability, or tort (including negligence
or otherwise) arising in any way out of the use of this software, even
if advised of the possibility of such damage.

********************************************************************************/
package rice.selector;

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

import rice.Destructable;
import rice.environment.logging.*;
import rice.environment.time.TimeSource;

/**
 * This class is the class which handles the selector, and listens for activity.
 * When activity occurs, it figures out who is interested in what has happened,
 * and hands off to that object.
 *
 * @version $Id: SelectorManager.java 3144 2006-03-15 16:19:17Z jstewart $
 * @author Alan Mislove
 */
public class SelectorManager extends Thread implements Timer, Destructable {

  // the underlying selector used
  /**
   * DESCRIBE THE FIELD
   */
  protected Selector selector;

  // a list of the invocations that need to be done in this thread
  /**
   * DESCRIBE THE FIELD
   */
  protected LinkedList invocations;

  // the list of handlers which want to change their key
  /**
   * DESCRIBE THE FIELD
   */
  protected HashSet modifyKeys;

  // the list of keys waiting to be cancelled
  /**
   * DESCRIBE THE FIELD
   */
  protected HashSet cancelledKeys;

  // the set used to store the timer events
  /**
   * DESCRIBE THE FIELD
   */
  protected TreeSet timerQueue = new TreeSet();

  // the next time the selector is schedeled to wake up
  /**
   * DESCRIBE THE FIELD
   */
  protected long wakeupTime = 0;

  /**
   * DESCRIBE THE FIELD
   */
  protected TimeSource timeSource;

  long lastTime = 0;

//  protected LogManager log;
  /**
   * DESCRIBE THE FIELD
   */
  protected Logger logger;

  /**
   * DESCRIBE THE FIELD
   */
  protected String instance;

  /**
   * DESCRIBE THE FIELD
   */
  protected boolean running = true;

  ArrayList loopObservers = new ArrayList();

  // the maximal time to sleep on a select operation
  /**
   * DESCRIBE THE FIELD
   */
  public static int TIMEOUT = 500;

  /**
   * Constructor, which is private since there is only one selector per JVM.
   *
   * @param instance DESCRIBE THE PARAMETER
   * @param timeSource DESCRIBE THE PARAMETER
   * @param log DESCRIBE THE PARAMETER
   */
  public SelectorManager(String instance,
                         TimeSource timeSource, LogManager log) {
    super(instance == null ? "Selector Thread" : "Selector Thread -- "
      + instance);
    this.instance = instance;
    this.logger = log.getLogger(getClass(), instance);
    this.invocations = new LinkedList();
    this.modifyKeys = new HashSet();
    this.cancelledKeys = new HashSet();
    this.timeSource = timeSource;

    // attempt to create selector
    try {
      selector = Selector.open();
    } catch (IOException e) {
      System.out.println("SEVERE ERROR (SelectorManager): Error creating selector "
        + e);
    }
    lastTime = timeSource.currentTimeMillis();
    start();
  }

  /**
   * Utility method which returns the SelectionKey attached to the given
   * channel, if one exists
   *
   * @param channel The channel to return the key for
   * @return The key
   */
  public SelectionKey getKey(SelectableChannel channel) {
    return channel.keyFor(selector);
  }

  /**
   * Debug method which returns the number of pending invocations
   *
   * @return The number of pending invocations
   */
  public int getNumInvocations() {
    return invocations.size();
  }

  /**
   * Method which synchroniously returns the first element off of the
   * invocations list.
   *
   * @return An item from the invocations list
   */
  protected synchronized Runnable getInvocation() {
    if (invocations.size() > 0) {
      return (Runnable) invocations.removeFirst();
    } else {
      return null;
    }
  }

  /**
   * Method which synchroniously returns on element off of the modifyKeys list
   *
   * @return An item from the invocations list
   */
  protected synchronized SelectionKey getModifyKey() {
    if (modifyKeys.size() > 0) {
      Object result = modifyKeys.iterator().next();
      modifyKeys.remove(result);
      return (SelectionKey) result;
    } else {
      return null;
    }
  }

  /**
   * Returns whether or not this thread of execution is the selector thread
   *
   * @return Whether or not this is the selector thread
   */
  public boolean isSelectorThread() {
    return (Thread.currentThread() == this);
  }

  /**
   * Returns the timer associated with this SelectorManager (in this case, it is
   * this).
   *
   * @return The associated timer
   */
  public Timer getTimer() {
    return this;
  }

  /**
   * Gets the Selector attribute of the SelectorManager object
   *
   * @return The Selector value
   */
  public Selector getSelector() {
    return selector;
  }

  /**
   * Method which asks the Selector Manager to add the given key to the
   * cancelled set. If noone calls register on this key during the rest of this
   * select() operation, the key will be cancelled. Otherwise, it will be
   * returned as a result of the register operation.
   *
   * @param key The key to cancel
   */
  public void cancel(SelectionKey key) {
    if (key == null) {
      throw new NullPointerException();
    }

    cancelledKeys.add(key);
  }

  /**
   * Registers a new channel with the selector, and attaches the given
   * SelectionKeyHandler as the handler for the newly created key. Operations
   * which the hanlder is interested in will be called as available.
   *
   * @param channel The channel to regster with the selector
   * @param handler The handler to use for the callbacks
   * @param ops The initial interest operations
   * @return The SelectionKey which uniquely identifies this channel
   * @exception IOException DESCRIBE THE EXCEPTION
   */
  public SelectionKey register(SelectableChannel channel,
                               SelectionKeyHandler handler, int ops) throws IOException {
    if ((channel == null) || (handler == null)) {
      throw new NullPointerException();
    }

    SelectionKey key = channel.register(selector, ops, handler);
    cancelledKeys.remove(key);

    return key;
  }

  /**
   * This method schedules a runnable task to be done by the selector thread
   * during the next select() call. All operations which modify the selector
   * should be done using this method, as they must be done in the selector
   * thread.
   *
   * @param d The runnable task to invoke
   */
  public synchronized void invoke(Runnable d) {
    if (d == null) {
      throw new NullPointerException();
    }

    invocations.add(d);
    selector.wakeup();
  }

  /**
   * Adds a selectionkey handler into the list of handlers which wish to change
   * their keys. Thus, modifyKeys() will be called on the next selection
   * operation
   *
   * @param key The key which is to be chanegd
   */
  public synchronized void modifyKey(SelectionKey key) {
    if (key == null) {
      throw new NullPointerException();
    }

    modifyKeys.add(key);
    selector.wakeup();
  }

  /**
   * This method is to be implemented by a subclass to do some task each loop.
   */
  protected void onLoop() {
  }

  /**
   * This method starts the socket manager listening for events. It is designed
   * to be started when this thread's start() method is invoked.
   */
  public void run() {
    try {
      //System.out.println("SelectorManager starting...");
      if (logger.level <= Logger.INFO) {
        logger.log("SelectorManager -- " + instance + " starting...");
      }

      lastTime = timeSource.currentTimeMillis();
      // loop while waiting for activity
      while (running) {
        notifyLoopListeners();

        // NOTE: This is so we aren't always holding the selector lock when we
        // get context switched
        Thread.yield();
        executeDueTasks();
        onLoop();
        doInvocations();
        doSelections();
        synchronized (selector) {
          int selectTime = SelectorManager.TIMEOUT;
          if (timerQueue.size() > 0) {
            TimerTask first = (TimerTask) timerQueue.first();
            selectTime = (int) (first.nextExecutionTime - timeSource.currentTimeMillis());
          }

          select(selectTime);

          if (cancelledKeys.size() > 0) {
            Iterator i = cancelledKeys.iterator();

            while (i.hasNext()) {
              ((SelectionKey) i.next()).cancel();
            }

            cancelledKeys.clear();