Java: sincronización de varios subprocesos según el paso del tiempo

Question

Supongamos que tengo un maestro que mantiene una lista de objetos SlaveThread. En cada paso de tiempo, deseo que el maestro ejecute SlaveThreads en paralelo; sin embargo, al final del paso de tiempo, quiero que los SlaveThreads esperen entre sí para completar el paso de tiempo actual antes de avanzar. Además, no quiero reinstalar SlaveThreads en cada paso de tiempo. Tengo 2 soluciones posibles, y no sé cómo hacer que ninguna funcione:

1) El método run() en SlaveThread está en un ciclo while (verdadero). Después de la ejecución de un solo bucle en el SlaveThread, voy a tener SlaveThread notificar al maestro (que no sé cómo hacerlo), y el Maestro hace algo como

try{ 
    for (int i = 0; i < numSlaveThreads; i++) { 
     while (!slaveThreads[i].getCompletedThisIter()) { 
     wait() 
     } 
     } 
    System.out.println("Joined"); 

}

antes avanzando al siguiente paso de tiempo. ¿Cómo haría esto? ¿Cómo puedo hacer que un solo SlaveThread notifique solo al maestro?

2) La ejecución() en Slave no está en el ciclo while (true), entonces tengo que llamar a start() en cada iteración. Pero el estado del hilo del esclavo en este punto terminará. ¿Cómo puedo llamar a start() nuevamente sin tener que volver a instalarlo?

Answer 1

Eso es exactamente lo que son las barreras, puede realizar esto con CyclicBarrier o CountDownLatch. Estos son sincronizadores utilizados para retrasar el progreso de los hilos hasta que se llegue al estado deseado, en su caso los hilos han completado su cálculo.

Aquí depende de los detalles de la forma en que desea realizar:

Cierres son para la espera de los acontecimientos; las barreras son para esperar otros hilos .

Para el CyclicBarrier que se hace de la siguiente manera:

// whereby count is the number of your slave threads 
this.barrier = new CyclicBarrier(count); 

Luego, en la definición Runnable de sus esclavos se va a insertar en el extremo de la computación: barrier.await()

public class Slaves implements Runnable { 

    // ... 

    @Override 
    public void run() { 

     while(condition) { 

     // computation 
     // ... 

     try { 
      // do not proceed, until all [count] threads 
      // have reached this position 
      barrier.await(); 
     } catch (InterruptedException ex) { 
      return; 
     } catch (BrokenBarrierException ex) { 
      return; 
     } 
     } 
    } 
} 

Sus hilos esclavos no continuarán, hasta que todos sus hilos hayan terminado el cálculo. De esta forma, no necesita realizar la señalización entre otro hilo maestro.

Si, sin embargo, tiene un código que desea ejecutar después de que todos los hilos hayan alcanzado esa posición (señalización maestra) puede pasar Runnable adicional al constructor CyclicBarrier, que se ejecutará después de que todos los hilos hayan llegado a la barrera .

this.barrier = new CyclicBarrier(count, 
    new Runnable() { 
     @Override 
     public void run() { 
     // signal your master thread, update values, etc. 
     } 
    } 
); 

Answer 2

Se puede usar una combinación de un ExecutorService para administrar sus hilos (es decir, a reciclar sus hilos sin tener que crear otros nuevos en cada ciclo) y un CyclicBarrier que sincronizar todos los esclavos.

Vea a continuación un ejemplo simple en el que el maestro inicia los esclavos en un bucle, asegurándose de que están hechos antes de volver a comenzar.Los esclavos, siendo un poco perezoso, simplemente dormir durante algún tiempo (no es realmente aleatoria):

public class Test { 

    private static final ExecutorService executor = Executors.newFixedThreadPool(5); 
    private static final CyclicBarrier barrier = new CyclicBarrier(5); //4 slaves + 1 master 

    public static void main(String[] args) throws InterruptedException { 
     Runnable master = new Runnable() { 
      @Override 
      public void run() { 
       try { 
        while (true) { 
         System.out.println("Starting slaves"); 
         for (int i = 100; i < 500; i += 100) { 
          executor.submit(getRunnable(i)); 
         } 
         barrier.await(); 
         System.out.println("All slaves done"); 
        } 
       } catch (InterruptedException | BrokenBarrierException ex) { 
        System.out.println("Bye Bye"); 
       } 
      } 
     }; 

     executor.submit(master); 
     Thread.sleep(2000); 
     executor.shutdownNow(); 

    } 

    public static Runnable getRunnable(final int sleepTime) { 
     return new Runnable() { 
      @Override 
      public void run() { 
       try { 
        System.out.println("Entering thread " + Thread.currentThread() + " for " + sleepTime + " ms."); 
        Thread.sleep(sleepTime); 
        System.out.println("Exiting thread " + Thread.currentThread()); 
        barrier.await(); 
       } catch (BrokenBarrierException | InterruptedException ex) { 
       } 
      } 
     }; 

    } 
}