假设队列使用单向链表实现的:
struct node{
struct node *next;
int data;
}
struct node *queue;//队列头
多个消费者(多线程)都需要向这个queue插入数据
为了说明问题的复杂性,先看看只有一个消费者时的情况,插入队列的操作非常简单:
Step1) new_head->next = queue->head;
Step2) queue->head = new_head;
加入了多线程,问题变得复杂,以step 2为例,多个线程可能会同时进行这个操作,因此结果是不可预知的。
解决办法1)任何线程在进行step1之前先获取锁,得到step 2完成后再释放锁,这种办法是最简单的,但锁的性能开销较大,还可以考虑改进的办法。
一个比较妙的思路是:
每次操作之前先确认别的生产者没有在改变队列的头部,如果没有别的生产者正在操作,当前生产者就可以操作了。
do{
old_head = queue->head;
new_head->next = old_head;
if (old_head == queue->head){
queue->head = new_head;
}
}while(queue->head != new_head)
注意循环终止条件:
当queue->head等于new_head时,说明本生产者已经成功操作了队列
否则,说明本轮有其他生产者操作了队列,下轮再做尝试,直到成功为止
这样看起来可以保证只有一个生产者来操作队列了(其他的生产者),现在的问题是第4行和第5行无法保证原子执行,也就是说存在多个线程的4条件都成立,紧接着又都执行了,这样还是会出现错误。
这个问题如何解决呢?4和5步如果能原子性的执行,问题就很大程度上得到了解决。幸运的是不同架构的cpu都提供了类似cas/cmpxchg的指令,保证操作的原子性
下面的c代码说明了cas的含义(这里的代码是示意性的,实际的指令是原子性的)
int compare_and_swap (int* reg, int oldval, int newval)
{
int old_reg_val = *reg;
if (old_reg_val == oldval)
*reg = newval;
return old_reg_val;
}
有了这个指令之后,上面的代码就可以改写成多生产者安全的
do{
old_head = queue->head;
new_head->next = old_head;
val=cmpxchg(&queue->head, old_head, new_head);
}while(val!=old_head)
注意循环终止的判断条件:
当val == old_head时,说明3,4步之间没有生产者更改过队列头,操作已经成功
当val != old_head时,说明已经3,4步之间已经有其他生产者操作过队列,此时当前的生产者需要重新尝试操作队列
为何将2,3步放到循环内部呢?为了说明这个,可以假设将2,3放到循环外面会如何?假设第一轮其他生产者操作了队列,我们需要重新来过,重新来过时,queue->head已经是其他线程更新过的了,如果放到循环外面,old_head无法更新,而val则会返回新的head,此时判断条件会永远失败,导致死循环。