Thread类源码(2)
第1节 JVM中thread.cpp
JVM中thread.cpp几个属性如下:
/** JVM源码对中断做个更加清晰的剖析 */
// JVM中thread.hpp源码:
// http://hg.openjdk.java.net/jdk8u/jdk8u/hotspot/file/5aa3d728164a/src/share/vm/runtime/thread.cpp
//用于synchronized同步块和Object.wait()
ParkEvent * _ParkEvent ;
//用于Thread.sleep()
ParkEvent * _SleepEvent ;
//用于unsafe.park()/unpark(),供java.util.concurrent.locks.LockSupport调用,
//因此它支持了java.util.concurrent的各种锁、条件变量等线程同步操作,是concurrent的实现基础
Parker* _parker;
源码地址如下:
http://hg.openjdk.java.net/jdk8u/jdk8u/hotspot/file/5aa3d728164a/src/share/vm/runtime/thread.cpp
第2节 interrupt方法源码
interrupt方法的jvm源码入口在jvm.cpp文件:
JVM_ENTRY(void, JVM_Interrupt(JNIEnv* env, jobject jthread))
JVMWrapper("JVM_Interrupt");
// Ensure that the C++ Thread and OSThread structures aren't freed before we operate
oop java_thread = JNIHandles::resolve_non_null(jthread);
MutexLockerEx ml(thread->threadObj() == java_thread ? NULL : Threads_lock);
// We need to re-resolve the java_thread, since a GC might have happened during the
// acquire of the lock
JavaThread* thr = java_lang_Thread::thread(JNIHandles::resolve_non_null(jthread));
if (thr != NULL) {
Thread::interrupt(thr);
}
JVM_END
源码地址如下:
http://hg.openjdk.java.net/jdk8u/jdk8u/hotspot/file/79920693f915/src/share/vm/prims/jvm.cpp
jvm.cpp中调用了Thread::interrupt,回到thread.hpp源码的源码:
void Thread::interrupt(Thread* thread) {
trace("interrupt", thread);
debug_only(check_for_dangling_thread_pointer(thread);)
os::interrupt(thread);
}
Thread::interrupt调用os::interrupt方法实现,不同操作系统有不同的实现,以Linux系统为例。
// Linux系统interrupt方法实现如下:
void os::interrupt(Thread* thread) {
assert(Thread::current() == thread || Threads_lock->owned_by_self(),
"possibility of dangling Thread pointer");
//获取系统native线程对象
OSThread* osthread = thread->osthread();
if (!osthread->interrupted()) {
//设置中断状态为true
osthread->set_interrupted(true);
// More than one thread can get here with the same value of osthread,
// resulting in multiple notifications. We do, however, want the store
// to interrupted() to be visible to other threads before we execute unpark().
//内存屏障,使osthread的interrupted状态对其它线程立即可见
OrderAccess::fence();
//前文说过,_SleepEvent用于Thread.sleep,线程调用了sleep方法,则通过unpark唤醒
ParkEvent * const slp = thread->_SleepEvent ;
if (slp != NULL) slp->unpark() ;
}
//_parker用于concurrent相关的锁,此处同样通过unpark唤醒
if (thread->is_Java_thread())
((JavaThread*)thread)->parker()->unpark();
//Object.wait()唤醒
ParkEvent * ev = thread->_ParkEvent ;
if (ev != NULL) ev->unpark() ;
}
源码地址如下:
http://hg.openjdk.java.net/jdk8u/jdk8u/hotspot/file/677234770800/src/os/linux/vm/os_linux.cpp
由此可见,interrupt除了设置了中断标识,还可以通过ParkEvent的unpark方法唤醒对象;另外要注意:
1. object.wait、Thread.sleep和Thread.join会抛出InterruptedException并清除中断状态;
2. Lock.lock()方法不会响应中断,Lock.lockInterruptibly()方法则会响应中断并抛出异常,区别在于park()等待被唤醒时lock会继续执行park()来等待锁,而 lockInterruptibly会抛出异常;
3. synchronized被唤醒后会尝试获取锁,失败则会通过循环继续park()等待,因此实际上是不会被interrupt()中断的;
4. 一般情况下,抛出异常时,会清空Thread的interrupt状态,在编程时需要注意。
第3节 isInterrupted()方法源码
thread.cpp中is_interrupted方法如下:
bool Thread::is_interrupted(Thread* thread, bool clear_interrupted) {
trace("is_interrupted", thread);
debug_only(check_for_dangling_thread_pointer(thread);)
// Note: If clear_interrupted==false, this simply fetches and
// returns the value of the field osthread()->interrupted().
return os::is_interrupted(thread, clear_interrupted);
}
以Linux操作系统为例,:is_interrupted方法如下:
bool os::is_interrupted(Thread* thread, bool clear_interrupted) {
assert(Thread::current() == thread || Threads_lock->owned_by_self(),
"possibility of dangling Thread pointer");
OSThread* osthread = thread->osthread();
bool interrupted = osthread->interrupted();
if (interrupted && clear_interrupted) {
osthread->set_interrupted(false);
// consider thread->_SleepEvent->reset() ... optional optimization
}
return interrupted;
}
第4节 wait()方法源码
Object.wait()源码分析:
public final void wait() throws InterruptedException {
wait(0);
}
Object.wait()方法会调用native wait()方法。
public final native void wait(long timeout) throws InterruptedException;
进入objectMonitor.cpp的源码。
void ObjectMonitor::wait(jlong millis, bool interruptible, TRAPS) {
Thread * const Self = THREAD ;
assert(Self->is_Java_thread(), "Must be Java thread!");
JavaThread *jt = (JavaThread *)THREAD;
DeferredInitialize () ;
// Throw IMSX or IEX.
CHECK_OWNER();
// 1. 调用thread::is_interrupted(Thread* thread, true)判断并清除线程中断状态
// 如果中断状态为true,抛出中断异常并结束
if (interruptible && Thread::is_interrupted(Self, true) && !HAS_PENDING_EXCEPTION) {
// post monitor waited event. Note that this is past-tense, we are done waiting.
if (JvmtiExport::should_post_monitor_waited()) {
// Note: 'false' parameter is passed here because the
// wait was not timed out due to thread interrupt.
JvmtiExport::post_monitor_waited(jt, this, false);
}
TEVENT (Wait - Throw IEX) ;
THROW(vmSymbols::java_lang_InterruptedException());
return ;
}
TEVENT (Wait) ;
assert (Self->_Stalled == 0, "invariant") ;
Self->_Stalled = intptr_t(this) ;
jt->set_current_waiting_monitor(this);
// create a node to be put into the queue
// Critically, after we reset() the event but prior to park(), we must check
// for a pending interrupt.
ObjectWaiter node(Self);
node.TState = ObjectWaiter::TS_WAIT ;
Self->_ParkEvent->reset() ;
OrderAccess::fence(); // ST into Event; membar ; LD interrupted-flag
// Enter the waiting queue, which is a circular doubly linked list in this case
// but it could be a priority queue or any data structure.
// _WaitSetLock protects the wait queue. Normally the wait queue is accessed only
// by the the owner of the monitor *except* in the case where park()
// returns because of a timeout of interrupt. Contention is exceptionally rare
// so we use a simple spin-lock instead of a heavier-weight blocking lock.
Thread::SpinAcquire (&_WaitSetLock, "WaitSet - add") ;
AddWaiter (&node) ;
Thread::SpinRelease (&_WaitSetLock) ;
if ((SyncFlags & 4) == 0) {
_Responsible = NULL ;
}
intptr_t save = _recursions; // record the old recursion count
_waiters++; // increment the number of waiters
_recursions = 0; // set the recursion level to be 1
exit (Self) ; // exit the monitor
guarantee (_owner != Self, "invariant") ;
// As soon as the ObjectMonitor's ownership is dropped in the exit()
// call above, another thread can enter() the ObjectMonitor, do the
// notify(), and exit() the ObjectMonitor. If the other thread's
// exit() call chooses this thread as the successor and the unpark()
// call happens to occur while this thread is posting a
// MONITOR_CONTENDED_EXIT event, then we run the risk of the event
// handler using RawMonitors and consuming the unpark().
//
// To avoid the problem, we re-post the event. This does no harm
// even if the original unpark() was not consumed because we are the
// chosen successor for this monitor.
if (node._notified != 0 && _succ == Self) {
node._event->unpark();
}
// The thread is on the WaitSet list - now park() it.
// On MP systems it's conceivable that a brief spin before we park
// could be profitable.
//
// TODO-FIXME: change the following logic to a loop of the form
// while (!timeout && !interrupted && _notified == 0) park()
int ret = OS_OK ;
int WasNotified = 0 ;
{ // State transition wrappers
OSThread* osthread = Self->osthread();
OSThreadWaitState osts(osthread, true);
{
ThreadBlockInVM tbivm(jt);
// Thread is in thread_blocked state and oop access is unsafe.
jt->set_suspend_equivalent();
if (interruptible && (Thread::is_interrupted(THREAD, false) || HAS_PENDING_EXCEPTION)) {
// Intentionally empty
} else
if (node._notified == 0) {
if (millis <= 0) {
// 2. 调用park()方法阻塞线程
Self->_ParkEvent->park () ;
} else {
// 2. 调用park()方法在超时时间内阻塞线程
ret = Self->_ParkEvent->park (millis) ;
}
}
// were we externally suspended while we were waiting?
if (ExitSuspendEquivalent (jt)) {
// TODO-FIXME: add -- if succ == Self then succ = null.
jt->java_suspend_self();
}
} // Exit thread safepoint: transition _thread_blocked -> _thread_in_vm
// Node may be on the WaitSet, the EntryList (or cxq), or in transition
// from the WaitSet to the EntryList.
// See if we need to remove Node from the WaitSet.
// We use double-checked locking to avoid grabbing _WaitSetLock
// if the thread is not on the wait queue.
//
// Note that we don't need a fence before the fetch of TState.
// In the worst case we'll fetch a old-stale value of TS_WAIT previously
// written by the is thread. (perhaps the fetch might even be satisfied
// by a look-aside into the processor's own store buffer, although given
// the length of the code path between the prior ST and this load that's
// highly unlikely). If the following LD fetches a stale TS_WAIT value
// then we'll acquire the lock and then re-fetch a fresh TState value.
// That is, we fail toward safety.
if (node.TState == ObjectWaiter::TS_WAIT) {
Thread::SpinAcquire (&_WaitSetLock, "WaitSet - unlink") ;
if (node.TState == ObjectWaiter::TS_WAIT) {
DequeueSpecificWaiter (&node) ; // unlink from WaitSet
assert(node._notified == 0, "invariant");
node.TState = ObjectWaiter::TS_RUN ;
}
Thread::SpinRelease (&_WaitSetLock) ;
}
// The thread is now either on off-list (TS_RUN),
// on the EntryList (TS_ENTER), or on the cxq (TS_CXQ).
// The Node's TState variable is stable from the perspective of this thread.
// No other threads will asynchronously modify TState.
guarantee (node.TState != ObjectWaiter::TS_WAIT, "invariant") ;
OrderAccess::loadload() ;
if (_succ == Self) _succ = NULL ;
WasNotified = node._notified ;
// Reentry phase -- reacquire the monitor.
// re-enter contended monitor after object.wait().
// retain OBJECT_WAIT state until re-enter successfully completes
// Thread state is thread_in_vm and oop access is again safe,
// although the raw address of the object may have changed.
// (Don't cache naked oops over safepoints, of course).
// post monitor waited event. Note that this is past-tense, we are done waiting.
if (JvmtiExport::should_post_monitor_waited()) {
JvmtiExport::post_monitor_waited(jt, this, ret == OS_TIMEOUT);
}
OrderAccess::fence() ;
assert (Self->_Stalled != 0, "invariant") ;
Self->_Stalled = 0 ;
assert (_owner != Self, "invariant") ;
ObjectWaiter::TStates v = node.TState ;
if (v == ObjectWaiter::TS_RUN) {
enter (Self) ;
} else {
guarantee (v == ObjectWaiter::TS_ENTER || v == ObjectWaiter::TS_CXQ, "invariant") ;
ReenterI (Self, &node) ;
node.wait_reenter_end(this);
}
// Self has reacquired the lock.
// Lifecycle - the node representing Self must not appear on any queues.
// Node is about to go out-of-scope, but even if it were immortal we wouldn't
// want residual elements associated with this thread left on any lists.
guarantee (node.TState == ObjectWaiter::TS_RUN, "invariant") ;
assert (_owner == Self, "invariant") ;
assert (_succ != Self , "invariant") ;
} // OSThreadWaitState()
jt->set_current_waiting_monitor(NULL);
guarantee (_recursions == 0, "invariant") ;
_recursions = save; // restore the old recursion count
_waiters--; // decrement the number of waiters
// Verify a few postconditions
assert (_owner == Self , "invariant") ;
assert (_succ != Self , "invariant") ;
assert (((oop)(object()))->mark() == markOopDesc::encode(this), "invariant") ;
if (SyncFlags & 32) {
OrderAccess::fence() ;
}
// 3. 检查是否有通知notify发生
// 从park()方法返回后,判断是否是因为中断返回,再次调用
// thread::is_interrupted(Thread* thread, true)判断并清除线程中断状态
// 如果中断状态为true,抛出中断异常并结束。
if (!WasNotified) {
// no, it could be timeout or Thread.interrupt() or both
// check for interrupt event, otherwise it is timeout
if (interruptible && Thread::is_interrupted(Self, true) && !HAS_PENDING_EXCEPTION) {
TEVENT (Wait - throw IEX from epilog) ;
THROW(vmSymbols::java_lang_InterruptedException());
}
}
// NOTE: Spurious wake up will be consider as timeout.
// Monitor notify has precedence over thread interrupt.
}
1.调用thread::is_interrupted(Thread* thread, true)判断并清除线程中断状态,如果中断状态为true,抛出中断异常并结束。
2. 阻塞线程
3. 从park()方法返回后,判断是否是因为中断返回,再次调用thread::is_interrupted(Thread* thread, true)判断并清除线程中断状态,如果中断状态为true,抛出中断异常并结束。
第5节 sleep方法源码
Thread.sleep()源码解析:
public static native void sleep(long millis) throws InterruptedException;
它会调用jvm.cpp中的sleep方法。
JVM_ENTRY(void, JVM_Sleep(JNIEnv* env, jclass threadClass, jlong millis))
JVMWrapper("JVM_Sleep");
if (millis < 0) {
THROW_MSG(vmSymbols::java_lang_IllegalArgumentException(), "timeout value is negative");
}
//1. 判断并清除线程中断状态,如果中断状态为true,抛出中断异常
if (Thread::is_interrupted (THREAD, true) && !HAS_PENDING_EXCEPTION) {
THROW_MSG(vmSymbols::java_lang_InterruptedException(), "sleep interrupted");
}
// Save current thread state and restore it at the end of this block.
// And set new thread state to SLEEPING.
JavaThreadSleepState jtss(thread);
#ifndef USDT2
HS_DTRACE_PROBE1(hotspot, thread__sleep__begin, millis);
#else /* USDT2 */
HOTSPOT_THREAD_SLEEP_BEGIN(
millis);
#endif /* USDT2 */
EventThreadSleep event;
if (millis == 0) {
// When ConvertSleepToYield is on, this matches the classic VM implementation of
// JVM_Sleep. Critical for similar threading behaviour (Win32)
// It appears that in certain GUI contexts, it may be beneficial to do a short sleep
// for SOLARIS
if (ConvertSleepToYield) {
os::yield();
} else {
ThreadState old_state = thread->osthread()->get_state();
thread->osthread()->set_state(SLEEPING);
// 2. 调用os::sleep方法休眠线程
os::sleep(thread, MinSleepInterval, false);
thread->osthread()->set_state(old_state);
}
} else {
ThreadState old_state = thread->osthread()->get_state();
thread->osthread()->set_state(SLEEPING);
// 2. 调用os::sleep方法休眠线程
if (os::sleep(thread, millis, true) == OS_INTRPT) {
// An asynchronous exception (e.g., ThreadDeathException) could have been thrown on
// us while we were sleeping. We do not overwrite those.
if (!HAS_PENDING_EXCEPTION) {
if (event.should_commit()) {
event.set_time(millis);
event.commit();
}
#ifndef USDT2
HS_DTRACE_PROBE1(hotspot, thread__sleep__end,1);
#else /* USDT2 */
HOTSPOT_THREAD_SLEEP_END(
1);
#endif /* USDT2 */
// TODO-FIXME: THROW_MSG returns which means we will not call set_state()
// to properly restore the thread state. That's likely wrong.
THROW_MSG(vmSymbols::java_lang_InterruptedException(), "sleep interrupted");
}
}
thread->osthread()->set_state(old_state);
}
if (event.should_commit()) {
event.set_time(millis);
event.commit();
}
#ifndef USDT2
HS_DTRACE_PROBE1(hotspot, thread__sleep__end,0);
#else /* USDT2 */
HOTSPOT_THREAD_SLEEP_END(
0);
#endif /* USDT2 */
JVM_END
JVM_ENTRY(jobject, JVM_CurrentThread(JNIEnv* env, jclass threadClass))
JVMWrapper("JVM_CurrentThread");
oop jthread = thread->threadObj();
assert (thread != NULL, "no current thread!");
return JNIHandles::make_local(env, jthread);
JVM_END
1. 方法开始时, 调用thread::is_interrupted(Thread* thread, true)判断并清除线程中断状态,如果中断状态为true,抛出中断异常并结束。
2. 调用os::sleep方法使线程休眠。
以Linux操作系统为例,打开os_linux.cpp源码
int os::sleep(Thread* thread, jlong millis, bool interruptible) {
assert(thread == Thread::current(), "thread consistency check");
ParkEvent * const slp = thread->_SleepEvent ;
slp->reset() ;
OrderAccess::fence() ;
if (interruptible) {
jlong prevtime = javaTimeNanos();
for (;;) {
//判断并清除线程中断状态
if (os::is_interrupted(thread, true)) {
//发生中断状态为true,返回OS_INTRP
return OS_INTRPT;
}
jlong newtime = javaTimeNanos();
if (newtime - prevtime < 0) {
// time moving backwards, should only happen if no monotonic clock
// not a guarantee() because JVM should not abort on kernel/glibc bugs
assert(!Linux::supports_monotonic_clock(), "time moving backwards");
} else {
millis -= (newtime - prevtime) / NANOSECS_PER_MILLISECS;
}
if(millis <= 0) {
return OS_OK;
}
prevtime = newtime;
{
assert(thread->is_Java_thread(), "sanity check");
JavaThread *jt = (JavaThread *) thread;
ThreadBlockInVM tbivm(jt);
OSThreadWaitState osts(jt->osthread(), false /* not Object.wait() */);
jt->set_suspend_equivalent();
// cleared by handle_special_suspend_equivalent_condition() or
// java_suspend_self() via check_and_wait_while_suspended()
//调用park方法
slp->park(millis);
// were we externally suspended while we were waiting?
jt->check_and_wait_while_suspended();
}
}
} else {
OSThreadWaitState osts(thread->osthread(), false /* not Object.wait() */);
jlong prevtime = javaTimeNanos();
for (;;) {
// It'd be nice to avoid the back-to-back javaTimeNanos() calls on
// the 1st iteration ...
jlong newtime = javaTimeNanos();
if (newtime - prevtime < 0) {
// time moving backwards, should only happen if no monotonic clock
// not a guarantee() because JVM should not abort on kernel/glibc bugs
assert(!Linux::supports_monotonic_clock(), "time moving backwards");
} else {
millis -= (newtime - prevtime) / NANOSECS_PER_MILLISECS;
}
if(millis <= 0) break ;
prevtime = newtime;
//调用park方法
slp->park(millis);
}
return OS_OK ;
}
}
os::sleep方法在死循环内调用park方法,他只在满足以下两种情形之一时从park方法返回并退出死循环,否则即使从park方法返回了,也认为唤醒无效,继续调用park方法:
1、调用thread.interrupt方法解除线程阻塞,让park方法返回。
从park()方法返回后,判断是否是因为中断返回,调用thread::is_interrupted(Thread* thread, true)判断并清除线程中断状态,如果中断状态为true,return返回OS_INTRPT,退出死循环。
2、到达指定睡眠时间,park方法自动返回。
从park()方法返回后,判断剩余时间millis是否小于等于0,如果是,则认为到达指定睡眠时间,return返回OS_OK,退出死循环。