feat: options added to configure no of workers (#262) (#263)

Author: mergify[bot]

cmd/config.cluster.yml

@@ -55,6 +55,7 @@ pubsub:
 task_queue:
   mode: remote # local or remote
   max_outstanding_messages_per_queue: 1000
+  no_of_workers_per_queue: 2 # this is lightweight goroutine based worker, so you can increase it as per your requirement
   amqp: # all the info should be filled if mode is remote
     protocol: amqp # amqp or amqps
     host: localhost

cmd/config.standalone.yml

@@ -55,6 +55,7 @@ pubsub:
 task_queue:
   mode: local # local or remote
   max_outstanding_messages_per_queue: 1000
+  no_of_workers_per_queue: 1 # this is lightweight goroutine based worker, so you can increase it as per your requirement
   amqp: # all the info should be filled if mode is remote
     protocol: amqp # amqp or amqps
     host: localhost

swiftwave_service/core/load_service_manager.go

@@ -98,6 +98,7 @@ func (manager *ServiceManager) Load(config system_config.Config) {
 			Type:                task_queue.Local,
 			Mode:                task_queue.Both, // TODO: option to configure this
 			MaxMessagesPerQueue: config.TaskQueueConfig.MaxOutstandingMessagesPerQueue,
+			NoOfWorkersPerQueue: config.TaskQueueConfig.NoOfWorkersPerQueue,
 		})
 		if err != nil {
 			panic(err)
@@ -107,6 +108,7 @@ func (manager *ServiceManager) Load(config system_config.Config) {
 		taskQueueClient, err := task_queue.NewClient(task_queue.Options{
 			Type:                task_queue.Remote,
 			Mode:                task_queue.Both, // TODO: option to configure this
+			NoOfWorkersPerQueue: config.TaskQueueConfig.NoOfWorkersPerQueue,
 			MaxMessagesPerQueue: config.TaskQueueConfig.MaxOutstandingMessagesPerQueue,
 			AMQPUri:             config.TaskQueueConfig.AMQPConfig.URI(),
 			AMQPVhost:           config.TaskQueueConfig.AMQPConfig.VHost,

system_config/types.go

@@ -72,6 +72,7 @@ type PubSubConfig struct {
 type TaskQueueConfig struct {
 	Mode                           TaskQueueMode `yaml:"mode"`
 	MaxOutstandingMessagesPerQueue int           `yaml:"max_outstanding_messages_per_queue"`
+	NoOfWorkersPerQueue            int           `yaml:"no_of_workers_per_queue"`
 	AMQPConfig                     AMQPConfig    `yaml:"amqp"`
 }
 

task_queue/init.go

@@ -37,6 +37,7 @@ func createLocalTaskQueueClient(options Options) (Client, error) {
 		queueToChannelMapping:       channelsMapping,
 		operationMode:               options.Mode,
 		maxMessagesPerQueue:         options.MaxMessagesPerQueue,
+		NoOfWorkersPerQueue:         options.NoOfWorkersPerQueue,
 		consumersWaitGroup:          &sync.WaitGroup{},
 	}, nil
 }
@@ -56,6 +57,7 @@ func createRemoteTaskQueueClient(options Options) (Client, error) {
 
 	return &remoteTaskQueue{
 		mutexQueueToFunctionMapping: mutex,
+		NoOfWorkersPerQueue:         options.NoOfWorkersPerQueue,
 		queueToFunctionMapping:      functionsMapping,
 		operationMode:               options.Mode,
 		amqpURI:                     options.AMQPUri,

task_queue/local_task_queue.go

@@ -90,8 +90,10 @@ func (l *localTaskQueue) StartConsumers(nowait bool) error {
 
 	// start consumers
 	for _, queueName := range queueNames {
-		wg.Add(1)
-		go l.listenForTasks(queueName, wg)
+		for i := 1; i <= l.NoOfWorkersPerQueue; i++ {
+			wg.Add(1)
+			go l.listenForTasks(queueName, wg)
+		}
 	}
 
 	if !nowait {

task_queue/remote_task_queue.go

@@ -124,8 +124,10 @@ func (r *remoteTaskQueue) StartConsumers(nowait bool) error {
 
 	// start consumers
 	for _, queueName := range queueNames {
-		wg.Add(1)
-		go r.listenForTasks(queueName, wg)
+		for i := 1; i <= r.NoOfWorkersPerQueue; i++ {
+			wg.Add(1)
+			go r.listenForTasks(queueName, wg)
+		}
 	}
 
 	if !nowait {

task_queue/types.go

@@ -27,6 +27,7 @@ type localTaskQueue struct {
 	queueToChannelMapping       map[string]chan ArgumentType
 	operationMode               Mode
 	maxMessagesPerQueue         int
+	NoOfWorkersPerQueue         int
 	consumersWaitGroup          *sync.WaitGroup
 }
 
@@ -38,6 +39,7 @@ type remoteTaskQueue struct {
 	amqpClientName              string
 	operationMode               Mode
 	consumersWaitGroup          *sync.WaitGroup
+	NoOfWorkersPerQueue         int
 	// internal use
 	amqpConnection *amqp.Connection
 	amqpChannel    *amqp.Channel
@@ -69,6 +71,7 @@ type Options struct {
 	Type                ServiceType
 	Mode                Mode
 	MaxMessagesPerQueue int // only applicable for local task queue
+	NoOfWorkersPerQueue int
 	// Extra options for remote task queue
 	AMQPUri        string
 	AMQPVhost      string