How to fix this issue: Panic: Synchronization: Negative WaitGroup counter

Question content

I sometimes run into this problem after running it again and again. I know this is related to counters. It throws this error when the done() method of sync.waitgroup is called more times than the add() method is called.

how to solve this problem?

My code creates batches of size 4 and does some processing on each batch, but I'm having trouble 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 for writing batch programs:

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

The early design is a bit complex, I will try to expand on this basic design.

Correct answer

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 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 effect, the goroutine will run forever until the program ends.

I don’t know if there are any other questions. The design is too complicated. Complex code is difficult to understand and error-prone. Consider redesigning it.

The above is the detailed content of How to fix this issue: Panic: Synchronization: Negative WaitGroup counter. For more information, please follow other related articles on the PHP Chinese website!