udp_nio.java.txt

来自「JGRoups源码」· 文本 代码 · 共 1,513 行 · 第 1/4 页

//     *
//     * @param dest The destination host. Will be local host if null
//     * @param port The destination port
//     */
//    void sendDummyPacket(InetAddress dest, int port) {
//        DatagramPacket packet;
//        byte[] buf={0};
//
//        if(dest == null) {
//            try {
//                dest=InetAddress.getLocalHost();
//            } catch(Exception e) {
//            }
//        }
//
//        if(trace) if(log.isInfoEnabled()) log.info("UDP_NIO.sendDummyPacket()", "sending packet to " + dest + ":" + port);
//
//        if(ucast_sock == null || dest == null) {
//            if(warn) log.warn("UDP_NIO.sendDummyPacket()", "send_sock was null or dest was null, cannot send dummy packet");
//            return;
//        }
//        packet=new DatagramPacket(buf, buf.length, dest, port);
//        try {
//            ucast_sock.send(packet);
//        } catch(Throwable e) {
//            if(log.isErrorEnabled()) log.error("UDP_NIO.sendDummyPacket()", "exception sending dummy packet to " +
//                    dest + ":" + port + ": " + e);
//        }
//    }





    void handleDownEvent(Event evt) {
        switch(evt.getType()) {

            case Event.TMP_VIEW:
            case Event.VIEW_CHANGE:
                synchronized(members) {
                    members.removeAllElements();
                    Vector tmpvec=((View)evt.getArg()).getMembers();
                    for(int i=0; i < tmpvec.size(); i++)
                        members.addElement(tmpvec.elementAt(i));
                }
                break;

            case Event.GET_LOCAL_ADDRESS:   // return local address -> Event(SET_LOCAL_ADDRESS, local)
                passUp(new Event(Event.SET_LOCAL_ADDRESS, local_addr));
                break;

            case Event.CONNECT:
                group_name=(String)evt.getArg();
                udp_hdr=new UdpHeader(group_name);

                // removed March 18 2003 (bela), not needed (handled by GMS)
                // changed July 2 2003 (bela): we discard CONNECT_OK at the GMS level anyway, this might
                // be needed if we run without GMS though
                passUp(new Event(Event.CONNECT_OK));
                break;

            case Event.DISCONNECT:
                passUp(new Event(Event.DISCONNECT_OK));
                break;

            case Event.CONFIG:
                 if(log.isInfoEnabled()) log.info("received CONFIG event: " + evt.getArg());
                handleConfigEvent((HashMap)evt.getArg());
                break;
        }
    }


    void handleConfigEvent(HashMap map) {
        if(map == null) return;
        if(map.containsKey("additional_data"))
            additional_data=(byte[])map.get("additional_data");
        if(map.containsKey("send_buf_size")) {
            mcast_send_buf_size=((Integer)map.get("send_buf_size")).intValue();
            ucast_send_buf_size=mcast_send_buf_size;
        }
        if(map.containsKey("recv_buf_size")) {
            mcast_recv_buf_size=((Integer)map.get("recv_buf_size")).intValue();
            ucast_recv_buf_size=mcast_recv_buf_size;
        }
    }


    /** Return the first non-loopback interface */
    public String determineDefaultBindInterface() throws SocketException {
        for(Enumeration en=NetworkInterface.getNetworkInterfaces(); en.hasMoreElements();) {
            NetworkInterface ni=(NetworkInterface)en.nextElement();
            for(Enumeration en2=ni.getInetAddresses(); en2.hasMoreElements();) {
                InetAddress bind_addr=(InetAddress)en2.nextElement();
                if(!bind_addr.isLoopbackAddress()) {
                    return bind_addr.getHostAddress();
                }
            }
        }
        return null;
    }

    public List determineAllBindInterfaces() throws SocketException {
        List ret=new ArrayList();
        for(Enumeration en=NetworkInterface.getNetworkInterfaces(); en.hasMoreElements();) {
            NetworkInterface ni=(NetworkInterface)en.nextElement();
            for(Enumeration en2=ni.getInetAddresses(); en2.hasMoreElements();) {
                InetAddress bind_addr=(InetAddress)en2.nextElement();
                ret.add(bind_addr.getHostAddress());
            }
        }

        return ret;
    }

    /* ----------------------------- End of Private Methods ---------------------------------------- */



    /* ----------------------------- Inner Classes ---------------------------------------- */


    /**
     * This thread fetches byte buffers from the packet_queue, converts them into messages and passes them up
     * to the higher layer (done in handleIncomingUdpPacket()).
     */
    class PacketHandler implements Runnable {
        Thread t=null;

        public void run() {
            byte[] data;
            SocketAddress sender;

            while(packet_queue != null && packet_handler != null) {
                try {
                    Object[] arr=(Object[])packet_queue.remove();
                    data=(byte[])arr[0];
                    sender=(SocketAddress)arr[1];
                } catch(QueueClosedException closed_ex) {
                     if(log.isInfoEnabled()) log.info("packet_handler thread terminating");
                    break;
                }
                handleIncomingUdpPacket(data, sender);
                data=null; // let's give the poor garbage collector a hand...
            }
        }

        void start() {
            if(t == null) {
                t=new Thread(this, "UDP_NIO.PacketHandler thread");
                t.setDaemon(true);
                t.start();
            }
        }

        void stop() {
            if(packet_queue != null)
                packet_queue.close(false); // should terminate the packet_handler thread too
            t=null;
            packet_queue=null;
        }
    }




    /**
     * Manages a multicast and unicast socket on a given interface (NIC). The multicast socket is used
     * to listen for incoming multicast packets, the unicast socket is used to (1) listen for incoming
     * unicast packets, (2) to send unicast packets and (3) to send multicast packets
     */
    public static class Connector implements Runnable {

        protected Thread t=null;

        protected SenderThread sender_thread=null;

        /** Interface on which ucast_sock and mcast_sender_sock are created */
        NetworkInterface bind_interface;


        /** Used for sending/receiving unicast/multicast packets. The reason we have to use a MulticastSocket versus a
         * DatagramSocket is that only MulticastSockets allow to set the interface over which a multicast
         * is sent: DatagramSockets consult the routing table to find the interface
         */
        MulticastSocket mcast_sock=null;

        /** Local port of the mcast_sock */
        SocketAddress localAddr=null;

        /** The receiver which handles incoming packets */
        Receiver receiver=null;

        /** Buffer for incoming unicast packets */
        protected byte[] receive_buffer=null;


        Queue send_queue=new Queue();

        static final Log mylog=LogFactory.getLog(Connector.class);
        static final boolean mywarn=mylog.isWarnEnabled();


        class SenderThread extends Thread {

            public SenderThread() {
                super(Util.getGlobalThreadGroup(), "SenderThread");
            }

            public void run() {
                Object[] arr;
                byte[] buf;
                SocketAddress dest;

                while(send_queue != null) {
                    try {
                        arr=(Object[])send_queue.remove();
                        buf=(byte[])arr[0];
                        dest=(SocketAddress)arr[1];
                        mcast_sock.send(new DatagramPacket(buf, buf.length, dest));
                    }
                    catch(QueueClosedException e) {
                        break;
                    }
                    catch(SocketException e) {
                        e.printStackTrace();
                    }
                    catch(IOException e) {
                        e.printStackTrace();
                    }

                }
            }
        }



        public Connector(NetworkInterface bind_interface, int local_bind_port,
                         int port_range,  int receive_buffer_size,
                         int receive_sock_buf_size, int send_sock_buf_size,
                         int ip_ttl, Receiver receiver) throws IOException {
            this.bind_interface=bind_interface;
            this.receiver=receiver;
            this.receive_buffer=new byte[receive_buffer_size];

            mcast_sock=createMulticastSocket(local_bind_port, port_range);

            // changed Bela Dec 31 2003: if loopback is disabled other members on the same machine won't be able
            // to receive our multicasts
            // mcast_sock.setLoopbackMode(true); // we don't want to receive our own multicasts
            mcast_sock.setReceiveBufferSize(receive_sock_buf_size);
            mcast_sock.setSendBufferSize(send_sock_buf_size);
            mcast_sock.setTimeToLive(ip_ttl);
            System.out.println("ttl=" + mcast_sock.getTimeToLive());
            mcast_sock.setNetworkInterface(this.bind_interface); // for outgoing multicasts
            localAddr=mcast_sock.getLocalSocketAddress();
            System.out.println("-- local_addr=" + localAddr);
            System.out.println("-- mcast_sock: send_bufsize=" + mcast_sock.getSendBufferSize() +
                    ", recv_bufsize=" + mcast_sock.getReceiveBufferSize());
        }


        public SocketAddress getLocalAddress() {
            return localAddr;
        }

        public NetworkInterface getBindInterface() {
            return bind_interface;
        }

        public void start() throws Exception {
            if(mcast_sock == null)
                throw new Exception("UDP_NIO.Connector.start(): connector has been stopped (start() cannot be called)");

            if(t != null && t.isAlive()) {
                if(mywarn) mylog.warn("connector thread is already running");
                return;
            }
            t=new Thread(this, "ConnectorThread for " + localAddr);
            t.setDaemon(true);
            t.start();

            sender_thread=new SenderThread();
            sender_thread.start();
        }

        /** Stops the connector. After this call, start() cannot be called, but a new connector has to
         * be created
         */
        public void stop() {
            if(mcast_sock != null)
                mcast_sock.close(); // terminates the thread if running
            t=null;
            mcast_sock=null;
        }



        /** Sends a message using mcast_sock */
        public void send(DatagramPacket packet) throws Exception {
            //mcast_sock.send(packet);

            byte[] buf=(byte[])packet.getData().clone();
            Object[] arr=new Object[]{buf, packet.getSocketAddress()};
            send_queue.add(arr);
        }

        public void run() {
            DatagramPacket packet=new DatagramPacket(receive_buffer, receive_buffer.length);
            while(t != null) {
                try {
                    packet.setData(receive_buffer, 0, receive_buffer.length);
                    ConnectorTable.receivePacket(packet, mcast_sock, receiver);
                }
                catch(Throwable th) {
                    if(th == null || mcast_sock == null || mcast_sock.isClosed())
                        break;
                    if(mylog.isErrorEnabled()) mylog.error("[" + localAddr + "] exception=" + th);
                    Util.sleep(300); // so we don't get into 100% cpu spinning (should NEVER happen !)
                }
            }
            t=null;
        }




        public String toString() {
            StringBuffer sb=new StringBuffer();
            sb.append("local_addr=").append(localAddr).append(", mcast_group=");
            return sb.toString();
        }





        // 27-6-2003 bgooren, find available port in range (start_port, start_port+port_range)
        private MulticastSocket createMulticastSocket(int local_bind_port, int port_range) throws IOException {
            MulticastSocket sock=null;
            int             tmp_port=local_bind_port;

            int max_port=tmp_port + port_range;
            while(tmp_port <= max_port) {
                try {
                    sock=new MulticastSocket(tmp_port);
                    break;
                }
                catch(Exception bind_ex) {
                    tmp_port++;
                }
            }
            if(sock == null)
                throw new IOException("could not create a MulticastSocket (port range: " + local_bind_port +
                        " - " + (local_bind_port+port_range));
            return sock;
        }
    }





    /** Manages a bunch of Connectors */
    public static class ConnectorTable implements Receiver, Runnable {

        Thread t=null;

        /** Socket to receive multicast packets. Will be bound to n interfaces */
        MulticastSocket mcast_sock=null;

        /** The multicast address which mcast_sock will join (e.g. 230.1.2.3:7500) */
        InetSocketAddress mcastAddr=null;

        Receiver receiver=null;

        /** Buffer for incoming packets */
        byte[] receive_buffer=null;