udp.java

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

            sb.append("\n").append(mcast_send_sockets.length).append(" mcast send sockets:\n");
            MulticastSocket s;
            for(int i=0; i < mcast_send_sockets.length; i++) {
                s=mcast_send_sockets[i];
                sb.append(s.getInterface().getHostAddress()).append(':').append(s.getLocalPort());
                sb.append(", send buffer size=").append(s.getSendBufferSize());
                sb.append(", receive buffer size=").append(s.getReceiveBufferSize()).append("\n");
            }
        }
        return sb.toString();
    }


    void setBufferSizes() {
        if(sock != null)
            setBufferSize(sock, ucast_send_buf_size, ucast_recv_buf_size);

        if(mcast_recv_sock != null)
            setBufferSize(mcast_recv_sock, mcast_send_buf_size, mcast_recv_buf_size);

        if(mcast_send_sock != null)
            setBufferSize(mcast_send_sock, mcast_send_buf_size, mcast_recv_buf_size);

        if(mcast_send_sockets != null) {
            for(int i=0; i < mcast_send_sockets.length; i++) {
                setBufferSize(mcast_send_sockets[i], mcast_send_buf_size, mcast_recv_buf_size);
            }
        }
    }

    private void setBufferSize(DatagramSocket sock, int send_buf_size, int recv_buf_size) {
        try {
            sock.setSendBufferSize(send_buf_size);
        }
        catch(Throwable ex) {
            if(warn) log.warn("failed setting send buffer size of " + send_buf_size + " in " + sock + ": " + ex);
        }

        try {
            sock.setReceiveBufferSize(recv_buf_size);
        }
        catch(Throwable ex) {
            if(warn) log.warn("failed setting receive buffer size of " + recv_buf_size + " in " + sock + ": " + ex);
        }
    }


    /**
     * Closed UDP unicast and multicast sockets
     */
    void closeSockets() {
        // 1. Close multicast socket
        closeMulticastSocket();

        // 2. Close socket
        closeSocket();
    }


    void closeMulticastSocket() {
        if(mcast_recv_sock != null) {
            try {
                if(mcast_addr != null) {
                    mcast_recv_sock.leaveGroup(mcast_addr.getIpAddress());
                }
                mcast_recv_sock.close(); // this will cause the mcast receiver thread to break out of its loop
                mcast_recv_sock=null;
                if(log.isDebugEnabled()) log.debug("multicast receive socket closed");
            }
            catch(IOException ex) {
            }
            mcast_addr=null;
        }

        if(mcast_send_sock != null) {
            mcast_send_sock.close();
            mcast_send_sock=null;
            if(log.isDebugEnabled()) log.debug("multicast send socket closed");
        }
        if(mcast_send_sockets != null) {
            MulticastSocket s;
            for(int i=0; i < mcast_send_sockets.length; i++) {
                s=mcast_send_sockets[i];
                s.close();
                if(log.isDebugEnabled()) log.debug("multicast send socket " + s + " closed");
            }
            mcast_send_sockets=null;
        }
    }


    private void closeSocket() {
        if(sock != null) {
            sock.close();
            sock=null;
            if(log.isDebugEnabled()) log.debug("socket closed");
        }
    }




    /**
     * Starts the unicast and multicast receiver threads
     */
    void startThreads() throws Exception {
        if(ucast_receiver == null) {
            //start the listener thread of the ucast_recv_sock
            ucast_receiver=new UcastReceiver();
            ucast_receiver.start();
             if(log.isDebugEnabled()) log.debug("created unicast receiver thread");
        }

        if(ip_mcast) {
            if(mcast_receiver != null) {
                if(mcast_receiver.isAlive()) {
                    if(log.isDebugEnabled()) log.debug("did not create new multicastreceiver thread as existing " +
                                                       "multicast receiver thread is still running");
                }
                else
                    mcast_receiver=null; // will be created just below...
            }

            if(mcast_receiver == null) {
                mcast_receiver=new Thread(Util.getGlobalThreadGroup(), this, "UDP mcast receiver");
                mcast_receiver.setPriority(Thread.MAX_PRIORITY); // needed ????
                mcast_receiver.setDaemon(true);
                mcast_receiver.start();
            }
        }
    }


    /**
     * Stops unicast and multicast receiver threads
     */
    void stopThreads() {
        Thread tmp;

        // 1. Stop the multicast receiver thread
        if(mcast_receiver != null) {
            if(mcast_receiver.isAlive()) {
                tmp=mcast_receiver;
                mcast_receiver=null;
                closeMulticastSocket();  // will cause the multicast thread to terminate
                tmp.interrupt();
                try {
                    tmp.join(100);
                }
                catch(Exception e) {
                }
                tmp=null;
            }
            mcast_receiver=null;
        }

        // 2. Stop the unicast receiver thread
        if(ucast_receiver != null) {
            ucast_receiver.stop();
            ucast_receiver=null;
        }
    }


    protected void setThreadNames() {
        super.setThreadNames();

        if(channel_name != null) {
            String tmp, prefix=Global.THREAD_PREFIX;
            if(mcast_receiver != null) {
                tmp=mcast_receiver.getName();
                if(tmp != null && tmp.indexOf(prefix) == -1) {
                    tmp+=prefix + channel_name + ")";
                    mcast_receiver.setName(tmp);
                }
            }
            if(ucast_receiver != null) {
                tmp=ucast_receiver.getName();
                if(tmp != null && tmp.indexOf(prefix) == -1) {
                    tmp+=prefix + channel_name + ")";
                    ucast_receiver.setName(tmp);
                }
            }
        }
    }

    protected void unsetThreadNames() {
        super.unsetThreadNames();
        if(channel_name != null) {
            String tmp, prefix=Global.THREAD_PREFIX;
            int index;

            tmp=mcast_receiver != null? mcast_receiver.getName() : null;
            if(tmp != null) {
                index=tmp.indexOf(prefix);
                if(index > -1) {
                    tmp=tmp.substring(0, index);
                    mcast_receiver.setName(tmp);
                }
            }
            tmp=ucast_receiver != null? ucast_receiver.getName() : null;
            if(tmp != null) {
                index=tmp.indexOf(prefix);
                if(index > -1) {
                    tmp=tmp.substring(0, index);
                    ucast_receiver.setName(tmp);
                }
            }
        }
    }


    protected void handleConfigEvent(HashMap map) {
        boolean set_buffers=false;
        super.handleConfigEvent(map);
        if(map == null) return;

        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;
            set_buffers=true;
        }
        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;
            set_buffers=true;
        }
        if(set_buffers)
            setBufferSizes();
    }



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

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




    public class UcastReceiver implements Runnable {
        boolean running=true;
        Thread thread=null;

        String getName() {
            return thread != null? thread.getName() : null;
        }

        void setName(String thread_name) {
            if(thread != null)
                thread.setName(thread_name);
        }

        public void start() {
            if(thread == null) {
                thread=new Thread(Util.getGlobalThreadGroup(), this, "UDP.UcastReceiverThread");
                thread.setDaemon(true);
                running=true;
                thread.start();
            }
        }


        public void stop() {
            Thread tmp;
            if(thread != null && thread.isAlive()) {
                running=false;
                tmp=thread;
                thread=null;
                closeSocket(); // this will cause the thread to break out of its loop
                tmp.interrupt();
                tmp=null;
            }
            thread=null;
        }


        public void run() {
            DatagramPacket  packet;
            byte            receive_buf[]=new byte[65535];
            int             offset, len;
            byte[]          data;
            InetAddress     sender_addr;
            int             sender_port;
            Address         sender;

            // moved out of loop to avoid excessive object creations (bela March 8 2001)
            packet=new DatagramPacket(receive_buf, receive_buf.length);

            while(running && thread != null && sock != null) {
                try {
                    packet.setData(receive_buf, 0, receive_buf.length);
                    sock.receive(packet);
                    sender_addr=packet.getAddress();
                    sender_port=packet.getPort();
                    offset=packet.getOffset();
                    len=packet.getLength();
                    data=packet.getData();
                    sender=new IpAddress(sender_addr, sender_port);

                    if(len > receive_buf.length) {
                        if(log.isErrorEnabled())
                            log.error("size of the received packet (" + len + ") is bigger than allocated buffer (" +
                                      receive_buf.length + "): will not be able to handle packet. " +
                                      "Use the FRAG protocol and make its frag_size lower than " + receive_buf.length);
                    }
                    receive(local_addr, sender, data, offset, len);
                }
                catch(SocketException sock_ex) {
                    if(log.isDebugEnabled()) log.debug("unicast receiver socket is closed, exception=" + sock_ex);
                    break;
                }
                catch(InterruptedIOException io_ex) { // thread was interrupted
                }
                catch(Throwable ex) {
                    if(log.isErrorEnabled())
                        log.error("[" + local_addr + "] failed receiving unicast packet", ex);
                    Util.sleep(100); // so we don't get into 100% cpu spinning (should NEVER happen !)
                }
            }
            if(log.isDebugEnabled()) log.debug("unicast receiver thread terminated");
        }
    }

}