evaluationclient.java

:<<数据挖掘--实用机器学习技术及java实现>>一书的配套源程序

    public void start() throws Exception
    {
        ConnectionService [] connect;
        ClientSideComputations cs = new ClientSideComputations();

        // Configure the client
        configure();

	connect = new ConnectionService[computers.size()];

        // Start a thread for each server
        for(int i = 0; i < computers.size(); i++)
        {
	    try 
	    {
		connect[i] = new ConnectionService((String)computers.get(i));
	    } catch (Exception e)
	    {
		e.printStackTrace();
	    }
            connect[i].start();
        }

        // Let the computer running the client also do some work to speed
        // things up
        cs.start();

        try
        {
            synchronized(evaluation)
            {
                // The main thread waits on the evaluation object.  It will
                // receive a signal when the cross-validation is complete.
                evaluation.wait();
            }
        }
        catch (InterruptedException e)
        {
            e.printStackTrace();
        }
    }

    /**
     * Performs part of the cross-fold validation on the computer running
     * Weka.  This yields a speed increase since this computer will
     * be calculating data at the same time the other computers are doing
     * so.
     *
     * @author Sebastian Celis ([e-mail])
     * @author Dave Musicant ([e-mail])
     */
    private class ClientSideComputations extends Thread
    {

        /**
         * This is the actual thread which performs the necessary
         * computations.
         */
        public void run()
        {
            int index;

            try
            {
                // Determine which section we should train on
                index = determineIndex();

                while(index != -1)
                {
                    // Do the cross validation
                    Evaluation anEvaluation = new Evaluation(data, null);
                    Instances train = data.trainCV(numFolds, index);
                    anEvaluation.setPriors(train);
                    classifier.buildClassifier(train);
                    Instances test = data.testCV(numFolds, index);
                    anEvaluation.evaluateModel(classifier, test);

                    synchronized(status)
                    {
                        synchronized(evaluation)
                        {
                            if(status[index] != Status.DONE)
                                evaluation.aggregate(anEvaluation);
                            status[index] = Status.DONE;
                        }
                    }

                    // Determine which section we should train on
                    index = determineIndex();
                }
                synchronized(evaluation)
                {
                    evaluation.notifyAll();
                }
            }
            catch (Exception e)
            {
                e.printStackTrace();
                return;
            }
        }
    }

    /**
     * One instance of this class will connect to a single computer
     * and send it the information necessary for computing cross-validations.
     * This class instructs the other computer concerning which folds it
     * should work on.  When the other computer finishes a fold, it sends
     * the results back to this class, which then compiles that information.
     *
     * @author Dave Musicant ([e-mail])
     * @author Sebastian Celis ([e-mail])
     */
    private class ConnectionService extends Thread
    {
        /** The address of the computer to connect to */
        private InetAddress serverAddress;

        /** The current fold to work on */
        private int index;

        /**
         * Gets the internet address of the computer to connect to by
         * looking up the server's name.
         *
         * @param serverName the name of computer to connect to
         */
        ConnectionService(String serverName)
        {
            try
            {
                this.serverAddress = InetAddress.getByName(serverName);
            }
            catch(UnknownHostException e)
            {
                // The server does not exist, so exit cleanly.
                return;
            }
        }

        /**
         * The actual thread that will connect to the other computer,
         * send data to it, and receive data from it.
         */
        public void run()
        {
            /** Used to send bytes to the server */
            BufferedOutputStream bos;

            /** Used to send objects to the server */
            ObjectOutputStream oos;

            /** Used to send ints to the server */
            DataOutputStream dos;

            /** Used to receive objects from the server */
            ObjectInputStream ois;

            /** The connection to the server */
            Socket sock;

            try
            {
                // Determines the server's InetAddress and then connects
                // to this server
                sock = new Socket(serverAddress, port);

                try
                {
                    try
                    {
                        index = determineIndex();
                        if(index != -1)
                        {
                            // Create the output streams
                            bos = new BufferedOutputStream(
                                      sock.getOutputStream());
                            oos = new ObjectOutputStream(bos);
                            dos = new DataOutputStream(bos);

                            // Write all necessary info to the server
                            dos.writeInt(Connection.CV_CLI);
                            oos.writeObject(data);
                            oos.writeObject(classifier);
                            oos.flush();
                            dos.writeInt(numFolds);
                            dos.writeInt(index);
                            dos.flush();

                            // Create the input stream and get the results
                            // back from the server
                            ois = new ObjectInputStream(
                                      new BufferedInputStream(
                                          sock.getInputStream()));
                            Evaluation newEvaluation
                                = (Evaluation)ois.readObject();

                            synchronized(status)
                            {
                                synchronized(evaluation)
                                {
                                    if(status[index] != Status.DONE)
                                        evaluation.aggregate(newEvaluation);
                                    status[index] = Status.DONE;
                                }
                            }

                            synchronized(otherComputers)
                            {
                                otherComputers.append(serverAddress+"\n");
                            }

                            // Determine which section we should train on
                            index = determineIndex();

                            while(index != -1)
                            {
                                // Write this number to the server
                                dos.writeInt(index);
                                dos.flush();

                                // Get the results from the server
                                newEvaluation
                                    = (Evaluation)ois.readObject();

                                synchronized(status)
                                {
                                    synchronized(evaluation)
                                    {
                                        if(status[index] != Status.DONE)
                                           evaluation.aggregate(newEvaluation);
                                        status[index] = Status.DONE;
                                    }
                                }

                                // Determine which section we should train on
                                index = determineIndex();
                            }

                            synchronized(evaluation)
                            {
                                evaluation.notifyAll();
                            }
                        }
                    }
                    catch (Exception e)
                    {
                        // There was a problem with this connection.
                        // One possibility is that the server ran out of
                        // memory.  Thus, we simply close the connection.
                        return;
                    }
                }
                finally
                {
                    // Close the connection no matter what exceptions were
                    // thrown
                    sock.close();
                }
            }
            catch(Exception e)
            {
                // We could not connect, so exit cleanly
                return;
            }
        }
    }
}