Etwas Pipe-Ähnliches zur Prozesskommunikation

[schoppenhauer_entfernt_]

Grüß Gott.

Ich hätte eine Frage, und zwar...

Ich möchte einige Threads, die mehr oder weniger nichts direkt miteinander zu tun haben (und damit schwierig bis unmöglich zu vereinigen sind), miteinander kommunizieren lassen, und zwar am b esten über irgendetwas pipe-ähnliches, z.B. einen Input- und einen Output-Stream, die irgendwie aneinander gekoppelt sind.
Momentan geht ebendies per loopback. Ich würde es aber sehr gerne ohne Netzwerkbeteiligung machen.

Geht das irgendwie?

 

[Murray]

Sieh Dir mal PipedInputStream und PipedOutputStream an

 

[foobar]

Für soetwas gibt es http://java.sun.com/j2se/1.4.2/docs/api/java/io/PipedInputStream.html

 

[André Uhres]

Versuch's mal hiermit:
http://java.sun.com/j2se/1.4.2/docs/api/java/io/PipedInputStream.html
http://java.sun.com/j2se/1.5.0/docs/api/java/io/PipedInputStream.html

 

[schoppenhauer_entfernt_]

Danke.
Ich hoffe mal, dass das auch geht.
Aber eine Frage hab ich noch: Ist es ein Problem, wenn mehrere PipedInputStreams vom gleichen PipedOutputStream lesen?

 

[André Uhres]

Ein PipedOutputStream kann an einen PipedInputStream gebunden werden um eine Kommunikationspipe zu bilden. Der PipedOutputStream ist das sendende Ende der Pipe. Die Daten werden von einem Thread an das PipedOutputStream Objekt geschrieben und von einem anderen Thread aus dem angebundenen PipedInputStream ausgelesen.

 

[schoppenhauer_entfernt_]

Ok. Aber es könnte ja sein, dass ich zwei Threads habe, die einfach dasselbe bekommen sollen (bzw. es ist auch so, denn sonst hätte ich es nicht gefragt). Die sollen nix zurücksenden können, sondern einfach nur dasselbe empfangen. Kann man das?

 

[André Uhres]

Du könntest es vielleicht mit einem Verteilerthread versuchen, der die Daten ausliest und an weitere Pipes verteilt.

 

[Murray]

Wenn es nicht unbedingt Streams sein müssen, ist für so eine "one-to-many"-Kommunikation auch Observer/Observable eine gute Lösung. Ansonsten brauchst Du wohl eine verteilende Instanz (so eine Art DelegatingPipedOutputStream), die die hineingeschriebenden Bytes n-mal an die n InputStreams weiterleitet. Soweit ich weiss, gibt es da im JDK aber nichts vorgefertigtes.

 

[schoppenhauer_entfernt_]

Ok. Danke.

 

[schoppenhauer_entfernt_]

Hm. Ok. Ein Problem hab ich doch noch - also bzw. das hab ich momentan noch nicht, aber es könnte auftreten (ehrlich!).

Wenn ich das so mache, dann kann es sein, dass einer der beiden Threads, an den ich sende, nicht weiterliest, sondern sehr lange auf sich warten lässt. Dann wird der Puffer des einen OutputStreams meines Verteilungsthreads voll sein, und die send-anweisung wird den gesamten Thread blockieren, während der andere Thread ggf. wartet.

Also ich hab mir da zwei Lösungen ausgedacht:

1. Jeder Thread hat auch gleichzeitig noch einen Pipe in die andere Richtung, der quasi sagt "sende mir jetzt so und so viele Bytes, ich kann das verkraften" - heißt, der thread "sagt", wie viel er braucht

2. Ich bastle selber einen beliebig vergrößerbaren Puffer
Das mit dem beliebig vergrößerbaren Puffer ist aber ein Problem. Denn ich verstehe nicht so ganz, wie ich feststellen kann, ob ein Puffer überhaupt voll ist. Bei InputStreams scheint das einfach zu gehen per in==out oder so.

Gibt es ne andere (ggf. bessere) Möglichkeit? Wenn nicht, welche der Beiden ist besser? Bzw. beide Möglichkeiten sind prinzipiell die Gleichen, ich brauche auf jeden Fall einen beliebig vergrößerbaren Puffer.

 

[Murray]

Die Verteiler-Instanz könnte für jeden Kanal einen eigenen Thread verwenden.

 

[schoppenhauer_entfernt_]

Die Verteiler-Instanz könnte für jeden Kanal einen eigenen Thread verwenden.

Da kapier ich jetzt nicht so ganz, was du meinst. Sorry.

Also wenn ich das richtig verstehe, soll ich quasi noch zu jedem der zwei PipedOutputStreams einen eigenen Thread "Dazwischen" schalten, der sich um die Puffer kümmert, richtig?
Aber bzw... das kann ich ja dann gleich im Verteilungsthread selber machen.

 

[Murray]

Ist der "Verteilungsthread" derjenige, der die Daten produziert? Der verwendet dann doch irgendeine Instanz, die die Daten dann auf die verschiedene (Piped)OutputStreams verteilt. Die würde doch vereinfacht in etwa so aussehen:

public class DispatchingrOutputStream extends OutputStream() {

  private ArrayList<PipedOutputStream> targets;

  /* ... */

  public void write ( int b) throws IOException {
     for ( PipedOutputStream target : targets) {
       target.write( b);
     }
  }
}

Wenn ich das richtig verstanden habe, befürchtest Du jetzt, dass einer der verbundenen PipedInputStream nicht schnell genug ausgelesen werden kann und die auf diesen Stream schreibende write-Methode daher blockiert, sobald der Puffer des InputStreams voll ist. Damit würde die Schleife nicht weiter abgearbeitet, so dass die anderen Streams nicht mehr bedient werden; ausserdem kehrt die gesamte write Methode dann nicht zurück, bevor nicht alle Daten geschrieben werden konnten. Damit würde ein einziger langsamer Consumer-Thread den Producer blockieren.
Wenn man das vermeiden will, dann müsste für jedes Target ein eigener Thread verwendet werden.
Das könnte in etwa so aussehen (nur so runtergeschrieben und nicht zur Übersetzung empfohlen):

public class DispatchingrOutputStream extends OutputStream() {

  class Target implements Runnable {

     private Thread t;
     private PipedOutputStream targetStream;
     private int data;

     public Target( PipedOutputStream targetStream) {
       this.targetStream = targetStream;
       t = new Thread( this).start();
     }

     public void run() {
        while ( true) {
          synchronized( this) {
            wait();
          }
          target.write( data);
       }
     }

     public synchronized void write( int b) throws IOException {
        data = b;
        notify();
     }
  }

  private ArrayList<Target> targets;

  /* ... */

  public void write ( int b) throws IOException {
     for ( Target target : targets) {
       target.write( b);
     }
  }
}

/EDIT: Oops, das war zu früh abgeschickt.

Wenn jetzt Targets eben nicht nur das eine zu schreibende Byte speichern, sondern ihren eigenen Puffer verwalten, dann kannst Du diesen Puffer eben auch dynamisch verändern.

 

[schoppenhauer_entfernt_]

Ok.

Auch wenn ich den unteren Code nicht wirklich (bzw. eigentlich garnicht) verstehe, danke. Ich hoffe, wenn ich mir das noch etwas länger anschaue, wirds mir hoffentlich klarer.