对于一个进程"睡眠"意味着什么? 当一个进程被置为睡眠, 它被标识为处于一个特殊的状态并且从调度器的运行队列中去除. 直到发生某些事情改变了那个状态, 这个进程将不被在任何 CPU 上调度, 并且, 因此, 将不会运行. 一个睡着的进程已被搁置到系统的一边, 等待以后发生事件.

对于一个 Linux 驱动使一个进程睡眠是一个容易做的事情. 但是, 有几个规则必须记住以安全的方式编码睡眠.

这些规则的第一个是: 当你运行在原子上下文时不能睡眠. 我们在第 5 章介绍过原子操作; 一个原子上下文只是一个状态, 这里多个步骤必须在没有任何类型的并发存取的情况下进行. 这意味着, 对于睡眠, 是你的驱动在持有一个自旋锁, seqlock, 或者 RCU 锁时不能睡眠. 如果你已关闭中断你也不能睡眠. 在持有一个旗标时睡眠是合法的, 但是你应当仔细查看这样做的任何代码. 如果代码在持有一个旗标时睡眠, 任何其他的等待这个旗标的线程也睡眠. 因此发生在持有旗标时的任何睡眠应当短暂, 并且你应当说服自己, 由于持有这个旗标, 你不能阻塞这个将最终唤醒你的进程.

另一件要记住的事情是, 当你醒来, 你从不知道你的进程离开 CPU 多长时间或者同时已经发生了什么改变. 你也常常不知道是否另一个进程已经睡眠等待同一个事件; 那个进程可能在你之前醒来并且获取了你在等待的资源. 结果是你不能关于你醒后的系统状态做任何的假设, 并且你必须检查来确保你在等待的条件是, 确实, 真的.

一个另外的相关的点, 当然, 是你的进程不能睡眠除非确信其他人, 在某处的, 将唤醒它. 做唤醒工作的代码必须也能够找到你的进程来做它的工作. 确保一个唤醒发生, 是深入考虑你的代码和对于每次睡眠, 确切知道什么系列的事件将结束那次睡眠. 使你的进程可能被找到, 真正地, 通过一个称为等待队列的数据结构实现的. 一个等待队列就是它听起来的样子:一个进程列表, 都等待一个特定的事件.

在 Linux 中, 一个等待队列由一个"等待队列头"来管理, 一个 wait_queue_head_t 类型的结构, 定义在<linux/wait.h>中. 一个等待队列头可被定义和初始化, 使用:

DECLARE_WAIT_QUEUE_HEAD(name); 

或者动态地, 如下:

wait_queue_head_t my_queue;
init_waitqueue_head(&my_queue);

我们将很快返回到等待队列结构, 但是我们知道了足够多的来首先看看睡眠和唤醒.

当一个进程睡眠, 它这样做以期望某些条件在以后会成真. 如我们之前注意到的, 任何睡眠的进程必须在它再次醒来时检查来确保它在等待的条件真正为真. Linux 内核中睡眠的最简单方式是一个宏定义, 称为 wait_event(有几个变体); 它结合了处理睡眠的细节和进程在等待的条件的检查. wait_event 的形式是:

wait_event(queue, condition)
wait_event_interruptible(queue, condition)
wait_event_timeout(queue, condition, timeout)
wait_event_interruptible_timeout(queue, condition, timeout)

在所有上面的形式中, queue 是要用的等待队列头. 注意它是"通过值"传递的. 条件是一个被这个宏在睡眠前后所求值的任意的布尔表达式; 直到条件求值为真值, 进程继续睡眠. 注意条件可能被任意次地求值, 因此它不应当有任何边界效应.

如果你使用 wait_event, 你的进程被置为不可中断地睡眠, 如同我们之前已经提到的, 它常常不是你所要的. 首选的选择是 wait_event_interruptible, 它可能被信号中断. 这个版本返回一个你应当检查的整数值; 一个非零值意味着你的睡眠被某些信号打断, 并且你的驱动可能应当返回 -ERESTARTSYS. 最后的版本(wait_event_timeout 和 wait_event_interruptible_timeout)等待一段有限的时间; 在这个时间期间(以嘀哒数表达的, 我们将在第 7 章讨论)超时后, 这个宏返回一个 0 值而不管条件是如何求值的.

图片的另一半, 当然, 是唤醒. 一些其他的执行线程(一个不同的进程, 或者一个中断处理, 也许)必须为你进行唤醒, 因为你的进程, 当然, 是在睡眠. 基本的唤醒睡眠进程的函数称为 wake_up. 它有几个形式(但是我们现在只看其中 2 个):

void wake_up(wait_queue_head_t *queue);
void wake_up_interruptible(wait_queue_head_t *queue);

wake_up 唤醒所有的在给定队列上等待的进程(尽管这个情形比那个要复杂一些, 如同我们之后将见到的). 其他的形式(wake_up_interruptible)限制它自己到处理一个可中断的睡眠. 通常, 这 2 个是不用区分的(如果你使用可中断的睡眠); 实际上, 惯例是使用 wake_up 如果你在使用 wait_event , wake_up_interruptible 如果你在使用 wait_event_interruptible.

我们现在知道足够多来看一个简单的睡眠和唤醒的例子. 在这个例子代码中, 你可找到一个称为 sleepy 的模块. 它实现一个有简单行为的设备:任何试图从这个设备读取的进程都被置为睡眠. 无论何时一个进程写这个设备, 所有的睡眠进程被唤醒. 这个行为由下面的 read 和 write 方法实现:

static DECLARE_WAIT_QUEUE_HEAD(wq);
static int flag = 0;

ssize_t sleepy_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_event_interruptible(wq, flag != 0);
        flag = 0;
        printk(KERN_DEBUG "awoken %i (%s)\n", current->pid, current->comm);
        return 0; /* EOF */
}
ssize_t sleepy_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);
        flag = 1;
        wake_up_interruptible(&wq);
        return count; /* succeed, to avoid retrial */

}

注意这个例子里 flag 变量的使用. 因为 wait_event_interruptible 检查一个必须变为真的条件, 我们使用 flag 来创建那个条件.

有趣的是考虑当 sleepy_write 被调用时如果有 2 个进程在等待会发生什么. 因为 sleepy_read 重置 flag 为 0 一旦它醒来, 你可能认为醒来的第 2 个进程会立刻回到睡眠. 在一个单处理器系统, 这几乎一直是发生的事情. 但是重要的是要理解为什么你不能依赖这个行为. wake_up_interruptible 调用将使 2 个睡眠进程醒来. 完全可能它们都注意到 flag 是非零, 在另一个有机会重置它之前. 对于这个小模块, 这个竞争条件是不重要的. 在一个真实的驱动中, 这种竞争可能导致少见的难于查找的崩溃. 如果正确的操作要求只能有一个进程看到这个非零值, 它将必须以原子的方式被测试. 我们将见到一个真正的驱动如何处理这样的情况. 但首先我们必须开始另一个主题.

在我们看全功能的 read 和 write 方法的实现之前, 我们触及的最后一点是决定何时使进程睡眠. 有时实现正确的 unix 语义要求一个操作不阻塞, 即便它不能完全地进行下去.

有时还有调用进程通知你他不想阻塞, 不管它的 I/O 是否继续. 明确的非阻塞 I/O 由 filp->f_flags 中的 O_NONBLOCK 标志来指示. 这个标志定义于 <linux/fcntl.h>, 被 <linux/fs.h>自动包含. 这个标志得名自"打开-非阻塞", 因为它可在打开时指定(并且起初只能在那里指定). 如果你浏览源码, 你会发现一些对一个 O_NDELAY 标志的引用; 这是一个替代 O_NONBLOCK 的名子, 为兼容 System V 代码而被接受的. 这个标志缺省地被清除, 因为一个等待数据的进程的正常行为仅仅是睡眠. 在一个阻塞操作的情况下, 这是缺省地, 下列的行为应当实现来符合标准语法:

这 2 句都假定有输入和输出缓冲; 实际上, 几乎每个设备驱动都有. 要求有输入缓冲是为了避免丢失到达的数据, 当无人在读时. 相反, 数据在写时不能丢失, 因为如果系统调用不能接收数据字节, 它们保留在用户空间缓冲. 即便如此, 输出缓冲几乎一直有用, 对于从硬件挤出更多的性能.

在驱动中实现输出缓冲所获得的性能来自减少了上下文切换和用户级/内核级切换的次数. 没有一个输出缓冲(假定一个慢速设备), 每次系统调用接收这样一个或几个字符, 并且当一个进程在 write 中睡眠, 另一个进程运行(那是一次上下文切换). 当第一个进程被唤醒, 它恢复(另一次上下文切换), 写返回(内核/用户转换), 并且这个进程重新发出系统调用来写入更多的数据(用户/内核转换); 这个调用阻塞并且循环继续. 增加一个输出缓冲可允许驱动在每个写调用中接收大的数据块, 性能上有相应的提高. 如果这个缓冲足够大, 写调用在第一次尝试就成功 -- 被缓冲的数据之后将被推到设备 -- 不必控制需要返回用户空间来第二次或者第三次写调用. 选择一个合适的值给输出缓冲显然是设备特定的.

我们不使用一个输入缓冲在 scull中, 因为数据当发出 read 时已经可用. 类似的, 不用输出缓冲, 因为数据被简单地拷贝到和设备关联的内存区. 本质上, 这个设备是一个缓冲, 因此额外缓冲的实现可能是多余的. 我们将在第 10 章见到缓冲的使用.

如果指定 O_NONBLOCK, read 和 write 的行为是不同的. 在这个情况下, 这个调用简单地返回 -EAGAIN(("try it agin")如果一个进程当没有数据可用时调用 read , 或者如果当缓冲中没有空间时它调用 write .

如你可能期望的, 非阻塞操作立刻返回, 允许这个应用程序轮询数据. 应用程序当使用 stdio 函数处理非阻塞文件中, 必须小心, 因为它们容易搞错一个的非阻塞返回为 EOF. 它们始终必须检查 errno.

自然地, O_NONBLOCK 也在 open 方法中有意义. 这个发生在当这个调用真正阻塞长时间时; 例如, 当打开(为读存取)一个 没有写者的(尚无)FIFO, 或者存取一个磁盘文件使用一个悬挂锁. 常常地, 打开一个设备或者成功或者失败, 没有必要等待外部的事件. 有时, 但是, 打开这个设备需要一个长的初始化, 并且你可能选择在你的 open 方法中支持 O_NONBLOCK , 通过立刻返回 -EAGAIN,如果这个标志被设置. 在开始这个设备的初始化进程之后. 这个驱动可能还实现一个阻塞 open 来支持存取策略, 通过类似于文件锁的方式. 我们将见到这样一个实现在"阻塞 open 作为对 EBUSY 的替代"一节, 在本章后面.

一些驱动可能还实现特别的语义给 O_NONBLOCK; 例如, 一个磁带设备的 open 常常阻塞直到插入一个磁带. 如果这个磁带驱动器使用 O_NONBLOCK 打开, 这个 open 立刻成功, 不管是否介质在或不在.

只有 read, write, 和 open 文件操作受到非阻塞标志影响.

最后, 我们看一个实现了阻塞 I/O 的真实驱动方法的例子. 这个例子来自 scullpipe 驱动; 它是 scull 的一个特殊形式, 实现了一个象管道的设备.

在驱动中, 一个阻塞在读调用上的进程被唤醒, 当数据到达时; 常常地硬件发出一个中断来指示这样一个事件, 并且驱动唤醒等待的进程作为处理这个中断的一部分. scullpipe 驱动不同, 以至于它可运行而不需要任何特殊的硬件或者一个中断处理. 我们选择来使用另一个进程来产生数据并唤醒读进程; 类似地, 读进程被用来唤醒正在等待缓冲空间可用的写者进程.

这个设备驱动使用一个设备结构, 它包含 2 个等待队列和一个缓冲. 缓冲大小是以常用的方法可配置的(在编译时间, 加载时间, 或者运行时间).

struct scull_pipe
{
        wait_queue_head_t inq, outq; /* read and write queues */
        char *buffer, *end; /* begin of buf, end of buf */
        int buffersize; /* used in pointer arithmetic */
        char *rp, *wp; /* where to read, where to write */
        int nreaders, nwriters; /* number of openings for r/w */
        struct fasync_struct *async_queue; /* asynchronous readers */
        struct semaphore sem;  /* mutual exclusion semaphore */
        struct cdev cdev;  /* Char device structure */
};

读实现既管理阻塞也管理非阻塞输入, 看来如此:

static ssize_t scull_p_read (struct file *filp, char __user *buf, size_t count, loff_t *f_pos)
{
        struct scull_pipe *dev = filp->private_data;
        if (down_interruptible(&dev->sem))
                return -ERESTARTSYS;

        while (dev->rp == dev->wp)
        { /* nothing to read */
                up(&dev->sem); /* release the lock */
                if (filp->f_flags & O_NONBLOCK)

                        return -EAGAIN;
                PDEBUG("\"%s\" reading: going to sleep\n", current->comm);
                if (wait_event_interruptible(dev->inq, (dev->rp != dev->wp)))
                        return -ERESTARTSYS; /* signal: tell the fs layer to handle it */ /* otherwise loop, but first reacquire the lock */
                if (down_interruptible(&dev->sem))
                        return -ERESTARTSYS;
        }
        /* ok, data is there, return something */

        if (dev->wp > dev->rp)
                count = min(count, (size_t)(dev->wp - dev->rp));
        else /* the write pointer has wrapped, return data up to dev->end */
                count = min(count, (size_t)(dev->end - dev->rp));
        if (copy_to_user(buf, dev->rp, count))
        {
                up (&dev->sem);
                return -EFAULT;
        }
        dev->rp += count;
        if (dev->rp == dev->end)

                dev->rp = dev->buffer; /* wrapped */
        up (&dev->sem);

        /* finally, awake any writers and return */
        wake_up_interruptible(&dev->outq);
        PDEBUG("\"%s\" did read %li bytes\n",current->comm, (long)count);
        return count;
}

如同你可见的, 我们在代码中留有一些 PDEBUG 语句. 当你编译这个驱动, 你可使能消息机制来易于跟随不同进程间的交互.

让我们仔细看看 scull_p_read 如何处理对数据的等待. 这个 while 循环在持有设备旗标下测试这个缓冲. 如果有数据在那里, 我们知道我们可立刻返回给用户, 不必睡眠, 因此整个循环被跳过. 相反, 如果这个缓冲是空的, 我们必须睡眠. 但是在我们可做这个之前, 我们必须丢掉设备旗标; 如果我们要持有它而睡眠, 就不会有写者有机会唤醒我们. 一旦这个确保被丢掉, 我们做一个快速检查来看是否用户已请求非阻塞 I/O, 并且如果是这样就返回. 否则, 是时间调用 wait_event_interruptible.

一旦我们过了这个调用, 某些东东已经唤醒了我们, 但是我们不知道是什么. 一个可能是进程接收到了一个信号. 包含 wait_event_interruptible 调用的这个 if 语句检查这种情况. 这个语句保证了正确的和被期望的对信号的反应, 它可能负责唤醒这个进程(因为我们处于一个可中断的睡眠). 如果一个信号已经到达并且它没有被这个进程阻塞, 正确的做法是让内核的上层处理这个事件. 到此, 这个驱动返回 -ERESTARTSYS 到调用者; 这个值被虚拟文件系统(VFS)在内部使用, 它或者重启系统调用或者返回 -EINTR 给用户空间. 我们使用相同类型的检查来处理信号, 给每个读和写实现.

但是, 即便没有一个信号, 我们还是不确切知道有数据在那里为获取. 其他人也可能已经在等待数据, 并且它们可能赢得竞争并且首先得到数据. 因此我们必须再次获取设备旗标; 只有这时我们才可以测试读缓冲(在 while 循环中)并且真正知道我们可以返回缓冲中的数据给用户. 全部这个代码的最终结果是, 当我们从 while 循环中退出时, 我们知道旗标被获得并且缓冲中有数据我们可以用.

仅仅为了完整, 我们要注意, scull_p_read 可以在另一个地方睡眠, 在我们获得设备旗标之后: 对 copy_to_user 的调用. 如果 scull 当在内核和用户空间之间拷贝数据时睡眠, 它在持有设备旗标中睡眠. 在这种情况下持有旗标是合理的因为它不能死锁系统(我们知道内核将进行拷贝到用户空间并且在不加锁进程中的同一个旗标下唤醒我们), 并且因为重要的是设备内存数组在驱动睡眠时不改变.

许多驱动能够满足它们的睡眠要求, 使用至今我们已涉及到的函数. 但是, 有时需要深入理解 Linux 等待队列机制如何工作. 复杂的加锁或者性能需要可强制一个驱动来使用低层函数来影响一个睡眠. 在本节, 我们在低层看而理解在一个进程睡眠时发生了什么.

void set_current_state(int new_state); 

current->state = TASK_INTERRUPTIBLE; 

if (!condition)
    schedule();

DEFINE_WAIT(my_wait); 

wait_queue_t my_wait;
init_wait(&my_wait);

void prepare_to_wait(wait_queue_head_t *queue, wait_queue_t *wait, int state); 

void finish_wait(wait_queue_head_t *queue, wait_queue_t *wait); 

/* How much space is free? */
static int spacefree(struct scull_pipe *dev)
{

        if (dev->rp == dev->wp)
                return dev->buffersize - 1;
        return ((dev->rp + dev->buffersize - dev->wp) % dev->buffersize) - 1;
}

static ssize_t scull_p_write(struct file *filp, const char __user *buf, size_t count,
                             loff_t *f_pos)
{

        struct scull_pipe *dev = filp->private_data;
        int result;
        if (down_interruptible(&dev->sem))
                return -ERESTARTSYS;

        /* Make sure there's space to write */
        result = scull_getwritespace(dev, filp);
        if (result)
                return result; /* scull_getwritespace called up(&dev->sem) */
        /* ok, space is there, accept something */
        count = min(count, (size_t)spacefree(dev));
        if (dev->wp >= dev->rp)
                count = min(count, (size_t)(dev->end - dev->wp)); /* to end-of-buf */
        else /* the write pointer has wrapped, fill up to rp-1 */
                count = min(count, (size_t)(dev->rp - dev->wp - 1));
        PDEBUG("Going to accept %li bytes to %p from %p\n", (long)count, dev->wp, buf);
        if (copy_from_user(dev->wp, buf, count))
        {
                up (&dev->sem);
                return -EFAULT;
        }
        dev->wp += count;
        if (dev->wp == dev->end)
                dev->wp = dev->buffer; /* wrapped */
        up(&dev->sem);

        /* finally, awake any reader */
        wake_up_interruptible(&dev->inq); /* blocked in read() and select() */

        /* and signal asynchronous readers, explained late in chapter 5 */
        if (dev->async_queue)
                kill_fasync(&dev->async_queue, SIGIO, POLL_IN);
        PDEBUG("\"%s\" did write %li bytes\n",current->comm, (long)count);
        return count;
}

/* Wait for space for writing; caller must hold device semaphore. On
 * error the semaphore will be released before returning. */
static int scull_getwritespace(struct scull_pipe *dev, struct file *filp)
{

        while (spacefree(dev) == 0)
        { /* full */
                DEFINE_WAIT(wait);

                up(&dev->sem);
                if (filp->f_flags & O_NONBLOCK)
                        return -EAGAIN;

                PDEBUG("\"%s\" writing: going to sleep\n",current->comm);
                prepare_to_wait(&dev->outq, &wait, TASK_INTERRUPTIBLE);
                if (spacefree(dev) == 0)
                        schedule();
                finish_wait(&dev->outq, &wait);
                if (signal_pending(current))

                        return -ERESTARTSYS; /* signal: tell the fs layer to handle it */
                if (down_interruptible(&dev->sem))
                        return -ERESTARTSYS;
        }
        return 0;

}

void prepare_to_wait_exclusive(wait_queue_head_t *queue, wait_queue_t *wait, int state); 

void sleep_on(wait_queue_head_t *queue);
void interruptible_sleep_on(wait_queue_head_t *queue);

我们已经见到了 scullpipe 驱动如何实现阻塞 I/O. 如果你想试一试, 这个驱动的源码可在剩下的本书例子中找到. 阻塞 I/O 的动作可通过打开 2 个窗口见到. 第一个可运行一个命令诸如 cat /dev/scullpipe. 如果你接着, 在另一个窗口拷贝文件到 /dev/scullpipe, 你可见到文件的内容出现在第一个窗口.

测试非阻塞的动作是技巧性的, 因为可用于 shell 的传统的程序不做非阻塞操作. misc-progs 源码目录包含下面简单的程序, 称为 nbtest, 来测试非阻塞操作. 所有它做的是拷贝它的输入到它的输出, 使用非阻塞 I/O 和在重试间延时. 延时时间在命令行被传递被缺省是 1 秒.

int main(int argc, char **argv)
{

        int delay = 1, n, m = 0;
        if (argc > 1)
                delay=atoi(argv[1]);
        fcntl(0, F_SETFL, fcntl(0,F_GETFL) | O_NONBLOCK); /* stdin */
        fcntl(1, F_SETFL, fcntl(1,F_GETFL) | O_NONBLOCK); /* stdout */

        while (1) {
                n = read(0, buffer, 4096);
                if (n >= 0)
                        m = write(1, buffer, n);
                if ((n < 0 || m < 0) && (errno != EAGAIN))
                        break;
                sleep(delay);
        }
        perror(n < 0 ? "stdin" : "stdout");
        exit(1);
}

如果你在一个进程跟踪工具, 如 strace 下运行这个程序, 你可见到每个操作的成功或者失败, 依赖是否当进行操作时有数据可用.