Concurrent implementation of the DMR parser.

concurrent-dmr-parser.go

func main() {
	flag.Parse()

	xmlFileName := filepath.Join("/tmp/", "out.xml")

	if _, err := os.Stat(*inFile); os.IsNotExist(err) {
		fmt.Printf("abort: file %q does not seem to exist\n", *inFile)
		return
	}
	
	xmlFile, err := os.Open(xmlFileName)
	if err != nil {
		fmt.Println("Unable to open file:", err)
		return
	}
	defer func() {
		if err := xmlFile.Close(); err != nil {
			panic(err)
		}
	}()

	// Pick a parser based on CLI flag.
	var parser engines.IDMRParser
	switch *engine {
	case "xml":
		parser = engines.NewXMLParser()
	case "string":
		parser = engines.NewStringParser()
	default:
		fmt.Printf("Invalid parser: %q\n", engine)
		return
	}

	// Nr. of workers = cpu core count - 1 for the main go routine.
	numWorkers := int(math.Max(1.0, float64(runtime.NumCPU()-1)))

	// Prepare channels for communicating parsed data and termination.
	lines, parsed, done := make(chan []string, numWorkers), make(chan string, numWorkers), make(chan int)

	// Start the number of workers (parsers) determined by numWorkers.
	fmt.Printf("Starting %v workers...\n", numWorkers)
	for i := 0; i < numWorkers; i++ {
		go parser.ParseExcerpt(i, lines, parsed, done)
	}

	// Main file scanner go routine.
	go func() {
		scanner := bufio.NewScanner(xmlFile)
		excerpt := []string{}
		grab := false
		defer func() {
			close(lines)
		}()
		for scanner.Scan() {
			line := strings.TrimSpace(scanner.Text())
			if strings.HasPrefix(line, "<ns:Statistik>") {
				grab = true
			} else if strings.HasPrefix(line, "</ns:Statistik>") {
				grab = false
				excerpt = append(excerpt, line)
				lines <- excerpt // On every closing elem. we send the excerpt to a worker and move on.
				excerpt = nil
			}
			if grab {
				excerpt = append(excerpt, line)
			}
		}
	}()

	var vehicles engines.VehicleList = make(map[string]struct{}) // For keeping track of unique vehicles.

	// Wait for parsed excerpts to come in, and ensure their uniqueness by using a map.
	waits := numWorkers
	for {
		select {
		case vehicle := <-parsed:
			if _, ok := vehicles[vehicle]; !ok {
				vehicles[vehicle] = struct{}{}
			}
		case <-done:
			waits--
			if waits == 0 {
				writeToFile(vehicles, *outFile) // Implementation left out on purpose.
				return
			}
		}
	}

}