Java 的 wait 和 notify 学习三部曲之一:JVM 源码分析
thread-A : start wait
thread-B : get lock
thread-C : c thread is start
thread-B : start notify
thread-B : release lock
thread-A : after wait, acquire lock again
thread-A : release lock
thread-C : get lock
thread-C : release lock
针对以上结果,问题来了:第一个问题:将以上代码反复执行多次,结果都是 B 释放锁之后 A 会先得到锁,这又是为什么呢?C 为何不能先拿到锁呢?
第二个问题:线程 C 自开始就执行了 monitorenter 指令,它能得到锁是容易理解的,但是线程 A 呢?在 wait()之后并没有没有 monitorenter 指令,那么它又是如何取得锁的呢?
wait()、notify()这些方法都是 native 方法,所以只有从 JVM 源码寻找答案了,本次阅读的是 openjdk8 的源码;
带上问题去看 JVM 源码
按照 demo 代码执行顺序,我整理了如下问题,带着这些问题去看 JVM 源码可以聚焦主线,不要被一些支线的次要的代码卡住(例如一些异常处理,监控和上报等):
线程 A 在 wait()的时候做了什么?
线程 C 启动后,由于此时线程 B 持有锁,那么线程 C 此时在干啥?
线程 B 在 notify()的时候做了什么?
线程 B 释放锁的时候做了什么?
源码中最重要的注释信息
在源码中有段注释堪称是整篇文章最重要的说明,请大家始终记住这段信息,处处都用得上:
ObjectWaiter 对象存在于 WaitSet、EntryList、cxq 等集合中,或者正在这些集合中移动
原文如下:
请务必记住这三个集合:WaitSet、EntryList、cxq
好了,接下来看源码分析问题吧:
线程 A 在 wait()的时候做了什么
打开 hotspot/src/share/vm/runtime/objectMonitor.cpp,看 ObjectMonitor::wait 方法:
如上图所示,有两处代码值得我们注意:
绿框中将当前线程包装成 ObjectWaiter 对象,并且状态为 TS_WAIT,这里对应的是 jstack 看到的线程状态 WAITING;
红框中调用了 AddWaiter 方法,跟进去看下:
这个 ObjectWaiter 对象被放入了 WaitSet 中,WaitSet 是个环形双向链表(circular doubly linked list)
回到 ObjectMonitor::wait 方法接着往下看,会发现关键代码如下图,当前线程通过 park()方法开始挂起(suspend):
至此,我们把 wait()方法要做的事情就理清了:
包装成 ObjectWaiter 对象,状态为 TS_WAIT;
ObjectWaiter 对象被放入_WaitSet 中;
当前线程挂起;
线程 B 持有锁的时候线程 C 在干啥
此时的线程 C 无法进入 synchronized{}代码块,用 jstack 看应该是 BLOCKED 状态,如下图:
我们看看 monitorenter 指令对应的源码吧,位置:openjdk/hotspot/src/share/vm/interpreter/interpreterRuntime.cpp
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
上面的代码有个 if (UseBiasedLocking)判断,是判断是否使用偏向锁的,本例中的锁显然已经不属于当前线程 C 了,所以我们还是直接看 slowenter(hobj, elem->lock(), CHECK)方法吧;
打开 openjdk/hotspot/src/share/vm/runtime/synchronizer.cpp:
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())) {
//如果处于有锁状态,就检查是不是当前线程持有锁,如果是当前线程持有的,就 return,然后就能执行同步代码块中的代码了
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);
}
线程 C 在上面代码中的执行顺序如下:
判断是否是无锁状态,如果是就通过 Atomic::cmpxchg_ptr 去竞争锁;
不是无锁状态,就检查当前锁是否是线程 C 持有;
不是线程 C 持有,调用 inflate 方法开始锁膨胀;
ObjectSynchronizer::inflate(THREAD, obj())->enter(THREAD);
来看看锁膨胀的源码:
![640?wx_fmt=png](https://img-blog.csdni
mg.cn/img_convert/4a5bd29bc6ddc405119e23b0fd389f30.png)
如上图,锁膨胀的代码太长,我们这里只看关键代码吧:红框中,如果当前状态已经是重量级锁,就通过 mark->monitor()方法取得 ObjectMonitor 指针再返回;绿框中,如果还不是重量级锁,就检查是否处于膨胀中状态(其他线程正在膨胀中),如果是膨胀中,就调用 ReadStableMark 方法进行等待,ReadStableMark 方法执行完毕后再通过 continue 继续检查,ReadStableMark 方法中还会调用 os::NakedYield()释放 CPU 资源;
如果红框和绿框的条件都没有命中,目前已经是轻量级锁了(不是重量级锁并且不处于锁膨胀状态),可以开始膨胀了,如下图:
简单来说,锁膨胀就是通过 CAS 将监视器对象 OjectMonitor 的状态设置为 INFLATING,如果 CAS 失败,就在此循环,再走前一副图中的的红框和绿框中的判断,如果 CAS 设置成功,会继续设置 ObjectMonitor 中的 header、owner 等字段,然后 inflate 方法返回监视器对象 OjectMonitor;
看看之前 slow_enter 方法中,调用 inflate 方法的代码如下:
ObjectSynchronizer::inflate(THREAD, obj())->enter(THREAD);
所以 inflate 方法返回监视器对象 OjectMonitor 之后,会立刻执行 OjectMonitor 的 enter 方法,这个方法中开始竞争锁了,方法在 openjdk/hotspot/src/share/vm/runtime/objectMonitor.cpp 文件中:
如上图,红框中表示 OjectMonitor 的 enter 方法一进来就通过 CAS 将 OjectMonitor 的 owner 设置为当前线程,绿框中表示设置成功的逻辑,第一个 if 表示重入锁的逻辑,第二个 if 表示第一次设置 owner 成功,都意味着竞争锁成功,而我们的线程 C 显然是竞争失败的,会进入下图中的无线循环,反复调用 EnterI 方法:
进入 EnterI 方法看看:
如上图,首先构造一个 ObjectWaiter 对象 node,后面的 for(;;)代码块中来是一段非常巧妙的代码,同一时刻可能有多个线程都竞争锁失败走进这个 EnterI 方法,所以在这个 for 循环中,用 CAS 将 cxq 地址放入 node 的 next,也就是把 node 放到 cxq 队列的首位,如果 CAS 失败,就表示其他线程把 node 放入到 cxq 的首位了,所以通过 for 循环再放一次,只要成功,此 node 就一定在最新的_cxq 队列的首位。
接下来的代码又是一个无限循环,如下图:
从上图可以看出,进入循环后先调用 TryLock 方法竞争一次锁,如果成功了就退出循环,否则就调用 Self->_ParkEvent->park 方法使线程挂起,这里有自旋锁的逻辑,也就是 park 方法带了时间参数,就会在挂起一段时间后自动唤醒,如果不是自旋的条件,就一直挂起等待被其他条件唤醒,线程被唤醒后又会执行 TryLock 方法竞争一次锁,竞争不到继续这个 for 循环;
到这里我们已经把线程 C 在 BLOCK 的时候的逻辑理清楚了,小结如下:
偏向锁逻辑,未命中;
如果是无锁状态,就通过 CAS 去竞争锁,此处由于锁已经被线程 B 持有,所以不是无锁状态;
不是无锁状态,而且锁不是线程 C 持有,执行锁膨胀,构造 OjectMonitor 对象;
竞争锁,竞争失败就将线程加入_cxq 队列的首位;
开始无限循环,竞争锁成功就退出循环,竞争失败线程挂起,等待被唤醒后继续竞争;
线程 B 在 notify()的时候做了什么
接下来该线程 B 执行 notify 了,代码是 objectMonitor.cpp 的 ObjectMonitor::notify 方法:
如上图所示,首先是 Policy 的赋值,其次是调用 DequeueWaiter()方法将 WaitSet 队列的第一个值取出并返回,还记得 WaitSet 么?所有 wait 的线程都被包装成 ObjectWaiter 对象然后放进来了;接下来对 ObjectWaiter 对象的处理方式,根据 Policy 的不同而不同:Policy == 0:放入 EntryList 队列的排头位置;Policy == 1:放入 EntryList 队列的末尾位置;Policy == 2:EntryList 队列为空就放入 EntryList,否则放入_cxq 队列的排头位置;
如上图所示,请注意把 ObjectWaiter 的地址写到 cxq 变量的时候要用 CAS 操作,因为此时可能有其他线程正在竞争锁,竞争失败的时候会将自己包装成 ObjectWaiter 对象加入到 cxq 中;
这里的代码有一处疑问,期待着读着您的指教:如果_EntryList 为空,就把 ObjectWaiter 放入 ObjectWaiter 中,为什么要这样做呢?
Policy == 3:放入 cxq 队列中,末尾位置;更新 cxq 变量的值的时候,同样要通过 CAS 注意并发问题;
评论