jvm源码分析之interrupt()

概述

线程的thread.interrupt()方法是中断线程。中断一个线程意味着在线程完成它的任务之前，停止它当前正在执行的操作。

如果线程堵塞在object.wait、Thread.join和Thread.sleep，将会清除线程的中断状态，并抛出InterruptedException;

如果线程堵塞在java.nio.channels.InterruptibleChannel的IO上，Channel将会被关闭，线程被置为中断状态，并抛出java.nio.channels.ClosedByInterruptException；

如果线程堵塞在java.nio.channels.Selector上，线程被置为中断状态，select方法会马上返回，类似调用wakeup的效果；

如果不是以上三种情况，thread.interrupt()方法仅仅是设置线程的中断状态为true。

interrupt方法的jvm源码分析

在jvm的Thread类中有三个成员变量：（javaThread就是继承了这个Thread类）

thread.hpp

public:ParkEvent * _ParkEvent ;                     // for synchronized()    ParkEvent * _SleepEvent ;                    // for Thread.sleep// JSR166 per-thread parker
private:Parker*    _parker;

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 operateoop 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 lockJavaThread* thr = java_lang_Thread::thread(JNIHandles::resolve_non_null(jthread));if (thr != NULL) {Thread::interrupt(thr);}
JVM_END

JVM_Interrupt对参数进行了校验，然后直接调用Thread::interrupt:

thread.cpp

void Thread::interrupt(Thread* thread) {trace("interrupt", thread);debug_only(check_for_dangling_thread_pointer(thread);)os::interrupt(thread);
}

Thread::interrupt调用os::interrupt方法实现：

os_linux.cpp

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()) {osthread->set_interrupted(true); //设置中断状态为true//内存屏障，使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() ;}

由此可见，interrupt()其实就是通过ParkEvent的unpark方法唤醒线程。

wait()响应中断

1、在调用ParkEvent的park方法之前，会先判断线程的中断状态，如果为true，清除线程的中断状态，并抛出InterruptedException，然后结束。

2、在调用ParkEvent的park方法阻塞在条件变量之后，当interrupt()调用ParkEvent的unpark方法唤醒线程，线程会从pthread_cond_wait()返回，从而解除阻塞，代码继续往下走，再次判断线程的中断状态，如果为true则清除线程的中断状态，并抛出InterruptedException，然后结束。

判断线程中断状态，并通过布尔参数决定是否清除线程中断状态，方法如下：

thread.cpp

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平台对os::is_interrupted()的实现为：

os_linux.cpp

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);  //清除线程中断状态，重置为false// consider thread->_SleepEvent->reset() ... optional optimization}return interrupted;
}

ObjectMonitor::wait()的实现如下：

方法开始时，调用thread::is_interrupted(Thread* thread, true)判断并清除线程中断状态,如果中断状态为true，抛出中断异常并结束。

从park()方法返回后，判断是否是因为中断返回，再次调用thread::is_interrupted(Thread* thread, true)判断并清除线程中断状态,如果中断状态为true，抛出中断异常并结束。

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();EventJavaMonitorWait event;// check for a pending interrupt  调用is_interrupted(thread,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);  }if (event.should_commit()) {post_monitor_wait_event(&event, 0, millis, false);}TEVENT (Wait - Throw IEX) ;THROW(vmSymbols::java_lang_InterruptedException()); //抛出InterruptedExceptionreturn ;   //直接结束，不执行下面逻辑}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);//将线程封装成waitor节点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) ; //添加到waitsetThread::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 1exit (true, Self) ;                    // exit the monitorguarantee (_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 wrappersOSThread* 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} elseif (node._notified == 0) {if (millis <= 0) {Self->_ParkEvent->park () ;  //调用park方法阻塞线程} else {ret = Self->_ParkEvent->park (millis) ; //调用park方法在超时时间内阻塞线程}}// 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 WaitSetassert(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);}if (event.should_commit()) {post_monitor_wait_event(&event, node._notifier_tid, millis, 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 postconditionsassert (_owner == Self       , "invariant") ;assert (_succ  != Self       , "invariant") ;assert (((oop)(object()))->mark() == markOopDesc::encode(this), "invariant") ;if (SyncFlags & 32) {OrderAccess::fence() ;}// check if the notification happened  判断是否发生通知if (!WasNotified) {   //如果不是发生通知，则可能是超时或者中断// no, it could be timeout or Thread.interrupt() or both 判断是否中断，否则为超时// check for interrupt event, otherwise it is timeout //调用is_interrupted()判断并清除中断状态if (interruptible && Thread::is_interrupted(Self, true) && !HAS_PENDING_EXCEPTION) {TEVENT (Wait - throw IEX from epilog) ;THROW(vmSymbols::java_lang_InterruptedException()); //抛出InterruptedException}}// NOTE: Spurious wake up will be consider as timeout.// Monitor notify has precedence over thread interrupt.
}

wait()返回只在以下三种情形下发生：通知、超时、中断。

Thread.interrupt可以看作一种特殊的通知信号：

os_linux.cpp

Object.wait(timo) will return because of// (a) notification// (b) timeout// (c) thread.interrupt//// Thread.interrupt and object.notify{All} both call Event::set.// That is, we treat thread.interrupt as a special case of notification.// The underlying Solaris implementation, cond_timedwait, admits// spurious/premature wakeups, but the JLS/JVM spec prevents the// JVM from making those visible to Java code.  As such, we must// filter out spurious wakeups.  We assume all ETIME returns are valid.

sleep()方法响应中断

Thread.sleep最终调用JVM_Sleep方法：

方法开始时，调用thread::is_interrupted(Thread* thread, true)判断并清除线程中断状态,如果中断状态为true，抛出中断异常并结束。

调用os::sleep方法返回后，判断返回值是否为OS_INTRPT，如果是则为发生中断，抛出中断异常。

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"); }
//判断并清除线程中断状态，如果中断状态为true,抛出中断异常
if (Thread::is_interrupted (THREAD, true) && !HAS_PENDING_EXCEPTION) { THROW_MSG(vmSymbols::java_lang_InterruptedException(), "sleep interrupted");
}
JavaThreadSleepState jtss(thread);
EventThreadSleep event;
if (millis == 0) {       if (ConvertSleepToYield) { os::yield(); } else {ThreadState old_state = thread->osthread()->get_state();thread->osthread()->set_state(SLEEPING); os::sleep(thread, MinSleepInterval, false);//sleep 1ms thread->osthread()->set_state(old_state); }} else {ThreadState old_state = thread->osthread()->get_state();//osthread->thread status mapping: // NEW->NEW //RUNNABLE->RUNNABLE //BLOCKED_ON_MONITOR_ENTER->BLOCKED //IN_OBJECT_WAIT,PARKED->WAITING //SLEEPING,IN_OBJECT_WAIT_TIMED,PARKED_TIMED->TIMED_WAITING ///TERMINATED->TERMINATED thread->osthread()->set_state(SLEEPING); //调用os::sleep方法，如果sleep()的返回值为OS_INTRPT，则为发生中断，抛出中断异常 if (os::sleep(thread, millis, true) == OS_INTRPT) { if (!HAS_PENDING_EXCEPTION) {if (event.should_commit()) { event.set_time(millis); event.commit(); } THROW_MSG(vmSymbols::java_lang_InterruptedException(), "sleep interrupted"); }} thread->osthread()->set_state(old_state);
}
if (event.should_commit()) { event.set_time(millis); event.commit();}
JVM_END

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，退出死循环。

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)) { //判断并清除线程中断状态return OS_INTRPT;    //发生中断状态为true，返回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 bugsassert(!Linux::supports_monotonic_clock(), "time moving backwards");} else {millis -= (newtime - prevtime) / NANOSECS_PER_MILLISEC;}if(millis <= 0) {  //如果剩余时间小于0，返回OS_OKreturn 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()slp->park(millis); //调用park方法// 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 bugsassert(!Linux::supports_monotonic_clock(), "time moving backwards");} else {millis -= (newtime - prevtime) / NANOSECS_PER_MILLISEC;}if(millis <= 0) break ;prevtime = newtime;slp->park(millis); //调用park方法}return OS_OK ;}
}

参考：

Thread.interrupt() API

Java线程源码解析之interrupt

Interrupting Java threads