From fbb9020bece17a0f76a76c36fb1c5a28e84d7973 Mon Sep 17 00:00:00 2001
From: Jerry <jerrycgh@gmail.com>
Date: Fri, 15 Sep 2023 13:06:47 -0700
Subject: [PATCH] Stop execution pool in rpc handler (#1005)

* Stop execution pool in rpc handler

All execution pools need to be closed properly. This fixes a potential goroutine leak caused by metric goutine created by each execution pool.

* Cancel only once
---
 rpc/execution_pool.go | 7 +++++--
 rpc/handler.go        | 1 +
 2 files changed, 6 insertions(+), 2 deletions(-)

diff --git a/rpc/execution_pool.go b/rpc/execution_pool.go
index daa431e9bc..378ed55fa2 100644
--- a/rpc/execution_pool.go
+++ b/rpc/execution_pool.go
@@ -21,7 +21,8 @@ type SafePool struct {
 	service   string       // the service using ep
 	processed atomic.Int64 // keeps count of total processed requests
 
-	close chan struct{}
+	close     chan struct{}
+	closeOnce sync.Once
 
 	// Skip sending task to execution pool
 	fastPath bool
@@ -103,7 +104,9 @@ func (s *SafePool) Size() int {
 }
 
 func (s *SafePool) Stop() {
-	close(s.close)
+	s.closeOnce.Do(func() {
+		close(s.close)
+	})
 
 	if s.executionPool.Load() != nil {
 		s.executionPool.Load().Stop()
diff --git a/rpc/handler.go b/rpc/handler.go
index a056176b8b..6a75e67502 100644
--- a/rpc/handler.go
+++ b/rpc/handler.go
@@ -302,6 +302,7 @@ func (h *handler) close(err error, inflightReq *requestOp) {
 	h.callWG.Wait()
 	h.cancelRoot()
 	h.cancelServerSubscriptions(err)
+	h.executionPool.Stop()
 }
 
 // addRequestOp registers a request operation.