Почему «ExecutorService» не может последовательно создавать потоки?

Question

Я пытаюсь переопределить мой параллельный код, используя CyclicBarrier, который является новым для меня. Я могу обойтись без него, но я время возможности проверки его против моего другого решения, проблема у меня есть тупиковая ситуация со следующим кодом:

//instance variables (fully initialised elsewhere). 
private final ExecutorService exec = Executors.newFixedThreadPool(4); 
private ArrayList<IListener> listeners = new ArrayList<IListener>(); 
private int[] playerIds; 

private class WorldUpdater { 
    final CyclicBarrier barrier1; 
    final CyclicBarrier barrier2; 
    volatile boolean anyChange; 
    List<Callable<Void>> calls = new ArrayList<Callable<Void>>(); 

    class SyncedCallable implements Callable<Void> { 
     final IListener listener; 

     private SyncedCallable(IListener listener) { 
      this.listener = listener; 
     } 

     @Override 
     public Void call() throws Exception { 
      listener.startUpdate(); 
      if (barrier1.await() == 0) { 
       anyChange = processCommons(); 
      } 
      barrier2.await(); 
      listener.endUpdate(anyChange); 
      return null; 
     } 
    } 

    public WorldUpdater(ArrayList<IListener> listeners, int[] playerIds) { 
     barrier2 = new CyclicBarrier(listeners.size()); 
     barrier1 = new CyclicBarrier(listeners.size()); 
     for (int i : playerIds) 
      calls.add(new SyncedCallable(listeners.get(i))); 
    } 

    void start(){ 
     try { 
      exec.invokeAll(calls); 
     } catch (InterruptedException e) { 
      e.printStackTrace(); 
     } 
    } 
} 


void someMethodCalledEveryFrame() { 
    //Calls some Fisher-something method that shuffles int[] 
    shufflePIDs(); 
    WorldUpdater updater = new WorldUpdater(listeners, playerIds); 
    updater.start(); 
} 

Я использую отладчик в Android Studio (IntelliJ), чтобы приостановить выполнение на этом сцена. Я получаю несколько потоков, показывающие мои await звонки, как последний из моего кода, который будет выполнен

->Unsafe.park

->LockSupport.park

->AbstractQueuedSynchronizer$ConditionObject.await

->CyclicBarrier.doWait

->CyclicBarrier.await

По крайней мере, один поток будет иметь этот стек:

->Unsafe.park.

->LockSupport.park

->AbstractQueuedSynchronizer$ConditionObject.await

->LinkedBlockingQueue.take

->ThreadPoolExecutor.getTask

->ThreadPoolExecutor.runWorker

->ThreadPoolExecutor$Worker.run

->Thread.run

Я замечаю, что CyclicBarrier не играет никакой роли в этих последних нисходящих потоках.

processCommonsявляется вызова exec.invokeAll (на 3-х слушателей), я полагаю, это означает, что я бегу из нитей. Но много раз это не так, пожалуйста, может кто-нибудь уточнить, почему ExecutorService не может постоянно планировать мои темы? У них есть свой стек и счетчик программ, поэтому я бы подумал, что это не проблема. Я только когда-либо бегу сразу. Кто-нибудь поможет мне в математике?

Answer 1

я сделал небольшой эксперимент и придумал:

 @Override 
     public Void call() throws Exception { 
      System.out.println("startUpdate, Thread:" + Thread.currentThread()); 
      listener.startUpdate(); 
      if (barrier1.await() == 0) { 
       System.out.println("processCommons, Thread:" + Thread.currentThread()); 
       anyChange = processCommons(); 
      } 
      barrier2.await(); 
      System.out.println("endUpdate, Thread:" + Thread.currentThread()); 
      listener.endUpdate(anyChange); 
      return null; 
     } 

который показал при использовании пула 3 с 3 listeners, я всегда буду висеть в processCommons, который содержит следующее:

List<Callable<Void>> calls = new ArrayList<Callable<Void>>(); 
    for (IListener listiner : listeners) 
     calls.add(new CommonsCallable(listener)); 
    try { 
     exec.invokeAll(calls); 
    } catch (InterruptedException e) { 
     e.printStackTrace(); 
    } 

С двумя нитями, ожидающими барьера, а третий пытается создать еще 3. Мне нужна была одна дополнительная нить в ExecutorService, а 2 на барьере можно было «переработать», как я задавал в своем вопросе.У меня есть ссылки на 6 потоков на этом этапе, когда exec удерживает только 4. Это может работать счастливо в течение многих минут.

private final ExecutorService exec = Executors.newFixedThreadPool(8); 

Должно быть лучше, но это не так.

Наконец я сделал точку останова степпинг в IntelliJ (спасибо!) IdeaC

Проблема заключается в том

 if (barrier1.await() == 0) { 
      anyChange = processCommons(); 
     } 
     barrier2.await(); 

Между 2 await вы можете получить несколько приостановленных потоков, которые на самом деле не достигли await. В случае 3 слушателей из пула из 4 требуется только одно «получить незапланированное» (или что-то еще), а barrier2 никогда не получит полный комплект. Но как насчет того, когда у меня есть пул из 8? Такое же поведение проявляется со всеми, кроме двух нитей, стопкой неопределенности:

-> Unsafe.park.

-> LockSupport.park

-> AbstractQueuedSynchronizer $ ConditionObject.await

-> LinkedBlockingQueue.take

-> ThreadPoolExecutor.getTask

-> ThreadPoolExecutor.runWorker

-> ThreadPoolExecutor $ Worker.run

-> Thread.run

Что может быть здесь, чтобы отключить все 5 потоков? Я должен был принять совет Джеймса Большого и избегать ворчания в этом сложном CyclicBarrier .-- ОБНОВЛЕНИЕ-- Это может бегать всю ночь без CyclicBarrier.

Answer 2

Какое значение listeners.size() при создании вашего WorldUpdater? Если его больше четырех, то ваши потоки никогда не пройдут через барьер.

ExecutorService имеет точно четыре резьбы. Не больше, не меньше. Вызывающие barrier1.await() и barrier2.await() не пройдут через барьер до тех пор, пока точно не listeners.size() нитей ждут.

Моя реакция кишки, было бы ошибкой в ​​пул-потоках использовать CyclicBarrier. CyclicBarrier полезен только тогда, когда вы точно знаете, сколько потоков будет использовать его. Но, когда вы используете пул потоков, вы часто делаете не знаете размер пула. Фактически, в реальном (например, коммерческом) приложении, если вы используете пул потоков, он, вероятно, вообще не был создан вашим кодом. Вероятно, он был создан где-то в другом месте и передан в ваш код как вложенная зависимость .