controller.java

简单的分布式算法

/*
 *****************************************************************************
 * $Id: Controller.java,v 1.4 2003/05/01 23:57:34 jheiss Exp $
 *****************************************************************************
 * This class allows runtime control of the load balancer
 *****************************************************************************
 * Copyright 2003 Jason Heiss
 * 
 * This file is part of Distributor.
 * 
 * Distributor 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.
 * 
 * Distributor 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 Distributor; if not, write to the Free Software
 * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307 USA
 *****************************************************************************
 */

package oss.distributor;

import java.io.*;
import java.net.*;
import java.util.List;
import java.util.LinkedList;
import java.util.Iterator;
import java.util.StringTokenizer;
import java.util.logging.Logger;
import java.util.logging.Level;
import java.text.ParseException;

class Controller implements Runnable
{
	Distributor distributor;
	Logger logger;
	int port;
	List conns;
	Thread thread;
	ServerSocket controllerServer;

	protected Controller(Distributor distributor, int port)
	{
		this.distributor = distributor;
		logger = distributor.getLogger();
		this.port = port;

		// Use a linked list so we can remove dead connections in the
		// middle of the list easily.
		conns = new LinkedList();

		// Create a thread for ourselves and start it
		thread = new Thread(this, getClass().getName());
		thread.start();
	}

	public void run()
	{
		try
		{
			controllerServer = new ServerSocket();
			controllerServer.bind(new InetSocketAddress("127.0.0.1", port));
		}
		catch (IOException e)
		{
			logger.warning(
				"Error binding controller socket: " + e.getMessage());
			return;
		}

		while(true)
		{
			try
			{
				Socket socket = controllerServer.accept();
				logger.fine("Control connection from " + socket);
				ControllerConnection conn =
					new ControllerConnection(socket, distributor);
				conns.add(conn);
			}
			catch (IOException e)
			{
				logger.warning("Error accepting connection to controller: " +
					e.getMessage());
				shutdown();
				return;
			}

			removeDeadConnections();
		}
	}

	private void shutdown()
	{
		try
		{
			controllerServer.close();
		}
		catch (IOException e)
		{
			logger.warning(
				"Error closing controller socket: " + e.getMessage());
		}

		ControllerConnection c;
		Iterator i = conns.iterator();
		while (i.hasNext())
		{
			c = (ControllerConnection) i.next();
			c.close();
			i.remove();
		}
	}

	private void removeDeadConnections()
	{
		ControllerConnection c;

		Iterator i = conns.iterator();
		while (i.hasNext())
		{
			c = (ControllerConnection) i.next();
			if (c.isClosed())
			{
				logger.finest("Removing closed connection " + c);
				i.remove();
			}
		}
	}
}

class ControllerConnection implements Runnable
{
	Socket socket;
	Distributor distributor;
	Logger logger;
	BufferedReader in;
	PrintWriter out;
	Thread thread;
	boolean closed = false;

	protected ControllerConnection(
		Socket socket, Distributor distributor) throws IOException
	{
		this.socket = socket;
		this.distributor = distributor;
		logger = distributor.getLogger();

		// Good thing we don't care about speed...
		in =
			new BufferedReader(
				new InputStreamReader(
					socket.getInputStream()));
		out = new PrintWriter(socket.getOutputStream(), true);

		// Create a thread for ourselves and start it
		thread = new Thread(this, getClass().getName());
		thread.start();
	}

	public void run()
	{
		try
		{
			String line;
			StringTokenizer st;
			String command;
			while(! closed)
			{
				// Print a prompt
				out.print("distributor> ");
				out.flush();

				// Read a command
				line = in.readLine();
				logger.finest("Read '" + line + "' from " + socket);

				if (line == null)
				{
					logger.fine("Control connection " + socket + " closed");
					close();
					return;
				}
				
				st = new StringTokenizer(line);
				command = st.nextToken();
				logger.finer("Read '" + command + "' command from " + socket);

				if (command.equals("stats"))
				{
					stats(st);
				}
				else if (command.equals("threads"))
				{
					threads(st);
				}
				else if (command.equals("add"))
				{
					addTarget(st);
				}
				else if (command.equals("remove"))
				{
					removeTarget(st);
				}
				else if (command.equals("addgroup"))
				{
					addTargetGroup(st);
				}
				else if (command.equals("removegroup"))
				{
					removeTargetGroup(st);
				}
				else if (command.equals("disable"))
				{
					disableTarget(st);
				}
				else if (command.equals("enable"))
				{
					enableTarget(st);
				}
				else if (command.equals("loglevel"))
				{
					setLogLevel(st);
				}
				else if (command.equals("help"))
				{
					help(st);
				}
				else if (command.equals("quit"))
				{
					close();
					return;
				}
				else
				{
					out.println("Unknown command:  " + command);
					out.println("Try 'help' for a list of commands");
				}
			}
		}
		catch (IOException e)
		{
			logger.warning(
				"Error while communicating with controller client: " +
				e.getMessage());
			close();
			return;
		}
	}

	protected void help(StringTokenizer st)
	{
		out.println("Commands:");
		out.println("stats");
		out.println("threads");
		out.println("add");
		out.println("remove");
		out.println("addgroup");
		out.println("removegroup");
		out.println("disable");
		out.println("enable");
		out.println("loglevel");
		out.println("help");
		out.println("quit");
	}

	protected void stats(StringTokenizer st)
	{
		List targetGroups = distributor.getTargetGroups();
		synchronized (targetGroups)
		{
			List targets;
			Target target;
			Iterator targetIter;

			int tgCounter = 0;
			Iterator tgIter = targetGroups.iterator();
			while (tgIter.hasNext())
			{
				out.println("Target group " + tgCounter + ":");
				tgCounter++;

				targets = (List) tgIter.next();
				synchronized (targets)
				{
					targetIter = targets.iterator();
					while (targetIter.hasNext())
					{
						target = (Target) targetIter.next();
						out.println("  " + target);
					}
				}
			}
		}
	}

	/*
	 * Print a list of the threads in Distributor
	 */
	protected void threads(StringTokenizer st)
	{
		Thread[] tarray = new Thread[Thread.activeCount()];
		Thread.enumerate(tarray);
		for (int i=0 ; i<tarray.length ; i++)
		{
			out.println(tarray[i]);
		}
	}

	protected void addTarget(StringTokenizer st)
	{
		if (st.countTokens() != 3)
		{
			out.println("Usage: add <target group> <hostname> <port>");
			return;
		}

		int tgIndex;
		InetAddress addr;
		int port;
		Target newTarget;
		try
		{
			tgIndex = Integer.parseInt(st.nextToken());
			addr = InetAddress.getByName(st.nextToken());
			port = Integer.parseInt(st.nextToken());

			newTarget = new Target(
				distributor, addr, port,
				distributor.getConnectionFailureLimit(),
				distributor.getTerminate());
		}
		catch (UnknownHostException e)
		{
			out.println("Host not found:  " + e.getMessage());
			return;
		}
		catch (NumberFormatException e)
		{
			out.println("Target group and port must be integers");
			return;
		}

		List targetGroups = distributor.getTargetGroups();

		synchronized (targetGroups)
		{
			if (tgIndex < 0 || tgIndex >= targetGroups.size())
			{
				out.println("Invalid target group");
				return;
			}

			Iterator tgIter = targetGroups.iterator();
			int tgCounter = 0;
			while (tgIter.hasNext())
			{
				if (tgCounter == tgIndex)
				{
					List targets = (List) tgIter.next();
					targets.add(newTarget);
					out.println("New target added");
					break;
				}

				tgCounter++;
			}
		}
	}

	protected void removeTarget(StringTokenizer st)
	{
		if (st.countTokens() != 3)
		{
			out.println("Usage: remove <target group> <hostname> <port>");
			return;
		}

		int tgIndex;
		InetAddress addr;
		int port;
		try
		{
			tgIndex = Integer.parseInt(st.nextToken());
			addr = InetAddress.getByName(st.nextToken());
			port = Integer.parseInt(st.nextToken());
		}
		catch (UnknownHostException e)
		{
			out.println("Host not found:  " + e.getMessage());
			return;
		}
		catch (NumberFormatException e)
		{
			out.println("Target group and port must be integers");
			return;
		}

		List targetGroups = distributor.getTargetGroups();

		List targets = null;
		synchronized (targetGroups)
		{
			if (tgIndex < 0 || tgIndex >= targetGroups.size())
			{
				out.println("Invalid target group");
				return;
			}

			Iterator tgIter = targetGroups.iterator();
			int tgCounter = 0;
			while (tgIter.hasNext())
			{
				if (tgCounter == tgIndex)
				{
					targets = (List) tgIter.next();
					break;
				}
				tgCounter++;
			}
		}

		boolean targetFound = false;
		Target target = null;
		synchronized (targets)
		{
			Iterator targetIter = targets.iterator();
			while (targetIter.hasNext())
			{
				target = (Target) targetIter.next();
				if (target.getInetAddress().equals(addr) &&
					target.getPort() == port)
				{
					targetIter.remove();
					targetFound = true;
					break;
				}
			}
		}
		if (targetFound)
		{
			target.terminateAll();
			out.println("Target removed");
		}
		else
		{
			out.println("No matching target found");
		}
	}

	protected void addTargetGroup(StringTokenizer st)
	{
		if (st.countTokens() != 1)
		{
			out.println("Usage: addgroup <target group>");
			return;
		}

		int tgIndex;
		try
		{
			tgIndex = Integer.parseInt(st.nextToken());
		}
		catch (NumberFormatException e)
		{
			out.println("Target group must be an integer");
			return;
		}

		List targetGroups = distributor.getTargetGroups();

		synchronized (targetGroups)
		{
			if (tgIndex < 0 || tgIndex > targetGroups.size())
			{
				out.println("Invalid target group");
				return;
			}

			targetGroups.add(tgIndex, new LinkedList());
			out.println("New target group added at position " + tgIndex);
		}
	}

	protected void removeTargetGroup(StringTokenizer st)
	{
		if (st.countTokens() != 1)
		{
			out.println("Usage: removegroup <target group>");
			return;
		}

		int tgIndex;
		try
		{
			tgIndex = Integer.parseInt(st.nextToken());
		}
		catch (NumberFormatException e)
		{
			out.println("Target group must be an integer");
			return;
		}

		List targetGroups = distributor.getTargetGroups();

		// Remove the target group from the list of target groups
		List targets;
		synchronized (targetGroups)
		{
			if (tgIndex < 0 || tgIndex >= targetGroups.size())
			{
				out.println("Invalid target group");
				return;
			}

			targets = (List) targetGroups.remove(tgIndex);
		}

		// Shutdown the connections to each target in the target group
		synchronized (targets)
		{
			Iterator targetIter = targets.iterator();
			Target target;
			while (targetIter.hasNext())
			{
				target = (Target) targetIter.next();
				target.terminateAll();
			}
		}

		out.println("Target group at position " + tgIndex + " removed");
	}

	protected void disableTarget(StringTokenizer st)
	{
		if (st.countTokens() != 3)
		{
			out.println("Usage: disable <target group> <hostname> <port>");
			return;
		}

		int tgIndex;
		InetAddress addr;
		int port;
		try
		{
			tgIndex = Integer.parseInt(st.nextToken());
			addr = InetAddress.getByName(st.nextToken());
			port = Integer.parseInt(st.nextToken());
		}
		catch (UnknownHostException e)
		{
			out.println("Host not found:  " + e.getMessage());
			return;
		}
		catch (NumberFormatException e)
		{
			out.println("Target group and port must be integers");
			return;
		}

		List targetGroups = distributor.getTargetGroups();

		List targets = null;
		synchronized (targetGroups)
		{
			if (tgIndex < 0 || tgIndex >= targetGroups.size())
			{
				out.println("Invalid target group");
				return;
			}

			Iterator tgIter = targetGroups.iterator();
			int tgCounter = 0;
			while (tgIter.hasNext())
			{
				if (tgCounter == tgIndex)
				{
					targets = (List) tgIter.next();
					break;
				}
				tgCounter++;
			}
		}

		boolean targetFound = false;
		synchronized (targets)
		{
			Iterator targetIter = targets.iterator();
			Target target;
			while (targetIter.hasNext())
			{
				target = (Target) targetIter.next();
				if (target.getInetAddress().equals(addr) &&
					target.getPort() == port)
				{
					target.disable();
					targetFound = true;
					out.println("Target disabled");
					break;
				}
			}
		}
		if (!targetFound)
		{
			out.println("No matching target found");
		}
	}

	protected void enableTarget(StringTokenizer st)
	{
		if (st.countTokens() != 3)
		{
			out.println("Usage: enable <target group> <hostname> <port>");
			return;
		}

		int tgIndex;
		InetAddress addr;
		int port;
		try
		{
			tgIndex = Integer.parseInt(st.nextToken());
			addr = InetAddress.getByName(st.nextToken());
			port = Integer.parseInt(st.nextToken());
		}
		catch (UnknownHostException e)
		{
			out.println("Host not found:  " + e.getMessage());
			return;
		}
		catch (NumberFormatException e)
		{
			out.println("Target group and port must be integers");
			return;
		}

		List targetGroups = distributor.getTargetGroups();

		List targets = null;
		synchronized (targetGroups)
		{
			if (tgIndex < 0 || tgIndex >= targetGroups.size())
			{
				out.println("Invalid target group");
				return;
			}

			Iterator tgIter = targetGroups.iterator();
			int tgCounter = 0;
			while (tgIter.hasNext())
			{
				if (tgCounter == tgIndex)
				{
					targets = (List) tgIter.next();
					break;
				}
				tgCounter++;
			}
		}

		boolean targetFound = false;
		synchronized (targets)
		{
			Iterator targetIter = targets.iterator();
			Target target;
			while (targetIter.hasNext())
			{
				target = (Target) targetIter.next();
				if (target.getInetAddress().equals(addr) &&
					target.getPort() == port)
				{
					target.enable();
					targetFound = true;
					out.println("Target enabled");
					break;
				}
			}
		}
		if (!targetFound)
		{
			out.println("No matching target found");
		}
	}

	protected void setLogLevel(StringTokenizer st)
	{
		if (st.countTokens() != 1)
		{
			out.println("Usage: loglevel " +
				"off|severe|warning|info|config|fine|finer|finest|all");
			return;
		}

		Level newLevel;
		String levelName = st.nextToken();
		try
		{
			newLevel = Distributor.parseLogLevel(levelName);
		}
		catch (ParseException e)
		{
			out.println("Unrecognized log level");
			return;
		}

		logger.setLevel(newLevel);
	}

	protected boolean isClosed()
	{
		return closed;
	}

	protected void close()
	{
		try
		{
			socket.close();
		}
		catch (IOException e)
		{
			logger.warning(
				"Error closing connection to controller client: " +
				e.getMessage());
		}
		closed = true;
	}

	public String toString()
	{
		return("ControllerConnection from " + socket);
	}
}