How to resolve this issue: panic: sync: negative WaitGroup counter

After running again and again I am getting sometimes this issue. I know it's related to counter related. When the Done() method of a sync.WaitGroup is called more times than the Add() method was called then it will throw this error.

How do I fix this?

My code created batches of size 4 and do some processing on each batch but I am getting issue with resolving this panic.

package main

import (
	"fmt"
	"sync"
)

func main() {
	// create input channel
	input := make(chan int)

	// create wait group
	var wg sync.WaitGroup

	// start batcher goroutine
	wg.Add(1)
	go batcher(input, &wg)

	// send input values to the batcher
	for i := 1; i <= 10; i++ {
		input <- i
	}

	// close input channel
	close(input)

	// wait for batcher goroutine to finish
	wg.Wait()
}

func batcher(input chan int, wg *sync.WaitGroup) {
	// create batch channel with buffer of size 4
	batch := make(chan int, 4)

	// create channel to synchronize worker goroutines
	done := make(chan bool)

	// create wait group for worker goroutines
	var workerWg sync.WaitGroup

	// start worker goroutines
	for i := 0; i < 4; i++ {
		workerWg.Add(1)
		go worker(batch, &workerWg, done)
	}

	// read input values and send to batch
	for value := range input {
		batch <- value
		if len(batch) == 4 {
			// wait for worker goroutines to finish processing batch
			workerWg.Wait()

			// send batch to worker goroutines
			for i := 0; i < 4; i++ {
				workerWg.Add(1)
				go sendBatch(batch, &workerWg, done)
			}
		}
	}

	// wait for worker goroutines to finish processing remaining batch
	workerWg.Wait()

	// close done channel to notify that all batches have been processed
	close(done)

	wg.Done()
}

func sendBatch(batch chan int, workerWg *sync.WaitGroup, done chan bool) {
	// process batch
	for value := range batch {
		fmt.Println("Processing value:", value)
	}

	// notify worker goroutines that batch has been processed
	workerWg.Done()

	select {
	case done <- true:
	default:
		// done channel has been closed
	}
}

func worker(batch chan int, workerWg *sync.WaitGroup, done chan bool) {
	// process batches received from batch channel
	for batch := range batch {
		// process batch
		fmt.Println("Processing batch:", batch)
		workerWg.Done()
	}

	// notify batcher goroutine that worker goroutine has finished
	select {
	case done <- true:
	default:
		// done channel has been closed
	}
}

Basic code to write batcher:

package main

import (
	"fmt"
	"sync"
)

func main() {
	input := make(chan int)
	output := make(chan []int)

	var wg sync.WaitGroup
	wg.Add(2)

	// Start the batcher goroutine
	go func() {
		batch := []int{}
		for value := range input {
			batch = append(batch, value)
			if len(batch) == 4 {
				output <- batch
				batch = []int{}
			}
		}
		if len(batch) > 0 {
			output <- batch
		}
		close(output)
		wg.Done()
	}()

	// Start the worker goroutine
	go func() {
		for batch := range output {
			sum := 0
			for _, value := range batch {
				sum += value
			}
			fmt.Printf("Sum of batch %v: %d\n", batch, sum)
		}
		wg.Done()
	}()

	// Send input values to the batcher
	for _, v := range []int{1, 2, 3, 4, 5, 6, 7, 8, 9, 10} {
		input <- v
	}
	close(input)

	// Wait for both goroutines to finish
	wg.Wait()
}

Sum of batch [1 2 3 4]: 10
Sum of batch [5 6 7 8]: 26
Sum of batch [9 10]: 19

Earlier design was bit complicated I will try to extend this basic one.

According to this code:

for i := 0; i < 4; i++ {
    workerWg.Add(1)
    go worker(batch, &workerWg, done)
}

I think workerWg.Done() should be moved outside of the loop:

  func worker(batch chan int, workerWg *sync.WaitGroup, done chan bool) {
+ 	defer workerWg.Done()
  	// process batches received from batch channel
  	for batch := range batch {
  		// process batch
  		fmt.Println("Processing batch:", batch)
- 		workerWg.Done()
  	}

  	// notify batcher goroutine that worker goroutine has finished
  	select {
  	case done <- true:
  	default:
  		// done channel has been closed
  	}
  }

But batch is not closed in the demo. So in fact, the goroutine will run for ever until the program is ended.

I'm not sure whether there is any other issue. The design is too complicated. Complicated code is hard to understand and prone to error. Consider redesigning it.