nakack.java

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

            index=members.indexOf(local_addr);

            if(index < 0) {

                    if(warn) log.warn("member " + local_addr + " not found in " + members);
                return;
            }
            seqno=seqnos[index];

                if(log.isInfoEnabled()) log.info("deleting stable messages [" +
                        deleted_up_to + " - " + seqno + ']');

            // delete sent messages that are stable (kept for retransmission requests from receivers)
            synchronized(sent_msgs) {
                for(long i=deleted_up_to; i <= seqno; i++) {
                    sent_msgs.remove(new Long(i));
                }
                deleted_up_to=seqno;
            }
            // delete received msgs that are stable
            for(int i=0; i < members.size(); i++) {
                sender=(Address)members.elementAt(i);
                recv_win=(NakReceiverWindow)received_msgs.get(sender);
                if(recv_win != null)
                    recv_win.stable(seqnos[i]);  // delete all messages with seqnos <= seqnos[i]
            }
        }


        void send(Message msg) {
            long id=getNextSeqno();
            ViewId vid_copy;

            if(vid == null)
                return;
            vid_copy=(ViewId)vid.clone(); /** todo No needs to copy vid */

            if(acking) {
                msg.putHeader(getName(), new NakAckHeader(NakAckHeader.NAK_ACK_MSG, id, vid_copy));
                sender_win.add(id, msg.copy(), (Vector)members.clone()); // msg must be copied !
            }
            else
                msg.putHeader(getName(), new NakAckHeader(NakAckHeader.NAK_MSG, id, vid_copy));

             if(log.isInfoEnabled()) log.info("sending msg #" + id);

            sent_msgs.put(new Long(id), msg.copy());
            passDown(new Event(Event.MSG, msg));
        }





        /** Re-broadcast message. Message already contains NakAckHeader (therefore also seqno).
         Wrap message (including header) in a new message and bcasts using ACKS. Every receiver
         acks the message, unwraps it to get the original message and delivers the original message
         (if not yet delivered).
         // send msgs in Vector arg again (they already have a NakAckHeader !)
         // -> use the same seq numbers
         // -> destination has to be set to null (broadcast), e.g.:
         //      dst=p (me !), src=q --> dst=null, src=q

         TODO:
         -----
         resend() has to wait until it received all ACKs from all recipients (for all msgs), or until
         members were suspected. Thus we can ensure that all members received outstanding msgs before
         we switch to a new view. Otherwise, because the switch to a new view resets NAK and ACK msg
         transmission, slow members might never receive all outstanding messages.
         */



        /**
         * 1. Set the destination address of the original msg to null
         * 2. Add a new header WRAPPED_MSG and send msg. The receiver will ACK the msg,
         * remove the header and deliver the msg
         */
        void resend(Message msg) {
            Message copy=msg.copy();
            NakAckHeader hdr=(NakAckHeader)copy.getHeader(getName());
            NakAckHeader wrapped_hdr;
            long id=hdr.seqno;

            if(vid == null) return;
            copy.setDest(null);  // broadcast, e.g. dst(p), src(q) --> dst(null), src(q)
            wrapped_hdr=new NakAckHeader(NakAckHeader.WRAPPED_MSG, hdr.seqno, hdr.vid);
            wrapped_hdr.sender=local_addr;
            copy.putHeader(WRAPPED_MSG_KEY, wrapped_hdr);
            sender_win.add(id, copy.copy(), (Vector)members.clone());
             if(log.isInfoEnabled()) log.info("resending " + copy.getHeader(getName()));
            passDown(new Event(Event.MSG, copy));
        }


        void waitUntilAllAcksReceived(long timeout) {
            sender_win.waitUntilAllAcksReceived(timeout);
        }


        void receive(long id, Message msg, Vector stable_msgs) {  /** todo Vector stable_msgs is not used in NAKer.receive() */
            Address sender=msg.getSrc();
            NakReceiverWindow win=(NakReceiverWindow)received_msgs.get(sender);
            Message msg_to_deliver;

            if(win == null) {
                win=new NakReceiverWindow(sender, this, 0);
                win.setRetransmitTimeouts(retransmit_timeout);
                received_msgs.put(sender, win);
            }

             if(log.isInfoEnabled()) log.info("received <" + sender + '#' + id + '>');

            win.add(id, msg);  // add in order, then remove and pass up as many msgs as possible
            while(true) {
                msg_to_deliver=win.remove();
                if(msg_to_deliver == null)
                    break;

                if(msg_to_deliver.getHeader(getName()) instanceof NakAckHeader)
                    msg_to_deliver.removeHeader(getName());
                passUp(new Event(Event.MSG, msg_to_deliver));
            }
        }


        void receiveAck(long id, Address sender) {

                if(log.isInfoEnabled()) log.info("received ack <-- ACK <" + sender + '#' + id + '>');
            sender_win.ack(id, sender);
        }


        /**
         * Implementation of interface AckMcastSenderWindow.RetransmitCommand.<p>
         * Called by retransmission thread of AckMcastSenderWindow. <code>msg</code> is already
         * a copy, so does not need to be copied again.
         */
        public void retransmit(long seqno, Message msg, Address dest) {

                if(log.isInfoEnabled()) log.info("retransmitting message " + seqno + " to " + dest +
                        ", header is " + msg.getHeader(getName()));

            // check whether dest is member of group. If not, discard retransmission message and
            // also remove it from sender_win (AckMcastSenderWindow)
            if(members != null) {
                if(!members.contains(dest)) {

                        if(log.isInfoEnabled()) log.info("retransmitting " + seqno + ") to " + dest + ": " + dest +
                                " is not a member; discarding retransmission and removing " +
                                dest + " from sender_win");
                    sender_win.remove(dest);
                    return;
                }
            }

            msg.setDest(dest);
            passDown(new Event(Event.MSG, msg));
        }


        /**
         * Implementation of Retransmitter.RetransmitCommand.<p>
         * Called by retransmission thread when gap is detected. Sends retr. request
         * to originator of msg
         */
        public void retransmit(long first_seqno, long last_seqno, Address sender) {

                if(log.isInfoEnabled()) log.info("retransmit([" + first_seqno + ", " + last_seqno +
                        "]) to " + sender + ", vid=" + vid);

            NakAckHeader hdr=new NakAckHeader(NakAckHeader.RETRANSMIT_MSG, first_seqno, (ViewId)vid.clone()); /** todo Not necessary to clone vid */
            Message retransmit_msg=new Message(sender, null, null);

            hdr.last_seqno=last_seqno;
            retransmit_msg.putHeader(getName(), hdr);
            passDown(new Event(Event.MSG, retransmit_msg));
        }


        // Retransmit from sent-table, called when RETRANSMIT message is received
        void retransmit(Address dest, long first_seqno, long last_seqno) {
            Message m, retr_msg;

            for(long i=first_seqno; i <= last_seqno; i++) {
                m=(Message)sent_msgs.get(new Long(i));
                if(m == null) {
                    if(warn) log.warn("(to " + dest + "): message with " + "seqno=" + i + " not found !");
                    continue;
                }

                retr_msg=m.copy();
                retr_msg.setDest(dest);

                try {
                    passDown(new Event(Event.MSG, retr_msg));
                }
                catch(Exception e) {
                    if(log.isDebugEnabled()) log.debug("exception is " + e);
                }
            }
        }


        void stop() {
            if(sender_win != null)
                sender_win.stop();
        }


        void reset() {
            NakReceiverWindow win;

            // Only reset if not coord: coord may have to retransmit the VIEW_CHANGE msg to slow members,
            // since VIEW_CHANGE results in retransmitter resetting, retransmission would be killed, and the
            // slow mbr would never receive a new view (see ./design/ViewChangeRetransmission.txt)
            if(!coordinator())
                sender_win.reset();

            sent_msgs.clear();
            for(Enumeration e=received_msgs.elements(); e.hasMoreElements();) {
                win=(NakReceiverWindow)e.nextElement();
                win.reset();
            }
            received_msgs.clear();
            seqno=0;
            deleted_up_to=0;
        }


        public void suspect(Address mbr) {
            NakReceiverWindow w;

            w=(NakReceiverWindow)received_msgs.get(mbr);
            if(w != null) {
                w.reset();
                received_msgs.remove(mbr);
            }

            sender_win.suspect(mbr); // don't keep retransmitting messages to mbr
        }


        String dumpContents() {
            StringBuffer ret=new StringBuffer();

            ret.append("\nsent_msgs: " + sent_msgs.size());

            ret.append("\nreceived_msgs: ");
            for(Enumeration e=received_msgs.keys(); e.hasMoreElements();) {
                Address key=(Address)e.nextElement();
                NakReceiverWindow w=(NakReceiverWindow)received_msgs.get(key);
                ret.append('\n' + w.toString());
            }

            ret.append("\nsender_win: " + sender_win.toString());


            return ret.toString();
        }


    }


    class OutOfBander implements AckMcastSenderWindow.RetransmitCommand {
        final AckMcastSenderWindow sender_win=new AckMcastSenderWindow(this, timer);
        final AckMcastReceiverWindow receiver_win=new AckMcastReceiverWindow();
        long seqno=0;


        void send(Message msg) {
            long id=seqno++;
            Vector stable_msgs=sender_win.getStableMessages();
            NakAckHeader hdr;

             if(log.isInfoEnabled()) log.info("sending msg #=" + id);

            hdr=new NakAckHeader(NakAckHeader.OUT_OF_BAND_MSG, id, null);
            hdr.stable_msgs=stable_msgs;
            msg.putHeader(getName(), hdr);

            // msg needs to be copied, otherwise it will be modified by the code below
            sender_win.add(id, msg.copy(), (Vector)members.clone());

            passDown(new Event(Event.MSG, msg));
        }


        void receive(long id, Message msg, Vector stable_msgs) {
            Address sender=msg.getSrc();

            // first thing: send ACK back to sender
            Message ack_msg=new Message(msg.getSrc(), null, null);
            NakAckHeader hdr=new NakAckHeader(NakAckHeader.OUT_OF_BAND_RSP, id, null);
            ack_msg.putHeader(getName(), hdr);


             if(log.isInfoEnabled()) log.info("received <" + sender + '#' + id + ">\n");

            if(receiver_win.add(sender, id))         // not received previously
                passUp(new Event(Event.MSG, msg));

            passDown(new Event(Event.MSG, ack_msg));  // send ACK
             if(log.isInfoEnabled()) log.info("sending ack <" + sender + '#' + id + ">\n");

            if(stable_msgs != null)
                receiver_win.remove(sender, stable_msgs);
        }


        void receiveAck(long id, Address sender) {
             if(log.isInfoEnabled()) log.info("received ack <" + sender + '#' + id + '>');
            sender_win.ack(id, sender);
        }


        /**
         * Called by retransmission thread of AckMcastSenderWindow. <code>msg</code> is already
         * a copy, so does not need to be copied again. All the necessary header are already present;
         * no header needs to be added ! The message is retransmitted as <em>unicast</em> !
         */
        public void retransmit(long seqno, Message msg, Address dest) {
             if(log.isInfoEnabled()) log.info("dest=" + dest + ", msg #" + seqno);
            msg.setDest(dest);
            passDown(new Event(Event.MSG, msg));
        }


        void reset() {
            sender_win.reset();       // +++ ?
            receiver_win.reset();     // +++ ?
        }


        void suspect(Address mbr) {
            sender_win.suspect(mbr);
            receiver_win.suspect(mbr);
        }


        void start() {
            sender_win.start();
        }

        void stop() {
            if(sender_win != null)
                sender_win.stop();
        }


        String dumpContents() {
            StringBuffer ret=new StringBuffer();
            ret.append("\nsender_win:\n" + sender_win.toString() +
                    "\nreceiver_win:\n" + receiver_win.toString());
            return ret.toString();
        }


    }


}