LockSupport(park&unpark)源码解析

LockSupport(park/unpark)源码解析 park方法

park()

public static void park() {
    UNSAFE.park(false, 0L);
}

park(Object blocker)

public static void park(Object blocker) {
    Thread t = Thread.currentThread();
    setBlocker(t, blocker);
    UNSAFE.park(false, 0L);
    setBlocker(t, null);
}

blocker是用来记录线程被阻塞时被谁阻塞的。用于线程监控和分析工具来定位原因的。

setBlocker(t, blocker)方法的作用是记录t线程是被broker阻塞的。

因此我们只关注最核心的方法，也就是UNSAFE.park(false, 0L)。

    //park
    public native void park(boolean isAbsolute, long time);
    
    //unpack
    public native void unpark(Object var1);

UNSAFE是一个非常强大的类，他的的操作是基于底层的，也就是可以直接操作内存，

Unsafe.park的底层实现原理

在Linux系统下，是用的Posix线程库pthread中的mutex（互斥量），condition（条件变量）来实现的。 mutex和condition保护了一个_counter的变量，当park时，这个变量被设置为0，当unpark时，这个变量被设置为1。

_counter字段，就是用来记录所谓的“许可”的

condition条件变量被用来阻塞一个线程，当条件不满足时，线程往往解开相应的互斥锁并等待条件发生变化。一旦其他的某个线程改变了条件变量，它将通知相应的条件变量唤醒一个或多个正被此条件变量阻塞的线程，这些线程将重新锁定互斥锁并重新测试条件是否满足。

每个Java线程都有一个Parker实例，Parker类是这样定义的：

park.hpp

/* In the future we'll want to think about eliminating Parker and using
 * ParkEvent instead.  There's considerable duplication between the two
 * services.
 */
/*
*将来我们会考虑除掉Parker，用ParkEvent代替。两者之间有相当多的重复服务。
*/
class Parker : public os::PlatformParker {
private:
  volatile int _counter ;
  Parker * FreeNext ;
  JavaThread * AssociatedWith ; // Current association

public:
  Parker() : PlatformParker() {
    _counter       = 0 ;
    FreeNext       = NULL ;
    AssociatedWith = NULL ;
  }
protected:
  ~Parker() { ShouldNotReachHere(); }
public:
  // For simplicity of interface with Java, all forms of park (indefinite,
  // relative, and absolute) are multiplexed into one call.
  void park(bool isAbsolute, jlong time);
  void unpark();

  // Lifecycle operators
  static Parker * Allocate (JavaThread * t) ;
  static void Release (Parker * e) ;
private:
  static Parker * volatile FreeList ;
  static volatile int ListLock ;

};

os::PlatformParker具体由操作系统实现，比如linux

os_linux.hpp

class PlatformParker : public CHeapObj {
  protected:
    enum {
        REL_INDEX = 0,
        ABS_INDEX = 1
    };
    int _cur_index;  // which cond is in use: -1, 0, 1  
    pthread_mutex_t _mutex [1] ;
    pthread_cond_t  _cond  [2] ; // one for relative times and one for abs.

  public:       // TODO-FIXME: make dtor private
    ~PlatformParker() { guarantee (0, "invariant") ; }

  public:
    PlatformParker() {
      int status;
      status = pthread_cond_init (&_cond[REL_INDEX], os::Linux::condAttr());
      assert_status(status == 0, status, "cond_init rel");
      status = pthread_cond_init (&_cond[ABS_INDEX], NULL);
      assert_status(status == 0, status, "cond_init abs");
      status = pthread_mutex_init (_mutex, NULL);
      assert_status(status == 0, status, "mutex_init");
      _cur_index = -1; // mark as unused
    }
};

①当调用park时，先尝试直接能否直接拿到“许可”，即_counter>0时，如果成功，则把_counter设置为0,并返回。

②如果不成功，则构造一个ThreadBlockInVM，然后检查_counter是不是>0，如果是，则把_counter设置为0，unlock mutex并返回:

③否则，再判断等待的时间，然后再调用pthread_cond_wait函数等待，如果等待返回，则把_counter设置为0，unlock mutex并返回：

os_linux.cpp

void Parker::park(bool isAbsolute, jlong time) {
  // Ideally we'd do something useful while spinning, such
  // as calling unpackTime().

  // Optional fast-path check:
  // Return immediately if a permit is available.
  // We depend on Atomic::xchg() having full barrier semantics
  // since we are doing a lock-free update to _counter.
    //①先尝试直接能否直接拿到“许可”
  if (Atomic::xchg(0, &_counter) > 0) return;

  Thread* thread = Thread::current();
  assert(thread->is_Java_thread(), "Must be JavaThread");
  JavaThread *jt = (JavaThread *)thread;

  // Optional optimization -- avoid state transitions if there's an interrupt pending.
  // Check interrupt before trying to wait
  if (Thread::is_interrupted(thread, false)) {
    return;//如果被interrupt中断，被唤醒返回
  }

  // Next, demultiplex/decode time arguments
  timespec absTime;
  if (time  0) {
      //传入了时间参数，将其存入absTime，并解析成absTime->tv_sec(秒)和absTime->tv_nsec(纳秒)存储起来，存的是绝对时间
    unpackTime(&absTime, isAbsolute, time);
  }


  // Enter safepoint region
  // Beware of deadlocks such as 6317397.
  // The per-thread Parker:: mutex is a classic leaf-lock.
  // In particular a thread must never block on the Threads_lock while
  // holding the Parker:: mutex.  If safepoints are pending both the
  // the ThreadBlockInVM() CTOR and DTOR may grab Threads_lock.
    //②构造一个ThreadBlockInVM，然后检查_counter是不是>0，如果是，则把_counter设置为0，unlock mutex并返回:
 //进入safepoint region，更改线程为阻塞状态
  ThreadBlockInVM tbivm(jt);

  // Don't wait if cannot get lock since interference arises from
  // unblocking.  Also. check interrupt before trying wait
  if (Thread::is_interrupted(thread, false) || pthread_mutex_trylock(_mutex) != 0) {
       //如果线程被中断，或者是在尝试给互斥变量加锁的过程中，加锁失败，比如被其它线程锁住了，直接返回
    return;
  }

//这里表示线程互斥变量锁成功了
  int status ;
    //许可为1，不需要等待
  if (_counter > 0)  { // no wait needed
    _counter = 0;//已经有许可了，用掉当前许可
      //对互斥变量解锁
    status = pthread_mutex_unlock(_mutex);
    assert (status == 0, "invariant") ;
    // Paranoia to ensure our locked and lock-free paths interact
    // correctly with each other and Java-level accesses.
      //使用内存屏障，确保 _counter赋值为0(写入操作)能够被内存屏障之后的读操作获取内存屏障事前的结果，也就是能够正确的读到0
    OrderAccess::fence();
    return;//返回
  }

#ifdef ASSERT
  // Don't catch signals while blocked; let the running threads have the signals.
  // (This allows a debugger to break into the running thread.)
  sigset_t oldsigs;
  sigset_t* allowdebug_blocked = os::Linux::allowdebug_blocked_signals();
  pthread_sigmask(SIG_BLOCK, allowdebug_blocked, &oldsigs);
#endif

//将java线程所拥有的操作系统线程设置成 CONDVAR_WAIT状态 ，表示在等待某个条件的发生
  OSThreadWaitState osts(thread->osthread(), false /* not Object.wait() */);
  jt->set_suspend_equivalent();
  // cleared by handle_special_suspend_equivalent_condition() or java_suspend_self()

  assert(_cur_index == -1, "invariant");
  if (time == 0) {
    _cur_index = REL_INDEX; // arbitrary choice when not timed
      //③判断等待的时间，然后再调用pthread_cond_wait函数等待，如果等待返回，则把_counter设置为0，unlock mutex并返回：
      //把调用线程放到等待条件的线程列表上，然后对互斥变量解锁，（这两是原子操作），这个时候线程进入等待，当它返回时，互斥变量再次被锁住。
  //成功返回0，否则返回错误编号
    status = pthread_cond_wait (&_cond[_cur_index], _mutex) ;
  } else {
    _cur_index = isAbsolute ? ABS_INDEX : REL_INDEX;
    status = os::Linux::safe_cond_timedwait (&_cond[_cur_index], _mutex, &absTime) ;
    if (status != 0 && WorkAroundNPTLTimedWaitHang) {
      pthread_cond_destroy (&_cond[_cur_index]) ;
      pthread_cond_init    (&_cond[_cur_index], isAbsolute ? NULL : os::Linux::condAttr());
    }
  }
  _cur_index = -1;
  assert_status(status == 0 || status == EINTR ||
                status == ETIME || status == ETIMEDOUT,
                status, "cond_timedwait");

#ifdef ASSERT
  pthread_sigmask(SIG_SETMASK, &oldsigs, NULL);
#endif
//等待结束后，许可被消耗，改为0  _counter = 0 ;
//释放互斥量的锁
  _counter = 0 ;
  status = pthread_mutex_unlock(_mutex) ;
  assert_status(status == 0, status, "invariant") ;
  // Paranoia to ensure our locked and lock-free paths interact
  // correctly with each other and Java-level accesses.
    
//加入内存屏障指令
  OrderAccess::fence();

  // If externally suspended while waiting, re-suspend
  if (jt->handle_special_suspend_equivalent_condition()) {
    jt->java_suspend_self();
  }
}

ThreadBlockInVM tbivm(jt) 这属于C++新建变量的语法，也就是调用构造函数新建了一个变量，变量名为tbivm,参数为jt。类的实现为

class ThreadBlockInVM : public ThreadStateTransition {
 public:
  ThreadBlockInVM(JavaThread *thread)
  : ThreadStateTransition(thread) {
    // Once we are blocked vm expects stack to be walkable    
    thread->frame_anchor()->make_walkable(thread);
   //把线程由运行状态转成阻塞状态
    trans_and_fence(_thread_in_vm, _thread_blocked);
  }
  ...
};

_thread_in_vm 表示线程当前在VM中执行， _thread_blocked表示线程当前阻塞了，他们是globalDefinitions.hpp中定义的枚举

//这个枚举是用来追踪线程在代码的那一块执行，用来给 safepoint code使用，有4种重要的类型，_thread_new/_thread_in_native/_thread_in_vm/_thread_in_Java。形如xxx_trans的状态都是中间状态，表示线程正在由一种状态变成另一种状态，这种方式使得 safepoint code在处理线程状态时，不需要对线程进行挂起，使得safe point code运行更快，而给定一个状态，通过+1就可以得到他的转换状态
enum JavaThreadState {
  _thread_uninitialized     =  0, // should never happen (missing initialization) 
_thread_new               =  2, // just starting up, i.e., in process of being initialized 
_thread_new_trans         =  3, // corresponding transition state (not used, included for completeness)  
_thread_in_native         =  4, // running in native code  . This is a safepoint region, since all oops will be in jobject handles
_thread_in_native_trans   =  5, // corresponding transition state  
_thread_in_vm             =  6, // running in VM 
_thread_in_vm_trans       =  7, // corresponding transition state 
_thread_in_Java           =  8, //  Executing either interpreted or compiled Java code running in Java or in stub code  
_thread_in_Java_trans     =  9, // corresponding transition state (not used, included for completeness) 
_thread_blocked           = 10, // blocked in vm 
_thread_blocked_trans     = 11, // corresponding transition state 
_thread_max_state         = 12  // maximum thread state+1 - used for statistics allocation
};

这就是整个park的过程，总结来说就是消耗“许可”的过程。

unpark

  /**
     * 如果给定线程的许可尚不可用，则使其可用。
     * 如果线程在 park 上受阻塞，则它将解除其阻塞状态。
     * 否则，保证下一次调用 park 不会受阻塞。
     * 如果给定线程尚未启动，则无法保证此操作有任何效果。 
     * @param thread: 要执行 unpark 操作的线程；该参数为 null 表示此操作没有任何效果。
     */
public static void unpark(Thread thread) {
    if (thread != null)
        UNSAFE.unpark(thread);
}

Unsafe.class

public native void unpark(Object var1);

Unsafe.unpark的底层实现原理

当unpark时，则简单多了，直接设置_counter为1，再unlock mutext返回。如果_counter之前的值是0，则还要调用pthread_cond_signal唤醒在park中等待的线程：

os_linux.cpp

void Parker::unpark() {
  int s, status ;
    //给互斥量加锁，如果互斥量已经上锁，则阻塞到互斥量被解锁
//park进入wait时，_mutex会被释放
  status = pthread_mutex_lock(_mutex);
  assert (status == 0, "invariant") ;
  s = _counter;存储旧的_counter许可
  _counter = 1;//设置许可为1
  if (s