I'm trying to create a log file parser which parses a file by concurrently reading the file in chunks of 100 kb. I'm using waitgroup wait for all the log entries to be processed. However, I'm getting the error panic: sync: negative WaitGroup counter while running this. Could someone help me understand why it's failing?
Code
package main
import (
"bufio"
"encoding/json"
"fmt"
"log"
"os"
"sync"
)
var filename string = "data/input.log"
const BUFFER = 100 * 1024
type logEntry struct {
Lines []string
CreatedAt string
LineCount int
Size int
Offset int64
}
func main() {
wg := &sync.WaitGroup{}
logChan := make(chan logEntry)
// get file size
fi, err := os.Stat(filename)
if err != nil {
log.Fatal(err)
}
// get the size
offset, size := int64(0), fi.Size()
for {
if offset > size {
break
}
go func(offset int64, wg *sync.WaitGroup) {
err := read(offset, logChan, wg)
if err != nil {
panic(err)
}
}(offset, wg)
offset += BUFFER
}
for l := range logChan {
fmt.Println("Removing from wg - ", l.LineCount)
wg.Done()
}
wg.Wait()
close(logChan)
}
func read(offset int64, logChan chan (logEntry), wg *sync.WaitGroup) error {
var l logEntry
file, err := os.Open(filename)
if err != nil {
return err
}
defer file.Close()
scanner := bufio.NewScanner(file)
scanner.Split(bufio.ScanLines)
for scanner.Scan() {
line := scanner.Bytes()
err := json.Unmarshal(line, &l)
// At any point if line parsing fails, throw error and dont continue with parsing rest of the logs
if err != nil {
return err
}
logChan <- l
fmt.Println("Adding to wg - ", l.LineCount)
wg.Add(1)
}
return nil
}
Output:
Adding to wg - 17
Removing from wg - 17
Removing from wg - 12
panic: sync: negative WaitGroup counter
goroutine 1 [running]:
sync.(*WaitGroup).Add(0x10fe1a0, 0xc00000e018)
/usr/local/Cellar/go@1.17/1.17.13/libexec/src/sync/waitgroup.go:74 +0x105
sync.(*WaitGroup).Done(...)
/usr/local/Cellar/go@1.17/1.17.13/libexec/src/sync/waitgroup.go:99
main.main()
main.go:57 +0x2b0
exit status 2