距那次该文预览早已一年多了。消亡的另一面可能将为的是更进一副棋的结晶。他们谈了大一年的缓存持续性此根底。无人晓得鸽了很久，有啥人还在等着预览呢。在那几天里，我也比过往对天然屏障的认知更进一副棋了一步棋。
曾我也想过两个问题，在现实生活中吗会碰着因为天然屏障缺位引致的BUG吗？而他们又该如何接纳天然屏障呢？怎么推论与否若是接纳天然屏障呢？当必要的时候，又该如何优先选择何种天然屏障类别呢？那不，巧了，消亡的一年中，我还吗fix了两个天然屏障缺位的BUG。刚好借助于那个规范两栖做战呵呵吧。
[w,r]mb和smp_[w,r]mb的优先选择LinuxMach中很难单纯三品种此外天然屏障号令的PCB。smp_[w,r]mb()和[w,r]mb()。标识符中他们时常看见都是"smp_"结尾的天然屏障。当他们写标识符时如何在那二者之间优先选择呢？我单纯说下我的优先选择国际尺度（能连结测度立场）。假设缓存操做体例的次序的不雅测者尽是CPU，那么请接纳smp_[w,r]mb()。假设缓存操做体例次序牵扯CPU和硬体电子设备（好比收集电子设备），请接纳[w,r]mb()。除此之外，[w,r]mb()比smp_[w,r]mb()的要求更严苛。因而操控性方面也是[w,r]mb()比smp_[w,r]mb()差。假设他们不写驱动力否则，只不外少少和电子设备关系亲近。因而一般都是优先选择smp_[w,r]mb()。
他们看两个wmb()和rmb()的接纳规范。他们必要到电子设备驱动力中找寻，就别忘了选两个我也不切当的存储设备驱动力吧（drivers/net/8139too.c）。
static netdev_tx_t rtl8139_start_xmit (struct sk_buff *skb, struct net_device *dev) { /* * Writing to TxStatus triggers a DMA transfer of the data * copied to tp->tx_buf[entry] above. Use a memory barrier * to make sure that the device sees the updated data. */ wmb(); RTL_W32_F (TxStatus0 + (entry * sizeof (u32)), tp->tx_flag | max(len, (unsigned int)ETH_ZLEN));}从那里的注解他们只不外能窥见RTL_W32_F()操做体例若是是策动一次电子设备DMA操做体例（通过写硬体暂存器实现）。wmb()的感化是确保载入DMA策动操做体例号令以后载入缓存的统计数据都早已载入缓存，确保DMA操做体例时能看见新一代的统计数据。单纯思索就是确保暂存器操做体例必需是最初发作的。他们继续找个rmb()的规范（drivers/net/bnx2.c）。
static int bnx2_rx_int(struct bnx2 *bp, struct bnx2_napi *bnapi, int budget) { sw_cons = rxr->rx_cons; sw_prod = rxr->rx_prod; /* Memory barrier necessary as speculative reads of the rx * buffer can be ahead of the index in the status block */ rmb(); while (sw_cons != hw_cons) { }那里的rmb()就是为的是避免while轮回里面的load操做体例在rmb()以后发作。也就说那里是有次序要求的。电子设备方面的规范就不多说了，究竟结果不是他们存眷的重点。
编译器天然屏障barrier()的优先选择编译器天然屏障barrier()不牵扯任何硬体号令。是最弱的一种天然屏障，只对编译器有效。他们什么情况下若是优先选择barrier()呢？
还记得以后的该文提到如何推论与否必要接纳硬体天然屏障号令的原则吗？你写的那段标识符（包罗缓存操做体例）在实正意义上存在并行施行，也就是可能将存在不行两个CPU不雅测者。假设你能确保不成能将同时存在多个CPU不雅测者（好比施行那段标识符的历程都是绑定到两个CPU），现实上你就不必要硬体号令级别天然屏障。此时你必要考虑的问题只要：与否必要接纳编译天然屏障。
void queued_spin_lock_slowpath(struct qspinlock *lock, u32 val) { node = this_cpu_ptr(&qnodes[0].mcs); idx = node->count++; /* * Ensure that we increment the head node->count before initialising * the actual node. If the compiler is kind enough to reorder these * stores, then an IRQ could overwrite our assignments. */ barrier(); node->locked = 0; node->next = NULL; }那里以qspinlock的标识符为例申明。那里的node是两个per cpu变量。根据per cpu变量的接纳规则，统一时间只会存在两个CPU写那块缓存。因而那里不存在多个CPU不雅测者。因而不必要CPU号令级此外天然屏障。而那里同样有次序的要求，因而编译器天然屏障barrier()足以。
smp_[w,r]mb()的优先选择如今回归到他们的重点存眷对象了。当同时存在多个CPU不雅测者，而且他们又有缓存操做体例次序的要求时，他们就若是考虑smp_[w,r]mb()的接纳。在以后的该文中他们提到下面标识符示例用来申明TSO乱序。
initial: X = Y = 0 CPU 1 CPU 2 =============================== =============================== X = 1; Y = 1; smp_mb(); smp_mb(); LOAD Y LOAD X因为smp_mb()的加持，那里CPU1和CPU2至少有两个CPU能看见此中两个变量的值是1。不成能将发作CPU1看见Y的值是0，同时CPU2看见X的值是0的情况。那里笼统的是两个模子，两个十分常见的模子。好比下面的标识符你必定时常碰见。
他们时常看见有个历程测验考试期待两个变量event_indicated为true，然撤退退却出轮回。不然继续schedule()。
for (;;) { set_current_state(TASK_UNINTERRUPTIBLE); if (event_indicated) break; schedule(); }另一段负责唤醒的标识符凡是长下面如许。置位event_indicated，然后唤醒历程。
event_indicated = 1; wake_up_process(event_daemon);他们想到达的目标只不外很单纯：当event_indicated为1时，历程必然是被唤醒的形态。他们把set_current_state()以及wake_up_process()部门细节展开如下：
CPU 1 (Sleeper) CPU 2 (Waker) =============================== =============================== current->state = TASK_UNINTERRUPTIBLE; event_indicate = 1 LOAD event_indicated if ((LOAD task->state) & TASK_NORMAL) task->state = TASK_RUNNINGset_current_state的目标是设置当前历程形态为TASK_UNINTERRUPTIBLE。然后推论event_indicated是都满足前提。而唤醒的逻辑是先无前提设置event_indicated。然后测验考试唤醒历程，但是唤醒前会先推论被唤醒的历程的形态state与否为TASK_UNINTERRUPTIBLE形态。假设不是TASK_UNINTERRUPTIBLE，申明历程早已处于唤醒形态，无需再次唤醒。所以那里牵扯2个统计数据，别离是current->state和event_indicated。为的是确保sleep历程被唤醒并退出while轮回，他们必需确保要么CPU1看见CPU2设置的event_indicate（然撤退退却出轮回），要么CPU2看见历程的state形态是TASK_UNINTERRUPTIBLE（然后施行唤醒操做体例），要么CPU1即看见event_indicate等于1而且CPU2看见历程的state形态是TASK_UNINTERRUPTIBLE（唤醒早已running的历程没有问题）。但是他们不期望看见的成果是：CPU1看见event_indicated的值是0，CPU看见历程的state形态不是TASK_UNINTERRUPTIBLE。那种情况下，sleep历程就错过了被唤醒。所以那里的模子笼统出来就是上面的XY模子。
为的是避免那种情况呈现，他们必要插入2条smp_mb()。
#define set_current_state(state_value) \ smp_store_mb(current->state, (state_value)) static int try_to_wake_up(struct task_struct *p, unsigned int state, int wake_flags) { /* * If we are going to wake up a thread waiting for CONDITION we * need to ensure that CONDITION=1 done by the caller can not be * reordered with p->state check below. This pairs with smp_store_mb() * in set_current_state() that the waiting thread does. */ raw_spin_lock_irqsave(&p->pi_lock, flags); smp_mb__after_spinlock(); if (!(p->state & state)) goto unlock; }set_current_state()的实现是smp_store_mb()，相当于写完state后插入smp_mb()。wake_up_process()最末会挪用try_to_wake_up()。他们在读取历程state前也插入了smp_mb()（smp_mb__after_spinlock）。
两栖做战分享他们fix两个io_uring的BUG。patch能看考Fix missing smp_mb() in io_cancel_async_work()。
diff --git a/fs/io_uring.c b/fs/io_uring.c index 2f46def7f5832..5d9583e3d0d25 100644 --- a/fs/io_uring.c +++ b/fs/io_uring.c @@ -2252,6 +2252,12 @@ static void io_sq_wq_submit_work(struct work_struct *work) if (!ret) { req->work_task = current; + + /* + * Pairs with the smp_store_mb() (B) in + * io_cancel_async_work(). + */ + smp_mb(); /* A */ if (req->flags & REQ_F_CANCEL) { ret = -ECANCELED; goto end_req; @@ -3730,7 +3736,15 @@ static void io_cancel_async_work(struct io_ring_ctx *ctx, req = list_first_entry(&ctx->task_list, struct io_kiocb, task_list); list_del_init(&req->task_list); - req->flags |= REQ_F_CANCEL; + + /* + * The below executes an smp_mb(), which matches with the + * smp_mb() (A) in io_sq_wq_submit_work() such that either + * we store REQ_F_CANCEL flag to req->flags or we see the + * req->work_task setted in io_sq_wq_submit_work(). + */ + smp_store_mb(req->flags, req->flags | REQ_F_CANCEL); /* B */ + if (req->work_task && (!files || req->files == files)) send_sig(SIGINT, req->work_task, 1); }单纯说呵呵那里的BUG，那里也同样设想2个变量，别离是req->flags和req->work_task。模子笼统简化如下：
CPU 1 CPU 2 =============================== =============================== STORE req->flags STORE req->work_task smp_mb(); smp_mb(); LOAD req->work_task LOAD req->flags不期望看见的成果：req->work_task == NULL && (req->flags & REQ_F_CANCEL) == 0。所以他们必要2个smp_mb()确保不呈现那种成果。
BTW，那个BUG也是那个模子，有兴趣也能看看。