Inside AbstractQueuedSynchronizer (4) (此版比较完整)

原文:

Inside AbstractQueuedSynchronizer (1)

Inside AbstractQueuedSynchronizer (2)

Inside AbstractQueuedSynchronizer (3)

Inside AbstractQueuedSynchronizer (4)

 

3.6 ConditionObject

    AbstractQueuedSynchronizer的内部类ConditionObject实现了Condition接口。Condition接口提供了跟Java语言内置的monitor机制类似的接口：await()/signal()/signalAll()，以及一些支持超时和回退的await版本。可以将任意个数的ConcitionObject关联到对应的synchronizer，例如通过调用ReentrantLock.newCondition()方法即可构造一个ConditionObject实例。每个ConditionObject实例内部都维护一个ConditionQueue，该队列的元素跟AbstractQueuedSynchronizer的WaitQueue一样，都是Node对象。

 

    ConditionObject的await()代码如下：

Java代码  

1. public final void await() throws InterruptedException {  
2.     if (Thread.interrupted())  
3.         throw new InterruptedException();  
4.     Node node = addConditionWaiter();  
5.     int savedState = fullyRelease(node);  
6.     int interruptMode = 0;  
7.     while (!isOnSyncQueue(node)) {  
8.         LockSupport.park(this);  
9.         if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)  
10.             break;  
11.     }  
12.     if (acquireQueued(node, savedState) && interruptMode != THROW_IE)  
13.         interruptMode = REINTERRUPT;  
14.     if (node.nextWaiter != null) // clean up if cancelled  
15.         unlinkCancelledWaiters();  
16.     if (interruptMode != 0)  
17.         reportInterruptAfterWait(interruptMode);  
18. }  

    以上代码中，可以看出ConditionObject的await语义跟Java语言内置的monitor机制是非常相似的（详见：http://whitesock.iteye.com/blog/162344 ）。首先addConditionWaiter()将当前线程加入到ConditionQueue中，然后fullyRelease(node)释放掉跟ConditionObject关联的synchronizer锁。如果某个线程在没有持有对应的synchronizer锁的情况下调用某个ConditionObject对象的await()方法，那么跟Object.wait()一样会抛出IllegalMonitorStateException。接下来while (!isOnSyncQueue(node)) {...}会保证在其它线程调用了该ConditionObject的signal()/siangalAll()之前，当前线程一直被阻塞（signal()/siangalAll()的行为稍后会介绍）。在被signal()/siangalAll()唤醒之后，await()通过acquireQueued(node, savedState)确保再次获得synchronizer的锁。

 

    ConditionObject的signal()代码如下：

Java代码  

1. public final void signal() {  
2.     if (!isHeldExclusively())  
3.         throw new IllegalMonitorStateException();  
4.     Node first = firstWaiter;  
5.     if (first != null)  
6.         doSignal(first);  
7. }  
8.   
9. private void doSignal(Node first) {  
10.     do {  
11.         if ( (firstWaiter = first.nextWaiter) == null)  
12.             lastWaiter = null;  
13.         first.nextWaiter = null;  
14.     } while (!transferForSignal(first) &&  
15.              (first = firstWaiter) != null);  
16. }  

   那么跟await()一样，如果某个线程在没有持有对应的synchronizer锁的情况下调用某个ConditionObject对象的signal()/siangalAll()方法，会抛出IllegalMonitorStateException。signal()主要的行为就是将ConditionQueue中对应的Node实例transfer到AbstractQueuedSynchronizer的WaitQueue中，以便在synchronizer release的过程中，该Node对应的线程可能被唤醒。

 

3.7 Timeout & Cancellation

    AbstractQueuedSynchronizer的acquireQueued()和doAcquire***()系列方法在acquire失败（超时或者中断）后，都会调用cancelAcquire(Node node)方法进行清理，其代码如下：

Java代码  

1. private void cancelAcquire(Node node) {  
2.     // Ignore if node doesn't exist  
3.     if (node == null)  
4.         return;  
5.   
6.     node.thread = null;  
7.   
8.     // Skip cancelled predecessors  
9.     Node pred = node.prev;  
10.     while (pred.waitStatus > 0)  
11.         node.prev = pred = pred.prev;  
12.   
13.     // predNext is the apparent node to unsplice. CASes below will  
14.     // fail if not, in which case, we lost race vs another cancel  
15.     // or signal, so no further action is necessary.  
16.     Node predNext = pred.next;  
17.   
18.     // Can use unconditional write instead of CAS here.  
19.     // After this atomic step, other Nodes can skip past us.  
20.     // Before, we are free of interference from other threads.  
21.     node.waitStatus = Node.CANCELLED;  
22.   
23.     // If we are the tail, remove ourselves.  
24.     if (node == tail && compareAndSetTail(node, pred)) {  
25.         compareAndSetNext(pred, predNext, null);  
26.     } else {  
27.         // If successor needs signal, try to set pred's next-link  
28.         // so it will get one. Otherwise wake it up to propagate.  
29.         int ws;  
30.         if (pred != head &&  
31.             ((ws = pred.waitStatus) == Node.SIGNAL ||  
32.              (ws <= 0 && compareAndSetWaitStatus(pred, ws, Node.SIGNAL))) &&  
33.             pred.thread != null) {  
34.             Node next = node.next;  
35.             if (next != null && next.waitStatus <= 0)  
36.                 compareAndSetNext(pred, predNext, next);  
37.         } else {  
38.             unparkSuccessor(node);  
39.         }  
40.   
41.         node.next = node; // help GC  
42.     }  
43. }  

    需要注意的是， cancelAcquire(Node node)方法是可能会被并发调用。while (pred.waitStatus > 0) {...}这段循环的作用就是清除当前Node之前的已经被标记为取消的节点，但是head节点除外（因为head节点保证不会被标记为Node.CANCELLED）。这段循环初看起来有并发问题，但是推敲一下之后发现：循环过程中函数参数node的waitStatus不会大于0，因此即使是多个线程并发执行这个循环，那么这些线程处理的都只是链表中互不重叠的一部分。接下来在node.waitStatus = Node.CANCELLED执行完毕之后，后续的操作都必须要避免并发问题。

 

    关于处理线程中断， ConditionObject的await()/signal()/signalAll()等方法符合JSR 133: Java Memory Model and Thread Specification Revision中规定的语义：如果中断在signal之前发生，那么await必须在重新获得synchronizer的锁之后，抛出InterruptedException；如果中断发生在signal之后发生，那么await必须要设定当前线程的中断状态，并且不能抛出InterruptedException。

 

4 Reference

The java.util.concurrent Synchronizer Framework

The Art of Multiprocessor Programming