再谈阻塞(2)：获取重量级锁从尝试到失败

发表于 | 分类于 | 阅读次数:
字数统计: 1,948 | 阅读时长 ≈ 9

本篇将从源码角度，了解竞争重量级锁失败的过程，解释了为什么t2启动后，立马调用getState方法，会显示处于RUNNABLE状态，也说明了在Java线程中的Blocked状态并非是完全挂起状态，入队_cxq前后都可能进行TryLock()和自旋。

竞争重量级锁之前

稍微熟悉synchronized就会知道，Java编译器在编译带有该关键字的代码块后，会插入monitorenter以及monitorexit指令到字节码中，monitorenter也就是进入内置锁机制的入口，看看虚拟机内部的部分实现：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
    ...
// traditional lightweight locking
if (!success) {
  markOop displaced = lockee->mark()->set_unlocked();
  entry->lock()->set_displaced_header(displaced);
  bool call_vm = UseHeavyMonitors;
  if (call_vm || Atomic::cmpxchg_ptr(entry, lockee->mark_addr(), displaced) != displaced) {
    // Is it simple recursive case?
    if (!call_vm && THREAD->is_lock_owned((address) displaced->clear_lock_bits())) {
      entry->lock()->set_displaced_header(NULL);
    } else {
      CALL_VM(InterpreterRuntime::monitorenter(THREAD, entry), handle_exception);
    }
  }
}
    ...

为了聚焦问题，所以上一篇已经强调过本系列主要研究的对象是重量级锁，获取轻量级锁失败后会去调用InterpreterRuntime::monitorenter方法，继续追踪有：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
//%note monitor_1
IRT_ENTRY_NO_ASYNC(void, InterpreterRuntime::monitorenter(JavaThread* thread, BasicObjectLock* elem))
#ifdef ASSERT
  thread->last_frame().interpreter_frame_verify_monitor(elem);
#endif
  if (PrintBiasedLockingStatistics) {
    Atomic::inc(BiasedLocking::slow_path_entry_count_addr());
  }
  Handle h_obj(thread, elem->obj());
  assert(Universe::heap()->is_in_reserved_or_null(h_obj()),
         "must be NULL or an object");
  if (UseBiasedLocking) {
    // Retry fast entry if bias is revoked to avoid unnecessary inflation
    ObjectSynchronizer::fast_enter(h_obj, elem->lock(), true, CHECK);
  } else {
    ObjectSynchronizer::slow_enter(h_obj, elem->lock(), CHECK);
  }
  assert(Universe::heap()->is_in_reserved_or_null(elem->obj()),
         "must be NULL or an object");
#ifdef ASSERT
  thread->last_frame().interpreter_frame_verify_monitor(elem);
#endif
IRT_END

fast_enter方法的作用注释已经充分的说明：

Retry fast entry if bias is revoked to avoid unnecessary inflation

所以主要看slow_enter方法，毕竟本人关注的是获取重量级锁的机制：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
void ObjectSynchronizer::slow_enter(Handle obj, BasicLock* lock, TRAPS) {
  markOop mark = obj->mark();
  assert(!mark->has_bias_pattern(), "should not see bias pattern here");
  if (mark->is_neutral()) {
    // Anticipate successful CAS -- the ST of the displaced mark must
    // be visible <= the ST performed by the CAS.
    lock->set_displaced_header(mark);
    if (mark == (markOop) Atomic::cmpxchg_ptr(lock, obj()->mark_addr(), mark)) {
      TEVENT (slow_enter: release stacklock) ;
      return ;
    }
    // Fall through to inflate() ...
  } else
  if (mark->has_locker() && THREAD->is_lock_owned((address)mark->locker())) {
    assert(lock != mark->locker(), "must not re-lock the same lock");
    assert(lock != (BasicLock*)obj->mark(), "don't relock with same BasicLock");
    lock->set_displaced_header(NULL);
    return;
  }
#if 0
  // The following optimization isn't particularly useful.
  if (mark->has_monitor() && mark->monitor()->is_entered(THREAD)) {
    lock->set_displaced_header (NULL) ;
    return ;
  }
#endif
  // The object header will never be displaced to this lock,
  // so it does not matter what the value is, except that it
  // must be non-zero to avoid looking like a re-entrant lock,
  // and must not look locked either.
  lock->set_displaced_header(markOopDesc::unused_mark());
  ObjectSynchronizer::inflate(THREAD, obj())->enter(THREAD);
}

可见在正式竞争重量级锁之前，还会有一个锁膨胀的过程，当调用enter方法后，就正式开始竞争重量级锁了。

本小节跳过了无锁、偏向锁、轻量级锁以及锁膨胀的实现，但在追踪代码的过程中，对竞争重量级锁之前的流程有了大致的了解。

t2的失败之旅

延用前篇最后列举的例子，看看t2是如何获取锁未遂的。

ObjectMonitor::enter()

ObjectMonitor::enter()可以说是真正开始竞争重量级锁的入口，当t2启动后尝试去拿锁，此时t1持有锁并处在操作系统的IO阻塞状态，这是一个竞争重量级锁的过程(在此之前会通过一系列判断后膨胀锁)，ObjectMonitor::enter一开始进行是否重入的检查：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
Thread * const Self = THREAD ;
void * cur ;
cur = Atomic::cmpxchg_ptr (Self, &_owner, NULL) ;
if (cur == NULL) {
   // Either ASSERT _recursions == 0 or explicitly set _recursions = 0.
   assert (_recursions == 0   , "invariant") ;
   assert (_owner      == Self, "invariant") ;
   // CONSIDER: set or assert OwnerIsThread == 1
   return ;
}
if (cur == Self) {
   // TODO-FIXME: check for integer overflow!  BUGID 6557169.
   _recursions ++ ;
   return ;
}

cmpxchg_ptr (Self, &_owner, NULL)的规则：

  • &_owner和NULL相等，则将Self指向的内容赋给&_owner，返回NULL；
  • &_owner和NULL不相等，则返回_owner；

第一个if语句用来处理CAS返回NULL的情形，此情形说明_owner指向的内容为空，没有线程持有锁，CAS将_owner指向当前线程，进行断言判断后直接返回；

第二个if语句用来处理其它的情形，&_owner和NULL不相等会返回_owner，此时如果_owner == Self，说明线程重入，那么重入游标_recursions将自增。

接下来的代码涉及到之前锁膨胀的内容：

1
2
3
4
5
6
7
8
9
if (Self->is_lock_owned ((address)cur)) {
  assert (_recursions == 0, "internal state error");
  _recursions = 1 ;
  // Commute owner from a thread-specific on-stack BasicLockObject address to
  // a full-fledged "Thread *".
  _owner = Self ;
  OwnerIsThread = 1 ;
  return ;
}

之前已经说明，竞争重量级锁之前的“故事”暂不深入。这里大致思想应该是轻量级锁膨胀为重量级锁后，若获得了轻量级锁又再次获得重量级锁，还是相当于锁重入了，所以_recursions设置为1，表示重入1次。

紧接着有：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
// We've encountered genuine contention.
assert (Self->_Stalled == 0, "invariant") ;
// 设置线程的_Stalled字段，自旋模式下的一个设定；
Self->_Stalled = intptr_t(this) ;
// Try one round of spinning *before* enqueueing Self
// and before going through the awkward and expensive state
// transitions.  The following spin is strictly optional ...
// Note that if we acquire the monitor from an initial spin
// we forgo posting JVMTI events and firing DTRACE probes.
// 设置Knob_SpinEarly来进行优化，
// 采用TrySpin方法自旋；
if (Knob_SpinEarly && TrySpin (Self) > 0) {
   assert (_owner == Self      , "invariant") ;
   assert (_recursions == 0    , "invariant") ;
   assert (((oop)(object()))->mark() == markOopDesc::encode(this), "invariant") ;
   Self->_Stalled = 0 ;
   return ;
}

自旋拿锁失败，可用断言验证此时Java线程的状态不为blocked状态：

1
assert (jt->thread_state() != _thread_blocked   , "invariant") ;

所以t2启动后一开始处于runnable状态。

1
2
// Change java thread status to indicate blocked on monitor enter.
    JavaThreadBlockedOnMonitorEnterState jtbmes(jt, this);

上面的代码将线程状态设置为blocked状态，此时没有拿到monitor锁，也没有在任何队列，不会突然的被JVMTI_EVENT_MONITOR_CONTENDED_ENTER事件unpark()。

后续会出现for循环：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
for (;;) {
  jt->set_suspend_equivalent();
  // cleared by handle_special_suspend_equivalent_condition()
  // or java_suspend_self()
  EnterI (THREAD) ;
  if (!ExitSuspendEquivalent(jt)) break ;
  //
  // We have acquired the contended monitor, but while we were
  // waiting another thread suspended us. We don't want to enter
  // the monitor while suspended because that would surprise the
  // thread that suspended us.
  //
      _recursions = 0 ;
  _succ = NULL ;
  exit (false, Self) ;
  jt->java_suspend_self();
}

这里先关注EnterI (THREAD)。

ObjectMonitor::EnterI()

该方法是最终没有拿到锁的线程入队的关键方法，但在入队前后，还会不停尝试获取锁。

进入该方法首先会断言判断线程状态为_thread_blocked，所以进入该方法时的状态确定是blocked状态，接着还会TryLock()和一轮TrySpin()。在通过断言确定没有拿到锁之后，开始入队操作：

1
2
3
4
ObjectWaiter node(Self) ;
Self->_ParkEvent->reset() ;
node._prev   = (ObjectWaiter *) 0xBAD ;
node.TState  = ObjectWaiter::TS_CXQ ;

可见，方法将线程包装成一个ObjectWaiter对象，继续下面的代码：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
// Push "Self" onto the front of the _cxq.
// Once on cxq/EntryList, Self stays on-queue until it acquires the lock.
// Note that spinning tends to reduce the rate at which threads
// enqueue and dequeue on EntryList|cxq.
ObjectWaiter * nxt ;
for (;;) {
    node._next = nxt = _cxq ;
    if (Atomic::cmpxchg_ptr (&node, &_cxq, nxt) == nxt) break ;
    // Interference - the CAS failed because _cxq changed.  Just retry.
    // As an optional optimization we retry the lock.
    if (TryLock (Self) > 0) {
        assert (_succ != Self         , "invariant") ;
        assert (_owner == Self        , "invariant") ;
        assert (_Responsible != Self  , "invariant") ;
        return ;
    }
}

cmpxchg_ptr()之前已经解析过了，在这里，当_cxq指向内容和nxt的相等时，node就赋给_cxq，完成入队并跳出循环；否则继续TryLock()，没拿到锁就循环再次尝试入队。将关键代码再单独拿出来：

1
2
node._next = nxt = _cxq ;
if (Atomic::cmpxchg_ptr (&node, &_cxq, nxt) == nxt) break ;

这里的逻辑表明，当有新来的node时，它的next总是指向当前头节点，而当cmpxchg_ptr成功后，头节点指针_cxq就会指向新来的node，所以后来者当头。

此时如若仍然没有获得锁，将进入最后的一个for循环中，这里列出主要代码：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
for (;;) {
  if (TryLock (Self) > 0) break ;
  assert (_owner != Self, "invariant") ;
  if ((SyncFlags & 2) && _Responsible == NULL) {
     Atomic::cmpxchg_ptr (Self, &_Responsible, NULL) ;
  }
  // park self
  if (_Responsible == Self || (SyncFlags & 1)) {
      TEVENT (Inflated enter - park TIMED) ;
      Self->_ParkEvent->park ((jlong) RecheckInterval) ;
      // Increase the RecheckInterval, but clamp the value.
      RecheckInterval *= 8 ;
      if (RecheckInterval > 1000) RecheckInterval = 1000 ;
  } else {
      TEVENT (Inflated enter - park UNTIMED) ;
      Self->_ParkEvent->park() ;
  }
  if (TryLock(Self) > 0) break ;
  ...
  if ((Knob_SpinAfterFutile & 1) && TrySpin (Self) > 0) break ;
  ...
  OrderAccess::fence() ;
}

代码中，每次循环中会进行数次TryLock()尝试拿锁，期间会执行Self->\_ParkEvent->park() ;将自己挂起。

小结

通过本文可以知道一个线程在拿锁失败的过程中是如何切换Java状态的。在解读源码的过程中，弄清楚了cmpxchg_ptr()的比较规则，而后续文章将深入队列的唤醒机制。