implement task queue

Author: jkralik

go.mod

@@ -15,6 +15,7 @@ require (
 	github.com/jessevdk/go-flags v1.4.0
 	github.com/kelseyhightower/envconfig v1.4.0
 	github.com/lestrrat-go/jwx v1.0.2
+	github.com/panjf2000/ants/v2 v2.4.3
 	github.com/patrickmn/go-cache v2.1.0+incompatible
 	github.com/pion/dtls/v2 v2.0.1-0.20200503085337-8e86b3a7d585
 	github.com/plgd-dev/go-coap/v2 v2.1.4-0.20201201213140-b8c428d8fccf

go.sum

@@ -31,7 +31,6 @@ github.com/go-acme/lego v2.7.2+incompatible/go.mod h1:yzMNe9CasVUhkquNvti5nAtPmG
 github.com/go-kit/kit v0.9.0/go.mod h1:xBxKIO96dXMWWy0MnWVtmwkA9/13aqxPnvrjFYMA2as=
 github.com/go-logfmt/logfmt v0.4.0/go.mod h1:3RMwSq7FuexP4Kalkev3ejPJsZTpXXBr9+V4qmtdjCk=
 github.com/go-ocf/go-coap/v2 v2.0.4-0.20200728125043-f38b86f047a7/go.mod h1:X9wVKcaOSx7wBxKcvrWgMQq1R2DNeA7NBLW2osIb8TM=
-github.com/go-ocf/kit v0.0.0-20200728130040-4aebdb6982bc h1:VfEKBNg9m0rHO88jv/fOCXW1MN+Z4imfGZhYF0G8484=
 github.com/go-ocf/kit v0.0.0-20200728130040-4aebdb6982bc/go.mod h1:TIsoMT/iB7t9P6ahkcOnsmvS83SIJsv9qXRfz/yLf6M=
 github.com/go-stack/stack v1.8.0/go.mod h1:v0f6uXyyMGvRgIKkXu+yp6POWl0qKG85gN/melR3HDY=
 github.com/gofrs/uuid v3.3.0+incompatible h1:8K4tyRfvU1CYPgJsveYFQMhpFd/wXNM7iK6rR7UHz84=
@@ -85,6 +84,8 @@ github.com/lestrrat-go/pdebug v0.0.0-20200204225717-4d6bd78da58d/go.mod h1:B06CS
 github.com/miekg/dns v1.1.29 h1:xHBEhR+t5RzcFJjBLJlax2daXOrTYtr9z4WdKEfWFzg=
 github.com/miekg/dns v1.1.29/go.mod h1:KNUDUusw/aVsxyTYZM1oqvCicbwhgbNgztCETuNZ7xM=
 github.com/opentracing/opentracing-go v1.1.0/go.mod h1:UkNAQd3GIcIGf0SeVgPpRdFStlNbqXla1AfSYxPUl2o=
+github.com/panjf2000/ants/v2 v2.4.3 h1:wHghL17YKFanB62QjPQ9o+DuM4q7WrQ7zAhoX8+eBXU=
+github.com/panjf2000/ants/v2 v2.4.3/go.mod h1:f6F0NZVFsGCp5A7QW/Zj/m92atWwOkY0OIhFxRNFr4A=
 github.com/patrickmn/go-cache v2.1.0+incompatible h1:HRMgzkcYKYpi3C8ajMPV8OFXaaRUnok+kx1WdO15EQc=
 github.com/patrickmn/go-cache v2.1.0+incompatible/go.mod h1:3Qf8kWWT7OJRJbdiICTKqZju1ZixQ/KpMGzzAfe6+WQ=
 github.com/pion/dtls/v2 v2.0.1-0.20200503085337-8e86b3a7d585 h1:0v1k/bHrth28TctdEWnrCgLehYn3nOvFAwOwtwmyC34=
@@ -96,7 +97,6 @@ github.com/pion/transport v0.10.0/go.mod h1:BnHnUipd0rZQyTVB2SBGojFHT9CBt5C5TcsJ
 github.com/pkg/errors v0.8.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
 github.com/pkg/errors v0.9.1 h1:FEBLx1zS214owpjy7qsBeixbURkuhQAwrK5UwLGTwt4=
 github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
-github.com/plgd-dev/go-coap/v2 v2.0.4-0.20200819112225-8eb712b901bc h1:kEOQOFCaNd7SDhf17MSQkSB519JeDbwhhrUWBffbsZI=
 github.com/plgd-dev/go-coap/v2 v2.0.4-0.20200819112225-8eb712b901bc/go.mod h1:+tCi9Q78H/orWRtpVWyBgrr4vKFo2zYtbbxUllerBp4=
 github.com/plgd-dev/go-coap/v2 v2.1.4-0.20201201213140-b8c428d8fccf h1:am6M4meMUVVNItVjtrCZSOVUwHNz95MkkmUn8TjS4Js=
 github.com/plgd-dev/go-coap/v2 v2.1.4-0.20201201213140-b8c428d8fccf/go.mod h1:DccQmYY6swDlNlOCQOAX+SXTI9laSfGytskmeeNWmms=
@@ -227,8 +227,8 @@ gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127/go.mod h1:Co6ibVJAznAaIkqp8
 gopkg.in/errgo.v2 v2.1.0/go.mod h1:hNsd1EY+bozCKY1Ytp96fpM3vjJbqLJn88ws8XvfDNI=
 gopkg.in/square/go-jose.v2 v2.5.1 h1:7odma5RETjNHWJnR32wx8t+Io4djHE1PqxCFx3iiZ2w=
 gopkg.in/square/go-jose.v2 v2.5.1/go.mod h1:M9dMgbHiYLoDGQrXy7OpJDJWiKiU//h+vD76mk0e1AI=
-gopkg.in/yaml.v2 v2.2.2 h1:ZCJp+EgiOT7lHqUV2J862kp8Qj64Jo6az82+3Td9dZw=
 gopkg.in/yaml.v2 v2.2.2/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
+gopkg.in/yaml.v2 v2.2.7/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
 honnef.co/go/tools v0.0.0-20190102054323-c2f93a96b099/go.mod h1:rf3lG4BRIbNafJWhAfAdb/ePZxsR/4RtNHQocxwk9r4=
 honnef.co/go/tools v0.0.0-20190523083050-ea95bdfd59fc/go.mod h1:rf3lG4BRIbNafJWhAfAdb/ePZxsR/4RtNHQocxwk9r4=
 honnef.co/go/tools v0.0.1-2019.2.3 h1:3JgtbtFHMiCmsznwGVTUWbgGov+pVqnlf1dEJTNAXeM=

task/queue/queue.go

@@ -0,0 +1,80 @@
+package queue
+
+import (
+	"container/list"
+	"fmt"
+	"sync"
+
+	"github.com/panjf2000/ants/v2"
+)
+
+// TaskQueue representation of task queue.
+type TaskQueue struct {
+	goPool             *ants.Pool
+	limit              int
+	numParallelRequest 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))
+	if err != nil {
+		return nil, err
+	}
+	return &TaskQueue{
+		queue:              list.New(),
+		goPool:             p,
+		limit:              limit,
+		numParallelRequest: numWorkers,
+	}, nil
+}
+
+func (q *TaskQueue) appendQueue(tasks []func()) error {
+	q.mutex.Lock()
+	defer q.mutex.Unlock()
+	if q.queue.Len()+len(tasks) > q.limit {
+		return fmt.Errorf("reached limit of max processed jobs")
+	}
+	for _, t := range tasks {
+		q.queue.PushBack(t)
+	}
+	return nil
+}
+
+func (q *TaskQueue) popQueue() func() {
+	q.mutex.Lock()
+	defer q.mutex.Unlock()
+	if q.queue.Len() == 0 {
+		return nil
+	}
+	return q.queue.Remove(q.queue.Front()).(func())
+}
+
+// Submit appends and execute task by taskQueue.
+func (q *TaskQueue) Submit(tasks ...func()) error {
+	err := q.appendQueue(tasks)
+	if err != nil {
+		return err
+	}
+	q.goPool.Submit(func() {
+		for {
+			task := q.popQueue()
+			if task == nil {
+				return
+			}
+			task()
+		}
+	})
+	return nil
+}
+
+// Release closes queue and release it.
+func (q *TaskQueue) Release() {
+	q.goPool.Release()
+	q.mutex.Lock()
+	defer q.mutex.Unlock()
+	q.queue.Init()
+}

task/queue/queue_test.go

@@ -0,0 +1,31 @@
+package queue_test
+
+import (
+	"testing"
+
+	"github.com/plgd-dev/kit/task/queue"
+	"github.com/stretchr/testify/require"
+)
+
+func TestTaskQueue_Submit(t *testing.T) {
+	_, err := queue.New(-1, 10)
+	require.Error(t, err)
+	q, err := queue.New(1, 2)
+	require.NoError(t, err)
+	err = q.Submit(func() {}, func() {}, func() {})
+	require.Error(t, err)
+	v := make(chan int)
+	err = q.Submit(func() { v <- 1 }, func() { v <- 2 })
+	require.NoError(t, err)
+	d := <-v
+	require.Equal(t, 1, d)
+	d = <-v
+	require.Equal(t, 2, d)
+	err = q.Submit(func() { v <- 3; v <- 4 })
+	require.NoError(t, err)
+	d = <-v
+	require.Equal(t, 3, d)
+	q.Release()
+	d = <-v
+	require.Equal(t, 4, d)
+}