glibc nptl库pthread_mutex_lock和pthread_mutex_unlock浅析

//pthread_mutex_t 互斥量属性
//PTHREAD_MUTEX_TIMED_NP，这是缺省值，也就是普通锁。首先进行一次CAS，如果失败则陷入内核态然后挂起线程
//PTHREAD_MUTEX_RECURSIVE_NP，可重入锁，允许同一个线程对同一个锁成功获得多次，并通过多次unlock解锁。如果是不同线程请求，则在加锁线程解锁时重新竞争。
// PTHREAD_MUTEX_ERRORCHECK_NP，检错锁，如果同一个线程请求同一个锁，则返回EDEADLK，否则与PTHREAD_MUTEX_TIMED_NP类型动作相同。这样就保证当不允许多次加锁时不会出现最简单情况下的死锁。
//PTHREAD_MUTEX_ADAPTIVE_NP，适应锁，此锁在多核处理器下首先进行自旋获取锁，如果自旋次数超过配置的最大次数，则也会陷入内核态挂起。
int
__pthread_mutex_lock (pthread_mutex_t *mutex)
{
  assert (sizeof (mutex->__size) >= sizeof (mutex->__data));
  //获取互斥锁类型
  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是一个宏
      LLL_MUTEX_LOCK (mutex);
      assert (mutex->__data.__owner == 0);
    }
  //如果具有可重入属性，也就是递归锁
  else if (__builtin_expect (PTHREAD_MUTEX_TYPE (mutex)
			     == PTHREAD_MUTEX_RECURSIVE_NP, 1))
    {
      /* Recursive mutex.  */
      //获取当前线程id
      pid_t id = THREAD_GETMEM (THREAD_SELF, tid);

      //判断当前线程是否已经持有这个锁，也就是互斥锁的__owner属性就是持有它的线程id
      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;
	}
      //如果是第一次获取锁则使用LLL_MUTEX_LOCK宏获取锁
      /* We have to get the mutex.  */
      LLL_MUTEX_LOCK (mutex);

      assert (mutex->__data.__owner == 0);
      //锁被持有的次数初始化为1
      mutex->__data.__count = 1;
    }
   //如果是PTHREAD_MUTEX_ADAPTIVE_NP类型的锁
  else if (__builtin_expect (PTHREAD_MUTEX_TYPE (mutex)
			  == PTHREAD_MUTEX_ADAPTIVE_NP, 1))
    {
    //非smp架构，则直接使用LLL_MUTEX_LOCK宏获取锁
      if (! __is_smp)
	goto simple;
    //如果是smp架构则使用LLL_MUTEX_TRYLOCK宏获取锁
      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获取锁
		  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
    {
      //这个分支就是PTHREAD_MUTEX_ERRORCHECK_NP类型的锁
      //获取线程id
      pid_t id = THREAD_GETMEM (THREAD_SELF, tid);
      assert (PTHREAD_MUTEX_TYPE (mutex) == PTHREAD_MUTEX_ERRORCHECK_NP);
      /* Check whether we already hold the mutex.  */
      //如果当前线程id和持有mutex锁的线程id一样，也就是同一个线程重复获取锁，就会返回错误
      if (__glibc_unlikely (mutex->__data.__owner == id))
	return EDEADLK;
      //如果不是同一个线程则按照一般的LLL_MUTEX_LOCK方式去获取锁
      goto simple;
    }

  pid_t id = THREAD_GETMEM (THREAD_SELF, tid);

  /* Record the ownership.  */
  //记录获取锁的线程id
  mutex->__data.__owner = id;
#ifndef NO_INCR
  ++mutex->__data.__nusers;
#endif

  LIBC_PROBE (mutex_acquired, 1, mutex);

  return 0;
}

# define LLL_MUTEX_LOCK(mutex) \
  lll_lock ((mutex)->__data.__lock, PTHREAD_MUTEX_PSHARED (mutex))
# define LLL_MUTEX_TRYLOCK(mutex) \
  lll_trylock ((mutex)->__data.__lock)
/* 
 * CAS操作的核心宏，cas操作判断(mutex)->__data.__lock的值是否为0，如果为0，则置为1，ZF=1
 * 1.判断是否在多线程环境下
 * 2.如果不是多线程环境则直接调用cmpxchgl指令进行cas操作，如果是多线程则需要在cmpxchgl指令前
 * 加上lock指令
 * 3.如果cas成功则跳到标号18，如果cas失败则调用__lll_lock_wait子程序
 */
# define __lll_lock_asm_start "cmpl $0, %%gs:%P6\n\t"			      \
			      "je 0f\n\t"				      \
			      "lock\n"					      \
			      "0:\tcmpxchgl %1, %2\n\t"
//LLL_PRIVATE为0，所以不会走第一个分支，走第二个分支
#define lll_lock(futex, private) \
  (void)								      \
    ({ int ignore1, ignore2;						      \
       if (__builtin_constant_p (private) && (private) == LLL_PRIVATE)	      \
	 __asm __volatile (__lll_lock_asm_start				      \
			   "jz 18f\n\t"				      \
			   "1:\tleal %2, %%ecx\n"			      \
			   "2:\tcall __lll_lock_wait_private\n" 	      \
			   "18:"					      \
			   : "=a" (ignore1), "=c" (ignore2), "=m" (futex)     \
			   : "0" (0), "1" (1), "m" (futex),		      \
			     "i" (MULTIPLE_THREADS_OFFSET)		      \
			   : "memory");					      \
       else								      \
	 {								      \
	   int ignore3;							      \
	   __asm __volatile (__lll_lock_asm_start			      \
			     "jz 18f\n\t"			 	      \
			     "1:\tleal %2, %%edx\n"			      \
			     "0:\tmovl %8, %%ecx\n"			      \
			     "2:\tcall __lll_lock_wait\n"		      \
			     "18:"					      \
			     : "=a" (ignore1), "=c" (ignore2),		      \
			       "=m" (futex), "=&d" (ignore3) 		      \
			     : "1" (1), "m" (futex),			      \
			       "i" (MULTIPLE_THREADS_OFFSET), "0" (0),	      \
			       "g" ((int) (private))			      \
			     : "memory");				      \
	 }								      \
    })
//子程序开始主要进行sys_futex系统调用
//从以上内联汇编代码可知，edx = futex，ecx = 0
//系统调用传参规则，在不超过6个参数的函数调用，从左到右参数分别使用ebx,	ecx
//edx,esi,edi,ebp表示		
__lll_lock_wait:
	pushl	%edx
	pushl	%ebx
	pushl	%esi
	//ebx是sys_futex系统调用的第一个参数为futex
	movl	%edx, %ebx
	//edx是第三个参数 值为2
	movl	$2, %edx
	xorl	%esi, %esi	/* No timeout.  */
	//ecx是第二个参数也就是 FUTEX_WAIT标志
	LOAD_FUTEX_WAIT (%ecx)
    //此时eax = 0，edx = 2，所以跳到标号2
	cmpl	%edx, %eax	/* NB:	 %edx == 2 */
	jne 2f

1:	movl	$SYS_futex, %eax
	ENTER_KERNEL
//将edx赋值给eax 此时eax = 2
2:	movl	%edx, %eax
//互斥锁和eax的值进行交换，互斥锁的值为2，eax = 1或0
	xchgl	%eax, (%ebx)	/* NB:	 lock is implied */
//eax和eax进行与操作，不等于0则跳到1，否则就执行出栈
//等于0说明其它线程释放了锁，此时不需要休眠而是重新去获取锁
	testl	%eax, %eax
	jnz	1b

	popl	%esi
	popl	%ebx
	popl	%edx
	ret

int
internal_function attribute_hidden
__pthread_mutex_unlock_usercnt (pthread_mutex_t *mutex, int decr)
{
	//获取锁类型
  int type = PTHREAD_MUTEX_TYPE_ELISION (mutex);
  //如果是普通类型
  if (__builtin_expect (type, PTHREAD_MUTEX_TIMED_NP)
      == PTHREAD_MUTEX_TIMED_NP)
    {
      /* Always reset the owner field.  */
    normal:
      mutex->__data.__owner = 0;
      if (decr)
	/* One less user.  */
	--mutex->__data.__nusers;

      /*使用lll_unlock宏进行解锁  */
      lll_unlock (mutex->__data.__lock, PTHREAD_MUTEX_PSHARED (mutex));

      LIBC_PROBE (mutex_release, 1, mutex);

      return 0;
    }
  else if (__glibc_likely (type == PTHREAD_MUTEX_TIMED_ELISION_NP))
    {
      /* Don't reset the owner/users fields for elision.  */
      return lll_unlock_elision (mutex->__data.__lock, mutex->__data.__elision,
				      PTHREAD_MUTEX_PSHARED (mutex));
    }
  else if (__builtin_expect (PTHREAD_MUTEX_TYPE (mutex)
			      == PTHREAD_MUTEX_RECURSIVE_NP, 1))
    {
      /* Recursive mutex.  */
      if (mutex->__data.__owner != THREAD_GETMEM (THREAD_SELF, tid))
	return EPERM;
      
      if (--mutex->__data.__count != 0)
	/* We still hold the mutex.  */
	return 0;
      goto normal;
    }
  else if (__builtin_expect (PTHREAD_MUTEX_TYPE (mutex)
			      == PTHREAD_MUTEX_ADAPTIVE_NP, 1))
    goto normal;
  else
    {
      /* Error checking mutex.  */
      assert (type == PTHREAD_MUTEX_ERRORCHECK_NP);
      if (mutex->__data.__owner != THREAD_GETMEM (THREAD_SELF, tid)
	  || ! lll_islocked (mutex->__data.__lock))
	return EPERM;
      goto normal;
    }
}

//1.将ebx设置为互斥量在用户空间的地址，sys_futex的第一个参数
//2.将互斥量置为0，表示释放了锁
//3.ecx是操作数FUTEX_WAKE，也就是sys_futex的op参数
//4.将edx设置为1 表示只会唤醒一个线程，sys_futex的第三个参数，然后开始sys_futex系统调用
__lll_unlock_wake:
	pushl	%ebx
	pushl	%ecx
	pushl	%edx

	movl	%eax, %ebx
	movl	$0, (%eax)
	LOAD_FUTEX_WAKE (%ecx)
	movl	$1, %edx	/* Wake one thread.  */
	movl	$SYS_futex, %eax
	ENTER_KERNEL

	popl	%edx
	popl	%ecx
	popl	%ebx
	ret

/*
 *pthread_mutex_lock，pthread_cond_wait,pthread_cond_signal都是调用此函数.如果在用户态CAS失败，
 *则进行系统调用进行挂起操作
 *uaddr表示互斥量在用户空间的地址，op表示操作类型，utime表示挂起时间
 */

asmlinkage long sys_futex(u32 __user *uaddr, int op, u32 val,
			  struct timespec __user *utime, u32 __user *uaddr2,
			  u32 val3)
{
	struct timespec t;
	//初始化timeout
	unsigned long timeout = MAX_SCHEDULE_TIMEOUT;
	u32 val2 = 0;
    //如果设置了休眠时间并且操作类型是FUTEX_WAIT
	if (utime && (op == FUTEX_WAIT || op == FUTEX_LOCK_PI)) {
		//将休眠时间从用户空间复制到变量t
		if (copy_from_user(&t, utime, sizeof(t)) != 0)
			return -EFAULT;
		//校验休眠时间
		if (!timespec_valid(&t))
			return -EINVAL;
		//如果操作类型是FUTEX_WAIT，则将休眠时间转换为系统的时钟
		if (op == FUTEX_WAIT)
			timeout = timespec_to_jiffies(&t) + 1;
		else {
			timeout = t.tv_sec;
			val2 = t.tv_nsec;
		}
	}
	/*
	 * requeue parameter in 'utime' if op == FUTEX_REQUEUE.
	 */
	if (op == FUTEX_REQUEUE || op == FUTEX_CMP_REQUEUE)
		val2 = (u32) (unsigned long) utime;
    //调用do_futex
	return do_futex(uaddr, op, val, timeout, uaddr2, val2, val3);
}

//根据操作类型选择具体的函数，比如加锁失败就会调用futex_wait挂起当前线程
long do_futex(u32 __user *uaddr, int op, u32 val, unsigned long timeout,
		u32 __user *uaddr2, u32 val2, u32 val3)
{
	int ret;

	switch (op) {
	case FUTEX_WAIT:
		ret = futex_wait(uaddr, val, timeout);
		break;
	case FUTEX_WAKE:
		ret = futex_wake(uaddr, val);
		break;
	case FUTEX_FD:
		/* non-zero val means F_SETOWN(getpid()) & F_SETSIG(val) */
		ret = futex_fd(uaddr, val);
		break;
	case FUTEX_REQUEUE:
		ret = futex_requeue(uaddr, uaddr2, val, val2, NULL);
		break;
	case FUTEX_CMP_REQUEUE:
		ret = futex_requeue(uaddr, uaddr2, val, val2, &val3);
		break;
	case FUTEX_WAKE_OP:
		ret = futex_wake_op(uaddr, uaddr2, val, val2, val3);
		break;
	case FUTEX_LOCK_PI:
		ret = futex_lock_pi(uaddr, val, timeout, val2, 0);
		break;
	case FUTEX_UNLOCK_PI:
		ret = futex_unlock_pi(uaddr);
		break;
	case FUTEX_TRYLOCK_PI:
		ret = futex_lock_pi(uaddr, 0, timeout, val2, 1);
		break;
	default:
		ret = -ENOSYS;
	}
	return ret;
}

//uaddr 互斥锁用户空间地址，val = 2，time是休眠时间
static int futex_wait(u32 __user *uaddr, u32 val, unsigned long time)
{
	//获取当前进程
	struct task_struct *curr = current;
	//创建一个等待队列
	DECLARE_WAITQUEUE(wait, curr);
	//哈希数组下标地址
	struct futex_hash_bucket *hb;
	//哈希元素
	struct futex_q q;
	u32 uval;
	int ret;

	q.pi_state = NULL;
 retry:
	down_read(&curr->mm->mmap_sem);
    //初始化futex_q的key成员，主要是给key的address，mm,offset赋值，address是互斥锁
    //在用户空间的地址进行4KB对齐的值，mm就是当前进程的mm，offset就是互斥锁和4KB对齐值的偏移量
	ret = get_futex_key(uaddr, &q.key);
	if (unlikely(ret != 0))
		goto out_release_sem;
    //获取哈希数组的下标地址
	hb = queue_lock(&q, -1, NULL);
	//以原子方式获取用户空间的互斥锁的值
	ret = get_futex_value_locked(&uval, uaddr);
    //如果获取失败
	if (unlikely(ret)) {
		queue_unlock(&q, hb);

		/*
		 * If we would have faulted, release mmap_sem, fault it in and
		 * start all over again.
		 */
		up_read(&curr->mm->mmap_sem);

		ret = get_user(uval, uaddr);

		if (!ret)
			goto retry;
		return ret;
	}
	ret = -EWOULDBLOCK;
	//如果互斥锁的值改变了则直接返回错误
	if (uval != val)
		goto out_unlock_release_sem;

	/* 如果互斥锁的值保持不变，则将q加入哈希表  */
	__queue_me(&q, hb);

	/*
	 *同步信号量
	 */
	up_read(&curr->mm->mmap_sem);

	/*
	 * There might have been scheduling since the queue_me(), as we
	 * cannot hold a spinlock across the get_user() in case it
	 * faults, and we cannot just set TASK_INTERRUPTIBLE state when
	 * queueing ourselves into the futex hash.  This code thus has to
	 * rely on the futex_wake() code removing us from hash when it
	 * wakes us up.
	 */

	/* add_wait_queue is the barrier after __set_current_state. */
	//将当前线程设置为休眠可中断状态
	__set_current_state(TASK_INTERRUPTIBLE);
	//加入等待队列
	add_wait_queue(&q.waiters, &wait);
	/*
	 * 如果等待队列不为null
	 */
	if (likely(!list_empty(&q.list)))
		//重新进行调度
		time = schedule_timeout(time);
	//将当前进程重新设置为运行态
	__set_current_state(TASK_RUNNING);

	/*
	 * NOTE: we don't remove ourselves from the waitqueue because
	 * we are the only user of it.
	 */

	/* 从哈希表删除key */
	if (!unqueue_me(&q))
		return 0;
	if (time == 0)
		return -ETIMEDOUT;
	/*
	 * We expect signal_pending(current), but another thread may
	 * have handled it for us already.
	 */
	return -EINTR;

 out_unlock_release_sem:
	queue_unlock(&q, hb);

 out_release_sem:
	up_read(&curr->mm->mmap_sem);
	return ret;
}

//futex哈希链表
struct futex_hash_bucket {
       spinlock_t              lock;
       struct list_head       chain;
};
//futex哈希数组
static struct futex_hash_bucket futex_queues[1<<FUTEX_HASHBITS];
//futext key联合体，分为共享，私有，二者结合，由于分析的是线程同步所以主要关注私有的
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;
};

struct futex_q {
	struct list_head list;
	//等待队列元素
	wait_queue_head_t waiters;

	/* 哈希表插入时的自旋锁 */
	spinlock_t *lock_ptr;

	/* 哈希元素的key */
	union futex_key key;

	/* For fd, sigio sent using these: */
	int fd;
	struct file *filp;

	/* Optional priority inheritance state: */
	struct futex_pi_state *pi_state;
	//哈希元素指向的进程
	struct task_struct *task;
};


static inline struct futex_hash_bucket *
queue_lock(struct futex_q *q, int fd, struct file *filp)
{
	struct futex_hash_bucket *hb;

	q->fd = fd;
	q->filp = filp;
    //初始化等待队列
	init_waitqueue_head(&q->waiters);
    
	get_key_refs(&q->key);
	//返回q->key哈希到的数组下标地址
	hb = hash_futex(&q->key);
	q->lock_ptr = &hb->lock;
    //加自旋锁
	spin_lock(&hb->lock);
	return hb;
}
static inline void __queue_me(struct futex_q *q, struct futex_hash_bucket *hb)
{
	//将q插入到哈希链表
	list_add_tail(&q->list, &hb->chain);
	//q的任务属性设置为当前进程
	q->task = current;
	//自旋锁解锁
	spin_unlock(&hb->lock);
}

fastcall signed long __sched schedule_timeout(signed long timeout)
{
	//定义定时器
	struct timer_list timer;
	unsigned long expire;

	switch (timeout)
	{
	//如果休眠时间是最大值，则不用设置定时器，直接调用schedule调用其它进程运行
	//此时需要由其它进程唤醒，否则会一直睡眠
	case MAX_SCHEDULE_TIMEOUT:
		schedule();
		goto out;
	default:
		/*
		 * 如果timeout小与0 则出错
		 */
		if (timeout < 0)
		{
			printk(KERN_ERR "schedule_timeout: wrong timeout "
				"value %lx from %p\n", timeout,
				__builtin_return_address(0));
			current->state = TASK_RUNNING;
			goto out;
		}
	}
    //如果timeout是正常定时，则设置过期时间
	expire = timeout + jiffies;
    //设置定时器,process_timeout是定时器超时后，回调的函数
	setup_timer(&timer, process_timeout, (unsigned long)current);
	__mod_timer(&timer, expire);
	//调用schedule调度其它进程运行
	schedule();
	//当前进程重新被调度运行时删除定时器
	del_singleshot_timer_sync(&timer);
    //计算超时时间
	timeout = expire - jiffies;
    //如果返回0，说明定时时间到了，否则说明定时时间没到就被唤醒了。
 out:
	return timeout < 0 ? 0 : timeout;
}

/*
 *  timer：定时器
 * function：定时器到期后触发的回调函数
 *  data：表示当前进程的task_struct
 *
 */
static inline void setup_timer(struct timer_list * timer,
				void (*function)(unsigned long),
				unsigned long data)
{
	timer->function = function;
	timer->data = data;
	init_timer(timer);
}
//定时器超时后回调的函数
static void process_timeout(unsigned long __data)
{
	//唤醒和定时器绑定的进程，主要操作就是将__data进程从等待队列，移动到运行队列
	//并且设置进程状态，然后重新调度
	wake_up_process((struct task_struct *)__data);
}

static void wake_futex(struct futex_q *q)
{
	//将哈希元素从哈希链表删除
	list_del_init(&q->list);
	if (q->filp)
		send_sigio(&q->filp->f_owner, q->fd, POLL_IN);
	/*
	 * 唤醒等待的线程
	 */
	wake_up_all(&q->waiters);
	/*
	 * The waiting task can free the futex_q as soon as this is written,
	 * without taking any locks.  This must come last.
	 *
	 * A memory barrier is required here to prevent the following store
	 * to lock_ptr from getting ahead of the wakeup. Clearing the lock
	 * at the end of wake_up_all() does not prevent this store from
	 * moving.
	 */
	wmb();
	q->lock_ptr = NULL;
}