sun/jist/JISTAgent.java

/*
 * CDDL HEADER START
 *
 * The contents of this file are subject to the terms of the
 * Common Development and Distribution License (the "License").
 * You may not use this file except in compliance with the License.
 *
 * You can obtain a copy of the license at usr/src/OPENSOLARIS.LICENSE
 * or http://www.opensolaris.org/os/licensing.
 * See the License for the specific language governing permissions
 * and limitations under the License.
 *
 * When distributing Covered Code, include this CDDL HEADER in each
 * file and include the License file at usr/src/OPENSOLARIS.LICENSE.
 * If applicable, add the following below this CDDL HEADER, with the
 * fields enclosed by brackets "[]" replaced with your own identifying
 * information: Portions Copyright [yyyy] [name of copyright owner]
 *
 * CDDL HEADER END
 */

/*
 * Copyright 2008 Sun Microsystems, Inc.  All rights reserved.
 * Use is subject to license terms.
 *
 * ident	"@(#)JISTAgent.java	1.25	08/12/04 SMI"
 */

package com.sun.jist;
import com.sun.jist.JISTData;
import com.sun.jist.JISTLogic;
import com.sun.jist.iscsi.iSCSIData;
import java.io.BufferedWriter;
import java.io.IOException;
import java.io.OutputStreamWriter;
import java.io.Writer;
import java.net.InetAddress;
import java.net.ServerSocket;
import java.net.Socket;
import java.net.SocketException;
import java.net.SocketTimeoutException;
import java.net.UnknownHostException;

/**
 * JIST Proxy Agent used to intercept communications for Queries,
 * Additions, Changes, and/or Deletions.
 * <p>
 * Verifies Compliance to IETF iSCSI and allows for SCSI fault
 * injection.
 * <p>
 * @author	Joel.Buckley (at) Sun.COM
 * @since	5.0
 */
public class JISTAgent extends JISTLogic {

/** Declare & Initialize "public final static" Constants. */

/**
 * Spin cycle in Milliseconds.
 * <p>
 * This is used for Read & Write Threads to avoid hangs.
 */
private final static long SPIN = 50L;

/** Data IN "READ" Direction. */
private final static byte DATA_IN = 0;

/** Data OUT "WRITE" Direction. */
private final static byte DATA_OUT = 1;

/**
 * JIST iSCSI Proxy Target Server Socket IP Address.
 * <p>
 * This is set by Java Environment Property "proxyIP=localhost".  By default,
 * this is set to localhost or 127.0.0.1.
 */
public final static String proxyIP;

/** Proxy Port Mutex. */
private final static Object proxyPortMutex = new Object();

/** Declare "static" Class Variables. */

/**
 * JIST iSCSI Proxy Management Server Socket Port.
 * <p>
 * This is set by Java Environment Property "mgntPort=ffffffff". By
 * default this is set to -1 and must be specified.
 */
public static volatile int mgntPort;

/**
 * JISTAgent Thread number, per JVM Counter.
 */
private static volatile int threadID;

/**
 * JIST iSCSI Proxy Target Server Socket Port.
 * <p>
 * This is set by Java Environment Property "proxyPort=ffffffff". By
 * default this is set to -1 and must be specified.
 */
public static volatile int[] proxyPort;

/**
 * Actual iSCSI Target Server Socket Port.
 * <p>
 * This is set by Java Environment Property "targetPort=cbc".  By
 * default, this is set to 3260 or 0x0cbc.
 */
public static volatile int[] targetPort;

/**
 * Actual iSCSI Target Server Socket IP Address.
 * <p>
 * This is set by Java Environment Property "targetIP=localhost".  By
 * default, this is set to localhost or 127.0.0.1.
 */
public static volatile String[] targetIP;

/** Declare non-"static" Instance Variables. */

/**
 * JISTAgent Thread number, instance specific.
 */
private volatile String myID;

/** Declare and Initialize non-"static" Instance Constants. */

/**
 * JISTAgent ServerSocket Thread.
 */
private final boolean serverSocket;

/**
 * JISTAgent Read or Write Thread.
 */
private final boolean readAgent;

/**
 * JIST iSCSI Proxy Target Server Index.
 * <p>
 * This value is used associate a ProxyPort with a TargetIP/TargetPort
 * combination.  This is necessary for iSCSI Target Port Redirection.
 */
private final int proxyIndex;

/**
 * JISTAgent Master Thread.
 */
private final JISTAgent master;

/** Initialize "static" Class Variables. */
static {
	/** Booleans. */

        /** I/O Paths. */

        /** Test Settings. */
	mgntPort = getIntProperty("mgntPort", -1);
	proxyPort = new int[1];
	proxyPort[0] = getIntProperty("proxyPort", -1);
	targetPort = new int[1];
	targetPort[0] = getIntProperty("targetPort", 3260);
	String ip;
	try {
		ip = InetAddress.getLocalHost().getHostAddress();
		ip = InetAddress.getByName(getStringProperty("proxyIP", ip))
		    .getHostAddress();
	} catch (UnknownHostException e) {
		ip = "127.0.0.1";
	}
	proxyIP = ip;
	targetIP = new String[1];
	try {
		ip = InetAddress.getLocalHost().getHostAddress();
		ip = InetAddress.getByName(getStringProperty("targetIP", ip))
		    .getHostAddress();
	} catch (UnknownHostException e) {
		ip = "127.0.0.1";
	}
	targetIP[0] = ip;

        /** Internal Variables - long. */

        /** Internal Variables - int. */
	threadID = 0;

        /** Internal Variables - short. */

        /** Internal Variables - byte. */

        /** Internal Variables - Arrays. */

        /** Internal Variables - String. */

        /** Internal Variables - misc. */
}

/** Initialize non-"static" Instance Variables. */
{
	/** Booleans. */

        /** I/O Paths. */

        /** Test Settings. */

        /** Internal Variables - long. */

        /** Internal Variables - int. */

        /** Internal Variables - short. */

        /** Internal Variables - byte. */

        /** Internal Variables - Arrays. */

        /** Internal Variables - String. */
	myID = "JISTAgent" + (++threadID);

        /** Internal Variables - misc. */
}

/**
 * First constructor for JIST Agent ServerSocket use.
 * <p>
 * Specifing a negative value for proxyIndex starts the JIST Proxy Management
 * Server.
 * <p>
 * Specifing a positive value for proxyIndex starts a JIST Proxy iSCSI Target
 * Server.
 * <p>
 * @param proxyIndex JIST Proxy iSCSI Target Server Index.
 */
public JISTAgent(int proxyIndex) {
	super();
	master = this;
	tester = this;
	serverSocket = true;
	readAgent = true;
	this.dataAgent = (proxyIndex >= 0);
	this.proxyIndex = proxyIndex;
}

/**
 * Second constructor to establish a new JIST Agent socket.
 * <p>
 * @param socket JISTAgent external Socket socket.
 * @param proxyIndex JIST Proxy iSCSI Target Server Index.
 */
private JISTAgent(Socket socket, int proxyIndex) {
	super();
	master = this;
	tester = this;
	serverSocket = false;
	readAgent = true;
	this.socket = socket;
	this.dataAgent = (proxyIndex >= 0);
	this.targetAgent = true;
	this.proxyIndex = proxyIndex;

	if (dataAgent) {
		if (mgntPort == -1) {
			// Create Initiator Read Thread
			this.peer = new JISTAgent(this, proxyIndex, dataAgent,
			    !targetAgent);
		} else {
			// Create Management Write Thread
			this.peer = new JISTAgent(this, proxyIndex, !dataAgent,
			    targetAgent);
		}
		Thread t = new Thread(peer);
		t.setPriority(Thread.NORM_PRIORITY + 1);
		t.start();
	}
}

/**
 * Third constructor to link JIST Initiator, Delta, & Target Agents.
 * <p>
 * In this constructor, the peer passes a reference to itself
 * such that bidirectional communications can be establish
 * between the JISTAgent instances.
 * <p>
 * @param peer JISTAgent Peer Thread asking for the object.
 * @param proxyIndex JIST Proxy iSCSI Target Server Index.
 * @param targetAgent JISTAgent Target or Initiator Thread flag.
 */
private JISTAgent(JISTAgent peer, int proxyIndex, boolean dataAgent,
    boolean targetAgent) {
	super();
	master = this;
	tester = this;
	serverSocket = false;
	this.peer = peer;
	this.proxyIndex = proxyIndex;
	this.dataAgent = dataAgent;
	this.targetAgent = targetAgent;
	dupBufferedWriter(peer);

	/*
	 * Data Agents default to readAgent as primary thread.
	 * Management Agents default to write agent as primary thread.
	 */
	this.readAgent = dataAgent;

	if (mgntPort != -1 && !dataAgent) {
		// Create Initiator Read Thread
		this.peer2 =
		    new JISTAgent(this, proxyIndex, !dataAgent, !targetAgent);
		Thread t = new Thread(peer2);
		t.setPriority(Thread.NORM_PRIORITY + 1);
		t.start();
	}
}

/**
 * Forth constructor to Start Read & Write Threads.
 * <p>
 * In this constructor, the peer passes a reference to itself
 * such that bidirectional communications can be establish
 * between the JISTAgent instances.
 * <p>
 * @param master JISTAgent Master Thread asking for the object.
 * @param dataAgent JISTAgent Data or Management Thread flag.
 * @param targetAgent JISTAgent Target or Initiator Thread flag.
 * @param readAgent JISTAgent Read or Write Thread flag.
 */
private JISTAgent(JISTAgent master, boolean dataAgent, boolean targetAgent,
    boolean readAgent) {
	super();
	serverSocket = false;
	this.master = master;
	this.tester = master;
	this.proxyIndex = master.proxyIndex;
	this.dataAgent = dataAgent;
	this.targetAgent = targetAgent;
	this.readAgent = readAgent;
	this.peer = master.peer;
	dupBufferedWriter(tester);
}

/**
 * Main entry point.
 */
public static void main(String[] args) {
	Thread t = new Thread(new JISTAgent(0));	// Proxy
	t.setPriority(Thread.NORM_PRIORITY + 1);
	t.start();
	new Thread(new JISTAgent(-1)).start();	// Management
	/* The above threads will prevent the object from exiting... */
}

/**
 * Main execution engines for listening ServerSocket, individual
 * socket reader, and individual socket writers.
 */
public void run() {
	/** Verify new Thread full name. */
	if (reportLevel > 8) {
		myID = new StringBuilder(myID)
		    .append(dataAgent ? ":\"Data " : ":\"Management ")
		    .append(targetAgent ? "Target " : "Initiator ")
		    .append(readAgent ? "Read " : "Write ")
		    .append(serverSocket ? "Server\"" : "Connection\"")
		    .toString();
	}
	Thread.currentThread().setName(myID);

	log(MILESTONE, "JISTAgent Thread Start", myID);
	if (serverSocket) {
		// Handle ServerSockets here.
		listen();
	} else if (!dataAgent && readAgent) {
		// Handle Management Connections.
		interact();
	} else if (readAgent) {
		// Start Write Connection Thread.
		Thread t = new Thread(new JISTAgent(
		    this, dataAgent, targetAgent, false));
		t.setPriority(Thread.NORM_PRIORITY + 1);
		t.start();

		// Data Read Connection Threads.
		if (master.openByteSocket(master, targetIP[proxyIndex],
		    targetPort[proxyIndex])) {
			JISTData obj;
			while ((obj = new iSCSIData()) != null &&
			    master.read(obj, 3600) && peer.writeQueue(obj));
		}
		master.deStress(master);
		master.close(master);
	} else {
		// Data Write Connection Threads.
		if (master.openByteSocket(master, targetIP[proxyIndex],
		    targetPort[proxyIndex])) {
			/** Core loop for Write Threads. No Printing Here. */
			while (running && tester.stressing &&
			    master.writeImmediate(
			    ((JISTData)master.dequeue())));
		}
		master.deStress(master);
		master.close(master);
	}
	log(MILESTONE, "JISTAgent Thread Complete", myID);
}

/**
 * Set Stressing to false to speed data socket thread shutdown.
 * <p>
 * @param customer Who is asking for the service.
 */
private final void deStress(JISTAgent customer) {
	tester.stressing = false;
	if (null != peer2 && peer2 != customer) {
		peer2.deStress(customer);
	}
	if (null != peer && peer != customer) {
		peer.deStress(customer);
	}
}

/**
 * Listen for incoming socket connections.
 * <p>
 * Note: All output goes to the console.
 */
private final void listen() {
	int localPort, remotePort;
	String remoteIP;
	ServerSocket ss = null;
	Thread t = null;

	if (proxyIndex < 0) {
		log(MILESTONE, "Security Notice", getMsg(107));
	}
	synchronized (proxyPortMutex) {
		if (!dataAgent) {
			localPort = proxyPort[0];
			remotePort = -1;
		} else if (proxyIndex < 0 || proxyIndex >= proxyPort.length) {
			localPort = -1;
			remotePort = -1;
		} else {
			localPort = proxyPort[proxyIndex];
			remotePort = targetPort[proxyIndex];
		}
	}

	if (!dataAgent && mgntPort < 0) {
		/** No Delta Agent.  Nothing else to do. */
		return;
	} else if (dataAgent && localPort < 0) {
		/** No Proxy Agent.  Nothing else to do. */
		return;
	} else if (mgntPort == localPort) {
		log(WARNING,
		    "JIST iSCSI Proxy Management & Target ports conflict",
		    getMsg(902));
		localPort = -1;
	} else if (dataAgent && localPort == remotePort) {
		synchronized (proxyPortMutex) {
			remoteIP = targetIP[proxyIndex];
		}
		try {
			if (InetAddress.getByName(remoteIP)
			    .isAnyLocalAddress()) {
				log(WARNING,
				    "JIST Proxy & Target Ports conflict.",
				    getMsg(902));
				localPort = -1;
			}
		} catch (UnknownHostException uhe) {
			log(WARNING, "iSCSI Target IP Address undefined",
			    getMsg(902));
			localPort = -1;
		}
	}

	/** Unlock thread that created this thread. */
	if (dataAgent && localPort < 0) {
		synchronized (proxyPortMutex) {
			proxyPort[proxyIndex] = localPort;
			proxyPortMutex.notifyAll();
		}
		return;
	}

	try {
		if (dataAgent) {
			ss = new ServerSocket(localPort);
			localPort = ss.getLocalPort();
			synchronized (proxyPortMutex) {
				proxyPort[proxyIndex] = localPort;
				proxyPortMutex.notifyAll();
			}
		} else {
			ss = new ServerSocket(mgntPort);
		}
		ss.setSoTimeout(1000); // 1 second timeout.
		while (running) {
			try {
				// New Target or Management Read Thread
				t = new Thread(new JISTAgent(ss.accept(),
				    proxyIndex));
				t.setPriority(Thread.NORM_PRIORITY + 1);
				t.start();
			} catch (SocketTimeoutException e) {
				// keep checking running variable...
			}
		}
	} catch (IOException e) {
		log(WARNING, "JISTAgent Server Socket, Unexpected Error", e);
	} finally {
		try {
			if (ss != null) {
				ss.close();
			}
		} catch (IOException e2) {
			log(DEBUG, "Close Server Socket Exception", e2);
		}
	}
}

/**
 * Interact with management socket connections.
 * <p>
 * Note:jj
 */
private final void interact() {
	final String complete = getMsg(196) + newline;
	final String prompt = getMsg(410);
	final String who = new StringBuilder(myID).append(" from ")
	    .append(socket.getInetAddress().getCanonicalHostName()).toString();
	String command;
	if (openCharSocket(this, "", -1)) {
		log(MILESTONE, "JISTAgent Thread Interacting", who);
		while (running) {
			if (!write(prompt)) {
				log(WARNING, "JISTAgent Thread Socket Closed",
				    who);
				break;
			}
			if ((command = readLine()) == null) {
				log(WARNING, "JISTAgent Thread Input Closed",
				    who);
				break;
			}
			if (command.equals("") || command.matches("\\s*")) {
				continue;
			}
			if (command.equals("exitall")) {
				log(MILESTONE, "JISTAgent Thread Exit All",
				    who);
				running = false;
				close(this);
				break;
			}
			if (command.equals("exit")) {
				log(MILESTONE, "JISTAgent Thread Exit", who);
				close(this);
				break;
			}
			if (!running) {
				log(MILESTONE, "JISTAgent Thread Not Running",
				    who);
				close(this);
				break;
			}
			if (command.startsWith("inject")) {
				String[] args = command.split("\\s");
				if (args.length > 1) {
					while (!JISTData.addMask(args[1])) {
						write(getMsg(411));
					}
				}
			} else if (command.startsWith("com.sun.jist")) {
				execute(command);
			} else if (!write(new StringBuilder(
			    "Unrecognized request \"").append(command)
			    .append("\", skipping.").toString())) {
				break;
			}
			if (!write(complete)) {
				log(WARNING, "JISTAgent Thread Output Closed",
				    who);
				break;
			}
		}
	} else {
		log(WARNING, "JISTAgent Thread Socket DOA", who);
	}
}

/**
 * Write JISTData Object to Byte Stream Output.
 * <p>
 * @param data JIST Data Object to be written.
 */
private final boolean writeQueue(JISTData data) {
	if (data == null) {
		// Read failure... initiate closing process.
		log(DEBUG, "JISTAgent Write Data Null",
		    new StringBuilder(myID).append(" Called By ")
		    .append(Thread.currentThread().getName()).toString());
		return false;
	}
	if (data.getImmediate() == 0 && master.enqueue(data)) {
		// Regular Command and available Queue.
		return true;
	} else {
		// PDU Immediate bit set or uninitialized Queue.
		return writeImmediate(data);
	}
}

/**
 * Internal write method behind write queue.
 * <p>
 * @param data JIST Data Object to be written.
 */
private final boolean writeImmediate(JISTData data) {
	if (data == null) {
		// Read failure... initiate closing process.
		log(DEBUG, "JISTAgent Write Data Null",
		    new StringBuilder(myID).append(" Called By ")
		    .append(Thread.currentThread().getName()).toString());
		return false;
	}
	if (dataAgent) {
		if (master.write(data) &&
		    data.getOpcode() != 0x026) {
			return true;
		} else {
			/**
			 * Initiate closing process on failed write or
			 * iSCSI Logout Request PDU.
			 */
			return false;
		}
	} else {
		// Data Manipulation done here...
		if (reportLevel > 8) {
			log(DEBUG, "JISTAgent iSCSI PDU Header, before Mask",
			    data.getHeader());
			log(DEBUG, "JISTAgent iSCSI PDU Payload, before Mask",
			    data.getPayload());
			switch (data.getOpcode()) {
			case(0x003):
			case(0x004):
			case(0x023):
			case(0x024):
				log(DEBUG, "JISTAgent iSCSI PDU " +
				    "Payload, before Mask, in ASCII",
				    new String(data.getPayload(), 0,
				    data.getPayload().length)
				    .replaceAll("\00", newline));
				break;
			}
		}


		JISTData atad; // Data backwards :)
		JISTData[] masked = data.mask();
		if (masked == null) {
			/** Attempting to Disconnect... Aburptly. */
			return false;
		}
		int index = masked.length;

		while (index > 0) {
			atad = masked[--index];

			if (atad == null) {
				continue;
			}

			if (reportLevel > 8) {
				log(DEBUG,
				    "JISTAgent iSCSI PDU Header, after Mask",
				    atad.getHeader());
				log(DEBUG,
				    "JISTAgent iSCSI PDU Payload, after Mask",
				    atad.getPayload());
				switch (data.getOpcode()) {
				case(0x003):
				case(0x004):
				case(0x023):
				case(0x024):
					log(DEBUG, "JISTAgent iSCSI PDU " +
					    "Payload, after Mask, in ASCII",
					    new String(atad.getPayload(), 0,
					    atad.getPayload().length)
					    .replaceAll("\00", newline));
					break;
				}
			}

			if (atad.getDirection() == DATA_IN) {
				if (!peer.writeQueue(atad)) {
					return false;
				}
			} else {
				if (!peer2.writeQueue(atad)) {
					return false;
				}
			}
		}

		return true;
	}
}

/**
 * Common method to lookup and, if necessary, increment the number of JIST
 * iSCSI Proxy Target Servers.
 * <p>
 * @param ip iSCSI Target IP Address.
 * @param port iSCSI Target Port Number.
 * @return proxyIndex Position where IP/Port combination was placed.
 */
public static int getProxyIndex(String ip, int port) {
	int inx;
	int[] proxyPortNew, targetPortNew;
	String[] targetIPNew;
	Thread t;

	synchronized (proxyPortMutex) {
		inx = targetIP.length;

		/** Is the IP:Port combination served by a Proxy? */
		while (--inx >= 0) {
			if (targetPort[inx] == port &&
			    targetIP[inx].equals(ip)) {
				return (inx);
			}
		}

		/** Well, Proxy not found... Create one... */
		inx = targetIP.length;

		proxyPortNew = new int[inx + 1];
		System.arraycopy(proxyPort, 0, proxyPortNew, 0,
		    proxyPort.length);
		proxyPortNew[inx] = 0; /** Temp. until Thread started below. */
		proxyPort = proxyPortNew;

		targetPortNew = new int[inx + 1];
		System.arraycopy(targetPort, 0, targetPortNew, 0,
		    targetPort.length);
		targetPortNew[inx] = port;
		targetPort = targetPortNew;

		targetIPNew = new String[inx + 1];
		System.arraycopy(targetIP, 0, targetIPNew, 0, targetIP.length);
		targetIPNew[inx] = ip;
		targetIP = targetIPNew;

		/** Start the JIST iSCSI Proxy Target Server. */
		t = new Thread(new JISTAgent(inx));	// Proxy
		t.setPriority(Thread.NORM_PRIORITY + 1);
	}

	/** Start the Thread outside the Synchronized Block. */
	t.start();

	synchronized (proxyPortMutex) {
		/** Wait for the real proxyPort to be assigned. */
		boolean i = false;
		while (proxyPort[inx] == 0) {
			try {
				proxyPortMutex.wait(SPIN);
			} catch (InterruptedException e) {
				i = true;
			}
		}
		if (i) {
			Thread.currentThread().interrupt();
		}
	}

	return (inx);
}

}  /* Class End */