X-Git-Url: http://info.iut-bm.univ-fcomte.fr/pub/gitweb/hpcvm.git/blobdiff_plain/f74ee61df3fbc8cf8371347069a715270a3e8a72..7cb1767605817324506b7500a36cf650bebcd010:/src/and/hpcvm/Server.java

diff --git a/src/and/hpcvm/Server.java b/src/and/hpcvm/Server.java
index 2bc7ab8..e865eb2 100644
--- a/src/and/hpcvm/Server.java
+++ b/src/and/hpcvm/Server.java
@@ -81,6 +81,10 @@ public class Server extends UnicastRemoteObject implements ServicesServer
 	private String working_directory ;
 	private long save_interleave ;
 	private Semaphore semaSave ;
+	private OperatingClients startingClients ;
+	private OperatingClients deployingClients ;
+	private LimitThread limitThread ;
+	private int maxThread;
 	
 	
 	protected Server() throws RemoteException 
@@ -95,49 +99,52 @@ public class Server extends UnicastRemoteObject implements ServicesServer
 	{
 		if( _stub != null )
 		{
-			String ip = "" ;
-			try {
-				ip = _stub.getIPHost() ;
-			} catch (RemoteException e) {
-				e.printStackTrace() ;
-				return 1 ;
-			}
-			
-			boolean exists = false ;
-			int i ;
-			
-			for( i = 0 ; i < clients.size() ; i++ )
+			synchronized( clients )
 			{
-				if( ip.equals( clients.get( i ).getIP() ) )
+				String ip = "" ;
+				try {
+					ip = _stub.getIPHost() ;
+				} catch (RemoteException e) {
+					e.printStackTrace() ;
+					return 1 ;
+				}
+
+				boolean exists = false ;
+				int i ;
+
+				for( i = 0 ; i < clients.size() ; i++ )
 				{
-					exists = true ;
-					System.out.println( "Client already connected!" ) ;
-					break ;
+					if( ip.equals( clients.get( i ).getIP() ) )
+					{
+						exists = true ;
+						System.out.println( "Client already connected!" ) ;
+						break ;
+					}
 				}
-			}
-			
-			if( exists )
-			{
-				System.out.println( "The client stub will be replaced." ) ;
-				clients.get( i ).setStub( _stub ) ;
-				System.out.println( "(reconnection of " + clients.get( i ).getName() + ")" ) ;
-				return 2 ;
-			} else {
-				System.out.println( "New connection!" ) ;
-				clients.add( new ConnectedClient( _stub ) ) ;
-				System.out.println( "(connection of " + clients.get( clients.size() - 1 ).getName() + ")" ) ;
-				generateVmIP( ip ) ;
-				
-				if( clients.size() == 0 )
+
+				if( exists )
 				{
-					System.out.println( "There is no client connected." ) ;
-				} else if( clients.size() == 1 ) {
-					System.out.println( "There is one client connected." ) ;
+					System.out.println( "The client stub will be replaced." ) ;
+					clients.get( i ).setStub( _stub ) ;
+					System.out.println( "(reconnection of " + clients.get( i ).getName() + ")" ) ;
+					return 2 ;
 				} else {
-					System.out.println( "There are " + clients.size() + " clients connected." ) ;
-				}
+					System.out.println( "New connection!" ) ;
+					clients.add( new ConnectedClient( _stub ) ) ;
+					System.out.println( "(connection of " + clients.get( clients.size() - 1 ).getName() + ")" ) ;
+					generateVmIP( ip ) ;
 
-				return 0 ;
+					if( clients.size() == 0 )
+					{
+						System.out.println( "There is no client connected." ) ;
+					} else if( clients.size() == 1 ) {
+						System.out.println( "There is one client connected." ) ;
+					} else {
+						System.out.println( "There are " + clients.size() + " clients connected." ) ;
+					}
+
+					return 0 ;
+				}
 			}
 		}
 		
@@ -187,9 +194,19 @@ public class Server extends UnicastRemoteObject implements ServicesServer
 			for( int i = 0 ; i < clients.size() ; i++ )
 			{
 				if( _ip.equals( clients.get( i ).getIP() ) ) 
-				{
-					clients.get( i ).setStatus( _status ) ;
-					System.out.println( "Client " + clients.get( i ).getName() + " changed its status to: " + _status ) ;
+				{					
+					if( _status.equalsIgnoreCase( "stopped" ) )
+					{
+						System.out.println( "Client " + clients.get( i ).getName() + " is stopped." ) ;
+						clients.get( i ).setStatus( "connected" ) ;
+						ComputingClient cc = clients.get( i ).getComputingClient() ;
+						computingClients.remove( cc ) ;
+						clients.get( i ).setComputingClient( null ) ;
+					} else {
+						clients.get( i ).setStatus( _status ) ;
+						System.out.println( "Client " + clients.get( i ).getName() + " changed its status to: " + _status ) ;
+					}
+					
 					break ;
 				}
 			}
@@ -207,6 +224,11 @@ public class Server extends UnicastRemoteObject implements ServicesServer
 		applications = new ArrayList<RunningApplication>() ;
 		monitor = null ;
 		
+		startingClients = new OperatingClients() ;
+		deployingClients = new OperatingClients() ;
+		limitThread = new LimitThread() ;
+		maxThread = 20 ;
+		
 		working_directory = "/localhome/vmware" ;
 		
 		save_interleave  = 30 * 60 * 1000 ;
@@ -524,7 +546,10 @@ public class Server extends UnicastRemoteObject implements ServicesServer
 							System.out.println( "Maybe it will come back later :)" ) ;
 						}
 						
-						it.remove() ;
+						synchronized( clients )
+						{
+							it.remove() ;
+						}
 						nb_disconnections++ ;
 						change = true ;
 					}
@@ -532,13 +557,16 @@ public class Server extends UnicastRemoteObject implements ServicesServer
 				
 				if( change )
 				{
-					if( clients.size() == 0 )
+					synchronized( clients )
 					{
-						System.out.println( "There is no client connected." ) ;
-					} else if( clients.size() == 1 ) {
-						System.out.println( "There is one client connected." ) ;
-					} else {
-						System.out.println( "There are " + clients.size() + " clients connected." ) ;
+						if( clients.size() == 0 )
+						{
+							System.out.println( "There is no client connected." ) ;
+						} else if( clients.size() == 1 ) {
+							System.out.println( "There is one client connected." ) ;
+						} else {
+							System.out.println( "There are " + clients.size() + " clients connected." ) ;
+						}
 					}
 				}
 				
@@ -663,7 +691,7 @@ public class Server extends UnicastRemoteObject implements ServicesServer
 				if( computingClients.get( i ).getClient().getIP().equalsIgnoreCase( _ip ) ) 
 				{
 					computingClients.get( i ).setLastSave( _saveName ) ;
-					System.out.println( "Save name successfully change for " + _ip ) ;
+					System.out.println( "Save name successfully changed for " + _ip ) ;
 					return 0 ;
 				}
 			}
@@ -679,35 +707,94 @@ public class Server extends UnicastRemoteObject implements ServicesServer
 		
 		if( nb > _nb )
 		{
-			ArrayList<ServicesClient> ac = new ArrayList<ServicesClient>() ;
-			ArrayList<ComputingClient> tmp = new ArrayList<ComputingClient>() ;
+			final ArrayList<ServicesClient> ac = new ArrayList<ServicesClient>() ;
+			final ArrayList<ComputingClient> tmp = new ArrayList<ComputingClient>() ;
 			
 			RunningApplication app = new RunningApplication( "Test" ) ;
 			
 			int i = 0 ;
+			boolean ok ;
 			
 			while( i < clients.size() && ac.size() < _nb )
 			{
-				if( clients.get(i).getStatus().equalsIgnoreCase( "connected" ) ) 
+				ok = false ;
+				if( clients.get( i ).getStatus().equalsIgnoreCase( "connected" ) ) 
 				{
-					int res = 1 ;
-					try {
-						res = clients.get( i ).getStub().startVM( 0 ) ;
-					} catch( RemoteException e ) {
-						e.printStackTrace();
+					synchronized( startingClients )
+					{
+						while( ac.size() + startingClients.getNb() >= _nb )
+						{
+							if( ac.size() == _nb ) break ;
+							
+							try {
+								startingClients.wait() ;
+							} catch (InterruptedException e) {
+								e.printStackTrace();
+							}
+						}
+						
+						if( ac.size() < _nb )
+						{
+							startingClients.inc() ;
+							ok = true ;
+						}
 					}
 					
-					if( res == 0 )
+					if( ok )
 					{
-						ac.add( clients.get( i ).getStub() ) ;
-						clients.get( i ).setStatus( "running" ) ;
-						ComputingClient cl = new ComputingClient( clients.get( i ) ) ;
-						clients.get( i ).setComputingClient( cl ) ;
-						computingClients.add( cl ) ;
-						tmp.add( cl ) ;
-					} else {
-						System.err.println( "Problem while launching the VM on " 
-								+ clients.get(i).getName() + "!" ) ;
+						synchronized( limitThread )
+						{
+							while( limitThread.getNb() >= maxThread )
+							{
+								try {
+									limitThread.wait() ;
+								} catch (InterruptedException e) {
+									e.printStackTrace();
+								}
+							} 
+							
+							limitThread.inc() ;
+						}
+						
+						final int ind = i ;
+						new Thread( new Runnable(){
+							
+							@Override
+							public void run() 
+							{
+								int res = 1 ;
+								try {
+									res = clients.get( ind ).getStub().startVM( 0 ) ;
+								} catch( RemoteException e ) {
+									e.printStackTrace();
+								}
+								
+								if( res == 0 )
+								{
+									ac.add( clients.get( ind ).getStub() ) ;
+									clients.get( ind ).setStatus( "running" ) ;
+									ComputingClient cl = new ComputingClient( clients.get( ind ) ) ;
+									clients.get( ind ).setComputingClient( cl ) ;
+									computingClients.add( cl ) ;
+									tmp.add( cl ) ;
+								} else {
+									System.err.println( "Problem while launching the VM on " 
+											+ clients.get(ind).getName() + "!" ) ;
+								}
+								
+								synchronized( limitThread )
+								{
+									limitThread.dec() ;
+									limitThread.notifyAll() ;
+								}
+								
+								synchronized( startingClients )
+								{
+									startingClients.dec() ;
+									startingClients.notifyAll() ;
+								}
+							}
+						}).start() ;
 					}
 				}
 				
@@ -973,10 +1060,9 @@ public class Server extends UnicastRemoteObject implements ServicesServer
 	}
 	
 	
-	public Integer deployVM( String _name, String _archive, String _directory )
+	public Integer deployVM( final String _name, final String _archive, final String _directory )
 	{
-		int pb = -1 ;
-		int nb = 0 ;
+		int nb = 0, pb = 0 ;
 		
 		if( _name != null && _name.length() > 0 && _archive != null && _name.length() > 0 
 				&& _directory != null && _directory.length() > 0 )
@@ -998,112 +1084,156 @@ public class Server extends UnicastRemoteObject implements ServicesServer
 			file = null ;
 			
 			// TODO do a better deployment !!
-			int ret ;
-			boolean error ;
+//			int ret ;
+//			boolean error, ok, server ;
+//			ArrayList<ConnectedClient> deployed = new ArrayList<ConnectedClient>() ;
 			
-			pb = 0 ;
+			boolean server = true ;
+			boolean ok ;
 			
 			for( int i = 0 ; i < clients.size() ; i++ )
 			{
-				ret = 1 ;
-				error = false ;
+//				ret = 1 ;
+				nb++ ;
+//				error = false ;
+//				ok = false ;
 				if( clients.get( i ).getStatus().equalsIgnoreCase( "connected" ) )
-				{
-					try {
-						ret = clients.get( i ).getStub().deployVM( _name, _archive, _directory ) ;
-					} catch( RemoteException e ) {
-						System.err.println( "Unable to deploy the VM on " + clients.get( i ).getName() + "!" ) ;
-						e.printStackTrace() ;
+				{					
+					synchronized( limitThread )
+					{
+						while( limitThread.getNb() >= maxThread )
+						{
+							try {
+								limitThread.wait() ;
+							} catch (InterruptedException e) {
+								e.printStackTrace();
+							}
+						} 
+	
+						limitThread.inc() ;
 					}
 					
-					// The client does not have the archive, we have to send it.
-					if( ret == 2 )
-					{
-						System.out.print( "Sending VM archive to " + clients.get( i ).getName() + " ... " ) ;
+					final int ind = i ;
+					new Thread( new Runnable() {
+
+						@Override
+						public void run() {
+							int ret = -1 ;
+							boolean error = true  ;
+							
+							try {
+								ret = clients.get( ind ).getStub().deployVM( _name, _archive, _directory ) ;
+							} catch( RemoteException e ) {
+								System.err.println( "Unable to deploy the VM on " + clients.get( ind ).getName() + "!" ) ;
+								e.printStackTrace() ;
+							}
 					
-						String wd = "" ;
-						String snIP = "" ;
-						error = false ;
+							// The client does not have the archive, we have to send it.
+							if( ret == 2 )
+							{
+								// Attention au multi-envois !!!
+								System.out.print( "Sending VM archive to " + clients.get( ind ).getName() + " ... " ) ;
 					
-						try {
-							wd = clients.get( i ).getStub().getWorkingDirectory() ;
-							snIP = clients.get( i ).getStub().getIPHost() ;
-						} catch (RemoteException e2) {
-							System.err.println( "Unable to retrieve information on " + clients.get( i ).getName() + "!" ) ;
-							e2.printStackTrace() ;
-							error = true ;
-							pb++ ;
-						}
+								String wd = "" ;
+								String snIP = "" ;
+								error = false ;
 					
-						String[] command = new String[]{ "/usr/bin/scp", working_directory + "/" + _archive,
-								snIP + ":" + wd } ;
+								try {
+									wd = clients.get( ind ).getStub().getWorkingDirectory() ;
+									snIP = clients.get( ind ).getStub().getIPHost() ;
+								} catch (RemoteException e2) {
+									System.err.println( "Unable to retrieve information on " + clients.get( ind ).getName() + "!" ) ;
+									e2.printStackTrace() ;
+									error = true ;
+								}
+					
+								String[] command = new String[]{ "/usr/bin/scp", working_directory + "/" + _archive,
+										snIP + ":" + wd } ;
 				
-						if( ! error )
-						{
-							try {
-								Process proc = Runtime.getRuntime().exec( command ) ;
-								proc.waitFor() ;
-			
-								if( proc.exitValue() == 0 )
+								if( ! error )
 								{
-									System.out.println( "Initial VM archive successfully sent." ) ;
-								} else {
-									System.err.println( "Initial VM archive not sent!" ) ;
-									System.err.println( "Error: " + proc.exitValue() ) ;
-									BufferedReader b = new BufferedReader( new InputStreamReader( proc.getErrorStream() ) ) ;
-							
-									String l ;
 									try {
-										while( (l = b.readLine()) != null )
+										Process proc = Runtime.getRuntime().exec( command ) ;
+										proc.waitFor() ;
+			
+										if( proc.exitValue() == 0 )
 										{
-											System.err.println( l ) ;
+											System.out.println( "Initial VM archive successfully sent." ) ;
+										} else {
+											System.err.println( "Initial VM archive not sent!" ) ;
+											System.err.println( "Error: " + proc.exitValue() ) ;
+											BufferedReader b = new BufferedReader( new InputStreamReader( proc.getErrorStream() ) ) ;
+							
+											String l ;
+											try {
+												while( (l = b.readLine()) != null )
+												{
+													System.err.println( l ) ;
+												}
+											} catch( IOException e ) {
+												e.printStackTrace() ;
+											}
+							
+											error = true ;
 										}
 									} catch( IOException e ) {
+										System.err.println( "Error during initial VM archive send command: " ) ;
+										e.printStackTrace() ;
+										error = true ;
+									} catch( InterruptedException e ) {
 										e.printStackTrace() ;
+										error = true ;
 									}
-							
-									error = true ;
-									pb++ ;
 								}
-							} catch( IOException e ) {
-								System.err.println( "Error during initial VM archive send command: " ) ;
-								e.printStackTrace() ;
-								error = true ;
-								pb++ ;
-							} catch( InterruptedException e ) {
-								e.printStackTrace() ;
-								error = true ;
-								pb++ ;
-							}
-						}
-				
 				
-						if( error )
-						{
-							continue ;
-						}
+								
+								if( ! error )
+								{				
+									// Second try ...
+									ret = 1 ;
+									try {
+										ret = clients.get( ind ).getStub().deployVM( _name, _archive, _directory ) ;
+									} catch( RemoteException e ) {
+										System.err.println( "Unable to deploy the VM on " + clients.get( ind ).getName() + "!" ) ;
+										e.printStackTrace() ;
+									}
+								}
+							}
 				
-						// Second try ...
-						ret = 1 ;
-						try {
-							ret = clients.get( i ).getStub().deployVM( _name, _archive, _directory ) ;
-						} catch( RemoteException e ) {
-							System.err.println( "Unable to deploy the VM on " + clients.get( i ).getName() + "!" ) ;
-							e.printStackTrace() ;
-							pb++ ;
+							if( ret == 0 )
+							{
+								System.out.println( "Initial VM archive successfully deployed on " + clients.get( ind ).getName() + "." ) ;
+								
+								synchronized( deployingClients )
+								{
+									deployingClients.inc() ;
+								}
+							}
+							
+							synchronized( limitThread )
+							{
+								limitThread.dec() ;
+								limitThread.notifyAll() ;
+							}
 						}
-					}
-				
-					if( ret == 0 )
-					{
-						System.out.println( "Initial VM archive successfully deployed on " + clients.get( i ).getName() + "." ) ;
-						nb++ ;
-					}
+					}).start() ;		
+				}
+			}
+		}
+		
+		synchronized( limitThread )
+		{
+			while( limitThread.getNb() > 0  )
+			{
+				try {
+					limitThread.wait() ;
+				} catch( InterruptedException e ) {
+					e.printStackTrace() ;
 				}
 			}
 		}
 		
-		if( pb > 0 )
+		if( nb - deployingClients.getNb() > 0 )
 		{
 			if( pb == 1 )
 				System.err.println( "** " + pb + " machine is not deployed!" ) ;
@@ -1120,6 +1250,34 @@ public class Server extends UnicastRemoteObject implements ServicesServer
 		return working_directory ;
 	}
 	
+	
+	private class OperatingClients
+	{
+		private int nb ;
+		
+		OperatingClients() { nb = 0 ; }
+		
+		protected void inc() { nb++ ; }
+		
+		protected void dec() { nb-- ; }
+		
+		protected int getNb() { return nb ; }
+	}
+	
+	
+	private class LimitThread
+	{
+		private int nb ;
+		
+		LimitThread() { nb = 0 ; }
+		
+		protected void inc() { nb++ ; }
+		
+		protected void dec() { nb-- ; }
+		
+		protected int getNb() { return nb ; }
+	}
+	
 }
 
 /** La programmation est un art, respectons ceux qui la pratiquent !! **/