/* * @(#)DBAglet.java * * A program send a bunch of aglets around some hosts to follow an Itineray. * Mutual Exclusion on the hosts is enfored by the co-ordinator. * The co-ordinator is the first aglet created. * * @author Daniel Ballinger * @version 1.0 $Date: 29/9/02 */ import java.net.URL; import java.util.Vector; import java.util.HashMap; import java.util.Iterator; import java.io.Serializable; import com.ibm.aglet.*; import com.ibm.aglet.event.*; //import com.ibm.agletx.util.SimpleItinerary; public class DBAglet extends Aglet implements MobilityListener { /* *********************** Co-ordinator attributes ************************* */ /** Port number */ private static final int port = 41839; /** * A list of host URLs as Strings, with no duplicates. * Usually the Co-ordinator will be created on the first host in the Itineray. * Subsequent aglets are dispatched to hosts in order. */ private static final String[] hosts = new String[]{ //"atp://majoribanks.mcs.vuw.ac.nz:"+port+"/", "atp://st-james.mcs.vuw.ac.nz:"+port+"/", //Sun/Solaris "atp://kore.mcs.vuw.ac.nz:"+port+"/", //Sun/Solaris "atp://depot.mcs.vuw.ac.nz:"+port+"/", //Sun/Solaris "atp://tahi.mcs.vuw.ac.nz:"+port+"/", //Sun/Solaris "atp://debretts.mcs.vuw.ac.nz:"+port+"/" //Sun/Solaris }; /** * Number of children to spawn, must be less than number of contexts. * Number exclues the co-ordinator that may also be mobile. */ public static final int children = 3; /** * Does the co-ordinator move with the other aglets. * If true, co-ordinator will hold locks of its own. * Otherwise it won't hold any locks. */ public static final boolean co_ordMoves = true; /** Indicates that the co-ordinator has finished. */ public static boolean co_ordDone; /** How many hosts to visit. */ public static final int itinSize = 10; /** Needs to store information for each known context. * [KEY]HostURL (as String) -> [VALUE]ContextStore. */ private HashMap contextControl; /** HashMap containing AgletProxies to all children and one to the coordinator. * [KEY]Aglet Number (int) -> [VALUE]AgletProxy. */ private HashMap proxies; /** HashMap containing the ContextStore all queued aglets are waiting from. * [KEY]Aglet Number (int) -> [VALUE]ContextStore. */ private HashMap waitingFor; /* *********************** Mobile aglet attributes ************************* */ /** Is this aglet the co-ordinator. Should only be one at a time. */ private boolean co_ordinator; /** * AgletProxy to the co-ordinator, may become invalid when co-ordinator moves. * In the event that it is invalid, the co-ordinator will send a new one when it arrives at its new location. */ private AgletProxy co_ordProxy; /** Unique identifer for this node. 0 for co-ordinator. */ private int num; /** List of places to visit. */ private Vector list; /** * The current position in the Itinerary. * Incremented when sending arrival notification to server. * May be incremented when requesting a dispatch (if at same location). */ private int position; /** How long to wait after requesting to move before requesting again. */ private int timeout; /** Will start following Itinerary when this is set true. */ private boolean active; /** print some debugging stuff */ public static final boolean loud = true; /** * The current aglet location. * When travelling to a new location this will be updated once the co-ordinator * is told of arrival at new location. */ private String currentHost; /** How long the aglet will display a message for on arrival. */ private int displayTime = 5000; /** * Called when the aglet is created. * @param ini An Object. * If the argument ini is null then a co-ordinator is being created. * Otherwise it is a child. */ public void onCreation(Object ini) { //Set important variables timeout = 500000; //Common attributes active = false; position = 0; addMobilityListener(this); //lisiten for things like onArrival... if(ini != null){ /* *********************** Child creation ************************* */ co_ordinator = false; //write("Creation args " + args.length,5000); Object[] args = (Object[])ini; //The arguments //Identifier number, co-ordinator proxy, itinerary list num = ((Integer)args[0]).intValue(); //Which Child number am I? co_ordProxy = (AgletProxy)args[1]; //Whos your daddy. list = (Vector)args[2]; // Itinerary //System.out.println("Itineray size expanded from"); //expand(list,itinSize); //increase the itineray size if needed. write("Created as child("+num+") with itinerary size = " + list.size(),5000); currentHost = null; //Will be setup after first dispatch. return; } /* *********************** Co-ordinator creation ************************* */ // I am the co-ordinator. I'll be running the show from now on. co_ordinator = true; co_ordDone = false; num = 0; write("Co-ordinator created.",3000); System.out.println("\n\n\n"); System.out.println("********************** CO-ORDINATOR SETUP **********************"); // Check the Number of hosts(n) verses the number of mobile aglets (including this co-ordinator if mobile). if((children+((co_ordMoves)?1:0))>=hosts.length){ System.err.println("WARNING: more nodes than hosts. ("+(children+((co_ordMoves)?1:0))+">="+hosts.length+")"); dispose(); } // Setup proxies proxies = new HashMap(); // This aglet needs to know about it's own proxy co_ordProxy = getProxy(); // First proxy is to the co-ordinator (which is also mobile). proxies.put(new Integer(num), co_ordProxy); // Setup waitingFor waitingFor = new HashMap(); // Add all known locations to the destination list // AND // Create a ContextStore record for each context list = new Vector(); contextControl = new HashMap(); for(int i = 0; i < hosts.length; i++){ list.addElement(hosts[i]); contextControl.put(hosts[i], new ContextStore(hosts[i], proxies, waitingFor, contextControl)); } System.out.println("---------------------- AGLET CREATION --------------------------"); AgletContext ac = getAgletContext();/** The current context for this aglet. Reset with each dispatch.*/ // Create the children for(int i = 1; i<=children;i++){ write("Creating child " + i, 1000); AgletProxy slave = null; try { shuffle(list); slave = ac.createAglet(getCodeBase(), "DBAglet", new Object[]{ new Integer(i), getProxy() ,list //Identifier number, co-ordinator proxy, itinerary list }); /* * It may be wise to scatter aglets on remote contexts on creation. * This will also init the currentHost for each aglet and setup the locks. */ //String aurl = ContextStore.findFreeContextStore(contextControl, slave); String dest = hosts[i]; ContextStore cs = (ContextStore)contextControl.get(dest); System.out.println("Have ContextStore to " + getHost(dest)); if(cs.getLock(new Integer(i))){ System.out.println("Dispatching "+i+" to " + getHost(dest)); slave = slave.dispatch(new URL(dest)); } else { System.out.println("Couldn't get lock for host " + getHost(dest)); } proxies.put(new Integer(i),slave); } catch (Exception ex) { handleException(ex); if (slave != null) { try { slave.dispose(); } catch (Exception exx) { exx.printStackTrace(); } } } } //If the co-ordinator is mobile if(co_ordMoves){ list = shuffle(list); System.out.println("Setting co-ordinator to move and locking current host resource."); currentHost = getAgletContext().getHostingURL().toString(); //Lock the context the co-ordinator is on. ContextStore cs = (ContextStore)contextControl.get(hosts[0]); if(cs.getLock(new Integer(0))){ System.out.println("The co-ordinator has locked its current context"); } else { System.out.println("Couldn't get lock for host " + getHost(hosts[0])); } } //Display the setup system state. System.out.println(ContextStore.dump(contextControl)); System.out.println("---------------------- FREE THE AGLETS -------------------------"); // Send children off into the World. // Go forth my pretties :) write("Allowing aglets to travel.",500); for(int i = (co_ordMoves)?0:1; i <= children; i++){ AgletProxy slave = (AgletProxy) proxies.get(new Integer(i)); write("Freeing aglet " + i); try { slave.sendOnewayMessage(new Message("activate")); } catch (Exception ex) { handleException(ex); } } System.out.println("Release phase done."); System.out.println("********************** CO-ORDINATOR SETUP DONE *****************"); } /** * Shuffles the instances in a Vector. * @return the shuffled vector, can make usage clearer. **/ private Vector shuffle(Vector v){ if(v.size()==0 || v.size()==1) return v; java.util.Random rand = new java.util.Random(); for (int i = 0; i < v.size(); i++){ Object o = v.remove(i); v.add(rand.nextInt(v.size()), o); } return v; } /** @param size the new size of the vector. */ private Vector expand(Vector v, int size){ if(v.size() >= size) return v; int initSize = v.size(); for(int i = initSize; i < size; i++){ v.add(v.get(i%initSize)); } return v; } public void run() {} /** Arrived at a new location. */ public void onArrival(MobilityEvent ev) { System.out.println("\n\n============================== Arrival("+num+") =========================================\n"); if(currentHost == null){ //This happens when the aglet is dispatched for the first time after creation. currentHost = getAgletContext().getHostingURL().toString(); //currentHost = currentHost.substring(0,currentHost.indexOf(":",6)); //Drop port number write("on Arrival from unknown location. Now at " + getHost(currentHost)); if(co_ordinator){ cwrite("ERROR, co_ordinator arrived with no current host record."); } //Do not request transport as not activated yet. return; } write("aglet for db arrived (wait "+displayTime+").",displayTime); String oldHost = currentHost; currentHost = getAgletContext().getHostingURL().toString(); write("Aglet has arrived at "+getHost(currentHost)+" from " + getHost(oldHost)); if(!meCheck() && co_ordinator){ //Mutual exclusion has failed. if(list == null || list.size()==0){ System.out.println("Note that the co-ordinator has an empty itinerary."); } System.out.println(ContextStore.dump(contextControl)); } //System.out.println("Arrival Status. Is coord:"co_ordinator +" done:"+ co_ordDone +" p:"+ position +" s:"+ list.size()); //Advance the position if we have arrived at the desired location. //This will not happen if the aglet has been moved as the result of deadlock. if(co_ordinator && (co_ordDone || position==list.size())) { write("Arrival of finished co-ordinator at " + getHost(currentHost)); } else if (position==list.size()){ write("Found aglet arrival with completed itinerary."); dispose(); } else if(currentHost.equals((String)list.get(position))){ position++; write("Have advanced to position("+position+") "+getHost(currentHost)+"."); } else { write("Have NOT advanced position("+position+"), must have sidestepped for deadlock reasons."); } if(co_ordinator) { //Inform aglets of new proxy co_ordProxy = getProxy(); cwrite("The co-ordinator has arrived at "+getHost(currentHost)+" from " +getHost(oldHost)+".",1000); sendCoproxy(); /* Release old co-ordinator lock. */ Integer id = new Integer(num); //Check the context to see if there is an aglet there. ContextStore cs = (ContextStore)contextControl.get(oldHost); if(cs == null){ System.err.println(">ARRIVED ERROR! Unable to find ContextStore for " + getHost(oldHost)); } else if(cs.returnLock(id)){ cwrite(">ARRIVED Lock returned from " + id + " for " + getHost(oldHost), 500); System.out.println(ContextStore.dump(contextControl)); //Free the next aglet if one is queued. if(cs.releaseQueued()){ cwrite(">ARRIVED Passed lock on to waiting aglet", 500); } } else { cwrite(">ARRIVED Problem returning lock from " + id + " for " + getHost(oldHost), 1000); } } //Move to next location requestTransport(); } /** @return true if ME is enforced on this context. */ private boolean meCheck(){ //Could perfrom mutual exclusion check here by asking the context for proxies to // all present aglets. int i = 0; for (java.util.Enumeration e = getAgletContext().getAgletProxies() ; e.hasMoreElements() ;) { //System.out.println(); e.nextElement(); i++; } if(i>1){ System.out.println("\n\n=============================== WARNING =========================================="); System.out.println("=WARNING, more that one aglet on this context("+ getHost(getAgletContext().getHostingURL().toString())+")." + i+"="); System.out.println("==================================================================================\n\n"); return false; } return true; } private void requestTransport(){ if (list == null){ write("No Itinerary present, waiting."); } else if (position==list.size()) { if(co_ordinator && co_ordDone)return; itineraryComplete(); } else if (active){ //write("On the move."); //if(co_ordinator) wait(5000); //Slow down the co_ordinator. sendMoveRequest(); } else { write("I don't know whats happening here."); } } public void onDispatching(MobilityEvent ev) { write("going to "+getHost(ev.getLocation()+"")+" from " + getHost(currentHost) ); //if(!((String) list.elementAt(position)).equals(ev.getLocation())) System.out.println(num + " must be under deadlock control."); if(currentHost!=null)System.out.println("\n============================== Dispatch("+num+") ========================================"); } private void itineraryComplete(){ write("Itinerary completed, I'm going now.", 5000); if(num == 0){ sendDispose(); //The coordinator should not dispose until all other aglets are done. } else { dispose(); } } public void onDisposing() { System.out.println("\n------------------------------ Disposing("+num+") ---------------------------------------"); write("disposing."); //have completed System.out.println("Position("+position+"/"+list.size()+") - Itineray completed:"); for(int i = 0; i < list.size(); i++){ System.out.println("i("+(i+1)+"): " + list.get(i)); } if(!co_ordinator){ //Mutual Exclusion is no longer enfored for the aglets dying location. sendDispose(); } else { //if(co_ordinator){ /* * This has the strange result that the co-ordinator has finished first, release it * locks but hangs around to service aglets that are still moving. */ //TODO //The co-ordinator must wait for all other aglets to complete before ending. //Could try passing responsibilities off to another aglet. while(true){ Vector alive = new Vector(proxies.values()); if(alive.size() == 0 && !co_ordMoves) { break; } else if(alive.size() > 0){ cwrite("WARNING: The co-ordinator must wait for "+alive.size()+" other aglets to finish."); wait(10000); Thread.currentThread().yield(); break;//Temp measure } else{ break; } } write("Co-ordinator is going."); } write("Done disposing."); System.out.println("============================== Disposing("+num+") DONE ================================="); } /** Inform the co-ordinator that this aglet no longer needs any locks.*/ private void sendDispose(){ try { //Send a message to release the lock for this location. Message message = new Message("finished"); message.setArg("id", new Integer(num)); message.setArg("loc", currentHost); System.out.println("DISPOSE Sending message to co-ordinator of my disposal at location " + getHost(currentHost)); //Short-circuit if the co-ordinator has moved. if(co_ordinator) receiveDispose(message); else co_ordProxy.sendOnewayMessage(message); } catch (Exception ex) {handleException(ex);} } private boolean receiveDispose(Message msg){ Integer id = (Integer)msg.getArg("id"); String loc = (String)msg.getArg("loc"); cwrite(">DISPOSE finished message for aglet id("+id+") from "+getHost(loc) , 500); if(loc==null){cwrite(">DISPOSE ERROR the given location is null for " + id);return true;} ContextStore cs = (ContextStore)contextControl.get(loc); if(cs==null){ cwrite(">DISPOSE ERROR ContextStore is null for " + id);return true;} if(id.intValue()!=0 || proxies.size()==0 ){ if(cs.returnLock(id)){ cwrite(">DISPOSE Lock returned from " + id + " for " + getHost(loc), 500); //Free the next aglet if one is queued. cs.releaseQueued(); } else { cwrite(">DISPOSE ERROR Problem returning lock from " + id + " for " + getHost(loc), 1000); } /* Remove the proxy from the hashmap */ proxies.remove(id); } else { cwrite(">DISPOSE Co-ordinator finished itinerary at " + getHost(loc), 500); co_ordDone = true; } if (co_ordDone && proxies.size()==1) { proxies.remove(new Integer(0)); //All aglets are done. } cwrite(">DISPOSE proxies.size()="+proxies.size() + " coordMoves=" + co_ordMoves); /* Drop the co_ordinator */ if(proxies.size() == 0){ cwrite(">DISPOSE Co-ordinator is no longer needed."+proxies.size()); dispose(); } if(proxies.size() == 1 && !co_ordMoves){ cwrite(">DISPOSE Co-ordinator(NM) is no longer needed."+proxies.size()); proxies.remove(new Integer(0)); dispose(); } else if (id.intValue()==0){ cwrite(">DISPOSE The co_ordinator is no-longer requesting moves as it has finished its intinery."+proxies.size()); co_ordDone = true; System.out.println("\n============================== CO-ORDINATOR DONE =========================================\n\n"); } return true; } public void onReverting(MobilityEvent ev) { throw new SecurityException(); } /** Inform the co-ordinator of this aglets desire to move to a new location. */ private void sendMoveRequest(){ //System.out.println("Sending move request."); //Continue along the Itinerary String nextStop = null; //Advance the itinerary if already at the correct host. while(position < list.size()){ nextStop = (String) list.elementAt(position); //System.out.println(" next:" + nextStop + " current:" + currentHost + " bool:" + !nextStop.equals(currentHost)); if(!nextStop.equals(currentHost)) {break;} System.out.println("already there."); position++; } if(position == list.size()){ write("no more locations to visit."); itineraryComplete(); } if(nextStop==null){System.err.println("WARNING: nextStop is null");} //Send a one way message to the co-ordinator requesting to move to nextStop. try { write("MOVE Sending request for move to " + getHost(nextStop)); Message message = new Message("move"); message.setArg("id", num); message.setArg("from", currentHost); //Not really needed, for debugging. message.setArg("dest", nextStop); //Short-circuit if the co-ordinator has moved. if(co_ordinator) receiveMoveRequest(message); else { if(co_ordProxy.isValid()){ write("Co-ordinator proxy is valid.",500); } else { write("Co-ordinator proxy is NOT valid, send anyway.",500); } co_ordProxy.sendOnewayMessage(message); write("request to move sent.",500); } } catch (InvalidAgletException e) {write("sendMoveRequest(): Could not message co_ord as " + e); } catch (Exception ex) {handleException(ex);} } /** The co-ordinator handles a request from an aglet who wants to move. */ private boolean receiveMoveRequest(Message msg){ /*synchronized*/ Integer id = (Integer)msg.getArg("id"); String dest = (String)msg.getArg("dest"); String from = (String)msg.getArg("from"); if(id.intValue() == 0) { cwrite(">MOVE WARNING: co-ordinator requesting moving."); } AgletProxy senderProxy = (AgletProxy) proxies.get(id); cwrite(">MOVE Request for aglet("+id+") from "+getHost(from)+" to " + getHost(dest), 1000); //Check the context to see if there is an aglet there. ContextStore dcs = (ContextStore)contextControl.get(dest); ContextStore fcs = (ContextStore)contextControl.get(from); Vector result = dcs.queueDispatch(id,fcs); if(result == null){ cwrite(">MOVE Dispatch "+id+" to "+getHost(dest)+" request queued."); } else if (result.size()==0){ cwrite(">MOVE Dispatched "+id+" to "+getHost(dest)+"."); //Experimental early lock release. if(fcs.releaseQueued()){ cwrite(">MOVE released further aglets after "+id); } } else { cwrite(">MOVE Deadlock detected for "+id+", will slide an aglet in list("+result+") to an alternate context."); shuffle(result); id = (Integer)result.get(0); //Special case when co-ordinator is blocking but not moving. if(result.size()==1 && id.intValue()==0){ cwrite("I'm blocking the path."); } else if(id.intValue()==0) id = (Integer)result.get(1); //Never move the co-cordinator to resolve standard deadlock try{ from = ((AgletProxy)proxies.get(id)).getAddress(); } catch(InvalidAgletException e) {handleException(e);} fcs = (ContextStore)contextControl.get(from);//where id is cwrite(">MOVE DEADLOCK Attempting to move " +id + " from " + getHost(from)); System.out.println(ContextStore.dump(contextControl)); //Find a free context. Vector hostsC = new Vector(contextControl.values()); shuffle(hostsC); //Try choosing a random host. Iterator itr = hostsC.iterator(); ContextStore sendtoContext = null; while(itr.hasNext()){ ContextStore cs = (ContextStore)itr.next(); //No point in sending to where they currently are. if(cs.getURL().equals(from)) continue; //Lockit if(cs.getLock(id)){ System.out.println("Have secured a new temp location for " + id + " " +getHost(cs.getURL())); sendtoContext = cs; break; } } if(sendtoContext == null){ System.err.println(warning("Couldn't find new context for " + id)); return false; } if(sendtoContext.deadlockDispatch(id)){ cwrite(">MOVE DEADLOCK corrected for " + id + " at " + getHost(from)); } //Deadlock corrected. } System.out.println(ContextStore.dump(contextControl)); return true; } /** Inform all proxies of the new co-ordinator proxy. */ private void sendCoproxy(){ Message message = new Message("coProxy", co_ordProxy); for(int i = 0; iCOPROXY Received an updated co-ordinator proxy at " +getHost(co_ordProxy.getAddress()), 500); } catch (InvalidAgletException e){write(">COPROXY Couldn't check co-ordinator location " + e);} if(!co_ordinator)requestTransport(); return true; } /** * An aglet is sending a message here. */ public boolean handleMessage(Message msg) { if (msg.sameKind("activate")) { write(">ACTIVATE I have been activated.", 500); active = true; //Start following Itinerary requestTransport(); //Move to next location } else if (msg.sameKind("dialog")) { //Used to correct messaging problems. write(">DIALOG The user is encouraging me("+num+") to move..."); if(co_ordinator) sendCoproxy(); requestTransport(); } else if (msg.sameKind("coProxy")) {return receiveCoproxy(msg);} else if (msg.sameKind("wakeup")) { write(">WAKEUP I got a kick up the bum!"); //Co-ordinator wants to know where I want to go. requestTransport(); } else { if(co_ordinator) { if (msg.sameKind("move")) {return receiveMoveRequest(msg);} /*else if (msg.sameKind("arrived")) {return receiveArrived(msg);}*/ else if (msg.sameKind("finished")) {return receiveDispose(msg);} else {write(">UNKNOWN Don't know how to handle message of kind:" + msg.getKind()); return false;} } else { write(">UNKNOWN Don't know how to handle message of kind:" + msg.getKind()); return false; } } return true; } /** Something has gone wrong for this aglet. */ private void handleException(Exception e){ if(e!=null)e.printStackTrace(); write(num+" CAUGHT EXCEPTION " + e); } private void wait(int time){ Thread t = Thread.currentThread(); System.out.println("Putting [" + t.getName() + " to sleep for "+time+" ms."); try{t.sleep(time);}catch (InterruptedException e){System.out.println("Thread("+t.getName()+") was rudely awakened.");} } /** ***** Pretty printing ***** */ public static String warning(String warn){ String r = "\n\n=============================== WARNING ==========================================\n" ; r += "= " + warn; r += "\n==================================================================================\n\n"; return r; } public static String getHost(String url){ if(url==null) return null; return url.substring(6,url.indexOf(".",6)); } private void cwrite(String msg){write("=C= "+msg);} private void cwrite(String msg, int wait){write("=C= "+msg,wait);} private void write(String msg){write(msg,-1);} private void write(String msg, int wait){ System.out.println("t[" + Thread.currentThread().getName()+ " " + num +" "+ msg); setText(num +" "+ msg); if(wait>0)waitMessage(wait); } } /** * A container used by the co-ordinator to store information about whats happening at each host.
* For each Context there is: * */ class ContextStore implements Serializable { /** The URL for that context. */ private String url; /** Proxies to those aglets waiting to come here. */ private Vector waiting; /** * The lock for this host. * May only be held by one aglet at a time. */ private boolean locked; /** The ID of the aglet that currently has the lock */ private Integer lockWith; /** Vector containing AgletProxies to all mobile aglets and one to the coordinator. * [KEY]Aglet Number (int) -> [VALUE]AgletProxy. */ private HashMap proxies; /** Vector containing the destination ContextStore for all queued aglets that are waiting for dispatch. * [KEY]Aglet Number (int) -> [VALUE]ContextStore. */ private HashMap waitingFor; /** Needs to store information for each known context. * [KEY]HostURL (as String) -> [VALUE]ContextStore. */ private HashMap contexts; /** Constructor for a ContextStore */ public ContextStore(String url, HashMap proxies, HashMap waitingFor, HashMap contexts){ this.url = url; this.proxies = proxies; this.waiting = new Vector(); this.waitingFor = waitingFor; this.locked = false; this.contexts = contexts; lockWith = null; } public String toString(){ String result = "CS(" + DBAglet.getHost(url) + ") Status:" + ((locked)?"locked":"unlocked") +" "+ ((lockWith!=null)?" by "+lockWith+" ":""); result += "Queue: "+((waiting.size()==0)?"empty":"|"+waiting.size()+"|"+" "+waiting+" "); return result; } public String getURL(){return url;} /** * Returns true if the lock is now allocated to the requesting aglet. * Otherwise return false; * @param the id of the requesting aglet. * @return if the lock has been allocated to the requesting aglet. */ public synchronized boolean getLock(Integer id){ System.out.println("ContextStore.getLock("+id+") "+ this); if(locked && lockWith.intValue()==id.intValue()){ //Odd situation where locked and trying to request again. Already holds this lock. System.err.println(DBAglet.warning("WARNING: aglet "+id+" has made a second request for a lock it already owns.")); return false; //False for attempting to find a free context reasons. } if(locked) return false; locked = true; lockWith = id; return true; } /** * Returns the lock. * Will return false if the lock is not out. */ public boolean returnLock(Integer id){ if(!locked) { System.err.println(DBAglet.warning("WARNING: lock returned when unlocked!")); return false; } if (id.intValue() != lockWith.intValue()){ System.err.println(DBAglet.warning("WARNING: lock returned by "+id+" but held by " + lockWith +" for " + DBAglet.getHost(url))); return false; } locked = false; lockWith = null; return true; } /** * Enqueue the request then attempt to dispatch the aglet. * @param id The aglet waiting to come to this context. * @param cs The current ContextStore for the aglet (where it will be leaving from). * @return null if the request is already queued. * null entered queue. * empty vector if dispatched. * list if deadlock detected. */ public Vector queueDispatch(Integer id, ContextStore cs){ System.out.println("Attempting to dispatch " + id + " from " + DBAglet.getHost(cs.getURL()) + " to " + DBAglet.getHost(url)); if((cs.lockWith) != null && (cs.lockWith).intValue() != id.intValue()){ System.out.println("ContextStore.dispatch(): Attempted to dispatch "+id+" from a context it was not at!"); return new Vector(); //pretend it was dispatched. } if(!queue(id)){ //Already queued to come here return null; } if(releaseQueued()){ //Send off the head of the queue //Dispatch successful. return new Vector(); } //Perfrom deadlock check. Vector list = new Vector(); if(deadlockCheck(id, list)){ System.out.println("Found deadlock for ("+id+") with the following list " + list); return list; //The deadlock list } return null; } /** In the event that the lock can't be acquired for this context the agglet is queued.*/ public boolean queue(Integer id){ waitingFor.put(id,this);//Aglet id is waiting for this context. //Need to check if already queued. if(waiting.contains(id)) { System.out.println("Aglet("+id+") already queued for " + DBAglet.getHost(url) + "|"+waiting.size()+"|"+" "+waiting+" "); //Could return false here if a repreated request indicates deadlock due to resending of timeout message. return true; } waiting.add(id); return true; } /** * Attempt to dispatch the first waiting aglet to this context. * @return true if the head aglet was dispatched. */ public boolean releaseQueued(){ //Try dispatch next waiting proxy. System.out.println("ContextStore.releaseQueued()" + this); if(waiting.size() > 0){ Integer id = (Integer)waiting.remove(0); //Fail if the aglet doesn't already have the lock and can't get it. if(!getLock(id)){ System.out.println("ContextStore.releaseQueued(): Could not get lock to release queue head("+id+")."); waiting.add(0,id); //re-enter the queue at head return false; } waitingFor.remove(id); //No longer waiting for a context. AgletProxy p = (AgletProxy)proxies.get(id); try{ String waitingFrom = p.getAddress(); //Where the aglet currently is. System.out.println("ContextStore.releaseQueued(): Sending first queued aglet("+id+") to " + DBAglet.getHost(url) + " from " + DBAglet.getHost(waitingFrom)); p = p.dispatch(new URL(url)); //When the aglet is dispatched the proxy must be updated. proxies.put(id,p);//Update the stored proxy. VERY IMPORTANT! //At this point the aglet will have arrived at its new location. //Release the previous lock held at this point and attempt to dispatch waiting aglets. ContextStore cs = (ContextStore)contexts.get(waitingFrom); cs.returnLock(id); if(cs.releaseQueued()){ System.out.println("sub release complete."); //Thread t = Thread.currentThread(); //try{t.sleep(1000);}catch (InterruptedException e){System.out.println("Thread("+t.getName()+") was rudely awakened.");} } //System.out.println(" complete."); return true; } catch(Exception e){System.err.println("releaseQueued(): ContextStore Exception: " + e);} } return false; //Nothing to dispatch } /** The aglet has the lock for this context and is coming here. * This code is very similar to that of releaseQueued - could refactor. */ public boolean deadlockDispatch(Integer id){ AgletProxy senderProxy = (AgletProxy)proxies.get(id); //Dispatch the aglet there. try{ String waitingFrom = senderProxy.getAddress(); //Where the aglet currently is ContextStore cs = (ContextStore)contexts.get(waitingFrom); //Dequeue this aglet from the waiting queues. waitingFor.remove(id); //(cs.waiting).remove(id); System.out.println("ContextStore.deadlockDispatch(): Sending "+id+" to " + DBAglet.getHost(url) + " from " + DBAglet.getHost(waitingFrom)); senderProxy = senderProxy.dispatch(new URL(url)); proxies.put(id,senderProxy);//Update the stored proxy. VERY IMPORTANT! //At this point the aglet will have arrived at its new location. //Release the lock held at this point. cs.returnLock(id); if(cs.releaseQueued()){ System.out.println("Deadlock - sub release complete."); //Thread t = Thread.currentThread(); //try{t.sleep(1000);}catch (InterruptedException e){System.out.println("Thread("+t.getName()+") was rudely awakened.");} } return true; } catch(Exception e){System.err.println("deadlockDispatch() ContextStore Exception: " + e);} return false; } /** * For this context, create a list of aglets waiting to come here or depending on aglets coming here. * @param id The Integer id of the aglet waiting to come to this context. * @param list The aglet ids seen to date. * @return true if deadlock has occured. */ private boolean deadlockCheck(Integer id, Vector list){ if(DBAglet.loud)System.out.println("deadlockCheck(): dest "+DBAglet.getHost(url)+"("+id+") with seen list " + list); /* * The id already appears in the list and is going to be appended again - Deadlock. */ if(list.contains(id)) return true; list.add(id); //The context is not occupied. if(lockWith == null) return false; //Special case where the co-ordinator has finished moving but is blocking the path. if(DBAglet.co_ordMoves && DBAglet.co_ordDone && lockWith.intValue() == 0){ System.out.println("Special case of blocking co-ordinator with no itinerary."); System.out.println(lockWith+" is blocking " + url); list.removeAllElements(); list.add(lockWith); return true; } // The contextStore that the aglet currently occupying this context is waiting to move to. ContextStore cs = (ContextStore)waitingFor.get(lockWith); if(cs == null) { if(DBAglet.loud)System.out.println("deadlockCheck(): Currently no recorded target context for " + lockWith); //This can occur when no move request has been received for the aglet at that context. //Send a message to that aglet requesting that they resend their dispatch request. if(lockWith.intValue() == 0) return false; //Except for the co-ordinator AgletProxy slowFellow = (AgletProxy)proxies.get(lockWith); try { System.out.println("WAKEUP deadlockCheck(): Sending message to " + lockWith + " in an attempt to find where they want to go."); slowFellow.sendOnewayMessage(new Message("wakeup")); } catch (InvalidAgletException e){ System.err.println("deadlockCheck(): Exception " + e); System.out.println("Assuming that it is dead."); //At this point I should remove any locks it holds and remove it from the proxy list cs.returnLock(lockWith); if(cs.releaseQueued()){ System.out.println("Have terminated " + lockWith + " at " + cs.getURL()); } proxies.remove(lockWith); } catch (Exception ex) { System.err.println("deadlockCheck(): Exception " + ex); } return false; } if(DBAglet.loud)System.out.println("deadlockCheck():"+ lockWith+" is waiting at " +DBAglet.getHost(url)+" for " + DBAglet.getHost(cs.url)); if(cs.deadlockCheck(lockWith,list)){return true;} if(DBAglet.loud)System.out.println("No deadlock found"); return false; } public static String dump(HashMap map){ try{ return dump(new Vector(map.values())); } catch(NullPointerException e){ System.err.println("ContextStore.dump(): minor exception " + e); return ("ContextStore.dump(): minor exception " + e); } } public static String dump(Vector v){ String result = "=============== Context File =================\n"; Iterator itr = v.iterator(); while(itr.hasNext()){ ContextStore cs = (ContextStore)itr.next(); result += "= " + cs+"\n"; } ContextStore cs2 = (ContextStore)v.get(0); itr = cs2.waitingFor.keySet().iterator(); while(itr.hasNext()){ Integer key = (Integer)itr.next(); ContextStore cs = (ContextStore)cs2.waitingFor.get(key); result += "= ("+key+") -> ("+DBAglet.getHost(cs.url)+")\n"; } result += "==============================================\n"; return result; } }