Expose ants options (#155)

* task/queue: allow to set optional ants.Option's

Author: jkralik

sync/task/queue/config.go

@@ -0,0 +1,26 @@
+package queue
+
+import (
+	"runtime"
+	"time"
+)
+
+// Config configuration for task queue.
+type Config struct {
+	// GoroutinePoolSize maximum number of running goroutine instances.
+	GoroutinePoolSize int `yaml:"goroutinePoolSize" json:"goroutinePoolSize"`
+	// Size size of queue. If it exhausted Submit returns error.
+	Size int `yaml:"size" json:"size"`
+	// MaxIdleTime sets up the interval time of cleaning up goroutines, 0 means never cleanup.
+	MaxIdleTime time.Duration `yaml:"maxIdleTime" json:"maxIdleTime"`
+}
+
+// SetDefaults set zero values to defautls.
+func (c *Config) SetDefaults() {
+	if c.GoroutinePoolSize == 0 {
+		c.GoroutinePoolSize = runtime.NumCPU() * 200
+	}
+	if c.Size == 0 {
+		c.Size = 2 * 1024 * 1024
+	}
+}

task/queue/queue.go sync/task/queue/queue.gotask/queue/queue.go renamed to sync/task/queue/queue.go

@@ -8,31 +8,32 @@ import (
 	"github.com/panjf2000/ants/v2"
 )
 
-// TaskQueue representation of task queue.
-type TaskQueue struct {
-	goPool             *ants.Pool
-	limit              int
-	numParallelRequest int
+// Queue representation of task queue.
+type Queue struct {
+	goPool *ants.Pool
+	limit  int
 
 	mutex sync.Mutex
 	queue *list.List
 }
 
-// New creates task queue which is processed by number of workers. Number of task is limited by limit.
-func New(numWorkers, limit int) (*TaskQueue, error) {
-	p, err := ants.NewPool(numWorkers, ants.WithPreAlloc(true), ants.WithNonblocking(true))
+// New creates task queue which is processed by goroutines.
+func New(cfg Config) (*Queue, error) {
+	if cfg.Size <= 0 {
+		return nil, fmt.Errorf("invalid value of Size")
+	}
+	p, err := ants.NewPool(cfg.GoroutinePoolSize, ants.WithPreAlloc(true), ants.WithExpiryDuration(cfg.MaxIdleTime), ants.WithNonblocking(true))
 	if err != nil {
 		return nil, err
 	}
-	return &TaskQueue{
-		queue:              list.New(),
-		goPool:             p,
-		limit:              limit,
-		numParallelRequest: numWorkers,
+	return &Queue{
+		queue:  list.New(),
+		goPool: p,
+		limit:  cfg.Size,
 	}, nil
 }
 
-func (q *TaskQueue) appendQueue(tasks []func()) error {
+func (q *Queue) appendQueue(tasks []func()) error {
 	q.mutex.Lock()
 	defer q.mutex.Unlock()
 	if q.queue.Len()+len(tasks) > q.limit {
@@ -44,7 +45,7 @@ func (q *TaskQueue) appendQueue(tasks []func()) error {
 	return nil
 }
 
-func (q *TaskQueue) popQueue() func() {
+func (q *Queue) popQueue() func() {
 	q.mutex.Lock()
 	defer q.mutex.Unlock()
 	if q.queue.Len() == 0 {
@@ -53,8 +54,8 @@ func (q *TaskQueue) popQueue() func() {
 	return q.queue.Remove(q.queue.Front()).(func())
 }
 
-// Submit appends and execute task by taskQueue.
-func (q *TaskQueue) Submit(tasks ...func()) error {
+// Submit appends and execute task by Queue.
+func (q *Queue) Submit(tasks ...func()) error {
 	err := q.appendQueue(tasks)
 	if err != nil {
 		return err
@@ -72,7 +73,7 @@ func (q *TaskQueue) Submit(tasks ...func()) error {
 }
 
 // Release closes queue and release it.
-func (q *TaskQueue) Release() {
+func (q *Queue) Release() {
 	q.goPool.Release()
 	q.mutex.Lock()
 	defer q.mutex.Unlock()

task/queue/queue_test.go sync/task/queue/queue_test.gotask/queue/queue_test.go renamed to sync/task/queue/queue_test.go

@@ -3,14 +3,22 @@ package queue_test
 import (
 	"testing"
 
-	"github.com/plgd-dev/kit/task/queue"
+	"github.com/plgd-dev/kit/sync/task/queue"
 	"github.com/stretchr/testify/require"
 )
 
 func TestTaskQueue_Submit(t *testing.T) {
-	_, err := queue.New(-1, 10)
+	_, err := queue.New(queue.Config{
+		NumWorkers: -1,
+		Size:       10,
+		PreAlloc:   true,
+	})
 	require.Error(t, err)
-	q, err := queue.New(1, 2)
+	q, err := queue.New(queue.Config{
+		NumWorkers: 1,
+		Size:       2,
+		PreAlloc:   true,
+	})
 	require.NoError(t, err)
 	err = q.Submit(func() {}, func() {}, func() {})
 	require.Error(t, err)