Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion FUTURE.md
Original file line number Diff line number Diff line change
Expand Up @@ -10,7 +10,7 @@
* [x] 支持设置 worker 任务队列大小
* [x] 支持设置 panic 处理函数
* [x] 支持自定义 sync.Locker 实现
* [ ] 完善单元测试,将覆盖率提高到 95% 以上
* [x] 完善单元测试,将覆盖率提高到 95% 以上

### v0.1.0

Expand Down
7 changes: 7 additions & 0 deletions HISTORY.md
Original file line number Diff line number Diff line change
@@ -1,5 +1,12 @@
## ✒ 历史版本的特性介绍 (Features in old versions)

### v0.2.1-alpha

> 此版本发布于 2025-06-25

* 支持选择 sync.Mutex 作为锁实现
* 单元测试覆盖率提高到 95%

### v0.2.0-alpha

> 此版本发布于 2025-06-25
Expand Down
4 changes: 2 additions & 2 deletions _icons/coverage.svg
Loading
Sorry, something went wrong. Reload?
Sorry, we cannot display this file.
Sorry, this file is invalid so it cannot be displayed.
8 changes: 7 additions & 1 deletion executor.go
Original file line number Diff line number Diff line change
Expand Up @@ -8,6 +8,8 @@ import (
"sync"
)

// 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

Expand All @@ -19,6 +21,7 @@ type Executor struct {
lock sync.Locker
}

// NewExecutor creates a new executor with given worker number and options.
func NewExecutor(workerNum int, opts ...Option) *Executor {
conf := newDefaultConfig(workerNum)
for _, opt := range opts {
Expand Down Expand Up @@ -49,7 +52,8 @@ func NewExecutor(workerNum int, opts ...Option) *Executor {
return executor
}

func (e *Executor) Submit(task func()) {
// Submit submits a task to be handled by workers.
func (e *Executor) Submit(task Task) {
e.lock.Lock()
defer e.lock.Unlock()

Expand All @@ -65,10 +69,12 @@ func (e *Executor) Submit(task func()) {
}
}

// Wait waits all tasks to be handled.
func (e *Executor) Wait() {
e.wg.Wait()
}

// Close closes the executor after handling all tasks.
func (e *Executor) Close() {
e.lock.Lock()
defer e.lock.Unlock()
Expand Down
19 changes: 15 additions & 4 deletions option.go
Original file line number Diff line number Diff line change
Expand Up @@ -21,15 +21,26 @@ func WithWorkerQueueSize(size int) Option {
}

// WithRecoverFunc sets the recover function.
func WithRecoverFunc(f func(r any)) Option {
func WithRecoverFunc(recoverFunc func(r any)) Option {
return func(conf *config) {
conf.recoverFunc = f
conf.recoverFunc = recoverFunc
}
}

// WithNewLockerFunc sets the new locker function.
func WithNewLockerFunc(f func() sync.Locker) Option {
func WithNewLockerFunc(newLockerFunc func() sync.Locker) Option {
return func(conf *config) {
conf.newLockerFunc = f
conf.newLockerFunc = newLockerFunc
}
}

// WithSyncMutex sets the new locker function returns sync.Mutex.
func WithSyncMutex() Option {
newLockerFunc := func() sync.Locker {
return new(sync.Mutex)
}

return func(conf *config) {
conf.newLockerFunc = newLockerFunc
}
}
12 changes: 12 additions & 0 deletions option_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -45,3 +45,15 @@ func TestWithNewLockerFunc(t *testing.T) {
t.Fatalf("conf.newLockerFunc %p != newLockerFunc %p", conf.newLockerFunc, newLockerFunc)
}
}

// go test -v -cover -run=^TestWithSyncMutex$
func TestWithSyncMutex(t *testing.T) {
workerNum := 16
conf := newDefaultConfig(workerNum)
WithSyncMutex()(conf)

lock := conf.newLockerFunc()
if _, ok := lock.(*sync.Mutex); !ok {
t.Fatalf("lock %T is not *sync.Mutex", lock)
}
}
7 changes: 3 additions & 4 deletions worker.go
Original file line number Diff line number Diff line change
Expand Up @@ -19,7 +19,7 @@ func newWorker(executor *Executor) *worker {
return w
}

func (w *worker) handle(task func()) {
func (w *worker) handle(task Task) {
defer func() {
if r := recover(); r != nil {
w.executor.conf.recover(r)
Expand All @@ -33,6 +33,7 @@ func (w *worker) work() {
w.executor.wg.Add(1)
go func() {
defer w.executor.wg.Done()
defer close(w.taskQueue)

for task := range w.taskQueue {
if task == nil {
Expand All @@ -41,13 +42,11 @@ func (w *worker) work() {

w.handle(task)
}

close(w.taskQueue)
}()
}

// Accept accepts a task to be handled.
func (w *worker) Accept(task func()) {
func (w *worker) Accept(task Task) {
if task == nil {
return
}
Expand Down
34 changes: 34 additions & 0 deletions worker_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -3,3 +3,37 @@
// license that can be found in the LICENSE file.

package goes

import "testing"

// go test -v -cover -run=^TestWorkerHandle$
func TestWorkerHandle(t *testing.T) {
got := 0
want := 666

executor := &Executor{
conf: &config{
recoverFunc: func(r any) {
got = r.(int)
},
},
}

worker := &worker{executor: executor}
worker.handle(func() {
panic(want)
})

if got != want {
t.Fatalf("got %d != want %d", got, want)
}

want = 123
worker.handle(func() {
got = 123
})

if got != want {
t.Fatalf("got %d != want %d", got, want)
}
}
Loading