操作系统实验报告-信号量的实现和应用

实验内容

在Linux-0.11中实现信号量，并编写生产者-消费者程序进行检验。

实验步骤

添加信号量结构体与相应的系统调用函数

在include/unistd.h中添加代码：

#define SEM_NAME_LEN 32                /* 信号量名称最大长度 */
typedef struct sem_t{
char name[SEM_NAME_LEN];        /* 信号量名称 */
unsigned int value;                /* 信号量的值 */
struct task_struct *s_wait;        /* 等待信号量的进程的pcb指针 */
struct sem_t *next;                /* 用于连接信号量形成链表 */
}sem_t;

sem_t *sem_open(const char *name, unsigned int value);    /* 打开或新建信号量 *//
int sem_wait(sem_t *sem);            /* 等待信号量至其值大于0，将其值减1；对应P原语 */
int sem_post(sem_t *sem);            /* 唤醒在信号量上等待的进程，将信号量值加1；对应V原语 */
int sem_unlink(const char *name);    /* 销毁信号量 */

接下来将上面定义的4个函数添加为系统调用（步骤同操作系统实验报告-系统调用），添加kernel/sem.c实现它们：

#include <linux/kernel.h>
#include <asm/system.h>
#include <linux/sched.h>
#include <asm/segment.h>
#include <unistd.h>

sem_t *sem_head = &((sem_t *){"\0", 0, NULL, NULL});    /* 链表头结点，方便统一操作 */

/* 将用户态中的ustr复制到内核态的kstr */
static inline int str_u2k(const char *ustr, char *kstr, unsigned int length)
{
char c;
int i;

for(i=0; (c=get_fs_byte(ustr++))!='\0' && i<length; i++)
*(kstr+i)=c;
*(kstr+i)='\0';

return i;
}

sem_t *sys_sem_open(const char *name, unsigned int value)
{
sem_t *sem_cur, *sem_pre;
char pname[SEM_NAME_LEN];

/* 将用户态参数name指向的信号量名称拷贝到内核态指针pname中 */
str_u2k(name, pname, SEM_NAME_LEN);

/* 遍历链表，检验信号量是否已存在 */
for(sem_pre=sem_head, sem_cur=sem_head->next; sem_cur && strcmp(pname, sem_cur->name);
sem_pre=sem_cur, sem_cur=sem_cur->next);

/* sem_cur为空，表明信号量不存在，分配一块内存新建一个信号量 */
if(!sem_cur)
{
printk("semaphore %s no found. created a new one. \n", pname);
sem_cur = (sem_t *)malloc(sizeof(sem_t));
strcpy(sem_cur->name, pname);
sem_cur->value = value;
sem_cur->next = NULL;
sem_pre->next = sem_cur;
}
printk("pid %d opens semaphore %s(value %u) OK. \n", current->pid, pname, sem_cur->value);
return sem_cur;
}

int sys_sem_wait(sem_t *sem)
{
cli();    /* 关闭中断 */
/* 进程等待直到信号量的值大于0 */
while(sem->value<=0)
sleep_on(&(sem->s_wait));
sem->value--;
sti();    /* 开启中断 */
return 0;
}

int sys_sem_post(sem_t *sem)
{
sem->value++;
/* 唤醒在信号量上等待的进程 */
if(sem->s_wait)
{
wake_up(&(sem->s_wait));
return 0;
}
return -1;
}

int sys_sem_unlink(const char *name)
{
sem_t *sem_cur, *sem_pre;
char pname[SEM_NAME_LEN];
int i;

str_u2k(name, pname, SEM_NAME_LEN);

for(sem_pre=sem_head, sem_cur=sem_head->next; sem_cur && strcmp(pname, sem_cur->name);
sem_pre=sem_cur, sem_cur=sem_cur->next);

/* 找不到则返回错误代码-1 */
if(!sem_cur)
return -1;

/* 找到了将其从链表中移除，并释放空间 */
sem_pre->next = sem_cur->next;
free(sem_cur);
printk("unlink semaphore %s OK. \n", pname);
return 0;
}

其中sys_sem_wait()和sys_sem_post()参考自kernel/blk_drv/ll_rw_blk.c：

static inline void lock_buffer(struct buffer_head * bh)
{
cli();
while (bh->b_lock)
sleep_on(&bh->b_wait);
bh->b_lock=1;
sti();
}

static inline void unlock_buffer(struct buffer_head * bh)
{
if (!bh->b_lock)
printk("ll_rw_block.c: buffer not locked\n\r");
bh->b_lock = 0;
wake_up(&bh->b_wait);
}

其中的sleep_on()为在kernel/sched.c中实现的函数：

void sleep_on(struct task_struct **p)
{
/* 参数p指向原等待进程pcb */

struct task_struct *tmp;

if (!p)
return;
if (current == &(init_task.task))
panic("task[0] trying to sleep");
tmp = *p;        /* 本地指针tmp指向原等待进程 */
*p = current;    /* 参数p指向当前进程，使其成为下一次调用此方法的等待进程 */
current->state = TASK_UNINTERRUPTIBLE;    /* 休眠进程 */
schedule();        /* 执行调度 */
/* 由于是不可中断睡眠，不会自动就绪，只能通过调用wake_up()来唤醒。
* 如果调度后又回到这里，说明是信号量的值已经大于0了，于是就调用了wake_up()将此进程唤醒
*/
if (tmp)    /* 将原等待进程也唤醒 */
tmp->state=0;
/* 等到原等待进程拿到CPU进入运行状态，
* 它也会将它以前调用此函数时产生的另一个本地指针tmp指向的等待进程唤醒。
* 就这样递归唤醒，就好像遍历唤醒了一条等待的进程队列
*/
}

下面是《Linux内核完全注释》里面的一张图，形象地描述了此函数中的指针变化：



 

wake_up()也是在kernel/sched.c中实现，是一个简单的唤醒判断：

void wake_up(struct task_struct **p)
{
if (p && *p) {
(**p).state=0;
*p=NULL;
}
}

编写生产者-消费者检验程序

生产者-消费者问题

生产者-消费者问题是互斥的一个经典例子，下面是实验指导书给出的功能要求：
建立一个生产者进程，N个消费者进程（N>1）；
用文件建立一个共享缓冲区；
生产者进程依次向缓冲区写入整数0,1,2,...,M，M>0；
消费者进程从缓冲区读数，每次读一个，并将读出的数字从缓冲区删除，然后将本进程ID和数字输出到标准输出；
缓冲区同时最多只能保存10个数。

为了增加可读性，我以句子的形式输出信息。

生产者-消费者问题的解决算法的伪代码描述：

Producer()
{
生产一个产品item;
P(Empty);
P(Mutex);
将item放到空闲缓存中;
V(Mutex);
V(Full);
}

Consumer()
{
P(Full);
P(Mutex);
从缓存区取出一个赋值给item;
V(Mutex);
V(Empty);
消费产品item;
}

新建pc.c文件，编写测试程序：

#define __LIBRARY__
#include <unistd.h>
#include <fcntl.h>
#include <stdio.h>

_syscall2(sem_t *,sem_open,const char *,name,unsigned int,value)
_syscall1(int,sem_wait,sem_t *,sem)
_syscall1(int,sem_post,sem_t *,sem)
_syscall1(int,sem_unlink,const char *,name)

const char *FILENAME = "/usr/root/buffer_file";    /* 消费生产的产品存放的缓冲文件的路径 */
const int NR_CONSUMERS = 5;                        /* 消费者的数量 */
const int NR_ITEMS = 50;                        /* 产品的最大量 */
const int BUFFER_SIZE = 10;                        /* 缓冲区大小，表示可同时存在的产品数量 */
sem_t *metux, *full, *empty;                    /* 3个信号量 */
unsigned int item_pro, item_used;                /* 刚生产的产品号；刚消费的产品号 */
int fi, fo;                                        /* 供生产者写入或消费者读取的缓冲文件的句柄 */

int main(int argc, char *argv[])
{
char *filename;
int pid;
int i;

filename = argc > 1 ? argv[1] : FILENAME;
/* O_TRUNC 表示：当文件以只读或只写打开时，若文件存在，则将其长度截为0（即清空文件）
* 0222 和 0444 分别表示文件只写和只读（前面的0是八进制标识）
*/
fi = open(filename, O_CREAT| O_TRUNC| O_WRONLY, 0222);    /* 以只写方式打开文件给生产者写入产品编号 */
fo = open(filename, O_TRUNC| O_RDONLY, 0444);            /* 以只读方式打开文件给消费者读出产品编号 */

metux = sem_open("METUX", 1);    /* 互斥信号量，防止生产消费同时进行 */
full = sem_open("FULL", 0);        /* 产品剩余信号量，大于0则可消费 */
empty = sem_open("EMPTY", BUFFER_SIZE);    /* 空信号量，它与产品剩余信号量此消彼长，大于0时生产者才能继续生产 */

item_pro = 0;

if ((pid = fork()))    /* 父进程用来执行消费者动作 */
{
printf("pid %d:\tproducer created....\n", pid);
/* printf()输出的信息会先保存到输出缓冲区，并没有马上输出到标准输出（通常为终端控制台）。
* 为避免偶然因素的影响，我们每次printf()都调用一下stdio.h中的fflush(stdout)
* 来确保将输出立刻输出到标准输出。
*/
fflush(stdout);

while (item_pro <= NR_ITEMS)    /* 生产完所需产品 */
{
sem_wait(empty);
sem_wait(metux);

/* 生产完一轮产品（文件缓冲区只能容纳BUFFER_SIZE个产品编号）后
* 将缓冲文件的位置指针重新定位到文件首部。
*/
if(!(item_pro % BUFFER_SIZE))
lseek(fi, 0, 0);

write(fi, (char *) &item_pro, sizeof(item_pro));        /* 写入产品编号 */
printf("pid %d:\tproduces item %d\n", pid, item_pro);
fflush(stdout);
item_pro++;

sem_post(full);        /* 唤醒消费者进程 */
sem_post(metux);
}
}
else    /* 子进程来创建消费者 */
{
i = NR_CONSUMERS;
while(i--)
{
if(!(pid=fork()))    /* 创建i个消费者进程 */
{
pid = getpid();
printf("pid %d:\tconsumer %d created....\n", pid, NR_CONSUMERS-i);
fflush(stdout);

while(1)
{
sem_wait(full);
sem_wait(metux);

/* read()读到文件末尾时返回0，将文件的位置指针重新定位到文件首部 */
if(!read(fo, (char *)&item_used, sizeof(item_used)))
{
lseek(fo, 0, 0);
read(fo, (char *)&item_used, sizeof(item_used));
}

printf("pid %d:\tconsumer %d consumes item %d\n", pid, NR_CONSUMERS-i+1, item_used);
fflush(stdout);

sem_post(empty);    /* 唤醒生产者进程 */
sem_post(metux);

if(item_used == NR_ITEMS)    /* 如果已经消费完最后一个商品，则结束 */
goto OK;
}
}
}
}
OK:
close(fi);
close(fo);
return 0;
}

我们先将虚拟硬盘挂载，将文件pc.c拷贝到虚拟硬盘下：

cd workspace/oslab/
sudo ./mount-hdc
cp pc.c hdc/usr/root/

编译运行linux-0.11：

cd linux-0.11
make
../run

在linux-0.11中，编译运行pc.c：

gcc -o pc pc.c
./pc > sem_output    # 这里我将输出重定向到文件sem_output，因为输出的内容比较多，而linux-0.11终端不能滚屏，
　　　　　　　　　　　　　　# 而且输出内容多了还会显示错乱（可以用Ctrl+L刷新屏幕），不能复制终端输出的内容

一定要记得把修改的数据写入磁盘：

sync

关闭linux-0.11，挂载虚拟磁盘，查看我们的文件（当然也可以在linux-0.11中直接查看，只是显示内容多时会错乱，需要反复按Ctrl+L刷新）：

cd ..
sudo ./mount-hdc
sudo less hdc/usr/root/sem_output

得到输出：

pid 20: producer created....
pid 20: produces item 0
pid 20: produces item 1
.......
pid 20: produces item 8
pid 20: produces item 9
pid 24: consumer 5 created....
pid 24: consumer 5 consumes item 0
pid 24: consumer 5 consumes item 1
pid 24: consumer 5 consumes item 2
......
pid 24: consumer 5 consumes item 7
pid 24: consumer 5 consumes item 8
pid 24: consumer 5 consumes item 9
pid 23: consumer 4 created....
......
pid 20: produces item 47
pid 20: produces item 48
pid 20: produces item 49
pid 21: consumer 2 consumes item 40
pid 21: consumer 2 consumes item 41
......
pid 21: consumer 2 consumes item 48
pid 21: consumer 2 consumes item 49
pid 20: produces item 50
pid 22: consumer 3 consumes item 50

可以看到得出正确结果。

再看一下缓冲文件：

sudo cat  hdc/usr/root/buffer_file

2^@^@^@)^@^@^@*^@^@^@+^@^@^@,^@^@^@-^@^@^@.^@^@^@/^@^@^@0^@^@^@1^@^@^@

它是一个数据文件，我们把它转成十六进制输出到终端：

sudo xxd hdc/usr/root/buffer_file

00000000: 3200 0000 2900 0000 2a00 0000 2b00 0000  2...)...*...+...
00000010: 2c00 0000 2d00 0000 2e00 0000 2f00 0000  ,...-......./...
00000020: 3000 0000 3100 0000                      0...1...

8个十六进制位 = 32个二进制位 = 4 byte = sizeof(unsigned int)，所以上面翻译为十进制则是：

00000000: 50 41 42 43  2...)...*...+...
00000010: 44 45 46 47  ,...-......./...
00000020: 48 49        0...1...

50是最后一轮的产品编号，覆盖掉了上一轮的40，也是正确的。

思考

1. 在pc.c中去掉所有与信号量有关的代码，再运行程序，执行效果有变化吗？为什么会这样？

删除所有sem_*()调用，在linux-0.11中编译运行得到的输出为：

pid 32: producer created....
pid 32: produces item 0
pid 32: produces item 1
pid 32: produces item 2
pid 32: produces item 3
......
pid 32: produces item 49
pid 32: produces item 50
pid 38: consumer 5 created....
pid 38: consumer 5 consumes item 50
pid 37: consumer 4 created....
pid 37: consumer 4 consumes item 41
pid 37: consumer 4 consumes item 42
......
pid 37: consumer 4 consumes item 49
pid 37: consumer 4 consumes item 50
pid 36: consumer 3 created....
pid 36: consumer 3 consumes item 41
pid 36: consumer 3 consumes item 42
......
pid 36: consumer 3 consumes item 49
pid 36: consumer 3 consumes item 50
pid 35: consumer 2 created....
pid 35: consumer 2 consumes item 41
pid 35: consumer 2 consumes item 42
.......
pid 35: consumer 2 consumes item 49
pid 35: consumer 2 consumes item 50
pid 34: consumer 1 created....
pid 34: consumer 1 consumes item 41
pid 34: consumer 1 consumes item 42
......
pid 34: consumer 1 consumes item 49
pid 34: consumer 1 consumes item 50

生产者进程生产完所有的商品，消费者才开始消费商品，并且都只能消费缓存区中的最终10件商品（从轮到它们时的文件位置指针开始直到消费了第50号商品）。这是因为没有信号量的约束，生产者不知道缓存区已经满了，仍然继续生产；也没有信号量告诉它是否有消费者要访问这块临界区（缓存文件），它就无所顾虑地生产完所有的商品。消费者也一样没有了信号量的约束，直接消费到了50号商品。

我觉得这个问题的目的在于让我们看到没有信号量时，消费品消费的顺序很乱、重复（脏数据导致），可能我的验证程序的设计思路与出题者的不一样。

2. 实验的设计者在第一次编写生产者——消费者程序的时候，是这么做的：

Producer()
{
P(Mutex);  //互斥信号量
生产一个产品item;
P(Empty);  //空闲缓存资源
将item放到空闲缓存中;
V(Full);  //产品资源
V(Mutex);
}

Consumer()
{
P(Mutex);
P(Full);
从缓存区取出一个赋值给item;
V(Empty);
消费产品item;
V(Mutex);
}

这样可行吗？如果可行，那么它和标准解法在执行效果上会有什么不同？如果不可行，那么它有什么问题使它不可行？ 

不可行。
假设Producer刚生产完一件商品，释放了Mutex，Mutex为1，此时缓存区满了，Empty为0；
然后OS执行调度，若被Producer拿到CPU，它拿到Mutex，使Mutex为0，而Empty为0，Producer让出CPU，等待Consumer执行V(Empty)；
而Consumer拿到CPU后，却要等待Producer执行V(Mutex)；
两者相互持有对方需要的资源，造成死锁。