← 返回首页
读书笔记

《Operating System:Three Easy Pieces》第三十二章 常见并发问题

有哪些类型的缺陷

非死锁缺陷

违反原子性原则

违反了多次内存访问中预期的可串行性(即代码段本意是原子的,但在执行中并没有强制实现原子性)

c
Thread 1::// 判断和赋值应该是原子性的if (thd->proc_info) if (thd->proc_info){ ... fputs(thd->proc_info, ...); ... } Thread 2:: thd->proc_info= NULL;

通过加解决

违反顺序缺陷

两个内存访问的预期顺序被打破了(即 A 应该在 B 之前执行,但是实际运行中却不是这个顺序)

c
1 pthread_mutex_t proc_info_lock = PTHREAD_MUTEX_INITIALIZER; 2 3 Thread 1:: 4 pthread_mutex_lock(&proc_info_lock); 5 if (thd->proc_info) { 6 ... 7 fputs(thd->proc_info, ...); 8 ... 9 } 10 pthread_mutex_unlock(&proc_info_lock); 11 12 Thread 2:: 13 pthread_mutex_lock(&proc_info_lock); 14 thd->proc_info = NULL; 15 pthread_mutex_unlock(&proc_info_lock);

如果 mState = mThread->State 语句先执行,则 mThread 为空。通过加条件变量(或信号量)解决

c
1 pthread_mutex_t mtLock = PTHREAD_MUTEX_INITIALIZER; 2 pthread_cond_t mtCond = PTHREAD_COND_INITIALIZER; 3 int mtInit = 0; 4 5 Thread 1:: 6 void init() { 7 ... 8 mThread = PR_CreateThread(mMain, ...); 9 10 // signal that the thread has been created... 11 pthread_mutex_lock(&mtLock); 12 mtInit = 1; 13 pthread_cond_signal(&mtCond); 14 pthread_mutex_unlock(&mtLock); 15 ... 16 } 17 18 Thread 2:: 19 void mMain(...) { 20 ... 21 // wait for the thread to be initialized... 22 pthread_mutex_lock(&mtLock); 23 while (mtInit == 0) 24 pthread_cond_wait(&mtCond, &mtLock); 25 pthread_mutex_unlock(&mtLock); 26 27 mState = mThread->State; 28 ... 29 }

死锁缺陷

c
Thread 1: lock(L1); lock(L2); Thread 2: lock(L2); lock(L1);

两个线程互相等待

为什么会发生死锁:

产生死锁的条件:

预防死锁

循环等待

也许最实用的预防技术(当然也是经常采用的),就是让代码不会产生循环等待。最直接的方法就是获取锁时提供一个全序(total ordering)。

假如系统共有两个锁(L1 和 L2),那么我们每次都先申请 L1 然后申请 L2,就可以避免死锁。这样严格的顺序避免了循环等待,也就不会产生死锁。

当然,更复杂的系统中不会只有两个锁,锁的全序可能很难做到。因此,偏序(partial ordering)可能是一种有用的方法,安排锁的获取并避免死锁。Linux 中的内存映射代码就是一个偏序锁的好例子。代码开头的注释表明了 10 组不同的加锁顺序,包括简单的关系,比如 i_mutex 早于 i_mmap_mutex,也包括复杂的关系,比如 i_mmap_mutex 早于private_lock,早于 swap_lock,早于 mapping->tree_lock。 你可以想到,全序和偏序都需要细致的锁策略的设计和实现。另外,顺序只是一种约定,粗心的程序员很容易忽略,导致死锁。

提示:通过锁的地址来强制锁的顺序 当一个函数要抢多个锁时,我们需要注意死锁。比如有一个函数:do_something(mutex t *m1, mutext *m2),如果函数总是先抢 m1,然后 m2,那么当一个线程调用 do_something(L1, L2),而另一个线程调用 do_something(L2, L1)时,就可能会产生死锁。 为了避免这种特殊问题,聪明的程序员根据锁的地址作为获取锁的顺序。按照地址从高到低,或者从低到高的顺序加锁,do_something()函数就可以保证不论传入参数是什么顺序,函数都会用固定的顺序加锁。具体的代码如下: if (m1 > m2) { // grab locks in high-to-low address order pthread_mutex_lock(m1); pthread_mutex_lock(m2); } else { pthread_mutex_lock(m2); pthread_mutex_lock(m1); } // Code assumes that m1 != m2 (it is not the same lock) 在获取多个锁时,通过简单的技巧,就可以确保简单有效的无死锁实现

持有并等待

死锁的持有并等待条件,可以通过原子地抢锁来避免。实践中,可以通过如下代码来实现

c
lock(prevention); lock(L1); lock(L2); ... unlock(prevention);

非抢占

解决非抢占就是要让线程能占用其他线程没在用的锁,不能占着茅坑不拉屎。主动去占用其他线程已经占用的锁不可能实现,需要让线程能认识到自己占着锁也没用,找机会主动释放一次

c
top:lock(L1); if (trylock(L2)==-1) { unlock(L1); goto top; }

假设线程要同时拥有 L1 和 L2 才能继续,获得 L1 后尝试获取 L2,如果失败就解锁 L1,重新开始。

需要加一定的随机延迟,防止活锁(livelock)(双方都不停放弃锁)。

类似于数据库中的事务和回滚

互斥

Herlihy 提出了设计各种无等待(wait-free)数据结构的思想,通过强大的硬件指令,我们可以构造出不需要锁的数据结构。

*address 的值等于 expected 值时,将其赋值为 new,这是个硬件提供的原子操作–比较并交换(compare-and-swap)指令:

c
int CompareAndSwap(int *address, int expected, int new) { if (*address == expected) { *address = new; return 1; // success } return 0; // failure }

假定我们想原子地给某个值增加特定的数量

c
intCompareAndSwap(int*address,int expected,int new) { if (*address== expected) { *address= new; return 1;// success } return 0;// failure }

无须获取锁,更新值,然后释放锁这些操作,我们使用比较并交换指令,反复尝试将值更新到新的值。这种方式没有使用锁,因此不会有死锁(有可能产生活锁)。

一个更复杂的例子:链表插入。这是在链表头部插入元素的代码:

c
void insert(int value) { node_t *n = malloc(sizeof(node_t)); assert(n != NULL); n->value = value; n->next = head; head = n; }

通过加锁解决同步:

c
void insert(int value) { node_t*n= malloc(sizeof(node_t)); assert(n!= NULL); n->value= value; lock(listlock);// begin critical sectio nn->next= head; head = n; unlock(listlock);// end of critical section }

比较并交换指令(compare-and-swap)来实现插入操作:

c
voidinsert(int value) { node_t*n= malloc(sizeof(node_t)); assert(n!= NULL); n->value= value; do{ n->next = head; }while (CompareAndSwap(&head, n->next, n)== 0); }

这段代码,首先把 next 指针指向当前的链表头(head),然后试着把新结点交换到链表头。但是,如果此时其他的线程成功地修改了 head 的值,这里的交换就会失败,导致这个线程根据新的 head 值重试。

通过调度避免死锁

除了死锁预防,某些场景更适合死锁避免(avoidance)。我们需要了解全局的信息,包括不同线程在运行中对锁的需求情况,从而使得后续的调度能够避免产生死锁。

只要 T1 和 T2 不同时运行,就不会产生死锁。,T3 和 T1 重叠,或者和 T2 重叠都是可以的。虽然 T3 会抢占锁 L2,但是由于它只用到一把锁,和其他线程并发执行都不会产生死锁。

我们再来看另一个竞争更多的例子。在这个例子中,对同样的资源(又是锁 L1 和 L2)有更多的竞争。锁和线程的竞争如表 32.3 所示。

特别是,线程 T1、T2 和 T3 执行过程中,都需要持有锁 L1 和 L2。下面是一种不会产生死锁的可行方案:

你可以看到,T1、T2 和 T3 运行在同一个处理器上,这种保守的静态方案会明显增加完成任务的总时间。尽管有可能并发运行这些任务,但为了避免死锁,我们没有这样做,付出了性能的代价。

检查和恢复

最后一种常用的策略就是允许死锁偶尔发生,检查到死锁时再采取行动。

很多数据库系统使用了死锁检测和恢复技术。死锁检测器会定期运行,通过构建资源图来检查循环。当循环(死锁)发生时,系统需要重启。如果还需要更复杂的数据结构相关的修复,那么需要人工参与。

参考

本文由 GJJ 创作,内容来源于 Notion 数据库,随时可在 Notion 中编辑更新。 本站由 DeepSeek-v4-flash 辅助构建,项目参考 NotionNext

← 返回首页
61
文章
6
标签
3
分类
962
运行天数