Slipped Conditions

所謂 Slipped conditions，就是說(shuō)， 從一個(gè)線程檢查某一特定條件到該線程操作此條件期間，這個(gè)條件已經(jīng)被其它線程改變，導(dǎo)致第一個(gè)線程在該條件上執(zhí)行了錯(cuò)誤的操作。這里有一個(gè)簡(jiǎn)單的例子：

public class Lock {
    private boolean isLocked = true;

    public void lock(){
      synchronized(this){
        while(isLocked){
          try{
            this.wait();
          } catch(InterruptedException e){
            //do nothing, keep waiting
          }
        }
      }

      synchronized(this){
        isLocked = true;
      }
    }

    public synchronized void unlock(){
      isLocked = false;
      this.notify();
    }
}

我們可以看到，lock()方法包含了兩個(gè)同步塊。第一個(gè)同步塊執(zhí)行 wait 操作直到 isLocked 變?yōu)?false 才退出，第二個(gè)同步塊將 isLocked 置為 true，以此來(lái)鎖住這個(gè) Lock 實(shí)例避免其它線程通過(guò) lock()方法。

我們可以設(shè)想一下，假如在某個(gè)時(shí)刻 isLocked 為 false， 這個(gè)時(shí)候，有兩個(gè)線程同時(shí)訪問(wèn) lock 方法。如果第一個(gè)線程先進(jìn)入第一個(gè)同步塊，這個(gè)時(shí)候它會(huì)發(fā)現(xiàn) isLocked 為 false，若此時(shí)允許第二個(gè)線程執(zhí)行，它也進(jìn)入第一個(gè)同步塊，同樣發(fā)現(xiàn) isLocked 是 false。現(xiàn)在兩個(gè)線程都檢查了這個(gè)條件為 false，然后它們都會(huì)繼續(xù)進(jìn)入第二個(gè)同步塊中并設(shè)置 isLocked 為 true。

這個(gè)場(chǎng)景就是 slipped conditions 的例子，兩個(gè)線程檢查同一個(gè)條件， 然后退出同步塊，因此在這兩個(gè)線程改變條件之前，就允許其它線程來(lái)檢查這個(gè)條件。換句話說(shuō)，條件被某個(gè)線程檢查到該條件被此線程改變期間，這個(gè)條件已經(jīng)被其它線程改變過(guò)了。

為避免 slipped conditions，條件的檢查與設(shè)置必須是原子的，也就是說(shuō)，在第一個(gè)線程檢查和設(shè)置條件期間，不會(huì)有其它線程檢查這個(gè)條件。

解決上面問(wèn)題的方法很簡(jiǎn)單，只是簡(jiǎn)單的把 isLocked = true 這行代碼移到第一個(gè)同步塊中，放在 while 循環(huán)后面即可：

public class Lock {
    private boolean isLocked = true;

    public void lock(){
      synchronized(this){
        while(isLocked){
          try{
            this.wait();
          } catch(InterruptedException e){
            //do nothing, keep waiting
          }
        }
        isLocked = true;
      }
    }

    public synchronized void unlock(){
      isLocked = false;
      this.notify();
    }
}

現(xiàn)在檢查和設(shè)置 isLocked 條件是在同一個(gè)同步塊中原子地執(zhí)行了。

一個(gè)更現(xiàn)實(shí)的例子

也許你會(huì)說(shuō)，我才不可能寫這么挫的代碼，還覺(jué)得 slipped conditions 是個(gè)相當(dāng)理論的問(wèn)題。但是第一個(gè)簡(jiǎn)單的例子只是用來(lái)更好的展示 slipped conditions。

饑餓和公平中實(shí)現(xiàn)的公平鎖也許是個(gè)更現(xiàn)實(shí)的例子。再看下嵌套管程鎖死中那個(gè)幼稚的實(shí)現(xiàn)，如果我們?cè)噲D解決其中的嵌套管程鎖死問(wèn)題，很容易產(chǎn)生 slipped conditions 問(wèn)題。 首先讓我們看下嵌套管程鎖死中的例子：

//Fair Lock implementation with nested monitor lockout problem
public class FairLock {
  private boolean isLocked = false;
  private Thread lockingThread = null;
  private List waitingThreads =
            new ArrayList();

  public void lock() throws InterruptedException{
    QueueObject queueObject = new QueueObject();

    synchronized(this){
      waitingThreads.add(queueObject);

      while(isLocked || waitingThreads.get(0) != queueObject){

        synchronized(queueObject){
          try{
            queueObject.wait();
          }catch(InterruptedException e){
            waitingThreads.remove(queueObject);
            throw e;
          }
        }
      }
      waitingThreads.remove(queueObject);
      isLocked = true;
      lockingThread = Thread.currentThread();
    }
  }

  public synchronized void unlock(){
    if(this.lockingThread != Thread.currentThread()){
      throw new IllegalMonitorStateException(
        "Calling thread has not locked this lock");
    }
    isLocked      = false;
    lockingThread = null;
    if(waitingThreads.size() > 0){
      QueueObject queueObject = waitingThread.get(0);
      synchronized(queueObject){
        queueObject.notify();
      }
    }
  }
}
public class QueueObject {}

我們可以看到 synchronized(queueObject)及其中的 queueObject.wait()調(diào)用是嵌在 synchronized(this)塊里面的，這會(huì)導(dǎo)致嵌套管程鎖死問(wèn)題。為避免這個(gè)問(wèn)題，我們必須將 synchronized(queueObject)塊移出 synchronized(this)塊。移出來(lái)之后的代碼可能是這樣的：

//Fair Lock implementation with slipped conditions problem
public class FairLock {
  private boolean isLocked = false;
  private Thread lockingThread  = null;
  private List waitingThreads =
            new ArrayList();

  public void lock() throws InterruptedException{
    QueueObject queueObject = new QueueObject();

    synchronized(this){
      waitingThreads.add(queueObject);
    }

    boolean mustWait = true;
    while(mustWait){

      synchronized(this){
        mustWait = isLocked || waitingThreads.get(0) != queueObject;
      }

      synchronized(queueObject){
        if(mustWait){
          try{
            queueObject.wait();
          }catch(InterruptedException e){
            waitingThreads.remove(queueObject);
            throw e;
          }
        }
      }
    }

    synchronized(this){
      waitingThreads.remove(queueObject);
      isLocked = true;
      lockingThread = Thread.currentThread();
    }
  }
}

注意：因?yàn)槲抑桓膭?dòng)了 lock()方法，這里只展現(xiàn)了 lock 方法。

現(xiàn)在 lock()方法包含了 3 個(gè)同步塊。

第一個(gè)，synchronized(this)塊通過(guò) mustWait = isLocked || waitingThreads.get(0) != queueObject 檢查內(nèi)部變量的值。

第二個(gè)，synchronized(queueObject)塊檢查線程是否需要等待。也有可能其它線程在這個(gè)時(shí)候已經(jīng)解鎖了，但我們暫時(shí)不考慮這個(gè)問(wèn)題。我們就假設(shè)這個(gè)鎖處在解鎖狀態(tài)，所以線程會(huì)立馬退出 synchronized(queueObject)塊。

第三個(gè)，synchronized(this)塊只會(huì)在 mustWait 為 false 的時(shí)候執(zhí)行。它將 isLocked 重新設(shè)回 true，然后離開(kāi) lock()方法。

設(shè)想一下，在鎖處于解鎖狀態(tài)時(shí)，如果有兩個(gè)線程同時(shí)調(diào)用 lock()方法會(huì)發(fā)生什么。首先，線程 1 會(huì)檢查到 isLocked 為 false，然后線程 2 同樣檢查到 isLocked 為 false。接著，它們都不會(huì)等待，都會(huì)去設(shè)置 isLocked 為 true。這就是 slipped conditions 的一個(gè)最好的例子。

解決 Slipped Conditions 問(wèn)題

要解決上面例子中的 slipped conditions 問(wèn)題，最后一個(gè) synchronized(this)塊中的代碼必須向上移到第一個(gè)同步塊中。為適應(yīng)這種變動(dòng)，代碼需要做點(diǎn)小改動(dòng)。下面是改動(dòng)過(guò)的代碼：

//Fair Lock implementation without nested monitor lockout problem,
//but with missed signals problem.
public class FairLock {
  private boolean isLocked = false;
  private Thread lockingThread  = null;
  private List waitingThreads =
            new ArrayList();

  public void lock() throws InterruptedException{
    QueueObject queueObject = new QueueObject();

    synchronized(this){
      waitingThreads.add(queueObject);
    }

    boolean mustWait = true;
    while(mustWait){
      synchronized(this){
        mustWait = isLocked || waitingThreads.get(0) != queueObject;
        if(!mustWait){
          waitingThreads.remove(queueObject);
          isLocked = true;
          lockingThread = Thread.currentThread();
          return;
        }
      }     

      synchronized(queueObject){
        if(mustWait){
          try{
            queueObject.wait();
          }catch(InterruptedException e){
            waitingThreads.remove(queueObject);
            throw e;
          }
        }
      }
    }
  }
}

我們可以看到對(duì)局部變量 mustWait 的檢查與賦值是在同一個(gè)同步塊中完成的。還可以看到，即使在 synchronized(this)塊外面檢查了 mustWait，在 while(mustWait)子句中，mustWait 變量從來(lái)沒(méi)有在 synchronized(this)同步塊外被賦值。當(dāng)一個(gè)線程檢查到 mustWait 是 false 的時(shí)候，它將自動(dòng)設(shè)置內(nèi)部的條件（isLocked），所以其它線程再來(lái)檢查這個(gè)條件的時(shí)候，它們就會(huì)發(fā)現(xiàn)這個(gè)條件的值現(xiàn)在為 true 了。

synchronized(this)塊中的 return;語(yǔ)句不是必須的。這只是個(gè)小小的優(yōu)化。如果一個(gè)線程肯定不會(huì)等待（即 mustWait 為 false），那么就沒(méi)必要讓它進(jìn)入到 synchronized(queueObject)同步塊中和執(zhí)行 if(mustWait)子句了。

細(xì)心的讀者可能會(huì)注意到上面的公平鎖實(shí)現(xiàn)仍然有可能丟失信號(hào)。設(shè)想一下，當(dāng)該 FairLock 實(shí)例處于鎖定狀態(tài)時(shí)，有個(gè)線程來(lái)調(diào)用 lock()方法。執(zhí)行完第一個(gè) synchronized(this)塊后，mustWait 變量的值為 true。再設(shè)想一下調(diào)用 lock()的線程是通過(guò)搶占式的，擁有鎖的那個(gè)線程那個(gè)線程此時(shí)調(diào)用了 unlock()方法，但是看下之前的 unlock()的實(shí)現(xiàn)你會(huì)發(fā)現(xiàn)，它調(diào)用了 queueObject.notify()。但是，因?yàn)?lock()中的線程還沒(méi)有來(lái)得及調(diào)用 queueObject.wait()，所以 queueObject.notify()調(diào)用也就沒(méi)有作用了，信號(hào)就丟失掉了。如果調(diào)用 lock()的線程在另一個(gè)線程調(diào)用 queueObject.notify()之后調(diào)用 queueObject.wait()，這個(gè)線程會(huì)一直阻塞到其它線程調(diào)用 unlock 方法為止，但這永遠(yuǎn)也不會(huì)發(fā)生。

公平鎖實(shí)現(xiàn)的信號(hào)丟失問(wèn)題在饑餓和公平一文中我們已有過(guò)討論，把 QueueObject 轉(zhuǎn)變成一個(gè)信號(hào)量，并提供兩個(gè)方法：doWait()和 doNotify()。這些方法會(huì)在 QueueObject 內(nèi)部對(duì)信號(hào)進(jìn)行存儲(chǔ)和響應(yīng)。用這種方式，即使 doNotify()在 doWait()之前調(diào)用，信號(hào)也不會(huì)丟失。