Ftr: Optimize logging for base worker pool (#68)

* ftr: work pool * rename WorkerPool -> FixedWorkerPool * go fmt * fix typo * benchmark for ConnectionPool * go fmt * benchmark for ConnectionPool * chan buffer extension for ConnectionPool benchmark * reset benchmark tests for TaskPool * reduce buffer of ConnectionPool * unbounded chan benchmark * reduce goroutine num * ConnectionPool adopts task queue array * apache license * fix shadow variable * remove done channel, add numWorkers * go fmt * fix wrong logger * add newBaseWorkerPool * remove ConnectionPoolConfig * update comments * update comments * fix unittests * Update base_worker_pool.go * Update base_worker_pool.go * fix newBaseWorkerPool bugs * go fmt * fix CountTaskSync bugs * fix CountTask bug * optimize code * go fmt * add SubmitSync test * go fmt * reduce the number of lines of logging for connection pool * merge logs * go fmt Co-authored-by: Xin.Zh <[email protected]>
dubbogo · Aug 11, 2021 · e1708fd · e1708fd
1 parent 8ef4d62
commit e1708fd
Showing 1 changed file with 9 additions and 10 deletions.
diff --git a/sync/base_worker_pool.go b/sync/base_worker_pool.go
@@ -102,6 +102,9 @@ func newBaseWorkerPool(config WorkerPoolConfig) *baseWorkerPool {
 	p.dispatch(config.NumWorkers, initWg)
 
 	initWg.Wait()
+	if p.logger != nil {
+		p.logger.Infof("all %d workers are started", p.NumWorkers())
+	}
 
 	return p
 }
@@ -112,11 +115,11 @@ func (p *baseWorkerPool) dispatch(numWorkers int, wg *sync.WaitGroup) {
 	}
 }
 
-func (p *baseWorkerPool) Submit(t task) error {
+func (p *baseWorkerPool) Submit(_ task) error {
 	panic("implement me")
 }
 
-func (p *baseWorkerPool) SubmitSync(t task) error {
+func (p *baseWorkerPool) SubmitSync(_ task) error {
 	panic("implement me")
 }
 
@@ -129,6 +132,9 @@ func (p *baseWorkerPool) Close() {
 		close(q)
 	}
 	p.wg.Wait()
+	if p.logger != nil {
+		p.logger.Infof("there are %d workers remained, all workers are closed", p.NumWorkers())
+	}
 }
 
 func (p *baseWorkerPool) IsClosed() bool {
@@ -153,20 +159,13 @@ func (p *baseWorkerPool) worker(workerId int, wg *sync.WaitGroup) {
 		p.wg.Done()
 	}()
 
-	if p.logger != nil {
-		p.logger.Infof("worker #%d is started\n", workerId)
-	}
-
 	chanId := workerId % len(p.taskQueues)
 
 	wg.Done()
 	for {
 		select {
 		case t, ok := <-p.taskQueues[chanId]:
 			if !ok {
-				if p.logger != nil {
-					p.logger.Infof("worker #%d is closed\n", workerId)
-				}
 				return
 			}
 			if t != nil {
@@ -175,7 +174,7 @@ func (p *baseWorkerPool) worker(workerId int, wg *sync.WaitGroup) {
 					defer func() {
 						if r := recover(); r != nil {
 							if p.logger != nil {
-								p.logger.Errorf("goroutine panic: %v\n%s\n", r, string(debug.Stack()))
+								p.logger.Errorf("goroutine panic: %v\n%s", r, string(debug.Stack()))
 							}
 						}
 					}()