Threads Thread blockieren

[calmar]

Hi,

hab folgendes Problem: ich habe eine Engine, die regelmäßig isAlive versenden soll (hab ich mit TimerTask gemacht), dazu hätte ich gerne einen zweiten Thread der Eingaben wie zb. exit (also über die Konsole) verarbeiten kann.

Das versenden der isAlive Messages erfolgt leider erst nachdem ich irgendwas in die Kosole getippt und bestätigt habe...

Hier mal mein Code:

public class SendAlivePacket extends TimerTask{
		
		private DatagramSocket socket;
		
		@Override
		public void run() {
			// get a datagram socket
			//DatagramSocket socket = null; 
			try {
				socket = new DatagramSocket();
			} catch (SocketException e) {
				System.out.println("Es konnte kein DatagramSocket erstellt werden!");
				System.exit(1);
				//e.printStackTrace();
			}

			byte[] buf = new byte[128];

			//Wenn der boolean true ist, dann ist die Engine nicht suspended und soll isAlive pakete schicken!
			if (isActivated == true){
				String messageForScheduler = String.valueOf(tcpPort)+" "+String.valueOf(maxConsumption)+
				" "+String.valueOf(minConsumption)+" "+String.valueOf(load);
				//System.out.println(messageForScheduler); 
				buf = messageForScheduler.getBytes();
				InetAddress address = null;
				try {
					address = InetAddress.getByName(schedulerHost);
				} catch (UnknownHostException e) {
					// TODO Auto-generated catch block
					e.printStackTrace();
				}
				DatagramPacket packet = new DatagramPacket(buf, buf.length, address, schedulerUDPPort);


				try {
					socket.send(packet);
//					System.out.println("isAlive packet gesendet!"); 
					//socket.close(); 
				} catch (IOException e) {
					// TODO Auto-generated catch block
					e.printStackTrace();
				}
			}
		}
	}

meine main:

public static void main(String[] args) {
		// TODO Auto-generated method stub
		
		ServerSocket serverSocket = null;
		
		checkingAndSettingArguments(args);
		isActivated = true; 
		pool = Executors.newCachedThreadPool();
		
		 try {
	            serverSocket = new ServerSocket(tcpPort);
	            System.out.println("Connection to port: "+tcpPort); 
	        } catch (IOException e) {
	            System.err.println("Could not listen on port: "+tcpPort+".");
	            System.exit(-1);
	        }
		        
	        //Hier wird der Timer gestartet, der in regelmaessigen Abstaenden isAlive 
	        //Pakete an den Scheduler sendet. 
	        Timer timer1 = new Timer();
	        timer1.schedule(new GenericTaskEngine().new SendAlivePacket(), alivePeriod, alivePeriod);

	        //Thread fuer interactiveCommand
	        pool.execute(new GenericTaskEngineInteractiveCommandThread());

	        //Solange der interactive Commannd laeuft, koennen Client-Anfragen bearbietet werden. 
	        while(isActivated){
	        	try {
	        		pool.execute(new MultiEngineThread(serverSocket.accept()));
	        	} catch (IOException e) {
	        		// TODO Auto-generated catch block
	        		e.printStackTrace();
	        	}
	        }

und zu guter letzt der thread für die user-eingaben

public class GenericTaskEngineInteractiveCommandThread 
extends Thread{

	private  BufferedReader in = new BufferedReader(
			new InputStreamReader(System.in));
	
	private boolean listening = true; 
	
	/**
	 * @return the listening
	 */
	public boolean isListening() {
		return listening;
	}


	@Override
	public void run() {
		
		while(listening == true){		
			System.out.println("Engine ready for interacitve commands!"); 
			String userEntry; 

			try {
				while((userEntry = in.readLine())!= null){
					System.out.println("User entered: "+userEntry); 
					this.processInteractiveCommand(userEntry);
					userEntry = null; 
				}
			} catch (IOException e) {
				System.out.println("Could not read interactive command!"); 
				listening = false; 
				e.printStackTrace();
			}
		}

		try {
			in.close();
			System.out.println("Interactive command thread closed!"); 
		} catch (IOException e) {
			System.out.println("Could not close interactive command thread!"); 
			e.printStackTrace();
		}	
	}
	
	/**
	 * Diese Methode verarbeitet die interaktiven Usereingaben. 
	 * Moeglich sind !engines, !companies, !exit. 
	 * @param userEntry
	 */
	private void processInteractiveCommand(String userEntry){
		System.out.println("Your entry: "+userEntry); 
		if(userEntry.toUpperCase().trim().equals("!EXIT")){
			listening = false; 
		}else if(userEntry.toUpperCase().trim().equals("!LOAD")){
			System.out.println("Current load: "+GenericTaskEngine.getLoad()+"%"); 
		}else{
			System.out.println("Unknown command!"); 
		}
	}
}

Sehe meinen Fehler leider selber nicht ;(.

Bitte um Eure Hilfe - Danke!!

 

[SlaterB]

woran liegt es denn exakt, was klappt alles, was geht noch nicht?
DatagramPaket usw. ist unnötig kompliziert, falls du nicht genau den Fehler dort vermutest, du schreibst ja eher dass irgendein 'Thread blockiert',
prüfe lieber mit System.out.println() wann was drankommt!

ungünstig ist auch so ein gestückelter Code mit unbekannten Klassen/ nicht definierten Variablen

hier zu einem Programm zusammengestellt, kopieren, Imports, läuft,
und dort erscheint alle 4 sec die Meldung, ob man nun was eintippt oder nicht, für mich kein Problem ersichtlich

public class Test {
	static boolean isActivated;

	public static void main(String[] args) throws Exception {
		ServerSocket serverSocket = null;

		// checkingAndSettingArguments(args);
		isActivated = true;
		ExecutorService pool = Executors.newCachedThreadPool();
		int tcpPort = 4567;
		try {

			serverSocket = new ServerSocket(tcpPort);
			System.out.println("Connection to port: " + tcpPort);
		} catch (IOException e) {
			System.err.println("Could not listen on port: " + tcpPort + ".");
			System.exit(-1);
		}

		// Hier wird der Timer gestartet, der in regelmaessigen Abstaenden
		// isAlive
		// Pakete an den Scheduler sendet.
		Timer timer1 = new Timer();
		timer1.schedule(new SendAlivePacket(), 4000, 4000);

		// Thread fuer interactiveCommand
		pool.execute(new GenericTaskEngineInteractiveCommandThread());

		// Solange der interactive Commannd laeuft, koennen Client-Anfragen
		// bearbietet werden.
		while (isActivated) {
			try {
				serverSocket.accept();
				System.out.println("accept");
			} catch (IOException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
		}
	}

	static class SendAlivePacket extends TimerTask {

		private DatagramSocket socket;

		@Override
		public void run() {
			// get a datagram socket
			// DatagramSocket socket = null;
			try {
				socket = new DatagramSocket();
			} catch (SocketException e) {
				System.out
						.println("Es konnte kein DatagramSocket erstellt werden!");
				System.exit(1);
				// e.printStackTrace();
			}

			byte[] buf = new byte[128];

			// Wenn der boolean true ist, dann ist die Engine nicht suspended
			// und
			// soll isAlive pakete schicken!
			if (isActivated == true) {
				String messageForScheduler = "test";
				// System.out.println(messageForScheduler);
				buf = messageForScheduler.getBytes();

				try {
					// socket.send(packet);
					System.out.println("isAlive packet gesendet!");
					// socket.close();
				} catch (Exception e) {
					// TODO Auto-generated catch block
					e.printStackTrace();
				}
			}
		}
	}

}

class GenericTaskEngineInteractiveCommandThread extends Thread {

	private BufferedReader in = new BufferedReader(new InputStreamReader(
			System.in));

	private boolean listening = true;

	/**
	 * @return the listening
	 */
	public boolean isListening() {
		return listening;
	}

	@Override
	public void run() {

		while (listening == true) {
			System.out.println("Engine ready for interacitve commands!");
			String userEntry;

			try {
				while ((userEntry = in.readLine()) != null) {
					System.out.println("User entered: " + userEntry);
					this.processInteractiveCommand(userEntry);
					userEntry = null;
				}
			} catch (IOException e) {
				System.out.println("Could not read interactive command!");
				listening = false;
				e.printStackTrace();
			}
		}

		try {
			in.close();
			System.out.println("Interactive command thread closed!");
		} catch (Exception e) {
			System.out.println("Could not close interactive command thread!");
			e.printStackTrace();
		}
	}

	/**
	 * Diese Methode verarbeitet die interaktiven Usereingaben. Moeglich sind
	 * !engines, !companies, !exit.
	 * 
	 * @param userEntry
	 */
	private void processInteractiveCommand(String userEntry) {
		System.out.println("Your entry: " + userEntry);
		if (userEntry.toUpperCase().trim().equals("!EXIT")) {
			listening = false;
		} else if (userEntry.toUpperCase().trim().equals("!LOAD")) {
			System.out.println("Current load: -42%");
		} else {
			System.out.println("Unknown command!");
		}
	}
}

 

[calmar]

hi,

danke fuer die antwort: ich kopiere mal die ganzen klassen rein...

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;


public class GenericTaskEngineInteractiveCommandThread 
extends Thread{

	private  BufferedReader in = new BufferedReader(
			new InputStreamReader(System.in));
	
	private boolean listening = true; 
	
	/**
	 * @return the listening
	 */
	public boolean isListening() {
		return listening;
	}


	@Override
	public void run() {
		
		while(listening == true){		
			System.out.println("Engine ready for interacitve commands!"); 
			String userEntry; 

			try {
				while((userEntry = in.readLine())!= null){
					System.out.println("User entered: "+userEntry); 
					this.processInteractiveCommand(userEntry);
					userEntry = null; 
				}
			} catch (IOException e) {
				System.out.println("Could not read interactive command!"); 
				listening = false; 
				e.printStackTrace();
			}
		}

		try {
			in.close();
			System.out.println("Interactive command thread closed!"); 
		} catch (IOException e) {
			System.out.println("Could not close interactive command thread!"); 
			e.printStackTrace();
		}	
	}
	
	/**
	 * Diese Methode verarbeitet die interaktiven Usereingaben. 
	 * Moeglich sind !load, !exit. 
	 * @param userEntry
	 */
	private void processInteractiveCommand(String userEntry){
		System.out.println("Your entry: "+userEntry); 
		if(userEntry.toUpperCase().trim().equals("!EXIT")){
			listening = false; 
		}else if(userEntry.toUpperCase().trim().equals("!LOAD")){
			System.out.println("Current load: "+GenericTaskEngine.getLoad()+"%"); 
		}else{
			System.out.println("Unknown command!"); 
		}
	}
}

import java.io.IOException;
import java.net.DatagramPacket;
import java.net.DatagramSocket;
import java.net.InetAddress;
import java.net.ServerSocket;
import java.net.SocketException;
import java.net.UnknownHostException;
import java.util.Timer;
import java.util.TimerTask;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class GenericTaskEngine {

	/**
	 * @param args
	 */
	
	private static int load = 0; 
	
	/**
	 * @return the load
	 */
	public static int getLoad() {
		return load;
	}

	/**
	 * @param load the load to set
	 */
	public static void setLoad(int load) {
		GenericTaskEngine.load = load;
	}

	private static int tcpPort; 
	private static String schedulerHost;
	private static int schedulerUDPPort; 
	private static int alivePeriod; 
	private static int minConsumption; 
	private static int maxConsumption; 
	private static String taskDir; 
	private static boolean isActivated; 
	private static ExecutorService pool;

	private static DatagramSocket socket;
	
	public class SendAlivePacket extends TimerTask{
		
		private DatagramSocket socket;
		
		@Override
		public void run() {
			// get a datagram socket
			//DatagramSocket socket = null; 
			try {
				socket = new DatagramSocket();
			} catch (SocketException e) {
				System.out.println("Es konnte kein DatagramSocket erstellt werden!");
				System.exit(1);
				//e.printStackTrace();
			}

			byte[] buf = new byte[128];

			//Wenn der boolean true ist, dann ist die Engine nicht suspended und soll isAlive pakete schicken!
			if (isActivated == true){
				String messageForScheduler = String.valueOf(tcpPort)+" "+String.valueOf(maxConsumption)+
				" "+String.valueOf(minConsumption)+" "+String.valueOf(load);
				//System.out.println(messageForScheduler); 
				buf = messageForScheduler.getBytes();
				InetAddress address = null;
				try {
					address = InetAddress.getByName(schedulerHost);
				} catch (UnknownHostException e) {
					// TODO Auto-generated catch block
					e.printStackTrace();
				}
				DatagramPacket packet = new DatagramPacket(buf, buf.length, address, schedulerUDPPort);


				try {
					socket.send(packet);
					System.out.println("isAlive packet gesendet!"); 
					//socket.close(); 
				} catch (IOException e) {
					// TODO Auto-generated catch block
					e.printStackTrace();
				}
			}
		}
	}

	/**
	 * @param args
	 */
	public static void main(String[] args) {
		// TODO Auto-generated method stub
		
		ServerSocket serverSocket = null;
		
		checkingAndSettingArguments(args);
		isActivated = true; 
		pool = Executors.newCachedThreadPool();
		
		 try {
	            serverSocket = new ServerSocket(tcpPort);
	            System.out.println("Connection to port: "+tcpPort); 
	        } catch (IOException e) {
	            System.err.println("Could not listen on port: "+tcpPort+".");
	            System.exit(-1);
	        }
		        
	        //Hier wird der Timer gestartet, der in regelmaessigen Abstaenden isAlive 
	        //Pakete an den Scheduler sendet. 
	        Timer timer1 = new Timer();
	        timer1.schedule(new GenericTaskEngine().new SendAlivePacket(), alivePeriod, alivePeriod);

	        //Thread fuer interactiveCommand
	        pool.execute(new GenericTaskEngineInteractiveCommandThread());

	        //Solange der interactive Commannd laeuft, koennen Client-Anfragen bearbietet werden. 
	        while(isActivated){
	        	try {
	        		pool.execute(new MultiEngineThread(serverSocket.accept()));
	        	} catch (IOException e) {
	        		// TODO Auto-generated catch block
	        		e.printStackTrace();
	        	}
	        }

		
			System.out.println("Shut-down generic task engine!"); 
			pool.shutdown();
			timer1.cancel();
			try {
				serverSocket.close();
				System.exit(1);
			} catch (IOException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
		
        
//TODO: korrektes Exit
		
		
        
	}

	/**
	 * Diese Methode prueft die Argumente und setzt sie. 
	 * Wenn die Argumente fehlerhaft sind, bzw. fehlen wird eine Fehlermeldung
	 * ausgegeben. 
	 * @param args
	 */
	private static void checkingAndSettingArguments(String[] args){		

		//System.out.println("Entering checkingAndSettingArguments!");

		boolean correctData = true; 

		if(args[0] == null || args[1] == null
				||args[2] == null || args[3] == null
				||args[4] == null || args[5] == null
				|| args[6] == null){
			correctData = false; 

		}else{
			try{
				tcpPort = Integer.valueOf(args[0]);
				schedulerHost = args[1];
				schedulerUDPPort = Integer.valueOf(args[2]);
				alivePeriod = Integer.valueOf(args[3]);
				minConsumption = Integer.valueOf(args[4]);
				maxConsumption = Integer.valueOf(args[5]);
				taskDir = args[6];
				System.out.println("tcpPort, schedulerHost, schedulerUDPPort, alivePeriod, " +
						"minConsumption, maxConsumption, taskdir: "+tcpPort+" "+schedulerHost
						+" "+schedulerUDPPort+" "+alivePeriod+" "+minConsumption+" "+maxConsumption+
						" "+taskDir);
				System.out.println("The arguments are correct and complete!"); 
			}catch(NumberFormatException nf){
				correctData = false; 
			}
		}  

		if(correctData == false){
			System.out.println("The arguments are incorrect or missing!");
			System.exit(1);
		}
	}
}

es haengt, nach der Ausgabe "Engine ready for interactive commands!"

Wenn ich dann bestätige läuft alles wie erwartet - dh die isAlive messages werden gesendet und ich kann User-Eingaben eintippen.

Lg

 

[SlaterB]

auf mein Posting, auf meine Vorschläge bist du ja nicht gerade eingegangen..,
aus Eigeninteresse habe ich deinen neuen Code erneut kopiert,
die main-Parameter durfte ich mir mühsam selber aus dem Screenshot zusammensammeln

neue Erkenntnisse habe ich aber nicht, selbst wenn ich das Datagram-Senden drin lasse, keine Ahnung ob das was macht,
kommen die isAlive-Meldungen

nach wie vor der Vorschlag an die, mehr System.out.println einzubauen, z.B. zu Beginn der run-Methode von SendAlivePacket
bzw. bis hin kurz vorm Senden,
wird das Senden blockiert oder kommt die run-Methode überhaupt nicht dran?
und nach Eingabe dann auf einmal doch?

mysteriös, für mich keine Fehlermöglichkeit sichtbar


edit:
eher ohne direkte Auswirkung bisher, aber für sich wahrscheinlich ein Fehler:

    private static DatagramSocket socket;
    
    public class SendAlivePacket extends TimerTask{
        
        private DatagramSocket socket;

 

[calmar]

so ich hab mir mein projekt jetzt auf einen anderen rechner gespielt (mit ubuntu statt windows) -

da läuft alles wie erwartet (dh. die isAlive messages werden mittels udp an einen server verschickt).

im windows hängt der thread genau an dieser stelle:

	address = InetAddress.getByName(schedulerHost);

(in der Klasse SendAlivePacket)

fehlermeldung erfolgt keine, nach einem enter in der Konsole läuft dann alles wie erwartet (im windows)

eigenartiges problem, aber ich ignorier es vorerst mal...

lg