前言
并发相关的缺陷是最容易制造的,也是最难找到的,为了响应现代硬件和应用程序的需求,Linux 内核已经发展到同时处理更多事情的时代。这种变革使得内核性能及伸缩性得到了相当大的提高,然而也极大提高了内核编程的复杂性。
一、scull 的缺陷
scull 内存管理代码的一些片段,深入到驱动程序的 write 逻辑时,我们发现,scull 必须判断所请求的内存是否已经分配好。如下面代码所示:
if (!dptr->data[s_pos]) { dptr->data[s_pos] = kmalloc(quantum, GFP_KERNEL); if (!dptr->data[s_pos]) goto out; }
假定有两个进程(我们称之为“A”和“B”)正在独立地尝试向同一个 scull 设备的相同偏移量写入数据,而且两个进程在同一时刻到达上述代码段中的第一个 if 判断语句。如果代码涉及的指针是 NULL,两个进程都会决定分配内存,而每个进程都会将结果指赋值给 dptr->data[s_pos]。因为两个进程对同一位置赋值,显然只有一个赋值会成功。如果进程 A 首先赋值,则它的赋值会被进程 B 覆盖,从而由 A分配的内存将丢失,从而永远不会返回到系统中。
上述事件过程就是一种竞态(race condition),竞态会导致对共享数据的非控制访问。对这里讨论的竞态,其结果是内存的泄漏,这种结构已经够糟糕的了,但某些竞态经常会导致系统崩溃、数据被破坏或者产生安全问题。
二、并发及其管理
在现代 Linux 系统中存在大量的并发来源,因此会导致可能的竞态,SMP(对称多处理)系统甚至可在不同的处理器上同时执行我们的代码。内核代码是可抢占的;因此,我们的驱动程序代码可能在任何时候丢失对处理器的独占,而拥有处理器的进程可能正在调用我们的驱动程序代码。
设备中断是异步事件,也会导致代码的并发执行。内核还提供了许多可延迟代码执行的机制,比如 workqueue(工作队列)、tasklet(小任务)以及 timer(定时器)等,这些机制使得代码可在任何时刻执行,而不管当前进程在做什么。
竞态通常作为对资源的共享访问结果而产生。当两个执行线程需要访问相同的数据结构(或硬件资源)时,混合的可能性就永远存在。因此在设计自己的驱动程序时第一个要记住的规则是,只要可能,就应该避免资源的共享,如果没有并发的访问,也就不会有竞态的产生,因此,仔细编写的内核代码应该具有最少的共享。
硬件资源本质上就是共享的,而软件资源经常需要对其他执行线程可用。
下面是资源共享的硬规则: 在单个执行线程之外共享硬件或软件资源的任何时候,因为另外一个线程可能产生对该资源的不一致观察,因此必须显式地管理对该资源的访问。访问管理的常见技术称为“锁定”或者“互斥”–确保一次只有一个执行线程可操作共享资源。
我们首先必须简要考虑另外一个重要的规则。当内核代码创建了一个可能和其他内核部分共享的对象时,该对象必须在还有其他组件引用自己时保持存在(并正确工作)。这一规则产生下面两个需求:在对象尚不能正确工作时,不能将其对内核可用,也就是说,对这类对象的应用必须得到跟踪。在大多数情况下,我们将发现内核会为我们处理引用计数,然而总是会有例外。
三、信号量和互斥体
我们的目的是使对 scull 数据结构的操作是原子的,这意味着在涉及到其他执行线程之前,整个操作就已经结束了。为此,我们必须建立临界区:在任意给定的时刻,代码只能被一个线程执行。
当一个 Linux 进程到达某个时间点,此时它不能进行任何处理时,它将进人休眠(或“阻塞”)状态,这将把处理器让给其他执行线程直到将来它能够继续完成自己的处理为止。在等待 I/O 完成时,进程经常会进入休眠状态。
我们可以使用一种锁定机制,当进程在等待对临界区的访问时,此机制可让进程进人休眠状态。
为了让我们的临界区正确工作,我们选择使用的锁定原语必须在其他拥有这个锁并休眠的情况下工作。在可能出现休眠的情况下。而目前,对于我们来说最合适的机制是信号量(semaphore)。
一个信号量本质上是一个整数值,它和一对函数联合使用,这一对函数通常称为 P 和 V。希望进入临界区的进程将在相关信号量上调用 P;如果信号量的值大于零,则该值会减小一,而进程可以继续。相反,如果信号量的值为零(或更小),进程必须等待直到其他人释放该信号量。对信号的解锁通过调用 V 完成,该函数增加信号量的值,并在必要时唤醒等待的进程。
当信号量用于互斥时(即避免多个进程同时在一个临界区中运行),信号量的值应初始化为 1。这种信号量在任何给定时刻只能由单个进程或线程拥有。在这种使用模式下,一个信号量有时也称为一个“互斥体(mutex)”,它是互斥(mutual exclusion)的简称。Linux 内核中几平所有的信号量均用于互斥。
1、Linux 信号量的实现
要使用信号量,内核代码必须包括 <asm/semaphore.h>。相关的类型是 struct semaphore;实际的信号量可通过几种途径来声明和初始化。其中之一是直接创建信号量,这通过 sema_init 完成:
void sema_init(struct semaphore *sem, int val);
其中 val 是赋予一个信号量的初始值。
不过,信号量通常被用于互斥模式。内核提供了一组辅助函数和宏。因此,我们可以用下面的方法之一来声明和初始化一个互斥体:
DECLARE_MUTEX(name); DECLARE_MUTEX_LOCKED(name);
上面两个宏的结果是,一个称为 name 的信号量变量被初始化为1(使用DECLARE MUTEX)或者0(使用DECLARE_MUTEX LOCKED)。在后面一种情况下,互斥体的初始状态是锁定的,也就是说,在允许任何线程访问之前,必须显式地解锁该互斥体。
如果互斥体必须在运行时被初始化(例如在动态分配互斥体的情况下),应使用下面的函数之一:
void init_MUTEX(struct semaphore *sem); void init_MUTEX_LOCKED(struct semaphore *sem);
P 函数被称为 down ——或者这个名字的其他变种。这里,“down”指的是该函数减小了信号量的值,它也许会将调用者置于休眠状态,然后等待信号量变得可用,之后授予调用者对被保护资源的访问。下面是 down 的三个版本:
void down(struct semaphore *sem); int down_interruptible(struct semaphore *sem); int down_trylock(struct semaphore *sem);
down 减小信号量的值,并在必要时一直等待。down interruptible 完成相同的工作,但操作是可中断的。可中断的版本几乎是我们始终要使用的版本,它允许等待在某个信号量上的用户空间进程可被用户中断。作为通常的规则,我们不应该使用非中断操作,除非没有其他可变通的办法。非中断操作是建立不可杀进程(ps 输出中的“D state”)的好方法。使用down interruptible 需要额外小心,如果操作被中断,该函数会返回非零值,而调用者不会拥有该信号量。对down interruptible 的正确使用需要始终检查返回值,并作出相应的响应。
最后一个版本(down_trylock)永远不会休眠;如果信号量在调用时不可获得,down_trylock 会立即返回一个非零值。
当一个线程成功调用上述 down 的某个版本之后,就称为该线程“拥有”(或“拿到”、“获取”)了该信号量。这样,该线程就被赋予访问由该信号量保护的临界区的权利。当互斥操作完成后,必须返回该信号量。Linux 等价于 V 的函数是 up:
void up(struct semaphore *sem);
调用 up 之后,调用者不再拥有该信号量。
任何拿到信号量的线程都必须通过一次(只有一次)对 up 的调用而释放该信号量。在出现错误的情况下,经常需要特别小心;如果在拥有一个信号量时发生错误,必须在将错误状态返回给调用者之前释放该信号量。我们很容易犯忘记释放信号量的错误,而其结果(进程在某些无关位置处被挂起)很难复现和跟踪。
2、在 scull 中使用信号量
信号量机制为 scull 提供了一种工具,它可以利用信号量避免在访问 scull_dev 数据结构时产生竞态。但我们必须正确使用这个工具。正确使用锁定机制的关键是,明确指定需要保护的资源,并确保每一个对这些资源的访问使用正确的锁定。在我们的示例驱动程序中,所有的信息都包含在 scull_dev 结构中,因此该结构就是我们锁定机构的逻辑范围。
该结构的定义如下:
struct scull_dev { struct scull_qset *data; /* Pointer to first quantum set */ int quantum; /* the current quantum size */ int qset; /* the current array size */ unsigned long size; /* amount of data stored here */ unsigned int access_key; /* used by sculluid and scullpriv */ struct semaphore sem; /* mutual exclusion semaphore */ struct cdev cdev; /* Char device structure */ };
该结构底部有一个称为 sem 的成员,它就是我们的信号量。我们决定对每个虚拟的 scull 设备使用单独的信号量。使用单个全局的信号量也是正确的。但是不同的 scull 设备并不共享资源,因此没有理由让一个进程在其他进程访问不同的 scull 设备时等待。为每个设备使用单独的信号量允许不同设备上的操作可以并行处理,从而可以提高性能。
信号量在使用前必须初始化。scull 在装载时通过下面的循环执行初始化:
for (i = 0; i < scull_nr_devs; i++) { scull_devices[i].quantum = scull_quantum; scull_devices[i].qset = scull_qset; init_MUTEX(&scull_devices[i].sem); scull_setup_cdev(&scull_devices[i], i); }
接下来,我们必须仔细检查代码,确保在不拥有该信号量的时候不会访问 scull_dev 数据结构。例如,scull_write 的开始处包含下面的代码:
if (down_interruptible(&dev->sem)) return -ERESTARTSYS;
如果我们返回 -ERESTARTSYS,则必须首先撤销已经做出的任何用户可见的修改,这样,系统调用可正确重试。如果无法撤销这些操作,则应该返回 -EINTR。
不管 scull_write 是否能够成功完成其他工作,它都必须释放信号量。如果一切正常,执行过程将到达该函数的最后几行:
out: up(&dev->sem); return retval;
上述代码释放信号量,并返回被调用的状态值。
3、读取者/写入者信号量
信号量对所有的调用者执行互斥,而不管每个线程到底想做什么。但是,许多任务可以划分为两种不同的工作类型:一些任务只需要读取受保护的数据结构,而其他的则必须做出修改。允许多个并发的读取者是可能的,只要它们之中没有哪个要做修改。这样做可以大大提高性能,因为只读任务可并行完成它们的工作,而不需要等待其他读取者退出临界区。
Linux 内核为这种情形提供了一种特殊的信号量类型称为 “rwsem”(或者“reader/writer
semaphore,读取者/写入者信号量”)。在驱动程序中使用 rwsem 的机会相对较少,但偶尔也比较有用。
使用 rwsem 的代码必须包括 <linux/rwsem.h>。读取者/写入者信号量相关的数据类型是 struct
rw_semaphore;一个 rwsem 对象必须在运行时通过下面的函数显式地初始化:
void init_rwsem(struct rw_semaphore *sem);
对只读访问,可用的接口如下:
void down_read(struct rw_semaphore *sem); int down_read_trylock(struct rw_semaphore *sem); void up_read(struct rw_semaphore *sem);
对 down_read 的调用提供了对受保护资源的只读访问,可和其他读取者并发地访问。注意down_read可能会将调用进程置于不可中断的休眠
。down_read_trylock 不会在读取访问不可获得时等待;它在授予访问时返回非零,其他情况下返回零。注意,down_read trylock的用法和其他大多数内核函数不同,其他函数会在成功时返回零
。由down_read 获得的 rwsem 对象最终必须通过 up_read 被释放。
针对写人者的接口类似于读取者接口:
void down_write(struct rw_semaphore *sem); int down_write_trylock(struct rw_semaphore *sem); void up_write(struct rw_semaphore *sem); void downgrade_write(struct rw_semaphore *sem);
down_write、down_write_trylock 和up_write 与读取者的对应函数行为相同,当然,它们提供的是写入访问。当某个快速改变获得了写入者锁,而其后是更长时间的只读访问的话,我们可以在结束修改之后调用downgrade_write,来允许其他读取者的访问。
一个 rwsem 可允许一个写入者或无限多个读取者拥有该信号量。写入者具有更高的优先级;当某个给定写入者试图进入临界区时,在所有写入者完成其工作之前,不会允许读取者获得访问。如果有大量的写入者竞争该信号量,则这种实现会导致读取者“饿死”即可能会长期拒绝读取者的访问。为此,最好在很少需要写访问且写入者只会短期拥有信号量的时候使用 rwsem。
四、Completions
内核编程中常见的一种模式是,在当前线程之外初始化某个活动,然后等待该活动的结束。在这种情况下,我们可以使用信号量来同步这两个任务、并如下所示来编写代码:
struct semaphore sem; init_MUTEX_LOCKED(&sem); start_external_task(&sem); down(&sem);
当外部任务完成其工作时,将调用 up(&sem)。
在通常的使用中,试图锁定某个信号量的代码会发现该信号量几乎总是可用;而如果存在针对该信号量的严重竞争,性能将受到影响,这时,我们需要重新审视锁定机制。然而,如果像上面那样使用信号量在任务完成时进行通信,则调用 down 的线程几平总是要等待,这样性能也同样会受到影响。
上述考虑导致 2.4.7 版内核中出现了“completion(完成)”接口。completion 是一种轻量级的机制,它允许一个线程告诉另一线程某个工作已经完成。为了使用 completion,代码必须包含
<linux/completion,h>。可以利用下面的接口创建completion:
DECLARE_COMPLETION(my_completion);
或者,如果必须动态地创建和初始化 completion,则使用下面的方法:
struct completion my_completion; /* ... */ init_completion(&my_completion);
要等待 completion,可进行如下调用:
void wait_for_completion(struct completion *c);
注意,该函数执行一个非中断的等待。如果代码调用了 wait for completion 且没有人会完成该任务,则将产生一个不可杀的进程。
另一方面,实际的 completion 事件可通过调用下面函数之一来触发:
void complete(struct completion *c); void complete_all(struct completion *c);
complete 只会唤醒一个等待线程,而complete_all 允许唤醒所有等待线程。
如果使用了 complete_all,则必须在重复使用该结构之前重新初始化它。下面这个宏可用来快速执行重新初始化:
INIT_COMPLETION(struct completion c);
任何试图从该设备读取的进程都将等待(使用wait_for_completion),直到其他进程写人该设备为止。实现这种行为的代码如下:
DECLARE_COMPLETION(comp); ssize_t complete_read (struct file *filp, char __user *buf, size_t count, loff_t *pos) { printk(KERN_DEBUG "process %i (%s) going to sleep\n",current->pid, current->comm); wait_for_completion(&comp); printk(KERN_DEBUG "awoken %i (%s)\n", current->pid, current->comm); return 0; /* EOF */ } ssize_t complete_write (struct file *filp, const char __user *buf, size_t count, loff_t *pos) { printk(KERN_DEBUG "process %i (%s) awakening the readers...\n", current->pid, current->comm); complete(&comp); return count; /* succeed, to avoid retrial */ }
同一时刻有多个进程从该设备“读取”是可能的。每次向该设备的写入将导致一个读取操作结束,但是没有办法知道会是哪个进程。
completion 机制的典型使用是模块退出时的内核线程终止。在这种原型中,某些驱动程序的内部工作由一个内核线程在 while(1) 循环中完成。当内核准备清除该模块时 exit 函数会告诉该线程退出并等待 completion。为了实现这个目的,内核包含了可用于这种线程的一个特殊函数:
void complete_and_exit(struct completion *c, long retval);