udp.java

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

                            }
                        }
                    }
                    else {
                        throw new Exception("both mcast_send_sock and mcast_send_sockets are null");
                    }
                }
            }
            else {
                if(sock != null)
                    sock.send(packet);
            }
        }
        catch(Exception ex) {
            throw new Exception("dest=" + dest + ":" + port + " (" + length + " bytes)", ex);
        }
    }


    /* ------------------------------------------------------------------------------- */



    /*------------------------------ Protocol interface ------------------------------ */

    public String getName() {
        return "UDP";
    }





    /**
     * Creates the unicast and multicast sockets and starts the unicast and multicast receiver threads
     */
    public void start() throws Exception {
        if(log.isDebugEnabled()) log.debug("creating sockets and starting threads");
        try {
            createSockets();
        }
        catch(Exception ex) {
            String tmp="problem creating sockets (bind_addr=" + bind_addr +
                    ", mcast_addr=" + mcast_addr + ")";
            throw new Exception(tmp, ex);
        }
        super.start();
        startThreads();
    }


    public void stop() {
        if(log.isDebugEnabled()) log.debug("closing sockets and stopping threads");
        stopThreads();  // will close sockets, closeSockets() is not really needed anymore, but...
        closeSockets(); // ... we'll leave it in there for now (doesn't do anything if already closed)
        super.stop();
    }





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


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





    /**
     * Create UDP sender and receiver sockets. Currently there are 2 sockets
     * (sending and receiving). This is due to Linux's non-BSD compatibility
     * in the JDK port (see DESIGN).
     */
    private void createSockets() throws Exception {
        InetAddress tmp_addr;

        // bind_addr not set, try to assign one by default. This is needed on Windows

        // changed by bela Feb 12 2003: by default multicast sockets will be bound to all network interfaces

        // CHANGED *BACK* by bela March 13 2003: binding to all interfaces did not result in a correct
        // local_addr. As a matter of fact, comparison between e.g. 0.0.0.0:1234 (on hostA) and
        // 0.0.0.0:1.2.3.4 (on hostB) would fail !
//        if(bind_addr == null) {
//            InetAddress[] interfaces=InetAddress.getAllByName(InetAddress.getLocalHost().getHostAddress());
//            if(interfaces != null && interfaces.length > 0)
//                bind_addr=interfaces[0];
//        }

        if(bind_addr == null && !use_local_host) {
            bind_addr=Util.getFirstNonLoopbackAddress();
        }
        if(bind_addr == null)
            bind_addr=InetAddress.getLocalHost();

        if(bind_addr != null)
            if(log.isInfoEnabled()) log.info("sockets will use interface " + bind_addr.getHostAddress());


        // 2. Create socket for receiving unicast UDP packets. The address and port
        //    of this socket will be our local address (local_addr)
        if(bind_port > 0) {
            sock=createDatagramSocketWithBindPort();
        }
        else {
            sock=createEphemeralDatagramSocket();
        }
        if(tos > 0) {
            try {
                sock.setTrafficClass(tos);
            }
            catch(SocketException e) {
                log.warn("traffic class of " + tos + " could not be set, will be ignored", e);
            }
        }

        if(sock == null)
            throw new Exception("UDP.createSocket(): sock is null");

        local_addr=new IpAddress(sock.getLocalAddress(), sock.getLocalPort());
        if(additional_data != null)
            ((IpAddress)local_addr).setAdditionalData(additional_data);


        // 3. Create socket for receiving IP multicast packets
        if(ip_mcast) {
            // 3a. Create mcast receiver socket
            mcast_recv_sock=new MulticastSocket(mcast_port);
            mcast_recv_sock.setTimeToLive(ip_ttl);
            tmp_addr=InetAddress.getByName(mcast_addr_name);
            mcast_addr=new IpAddress(tmp_addr, mcast_port);

            if(receive_on_all_interfaces || (receive_interfaces != null && receive_interfaces.size() > 0)) {
                List interfaces;
                if(receive_interfaces != null)
                    interfaces=receive_interfaces;
                else
                    interfaces=Util.getAllAvailableInterfaces();
                bindToInterfaces(interfaces, mcast_recv_sock, mcast_addr.getIpAddress());
            }
            else {
                if(bind_addr != null)
                    mcast_recv_sock.setInterface(bind_addr);
                 mcast_recv_sock.joinGroup(tmp_addr);
            }

            // 3b. Create mcast sender socket
            if(send_on_all_interfaces || (send_interfaces != null && send_interfaces.size() > 0)) {
                List interfaces;
                NetworkInterface intf;
                if(send_interfaces != null)
                    interfaces=send_interfaces;
                else
                    interfaces=Util.getAllAvailableInterfaces();
                mcast_send_sockets=new MulticastSocket[interfaces.size()];
                int index=0;
                for(Iterator it=interfaces.iterator(); it.hasNext();) {
                    intf=(NetworkInterface)it.next();
                    mcast_send_sockets[index]=new MulticastSocket();
                    mcast_send_sockets[index].setNetworkInterface(intf);
                    mcast_send_sockets[index].setTimeToLive(ip_ttl);
                    if(tos > 0) {
                        try {
                            mcast_send_sockets[index].setTrafficClass(tos);
                        }
                        catch(SocketException e) {
                            log.warn("traffic class of " + tos + " could not be set, will be ignored", e);
                        }
                    }
                    index++;
                }
            }
            else {
                mcast_send_sock=new MulticastSocket();
                mcast_send_sock.setTimeToLive(ip_ttl);
                if(bind_addr != null)
                    mcast_send_sock.setInterface(bind_addr);

                if(tos > 0) {
                    try {
                        mcast_send_sock.setTrafficClass(tos); // high throughput
                    }
                    catch(SocketException e) {
                        log.warn("traffic class of " + tos + " could not be set, will be ignored", e);
                    }
                }
            }
        }

        setBufferSizes();
        if(log.isInfoEnabled()) log.info("socket information:\n" + dumpSocketInfo());
    }


//    private void bindToAllInterfaces(MulticastSocket s, InetAddress mcastAddr) throws IOException {
//        SocketAddress tmp_mcast_addr=new InetSocketAddress(mcastAddr, mcast_port);
//        Enumeration en=NetworkInterface.getNetworkInterfaces();
//        while(en.hasMoreElements()) {
//            NetworkInterface i=(NetworkInterface)en.nextElement();
//            for(Enumeration en2=i.getInetAddresses(); en2.hasMoreElements();) {
//                InetAddress addr=(InetAddress)en2.nextElement();
//                // if(addr.isLoopbackAddress())
//                // continue;
//                s.joinGroup(tmp_mcast_addr, i);
//                if(trace)
//                    log.trace("joined " + tmp_mcast_addr + " on interface " + i.getName() + " (" + addr + ")");
//                break;
//            }
//        }
//    }


    /**
     *
     * @param interfaces List<NetworkInterface>. Guaranteed to have no duplicates
     * @param s
     * @param mcastAddr
     * @throws IOException
     */
    private void bindToInterfaces(List interfaces, MulticastSocket s, InetAddress mcastAddr) throws IOException {
        SocketAddress tmp_mcast_addr=new InetSocketAddress(mcastAddr, mcast_port);
        for(Iterator it=interfaces.iterator(); it.hasNext();) {
            NetworkInterface i=(NetworkInterface)it.next();
            for(Enumeration en2=i.getInetAddresses(); en2.hasMoreElements();) {
                InetAddress addr=(InetAddress)en2.nextElement();
                s.joinGroup(tmp_mcast_addr, i);
                if(trace)
                    log.trace("joined " + tmp_mcast_addr + " on " + i.getName() + " (" + addr + ")");
                break;
            }
        }
    }



    /** Creates a DatagramSocket with a random port. Because in certain operating systems, ports are reused,
     * we keep a list of the n last used ports, and avoid port reuse */
    private DatagramSocket createEphemeralDatagramSocket() throws SocketException {
        DatagramSocket tmp;
        int localPort=0;
        while(true) {
            tmp=new DatagramSocket(localPort, bind_addr); // first time localPort is 0
            if(num_last_ports <= 0)
                break;
            localPort=tmp.getLocalPort();
            if(last_ports_used == null)
                last_ports_used=new BoundedList(num_last_ports);
            if(last_ports_used.contains(new Integer(localPort))) {
                if(log.isDebugEnabled())
                    log.debug("local port " + localPort + " already seen in this session; will try to get other port");
                try {tmp.close();} catch(Throwable e) {}
                localPort++;
            }
            else {
                last_ports_used.add(new Integer(localPort));
                break;
            }
        }
        return tmp;
    }




    /**
     * Creates a DatagramSocket when bind_port > 0. Attempts to allocate the socket with port == bind_port, and
     * increments until it finds a valid port, or until port_range has been exceeded
     * @return DatagramSocket The newly created socket
     * @throws Exception
     */
    private DatagramSocket createDatagramSocketWithBindPort() throws Exception {
        DatagramSocket tmp=null;
        // 27-6-2003 bgooren, find available port in range (start_port, start_port+port_range)
        int rcv_port=bind_port, max_port=bind_port + port_range;
        while(rcv_port <= max_port) {
            try {
                tmp=new DatagramSocket(rcv_port, bind_addr);
                break;
            }
            catch(SocketException bind_ex) {	// Cannot listen on this port
                rcv_port++;
            }
            catch(SecurityException sec_ex) { // Not allowed to listen on this port
                rcv_port++;
            }

            // Cannot listen at all, throw an Exception
            if(rcv_port >= max_port + 1) { // +1 due to the increment above
                throw new Exception("cannot create a socket on any port in range " +
                        bind_port + '-' + (bind_port + port_range));
            }
        }
        return tmp;
    }


    private String dumpSocketInfo() throws Exception {
        StringBuffer sb=new StringBuffer(128);
        sb.append("local_addr=").append(local_addr);
        sb.append(", mcast_addr=").append(mcast_addr);
        sb.append(", bind_addr=").append(bind_addr);
        sb.append(", ttl=").append(ip_ttl);

        if(sock != null) {
            sb.append("\nsock: bound to ");
            sb.append(sock.getLocalAddress().getHostAddress()).append(':').append(sock.getLocalPort());
            sb.append(", receive buffer size=").append(sock.getReceiveBufferSize());
            sb.append(", send buffer size=").append(sock.getSendBufferSize());
        }

        if(mcast_recv_sock != null) {
            sb.append("\nmcast_recv_sock: bound to ");
            sb.append(mcast_recv_sock.getInterface().getHostAddress()).append(':').append(mcast_recv_sock.getLocalPort());
            sb.append(", send buffer size=").append(mcast_recv_sock.getSendBufferSize());
            sb.append(", receive buffer size=").append(mcast_recv_sock.getReceiveBufferSize());
        }

         if(mcast_send_sock != null) {
            sb.append("\nmcast_send_sock: bound to ");
            sb.append(mcast_send_sock.getInterface().getHostAddress()).append(':').append(mcast_send_sock.getLocalPort());
            sb.append(", send buffer size=").append(mcast_send_sock.getSendBufferSize());
            sb.append(", receive buffer size=").append(mcast_send_sock.getReceiveBufferSize());
        }
        if(mcast_send_sockets != null) {