Concurrencia de Java en la práctica: condición de carrera en BoundedExecutor?

Question

Hay algo extraño acerca de la implementación del BoundedExecutor en el libro Java Concurrency in Practice.

Se supone que debe estrangular el envío de tareas al Ejecutor al bloquear el hilo de envío cuando hay suficientes hilos en cola o ejecutándose en el Ejecutor.

Esta es la aplicación (después de añadir el volver a lanzar falta en la cláusula catch):

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; 
     } 
    } 

Cuando una instancia del BoundedExecutor con un Executors.newCachedThreadPool() y un límite de 4, yo esperaría que el número de hilos instanciados por el grupo de subprocesos en caché nunca excederá 4. En la práctica, sin embargo, sí lo hace. He conseguido este pequeño programa de prueba para crear hasta 11 hilos:

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)); 
} 

Creo que hay un pequeño marco de tiempo entre la liberación del semáforo y la terminación de tareas, donde otro hilo puede adquirir un permiso y envíe una tarea mientras el hilo de liberación aún no ha terminado. En otras palabras, tiene una condición de carrera.

¿Alguien puede confirmar esto?

Answer 1

Veo tanto como 9 hilos creados a la vez. Sospecho que hay una condición de carrera que hace que haya más hilo de lo necesario.

Esto podría deberse a que hay antes y después de ejecutar el trabajo de tarea por hacer. Esto significa que a pesar de que solo hay 4 subprocesos dentro de su bloque de código, hay una serie de subprocesos que detienen una tarea previa o se preparan para comenzar una nueva tarea.

es decir, el hilo tiene una liberación() mientras se está ejecutando. Aunque es lo último que haces, no es lo último que hace antes de adquirir una nueva tarea.

Answer 2

Tiene razón en su análisis de la condición de carrera. No hay garantías de sincronización entre ExecutorService & el semáforo.

Sin embargo, no sé si el límite del número de subprocesos se usa para BoundedExecutor. Creo que es más para estrangular el número de tareas enviadas al servicio. Imagine si tiene 5 millones de tareas que debe enviar, y si envía más de 10.000 de ellas, se quedará sin memoria.

Bueno, solo tendrá 4 subprocesos ejecutándose en un momento dado, ¿por qué querría intentar y poner en cola todas las tareas de 5 millones? Puede usar una construcción similar a esta para reducir el número de tareas puestas en cola en un momento dado. Lo que debe sacarse de esto es que en un momento dado solo hay 4 tareas ejecutándose.

Obviamente, la resolución de esto es usar un Executors.newFixedThreadPool(4).

Answer 3

BoundedExecutor fue pensado como una ilustración de cómo acelerar el envío de tareas, no como una forma de colocar un límite en el tamaño del grupo de subprocesos. Hay formas más directas de lograr esto último, como se señaló al menos un comentario.

Pero las otras respuestas no mencionan el texto en el libro que dice utilizar una cola sin límites y para

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]

Al mencionar colas sin límites y el tamaño de la piscina, estábamos dando a entender (al parecer no es muy clara) el uso de un grupo de subprocesos de tamaño limitado.

Lo que siempre me ha molestado acerca de BoundedExecutor, sin embargo, es que no implementa la interfaz ExecutorService. Una forma moderna de lograr una funcionalidad similar y aún implementar las interfaces estándar sería utilizar el método listeningDecorator de Guava y la clase ForwardingListeningExecutorService.