Golang 中的回环栅栏

这篇文章中我们会研究一个基本的同步问题。并使用 Golang 中原生的 Buffered Channels 来为这个问题找到一个简洁的解决方案。

问题

现在假设我们我们有一堆 workers。为了充分发挥 CPU 多核的能力，我们让每个 worker 运行在单独的 goroutine 中：

for i := 0; i < workers; i++ {
    go worker()
}

worker 需要做一系列的工作 job：

func worker() {
    for i := 0; i < 3; i++ {
        job()
    }
}

每次 job 前都需要在所有的 worker 上同步地先进行一次准备 bootstrap 的过程。也就是说，每个 worker 在执行 job 前，需要等待所有其他 worker 都完成 bootstrap 的准备。

func worker() {
    for i := 0; i < 3; i++ {
        bootstrap()
        # wait for other workers to bootstrap
        job()
    }
}

还有件事。如果至少有一个 worker 仍在执行 job，则所有 worker 的下一次的 bootstrap 都不能开始。换句话说，每次的 bootstrap 都是为紧接着的 job 部分做准备的，所以不能在上一次的 job 尚未结束之前就开始下一次的 bootstrap：

func worker() {
    for i := 0; i < 3; i++ {
        # wait for all workers to finish previous loop
        bootstrap()
        # wait for other workers to bootstrap
        job()
    }
}

我们的 bootstrap 部分内容为增长一个共享的计数器。job 部分为等待一段时间并打印计数器的内容：

type counter struct {
    c int
    sync.Mutex
}

func (c *counter) Incr() {
    c.Lock()
    c.c += 1
    c.Unlock()
}

func (c *counter) Get() (res int) {
    c.Lock()
    res = c.c
    c.Unlock()
    return
}

func worker(c *counter) {
    for i := 0; i < 3; i++ {
        # wait for all workers to finish previous loop
        c.Incr()
        # wait for other workers to do bootstrap
        time.Sleep(time.Duration(rand.Intn(100)) * time.Millisecond)
        fmt.Println(c.Get())
    }
}

我们的目标是编写一个程序并以下列情形打印数字：

• 只有 n, 2*n 和 3*n 的数字被打印（因为每个 worker 循环 3 次）
• 每次打印的数字不会比之前的数字小
• 每个数字会被打印 n 次

如果有 3 个 worker，则期望输出如下：

3
3
3
6
6
6
9
9
9

2 个 worker 的期望输出：

2
2
4
4
6
6

2 个 worker 不合法的输出可能会是这样：

2
4
2
4
6
6

想一想可能的解决办法。下面几行我故意留空并不急着给大家答案。

.

.

.

.

.

.

.

.

.

.

.

.

.

.

workers 会用一个名为回环栅栏（reusable barrier，类似 Java 中的 CyclicBarrier）的数据结构来实现同步。每个栅栏包含 2 扇门。第 1 扇门放置于增长计数器之前，一开始是关闭的。关闭的门意味着到达这扇门的 worker 会被阻塞。一旦所有的 worker 到达了第 1 扇门：

• 第 2 扇门（放置于增长计数器之后）会关闭
• 第 1 扇门开启

所有的计数器会通过并增长计数器，接着成功到达第 2 扇门。一旦所有的 worker 都到达了第 2 扇门：

• 第 1 扇门关闭
• 第 2 扇门开启

worker 此时可以开始执行 job 并接着在下一次循环中再次抵达第 1 扇门。循环再次开始。整个过程如图所示：

     1st gate     2nd gate
        v             v
-w1-->  |             |
 --w2-->|
--w3--> |          
 --w4-->|             
-w5-->  |             |
 --w1-->|             |
 --w2-->|             
 --w3-->|
 --w4-->|             
 --w5-->|             |
        |      --w1-->|
               --w2-->|
            --w3-->   |
             --w4-->  |
        |      --w5-->|
        |      --w1-->|
               --w2-->|
               --w3-->|
               --w4-->|
        |      --w5-->|
--w1--> |             |
        |              --w2-->
 --w3-->|
--w4--> |             
        |             | --w5-->

下文有两种解决方案，实现略有不同。下面的代码可用来对两种方案进行测试：

package main

import (
    "fmt"
    "math/rand"
    "sync"
    "time"
    // Set this import spec to point
    // to the copy of one of proposed
    // solutions.
    "path/to/package/barrier"
)

func init() {
    rand.Seed(time.Now().Unix())
}

type counter struct {
    c int
    sync.Mutex
}

func (c *counter) Incr() {
    c.Lock()
    c.c += 1
    c.Unlock()
}

func (c *counter) Get() (res int) {
    c.Lock()
    res = c.c
    c.Unlock()
    return
}

func worker(c *counter, br *barrier.Barrier, wg *sync.WaitGroup) {
    for i := 0; i < 3; i++ {
        br.Before()
        c.Incr()
        br.After()
        time.Sleep(time.Duration(rand.Intn(100)) * time.Millisecond)
        fmt.Println(c.Get())
    }
    wg.Done()
}

func main() {
    var wg sync.WaitGroup
    workers := 3
    br := barrier.New(workers)
    c := counter{}
    for i := 0; i < workers; i++ {
        wg.Add(1)
        go worker(&c, br, &wg)
    }
    wg.Wait()
}

栅栏必须实现 Before 和 After 两个方法，各自对应第 1 和第 2 扇门。

解决方案 1

我们需要容量为 1 的 buffered channel：

ch := make(chan int, 1)

门的逻辑可以通过先从 channel 中接收数据，然后再次向其中发送数据来实现：

<-ch
ch <- 1

如果 channel 中包含元素数量为 1，则表示门是开的。它会让一个 worker 通过并往 channel 中放入新的元素以使另外一个 worker 通过，依此类推。

如果 channel 中没有元素了则表示门关闭了。接着 worker 从 channel 中接收元素就会被阻塞。

// github.com/mlowicki/barrier
package barrier 

import "sync"

type Barrier struct {
    c      int
    n      int
    m      sync.Mutex
    before chan int
    after  chan int
}

func New(n int) *Barrier {
    b := Barrier{
        n:      n,
        before: make(chan int, 1),
        after:  make(chan int, 1),
    }
    // close 1st gate
    b.after <- 1  
    return &b
}

func (b *Barrier) Before() {
    b.m.Lock()
    b.c += 1
    if b.c == b.n {
        // close 2nd gate
        <-b.after
        // open 1st gate
        b.before <- 1
    }
    b.m.Unlock()
    <-b.before
    b.before <- 1
}

func (b *Barrier) After() {
    b.m.Lock()
    b.c -= 1
    if b.c == 0 {
       // close 1st gate
       <-b.before
       // open 2st gate
       b.after <- 1
    }
    b.m.Unlock()
    <-b.after
    b.after <- 1
}

解决方案 2

这个方案使用了容量和 worker 数量 n 相等的 buffered channel。现在我们不再让 worker 一个接一个地依次通过，而是在 channel 中放入 n 个元素来使所有的 worker 一次性通过：

// github.com/mlowicki/barrier2
package barrier

import "sync"

type Barrier struct {
    c      int
    n      int
    m      sync.Mutex
    before chan int
    after  chan int
}

func New(n int) *Barrier {
    b := Barrier{
        n:      n,
        before: make(chan int, n),
        after:  make(chan int, n),
    }
    return &b
}

func (b *Barrier) Before() {
    b.m.Lock()
    b.c += 1
    if b.c == b.n {
        // open 2nd gate
        for i := 0; i < b.n; i++ {
            b.before <- 1
        }
    }
    b.m.Unlock()
    <-b.before
}

func (b *Barrier) After() {
    b.m.Lock()
    b.c -= 1
    if b.c == 0 {
        // open 1st gate
        for i := 0; i < b.n; i++ {
            b.after <- 1
        }
    }
    b.m.Unlock()
    <-b.after
}

参考

• "The Little Book of Semaphores" --- Allen B. Downey