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

昨日は、 @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

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 さんです。ぜひお楽しみに！