perf(convert): parallelize batch conversion across worker pool

Batch convert processed files strictly sequentially. Parsing/converting
is CPU-bound and stateless per file, so run it across a GOMAXPROCS-sized
worker pool; zip writing and progress updates stay serialized in
original file order.
This commit is contained in:
Mikhail Chusavitin
2026-08-15 12:43:21 +03:00
parent 42cfa3aa94
commit 4fa16c78a5
+67 -27
View File
@@ -15,6 +15,7 @@ import (
"os"
"path/filepath"
"regexp"
"runtime"
"sort"
"strconv"
"strings"
@@ -1423,40 +1424,79 @@ func (s *Server) runConvertJob(jobID, tempDir string, inputFiles []convertInputF
success := 0
totalProcess := len(inputFiles)
for i, in := range inputFiles {
s.jobManager.AppendJobLog(jobID, fmt.Sprintf("Processing %s", in.Name))
payload, err := os.ReadFile(in.Path)
if err != nil {
failures = append(failures, fmt.Sprintf("%s: %v", in.Name, err))
continue
}
type convertOutcome struct {
name string
data any
fail string
}
result, _, _, err := s.analyzeUploadedFile(in.Name, in.MIMEType, payload)
if err != nil {
failures = append(failures, fmt.Sprintf("%s: %v", in.Name, err))
progress := ((i + 1) * 100) / totalProcess
s.jobManager.UpdateJobStatus(jobID, CollectStatusRunning, progress, "")
continue
workers := runtime.GOMAXPROCS(0)
if workers > totalProcess {
workers = totalProcess
}
if workers < 1 {
workers = 1
}
jobs := make(chan int)
results := make([]convertOutcome, totalProcess)
var wg sync.WaitGroup
for w := 0; w < workers; w++ {
wg.Add(1)
go func() {
defer wg.Done()
for i := range jobs {
in := inputFiles[i]
s.jobManager.AppendJobLog(jobID, fmt.Sprintf("Processing %s", in.Name))
payload, err := os.ReadFile(in.Path)
if err != nil {
results[i] = convertOutcome{name: in.Name, fail: fmt.Sprintf("%s: %v", in.Name, err)}
continue
}
result, _, _, err := s.analyzeUploadedFile(in.Name, in.MIMEType, payload)
if err != nil {
results[i] = convertOutcome{name: in.Name, fail: fmt.Sprintf("%s: %v", in.Name, err)}
continue
}
if result == nil || result.Hardware == nil {
results[i] = convertOutcome{name: in.Name, fail: fmt.Sprintf("%s: no hardware data", in.Name)}
continue
}
reanimatorData, err := exporter.ConvertToReanimator(result)
if err != nil {
results[i] = convertOutcome{name: in.Name, fail: fmt.Sprintf("%s: %v", in.Name, err)}
continue
}
results[i] = convertOutcome{name: in.Name, data: reanimatorData}
}
}()
}
go func() {
defer close(jobs)
for i := range inputFiles {
jobs <- i
}
if result == nil || result.Hardware == nil {
failures = append(failures, fmt.Sprintf("%s: no hardware data", in.Name))
}()
wg.Wait()
for i, out := range results {
if out.data == nil {
failures = append(failures, out.fail)
progress := ((i + 1) * 100) / totalProcess
s.jobManager.UpdateJobStatus(jobID, CollectStatusRunning, progress, "")
continue
}
reanimatorData, err := exporter.ConvertToReanimator(result)
if err != nil {
failures = append(failures, fmt.Sprintf("%s: %v", in.Name, err))
progress := ((i + 1) * 100) / totalProcess
s.jobManager.UpdateJobStatus(jobID, CollectStatusRunning, progress, "")
continue
}
entryPath := sanitizeZipPath(in.Name)
entryPath := sanitizeZipPath(out.name)
entry, err := zw.Create(entryPath + ".reanimator.json")
if err != nil {
failures = append(failures, fmt.Sprintf("%s: %v", in.Name, err))
failures = append(failures, fmt.Sprintf("%s: %v", out.name, err))
progress := ((i + 1) * 100) / totalProcess
s.jobManager.UpdateJobStatus(jobID, CollectStatusRunning, progress, "")
continue
@@ -1464,8 +1504,8 @@ func (s *Server) runConvertJob(jobID, tempDir string, inputFiles []convertInputF
encoder := json.NewEncoder(entry)
encoder.SetIndent("", " ")
if err := encoder.Encode(reanimatorData); err != nil {
failures = append(failures, fmt.Sprintf("%s: %v", in.Name, err))
if err := encoder.Encode(out.data); err != nil {
failures = append(failures, fmt.Sprintf("%s: %v", out.name, err))
} else {
success++
}