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.