symphonynode.java

p2p仿真器。开发者可以工作在覆盖层中进行创造和测试逻辑算法或者创建和测试新的服务。PlanetSim还可以将仿真代码平稳转换为在Internet上的实验代码

package planet.symphony;
import java.util.Collection;
import java.util.HashSet;
import java.util.Hashtable;
import java.util.Iterator;
import java.util.List;
import java.util.Random;
import java.util.Set;
import java.util.TreeSet;
import java.util.Vector;

import planet.commonapi.Id;
import planet.commonapi.Message;
import planet.commonapi.Node;
import planet.commonapi.NodeHandle;
import planet.commonapi.RouteMessage;
import planet.commonapi.exception.InitializationException;
import planet.commonapi.results.ResultsConstraint;
import planet.generic.commonapi.behaviours.BehavioursPatternImpl;
import planet.generic.commonapi.factory.GenericFactory;
import planet.results.LinkStateResults;
import planet.simulate.Globals;
import planet.simulate.Logger;
import planet.simulate.Results;
import planet.symphony.messages.JoinMessage;
import planet.symphony.messages.NeighbourMessagePool;
import planet.util.Properties;
import planet.util.timer.TimerTaskImpl;

/**
 * 
 * This node is an implementation of the Symphony overlay. 
 * 
 * @author <a href="mailto:[e-mail]">Helio Tejedor</a>
 * @author <a href="mailto:[e-mail]">Marc Sanchez</a>
 * @author <a href="mailto:[e-mail]">Ruben Mondejar</a>
 * @author <a href="mailto:[e-mail]">Carles Pairot</a>
 * @author <a href="mailto:[e-mail]">Jordi Pujol</a>
 */

public class SymphonyNode
	extends planet.generic.commonapi.NodeImpl 
{
	
	/* ******************  CONSTANTS FOR MODE OF ROUTEMESSAGE *******/
	
	/**
	 * Mode value: Defines the start of communication that requires 
	 * response (REPLY). 
	 */
	public final static int REQUEST					= 0;
	/**
	 * Mode value: Defines the response of a communication.
	 */
	public final static int REPLY					= 1;
	/**
	 * Mode value: Defines a message's mode that requires only
	 * communication on one way.
	 */
	public final static int REFRESH  				= 2;		
	
	/* END ******************  CONSTANTS FOR MODE OF ROUTEMESSAGE *******/
	
	/* ******************  CONSTANTS FOR TYPE OF ROUTEMESSAGE *******/
	/**
	 * Type value: First message that is send by any node to get its
	 * inmediate successor.
	 */
	 public final static int QUERY_JOIN 			 = 0;
     public final static int SET_INFO   			 = 1;
     public final static int QUERY_CONNECT 			 = 2;
     public final static int ACCEPT_CONNECT 		 = 3;
     public final static int CANCEL_CONNECT 		 = 4;
     public final static int CLOSE_LONG_CONNECT 	 = 5;
     public final static int CLOSE_NEIGHBOUR_CONNECT = 6;
     /**
      * Type value: It identifies that this message contains an application 
      * level Message. 
      */
     public final static int DATA                     = 7;    
     
	/* END ******************  CONSTANTS FOR TYPE OF ROUTEMESSAGE *******/
	
	/* ******************  CONSTANTS FOR TYPE/MODE OF ROUTEMESSAGE *******/
	/**
	 * This String contains a string representation of each type
	 * value of the RouteMessage.
	 */
	public final static String[] TYPES = {
			"QUERY_JOIN", "SET_INFO", 
			"QUERY_CONNECT", "ACCEPT_CONNECT", 
			"CANCEL_CONNECT", "CLOSE_LONG_CONNECT", 
			"CLOSE_NEIGHBOUR_CONNECT", "DATA"};
	/**
	 * This String contains a string representation of each mode
	 * value of the RouteMessage.
	 */
	public final static String[] MODES = {"REQUEST","REPLY","REFRESH" };
	
	/* END ******************  CONSTANTS FOR TYPE/MODE OF ROUTEMESSAGE *******/
	
	
	
	/**
	 * To build new Id to find a new long distance.
	 */
	private static Random r = new Random();

	
	private List incommingSet;
	private List outcommingSet;
	public SortedKList neighbourSet; // with order !!!
    
	private double n;
	
	/* UpdateOutcommingList */
	public int retriesNewLongDistance;
    private boolean requestedNewLongDistance;
	
	public boolean alive;
	public boolean fixedNeighbours; 
    
    /* When the flag 'stabilized' is activated, makes F*24 stabilization processes
     * to guarantee that the neighbour set is the most possible correct.
     */
    public boolean statisticStabilized;
    public int statisticStabilizationSteps;
    
    private planet.commonapi.behaviours.BehavioursPool behPool; 
    
    private boolean modifiedNeighbours = true;
    
    /* only for temporal uses on the methods*/
    private Vector stabilizeVector = null;
    private NodeHandle stabilizeNH = null;
    private Vector estimationVector = null;
    private NodeHandle estimationNH = null;
    
    
    /**
     * Gets the maximum number of successors per node.
     * @return the maximum number of successors per node.
     */
    public static int getSuccessorsNumber()
    {
        return ((SymphonyProperties)Properties.overlayPropertiesInstance).maxSuccessorList;
    }
    
    /**
     * Gets the maximum number of long distance links per node.
     * @return the maximum number of long distance links per node.
     */
    public static int getLongDistanceNumber()
    {
        return ((SymphonyProperties)Properties.overlayPropertiesInstance).maxLongDistance;
    }
    
    
    /**
     * Initialize the internal data structure. 
     * @throws InitializationException
     */
    public SymphonyNode() throws InitializationException {
        super();
        alive = true;
        outcommingSet = new Vector(SymphonyNode.getLongDistanceNumber(),1);
        incommingSet = new Vector(SymphonyNode.getLongDistanceNumber(),1);
        fixedNeighbours = false;
        statisticStabilized = false;
        modifiedNeighbours = true;
        statisticStabilizationSteps = SymphonyNode.getSuccessorsNumber()*24;
        behPool = GenericFactory.getDefaultBehavioursPool();

        // Stabilize Timer
        setTimer(new StabilizeTask(),  ((SymphonyProperties)Properties.overlayPropertiesInstance).stabilizeSteps,
                ((SymphonyProperties)Properties.overlayPropertiesInstance).stabilizeSteps);
    }
	
    /**
     * Return the endpoints, based on (key,value) structure, where
     * the <b>key</b> is the Application name, and the <b>value</b> the own EndPoint instance.
     * @return The Hashtable with all the endpoint instances.
     */
	public Hashtable getEndPoints() {
		return this.endpoints;
	}
	
    /**
     * Builds a RouteMessage with the specified data.
     * @param from Communication source.
     * @param to   Communitacion destination.
     * @param type Communication type.
     * @param mode Communication mode, from the specified type.
     * @param m    Data to be sent in this communication.
     * @return The built RouteMessage or null if any error has ocurred.
     */
	public RouteMessage buildMessage(NodeHandle from, NodeHandle to, int type, int mode, Message m) {
		return buildMessage(from, to, route(to), type, mode, m, GenericFactory.generateKey());
	}
	
    /**
     * Builds a new RouteMessage with the specified data.
     * @param from Communication source.
     * @param to   Communication destination.
     * @param type Communication type.
     * @param mode Communication mode, from the specified type.
     * @param m    Data to be sent in this communication.
     * @param key  Unique communication key.
     * @return The built RouteMessage or null if any error has ocurred.
     */
	public RouteMessage buildMessage(NodeHandle from, NodeHandle to, int type, int mode, Message m, String key) {
		return buildMessage(from, to, route(to), type, mode, m, key);
	}
	
    /**
     * Builds a nw RouteMessage with the specified data.
     * @param from    Communication source.
     * @param to      Communication destination.
     * @param nextHop Communication next hop.
     * @param type    Communication type.
     * @param mode    Communication mode, from the specified type.
     * @param m       Data to be sent in this communication.
     * @return        The built RouteMessage or null if any error has ocurred.
     */
	public RouteMessage buildMessage(NodeHandle from, NodeHandle to, NodeHandle nextHop, int type, int mode, Message m) {
		return buildMessage(from, to, nextHop, type, mode, m, GenericFactory.generateKey());
	}
	
    /**
     * Builds a RouteMessage with the specified data.
     * @param from    Communication source.
     * @param to      Communication destination.
     * @param nextHop Communication next hop.
     * @param type    Communication type.
     * @param mode    Communication mode, from the specified type.
     * @param m       Data to be sent in this communication.
     * @param key     Unique communication key.
     * @return        The built RouteMessage or null if any error has ocurred.
     */
	public RouteMessage buildMessage(NodeHandle from, NodeHandle to, NodeHandle nextHop, int type, int mode, Message m, String key) {
		RouteMessage msg = null;
		try {
		    msg = GenericFactory.getMessage(key,from,to,nextHop,m,type,mode,null);
			//planet.results.LinkStateResults.newMessage(msg);
			//planet.results.LinkStateResults.updateOutcoming(this.id);
		} catch (InitializationException e) {
			key = null;
			Logger.log("ERROR: Cannot get a RouteMessage of MessagePool\n" + e.getMessage(), Logger.ERROR_LOG);
            GenericFactory.freeMessage(msg);
            msg = null;
		}
		return msg;
	}
    
    /**
     * Builds a new RouteMessage with all the values appeared in <b>toCopy</b>, and
     * the specified <b>nextHop</b>.
     * @param toCopy Message to be cloned.
     * @param nextHop Next hop in the route.
     * @return A valid RouteMessage or null if there are any error.
     */
    public RouteMessage buildMessage(RouteMessage toCopy, NodeHandle nextHop)
    {
        RouteMessage msg = null;
        try {
            msg = GenericFactory.getMessage(GenericFactory.generateKey()/*toCopy.getKey()*/,toCopy.getSource(),toCopy.getDestination(),nextHop,toCopy.getMessage(),toCopy.getType(),toCopy.getMode(),toCopy.getApplicationId());
            //planet.results.LinkStateResults.newMessage(msg);
            //planet.results.LinkStateResults.updateOutcoming(this.id);
        } catch (InitializationException e)
        {
            Logger.log("ERROR: Cannot get a RouteMessage of MessagePool\n" + e.getMessage(), Logger.ERROR_LOG);
            GenericFactory.freeMessage(msg);
            msg = null;
        }
        return msg;
    }
	
    /**
     * Build and send a new RouteMessage with the specified data.
     * @param from Communication source.
     * @param to   Communication destination.
     * @param type Communication type.
     * @param mode Communication mode, from the specified type.
     * @param m    Data to be sent in this communication.
     * @return     The generated unique key for this communication or null if cannot be sent.
     */
    private String sendMessage(NodeHandle from, NodeHandle to, int type, int mode, Message m) {
		return sendMessage(from, to, route(to), type, mode, m, GenericFactory.generateKey());
	}
	
    /**
     * Build and send a new RouteMessage with the specified data.
     * @param from Communication source.
     * @param to   Communication destination.
     * @param type Communication type.
     * @param mode Communication mode, from the specified type.
     * @param m    Data to be sent in this communication.
     * @param key  Unique communication key.
     * @return     The same key or null if cannot be sent.
     */
	private String sendMessage(NodeHandle from, NodeHandle to, int type, int mode, Message m, String key) {
		return sendMessage(from, to, route(to), type, mode, m, key);
	}
	
    /**
     * Build and send a new RouteMessage with the specified data.
     * @param from    Communication source.
     * @param to      Communication destination.
     * @param nextHop Communication next hop.
     * @param type    Communication type.
     * @param mode    Communication mode, from the specified type.
     * @param m       Data to be sent in this communication.
     * @return        The generated unique key for this communication or null if 
     *                cannot be sent.
     */
	private String sendMessage(NodeHandle from, NodeHandle to, NodeHandle nextHop, int type, int mode, Message m) {
		return sendMessage(from, to, nextHop, type, mode, m, GenericFactory.generateKey());
	}
	
    /**
     * Build and send a new RouteMessage with the specified data.
     * @param from    Communication source.
     * @param to      Communication destination.
     * @param nextHop Communication next hop.
     * @param type    Communication type.
     * @param mode    Communication mode, from the specified type.
     * @param m       Data to be sent in this communication.
     * @param key     Unique communication key.
     * @return        The same key or null if cannot be sent.
     */
	private String sendMessage(NodeHandle from, NodeHandle to, NodeHandle nextHop, int type, int mode, Message m, String key) {
	    String toReturn = key;
        RouteMessage msg = buildMessage(from,to,nextHop,type,mode,m,key);
        if (msg!=null)
        {
			//planet.results.LinkStateResults.newMessage(msg);
			//planet.results.LinkStateResults.updateOutcoming(this.id);
			sendMessage(msg);
		}
        else toReturn = null;
		return toReturn;
	}
	
    /**
     * Resends the RouteMessage to the specified <b>msg</b> next hop. 
     * @param msg RouteMessage to be resent.
     * @return The key of the RouteMessage or null if cannot be sent.
     */
	private String resendMessage(RouteMessage msg) {
		return resendMessage(msg, msg.getNextHopHandle());
	}
	
    /**
     * Resends the RouteMessage to the specified <b>nextHop</B>.
     * @param msg     RouteMessage to be resent.
     * @param nextHop Communication next hop.
     * @return        The routemessage key or null if cannot be sent.
     */
	private String resendMessage(RouteMessage msg, NodeHandle nextHop) {
		String key = null;
		key = msg.getKey();
		msg.setNextHopHandle(nextHop);
		//planet.results.LinkStateResults.newMessage(msg);
		//planet.results.LinkStateResults.updateOutcoming(this.id);
		if (!sendMessage(msg)) key = null;
		return key;
	}
	
    /**
     * Closes all stale long connections appeared in the <b>toRemoveFrom</b> list.
     * @param toRemoveFrom Long connections list.
     */
    private void closeStaleLongConnections(List toRemoveFrom) {
        int index = 0;
        NodeHandle candidate = null;
        
        while (index < toRemoveFrom.size())
        {
            candidate = (NodeHandle) toRemoveFrom.get(index);   
            if (neighbourSet.contains(candidate))
            {
                sendMessage(buildMessage(getLocalHandle(), candidate, 
                        SymphonyNode.CLOSE_LONG_CONNECT, SymphonyNode.REFRESH, null));
                toRemoveFrom.remove(index);
            } else
                index ++;
        }
    } 
    
    /**
     * Inform to stale neighbour that its connection has been closed.
     * @param closed
     */
    private void sendClosedNeighbour(Vector closed)
    {