diff options
| -rw-r--r-- | .gitignore | 4 | ||||
| -rw-r--r-- | config/config-b2.sample.json | 3 | ||||
| -rw-r--r-- | config/config-local-unix.sample.json | 3 | ||||
| -rw-r--r-- | config/config.go | 13 | ||||
| -rw-r--r-- | main.go | 12 |
5 files changed, 21 insertions, 14 deletions
@@ -28,8 +28,8 @@ go.work.sum .env # Editor/IDE -# .idea/ -# .vscode/ +.idea/ +.vscode/ config*.json !config*.sample.json diff --git a/config/config-b2.sample.json b/config/config-b2.sample.json index d3d772e..c48b1b2 100644 --- a/config/config-b2.sample.json +++ b/config/config-b2.sample.json @@ -1,6 +1,7 @@ { "ForceRewrite": false, - "MaxConcurrentJobs": 4, + "MaxProcessThreads": 4, + "MaxPreProcessThreads": 12, "LogLevel": "info", "Input": { "Storage": { diff --git a/config/config-local-unix.sample.json b/config/config-local-unix.sample.json index 637a05a..36d8114 100644 --- a/config/config-local-unix.sample.json +++ b/config/config-local-unix.sample.json @@ -1,6 +1,7 @@ { "ForceRewrite": false, - "MaxConcurrentJobs": 4, + "MaxProcessThreads": 4, + "MaxPreProcessThreads": 12, "LogLevel": "debug", "Input": { "Storage": { diff --git a/config/config.go b/config/config.go index d2b7072..5b05ce9 100644 --- a/config/config.go +++ b/config/config.go @@ -10,12 +10,13 @@ import ( ) type Config struct { - Input InputConfig `json:"Input" validate:"required"` - Converter ConverterConfig `json:"Converter" validate:"required"` - Output OutputConfig `json:"Output" validate:"required"` - MaxConcurrentJobs int `json:"MaxConcurrentJobs" validate:"required,min=1"` - ForceRewrite bool `json:"ForceRewrite" validate:"required"` - LogLevel slog.Level `json:"LogLevel" validate:"required"` + Input InputConfig `json:"Input" validate:"required"` + Converter ConverterConfig `json:"Converter" validate:"required"` + Output OutputConfig `json:"Output" validate:"required"` + MaxProcessThreads int `json:"MaxProcessThreads" validate:"required,min=1"` + MaxPreProcessThreads int `json:"MaxPreProcessThreads" validate:"min=1;gtefield=MaxProcessThreads"` + ForceRewrite bool `json:"ForceRewrite" validate:"required"` + LogLevel slog.Level `json:"LogLevel" validate:"required"` } type InputConfig struct { @@ -86,11 +86,13 @@ func main() { default: } - semaphore := make(chan struct{}, cfg.MaxConcurrentJobs) + processSemaphore := make(chan struct{}, cfg.MaxProcessThreads) + queueSemaphore := make(chan struct{}, cfg.MaxPreProcessThreads) var wg sync.WaitGroup wg.Add(len(files)) for i, file := range files { + queueSemaphore <- struct{}{} go func(index int, inputName string) { outputName := conv.DeductOutputPath(inputName) fileLogger := generalLogger.With(slog.String("input_path", inputName), slog.String("output_path", outputName), slog.Int("file_index", index)) @@ -98,9 +100,8 @@ func main() { threadSigTermChannel := make(chan os.Signal, 1) signal.Notify(threadSigTermChannel, os.Interrupt, syscall.SIGTERM) - defer func() { <-semaphore }() - defer wg.Done() - semaphore <- struct{}{} + defer func() { <-queueSemaphore; wg.Done() }() + select { case <-threadSigTermChannel: fileLogger.Info("exiting due to termination signal") @@ -128,6 +129,9 @@ func main() { } } + processSemaphore <- struct{}{} + defer func() { <-processSemaphore }() + fileLogger.Info("start to process file", slog.String("input_hash", inputMetadata.Hash), slog.String("original_input_hash", originalInputHash)) reader, err := inputClient.GetReader(inputName) if err != nil { |