FreeBSD内核锁机制一

FreeBSD内核锁机制一（转）

English:

2008-01-07 9:41

在FreeBSD5.0中，有很多类型的锁：互斥体(struct mtx)、共享/独占锁(struct sx)、lockmgr锁(struct lock)、条件变量(struct cv)和信号量(struct sema)。本文将探讨这些锁机制的含义、应用和实现。另外，有一种文件锁(struct lockf)用于文件字段的保护，嵌在inode结构中使用，是属于另外的范畴，这里不做说明，有兴趣可以参考flock(2)和VOP_ADVLOCK (9)。
1 基本锁机制
struct lock_class {
const char *lc_name;
u_int lc_flags;
};
本文讨论的锁机制，任何一种类型都直接/间接的包含该结构。每一个锁都有一个类(lock_class)，该类描述了某个锁类(lc_name)和基本特性(lc_flags)。
其 中，lc_flags的宏定义：LC_SLEEPLOCK是普通锁；LC_SPINLOCK是自旋锁；LC_SLEEPABLE说明可以在对该锁加锁的情 况下，休眠；LC_RECURSABLE说明该锁是否支持递归；LC_UPGRADABLE说明该锁支持从共享锁变为独占锁，抑或相反。
目前，系统支持如下的lock_class：
struct lock_class lock_class_mtx_sleep = {
"sleep mutex",
LC_SLEEPLOCK | LC_RECURSABLE
};
struct lock_class lock_class_mtx_spin = {
"spin mutex",
LC_SPINLOCK | LC_RECURSABLE
};
struct lock_class lock_class_sx = {
"sx",
LC_SLEEPLOCK | LC_SLEEPABLE | LC_RECURSABLE | LC_UPGRADABLE
};
前两个说明了互斥体支持的两个类(MTX_DEF、MTX_SPIN)；后一个说明了共享/独占锁的类。
struct lock_object {
struct lock_class *lo_class;
const char *lo_name; /* Individual lock name. */
const char *lo_type; /* General lock type. */
u_int lo_flags;
TAILQ_ENTRY(lock_object) lo_list; /* List of all locks in system. */
struct witness *lo_witness;
};
用于存储一个具体锁的数据结构，lo_name是单个具体锁的名字；如果在锁初始化时指定了类型描述的字符串，则赋给lo_type，否则lo_type的值和lo_name一样。
成员lo_flags的值，及其说明如下：
#define LO_CLASSFLAGS 0x0000ffff /* Class specific flags. */
#define LO_INITIALIZED 0x00010000 /* Lock has been initialized. */
#define LO_WITNESS 0x00020000 /* Should witness monitor this lock. */
#define LO_QUIET 0x00040000 /* Don't log locking operations. */
#define LO_RECURSABLE 0x00080000 /* Lock may recurse. */
#define LO_SLEEPABLE 0x00100000 /* Lock may be held while sleeping. */
#define LO_UPGRADABLE 0x00200000 /* Lock may be upgraded/downgraded. */
#define LO_DUPOK 0x00400000 /* Don't check for duplicate acquires */
余下两个成员，lo_list用于加入全局all_locks链表中，lo_witness用于指向witness数据结构。这两个成员的作用是用于witness模块，跟踪锁的获得、释放。这个部分需要参考witness(9)。
本节介绍了FreeBSD锁机制的最基本的结构和意义，下面介绍各种锁机制的基本含义。
2互斥体(Mutex)
FreeBSD5.0进程同步控制的基本和主要方法是Mutex。在设计时，主要有如下考虑：
Ø 申请和释放一个无争议的互斥体应该尽可能的简单。
Ø 互斥体的数据结构应该有足够的信息和存储空间来支持优先级的继承。
Ø 一个进程应该可以递归地使用某一个互斥体(比如最常用的全局互斥体Giant)。
根据Mutex阻塞时，是否做上下文切换，可以分为两类：MTX_DEF(做切换)和MTX_SPIN(不做切换)。
MTX_DEF：当内核线程不能获得锁时，该线程会放弃CPU资源；采用这样类型的锁，不必考虑产生死锁的情况。
MTX_SPIN：当内核线程不能获得锁时，该线程不会放弃CPU资源，它会自旋，等待直到请求的锁被其它CPU释放。这样做，可能会产生死锁的情况，因此，我们需要在使用这种类型的锁屏蔽本地CPU的所有中断。
关于互斥体的其他类型可以和以上两种类型并存，它们说明了互斥体的其它属性。
MTX_RECURSE：具有该标志的锁，可以被递归使用。
MTX_QUIET：对该锁不做任何日志操作。
MTX_NOWITNESS：通知witness，不必跟踪该锁。
MTX_DUPOK：当该锁被复制时，witness不必写日志消息。
2.1 互斥体数据结构
struct mtx {
struct lock_object mtx_object; /* Common lock properties. */
volatile uintptr_t mtx_lock; /* Owner and flags. */
volatile u_int mtx_recurse; /* Number of recursive holds. */
TAILQ_HEAD(, thread) mtx_blocked; /* Threads blocked on us. */
LIST_ENTRY(mtx) mtx_contested; /* Next contested mtx. */
/* Fields used only if MUTEX_PROFILING is configured: */
u_int64_t mtx_acqtime;
const char *mtx_filename;
int mtx_lineno;
};
成员mtx_object的意义和前面介绍lock_object结构所说的一样，其lo_class只能是lock_class_mtx_sleep或lock_class_mtx_spin。
成员mtx_lock是对互斥体锁状态的额外说明，其值如下：
#define MTX_RECURSED 0x00000001 /* lock recursed (for MTX_DEF only) */
#define MTX_CONTESTED 0x00000002 /* lock contested (for MTX_DEF only) */
#define MTX_UNOWNED 0x00000004 /* Cookie for free mutex */
成员mtx_recurse是统计递归调用该锁次数的计数器。
成员mtx_blocked是阻塞于该锁的线程链表的头指针。
成员mtx_contested作为一个实体，链接在thread结构中td_contested指向的该线程的竞争互斥体链表中。
余下的三个成员用于Mutex调试。
2.2 Mutex接口函数
函数void mtx_init(struct mtx *mutex, const char *name, const char *type, int opts)：
在互斥体锁能够被使用之前，该锁mutex必须通过mtx_init初始化；入参name和type都是说明性的字符串，用于调试和witness；opts指明了该锁的类型和特性。
函数void mtx_lock_flags(struct mtx *mutex, int flags);
该函数是为目前正在运行的内核线程A申请一个相互排斥的MTX_DEF类型锁，如果正在运行的另一个内核线程B持有该锁，则线程A会放弃CPU资源，直到该锁能被再次使用，通常在等待该锁，线程A会sleep。函数mtx_lock()是其一个特例。
函数void mtx_lock_spin_flags(struct mtx *mutex, int flags);
该 函数是为目前正在运行的内核线程A申请一个相互排斥的MTX_SPIN类型锁，如果正在运行的另一个内核线程B持有该锁，则线程A会自旋直到该锁能够被再 次使用，在线程A自旋及获得该锁的过程中，对于线程A而言，所有的中断应该屏蔽。函数mtx_lock_spin()是其一个特例。
函数int mtx_trylock_flags(struct mtx *mutex, int flags)：
该函数试图获得一个互斥体mutex，如果不能，则立即返回0，如果可以，则获得该锁，并返回非零值。函数mtx_trylock()是其一个特例。
函数void mtx_unlock_flags(struct mtx *mutex, int flags)：
该函数释放其持有的相互排斥的MTX_DEF类型锁，如果有一个高优先级的线程正在等待该锁，则释放该锁的线程释放CPU资源以便高优先级的线程能够马上获得该锁，并执行。只有在释放该锁的线程处于临界区时，是例外。函数mtx_unlock()是其一个特例。
函数void mtx_unlock_spin_flags(struct mtx *mutex, int flags)：
该函数释放其持有的相互排斥的MTX_ SPIN类型锁，恢复在获得该锁前的中断状态。函数mtx_unlock_spin()是其一个特例。
函数void mtx_destroy(struct mtx *mutex)：
该函数用于销毁一个互斥体锁，并释放或重写该锁的所有数据。一个没有初始化(mtx_init)的锁是不应该被销毁的。如果该锁只有一个持有统计数时，是允许被销毁的。当该锁处于递归调用中，或是有另外的进程阻塞于该锁时，该锁是不允许被销毁的。
宏int mtx_initialized(struct mtx *mutex)：如果该锁已经初始化了，返回非零，否则返回零。宏int mtx_owned(struct mtx *mutex)：如果当前线程持有该锁，返回非零，否则返回零。宏int mtx_recursed(struct mtx *mutex)：如果该锁正在被递归使用，则返回非零，否则返回零。
前面，我们提到的mtx_xxx_flag函数系列，其flags入参说明相关锁作是否需要log日志(MTX_QUIET)；相应的mtx_xxx函数系列，是不须log日志。
另外，函数void mtx_assert(struct mtx *mutex, int what)：是用于诊断的，于互斥体锁机制的原理关系并不紧密，因此，不作过多论述。
宏MTX_SYSINIT是对互斥体锁初始化的一个封装。
2.3 系统对互斥体子系统以及互斥体池的初始化
我们以i386体系为例，在内核加载的很早的时候(在与平台相关的初始化阶段，产生局部描述符表(LDT)之前)，互斥体锁子系统就已经初始化了(在init386()函数中调用mutex_init()函数实现)。
/* Setup thread0 so that mutexes work. */
LIST_INIT(&thread0.td_contested);
/* Initialize mutexes. */
mtx_init(&Giant, "Giant", NULL, MTX_DEF | MTX_RECURSE);
mtx_init(&sched_lock, "sched lock", NULL, MTX_SPIN | MTX_RECURSE);
mtx_init(&proc0.p_mtx, "process lock", NULL, MTX_DEF | MTX_DUPOK);
mtx_lock(&Giant);
初始化第一个线程的竞争互斥体链表。
初始化Giant全局锁，支持递归调用的MTX_DEF类型。锁Giant被广泛地用在内核的几乎每个角落，保护内核一般性代码。
初始化sched_lock全局锁，支持递归调用的MTX_SPIN类型。锁sched_lock用于保护内核调度队列。
以 上两个锁都是可递归的，这里，需要补充说明一下可递归锁的意义，一个非递归锁加锁后，是不能够被再次加锁；而当递归锁遇到这样的情况，首先检查加锁的持有 者，如果是当前进程，那么对递归计数器加1，如果不是当前进程，则按非递归锁那样处理；当解锁一个递归锁时，递归计数器减1，当计数器为0时，正真解锁 (和非递归锁调用mtx_unlock_xxx()一样)。
初始化proc0变量的内部锁p_mtx，该锁用于保护对proc结构的操作。
最 后对Giant加锁。为什么必需在这里就对Giant加锁？因为Giant必须先于其他任何互斥体先获得锁。即：在持有其他锁的情况下，是不可能以非递归 的方式获得Giant(即第一次调用mtx_lock(&Giant))；在持有Giant的情况下，时可以获得其他互斥体；在持有其他锁的情况 下，是可以以递归方式获得Giant。
在内核加载的稍后阶段，会初始化互斥体池(在mi_startup()函数中，通过加载SI_SUB_MTX_POOL子模块实现)。
初始化一个长度为MTX_POOL_SIZE(128)的互斥锁数组mtx_pool_ary。互斥锁池主要用在lock和sx中，参考lockinit(9)和sx(9)。以后涉及到一般性锁和共享/独占锁时，再讨论其用法。
2.4 互斥体加锁/解锁的过程
互斥体锁的初始化和销毁函数非常简单，这里就不做论述了，有兴趣的读者可以参考源码的实现。本节主要论述互斥体锁的加锁/解锁的过程。
MTX_DEF：
获得/释放互斥体锁(MTX_DEF)的原子操作：
获 得互斥体锁是通过_obtain_lock宏(展开后是atomic_cmpset_int()嵌入汇编的内联函数)实现的，其思想是：用mtx的成员 mtx_lock与MTX_UNOWNED比较，如果相同，则说明没有加锁，则将当前线程(curthread)的地址传给mtx_lock，并返回非 零；否则，说明该锁已经被加锁，不做任何操作，返回零。
释放互斥体锁是通过_release_lock宏(展开后是 atomic_cmpset_int()嵌入汇编的内联函数)实现的，其思想是：用mtx的成员mtx_lock与当前线程(curthread)比较， 如果相同，则说明该锁被当前线程加锁，则将MTX_UNOWNED传给mtx_lock，并返回非零；否则，说明该锁不是被当前线程加锁，不做任何操作， 返回零。
如果能正常加锁/解锁，则无需执行后面的函数。我们现在考虑不能加锁/解锁的情况，根据MTX_DEF的特性，当前线程应该试图sleep，放弃其CPU资源，这是分别通过函数_mtx_lock_sleep/_mtx_unlock_sleep实现的。
函数_mtx_lock_sleep()：(省略了和锁操作日志、witness相关代码，以线程A为调用者)
if ((m->mtx_lock & MTX_FLAGMASK) == (uintptr_t)td) {
m->mtx_recurse++;
atomic_set_ptr(&m->mtx_lock, MTX_RECURSED);
return;
}
如果该锁支持递归，而且该锁的目前持有者是当前线程，则递归计数器加1，直接返回。
while (!_obtain_lock(m, td)) {
…
}
试图获得锁，如果不行则休息一定时间，再次试图获得锁；不停循环，直至获得该锁。我们接下来分析while语句里的代码。
mtx_lock_spin(&sched_lock);
if ((v = m->mtx_lock) == MTX_UNOWNED) {
mtx_unlock_spin(&sched_lock);
ia32_pause();
continue;
}
开 始准备调度其他线程运行，这需要在sched_lock锁的保护下进行。在对sched_lock加锁的过程中，有可能该锁m已经被解锁了，因此，再次通 过_obtain_lock()试图获得该锁。其中ia32_pause()是i386体系的一个暂停操作语句。之所以将mtx_lock记录v值，是因 为在随后的操作中，可能mtx_lock会被改变(由调度产生)。
if (v == MTX_CONTESTED) {
td1 = TAILQ_FIRST(&m->mtx_blocked);
m->mtx_lock = (uintptr_t)td | MTX_CONTESTED;
if (td1->td_priority < td->td_priority)
td->td_priority = td1->td_priority;
mtx_unlock_spin(&sched_lock);
return;
}
当v的值为MTX_CONTESTED，说明还有其他线程B、C阻塞在该锁上，则将mtx_lock设置为线程A所持有，并改变线程A的优先级，完成线程调度，返回。
if ((v & MTX_CONTESTED) == 0 &&
!atomic_cmpset_ptr(&m->mtx_lock, (void *)v, (void *)(v | MTX_CONTESTED))) {
mtx_unlock_spin(&sched_lock);
ia32_pause();
continue;
}
如 果v值还没有设置MTX_CONTESTED，并且v值和mtx_lock相同，说明该线程正阻塞于该锁，应该设置为MTX_CONTESTED的状态， 继续执行if以外的语句。如果v值已经设置了MTX_CONTESTED，则继续执行if以外语句。如果v值还没设置MTX_CONTESTED，并且 mtx_lock和v值不相同，暂停操作，再次试图获得该锁。
if (TAILQ_EMPTY(&m->mtx_blocked)) {
td1 = mtx_owner(m);
LIST_INSERT_HEAD(&td1->td_contested, m, mtx_contested);
TAILQ_INSERT_TAIL(&m->mtx_blocked, td, td_lockq);
} else {
TAILQ_FOREACH(td1, &m->mtx_blocked, td_lockq)
if (td1->td_priority > td->td_priority)
break;
if (td1)
TAILQ_INSERT_BEFORE(td1, td, td_lockq);
else
TAILQ_INSERT_TAIL(&m->mtx_blocked, td, td_lockq);
}
当代码执行到这里，说明该锁被其他线程持有，将该线程A插入该m的阻塞队列mtx_blocked中。
d->td_blocked = m;
td->td_lockname = m->mtx_object.lo_name;
TD_SET_LOCK(td);
propagate_priority(td);
td->td_proc->p_stats->p_ru.ru_nvcsw++;
mi_switch();
mtx_unlock_spin(&sched_lock);
最后，完成一些线程相关的设置，然后调用mi_switch()切换线程。
函数_mtx_unlock_sleep()：
if (mtx_recursed(m)) {
if (--(m->mtx_recurse) == 0)
atomic_clear_ptr(&m->mtx_lock, MTX_RECURSED);
return;
}
如果该锁m是支持递归的，则递归计数器减1，如果计数器为零，则，清除mtx_lock的设置，返回。
mtx_lock_spin(&sched_lock);
td1 = TAILQ_FIRST(&m->mtx_blocked);
TAILQ_REMOVE(&m->mtx_blocked, td1, td_lockq);
选出阻塞于该锁的第一个线程A，并将其移出mtx_blocked队列。
if (TAILQ_EMPTY(&m->mtx_blocked)) {
LIST_REMOVE(m, mtx_contested);
_release_lock_quick(m);
} else
atomic_store_rel_ptr(&m->mtx_lock, (void *)MTX_CONTESTED);
如果mtx_blocked队列为空，则将mtx_lock设置为MTX_UNOWNED，否则设置为MTX_CONTESTED。
pri = PRI_MAX;
LIST_FOREACH(m1, &td->td_contested, mtx_contested) {
int cp = TAILQ_FIRST(&m1->mtx_blocked)->td_priority;
if (cp < pri)
pri = cp;
}
if (pri > td->td_base_pri)
pri = td->td_base_pri;
td->td_priority = pri;
遍历当前线程B的所有竞争锁，找出一个优先级最高的(td_priority最小)，并将该值赋给pri变量。并调整线程B的活动优先级。
td1->td_blocked = NULL;
TD_CLR_LOCK(td1);
if (!TD_CAN_RUN(td1)) {
mtx_unlock_spin(&sched_lock);
return;
}
setrunqueue(td1);
如果线程A可以运行，则加入可运行线程队列，等待分时调度。如果不可运行，说明线程A还在等待别的资源。
if (td->td_critnest == 1 && td1->td_priority < pri) {
td->td_proc->p_stats->p_ru.ru_nivcsw++;
mi_switch();
}
mtx_unlock_spin(&sched_lock);
如果线程B进入临界区，并且线程A的优先级高于线程B，则通过mi_switch()做上下文切换。
MTX_SPIN：
对MTX_SPIN类型的锁加锁，展开mtx_lock_spin()得到：
critical_enter();
if (!_obtain_lock((mp), (tid))) {
if ((mp)->mtx_lock == (uintptr_t)(tid))
(mp)->mtx_recurse++;
else
_mtx_lock_spin((mp), (opts), (file), (line));
}
假 设该函数的调用者是线程A，通过critical_enter()函数，对当前线程A的td_critnest成员加1，说明线程A进入临界区。如果能获 得该锁，则成功返回。如果不能：如果该锁的持有者是线程A，则将递归计数器加1，如果不是，则调用_mtx_lock_spin函数处理。下面讨论该函 数。
该函数是一个for(;循环主体，我们讨论for语句里的内容，即自旋锁的自旋主体。
if (_obtain_lock(m, curthread))
break;
critical_exit();
试图过得该锁，如果可以，停止自旋。如果不能，继续自旋。通过critical_exit()函数退出临界区，这样允许我们在自旋的时候，如果有中断产生，中断能够有机会处理。
while (m->mtx_lock != MTX_UNOWNED) {
if (i++ < 10000000) {
ia32_pause();
continue;
}
if (i < 60000000)
DELAY(1);
else
panic("spin lock %s held by %p for > 5 seconds",
m->mtx_object.lo_name, (void *)m->mtx_lock);
ia32_pause();
}
critical_enter();
自旋，如果，自旋时间大于5秒，则认为出错。如果mtx_lock的值为MTX_UNOWNED，则说明该锁已被释放。因此，可以进入临界区，再次试图获得该锁。
对MTX_SPIN类型的锁加锁，展开mtx_unlock_spin()得到：
if (mtx_recursed((mp)))
(mp)->mtx_recurse--;
else
_release_lock_quick((mp));
critical_exit();
如果该锁递归值不为0，则递归计数器减1。如果为零，则将该锁的mtx_lock设置为MTX_UNOWNED。最后退出临界区。
3 条件变量(Condition variables)
条件变量构建于互斥体之上，和互斥体联合使用，等待条件发生。我们可以理解为基于锁的msleep机制(参考：msleep(9))。
3.1 条件变量数据结构
struct cv {
struct cv_waitq cv_waitq;
struct mtx *cv_mtx;
const char *cv_description;
};
成员cv_waitq是阻塞于该条件变量的线程链表表头；成员cv_mtx应该指向通过函数系列cv_*wait*()传入的参数(mtx类型指针)，目前只有在定义了INVARIANTS的时候有意义；成员cv_description是该条件变量的说明性文字。
3.2 条件变量的接口函数
函数void cv_init(struct cv *cvp, const char *desc)：
通过该函数初始化条件变量cvp，其说明性文字是通过desc指针获得。
TAILQ_INIT(&cvp->cv_waitq);
cvp->cv_mtx = NULL;
cvp->cv_description = desc;
这段初始化的代码十分简单。
函数void cv_destroy(struct cv *cvp)：
通过该函数销毁一个条件变量cvp。
KASSERT(cv_waitq_empty(cvp), ("%s: cv_waitq non-empty", __func__));
通过宏cv_waitq_empty判断阻塞于该条件变量的线程队列cv_waitq是否为空。如果不为空则出错。
当线程等待条件变量发生时，阻塞等待；当条件变量满足后，通过信号通知。用于阻塞等待的函数系列：cv_wait()、cv_wait_sig()、cv_timedwait()和cv_timedwait_sig()。
函数void cv_wait(struct cv *cvp, struct mtx *mp)：
该函数用于等待一个条件变量。当等待条件变量时，将当前线程放置在该条件变量的等待队列(cv_waitq)中，然后挂起该线程。
mtx_lock_spin(&sched_lock);
CV_WAIT_VALIDATE(cvp, mp);
DROP_GIANT();
mtx_unlock(mp);
cv_waitq_add(cvp, td);
cv_switch(td);
mtx_unlock_spin(&sched_lock);
PICKUP_GIANT();
mtx_lock(mp);
这 段代码简单的说，就是在调度自旋锁的保护下，将当前线程挂起，并做上下文切换。宏CV_WAIT_VALIDATE的意义：如果该条件变量cvp目前没有 阻塞线程，则将成员cv_mtx指向互斥体mp；如果有阻塞线程，而且其cv_mtx指向的互斥体和mp不一样，则异常出错。
通过 cv_waitq_add()函数，将td线程加入cvp的阻塞队列中，并完成td的相关设置。通过cv_switch()函数完成td的相关工作，并调 用mi_switch()做上下文切换。之所以要对mp解锁，是给阻塞于该mp的其他线程运行的机会。在等待到条件变量，并且轮到自己的时间片，则善后处 理，并对mp再次加锁。
宏DROP_GIANT和PICKUP_GIANT是有关Giant互斥体操作的一对宏，必须联合使用。这对宏的作用 是：在该对宏的代码中，存在主动上下文切换，因此，需要先把Giant解锁，在处理完中间代码后，再将Giant恢复到DROP_GIANT宏之前的状 态。其宏展开如下：
do {
int _giantcnt;
for (_giantcnt = 0; mtx_owned(&Giant); _giantcnt++)
mtx_unlock(&Giant)
我们中间代码，关于这对宏，我省略了witness和断言的相关代码，这样更清晰些。
while (_giantcnt--)
mtx_lock(&Giant);
} while (0)
函数int cv_wait_sig(struct cv *cvp, struct mtx *mp)：
该函数同样用于等待条件变量，当等待条件变量时，允许信号的中断。给出其主要代码，和cv_wait()函数相同的部分就不讨论了。
mtx_lock_spin(&sched_lock);
CV_WAIT_VALIDATE(cvp, mp);
DROP_GIANT();
mtx_unlock(mp);
cv_waitq_add(cvp, td);
sig = cv_switch_catch(td);
mtx_unlock_spin(&sched_lock);
PROC_LOCK(p);
if (sig == 0)
sig = cursig(td); /* XXXKSE */
if (sig != 0) {
if (SIGISMEMBER(p->p_sigacts->ps_sigintr, sig))
rval = EINTR;
else
rval = ERESTART;
}
PROC_UNLOCK(p);
if (p->p_flag & P_WEXIT)
rval = EINTR;
PICKUP_GIANT();
mtx_lock(mp);
在 这里，是通过cv_switch_catch()调用，决定是否需要上下文切换，在该函数里，信号中断是允许的。在宏PROC_LOCK(进程结构互斥体 保护机制)和PROC_UNLOCK的保护下，会再给一次机会给当前线程处理信号中断。信号值会作为cv_wait_sig()函数的返回值，由调用者决 定如何处理。
另外两个函数：函数int cv_timedwait(struct cv *cvp, struct mtx *mp, int timo)和函数int cv_timedwait_sig(struct cv *cvp, struct mtx *mp, int timo)和前面两个函数功能分别对应，只不过加入了等待的最大时间(timo/hz秒)的限制。
接下来，我们讨论当条件变量满足时，信号通知的相关内容，涉及函数cv_signal()和cv_broadcast()。
函数void cv_signal(struct cv *cvp)：
mtx_lock_spin(&sched_lock);
if (!TAILQ_EMPTY(&cvp->cv_waitq)) {
CV_SIGNAL_VALIDATE(cvp);
cv_wakeup(cvp);
}
mtx_unlock_spin(&sched_lock);
如果阻塞于该条件变量的线程队列为空，则不用做任何事。如果不为空，通过cv_wakeup()函数唤醒线程队列的第一个线程。由于该函数的处理主体是cv_wakeup()函数，下面讨论它的实现。
td = TAILQ_FIRST(&cvp->cv_waitq);
cv_waitq_remove(td);
TD_CLR_SLEEPING(td);
setrunnable(td);
得到阻塞于该条件变量的第一个线程，将其移出阻塞队列，并修改其状态值，通过调用函数setrunnable()，将该线程加入线程的运行队列中，等待调度。
函数 void cv_broadcast(struct cv *cvp)：
mtx_lock_spin(&sched_lock);
CV_SIGNAL_VALIDATE(cvp);
while (!TAILQ_EMPTY(&cvp->cv_waitq))
cv_wakeup(cvp);
mtx_unlock_spin(&sched_lock);
该函数唤醒阻塞于该条件变量的所有线程。
在前面，我们讨论当线程阻塞于某个条件变量，涉及到两个上下文切换函数的调用：cv_switch()和cv_switch_catch()。这两个函数主要涉及内核调度和线程状态转换等知识点，和锁关系不大，这里就讨论了。
4共享/独占锁(Share/exclusive locks)
共享/独占锁(sx)是非常有效的读写锁。之所以称之为sx是因为考虑以后添加额外的功能，不仅仅是读写功能，但是目前仅用于读写控制。由于sx锁机制的花费比互斥体高许多，因此必须谨慎使用。
4.1 共享/独占锁的数据结构
struct sx {
struct lock_object sx_object; /* Common lock properties. */
struct mtx *sx_lock; /* General protection lock. */
int sx_cnt; /* -1: xlock, > 0: slock count. */
struct cv sx_shrd_cv; /* slock waiters. */
int sx_shrd_wcnt; /* Number of slock waiters. */
struct cv sx_excl_cv; /* xlock waiters. */
int sx_excl_wcnt; /* Number of xlock waiters. */
struct thread *sx_xholder; /* Thread presently holding xlock. */
};
成 员sx_object说明该共享/独占锁的一般性属性，正如我们第一节提及的，其lock_class是lock_class_sx。成员sx_lock 是互斥体锁指针，用于保护对该共享/独占锁成员的操作。成员sx_cnt是锁的统计，初始化状态是0，如果是-1，则说明该锁是独占锁；如果是大于0的 值，则说明该锁是共享锁，并统计了其引用的次数。成员sx_shrd_cv是用于共享锁的条件变量控制，相应的成员sx_shrd_wcnt记录了等待该 共享锁的线程数目。成员sx_excl_cv是用于独占锁的条件变量控制，相应的成员sx_excl_wcnt记录了等待该独占锁的线程数目。成员 sx_xholder指向了当前支持有该独占锁的线程。
4.2 共享/独占锁的接口函数
函数void sx_init(struct sx *sx, const char *description)：
lock = &sx->sx_object;
bzero(sx, sizeof(*sx));
lock->lo_class = &lock_class_sx;
lock->lo_type = lock->lo_name = description;
lock->lo_flags = LO_WITNESS | LO_RECURSABLE | LO_SLEEPABLE |
LO_UPGRADABLE;
sx->sx_lock = mtx_pool_find(sx);
sx->sx_cnt = 0;
cv_init(&sx->sx_shrd_cv, description);
sx->sx_shrd_wcnt = 0;
cv_init(&sx->sx_excl_cv, description);
sx->sx_excl_wcnt = 0;
sx->sx_xholder = NULL;
共享/独占锁是通过该函数初始化的，这段代码非常清晰，唯一需要说明的是，该结构的内部互斥体sx_lock是从互斥体池中获得，参考第二节。
函数void sx_destroy(struct sx *sx)：
sx->sx_lock = NULL;
cv_destroy(&sx->sx_shrd_cv);
cv_destroy(&sx->sx_excl_cv);
通过该函数销毁一个共享/独占锁。因为sx_lock是从互斥体池中获得，所以只需断掉链接，就可以了。
共享锁的管理主要由三个函数组成：sx_slock()、sx_try_slock()和sx_sunlock()。
函数void sx_slock(struct sx *sx)：
mtx_lock(sx->sx_lock);
while (sx->sx_cnt < 0) {
sx->sx_shrd_wcnt++;
cv_wait(&sx->sx_shrd_cv, sx->sx_lock);
sx->sx_shrd_wcnt--;
}
sx->sx_cnt++;
mtx_unlock(sx->sx_lock);
该 函数申请一个共享锁sx，在互斥体sx_lock的保护下，如果sx_cnt小于0，说明该sx正被独占，因此共享申请等待计数器 sx_shrd_wcnt加1，调用cv_wait函数，等待共享锁的条件变量满足，当条件满足后，共享申请等待计数器sx_shrd_wcnt减1。获 得共享锁后，共享锁持有者计数器sx_cnt加1。
函数int sx_try_slock(struct sx *sx)：
mtx_lock(sx->sx_lock);
if (sx->sx_cnt >= 0) {
sx->sx_cnt++;
mtx_unlock(sx->sx_lock);
return (1);
} else {
return (0);
}
该函数试图获得共享锁，如果可以则将sx_cnt计数器加1，返回非零值，表示成功；如果不能，则说明该锁正被独占，直接返回0，表示失败。
函数void sx_sunlock(struct sx *sx)：
mtx_lock(sx->sx_lock);
sx->sx_cnt--;
if (sx->sx_excl_wcnt > 0) {
if (sx->sx_cnt == 0)
cv_signal(&sx->sx_excl_cv);
} else if (sx->sx_shrd_wcnt > 0)
cv_broadcast(&sx->sx_shrd_cv);
mtx_unlock(sx->sx_lock);
该 函数释放共享锁，首先将共享持有者计数器sx_cnt减1；如果sx_excl_wcnt大于0，说明现在有申请独占锁的线程阻塞，如果sx_cnt为 0，说明目前该锁完全自由，即：没有被共享占有，则向独占等待条件变量sx_excl_cv发一个信号，通知它可以试图申请独占锁。在没有独占锁申请的情 况下，如果有共享锁申请等待，即sx_shrd_wcnt大于0，则向共享等待条件变量sx_shrd_cv发一个广播信号，唤醒所有等待在该条件变量的 线程。