Basic Of Concurrency(十三: Slipped Conditions)-蒲公英云

什么是Slipped Conditions？

Slipped Conditions是指一个线程对一个确切的条件进行检查到操作期间,如果条件被其他线程访问到的话就会给第一个线程的执行结果造成影响。下面是一个简单的实例：

public class Lock {
    private boolean isLocked = false;
    public void lock() {
        synchronized (this) {
            while (isLocked) {
                try {
                    wait();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }
        synchronized (this) {
            isLocked = true;
        }
    }
    public synchronized void unLock() {
        isLocked = false;
        notify();
    }
}
复制代码

我们可以注意到lock()方法中有两个同步块。第一个同步块会让线程一直等待直到isLocked为false为止。第二个同步块用来设置isLocked为true，以便让当前线程取得Lock实例阻塞其他线程进入临界区。

想象一下当isLocked为false时，有两个线程同时调用lock()方法。如果第一个线程抢先进入到第一个同步块中，它会检查isLocked并发现它为false并退出同步块.如果这时第二个线程刚好被允许执行进入第一个同步块，它同样会检查isLocked并发现它为false。这样两个线程同时读取到条件为false。然后两个线程都会进入到第二个同步块，设置isLocked为true并继续运行。问题在于第二个线程在第一个线程检查和设置isLocked之间的时间点就访问了isLocked.以至于最后两个线程都能退出lock()方法执行到临界区的代码.

这种情况我们称为slipped conditions.所有的线程都能在抢先运行的线程改变条件前访问并检查条件,从而退出同步代码块.换句话说,条件的访问时机被拉长了.条件在被线程改变前,其他线程都能够访问到它.

为了解决Slipped Conditions问题，我们需要让线程以原子的方式来检查和设置条件，这样才能保证线程在执行检查和设置的过程中不会有线程能够访问到条件。

解决上文例子中提到的问题比较简单，只需要将isLocked=true;这一行移动到第一个同步块while()循环下方即可。如下所示：

public class Lock {
    private boolean isLocked = false;
    public void lock() {
        synchronized (this) {
            while (isLocked) {
                try {
                    wait();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            isLocked = true;
        }
    }
    public synchronized void unLock() {
        isLocked = false;
        notify();
    }
}
复制代码

现在我们可以看到isLocked的检查和设置都被放在同一个同步代码块中来保证原子操作。

一个更加完整的实例

也许你觉的你永远不会实现一个像上文中提到的一摸一样的Lock,所以对Slipped Conditions问题的出现存在争议.觉得它只是理论上的问题.但实际它是真实发生的.上文提到的实例是为凸显Slipped Conditions问题而精简设计的. 一个更加完整和真实的实例是实现一个FairLock.FairLock的实现在饥饿与公平一文中有提及.让我们回头看Nested Monitor Lockout问题,在解决它的过程中很容易遇到Slipped Conditons问题.首先我们看一个有nested monitor lockout问题的实例.

public class FairLock {
    private boolean isLocked = false;
    private Thread lockingThread;
    private List<QueueObject> 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 = waitingThreads.get(0);
            synchronized (queueObject) {
                queueObject.notify();
            }
        }
    }
复制代码

我们会注意到synchronized(queueObject)同步块连同同步块中的queueObject.wait()调用都被嵌套在synchronized(this)中.这将产生nested monitor lockout问题.为了解决这个问题,我们需要将synchronized(queueObject)同步代码块在synchronized(this)同步代码块中移除.如下所示:

public class FairLock {
    private boolean isLocked = false;
    private Thread lockingThread;
    private List<QueueObject> 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();
        }
    }
}
复制代码

注意:我们只修改lock()方法,因此我们只看lock()方法的改动即可.

我们可以注意到此时lock()方法中有四个同步代码块.我们可以先忽略以下代码块:

synchronized(this){
      waitingThreads.add(queueObject);
    }
复制代码

除去这里示例的代码块,一共还有3个.

第一个synchronized(this)同步代码块中用于检查mustWait = isLocked || waitingThreads.get(0) != queueObject.表达式的值.
第二个synchronized(queueObject)同步代码块用于检查线程是否需要调用queueObject.wait()以进入等待状态.在此期间上一个线程可能还没有取得FairLock实例.不过我们可以先忽略这一点.现在我们需要关注的是当前FairLock实例还没有被锁住,所以线程可以退出sychronized(queueObject)同步代码块.

第三个synchronized(this)同步代码块只有在mustWait = false时才能被访问到.它将条件isLocked设置回true并且退出lock()方法调用.

当FairLock未被锁住的情况下,有两个线程同时调用lock()方法.首先线程1检查isLocked发现它为false.线程2也是如此.然后两个线程不会进入等待状态而是同时设置isLocked状态为true.这是一个典型的slipped conditions实例.

解决Slipped Conditions问题

为了解决上文实例的slipped conditions问题,我们需要将最后一个synchronized(this)同步块中的内容移动到第二个同步块中去.原先的代码自然需要做出一点小改动来适应这次移动.看起来是这样的:

// 当前FairLock没有nested monitor lockout问题
// 当仍然有信号丢失问题
public class FairLock {
    private boolean isLocked = false;
    private Thread lockingThread;
    private List<QueueObject> 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;
                // 移动代码块，start
                if(!mustWait){
                    waitingThreads.remove(queueObject);
                    isLocked = true;
                    lockingThread = Thread.currentThread();
                    return;
                }
                // 移动代码块，end
            }
            synchronized (queueObject) {
                if (mustWait) {
                    try {
                        queueObject.wait();
                    } catch (InterruptedException e) {
                        waitingThreads.remove(queueObject);
                        throw e;
                    }
                }
            }
        }
    }
}
复制代码

现在我们注意到mustWait条件表达式的检查和设置被放置在同一个同步代码块中.同时需要注意的是,尽管mustWait本地变量在synchronized(this)外部被while(mustWait)当作条件使用,但mustWait的值始终没有在外部被更改.一旦线程解析到mustWait的值为false时,会在同一个原子操作中将mustWait设置为true.以便让其他线程在解析条件表达式时得到true值.

return;语句在synchronized(this)同步块中并不是必须的.这只是一个小优化.如果线程已经知道mustWait=false,则没有必要继续往下执行,进入synchronized (queueObject)同步代码块再去判断mustWait=false后退出.这有点类似快速失败.

如果你善于观察的话,仍然会发现当前FairLock实现会有信号丢失问题(你不看代码也会看注释吧,哈哈…).跟之前的信号丢失问题一样,若mustWait=true.则调用线程将会进入synchronized (queueObject)同步代码块准备调用queueObject.wait();若此时其他线程抢先调用unLock()方法并进入unLock()中的synchronized (queueObject)同步代码块中成功调用了queueObject.notify(),则此调用会失效并且信号丢失,因为在queueObject对象锁上还没有任何线程调用wait()方法等待notify()的唤醒.这样线程lock()方法中的线程在信号丢失后再进入synchronized (queueObject)同步代码调用wait()方法,可能会永远等待下去,除非有其他线程再次调用了当前queueObject的notify()方法.

信号丢失问题已经在之前的饥饿与公平一文中给出了解决方法.只需要将QueueObject.class替换成一个Semaphore.class即可.将queueObject的wait()和notify()方法调用替换成Semaphore的doWait()和doNotify()调用即可.这些调用会将信号存储在Semaphore对象中.这样就算doNotify()在doWait()之前调用了也不会有信号丢失问题.