并发编程网 - ifeve.com

一个因中断或者超时的调用可能会引起数据丢失和CPU爆满。

前几天读LinkedTransferQueue（以下简称ltq)的源码，想加深下对松弛型双重队列的理解，无意中发现了这个问题:），经过仔细检查后确认了这是个bug，存在于JDK1.7.0_40和刚发布的JDK8中，去google和oracle官方似乎也没有搜索到这个问题。

重现bug：先来重现下这个bug，由于对并发线程的执行顺序预先不能做任何假设，所以很可能根本就不存在所谓的重现错误的“测试用例”，或者说这个测试用例应该是某种“执行顺序”。所以我一开始的做法是copy了一份ltq的源码，通过某个地方加自旋…但是这种方法毕竟要修改源码，后来我发现直接debug进源码就可以轻易重现bug了。

[code lang=”java”]
LinkedTransferQueue:xfer(E e, boolean haveData, int how, long nanos)

if (how != NOW) { // No matches available
if (s == null)
s = new Node(e, haveData);
Node pred = tryAppend(s, haveData);
if (pred == null)
continue retry; // lost race vs opposite mode
if (how != ASYNC)
return awaitMatch(s, pred, e, (how == TIMED), nanos);
}
return e; // not waiting

[/code]

在以上06行Node pred = tryAppend(s, havaData)  断点（我是windows下用eclipse调试）；
debug以下代码：

[code lang=”java”]
public static void main(String[] args) {
final BlockingQueue<Long> queue = new LinkedTransferQueue<Long>();

Runnable offerTask = new Runnable(){
public void run(){
queue.offer(8L);
System.out.println("offerTask thread has gone!");
}
};
Runnable takeTask = new Runnable(){
public void run(){
try {
System.out.println(Thread.currentThread().getId() + " " +queue.take());
} catch (InterruptedException e) {
e.printStackTrace();
}
}
};
Runnable takeTaskInterrupted = new Runnable(){
public void run(){
Thread.currentThread().interrupt();
try {
System.out.println(Thread.currentThread().getId() + " " +queue.take());
} catch (InterruptedException e) {
System.out.println(e + " "+Thread.currentThread().getId());
}
}
};

new Thread(offerTask).start();
new Thread(takeTask).start();
new Thread(takeTaskInterrupted).start();
}
[/code]

执行到断点处之后，在Debug界面里面有Thread-0、Thread-1、Thread-2三个线程分别指代代码中的offerTask、takeTask、takeTaskInterrupted三者。现在执行三步：

step 1: Resume Thread-1（没有输出，线程Thread-1自己挂起，等待数据）
step 2: Resume Thread-2（看到类似于 java.lang.InterruptedException 15 的输出）
step 3: Resume Thread-0（输出：offerTask thread has gone!）

offer线程已经执行完毕，然后我们的64L呢，明明Thread-1在等待数据，数据丢失了吗？其实不是，只不过take线程现在无法取得offer线程提交的数据了。

如果你觉得上面的数据丢失还不是什么大问题请在上面的示例下添加如下代码（和你CPU核心数相同的代码行:）

[code lang=”java”]
…………..

new Thread(takeTask).start();
new Thread(takeTask).start();
new Thread(takeTask).start();
new Thread(takeTask).start();

[/code]

把上面的3个step重新按顺序执行一遍，建议先打开任务管理器，接着忽略断点，让接下来这几个线程跑:）
CPU爆满了吧…其实是被这几个线程占据了，你去掉几行代码，CPU使用率会有相应的调整。
所以这个bug可能会引起数据暂时遗失和CPU爆满， 只不过貌似发生这种情况的概率极低。

原因：为什么会出现这个bug呢，要想了解原因必须先深入分析ltq内部所使用的数据结构和并发策略，ltq内部采用的是一种非常不同的队列，即松弛型双重队列（Dual Queues with Slack）。

数据结构：

松弛的意思是说，它的head和tail节点相较于其他并发列队要求上更放松，构造它的目的是减少CAS操作的次数（相应的会增加next域的引用次数），举个例子：某个瞬间tail指向的节点后面已经有6个节点了（以下图借用源码的注释），而其他并发队列真正的尾节点最多只能是tail的下一个节点。

* head                      tail
* |                               |
* v                             v
* M -> M -> U -> U -> U -> U->U->U->U->U

收缩的方式是大部分情况下不会用tail的next来设置tail节点，而是第一次收缩N个next(N>=2)，然后查看能否2个一次来收缩tail。（head类似，并且head改变一次会导致前“head”节点的next域断裂即如下图）

*”prehead”                head                 tail
*     |                              |                      |
*     v                            v                      v
*    M      M-> U -> U -> U -> U->U->U->U->U

双重是指有两种类型相互对立的节点(Node.isData==false || true)，并且我理解的每种节点都有三种状态：

1  INIT(节点构造完成，刚进入队列的状态)

2 MATCHED(节点备置为“满足”状态，即入队节点标识的线程成功取得或者传递了数据)

3 CANCELED(节点被置为取消状态，即入队节点标识的线程因为超时或者中断决定放弃等待）

（bug的原因就是现有代码中将2、3都当做MATCHED处理，后面会看到把3独立出来就修复了这个问题）

并发策略：

既然使用了松弛的双重队列，那么当take、offer等方法被调用时执行的策略也稍微不同。

就我们示例中的代码的流程来看，Thread-0、Thread-1、Thread-2几乎同时进入到了xfer的调用，发现队列为空，所以都构造了自己的node希望入队，于是三者都从tail开始加入自己的node，我们在这里的顺序是Thread-1->Thread-2->Thread-0，因为想要入队还要和当前的tail节点进行匹配得到“认可”才能尝试入队，队列为空Thread-1理所当然入队成功并且挂起了自己的线程（park)等待相对的调用来唤醒自己（unpark)，然后Thread-2发现队列末尾的node和自己是同一类型的，于是通过了测试把自己也加入了队列，由于本身是中断的所以让自己进入MATCHED状态（bug就是这里了，上面说过CANCEL被当做MATCHED状态处理），接着我们提交数据的Thread-0来了，发现末尾节点的类型虽然对立但却是MATCHED状态（假如不为MATCHED会有退回再从head来探测一次的机会），所以认为队列已经为空，前面的调用已经被匹配完了，然后把自己的node入队，这样就形成了如下所示的场景：

*                                            Thread-1         Thread-2          Thread-0
*                                                 |                        |                         |
*                                                 v                       v                        v
*                                       REQUEST ->     MATCHED   ->         DATA

好了， 现在Thread-3来了，先探测尾部发现Thread-0的node是类型相反的，于是退回从头部开始重新探测，但是又发现Thread-1的node的类型是相同的，于是再次去探测尾部看看能否入队…….结果造成CPU是停不下来的。

修复：

如上面所说，错误的本质在于当尾部的节点是CANCELED（取消）状态时不能作为被匹配完成的MATCHED状态处理，应该让后来者回退到head去重新测试一次所以重点是对源码做出如下修改（修改放在注释中）：

[code lang=”java”]

static final class Node {
final boolean isData; // false if this is a request node
volatile Object item; // initially non-null if isData; CASed to match
volatile Node next;
volatile Thread waiter; // null until waiting

/*

static final Object CANCEL = new Object();

final void forgetWaiter(){
UNSAFE.putObject(this, waiterOffset, null);
}

final boolean isCanceled(){
return item == CANCEL;
}

*/

[/code]

在Node节点代码中加入标识取消的对象CANCEL。

[code lang=”java”]

private E xfer(E e, boolean haveData, int how, long nanos) {

if (item != p && (item != null) == isData  /*&& item!=Node.CANCEL*/) { // unmatched
if (isData == haveData) // can’t match

[/code]

在xfer函数中添加对于为状态为取消的判断。

[code lang=”java”]

private E xfer(E e, boolean haveData, int how, long nanos) {

Node pred = tryAppend(/*s,*/ haveData);

…..

}

private Node tryAppend(Node s, boolean haveData) {

else if (p.cannotPrecede(/*s, */haveData))

else {
/* if(p.isCanceled())
p.forgetContents();*/
if (p != t) { // update if slack now >= 2

[/code]

添加对于前置节点为取消状态时当前节点的入队策略

[code lang=”java”]

final boolean cannotPrecede(boolean haveData) {
boolean d = isData;
Object x;
return d != haveData && (x = item) != this && (x != null) == d;
}

final boolean cannotPrecede(Node node, boolean haveData) {
boolean d = isData;
if(d != haveData){
Object x = item;
if(x != this && (x!=null) == d && x!= Node.CANCEL)
return true;
if(item == CANCEL){
if(node.next != this){
node.next = this;
return true;
}
this.forgetContents();
}
}
node.next = null;
return false;
}

[/code]

这一步是关键， 当我们入队时发现前置节点是类型对立并且取消状态时，我们就需要多一次的回退探测，所以借用了一下next域来标识这个CANCEL节点，下次经过时或者可以确认它可以当做MATCHED处理（它前面没有INIT节点）或者已经有别的节点粘接在它后面，我们就进而处理那个节点，总之当我们总是能够得到正确的行为。

[code lang=”java”]

private E awaitMatch(Node s, Node pred, E e, boolean timed, long nanos) {

if ((w.isInterrupted() || (timed && nanos <= 0)) &&
s.casItem(e, /*Node.CANCEL*/)) { // cancel
unsplice(pred, s);
return e;
}

[/code]

这一处关键点把item的值从原来的s本身修改为我们新增的CANCEL。

额  代码好乱，关于这个bug定位应该没问题，后面的原因很多方面都没讲，剩下的还有很多处大大小小的修改=_=，整个修改之后的LinkedTransferQueue在github上，大家有兴趣的话可以参考下，已经通过了  JSR166测试套件。