nakack.java

来自「JGRoups源码」· Java 代码 · 共 1,094 行 · 第 1/3 页

            case Event.GET_MSGS_RECEIVED:  // return the highest seqnos delivered (=consumed by the application)
                long[] h=naker.getHighestSeqnosDelivered();
                passUp(new Event(Event.GET_MSGS_RECEIVED_OK, h));
                break;
        }


        passDown(evt);
    }


    boolean coordinator() {
        if(members == null || members.size() < 1 || local_addr == null)
            return false;
        return local_addr.equals(members.elementAt(0));
    }

    /**
     * Rebroadcasts the messages given as arguments
     */
    void rebroadcastMsgs(Vector v) {
        Vector final_v;
        Message m1, m2;
        NakAckHeader h1, h2;

        if(v == null) return;
        final_v=new Vector(v.size());

        // weed out duplicates
        //TODO Check!!!!!
        for(int i=0; i < v.size(); i++) {
            boolean present=false;
            m1=(Message)v.elementAt(i);
            h1=m1 != null? (NakAckHeader)m1.getHeader(getName()) : null;
            if(m1 == null || h1 == null) { // +++ remove
                if(log.isErrorEnabled()) log.error("message is null");
                continue;
            }

            for(int j=0; j < final_v.size(); j++) {
                m2=(Message)final_v.elementAt(j);
                h2=m2 != null? (NakAckHeader)m2.getHeader(getName()) : null;
                if(m2 == null || h2 == null) { // +++ remove
                    if(log.isErrorEnabled()) log.error("message m2 is null");
                    continue;
                }
                if(h1.seqno == h2.seqno && m1.getSrc() != null && m2.getSrc() != null &&
                        m1.getSrc().equals(m2.getSrc())) {
                    present=true;
                }
            }
            if(!present)
                final_v.addElement(m1);
        }

        if(warn) log.warn("rebroadcasting " + final_v.size() + " messages");

        /* Now re-broadcast messages using original NakAckHeader (same seqnos, same sender !) */
        for(int i=0; i < final_v.size(); i++) {
            m1=(Message)final_v.elementAt(i);
            naker.resend(m1);
        }

        // Wait until all members have acked reception of outstanding msgs. This will empty our
        // retransmission table (AckMcastSenderWindow)
        naker.waitUntilAllAcksReceived(rebroadcast_timeout);
        passUp(new Event(Event.REBROADCAST_MSGS_OK));
    }


    /**
     * Deliver all messages in the queue where <code>msg.vid == vid</code> holds. Messages were stored
     * in the queue because their vid was greater than the current view.
     */
    void deliverQueuedMessages() {
        NakAckHeader hdr;
        Message tmpmsg;
        int rc;

        while(queued_msgs.size() > 0) {
            tmpmsg=(Message)queued_msgs.removeFromHead();
            hdr=(NakAckHeader)tmpmsg.getHeader(getName());
            rc=hdr.vid.compareTo(vid);
            if(rc == 0) {               // same vid -> OK
                up(new Event(Event.MSG, tmpmsg));
            }
            else if(rc > 0) {
            }
            else
            /** todo Maybe messages from previous vids are stored while client */
                ; // can't be the case; only messages for future views are stored !
        }
    }


    public boolean setProperties(Properties props) {
        String str;
        long[] tmp;

        super.setProperties(props);
        str=props.getProperty("retransmit_timeout");
        if(str != null) {
            tmp=Util.parseCommaDelimitedLongs(str);
            props.remove("retransmit_timeout");
            if(tmp != null && tmp.length > 0)
                retransmit_timeout=tmp;
        }

        str=props.getProperty("rebroadcast_timeout");
        if(str != null) {
            rebroadcast_timeout=Long.parseLong(str);
            props.remove("rebroadcast_timeout");
        }

        if(props.size() > 0) {
            log.error("NAKACK.setProperties(): these properties are not recognized: " + props);

            return false;
        }
        return true;
    }


    class NAKer implements Retransmitter.RetransmitCommand, AckMcastSenderWindow.RetransmitCommand {
        long seqno=0;                       // current message sequence number
        final Hashtable received_msgs=new Hashtable(); // ordered by sender -> NakReceiverWindow
        final Hashtable sent_msgs=new Hashtable();     // ordered by seqno (sent by me !) - Messages
        final AckMcastSenderWindow sender_win=new AckMcastSenderWindow(this, timer);
        boolean acking=false;                  // require acks when sending msgs
        long deleted_up_to=0;


        // Used to periodically retransmit the last message
        final LastMessageRetransmitter last_msg_xmitter=new LastMessageRetransmitter();


        private class LastMessageRetransmitter implements TimeScheduler.Task {
            boolean stopped=false;
            int num_times=2;  // number of times a message is retransmitted
            long last_xmitted_seqno=0;


            public void stop() {
                stopped=true;
            }

            public boolean cancelled() {
                return stopped;
            }

            public long nextInterval() {
                return retransmit_timeout[0];
            }


            /**
             * Periodically retransmits the last seqno to all members. If the seqno doesn't change (ie. there
             * were no new messages sent) then the retransmitter task doesn't retransmit after 'num_times' times.
             */
            public void run() {
                synchronized(sent_msgs) {
                    long prevSeqno=seqno - 1;

                    if(prevSeqno == last_xmitted_seqno) {

                            if(log.isInfoEnabled()) log.info("prevSeqno=" + prevSeqno + ", last_xmitted_seqno=" +
                                    last_xmitted_seqno + ", num_times=" + num_times);
                        if(--num_times <= 0)
                            return;
                    }
                    else {
                        num_times=3;
                        last_xmitted_seqno=prevSeqno;
                    }

                    if((prevSeqno >= 0) && (prevSeqno > deleted_up_to)) {

                            if(log.isInfoEnabled()) log.info("retransmitting last message " + prevSeqno);
                        retransmit(null, prevSeqno, prevSeqno);
                    }
                }
            }

        }


        NAKer() {
            if(timer != null)
                timer.add(last_msg_xmitter, true); // fixed-rate scheduling
            else
                if(log.isErrorEnabled()) log.error("timer is null");
        }


        long getNextSeqno() {
            return seqno++;
        }


        long getHighestSeqnoSent() {
            long highest_sent=-1;
            for(Enumeration e=sent_msgs.keys(); e.hasMoreElements();)
                highest_sent=Math.max(highest_sent, ((Long)e.nextElement()).longValue());
            return highest_sent;
        }


        /**
         * Returns an array of the highest sequence numbers consumed by the application so far,
         * its order corresponding with <code>mbrs</code>. Used by coordinator as argument when
         * sending initial FLUSH request to members
         */
        long[] getHighestSeqnosDelivered() {
            long[] highest_deliv=members != null? new long[members.size()] : null;
            Address mbr;
            NakReceiverWindow win;

            if(highest_deliv == null) return null;

            for(int i=0; i < highest_deliv.length; i++) highest_deliv[i]=-1;

            synchronized(members) {
                for(int i=0; i < members.size(); i++) {
                    mbr=(Address)members.elementAt(i);
                    win=(NakReceiverWindow)received_msgs.get(mbr);
                    if(win != null)
                        highest_deliv[i]=win.getHighestDelivered();
                }
            }
            return highest_deliv;
        }


        /**
         * Return all messages sent by us that are higher than <code>seqno</code>
         */
        List getSentMessagesHigherThan(long seqno) {
            List retval=new List();
            Long key;

            for(Enumeration e=sent_msgs.keys(); e.hasMoreElements();) {
                key=(Long)e.nextElement();
                if(key.longValue() > seqno)
                    retval.add(sent_msgs.get(key));
            }
            return retval;
        }


        /**
         * Returns a message digest: for each member P in <code>highest_seqnos</code>, the highest seqno
         * received from P is added to the digest's array. If P == this, then the highest seqno
         * <em>sent</em> is added: this makes sure that messages sent but not yet received are also
         * re-broadcast (because they are also unstable).<p>If my highest seqno for a member P is
         * higher than the one in <code>highest_seqnos</code>, then all messages from P (received or sent)
         * whose seqno is higher are added to the digest's messages. The coordinator will use all digests
         * to compute a set of messages than need to be re-broadcast to the members before installing
         * a new view.
         */
        Digest computeMessageDigest(long[] highest_seqnos) {
            Digest digest=highest_seqnos != null? new Digest(highest_seqnos.length) : null;
            Address sender;
            NakReceiverWindow win;
            List unstable_msgs;
            int own_index;
            long highest_seqno_sent=-1, highest_seqno_received=-1;

            if(digest == null) {

                    if(warn) log.warn("highest_seqnos is null, cannot compute digest !");
                return null;
            }

            if(highest_seqnos.length != members.size()) {

                    if(warn) log.warn("the mbrship size and the size " +
                            "of the highest_seqnos array are not equal, cannot compute digest !");
                return null;
            }

            System.arraycopy(highest_seqnos, 0, digest.highest_seqnos, 0, digest.highest_seqnos.length);

            for(int i=0; i < highest_seqnos.length; i++) {
                sender=(Address)members.elementAt(i);
                if(sender == null) continue;
                win=(NakReceiverWindow)received_msgs.get(sender);
                if(win == null) continue;
                digest.highest_seqnos[i]=win.getHighestReceived();
                unstable_msgs=win.getMessagesHigherThan(highest_seqnos[i]);
                for(Enumeration e=unstable_msgs.elements(); e.hasMoreElements();)
                    digest.msgs.add(e.nextElement());
            }


            /** If our highest seqno <em>sent</em> is higher than the one <em>received</em>, we have to
             (a) set it in the digest and (b) add the corresponding messages **/

            own_index=members.indexOf(local_addr);
            if(own_index == -1) {

                    if(warn) log.warn("no own address in highest_seqnos");
                return digest;
            }
            highest_seqno_received=digest.highest_seqnos[own_index];
            highest_seqno_sent=getHighestSeqnoSent();

            if(highest_seqno_sent > highest_seqno_received) {
                // (a) Set highest seqno sent in digest
                digest.highest_seqnos[own_index]=highest_seqno_sent;

                // (b) Add messages between highest_seqno_received and highest_seqno_sent
                unstable_msgs=getSentMessagesHigherThan(highest_seqno_received);
                for(Enumeration e=unstable_msgs.elements(); e.hasMoreElements();)
                    digest.msgs.add(e.nextElement());
            }

            return digest;
        }


        /**
         * For each non-null member m in <code>range</code>, get messages with sequence numbers between
         * range[m][0] and range[m][1], excluding range[m][0] and including range[m][1].
         */
        List getMessagesInRange(long[][] range) {
            List retval=new List();
            List tmp;
            NakReceiverWindow win;
            Address sender;

            for(int i=0; i < range.length; i++) {
                if(range[i] != null) {
                    sender=(Address)members.elementAt(i);
                    if(sender == null) continue;
                    win=(NakReceiverWindow)received_msgs.get(sender);
                    if(win == null) continue;
                    tmp=win.getMessagesInRange(range[i][0], range[i][1]);
                    if(tmp == null || tmp.size() < 1) continue;
                    for(Enumeration e=tmp.elements(); e.hasMoreElements();)
                        retval.add(e.nextElement());
                }
            }
            return retval;
        }


        void setAcks(boolean f) {
            acking=f;
        }


        /**
         * Vector with messages (ordered by sender) that are stable and can be discarded.
         * This applies to NAK-based sender and receivers.
         */
        void stable(long[] seqnos) {
            int index;
            long seqno;
            NakReceiverWindow recv_win;
            Address sender;

            if(members == null || local_addr == null) {
                 if(warn) log.warn("members or local_addr are null !");
                return;
            }