Notifica thread in caso di modifiche contatore

Question

Sto provando a progettare una classe come codice Kata per me che ha una proprietà valore che può essere impostata e la classe può emettere istanze ValueListener. L'idea è che esiste un'istanza di ValueHolder con molti thread client che accedono contemporaneamente. Ogni thread client ha richiesto un ValueWatcher e ha chiamato waitForValue().

Quello con cui sto davvero combattendo è quale condizione dovrei usare nel ciclo while attorno all'attesa() per evitare notifiche spurie (cioè il valore non è cambiato). Vedo che questo design può rendere possibile che le istanze di ValueWatcher perdano gli aggiornamenti, ma in questo momento sono meno preoccupato di ciò.

Apprezzerebbero qualsiasi guida in offerta!

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 interessante. Ecco una soluzione in cima alla mia testa. Avere un numero di versione insieme al valore che viene modificato. Ogni volta che il valore viene aggiornato, anche il numero di versione viene incrementato in modo che gli oggetti ValueWatcher possano quindi verificare se la versione è aumentata, il che significa che è stata apportata una modifica.

Edit: inizialmente ho avuto un AtomicLong ma sto rubando l'idea di un oggetto wrapper da @ 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; 
} 

---

Inoltre, anche se wakeups spurie sono possibili è importante ricordare che il testo:

invocano sempre attesa all'interno di un ciclo che i test per la condizione di essere aspettato. Non dare per scontato che l'interruzione riguardasse la particolare condizione che si stava aspettando o che la condizione fosse ancora vera.

È più sulle condizioni di gara e sui modelli produttore/consumatore che su wakeups spuri. Vedi 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

Tutto ciò che interessa veramente è se il valore viene modificato dopo aver immesso il metodo e il relativo blocco sincronizzato. Quindi prendi un timestamp dell'ultima volta in cui un valore è stato modificato e prosegui solo quando l'ultimo timestamp aggiornato> quindi quando hai inserito.

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

L'idea è di tenere traccia di chi aspetta e ogni osservatore ha una coda di valori che sono stati impostati dall'ultima chiamata di waitForValue(). Ciò elimina la necessità di archiviare value in ValueHolder, il che è una buona cosa quando si riavvia uno ValueWatcher, che potrebbe essere stato modificato più volte. Lo svantaggio di questo metodo è che, come puoi vedere, la creazione di nuovi watcher bloccherà fino a quando il monitor non sarà libero.

Answer 5

Che dire di ogni ascoltatore con uno BlockingQueue che fornisce al thread di impostazione del valore come parte della sua registrazione come listener? Quindi, quando il valore viene modificato, il thread di impostazione del valore scorre semplicemente su ciascuna di queste code, assegnandogli il nuovo valore. Potresti voler usare BlockingQueue.offer in quel ciclo, in modo che se un thread non è ancora pronto per ricevere il nuovo valore, non impedirà ad altri thread di riceverlo.

Questo potrebbe non essere l'approccio più efficiente, ma è semplice, e la struttura concorrente (cioè la parte difficile) è ben collaudata e mantenuta per voi. E non è nemmeno inefficiente.