Repository: cloudwego/hertz
Pull request #625 (2023-02-20 12:34:25)

fix: connection pool 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
-->
fix
#### 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.
修复链接池泄漏

#### (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: When the client uses bodystream to receive data, if the user uses netpoll and does not close the bodyStream properly, it can cause a link pool leak. Use `runtime.SetFinalizer` as a bailout to close the bodyStream.
zh(optional): 在 client 使用 bodystream 接收数据的时候，如果用户使用 netpoll 且没有正确关闭 bodyStream，则会造成链接池泄漏。使用 `runtime.SetFinalizer` 作为保底措施来关闭bodyStream。

#### 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/625?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=cloudwego) Report
Base: **72.07**% // Head: **72.50**% // Increases project coverage by **`+0.43%`** :tada:
> Coverage data is based on head [(`3fcaaaf`)](https://codecov.io/gh/cloudwego/hertz/pull/625?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=cloudwego) compared to base [(`6b1ebb4`)](https://codecov.io/gh/cloudwego/hertz/commit/6b1ebb449c98423b7dd2a9b958f3670c5032c9f8?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=cloudwego).
> Patch coverage: 100.00% of modified lines in pull request are covered.

> :exclamation: Current head 3fcaaaf differs from pull request most recent head 8f472d2. Consider uploading reports for the commit 8f472d2 to get more accurate results

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


```diff
@@             Coverage Diff             @@
##           develop     #625      +/-   ##
===========================================
+ Coverage    72.07%   72.50%   +0.43%     
===========================================
  Files           96       96              
  Lines         9389     9391       +2     
===========================================
+ Hits          6767     6809      +42     
+ Misses        2189     2152      -37     
+ Partials       433      430       -3     
```


| [Impacted Files](https://codecov.io/gh/cloudwego/hertz/pull/625?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=cloudwego) | Coverage Δ | |
|---|---|---|
| [pkg/protocol/http1/resp/response.go](https://codecov.io/gh/cloudwego/hertz/pull/625?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=cloudwego#diff-cGtnL3Byb3RvY29sL2h0dHAxL3Jlc3AvcmVzcG9uc2UuZ28=) | `71.11% <100.00%> (+0.43%)` | :arrow_up: |
| [pkg/protocol/http1/client.go](https://codecov.io/gh/cloudwego/hertz/pull/625?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=cloudwego#diff-cGtnL3Byb3RvY29sL2h0dHAxL2NsaWVudC5nbw==) | `56.34% <0.00%> (+7.69%)` | :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/625?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/protocol/http1/resp/response.go b/pkg/protocol/http1/resp/response.go
index 1f909aa..819d39d 100644
--- a/pkg/protocol/http1/resp/response.go
+++ b/pkg/protocol/http1/resp/response.go
@@ -45,6 +45,7 @@ import (
 	"errors"
 	"fmt"
 	"io"
+	"runtime"
 	"sync"
 
 	"github.com/cloudwego/hertz/pkg/common/bytebufferpool"
@@ -138,6 +139,7 @@ type clientRespStream struct {
 }
 
 func (c *clientRespStream) Close() (err error) {
+	runtime.SetFinalizer(c, nil)
 	ext.ReleaseBodyStream(c.r)
 	if c.closeCallback != nil {
 		err = c.closeCallback()
@@ -166,6 +168,7 @@ func convertClientRespStream(bs io.Reader, fn func() error) *clientRespStream {
 	clientStream := clientRespStreamPool.Get().(*clientRespStream)
 	clientStream.r = bs
 	clientStream.closeCallback = fn
+	runtime.SetFinalizer(clientStream, (*clientRespStream).Close)
 	return clientStream
 }
 


Test patch:
diff --git a/pkg/protocol/http1/client_test.go b/pkg/protocol/http1/client_test.go
index f61e491..1e4e013 100644
--- a/pkg/protocol/http1/client_test.go
+++ b/pkg/protocol/http1/client_test.go
@@ -56,9 +56,8 @@ import (
 	"time"
 
 	"github.com/cloudwego/hertz/pkg/common/config"
-	"github.com/cloudwego/hertz/pkg/common/test/assert"
-
 	errs "github.com/cloudwego/hertz/pkg/common/errors"
+	"github.com/cloudwego/hertz/pkg/common/test/assert"
 	"github.com/cloudwego/hertz/pkg/common/test/mock"
 	"github.com/cloudwego/hertz/pkg/network"
 	"github.com/cloudwego/hertz/pkg/protocol"
diff --git a/pkg/protocol/http1/client_unix_test.go b/pkg/protocol/http1/client_unix_test.go
new file mode 100644
index 0000000..ff65bc4
--- /dev/null
+++ b/pkg/protocol/http1/client_unix_test.go
@@ -0,0 +1,63 @@
+// Copyright 2023 CloudWeGo Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+//
+//go:build aix || darwin || dragonfly || freebsd || linux || netbsd || openbsd || solaris
+// +build aix darwin dragonfly freebsd linux netbsd openbsd solaris
+
+package http1
+
+import (
+	"context"
+	"net/http"
+	"runtime"
+	"testing"
+	"time"
+
+	"github.com/cloudwego/hertz/pkg/common/test/assert"
+	"github.com/cloudwego/hertz/pkg/network/netpoll"
+	"github.com/cloudwego/hertz/pkg/protocol"
+	"github.com/cloudwego/hertz/pkg/protocol/consts"
+)
+
+func TestGcBodyStream(t *testing.T) {
+	srv := &http.Server{Addr: ":11001", Handler: http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) {
+		w.Write([]byte("hello\n"))
+		time.Sleep(time.Second)
+		w.Write([]byte("world\n"))
+	})}
+	go srv.ListenAndServe()
+	time.Sleep(100 * time.Millisecond)
+
+	c := &HostClient{
+		ClientOptions: &ClientOptions{
+			Dialer:             netpoll.NewDialer(),
+			ResponseBodyStream: true,
+		},
+		Addr: "127.0.0.1:11001",
+	}
+
+	for i := 0; i < 10; i++ {
+		req, resp := protocol.AcquireRequest(), protocol.AcquireResponse()
+		req.SetRequestURI("http://127.0.0.1:11001")
+		req.SetMethod(consts.MethodPost)
+		err := c.Do(context.Background(), req, resp)
+		if err != nil {
+			t.Errorf("client Do error=%v", err.Error())
+		}
+	}
+
+	runtime.GC()
+	c.CloseIdleConnections()
+	assert.DeepEqual(t, 0, c.ConnPoolState().TotalConnNum)
+}