Allow configuring processor concurrency (#281)

bitmagnet.io/setup/configuration.md

@@ -17,9 +17,10 @@ nav_order: 2
 - `log.level` (default: `info`): If you're developing or just curious then you may want to set this to `debug`; note that `debug` output will be very verbose.
 - `log.development` (default: `false`): If you're developing you may want to enable this flag to enable more verbose output such as stack traces.
 - `log.json` (default: `false`): By default logs are output in a pretty format with colors; enable this flag if you'd prefer plain JSON.
-- `log.file_rotator.enabled` (default: `false`): If true, logs will be output to rotating log files at level `log.file_rotator.level` in the `log.file_rotator.path` directory, allowing forwarding to a logs aggregator (see [the observability guide](/internals-development/observability-telemetry.html)).
+- `log.file_rotator.enabled` (default: `false`): If true, logs will be output to rotating log files at level `log.file_rotator.level` in the `log.file_rotator.path` directory, allowing forwarding to a logs aggregator (see [the observability guide](/guides/observability-telemetry.html)).
 - `http_server.options` (default `["*"]`): A list of enabled HTTP server components. By default all are enabled. Components include: `cors`, `pprof`, `graphql`, `import`, `prometheus`, `torznab`, `status`, `webui`.
 - `dht_crawler.scaling_factor` (default: `10`): There are various rate and concurrency limits associated with the DHT crawler. This parameter is a rough proxy for resource usage of the crawler; concurrency and buffer size of the various pipeline channels are multiplied by this value. Diminishing returns may result from exceeding the default value of 10. Since the software has not been tested on a wide variety of hardware and network conditions your mileage may vary here...
+- `processor.concurrency` (default: `3`): Defines the maximum number of torrents to be processed/classified simultaneously. If you experience slowdowns when the queue is working, try decreasing this to 1; conversely, if running on more powerful hardware and you'd like to work through the queue more quickly, try increasing the value.
 
 To see a full list of available configuration options using the CLI, run:
 

internal/processor/config.go

@@ -0,0 +1,11 @@
+package processor
+
+type Config struct {
+	Concurrency uint
+}
+
+func NewDefaultConfig() Config {
+	return Config{
+		Concurrency: 3,
+	}
+}

internal/processor/processorfx/module.go

@@ -1,6 +1,7 @@
 package processorfx
 
 import (
+	"github.com/bitmagnet-io/bitmagnet/internal/boilerplate/config/configfx"
 	"github.com/bitmagnet-io/bitmagnet/internal/processor"
 	batchqueue "github.com/bitmagnet-io/bitmagnet/internal/processor/batch/queue"
 	"github.com/bitmagnet-io/bitmagnet/internal/processor/hook_0_9_0"
@@ -11,6 +12,7 @@ import (
 func New() fx.Option {
 	return fx.Module(
 		"processor",
+		configfx.NewConfigModule[processor.Config]("processor", processor.NewDefaultConfig()),
 		fx.Provide(
 			processor.New,
 			processorqueue.New,

internal/processor/queue/handler.go

@@ -13,6 +13,7 @@ import (
 
 type Params struct {
 	fx.In
+	Config    processor.Config
 	Processor lazy.Lazy[processor.Processor]
 }
 
@@ -34,7 +35,7 @@ func New(p Params) Result {
 					return err
 				}
 				return pr.Process(ctx, *msg)
-			}, handler.JobTimeout(time.Second*60*10), handler.Concurrency(3)), nil
+			}, handler.JobTimeout(time.Second*60*10), handler.Concurrency(int(p.Config.Concurrency))), nil
 		}),
 	}
 }