Threads eines Pools aufeinander warten lassen

[snOOfy]

Hallo!

Ich habe einen Threadpool, der mehrere Worker startet, die parallel ihre Operationen ausführen. Zu Testzwecken möchte ich nun erreichen, dass immer nur einer der Worker gleichzeitig aktiv ist, während die anderen warten.
Wenn der aktive Worker eine Operation beendet hat, soll er den nächsten benachrichtigen, dass dieser nun eine Operation durchführen darf. Wenn es also z.B. 4 Worker gibt, sollen diese in der Reihenfolge 1, 2, 3, 4, 1, 2, 3, 4, ... jeweils eine Operation ausführen.

Das Benachrichtigen habe ich mit einer statischen Integervariable gelöst, in der der aktive Worker jeweils die Nummer des nächsten speichert. Das Problem: Ich schaffe es nicht, die Worker zu pausieren, da wait() in der run()-Funktion des Workers zu einer IllegalMonitorStateException führt. Hat jemand eine Idee, wie man das machen könnte?

 

[SlaterB]

wait() muss an einem Objekt aufgerufen werden, in einem synchronized-Block an diesem Monitor,
dann funktioniert auch notify()
siehe z.B.
Galileo Computing :: Java ist auch eine Insel (8. Auflage) – 11 Threads und nebenläufige Programmierung

unabhängiges Warten wäre sleep()

-------


allgemein klingt aber die kontrollierte isolierte Wiederholung von 4 Aktionen nach einer perfekten Beschäftigung für EINEN Thread,
wieso bloß 4?!

edit:
ach so ja, zu Testzwecken

 

[snOOfy]

allgemein klingt aber die kontrollierte isolierte Wiederholung von 4 Aktionen nach einer perfekten Beschäftigung für EINEN Thread

Ja das stimmt. In der Originalversion meines Programmes lief das alles in einem Thread. Zwecks Laufzeitoptimierung sollen nun Berechnungen auf mehrere CPUs verteilt werden. Durch die Parallelität wird ein globaler Zufallsgenerator in einer anderen Reihenfolge aufgerufen als im Original, wodurch sich das Ergebnis leicht verändert. Ich bin aber nicht sicher, ob das wirklich nur an diesem Zufallsgenerator liegt, oder ob durch die Umstellung auf mehrere Threads, die das Programm grundlegend umstrukturiert hat, nicht noch andere Fehler aufgetreten sind. Daher möchte ich als Test die Threads in der gleichen Reihenfolge arbeiten lassen wie die Originalversion, um anschließend die Ergebnisse zu vergleichen (ist wahrscheinlich etwas umständlich beschrieben, sorry)

Ich habe in deinem Link den folgenden Abschnitt angeschaut:

Galileo Computing :: Java ist auch eine Insel (8. Auflage) – 11.6 Synchronisation über Warten und Benachrichtigen

Dort wird das Problem zwar beschrieben, ich kann für mich daraus aber keine Lösung basteln... wie muss ich das mit dem Lock regeln, damit der eine Worker wait() aufruen und auf einen anderen warten kann?

Das mit sleep habe ich in einer älteren Version versucht, das war aber nicht gut, da ich eine Angabe in Millisekunden machen musste, wie lange gewartet werden soll. Es soll aber genau so lange gewartet werden, bis vom anderen Worker die Beachrichtigung kommt.

 

[SlaterB]

> wie muss ich das mit dem Lock regeln, damit der eine Worker wait() aufruen und auf einen anderen warten kann?
na auf korrekte Weise

es müssen einfach nur alle das gleiche Objekt kennen und dann daran die Methoden aufrufen,
wie man ein Objekt verteilt müsste doch bekannt sein,
im Konstruktor oder per set() übergeben, irgendwo anders per get() abfragen, statische Variable,

 

[snOOfy]

Hmm ich habe es noch nicht hinbekommen. Um das ganze zu vereinfachen, habe ich nun ein Primzahlberechnungsprogramm geschrieben, dessen Threadsystem exakt dem meines anderen Programmes entspricht. Hier der Programmtext:

// Simulator.java

package prime;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class Simulator extends Thread {
   private  ExecutorService   service;
   public   static final int  THREADS = 4, START = 500000001, STOP = 500002000;
   public   boolean[]         done;
   private  boolean           cDone;
   
   public Simulator() {
      service = Executors.newCachedThreadPool();
      done = new boolean[Simulator.THREADS];
   }
  
  public static void main (String args[]) {
     
     Simulator sim = new Simulator();
     System.out.println("start");
     sim.cDone = false;
     for (int ti = 0; ti < Simulator.THREADS; ++ ti)
        sim.done[ti] = false;

     synchronized (sim) {
        for (int ti = 0; ti < Simulator.THREADS; ++ ti) 
           sim.service.execute(new CalcRunnable(ti, sim));

        while (!sim.cDone) {
           try {
              sim.wait();
              sim.cDone = true;
              for (int ti = 0; ti < Simulator.THREADS; ++ ti)
                 if (!sim.done[ti]) 
                    sim.cDone = false;
                 
           } catch (InterruptedException e) {
              e.printStackTrace();
           }
        }
     } // end sync
     
     System.out.println("stop");
     System.exit(0);
  }
}


// CalcRunnable.java

package prime;

public class CalcRunnable implements Runnable {
   
   private Simulator sim;
   private int number;
   
   public CalcRunnable(int number, Simulator sim) {
      this.number = number;
      this.sim = sim;
   }

  public void run() {
     try {
        System.out.println(Thread.currentThread().getName() + " running...");
        for (int i = Simulator.START + number * 2; i < Simulator.STOP; i += Simulator.THREADS * 2) {
           if (isPrime(i))
              System.out.println(Thread.currentThread().getName() + " -> " + i + " is prime");
        }
        sim.done[number] = true;
     }
     finally {
        synchronized (sim) {
           sim.notify();
        }
     }
  }

  public static boolean isPrime(long number){
     int bPrime = 0;
     for (long i = 1; i < Math.sqrt(number); ++ i)
        if (number % i == 0)
           ++ bPrime;
     return (bPrime <= 1);
  }
}

Wenn man das so startet, prüfen die 4 Threads in mehr oder weniger zufälliger Reihenfolge die Primzahlen im angegebenen Bereich. Wie kann ich das jetzt so hinkriegen, dass sie dabei jeweils aufeinander warten, so dass die Reihenfolge wieder stimmt?

 

[Marco13]

Du kannst dir auch mal java.util.concurrent (Java 2 Platform SE 5.0) und speziell java.util.concurrent.locks (Java 2 Platform SE 5.0) ansehen. Spätestens wenn z.B. "immer genau 4 Threads" arbeiten sollen, oder die synchronizsation in anderer Hinsicht "komplizierter" wird (d.h. z.B. der kritische Abschnitt nicht oder nur schlecht durch EIN "synchronized" eingeschlossen werden kann) sind solche Sachen wie Counting Latches, Conditions und Locks sehr hilfreich: Die "verstecken" praktisch die synchronized/wait/notify-Mechanismen in flexibler handhabbaren Objekten.

 

[snOOfy]

Danke für den Hinweis! Die Funktionen klingen teilweise ganz vielversprechend, aber ich würde trotzdem gerne die "primitive" Version nutzen, da es sich wie gesagt nur um einen Test handelt. Wenn ich die Threads in meinem Testprogramm anders verwalte als im Original, dann macht der Test nicht mehr so viel Sinn.

 

[Marco13]

Habe es mal "minimal" geändert, so dass es wohl das tut, was es soll. Einiges davon sollte man "schöner" machen, und in der "run"-Methode muss man sich noch genauer überlegen, WORUM das "synchronized(sim)" stehen sollte, aber vom Prinzip her:

// Von [url]http://www.java-forum.org/allgemeine-java-themen/88293-threads-eines-pools-aufeinander-warten-lassen.html[/url]
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class PrimeSimulator extends Thread {
   private  ExecutorService   service;
   public   static final int  THREADS = 4, START = 500000001, STOP = 500002000;
   public   boolean[]         done;
   private  boolean           cDone;

   public PrimeSimulator() {
      service = Executors.newCachedThreadPool();
      done = new boolean[PrimeSimulator.THREADS];
   }

  public static void main (String args[]) {

     PrimeSimulator sim = new PrimeSimulator();
     System.out.println("start");
     sim.cDone = false;
     for (int ti = 0; ti < PrimeSimulator.THREADS; ++ ti)
        sim.done[ti] = false;

     synchronized (sim) {
        for (int ti = 0; ti < PrimeSimulator.THREADS; ++ ti)
           sim.service.execute(new CalcRunnable(ti, sim));

        while (!sim.cDone) {
           try {
              sim.wait();
              sim.cDone = true;
              for (int ti = 0; ti < PrimeSimulator.THREADS; ++ ti)
                 if (!sim.done[ti])
                    sim.cDone = false;

           } catch (InterruptedException e) {
              e.printStackTrace();
           }
        }
     } // end sync

     System.out.println("stop");
     System.exit(0);
  }
}


class CalcRunnable implements Runnable {

   private PrimeSimulator sim;
   private int number;
   private int index;

   public CalcRunnable(int number, PrimeSimulator sim) {
      this.number = number;
      this.sim = sim;
   }

  public void run() {
     try {

        if (number > 0)
        {
            synchronized (sim)
            {
                while (!sim.done[number-1])
                {
                    System.out.println(Thread.currentThread().getName() +
                       " waiting until "+(number+1-1)+" finished");
                    try
                    {
                        sim.wait();
                    }
                    catch (InterruptedException e)
                    {
                        Thread.currentThread().interrupt();
                    }
                }
                System.out.println(Thread.currentThread().getName() +
                    " has waited until "+(number+1-1)+" finished and starts now");
            }
        }

        System.out.println(Thread.currentThread().getName() + " running...");
        for (int i = PrimeSimulator.START + number * 2; i < PrimeSimulator.STOP; i += PrimeSimulator.THREADS * 2) {
           if (isPrime(i))
              System.out.println(Thread.currentThread().getName() + " -> " + i + " is prime");
        }
        sim.done[number] = true;
     }
     finally {
        synchronized (sim) {
           System.out.println(Thread.currentThread().getName() + " finished");
           sim.notifyAll(); // notify ALL!
        }
     }
  }

  public static boolean isPrime(long number){
     int bPrime = 0;
     for (long i = 1; i < Math.sqrt(number); ++ i)
        if (number % i == 0)
           ++ bPrime;
     return (bPrime <= 1);
  }
}

 

[snOOfy]

Vielen Dank! Ich hatte gerade angefangen, ein eigenes Lock-Objekt zu implementieren, das durch die Threads durchgereicht werden kann... aber deine Idee, einfach den Simulator dafür zu benutzen scheint mir die bessere Variante zu sein.

Es war so gedacht, dass die Primzahlen in der richtigen Reihenfolge ausgegeben werden. Dazu soll ein Thread nich seine gesamte Schleife, sondern nur eine einzige Zahl prüfen und dann an den nächsten Thread übergeben. Das done[] wird im PrimeSimulator benutzt, um festzustellen, wann alle Threads komplett fertig sind. Um festzustellen, ob ein einzelner Schritt fertig ist, habe ich nun ein weiteres Array eingeführt, stepDone[].
Das Programm tut nun, was es soll, allerdings ist die Struktur ein wenig kompliziert geworden... vielleicht hast du ja noch einen Verbesserungsvorschlag?
Hier der abgeänderte Programmtext:

// PrimeSimulator.java

package prime;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
 
public class PrimeSimulator extends Thread {
   private  ExecutorService   service;
   public   static final int  THREADS = 4, START = 500000001, STOP = 500000200;
   public   boolean[]         done;
   public   boolean[]         stepDone;
   private  boolean           cDone;
 
   public PrimeSimulator() {
      service = Executors.newCachedThreadPool();
      done = new boolean[PrimeSimulator.THREADS];
      stepDone = new boolean[PrimeSimulator.THREADS];
   }
 
  public static void main (String args[]) {
 
     PrimeSimulator sim = new PrimeSimulator();
     System.out.println("start");
     sim.cDone = false;
     for (int ti = 0; ti < PrimeSimulator.THREADS; ++ ti) {
        sim.done[ti] = false;
        sim.stepDone[ti] = false;
     }
     sim.stepDone[PrimeSimulator.THREADS - 1] = true; // so that first one can start
 
     synchronized (sim) {
        for (int ti = 0; ti < PrimeSimulator.THREADS; ++ ti)
           sim.service.execute(new CalcRunnable(ti, sim));
 
        while (!sim.cDone) {
           try {
              sim.wait();
              sim.cDone = true;
              for (int ti = 0; ti < PrimeSimulator.THREADS; ++ ti)
                 if (!sim.done[ti])
                    sim.cDone = false;
 
           } catch (InterruptedException e) {
              e.printStackTrace();
           }
        }
     } // end sync
 
     System.out.println("stop");
     System.exit(0);
  }
}


// CalcRunnable.java

package prime;

class CalcRunnable implements Runnable {
 
   private PrimeSimulator sim;
   private int number;
 
   public CalcRunnable(int number, PrimeSimulator sim) {
      this.number = number;
      this.sim = sim;
   }
 
  public void run() {
     System.out.println(Thread.currentThread().getName() + " running...");
     try {
        for (int i = PrimeSimulator.START + number * 2; i < PrimeSimulator.STOP; i += PrimeSimulator.THREADS * 2) {
           try {
              int lastNumber = number - 1; // number of previously active thread
              if (lastNumber < 0)
                 lastNumber = PrimeSimulator.THREADS - 1;

               synchronized (sim) {
                   while (!sim.stepDone[lastNumber]) {
                       // waiting until previous thread finished step
                       try {
                           sim.wait();
                       }
                       catch (InterruptedException e) {
                           Thread.currentThread().interrupt();
                       }
                   }
              }
              //  has waited until previous thread finished step and starts now
              if (isPrime(i))
                 System.out.println(Thread.currentThread().getName() + " -> " + i + " is prime");
   
              sim.stepDone[number] = true;
              sim.stepDone[lastNumber] = false;
           }
           finally {
              synchronized (sim) {
                 // only one step finished, changing to next thread
                 sim.notifyAll(); 
              }
           } 
        } // end for
        sim.done[number] = true;
     } // end try
     finally {
        synchronized (sim) {
           System.out.println(Thread.currentThread().getName() + " finished completely");
           sim.notifyAll();
        } 
     } 
  } // end run
 
  public static boolean isPrime(long number){
     int bPrime = 0;
     for (long i = 1; i < Math.sqrt(number); ++ i)
        if (number % i == 0)
           ++ bPrime;
     return (bPrime <= 1);
  }
}

 

[Marco13]

Inwieweit Verbesserungsvorschläge mit einem Beibehalten der "Gesamtstruktur" und den gewünschten Zielen vereinbar sind, weiß ich nicht.

Mir ist nämlich nicht klar, was genau das Ziel ist - du hattest angedeutet, dass das ein vereinfachtes Beispiel sein soll, das nur "von der Struktur her" deinem eigentlichen Problem entspricht - und wie SlaterB schon gesagt hat: Mehrere Threads zu sequenteiller Arbeit zu "zwingen" macht so erstmal ja nicht viel Sinn....

Vielleicht könntest du nochmal in wenigen Stichpunkten zusammengefasst sagen, wie der Ablauf sein soll? Im Moment erscheint mir das etwa wie
- Es werden merhere Threads gestartet
- Irgendein Thread prüft eine Zahl auf prime
- Sobald der Thread fertig ist, prüft der nächste Thread die nächste Zahl auf prime

Es klingt im Moment ein bißchen, als könnte man das mit einer LinkedBlockingQueue lösen, aber bin nicht 100% sicher, ob das gewünscht ist...

 

[snOOfy]

Also, das Originalprogramm:
Es handelt sich um einen zellularen Automaten. In der ersten Version werden alle Zellen nacheinander in einem Thread behandelt. Dabei wird auf einen Zufallsgenerator zugegriffen, der aber durch die feste Reihenfolge der Zellen im Automaten und einem festgelegten Seed-Wert bei jedem Start des Programmes zum gleichen Ergebnis führt. Diese Reproduzierbarkeit des Simulationsergebnisses ist für mich wünschenswert.

Beim Versuch, das Programm zu parallelisieren, habe ich das "Spielfeld" des Automaten in vier Bereiche eingeteilt (hauptsächlich, weil mein PC vier CPU-Kerne hat), deren Zellen dann parallel von vier Threads abgearbeitet werden. Daraus folgt aber, dass sich nun bei jedem Start des Programmes ein anderes Endergebnis ergibt, da die Threads in zufälliger Reihenfolge Zahlen aus dem Zufallsgenerator entnehmen. Das wäre noch akzeptabel, wenn ich sicher wäre, dass die Unterschiede im Simulationsergebnis wirklich nur am Zufallsgenerator liegen und nicht an der Umstellung des Algorithmus auf Threads. Deswegen möchte ich eine Testversion des Programmes haben, in der die Zellen wie in der finalen Version auf Threads verteilt werden, von den Threads aber in der Originalreihenfolge bearbeitet werden. Dass dadurch die Threads aufeinander warten müssen und der Geschwindigkeitsvorteil verloren geht ist mir klar, es handelt sich wie gesagt nur um einen Test.

Da der Simulator sehr kompliziert ist, habe ich nun dieses Primzahl-Berechnungsprogramm geschrieben. Wie im zellularen Automaten werden vier Threads von einem Verwaltungsthread aus gestartet. Sie erhalten als "Bereich" eine Menge an Zahlen, die sie testen müssen:
Thread1: 1, 9, 17, ...
Thread2: 3, 11, 19, ...
Thread3: 5, 13, 21, ...
Thread4: 7, 15, 23, ...
Wie im zellularen Automaten lässt sich die Originalreihenfolge wieder herstellen, indem die Threads wie folgt jeweils eine Zahl abarbeiten:
(Thread1: 1), (Thread2: 3), (Thread3: 5), (Thread4: 7), (Thread1: 9), ...
Dazu ist es notwendig, dass der aktive Thread jeweils wieder den nächsten aktiviert.
Wenn dieses Verfahren gut funktioniert, möchte ich es dann als Test für den Simulator nutzen.

Sorry, ist doch ein bisschen länger als Stichpunkte geworden, ich hoffe ich habe es verständlich erklärt.

 

[Marco13]

OK, naja, der Hauptgrund für die Frage war, dass diese public boolean "done"/"stepDone"-Arrays eben strukturell IMHO nicht so schön sind und leicht ausufern könnten, wenn noch weitere Infos dazukommen, die eigentlich zu den CalcRunnables gehören: Es wäre IMHO etwas sauberer, diese Informationen als private fields der CalcRunnable-Klasse zu speichern, und im "PrimeSimulator" dann nur noch eine List<CalcRunnable> zu haben. Das wäre aber auch nicht "kürzer" oder "einfacher", aber evtl. leichter handhabbar. Ideal wäre natürlich (wenn ich das richtig verstanden habe) wenn man "von außen" keinen Unterschied sehen würde, ob der "Testmodus" (MIT dem stepDone) oder der echt parallele Modus verwendet wird. Aber wie gesagt, das sind Sachen, die man sich dann im Rahmen des realen Anwendungsfalles überlegen muss...