distributedserver.java

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

/*
 *    This program is free software; you can redistribute it and/or modify
 *    it under the terms of the GNU General Public License as published by
 *    the Free Software Foundation; either version 2 of the License, or
 *    (at your option) any later version.
 *
 *    This program is distributed in the hope that it will be useful,
 *    but WITHOUT ANY WARRANTY; without even the implied warranty of
 *    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 *    GNU General Public License for more details.
 *
 *    You should have received a copy of the GNU General Public License
 *    along with this program; if not, write to the Free Software
 *    Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA.
 */

/*
 *    DistributedServer.java
 *    Copyright (C) 2002 Sebastian Celis, Dave Musicant
 *
 */

package weka.core;

import java.io.*;
import java.text.DateFormat;
import java.net.*;
import java.util.*;
import weka.estimators.*;
import weka.classifiers.*;
import weka.classifiers.evaluation.NominalPrediction;
import weka.attributeSelection.*;
import weka.gui.visualize.Plot2D;
import java.util.zip.GZIPInputStream;
import java.util.zip.GZIPOutputStream;

/**
 * Used by the client to tell the server if it is being run from
 * the command line or the graphical interface.
 */
interface Connection
{
    int
        CV_CLI = 0,
        CV_GUI = 1;
}

/**
 * Class for running WEKA code on multiple machines. <p>
 *
 * This is the class that actually calculates folds on a specific
 * cross-validation.  After it gets results, these results are sent back
 * to the client machine to be aggregated. <p>
 *
 * ------------------------------------------------------------------- <p>
 *
 * Example code from within an application:
 * <code> <pre>
 * int port = ... the port to listen on
 *
 * DistributedServer ds = new DistributedServer(port);
 * ds.start();
 * </pre> </code>
 *
 *
 * @author Dave Musicant ([e-mail])
 * @author Sebastian Celis ([e-mail])
 */
public class DistributedServer
{
    /** The port number the server will be listening on */
    private int port;

    /**
     * Sets the port that the server will listen on.
     *
     * @param port the port number the server will listen on
     */
    public DistributedServer(int port)
    {
        this.port = port;
    }

    /**
     * One instance of this class is used for each connection made to a
     * client computer running Weka.  This thread will continually run,
     * computing different folds in the cross-validation until the client
     * disconnects.
     *
     * @author Dave Musicant ([e-mail])
     * @author Sebastian Celis ([e-mail])
     */
    private class ConnectionThread extends Thread
    {
        /** The socket to listen on */
        private Socket sock;

        /** Used to read data from the socket */
        DataInputStream dis;

        /** Used to read objects from the socket */
        ObjectInputStream ois;

        /**
         * Sets the socket used for this conenction.
         *
         * @param sock the socket for the connection
         */
        ConnectionThread(Socket sock)
        {
            this.sock = sock;
        }

        /**
         * The actual method of the thread that will perform
         * cross validations requested by a client machine and
         * send the results of these cross validations back to it.
         */
        public void run()
        {
            BufferedInputStream bis;
            int connection;

	    try
	    {
		try
		{
		    // Create the data input stream
		    bis = new BufferedInputStream(sock.getInputStream());
		    dis = new DataInputStream(bis);
		    ois = new ObjectInputStream(bis);

		    // Find out who is connecting
		    connection = dis.readInt();

		    switch(connection)
		    {
                        case Connection.CV_CLI:
			    crossValidateCLI();
			    break;
                        case Connection.CV_GUI:
			    crossValidateGUI();
			    break;
		    }
		}
		catch(Exception e)
		{
		    System.out.print(new Date() + ": ");
		    System.out.println("Connection with " +
				       sock.getInetAddress().getHostName() +
				       " closed.");
		    return;
		}
		finally
		{
		    // Close the socket no matter what exceptions occurred
		    sock.close();
		}
	    }
	    catch(Exception e)
	    {
		System.out.print(new Date() + ": ");
		System.out.println("Connection with " +
				   sock.getInetAddress().getHostName() +
				   " closed.");
		return;
	    }
        }

        /**
         * Processes parallel fold requests for cross-validation that is being
         * run through Weka's command-line interface.
         *
         * @exception Exception if anything goes wrong with the
         * client-server connection
         */
        private void crossValidateCLI() throws Exception
        {
            ObjectOutputStream oos;
            Instances data;
            Evaluation evaluation;
            Classifier classifier;
            int index;
            int numFolds;

            // Get the necessary information from the client
            data = (Instances) ois.readObject();
            classifier = (Classifier) ois.readObject();
            numFolds = dis.readInt();
            index = dis.readInt();

            // Make the evaluation object out of the data created
            evaluation = new Evaluation(data, null);

            // Run the fold
            Instances train = data.trainCV(numFolds, index);
            evaluation.setPriors(train);
            classifier.buildClassifier(train);
            Instances test = data.testCV(numFolds, index);
            evaluation.evaluateModel(classifier, test);

            // Create the OutputObjectStream
            oos = new ObjectOutputStream(
                      new BufferedOutputStream(
                          sock.getOutputStream()));

            // Send the results back to the client
            oos.writeObject(evaluation);
            oos.flush();

            while(true)
            {
                // Find out which fold we should run
                index = dis.readInt();

                // Create a new evaluation object
                evaluation = new Evaluation(data, null);

                // Run the fold
                train = data.trainCV(numFolds, index);
                evaluation.setPriors(train);
                classifier.buildClassifier(train);
                test = data.testCV(numFolds, index);
                evaluation.evaluateModel(classifier, test);

                // Send the object back to the client
                oos.writeObject(evaluation);
                oos.flush();
            }
        }

        /**
         * Processes parallel fold requests for cross-validation that is being
         * run through Weka's graphical interface.
         *
         * @exception Exception if anything goes wrong with the
         * client-server connection
         */
        private void crossValidateGUI() throws Exception
        {
            ObjectOutputStream oos;
            Instances data;
            Evaluation evaluation;
            Classifier classifier;
            int index;
            int numFolds;
            int connection;
            FastVector predictions = null;
            FastVector hv = null;
            FastVector plotShape = null;
            FastVector plotSize = null;
            Instances predInstances = null;

            // Get the necessary information from the client
            data = (Instances) ois.readObject();
            classifier = (Classifier) ois.readObject();
            numFolds = dis.readInt();
            index = dis.readInt();

            // Create the evaluation object to be used for this fold
            evaluation = new Evaluation(data, null);

            // Initialize the necessary variables
            plotShape = new FastVector();
            plotSize = new FastVector();
            if (data.classAttribute().isNominal() &&
                classifier instanceof DistributionClassifier)
            {
                predictions = new FastVector();
            }
            hv = new FastVector();
            Attribute predictedClass;
            Attribute classAt =
                data.attribute(data.classIndex());
            if (classAt.isNominal())
            {
                FastVector attVals = new FastVector();
                for (int i = 0; i < classAt.numValues(); i++)