tp.java

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

    /**
     * Bundles smaller messages into bigger ones. Collects messages in a list until
     * messages of a total of <tt>max_bundle_size bytes</tt> have accumulated, or until
     * <tt>max_bundle_timeout</tt> milliseconds have elapsed, whichever is first. Messages
     * are unbundled at the receiver.
     */
//    private class BundlingOutgoingPacketHandler extends OutgoingPacketHandler {
//        /** HashMap<Address, List<Message>>. Keys are destinations, values are lists of Messages */
//        final HashMap       msgs=new HashMap(11);
//        long                count=0;    // current number of bytes accumulated
//        int                 num_msgs=0;
//        long                start=0;
//        long                wait_time=0; // wait for removing messages from the queue
//
//
//
//        private void init() {
//            wait_time=start=count=0;
//        }
//
//        void start() {
//            init();
//            super.start();
//            t.setName("BundlingOutgoingPacketHandler");
//        }
//
//        void stop() {
//            // bundleAndSend();
//            super.stop();
//        }
//
//        public void run() {
//            Message msg;
//            long    length;
//            while(t != null && Thread.currentThread().equals(t)) {
//                try {
//                    msg=(Message)outgoing_queue.poll(wait_time);
//                    if(msg == null)
//                        throw new TimeoutException();
//                    length=msg.size();
//                    checkLength(length);
//                    if(start == 0)
//                        start=System.currentTimeMillis();
//
//                    if(count + length >= max_bundle_size) {
//                        bundleAndSend();
//                        count=0;
//                        start=System.currentTimeMillis();
//                    }
//
//                    addMessage(msg);
//                    count+=length;
//
//                    wait_time=max_bundle_timeout - (System.currentTimeMillis() - start);
//                    if(wait_time <= 0) {
//                        bundleAndSend();
//                        init();
//                    }
//                }
//                catch(QueueClosedException queue_closed_ex) {
//                    bundleAndSend();
//                    break;
//                }
//                catch(TimeoutException timeout_ex) {
//                    bundleAndSend();
//                    init();
//                }
//                catch(Throwable ex) {
//                    log.error("failure in bundling", ex);
//                }
//            }
//            if(trace) log.trace("BundlingOutgoingPacketHandler thread terminated");
//        }
//
//
//
//
//        private void checkLength(long len) throws Exception {
//            if(len > max_bundle_size)
//                throw new Exception("message size (" + len + ") is greater than max bundling size (" + max_bundle_size +
//                        "). Set the fragmentation/bundle size in FRAG and TP correctly");
//        }
//
//
//        private void addMessage(Message msg) { // no sync needed, never called by multiple threads concurrently
//            List    tmp;
//            Address dst=msg.getDest();
//            tmp=(List)msgs.get(dst);
//            if(tmp == null) {
//                tmp=new List();
//                msgs.put(dst, tmp);
//            }
//            tmp.add(msg);
//            num_msgs++;
//        }
//
//
//
//        private void bundleAndSend() {
//            Map.Entry      entry;
//            Address        dst;
//            Buffer         buffer;
//            List           l;
//            long           stop_time=System.currentTimeMillis();
//
//            if(msgs.size() == 0)
//                return;
//
//            try {
//                if(trace) {
//                    StringBuffer sb=new StringBuffer("sending ").append(num_msgs).append(" msgs (");
//                    sb.append(count).append(" bytes, ").append(stop_time-start).append("ms)");
//                    sb.append(" to ").append(msgs.size()).append(" destination(s)");
//                    if(msgs.size() > 1) sb.append(" (dests=").append(msgs.keySet()).append(")");
//                    log.trace(sb.toString());
//                }
//                boolean multicast;
//                for(Iterator it=msgs.entrySet().iterator(); it.hasNext();) {
//                    entry=(Map.Entry)it.next();
//                    l=(List)entry.getValue();
//                    if(l.size() == 0)
//                        continue;
//                    dst=(Address)entry.getKey();
//                    multicast=dst == null || dst.isMulticastAddress();
//                    synchronized(out_stream) {
//                        try {
//                            buffer=listToBuffer(l, multicast);
//                            doSend(buffer, dst, multicast);
//                        }
//                        catch(Throwable e) {
//                            if(log.isErrorEnabled()) log.error("exception sending msg", e);
//                        }
//                    }
//                }
//            }
//            finally {
//                msgs.clear();
//                num_msgs=0;
//            }
//        }
//    }



    private class Bundler {
        /** HashMap<Address, List<Message>>. Keys are destinations, values are lists of Messages */
        final HashMap       msgs=new HashMap(36);
        long                count=0;    // current number of bytes accumulated
        int                 num_msgs=0;
        long                start=0;
        BundlingTimer       bundling_timer=null;


        private synchronized void send(Message msg, Address dest) throws Exception {
            long length=msg.size();
            checkLength(length);

            if(start == 0)
                start=System.currentTimeMillis();
            if(count + length >= max_bundle_size) {
                cancelTimer();
                bundleAndSend();  // clears msgs and resets num_msgs
            }

            addMessage(msg, dest);
            count+=length;
            startTimer(); // start timer if not running
        }

        /** Never called concurrently with cancelTimer - no need for synchronization */
        private void startTimer() {
            if(bundling_timer == null || bundling_timer.cancelled()) {
                bundling_timer=new BundlingTimer();
                timer.add(bundling_timer);
            }
        }

        /** Never called concurrently with startTimer() - no need for synchronization */
        private void cancelTimer() {
            if(bundling_timer != null) {
                bundling_timer.cancel();
                bundling_timer=null;
            }
        }

        private void addMessage(Message msg, Address dest) { // no sync needed, never called by multiple threads concurrently
            List    tmp;
            synchronized(msgs) {
                tmp=(List)msgs.get(dest);
                if(tmp == null) {
                    tmp=new List();
                    msgs.put(dest, tmp);
                }
                tmp.add(msg);
                num_msgs++;
            }
        }



        private void bundleAndSend() {
            Map.Entry      entry;
            Address        dst;
            Buffer         buffer;
            List           l;
            Map copy;

            synchronized(msgs) {
                if(msgs.size() == 0)
                    return;
                copy=new HashMap(msgs);
                if(trace) {
                    long stop=System.currentTimeMillis();
                    double percentage=100.0 / max_bundle_size * count;
                    StringBuffer sb=new StringBuffer("sending ").append(num_msgs).append(" msgs (");
                    num_msgs=0;
                    sb.append(count).append(" bytes (" + f.format(percentage) + "% of max_bundle_size), collected in "+
                            + (stop-start) + "ms) to ").append(copy.size()).
                            append(" destination(s)");
                    if(copy.size() > 1) sb.append(" (dests=").append(copy.keySet()).append(")");
                    log.trace(sb.toString());
                }
                msgs.clear();
                count=0;
            }

            try {
                boolean multicast;
                for(Iterator it=copy.entrySet().iterator(); it.hasNext();) {
                    entry=(Map.Entry)it.next();
                    l=(List)entry.getValue();
                    if(l.size() == 0)
                        continue;
                    dst=(Address)entry.getKey();
                    multicast=dst == null || dst.isMulticastAddress();
                    synchronized(out_stream) {
                        try {
                            buffer=listToBuffer(l, multicast);
                            doSend(buffer, dst, multicast);
                        }
                        catch(Throwable e) {
                            if(log.isErrorEnabled()) log.error("exception sending msg: " + e.toString(), e.getCause());
                        }
                    }
                }
            }
            finally {
                start=0;
            }
        }



        private void checkLength(long len) throws Exception {
            if(len > max_bundle_size)
                throw new Exception("message size (" + len + ") is greater than max bundling size (" + max_bundle_size +
                        "). Set the fragmentation/bundle size in FRAG and TP correctly");
        }

        private class BundlingTimer implements TimeScheduler.Task {
            boolean cancelled=false;

            void cancel() {
                cancelled=true;
            }

            public boolean cancelled() {
                return cancelled;
            }

            public long nextInterval() {
                return max_bundle_timeout;
            }

            public void run() {
                bundleAndSend();
                cancelled=true;
            }
        }
    }



    private class DiagnosticsHandler implements Runnable {
        Thread t=null;
        MulticastSocket diag_sock=null;

        DiagnosticsHandler() {
        }

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

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

        void start() throws IOException {
            diag_sock=new MulticastSocket(diagnostics_port);
            java.util.List interfaces=Util.getAllAvailableInterfaces();
            bindToInterfaces(interfaces, diag_sock);

            if(t == null || !t.isAlive()) {
                t=new Thread(Util.getGlobalThreadGroup(), this, "DiagnosticsHandler");
                t.setDaemon(true);
                t.start();
            }
        }

        void stop() {
            if(diag_sock != null)
                diag_sock.close();
            t=null;
        }

        public void run() {
            byte[] buf=new byte[1500]; // MTU on most LANs
            DatagramPacket packet;
            while(!diag_sock.isClosed() && Thread.currentThread().equals(t)) {
                packet=new DatagramPacket(buf, 0, buf.length);
                try {
                    diag_sock.receive(packet);
                    handleDiagnosticProbe(packet.getSocketAddress(), diag_sock,
                                          new String(packet.getData(), packet.getOffset(), packet.getLength()));
                }
                catch(IOException e) {
                }
            }
        }

        private void bindToInterfaces(java.util.List interfaces, MulticastSocket s) {
            SocketAddress group_addr=new InetSocketAddress(diagnostics_addr, diagnostics_port);
            for(Iterator it=interfaces.iterator(); it.hasNext();) {
                NetworkInterface i=(NetworkInterface)it.next();
                try {
                    if (i.getInetAddresses().hasMoreElements()) { // fix for VM crash - suggested by [e-mail]
                        s.joinGroup(group_addr, i);
                        if(trace)
                            log.trace("joined " + group_addr + " on " + i.getName());
                    }
                }
                catch(IOException e) {
                    log.warn("failed to join " + group_addr + " on " + i.getName() + ": " + e);
                }
            }
        }
    }


}