动态增加 worker

Author: FishGoddess

README.en.md

@@ -75,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

@@ -75,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
 ```
 

_icons/coverage.svg

@@ -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.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)
 	}

config_test.go

@@ -45,23 +45,27 @@ 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()
@@ -84,6 +88,19 @@ func (e *Executor) Submit(task Task) error {
 		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
 }

executor.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.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 {

option_test.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.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)
 	}