长时间运行的副本到bytes.Buffer

Question

I have a long lived io.Reader which returns some data every few seconds (never EOF), and a goroutine which does an io.Copy from that reader to a bytes.Buffer (also never terminates). Something like so:

var src io.Reader
var buf bytes.Buffer

func main() {
    go io.Copy(&buf, src)
    // Do stuff. Read from the buffer periodically.
}

What I don't understand is that I see strange results when I try to read from that buffer. It doesn't matter whether I call buf.Bytes() or ioutil.ReadAll(&buf) or anything, I just see the first bytes written to the buffer over and over again.

https://play.golang.org/p/yn0JPrvohV

My question is, what am I doing wrong? Can I use bytes.Buffer in this way (io.Copy to it and read periodically)?

Answer 1

You can't synchronize your Read calls with the writes that are happening on the bytes.Buffer in the io.Copy. Even if you wrap the bytes.Buffer in a struct to lock the Read/Write methods, you are going to deadlock when the Copy is waiting on a Write while the ReadAll is blocked on the Read. You either need to do the copy manually, and properly serialize all access, or separate the reads and writes with an io.Pipe.

If you use a FIFO (io.Pipe) to synchronize reads and writes, you don't need any extra locks or channels to tail the first io.Reader. Here's an example read function that either prints when its buffer is full, or waits some interval since the last print statement:

func read(r io.Reader) {
    buf := make([]byte, 1024)
    pos := 0
    lastPrint := time.Now()
    for {
        n, err := r.Read(buf[pos:])
        if n > 0 {
            pos += n
        }

        if pos >= len(buf) || time.Since(lastPrint) > 125*time.Millisecond && pos > 0 {
            fmt.Println("read:", buf[:pos])
            lastPrint = time.Now()
            pos = 0
        }

        if err != nil {
            fmt.Println(err)
            break
        }
    }

    if pos > 0 {
        fmt.Println("read:", buf[:pos])
    }
}

func main() {
    pr, pw := io.Pipe()
    go func() {
        io.Copy(pw, &trickle{})
        pw.Close()
    }()
    read(pr)
}

https://play.golang.org/p/8NeV3v0LOU