本地自旋锁与信号量/多服务台自旋队列-spin wait风格的信号量

周日傍晚，我去家附近的超市(...)买苏打水，准备自制青柠苏打，我感受我作的比买的那个巴黎水要更爽口。因为天气太热，不少人都去超市避暑去了，超市 也不撵人，这仿佛是他们的策略，人过来避暑了，走的时候不免要买些东西的，就跟不少美女在公交地铁上看淘宝消磨时光，而后就下单了...这是多么容易一件 事，反之开车的美女网购就少不少。对于超市的避暑者，要比公交车上下单更麻烦些，由于有一个成本问题，这就是排队成本。
       其实这是一个典型的多服务台排队问题，可是超市处理的并很差，存在队头拥塞问题，我就好几回遇到过。好几回，我排的那个队，前面结帐出现了纠纷，咱们后面 的就必须等待，眼睁睁看着旁边的结帐队伍向前推动，可是这种排队方案足够简单，把调度任务交给了排队者本人，结帐的人想排到哪一个队列就排到哪一个队列，判断 一个队列是否会拥塞也有不少办法，好比看购物的多少，是否有衣物(锁卡拔出纠纷)，是否有称重的东西(会忘记称重)，是否有打折物，是否有老年人，收银员 的手法是否娴熟等，全靠本身的判断，无异于一场***。  我改造的Open×××多线程实现就是这种。
       银行服务以及饭店的排队服务就要好不少，顾客排队时，自取一个号码，排入单一的队列，由空闲服务台叫号，这就是一个调度系统。这种单队列多服务台是不会出 现队头拥塞的，等候的顾客持ticket排队，自己没必要排在队伍里，而ticket号逻辑上组成一个虚拟的队列，没叫到号的能够暂时干点别的，自身没必要排 队。

       暂时干别的？并不意味着你能够离开，特别是业务处理流程很快的状况下。你离开大厅，刚走出去，准备去旁边的小店逛逛，结果听到叫到你的号了，赶忙返回，其 实还不如不出去呢。可是对于等待比较久的叫号系统，那却是能够暂时出去。出去再返回的过程意味着体力开销，可是若是出去的时间久，能够完成另外一件重要的 事，意味着为这另外这件事的收益付出的体力开销是值得的。

       知道我想到什么了吗？我想到了信号量。信号量就是一个单队列多服务台排队系统，信号量的初始值就是服务台的数量。一个执行流被服务意味着少了一个可服务的 服务台，这就是down操做，而up操做则是一个服务台从新变成空闲的信号，这意味着有一个新的排队者能够获得服务了，我能够把”服务“理解成进入临界 区。

       我在想一个问题，为何信号量必定要设计成sleep-wait的模式，为何就没有spin-wait的模式啊。而我目前面临的问题，若是使用 sleep-wait，切换开销太大，perf显示的头几名大头都在schedule，wake up，之类的，也就是说，你切换出去了，没多久就又把你叫回来了，好在Linux调度系统基于CFS彻底公平机制，抖动不会太厉害，不过这么切换一次形成 的开销也不算小，起码等到再次切换回来的时候，cache变凉了。

回顾Linux版的ticket自旋锁，我以为全部的排队者以及持锁者 touch同一个变量，该变量会cache到全部的当事者cpu的cache中，被持锁者以及争锁者read/write时，会涉及到多个处理器之间的 cache一致性问题，这也是一笔很大的底层开销。因而我设计了一个本地接力自旋锁改变了这个局面，保持每个争锁者都只touch一个别的争锁者不会 touch的变量，且cache line要着色以保证不会cache到同一line，此外，持锁者在释放锁的时候，只会write下一个争锁者的本地变量。这样就确保了cache一致性 被最少的触发。
       本着这个新的自旋锁设计，结合我在超市的经历，我想把我这个自旋锁发展成一个能够有多个CPU持有锁的自旋队列。后来我忽然发现，这不就是信号量嘛... 惋惜信号量并无如期被我所用，由于Linux实现的信号量是sleep-wait机制的，我须要的是spin-wait，由于我知道一个数据包的发送是 很快的，之因此引入队列，构建VOQ，是由于我想避开N加速比问题，然而个人算法是软实现，根本不存在N加速比问题，因此后来我想取消VOQ，又怕引起队 头拥塞，因此采用了多服务台单队列机制，为了实现这个，我本能够采用信号量的，可是又不想sleep，因此采用极其复杂的多个spin lock的机制，超市排队引起的遐想致使我想到用spin-wait来实现信号量，事实上，简单测试以后，发现效果还真不错。

先看一下Linux原生的信号量实现，代码比较简单。顺便说一句，这篇文章并不意味着我又开始源代码分析了，而是也许它意味着某种终结，先后的呼应。

/*
 * 为了突出重点问题,不至于迷失在代码细节.我作了如下的假设:
 * 1.我省去了操做信号量自己的自旋锁,我假设P/V操做过程的任意序列都是原子的.
 * 2.我取消了超时参数以及state,我假设除非获得信号量,不然必定等下去,我还假设睡眠不会被打断,除非有人唤醒.
 * 3.我取消了inline，由于我想突出围绕本地栈变量本地自旋，这样不会cache pingpong.
 */
struct semaphore {
    raw_spinlock_t        lock;
    unsigned int        count;
    struct list_head    wait_list;
};

struct semaphore_waiter {
    struct list_head list;
    struct task_struct *task;
    // 本地局部检测变量
    bool up;
};


static int down(struct semaphore *sem)
{
    if (likely(sem->count > 0)) {
        sem->count--;
    }
    else {
        struct task_struct *task = current;
        struct semaphore_waiter waiter;
        // 栈上的排队体,至关于ticket,得到信号量(函数返回)后就没有用了
        list_add_tail(&waiter.list, &sem->wait_list);
        waiter.task = task;
        waiter.up = false;

        for (;;) {
            __set_task_state(task, TASK_UNINTERRUPTIBLE);
            schedule();

            // 本地栈变量的检测，减小了多处理器之间的cache同步，不会cache乒乓
            // ********************************************************************
            // 可是要想到一种状况，若是多个进程试图写这个变量，仍是要有锁操做的。
            // 虽然个人假设是全部操做以及操做序列都是原子的，可是在up操做中，持有信
            // 号量的进程只是简单的wake up了队列，而这并不能确保被唤醒的task就必定可
            // 以获得执行，中间还有一个schedule层呢。鉴于这种复杂的局面，我想到了不
            // sleep，而是本地自旋版本的信号量，无论怎样，它确实解决了个人问题。
            // [事实上，因为sem自己拥有一把自旋锁，这就禁止了多个“服务台”同时召唤
            //  同一个等待者的局面，而我在个人描述中，忽略了这把自旋锁，这是为何呢？
            //  由于，我想为个人自旋信号量版本贴金，否则人家都把问题解决了，我还扯啥
            //  玩意儿啊！]
            // ********************************************************************
            // 这种状况在spin lock下不会存在，由于同时只有一个进程会持有lock，
            // 不可能多个进程同时操做。

            if (waiter.up) {
                return 0;
            }
        }
    }
}

void up(struct semaphore *sem)
{
    unsigned long flags;
    if (likely(list_empty(&sem->wait_list))) {
        sem->count++;
    } else {
        struct semaphore_waiter *waiter = list_first_entry(&sem->wait_list,
                            struct semaphore_waiter, list);
        // 标准的Linux kernel中，该操做被spin lock保护，这意味着不可能多个服务台同时将
        // 服务给与同一个等待者。
        list_del(&waiter->list);
        waiter->up = true;
        // 简单wake up进程，它什么时候投入运行，看调度器什么时候调度它了。
        wake_up_process(waiter->task);
    }
}

因为我忽略了信号量自己的保护自旋锁，当你详细分析上述实现的时候，会发现不少竞争条件，好比同时多个服务台召唤一个等待者，可是 不要紧，该说的我都写到冗长的注释里面了。我之因此忽略信号量的自旋锁，是由于我想把信号量该形成一个通用的自旋等待队列，自旋锁只是其中一个特殊状况， 该状况对应只有一个服务台的情形。
       若是看懂了原生的实现，那么改造后的实现应该是如下的样子：

/*
 * 我引入了BEGIN_ATOMIC和END_ATOMIC两个宏，由于我不想贴汇编码，因此这两个宏的意思就是它们之间的代码都是由
 * lock前缀修饰的，锁总线。
 * 此外，什么事情都没有作，只是改了名称。若是想初始化一个标准的排队自旋锁，将初始化宏的val设置成1便可。
 */
struct spin_semaphore {
    unsigned int        count;
    struct list_head    wait_list;
};

struct spin_semaphore_waiter {
    struct list_head list;
    struct task_struct *task;
    // 本地局部检测变量
    bool up;
};


static int spin_down(struct spin_semaphore *sem)
{
    if (likely(sem->count > 0)) {
        sem->count--;
    }
    else {
        struct task_struct *task = current;
        struct spin_semaphore_waiter waiter;
BEGIN_ATOMIC
        list_add_tail(&waiter.list, &sem->wait_list);
        waiter.task = task;
        waiter.up = false;
END_ATOMIC

        for (;;) {
            cpu_relax();  // PAUSE
            if (waiter.up) {
                return 0;
            }
        }
    }
}

void up(struct spin_semaphore *sem)
{
    unsigned long flags;
BEGIN_ATOMIC
    if (likely(list_empty(&sem->wait_list))) {
        sem->count++;
END_ATOMIC
    }
    else {
        struct spin_semaphore_waiter *waiter = list_first_entry(&sem->wait_list,
                            struct spin_semaphore_waiter, list);
        list_del(&waiter->list);
        waiter->up = true;
END_ATOMIC
    }
}

全部名称加上了spin_前缀修饰。不错，这个应该是和Windows NT内核的排队自旋锁的实现很接近了。在此不谈优化，然而实际使用时，应该是先用汇编编码，而后汇编码优化它了。