打通JAVA与内核系列之一ReentrantLock锁的实现原理

一  JDK层

1  AbstractQueuedSynchronizer

代码节选自：java.util.concurrent.locks.ReentrantLock.java /** Synchronizer providing all implementation mechanics */    private final Sync sync;    /**     * Base of synchronization control for this lock. Subclassed     * into fair and nonfair versions below. Uses AQS state to     * represent the number of holds on the lock.     */    abstract static class Sync extends AbstractQueuedSynchronizer {......}
    public void lock() {        sync.lock();    }

    /**     * Acquires in exclusive mode, ignoring interrupts.  Implemented     * by invoking at least once {@link #tryAcquire},     * returning on success.  Otherwise the thread is queued, possibly     * repeatedly blocking and unblocking, invoking {@link     * #tryAcquire} until success.  This method can be used     * to implement method {@link Lock#lock}.     *     * @param arg the acquire argument.  This value is conveyed to     *        {@link #tryAcquire} but is otherwise uninterpreted and     *        can represent anything you like.     */    public final void acquire(int arg) {        if (!tryAcquire(arg) &&            acquireQueued(addWaiter(Node.EXCLUSIVE), arg))            selfInterrupt();    }

/**     * Creates and enqueues node for current thread and given mode.     *     * @param mode Node.EXCLUSIVE for exclusive, Node.SHARED for shared     * @return the new node     */    private Node addWaiter(Node mode) {        //创建节点,设置关联线程和模式(独占或共享)        Node node = new Node(Thread.currentThread(), mode);        // Try the fast path of enq; backup to full enq on failure        Node pred = tail;        // 如果尾节点不为空,说明同步队列已经初始化过        if (pred != null) {            //新节点的前驱节点设置为尾节点            node.prev = pred;            // 设置新节点为尾节点            if (compareAndSetTail(pred, node)) {                //老的尾节点的后继节点设置为新的尾节点。 所以同步队列是一个双向列表。                pred.next = node;                return node;            }        }        //如果尾节点为空,说明队列还未初始化,需要初始化head节点并加入新节点        enq(node);        return node;    }

/**     * Inserts node into queue, initializing if necessary. See picture above.     * @param node the node to insert     * @return node's predecessor     */    private Node enq(final Node node) {        for (;;) {            Node t = tail;            if (t == null) { // Must initialize                // 如果tail为空,则新建一个head节点,并且tail和head都指向这个head节点                //队列头节点称作“哨兵节点”或者“哑节点”，它不与任何线程关联                if (compareAndSetHead(new Node()))                    tail = head;            } else {                //第二次循环进入这个分支，                node.prev = t;                if (compareAndSetTail(t, node)) {                    t.next = node;                    return t;                }            }        }    }

 

 /**     * Acquires in exclusive uninterruptible mode for thread already in     * queue. Used by condition wait methods as well as acquire.     *     * @param node the node     * @param arg the acquire argument     * @return {@code true} if interrupted while waiting     */    final boolean acquireQueued(final Node node, int arg) {        boolean failed = true;        try {            boolean interrupted = false;            for (;;) {                //获取当前node的前驱node                final Node p = node.predecessor();                //如果前驱node是head node，说明自己是第一个排队的线程，则尝试获锁                if (p == head && tryAcquire(arg)) {                    //把获锁成功的当前节点变成head node（哑节点）。                    setHead(node);                    p.next = null; // help GC                    failed = false;                    return interrupted;                }                if (shouldParkAfterFailedAcquire(p, node) &&                    parkAndCheckInterrupt())                    interrupted = true;            }        } finally {            if (failed)                cancelAcquire(node);        }    }

 

2  ReentrantLock

/**     * Sync object for fair locks     */    static final class FairSync extends Sync {        private static final long serialVersionUID = -3000897897090466540L;
        final void lock() {            acquire(1);        }
        /**         * Fair version of tryAcquire.  Don't grant access unless         * recursive call or no waiters or is first.         */        protected final boolean tryAcquire(int acquires) {            final Thread current = Thread.currentThread();            int c = getState();            if (c == 0) {                if (!hasQueuedPredecessors() &&                    compareAndSetState(0, acquires)) {                    setExclusiveOwnerThread(current);                    return true;                }            }            else if (current == getExclusiveOwnerThread()) {                int nextc = c + acquires;                if (nextc < 0)                    throw new Error("Maximum lock count exceeded");                setState(nextc);                return true;            }            return false;        }    }

1. 如果此时state（private volatile int state）是0，那么就表示这个时候没有人占有锁。但因为是公平锁，所以还要判断自己是不是首节点，然后才尝试把状态设置为1，假如成功的话，就成功的占有了锁。compareAndSetState 也是通过CAS来实现。CAS 是原子操作，而且state的类型是volatile，所以state 的值是线程安全的。

   
   

2. 如果此时状态不是0，那么再判断当前线程是不是锁的owner，如果是的话，则state 递增，当state溢出时，会抛错。如果没溢出，则返回true，表示成功获取锁。

   
   

3. 上述都不满足，则返回false，获取锁失败。

二  JVM层

public native void unpark(Object var1);
public native void park(boolean var1, long var2);

UNSAFE_ENTRY(void, Unsafe_Park(JNIEnv *env, jobject unsafe, jboolean isAbsolute, jlong time))  UnsafeWrapper("Unsafe_Park");  EventThreadPark event;#ifndef USDT2  HS_DTRACE_PROBE3(hotspot, thread__park__begin, thread->parker(), (int) isAbsolute, time);#else /* USDT2 */   HOTSPOT_THREAD_PARK_BEGIN(                             (uintptr_t) thread->parker(), (int) isAbsolute, time);#endif /* USDT2 */  JavaThreadParkedState jtps(thread, time != 0);  thread->parker()->park(isAbsolute != 0, time);#ifndef USDT2  HS_DTRACE_PROBE1(hotspot, thread__park__end, thread->parker());#else /* USDT2 */  HOTSPOT_THREAD_PARK_END(                          (uintptr_t) thread->parker());#endif /* USDT2 */  if (event.should_commit()) {    const oop obj = thread->current_park_blocker();    if (time == 0) {      post_thread_park_event(&event, obj, min_jlong, min_jlong);    } else {      if (isAbsolute != 0) {        post_thread_park_event(&event, obj, min_jlong, time);      } else {        post_thread_park_event(&event, obj, time, min_jlong);      }    }  }UNSAFE_END

class Parker : public os::PlatformParker {private:  volatile int _counter ;  ...public:  void park(bool isAbsolute, jlong time);  void unpark();  ...}class PlatformParker : public CHeapObj<mtInternal> {  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方法：

void Parker::park(bool isAbsolute, jlong time) {  // 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;
  if (Thread::is_interrupted(thread, false)) {    return;  }
  // Next, demultiplex/decode time arguments  timespec absTime;  if (time < 0 || (isAbsolute && time == 0) ) { // don't wait at all    return;  }  if (time > 0) {    unpackTime(&absTime, isAbsolute, time);  }
  ////进入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;  }  //到这里，意味着pthread_mutex_trylock(_mutex)成功  int status ;  if (_counter > 0)  { // no wait needed    _counter = 0;    status = pthread_mutex_unlock(_mutex);    assert (status == 0, "invariant") ;    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  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    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
  _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();  }}

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

   
   

2. 如果不成功，则把线程的状态设置成_thread_in_vm并且_thread_blocked。_thread_in_vm 表示线程当前在JVM中执行，_thread_blocked表示线程当前阻塞了。

   
   

3. 拿到mutex之后，再次检查_counter是不是>0，如果是，则把_counter设置为0，unlock mutex并返回

   
   

4. 如果_counter还是不大于0，则判断等待的时间是否等于0，然后调用相应的pthread_cond_wait系列函数进行等待，如果等待返回（即有人进行unpark，则pthread_cond_signal来通知），则把_counter设置为0，unlock mutex并返回。

三  GLIBC 层

#include <pthread.h>#include <stdio.h>#include <stdlib.h>
pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER; /*初始化互斥锁*/pthread_cond_t cond = PTHREAD_COND_INITIALIZER;  //初始化条件变量 
void *thread1(void *);void *thread2(void *);
int i=1;int main(void){        pthread_t t_a;        pthread_t t_b;        pthread_create(&t_a,NULL,thread1,(void *)NULL);/*创建进程t_a*/        pthread_create(&t_b,NULL,thread2,(void *)NULL); /*创建进程t_b*/        pthread_join(t_b, NULL);/*等待进程t_b结束*/        pthread_mutex_destroy(&mutex);        pthread_cond_destroy(&cond);        exit(0);}void *thread1(void *junk){        for(i=1;i<=9;i++)        {             pthread_mutex_lock(&mutex);//             if(i%3==0)                 pthread_cond_signal(&cond);/*条件改变，发送信号，通知t_b进程*/              else                         printf("thead1:%d/n",i);              pthread_mutex_unlock(&mutex);//*解锁互斥量*/              printf("Up Unlock Mutex/n");                    sleep(1);        }}void *thread2(void *junk){        while(i<9)        {            pthread_mutex_lock(&mutex);            if(i%3!=0)                   pthread_cond_wait(&cond,&mutex);/*等待*/            printf("thread2:%d/n",i);            pthread_mutex_unlock(&mutex);            printf("Down Ulock Mutex/n");            sleep(1);         }
}

1  pthread_mutex_lock

intPTHREAD_MUTEX_LOCK (pthread_mutex_t *mutex){  /* See concurrency notes regarding mutex type which is loaded from __kind     in struct __pthread_mutex_s in sysdeps/nptl/bits/thread-shared-types.h.  */  unsigned int type = PTHREAD_MUTEX_TYPE_ELISION (mutex);
  LIBC_PROBE (mutex_entry, 1, mutex);
  if (__builtin_expect (type & ~(PTHREAD_MUTEX_KIND_MASK_NP                                 | PTHREAD_MUTEX_ELISION_FLAGS_NP), 0))    return __pthread_mutex_lock_full (mutex);
  if (__glibc_likely (type == PTHREAD_MUTEX_TIMED_NP))    {      FORCE_ELISION (mutex, goto elision);    simple:      /* Normal mutex.  */      LLL_MUTEX_LOCK_OPTIMIZED (mutex);      assert (mutex->__data.__owner == 0);    }#if ENABLE_ELISION_SUPPORT  else if (__glibc_likely (type == PTHREAD_MUTEX_TIMED_ELISION_NP))    {  elision: __attribute__((unused))      /* This case can never happen on a system without elision,         as the mutex type initialization functions will not         allow to set the elision flags.  */      /* Don't record owner or users for elision case.  This is a         tail call.  */      return LLL_MUTEX_LOCK_ELISION (mutex);    }#endif  else if (__builtin_expect (PTHREAD_MUTEX_TYPE (mutex)                             == PTHREAD_MUTEX_RECURSIVE_NP, 1))    {      /* Recursive mutex.  */      pid_t id = THREAD_GETMEM (THREAD_SELF, tid);
      /* Check whether we already hold the mutex.  */      if (mutex->__data.__owner == id)        {          /* Just bump the counter.  */          if (__glibc_unlikely (mutex->__data.__count + 1 == 0))            /* Overflow of the counter.  */            return EAGAIN;
          ++mutex->__data.__count;
          return 0;        }
      /* We have to get the mutex.  */      LLL_MUTEX_LOCK_OPTIMIZED (mutex);
      assert (mutex->__data.__owner == 0);      mutex->__data.__count = 1;    }  else if (__builtin_expect (PTHREAD_MUTEX_TYPE (mutex)                          == PTHREAD_MUTEX_ADAPTIVE_NP, 1))    {      if (LLL_MUTEX_TRYLOCK (mutex) != 0)        {          int cnt = 0;          int max_cnt = MIN (max_adaptive_count (),                             mutex->__data.__spins * 2 + 10);          do            {              if (cnt++ >= max_cnt)                {                  LLL_MUTEX_LOCK (mutex);                  break;                }              atomic_spin_nop ();            }          while (LLL_MUTEX_TRYLOCK (mutex) != 0);
          mutex->__data.__spins += (cnt - mutex->__data.__spins) / 8;        }      assert (mutex->__data.__owner == 0);    }  else    {      pid_t id = THREAD_GETMEM (THREAD_SELF, tid);      assert (PTHREAD_MUTEX_TYPE (mutex) == PTHREAD_MUTEX_ERRORCHECK_NP);      /* Check whether we already hold the mutex.  */      if (__glibc_unlikely (mutex->__data.__owner == id))        return EDEADLK;      goto simple;    }
  pid_t id = THREAD_GETMEM (THREAD_SELF, tid);
  /* Record the ownership.  */  mutex->__data.__owner = id;#ifndef NO_INCR  ++mutex->__data.__nusers;#endif
  LIBC_PROBE (mutex_acquired, 1, mutex);
  return 0;}

typedef union{  struct __pthread_mutex_s  {    int __lock;    unsigned int __count;    int __owner;    unsigned int __nusers;    int __kind;    int __spins;    __pthread_list_t __list;  } __data;  ......} pthread_mutex_t;

/* Mutex types.  */enum{   PTHREAD_MUTEX_TIMED_NP,  PTHREAD_MUTEX_RECURSIVE_NP,  PTHREAD_MUTEX_ERRORCHECK_NP,  PTHREAD_MUTEX_ADAPTIVE_NP#if defined __USE_UNIX98 || defined __USE_XOPEN2K8  ,  PTHREAD_MUTEX_NORMAL = PTHREAD_MUTEX_TIMED_NP,  PTHREAD_MUTEX_RECURSIVE = PTHREAD_MUTEX_RECURSIVE_NP,  PTHREAD_MUTEX_ERRORCHECK = PTHREAD_MUTEX_ERRORCHECK_NP,  PTHREAD_MUTEX_DEFAULT = PTHREAD_MUTEX_NORMAL#endif#ifdef __USE_GNU  /* For compatibility.  */  , PTHREAD_MUTEX_FAST_NP = PTHREAD_MUTEX_TIMED_NP#endif};

• PTHREAD_MUTEX_TIMED_NP，这是缺省值，也就是普通锁。　

• PTHREAD_MUTEX_RECURSIVE_NP，可重入锁，允许同一个线程对同一个锁成功获得多次，并通过多次unlock解锁。

• PTHREAD_MUTEX_ERRORCHECK_NP，检错锁，如果同一个线程重复请求同一个锁，则返回EDEADLK，否则与PTHREAD_MUTEX_TIMED_NP类型相同。

• PTHREAD_MUTEX_ADAPTIVE_NP，自适应锁，自旋锁与普通锁的混合。　

# define LLL_MUTEX_LOCK_OPTIMIZED(mutex) lll_mutex_lock_optimized (mutex)

lll_mutex_lock_optimized (pthread_mutex_t *mutex){  /* The single-threaded optimization is only valid for private     mutexes.  For process-shared mutexes, the mutex could be in a     shared mapping, so synchronization with another process is needed     even without any threads.  If the lock is already marked as     acquired, POSIX requires that pthread_mutex_lock deadlocks for     normal mutexes, so skip the optimization in that case as     well.  */  int private = PTHREAD_MUTEX_PSHARED (mutex);  if (private == LLL_PRIVATE && SINGLE_THREAD_P && mutex->__data.__lock == 0)    mutex->__data.__lock = 1;  else    lll_lock (mutex->__data.__lock, private);}

#define lll_lock(futex, private)        \  __lll_lock (&(futex), private)

#define __lll_lock(futex, private)                                      \  ((void)                                                               \   ({                                                                   \     int *__futex = (futex);                                            \     if (__glibc_unlikely                                               \         (atomic_compare_and_exchange_bool_acq (__futex, 1, 0)))        \       {                                                                \         if (__builtin_constant_p (private) && (private) == LLL_PRIVATE) \           __lll_lock_wait_private (__futex);                           \         else                                                           \           __lll_lock_wait (__futex, private);                          \       }                                                                \   }))

void__lll_lock_wait (int *futex, int private){  if (atomic_load_relaxed (futex) == 2)    goto futex;
  while (atomic_exchange_acquire (futex, 2) != 0)    {    futex:      LIBC_PROBE (lll_lock_wait, 1, futex);      futex_wait ((unsigned int *) futex, 2, private); /* Wait if *futex == 2.  */    }}

• 0，代表当前锁空闲无锁，可以进行快速上锁，不需要进内核。

• 1，代表有线程持有当前锁，如果这时有其它线程需要上锁，就必须标记futex为“锁竞争”，然后通过futex系统调用进内核把当前线程挂起。

• 2，代表锁竞争，有其它线程将要或正在内核的futex系统中排队等待锁。

2  pthread_cond_wait

四  内核层

struct futex_hash_bucket {        atomic_t waiters;        spinlock_t lock;        struct plist_head chain;} ____cacheline_aligned_in_smp;

struct futex_q {        struct plist_node list;        struct task_struct *task;        spinlock_t *lock_ptr;        union futex_key key;   //唯一标识uaddr的key值        struct futex_pi_state *pi_state;        struct rt_mutex_waiter *rt_waiter;        union futex_key *requeue_pi_key;        u32 bitset;};

union futex_key {         struct {                unsigned long pgoff;                struct inode *inode;                int offset;        } shared;        struct {                unsigned long address;                struct mm_struct *mm;                int offset;        } private;         struct {                unsigned long word;                void *ptr;                int offset;        } both;};

static struct {        struct futex_hash_bucket *queues;        unsigned long            hashsize;} __futex_data __read_mostly __aligned(2*sizeof(long));
#define futex_queues   (__futex_data.queues)#define futex_hashsize (__futex_data.hashsize)

static int __init futex_init(void){        unsigned int futex_shift;        unsigned long i;
#if CONFIG_BASE_SMALL        futex_hashsize = 16;#else        futex_hashsize = roundup_pow_of_two(256 * num_possible_cpus());#endif
        futex_queues = alloc_large_system_hash("futex", sizeof(*futex_queues),                                               futex_hashsize, 0,                                               futex_hashsize < 256 ? HASH_SMALL : 0,                                               &futex_shift, NULL,                                               futex_hashsize, futex_hashsize);        futex_hashsize = 1UL << futex_shift;
        futex_detect_cmpxchg();
        for (i = 0; i < futex_hashsize; i++) {                atomic_set(&futex_queues[i].waiters, 0);                plist_head_init(&futex_queues[i].chain);                spin_lock_init(&futex_queues[i].lock);        }
        return 0;}

static int futex_wait(u32 __user *uaddr, unsigned int flags, u32 val,                      ktime_t *abs_time, u32 bitset){        struct hrtimer_sleeper timeout, *to = NULL;        struct restart_block *restart;        struct futex_hash_bucket *hb;        struct futex_q q = futex_q_init;        int ret;
        if (!bitset)                return -EINVAL;        q.bitset = bitset;
        if (abs_time) {                to = &timeout;                hrtimer_init_on_stack(&to->timer, (flags & FLAGS_CLOCKRT) ?                                      CLOCK_REALTIME : CLOCK_MONOTONIC,                                      HRTIMER_MODE_ABS);                hrtimer_init_sleeper(to, current);                hrtimer_set_expires_range_ns(&to->timer, *abs_time,                                             current->timer_slack_ns);        }
retry:        /*         * Prepare to wait on uaddr. On success, holds hb lock and increments         * q.key refs.         */        ret = futex_wait_setup(uaddr, val, flags, &q, &hb);        if (ret)                goto out;
        /* queue_me and wait for wakeup, timeout, or a signal. */        futex_wait_queue_me(hb, &q, to);
        /* If we were woken (and unqueued), we succeeded, whatever. */        ret = 0;        /* unqueue_me() drops q.key ref */        if (!unqueue_me(&q))                goto out;        ret = -ETIMEDOUT;        if (to && !to->task)                goto out;
        /*         * We expect signal_pending(current), but we might be the         * victim of a spurious wakeup as well.         */        if (!signal_pending(current))                goto retry;
        ret = -ERESTARTSYS;        if (!abs_time)                goto out;
        restart = &current->restart_block;        restart->fn = futex_wait_restart;        restart->futex.uaddr = uaddr;        restart->futex.val = val;        restart->futex.time = *abs_time;        restart->futex.bitset = bitset;        restart->futex.flags = flags | FLAGS_HAS_TIMEOUT;
        ret = -ERESTART_RESTARTBLOCK;
out:        if (to) {                hrtimer_cancel(&to->timer);                destroy_hrtimer_on_stack(&to->timer);        }        return ret;}

* * futex_wait_setup() - Prepare to wait on a futex * @uaddr:      the futex userspace address * @val:        the expected value * @flags:      futex flags (FLAGS_SHARED, etc.) * @q:          the associated futex_q * @hb:         storage for hash_bucket pointer to be returned to caller * * Setup the futex_q and locate the hash_bucket.  Get the futex value and * compare it with the expected value.  Handle atomic faults internally. * Return with the hb lock held and a q.key reference on success, and unlocked * with no q.key reference on failure. * * Return: *  -  0 - uaddr contains val and hb has been locked; *  - <1 - -EFAULT or -EWOULDBLOCK (uaddr does not contain val) and hb is unlocked */static int futex_wait_setup(u32 __user *uaddr, u32 val, unsigned int flags,                           struct futex_q *q, struct futex_hash_bucket **hb){        u32 uval;        int ret;        retry:        //初始化futex_q， 把uaddr设置到futex_key的字段中，将来futex_wake时也是通过这个key来查找futex。        ret = get_futex_key(uaddr, flags & FLAGS_SHARED, &q->key, VERIFY_READ);        if (unlikely(ret != 0))                return ret;
retry_private:        //根据key计算hash，然后在数组里找到对应的futex_hash_bucket        *hb = queue_lock(q);        //原子地将uaddr的值读到uval中        ret = get_futex_value_locked(&uval, uaddr);
        if (ret) {                queue_unlock(*hb);
                ret = get_user(uval, uaddr);                if (ret)                        goto out;
                if (!(flags & FLAGS_SHARED))                        goto retry_private;
                put_futex_key(&q->key);                goto retry;        }        //如果当前uaddr指向的值不等于val，即说明其他进程修改了  //uaddr指向的值，等待条件不再成立，不用阻塞直接返回。       if (uval != val) {                queue_unlock(*hb);                ret = -EWOULDBLOCK;        }
out:        if (ret)                put_futex_key(&q->key);        return ret;}

/** * futex_wait_queue_me() - queue_me() and wait for wakeup, timeout, or signal * @hb:         the futex hash bucket, must be locked by the caller * @q:          the futex_q to queue up on * @timeout:    the prepared hrtimer_sleeper, or null for no timeout */static void futex_wait_queue_me(struct futex_hash_bucket *hb, struct futex_q *q,                                struct hrtimer_sleeper *timeout){        /*         * The task state is guaranteed to be set before another task can         * wake it. set_current_state() is implemented using smp_store_mb() and         * queue_me() calls spin_unlock() upon completion, both serializing         * access to the hash list and forcing another memory barrier.         */        set_current_state(TASK_INTERRUPTIBLE);        queue_me(q, hb);
        /* Arm the timer */        if (timeout)                hrtimer_start_expires(&timeout->timer, HRTIMER_MODE_ABS);
        /*         * If we have been removed from the hash list, then another task         * has tried to wake us, and we can skip the call to schedule().         */        if (likely(!plist_node_empty(&q->list))) {                /*                 * If the timer has already expired, current will already be                 * flagged for rescheduling. Only call schedule if there                 * is no timeout, or if it has yet to expire.                 */                if (!timeout || timeout->task)                        freezable_schedule();        }        __set_current_state(TASK_RUNNING);}

1. 将当前进程插入到等待队列，就是把futex_q 挂到futex_hash_bucket上
2. 启动定时任务
3. 主动触发内核进程调度

五  总结