Java Concurrency in der Praxis: Race-Bedingung in BoundedExecutor?

Question

Es gibt etwas Seltsames an der Implementierung des BoundedExecutor in dem Buch Java Concurrency in Practice.

Es soll die Aufgabenübergabe an den Executor abbremsen, indem der übergebende Thread blockiert wird, wenn genügend Threads entweder in der Warteschlange stehen oder im Executor ausgeführt werden.

Dies ist die Implementierung (nach dem fehlenden rethrow in der catch-Klausel hinzufügen):

public class BoundedExecutor { 
    private final Executor exec; 
    private final Semaphore semaphore; 

    public BoundedExecutor(Executor exec, int bound) { 
     this.exec = exec; 
     this.semaphore = new Semaphore(bound); 
    } 

    public void submitTask(final Runnable command) throws InterruptedException, RejectedExecutionException { 
     semaphore.acquire(); 

     try { 
      exec.execute(new Runnable() { 
       @Override public void run() { 
        try { 
         command.run(); 
        } finally { 
         semaphore.release(); 
        } 
       } 
      }); 
     } catch (RejectedExecutionException e) { 
      semaphore.release(); 
      throw e; 
     } 
    } 

Als ich die BoundedExecutor mit einem Executors.newCachedThreadPool() und einer Grenze von 4 instanziiert, würde ich die Anzahl der Threads erwartet instanziiert durch der Cache-Thread-Pool sollte nie größer als 4 sein. In der Praxis ist dies jedoch der Fall. Ich habe dieses kleine Testprogramm bekommen, so viel wie 11 Threads zu erstellen:

public static void main(String[] args) throws Exception { 
    class CountingThreadFactory implements ThreadFactory { 
     int count; 

     @Override public Thread newThread(Runnable r) { 
      ++count; 
      return new Thread(r); 
     }   
    } 

    List<Integer> counts = new ArrayList<Integer>(); 

    for (int n = 0; n < 100; ++n) { 
     CountingThreadFactory countingThreadFactory = new CountingThreadFactory(); 
     ExecutorService exec = Executors.newCachedThreadPool(countingThreadFactory); 

     try { 
      BoundedExecutor be = new BoundedExecutor(exec, 4); 

      for (int i = 0; i < 20000; ++i) { 
       be.submitTask(new Runnable() { 
        @Override public void run() {} 
       }); 
      } 
     } finally { 
      exec.shutdown(); 
     } 

     counts.add(countingThreadFactory.count); 
    } 

    System.out.println(Collections.max(counts)); 
} 

Ich denke, es ist ein winzig kleiner Zeitrahmen zwischen der Freigabe des Semaphors und die Aufgabe endet, wo ein anderer Thread eine Genehmigung erwerben kann und Aufgabe einreichen, solange der Freigabe-Thread noch nicht beendet ist. Mit anderen Worten, es hat eine Wettlaufbedingung.

Kann jemand das bestätigen?

Answer 1

Ich sehe so viel wie 9 Threads auf einmal erstellt. Ich vermute, dass es eine Race-Bedingung gibt, die dazu führt, dass mehr Thread als erforderlich ist.

Dies könnte sein, weil es vor und nach dem Ausführen der Aufgabe Arbeit zu tun gibt. Das bedeutet, dass, obwohl nur 4 Threads in Ihrem Codeblock vorhanden sind, eine Reihe von Threads eine vorherige Aufgabe beendet oder bereit ist, eine neue Aufgabe zu starten.

d. H. Der Thread macht eine Freigabe(), während es noch läuft. Auch wenn es das letzte ist, was du tust, ist es nicht das letzte, was es macht, bevor du eine neue Aufgabe erwirbst.

Answer 2

Sie haben Ihre Analyse des Race Condition korrekt durchgeführt. Es gibt keine Synchronisationsgarantien zwischen dem ExecutorService & des Semaphore.

Allerdings weiß ich nicht, ob der BoundedExecutor für die Drosselung der Anzahl der Threads verwendet wird. Ich denke, es ist mehr für die Drosselung der Anzahl der Aufgaben an den Dienst übermittelt. Stellen Sie sich vor, Sie haben 5 Millionen Aufgaben, die Sie einreichen müssen, und wenn Sie mehr als 10.000 davon einreichen, haben Sie zu wenig Speicherplatz.

Nun, Sie werden immer nur 4 Threads haben, die zu einer bestimmten Zeit ausgeführt werden, warum sollten Sie versuchen, alle 5 Millionen Tasks anzuordnen? Sie können ein ähnliches Konstrukt verwenden, um die Anzahl der Aufgaben zu begrenzen, die zu einem bestimmten Zeitpunkt in der Warteschlange stehen. Was Sie daraus machen sollten ist, dass zu jedem Zeitpunkt nur 4 Aufgaben ausgeführt werden.

Offensichtlich ist die Auflösung zu diesem einen Executors.newFixedThreadPool(4) zu verwenden.

Answer 3

BoundedExecutor war in der Tat als eine Veranschaulichung der Drohung Aufgabe Aufgabe, nicht als eine Möglichkeit, eine Grenze auf Thread-Pool-Größe zu platzieren gedacht. Es gibt direktere Wege, Letzteres zu erreichen, wie zumindest ein Kommentar herausstellte.

Aber auch die anderen Antworten nicht den Text in dem Buch erwähnt, das sagt eine unbegrenzte Warteschlange zu verwenden und zu

set the bound on the semaphore to be equal to the pool size plus the number of queued tasks you want to allow, since the semaphore is bounding the number of tasks both currently executing and awaiting execution. [JCiP, end of section 8.3.3]

Durch unbeschränkte Warteschlangen und Poolgröße zu nennen, wurden implizieren wir (offenbar nicht ganz klar) die Verwendung eines Thread-Pools begrenzter Größe.

Was mich bei BoundedExecutor immer gestört hat, ist, dass es die ExecutorService-Schnittstelle nicht implementiert. Ein moderner Weg, um eine ähnliche Funktionalität zu erreichen und dennoch die Standardschnittstellen zu implementieren, wäre die Verwendung der Guava-Methode listeningDecorator und der ForwardingListeningExecutorService-Klasse.