Upgrade to Pro — share decks privately, control downloads, hide ads and more …

Effective streaming in Golang

avvmoto
June 01, 2018

Effective streaming in Golang

Golang で Stream 処理を書く上で重要になる io についてです。
- io interface の実装とすることの重要性
- io interface の紹介
- メモリ使用量を減らすためのちょっとしたテクニック紹介
- io 関連 package の紹介
についてお話します。

avvmoto

June 01, 2018
Tweet

More Decks by avvmoto

Other Decks in Programming

Transcript

  1. io とは データの入出力の流れは Stream として抽象化されていて、 Golang では package io に

    て共通化されています。 例えば os.File, bytes.Buffer など、ファイルや、メモリ上のバッファへの書き込みを io.Write はラップしています。 Golang 開発者は io だけを知っておけば、各種異なるライブラリーごとにインターフェー スを学習する必要がなくなっています。
  2. 適切に package io を利用しよう io は io のまま利用することで、メモリ使用量を減らすことができます。 例えば、 io.Reader

    を ioutil.ReadAll で []byte に都度展開するのは微妙です。 メモリのアロケートはコスト自体が高いので、避けるべきです。 また、コストの高いGC の頻度を減らすことができます。 まずはじめに二件、io を適切に利用することで、メモリ使用量を減らしている例を見て みましょう。
  3. 例1 JSON を struct へ Decoding する JSON を一度メモリ上へ展開する必要はありません。 io.Reader

    はそのままストリームとして使いましょう。 NG ( 一度メモリ上に JSON を展開している ) func BenchmarkJSON_ReadAll(b *testing.B) { b.ResetTimer() for i := 0; i < b.N; i++ { data, err := ioutil.ReadAll(r1) // ← 一度メモリ上に展開している if err != nil { b.Fatal(err) } err = json.Unmarshal(data, &u1) if err != nil { b.Fatal(err) } r1.Seek(0, io.SeekStart) } } Run
  4. 例1 OK ( io.Reader を受け付ける json.NewDecoder を利用 ) Benchmark $

    go test -bench . -benchmem goos: darwin goarch: amd64 BenchmarkJSON_Decoder-8 1 1026901580 ns/op 75107424 B/op 2000022 allocs/op BenchmarkJSON_ReadAll-8 1 1064282789 ns/op 150794144 B/op 2000085 allocs/op PASS ok func BenchmarkJSON_Decoder(b *testing.B) { b.ResetTimer() for i := 0; i < b.N; i++ { err := json.NewDecoder(r2).Decode(&u2) // ← allocate が無くなった! if err != nil { b.Fatal(err) } r2.Seek(0, io.SeekStart) } } Run
  5. 中間まとめ: package io を適切に使おう このように、io は io のまま扱うことで、メモリのアロケーションを減らし、効率のよ い実装ができます。 io

    を io のまま扱うことができるように、 package io やその他標準ライブラリで、いく つか便利な方法が提供されています。 便利なので把握しておきましょう。
  6. 例2 大きなサイズの []byte を返す関数を書く時は、 io.WriterTo の実装を検討しましょ う。 NG ( User

    という struct の JSON encoding を []byte として返す ) func (u *User) Bytes() []byte { b := new(bytes.Buffer) // ← 避けられるメモリ確保!! json.NewEncoder(b).Encode(u) return b.Bytes() } この例では、返り値の []byte をメモリ上に確保してから、値を返却しています。
  7. 例2 大きなサイズの []byte を返す関数を書く時は、 io.WriterTo の実装を検討しましょ う。 OK ( User

    という struct の JSON encoding を返す ) var _ io.WriterTo = &User{} func (u *User) WriteTo(w io.Writer) (int64, error) { err := json.NewEncoder(w).Encode(u) // ← メモリ確保していない return u.Size(), err } 返り値となる []byte を書き込む io.Writer を引数として受け取り、そこに結果を書き 込んでいます。 前の例であったメモリ確保はなくなっています。
  8. 中間まとめ: package io を精読しよう 自分の書いたコードの利用者は、データの入出力に関して package io を wrap している

    ことをきっと期待していることでしょう。 例えば、先程の 構造体 User は、io.WriterTo をラップしているため、 io.WriterTo を引数とする他の標準ライブラリーにそのまま渡せます。 そして不要なバッファの確保を避けることができます。 このため、 package io にどのような interface があるか把握しておきましょう。
  9. io.Reader, io.Writer 広く用いられている interface type Reader interface { Read(p []byte)

    (n int, err error) } type Writer interface { Write(p []byte) (n int, err error) } これらのバッファ機能を提供する package bufio がある
  10. io.ReaderAt, io.WriterAt ランダムリード・ランダムライトしたい時用 type ReaderAt interface { ReadAt(p []byte, off

    int64) (n int, err error) } type WriterAt interface { WriteAt(p []byte, off int64) (n int, err error) } bu o で ReaderAt のバッファ機能は提供されてないので、拙作 avvmoto/buf-readerat (https://github.com/avvmoto/buf-readerat) をどうぞ
  11. io.WriteString func WriteString(w Writer, s string) (n int, err error)

    io.Writer に string を書き込んでくれる関数 自作 struct であっても、WriteString メソッドを生やしておこう。そうすると、 io.WriteString を使う時、 string から []byte への変換が走らず、効率がよい // WriteString writes the contents of the string s to w, which accepts a slice of bytes. // If w implements a WriteString method, it is invoked directly. // Otherwise, w.Write is called exactly once. func WriteString(w Writer, s string) (n int, err error) { if sw, ok := w.(stringWriter); ok { return sw.WriteString(s) } return w.Write([]byte(s)) }
  12. io.Pipe io.Pipe を利用しない例 一度書き込んだ内容をバッファしている func ExamplePipeNG() { print := func(r

    io.Reader) { buf := new(bytes.Buffer) buf.ReadFrom(r) fmt.Print(buf.String()) } b := new(bytes.Buffer) // ← 不要なアロケート!! fmt.Fprint(b, "some text to be read\n") print(b) // Output: // some text to be read } Run
  13. io.Pipe io.Pipe を利用した例 内部バッファがなくなった golang.org/src/io/example_test.go (https://golang.org/src/io/example_test.go) より引用 func ExamplePipe() {

    r, w := io.Pipe() go func() { fmt.Fprint(w, "some text to be read\n") w.Close() }() buf := new(bytes.Buffer) buf.ReadFrom(r) fmt.Print(buf.String()) // Output: // some text to be read } Run
  14. io.TeeReader func TeeReader(r Reader, w Writer) Reader io.TeeReader は r

    から読み取った内容を、指定した w に書き込む io.Reader を返す io.Reader を引数として渡している時、渡っているデータをプリントデバッグするときに 便利 似た目的で、 testing/iotest に NewReadLogger (https://golang.org/pkg/testing/iotest/#NewReadLogger) というのも ある( 後述) func ExampleTeeReader() { var r io.Reader = strings.NewReader(`{"id":100, "name":"imoty"}`) r = io.TeeReader(r, os.Stderr) // Print デバッグするときだけ、この行を挿入 var u User json.NewDecoder(r).Decode(&u) fmt.Printf("%+v", u) // Output: // {ID:100 Name:imoty} } Run
  15. io.Copy func Copy(dst Writer, src Reader) (written int64, err error)

    io.Reader から io.EOF が出るまで読み取り、io.Writer に書き込む 内部き 32 k byte のバッファを持つ。指定したバッファを使いたい時は、 io.CopyBuffer を使う。 n byte だけコピーする io.CopyN もある golang.org/src/io/example_test.go (https://golang.org/src/io/example_test.go) より引用 func ExampleCopy() { r := strings.NewReader("some io.Reader stream to be read\n") if _, err := io.Copy(os.Stdout, r); err != nil { log.Fatal(err) } // Output: // some io.Reader stream to be read } Run
  16. io.Copy 利用例 JSON encoding を http.ResponseWriter へ書き込む golang.org/src/io/example_test.go (https://golang.org/src/io/example_test.go) より引用

    func ExampleCopy() { r := strings.NewReader("some io.Reader stream to be read\n") if _, err := io.Copy(os.Stdout, r); err != nil { log.Fatal(err) } // Output: // some io.Reader stream to be read } Run
  17. io.WriterTo ( 再掲) type WriterTo interface { WriteTo(w Writer) (n

    int64, err error) } WriteTo は w へ、書き込むべきデータがなくなるかエラーが発生するまで、データを書 き込んでくれる io.Copy は WriterTo が実装されていれば、Read ではなく WriterTo を利用してくれる WriteTo を実装しておくと、書き込むときのバッファのサイズ等、書き込む処理を明示 的に自分で書くことができるので、 io.Copy の効率が良くなる可能性がある
  18. io.ReaderFrom golang.org/pkg/io/#ReaderFrom (https://golang.org/pkg/io/#ReaderFrom) type ReaderFrom interface { ReadFrom(r Reader) (n

    int64, err error) } ReadFrom は、 EOF またはエラーが返るまでデータを r から読み取る io.Copy は ReadFrom が実装されていれば、これを利用してくれる そのため自作 Struct であっても、実装しておくと io.Copy で効率がよくなる可能性があ る (WriterTo と両方実装されていれば、そちらを優先して利用)
  19. io.ReadFull golang.org/pkg/io/#ReadFull (https://golang.org/pkg/io/#ReadFull) func ReadFull(r Reader, buf []byte) (n int,

    err error) { return ReadAtLeast(r, buf, len(buf)) } ReadFull はちょうど len(buf) bytes だけ r から buf へデータを読み取る あらかじめ読み込むサイズが分かっていて( ヘッダとかで) 、書き込み先のバッファを持っ ているという限定的な場合に利用すると、メモリアロケートをへらすことができる encoding/binary , encoding/gob 等で利用例あり
  20. io.ReadFull NG func ExampleReadFullNG() { src := strings.NewReader("beefcafedeadbeef") result :=

    make([]byte, 16) // 書き込み領域が、何らかの理由で既に確保されている result, _ = ioutil.ReadAll(src) // ← ReadAll が []byte を別途 allocate している fmt.Printf("%s", result) // Output: // beefcafedeadbeef } Run
  21. io.ReadFull OK func ExampleReadFullOK() { src := strings.NewReader("beefcafedeadbeef") result :=

    make([]byte, 16) // 書き込み領域が、何らかの理由で既に確保されている io.ReadFull(src, result) fmt.Printf("%s", result) // Output: // beefcafedeadbeef } Run
  22. bytes.Bu er golang.org/pkg/bytes/#Bu er (https://golang.org/pkg/bytes/#Bu er) みんなご存知 Package io が

    一通り実装されている。Write, WriteTo, WriteString, Read, ReadFrom と か。 ReadAt, WriteAt はない。 初期化でバッファを指定するなら NewBuffer 初期化のバッファとしてstring 使うなら NewBufferString 内部バッファー以上に書き込んだら、書き込んだ分だけ都度 allcoate するので注意
  23. strings.Reader type Reader func NewReader(s string) *Reader func (r *Reader)

    Len() int func (r *Reader) Read(b []byte) (n int, err error) func (r *Reader) ReadAt(b []byte, off int64) (n int, err error) func (r *Reader) ReadByte() (byte, error) func (r *Reader) ReadRune() (ch rune, size int, err error) func (r *Reader) Reset(s string) func (r *Reader) Seek(offset int64, whence int) (int64, error) func (r *Reader) Size() int64 func (r *Reader) UnreadByte() error func (r *Reader) UnreadRune() error func (r *Reader) WriteTo(w io.Writer) (n int64, err error) string を読みとって、Read 系の io を提供
  24. strings.Builder golang.org/pkg/strings/#Builder (https://golang.org/pkg/strings/#Builder) type Builder func (b *Builder) Grow(n int)

    func (b *Builder) Len() int func (b *Builder) Reset() func (b *Builder) String() string func (b *Builder) Write(p []byte) (int, error) func (b *Builder) WriteByte(c byte) error func (b *Builder) WriteRune(r rune) (int, error) func (b *Builder) WriteString(s string) (int, error) go 1.9 で追加 Write メソッドを使って string を効率的に組み立てることができる bytes.Buffer 使ってもよいが、用途が buf.Write() して buf.String() するだけの場 合、特にメモリコピーを減らしてくれる base64 encoding する場合がまさにこのケース
  25. strings.Builder func ExampleStringBuilder() { input := []byte("foo\x00bar") w := &strings.Builder{}

    encoder := base64.NewEncoder(base64.StdEncoding, w) // ← Write() を利用 encoder.Write(input) encoder.Write(input) encoder.Close() fmt.Print(w.String()) // ← string() を利用 // Output: // Zm9vAGJhcmZvbwBiYXI= } Run
  26. ioutil.NopCloser func NopCloser(r io.Reader) io.ReadCloser io.Reader を io.ReadCloser にしてくれる Endpoint

    Test でダミーリクエスト作る時などに便利 var req *http.Request req.Body = ioutil.NopCloser(strings.NewReader("dummy request"))
  27. ioutil その他 func ReadDir(dirname string) ([]os.FileInfo, error) func ReadFile(filename string)

    ([]byte, error) func TempDir(dir, prefix string) (name string, err error) func TempFile(dir, prefix string) (f *os.File, err error) func WriteFile(filename string, data []byte, perm os.FileMode) error ファイルの読み書き、temp le など
  28. package hash/crc32 golang.org/pkg/hash/crc32/ (https://golang.org/pkg/hash/crc32/) func Checksum(data []byte, tab *Table) uint32

    func ChecksumIEEE(data []byte) uint32 func New(tab *Table) hash.Hash32 func NewIEEE() hash.Hash32 func Update(crc uint32, tab *Table, p []byte) uint32 type Table func MakeTable(poly uint32) *Table Package hash/crc32 は一見、 []byte しか受付なさそうだが・・・ hash.Hash32 が interface で、 io.Writer を Wrap している! type Hash interface { io.Writer // ... 略 ... }
  29. package hash io.Writer を wrap していることで、データソースが io.Reader の場合そのまま渡せる この例では io.Copy

    を利用 CRC32 を計算したいバイト列を、一度に全部メモリ上に展開することなく、計算可能 hash/crc32 だけでなく、 crypt/md5 など他のハッシュの実装も同様 func ExampleHash() { src := strings.NewReader("foo\x00bar") crc := crc32.NewIEEE() io.Copy(crc, src) // ← crc wraps io.Writer fmt.Printf("%v", crc.Sum(nil)) // Output: // [30 116 174 10] } Run
  30. NewReadLogger, NewWriteLogger func NewReadLogger(prefix string, r io.Reader) io.Reader func NewWriteLogger(prefix

    string, w io.Writer) io.Writer Read のたびに、 standard error に Read の内容を stderr に表示してくれる 内部的には、 log.Printf が利用されている NewReadLogger は、 io.TeeReader でログを出すのと同様のユースケースで利用でき る。
  31. iotest.NewReadLogger 例 io.Reader の内容を printf debug するときに便利 ただしフォーマットは %x (base16)

    で Printf されるので、厳しい。 2018/03/30 16:44:30 myprefix 7b226964223a3130302c20226e616d65223a22696d6f7479227d io.Reader が error か返す場合、error の内容も表示してくれるので、その点は便利 func ExampleNewReadLogger() { var r io.Reader = strings.NewReader(`{"id":100, "name":"imoty"}`) // r = iotest.NewReadLogger("myprefix", r) // Print デバッグするときだけ、この行を挿入 var u User json.NewDecoder(r).Decode(&u) fmt.Printf("%+v", u) // Output: // {ID:100 Name:imoty} } Run
  32. package testing/iotest その他 func DataErrReader(r io.Reader) io.Reader func HalfReader(r io.Reader)

    io.Reader func OneByteReader(r io.Reader) io.Reader func TimeoutReader(r io.Reader) io.Reader func TruncateWriter(w io.Writer, n int64) io.Writer タイムアウト等、意図的にエラーを返す Reader(Writer) へラップしてくれる ネットワーク越しのIO のエラーケースのテストに使えそう
  33. package golang.org/x/text/transform godoc.org/golang.org/x/text/transform (https://godoc.org/golang.org/x/text/transform) 主にバイト列の変換を行う機能を提供 type Transformer interface { Transform(dst,

    src []byte, atEOF bool) (nDst, nSrc int, err error) Reset() } func Chain(t ...Transformer) Transformer func RemoveFunc(f func(r rune) bool) Transformer 実装例には golang.org/x/text/encoding/japanese.EUCJP, golang.org/x/text/unicode/norm.Form など Transformer を Chain できるので、複数の Transform を続けて変換することができる
  34. transform 利用方法 定義済みの Transformer を用いて、ストリームのままバイトの変換処理が書ける type Writer func NewWriter(w io.Writer,

    t Transformer) *Writer func (w *Writer) Close() error func (w *Writer) Write(data []byte) (n int, err error) 汎用的なバイト列変換処理を書きたくなったら思い出そう
  35. まとめ package io で定義される interface や便利機能を把握し、効率よくストリーミング処理の 実装・設計をしましょう。 []byte ではなく、 io

    の interface を引き回すことで、メモリ使用量が減る可能性があり ます。 Bytes() などのメソッドが生えていたら、アンチパターンかな? と疑ってみましょう。
  36. 参考資料 io - The Go Programming Language (https://golang.org/pkg/io/) Golang でのstream

    の扱い方を学ぶ (http://christina04.hatenablog.com/entry/2017/01/06/190000) Streaming IO in Go (https://medium.com/learning-the-go-programming-language/streaming-io-in-go-d93507931185) メルカリ カウルのマスタデータの更新 (https://www.slideshare.net/takuyaueda967/ss-82030886)