Unverified Commit 99c2ec70 authored by Laurence's avatar Laurence Committed by GitHub

Fix: default not to start gr pool limit (#72)

* fix: default not start gr pool

* fix: ut
parent b640a570
...@@ -36,6 +36,7 @@ type WorkerPoolConfig struct { ...@@ -36,6 +36,7 @@ type WorkerPoolConfig struct {
NumQueues int NumQueues int
QueueSize int QueueSize int
Logger gxlog.Logger Logger gxlog.Logger
Enable bool
} }
// baseWorkerPool is a worker pool with multiple queues. // baseWorkerPool is a worker pool with multiple queues.
...@@ -69,6 +70,7 @@ type baseWorkerPool struct { ...@@ -69,6 +70,7 @@ type baseWorkerPool struct {
taskQueues []chan task taskQueues []chan task
numWorkers *atomic.Int32 numWorkers *atomic.Int32
enable bool
wg *sync.WaitGroup wg *sync.WaitGroup
} }
...@@ -94,6 +96,11 @@ func newBaseWorkerPool(config WorkerPoolConfig) *baseWorkerPool { ...@@ -94,6 +96,11 @@ func newBaseWorkerPool(config WorkerPoolConfig) *baseWorkerPool {
taskQueues: taskQueues, taskQueues: taskQueues,
numWorkers: new(atomic.Int32), numWorkers: new(atomic.Int32),
wg: new(sync.WaitGroup), wg: new(sync.WaitGroup),
enable: config.Enable,
}
if !config.Enable {
return p
} }
initWg := new(sync.WaitGroup) initWg := new(sync.WaitGroup)
......
...@@ -45,6 +45,11 @@ func (p *ConnectionPool) Submit(t task) error { ...@@ -45,6 +45,11 @@ func (p *ConnectionPool) Submit(t task) error {
return perrors.New("task shouldn't be nil") return perrors.New("task shouldn't be nil")
} }
if !p.enable {
go t()
return nil
}
// put the task to a queue using Round Robin algorithm // put the task to a queue using Round Robin algorithm
taskId := atomic.AddUint32(&p.taskId, 1) taskId := atomic.AddUint32(&p.taskId, 1)
select { select {
......
...@@ -36,6 +36,7 @@ func TestConnectionPool(t *testing.T) { ...@@ -36,6 +36,7 @@ func TestConnectionPool(t *testing.T) {
NumQueues: runtime.NumCPU(), NumQueues: runtime.NumCPU(),
QueueSize: 10, QueueSize: 10,
Logger: nil, Logger: nil,
Enable: true,
}) })
var count int64 var count int64
wg := new(sync.WaitGroup) wg := new(sync.WaitGroup)
...@@ -60,6 +61,7 @@ func TestConnectionPool(t *testing.T) { ...@@ -60,6 +61,7 @@ func TestConnectionPool(t *testing.T) {
NumQueues: 1, NumQueues: 1,
QueueSize: 0, QueueSize: 0,
Logger: nil, Logger: nil,
Enable: true,
}) })
wg := new(sync.WaitGroup) wg := new(sync.WaitGroup)
...@@ -85,6 +87,7 @@ func TestConnectionPool(t *testing.T) { ...@@ -85,6 +87,7 @@ func TestConnectionPool(t *testing.T) {
NumWorkers: runtime.NumCPU(), NumWorkers: runtime.NumCPU(),
NumQueues: runtime.NumCPU(), NumQueues: runtime.NumCPU(),
QueueSize: 100, QueueSize: 100,
Enable: true,
Logger: nil, Logger: nil,
}) })
...@@ -102,6 +105,7 @@ func TestConnectionPool(t *testing.T) { ...@@ -102,6 +105,7 @@ func TestConnectionPool(t *testing.T) {
p := NewConnectionPool(WorkerPoolConfig{ p := NewConnectionPool(WorkerPoolConfig{
NumWorkers: 0, NumWorkers: 0,
NumQueues: runtime.NumCPU(), NumQueues: runtime.NumCPU(),
Enable: true,
QueueSize: 100, QueueSize: 100,
Logger: nil, Logger: nil,
}) })
...@@ -111,6 +115,7 @@ func TestConnectionPool(t *testing.T) { ...@@ -111,6 +115,7 @@ func TestConnectionPool(t *testing.T) {
p = NewConnectionPool(WorkerPoolConfig{ p = NewConnectionPool(WorkerPoolConfig{
NumWorkers: 1, NumWorkers: 1,
NumQueues: 0, NumQueues: 0,
Enable: true,
QueueSize: 0, QueueSize: 0,
Logger: nil, Logger: nil,
}) })
...@@ -123,6 +128,7 @@ func TestConnectionPool(t *testing.T) { ...@@ -123,6 +128,7 @@ func TestConnectionPool(t *testing.T) {
NumQueues: 1, NumQueues: 1,
QueueSize: -1, QueueSize: -1,
Logger: nil, Logger: nil,
Enable: true,
}) })
err = p.Submit(func() {}) err = p.Submit(func() {})
...@@ -134,6 +140,7 @@ func TestConnectionPool(t *testing.T) { ...@@ -134,6 +140,7 @@ func TestConnectionPool(t *testing.T) {
p := NewConnectionPool(WorkerPoolConfig{ p := NewConnectionPool(WorkerPoolConfig{
NumWorkers: 1, NumWorkers: 1,
NumQueues: 1, NumQueues: 1,
Enable: true,
QueueSize: 0, QueueSize: 0,
Logger: nil, Logger: nil,
}) })
...@@ -149,6 +156,7 @@ func TestConnectionPool(t *testing.T) { ...@@ -149,6 +156,7 @@ func TestConnectionPool(t *testing.T) {
NumQueues: runtime.NumCPU(), NumQueues: runtime.NumCPU(),
QueueSize: 10, QueueSize: 10,
Logger: nil, Logger: nil,
Enable: true,
}) })
task, v := newCountTask() task, v := newCountTask()
...@@ -174,6 +182,7 @@ func TestConnectionPool(t *testing.T) { ...@@ -174,6 +182,7 @@ func TestConnectionPool(t *testing.T) {
NumQueues: runtime.NumCPU(), NumQueues: runtime.NumCPU(),
QueueSize: 10, QueueSize: 10,
Logger: nil, Logger: nil,
Enable: true,
}) })
task, v := newCountTask() task, v := newCountTask()
...@@ -192,6 +201,7 @@ func BenchmarkConnectionPool(b *testing.B) { ...@@ -192,6 +201,7 @@ func BenchmarkConnectionPool(b *testing.B) {
NumWorkers: 100, NumWorkers: 100,
NumQueues: runtime.NumCPU(), NumQueues: runtime.NumCPU(),
QueueSize: 100, QueueSize: 100,
Enable: true,
Logger: nil, Logger: nil,
}) })
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment