fd_sock.java

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

            retval+=Global.INT_SIZE; // mbrs size
            if(mbrs != null) {
                for(int i=0; i < mbrs.size(); i++) {
                    retval+=Util.size((Address)mbrs.elementAt(i));
                }
            }

            return retval;
        }

        public void writeTo(DataOutputStream out) throws IOException {
            int size;
            out.writeByte(type);
            Util.writeAddress(mbr, out);
            Util.writeStreamable(sock_addr, out);
            size=cachedAddrs != null? cachedAddrs.size() : 0;
            out.writeInt(size);
            if(size > 0) {
                for(Iterator it=cachedAddrs.entrySet().iterator(); it.hasNext();) {
                    Map.Entry entry=(Map.Entry)it.next();
                    Address key=(Address)entry.getKey();
                    IpAddress val=(IpAddress)entry.getValue();
                    Util.writeAddress(key, out);
                    Util.writeStreamable(val, out);
                }
            }
            size=mbrs != null? mbrs.size() : 0;
            out.writeInt(size);
            if(size > 0) {
                for(Iterator it=mbrs.iterator(); it.hasNext();) {
                    Address address=(Address)it.next();
                    Util.writeAddress(address, out);
                }
            }
        }

        public void readFrom(DataInputStream in) throws IOException, IllegalAccessException, InstantiationException {
            int size;
            type=in.readByte();
            mbr=Util.readAddress(in);
            sock_addr=(IpAddress)Util.readStreamable(IpAddress.class, in);
            size=in.readInt();
            if(size > 0) {
                if(cachedAddrs == null)
                    cachedAddrs=new Hashtable();
                for(int i=0; i < size; i++) {
                    Address key=Util.readAddress(in);
                    IpAddress val=(IpAddress)Util.readStreamable(IpAddress.class, in);
                    cachedAddrs.put(key, val);
                }
            }
            size=in.readInt();
            if(size > 0) {
                if(mbrs == null)
                    mbrs=new Vector();
                for(int i=0; i < size; i++) {
                    Address addr=Util.readAddress(in);
                    mbrs.add(addr);
                }
            }
        }

    }


    /**
     * Handles the server-side of a client-server socket connection. Waits until a client connects, and then loops
     * until that client closes the connection. Note that there is no new thread spawned for the listening on the
     * client socket, therefore there can only be 1 client connection at the same time. Subsequent clients attempting
     * to create a connection will be blocked until the first client closes its connection. This should not be a problem
     * as the ring nature of the FD_SOCK protocol always has only 1 client connect to its right-hand-side neighbor.
     */
    private class ServerSocketHandler implements Runnable {
        Thread acceptor=null;
        /** List<ClientConnectionHandler> */
        final List clients=new ArrayList();


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

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

        ServerSocketHandler() {
            start();
        }

        final void start() {
            if(acceptor == null) {
                acceptor=new Thread(Util.getGlobalThreadGroup(), this, "ServerSocket acceptor thread");
                acceptor.setDaemon(true);
                acceptor.start();
            }
        }


        final void stop() {
            if(acceptor != null && acceptor.isAlive()) {
                try {
                    srv_sock.close(); // this will terminate thread, peer will receive SocketException (socket close)
                }
                catch(Exception ex) {
                }
            }
            synchronized(clients) {
                for(Iterator it=clients.iterator(); it.hasNext();) {
                    ClientConnectionHandler handler=(ClientConnectionHandler)it.next();
                    handler.stopThread();
                }
                clients.clear();
            }
            acceptor=null;
        }


        /** Only accepts 1 client connection at a time (saving threads) */
        public void run() {
            Socket client_sock;
            while(acceptor != null && srv_sock != null) {
                try {
                    if(trace) // +++ remove
                        log.trace("waiting for client connections on " + srv_sock.getInetAddress() + ":" +
                                  srv_sock.getLocalPort());
                    client_sock=srv_sock.accept();
                    if(trace) // +++ remove
                        log.trace("accepted connection from " + client_sock.getInetAddress() + ':' + client_sock.getPort());
                    ClientConnectionHandler client_conn_handler=new ClientConnectionHandler(client_sock, clients);
                    synchronized(clients) {
                        clients.add(client_conn_handler);
                    }
                    client_conn_handler.start();
                }
                catch(IOException io_ex2) {
                    break;
                }
            }
            acceptor=null;
        }
    }



    /** Handles a client connection; multiple client can connect at the same time */
    private static class ClientConnectionHandler extends Thread {
        Socket      client_sock=null;
        InputStream in;
        final Object mutex=new Object();
        final List clients=new ArrayList();

        ClientConnectionHandler(Socket client_sock, List clients) {
            setName("ClientConnectionHandler");
            setDaemon(true);
            this.client_sock=client_sock;
            this.clients.addAll(clients);
        }

        void stopThread() {
            synchronized(mutex) {
                if(client_sock != null) {
                    try {
                        OutputStream out=client_sock.getOutputStream();
                        out.write(NORMAL_TERMINATION);
                        out.flush();
                        closeClientSocket();
                    }
                    catch(Throwable t) {
                    }
                }
            }
        }

        void closeClientSocket() {
            synchronized(mutex) {
                Util.close(client_sock);
                client_sock=null;
            }
        }

        public void run() {
            try {
                synchronized(mutex) {
                    if(client_sock == null)
                        return;
                    in=client_sock.getInputStream();
                }
                int b=0;
                do {
                    b=in.read();
                }
                while(b != ABNORMAL_TERMINATION && b != NORMAL_TERMINATION);
            }
            catch(IOException ex) {
            }
            finally {
                Socket sock=client_sock; // PATCH: avoid race condition causing NPE
                if (sock != null && !sock.isClosed())
                    closeClientSocket();
                synchronized(clients) {
                    clients.remove(this);
                }
            }
        }
    }


    /**
     * Task that periodically broadcasts a list of suspected members to the group. Goal is not to lose
     * a SUSPECT message: since these are bcast unreliably, they might get dropped. The BroadcastTask makes
     * sure they are retransmitted until a view has been received which doesn't contain the suspected members
     * any longer. Then the task terminates.
     */
    private class BroadcastTask implements TimeScheduler.Task {
        final Vector suspected_mbrs=new Vector(7);
        boolean stopped=false;


        /** Adds a suspected member. Starts the task if not yet running */
        public void addSuspectedMember(Address mbr) {
            if(mbr == null) return;
            if(!members.contains(mbr)) return;
            synchronized(suspected_mbrs) {
                if(!suspected_mbrs.contains(mbr)) {
                    suspected_mbrs.addElement(mbr);
                    if(log.isDebugEnabled()) log.debug("mbr=" + mbr + " (size=" + suspected_mbrs.size() + ')');
                }
                if(stopped && suspected_mbrs.size() > 0) {
                    stopped=false;
                    timer.add(this, true);
                }
            }
        }


        public void removeSuspectedMember(Address suspected_mbr) {
            if(suspected_mbr == null) return;
            if(log.isDebugEnabled()) log.debug("member is " + suspected_mbr);
            synchronized(suspected_mbrs) {
                suspected_mbrs.removeElement(suspected_mbr);
                if(suspected_mbrs.size() == 0)
                    stopped=true;
            }
        }


        public void removeAll() {
            synchronized(suspected_mbrs) {
                suspected_mbrs.removeAllElements();
                stopped=true;
            }
        }


        /**
         * Removes all elements from suspected_mbrs that are <em>not</em> in the new membership
         */
        public void adjustSuspectedMembers(Vector new_mbrship) {
            Address suspected_mbr;

            if(new_mbrship == null || new_mbrship.size() == 0) return;
            synchronized(suspected_mbrs) {
                for(Iterator it=suspected_mbrs.iterator(); it.hasNext();) {
                    suspected_mbr=(Address) it.next();
                    if(!new_mbrship.contains(suspected_mbr)) {
                        it.remove();
                        if(log.isDebugEnabled())
                            log.debug("removed " + suspected_mbr + " (size=" + suspected_mbrs.size() + ')');
                    }
                }
                if(suspected_mbrs.size() == 0)
                    stopped=true;
            }
        }


        public boolean cancelled() {
            return stopped;
        }


        public long nextInterval() {
            return suspect_msg_interval;
        }


        public void run() {
            Message suspect_msg;
            FdHeader hdr;

            if(log.isDebugEnabled())
                log.debug("broadcasting SUSPECT message (suspected_mbrs=" + suspected_mbrs + ") to group");

            synchronized(suspected_mbrs) {
                if(suspected_mbrs.size() == 0) {
                    stopped=true;
                    if(log.isDebugEnabled()) log.debug("task done (no suspected members)");
                    return;
                }

                hdr=new FdHeader(FdHeader.SUSPECT);
                hdr.mbrs=(Vector) suspected_mbrs.clone();
            }
            suspect_msg=new Message();       // mcast SUSPECT to all members
            suspect_msg.putHeader(name, hdr);
            passDown(new Event(Event.MSG, suspect_msg));
            if(log.isDebugEnabled()) log.debug("task done");
        }
    }


}