v0.2.4-alpha

FUTURE.md

@@ -8,6 +8,7 @@
 * [ ] 支持任务数量优先调度策略
 * [ ] 支持任务时间优先调度策略
 * [ ] 支持 worker 动态扩缩容
+* [x] 支持查询可用的 worker 数量
 * [x] 支持设置 worker 任务队列大小
 * [x] 支持设置 panic 处理函数
 * [x] 支持自定义 sync.Locker 实现

HISTORY.md

@@ -1,5 +1,12 @@
 ## ✒ 历史版本的特性介绍 (Features in old versions)
 
+### v0.2.4-alpha
+
+> 此版本发布于 2025-06-28
+
+* 支持查询可用的 worker 数量
+* 懒启动 worker 机制，避免浪费 worker 资源
+
 ### v0.2.3-alpha
 
 > 此版本发布于 2025-06-28

README.en.md

@@ -15,6 +15,7 @@
 * Limits the number of simultaneous goroutines and reuses them by Executor.
 * Supports multiple scheduling strategies, including round robin, random, etc.
 * Supports spin lock with backoff strategy.
+* Supports getting the number of workers available in the executor.
 
 _Check [HISTORY.md](./HISTORY.md) and [FUTURE.md](./FUTURE.md) to know about more information._
 
@@ -74,11 +75,11 @@ goarch: amd64
 cpu: AMD EPYC 7K62 48-Core Processor
 
 BenchmarkLimiter-2               2417040               498.5 ns/op            24 B/op          1 allocs/op
-BenchmarkExecutor-2             23793781                49.9 ns/op             0 B/op          0 allocs/op
+BenchmarkExecutor-2             20458502                58.3 ns/op             0 B/op          0 allocs/op
 BenchmarkAntsPool-2              4295964               271.7 ns/op             0 B/op          0 allocs/op
 
 BenchmarkLimiterTime-2:  num is 1000000, cost is 300.936441ms
-BenchmarkExecutorTime-2: num is 1000000, cost is  51.350509ms
+BenchmarkExecutorTime-2: num is 1000000, cost is  63.026947ms
 BenchmarkAntsPoolTime-2: num is  999744, cost is 346.972287ms
 ```
 

README.md

@@ -15,6 +15,7 @@
 * 支持限制同时执行的协程数，且复用协程，使用 Executor。
 * 支持多种调度策略，包括轮询、随机等。
 * 支持使用退避策略的自旋锁。
+* 支持查询可用的 worker 数量。
 
 _历史版本的特性请查看 [HISTORY.md](./HISTORY.md)。未来版本的新特性和计划请查看 [FUTURE.md](./FUTURE.md)。_
 
@@ -74,11 +75,11 @@ goarch: amd64
 cpu: AMD EPYC 7K62 48-Core Processor
 
 BenchmarkLimiter-2               2417040               498.5 ns/op            24 B/op          1 allocs/op
-BenchmarkExecutor-2             23793781                49.9 ns/op             0 B/op          0 allocs/op
+BenchmarkExecutor-2             20458502                58.3 ns/op             0 B/op          0 allocs/op
 BenchmarkAntsPool-2              4295964               271.7 ns/op             0 B/op          0 allocs/op
 
 BenchmarkLimiterTime-2:  num is 1000000, cost is 300.936441ms
-BenchmarkExecutorTime-2: num is 1000000, cost is  51.350509ms
+BenchmarkExecutorTime-2: num is 1000000, cost is  63.026947ms
 BenchmarkAntsPoolTime-2: num is  999744, cost is 346.972287ms
 ```
 

config.go

@@ -15,7 +15,7 @@ type config struct {
 	workerQueueSize  int
 	recoverFunc      func(r any)
 	newLockerFunc    func() sync.Locker
-	newSchedulerFunc func(workers []*worker) scheduler
+	newSchedulerFunc func(workers ...*worker) scheduler
 }
 
 func newDefaultConfig(workerNum int) *config {
@@ -42,10 +42,10 @@ func (c *config) newLocker() sync.Locker {
 	return c.newLockerFunc()
 }
 
-func (c *config) newScheduler(workers []*worker) scheduler {
+func (c *config) newScheduler(workers ...*worker) scheduler {
 	if c.newSchedulerFunc == nil {
-		return newRoundRobinScheduler(workers)
+		return newRoundRobinScheduler(workers...)
 	}
 
-	return c.newSchedulerFunc(workers)
+	return c.newSchedulerFunc(workers...)
 }

config_test.go

@@ -68,19 +68,19 @@ func TestConfigNewScheduler(t *testing.T) {
 	conf := newDefaultConfig(workerNum)
 
 	workers := make([]*worker, workerNum)
-	got := conf.newScheduler(workers)
+	got := conf.newScheduler(workers...)
 
 	if _, ok := got.(*roundRobinScheduler); !ok {
 		t.Fatalf("got %T is not *roundRobinScheduler", got)
 	}
 
 	want := &roundRobinScheduler{}
-	conf.newSchedulerFunc = func(workers []*worker) scheduler {
+	conf.newSchedulerFunc = func(workers ...*worker) scheduler {
 		want.workers = workers
 		return want
 	}
 
-	got = conf.newScheduler(workers)
+	got = conf.newScheduler(workers...)
 	if fmt.Sprintf("%p", got) != fmt.Sprintf("%p", want) {
 		t.Fatalf("got %p != want %p", got, want)
 	}

executor.go

@@ -10,8 +10,8 @@ import (
 )
 
 var (
-	ErrExecutorClosed = errors.New("goes: executor is closed")
-	ErrWorkerIsNil    = errors.New("goes: worker is nil")
+	ErrExecutorIsClosed = errors.New("goes: executor is closed")
+	ErrWorkerIsNil      = errors.New("goes: worker is nil")
 )
 
 // Task is a function can be executed by executor.
@@ -45,37 +45,62 @@ func NewExecutor(workerNum int, opts ...Option) *Executor {
 		panic("goes: worker's queue size <= 0")
 	}
 
-	executor := &Executor{
-		conf:   conf,
-		closed: false,
-		lock:   conf.newLocker(),
-	}
-
 	workers := make([]*worker, 0, conf.workerNum)
-	for range conf.workerNum {
-		worker := newWorker(executor)
-		workers = append(workers, worker)
+	executor := &Executor{
+		conf:      conf,
+		workers:   workers,
+		scheduler: conf.newScheduler(),
+		closed:    false,
+		lock:      conf.newLocker(),
 	}
 
-	executor.workers = workers
-	executor.scheduler = conf.newScheduler(workers)
+	executor.spawnWorker()
 	return executor
 }
 
+func (e *Executor) spawnWorker() *worker {
+	worker := newWorker(e)
+
+	e.workers = append(e.workers, worker)
+	e.scheduler.Set(e.workers)
+	return worker
+}
+
+// AvailableWorkers returns the number of workers available in the executor.
+func (e *Executor) AvailableWorkers() int {
+	e.lock.Lock()
+	defer e.lock.Unlock()
+
+	return len(e.workers)
+}
+
 // Submit submits a task to be handled by workers.
 func (e *Executor) Submit(task Task) error {
 	e.lock.Lock()
 	defer e.lock.Unlock()
 
 	if e.closed {
-		return ErrExecutorClosed
+		return ErrExecutorIsClosed
 	}
 
 	worker := e.scheduler.Get()
 	if worker == nil {
 		return ErrWorkerIsNil
 	}
 
+	// We don't need to create a new worker if we got a worker with no tasks.
+	if worker.WaitingTasks() <= 0 || len(e.workers) >= e.conf.workerNum {
+		worker.Accept(task)
+		return nil
+	}
+
+	// The number of workers has reached the limit, so we can only use the worker we got.
+	if len(e.workers) >= e.conf.workerNum {
+		worker.Accept(task)
+		return nil
+	}
+
+	worker = e.spawnWorker()
 	worker.Accept(task)
 	return nil
 }
@@ -94,6 +119,6 @@ func (e *Executor) Close() {
 		worker.Done()
 	}
 
-	e.wg.Wait()
+	e.Wait()
 	e.closed = true
 }

executor_test.go

@@ -55,8 +55,8 @@ func TestExecutorError(t *testing.T) {
 	executor.Close()
 
 	err := executor.Submit(func() {})
-	if err != ErrExecutorClosed {
-		t.Fatalf("err %v != ErrExecutorClosed %v", err, ErrExecutorClosed)
+	if err != ErrExecutorIsClosed {
+		t.Fatalf("err %v != ErrExecutorIsClosed %v", err, ErrExecutorIsClosed)
 	}
 
 	executor = NewExecutor(workerNum)
@@ -67,3 +67,52 @@ func TestExecutorError(t *testing.T) {
 		t.Fatalf("err %v != ErrWorkerIsNil %v", err, ErrWorkerIsNil)
 	}
 }
+
+// go test -v -cover -run=^TestExecutorAvailableWorkers$
+func TestExecutorAvailableWorkers(t *testing.T) {
+	workerNum := 16
+	executor := NewExecutor(workerNum)
+
+	if len(executor.workers) != 1 {
+		t.Fatalf("len(executor.workers) %d != 1", len(executor.workers))
+	}
+
+	if executor.AvailableWorkers() != 1 {
+		t.Fatalf("executor.AvailableWorkers() %d != 1", executor.AvailableWorkers())
+	}
+
+	executor.workers = make([]*worker, workerNum)
+
+	if executor.AvailableWorkers() != workerNum {
+		t.Fatalf("executor.AvailableWorkers() %d != workerNum %d", executor.AvailableWorkers(), workerNum)
+	}
+}
+
+// go test -v -cover -run=^TestExecutorSpawnWorker$
+func TestExecutorSpawnWorker(t *testing.T) {
+	workerNum := 16
+	executor := NewExecutor(workerNum)
+
+	if executor.AvailableWorkers() != 1 {
+		t.Fatalf("executor.AvailableWorkers() %d != 1", executor.AvailableWorkers())
+	}
+
+	for range workerNum {
+		executor.Submit(func() {})
+		time.Sleep(time.Millisecond)
+	}
+
+	if executor.AvailableWorkers() != 1 {
+		t.Fatalf("executor.AvailableWorkers() %d != 1", executor.AvailableWorkers())
+	}
+
+	for range workerNum * 2 {
+		executor.Submit(func() {
+			time.Sleep(time.Millisecond)
+		})
+	}
+
+	if executor.AvailableWorkers() != workerNum {
+		t.Fatalf("executor.AvailableWorkers() %d != workerNum %d", executor.AvailableWorkers(), workerNum)
+	}
+}

option.go

@@ -62,8 +62,8 @@ func WithSyncMutex() Option {
 
 // WithRoundRobinScheduler sets the new scheduler function using round robin strategy.
 func WithRoundRobinScheduler() Option {
-	newSchedulerFunc := func(workers []*worker) scheduler {
-		return newRoundRobinScheduler(workers)
+	newSchedulerFunc := func(workers ...*worker) scheduler {
+		return newRoundRobinScheduler(workers...)
 	}
 
 	return func(conf *config) {
@@ -73,8 +73,8 @@ func WithRoundRobinScheduler() Option {
 
 // WithRandomScheduler sets the new scheduler function using random strategy.
 func WithRandomScheduler() Option {
-	newSchedulerFunc := func(workers []*worker) scheduler {
-		return newRandomScheduler(workers)
+	newSchedulerFunc := func(workers ...*worker) scheduler {
+		return newRandomScheduler(workers...)
 	}
 
 	return func(conf *config) {

option_test.go

@@ -79,7 +79,7 @@ func TestWithRoundRobinScheduler(t *testing.T) {
 	WithRoundRobinScheduler()(conf)
 
 	workers := make([]*worker, 0, workerNum)
-	got := conf.newSchedulerFunc(workers)
+	got := conf.newSchedulerFunc(workers...)
 
 	scheduler, ok := got.(*roundRobinScheduler)
 	if !ok {
@@ -98,7 +98,7 @@ func TestWithRandomScheduler(t *testing.T) {
 	WithRandomScheduler()(conf)
 
 	workers := make([]*worker, 0, workerNum)
-	got := conf.newSchedulerFunc(workers)
+	got := conf.newSchedulerFunc(workers...)
 
 	scheduler, ok := got.(*randomScheduler)
 	if !ok {

random.go

@@ -14,7 +14,7 @@ type randomScheduler struct {
 	random  *rand.Rand
 }
 
-func newRandomScheduler(workers []*worker) *randomScheduler {
+func newRandomScheduler(workers ...*worker) *randomScheduler {
 	scheduler := &randomScheduler{
 		workers: workers,
 		random:  rand.New(rand.NewSource(time.Now().Unix())),

random_test.go

@@ -17,7 +17,7 @@ func TestRandomScheduler(t *testing.T) {
 		workers = append(workers, new(worker))
 	}
 
-	scheduler := newRandomScheduler(workers)
+	scheduler := newRandomScheduler(workers...)
 	if fmt.Sprintf("%p", scheduler.workers) != fmt.Sprintf("%p", workers) {
 		t.Fatalf("scheduler.workers %p != workers %p", scheduler.workers, workers)
 	}

round_robin.go

@@ -9,7 +9,7 @@ type roundRobinScheduler struct {
 	index   int
 }
 
-func newRoundRobinScheduler(workers []*worker) *roundRobinScheduler {
+func newRoundRobinScheduler(workers ...*worker) *roundRobinScheduler {
 	scheduler := &roundRobinScheduler{
 		workers: workers,
 		index:   -1,

round_robin_test.go

@@ -17,7 +17,7 @@ func TestRoundRobinScheduler(t *testing.T) {
 		workers = append(workers, new(worker))
 	}
 
-	scheduler := newRoundRobinScheduler(workers)
+	scheduler := newRoundRobinScheduler(workers...)
 	if fmt.Sprintf("%p", scheduler.workers) != fmt.Sprintf("%p", workers) {
 		t.Fatalf("scheduler.workers %p != workers %p", scheduler.workers, workers)
 	}

worker.go

@@ -27,6 +27,11 @@ func newWorker(executor *Executor) *worker {
 	return w
 }
 
+// WaitingTasks returns the number of tasks waiting in the worker.
+func (w *worker) WaitingTasks() int {
+	return len(w.taskQueue)
+}
+
 func (w *worker) handle(task Task) {
 	defer func() {
 		if r := recover(); r != nil {

worker_test.go

@@ -37,3 +37,28 @@ func TestWorkerHandle(t *testing.T) {
 		t.Fatalf("got %d != want %d", got, want)
 	}
 }
+
+// go test -v -cover -run=^TestWorkerWaitingTasks$
+func TestWorkerWaitingTasks(t *testing.T) {
+	taskQueue := make(chan Task, 4)
+	worker := &worker{taskQueue: taskQueue}
+
+	if worker.WaitingTasks() != len(taskQueue) {
+		t.Fatalf("got %d != want %d", worker.WaitingTasks(), len(taskQueue))
+	}
+
+	if worker.WaitingTasks() != 0 {
+		t.Fatalf("got %d != 0", worker.WaitingTasks())
+	}
+
+	taskQueue <- nil
+	taskQueue <- nil
+
+	if worker.WaitingTasks() != len(taskQueue) {
+		t.Fatalf("got %d != want %d", worker.WaitingTasks(), len(taskQueue))
+	}
+
+	if worker.WaitingTasks() != 2 {
+		t.Fatalf("got %d != 2", worker.WaitingTasks())
+	}
+}