Synchronize Buffered channel and Waitgroup

Question

I am having issue while using waitgroup with the buffered channel. The problem is waitgroup closes before channel is read completely, which make my channel is half read and break in between.

func main() {
    var wg sync.WaitGroup
    var err error

    start := time.Now()
    students := make([]studentDetails, 0)
    studentCh := make(chan studentDetail, 10000)
    errorCh := make(chan error, 1)

    wg.Add(1)

    go s.getDetailStudents(rCtx, studentCh , errorCh, &wg, s.Link, false)
    go func(ch chan studentDetail, e chan error) {
    
    LOOP:
        for {
            select {
            case p, ok := <-ch:
                if ok {
                    L.Printf("Links %s: [%s]\n", p.title, p.link)
                    students = append(students, p)
                } else {
                    L.Print("Closed channel")
                    break LOOP
                }
            case err = <-e:
                if err != nil {
                    break
                }
            }
        }
    }(studentCh, errorCh)
    wg.Wait()
    close(studentCh)
    close(errorCh)
    L.Warnln("closed: all wait-groups completed!")
    L.Warnf("total items fetched: %d", len(students))

    elapsed := time.Since(start)
    L.Warnf("operation took %s", elapsed)
}

The problem is this function is recursive. I mean some http call to fetch students and then make more calls depending on condition.

func (s Student) getDetailStudents(rCtx context.Context, content chan<- studentDetail, errorCh chan<- error, wg *sync.WaitGroup, url string, subSection bool) {
    util.MustNotNil(rCtx)
    L := logger.GetLogger(rCtx)
    defer func() {
        L.Println("Closing all waitgroup!")
        wg.Done()
    }()

    wc := getWC()
    httpClient := wc.Registry.MustHTTPClient()
    res, err := httpClient.Get(url)
    if err != nil {
        L.Fatal(err)
    }
    defer res.Body.Close()
    if res.StatusCode != 200 {
        L.Errorf("status code error: %d %s", res.StatusCode, res.Status)
        errorCh <- errors.New("service_status_code")
        return
    }

    // parse response and return error if found some through errorCh as done above.
    // decide page subSection based on response if it is more.
    if !subSection {
        wg.Add(1)
        go s.getDetailStudents(rCtx, content, errorCh, wg, link, true)
        // L.Warnf("total pages found %d", pageSub.Length()+1)
    }

    // Find students from response list and parse each Student
    students := s.parseStudentItemList(rCtx, item)
    for _, student := range students {
        content <- student
    }
 
    L.Warnf("Calling HTTP Service for %q with total %d record", url, elementsSub.Length())
}

Variables are changed to avoid original code base.

The problem is students are read randomly as soon as Waitgroup complete. I am expecting to hold the execution until all students are read, In case of error it should break as soon error encounter.

Answer 1

You need to know when the receiving goroutine completes. The WaitGroup does that for the generating goroutine. So, you can use two waitgroups:

wg.Add(1)
go s.getDetailStudents(rCtx, studentCh , errorCh, &wg, s.Link, false)
wgReader.Add(1)
go func(ch chan studentDetail, e chan error) {
    defer wgReader.Done()
    ...
}
wg.Wait()
close(studentCh)
close(errorCh)
wgReader.Wait() // Wait for the readers to complete