Merge pull request #1 from lxzan/dev

Zero Allocs FIFO

.github/workflows/go.yml

@@ -22,3 +22,5 @@ jobs:
         go-version: 1.20.3
     - name: Test
       run: go test -v ./...
+    - name: Bench
+      run: make bench

.gitignore

@@ -16,5 +16,3 @@ vendor/
 examples/
 .idea/
 .vacode/
-go.work
-go.work.sum

Makefile

@@ -1,5 +1,5 @@
 test:
-	go test ./...
+	go test -count=1 ./...
 
 bench:
 	go test -benchmem -run=^$$ -bench . github.com/lxzan/concurrency/benchmark

README.md

@@ -8,7 +8,7 @@
 ### Install
 
 ```bash
-GOPROXY=https://goproxy.cn go get -v github.com/lxzan/concurrency@latest
+go get -v github.com/lxzan/concurrency@latest
 ```
 
 #### Usage
@@ -52,6 +52,7 @@ func main() {
 package main
 
 import (
+	"context"
 	"fmt"
 	"github.com/lxzan/concurrency/queues"
 	"sync/atomic"
@@ -67,7 +68,7 @@ func main() {
 			atomic.AddInt64(&sum, x)
 		})
 	}
-	w.Stop()
+	w.Stop(context.Background())
 	fmt.Printf("sum=%d\n", sum)
 }
 ```
@@ -80,16 +81,16 @@ func main() {
 
 ```
 go test -benchmem -run=^$ -bench . github.com/lxzan/concurrency/benchmark
-goos: darwin
-goarch: arm64
+goos: linux
+goarch: amd64
 pkg: github.com/lxzan/concurrency/benchmark
-Benchmark_Fib-8                  1534509               775.5 ns/op             0 B/op          0 allocs/op
-Benchmark_StdGo-8                    390           3078647 ns/op          160585 B/op      10002 allocs/op
-Benchmark_QueuesSingle-8             262           4388264 ns/op          345144 B/op      10898 allocs/op
-Benchmark_QueuesMultiple-8           470           2630718 ns/op          323923 B/op      10964 allocs/op
-Benchmark_Ants-8                     178           6708482 ns/op          160374 B/op      10004 allocs/op
-Benchmark_GoPool-8                   348           3487154 ns/op          194926 B/op      10511 allocs/op
+cpu: AMD Ryzen 5 PRO 4650G with Radeon Graphics
+Benchmark_Fib-12                 1000000              1146 ns/op               0 B/op          0 allocs/op
+Benchmark_StdGo-12                  3661            317905 ns/op           16064 B/op       1001 allocs/op
+Benchmark_QueuesSingle-12           2178            532224 ns/op           67941 B/op       1098 allocs/op
+Benchmark_QueuesMultiple-12         3691            317757 ns/op           61648 B/op       1256 allocs/op
+Benchmark_Ants-12                   1569            751802 ns/op           22596 B/op       1097 allocs/op
+Benchmark_GoPool-12                 2910            406935 ns/op           19042 B/op       1093 allocs/op
 PASS
-ok      github.com/lxzan/concurrency/benchmark  10.107s
-
+ok      github.com/lxzan/concurrency/benchmark  7.271s
 ```

benchmark/benchmark_test.go

@@ -10,10 +10,17 @@ import (
 
 const (
 	Concurrency = 16
-	M           = 10000
+	M           = 1000
 	N           = 13
 )
 
+func newJob(wg *sync.WaitGroup) func() {
+	return func() {
+		fib(N)
+		wg.Done()
+	}
+}
+
 func Benchmark_Fib(b *testing.B) {
 	for i := 0; i < b.N; i++ {
 		fib(N)
@@ -24,43 +31,43 @@ func Benchmark_StdGo(b *testing.B) {
 	for i := 0; i < b.N; i++ {
 		wg := &sync.WaitGroup{}
 		wg.Add(M)
+		job := newJob(wg)
 		for j := 0; j < M; j++ {
-			go func() {
-				fib(N)
-				wg.Done()
-			}()
+			go job()
 		}
 		wg.Wait()
 	}
 }
 
 func Benchmark_QueuesSingle(b *testing.B) {
-	q := queues.New(queues.WithConcurrency(Concurrency))
+	q := queues.New(
+		queues.WithConcurrency(Concurrency),
+		queues.WithSharding(1),
+	)
 
 	for i := 0; i < b.N; i++ {
 		wg := &sync.WaitGroup{}
 		wg.Add(M)
+		job := newJob(wg)
 		for j := 0; j < M; j++ {
-			q.Push(func() {
-				fib(N)
-				wg.Done()
-			})
+			q.Push(job)
 		}
 		wg.Wait()
 	}
 }
 
 func Benchmark_QueuesMultiple(b *testing.B) {
-	q := queues.New(queues.WithConcurrency(1), queues.WithMultiple(Concurrency))
+	q := queues.New(
+		queues.WithConcurrency(1),
+		queues.WithSharding(Concurrency),
+	)
 
 	for i := 0; i < b.N; i++ {
 		wg := &sync.WaitGroup{}
 		wg.Add(M)
+		job := newJob(wg)
 		for j := 0; j < M; j++ {
-			q.Push(func() {
-				fib(N)
-				wg.Done()
-			})
+			q.Push(job)
 		}
 		wg.Wait()
 	}
@@ -73,11 +80,9 @@ func Benchmark_Ants(b *testing.B) {
 	for i := 0; i < b.N; i++ {
 		wg := &sync.WaitGroup{}
 		wg.Add(M)
+		job := newJob(wg)
 		for j := 0; j < M; j++ {
-			q.Submit(func() {
-				fib(N)
-				wg.Done()
-			})
+			q.Submit(job)
 		}
 		wg.Wait()
 	}
@@ -89,11 +94,9 @@ func Benchmark_GoPool(b *testing.B) {
 	for i := 0; i < b.N; i++ {
 		wg := &sync.WaitGroup{}
 		wg.Add(M)
+		job := newJob(wg)
 		for j := 0; j < M; j++ {
-			q.Go(func() {
-				fib(N)
-				wg.Done()
-			})
+			q.Go(job)
 		}
 		wg.Wait()
 	}

go.work

@@ -0,0 +1,6 @@
+go 1.20
+
+use (
+	.
+	./benchmark
+)

groups/group.go

@@ -13,16 +13,7 @@ const (
 	defaultWaitTimeout = 60 * time.Second // 默认线程同步等待超时
 )
 
-var ErrWaitTimeout = errors.New("wait timeout")
-
 type (
-	options struct {
-		timeout     time.Duration
-		concurrency int64
-		caller      Caller
-		cancel      bool
-	}
-
 	Caller func(args any, f func(any) error) error
 
 	Group[T any] struct {
@@ -34,6 +25,7 @@ type (
 		taskDone  int64              // 已完成任务数量
 		taskTotal int64              // 总任务数量
 		OnMessage func(args T) error // 任务处理
+		OnError   func(err error)    // 错误处理
 	}
 )
 
@@ -58,6 +50,7 @@ func New[T any](opts ...Option) *Group[T] {
 	c.OnMessage = func(args T) error {
 		return nil
 	}
+	c.OnError = func(err error) {}
 	return c
 }
 
@@ -105,6 +98,7 @@ func (c *Group[T]) do(args T) {
 		c.mu.Lock()
 		c.errs = append(c.errs, err)
 		c.mu.Unlock()
+		c.OnError(err)
 	}
 
 	if c.incrAndIsDone() {
@@ -161,6 +155,6 @@ func (c *Group[T]) Start() error {
 		return errors.Join(c.errs...)
 	case <-ctx.Done():
 		c.clear()
-		return ErrWaitTimeout
+		return ctx.Err()
 	}
 }

groups/options.go

@@ -7,6 +7,13 @@ import (
 	"unsafe"
 )
 
+type options struct {
+	timeout     time.Duration
+	concurrency int64
+	caller      Caller
+	cancel      bool
+}
+
 type Option func(o *options)
 
 // WithTimeout 设置任务超时时间

internal/queue.go

@@ -0,0 +1,126 @@
+package internal
+
+type (
+	pointer uint32
+
+	element[T any] struct {
+		prev, addr, next pointer
+		Value            T
+	}
+
+	Queue[T any] struct {
+		head, tail pointer      // 头尾指针
+		length     int          // 长度
+		stack      stack        // 回收站
+		elements   []element[T] // 元素列表
+		template   element[T]   // 空值模板
+	}
+)
+
+func NewQueue[T any](capacity uint32) *Queue[T] {
+	return &Queue[T]{elements: make([]element[T], 1, 1+capacity)}
+}
+
+func (c *Queue[T]) get(addr pointer) *element[T] {
+	if addr > 0 {
+		return &(c.elements[addr])
+	}
+	return nil
+}
+
+func (c *Queue[T]) getElement() *element[T] {
+	if c.stack.Len() > 0 {
+		addr := c.stack.Pop()
+		v := c.get(addr)
+		v.addr = addr
+		return v
+	}
+
+	addr := pointer(len(c.elements))
+	c.elements = append(c.elements, c.template)
+	v := c.get(addr)
+	v.addr = addr
+	return v
+}
+
+func (c *Queue[T]) Reset() {
+	c.head, c.tail, c.length = 0, 0, 0
+	c.stack = c.stack[:0]
+	c.elements = c.elements[:1]
+}
+
+func (c *Queue[T]) Len() int {
+	return c.length
+}
+
+func (c *Queue[T]) Front() (value T) {
+	if c.head > 0 {
+		return c.get(c.head).Value
+	}
+	return value
+}
+
+func (c *Queue[T]) Push(value T) {
+	ele := c.getElement()
+	ele.Value = value
+	c.length++
+
+	if c.tail != 0 {
+		tail := c.get(c.tail)
+		tail.next = ele.addr
+		ele.prev = tail.addr
+		c.tail = ele.addr
+		return
+	}
+
+	c.head = ele.addr
+	c.tail = ele.addr
+}
+
+func (c *Queue[T]) Pop() (value T) {
+	ele := c.get(c.head)
+	if ele == nil {
+		return value
+	}
+
+	c.head = ele.next
+	if head := c.get(c.head); head != nil {
+		head.prev = 0
+	}
+
+	c.length--
+	value = ele.Value
+	c.stack.Push(ele.addr)
+	*ele = c.template
+	if c.length == 0 {
+		c.tail = 0
+		c.Reset()
+	}
+
+	return value
+}
+
+func (c *Queue[T]) Range(f func(v T) bool) {
+	for i := c.get(c.head); i != nil; i = c.get(i.next) {
+		if !f(i.Value) {
+			break
+		}
+	}
+}
+
+type stack []pointer
+
+func (c *stack) Len() int {
+	return len(*c)
+}
+
+func (c *stack) Push(v pointer) {
+	*c = append(*c, v)
+}
+
+func (c *stack) Pop() pointer {
+	var n = c.Len()
+	var v = (*c)[n-1]
+	*c = (*c)[:n-1]
+	return v
+}