chordimpl.java

一个对等计算发生器

/***************************************************************************
 *                                                                         *
 *                             ChordFuture.java                            *
 *                            -------------------                          *
 *   date                 : 16.08.2005                                     *
 *   copyright            : (C) 2004/2005 Distributed and                  *
 *								Mobile Systems Group                       *
 *                              Lehrstuhl fuer Praktische Informatik       *
 *                              Universitaet Bamberg                       *
 *                              http://www.lspi.wiai.uni-bamberg.de/       *
 *   email                : [e-mail]              *
 *   						[e-mail]            *
 *                                                                         *
 *                                                                         *
 ***************************************************************************/

/***************************************************************************
 *                                                                         *
 *   This program is free software; you can redistribute it and/or modify  *
 *   it under the terms of the GNU General Public License as published by  *
 *   the Free Software Foundation; either version 2 of the License, or     *
 *   (at your option) any later version.                                   *
 *                                                                         *
 *   A copy of the license can be found in the license.txt file supplied   *
 *   with this software or at: http://www.gnu.org/copyleft/gpl.html        *
 *                                                                         *
 ***************************************************************************/
package de.uniba.wiai.lspi.chord.service.impl;

import java.io.Serializable;
import java.util.HashSet;
import java.util.Set;
import java.util.concurrent.Executor;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

import de.uniba.wiai.lspi.chord.com.CommunicationException;
import de.uniba.wiai.lspi.chord.com.Entry;
import de.uniba.wiai.lspi.chord.com.Node;
import de.uniba.wiai.lspi.chord.com.Proxy;
import de.uniba.wiai.lspi.chord.com.RefsAndEntries;
import de.uniba.wiai.lspi.chord.data.ID;
import de.uniba.wiai.lspi.chord.data.URL;
import de.uniba.wiai.lspi.chord.service.AsynChord;
import de.uniba.wiai.lspi.chord.service.Chord;
import de.uniba.wiai.lspi.chord.service.ChordCallback;
import de.uniba.wiai.lspi.chord.service.ChordFuture;
import de.uniba.wiai.lspi.chord.service.ChordRetrievalFuture;
import de.uniba.wiai.lspi.chord.service.Key;
import de.uniba.wiai.lspi.chord.service.Report;
import de.uniba.wiai.lspi.chord.service.ServiceException;
import de.uniba.wiai.lspi.util.logging.Logger;

/**
 * Implements all operations which can be invoked on the local node.
 * 
 * @author Karsten Loesing
 * @version 1.0.1
 */
public final class ChordImpl implements Chord, Report, AsynChord {

	/**
	 * Time in seconds until the stabilize task is started for the first time.
	 */
	private static final int STABILIZE_TASK_START = Integer
			.parseInt(System
					.getProperty("de.uniba.wiai.lspi.chord.service.impl.ChordImpl.StabilizeTask.start"));

	/**
	 * Time in seconds between two invocations of the stabilize task.
	 */
	private static final int STABILIZE_TASK_INTERVAL = Integer
			.parseInt(System
					.getProperty("de.uniba.wiai.lspi.chord.service.impl.ChordImpl.StabilizeTask.interval"));

	/**
	 * Time in seconds until the fix finger task is started for the first time.
	 */
	private static final int FIX_FINGER_TASK_START = Integer
			.parseInt(System
					.getProperty("de.uniba.wiai.lspi.chord.service.impl.ChordImpl.FixFingerTask.start"));

	/**
	 * Time in seconds between two invocations of the fix finger task.
	 */
	private static final int FIX_FINGER_TASK_INTERVAL = Integer
			.parseInt(System
					.getProperty("de.uniba.wiai.lspi.chord.service.impl.ChordImpl.FixFingerTask.interval"));

	/**
	 * Time in seconds until the check predecessor task is started for the first
	 * time.
	 */
	private static final int CHECK_PREDECESSOR_TASK_START = Integer
			.parseInt(System
					.getProperty("de.uniba.wiai.lspi.chord.service.impl.ChordImpl.CheckPredecessorTask.start"));

	/**
	 * Time in seconds between two invocations of the check predecessor task.
	 */
	private static final int CHECK_PREDECESSOR_TASK_INTERVAL = Integer
			.parseInt(System
					.getProperty("de.uniba.wiai.lspi.chord.service.impl.ChordImpl.CheckPredecessorTask.interval"));

	/**
	 * Number of references in the successor list.
	 */
	private static final int NUMBER_OF_SUCCESSORS = Integer
			.parseInt(System
					.getProperty("de.uniba.wiai.lspi.chord.service.impl.ChordImpl.successors"));

	/**
	 * Object logger.
	 */
	protected Logger logger;

	/**
	 * Reference on that part of the node implementation which is accessible by
	 * other nodes; if <code>null</code>, this node is not connected
	 */
	private NodeImpl localNode;

	/**
	 * Entries stored at this node, including replicas.
	 */
	private Entries entries;

	/**
	 * Executor service for local maintenance tasks.
	 */
	private ScheduledExecutorService maintenanceTasks;

	/**
	 * Executor service for asynch requests.
	 */
	private ExecutorService asyncExecutor;

	/**
	 * References to remote nodes.
	 */
	protected References references;

	/**
	 * Reference on hash function (singleton instance).
	 */
	private HashFunction hashFunction;

	/**
	 * This node's URL.
	 */
	private URL localURL;

	/**
	 * This node's ID.
	 */
	private ID localID;

	/* constructor */

	/**
	 * Creates a new instance of ChordImpl which initially is disconnected.
	 * Constructor is hidden. Only constructor.
	 */
	public ChordImpl() {
		this.logger = Logger.getLogger(ChordImpl.class + ".unidentified");
		this.logger.debug("Logger initialized.");

		this.maintenanceTasks = new ScheduledThreadPoolExecutor(3);
		this.asyncExecutor = Executors.newCachedThreadPool();
		this.hashFunction = HashFunction.getHashFunction();
	}

	/**
	 * @return The Executor executing asynchronous request. 
	 */
	final Executor getAsyncExecutor() {
		if (this.asyncExecutor == null) {
			throw new NullPointerException("ChordImpl.asyncExecutor is null!");
		}
		return this.asyncExecutor;
	}

	/* implementation of Chord interface */

	public final URL getURL() {
		return this.localURL;
	}

	public final void setURL(URL nodeURL) {

		if (nodeURL == null) {
			NullPointerException e = new NullPointerException(
					"Cannot set URL to null!");
			this.logger.error("Null pointer", e);
			throw e;
		}

		if (this.localNode != null) {
			IllegalStateException e = new IllegalStateException(
					"URL cannot be set after creating or joining a network!");
			this.logger.error("Illegal state.", e);
			throw e;
		}

		this.localURL = nodeURL;

		this.logger.info("URL was set to " + nodeURL);
	}

	public final ID getID() {
		return this.localID;
	}

	public final void setID(ID nodeID) {

		if (nodeID == null) {
			NullPointerException e = new NullPointerException(
					"Cannot set ID to null!");
			this.logger.error("Null pointer", e);
			throw e;
		}

		if (this.localNode != null) {
			IllegalStateException e = new IllegalStateException(
					"ID cannot be set after creating or joining a network!");
			this.logger.error("Illegal state.", e);
			throw e;
		}

		this.localID = nodeID;
		this.logger = Logger.getLogger(ChordImpl.class + "." + this.localID);
	}

	public final void create() throws ServiceException {

		// is node already connected?
		if (this.localNode != null) {
			throw new ServiceException(
					"Cannot create network; node is already connected!");
		}

		// has nodeURL been set?
		if (this.localURL == null) {
			throw new ServiceException("Node URL is not set yet!");
		}

		// if necessary, generate nodeID out of nodeURL
		if (this.getID() == null) {
			this.setID(this.hashFunction.createUniqueNodeID(this.localURL));
		}

		// establish connection
		this.createHelp();

	}

	public final void create(URL localURL1) throws ServiceException {

		// check if parameters are valid
		if (localURL1 == null) {
			throw new NullPointerException(
					"At least one parameter is null which is not permitted!");
		}

		// is node already connected?
		if (this.localNode != null) {
			throw new ServiceException(
					"Cannot create network; node is already connected!");
		}

		// set nodeURL
		this.localURL = localURL1;

		// if necessary, generate nodeID out of nodeURL
		if (this.getID() == null) {
			this.setID(this.hashFunction.createUniqueNodeID(this.localURL));
		}

		// establish connection
		this.createHelp();

	}

	public final void create(URL localURL1, ID localID1)
			throws ServiceException {

		// check if parameters are valid
		if (localURL1 == null || localID1 == null) {
			throw new IllegalArgumentException(
					"At least one parameter is null which is not permitted!");
		}

		// is node already connected?
		if (this.localNode != null) {
			throw new ServiceException(
					"Cannot create network; node is already connected!");
		}

		// set nodeURL
		this.localURL = localURL1;

		// set nodeID
		this.setID(localID1);

		// establish connection
		this.createHelp();

	}

	/**
	 * Performs all necessary tasks for creating a new Chord ring. Assumes that
	 * localID and localURL are correctly set. Is invoked by the methods
	 * {@link #create()}, {@link #create(URL)}, and {@link #create(URL, ID)}
	 * only.
	 */
	private final void createHelp() {

		this.logger.debug("Help method for creating a new Chord ring invoked.");

		// create local repository for entries
		this.entries = new Entries();

		// create local repository for node references
		this.references = new References(this.getID(), NUMBER_OF_SUCCESSORS,
				this.entries);

		// create NodeImpl instance for communication
		this.localNode = new NodeImpl(this, this.getID(), this.localURL,
				this.references, this.entries);

		// create tasks for fixing finger table, checking predecessor and
		// stabilizing
		this.createTasks();

		// accept content requests from outside
		this.localNode.acceptEntries();

	}

	/**
	 * Creates the tasks that must be executed periodically to maintain the
	 * Chord overlay network and schedules them with help of a
	 * {@link ScheduledExecutorService}.
	 */
	private final void createTasks() {

		// start thread which periodically stabilizes with successor
		this.maintenanceTasks.scheduleWithFixedDelay(new StabilizeTask(
				this.localNode, this.references),
				ChordImpl.STABILIZE_TASK_START,
				ChordImpl.STABILIZE_TASK_INTERVAL, TimeUnit.SECONDS);

		// start thread which periodically attempts to fix finger table
		this.maintenanceTasks.scheduleWithFixedDelay(new FixFingerTask(
				this.localNode, this.getID(), this.references),
				ChordImpl.FIX_FINGER_TASK_START,
				ChordImpl.FIX_FINGER_TASK_INTERVAL, TimeUnit.SECONDS);

		// start thread which periodically checks whether predecessor has
		// failed
		this.maintenanceTasks.scheduleWithFixedDelay(new CheckPredecessorTask(
				this.references), ChordImpl.CHECK_PREDECESSOR_TASK_START,
				ChordImpl.CHECK_PREDECESSOR_TASK_INTERVAL, TimeUnit.SECONDS);
	}

	public final void join(URL bootstrapURL) throws ServiceException {

		// check if parameters are valid
		if (bootstrapURL == null) {
			throw new NullPointerException(
					"At least one parameter is null which is not permitted!");
		}

		// is node already connected?
		if (this.localNode != null) {
			throw new ServiceException(
					"Cannot join network; node is already connected!");
		}

		// has nodeURL been set?
		if (this.localURL == null) {
			throw new ServiceException("Node URL is not set yet! Please "
					+ "set URL with help of setURL(URL) "
					+ "before invoking join(URL)!");
		}

		// if necessary, generate nodeID out of nodeURL
		if (this.getID() == null) {
			this.setID(this.hashFunction.createUniqueNodeID(this.localURL));
		}

		// establish connection
		this.joinHelp(bootstrapURL);

	}

	public final void join(URL localURL1, URL bootstrapURL)
			throws ServiceException {

		// check if parameters are valid
		if (localURL1 == null || bootstrapURL == null) {
			throw new NullPointerException(
					"At least one parameter is null which is not permitted!");
		}

		// is node already connected?
		if (this.localNode != null) {
			throw new ServiceException(
					"Cannot join network; node is already connected!");
		}

		// set nodeURL
		this.localURL = localURL1;

		// if necessary, generate nodeID out of nodeURL
		if (this.getID() == null) {
			this.setID(this.hashFunction.createUniqueNodeID(this.localURL));
		}

		// establish connection
		this.joinHelp(bootstrapURL);

	}

	public final void join(URL localURL1, ID localID1, URL bootstrapURL)
			throws ServiceException {

		// check if parameters are valid
		if (localURL1 == null || localID1 == null || bootstrapURL == null) {
			throw new NullPointerException(
					"At least one parameter is null which is not permitted!");
		}

		// is node already connected?
		if (this.localNode != null) {
			throw new ServiceException(
					"Cannot join network; node is already connected!");
		}

		// set nodeURL
		this.localURL = localURL1;

		// set nodeID
		this.setID(localID1);

		// establish connection
		this.joinHelp(bootstrapURL);

	}

	/**
	 * Performs all necessary tasks for joining an existing Chord ring. Assumes
	 * that localID and localURL are correctly set. Is invoked by the methods
	 * {@link #join(URL)}, {@link #join(URL, URL)}, and
	 * {@link #join(URL, ID, URL)} only.
	 * 
	 * @param bootstrapURL
	 *            URL of bootstrap node.
	 * @throws ServiceException
	 *             If anything goes wrong during the join process.
	 */
	private final void joinHelp(URL bootstrapURL) throws ServiceException {

		// create local repository for entries
		this.entries = new Entries();

		// create local repository for node references
		this.references = new References(this.getID(), NUMBER_OF_SUCCESSORS,
				this.entries);