Linux进程间通讯（五）信号量

小咪咪 2023-05-31 02:54 4阅读 0赞

**Linux进程间通讯**

[Linux进程间通讯（一）信号（上）][Linux]

[Linux进程间通讯（二）信号（下）][Linux 1]

[Linux进程间通讯（三）管道][Linux 2]

[Linux进程间通讯（四）共享内存][Linux 3]

[Linux进程间通讯（五）信号量 ][Linux_]

# Linux进程间通讯（五）信号量 #

### 文章目录 ###

*  Linux进程间通讯（五）信号量
 *   *  一、信号量集的创建
     *  二、信号量的设置
     *  三、信号量的PV操作
     *  四、SEM\_UNDO标志的含义
     *  五、总结

## 一、信号量集的创建 ##

信号量集的创建需要通过系统调用 semget，其定义如下（这里需要注意的是，semget创建的是一个信号量集合，也就是一个集合是可以包含多个信号量的）

SYSCALL_DEFINE3(semget, key_t, key, int, nsems, int, semflg)
    {
        
    	struct ipc_namespace *ns;
    	static const struct ipc_ops sem_ops = {
        
    		.getnew = newary,
    		.associate = sem_security,
    		.more_checks = sem_more_checks,
    	};
    	struct ipc_params sem_params;
    	ns = current->nsproxy->ipc_ns;
    	sem_params.key = key;
    	sem_params.flg = semflg;
    	sem_params.u.nsems = nsems;
    	return ipcget(ns, &sem_ids(ns), &sem_ops, &sem_params);
    }

*  key：表示信号量的id
 *  msems：表示这个信号量集合中的信号量个数

这部分的操作和上一篇文章分析共享内存时创建共享内存十分相似。首先将请求封装成 sem\_params，然后调用通用的 ipcget，指定信号量对应的 ipc 为 sem\_ids，指定信号量对应的操作为 sem\_ops，这一次请求的操作为 sem\_params

ipcget 在上一篇文章已经解析过了，如果设置了 IPC\_PRIVATE，那么就永远创建新的信号量；否则就会调用 ipcget\_public

在 ipcget\_public 中，首先会通过 id 去 sem\_ids 中查找信号量是否存在，如果存在就返回，如果不存在，那么就创建一个新的

当要创建新的信号量的时候，对应的函数就是 newary 函数，其定义如下

static int newary(struct ipc_namespace *ns, struct ipc_params *params)
    {
        
    	int retval;
    	struct sem_array *sma;
    	key_t key = params->key;
    	int nsems = params->u.nsems;
    	int semflg = params->flg;
    	int i;
    ......
    	sma = sem_alloc(nsems);
    ......
    	sma->sem_perm.mode = (semflg & S_IRWXUGO);
    	sma->sem_perm.key = key;
    	sma->sem_perm.security = NULL;
    ......
    	for (i = 0; i < nsems; i++) {
        
    		INIT_LIST_HEAD(&sma->sems[i].pending_alter);
    		INIT_LIST_HEAD(&sma->sems[i].pending_const);
    		spin_lock_init(&sma->sems[i].lock);
    	}
    	sma->complex_count = 0;
    	sma->use_global_lock = USE_GLOBAL_LOCK_HYSTERESIS;
    	INIT_LIST_HEAD(&sma->pending_alter);
    	INIT_LIST_HEAD(&sma->pending_const);
    	INIT_LIST_HEAD(&sma->list_id);
    	sma->sem_nsems = nsems;
    	sma->sem_ctime = get_seconds();
    	retval = ipc_addid(&sem_ids(ns), &sma->sem_perm, ns->sc_semmni);
    ......
    	ns->used_sems += nsems;
    ......
    	return sma->sem_perm.id;
    }

newary 的第一步就是分配一个 struct sem\_array 对象，它表示一个信号量集对象，然后是设置信号量的属性，如 key、权限等

struct sem\_array 中其实定义了多个信号量，具体多少个是通过 semget 系统调用指定的，其定义如下

struct sem_array {
        
    	struct kern_ipc_perm	____cacheline_aligned_in_smp
    				sem_perm;	/* permissions .. see ipc.h */
    	time_t			sem_otime;	/* last semop time */
    	time_t			sem_ctime;	/* last change time */
    	struct sem		*sem_base;	/* ptr to first semaphore in array，信号量数组 */
    	struct list_head	pending_alter;	/* pending operations to be processed */
    	struct list_head	list_id;	/* undo requests on this array */
    	int			sem_nsems;	/* no. of semaphores in array，信号量个数 */
    	int			complex_count;	/* pending complex operations */
    };

*  sem\_nsems：表示该信号量集的信号量数量
 *  sem\_base：表示信号量数组

每个信号量都使用 struct sem 来表示，定义如下

struct sem {
        
    	int	semval;		/* current value */
    	/*
    	 * PID of the process that last modified the semaphore. For
    	 * Linux, specifically these are:
    	 *  - semop
    	 *  - semctl, via SETVAL and SETALL.
    	 *  - at task exit when performing undo adjustments (see exit_sem).
    	 */
    	int	sempid;
    	spinlock_t	lock;	/* spinlock for fine-grained semtimedop */
    	struct list_head pending_alter; /* pending single-sop operations that alter the semaphore */
    	struct list_head pending_const; /* pending single-sop operations that do not alter the semaphore*/
    	time_t	sem_otime;	/* candidate for sem_otime */
    } ____cacheline_aligned_in_smp;

*  sem\_val：表示该信号量的值
    
    外，可以看到 struct sem\_array 和 struct sem 中，都有 struct list\_head pending\_alter 链表，维护着对整个信号量集或者某个信号量的操作队列

newary 的第二步就是初始化这些链表

newary 的第三步就是将新创建的 struct sem\_array 挂载到 sem\_ids 中

信号量集创建到此就结束了，下面来看看关于信号量的操作

## 二、信号量的设置 ##

首先通过系统调用 semctl 来对信号量进行设置，其定义如下

SYSCALL_DEFINE4(semctl, int, semid, int, semnum, int, cmd, unsigned long, arg)
    {
        
    	int version;
    	struct ipc_namespace *ns;
    	void __user *p = (void __user *)arg;
    	ns = current->nsproxy->ipc_ns;
    	switch (cmd) {
        
    	case IPC_INFO:
    	case SEM_INFO:
    	case IPC_STAT:
    	case SEM_STAT:
    		return semctl_nolock(ns, semid, cmd, version, p);
    	case GETALL:
    	case GETVAL:
    	case GETPID:
    	case GETNCNT:
    	case GETZCNT:
    	case SETALL:
    		return semctl_main(ns, semid, semnum, cmd, p);
    	case SETVAL:
    		return semctl_setval(ns, semid, semnum, arg);
    	case IPC_RMID:
    	case IPC_SET:
    		return semctl_down(ns, semid, cmd, version, p);
    	default:
    		return -EINVAL;
    	}
    }

如果是 SETALL，表示要设置信号量集所有的信号量，调用的是 semctl\_main

如果是 SETVAL，表示要设置信号量集中的某个信号量，调用的是 semctl\_setval

semctl\_setval 的定义如下

static int semctl_setval(struct ipc_namespace *ns, int semid, int semnum,
    		unsigned long arg)
    {
        
    	struct sem_undo *un;
    	struct sem_array *sma;
    	struct sem *curr;
    	int err, val;
    	DEFINE_WAKE_Q(wake_q);
    ......
    	sma = sem_obtain_object_check(ns, semid);
    ......
    	curr = &sma->sems[semnum];
    ......
    	curr->semval = val;
    	curr->sempid = task_tgid_vnr(current);
    	sma->sem_ctime = get_seconds();
    	/* maybe some queued-up processes were waiting for this */
    	do_smart_update(sma, NULL, 0, 0, &wake_q);
    ......
    	wake_up_q(&wake_q);
    	return 0;
    }

首先根据指定的编号找到信号量集中对应的信号量，然后直接修改信号量的值

信号量设置完毕后，我们来看看信号量的PV操作

## 三、信号量的PV操作 ##

信号量的PV操作，都是使用系统调用 sem\_op，其定义如下

SYSCALL_DEFINE3(semop, int, semid, struct sembuf __user *, tsops,
    		unsigned, nsops)
    {
        
    	return sys_semtimedop(semid, tsops, nsops, NULL);
    }
    
    SYSCALL_DEFINE4(semtimedop, int, semid, struct sembuf __user *, tsops,
    		unsigned, nsops, const struct timespec __user *, timeout)
    {
        
    	int error = -EINVAL;
    	struct sem_array *sma;
    	struct sembuf fast_sops[SEMOPM_FAST];
    	struct sembuf *sops = fast_sops, *sop;
    	struct sem_undo *un;
    	int max, locknum;
    	bool undos = false, alter = false, dupsop = false;
    	struct sem_queue queue;
    	unsigned long dup = 0, jiffies_left = 0;
    	struct ipc_namespace *ns;
    
    	ns = current->nsproxy->ipc_ns;
    ......
    	if (copy_from_user(sops, tsops, nsops * sizeof(*tsops))) {
        
    		error =  -EFAULT;
    		goto out_free;
    	}
    
    	if (timeout) {
        
    		struct timespec _timeout;
    		if (copy_from_user(&_timeout, timeout, sizeof(*timeout))) {
        
    		}
    		jiffies_left = timespec_to_jiffies(&_timeout);
    	}
    ......
    	/* On success, find_alloc_undo takes the rcu_read_lock */
    	un = find_alloc_undo(ns, semid);
    ......
    	sma = sem_obtain_object_check(ns, semid);
    ......
    	queue.sops = sops;
    	queue.nsops = nsops;
    	queue.undo = un;
    	queue.pid = task_tgid_vnr(current);
    	queue.alter = alter;
    	queue.dupsop = dupsop;
    
    	error = perform_atomic_semop(sma, &queue);
    	if (error == 0) {
         /* non-blocking succesfull path */
    		DEFINE_WAKE_Q(wake_q);
    ......
    		do_smart_update(sma, sops, nsops, 1, &wake_q);
    ......
    		wake_up_q(&wake_q);
    		goto out_free;
    	}
    	/*
    	 * We need to sleep on this operation, so we put the current
    	 * task into the pending queue and go to sleep.
    	 */
    	if (nsops == 1) {
        
    		struct sem *curr;
    		curr = &sma->sems[sops->sem_num];
    ......
    		list_add_tail(&queue.list,
    						&curr->pending_alter);
    ......
    	} else {
        
    ......
    		list_add_tail(&queue.list, &sma->pending_alter);
    ......
    	}
    
    	do {
        
    		queue.status = -EINTR;
    		queue.sleeper = current;
    
    		__set_current_state(TASK_INTERRUPTIBLE);
    		if (timeout)
    			jiffies_left = schedule_timeout(jiffies_left);
    		else
    			schedule();
    ......
    		/*
    		 * If an interrupt occurred we have to clean up the queue.
    		 */
    		if (timeout && jiffies_left == 0)
    			error = -EAGAIN;
    	} while (error == -EINTR && !signal_pending(current)); /* spurious */
    ......
    }

semop 会调用 sys\_semtimedop，这个函数比较复杂

sys\_semtimedop 做的第一件事就是将用户空间设置的 struct sembuf 操作拷贝内核空间，然后如果设置了timeout，也会将其拷贝到内核空间

sys\_semtimedop 做的第二件事是根据 id 在 sem\_ids 中找到相应的信号量集 struct sem\_array。然后创建一个 struct sem\_queue 表示当前信号量的操作，然后调用 perform\_atomic\_semop 来完成操作，perform\_atomic\_semop 的定义如下

static int perform_atomic_semop(struct sem_array *sma, struct sem_queue *q)
    {
        
    	int result, sem_op, nsops;
    	struct sembuf *sop;
    	struct sem *curr;
    	struct sembuf *sops;
    	struct sem_undo *un;
    
    	sops = q->sops;
    	nsops = q->nsops;
    	un = q->undo;
    
    	for (sop = sops; sop < sops + nsops; sop++) {
        
    		curr = &sma->sems[sop->sem_num];
    		sem_op = sop->sem_op;
    		result = curr->semval;
    ......
    		result += sem_op;
    		if (result < 0)
    			goto would_block;
    ......
    		if (sop->sem_flg & SEM_UNDO) {
        
    			int undo = un->semadj[sop->sem_num] - sem_op;
    .....
    		}
    	}
    
    	for (sop = sops; sop < sops + nsops; sop++) {
        
    		curr = &sma->sems[sop->sem_num];
    		sem_op = sop->sem_op;
    		result = curr->semval;
    
    		if (sop->sem_flg & SEM_UNDO) {
        
    			int undo = un->semadj[sop->sem_num] - sem_op;
    			un->semadj[sop->sem_num] = undo;
    		}
    		curr->semval += sem_op;
    		curr->sempid = q->pid;
    	}
    	return 0;
    would_block:
    	q->blocking = sop;
    	return sop->sem_flg & IPC_NOWAIT ? -EAGAIN : 1;
    }

perform\_atomic\_semop 会对信号量做两次遍历，第一次遍历用于测试，第二次遍历用于更新

第一次遍历，计算信号量的值是否可以满足此操作的要求，如果不满足，那可能需要睡眠等待，之后会跳到 would\_block，设置 q->blocking 为不满足的那个信号量，表示当前操作 block 在这个信号量上。如果设置了非阻塞，那么就返回 -EAGAIN，否则，返回1，表示需要当前进程需要阻塞等待

如果在第一次遍历中，信号量的值满足这个操作，那么就会进行第二次遍历，更新所有信号量的值

接下来回到 semtimedop 函数，看一看它干的第三件事，如果需要睡眠等待，那么它会怎么做呢？

如果需要睡眠，那么就需要判断这个操作是对整个信号量集，还是对某一个信号量。如果是对整一个信号量集，那么就将这个 queue 挂到这个信号量集的 pending\_alter 中。如果是对某一个信号量，那么就将这个 queue 挂到对应信号量的 pending\_alter 中

接下来就进入 do-while 的睡眠等待，如果没有设置超时时间，那么就调用 schedule 让出CPU，如果设置了，那么就调用 schedule\_timeout。需要注意的是，这里将进程设置为 TASK\_INTERRUPTIBLE 状态，表示可以被信号唤醒，这表示我们可以通过信号来终止一个正在等待信号量的进程

下面我们来看看 semtimedop 干的第四件事，如果不需要等待，应该怎么做呢？

如果不需要等待，就说明信号量已经改变了，现在看看是否能唤醒某些等待信号量的进程

首先会使用 DEFINE\_WAKE\_Q(wake\_q) 定义一个 wake\_q，然后通过 do\_smart\_update，看可以激活指定信号量上的哪些 struct sem\_queue，然后将可以唤醒的加入 wake\_q 中，之后再通过 wake\_up\_q 来唤醒这些进程

下面来看看 do\_smart\_update 事如何实现的，do\_smart\_update 会调用 update\_queue 来实现，其定义如下

static int update_queue(struct sem_array *sma, int semnum, struct wake_q_head *wake_q)
    {
        
      struct sem_queue *q, *tmp;
      struct list_head *pending_list;
      int semop_completed = 0;
    
      if (semnum == -1)
        pending_list = &sma->pending_alter;
      else
        pending_list = &sma->sems[semnum].pending_alter;
    
    again:
      list_for_each_entry_safe(q, tmp, pending_list, list) {
        
        int error, restart;
    ......
        error = perform_atomic_semop(sma, q);
    
        /* Does q->sleeper still need to sleep? */
        if (error > 0)
          continue;
    
        unlink_queue(sma, q);
    ......
        wake_up_sem_queue_prepare(q, error, wake_q);
    ......
      }
      return semop_completed;
    }
    
    static inline void wake_up_sem_queue_prepare(struct sem_queue *q, int error,
                   struct wake_q_head *wake_q)
    {
        
      wake_q_add(wake_q, q->sleeper);
    ......
    }

首先会取得整个信号量集的 pending\_list，或者某个信号量的 pending\_list，这个链表上面挂着 struct sem\_queue，struct sem\_queue 表示一个正在睡眠等待信号量的进程的操作

在取得 pending\_list 链表时候，会遍历这个链表，然后通过 perform\_atomic\_semop 来判断现在信号量的值是否满足这个操作，如果满足，就将这个 struct sem\_queue 从 pending\_list 中删除，然后通过 wake\_up\_sem\_queue\_prepare 将这个操作 q->sleeper 加入到 wake\_q 中

其中 q->sleeper 是一个 task\_struct，表示这个信号量等待的进程

接下来就是通过 wake\_up\_q 来依次唤醒 wake\_q 上面的进程

void wake_up_q(struct wake_q_head *head)
    {
        
      struct wake_q_node *node = head->first;
    
      while (node != WAKE_Q_TAIL) {
        
        struct task_struct *task;
    
        task = container_of(node, struct task_struct, wake_q);
    
        node = node->next;
        task->wake_q.next = NULL;
    
        wake_up_process(task);
        put_task_struct(task);
      }
    }

对于信号量的大体操作到这里也就结束了

下面思考一个问题，信号量在内核中是属于全局变量，也就是所有的进程都可以访问它，如果有一个进程通过P操作获取了一个信号量，当时当它没来得及释放信号量的时候，发生了异常退出。这就会导致其它进程都无法获取信号量而睡眠等待

## 四、SEM\_UNDO标志的含义 ##

为了解决这个问题，可以在每个操作设置 SEM\_UNDO 标志，每次进程操作信号量的时候，都会保存这个进程对应反操作，当进程异常退出的时候，就会执行这些反操作，从而撤销了这个进程对信号量的影响

在进程的 task\_struct 里有一个对于信号量的成员 struct sysv\_sem，里面有一个变量 struct sem\_undo\_list 保存该进程所有对信号量的操作对应的 undo 操作

struct task_struct {
        
    ......
    struct sysv_sem      sysvsem;
    ......
    }
    
    struct sysv_sem {
        
      struct sem_undo_list *undo_list;
    };
    
    struct sem_undo {
        
      struct list_head  list_proc;  /* per-process list: *
                 * all undos from one process
                 * rcu protected */
      struct rcu_head    rcu;    /* rcu struct for sem_undo */
      struct sem_undo_list  *ulp;    /* back ptr to sem_undo_list */
      struct list_head  list_id;  /* per semaphore array list:
                 * all undos for one array */
      int      semid;    /* semaphore set identifier */
      short      *semadj;  /* array of adjustments */
                /* one per semaphore */
    };
    
    struct sem_undo_list {
        
      atomic_t    refcnt;
      spinlock_t    lock;
      struct list_head  list_proc;
    };

下面举个例子来说明 undo 机制的原理

下面内核中有两个信号量集 semaphore1 和 semaphore2.semaphore1 中有三个信号量，它们的初始值都为3。semaphore2 中有四个信号量，它们的初始值都为4

有两个进程来对这两个信号量集操作

![img][]

进程1首先对 semaphore1 进行操作，分别对信号量 +1，+2，-3，之后信号量对应的值就变为 4，5，0，对应图中的（2）。然后在进程1的 sem\_undo 链表中，就保存了这个操作的反向操作 -1，-2，+3

进程2对 sempaphore2 进行操作，分别对信号量 -3，+2，+1，之后信号量对应的值就变为 1，7，1，对应图中的（3），然后进程2的sem\_undo 链表中，就保存了这个操作的反向操作

之后进程1和进程2对 semaphore2 进行操作也是同样的过程

当进程异常退出后，就会执行链表上的反向操作，可以想象的是，进程之前的操作都被撤销了，这就不会导致信号量的值不正确，而使得其它进程睡眠等待

## 五、总结 ##

*  信号量是 ipc 的一种方式，它在内核中使用 sem\_ids 来维护所有的信号量
 *  可以使用 shemget来创建信号量，信号量对象在内核中使用 struct sem\_array 来表示，它其实是一个信号量集合，可以包含多个信号量，每个信号量在内核中使用 struct sem 表示，都有自己对应的值
 *  可以通过 semctl 来设置信号量
 *  可以通过 semop 操作信号量，其实它们就是修改对应信号量的值，每一个操作都对应一个 sem\_queue，如果当前信号量的值不满足这个操作需求，那么就会将其加入对应信号量的链表中，然后当前进程睡眠等待
 *  当有信号量操作完成后，会判断当前信号量的值是否满足对应信号量链表上的操作，如果满足，就唤醒对应的进程
 *  如果在 semop 设置了 SEM\_UNDO 标志，那么就表示当进程异常退出的时候，要撤销这个操作。内核的实现是将这个 semop 操作的反操作保存到 task\_struct 的 undo 链表中，当进程异常退出的时候，就会遍历该链表来撤销进程对信号量的操作

[Linux]: https://blog.csdn.net/weixin_42462202/article/details/102630450
[Linux 1]: https://blog.csdn.net/weixin_42462202/article/details/102632425
[Linux 2]: https://blog.csdn.net/weixin_42462202/article/details/102640895
[Linux 3]: https://blog.csdn.net/weixin_42462202/article/details/102670752
[Linux_]: https://blog.csdn.net/weixin_42462202/article/details/102681512
[img]: /images/20230531/36f66f651a664a60abe3119ada3a9782.png