agentmobilityservice.java

java实现的P2P多agent中间件

/*****************************************************************
 JADE - Java Agent DEvelopment Framework is a framework to develop
 multi-agent systems in compliance with the FIPA specifications.
 Copyright (C) 2000 CSELT S.p.A.
 
 GNU Lesser General Public License
 
 This library is free software; you can redistribute it and/or
 modify it under the terms of the GNU Lesser General Public
 License as published by the Free Software Foundation,
 version 2.1 of the License.
 
 This library 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
 Lesser General Public License for more details.
 
 You should have received a copy of the GNU Lesser General Public
 License along with this library; if not, write to the
 Free Software Foundation, Inc., 59 Temple Place - Suite 330,
 Boston, MA  02111-1307, USA.
 *****************************************************************/

package jade.core.mobility;

//#MIDP_EXCLUDE_FILE

import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.io.ObjectStreamClass;
import java.io.File;
import java.io.FileInputStream;
import java.io.DataInputStream;
import java.io.InterruptedIOException;
import java.util.StringTokenizer;
import java.util.zip.*;

import jade.core.ServiceFinder;
import jade.core.HorizontalCommand;
import jade.core.VerticalCommand;
import jade.core.Command;
import jade.core.GenericCommand;
import jade.core.Service;
import jade.core.ServiceHelper;
import jade.core.BaseService;
import jade.core.ServiceException;
import jade.core.Sink;
import jade.core.Filter;
import jade.core.Node;
import jade.core.LifeCycle;

import jade.core.Profile;
import jade.core.Agent;
import jade.core.Agent.Interrupted;
import jade.core.AID;
import jade.core.CaseInsensitiveString;
import jade.core.ContainerID;
import jade.core.Location;
import jade.core.AgentContainer;
import jade.core.MainContainer;
import jade.core.AgentDescriptor;

import jade.core.ProfileException;
import jade.core.IMTPException;
import jade.core.NameClashException;
import jade.core.NotFoundException;

import jade.core.management.AgentManagementService;
import jade.core.management.AgentManagementSlice;
//#J2ME_EXCLUDE_BEGIN
import jade.core.management.CodeLocator;
//#J2ME_EXCLUDE_END

import jade.lang.acl.ACLMessage;

import jade.security.Credentials;
import jade.security.JADEPrincipal;
import jade.security.JADESecurityException;
import jade.security.CredentialsHelper;

import jade.util.leap.List;
import jade.util.leap.ArrayList;
import jade.util.leap.Map;
import jade.util.leap.HashMap;
import jade.util.Logger;

/**
 The JADE service to manage mobility-related agent life cycle: migration
 and clonation.
 
 @author Giovanni Rimassa - FRAMeTech s.r.l.
 @author Giovanni Caire - TILAB
 */
public class AgentMobilityService extends BaseService {
	public static final String NAME = AgentMobilitySlice.NAME;
	
	public static final int AP_TRANSIT = 7;
	public static final int AP_COPY = 8;
	public static final int AP_GONE = 9;
	
	
	private static final String[] OWNED_COMMANDS = new String[] {
		AgentMobilityHelper.REQUEST_MOVE,
		AgentMobilityHelper.REQUEST_CLONE,
		AgentMobilityHelper.INFORM_MOVED,
		AgentMobilityHelper.INFORM_CLONED
	};
	
	private static final int SIZE_JAR_BUFFER = 4096;
	
	static final boolean MIGRATION = false;
	static final boolean CLONING = true;
	
	static final boolean CREATE_AND_START = true;
	static final boolean CREATE_ONLY = false;
	
	static final boolean TRANSFER_ABORT = false;
	static final boolean TRANSFER_COMMIT = true;
	
	
	public void init(AgentContainer ac, Profile p) throws ProfileException {
		super.init(ac, p);
		
		myContainer = ac;
	}
	
	public String getName() {
		return AgentMobilitySlice.NAME;
	}
	
	public Class getHorizontalInterface() {
		return AgentMobilitySlice.class;
	}
	
	public Service.Slice getLocalSlice() {
		return localSlice;
	}
	
	public ServiceHelper getHelper(Agent a) {
		return new AgentMobilityHelperImpl();
	}
	
	//#J2ME_EXCLUDE_BEGIN
	public Filter getCommandFilter(boolean direction) {
		if (direction == Filter.OUTGOING) {
            return _outFilter;
		} else return null;
	}
	//#J2ME_EXCLUDE_END
	
	public Sink getCommandSink(boolean side) {
		if(side == Sink.COMMAND_SOURCE) {
			return senderSink;
		}
		else {
			return receiverSink;
		}
	}
	
	public String[] getOwnedCommands() {
		return OWNED_COMMANDS;
	}
	
	/**
	 * Retrieve the name of the container where the classes of a given agent can be found
	 */
	public String getClassSite(Agent a) {
		return (String) sites.get(a);
	}
	
	// This inner class handles the messaging commands on the command
	// issuer side, turning them into horizontal commands and
	// forwarding them to remote slices when necessary.
	private class CommandSourceSink implements Sink {
		
		public void consume(VerticalCommand cmd) {
			try {
				String name = cmd.getName();
				if(name.equals(AgentMobilityHelper.REQUEST_MOVE)) {
					handleRequestMove(cmd);
				}
				else if(name.equals(AgentMobilityHelper.REQUEST_CLONE)) {
					handleRequestClone(cmd);
				}
				else if(name.equals(AgentMobilityHelper.INFORM_MOVED)) {
					handleInformMoved(cmd);
				}
				else if(name.equals(AgentMobilityHelper.INFORM_CLONED)) {
					handleInformCloned(cmd);
				}
			}
			catch(IMTPException imtpe) {
				cmd.setReturnValue(imtpe);
			}
			catch(NotFoundException nfe) {
				cmd.setReturnValue(nfe);
			}
			catch(NameClashException nce) {
				cmd.setReturnValue(nce);
			}
			catch(JADESecurityException ae) {
				cmd.setReturnValue(ae);
			}
			catch(ServiceException se) {
				cmd.setReturnValue(new IMTPException("Service error", se));
			}
		}
		
		
		// Vertical command handler methods
		
		private void handleRequestMove(VerticalCommand cmd) throws IMTPException, ServiceException, NotFoundException {
			Object[] params = cmd.getParams();
			AID agentID = (AID)params[0];
			Location where = (Location)params[1];
			
			MainContainer impl = myContainer.getMain();
			if(impl != null) {
				ContainerID cid = impl.getContainerID(agentID);
				AgentMobilitySlice targetSlice = (AgentMobilitySlice)getSlice(cid.getName());
				try {
					targetSlice.moveAgent(agentID, where);
				}
				catch(IMTPException imtpe) {
					// Try to get a newer slice and repeat...
					targetSlice = (AgentMobilitySlice)getFreshSlice(cid.getName());
					targetSlice.moveAgent(agentID, where);
				}
			}
			else {
				// Do nothing for now, but could also route the command to the main slice, thus enabling e.g. AMS replication
			}
		}
		
		private void handleRequestClone(VerticalCommand cmd) throws IMTPException, ServiceException, NotFoundException {
			Object[] params = cmd.getParams();
			AID agentID = (AID)params[0];
			Location where = (Location)params[1];
			String newName = (String)params[2];
			
			MainContainer impl = myContainer.getMain();
			if(impl != null) {
				ContainerID cid = impl.getContainerID(agentID);
				AgentMobilitySlice targetSlice = (AgentMobilitySlice)getSlice(cid.getName());
				try {
					targetSlice.copyAgent(agentID, where, newName);
				}
				catch(IMTPException imtpe) {
					// Try to get a newer slice and repeat...
					targetSlice = (AgentMobilitySlice)getFreshSlice(cid.getName());
					targetSlice.copyAgent(agentID, where, newName);
				}
			}
			else {
				// Do nothing for now, but could also route the command to the main slice, thus enabling e.g. AMS replication
			}
		}
		
		private void handleInformMoved(VerticalCommand cmd) throws IMTPException, ServiceException, JADESecurityException, NotFoundException {
			Object[] params = cmd.getParams();
			AID agentID = (AID)params[0];
			Location where = (Location)params[1];
			
			if(myLogger.isLoggable(Logger.CONFIG))
				myLogger.log(Logger.CONFIG,"Moving agent " + agentID.getName() + " on container " + where.getName());
			
			Agent a = myContainer.acquireLocalAgent(agentID);
			if (a == null) {
				myLogger.log(Logger.SEVERE,"Internal error: handleMove() called with a wrong name (" + agentID.getName() + ") !!!");
				return;
			}
			
			int transferState = 0;
			List messages = new ArrayList();
			AgentMobilitySlice dest = null;
			try {
				// If the destination container is the same as this one, there is nothing to do
				if (CaseInsensitiveString.equalsIgnoreCase(where.getName(), myContainer.here().getName())) {
					return;
				}
				
				dest = (AgentMobilitySlice) getSlice(where.getName());
				if (dest == null) {
					myLogger.log(Logger.SEVERE,"Destination "+where.getName()+" does not exist or does not support mobility");
					return;
				}
				if(myLogger.isLoggable(Logger.FINE)) {
					myLogger.log(Logger.FINE,"Destination container for agent " + agentID + " found");
				}
				
				transferState = 1;
				
				// Serialize the agent
				ByteArrayOutputStream out = new ByteArrayOutputStream();
				ObjectOutputStream encoder = new ObjectOutputStream(out);
				encoder.writeObject(a);
				byte[] bytes = out.toByteArray();
				if(myLogger.isLoggable(Logger.FINE)) {
					myLogger.log(Logger.FINE,"Agent " + agentID.getName() + " correctly serialized");
				}
				
				// Gets the container where the agent classes can be retrieved (possibly the agent arrived in this container from another container)
				String classSiteName = (String)sites.get(a);			
				if (classSiteName == null) {
					// The agent was born on this container
					classSiteName = getLocalNode().getName();
				}
				
				// Create the agent on the destination container
				try {
					dest.createAgent(agentID, bytes, classSiteName, MIGRATION, CREATE_ONLY);
				}
				catch(IMTPException imtpe) {
					// Try to get a newer slice and repeat...
					dest = (AgentMobilitySlice)getFreshSlice(where.getName());
					dest.createAgent(agentID, bytes, classSiteName, MIGRATION, CREATE_ONLY);
				}
				
				transferState = 2;
				if(myLogger.isLoggable(Logger.FINE)) {
					myLogger.log(Logger.FINE,"Agent " + agentID.getName() + " correctly created on destination container");
				}
				
				AgentMobilitySlice mainSlice = (AgentMobilitySlice)getSlice(MAIN_SLICE);
				
				// Perform an atomic transaction for agent identity transfer
				// From now on, messages for the moving agent will be routed to the
				// destination container
				boolean transferResult = false;
				try {
					transferResult = mainSlice.transferIdentity(agentID, (ContainerID) myContainer.here(), (ContainerID) where);
				}
				catch(IMTPException imtpe) {
					// Try to get a newer slice and repeat...
					mainSlice = (AgentMobilitySlice)getFreshSlice(MAIN_SLICE);
					transferResult = mainSlice.transferIdentity(agentID, (ContainerID) myContainer.here(), (ContainerID) where);
				}
				
				transferState = 3;
				
				if (transferResult == TRANSFER_COMMIT) {
					if(myLogger.isLoggable(Logger.FINE)) {
						myLogger.log(Logger.FINE,"Identity of agent " + agentID.getName() + " correctly transferred");
					}
					
					// Send received messages to the destination container. Note that
					// there is no synchronization problem as the agent is locked in the LADT
					myContainer.fillListFromMessageQueue(messages, a);
					
					dest.handleTransferResult(agentID, transferResult, messages);
					
					try {
						// Cause the termination of the agent thread
						a.changeStateTo(new LifeCycle(AP_GONE) {
							public boolean alive() {
								return false;
							}
						});
						
						// Remove the gone agent from the LADT
						myContainer.removeLocalAgent(a.getAID());
					}
					catch (Exception e) {
						// Should never happen
						e.printStackTrace();
					}
					sites.remove(a);
					if(myLogger.isLoggable(Logger.FINE)) {
						myLogger.log(Logger.FINE,"Agent " + agentID.getName() + " correctly gone");
					}
				}
				else {
					myLogger.log(Logger.WARNING,"Error transferring identity of agent " + agentID.getName());
					
					a.restoreBufferedState();
					dest.handleTransferResult(agentID, transferResult, messages);
					myLogger.log(Logger.WARNING,"Migration of agent " + agentID.getName() + "aborted");
				}
			}
			//#DOTNET_EXCLUDE_BEGIN
			catch (IOException ioe) {
				// Error in agent serialization
				myLogger.log(Logger.SEVERE,"Error in agent serialization. Abort transfer. " + ioe);
			}
			catch (JADESecurityException ae) {
				// Permission to move not owned
				myLogger.log(Logger.SEVERE,"Permission to move not owned. Abort transfer. " + ae.getMessage());
			}
			catch(NotFoundException nfe) {
				if(transferState == 0) {
					myLogger.log(Logger.SEVERE,"Destination container does not exist. Abort transfer. " + nfe.getMessage());
				}
				else if(transferState == 2) {
					myLogger.log(Logger.SEVERE,"Transferring agent does not seem to be part of the platform. Abort transfer. " + nfe.getMessage());
				}
				else if(transferState == 3) {
					// PANIC !!!
					myLogger.log(Logger.SEVERE,"Transferred agent not found on destination container. Can't roll back. " + nfe.getMessage());
				}
			}
			catch(NameClashException nce) {
				// This should not happen, because the agent is not changing its name but just its location...