Issues with goroutines Synchronization using channnels

user_508214 0
Viewed 61

I am using the below code to sync goroutines. Recently while investigating a bug I found that the below code is not working always. Approx one out of five times it fails. The channel quit gets the message before my out channel. I am able to consistently reproduce this issue in my local (not in go-playground) and in k8s environments. As a workaround, I am now using sync.Map to synchronize.

Is there a way to fix the below code?

package main

import (
    "fmt"
    "io/ioutil"
    "log"
    "os"
    "path"
    "sync"
    "sync/atomic"
    "time"
)

func main() {
    //test setup
    filePaths := []string{
        path.Join(os.TempDir(), fmt.Sprint("f1-", time.Now().Nanosecond())),
        path.Join(os.TempDir(), fmt.Sprint("f2-", time.Now().Nanosecond())),
        path.Join(os.TempDir(), fmt.Sprint("f3-", time.Now().Nanosecond())),
        path.Join(os.TempDir(), fmt.Sprint("f4-", time.Now().Nanosecond())),
        path.Join(os.TempDir(), fmt.Sprint("f5-", time.Now().Nanosecond())),
        path.Join(os.TempDir(), fmt.Sprint("f6-", time.Now().Nanosecond())),
    }
    for _, filePath := range filePaths {
        f, err := os.OpenFile(filePath, os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0600)
        if err != nil {
            log.Fatal(err)
        }
        _, err = f.WriteString("There are many variations of passages of Lorem Ipsum available, but the majority have suffered alteration in some form, by injected humour, or randomised words which don't look even slightly believable. If you are going to use a passage of Lorem Ipsum, you need to be sure there isn't anything embarrassing hidden in the middle of text. All the Lorem Ipsum generators on the Internet tend to repeat predefined chunks as necessary, making this the first true generator on the Internet. It uses a dictionary of over 200 Latin words, combined with a handful of model sentence structures, to generate Lorem Ipsum which looks reasonable. The generated Lorem Ipsum is therefore always free from repetition, injected humour, or non-characteristic words etc.")
        if err != nil {
            log.Fatal(err)
        }
        err = f.Close()
        if err != nil {
            log.Fatal(err)
        }
    }
    for  {
        responses, err := getContents(filePaths)
        if err != nil {
            log.Fatal(err)
        }
        if len(responses) != len(filePaths) {
            log.Fatalf("Responses Does not Match, need %d Got %d",len(filePaths), len(responses))
        }
        time.Sleep(1 * time.Second)
    }

}


func  getContents(fileNames []string) ([][]byte, error) {
    wg := sync.WaitGroup{}
    var responseBytes [][]byte
    out := make(chan []byte, 10)
    defer close(out)
    quit := make(chan int)
    var opsFileRead uint64
    var opsChannelGot uint64
    defer close(quit)
    go func() {
        for {
            select {
            case bts := <-out:
                if len(bts) > 0 {
                    atomic.AddUint64(&opsChannelGot, 1)
                    responseBytes = append(responseBytes, bts)
                }
                break
            case <-quit:
                fmt.Printf("I quit, i read %d, i got %d\n", opsFileRead, opsChannelGot)
                return
            }
        }
    }()
    for _, fileName := range fileNames {
        wg.Add(1)
        go func(fName string, out chan []byte, wg *sync.WaitGroup) {
            defer wg.Done()
            data, err := ioutil.ReadFile(fName)
            if err != nil {
                log.Fatal(err)
            }
            out <- data
            atomic.AddUint64(&opsFileRead, 1)
        }(fileName, out, &wg)
    }
    wg.Wait()
    quit <- 1
    return responseBytes, nil
}
go
1 Answers

The out channel can contain values when quit is received. Fix by making out an unbuffered channel:

out := make(chan []byte)

This ensures that values sent from the workers are received before quit:

  • send/receive on the unbuffered channel happens before the call to wg.Done()
  • all calls to wg.Done() happen before wg.Wait() returns
  • wg.Wait() returns before a value is sent to quit

Therefore, the values are received from out before the value is sent to quit.

Another approach is to close the out channel to signal the result collector that the workers are done:

func getContents(fileNames []string) ([][]byte, error) {
    wg := sync.WaitGroup{}
    var responseBytes [][]byte
    out := make(chan []byte)
    var opsFileRead uint64
    var opsChannelGot uint64

    for _, fileName := range fileNames {
        wg.Add(1)
        go func(fName string, out chan []byte, wg *sync.WaitGroup) {
            defer wg.Done()
            data, err := ioutil.ReadFile(fName)
            if err != nil {
                log.Fatal(err)
            }
            out <- data
            atomic.AddUint64(&opsFileRead, 1)
        }(fileName, out, &wg)
    }

    // Close out after workers are done.
    go func() {
        wg.Wait()
        close(out)
    }()

    // Loop over outputs until done.
    for bts := range out {
        if len(bts) > 0 {
            atomic.AddUint64(&opsChannelGot, 1)
            responseBytes = append(responseBytes, bts)
        }
    }

    fmt.Printf("I quit, i read %d, i got %d\n", opsFileRead, opsChannelGot)

    return responseBytes, nil
}
user_57289910
Related