分析JVM源碼之Thread.interrupt系統級別線程打斷

@Test
public void interruptedTest() throws InterruptedException {
    Thread sleep = new Thread(() -> {
        try {
            log.info("sleep thread start");
            TimeUnit.SECONDS.sleep(1);
            log.info("sleep thread end");
        } catch (InterruptedException e) {
            log.info("sleep thread interrupted");
        }
    }, "sleep_thread");
    sleep.start();

    TimeUnit.MILLISECONDS.sleep(100);
    log.info("ready to interrupt sleep");
    sleep.interrupt();
}

@Test
public void normalTest() throws InterruptedException {
    Thread normal = new Thread(() -> {
        log.info("normal thread start");
        int i = 0;
        while (true) {
            i++;
        }
    }, "normal_thread");
    normal.start();
    TimeUnit.MILLISECONDS.sleep(100);
    log.info("ready to interrupt normal");
    normal.interrupt();
}

public void interrupt() {
    if (this != Thread.currentThread())
        checkAccess();

    synchronized (blockerLock) {
        Interruptible b = blocker;
        if (b != null) {
            //打斷的主要方法，該方法的主要作用是設置一個打斷標記
            interrupt0();
            b.interrupt(this);
            return;
        }
    }
    interrupt0();
}

private native void interrupt0();

static JNINativeMethod methods[] = {
    ...
    {"interrupt0",       "()V",        (void *)&JVM_Interrupt}
    ...
};

JVM_ENTRY(void, JVM_Interrupt(JNIEnv* env, jobject jthread))
  JVMWrapper("JVM_Interrupt");
  ...
  if (thr != NULL) {
    //執行線程打斷操作
    Thread::interrupt(thr);
  }
JVM_END

void Thread::interrupt(Thread* thread) {
  //執行os層面的打斷
  os::interrupt(thread);
}

void os::interrupt(Thread* thread) {
  ...
  //獲得c++線程對應的系統線程
  OSThread* osthread = thread->osthread();
  //如果系統線程的打斷標記是false，意味著還未被打斷
  if (!osthread->interrupted()) {
    //將系統線程的打斷標記設為true
    osthread->set_interrupted(true);
    //這個涉及到內存屏障，本文不展開
    OrderAccess::fence();
    //這裡獲取一個_SleepEvent，並調用其unpark()方法
    ParkEvent * const slp = thread->_SleepEvent ;
    if (slp != NULL) slp->unpark() ;
  }

  //這裡依據JSR166標準，即使打斷標記為true，依然要調用下面的2個unpark
  if (thread->is_Java_thread())
    //如果是一個java線程，這裡獲取一個parker對象，並調用其unpark()方法
    ((JavaThread*)thread)->parker()->unpark();

  ParkEvent * ev = thread->_ParkEvent ;
  //這裡獲取一個_ParkEvent，並調用其unpark()方法
  if (ev != NULL) ev->unpark() ;
}

ParkEvent * _ParkEvent;    // for synchronized()
ParkEvent * _SleepEvent;   // for Thread.sleep

class ParkEvent : public os::PlatformEvent {
  ...
}

class PlatformEvent : public CHeapObj<mtInternal> {
 private:
  ...
  pthread_mutex_t _mutex[1];
  pthread_cond_t  _cond[1];
  ...
  void park();
  void unpark();
  int  park(jlong millis); // relative timed-wait only
  ...
};

void os::PlatformEvent::park() { 
    ...
    //鎖住信號量
    int status = pthread_mutex_lock(_mutex);
    while (_Event < 0) {
      //釋放信號量，並在條件變量上等待
      status = pthread_cond_wait(_cond, _mutex);
    }
    //釋放信號量
    status = pthread_mutex_unlock(_mutex);
}

synchronize(obj){
  obj.wait();
}

ReentrantLock lock = new ReentrantLock();
Condition condition = lock.newCondition();
lock.lock();
condition.wait();
lock.unlock();

void os::PlatformEvent::unpark() {
  ...
  if (AnyWaiters != 0) {
    //喚醒條件變量
    status = pthread_cond_signal(_cond);
  }
  ...
}

public:
  Parker*     parker() { return _parker; }

class Parker : public os::PlatformParker {
    pthread_mutex_t _mutex[1];
    pthread_cond_t  _cond[2]; // one for relative times and one for abs.
}

void Parker::park(bool isAbsolute, jlong time) {
  ...
  if (time == 0) {
    //這裡是直接長時間等待
    _cur_index = REL_INDEX; 
    status = pthread_cond_wait(&_cond[_cur_index], _mutex);
  } else {
    //這裡會根據時間是否是絕對時間，分別等待在不同的條件上
    _cur_index = isAbsolute ? ABS_INDEX : REL_INDEX;
    status = pthread_cond_timedwait(&_cond[_cur_index], _mutex, &absTime);
  }
  ...
}

void Parker::unpark() {
  int status = pthread_mutex_lock(_mutex);
  ...
  //因為在等待的時候會有2個等待對象，所以需要先獲取正確的索引
  int index = _cur_index;
  ...
  status = pthread_mutex_unlock(_mutex);
  if (s < 1 && index != -1) {
    //喚醒線程
    status = pthread_cond_signal(&_cond[index]);
  }
  ...
}

import java.util.concurrent.TimeUnit;
import java.time.LocalDateTime;

public class MyThread {

    static {
        //設置查找路徑為當前項目路徑
        System.setProperty("java.library.path", ".");
        //加載動態庫的名稱
        System.loadLibrary("MyThread");
    }

    public native void startAndPark();

    public native void interrupt();

    public static void main(String[] args) throws InterruptedException {
        MyThread thread = new MyThread();
        //啟動線程打印一段文字，並睡眠
        thread.startAndPark();
        //1秒後主線程打斷子線程
        TimeUnit.MILLISECONDS.sleep(1000);
        System.out.println(LocalDateTime.now() + "：Main---準備打斷線程");
        //打斷子線程
        thread.interrupt();
        System.out.println(LocalDateTime.now() + "：Main---打斷完成");
    }
}

#include <pthread.h>
#include <stdio.h>
#include "MyThread.h"
#include "time.h"

pthread_t pid;
pthread_mutex_t _mutex = PTHREAD_MUTEX_INITIALIZER;
pthread_cond_t  _cond = PTHREAD_COND_INITIALIZER; 

//打印時間
void printTime(){
    char strTm[50] = { 0 };
	  time_t currentTm;
	  time(&currentTm);
	  strftime(strTm, sizeof(strTm), "%x %X", localtime(&currentTm));
	  puts(strTm);
}

//子線程執行的方法
void* thread_entity(void* arg){
    printTime();
    printf("MyThread---啟動\n");
    printTime();
    printf("MyThread---準備休眠\n");
    //阻塞線程，等待喚醒
    pthread_cond_wait(&_cond, &_mutex);
    printTime();
    printf("MyThread---休眠被打斷\n");
}
//對應MyThread中的startAndPark方法
JNIEXPORT void JNICALL Java_MyThread_startAndPark(JNIEnv *env, jobject c1){
    //創建一個子線程
    pthread_create(&pid, NULL, thread_entity, NULL);
}
//對應MyThread中的interrupt方法
JNIEXPORT void JNICALL Java_MyThread_interrupt(JNIEnv *env, jobject c1){
    //喚醒線程
    pthread_cond_signal(&_cond);
}