Golang multithreading and pipeline

Question

When you run an anonymous function in the router, it does not work correctly. (it opens and parsit not all files), although if you do not run it in a separate thread, then everything works fine. Help me figure out what I'm doing wrong.

names := make(map[string]bool) var mutex = &sync.Mutex{} for _, file := range f { out := make(chan int) go func() { if file.Name[:2] == "HM"{ names[file.Name] = true mutex.Lock() xmlFile, err := os.Open(filepath.Join(file.PathUnzipFile)) if err != nil { fmt.Printf("[ERROR] невозможно открыть файл %e \n", err) } mutex.Unlock() defer xmlFile.Close() var hFile ZL_LIST decoder := xml.NewDecoder(xmlFile) decoder.CharsetReader = func(charset string, input io.Reader) (io.Reader, error) { switch charset { case "windows-1251": return charmap.Windows1251.NewDecoder().Reader(input), nil default: return nil, fmt.Errorf("unknown charset: %s", charset) } } err1 := decoder.Decode(&hFile) if err1 != nil { fmt.Errorf("[ERROR] невозможно декодировать файл %e", err) } } close(out) }() }

Accepted Answer · 2018-07-11T04:31:41

It was necessary to use the WaitGroup action waiting group and pass the copy itself directly to the anonymous function. Do not consider it an advertisement, but found the solution in the book Go in practice on page 97.

 func (ZL_LIST) Parse(FilesName []*Files) { unzipPath, err := filepath.Abs("tmp") files, err := ioutil.ReadDir(unzipPath) f := make([]*Files, 0) for _, file := range files { fileZip := new(Files) if !file.IsDir() && file.Name()[:1] != "." { fileZip.Name = file.Name() fileZip.PathUnzipFile = filepath.Join(unzipPath, file.Name()) f = append(f, fileZip) } } if err != nil { fmt.Printf("[ERROR] ошибка при получении файлов %v", err) } names := make(nameMap, 1) var wg sync.WaitGroup var countHmLmFiles int = -1 var countHmFiles int = 0 var countLmFiles int = 0 var file *Files var l sync.Mutex for countHmLmFiles, file = range f { wg.Add(1) go func(filename *Files) { parseHmLm(filename, &countHmFiles, &countLmFiles, &names, &l) //parseHmLm(filename) wg.Done() }(file) } wg.Wait() fmt.Printf("[INFO] Прочитано %d файловб из них HM файлов %d и LM файлов %d ", countHmLmFiles+1, countHmFiles, countLmFiles) } func parseHmLm(filename *Files, countHmFiles *int, countLmFiles *int, names *nameMap, l *sync.Mutex) { fmt.Println("[INFO] считывается файл ", filename.Name, " с заголовком ", filename.Name[:2]) xmlFile, err := os.Open(filepath.Join(filename.PathUnzipFile)) if err != nil { fmt.Printf("[ERROR] невозможно открыть файл %e \n", err) } defer xmlFile.Close() switch filename.Name[:2] { case "HM": l.Lock() (*names)[filename.Name] = true (*countHmFiles) += 1 l.Unlock() var hFile ZL_LIST decoder := xml.NewDecoder(xmlFile) decoder.CharsetReader = charset err := decoder.Decode(&hFile) if err != nil { fmt.Errorf("[ERROR] невозможно декодировать файл %e", err) } case "LM": l.Lock() (*countLmFiles) += 1 l.Unlock() var lFile PERS_LIST decoder := xml.NewDecoder(xmlFile) decoder.CharsetReader = charset err := decoder.Decode(&lFile) if err != nil { fmt.Errorf("[ERROR] невозможно декодировать файл %e", err) } } } func charset(charset string, input io.Reader) (io.Reader, error) { switch charset { case "windows-1251": return charmap.Windows1251.NewDecoder().Reader(input), nil default: return nil, fmt.Errorf("unknown charset: %s", charset) } }

You have a lot of racing there. Try running with -race with GORACE="halt_on_error=1" . - Ainar-G
Yes, I already understood it ... I am working on it now - Evgeny Gusev

Golang multithreading and pipeline

1 answer 1

More articles: