From 8d66639cdc4c76f9a2f0b936ef3a7eea1e283481 Mon Sep 17 00:00:00 2001
From: Paul Pan <i@0x7f.app>
Date: Sat, 6 Jan 2024 15:06:12 +0800
Subject: [PATCH] feat: #6 [1] Add pool in runner

---
 internal/app/runner/runner.go      | 4 +---
 internal/service/runner/service.go | 7 +++++++
 2 files changed, 8 insertions(+), 3 deletions(-)

diff --git a/internal/app/runner/runner.go b/internal/app/runner/runner.go
index 4884196..78cda77 100644
--- a/internal/app/runner/runner.go
+++ b/internal/app/runner/runner.go
@@ -6,12 +6,10 @@ import (
 	"git.0x7f.app/WOJ/woj-server/internal/misc/config"
 	"git.0x7f.app/WOJ/woj-server/internal/misc/log"
 	"git.0x7f.app/WOJ/woj-server/internal/model"
-	"git.0x7f.app/WOJ/woj-server/pkg/utils"
 	"git.0x7f.app/WOJ/woj-server/pkg/zapasynq"
 	"github.com/hibiken/asynq"
 	"github.com/samber/do"
 	"go.uber.org/zap"
-	"runtime"
 )
 
 func RunRunner(i *do.Injector) error {
@@ -34,7 +32,7 @@ func RunRunner(i *do.Injector) error {
 			DB:       conf.Redis.QueueDb,
 		},
 		asynq.Config{
-			Concurrency: utils.If(runtime.NumCPU() > 1, runtime.NumCPU()-1, 1),
+			Concurrency: 1, // there's a worker pool in runner service
 			Logger:      zapasynq.New(rlog),
 			Queues:      map[string]int{model.QueueRunner: 1},
 		},
diff --git a/internal/service/runner/service.go b/internal/service/runner/service.go
index 820c214..112420d 100644
--- a/internal/service/runner/service.go
+++ b/internal/service/runner/service.go
@@ -4,8 +4,11 @@ import (
 	"git.0x7f.app/WOJ/woj-server/internal/e"
 	"git.0x7f.app/WOJ/woj-server/internal/misc/config"
 	"git.0x7f.app/WOJ/woj-server/internal/misc/log"
+	"git.0x7f.app/WOJ/woj-server/pkg/pool"
+	"git.0x7f.app/WOJ/woj-server/pkg/utils"
 	"github.com/samber/do"
 	"go.uber.org/zap"
+	"runtime"
 )
 
 var _ Service = (*service)(nil)
@@ -30,14 +33,18 @@ type Service interface {
 }
 
 func NewService(i *do.Injector) (Service, error) {
+	concurrency := utils.If(runtime.NumCPU() > 1, runtime.NumCPU()-1, 1)
+
 	return &service{
 		log:     do.MustInvoke[log.Service](i).GetLogger("runner"),
+		pool:    pool.NewTaskPool(concurrency, concurrency),
 		verbose: do.MustInvoke[config.Service](i).GetConfig().Development,
 	}, nil
 }
 
 type service struct {
 	log     *zap.Logger
+	pool    *pool.TaskPool
 	verbose bool
 }