【Go言語】無駄にマルチスレッドな行数カウンタを作ってみる

はじめに

普段から無駄なものを生成する癖が抜けません。
今回は無駄にマルチスレッド化した行数カウンタのコードを記事にしました。

ファイルによってはwc -lよりも高速なので、最速で行数を数えたい人の役には立つかもしれません。

Goは独学で勉強しています。誤りや、もっと良い書き方などご指摘をお待ちしています。

参考書

「Go言語による平行処理」(なんでもっと早く読まなかったのか、、、)
io.Readerをこすれ(Goの隠れた？すばらしさを教えていただきました)

実装する機能

引数で指定したファイルの行数を数えて画面表示します

引数では次の値を指定可能とします
・ファイル名(-f)
・分割読込の分割数(-s)
・並列処理のためのスレッド数(-t)
・読込バッファのサイズ(-b)

特徴

・goroutineとチャンネルを使ったマルチスレッド処理で行数を数えます
・ファイルはio.Readerでこすります(io.Readerをこすれ)
・Seekで読込開始位置をずらし、無駄にファイルの分割読込を行います
・読込バッファを直接検索するのでエコです

処理の流れ

大まかな処理の流れは次の通りです。

1. コマンドライン引数から既出のオプション情報を受け取る
2. 対象ファイルのファイルサイズ（バイト数）を取得する
3. ファイルサイズを分割数で割り、各処理が担当する読込開始位置と読込サイズを決定する
4. goroutineを指定スレッド数起動し、読込開始位置から行数カウントを実施する
5. チャネルを経由して各goroutineのカウント結果を収集し、全ての処理が終了したらコンソールに全体の行数を表示する

リポジトリ

全体のコードはこちら

コードの解説

お膳立て

必要なライブラリをインポートします。
引数の受け取りにはflag、ロギングには謹製のglogを使用します。

package main
    
import (
    "bytes"                            
    "flag"          
    "fmt"
    "io"
    "math"
    "os"
    "sync"
    "time"
            
    "github.com/golang/glog"
)

引数を受け取るための構造体を定義して変数を宣言します。

// Arg is struct for commandline arg.
type Arg struct {          
    // 処理対象のファイル                                                                       
    targetFile string
        
    // 分割カウントする際の分割数
    splitNum int
        
    // 同時実行するスレッド(の最大)数        
    maxThreads int
  
    // ファイル読み込み用Bufferのサイズ                               
    buffersize int
}                                                                     
                                                                         
var (
    arg Arg
)

init関数に引き数展開処理などを書きます。

glogは独自に引き数を取るので、flagを使ってglog独自の引き数を再定義して初期化しています。
コマンドヘルプを表示させると、glogの引き数の説明も一緒に表示されて紛らわしいので、本処理の引き数説明には「(go-lc) 」と入れて区別しています（この辺、もっとスマートな書き方がありそうです）。

func init() {
    // ヘルプメッセージを設定
    flag.Usage = func() {  
        fmt.Fprintf(os.Stderr, "%s\n", fmt.Sprintf("%s -f TARGETFILE [options] [glog options]", 
os.Args[0]))
        flag.PrintDefaults()
    }               
 
    // loggerの初期設定
    _ = flag.Set("stderrthreshold", "INFO")
    _ = flag.Set("v", "0")
        
    // コマンドラインオプションの設定
    flag.StringVar(&arg.targetFile, "f", "", "(go-lc) Target File.")
    flag.IntVar(&arg.splitNum, "s", 2, "(go-lc) Num of File split.")
    flag.IntVar(&arg.maxThreads, "t", 2, "(go-lc) Max Num of Threads.")
    flag.IntVar(&arg.buffersize, "b", 1024*1024, "(go-lc) Size of ReadBuffer(default=1024*1024).")
}

この構成でのコマンドヘルプは次のように表示されます。
glogのヘルプが結構やかましいです。

(やかましいですが、ログレベルの変更処理などを自分で実装しなくて良いのでちょっとしたツール用途では楽です)

> ./bin/go-lc --help
./bin/go-lc -f TARGETFILE [options] [glog options]
  -alsologtostderr
        log to standard error as well as files
  -b int
        (go-lc) Size of ReadBuffer (default 1048576)
  -f string
        (go-lc) Target File
  -log_backtrace_at value
        when logging hits line file:N, emit a stack trace
  -log_dir string
        If non-empty, write log files in this directory
  -logtostderr
        log to standard error instead of files
  -s int
        (go-lc) Num of File split (default 2)
  -stderrthreshold value
        logs at or above this threshold go to stderr
  -t int
        (go-lc) Max Num of Threads (default 2)
  -v value
        log level for V logs
  -vmodule value
        comma-separated list of pattern=N settings for file-filtered logging

メイン処理

メイン処理の説明を先にします。
とは言え、集計を行うgetNumOfLines関数を呼んで結果を受け取り、画面表示しているだけです。

func main() {
    flag.Parse()
            
    glog.V(1).Infof("Start")
            
    // 処理時間算出用のタイマを開始
    startTime := time.Now() 

    // 集計処理を実行
    numOfLines, _ := getNumOfLines(arg.targetFile, arg.splitNum, arg.maxThreads, arg.buffersize)
            
    // 処理時間を表示
    glog.V(1).Infof("End(%s)", time.Since(startTime))
            
    fmt.Printf("%d\n", numOfLines)
}

getFileSize()

ファイルサイズを取得する関数です。
処理はコメントを参照してください。

func getFileSize(filename string) (int, error) {
    // 対象ファイルを開く
    fh, err := os.OpenFile(filename, 0, 0)
    if err != nil {  
        return 0, err
    }
    defer fh.Close()

    // ファイル情報を取得
    fileinfo, err := fh.Stat()
    if err != nil {
        return 0, err
    }

    // ファイルのバイト数を取得して返す
    return int(fileinfo.Size()), nil
}

getNumOfLines()

mainで読んでいたgetNumOfLines()です。
ちょっと長い関数なので、分けて説明します。

冒頭部分ではファイル全体に対して実施されるバッファ読み出し回数を計算します。
次の式です。

読み出し回数　＝　ファイルサイズ　／　バッファサイズ　
（割り切れなかった場合は＋１）

func getNumOfLines(filename string, splitNum int, maxThreads int, buffersize int) (int, error) {
    // ファイルサイズを取得
    fsize, err := getFileSize(filename)
    if err != nil {  
        return 0, err
    }

    // loglevel = 1で情報表示
    glog.V(1).Infof("FileSize   : %10d byte", fsize)
    glog.V(1).Infof("Read buffer: %10d byte", buffersize)
    glog.V(1).Infof("Max Threads: %d", maxThreads)
    glog.V(1).Infof("Split Num  : %d", splitNum)

    // buffersizeの単位で何回読み込みができるかを算出。
    var readCountTotal int = int(math.Trunc(float64(fsize) / float64(buffersize)))

    // 余りがあった場合、読み込み回数に1を加算
    if fsize-(readCountTotal*buffersize) > 0 {
        readCountTotal++
    }

続いて、マルチスレッド化に関するお膳立てをします。
チャネルの使い方を工夫すればwaitgroupは不要かもしれませんが、分かりやすさのため利用します。

    // 終了待機用グループを初期化
    wg := &sync.WaitGroup{}

    // goroutineの同時実行数を制限するためのチャンネル
    jc := make(chan interface{}, maxThreads)
    defer close(jc)

    // 各goroutineの行数カウント結果を受け取るすチャンネル
    counterCh := make(chan int, maxThreads)

    // 各goroutineの終了待ち受けgoroutineから、
    // main処理に集計結果を返すためのチャンネル
    resultCh := make(chan int)
    defer close(resultCh)

    // 結果受信用goroutineを起動
    // 終了条件はclose(counterCh)
    go func(counterCh <-chan int) {
        cAll := 0
        for c := range counterCh {
            cAll += c

            glog.V(2).Infof("[receiver] receive: %d\n", c)
        }
     
        resultCh <- cAll
    }(counterCh)

行数をカウントするgoroutine(countWorker)を起動するループです。
byteOffsetが読み出し開始位置です。
eachReadCountがバッファの読み出し回数ですが、算出方法は次のサイトのものを使用しています。
整数をできるだけ均等になるように n分割
頭の良い人はいるものです。

同時起動数の制御でjcというチャネルを使用しています。定番の使い方ですが次のURLあたりが参考になると思います。
Goroutineの最大数を制御する方法
Go言語でCPU数に応じて並列処理数を制限する
ちなみに上記２サイトはbool型やint型を使用していますが「Go言語による並行処理(O'Reilly)」によると「起動数制御用のチャネルには容量0のinterfaceが良いよ！」とのことです。

    // 行数カウントgoroutineに渡す読み込み開始位置(0は#1のgoroutineのため)        
    var byteOffset int64 = 0

    // 行数カウントgoroutineを起動するためのループ
    for i := 0; i < splitNum; i++ {
        // countLinesInThread内で、何回buffer読み出しを行うか
        eachReadCount := int(math.Trunc(float64(readCountTotal+i) / float64(splitNum)))

        // goroutineの起動数配列を1つ埋める
        jc <- true

        // waitgroupを1つ増やす
        wg.Add(1)

        // 行数カウントgoroutineを起動
        go countWorker(filename, eachReadCount, byteOffset, buffersize, wg, jc, counterCh)

        // 読み込みオフセットを進める
        byteOffset += int64(eachReadCount * buffersize)
    }

    wg.Wait()
    close(counterCh)

    return <-resultCh, nil
}

countWorker()

次にgoroutineの本体であるcountWorker()を定義します。
実際は、filenameで与えられたファイルを開き、f.Seekで読み込み位置をずらしてからgetNumOfCharsOnIoを読んでいるだけです。同じファイルを複数回openする部分が少し冗長にも感じますが分割読込のためのSeek処理があるので、こうしています（このあたり、分割する意味あるのか？、感が漂っていますが気にしないことにします。大人の開発現場では初志貫徹が大切なケースもまま御座います）。

func countWorker(filename string, eachReadCount int, byteOffset int64, buffersize int,
    wg *sync.WaitGroup, jc <-chan interface{}, counter chan<- int) {
    var c int = 0

    defer func() {
        // 無名関数はアウタースコープの変数にアクセスできるため問題ない
        counter <- c
        wg.Done()
        <-jc
    }()

    // loglevel=2で情報表示
    glog.V(2).Infof("[countWorker] start (offset: %d, read size: %d)\n", byteOffset, eachReadCount*buffersize)
    // 対象ファイルを再度開く
    // 元のファイルハンドラを使用するとSeekの読み出しカーソルがおかしくなるため
    f, err := os.OpenFile(filename, 0, 0)
    if err != nil {
        return
    }
    defer f.Close()

    // 指定された読み込み開始位置まで移動
    _, err = f.Seek(byteOffset, 0)
    if err != nil {
        return
    }

    // getNumOfCharsOnIoには次のように作成したbufioを渡すこともできます
    // 今回はio.Readerから読み出すデータのサイズを変更できるのでメリットが少ないと考え不使用
    // br := bufio.NewReaderSize(f, 1024*1024)

    c, err = getNumOfCharsOnIo(f, buffersize, eachReadCount)
    if err != nil {
        panic(err)
    }
}

getNumOfCharsOnIo()

最後に、実際に行数を数えているgetNumOfCharsOnIo()を定義します。
名記事「io.Readerをすこれ」と似たような処理をしていますが、改行を数える部分でbytes.IndexByte()を使用しています。UTF-8の仕組みを考えると大丈夫と思っていますが、もしかするとコメントアウトしてあるIndexRuneを使うなどしたほうが安全かもしれません。幾つかの実ファイルでwcの結果と比べてみても同じだったのですが、もしかすると結果が相違するケースがあるかもしれません。

// io.Readerからbuffersizeづつ読み出し、targetStrの出現数を数える処理をrepeatCount回繰り返します
func getNumOfCharsOnIo(r io.Reader, buffersize int, repeatCount int) (int, error) {
    // 読み込みバッファを初期化
    buf := make([]byte, buffersize)

    // 行数を格納する変数
    var c int = 0

    // 開始位置から、buffersizeづつバイト列を読み込んでbufに代入
    for j := 0; j < repeatCount; j++ {
        n, err := r.Read(buf)
        // 読み込みサイズが0だった場合
        if n == 0 {
            return c, err
        }

        // Readエラー時の処理
        if err != nil {
            return c, err
        }

        // Bufferの中身を走査するためのオフセット
        of := 0

        // Buffer内の改行を数える処理
        for {
            // nでサイズを指定しているのは、bufを使いまわしているから
            // index := bytes.IndexRune(buf[of:n], rune('\n'))
            index := bytes.IndexByte(buf[of:n], '\n')

            // 以降改行が存在しない状態で、ループを抜ける
            if index == -1 {
                break
            }

            // (改行の)カウンタをインクリメント
            c++

            // 発見位置+1までオフセットを進める
            of += index + 1
        }
    }

    return c, nil
}

殆ど、インラインコメントに任せてしまい、ざつな説明でしたがコードについては以上になります。

全体はこちら

利用例

オプションを取るプログラムなので、簡単に使い方を例示します。

ヘルプを表示

> ./bin/go-lc --help
./bin/go-lc -f TARGETFILE [options] [glog options]
  -alsologtostderr
        log to standard error as well as files
  -b int
        (go-lc) Size of ReadBuffer (default 1048576)
  -f string
        (go-lc) Target File
  -log_backtrace_at value
        when logging hits line file:N, emit a stack trace
  -log_dir string
        If non-empty, write log files in this directory
  -logtostderr
        log to standard error instead of files
  -s int
        (go-lc) Num of File split (default 2)
  -stderrthreshold value
        logs at or above this threshold go to stderr
  -t int
        (go-lc) Max Num of Threads (default 2)
  -v value
        log level for V logs
  -vmodule value
        comma-separated list of pattern=N settings for file-filtered logging

とりあえず行数をカウント

> ./bin/go-lc -f /mnt/v01/resource/wikipedia/jawiki/20191001/extract/jawiki-20191001-categorylinks.sql
1428

loglevelを変更して、色々表示しながら行数をカウント

デフォルトで2分割、2スレッドで実行します。

> ./bin/go-lc -f /mnt/v01/resource/wikipedia/jawiki/20191001/extract/jawiki-20191001-categorylinks.sql -v 5
I1216 20:47:23.781456   12091 main.go:233] Start
I1216 20:47:23.781785   12091 main.go:79] FileSize   : 1426087753 byte
I1216 20:47:23.781801   12091 main.go:80] Read buffer:    1048576 byte
I1216 20:47:23.781816   12091 main.go:81] Max Threads: 2
I1216 20:47:23.781826   12091 main.go:82] Split Num  : 2
I1216 20:47:23.781871   12091 main.go:160] [countWorker] start (offset: 713031680, read size: 714080256)
I1216 20:47:23.781953   12091 main.go:160] [countWorker] start (offset: 0, read size: 713031680)
I1216 20:47:23.957093   12091 main.go:115] [receiver] receive: 699
I1216 20:47:23.969989   12091 main.go:115] [receiver] receive: 729
I1216 20:47:23.970048   12091 main.go:242] End(188.280638ms)
1428

約1.4GBのファイルを0.188秒で処理しています。

分割数(-s)とスレッド数(-t)を指定してカウント

> ./bin/go-lc -f /mnt/v01/resource/wikipedia/jawiki/20191001/extract/jawiki-20191001-categorylinks.sql -v 5 -s 4 -t 4
I1216 20:51:51.827208   13285 main.go:233] Start
I1216 20:51:51.827519   13285 main.go:79] FileSize   : 1426087753 byte
I1216 20:51:51.827534   13285 main.go:80] Read buffer:    1048576 byte
I1216 20:51:51.827553   13285 main.go:81] Max Threads: 4
I1216 20:51:51.827565   13285 main.go:82] Split Num  : 4
I1216 20:51:51.827607   13285 main.go:160] [countWorker] start (offset: 1069547520, read size: 357564416)
I1216 20:51:51.827706   13285 main.go:160] [countWorker] start (offset: 713031680, read size: 356515840)
I1216 20:51:51.827646   13285 main.go:160] [countWorker] start (offset: 356515840, read size: 356515840)
I1216 20:51:51.827642   13285 main.go:160] [countWorker] start (offset: 0, read size: 356515840)
I1216 20:51:51.938578   13285 main.go:115] [receiver] receive: 343
I1216 20:51:51.939430   13285 main.go:115] [receiver] receive: 356
I1216 20:51:51.952839   13285 main.go:115] [receiver] receive: 386
I1216 20:51:51.956868   13285 main.go:115] [receiver] receive: 343
I1216 20:51:51.956899   13285 main.go:242] End(129.400448ms)
1428

先ほどのファイルで、0.129秒まで速くなりました。

読み込みバッファサイズも変えてみる

> ./bin/go-lc -f /mnt/v01/resource/wikipedia/jawiki/20191001/extract/jawiki-20191001-categorylinks.sql -v 5 -s 4 -t 4 -
b 1024
I1216 20:53:02.522702   13459 main.go:233] Start
I1216 20:53:02.523194   13459 main.go:79] FileSize   : 1426087753 byte
I1216 20:53:02.523217   13459 main.go:80] Read buffer:       1024 byte
I1216 20:53:02.523222   13459 main.go:81] Max Threads: 4
I1216 20:53:02.523229   13459 main.go:82] Split Num  : 4
I1216 20:53:02.523275   13459 main.go:160] [countWorker] start (offset: 1069565952, read size: 356521984)
I1216 20:53:02.523351   13459 main.go:160] [countWorker] start (offset: 0, read size: 356521984)
I1216 20:53:02.523442   13459 main.go:160] [countWorker] start (offset: 356521984, read size: 356521984)
I1216 20:53:02.526218   13459 main.go:160] [countWorker] start (offset: 713043968, read size: 356521984)
I1216 20:53:03.146721   13459 main.go:115] [receiver] receive: 343
I1216 20:53:03.149466   13459 main.go:115] [receiver] receive: 386
I1216 20:53:03.186216   13459 main.go:115] [receiver] receive: 356
I1216 20:53:03.190404   13459 main.go:115] [receiver] receive: 343
I1216 20:53:03.190443   13459 main.go:242] End(667.278999ms)
1428

バッファを小さくし、非効率になったため0.667秒と遅くなりました。

結果

さて、こんなに無駄な処理を書いて、どれだけ早くなるか結果が気になる方もおられると思います。
利用例を見ると4分割4スレッドのバッファサイズ1048576(1024*1024)の結果が良いですね。

同じファイルを対象にした、wc -l コマンドの結果は

time wc -l /mnt/v01/resource/wikipedia/jawiki/20191001/extract/jawiki-20191001-categorylinks.sql

(1回目)
1428 /mnt/v01/resource/wikipedia/jawiki/20191001/extract/jawiki-20191001-categorylinks.sql
0.04user 0.26system 0:00.30elapsed 100%CPU (0avgtext+0avgdata 2104maxresident)k
0inputs+0outputs (0major+76minor)pagefaults 0swaps

(2回目)
time wc -l /mnt/v01/resource/wikipedia/jawiki/20191001/extract/jawiki-20191001-categorylinks.sql
1428 /mnt/v01/resource/wikipedia/jawiki/20191001/extract/jawiki-20191001-categorylinks.sql
0.03user 0.22system 0:00.26elapsed 99%CPU (0avgtext+0avgdata 2068maxresident)k
0inputs+0outputs (0major+75minor)pagefaults 0swaps

(3回目)
1428 /mnt/v01/resource/wikipedia/jawiki/20191001/extract/jawiki-20191001-categorylinks.sql
0.03user 0.22system 0:00.26elapsed 99%CPU (0avgtext+0avgdata 2124maxresident)k
0inputs+0outputs (0major+75minor)pagefaults 0swaps

(4回目)
1428 /mnt/v01/resource/wikipedia/jawiki/20191001/extract/jawiki-20191001-categorylinks.sql
0.04user 0.20system 0:00.25elapsed 99%CPU (0avgtext+0avgdata 2104maxresident)k
0inputs+0outputs (0major+78minor)pagefaults 0swaps

(5回目)
1428 /mnt/v01/resource/wikipedia/jawiki/20191001/extract/jawiki-20191001-categorylinks.sql
0.04user 0.23system 0:00.27elapsed 100%CPU (0avgtext+0avgdata 2068maxresident)k
0inputs+0outputs (0major+75minor)pagefaults 0swaps

平均: 0.268s

５回の平均で0.268秒でした。

マシン環境は全てGoogle Cloud PlatformのCompute Engine(n1-standard-4（vCPU x 4、メモリ 15 GB）)で、対象ファイルは1.4GBのWikidumpのsqlファイルです。

4分割で0.129秒だったので、wc -lコマンドよりも良い結果です。
無駄を重ねた甲斐がありました。

ご注意

実のところ、上記の結果はGCEのファイルキャッシュが効いている恐れが強いです。20GBの別のファイルで試すと次のように3分程度かかりました。このサイズでもwcより早いですが、1.4GBを0.3秒で処理しているのでもっと早くても良さそうです。なので数GB程度のファイルはキャッシュされているのかもしれません。つまりキャッシュされる前と後では、効率云々以上の差が出る恐れがあります。それにGCEはディスクがネットワーク越しにマウントされているため、速度がネットワークの状況にも左右されます。上記結果はあくまでも参考程度に見てください。

約20GBのファイルを対象にwc -lと本プログラムを実行した結果を列挙します。

> time wc -l /mnt/v01/resource/wikipedia/enwiki/20191001/extract/enwiki-20191001-categorylinks.sql

(wc 1回目)
19525 /mnt/v01/resource/wikipedia/enwiki/20191001/extract/enwiki-20191001-categorylinks.sql
0.91user 8.21system 3:58.40elapsed 3%CPU (0avgtext+0avgdata 2116maxresident)k
31893064inputs+0outputs (0major+75minor)pagefaults 0swaps

(wc 2回目)
19525 /mnt/v01/resource/wikipedia/enwiki/20191001/extract/enwiki-20191001-categorylinks.sql
0.86user 7.86system 3:09.32elapsed 4%CPU (0avgtext+0avgdata 2044maxresident)k
26220800inputs+0outputs (0major+77minor)pagefaults 0swaps

(go-lc 1回目)
bin/go-lc -f /mnt/v01/resource/wikipedia/enwiki/20191001/extract/enwiki-20191001-categorylinks.sql -v 5 -t 4 -s 4
I1216 21:19:05.381636   14202 main.go:233] Start
I1216 21:19:05.384584   14202 main.go:79] FileSize   : 20234931295 byte
I1216 21:19:05.384601   14202 main.go:80] Read buffer:    1048576 byte
I1216 21:19:05.384605   14202 main.go:81] Max Threads: 4
I1216 21:19:05.384609   14202 main.go:82] Split Num  : 4
I1216 21:19:05.384704   14202 main.go:160] [countWorker] start (offset: 15176040448, read size: 5059379200)
I1216 21:19:05.384733   14202 main.go:160] [countWorker] start (offset: 0, read size: 5058330624)
I1216 21:19:05.384786   14202 main.go:160] [countWorker] start (offset: 5058330624, read size: 5058330624)
I1216 21:19:05.384859   14202 main.go:160] [countWorker] start (offset: 10116661248, read size: 5059379200)
I1216 21:19:06.836037   14202 main.go:115] [receiver] receive: 4881
I1216 21:20:49.994339   14202 main.go:115] [receiver] receive: 4866
I1216 21:20:56.968630   14202 main.go:115] [receiver] receive: 4910
I1216 21:21:00.825423   14202 main.go:115] [receiver] receive: 4868
I1216 21:21:00.825466   14202 main.go:242] End(1m55.440902834s)
19525

(go-lc 2回目)
bin/go-lc -f /mnt/v01/resource/wikipedia/enwiki/20191001/extract/enwiki-20191001-categorylinks.sql -v 5 -t 4 -s 4
I1216 21:21:19.065087   14343 main.go:233] Start
I1216 21:21:19.066146   14343 main.go:79] FileSize   : 20234931295 byte
I1216 21:21:19.066164   14343 main.go:80] Read buffer:    1048576 byte
I1216 21:21:19.066169   14343 main.go:81] Max Threads: 4
I1216 21:21:19.066182   14343 main.go:82] Split Num  : 4
I1216 21:21:19.066232   14343 main.go:160] [countWorker] start (offset: 15176040448, read size: 5059379200)
I1216 21:21:19.066234   14343 main.go:160] [countWorker] start (offset: 0, read size: 5058330624)
I1216 21:21:19.066314   14343 main.go:160] [countWorker] start (offset: 5058330624, read size: 5058330624)
I1216 21:21:19.066377   14343 main.go:160] [countWorker] start (offset: 10116661248, read size: 5059379200)
I1216 21:21:20.477393   14343 main.go:115] [receiver] receive: 4881
I1216 21:23:04.790516   14343 main.go:115] [receiver] receive: 4910
I1216 21:23:35.783612   14343 main.go:115] [receiver] receive: 4868
I1216 21:23:53.859878   14343 main.go:115] [receiver] receive: 4866
I1216 21:23:53.859920   14343 main.go:242] End(2m34.793812658s)
19525

さいごに

今回の処理はWikidumpのような大きなテキストファイルを、全体をロードすることなく高速に処理するテストも兼ねて作成しました。

読み込みハンドラを別々に取得しているので、試しにGCEに4台のディスクをマウントして同じファイルをコピーし、ディスク分散で読んでみましたが、NASとの通信速度に制限があるのか期待した効果は得られませんでした（１台で分割読み込みした場合とほとんど同じ速度でした）。GCPのストレージサービスであるGCSはバイト単位の読み出しに対応しているので、あるいはGCSを対象に分割読み込みをかけると早いかもしれません。このあたりはio.Readerの恩恵というか、ファイルからの読み込みとGCSのAPI経由とでコードを使いまわせるGoのメリットがありそうです。

車輪を複雑にして再開発しているような記事に、最後までお付き合いいただきありがとうございました。