Java: due WAITING + un thread BLOCCATO, notify() porta a un livelock, notifyAll() no, perché?

Question

Stavo cercando di implementare qualcosa di simile all'interfaccia BlockingQueue limitata di Java usando "primitive" di sincronizzazione Java (sincronizzato, wait(), notify()) quando mi sono imbattuto in qualche comportamento che non capisco.

Creo una coda in grado di memorizzare 1 elemento, creare due thread che attendono il recupero di un valore dalla coda, avviarli, quindi provare a inserire due valori nella coda in un blocco sincronizzato nel thread principale. La maggior parte delle volte funziona, ma a volte i due thread in attesa di un valore iniziano a svegliarsi apparentemente l'un l'altro e non lasciare che il thread principale entri nel blocco sincronizzato.

Ecco il mio codice (semplificato):

import java.util.LinkedList; 
import java.util.Queue; 

public class LivelockDemo { 
    private static final int MANY_RUNS = 10000; 

    public static void main(String[] args) throws InterruptedException { 
     for (int i = 0; i < MANY_RUNS; i++) { // to increase the probability 
      final MyBoundedBlockingQueue ctr = new MyBoundedBlockingQueue(1); 

      Thread t1 = createObserver(ctr, i + ":1"); 
      Thread t2 = createObserver(ctr, i + ":2"); 

      t1.start(); 
      t2.start(); 

      System.out.println(i + ":0 ready to enter synchronized block"); 
      synchronized (ctr) { 
       System.out.println(i + ":0 entered synchronized block"); 
       ctr.addWhenHasSpace("hello"); 
       ctr.addWhenHasSpace("world"); 
      } 

      t1.join(); 
      t2.join(); 

      System.out.println(); 
     } 
    } 

    public static class MyBoundedBlockingQueue { 
     private Queue<Object> lst = new LinkedList<Object>();; 

     private int limit; 

     private MyBoundedBlockingQueue(int limit) { 
      this.limit = limit; 
     } 

     public synchronized void addWhenHasSpace(Object obj) throws InterruptedException { 
      boolean printed = false; 
      while (lst.size() >= limit) { 
       printed = __heartbeat(':', printed); 
       notify(); 
       wait(); 
      } 
      lst.offer(obj); 
      notify(); 
     } 

     // waits until something has been set and then returns it 
     public synchronized Object getWhenNotEmpty() throws InterruptedException { 
      boolean printed = false; 
      while (lst.isEmpty()) { 
       printed = __heartbeat('.', printed); // show progress 
       notify(); 
       wait(); 
      } 
      Object result = lst.poll(); 
      notify(); 
      return result; 
     } 

     // just to show progress of waiting threads in a reasonable manner 
     private static boolean __heartbeat(char c, boolean printed) { 
      long now = System.currentTimeMillis(); 
      if (now % 1000 == 0) { 
       System.out.print(c); 
       printed = true; 
      } else if (printed) { 
       System.out.println(); 
       printed = false; 
      } 
      return printed; 
     } 
    } 

    private static Thread createObserver(final MyBoundedBlockingQueue ctr, 
      final String name) { 
     return new Thread(new Runnable() { 
      @Override 
      public void run() { 
       try { 
        System.out.println(name + ": saw " + ctr.getWhenNotEmpty()); 
       } catch (InterruptedException e) { 
        e.printStackTrace(System.err); 
       } 
      } 
     }, name); 
    } 
} 

Ecco quello che vedo quando si "blocchi":

(skipped a lot) 

85:0 ready to enter synchronized block 
85:0 entered synchronized block 
85:2: saw hello 
85:1: saw world 

86:0 ready to enter synchronized block 
86:0 entered synchronized block 
86:2: saw hello 
86:1: saw world 

87:0 ready to enter synchronized block 
............................................ 

.......................................................................... 

.................................................................................. 
(goes "forever") 

Tuttavia, se cambio la notify() chiama all'interno del while (. ..) loop di addWhenHasSpace e getWhenNotEmpty metodi per notifyAll(), passa "sempre".

La mia domanda è questa: perché il comportamento varia tra i metodi notify() e notifyAll() in questo caso, e anche perché il comportamento di notify() è così com'è?

mi aspetterei entrambi i metodi si comportino allo stesso modo, in questo caso (due thread in attesa, uno bloccato), perché:

1. mi sembra che in questo caso notifyAll() sarebbe svegliarsi solo fino l'altro thread, come notify();
2. sembra che la scelta del metodo che risveglia un thread influenzi il modo in cui il thread viene risvegliato (e diventa RUNNABLE suppongo) e il thread principale (che è stato BLOCCATO) più tardi competere per il blocco - non qualcosa che mi aspetterei da javadoc così come la ricerca su internet sull'argomento.

O forse sto facendo qualcosa di sbagliato del tutto?

Answer 1

Sembra esserci una sorta di equità/barging in corso utilizzando il blocco intrinseco - probabilmente a causa di qualche ottimizzazione. Sto indovinando, che il codice nativo controlla per vedere se il thread corrente ha notificato il monitor che sta per aspettare e gli permette di vincere.

Sostituire synchronized con ReentrantLock e dovrebbe funzionare come previsto. Il diverso qui è come il ReentrantLock gestisce i camerieri di un blocco a cui è stata notificata.

---

Aggiornamento:

Interessante trovare qui. Quello che state vedendo è una gara tra la main filo entrare

 synchronized (ctr) { 
      System.out.println(i + ":0 entered synchronized block"); 
      ctr.addWhenHasSpace("hello"); 
      ctr.addWhenHasSpace("world"); 
     } 

mentre gli altri due fili entrare rispettivi synchronized regioni. Se il thread principale non entra nella sua area di sincronizzazione prima di almeno uno dei due, si verificherà questo output di live-lock che si sta descrivendo.

Ciò che sembra accadere è che se entrambi i thread dei consumatori colpiscono prima il blocco di sincronizzazione, eseguiranno ping-pong l'uno con l'altro per notify e wait. Potrebbe essere il caso in cui la JVM fornisce thread in attesa di priorità sul monitor mentre i thread sono bloccati.

Answer 2

Senza guardare troppo nel codice, posso vedere che si sta utilizzando una singola variabile di condizione per implementare una coda con un produttore e più di un consumatore. Questa è una ricetta per i problemi: se c'è una sola variabile di condizione, quando un consumatore chiama notify(), non c'è modo di sapere se riattivare il produttore o riattivare l'altro utente.

ci sono due vie d'uscita da quella trappola: Il più semplice è di usare sempre notifyAll().

L'altro modo è quello di smettere di usare synchronized, wait() e notify(), e invece utilizzare le strutture in java.util.concurrent. serrature.

Un singolo oggetto ReentrantLock può fornire due (o più) variabili di condizione. Usane uno esclusivamente per consentire al produttore di informare i consumatori e utilizzare l'altro esclusivamente per informare i consumatori sul produttore.

Nota: I nomi cambiano quando si passa ad usare ReentrantLocks: o.wait() diventa c.await(), e o.notify() diventa c.signal().