Notificar subprocesos cuando el contador cambia

Question

Estoy tratando de diseñar una clase como Code Kata para mí que tiene una propiedad de valor que se puede establecer, y la clase puede emitir instancias de ValueListener. La idea es que haya una instancia de ValueHolder con muchos subprocesos de cliente accediendo a ella simultáneamente. Cada subproceso de cliente ha solicitado un ValueWatcher y ha llamado a waitForValue().

Lo que realmente estoy luchando es qué condición debo usar en el ciclo while() para evitar notificaciones espurias (es decir, el valor no ha cambiado). Puedo ver que este diseño puede hacer posible que las instancias de ValueWatcher pierdan actualizaciones, pero estoy menos preocupado por eso en esta etapa.

¡Agradecería cualquier orientación que se ofrezca!

public class ValueHolder { 

    private int value = 0; 
    private final Object monitor = new Object(); 

    public void setValue(int value) { 
    synchronized(monitor) { 
     this.value = value; 
     monitor.notifyAll(); 
    } 
    } 

    ValueWatcher createChangeWatcher() { 
    return new ValueWatcher(); 
    } 

    private class ValueWatcher { 
    public int waitForValue() { 
     synchronized(monitor) { 
     while (==== ??? =====) { 
      monitor.wait(); 
      return value; 
     } 
     } 
    } 
    }  
} 

Answer 1

Problema interesante. Aquí hay una solución en la parte superior de mi cabeza. Tener un número de versión junto con el valor que se está cambiando. Cada vez que se actualiza el valor, el número de versión también se incrementa para que los objetos ValueWatcher puedan verificar si la versión se ha ejecutado, lo que significa que se ha producido un cambio.

Editar: inicialmente tenían una AtomicLong pero yo estoy robando la idea de un objeto envoltorio de @John Vint.

private final VersionValue versionValue = new VersionValue(); 

public void setValue(int value) { 
    synchronized (monitor) { 
     versionValue.value = value; 
     versionValue.version++; 
     monitor.notifyAll(); 
    } 
} 

private class ValueWatcher { 
    private long localVersion = 0; 
    public int waitForValue() { 
     synchronized (monitor) { 
      while (true) { 
       if (localVersion < versionValue.version) { 
        // NOTE: the value might have been set twice here 
        localVersion = versionValue.version; 
        return versionValue.value; 
       } 
       monitor.wait(); 
      } 
     } 
    } 
} 

private static class VersionValue { 
    int value; 
    long version; 
} 

---

También, aunque activaciones falsas son posibles es importante recordar que el texto:

invocan siempre espera dentro de un bucle que las pruebas para la enfermedad que se esperaban. No suponga que la interrupción fue para la condición particular que estaba esperando, o que la condición sigue siendo verdadera.

Más acerca de las condiciones de carrera y los modelos de productor/consumidor que los despertadores espurios. Ver my page here about that.

Answer 2

private class ValueWatcher { 
    private int oldValue = 0; 

    public int waitForValue() { 
    synchronized(monitor) { 
     while (value == oldValue) { 
     monitor.wait(); 
     } 
     oldValue = value 
     return oldValue; 
    } 
    } 
} 

Answer 3

Lo único que realmente importa es si el valor cambió después de ingresar el método y su bloque sincronizado. Por lo tanto, tome las marcas de tiempo de la última vez que se modificó un valor y solo continúe cuando la última marca de tiempo actualizada> luego cuando ingresó.

private final StampedValue stamped = new StampedValue(); 

    public void setValue(int value) { 
     synchronized (monitor) { 
      this.stamped.value = value; 
      this.stamped.lastUpdated = System.currentTimeMillis(); 
      monitor.notifyAll(); 
     } 
    } 
    private class ValueWatcher { 

     public int waitForValue() { 
      synchronized(monitor) { 
       long enteredOn = System.currentTimeMillis();  
       while (enteredOn > stamped.lastUpdated) { 
        monitor.wait(); 
       } 
       return stamped.value; 
      } 
     } 
    } 
    private class StampedValue { 
     long lastUpdated = System.currentTimeMillis(); 
     int value; 
    } 

Answer 4

public class ValueHolder { 

    private final Object monitor = new Object(); 
    private LinkedList<WeakReference<ValueWatcher>> waiters = new LinkedList<WeakReference<ValueWatcher>>(); 

    public void setValue(int value) { 
     synchronized (monitor) { 
      Iterator<WeakReference<ValueWatcher>> it = waiters.iterator(); 
      while (it.hasNext()) { 
       WeakReference<ValueWatcher> ref = it.next(); 
       if (ref.get() == null) 
        it.remove(); 
       else 
        ref.get().waitingList.add(value); 
      } 
      monitor.notifyAll(); 
     } 
    } 

    ValueWatcher createChangeWatcher() { 
     ValueWatcher ret = new ValueWatcher(); 
     synchronized(monitor) { 
      waiters.add(new WeakReference<ValueWatcher>(ret)); 
     } 
     return ret; 
    } 

    private class ValueWatcher { 

     private Queue<Integer> waitingList = new LinkedList<Integer>(); 

     public int waitForValue() { 
      synchronized (monitor) { 
       while (waitingList.isEmpty()) { 
        monitor.wait(); 
       } 
       return waitingList.poll(); 
      } 
     } 
    } 
} 

La idea es que no perder de vista que está esperando y cada observador tiene una cola de los valores que se han establecido desde la última invocación de waitForValue(). Esto elimina la necesidad de almacenar value en ValueHolder, lo cual es bueno ya que cuando se despierta ValueWatcher, eso podría haber cambiado varias veces. El inconveniente de este método es que, como pueden ver, la creación de nuevos observadores se bloqueará hasta que el monitor esté libre.

Answer 5

¿Qué pasa con cada oyente que tiene un BlockingQueue que le da a la secuencia de determinación de valor como parte de su registro como un oyente? Luego, cuando se cambia el valor, el hilo de configuración de valor simplemente se repite en cada una de esas colas, dándole el nuevo valor. Es posible que desee utilizar BlockingQueue.offer en ese bucle, de modo que si un hilo aún no está listo para recibir el nuevo valor, no detendrá que otros hilos lo reciban.

Puede que este no sea el enfoque más eficiente, pero es simple, y la estructura concurrente (es decir, la parte difícil) está bien probada y se mantiene para usted. Y tampoco es tan ineficiente.