深入浅出 Barriers 实现（一）

Barriers，字面意思为“壁垒，屏障，栅栏”，在计算机领域中 Barriers 也有它独特的含义，具体来讲，在并行程序中，Barriers 是一种同步的手段，可被视为一种线程同步原语，如一组线程/进程的 Barrier 可以用来同步该线程/进程组，只有当该线程/进程组中所有线程到达屏障点（可称之为同步点）时，整个程序才得以继续执行。如比较熟悉的 Memory Barriers(Wikipedia)，Memory Barriers可翻译为内存屏障。由于现代绝大多数的CPU都采用流水线和乱序执行(out-of-order)，一条指令的执行一般分为：取指，译码，访存，执行，写回的若干个阶段，多条指令可能同时存在与流水线中并同时被执行，由于CPU具备乱序执行，往往后取的指令被提前执行了，还有，在多处理器的条件下，CPU 可能共享 Cache，此时就会发生 Cache 不一致的情况，再加上编译器优化，问题就变得很复杂了，如何保证指令执行的一致性呢？内存屏障就发挥了很大作用。在 Linux 内核中，最简单的内存屏障原语是（更多有关内核屏障请参看文档）：

#define barrier() __asm__ __volatile__("":::"memory")

barrier() 的作用就是告诉编译器，内存中变量的值已经改变了，之前保存与寄存器或cache中的变量副本无效，如果访问该变量需要直接去内存中读取。内存屏障主要有：读屏障，写屏障，优化屏障，通用屏障等，以读屏障为例，它用于保证读操作有序，屏障之前的读操作一定会由于屏障之后的读操作完成，写屏障与此类似，只是用于限制写操作。优化屏障用于限制编译器的指令重排，关于优化屏障，可以参考以下系列文章，1, 2, 3 。

好了，上面讲的内容只是开始，现在就让我们来一步一步动手实现并优化我们自己的 Barriers 吧 :-)

前面讲过，Barriers 是一种基本的同步原语，它可以告诉你一组线程在什么时候完成了各自的任务可以接下来进行其他的工作，即一旦所有的线程都到达了屏障点，它们才能够继续执行下去，否则先到达屏障点的线程就会在此处等待其他线程的到来，鉴于此，屏障操作也是一个相当重量级的同步操作。但是，很多并行算法都可以通过屏障来把复杂的计算任务分阶段执行后进行同步，以此来简化算法的设计。

pthread 库本身支持 pthread_barrier_t 数据类型，并且提供了pthread_barrier_init, pthread_barrier_destroy, pthread_barrier_wait API, 为了使用 pthread 的 barrier, 你必须先初始化 barrier，参数为需要同步的线程数目（以及其他可选的属性，如 barrier 是用于线程间的同步还是线程内的同步），然后各个线程通过调用 pthread_barrier_wait 来进行同步，最后调用 pthread_barrier_destroy 销毁创建的 barrier。

销毁 barrier 原语需要特殊处理，而许多其他的同步原语可能根本就不需要销毁操作，因为他们并不分配内存，并且其依赖 的futex 系统调用也不需要显式地释放资源， 虽然 barrier 也不分配内存同时也使用了 futex 系统调用( 基于 Linux 的实现)，但是仍然需要一些工作要做。

但问题是，barrier 销毁的销毁操作可能发生在其他线程离开 barriers 时，那些线程需要检查 barrier 的状态，如果此时 barrier 使用的内存被释放了，那么程序的行为就不确定了，不过大部分情况下都可能发生 segfault。因此，销毁 barrier 的线程需要等待其他的线程都退出了 barrier 后才能执行销毁操作。为了达到该要求，我们需要做一些额外的同步，不过幸运的是，我们可以使用条件变量和互斥量来完成该工作，本文稍后会给出一个最简单的例子。

我们可以通过 pthread 的互斥量和条件变量来实现简单的 barrier，大致代码如下：

•  pthread-barrier.h

*
 * =============================================================================
 *
 *       Filename:  pthread-barrier.h
 *
 *    Description:  pthread barrier implementation.
 *
 *        Created:  12/26/2012 11:08:06 AM
 *
 *         Author:  Fu Haiping (forhappy), haipingf@gmail.com
 *        Company:  ICT ( Institute Of Computing Technology, CAS )
 *
 * =============================================================================
 */
#ifndef _PTHREAD_BARRIER_H_
#define _PTHREAD_BARRIER_H_

#include <pthread.h>

typedef struct barrier_s_ barrier_t;

struct barrier_s_ {
    unsigned count;
    unsigned total;
    pthread_mutex_t m;
    pthread_cond_t cv;
};

#define BARRIER_FLAG (1UL<<31)

extern void barrier_init(barrier_t *b, unsigned count);

extern int barrier_wait(barrier_t *b);

extern void barrier_destroy(barrier_t *b);

#endif /* _PTHREAD_BARRIER_H_ */

• pthread-barrier.c

/*
 * =============================================================================
 *
 *       Filename:  pthread-barrier.c
 *
 *    Description:  pthread barrier implementation.
 *
 *        Created:  12/26/2012 11:08:14 AM
 *
 *         Author:  Fu Haiping (forhappy), haipingf@gmail.com
 *        Company:  ICT ( Institute Of Computing Technology, CAS )
 *
 * =============================================================================
 */

#include "pthread-barrier.h"

void barrier_destroy(barrier_t *b)
{
    pthread_mutex_lock(&b->m);

    while (b->total > BARRIER_FLAG)
    {
        /* 等待所有线程退出 barrier. */
        pthread_cond_wait(&b->cv, &b->m);
    }

    pthread_mutex_unlock(&b->m);

    pthread_cond_destroy(&b->cv);
    pthread_mutex_destroy(&b->m);
}

void barrier_init(barrier_t *b, unsigned count)
{
    pthread_mutex_init(&b->m, NULL);
    pthread_cond_init(&b->cv, NULL);
    b->count = count;
    b->total = BARRIER_FLAG;
}

int barrier_wait(barrier_t *b)
{
    pthread_mutex_lock(&b->m);

    while (b->total > BARRIER_FLAG)
    {
        pthread_cond_wait(&b->cv, &b->m);
    }

    /* 是否为第一个到达 barrier 的线程? */
    if (b->total == BARRIER_FLAG) b->total = 0;

    b->total++;

    if (b->total == b->count)
    {
        b->total += BARRIER_FLAG - 1;
        pthread_cond_broadcast(&b->cv);

        pthread_mutex_unlock(&b->m);

        return PTHREAD_BARRIER_SERIAL_THREAD;
    }
    else
    {
        while (b->total < BARRIER_FLAG)
        {
            pthread_cond_wait(&b->cv, &b->m);
        }

        b->total--;

        /* 唤醒所有进入 barrier 的线程. */
        if (b->total == BARRIER_FLAG) pthread_cond_broadcast(&b->cv);

        pthread_mutex_unlock(&b->m);

        return 0;
    }
}

 测试程序如下：

#include <time.h>
#include <stdio.h>
#include <stdlib.h>
#include <pthread.h>

#include "pthread-barrier.h"
#include "futex-barrier.h"
#include "ticket-barrier.h"
#include "fast-barrier.h"

// #define USE_MUTEX_CV_BARRIER

#if defined(USE_MUTEX_CV_BARRIER)
  #define pthread_barrier_t barrier_t
  #define pthread_barrier_init(B, A, N) (barrier_init((B), (N)), 0)
  #define pthread_barrier_destroy(B) (barrier_destroy(B), 0)
  #define pthread_barrier_wait barrier_wait

#endif

/* Number of threads to use */
#define N 2

/* Number of times to call barrier primitive */
#define COUNT 1000000

static pthread_barrier_t barrier;

static void *do_barrier_bench(void *arg)
{
    int i;

    (void) arg;

    for (i = 0; i < COUNT; i++)
    {
        pthread_barrier_wait(&barrier);
    }

    return NULL;
}

int main(void)
{
    pthread_t th[N - 1];
    clock_t start, end;
    int i;

    pthread_barrier_init(&barrier, NULL, N);

    for (i = 0; i < N - 1; i++) {
        if (pthread_create(&th[i], NULL, do_barrier_bench, NULL)) {
            fprintf(stderr, "pthread_create failed");
            exit(EXIT_FAILURE);
        }
    }
    start = clock();

    do_barrier_bench(NULL);

    end = clock();

    pthread_barrier_destroy(&barrier);

    for (i = 0; i < N - 1; i++) {
        if (pthread_join(th[i], NULL)) {
            fprintf(stderr, "pthread_join failed");
            exit(EXIT_FAILURE);
        }
    }

    puts("bench OK");

    printf("time elasped %lds.\n", (end - start) / CLOCKS_PER_SEC);
    return 0;
}