mspastryprotocol.java

pastry 协议在peersim下的仿真环境。测试无误。在eclipse下打开源文件夹为工程即可使用和做仿真实验。

package peersim.pastry;

/**
 *
 *
 * <p>Title: MSPASTRY</p>
 *
 * <p>Description: MsPastry implementation for PeerSim</p>
 *
 * <p>Copyright: Copyright (c) 2007</p>
 *
 * <p>Company: The Pastry Group</p>
 *
 * @author Elisa Bisoffi, Manuel Cortella
 * @version 1.0
 */

import java.math.*;

import peersim.config.*;
import peersim.core.*;
import peersim.edsim.*;
import peersim.transport.*;
import java.util.Comparator;

//__________________________________________________________________________________________________
public class MSPastryProtocol implements Cloneable, EDProtocol {
    //______________________________________________________________________________________________
    /**
     * Event Handler container for managing the receiving of a message
     *
     * <p>Title: MSPASTRY</p>
     *
     * <p>Description: MsPastry implementation for PeerSim</p>
     *
     * <p>Copyright: Copyright (c) 2007</p>
     *
     * <p>Company: The Pastry Group</p>
     *
     * @author Elisa Bisoffi, Manuel Cortella
     * @version 1.0
     */
    public static interface Listener {
        /**
         * This method is called every time a message is received
         * @param m Message
         */
        public void receive(Message m);
    }

    /**
     * Listener assingned to the receiving of a message. If null it is not called
     */
    private Listener listener;

    //______________________________________________________________________________________________
    /**
     * allows to change/clear the listener
     * @param l Listener
     */
    public void setListener(Listener l) {
    	listener = l;
    }

    //______________________________________________________________________________________________
    private static final String PAR_TRANSPORT = "transport";
    private static String prefix = null;
    private UnreliableTransport transport;
    private int tid;
    private int mspastryid;
    private boolean cleaningScheduled = false;

    /**
     * allow to call the cleaning service initializer only once
     */
    private static boolean _ALREADY_INSTALLED = false;

    //______________________________________________________________________________________________
    /**
     * nodeId of this pastry node
     */
    public BigInteger nodeId;

    /**
     * routing table of this pastry node
     */
    public RoutingTable routingTable;

    /**
     * leaf set of this pastry node
     */
    public LeafSet leafSet;


    //______________________________________________________________________________________________
    /**
     * Replicate this object by returning an identical copy.
     * it put the eye on the fact that only the peersim initializer call this
     * method and we expects to replicate every time a non-initialized table.
     * Thus the method clone() do not fill any particular field;
     * @return Object
     */
    public Object clone() {

        MSPastryProtocol dolly = new MSPastryProtocol(MSPastryProtocol.prefix);
        dolly.routingTable = (RoutingTable)this.routingTable.clone();
        dolly.leafSet = (LeafSet)this.leafSet.clone();
        //dolly.nodeId is not copied, because ID is unique!
        return dolly;
    }


    //______________________________________________________________________________________________
    /**
     * Used only by the initializer when creating the prototype Every other instance call CLONE to
     * create the new object. clone could not use this constructor, preferring a more quick
     * constructor
     *
     * @param prefix String
     */
    public MSPastryProtocol(String prefix) {
        this.nodeId = null;              // empty nodeId
        MSPastryProtocol.prefix = prefix;

        _init();

        routingTable = new RoutingTable(MSPastryCommonConfig.BITS / MSPastryCommonConfig.B, Util.pow2(MSPastryCommonConfig.B));
        leafSet = new LeafSet(BigInteger.ZERO, MSPastryCommonConfig.L);

        tid = Configuration.getPid(prefix + "." + PAR_TRANSPORT);

    }

    //______________________________________________________________________________________________
    /**
     * This subrouting is called only once and allow to inizialize the internal state of the
     * MSPastreyProtocol. Every node shares the same configuration, so it is sufficient caling
     * ont this this sub in order to set up the configuration
     */
    private void _init() {
        if (_ALREADY_INSTALLED) return;

        int b=0, l=0, base=0;
        final String PAR_B = "B";
        final String PAR_L = "L";


        b = Configuration.getInt(prefix + "." + PAR_B, 4);
        l = Configuration.getInt(prefix + "." + PAR_L, MSPastryCommonConfig.BITS / b);
        base = Util.pow2(b);

        MSPastryCommonConfig.B = b;
        MSPastryCommonConfig.L = l;
        MSPastryCommonConfig.BASE = base;

        e(MSPastryCommonConfig.info()+"\n");

        _ALREADY_INSTALLED = true;
    }

    //______________________________________________________________________________________________
    /**
     * called when a Lookup message is ready to be passed through the upper/application level, by
     * calling the "message received" event handler (listener). It also provide some statistical
     * update
     * @param m Message
     */
    private void deliver(Message m) {
        //statistiche utili all'observer
        MSPastryObserver.hopStore.add(m.nrHops-1);
        long timeInterval = (CommonState.getTime())-(m.timestamp);
        MSPastryObserver.timeStore.add(timeInterval);

        if (listener != null) {
            listener.receive(m);
        }

    }

    //______________________________________________________________________________________________
    /**
     * given one nodeId, it search through the network its node reference, by performing binary serach
     * (we concern about the ordering of the network).
     * @param searchNodeId BigInteger
     * @return Node
     */
    private Node nodeIdtoNode(BigInteger searchNodeId) {
        if (searchNodeId==null) return null;

        int inf = 0;
        int sup = Network.size() - 1;
        int m;

        while (inf <= sup) {
            m = (inf + sup) / 2;

            BigInteger mId = ((MSPastryProtocol) Network.get(m).getProtocol(mspastryid)).nodeId;

            if (mId.equals(searchNodeId))
                return Network.get(m);

            if (mId.compareTo(searchNodeId) < 0)
                inf = m + 1;
            else
                sup = m - 1;
        }

        /**
         * La ricerca binaria � veloce ed efficiente, ma qualche volta pu� capitare che l'array dei
         * nodi di Network non sia ordinato, quindi si applica ora la ricerca sequenziale.
         * Se nemmeno la ricerca sequenziale trova il nodo cercato, viene ritornato null
         */
        BigInteger mId;
        for (int i= Network.size()-1; i >= 0 ; i--) {
              mId = ((MSPastryProtocol) Network.get(i).getProtocol(mspastryid)).nodeId;
             if  (mId.equals(searchNodeId))
                    return Network.get(i);
        }

        return null;
    }

    //______________________________________________________________________________________________
    /**
     * see MSPastry protocol "ReceiveRoute" primitive
     * @param m Message
     */
    public void receiveRoute(Message m) {
        switch (m.messageType) {
        case Message.MSG_LOOKUP:
            deliver(m);
            o(" [rr] Delivered message [m.id="+m.id+"] [src=dest=" + RoutingTable.truncateNodeId(nodeId) + "[in "+ ((CommonState.getTime())-(m.timestamp))+" msecs] ["+m.nrHops+" hops]");
            break;
        case Message.MSG_JOINREQUEST:

            m.messageType = Message.MSG_JOINREPLY;
            // ((Message.BodyJoinRequestReply)m.body).joiner = null; //not necessary anymore
            ((Message.BodyJoinRequestReply)m.body).ls = this.leafSet;
            //  ((Message.BodyJoinRequestReply)m.body).rt = ...LEAVE AS IS...

            transport = (UnreliableTransport) (Network.prototype).getProtocol(tid);
            transport.send(nodeIdtoNode(this.nodeId), nodeIdtoNode(m.dest), m, mspastryid);
            break;

        }
    }

    //______________________________________________________________________________________________
    /**
     * see MSPastry protocol "Route" primitive
     * @param m Message
     * @param srcNode Node
     */
    private void route(Message m, Node srcNode) {

        // u("["+RoutingTable.truncateNodeId(nodeId)+"] received msg:[" + m.id + "] to route, with track: <");        o(m.traceToString(false) + ">");

        BigInteger nexthop = null;

        //leave a track of the transit of the message over this node
        m.nrHops++;
        if(m.trackSize<m.tracks.length)
            m.trackSize++;
        m.tracks[m.trackSize-1]= this.nodeId;

        BigInteger curdist;
        BigInteger[] allNodes;
        BigInteger mindist;



        if (leafSet.encompass(m.dest)) {
            // il nodeID j in Li t.c. |k-j| � minimo
            int near = 0;
            allNodes = leafSet.listAllNodes();

            if (allNodes.length != 0)  {

                mindist = m.dest.subtract(allNodes[near]).abs();
                for (int i = 1; i < allNodes.length; i++) {
                    curdist = m.dest.subtract(allNodes[i]).abs();
                    if (mindist.compareTo(curdist) > 0) {
                        mindist = curdist;
                        near = i;
                    }
                }
                nexthop = allNodes[near];
            } else  nexthop = this.nodeId;

        }
        else {
            int r = Util.prefixLen(m.dest, this.nodeId);
            if (r == MSPastryCommonConfig.DIGITS) {
                deliver(m);
                o("  [route]   Delivered message src=dest=" + RoutingTable.truncateNodeId(nodeId));
                return;
            }

            nexthop = this.routingTable.get(r, Util.charToIndex(Util.put0(m.dest).charAt(r))  );

            if (nexthop == null) {
                //il nodeID j in (Li U Ri) t.c. |k-j| < |k-i| && prefixLen(k,j)>=r

                BigInteger[] l = this.leafSet.listAllNodes();


                for (int jrow = 0; jrow < routingTable.rows; jrow++) {
                    for (int jcol = 0; jcol < routingTable.cols; jcol++) {
                        BigInteger nodejj = routingTable.get(jrow, jcol);
                        if (nodejj!=null)
                         if (cond1(m.dest, this.nodeId,nodejj) &&
                             cond2(m.dest, nodejj, r)) {
                            nexthop = nodejj;
                            break;
                        }
                    }
                }

                if (nexthop == null)
                for (int j = 0; j < l.length; j++) {
                    if (cond1(m.dest, this.nodeId, l[j]) &&
                        cond2(m.dest, l[j], r)) {
                        nexthop = l[j];
                        break;
                    }
                }


            } // end if (nexthop==null)
        }


        o(String.format("[%s].route([type=%s][src:%s][dest:%s][m.id=%d]): [nexthop:%s]",
            RoutingTable.truncateNodeId(nodeId),
            m.messageTypetoString(),
            "", // RoutingTable.truncateNodeId(src.nodeId),
            RoutingTable.truncateNodeId(m.dest),
            m.id,
            RoutingTable.truncateNodeId(nexthop)
                ));

        if ( m.body instanceof Message.BodyJoinRequestReply && false)
         o("m.RT "+((Message.BodyJoinRequestReply)(m.body)).rt);




        /** !!!
         *  (this.nodeId.equals(m.dest)) � troppo limitativo, noi vogliamo vedere se "io" sono
         * quello pi� (numericammente) vicino possibile.
         * Poich� supponiamo di avvicinarci progressivamente, questo si traduce nel controllare
         * se sono pi� vicino dell'ultimo nodo attraversato (m.traks[last])
         *
         * l'hop lo facciamo solo se facendolo... ridurremo la distanza,
         * la distanza tra destinatario e me
         *   rispetto alla distanza fra destinatario e precedente
         */
        if ((m.trackSize > 0) && (nexthop != null)) {
            BigInteger src = m.tracks[m.trackSize-1]; if (!Util.nearer(m.dest,nexthop,src))
            //if (!Util.nearer(m.dest,nexthop,src.nodeId))
             nexthop = this.nodeId;
        }