Repository: cloudwego/hertz
Pull request #499 (2022-12-28 07:02:01)

optimize: prevent hostclient observation resource leak
#### What type of PR is this?
<!--
Add one of the following kinds:

build: Changes that affect the build system or external dependencies (example scopes: gulp, broccoli, npm)
ci: Changes to our CI configuration files and scripts (example scopes: Travis, Circle, BrowserStack, SauceLabs)
docs: Documentation only changes
feat: A new feature
optimize: A new optimization
fix: A bug fix
perf: A code change that improves performance
refactor: A code change that neither fixes a bug nor adds a feature
style: Changes that do not affect the meaning of the code (white space, formatting, missing semi-colons, etc)
test: Adding missing tests or correcting existing tests
chore: Changes to the build process or auxiliary tools and libraries such as documentation generation
-->
feat
#### Check the PR title.
<!--
The description of the title will be attached in Release Notes, 
so please describe it from user-oriented, what this PR does / why we need it.
Please check your PR title with the below requirements:
-->
- [ ] This PR title match the format: \<type\>(optional scope): \<description\>
- [ ] The description of this PR title is user-oriented and clear enough for others to understand.


#### (Optional) Translate the PR title into Chinese.
防止 hostclient 监控资源泄漏

#### (Optional) More detail description for this PR(en: English/zh: Chinese).
<!--
Provide more detailed info for review. If it is a perf type PR, perf data is suggested to give.
-->
en: prevent hostclient opservation resource leak
zh(optional): 防止 hostclient 监控资源泄漏

#### Which issue(s) this PR fixes:
<!--
Automatically closes linked issue when PR is merged.
Eg: `Fixes #<issue number>`, or `Fixes (paste link of issue)`.
-->


codecov[bot]: # [Codecov](https://codecov.io/gh/cloudwego/hertz/pull/499?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=cloudwego) Report
Base: **69.52**% // Head: **69.54**% // Increases project coverage by **`+0.01%`** :tada:
> Coverage data is based on head [(`c62ecf9`)](https://codecov.io/gh/cloudwego/hertz/pull/499?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=cloudwego) compared to base [(`1b9a45e`)](https://codecov.io/gh/cloudwego/hertz/commit/1b9a45e7ae924c8be82e482ecb8523eac4c17636?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=cloudwego).
> Patch coverage: 41.17% of modified lines in pull request are covered.

<details><summary>Additional details and impacted files</summary>


```diff
@@             Coverage Diff             @@
##           develop     #499      +/-   ##
===========================================
+ Coverage    69.52%   69.54%   +0.01%     
===========================================
  Files           93       93              
  Lines         8802     8813      +11     
===========================================
+ Hits          6120     6129       +9     
- Misses        2318     2320       +2     
  Partials       364      364              
```


| [Impacted Files](https://codecov.io/gh/cloudwego/hertz/pull/499?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=cloudwego) | Coverage Δ | |
|---|---|---|
| [pkg/protocol/http1/client.go](https://codecov.io/gh/cloudwego/hertz/pull/499/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=cloudwego#diff-cGtnL3Byb3RvY29sL2h0dHAxL2NsaWVudC5nbw==) | `48.65% <0.00%> (-0.48%)` | :arrow_down: |
| [pkg/app/client/client.go](https://codecov.io/gh/cloudwego/hertz/pull/499/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=cloudwego#diff-cGtnL2FwcC9jbGllbnQvY2xpZW50Lmdv) | `74.39% <100.00%> (+0.15%)` | :arrow_up: |
| [pkg/app/client/option.go](https://codecov.io/gh/cloudwego/hertz/pull/499/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=cloudwego#diff-cGtnL2FwcC9jbGllbnQvb3B0aW9uLmdv) | `96.80% <100.00%> (+0.14%)` | :arrow_up: |
| [pkg/common/config/client\_option.go](https://codecov.io/gh/cloudwego/hertz/pull/499/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=cloudwego#diff-cGtnL2NvbW1vbi9jb25maWcvY2xpZW50X29wdGlvbi5nbw==) | `100.00% <100.00%> (ø)` | |
| [pkg/common/compress/compress.go](https://codecov.io/gh/cloudwego/hertz/pull/499/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=cloudwego#diff-cGtnL2NvbW1vbi9jb21wcmVzcy9jb21wcmVzcy5nbw==) | `83.80% <0.00%> (+2.85%)` | :arrow_up: |

Help us with your feedback. Take ten seconds to tell us [how you rate us](https://about.codecov.io/nps?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=cloudwego). Have a feature suggestion? [Share it here.](https://app.codecov.io/gh/feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=cloudwego)

</details>

[:umbrella: View full report at Codecov](https://codecov.io/gh/cloudwego/hertz/pull/499?src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=cloudwego).   
:loudspeaker: Do you have feedback about the report comment? [Let us know in this issue](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=cloudwego).


Patch:
diff --git a/pkg/app/client/client.go b/pkg/app/client/client.go
index 61c9894..e128519 100644
--- a/pkg/app/client/client.go
+++ b/pkg/app/client/client.go
@@ -538,6 +538,7 @@ func (c *Client) mCleaner() {
 	mustStop := false
 
 	for {
+		time.Sleep(10 * time.Second)
 		c.mLock.Lock()
 		for k, v := range c.m {
 			shouldRemove := v.ShouldRemove()
@@ -560,7 +561,6 @@ func (c *Client) mCleaner() {
 		if mustStop {
 			break
 		}
-		time.Sleep(10 * time.Second)
 	}
 }
 
@@ -638,5 +638,6 @@ func newHttp1OptionFromClient(c *Client) *http1.ClientOptions {
 		RetryConfig:                   c.options.RetryConfig,
 		RetryIfFunc:                   c.RetryIfFunc,
 		StateObserve:                  c.options.HostClientStateObserve,
+		ObservationInterval:           c.options.ObservationInterval,
 	}
 }
diff --git a/pkg/app/client/option.go b/pkg/app/client/option.go
index 37e0864..34f4196 100644
--- a/pkg/app/client/option.go
+++ b/pkg/app/client/option.go
@@ -150,9 +150,15 @@ func WithWriteTimeout(t time.Duration) config.ClientOption {
 }
 
 // WithConnStateObserve sets the connection state observation function.
-func WithConnStateObserve(hs config.HostClientStateFunc) config.ClientOption {
+// The first param is used to set hostclient state func.
+// The second param is used to set observation interval, default value is 5 seconds.
+// Warn: Do not start go routine in HostClientStateFunc.
+func WithConnStateObserve(hs config.HostClientStateFunc, interval ...time.Duration) config.ClientOption {
 	return config.ClientOption{F: func(o *config.ClientOptions) {
 		o.HostClientStateObserve = hs
+		if len(interval) > 0 {
+			o.ObservationInterval = interval[0]
+		}
 	}}
 }
 
diff --git a/pkg/common/config/client_option.go b/pkg/common/config/client_option.go
index a413cd8..4513383 100644
--- a/pkg/common/config/client_option.go
+++ b/pkg/common/config/client_option.go
@@ -34,9 +34,6 @@ type ConnPoolState struct {
 	WaitConnNum int
 	// HostClient Addr
 	Addr string
-	// Whether HostClient has been removed by Client. If it is true,
-	// you should let HostClientStateFunc return.
-	Closed bool
 }
 
 type HostClientState interface {
@@ -128,6 +125,9 @@ type ClientOptions struct {
 	RetryConfig *retry.Config
 
 	HostClientStateObserve HostClientStateFunc
+
+	// StateObserve execution interval
+	ObservationInterval time.Duration
 }
 
 func NewClientOptions(opts []ClientOption) *ClientOptions {
@@ -136,6 +136,7 @@ func NewClientOptions(opts []ClientOption) *ClientOptions {
 		MaxConnsPerHost:     consts.DefaultMaxConnsPerHost,
 		MaxIdleConnDuration: consts.DefaultMaxIdleConnDuration,
 		KeepAlive:           true,
+		ObservationInterval: time.Second * 5,
 	}
 	options.Apply(opts)
 
diff --git a/pkg/protocol/http1/client.go b/pkg/protocol/http1/client.go
index 4acdd49..4caddb0 100644
--- a/pkg/protocol/http1/client.go
+++ b/pkg/protocol/http1/client.go
@@ -121,7 +121,7 @@ type HostClient struct {
 
 	connsCleanerRun bool
 
-	closed bool
+	closed chan struct{}
 }
 
 func (c *HostClient) SetDynamicConfig(dc *client.DynamicConfig) {
@@ -129,9 +129,19 @@ func (c *HostClient) SetDynamicConfig(dc *client.DynamicConfig) {
 	c.ProxyURI = dc.ProxyURI
 	c.IsTLS = dc.IsTLS
 
-	// start observation after setting addr
+	// start observation after setting addr to avoid race
 	if c.StateObserve != nil {
-		go c.StateObserve(c)
+		go func() {
+			t := time.NewTicker(c.ObservationInterval)
+			for {
+				select {
+				case <-c.closed:
+					return
+				case <-t.C:
+					c.StateObserve(c)
+				}
+			}
+		}()
 	}
 }
 
@@ -178,7 +188,6 @@ func (c *HostClient) ConnPoolState() config.ConnPoolState {
 		PoolConnNum:  len(c.conns),
 		TotalConnNum: c.connsCount,
 		Addr:         c.Addr,
-		Closed:       c.closed,
 	}
 
 	if c.connsWait != nil {
@@ -608,9 +617,7 @@ func (c *HostClient) doNonNilReqResp(req *protocol.Request, resp *protocol.Respo
 }
 
 func (c *HostClient) Close() error {
-	c.connsLock.Lock()
-	defer c.connsLock.Unlock()
-	c.closed = true
+	close(c.closed)
 	return nil
 }
 
@@ -1145,6 +1152,7 @@ func (q *wantConnQueue) clearFront() (cleaned bool) {
 func NewHostClient(c *ClientOptions) client.HostClient {
 	hc := &HostClient{
 		ClientOptions: c,
+		closed:        make(chan struct{}),
 	}
 
 	return hc
@@ -1258,5 +1266,9 @@ type ClientOptions struct {
 
 	RetryIfFunc client.RetryIfFunc
 
+	// Observe hostclient state
 	StateObserve config.HostClientStateFunc
+
+	// StateObserve execution interval
+	ObservationInterval time.Duration
 }


Test patch:
diff --git a/pkg/app/client/client_test.go b/pkg/app/client/client_test.go
index 212a7c8..b769ec8 100644
--- a/pkg/app/client/client_test.go
+++ b/pkg/app/client/client_test.go
@@ -2158,17 +2158,26 @@ func TestClientState(t *testing.T) {
 
 	time.Sleep(time.Millisecond)
 
-	client, _ := NewClient(WithConnStateObserve(func(hcs config.HostClientState) {
-		time.Sleep(time.Second * 1)
-		assert.DeepEqual(t, 1, hcs.ConnPoolState().TotalConnNum)
-		assert.DeepEqual(t, 1, hcs.ConnPoolState().PoolConnNum)
-		assert.DeepEqual(t, "127.0.0.1:11000", hcs.ConnPoolState().Addr)
-		// sleep 20s so that client cleaner has removed hostclient.
-		// In this case, the closed should be true
-		time.Sleep(time.Second * 20)
-		assert.DeepEqual(t, true, hcs.ConnPoolState().Closed)
-	}))
+	state := int32(0)
+	client, _ := NewClient(
+		WithConnStateObserve(func(hcs config.HostClientState) {
+			switch atomic.LoadInt32(&state) {
+			case int32(0):
+				assert.DeepEqual(t, 1, hcs.ConnPoolState().TotalConnNum)
+				assert.DeepEqual(t, 1, hcs.ConnPoolState().PoolConnNum)
+				assert.DeepEqual(t, "127.0.0.1:11000", hcs.ConnPoolState().Addr)
+				atomic.StoreInt32(&state, int32(1))
+			case int32(1):
+				assert.DeepEqual(t, 0, hcs.ConnPoolState().TotalConnNum)
+				assert.DeepEqual(t, 0, hcs.ConnPoolState().PoolConnNum)
+				assert.DeepEqual(t, "127.0.0.1:11000", hcs.ConnPoolState().Addr)
+				atomic.StoreInt32(&state, int32(2))
+				return
+			case int32(2):
+				t.Fatal("It shouldn't go to here")
+			}
+		}, time.Second*9))
 
 	client.Get(context.Background(), nil, "http://127.0.0.1:11000")
-	time.Sleep(time.Second * 25)
+	time.Sleep(time.Second * 22)
 }
diff --git a/pkg/app/client/option_test.go b/pkg/app/client/option_test.go
index 39ac268..f744210 100644
--- a/pkg/app/client/option_test.go
+++ b/pkg/app/client/option_test.go
@@ -44,6 +44,7 @@ func TestClientOptions(t *testing.T) {
 			retry.WithDelayPolicy(retry.CombineDelay(retry.FixedDelayPolicy, retry.BackOffDelayPolicy, retry.RandomDelayPolicy)),
 		),
 		WithWriteTimeout(time.Second),
+		WithConnStateObserve(nil, time.Second),
 	})
 	assert.DeepEqual(t, 100*time.Millisecond, opt.DialTimeout)
 	assert.DeepEqual(t, 128, opt.MaxConnsPerHost)
@@ -58,5 +59,6 @@ func TestClientOptions(t *testing.T) {
 	assert.DeepEqual(t, 100*time.Millisecond, opt.RetryConfig.Delay)
 	assert.DeepEqual(t, 5*time.Second, opt.RetryConfig.MaxDelay)
 	assert.DeepEqual(t, 1*time.Second, opt.RetryConfig.MaxJitter)
+	assert.DeepEqual(t, 1*time.Second, opt.ObservationInterval)
 	assert.DeepEqual(t, fmt.Sprint(retry.CombineDelay(retry.FixedDelayPolicy, retry.BackOffDelayPolicy, retry.RandomDelayPolicy)), fmt.Sprint(opt.RetryConfig.DelayPolicy))
 }