Repository: valyala/fasthttp
Pull request #1243 (2022-03-10 19:31:27)

support adding/removing clients from LBClient
addresses https://github.com/valyala/fasthttp/issues/504 and is a followup to https://github.com/valyala/fasthttp/pull/556, addressing the outstanding feedback.

Adds support for adding/removing clients from LBClient utilizing a mutex as well as a user-defined callback for determining which HostClients to be removed


Patch:
diff --git a/.github/workflows/lint.yml b/.github/workflows/lint.yml
index 5130b05..6c2c7b0 100644
--- a/.github/workflows/lint.yml
+++ b/.github/workflows/lint.yml
@@ -11,9 +11,12 @@ jobs:
     - uses: actions/checkout@v2
     - uses: actions/setup-go@v2
       with:
-        go-version: 1.17.x
+        go-version: 1.18.x
     - run: go version
     - run: diff -u <(echo -n) <(gofmt -d .)
-    - uses: golangci/golangci-lint-action@v2
-      with:
-        version: v1.28.3
+    - name: Run golangci-lint
+      run: | # https://github.com/golangci/golangci-lint/pull/2438
+        export PATH=$PATH:$(go env GOPATH)/bin
+        go install github.com/golangci/golangci-lint/cmd/golangci-lint@v1.44.2
+        golangci-lint run
+
diff --git a/.github/workflows/security.yml b/.github/workflows/security.yml
index 13b8fa7..07c61ed 100644
--- a/.github/workflows/security.yml
+++ b/.github/workflows/security.yml
@@ -8,9 +8,11 @@ jobs:
   test:
     strategy:
       matrix:
-        go-version: [1.17.x]
+        go-version: [1.18.x]
         platform: [ubuntu-latest]
     runs-on: ${{ matrix.platform }}
+    env:
+      GO111MODULE: on
     steps:
       - name: Install Go
         uses: actions/setup-go@v1
@@ -18,5 +20,8 @@ jobs:
           go-version: ${{ matrix.go-version }}
       - name: Checkout code
         uses: actions/checkout@v2
-      - name: Security
-        run: go get github.com/securego/gosec/cmd/gosec; `go env GOPATH`/bin/gosec -exclude=G104,G304 ./...
+      - name: Run Gosec Security Scanner
+        run: | # https://github.com/securego/gosec/issues/469#issuecomment-1070608395
+          export PATH=$PATH:$(go env GOPATH)/bin
+          go install github.com/securego/gosec/v2/cmd/gosec@latest
+          gosec -exclude=G104,G304,G402 ./...
diff --git a/args.go b/args.go
index a8e4394..7c55943 100644
--- a/args.go
+++ b/args.go
@@ -230,7 +230,7 @@ func (a *Args) SetBytesKV(key, value []byte) {
 
 // SetNoValue sets only 'key' as argument without the '='.
 //
-// Only key in argumemt, like key1&key2
+// Only key in argument, like key1&key2
 func (a *Args) SetNoValue(key string) {
 	a.args = setArg(a.args, key, "", argsNoValue)
 }
diff --git a/bytesconv.go b/bytesconv.go
index bf582af..a12bb95 100644
--- a/bytesconv.go
+++ b/bytesconv.go
@@ -11,7 +11,6 @@ import (
 	"math"
 	"net"
 	"reflect"
-	"strings"
 	"sync"
 	"time"
 	"unsafe"
@@ -19,29 +18,24 @@ import (
 
 // AppendHTMLEscape appends html-escaped s to dst and returns the extended dst.
 func AppendHTMLEscape(dst []byte, s string) []byte {
-	if strings.IndexByte(s, '<') < 0 &&
-		strings.IndexByte(s, '>') < 0 &&
-		strings.IndexByte(s, '"') < 0 &&
-		strings.IndexByte(s, '\'') < 0 {
+	var (
+		prev int
+		sub  string
+	)
 
-		// fast path - nothing to escape
-		return append(dst, s...)
-	}
-
-	// slow path
-	var prev int
-	var sub string
 	for i, n := 0, len(s); i < n; i++ {
 		sub = ""
 		switch s[i] {
+		case '&':
+			sub = "&amp;"
 		case '<':
 			sub = "&lt;"
 		case '>':
 			sub = "&gt;"
 		case '"':
-			sub = "&quot;"
+			sub = "&#34;" // "&#34;" is shorter than "&quot;".
 		case '\'':
-			sub = "&#39;"
+			sub = "&#39;" // "&#39;" is shorter than "&apos;" and apos was not in HTML until HTML5.
 		}
 		if len(sub) > 0 {
 			dst = append(dst, s[prev:i]...)
diff --git a/bytesconv_32.go b/bytesconv_32.go
index 6a6fec2..b574883 100644
--- a/bytesconv_32.go
+++ b/bytesconv_32.go
@@ -1,5 +1,5 @@
-//go:build !amd64 && !arm64 && !ppc64 && !ppc64le
-// +build !amd64,!arm64,!ppc64,!ppc64le
+//go:build !amd64 && !arm64 && !ppc64 && !ppc64le && !s390x
+// +build !amd64,!arm64,!ppc64,!ppc64le,!s390x
 
 package fasthttp
 
diff --git a/bytesconv_64.go b/bytesconv_64.go
index 1300d5a..94d0ec6 100644
--- a/bytesconv_64.go
+++ b/bytesconv_64.go
@@ -1,5 +1,5 @@
-//go:build amd64 || arm64 || ppc64 || ppc64le
-// +build amd64 arm64 ppc64 ppc64le
+//go:build amd64 || arm64 || ppc64 || ppc64le || s390x
+// +build amd64 arm64 ppc64 ppc64le s390x
 
 package fasthttp
 
diff --git a/client.go b/client.go
index b36ca40..aea7411 100644
--- a/client.go
+++ b/client.go
@@ -618,6 +618,14 @@ type RetryIfFunc func(request *Request) bool
 // TransportFunc wraps every request/response.
 type TransportFunc func(*Request, *Response) error
 
+// ConnPoolStrategyType define strategy of connection pool enqueue/dequeue
+type ConnPoolStrategyType int
+
+const (
+	FIFO ConnPoolStrategyType = iota
+	LIFO
+)
+
 // HostClient balances http requests among hosts listed in Addr.
 //
 // HostClient may be used for balancing load among multiple upstream hosts.
@@ -772,6 +780,9 @@ type HostClient struct {
 	// Transport defines a transport-like mechanism that wraps every request/response.
 	Transport TransportFunc
 
+	// Connection pool strategy. Can be either LIFO or FIFO (default).
+	ConnPoolStrategy ConnPoolStrategyType
+
 	clientName  atomic.Value
 	lastUseTime uint32
 
@@ -1437,12 +1448,12 @@ func (c *HostClient) doNonNilReqResp(req *Request, resp *Response) (bool, error)
 	if err == nil {
 		err = bw.Flush()
 	}
-	if err != nil {
-		c.releaseWriter(bw)
+	c.releaseWriter(bw)
+	isConnRST := isConnectionReset(err)
+	if err != nil && !isConnRST {
 		c.closeConn(cc)
 		return true, err
 	}
-	c.releaseWriter(bw)
 
 	if c.ReadTimeout > 0 {
 		// Set Deadline every time, since golang has fixed the performance issue
@@ -1462,22 +1473,22 @@ func (c *HostClient) doNonNilReqResp(req *Request, resp *Response) (bool, error)
 	}
 
 	br := c.acquireReader(conn)
-	if err = resp.ReadLimitBody(br, c.MaxResponseBodySize); err != nil {
-		c.releaseReader(br)
+	err = resp.ReadLimitBody(br, c.MaxResponseBodySize)
+	c.releaseReader(br)
+	if err != nil {
 		c.closeConn(cc)
 		// Don't retry in case of ErrBodyTooLarge since we will just get the same again.
 		retry := err != ErrBodyTooLarge
 		return retry, err
 	}
-	c.releaseReader(br)
 
-	if resetConnection || req.ConnectionClose() || resp.ConnectionClose() {
+	if resetConnection || req.ConnectionClose() || resp.ConnectionClose() || isConnRST {
 		c.closeConn(cc)
 	} else {
 		c.releaseConn(cc)
 	}
 
-	return false, err
+	return false, nil
 }
 
 var (
@@ -1497,6 +1508,10 @@ var (
 	// to broken server.
 	ErrConnectionClosed = errors.New("the server closed connection before returning the first response byte. " +
 		"Make sure the server returns 'Connection: close' response header before closing the connection")
+
+	// ErrConnPoolStrategyNotImpl is returned when HostClient.ConnPoolStrategy is not implement yet.
+	// If you see this error, then you need to check your HostClient configuration.
+	ErrConnPoolStrategyNotImpl = errors.New("connection pool strategy is not implement")
 )
 
 type timeoutError struct{}
@@ -1544,10 +1559,20 @@ func (c *HostClient) acquireConn(reqTimeout time.Duration, connectionClose bool)
 			}
 		}
 	} else {
-		n--
-		cc = c.conns[n]
-		c.conns[n] = nil
-		c.conns = c.conns[:n]
+		switch c.ConnPoolStrategy {
+		case LIFO:
+			n--
+			cc = c.conns[n]
+			c.conns[n] = nil
+			c.conns = c.conns[:n]
+		case FIFO:
+			cc = c.conns[0]
+			copy(c.conns, c.conns[1:])
+			c.conns[n-1] = nil
+			c.conns = c.conns[:n-1]
+		default:
+			return nil, ErrConnPoolStrategyNotImpl
+		}
 	}
 	c.connsLock.Unlock()
 
@@ -2606,8 +2631,8 @@ func (c *pipelineConnClient) init() {
 			for {
 				if err := c.worker(); err != nil {
 					c.logger().Printf("error in PipelineClient(%q): %s", c.Addr, err)
-					if netErr, ok := err.(net.Error); ok && netErr.Temporary() {
-						// Throttle client reconnections on temporary errors
+					if netErr, ok := err.(net.Error); ok && netErr.Timeout() {
+						// Throttle client reconnections on timeout errors
 						time.Sleep(time.Second)
 					}
 				} else {
@@ -2891,14 +2916,15 @@ func (c *pipelineConnClient) getClientName() []byte {
 
 var errPipelineConnStopped = errors.New("pipeline connection has been stopped")
 
-func acquirePipelineWork(pool *sync.Pool, timeout time.Duration) *pipelineWork {
+func acquirePipelineWork(pool *sync.Pool, timeout time.Duration) (w *pipelineWork) {
 	v := pool.Get()
-	if v == nil {
-		v = &pipelineWork{
+	if v != nil {
+		w = v.(*pipelineWork)
+	} else {
+		w = &pipelineWork{
 			done: make(chan struct{}, 1),
 		}
 	}
-	w := v.(*pipelineWork)
 	if timeout > 0 {
 		if w.t == nil {
 			w.t = time.NewTimer(timeout)
diff --git a/go.mod b/go.mod
index f5cfe1b..a9cd927 100644
--- a/go.mod
+++ b/go.mod
@@ -1,6 +1,6 @@
 module github.com/valyala/fasthttp
 
-go 1.12
+go 1.15
 
 require (
 	github.com/andybalholm/brotli v1.0.4
diff --git a/http.go b/http.go
index 47431cd..75b40c9 100644
--- a/http.go
+++ b/http.go
@@ -1291,6 +1291,9 @@ func (resp *Response) ReadLimitBody(r *bufio.Reader, maxBodySize int) error {
 	if !resp.mustSkipBody() {
 		err = resp.ReadBody(r, maxBodySize)
 		if err != nil {
+			if isConnectionReset(err) {
+				return nil
+			}
 			return err
 		}
 	}
@@ -1298,6 +1301,9 @@ func (resp *Response) ReadLimitBody(r *bufio.Reader, maxBodySize int) error {
 	if resp.Header.ContentLength() == -1 {
 		err = resp.Header.ReadTrailer(r)
 		if err != nil && err != io.EOF {
+			if isConnectionReset(err) {
+				return nil
+			}
 			return err
 		}
 	}
diff --git a/lbclient.go b/lbclient.go
index 46d14b7..34ff719 100644
--- a/lbclient.go
+++ b/lbclient.go
@@ -50,6 +50,7 @@ type LBClient struct {
 	cs []*lbClient
 
 	once sync.Once
+	mu   sync.RWMutex
 }
 
 // DefaultLBClientTimeout is the default request timeout used by LBClient
@@ -80,6 +81,8 @@ func (cc *LBClient) Do(req *Request, resp *Response) error {
 }
 
 func (cc *LBClient) init() {
+	cc.mu.Lock()
+	defer cc.mu.Unlock()
 	if len(cc.Clients) == 0 {
 		panic("BUG: LBClient.Clients cannot be empty")
 	}
@@ -91,9 +94,44 @@ func (cc *LBClient) init() {
 	}
 }
 
+// AddClient adds a new client to the balanced clients
+// returns the new total number of clients
+func (cc *LBClient) AddClient(c BalancingClient) int {
+	cc.mu.Lock()
+	cc.cs = append(cc.cs, &lbClient{
+		c:           c,
+		healthCheck: cc.HealthCheck,
+	})
+	cc.mu.Unlock()
+	return len(cc.cs)
+}
+
+// RemoveClients removes clients using the provided callback
+// if rc returns true, the passed client will be removed
+// returns the new total number of clients
+func (cc *LBClient) RemoveClients(rc func(BalancingClient) bool) int {
+	cc.mu.Lock()
+	n := 0
+	for _, cs := range cc.cs {
+		if rc(cs.c) {
+			continue
+		}
+		cc.cs[n] = cs
+		n++
+	}
+	for i := n; i < len(cc.cs); i++ {
+		cc.cs[i] = nil
+	}
+	cc.cs = cc.cs[:n]
+
+	cc.mu.Unlock()
+	return len(cc.cs)
+}
+
 func (cc *LBClient) get() *lbClient {
 	cc.once.Do(cc.init)
 
+	cc.mu.RLock()
 	cs := cc.cs
 
 	minC := cs[0]
@@ -108,6 +146,7 @@ func (cc *LBClient) get() *lbClient {
 			minT = t
 		}
 	}
+	cc.mu.RUnlock()
 	return minC
 }
 
diff --git a/nocopy.go b/nocopy.go
index 9664cb0..5e41bdf 100644
--- a/nocopy.go
+++ b/nocopy.go
@@ -7,5 +7,5 @@ package fasthttp
 // and also: https://stackoverflow.com/questions/52494458/nocopy-minimal-example
 type noCopy struct{} //nolint:unused
 
-func (*noCopy) Lock()   {}
-func (*noCopy) Unlock() {}
+func (*noCopy) Lock()   {} //nolint:unused
+func (*noCopy) Unlock() {} //nolint:unused
diff --git a/server.go b/server.go
index 3b58542..a970e82 100644
--- a/server.go
+++ b/server.go
@@ -482,7 +482,7 @@ func TimeoutWithCodeHandler(h RequestHandler, timeout time.Duration, msg string,
 	}
 }
 
-//RequestConfig configure the per request deadline and body limits
+// RequestConfig configure the per request deadline and body limits
 type RequestConfig struct {
 	// ReadTimeout is the maximum duration for reading the entire
 	// request body.
@@ -1915,8 +1915,8 @@ func acceptConn(s *Server, ln net.Listener, lastPerIPErrorTime *time.Time) (net.
 			if c != nil {
 				panic("BUG: net.Listener returned non-nil conn and non-nil error")
 			}
-			if netErr, ok := err.(net.Error); ok && netErr.Temporary() {
-				s.logger().Printf("Temporary error when accepting new connections: %s", netErr)
+			if netErr, ok := err.(net.Error); ok && netErr.Timeout() {
+				s.logger().Printf("Timeout error when accepting new connections: %s", netErr)
 				time.Sleep(time.Second)
 				continue
 			}
@@ -2138,7 +2138,7 @@ func (s *Server) serveConn(c net.Conn) (err error) {
 		if connRequestNum > 1 {
 			if d := s.idleTimeout(); d > 0 {
 				if err := c.SetReadDeadline(time.Now().Add(d)); err != nil {
-					panic(fmt.Sprintf("BUG: error in SetReadDeadline(%s): %s", d, err))
+					break
 				}
 			}
 		}
@@ -2178,15 +2178,17 @@ func (s *Server) serveConn(c net.Conn) (err error) {
 		ctx.Response.secureErrorLogMessage = s.SecureErrorLogMessage
 
 		if err == nil {
+			s.setState(c, StateActive)
+
 			if s.ReadTimeout > 0 {
 				if err := c.SetReadDeadline(time.Now().Add(s.ReadTimeout)); err != nil {
-					panic(fmt.Sprintf("BUG: error in SetReadDeadline(%s): %s", s.ReadTimeout, err))
+					break
 				}
 			} else if s.IdleTimeout > 0 && connRequestNum > 1 {
 				// If this was an idle connection and the server has an IdleTimeout but
 				// no ReadTimeout then we should remove the ReadTimeout.
 				if err := c.SetReadDeadline(zeroTime); err != nil {
-					panic(fmt.Sprintf("BUG: error in SetReadDeadline(zeroTime): %s", err))
+					break
 				}
 			}
 			if s.DisableHeaderNamesNormalizing {
@@ -2230,7 +2232,7 @@ func (s *Server) serveConn(c net.Conn) (err error) {
 						writeTimeout = reqConf.WriteTimeout
 					}
 				}
-				//read body
+				// read body
 				if s.StreamRequestBody {
 					err = ctx.Request.readBodyStream(br, maxRequestBodySize, s.GetOnly, !s.DisablePreParseMultipartForm)
 				} else {
@@ -2238,11 +2240,6 @@ func (s *Server) serveConn(c net.Conn) (err error) {
 				}
 			}
 
-			if err == nil {
-				// If we read any bytes off the wire, we're active.
-				s.setState(c, StateActive)
-			}
-
 			if (s.ReduceMemoryUsage && br.Buffered() == 0) || err != nil {
 				releaseReader(s, br)
 				br = nil
@@ -2637,11 +2634,11 @@ func (s *Server) acquireCtx(c net.Conn) (ctx *RequestCtx) {
 		ctx = new(RequestCtx)
 		ctx.Request.keepBodyBuffer = keepBodyBuffer
 		ctx.Response.keepBodyBuffer = keepBodyBuffer
+		ctx.s = s
 	} else {
 		ctx = v.(*RequestCtx)
 	}
 
-	ctx.s = s
 	ctx.c = c
 
 	return ctx
diff --git a/tcp.go b/tcp.go
new file mode 100644
index 0000000..54d3033
--- /dev/null
+++ b/tcp.go
@@ -0,0 +1,13 @@
+//go:build !windows
+// +build !windows
+
+package fasthttp
+
+import (
+	"errors"
+	"syscall"
+)
+
+func isConnectionReset(err error) bool {
+	return errors.Is(err, syscall.ECONNRESET)
+}
diff --git a/tcp_windows.go b/tcp_windows.go
new file mode 100644
index 0000000..5c33025
--- /dev/null
+++ b/tcp_windows.go
@@ -0,0 +1,13 @@
+//go:build windows
+// +build windows
+
+package fasthttp
+
+import (
+	"errors"
+	"syscall"
+)
+
+func isConnectionReset(err error) bool {
+	return errors.Is(err, syscall.WSAECONNRESET)
+}


Test patch:
diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml
index 068a6f5..429c155 100644
--- a/.github/workflows/test.yml
+++ b/.github/workflows/test.yml
@@ -8,7 +8,7 @@ jobs:
   test:
     strategy:
       matrix:
-        go-version: [1.15.x, 1.16.x, 1.17.x]
+        go-version: [1.15.x, 1.16.x, 1.17.x, 1.18.x]
         os: [ubuntu-latest, macos-latest, windows-latest]
     runs-on: ${{ matrix.os }}
     steps:
diff --git a/bytesconv_32_test.go b/bytesconv_32_test.go
index cec5aa9..3f5d5de 100644
--- a/bytesconv_32_test.go
+++ b/bytesconv_32_test.go
@@ -1,5 +1,5 @@
-//go:build !amd64 && !arm64 && !ppc64 && !ppc64le
-// +build !amd64,!arm64,!ppc64,!ppc64le
+//go:build !amd64 && !arm64 && !ppc64 && !ppc64le && !s390x
+// +build !amd64,!arm64,!ppc64,!ppc64le,!s390x
 
 package fasthttp
 
diff --git a/bytesconv_64_test.go b/bytesconv_64_test.go
index 5351591..0689809 100644
--- a/bytesconv_64_test.go
+++ b/bytesconv_64_test.go
@@ -1,5 +1,5 @@
-//go:build amd64 || arm64 || ppc64 || ppc64le
-// +build amd64 arm64 ppc64 ppc64le
+//go:build amd64 || arm64 || ppc64 || ppc64le || s390x
+// +build amd64 arm64 ppc64 ppc64le s390x
 
 package fasthttp
 
diff --git a/bytesconv_test.go b/bytesconv_test.go
index 4c35371..cf4a12e 100644
--- a/bytesconv_test.go
+++ b/bytesconv_test.go
@@ -4,20 +4,48 @@ import (
 	"bufio"
 	"bytes"
 	"fmt"
+	"html"
 	"net"
+	"net/url"
 	"testing"
 	"time"
 
 	"github.com/valyala/bytebufferpool"
 )
 
+func TestAppendQuotedArg(t *testing.T) {
+	t.Parallel()
+
+	// Sync with url.QueryEscape
+	allcases := make([]byte, 256)
+	for i := 0; i < 256; i++ {
+		allcases[i] = byte(i)
+	}
+	res := string(AppendQuotedArg(nil, allcases))
+	expect := url.QueryEscape(string(allcases))
+	if res != expect {
+		t.Fatalf("unexpected string %q. Expecting %q.", res, expect)
+	}
+}
+
 func TestAppendHTMLEscape(t *testing.T) {
 	t.Parallel()
 
+	// Sync with html.EscapeString
+	allcases := make([]byte, 256)
+	for i := 0; i < 256; i++ {
+		allcases[i] = byte(i)
+	}
+	res := string(AppendHTMLEscape(nil, string(allcases)))
+	expect := string(html.EscapeString(string(allcases)))
+	if res != expect {
+		t.Fatalf("unexpected string %q. Expecting %q.", res, expect)
+	}
+
 	testAppendHTMLEscape(t, "", "")
 	testAppendHTMLEscape(t, "<", "&lt;")
 	testAppendHTMLEscape(t, "a", "a")
-	testAppendHTMLEscape(t, `><"''`, "&gt;&lt;&quot;&#39;&#39;")
+	testAppendHTMLEscape(t, `><"''`, "&gt;&lt;&#34;&#39;&#39;")
 	testAppendHTMLEscape(t, "fo<b x='ss'>a</b>xxx", "fo&lt;b x=&#39;ss&#39;&gt;a&lt;/b&gt;xxx")
 }
 
diff --git a/client_test.go b/client_test.go
index 78eda2e..e559305 100644
--- a/client_test.go
+++ b/client_test.go
@@ -988,14 +988,14 @@ func testPipelineClientDo(t *testing.T, c *PipelineClient) {
 				time.Sleep(10 * time.Millisecond)
 				continue
 			}
-			t.Fatalf("unexpected error on iteration %d: %s", i, err)
+			t.Errorf("unexpected error on iteration %d: %s", i, err)
 		}
 		if resp.StatusCode() != StatusOK {
-			t.Fatalf("unexpected status code: %d. Expecting %d", resp.StatusCode(), StatusOK)
+			t.Errorf("unexpected status code: %d. Expecting %d", resp.StatusCode(), StatusOK)
 		}
 		body := string(resp.Body())
 		if body != "OK" {
-			t.Fatalf("unexpected body: %q. Expecting %q", body, "OK")
+			t.Errorf("unexpected body: %q. Expecting %q", body, "OK")
 		}
 
 		// sleep for a while, so the connection to the host may expire.
@@ -1705,10 +1705,10 @@ func testClientDoTimeoutError(t *testing.T, c *Client, n int) {
 	for i := 0; i < n; i++ {
 		err := c.DoTimeout(&req, &resp, time.Millisecond)
 		if err == nil {
-			t.Fatalf("expecting error")
+			t.Errorf("expecting error")
 		}
 		if err != ErrTimeout {
-			t.Fatalf("unexpected error: %s. Expecting %s", err, ErrTimeout)
+			t.Errorf("unexpected error: %s. Expecting %s", err, ErrTimeout)
 		}
 	}
 }
@@ -1718,16 +1718,16 @@ func testClientGetTimeoutError(t *testing.T, c *Client, n int) {
 	for i := 0; i < n; i++ {
 		statusCode, body, err := c.GetTimeout(buf, "http://foobar.com/baz", time.Millisecond)
 		if err == nil {
-			t.Fatalf("expecting error")
+			t.Errorf("expecting error")
 		}
 		if err != ErrTimeout {
-			t.Fatalf("unexpected error: %s. Expecting %s", err, ErrTimeout)
+			t.Errorf("unexpected error: %s. Expecting %s", err, ErrTimeout)
 		}
 		if statusCode != 0 {
-			t.Fatalf("unexpected statusCode=%d. Expecting %d", statusCode, 0)
+			t.Errorf("unexpected statusCode=%d. Expecting %d", statusCode, 0)
 		}
 		if body == nil {
-			t.Fatalf("body must be non-nil")
+			t.Errorf("body must be non-nil")
 		}
 	}
 }
@@ -2326,14 +2326,14 @@ func testClientGet(t *testing.T, c clientGetter, addr string, n int) {
 		statusCode, body, err := c.Get(buf, uri)
 		buf = body
 		if err != nil {
-			t.Fatalf("unexpected error when doing http request: %s", err)
+			t.Errorf("unexpected error when doing http request: %s", err)
 		}
 		if statusCode != StatusOK {
-			t.Fatalf("unexpected status code: %d. Expecting %d", statusCode, StatusOK)
+			t.Errorf("unexpected status code: %d. Expecting %d", statusCode, StatusOK)
 		}
 		resultURI := string(body)
 		if resultURI != uri {
-			t.Fatalf("unexpected uri %q. Expecting %q", resultURI, uri)
+			t.Errorf("unexpected uri %q. Expecting %q", resultURI, uri)
 		}
 	}
 }
@@ -2346,17 +2346,17 @@ func testClientDoTimeoutSuccess(t *testing.T, c *Client, addr string, n int) {
 		uri := fmt.Sprintf("%s/foo/%d?bar=baz", addr, i)
 		req.SetRequestURI(uri)
 		if err := c.DoTimeout(&req, &resp, time.Second); err != nil {
-			t.Fatalf("unexpected error: %s", err)
+			t.Errorf("unexpected error: %s", err)
 		}
 		if resp.StatusCode() != StatusOK {
-			t.Fatalf("unexpected status code: %d. Expecting %d", resp.StatusCode(), StatusOK)
+			t.Errorf("unexpected status code: %d. Expecting %d", resp.StatusCode(), StatusOK)
 		}
 		resultURI := string(resp.Body())
 		if strings.HasPrefix(uri, "https") {
 			resultURI = uri[:5] + resultURI[4:]
 		}
 		if resultURI != uri {
-			t.Fatalf("unexpected uri %q. Expecting %q", resultURI, uri)
+			t.Errorf("unexpected uri %q. Expecting %q", resultURI, uri)
 		}
 	}
 }
@@ -2368,17 +2368,17 @@ func testClientGetTimeoutSuccess(t *testing.T, c *Client, addr string, n int) {
 		statusCode, body, err := c.GetTimeout(buf, uri, time.Second)
 		buf = body
 		if err != nil {
-			t.Fatalf("unexpected error when doing http request: %s", err)
+			t.Errorf("unexpected error when doing http request: %s", err)
 		}
 		if statusCode != StatusOK {
-			t.Fatalf("unexpected status code: %d. Expecting %d", statusCode, StatusOK)
+			t.Errorf("unexpected status code: %d. Expecting %d", statusCode, StatusOK)
 		}
 		resultURI := string(body)
 		if strings.HasPrefix(uri, "https") {
 			resultURI = uri[:5] + resultURI[4:]
 		}
 		if resultURI != uri {
-			t.Fatalf("unexpected uri %q. Expecting %q", resultURI, uri)
+			t.Errorf("unexpected uri %q. Expecting %q", resultURI, uri)
 		}
 	}
 }
@@ -2394,14 +2394,14 @@ func testClientPost(t *testing.T, c clientPoster, addr string, n int) {
 		statusCode, body, err := c.Post(buf, uri, &args)
 		buf = body
 		if err != nil {
-			t.Fatalf("unexpected error when doing http request: %s", err)
+			t.Errorf("unexpected error when doing http request: %s", err)
 		}
 		if statusCode != StatusOK {
-			t.Fatalf("unexpected status code: %d. Expecting %d", statusCode, StatusOK)
+			t.Errorf("unexpected status code: %d. Expecting %d", statusCode, StatusOK)
 		}
 		s := string(body)
 		if s != argsS {
-			t.Fatalf("unexpected response %q. Expecting %q", s, argsS)
+			t.Errorf("unexpected response %q. Expecting %q", s, argsS)
 		}
 	}
 }
@@ -2837,6 +2837,6 @@ func TestHttpsRequestWithoutParsedURL(t *testing.T) {
 
 	_, err := client.doNonNilReqResp(req, &Response{})
 	if err != nil {
-		t.Fatalf("https requests with IsTLS client must succeed")
+		t.Fatal("https requests with IsTLS client must succeed")
 	}
 }
diff --git a/client_unix_test.go b/client_unix_test.go
new file mode 100644
index 0000000..f369111
--- /dev/null
+++ b/client_unix_test.go
@@ -0,0 +1,136 @@
+//go:build !windows
+// +build !windows
+
+package fasthttp
+
+import (
+	"io"
+	"io/ioutil"
+	"net"
+	"net/http"
+	"strings"
+	"testing"
+)
+
+// See issue #1232
+func TestRstConnResponseWhileSending(t *testing.T) {
+	const expectedStatus = http.StatusTeapot
+	const payload = "payload"
+
+	srv, err := net.Listen("tcp", "127.0.0.1:0")
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer srv.Close()
+
+	go func() {
+		for {
+			conn, err := srv.Accept()
+			if err != nil {
+				return
+			}
+
+			// Read at least one byte of the header
+			// Otherwise we would have an unsolicited response
+			_, err = ioutil.ReadAll(io.LimitReader(conn, 1))
+			if err != nil {
+				t.Error(err)
+			}
+
+			// Respond
+			_, err = conn.Write([]byte("HTTP/1.1 418 Teapot\r\n\r\n"))
+			if err != nil {
+				t.Error(err)
+			}
+
+			// Forcefully close connection
+			err = conn.(*net.TCPConn).SetLinger(0)
+			if err != nil {
+				t.Error(err)
+			}
+			conn.Close()
+		}
+	}()
+
+	svrUrl := "http://" + srv.Addr().String()
+	client := HostClient{Addr: srv.Addr().String()}
+
+	for i := 0; i < 100; i++ {
+		req := AcquireRequest()
+		defer ReleaseRequest(req)
+		resp := AcquireResponse()
+		defer ReleaseResponse(resp)
+
+		req.Header.SetMethod("POST")
+		req.SetBodyStream(strings.NewReader(payload), len(payload))
+		req.SetRequestURI(svrUrl)
+
+		err = client.Do(req, resp)
+		if err != nil {
+			t.Fatal(err)
+		}
+		if expectedStatus != resp.StatusCode() {
+			t.Fatalf("Expected %d status code, but got %d", expectedStatus, resp.StatusCode())
+		}
+	}
+}
+
+// See issue #1232
+func TestRstConnClosedWithoutResponse(t *testing.T) {
+	const payload = "payload"
+
+	srv, err := net.Listen("tcp", "127.0.0.1:0")
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer srv.Close()
+
+	go func() {
+		for {
+			conn, err := srv.Accept()
+			if err != nil {
+				return
+			}
+
+			// Read at least one byte of the header
+			// Otherwise we would have an unsolicited response
+			_, err = ioutil.ReadAll(io.LimitReader(conn, 1))
+			if err != nil {
+				t.Error(err)
+			}
+
+			// Respond with incomplete header
+			_, err = conn.Write([]byte("Http"))
+			if err != nil {
+				t.Error(err)
+			}
+
+			// Forcefully close connection
+			err = conn.(*net.TCPConn).SetLinger(0)
+			if err != nil {
+				t.Error(err)
+			}
+			conn.Close()
+		}
+	}()
+
+	svrUrl := "http://" + srv.Addr().String()
+	client := HostClient{Addr: srv.Addr().String()}
+
+	for i := 0; i < 100; i++ {
+		req := AcquireRequest()
+		defer ReleaseRequest(req)
+		resp := AcquireResponse()
+		defer ReleaseResponse(resp)
+
+		req.Header.SetMethod("POST")
+		req.SetBodyStream(strings.NewReader(payload), len(payload))
+		req.SetRequestURI(svrUrl)
+
+		err = client.Do(req, resp)
+
+		if !isConnectionReset(err) {
+			t.Fatal("Expected connection reset error")
+		}
+	}
+}