Socket-Thread: Designfrage

[tuxedo]

Hallo zusammen,

hab mal eine kleine Designfrage:

Ich hab eine Klasse (die als Thread realisiert ist) die eingehende Anfragen vom Socket liest und "beantwortet". Das geschieht in der "Endlosschleife" des Threads. Parallel dazu habe ich 2 Methoden die selbst Anfragen an den "Gegenüber" senden und eine Antwort erwarten. Das ganze passt jetzt nicht so einfach zusammen weil ich ja a) in der Endlosschleife vom Socket lese und b) in einer der beiden Methoden sende und danach ebenfalls lese.

Ich versuch mal die aktuelle Situation zu skizzieren:

class SocketThread extends Thread {

public void run() {
   while (!interrupted()){
      packetHandler.process(objectInputStream.readObject());
   }
}

public Object sendMethodA(Object someDataA){
   objectOutputStream.writeObject(someData);
   return objectInputStream.readObject();
}

public Object sendMethodB(Object someDataB1, Object someDataB2){
   objectOutputStream.writeObject(someDataB1);
   objectOutputStream.writeObject(someDataB2);
   return objectInputStream.readObject();
}

Man sieht schnell dass da was in die Hose geht. Ich kann ja nicht an zwei Stellen vom Stream lesen ohne dass ich Konsistentprobleme bekomme. Das ganze muss nun irgendwie asynchron laufen:

Empfangen geschieht ausschließlich an der einen Stelle (Schleife), senden an einer anderen (weitere Methoden).

Mein erster Ansatz wäre jetzt folgender (am Beispiel "sendMethodA()"):

public Object sendMethodA(Object someDataA){
   int anfrageId = this.generateAnfrageId();

   objectOutputStream.writeObject(someData);

   while (!receivedAnswers.containsKey(anfrageId){
      Thread.sleep(1);
   }

   return receivedAnswers.get(anfrageId);
}

Ich lass mich eine eindeutige ID für meine Anfragen generieren, sende meine Anfrage und warte dann, bis mir der PacketHandler (der, der ion der endlosschleife in der run() Methode läuft) die passende Antwort dazu in einer HashMap (oder sonstwas) zur Verfügung stellen kann. Dann nehme ich die Antwort und liefere sie als return-Wert. Das ganze ließe sich natürlich noch mit einem TimeOut belegen.

Prinzipiell müsste das funktionieren. Doch mich stört daran, dass ich ein Sleep() einbauen muss. Zudem weiß ich nicht wie effizient es ist eine HashMap auf diesem Wege nach dem "vorhandensein" der passenden Antwort zu fragen.
Am geschicktesten wäre es, wenn sich die Methode nach dem senden schlafen legen würde, und erst dann - getriggert vom PacketHandler - weiter macht wenn die Antwort da ist.

Hat da jemand nen Tipp für mich? Sollte halt möglichst effizient sein.

- Alex

 

[schalentier]

Also ueblicherweise wuerde man einen Thread nehmen (oder einen Mainloop und dort "pollen") und da das Empfangen realisieren. Dazu das Empfangene in einzelne Meldungsobjekte (Packets) zerlegen und diese in eine Warteschlange packen.

An einer anderen Stelle im Programm (zweiter Thread oder ebenfalls im Mainloop pollen) wird diese Warteschlange abgearbeitet, also immer das letzte Packet nehmen und verarbeiten.

Deine Methoden (sendMethodA und B) musst du dann auftrennen in eine zum Senden und eine zum Verarbeiten der Packets. D.h. das ganze wird damit asynchron. Die sendMethods senden nur und anschliessend laeuft dein Programm weiter. Reagiert wird erst, wenn das Antwortpacket eingetroffen ist.

Wenn du mit Threads arbeitest muss die Queue natuerlich synchronisiert werden (so dass nicht gleichzeitig in die Queue geschrieben und aus ihr gelesen wird).

 

[tuxedo]

Also ueblicherweise wuerde man einen Thread nehmen (oder einen Mainloop und dort "pollen") und da das Empfangen realisieren. Dazu das Empfangene in einzelne Meldungsobjekte (Packets) zerlegen und diese in eine Warteschlange packen.

Das hört sich nach dem was ich in meinem ersten Ansatz versucht habe zu formulieren: Einen Thread der in einer Schleife nix weiter macht wie Pakete empfangen und in einer Liste/Warteschlange zu packen.

An einer anderen Stelle im Programm (zweiter Thread oder ebenfalls im Mainloop pollen) wird diese Warteschlange abgearbeitet, also immer das letzte Packet nehmen und verarbeiten.

Deine Methoden (sendMethodA und B) musst du dann auftrennen in eine zum Senden und eine zum Verarbeiten der Packets. D.h. das ganze wird damit asynchron. Die sendMethods senden nur und anschliessend laeuft dein Programm weiter. Reagiert wird erst, wenn das Antwortpacket eingetroffen ist.

Naja, das ist ein kleines Problem. Ich will da eigentlich mit meinen Sende-Methoden (sind wirklich nur 2) das Frage-Antwort-Prinzip verfolgen. Vergleichbar mit der Anfrage eins Browsers an einen Webserver: Anfrage geht raus, Webserver antwortet, Seite wird angezeigt.
D.h. ich brauche am Ende der Methode einen passenden Return-Wert.

[/quote]
Wenn du mit Threads arbeitest muss die Queue natuerlich synchronisiert werden (so dass nicht gleichzeitig in die Queue geschrieben und aus ihr gelesen wird).[/quote]

Das ist klar.


Wenn ich die Sende-Methoden, die UNBEDINGT eine Antwort für ihren Returnwert brauchen in einen eigenen Thread auslagere:

Kann ich dann irgendwie den Thread nach dem Senden anhalten und von außen wieder wecken?
Weil dann könnte ich folgendes tun:

Der Sende-Thread registriert sich beim PacketHandler und teil dem mit: "Wenn du ein Paket empfängst das die AnfrageID XYZ hat: Bitte mich wieder aufwecken". Eben ähnlich einem Eventlistener. Nur dass ich die Ausführung der Methode durch einen Thread "schlafen lege" bis das Event eintritt.

Thread#suspend() und Thread#resume() sind ja leider deprecated .. Gibts da ein brauchbares "workaround"?

Wenn das geht, dann hab ich ja erreicht was ich wollte...Ich könnte das auch mit einem Sleep und einem polling der Warteschlange der empfangenen Pakete machen. Aber das wäre wohl weniger schnell/performant, oder? Mir gehts hier um jede Millisekunde.

- Alex

 

[schalentier]

Pfff... also 1 Thread pro Methodenaufruf, der einfach nur auf die Antwort wartet, wird wohl ziemlich langsam werden. Ich hab jetzt keine richtigen Thread-Benchmarks, auf die ich verweisen koennte, aber imo basieren die Java-Threads auf Betriebssystem-Prozessen. Und die brauchen ihre Zeit zum initialisieren usw.

Kann dabei evtl. RMI irgendwie helfen? Ich bin da nich ganz so firm... aber sind RMI's nicht auch synchronisiert? Google sollte helfen... (bin grad aufm Sprung ^^).

Das ganze schreit aber nach Erlang. Erlang kann genau das, was du willst und zwar sehr schnell und verteilt auf mehrere Cores/Prozessoren/Maschinen. Wenn du also mal was neues lernen willst... nimm Erlang dafuer ;-)

 

[tuxedo]

Ich glaube das hier ist genau das was ich brauche:

public class MyThreadTest {
	
	
	public class MyThread extends Thread {
		boolean sleepMode = false;

		// This method is called when the thread runs
		public void run() {

				// Do work
				System.out.println("Arbeit A");
				
				goToSleep();

				// Do work
				System.out.println("Arbeit B");

		}
		
		public void goToSleep(){
			System.out.println("goToSleep()");
			sleepMode = true;
			synchronized (this) {
				while (sleepMode) {
					try {
						System.out.println("wait");
						wait();
					} catch (Exception e) {
						e.printStackTrace();
					}
				}
			}
		}
		
		
	}

	public static void main(String[] args) throws InterruptedException {

		// Create and start the thread
		MyThread thread = new MyThreadTest().new MyThread();
		
		// Thread starten und Arbeit A beginnen
		thread.start();
		
		// ein bisschen warten bis Arbeit A erledigt ist
		Thread.sleep(5000);


		// Aufwecken und Arbeit B erledigen lassen
		System.out.println("wecke auf");	
		synchronized (thread) {
			thread.sleepMode=false;
			thread.notify();	
		}
		

			
			

	}

	

}

Damit kann ich prima den Thread schlafen legen und auch wieder aufwecken. Warum ich beim Aufwecken unbedingt das "synchronized { ... }" brauche weiß ich noch nicht... Aber funktionieren tuts schonmal so.

- Alex

 

[Niki]

Meiner Meinung nach musst du vor dem Senden ein Object erstellen (dies ist dein syncMonitor). dieses Object stellst du mit einer generierten ID in eine HashMap. Dann rufst du das Senden auf (im synchronized-block des Objects) und legst dich anschließend in den wait pool.
Der Empfänger-Thread bekommt dann die ID zurück, schaut nach ob es ein Objekt in der HashMap dafür gibt und holt sich dieses. Das Ergebnis des Empfängers wird wieder mit der ID als Schlüssel in eine Map gespeichert. Anschließend ruft der Empfänger auf dem Object die notify Methode auf und löscht das Objekt aus der Map raus:

public static final HashMap<Integer, Object> ID_MONITOR_MAP = new HashMap<Integer, Object>();

public static final HashMap<Integer, Object> RESULTS = new HashMap<Integer, Object>();

public Object send(Object o) throws Exception{
  Integer id = generiereID();
  Object monitor = new Object();
  ID_MONITOR_MAP.put(id, monitor);
  synchronized(monitor){
    objectOutputStream.writeObject(o);
    monitor.wait();
  }
  Object result = RESULTS.get(id);
  RESULTS.remove(id);
  return result;
}

//hier der Thread...
public void run(){
  while(true){
    Object o = objectInputStream.readObject();
   //irgendwie muss ja der Schlüssel zurück:
    Integer key = ...;
    RESULTS.put(key, o);
    Object monitor = ID_MONITOR_MAP.get(key);
    synchronized(monitor){
      monitor.notify();
    }
    ID_MONITOR_MAP.remove(key);
  }
}

Könnte klappen, oder?

 

[tuxedo]

Nachtrag:

Nein, nicht nur ein Thread ...

Ich hab ein Thread der pollt den Socket nach neuen Nachrichten. Und ich hab je einen Thread für das jeweilige Nachricht senden. Nur wartet der Methodenaufruf der die Nachricht absendet, bis der Empfang-Thread SEINE Nachricht empfangen hat. Und erst dann beendet sich die Sende-Methode im SendeThread und liefert das Ergebis der Anfrage zurück.

RMI? Nein, das ist nicht wirklich eine Lösung. Grund:

Ich versuche gerade einen RMI-Ersatz zu schreiben ;-) Einer der nur eine einzige Socketverbindung benutzt.

- Alex

 

[Niki]

Bei meiner Lösung kannst du soviele Thread wie du willst verwenden, die senden Methode wartet solange bis sie vom Empfänger benachrichtigt wird.

 

[ms]

@Alex
Den Ansatz mit wait() und notify() wollte ich auch schon vorschlagen.

ms

 

[tuxedo]

@Niki

Jupp, sowas bastle ich gerade. Nur ist dein Ansatz noch n bisschen "einfacher". Hab da gerade irgendwie zu komplex angefangen.

Danke soweit für die Ratschläge ...

- Alex

 

[tuxedo]

Danke Leute,

die Sache hat bestens geklappt. Hab nur noch was an einer zweiten Stelle entkoppeln müssen.

Mein RMI-Ersatz läuft schon prima. "Normales" Methodenaufrufen auf dem Server, sowie Callbacks funzen astrein.

ENDLICH was RMI-artiges was nur eine Socketverbindung benutzt *chacka*. Muss noch ein wenig feinarbeit machen. Dann schau ich mal unter welche Lizenz ich das ganze Stelle.

- Alex

 

[tuxedo]

Hallo nochmal ...

Hab ein kleines Problemchen. Habs so gelöst wie Niki vorgeschlagen hat.

Doch dauert das aufwecken, wenn ich client und server auf getrennten maschinen laufen lasse, um 300..400ms .....

Lasse ich Client und Server lokal lauf, so hab ich meistens 0ms aufweckzeit (also die zeit von schlafebn gehen, bis wieder aufwachen).

Der Client schickt ja eine Frage zum Server. Dann schläft er, bis er die Antwort empfangen hat und aufgeweckt wird. Der Server antwortet jedoch sowas von unmittelbar, dass es nicht am Server liegen kann. Die Nachricht ist also schon wieder da, nur dauert das Wecken zu lange ...

Any ideas?

- Alex

 

[Niki]

Was genau dauert so lange? Nur das aufwecken also MONITOR.notify() oder das deserialisieren des Objektes?

 

[lhein]

Also wenn ich den Thread hier so lese, dann drängt sich mir unmittelbar die Frage auf, warum Du nicht eine Lib wie xSocket verwendest, die Dir schon einen Non-Blocking NIO Server an die Hand gibt und einfach dort Deinen Handler integrierst. Damit hättest Du den Server mal recht einfach realisiert und kannst sicher sein, daß der optimal funktioniert.
Bliebe dann nur noch der Client.

Oder hab ich hier was übersehen?

lr

 

[tuxedo]

@LR

Naja, ich habs so modular aufgebaut, dass ich recht einfach auf eine andere socket-realisierung aufbauen kann.

@Niki

Hab den schuldigen.... Nicht das wecken dauert so lange, sondern das Netzwerk war "lahm". Bzw. nicht das Netzwerk selbst.

Ich hab gecheckt wie lange der Server rbaucht um die Anfrage entgegen zu nehmen, zu bearbeiten und die Antwort zu senden. Belief sich auf fast 0ms...

Nach ein wenig rumprobieren bin ich drauf gekommen: Die Standard TCP-Einstellungen ist schuld.

Wenn man den Nagle-Aglorithmus abschaltet und das ganze noch etwas feintuned, dann geht's rucki-zucki. Das Delay von 300...400ms ist nun komplett weg.

Hab einfach nach dem herstellen der Socketverbindung beim Client als auch beim Server folgendes eingebaut:

socket.setTcpNoDelay(true); // nagle-algo abschalten, siehe [url]http://de.wikipedia.org/wiki/Nagle-Algorithmus[/url]
			socket.setTrafficClass(0x10); // low delay
			socket.setPerformancePreferences(0, 2, 1); // prio: geringe latenz > bandbreite > verbindungzeit

Bin mir nur nicht ganz sicher ob das "sicher" ist oder ob ich mir damit nicht noch Probleme einfange. Bei meiner Diplomarbeit (da ging um Audiokonferenz mit Java) hab ich das genau so gemacht. Aber da isses wurscht ob ein einzelnes Paket "abraucht" oder nicht.

- Alex

p.s. flush() hatte ich schon überall drin ... Daran lags also nicht.