jdk unsafe

unsafe

park&unpark

结构体

class Parker : public os::PlatformParker {
    volatile int _counter;
};

park

UNSAFE_ENTRY(void, Unsafe_Park(JNIEnv *env, jobject unsafe, jboolean isAbsolute, jlong time)) {
  HOTSPOT_THREAD_PARK_BEGIN((uintptr_t) thread->parker(), (int) isAbsolute, time);
  // --------------------------------------------------
  EventThreadPark event;
  JavaThreadParkedState jtps(thread, time != 0);
  // ============================
  // 线程对应的 parker 对象，调用 park 方法
  thread->parker()->park(isAbsolute != 0, time);
  // ============================
  // 从 park 状态恢复了
  // event 是个 JfrEvent 类型事件，Jfr 就是 JavaFlightRecorder，毫无疑问只有启用了 Jfr 这里才会生效。
  if (event.should_commit()) {
        // ......
  }
  // --------------------------------------------------
  HOTSPOT_THREAD_PARK_END((uintptr_t) thread->parker());
} UNSAFE_END

void Parker::park(bool isAbsolute, jlong time) {
    // 原子操作将 counter 置为0，并返回旧值。如果旧值 > 0 说明已经 park 了。
    if (Atomic::xchg(0, &_counter) > 0) return; 
    Thread* thread = Thread::current();
    JavaThread *jt = (JavaThread *)thread;
    // 是否已经被中断过了
    if (Thread::is_interrupted(thread, false)) {
        return;
    }
    struct timespec absTime;
    if (time < 0 || (isAbsolute && time == 0)) { // don't wait at all
        // 时间参数不合法，会导致不进行park
        return;
    }
    if (time > 0) {
        to_abstime(&absTime, time, isAbsolute);
    }
    // 安全点相关
    ThreadBlockInVM tbivm(jt);
    if (Thread::is_interrupted(thread, false) ||
                pthread_mutex_trylock(_mutex) != 0) {
        // 如果已经被中断了，或者没有被中断，但是尝试加线程锁失败。
        return;
    }
    int status;
    // park 的第一个操作就会把 _counter 置为0。这里又不为0了
    if (_counter > 0) { // no wait needed
        _counter = 0;
        // 解锁
        status = pthread_mutex_unlock(_mutex);
        // 双屏障
        OrderAccess::fence();
        return;
    }
    OSThreadWaitState osts(thread->osthread(), false /* not Object.wait() */);
    jt->set_suspend_equivalent();
    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);
    }
    // 走到这里就说明被唤醒了
    _cur_index = -1; 
    _counter = 0;
    // 解锁
    status = pthread_mutex_unlock(_mutex);
    OrderAccess::fence();
    // debug 相关
    if (jt->handle_special_suspend_equivalent_condition()) {
        jt->java_suspend_self();
    }
}

用于返回 thread 的 interrupted 标记，可以选择是否重置 interrupted 标记

bool os::is_interrupted(Thread* thread, bool clear_interrupted) {
    OSThread* osthread = thread->osthread();
    bool interrupted = osthread->interrupted();
    if (interrupted && clear_interrupted) {
        osthread->set_interrupted(false);
    }
    return interrupted;
}

unpark

UNSAFE_ENTRY(void, Unsafe_Unpark(JNIEnv *env, jobject unsafe, jobject jthread)) {
  Parker* p = NULL;
  if (jthread != NULL) {
    ThreadsListHandle tlh;
    JavaThread* thr = NULL;
    oop java_thread = NULL;
    // 将java对象转化成对应jvm的 JavaThread 对象
    (void) tlh.cv_internal_thread_to_JavaThread(jthread, &thr, &java_thread);
    if (java_thread != NULL) {
      if (thr != NULL) {
          // 拿到线程对应的 parker 对象
        p = thr->parker();
      }
    }
  if (p != NULL) {
      // 看不出来作用是啥
      HOTSPOT_THREAD_UNPARK((uintptr_t) p);
      // 调用 parker 的 unpark 方法
      p->unpark();
  }
} UNSAFE_END

void Parker::unpark() {
    // 对线程互斥量加锁
    int status = pthread_mutex_lock(_mutex);
    const int s = _counter;
    // 因为加锁了，所以可以直接改为1
    _counter = 1;
    int index = _cur_index;
    // 释放线程互斥量
    status = pthread_mutex_unlock(_mutex);
    if (s < 1 && index != -1) {
        // 旧值为0，说明线程正在 park, 需要唤醒。 
        // index 对应的是等待的条件的索引，是在park时赋值的
        status = pthread_cond_signal(&_cond[index]);
    }
}