Merge pull request #5 from FishGoddess/develop

v0.2.2-alpha

Author: FishGoddess

FUTURE.md

@@ -4,7 +4,7 @@
 
 * [x] 增加异步任务执行器
 * [x] 支持轮询调度策略
-* [ ] 支持随机调度策略
+* [x] 支持随机调度策略
 * [ ] 支持任务数量优先调度策略
 * [ ] 支持任务时间优先调度策略
 * [x] 支持设置 worker 任务队列大小

HISTORY.md

@@ -1,5 +1,12 @@
 ## ✒ 历史版本的特性介绍 (Features in old versions)
 
+### v0.2.2-alpha
+
+> 此版本发布于 2025-06-26
+
+* 抽象出 workers 接口层
+* 增加随机调度策略
+
 ### v0.2.1-alpha
 
 > 此版本发布于 2025-06-25

README.en.md

@@ -11,9 +11,10 @@
 
 ### 🥇 Features
 
-* Supports spin lock with backoff strategy.
 * Limits the number of simultaneous goroutines and not reuses them by Limiter.
 * 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.
 
 _Check [HISTORY.md](./HISTORY.md) and [FUTURE.md](./FUTURE.md) to know about more information._
 

README.md

@@ -11,9 +11,10 @@
 
 ### 🥇 功能特性
 
-* 内置退避策略的自旋锁。
 * 支持只限制同时执行的协程数，但不复用协程，使用 Limiter。
 * 支持限制同时执行的协程数，且复用协程，使用 Executor。
+* 支持多种调度策略，包括轮询、随机等。
+* 支持使用退避策略的自旋锁。
 
 _历史版本的特性请查看 [HISTORY.md](./HISTORY.md)。未来版本的新特性和计划请查看 [FUTURE.md](./FUTURE.md)。_
 

_icons/coverage.svg

@@ -15,6 +15,7 @@ type config struct {
 	workerQueueSize int
 	recoverFunc     func(r any)
 	newLockerFunc   func() sync.Locker
+	newWorkersFunc  func(workerNum int) workers
 }
 
 func newDefaultConfig(workerNum int) *config {
@@ -23,6 +24,7 @@ func newDefaultConfig(workerNum int) *config {
 		workerQueueSize: 64,
 		recoverFunc:     nil,
 		newLockerFunc:   nil,
+		newWorkersFunc:  nil,
 	}
 }
 
@@ -39,3 +41,11 @@ func (c *config) newLocker() sync.Locker {
 
 	return c.newLockerFunc()
 }
+
+func (c *config) newWorkers() workers {
+	if c.newWorkersFunc == nil {
+		return newRoundRobinWorkers(c.workerNum)
+	}
+
+	return c.newWorkersFunc(c.workerNum)
+}

config.go

@@ -8,6 +8,8 @@ import (
 	"fmt"
 	"sync"
 	"testing"
+
+	"github.com/FishGoddess/goes/pkg/spinlock"
 )
 
 // go test -v -cover -run=^TestNewDefaultConfig$
@@ -43,16 +45,50 @@ func TestConfigRecover(t *testing.T) {
 func TestConfigNewLocker(t *testing.T) {
 	workerNum := 16
 	conf := newDefaultConfig(workerNum)
-	conf.newLocker()
+
+	got := conf.newLocker()
+	if _, ok := got.(*spinlock.Lock); !ok {
+		t.Fatalf("got %T is not *spinlock.Lock", got)
+	}
 
 	want := &sync.Mutex{}
 	conf.newLockerFunc = func() sync.Locker {
 		return want
 	}
 
-	got := conf.newLocker()
+	got = conf.newLocker()
+	if fmt.Sprintf("%p", got) != fmt.Sprintf("%p", want) {
+		t.Fatalf("got %p != want %p", got, want)
+	}
+}
 
+// go test -v -cover -run=^TestConfigNewWorkers$
+func TestConfigNewWorkers(t *testing.T) {
+	workerNum := 16
+	conf := newDefaultConfig(workerNum)
+
+	got := conf.newWorkers()
+	if _, ok := got.(*roundRobinWorkers); !ok {
+		t.Fatalf("got %T is not *roundRobinWorkers", got)
+	}
+
+	want := &roundRobinWorkers{}
+	conf.newWorkersFunc = func(workerNum int) workers {
+		want.workers = make([]*worker, workerNum)
+		return want
+	}
+
+	got = conf.newWorkers()
 	if fmt.Sprintf("%p", got) != fmt.Sprintf("%p", want) {
 		t.Fatalf("got %p != want %p", got, want)
 	}
+
+	rrWorkers, ok := got.(*roundRobinWorkers)
+	if !ok {
+		t.Fatalf("got %T is not *roundRobinWorkers", got)
+	}
+
+	if len(rrWorkers.workers) != workerNum {
+		t.Fatalf("len(rrWorkers.workers) %d != workerNum %d", len(rrWorkers.workers), workerNum)
+	}
 }

config_test.go

@@ -5,16 +5,24 @@
 package goes
 
 import (
+	"errors"
 	"sync"
 )
 
+var (
+	ErrExecutorClosed = errors.New("goes: executor is closed")
+	ErrWorkerIsNil    = errors.New("goes: worker is nil")
+)
+
+// Task is a function can be executed by executor.
+type Task = func()
+
 // Executor executes tasks concurrently using limited goroutines.
 // You can specify the number of workers and the queue size of each worker.
 type Executor struct {
 	conf *config
 
-	workers []*worker
-	index   int
+	workers workers
 	closed  bool
 
 	wg   sync.WaitGroup
@@ -38,35 +46,35 @@ func NewExecutor(workerNum int, opts ...Option) *Executor {
 
 	executor := &Executor{
 		conf:    conf,
-		workers: make([]*worker, 0, conf.workerNum),
-		index:   0,
+		workers: conf.newWorkers(),
 		closed:  false,
 		lock:    conf.newLocker(),
 	}
 
 	for range conf.workerNum {
 		worker := newWorker(executor)
-		executor.workers = append(executor.workers, worker)
+		executor.workers.Add(worker)
 	}
 
 	return executor
 }
 
 // Submit submits a task to be handled by workers.
-func (e *Executor) Submit(task Task) {
+func (e *Executor) Submit(task Task) error {
 	e.lock.Lock()
 	defer e.lock.Unlock()
 
 	if e.closed {
-		return
+		return ErrExecutorClosed
 	}
 
-	worker := e.workers[e.index]
-	worker.Accept(task)
-
-	if e.index++; e.index >= len(e.workers) {
-		e.index = 0
+	worker := e.workers.Next()
+	if worker == nil {
+		return ErrWorkerIsNil
 	}
+
+	worker.Accept(task)
+	return nil
 }
 
 // Wait waits all tasks to be handled.
@@ -79,10 +87,7 @@ func (e *Executor) Close() {
 	e.lock.Lock()
 	defer e.lock.Unlock()
 
-	for _, worker := range e.workers {
-		worker.Done()
-	}
-
+	e.workers.Done()
 	e.wg.Wait()
 	e.closed = true
 }

executor.go

@@ -47,3 +47,23 @@ func TestExecutor(t *testing.T) {
 		t.Fatalf("gotTotalCount %d != totalCount %d", gotTotalCount, totalCount)
 	}
 }
+
+// go test -v -cover -run=^TestExecutorError$
+func TestExecutorError(t *testing.T) {
+	workerNum := 16
+	executor := NewExecutor(workerNum)
+	executor.Close()
+
+	err := executor.Submit(func() {})
+	if err != ErrExecutorClosed {
+		t.Fatalf("err %v != ErrExecutorClosed %v", err, ErrExecutorClosed)
+	}
+
+	executor = NewExecutor(workerNum)
+	executor.workers = &roundRobinWorkers{}
+
+	err = executor.Submit(func() {})
+	if err != ErrWorkerIsNil {
+		t.Fatalf("err %v != ErrWorkerIsNil %v", err, ErrWorkerIsNil)
+	}
+}

executor_test.go

@@ -4,7 +4,11 @@
 
 package goes
 
-import "sync"
+import (
+	"sync"
+
+	"github.com/FishGoddess/goes/pkg/spinlock"
+)
 
 // Option is for setting config.
 type Option func(conf *config)
@@ -34,7 +38,18 @@ func WithNewLockerFunc(newLockerFunc func() sync.Locker) Option {
 	}
 }
 
-// WithSyncMutex sets the new locker function returns sync.Mutex.
+// WithSpinLock sets the new locker function returns spin lock.
+func WithSpinLock() Option {
+	newLockerFunc := func() sync.Locker {
+		return spinlock.New()
+	}
+
+	return func(conf *config) {
+		conf.newLockerFunc = newLockerFunc
+	}
+}
+
+// WithSyncMutex sets the new locker function returns sync mutex.
 func WithSyncMutex() Option {
 	newLockerFunc := func() sync.Locker {
 		return new(sync.Mutex)
@@ -44,3 +59,25 @@ func WithSyncMutex() Option {
 		conf.newLockerFunc = newLockerFunc
 	}
 }
+
+// WithRoundRobinWorkers sets the new workers function using round robin strategy.
+func WithRoundRobinWorkers() Option {
+	newWorkersFunc := func(workerNum int) workers {
+		return newRoundRobinWorkers(workerNum)
+	}
+
+	return func(conf *config) {
+		conf.newWorkersFunc = newWorkersFunc
+	}
+}
+
+// WithRandomWorkers sets the new workers function using random strategy.
+func WithRandomWorkers() Option {
+	newWorkersFunc := func(workerNum int) workers {
+		return newRandomWorkers(workerNum)
+	}
+
+	return func(conf *config) {
+		conf.newWorkersFunc = newWorkersFunc
+	}
+}