purge

Author: FishGoddess

FUTURE.md

@@ -12,6 +12,8 @@
 * [x] 支持设置 worker 任务队列大小
 * [x] 支持设置 panic 处理函数
 * [x] 支持自定义 sync.Locker 实现
+* [ ] 增加 worker 内存复用池
+* [ ] 增加 workers 切片内存复用池
 * [x] 完善单元测试，将覆盖率提高到 95% 以上
 
 ### v0.1.0

config.go

@@ -14,6 +14,8 @@ import (
 type config struct {
 	workerNum        int
 	workerQueueSize  int
+	workerLifetime   time.Duration
+	purgeInterval    time.Duration
 	nowFunc          func() time.Time
 	recoverFunc      func(r any)
 	newLockerFunc    func() sync.Locker
@@ -24,6 +26,8 @@ func newDefaultConfig(workerNum int) *config {
 	return &config{
 		workerNum:        workerNum,
 		workerQueueSize:  64,
+		workerLifetime:   0,
+		purgeInterval:    0,
 		nowFunc:          nil,
 		recoverFunc:      nil,
 		newLockerFunc:    nil,

executor.go

@@ -7,6 +7,7 @@ package goes
 import (
 	"errors"
 	"sync"
+	"time"
 )
 
 var (
@@ -24,6 +25,7 @@ type Executor struct {
 
 	workers   []*worker
 	scheduler scheduler
+	closeCh   chan struct{}
 	closed    bool
 
 	wg   sync.WaitGroup
@@ -50,11 +52,13 @@ func NewExecutor(workerNum int, opts ...Option) *Executor {
 		conf:      conf,
 		workers:   workers,
 		scheduler: conf.newScheduler(),
+		closeCh:   make(chan struct{}, 1),
 		closed:    false,
 		lock:      conf.newLocker(),
 	}
 
 	executor.spawnWorker()
+	executor.runPurgeTask()
 	return executor
 }
 
@@ -66,6 +70,54 @@ func (e *Executor) spawnWorker() *worker {
 	return worker
 }
 
+func (e *Executor) purgeActive() bool {
+	return e.conf.purgeInterval > 0 && e.conf.workerLifetime > 0
+}
+
+func (e *Executor) purgeWorkers() {
+	e.lock.Lock()
+	defer e.lock.Unlock()
+
+	now := e.conf.now()
+	oldWorkers := e.workers
+	newWorkers := make([]*worker, 0, len(oldWorkers))
+	for _, worker := range oldWorkers {
+		if worker.WaitingTasks() <= 0 && now.Sub(worker.AcceptTime()) >= e.conf.workerLifetime {
+			worker.Done()
+		} else {
+			newWorkers = append(newWorkers, worker)
+		}
+	}
+
+	e.workers = newWorkers
+	e.scheduler.Set(e.workers)
+
+	// Avoid gc leaks.
+	for i := range oldWorkers {
+		oldWorkers[i] = nil
+	}
+}
+
+func (e *Executor) runPurgeTask() {
+	if !e.purgeActive() {
+		return
+	}
+
+	go func() {
+		ticker := time.NewTicker(e.conf.purgeInterval)
+		defer ticker.Stop()
+
+		for {
+			select {
+			case <-ticker.C:
+				e.purgeWorkers()
+			case <-e.closeCh:
+				return
+			}
+		}
+	}()
+}
+
 // AvailableWorkers returns the number of workers available.
 func (e *Executor) AvailableWorkers() int {
 	e.lock.Lock()
@@ -97,9 +149,11 @@ func (e *Executor) Submit(task Task) error {
 		worker = e.spawnWorker()
 	}
 
-	worker.SetAcceptTime(e.conf.now())
-	e.lock.Unlock()
+	if e.purgeActive() {
+		worker.SetAcceptTime(e.conf.now())
+	}
 
+	e.lock.Unlock()
 	worker.Accept(task)
 	return nil
 }