Repository: valyala/fasthttp
Pull request #770 (2020-04-01 23:29:27)

Added fastws (alphabetically) to related projects

dgrr: @erikdubbelboer 

Patch:
diff --git a/.gitignore b/.gitignore
index 7b58ce4..220a3f8 100644
--- a/.gitignore
+++ b/.gitignore
@@ -1,3 +1,4 @@
 tags
 *.pprof
 *.fasthttp.gz
+.idea
\ No newline at end of file
diff --git a/README.md b/README.md
index 95270a3..1678960 100644
--- a/README.md
+++ b/README.md
@@ -485,6 +485,8 @@ uintBuf := fasthttp.AppendUint(nil, 1234)
     powerful routing package for fasthttp servers.
   * [fasthttprouter](https://github.com/buaazp/fasthttprouter) - a high
     performance fasthttp request router that scales well.
+  * [fastws](https://github.com/fasthttp/fastws) - Bloatless WebSocket package made for fasthttp
+    to handle Read/Write operations concurrently.
   * [gramework](https://github.com/gramework/gramework) - a web framework made by one of fasthttp maintainers
   * [lu](https://github.com/vincentLiuxiang/lu) - a high performance
     go middleware web framework which is based on fasthttp.
diff --git a/client.go b/client.go
index 3a7108c..7665312 100644
--- a/client.go
+++ b/client.go
@@ -284,6 +284,11 @@ type Client struct {
 	// extra slashes are removed, special characters are encoded.
 	DisablePathNormalizing bool
 
+	// Maximum duration for waiting for a free connection.
+	//
+	// By default will not waiting, return ErrNoFreeConns immediately
+	MaxConnWaitTimeout time.Duration
+
 	mLock sync.Mutex
 	m     map[string]*HostClient
 	ms    map[string]*HostClient
@@ -487,6 +492,7 @@ func (c *Client) Do(req *Request, resp *Response) error {
 			MaxResponseBodySize:           c.MaxResponseBodySize,
 			DisableHeaderNamesNormalizing: c.DisableHeaderNamesNormalizing,
 			DisablePathNormalizing:        c.DisablePathNormalizing,
+			MaxConnWaitTimeout:            c.MaxConnWaitTimeout,
 		}
 		m[string(host)] = hc
 		if len(m) == 1 {
@@ -687,12 +693,18 @@ type HostClient struct {
 	// extra slashes are removed, special characters are encoded.
 	DisablePathNormalizing bool
 
+	// Maximum duration for waiting for a free connection.
+	//
+	// By default will not waiting, return ErrNoFreeConns immediately
+	MaxConnWaitTimeout time.Duration
+
 	clientName  atomic.Value
 	lastUseTime uint32
 
 	connsLock  sync.Mutex
 	connsCount int
 	conns      []*clientConn
+	connsWait  *wantConnQueue
 
 	addrsLock sync.Mutex
 	addrs     []string
@@ -1119,6 +1131,7 @@ func clientDoDeadline(req *Request, resp *Response, deadline time.Time, c client
 	var timedout bool
 
 	go func() {
+		reqCopy.timeout = timeout
 		errDo := c.Do(reqCopy, respCopy)
 		mu.Lock()
 		{
@@ -1282,7 +1295,7 @@ func (c *HostClient) doNonNilReqResp(req *Request, resp *Response) (bool, error)
 		req.SetConnectionClose()
 	}
 
-	cc, err := c.acquireConn()
+	cc, err := c.acquireConn(req.timeout)
 	if err != nil {
 		return false, err
 	}
@@ -1409,8 +1422,7 @@ func (c *HostClient) SetMaxConns(newMaxConns int) {
 	c.connsLock.Unlock()
 }
 
-func (c *HostClient) acquireConn() (*clientConn, error) {
-	var cc *clientConn
+func (c *HostClient) acquireConn(reqTimeout time.Duration) (cc *clientConn, err error) {
 	createConn := false
 	startCleaner := false
 
@@ -1442,7 +1454,47 @@ func (c *HostClient) acquireConn() (*clientConn, error) {
 		return cc, nil
 	}
 	if !createConn {
-		return nil, ErrNoFreeConns
+		if c.MaxConnWaitTimeout <= 0 {
+			return nil, ErrNoFreeConns
+		}
+
+		// reqTimeout    c.MaxConnWaitTimeout   wait duration
+		//     d1                 d2            min(d1, d2)
+		//  0(not set)            d2            d2
+		//     d1            0(don't wait)      0(don't wait)
+		//  0(not set)            d2            d2
+		timeout := c.MaxConnWaitTimeout
+		timeoutOverridden := false
+		// reqTimeout == 0 means not set
+		if reqTimeout > 0 && reqTimeout < timeout {
+				timeout = reqTimeout
+				timeoutOverridden = true
+		}
+
+		// wait for a free connection
+		tc := AcquireTimer(timeout)
+		defer ReleaseTimer(tc)
+
+		w := &wantConn{
+			ready: make(chan struct{}, 1),
+		}
+		defer func() {
+			if err != nil {
+				w.cancel(c, err)
+			}
+		}()
+
+		c.queueForIdle(w)
+
+		select {
+		case <-w.ready:
+			return w.conn, w.err
+		case <-tc.C:
+			if timeoutOverridden {
+				return nil, ErrTimeout
+			}
+			return nil, ErrNoFreeConns
+		}
 	}
 
 	if startCleaner {
@@ -1459,6 +1511,33 @@ func (c *HostClient) acquireConn() (*clientConn, error) {
 	return cc, nil
 }
 
+func (c *HostClient) queueForIdle(w *wantConn) {
+	c.connsLock.Lock()
+	defer c.connsLock.Unlock()
+	if c.connsWait == nil {
+		c.connsWait = &wantConnQueue{}
+	}
+	c.connsWait.clearFront()
+	c.connsWait.pushBack(w)
+}
+
+func (c *HostClient) dialConnFor(w *wantConn) {
+	conn, err := c.dialHostHard()
+
+	if err != nil {
+		w.tryDeliver(nil, err)
+		c.decConnsCount()
+		return
+	}
+
+	cc := acquireClientConn(conn)
+	delivered := w.tryDeliver(cc, nil)
+	if !delivered {
+		// not delivered, return idle connection
+		c.releaseConn(cc)
+	}
+}
+
 func (c *HostClient) connsCleaner() {
 	var (
 		scratch             []*clientConn
@@ -1522,9 +1601,30 @@ func (c *HostClient) closeConn(cc *clientConn) {
 }
 
 func (c *HostClient) decConnsCount() {
+	if c.MaxConnWaitTimeout <= 0 {
+		c.connsLock.Lock()
+		c.connsCount--
+		c.connsLock.Unlock()
+		return
+	}
+
 	c.connsLock.Lock()
-	c.connsCount--
-	c.connsLock.Unlock()
+	defer c.connsLock.Unlock()
+	dialed := false
+	if q := c.connsWait; q != nil && q.len() > 0 {
+		for q.len() > 0 {
+			w := q.popFront()
+			if w.waiting() {
+				go c.dialConnFor(w)
+				dialed = true
+				break
+			}
+		}
+	}
+	if !dialed {
+		c.connsCount--
+	}
+
 }
 
 func acquireClientConn(conn net.Conn) *clientConn {
@@ -1548,9 +1648,29 @@ var clientConnPool sync.Pool
 
 func (c *HostClient) releaseConn(cc *clientConn) {
 	cc.lastUseTime = time.Now()
+	if c.MaxConnWaitTimeout <= 0 {
+		c.connsLock.Lock()
+		c.conns = append(c.conns, cc)
+		c.connsLock.Unlock()
+		return
+	}
+
+	// try to deliver an idle connection to a *wantConn
 	c.connsLock.Lock()
-	c.conns = append(c.conns, cc)
-	c.connsLock.Unlock()
+	defer c.connsLock.Unlock()
+	delivered := false
+	if q := c.connsWait; q != nil && q.len() > 0 {
+		for q.len() > 0 {
+			w := q.popFront()
+			if w.waiting() {
+				delivered = w.tryDeliver(cc, nil)
+				break
+			}
+		}
+	}
+	if !delivered {
+		c.conns = append(c.conns, cc)
+	}
 }
 
 func (c *HostClient) acquireWriter(conn net.Conn) *bufio.Writer {
@@ -1776,6 +1896,137 @@ func addMissingPort(addr string, isTLS bool) string {
 	return net.JoinHostPort(addr, strconv.Itoa(port))
 }
 
+// A wantConn records state about a wanted connection
+// (that is, an active call to getConn).
+// The conn may be gotten by dialing or by finding an idle connection,
+// or a cancellation may make the conn no longer wanted.
+// These three options are racing against each other and use
+// wantConn to coordinate and agree about the winning outcome.
+//
+// inspired by net/http/transport.go
+type wantConn struct {
+	ready chan struct{}
+	mu    sync.Mutex // protects conn, err, close(ready)
+	conn  *clientConn
+	err   error
+}
+
+// waiting reports whether w is still waiting for an answer (connection or error).
+func (w *wantConn) waiting() bool {
+	select {
+	case <-w.ready:
+		return false
+	default:
+		return true
+	}
+}
+
+// tryDeliver attempts to deliver conn, err to w and reports whether it succeeded.
+func (w *wantConn) tryDeliver(conn *clientConn, err error) bool {
+	w.mu.Lock()
+	defer w.mu.Unlock()
+
+	if w.conn != nil || w.err != nil {
+		return false
+	}
+	w.conn = conn
+	w.err = err
+	if w.conn == nil && w.err == nil {
+		panic("fasthttp: internal error: misuse of tryDeliver")
+	}
+	close(w.ready)
+	return true
+}
+
+// cancel marks w as no longer wanting a result (for example, due to cancellation).
+// If a connection has been delivered already, cancel returns it with c.releaseConn.
+func (w *wantConn) cancel(c *HostClient, err error) {
+	w.mu.Lock()
+	if w.conn == nil && w.err == nil {
+		close(w.ready) // catch misbehavior in future delivery
+	}
+
+	conn := w.conn
+	w.conn = nil
+	w.err = err
+	w.mu.Unlock()
+
+	if conn != nil {
+		c.releaseConn(conn)
+	}
+}
+
+// A wantConnQueue is a queue of wantConns.
+//
+// inspired by net/http/transport.go
+type wantConnQueue struct {
+	// This is a queue, not a deque.
+	// It is split into two stages - head[headPos:] and tail.
+	// popFront is trivial (headPos++) on the first stage, and
+	// pushBack is trivial (append) on the second stage.
+	// If the first stage is empty, popFront can swap the
+	// first and second stages to remedy the situation.
+	//
+	// This two-stage split is analogous to the use of two lists
+	// in Okasaki's purely functional queue but without the
+	// overhead of reversing the list when swapping stages.
+	head    []*wantConn
+	headPos int
+	tail    []*wantConn
+}
+
+// len returns the number of items in the queue.
+func (q *wantConnQueue) len() int {
+	return len(q.head) - q.headPos + len(q.tail)
+}
+
+// pushBack adds w to the back of the queue.
+func (q *wantConnQueue) pushBack(w *wantConn) {
+	q.tail = append(q.tail, w)
+}
+
+// popFront removes and returns the wantConn at the front of the queue.
+func (q *wantConnQueue) popFront() *wantConn {
+	if q.headPos >= len(q.head) {
+		if len(q.tail) == 0 {
+			return nil
+		}
+		// Pick up tail as new head, clear tail.
+		q.head, q.headPos, q.tail = q.tail, 0, q.head[:0]
+	}
+
+	w := q.head[q.headPos]
+	q.head[q.headPos] = nil
+	q.headPos++
+	return w
+}
+
+// peekFront returns the wantConn at the front of the queue without removing it.
+func (q *wantConnQueue) peekFront() *wantConn {
+
+	if q.headPos < len(q.head) {
+		return q.head[q.headPos]
+	}
+	if len(q.tail) > 0 {
+		return q.tail[0]
+	}
+	return nil
+
+}
+
+// cleanFront pops any wantConns that are no longer waiting from the head of the
+// queue, reporting whether any were popped.
+func (q *wantConnQueue) clearFront() (cleaned bool) {
+	for {
+		w := q.peekFront()
+		if w == nil || w.waiting() {
+			return cleaned
+		}
+		q.popFront()
+		cleaned = true
+	}
+}
+
 // PipelineClient pipelines requests over a limited set of concurrent
 // connections to the given Addr.
 //
diff --git a/http.go b/http.go
index 9b1bcc0..fbb54c7 100644
--- a/http.go
+++ b/http.go
@@ -11,6 +11,7 @@ import (
 	"net"
 	"os"
 	"sync"
+	"time"
 
 	"github.com/valyala/bytebufferpool"
 )
@@ -49,6 +50,10 @@ type Request struct {
 
 	// To detect scheme changes in redirects
 	schemaUpdate bool
+
+	// Request timeout. Usually set by DoDealine or DoTimeout
+	// if <= 0, means not set
+	timeout time.Duration
 }
 
 // Response represents HTTP response.
@@ -845,6 +850,7 @@ func readMultipartForm(r io.Reader, boundary string, size, maxInMemoryFileSize i
 func (req *Request) Reset() {
 	req.Header.Reset()
 	req.resetSkipHeader()
+	req.timeout = 0
 }
 
 func (req *Request) resetSkipHeader() {


Test patch:
diff --git a/client_test.go b/client_test.go
index cdf2250..59bc413 100644
--- a/client_test.go
+++ b/client_test.go
@@ -2120,3 +2120,264 @@ func TestClientTLSHandshakeTimeout(t *testing.T) {
 		t.Errorf("resulting error not a timeout: %v\nType %T: %#v", err, err, err)
 	}
 }
+
+func TestHostClientMaxConnWaitTimeoutSuccess(t *testing.T) {
+	var (
+		emptyBodyCount uint8
+		ln             = fasthttputil.NewInmemoryListener()
+		wg             sync.WaitGroup
+	)
+
+	s := &Server{
+		Handler: func(ctx *RequestCtx) {
+			if len(ctx.PostBody()) == 0 {
+				emptyBodyCount++
+			}
+			time.Sleep(5 * time.Millisecond)
+			ctx.WriteString("foo") //nolint:errcheck
+		},
+	}
+	serverStopCh := make(chan struct{})
+	go func() {
+		if err := s.Serve(ln); err != nil {
+			t.Errorf("unexpected error: %s", err)
+		}
+		close(serverStopCh)
+	}()
+
+	c := &HostClient{
+		Addr: "foobar",
+		Dial: func(addr string) (net.Conn, error) {
+			return ln.Dial()
+		},
+		MaxConns:           1,
+		MaxConnWaitTimeout: 200 * time.Millisecond,
+	}
+
+	for i := 0; i < 5; i++ {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+
+			req := AcquireRequest()
+			req.SetRequestURI("http://foobar/baz")
+			req.Header.SetMethod(MethodPost)
+			req.SetBodyString("bar")
+			resp := AcquireResponse()
+
+			if err := c.Do(req, resp); err != nil {
+				t.Errorf("unexpected error: %s", err)
+			}
+
+			if resp.StatusCode() != StatusOK {
+				t.Errorf("unexpected status code %d. Expecting %d", resp.StatusCode(), StatusOK)
+			}
+
+			body := resp.Body()
+			if string(body) != "foo" {
+				t.Errorf("unexpected body %q. Expecting %q", body, "abcd")
+			}
+		}()
+	}
+	wg.Wait()
+
+	if c.connsWait.len() > 0 {
+		t.Errorf("connsWait has %v items remaining", c.connsWait.len())
+	}
+	if err := ln.Close(); err != nil {
+		t.Fatalf("unexpected error: %s", err)
+	}
+	select {
+	case <-serverStopCh:
+	case <-time.After(time.Second):
+		t.Fatalf("timeout")
+	}
+
+	if emptyBodyCount > 0 {
+		t.Fatalf("at least one request body was empty")
+	}
+}
+
+func TestHostClientMaxConnWaitTimeoutError(t *testing.T) {
+	var (
+		emptyBodyCount uint8
+		ln             = fasthttputil.NewInmemoryListener()
+		wg             sync.WaitGroup
+	)
+
+	s := &Server{
+		Handler: func(ctx *RequestCtx) {
+			if len(ctx.PostBody()) == 0 {
+				emptyBodyCount++
+			}
+			time.Sleep(5 * time.Millisecond)
+			ctx.WriteString("foo") //nolint:errcheck
+		},
+	}
+	serverStopCh := make(chan struct{})
+	go func() {
+		if err := s.Serve(ln); err != nil {
+			t.Errorf("unexpected error: %s", err)
+		}
+		close(serverStopCh)
+	}()
+
+	c := &HostClient{
+		Addr: "foobar",
+		Dial: func(addr string) (net.Conn, error) {
+			return ln.Dial()
+		},
+		MaxConns:           1,
+		MaxConnWaitTimeout: 10 * time.Millisecond,
+	}
+
+	var errNoFreeConnsCount uint32
+	for i := 0; i < 5; i++ {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+
+			req := AcquireRequest()
+			req.SetRequestURI("http://foobar/baz")
+			req.Header.SetMethod(MethodPost)
+			req.SetBodyString("bar")
+			resp := AcquireResponse()
+
+			if err := c.Do(req, resp); err != nil {
+				if err != ErrNoFreeConns {
+					t.Errorf("unexpected error: %s. Expecting %s", err, ErrNoFreeConns)
+				}
+				atomic.AddUint32(&errNoFreeConnsCount, 1)
+			} else {
+				if resp.StatusCode() != StatusOK {
+					t.Errorf("unexpected status code %d. Expecting %d", resp.StatusCode(), StatusOK)
+				}
+
+				body := resp.Body()
+				if string(body) != "foo" {
+					t.Errorf("unexpected body %q. Expecting %q", body, "abcd")
+				}
+			}
+
+		}()
+	}
+	wg.Wait()
+
+	if c.connsWait.len() > 0 {
+		t.Errorf("connsWait has %v items remaining", c.connsWait.len())
+	}
+	if errNoFreeConnsCount == 0 {
+		t.Errorf("unexpected errorCount: %d. Expecting > 0", errNoFreeConnsCount)
+	}
+	if err := ln.Close(); err != nil {
+		t.Fatalf("unexpected error: %s", err)
+	}
+	select {
+	case <-serverStopCh:
+	case <-time.After(time.Second):
+		t.Fatalf("timeout")
+	}
+
+	if emptyBodyCount > 0 {
+		t.Fatalf("at least one request body was empty")
+	}
+}
+
+func TestHostClientMaxConnWaitTimeoutWithEarlierDeadline(t *testing.T) {
+	var (
+		emptyBodyCount uint8
+		ln             = fasthttputil.NewInmemoryListener()
+		wg             sync.WaitGroup
+		// make deadline reach earlier than conns wait timeout
+		sleep              = 100 * time.Millisecond
+		timeout            = 10 * time.Millisecond
+		maxConnWaitTimeout = 50 * time.Millisecond
+	)
+
+	s := &Server{
+		Handler: func(ctx *RequestCtx) {
+			if len(ctx.PostBody()) == 0 {
+				emptyBodyCount++
+			}
+			time.Sleep(sleep)
+			ctx.WriteString("foo") //nolint:errcheck
+		},
+	}
+	serverStopCh := make(chan struct{})
+	go func() {
+		if err := s.Serve(ln); err != nil {
+			t.Errorf("unexpected error: %s", err)
+		}
+		close(serverStopCh)
+	}()
+
+	c := &HostClient{
+		Addr: "foobar",
+		Dial: func(addr string) (net.Conn, error) {
+			return ln.Dial()
+		},
+		MaxConns:           1,
+		MaxConnWaitTimeout: maxConnWaitTimeout,
+	}
+
+	var errTimeoutCount uint32
+	for i := 0; i < 5; i++ {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+
+			req := AcquireRequest()
+			req.SetRequestURI("http://foobar/baz")
+			req.Header.SetMethod(MethodPost)
+			req.SetBodyString("bar")
+			resp := AcquireResponse()
+
+			if err := c.DoDeadline(req, resp, time.Now().Add(timeout)); err != nil {
+				if err != ErrTimeout {
+					t.Errorf("unexpected error: %s. Expecting %s", err, ErrTimeout)
+				}
+				atomic.AddUint32(&errTimeoutCount, 1)
+			} else {
+				if resp.StatusCode() != StatusOK {
+					t.Errorf("unexpected status code %d. Expecting %d", resp.StatusCode(), StatusOK)
+				}
+
+				body := resp.Body()
+				if string(body) != "foo" {
+					t.Errorf("unexpected body %q. Expecting %q", body, "abcd")
+				}
+			}
+
+		}()
+	}
+	wg.Wait()
+
+	c.connsLock.Lock()
+	for {
+		w := c.connsWait.popFront()
+		if w == nil {
+			break
+		}
+		w.mu.Lock()
+		if w.err != nil && w.err != ErrTimeout {
+			t.Errorf("unexpected error: %s. Expecting %s", w.err, ErrTimeout)
+		}
+		w.mu.Unlock()
+	}
+	c.connsLock.Unlock()
+	if errTimeoutCount == 0 {
+		t.Errorf("unexpected errTimeoutCount: %d. Expecting > 0", errTimeoutCount)
+	}
+	if err := ln.Close(); err != nil {
+		t.Fatalf("unexpected error: %s", err)
+	}
+	select {
+	case <-serverStopCh:
+	case <-time.After(time.Second):
+		t.Fatalf("timeout")
+	}
+
+	if emptyBodyCount > 0 {
+		t.Fatalf("at least one request body was empty")
+	}
+}
diff --git a/client_timing_wait_test.go b/client_timing_wait_test.go
new file mode 100644
index 0000000..1d1f767
--- /dev/null
+++ b/client_timing_wait_test.go
@@ -0,0 +1,166 @@
+// +build go1.11
+
+package fasthttp
+
+import (
+	"io/ioutil"
+	"net"
+	"net/http"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/valyala/fasthttp/fasthttputil"
+)
+
+func newFasthttpSleepEchoHandler(sleep time.Duration) RequestHandler {
+	return func(ctx *RequestCtx) {
+		time.Sleep(sleep)
+		ctx.Success("text/plain", ctx.RequestURI())
+	}
+}
+
+func BenchmarkClientGetEndToEndWaitConn1Inmemory(b *testing.B) {
+	benchmarkClientGetEndToEndWaitConnInmemory(b, 1)
+}
+
+func BenchmarkClientGetEndToEndWaitConn10Inmemory(b *testing.B) {
+	benchmarkClientGetEndToEndWaitConnInmemory(b, 10)
+}
+
+func BenchmarkClientGetEndToEndWaitConn100Inmemory(b *testing.B) {
+	benchmarkClientGetEndToEndWaitConnInmemory(b, 100)
+}
+
+func BenchmarkClientGetEndToEndWaitConn1000Inmemory(b *testing.B) {
+	benchmarkClientGetEndToEndWaitConnInmemory(b, 1000)
+}
+
+func benchmarkClientGetEndToEndWaitConnInmemory(b *testing.B, parallelism int) {
+	ln := fasthttputil.NewInmemoryListener()
+
+	ch := make(chan struct{})
+	sleepDuration := 50 * time.Millisecond
+	go func() {
+
+		if err := Serve(ln, newFasthttpSleepEchoHandler(sleepDuration)); err != nil {
+			b.Errorf("error when serving requests: %s", err)
+		}
+		close(ch)
+	}()
+
+	c := &Client{
+		MaxConnsPerHost:    1,
+		Dial:               func(addr string) (net.Conn, error) { return ln.Dial() },
+		MaxConnWaitTimeout: 5 * time.Second,
+	}
+
+	requestURI := "/foo/bar?baz=123&sleep=10ms"
+	url := "http://unused.host" + requestURI
+	b.SetParallelism(parallelism)
+	b.RunParallel(func(pb *testing.PB) {
+		var buf []byte
+		for pb.Next() {
+			statusCode, body, err := c.Get(buf, url)
+			if err != nil {
+				if err != ErrNoFreeConns {
+					b.Fatalf("unexpected error: %s", err)
+				}
+			} else {
+				if statusCode != StatusOK {
+					b.Fatalf("unexpected status code: %d. Expecting %d", statusCode, StatusOK)
+				}
+				if string(body) != requestURI {
+					b.Fatalf("unexpected response %q. Expecting %q", body, requestURI)
+				}
+			}
+			buf = body
+		}
+	})
+
+	ln.Close()
+	select {
+	case <-ch:
+	case <-time.After(time.Second):
+		b.Fatalf("server wasn't stopped")
+	}
+}
+
+func newNethttpSleepEchoHandler(sleep time.Duration) http.HandlerFunc {
+	return func(w http.ResponseWriter, r *http.Request) {
+		time.Sleep(sleep)
+		w.Header().Set(HeaderContentType, "text/plain")
+		w.Write([]byte(r.RequestURI)) //nolint:errcheck
+	}
+}
+
+func BenchmarkNetHTTPClientGetEndToEndWaitConn1Inmemory(b *testing.B) {
+	benchmarkNetHTTPClientGetEndToEndWaitConnInmemory(b, 1)
+}
+
+func BenchmarkNetHTTPClientGetEndToEndWaitConn10Inmemory(b *testing.B) {
+	benchmarkNetHTTPClientGetEndToEndWaitConnInmemory(b, 10)
+}
+
+func BenchmarkNetHTTPClientGetEndToEndWaitConn100Inmemory(b *testing.B) {
+	benchmarkNetHTTPClientGetEndToEndWaitConnInmemory(b, 100)
+}
+
+func BenchmarkNetHTTPClientGetEndToEndWaitConn1000Inmemory(b *testing.B) {
+	benchmarkNetHTTPClientGetEndToEndWaitConnInmemory(b, 1000)
+}
+
+func benchmarkNetHTTPClientGetEndToEndWaitConnInmemory(b *testing.B, parallelism int) {
+	ln := fasthttputil.NewInmemoryListener()
+
+	ch := make(chan struct{})
+	sleep := 50 * time.Millisecond
+	go func() {
+		if err := http.Serve(ln, newNethttpSleepEchoHandler(sleep)); err != nil && !strings.Contains(
+			err.Error(), "use of closed network connection") {
+			b.Errorf("error when serving requests: %s", err)
+		}
+		close(ch)
+	}()
+
+	c := &http.Client{
+		Transport: &http.Transport{
+			Dial:            func(_, _ string) (net.Conn, error) { return ln.Dial() },
+			MaxConnsPerHost: 1,
+		},
+		Timeout: 5 * time.Second,
+	}
+
+	requestURI := "/foo/bar?baz=123"
+	url := "http://unused.host" + requestURI
+	b.SetParallelism(parallelism)
+	b.RunParallel(func(pb *testing.PB) {
+		for pb.Next() {
+			resp, err := c.Get(url)
+			if err != nil {
+				if netErr, ok := err.(net.Error); !ok || !netErr.Timeout() {
+					b.Fatalf("unexpected error: %s", err)
+				}
+			} else {
+				if resp.StatusCode != http.StatusOK {
+					b.Fatalf("unexpected status code: %d. Expecting %d", resp.StatusCode, http.StatusOK)
+				}
+				body, err := ioutil.ReadAll(resp.Body)
+				resp.Body.Close()
+				if err != nil {
+					b.Fatalf("unexpected error when reading response body: %s", err)
+				}
+				if string(body) != requestURI {
+					b.Fatalf("unexpected response %q. Expecting %q", body, requestURI)
+				}
+			}
+		}
+	})
+
+	ln.Close()
+	select {
+	case <-ch:
+	case <-time.After(time.Second):
+		b.Fatalf("server wasn't stopped")
+	}
+}