Search code examples
gochannelgoroutine

goroutine reading from channel with dynamic loop getting waitgroup is reused before previous was has returned


I am working on a small utility which needs to iterate over a dynamic range of items (can be 100 or can be 100000) and put the items in a channel. From that channel another function reads the items and does some processing on each item individually. I am trying to use sync.WaitGroup to make sure that my utility does not exit before all the items in the channels are processed. As I am fairly new to channels and waitgroups, I am stuck with an error panic: sync: WaitGroup is reused before previous Wait has returned

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

package main

import (
    "fmt"
    "github.com/dchest/uniuri"
    "math/rand"
    "sync"
    "time"
)

var wg sync.WaitGroup
var count = 0

func printMe(msg string) {
    time.Sleep(1 * time.Second)
    fmt.Println(count, msg)
}

func makeMePrint(ch chan string) {
    for s := range ch {
        count++
        wg.Add(1)
        printMe(s)
        wg.Done()
    }

}

func writePrint(ch chan<- string) {
    fmt.Println("Starting to insert data in channel")
    for i := 1; i <= rand.Intn(30); i++ {
        s := uniuri.New()
        ch <- s
    }
    fmt.Println("We are done inserting all data in the channel")
    close(ch)
}

func main() {

    var ch = make(chan string)

    go writePrint(ch)
    go makeMePrint(ch)
    time.Sleep(1 * time.Second)
    wg.Wait()
}

This is the main idea of what I am working on (not the exact code but exactly the same architecture with same number of functions).

How can I make sure that the utility only exits only when all the items in the channel are processes.

Any help is appreciated.


Solution

  • I finally got it to work.

    package main
    
    import (
        "fmt"
        "github.com/dchest/uniuri"
        "math/rand"
        "sync"
        "time"
    )
    
    var count = 0
    
    func printMe(msg string) {
        time.Sleep(1 * time.Second)
        fmt.Println(count, msg)
    }
    
    func makeMePrint(wg *sync.WaitGroup, ch chan string) {
        wg.Add(1)
        defer wg.Done()
    
        for s := range ch {
            count++
            printMe(s)
    
        }
    
    }
    
    func writePrint(wg *sync.WaitGroup, ch chan<- string) {
        wg.Add(1)
        defer wg.Done()
    
        fmt.Println("Starting to insert data in channel")
        for i := 1; i <= rand.Intn(30); i++ {
            s := uniuri.New()
            ch <- s
        }
        fmt.Println("We are done inserting all data in the channel")
        close(ch)
    }
    
    func main() {
        wg := &sync.WaitGroup{}
        var ch = make(chan string)
    
        go writePrint(wg, ch)
        go makeMePrint(wg, ch)
        time.Sleep(1 * time.Second)
        wg.Wait()
    }