tp.java

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

        }
        catch(QueueClosedException closed_ex) {
        }
        catch(InterruptedException interruptedEx) {
        }
        catch(Throwable e) {
            if(log.isErrorEnabled()) {
                String dst=msg.getDest() == null? "null" : msg.getDest().toString();
                log.error("failed sending message to " + dst + " (" + msg.getLength() + " bytes)", e.getCause());
            }
        }
    }



    /*--------------------------- End of Protocol interface -------------------------- */


    /* ------------------------------ Private Methods -------------------------------- */



    /**
     * If the sender is null, set our own address. We cannot just go ahead and set the address
     * anyway, as we might be sending a message on behalf of someone else ! E.gin case of
     * retransmission, when the original sender has crashed, or in a FLUSH protocol when we
     * have to return all unstable messages with the FLUSH_OK response.
     */
    private void setSourceAddress(Message msg) {
        if(msg.getSrc() == null)
            msg.setSrc(local_addr);
    }

    /**
     * Subclasses must call this method when a unicast or multicast message has been received.
     * Declared final so subclasses cannot override this method.
     *
     * @param dest
     * @param sender
     * @param data
     * @param offset
     * @param length
     */
    protected final void receive(Address dest, Address sender, byte[] data, int offset, int length) {
        if(data == null) return;

//        if(length == 4) {  // received a diagnostics probe
//            if(data[offset] == 'd' && data[offset+1] == 'i' && data[offset+2] == 'a' && data[offset+3] == 'g') {
//                handleDiagnosticProbe(sender);
//                return;
//            }
//        }

        boolean mcast=dest == null || dest.isMulticastAddress();
        if(trace){
            StringBuffer sb=new StringBuffer("received (");
            sb.append(mcast? "mcast) " : "ucast) ").append(length).append(" bytes from ").append(sender);
            log.trace(sb.toString());
        }

        try {
            if(use_incoming_packet_handler) {
                byte[] tmp=new byte[length];
                System.arraycopy(data, offset, tmp, 0, length);
                incoming_packet_queue.add(new IncomingQueueEntry(dest, sender, tmp, offset, length));
            }
            else
                handleIncomingPacket(dest, sender, data, offset, length);
        }
        catch(Throwable t) {
            if(log.isErrorEnabled())
                log.error(new StringBuffer("failed handling data from ").append(sender), t);
        }
    }


    /**
     * Processes a packet read from either the multicast or unicast socket. Needs to be synchronized because
     * mcast or unicast socket reads can be concurrent.
     * Correction (bela April 19 2005): we access no instance variables, all vars are allocated on the stack, so
     * this method should be reentrant: removed 'synchronized' keyword
     */
    private void handleIncomingPacket(Address dest, Address sender, byte[] data, int offset, int length) {
        Message                msg=null;
        List                   l=null;  // used if bundling is enabled
        short                  version;
        boolean                is_message_list, multicast;
        byte                   flags;

        try {
            synchronized(in_stream) {
                in_stream.setData(data, offset, length);
                buf_in_stream.reset(length);
                version=dis.readShort();
                if(Version.compareTo(version) == false) {
                    if(warn) {
                        StringBuffer sb=new StringBuffer();
                        sb.append("packet from ").append(sender).append(" has different version (").append(version);
                        sb.append(") from ours (").append(Version.printVersion()).append("). ");
                        if(discard_incompatible_packets)
                            sb.append("Packet is discarded");
                        else
                            sb.append("This may cause problems");
                        log.warn(sb);
                    }
                    if(discard_incompatible_packets)
                        return;
                }

                flags=dis.readByte();
                is_message_list=(flags & LIST) == LIST;
                multicast=(flags & MULTICAST) == MULTICAST;

                if(is_message_list)
                    l=bufferToList(dis, dest, multicast);
                else
                    msg=bufferToMessage(dis, dest, sender, multicast);
            }

            LinkedList msgs=new LinkedList();
            if(is_message_list) {
                for(Enumeration en=l.elements(); en.hasMoreElements();)
                    msgs.add(en.nextElement());
            }
            else
                msgs.add(msg);

            Address src;
            for(Iterator it=msgs.iterator(); it.hasNext();) {
                msg=(Message)it.next();
                src=msg.getSrc();
                if(loopback) {
                    if(multicast && src != null && local_addr.equals(src)) { // discard own loopback multicast packets
                        it.remove();
                    }
                }
                else
                    handleIncomingMessage(msg);
            }
            if(incoming_msg_queue != null && msgs.size() > 0)
                incoming_msg_queue.addAll(msgs);
        }
        catch(QueueClosedException closed_ex) {
            ; // swallow exception
        }
        catch(Throwable t) {
            if(log.isErrorEnabled())
                log.error("failed unmarshalling message", t);
        }
    }



    private void handleIncomingMessage(Message msg) {
        Event      evt;
        TpHeader   hdr;

        if(stats) {
            num_msgs_received++;
            num_bytes_received+=msg.getLength();
        }

        evt=new Event(Event.MSG, msg);
        if(trace) {
            StringBuffer sb=new StringBuffer("message is ").append(msg).append(", headers are ").append(msg.getHeaders());
            log.trace(sb);
        }

        /* Because Protocol.up() is never called by this bottommost layer, we call up() directly in the observer.
        * This allows e.g. PerfObserver to get the time of reception of a message */
        if(observer != null)
            observer.up(evt, up_queue.size());

        hdr=(TpHeader)msg.getHeader(name); // replaced removeHeader() with getHeader()
        if(hdr != null) {

            /* Discard all messages destined for a channel with a different name */
            String ch_name=hdr.channel_name;

            // Discard if message's group name is not the same as our group name unless the
            // message is a diagnosis message (special group name DIAG_GROUP)
            if(ch_name != null && channel_name != null && !channel_name.equals(ch_name) &&
                    !ch_name.equals(Util.DIAG_GROUP)) {
                if(warn)
                    log.warn(new StringBuffer("discarded message from different group \"").append(ch_name).
                            append("\" (our group is \"").append(channel_name).append("\"). Sender was ").append(msg.getSrc()));
                return;
            }
        }
        else {
            if(trace)
                log.trace(new StringBuffer("message does not have a transport header, msg is ").append(msg).
                          append(", headers are ").append(msg.getHeaders()).append(", will be discarded"));
            return;
        }
        passUp(evt);
    }


    /** Internal method to serialize and send a message. This method is not reentrant */
    private void send(Message msg, Address dest, boolean multicast) throws Exception {
        if(enable_bundling) {
            bundler.send(msg, dest);
            return;
        }

        // Needs to be synchronized because we can have possible concurrent access, e.g.
        // Discovery uses a separate thread to send out discovery messages
        // We would *not* need to sync between send(), OutgoingPacketHandler and BundlingOutgoingPacketHandler,
        // because only *one* of them is enabled
        Buffer   buf;
        synchronized(out_stream) {
            buf=messageToBuffer(msg, multicast);
            doSend(buf, dest, multicast);
        }
    }


    private void doSend(Buffer buf, Address dest, boolean multicast) throws Exception {
        if(stats) {
            num_msgs_sent++;
            num_bytes_sent+=buf.getLength();
        }
        if(multicast) {
            sendToAllMembers(buf.getBuf(), buf.getOffset(), buf.getLength());
        }
        else {
            sendToSingleMember(dest, buf.getBuf(), buf.getOffset(), buf.getLength());
        }
    }



    /**
     * This method needs to be synchronized on out_stream when it is called
     * @param msg
     * @return
     * @throws java.io.IOException
     */
    private Buffer messageToBuffer(Message msg, boolean multicast) throws Exception {
        Buffer retval;
        byte flags=0;

        out_stream.reset();
        buf_out_stream.reset(out_stream.getCapacity());
        dos.reset();
        dos.writeShort(Version.version); // write the version
        if(multicast)
            flags+=MULTICAST;
        dos.writeByte(flags);
        // preMarshalling(msg, dest, src);  // allows for optimization by subclass
        msg.writeTo(dos);
        // postMarshalling(msg, dest, src); // allows for optimization by subclass
        dos.flush();
        retval=new Buffer(out_stream.getRawBuffer(), 0, out_stream.size());
        return retval;
    }

    private Message bufferToMessage(DataInputStream instream, Address dest, Address sender, boolean multicast) throws Exception {
        Message msg=new Message(false); // don't create headers, readFrom() will do this
        msg.readFrom(instream);
        postUnmarshalling(msg, dest, sender, multicast); // allows for optimization by subclass
        return msg;
    }



    private Buffer listToBuffer(List l, boolean multicast) throws Exception {
        Buffer retval;
        Address src;
        Message msg;
        byte flags=0;
        int len=l != null? l.size() : 0;
        boolean src_written=false;
        out_stream.reset();
        buf_out_stream.reset(out_stream.getCapacity());
        dos.reset();
        dos.writeShort(Version.version);
        flags+=LIST;
        if(multicast)
            flags+=MULTICAST;
        dos.writeByte(flags);
        dos.writeInt(len);
        for(Enumeration en=l.elements(); en.hasMoreElements();) {
            msg=(Message)en.nextElement();
            src=msg.getSrc();
            if(!src_written) {
                Util.writeAddress(src, dos);
                src_written=true;
            }
            // msg.setSrc(null);
            msg.writeTo(dos);
            // msg.setSrc(src);
        }
        dos.flush();
        retval=new Buffer(out_stream.getRawBuffer(), 0, out_stream.size());
        return retval;
    }

    private List bufferToList(DataInputStream instream, Address dest, boolean multicast) throws Exception {
        List                    l=new List();
        DataInputStream         in=null;
        int                     len;
        Message                 msg;
        Address                 src;

        try {
            len=instream.readInt();
            src=Util.readAddress(instream);
            for(int i=0; i < len; i++) {
                msg=new Message(false); // don't create headers, readFrom() will do this
                msg.readFrom(instream);
                postUnmarshallingList(msg, dest, multicast);
                msg.setSrc(src);
                l.add(msg);
            }
            return l;
        }
        finally {
            Util.close(in);
        }
    }




    /**
     *
     * @param s
     * @return List<NetworkInterface>
     */
    private java.util.List parseInterfaceList(String s) throws Exception {
        java.util.List interfaces=new ArrayList(10);
        if(s == null)
            return null;

        StringTokenizer tok=new StringTokenizer(s, ",");
        String interface_name;
        NetworkInterface intf;

        while(tok.hasMoreTokens()) {
            interface_name=tok.nextToken();

            // try by name first (e.g. (eth0")
            intf=NetworkInterface.getByName(interface_name);

            // next try by IP address or symbolic name
            if(intf == null)
                intf=NetworkInterface.getByInetAddress(InetAddress.getByName(interface_name));

            if(intf == null)
                throw new Exception("interface " + interface_name + " not found");
            if(interfaces.contains(intf)) {
                log.warn("did not add interface " + interface_name + " (already present in " + print(interfaces) + ")");
            }
            else {