Go言語による並行処理の理解を深めるために、ライブラリcheggaaa/pbを読んでみた #Go

昨日は、 @srttk さんの「net/http でサーバーを立て、いくつかのパターンをパースしてみる」でした。
標準パッケージで、ライブラリが実装していること（URLに含まれるパラメータのパース）の実現のための方法が分かりました！「標準パッケージでどうやるの」と考え模索することでGo言語におけるノウハウが得られますよね！自分も定期的に、そのような時間を作りたいとおもいました。

これは何？

cheggaaa/pb(progress barのライブラリ)のコードを読んでみたメモです。
並行処理についてのノウハウをまとめました。

モチベーション

Go言語による並行処理を読んで、並行処理についての理解を深めたかったからです。
なにかサクッと作りたいな〜と思い、ヌルヌル動くプログレスバーを作ろうと思いつくも、すでに良いライブラリ（cheggaaa/pb）があることを知り、これを読んで理解を深めたいと思いました。

pbのコードを読んでみた

使い方（READMEより）

下記のとおり、bar := pb.StartNew(するとプログレスバーの表示が始まり、処理が進むごとにbar.Increment()するだけ！cliツールを作成する際に、かなり簡単に利用できそうです。
(公式READMEより)

package main

import (
    "gopkg.in/cheggaaa/pb.v1"
    "time"
)

func main() {
    count := 100000
    bar := pb.StartNew(count)
    for i := 0; i < count; i++ {
        bar.Increment()
        time.Sleep(time.Millisecond)
    }
    bar.FinishPrint("The End!")
}

上記のサンプルは、下記のように動きます。

このライブラリは、主に下記の実装がされているようでした。

プログレスバー初期化
プログレスバー表示開始
処理の進捗をプログレスバーに反映
プログレスバーの再描画

プログレスバー初期化

このライブラリの主な情報は、構造体ProgressBarに保持されます。
たとえば、進捗度合いを管理するための情報として、処理全体の値Total　と、現時点の進捗値current をもっています。

// at https://github.com/cheggaaa/pb/blob/751f9183c336d1bb8ef77beb95243f367d1d90e1/pb.go#L68

type ProgressBar struct {
current  int64 // current must be first member of struct (https://code.google.com/p/go/issues/detail?id=5278)
//〜省略〜
Total    int64
//〜省略〜
}

この構造体の初期化処理で、Totalを設定できるようになっています。

// at https://github.com/cheggaaa/pb/blob/751f9183c336d1bb8ef77beb95243f367d1d90e1/pb.go#L38

// Create new progress bar object using int64 as total
func New64(total int64) *ProgressBar {
    pb := &ProgressBar{
        Total:           total,
//〜省略〜
    }
    return pb.Format(FORMAT)
}

プログレスバー表示開始

bar := pb.StartNew(では、Newしたあと、Start()を実行するようです

// at https://github.com/cheggaaa/pb/blob/751f9183c336d1bb8ef77beb95243f367d1d90e1/pb.go#L56

// Create new object and start
func StartNew(total int) *ProgressBar {
    return New(total).Start()
}

Start()はざっくり、
1.pb.Update()で初回のプログレスバーを描画して、
2.go pb.refresher()という再描画タスクを実行する処理のようです。

// at https://github.com/cheggaaa/pb/blob/751f9183c336d1bb8ef77beb95243f367d1d90e1/pb.go#L114

// Start print
func (pb *ProgressBar) Start() *ProgressBar {
    pb.startTime = time.Now()
    pb.startValue = atomic.LoadInt64(&pb.current)
    if atomic.LoadInt64(&pb.Total) == 0 {
        pb.ShowTimeLeft = false
        pb.ShowPercent = false
        pb.AutoStat = false
    }
    if !pb.ManualUpdate {
        pb.Update() // Initial printing of the bar before running the bar refresher.
        go pb.refresher()
    }
    return pb
}

処理の進捗をプログレスバーに反映

このライブラリ利用者は、bar.Increment()とすれば、プログレスバーに進捗状況を反映できました。これは下記のように、&pb.currentのフィールドに1を加算している処理のようです。

// at https://github.com/cheggaaa/pb/blob/751f9183c336d1bb8ef77beb95243f367d1d90e1/pb.go#L136
// Increment current value
func (pb *ProgressBar) Increment() int {
    return pb.Add(1)
}

// at https://github.com/cheggaaa/pb/blob/751f9183c336d1bb8ef77beb95243f367d1d90e1/pb.go#L151
// Add to current value
func (pb *ProgressBar) Add(add int) int {
    return int(pb.Add64(int64(add)))
}

func (pb *ProgressBar) Add64(add int64) int64 {
    return atomic.AddInt64(&pb.current, add)
}

プログレスバーの再描画

Start()にて、go pb.refresher()が実行されていました。ここでプログレスバーが再描画されるようです。
pb.refresher()は、pb.RefreshRateごとに、Update()を実行する関数のようです。

// at https://github.com/cheggaaa/pb/blob/751f9183c336d1bb8ef77beb95243f367d1d90e1/pb.go#L494

// Internal loop for refreshing the progressbar
func (pb *ProgressBar) refresher() {
    for {
        select {
        case <-pb.finish:
            return
        case <-time.After(pb.RefreshRate):
            pb.Update()
        }
    }
}

Update()は、Write the current state of the progressbarのコメントどおり、プログレスバーを描画する関数のようです。実際に描画する処理は、pb.write(に集約されていそうです。

// at https://github.com/cheggaaa/pb/blob/751f9183c336d1bb8ef77beb95243f367d1d90e1/pb.go#L442:24

// Write the current state of the progressbar
func (pb *ProgressBar) Update() {
    c := atomic.LoadInt64(&pb.current)
    p := atomic.LoadInt64(&pb.previous)
    t := atomic.LoadInt64(&pb.Total)
    if p != c {
        pb.mu.Lock()
        pb.changeTime = time.Now()
        pb.mu.Unlock()
        atomic.StoreInt64(&pb.previous, c)
    }
    pb.write(t, c)
    if pb.AutoStat {
        if c == 0 {
            pb.startTime = time.Now()
            pb.startValue = 0
        } else if c >= t && pb.isFinish != true {
            pb.Finish()
        }
    }
}

pb.write()は長いですが、
* 1.変数outにプログレスバーを設定し、
* 2.outに設定された文字列を出力　しているようです

// at https://github.com/cheggaaa/pb/blob/751f9183c336d1bb8ef77beb95243f367d1d90e1/pb.go#L279
func (pb *ProgressBar) write(total, current int64) {
//〜省略〜
            totalS := Format(total).To(pb.Units).Width(pb.UnitsWidth)
            countersBox = fmt.Sprintf(" %s / %s ", current, totalS)
// 1.1変数`out`にプログレスバーを設定
// at https://github.com/cheggaaa/pb/blob/751f9183c336d1bb8ef77beb95243f367d1d90e1/pb.go#L400
    out = pb.prefix + timeSpentBox + countersBox + barBox + percentBox + speedBox + timeLeftBox + pb.postfix

// 2.outに設定された文字列を出力
// at https://github.com/cheggaaa/pb/blob/751f9183c336d1bb8ef77beb95243f367d1d90e1/pb.go#L418
    switch {
//〜省略〜
    case pb.Callback != nil:
        pb.Callback(out + end)
    case !pb.NotPrint:
        fmt.Print("\r" + out + end)
    }

並行処理に関するノウハウ

上記のように、このライブラリの基本動作が分かったので、並行処理に関する学びをまとめます。

基本コンセプト

再描画のためのgoroutineを走らせ、Durationごとに再描画させている」という方針だということがわかりました。
ツールを作る際に、この辺の設計コンセプトをつくるのに経験が必要だと思いました。

sync.Mutex

sync.Mutex
mu.Lock()できるのは、１箇所のみ。他の箇所でLockされているときにLockがcallされたら、ほか箇所のLockがmu.Unlock()されるまでブロックされる。

sync/atomic

Package atomic provides low-level atomic memory primitives useful for implementing synchronization algorithms.

これは、goroutine内部における、主に数値系の競合を避けるためのパッケージのようです。
sync/atomicについて、golang.tokyo#14で発表されていた morikuni さんの資料ホリネズミでもわかるGoroutine入門でも言及されているようですので、こちらを読むと理解が深まりました。

まとめ

コードを書く時間も学習のためには必要だけど、良いコードを読む時間も大切！
morikuni さんの資料ホリネズミでもわかるGoroutine入門がめっちゃわかりやすかった
他に並行処理に関して良い学習方法があれば、教えてください！

明日は、@Baki33 さんです。ぜひお楽しみに！