JAVA 并发编程 - StampedLock 可重入性与 CPU 暴涨问题分析

  • 时间:
  • 分类: JDK,JAVA

前面一篇文章学习了StampedLock的简单使用，后来搜索资料时发现有人说StampedLock是不可重入的锁，于是随便写了个 Demo 验证了一把发现的确如此：

  1. import java.util.concurrent.locks.StampedLock;
  3. public class StampedLockTest {
  5.     public static void main(String[] args) {
  7.         StampedLock sl = new StampedLock();
  8.         long stamp1 = sl.readLock();
  9.         long stamp2 = sl.readLock();
  10.         System.out.println(stamp1 + ".................." + stamp2);
  11.         sl.unlock(stamp1);
  12.         sl.unlock(stamp2);
  14.         stamp1 = sl.writeLock();
  15.         stamp2 = sl.writeLock();
  16.         System.out.println(stamp1 + ".................." + stamp2);
  17.         sl.unlock(stamp1);
  18.         sl.unlock(stamp2);
  19.     }
  20. }

运行以上程序，发现只打印了一次stamp1和stamp2，那是因为StampedLock读锁之间是没有冲突的，所以不回阻塞，而写锁之间是冲突的，但是在同一个线程中，写锁未释放之前再次申请锁是不会成功的，所以stamp2 = sl.writeLock();将会一直等待导致死锁，即StampedLock的读写锁是不可重入的（读锁之间并不冲突），上面程序只验证了writeLock和writeLock是不可重入的，实际上writeLock和readLock也是不可重入的。

还有一个问题就是StampedLock导致 CPU 爆满的问题，有人说是 BUG，查看源码后还不如说它设计就是如此，首先，查看两个关键的方法：

  1. public long readLock() {
  2.     long s = state, next;  // bypass acquireRead on common uncontended case
  3.     return ((whead == wtail && (s & ABITS) < RFULL &&
  4.             U.compareAndSwapLong(this, STATE, s, next = s + RUNIT)) ?
  5.             next : acquireRead(false, 0L));
  6. }
  8. public long writeLock() {
  9.     long s, next;  // bypass acquireWrite in fully unlocked case only
  10.     return ((((s = state) & ABITS) == 0L &&
  11.             U.compareAndSwapLong(this, STATE, s, next = s + WBIT)) ?
  12.             next : acquireWrite(false, 0L));
  13. }

可以看到申请锁的操作在进行一次 CAS 操作失败后都会调用acquireRead(false, 0L)或者acquireWrite(false, 0L)方法去申请锁，继续查看这两个方法的源码：

  1. /**
  2.  * See above for explanation.
  3.  *
  4.  * @param interruptible true if should check interrupts and if so
  5.  * return INTERRUPTED
  6.  * @param deadline if nonzero, the System.nanoTime value to timeout
  7.  * at (and return zero)
  8.  * @return next state, or INTERRUPTED
  9.  */
  10. private long acquireWrite(boolean interruptible, long deadline) {
  11.     StampedLock.WNode node = null, p;
  12.     for (int spins = -1;;) { // spin while enqueuing
  13.         long m, s, ns;
  14.         if ((m = (s = state) & ABITS) == 0L) {
  15.             if (U.compareAndSwapLong(this, STATE, s, ns = s + WBIT))
  16.                 return ns;
  17.         }
  18.         else if (spins < 0)
  19.             spins = (m == WBIT && wtail == whead) ? SPINS : 0;
  20.         else if (spins > 0) {
  21.             if (LockSupport.nextSecondarySeed() >= 0)
  22.                 --spins;
  23.         }
  24.         else if ((p = wtail) == null) { // initialize queue
  25.             StampedLock.WNode hd = new StampedLock.WNode(WMODE, null);
  26.             if (U.compareAndSwapObject(this, WHEAD, null, hd))
  27.                 wtail = hd;
  28.         }
  29.         else if (node == null)
  30.             node = new StampedLock.WNode(WMODE, p);
  31.         else if (node.prev != p)
  32.             node.prev = p;
  33.         else if (U.compareAndSwapObject(this, WTAIL, p, node)) {
  34.             p.next = node;
  35.             break;
  36.         }
  37.     }
  39.     for (int spins = -1;;) {
  40.         StampedLock.WNode h, np, pp; int ps;
  41.         if ((h = whead) == p) {
  42.             if (spins < 0)
  43.                 spins = HEAD_SPINS;
  44.             else if (spins < MAX_HEAD_SPINS)
  45.                 spins <<= 1;
  46.             for (int k = spins;;) { // spin at head
  47.                 long s, ns;
  48.                 if (((s = state) & ABITS) == 0L) {
  49.                     if (U.compareAndSwapLong(this, STATE, s,
  50.                             ns = s + WBIT)) {
  51.                         whead = node;
  52.                         node.prev = null;
  53.                         return ns;
  54.                     }
  55.                 }
  56.                 else if (LockSupport.nextSecondarySeed() >= 0 &&
  57.                         --k <= 0)
  58.                     break;
  59.             }
  60.         }
  61.         else if (h != null) { // help release stale waiters
  62.             StampedLock.WNode c; Thread w;
  63.             while ((c = h.cowait) != null) {
  64.                 if (U.compareAndSwapObject(h, WCOWAIT, c, c.cowait) &&
  65.                         (w = c.thread) != null)
  66.                     U.unpark(w);
  67.             }
  68.         }
  69.         if (whead == h) {
  70.             if ((np = node.prev) != p) {
  71.                 if (np != null)
  72.                     (p = np).next = node;   // stale
  73.             }
  74.             else if ((ps = p.status) == 0)
  75.                 U.compareAndSwapInt(p, WSTATUS, 0, WAITING);
  76.             else if (ps == CANCELLED) {
  77.                 if ((pp = p.prev) != null) {
  78.                     node.prev = pp;
  79.                     pp.next = node;
  80.                 }
  81.             }
  82.             else {
  83.                 long time; // 0 argument to park means no timeout
  84.                 if (deadline == 0L)
  85.                     time = 0L;
  86.                 else if ((time = deadline - System.nanoTime()) <= 0L)
  87.                     return cancelWaiter(node, node, false);
  88.                 Thread wt = Thread.currentThread();
  89.                 U.putObject(wt, PARKBLOCKER, this);
  90.                 node.thread = wt;
  91.                 if (p.status < 0 && (p != h || (state & ABITS) != 0L) &&
  92.                         whead == h && node.prev == p)
  93.                     U.park(false, time);  // emulate LockSupport.park
  94.                 node.thread = null;
  95.                 U.putObject(wt, PARKBLOCKER, null);
  96.                 if (interruptible && Thread.interrupted())
  97.                     return cancelWaiter(node, node, true);
  98.             }
  99.         }
  100.     }
  101. }
  103. /**
  104.  * See above for explanation.
  105.  *
  106.  * @param interruptible true if should check interrupts and if so
  107.  * return INTERRUPTED
  108.  * @param deadline if nonzero, the System.nanoTime value to timeout
  109.  * at (and return zero)
  110.  * @return next state, or INTERRUPTED
  111.  */
  112. private long acquireRead(boolean interruptible, long deadline) {
  113.     StampedLock.WNode node = null, p;
  114.     for (int spins = -1;;) {
  115.         StampedLock.WNode h;
  116.         if ((h = whead) == (p = wtail)) {
  117.             for (long m, s, ns;;) {
  118.                 if ((m = (s = state) & ABITS) < RFULL ?
  119.                         U.compareAndSwapLong(this, STATE, s, ns = s + RUNIT) :
  120.                         (m < WBIT && (ns = tryIncReaderOverflow(s)) != 0L))
  121.                     return ns;
  122.                 else if (m >= WBIT) {
  123.                     if (spins > 0) {
  124.                         if (LockSupport.nextSecondarySeed() >= 0)
  125.                             --spins;
  126.                     }
  127.                     else {
  128.                         if (spins == 0) {
  129.                             StampedLock.WNode nh = whead, np = wtail;
  130.                             if ((nh == h && np == p) || (h = nh) != (p = np))
  131.                                 break;
  132.                         }
  133.                         spins = SPINS;
  134.                     }
  135.                 }
  136.             }
  137.         }
  138.         if (p == null) { // initialize queue
  139.             StampedLock.WNode hd = new StampedLock.WNode(WMODE, null);
  140.             if (U.compareAndSwapObject(this, WHEAD, null, hd))
  141.                 wtail = hd;
  142.         }
  143.         else if (node == null)
  144.             node = new StampedLock.WNode(RMODE, p);
  145.         else if (h == p || p.mode != RMODE) {
  146.             if (node.prev != p)
  147.                 node.prev = p;
  148.             else if (U.compareAndSwapObject(this, WTAIL, p, node)) {
  149.                 p.next = node;
  150.                 break;
  151.             }
  152.         }
  153.         else if (!U.compareAndSwapObject(p, WCOWAIT,
  154.                 node.cowait = p.cowait, node))
  155.             node.cowait = null;
  156.         else {
  157.             for (;;) {
  158.                 StampedLock.WNode pp, c; Thread w;
  159.                 if ((h = whead) != null && (c = h.cowait) != null &&
  160.                         U.compareAndSwapObject(h, WCOWAIT, c, c.cowait) &&
  161.                         (w = c.thread) != null) // help release
  162.                     U.unpark(w);
  163.                 if (h == (pp = p.prev) || h == p || pp == null) {
  164.                     long m, s, ns;
  165.                     do {
  166.                         if ((m = (s = state) & ABITS) < RFULL ?
  167.                                 U.compareAndSwapLong(this, STATE, s,
  168.                                         ns = s + RUNIT) :
  169.                                 (m < WBIT &&
  170.                                         (ns = tryIncReaderOverflow(s)) != 0L))
  171.                             return ns;
  172.                     } while (m < WBIT);
  173.                 }
  174.                 if (whead == h && p.prev == pp) {
  175.                     long time;
  176.                     if (pp == null || h == p || p.status > 0) {
  177.                         node = null; // throw away
  178.                         break;
  179.                     }
  180.                     if (deadline == 0L)
  181.                         time = 0L;
  182.                     else if ((time = deadline - System.nanoTime()) <= 0L)
  183.                         return cancelWaiter(node, p, false);
  184.                     Thread wt = Thread.currentThread();
  185.                     U.putObject(wt, PARKBLOCKER, this);
  186.                     node.thread = wt;
  187.                     if ((h != pp || (state & ABITS) == WBIT) &&
  188.                             whead == h && p.prev == pp)
  189.                         U.park(false, time);
  190.                     node.thread = null;
  191.                     U.putObject(wt, PARKBLOCKER, null);
  192.                     if (interruptible && Thread.interrupted())
  193.                         return cancelWaiter(node, p, true);
  194.                 }
  195.             }
  196.         }
  197.     }
  199.     for (int spins = -1;;) {
  200.         StampedLock.WNode h, np, pp; int ps;
  201.         if ((h = whead) == p) {
  202.             if (spins < 0)
  203.                 spins = HEAD_SPINS;
  204.             else if (spins < MAX_HEAD_SPINS)
  205.                 spins <<= 1;
  206.             for (int k = spins;;) { // spin at head
  207.                 long m, s, ns;
  208.                 if ((m = (s = state) & ABITS) < RFULL ?
  209.                         U.compareAndSwapLong(this, STATE, s, ns = s + RUNIT) :
  210.                         (m < WBIT && (ns = tryIncReaderOverflow(s)) != 0L)) {
  211.                     StampedLock.WNode c; Thread w;
  212.                     whead = node;
  213.                     node.prev = null;
  214.                     while ((c = node.cowait) != null) {
  215.                         if (U.compareAndSwapObject(node, WCOWAIT,
  216.                                 c, c.cowait) &&
  217.                                 (w = c.thread) != null)
  218.                             U.unpark(w);
  219.                     }
  220.                     return ns;
  221.                 }
  222.                 else if (m >= WBIT &&
  223.                         LockSupport.nextSecondarySeed() >= 0 && --k <= 0)
  224.                     break;
  225.             }
  226.         }
  227.         else if (h != null) {
  228.             StampedLock.WNode c; Thread w;
  229.             while ((c = h.cowait) != null) {
  230.                 if (U.compareAndSwapObject(h, WCOWAIT, c, c.cowait) &&
  231.                         (w = c.thread) != null)
  232.                     U.unpark(w);
  233.             }
  234.         }
  235.         if (whead == h) {
  236.             if ((np = node.prev) != p) {
  237.                 if (np != null)
  238.                     (p = np).next = node;   // stale
  239.             }
  240.             else if ((ps = p.status) == 0)
  241.                 U.compareAndSwapInt(p, WSTATUS, 0, WAITING);
  242.             else if (ps == CANCELLED) {
  243.                 if ((pp = p.prev) != null) {
  244.                     node.prev = pp;
  245.                     pp.next = node;
  246.                 }
  247.             }
  248.             else {
  249.                 long time;
  250.                 if (deadline == 0L)
  251.                     time = 0L;
  252.                 else if ((time = deadline - System.nanoTime()) <= 0L)
  253.                     return cancelWaiter(node, node, false);
  254.                 Thread wt = Thread.currentThread();
  255.                 U.putObject(wt, PARKBLOCKER, this);
  256.                 node.thread = wt;
  257.                 if (p.status < 0 &&
  258.                         (p != h || (state & ABITS) == WBIT) &&
  259.                         whead == h && node.prev == p)
  260.                     U.park(false, time);
  261.                 node.thread = null;
  262.                 U.putObject(wt, PARKBLOCKER, null);
  263.                 if (interruptible && Thread.interrupted())
  264.                     return cancelWaiter(node, node, true);
  265.             }
  266.         }
  267.     }
  268. }

可以看到，其内部实现都是两个死循环，在死循环中，在它挂起线程时，使用的是Unsafe.park()函数，而park()函数在遇到线程中断时会直接返回，而不是像Thread.sleep()那样会抛出异常。虽然这两个申请锁的方法第一个参数interruptible表示是否处理中断，但是readLock()与writeLock()中都是传的 false，所以并不会处理中断，也就是代码中if (interruptible && Thread.interrupted())后面的Thread.interrupted()不会执行，中断转态不会清除，那么这样一来就会导致线程在调用park()函数挂起时遇到中断会再次陷入循环，并且中断转态没有清除，后面的循环中调用park()都将立即返回导致无法将线程挂起，所以当退出条件得不到满足条件时，将导致 CPU 暴涨。

下面演示程序模拟了 CPU 暴涨的情况：

  1. import java.util.concurrent.locks.LockSupport;
  2. import java.util.concurrent.locks.StampedLock;
  4. public class StampedLockTest {
  6.     public static void main(String[] args) throws InterruptedException {
  8.         Thread[] threads = new Thread[3];
  9.         StampedLock sl = new StampedLock();
  11.         //启动一个写线程长时间占用锁
  12.         new Thread(() -> {
  13.             long stamp = sl.writeLock();
  14.             LockSupport.parkNanos(60000000000000L);
  15.             sl.unlock(stamp);
  16.         }).start();
  17.         Thread.sleep(100);
  19.         //启动三个读线程
  20.         for (int i = 0; i < 3; i++) {
  21.             threads[i] = new Thread(() -> {
  22.                 long stamp = sl.readLock();
  23.                 System.out.println(Thread.currentThread().getName() + "获得锁.");
  24.                 sl.unlock(stamp);
  25.             });
  26.             threads[i].start();
  27.         }
  28.         //5秒后等到三个线程因写锁被占用导致挂起后进行中断
  29.         Thread.sleep(5000);
  30.         for (int i = 0 ; i < 3; i++) {
  31.             threads[i].interrupt();
  32.         }
  33.     }
  34. }

标签: JAVA 并发编程

发表评论：

icon_mrgreen.gificon_neutral.gificon_twisted.gificon_arrow.gificon_eek.gificon_smile.gificon_confused.gificon_cool.gificon_evil.gificon_biggrin.gificon_idea.gificon_redface.gificon_razz.gificon_rolleyes.gificon_wink.gificon_cry.gificon_surprised.gificon_lol.gificon_mad.gificon_sad.gificon_exclaim.gificon_question.gif