0
\$\begingroup\$

The code below implements ProgressWriter, a wrapper around io.Writer that can track the number of bytes written, and a function MonitorProgress that, as a goroutine, uses a Ticker to log progress incrementally.

Currently, I'm using the atomic package to synchronize writes to and reads from the totalBytes field. Is this the best I can do? It seems as though there should be some way to avoid this by posting the progress to a channel, but it's not obvious what that would be.

  • I tried an implementation that had every Write post progress to an unbuffered channel, but that caused Write to block waiting for a channel read.
  • I wrote a version that posted progress from a second goroutine, but then realized I was just pushing the shared-memory problem to that goroutine instead of the MonitorProgress one.
  • I tried a non-blocking version with switch/default that would only post progress when there was a reader available, but ran into timing issues where the reader would never get the most recent progress.
  • I considered writing to a buffered channel, but it seemed like:
    1. it wasn't clear what the size of the buffer should be
    2. the reader logic would get messy trying to make sure we only logged the most recent progress

Is there a channel pattern I'm missing here? It seems unlikely that I've stumbled into the one case where "communicate by sharing memory" is actually correct.

package main

import (
    "io"
    "io/ioutil"
    "log"
    "math/rand"
    "sync/atomic"
    "time"
)

type ProgressWriter struct {
    totalBytes    *int64
    expectedBytes int64
    out           io.Writer
}

func NewProgressWriter(expectedBytes int64, out io.Writer) *ProgressWriter {
    zero := int64(0)
    return &ProgressWriter{&zero, expectedBytes, out}
}

func (w *ProgressWriter) updateTotal(additionalBytes int) {
    atomic.AddInt64(w.totalBytes, int64(additionalBytes))
}

func (w *ProgressWriter) TotalBytes() int64 {
    return atomic.LoadInt64(w.totalBytes)
}

func (w *ProgressWriter) Write(p []byte) (n int, err error) {
    n, err = w.out.Write(p)
    w.updateTotal(n)
    return
}

func (w *ProgressWriter) MonitorProgress(interval time.Duration) {
    ticker := time.NewTicker(interval)
    defer ticker.Stop()

    nsStart := time.Now().UnixNano()
    for {
        select {
        case _ = <-ticker.C:
            currentBytes := w.TotalBytes()
            sElapsed := float64(time.Now().UnixNano()-nsStart) / float64(time.Second)
            log.Printf("%d of %d bytes (%f s elapsed)", currentBytes, w.expectedBytes, sElapsed)
            if currentBytes >= w.expectedBytes {
                return
            }
        default:
            // wait for next tick
        }
    }
}

func main() {
    expectedBytes := int64(1024 * 1024)
    input := io.LimitReader(rand.New(rand.NewSource(1)), expectedBytes)
    output := NewProgressWriter(expectedBytes, ioutil.Discard)
    go output.MonitorProgress(time.Second)
    for output.TotalBytes() < expectedBytes {
        p := make([]byte, 4096)
        _, err := input.Read(p)
        if err != nil {
            panic(err)
        }
        _, err = output.Write(p)
        if err != nil {
            panic(err)
        }
        time.Sleep(10 * time.Millisecond)
    }
}
\$\endgroup\$

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Browse other questions tagged or ask your own question.