Waiting for goroutines to finish

Question

I have a program that starts a bunch of goroutines, feeds them data and when there is no more data, waits for the goroutines to finish. I use sync.WaitGroup. I am unsure if I am using correctly, in an idiomatic way or if there is a better solution.

Background

My real code reads a log file of disk-space and numbers of files backed up etc for several servers. The records are intermingled. From this data it produces a web-page report per server with graphs etc. It works but I believe I can more cleanly separate the functions if I use concurrency. This is a test program before I go ahead and refactor.

Code (72 lines including data)

package main

import (
    "fmt"
    "strconv"
    "strings"
    "sync"
)

func main() {
    p := newPool()
    defer p.Close()

    for _, line := range strings.Split(logData, "\n") {
        f := strings.Fields(line)
        yr, _ := strconv.Atoi(f[0])
        sub := f[1]
        amt, _ := strconv.Atoi(f[2])
        r := logRec{yr, sub, amt}
        p.Report(r)
    }
}

type logRec struct {
    when    int
    subject string
    amount  int
}

type pool struct {
    reporters map[string]chan logRec
    wg        *sync.WaitGroup
}

func newPool() pool {
    rs := make(map[string]chan logRec)
    var wg sync.WaitGroup
    return pool{reporters: rs, wg: &wg}
}

func (p *pool) Report(rec logRec) {
    if _, ok := p.reporters[rec.subject]; !ok {
        c := make(chan logRec)
        p.wg.Add(1)
        go func() {
            defer p.wg.Done()
            tot := 0
            for r := range c {
                tot += r.amount
            }
            fmt.Printf("Total for %s is %d\n", rec.subject, tot)
        }()
        p.reporters[rec.subject] = c
    }
    p.reporters[rec.subject] <- rec
}

func (p pool) Close() {
    for _, c := range p.reporters {
        close(c)
    }

    p.wg.Wait()
}

const logData = `2018 apples 9
2018 oranges 5
2019 apples 27
2019 lemons 13
2019 oranges 2
2020 oranges 1
2020 apples 16
2020 lemons 3`

Output

Total for oranges is 8
Total for lemons is 16
Total for apples is 52

---

Question:

Can I improve this? Is there a better way to wait or a more idiomatic use of WaitGroups?

Answer 1

Yes, since you are here for code review:

1. Never ever ignore errors: yr, _ := strconv.Atoi(f[0])

        yr, err := strconv.Atoi(f[0])
        if err != nil {
            log.Fatal(err)
        }

2. You may use just one strings.Split/strings.Fields this way:

    f := strings.Fields(logData)
    for i := 3; i <= len(f); i += 3 {
        yr, err := strconv.Atoi(f[i-3])
        if err != nil {
            log.Fatal(err)
        }
        amt, err := strconv.Atoi(f[i-1])
        if err != nil {
            log.Fatal(err)
        }
        r := logRec{yr, f[i-2], amt}
        p.Report(r)
    }

3. You are over complicating a simple sum function, You don't need channels here, just a map does the job: It is this simple:

package main

import (
    "fmt"
    "log"
    "strconv"
    "strings"
)

func main() {
    m := map[string]int{}
    f := strings.Fields(logData)
    for i := 3; i <= len(f); i += 3 {
        amt, err := strconv.Atoi(f[i-1])
        if err != nil {
            log.Fatal(err)
        }
        m[f[i-2]] += amt
    }
    fmt.Println(m)
}

const logData = `2018 apples 9
2018 oranges 5
2019 apples 27
2019 lemons 13
2019 oranges 2
2020 oranges 1
2020 apples 16
2020 lemons 3`

4. For concurrency sake: You only need concurrent map, nothing more:

package main

import (
    "fmt"
    "log"
    "strconv"
    "strings"
    "sync"
)

type concurrentMap struct {
    sync.Mutex
    m map[string]int
}

func (p *concurrentMap) sum(item string, amount int) {
    p.Lock()
    p.m[item] += amount
    p.Unlock()
}

var data = &concurrentMap{m: map[string]int{}}

func main() {
    f := strings.Fields(logData)
    for i := 3; i <= len(f); i += 3 {
        amt, err := strconv.Atoi(f[i-1])
        if err != nil {
            log.Fatal(err)
        }
        data.sum(f[i-2], amt)
    }
    fmt.Println(data.m)
}

const logData = `2018 apples 9
2018 oranges 5
2019 apples 27
2019 lemons 13
2019 oranges 2
2020 oranges 1
2020 apples 16
2020 lemons 3`

5. Long running task example:

package main

import (
    "fmt"
    "log"
    "strconv"
    "strings"
    "sync"
    "time"
)

type concurrentMap struct {
    m map[string]int
    sync.Mutex
}

func (p *concurrentMap) sum(item string, amount int) {
    p.Lock()
    p.m[item] += amount
    p.Unlock()
}
func (p *concurrentMap) show() {
    p.Lock()
    for k, v := range p.m {
        fmt.Printf("Total for %s is %d\n", k, v)
    }
    p.Unlock()
}

var data = &concurrentMap{m: map[string]int{}}

func main() {
    go data.sum("oranges", 100)
    go data.sum("apples", 100)
    go data.sum("lemons", 100)

    go func() {
        f := strings.Fields(logData)
        for i := 3; i <= len(f); i += 3 {
            amt, err := strconv.Atoi(f[i-1])
            if err != nil {
                log.Fatal(err)
            }
            data.sum(f[i-2], amt)
            time.Sleep(1000 * time.Millisecond) // e.g. slow hard disk
        }
    }()

    t := time.NewTicker(1000 * time.Millisecond)
    defer t.Stop()
    for range t.C {
        data.show()
        fmt.Println()
    }
}

const logData = `2018 apples 9
2018 oranges 5
2019 apples 27
2019 lemons 13
2019 oranges 2
2020 oranges 1
2020 apples 16
2020 lemons 3`

And finally if your tasks will finish in time, you may use sync.WaitGroup:

package main

import (
    "fmt"
    "log"
    "strconv"
    "strings"
    "sync"
    "time"
)

type concurrentMap struct {
    m map[string]int
    sync.Mutex
}

func (p *concurrentMap) sum(item string, amount int) {
    p.Lock()
    p.m[item] += amount
    p.Unlock()
}
func (p *concurrentMap) show() {
    p.Lock()
    for k, v := range p.m {
        fmt.Printf("Total for %s is %d\n", k, v)
    }
    p.Unlock()
}

var data = &concurrentMap{m: map[string]int{}}

func main() {
    finished := &sync.WaitGroup{}
    finished.Add(1)
    go func() {
        defer finished.Done()
        data.sum("oranges", 100)
        data.sum("apples", 100)
        data.sum("lemons", 100)
    }()

    finished.Add(1)
    go func() {
        defer finished.Done()
        f := strings.Fields(logData)
        for i := 3; i <= len(f); i += 3 {
            amt, err := strconv.Atoi(f[i-1])
            if err != nil {
                log.Fatal(err)
            }
            data.sum(f[i-2], amt)
            time.Sleep(100 * time.Millisecond) // e.g. slow hard disk
        }
    }()

    finished.Wait()
    data.show()
}

const logData = `2018 apples 9
2018 oranges 5
2019 apples 27
2019 lemons 13
2019 oranges 2
2020 oranges 1
2020 apples 16
2020 lemons 3`

That is all.
I hope this helps.