Repository: valyala/fasthttp
Pull request #764 (2020-03-13 14:34:51)

feat: make client to wait when no free connections
Add a new field `MaxConnWaitTimeout` on Client. 
* If `MaxConnWaitTimeout` == 0, when all connections are busy, will return `ErrNoFreeConns` immediately (just like what it already does). 
* If `MaxConnWaitTimeout` > 0, when all connections are busy, Client will wait at most `MaxConnWaitTimeout` for a chance to get a connection, if Client failed, will return `ErrNoFreeConns`



chanjarster: > First thing I notice is a lot of potential heap allocations which can slow things down. Did you try doing any benchmarks or checking the number of allocations with a test like this?:
> https://github.com/valyala/fasthttp/blob/38aa88ab52c4b6f61722e2e7809c17b6f7df2b2a/allocation_test.go#L40-L72

I've noticed that the current benchmark and allocation test are based on fast server. And, by default (which `MaxConnWaitTimeout=0`) this implementation doesn't queue the request and return ErrNoFreeConns immediately.

I'll provide a allocation test based on slow server and enable `MaxConnWaitTimeout` later.
chanjarster: Hi @erikdubbelboer , I add `ClientGetEndToEndWaitConn` benchmarks to show heap allocations (see commit cb96d97), below is my test result.

```txt
$ GOMAXPROCS=4 go test -bench='kClientGetEndToEndWaitConn' -benchmem -benchtime=10s
goos: darwin
goarch: amd64
pkg: github.com/valyala/fasthttp
BenchmarkClientGetEndToEndWaitConn1Inmemory-4      	     237	  50495156 ns/op	     917 B/op	       4 allocs/op
BenchmarkClientGetEndToEndWaitConn10Inmemory-4     	     397	  25392398 ns/op	     397 B/op	       3 allocs/op
BenchmarkClientGetEndToEndWaitConn100Inmemory-4    	    4700	   2565375 ns/op	     193 B/op	       2 allocs/op
BenchmarkClientGetEndToEndWaitConn1000Inmemory-4   	   41949	    263421 ns/op	     183 B/op	       2 allocs/op
BenchmarkClientGetEndToEndWaitConn10kInmemory-4    	  414907	     29407 ns/op	     183 B/op	       2 allocs/op
PASS
ok  	github.com/valyala/fasthttp	87.533s
```
chanjarster: Hi @erikdubbelboer , I add some benchmarks to show heap allocations (see commit cb96d97 and b3a83d1) when in situation of waiting for a free connection, below is my test result:

fasthttp:

```txt
$ GOMAXPROCS=4 go test -bench='kClientGetEndToEndWaitConn' -benchmem -benchtime=10s
goos: darwin
goarch: amd64
pkg: github.com/valyala/fasthttp
BenchmarkClientGetEndToEndWaitConn1Inmemory-4      	     237	  50495156 ns/op	     917 B/op	       4 allocs/op
BenchmarkClientGetEndToEndWaitConn10Inmemory-4     	     397	  25392398 ns/op	     397 B/op	       3 allocs/op
BenchmarkClientGetEndToEndWaitConn100Inmemory-4    	    4700	   2565375 ns/op	     193 B/op	       2 allocs/op
BenchmarkClientGetEndToEndWaitConn1000Inmemory-4   	   41949	    263421 ns/op	     183 B/op	       2 allocs/op
BenchmarkClientGetEndToEndWaitConn10kInmemory-4    	  414907	     29407 ns/op	     183 B/op	       2 allocs/op
PASS
ok  	github.com/valyala/fasthttp	87.533s
```

net/http:

```txt
$ GOMAXPROCS=4 go test -bench='HTTPClientGetEndToEndWaitConn' -benchmem -benchtime=10s
goos: darwin
goarch: amd64
pkg: github.com/valyala/fasthttp
BenchmarkNetHTTPClientGetEndToEndWaitConn1Inmemory-4      	     951	  12644110 ns/op	    9435 B/op	      76 allocs/op
BenchmarkNetHTTPClientGetEndToEndWaitConn10Inmemory-4     	    9501	   1262682 ns/op	   10328 B/op	      79 allocs/op
BenchmarkNetHTTPClientGetEndToEndWaitConn100Inmemory-4    	   91471	    133340 ns/op	   12020 B/op	      84 allocs/op
BenchmarkNetHTTPClientGetEndToEndWaitConn1000Inmemory-4   	  277612	     39097 ns/op	   13705 B/op	      89 allocs/op
PASS
ok  	github.com/valyala/fasthttp	74.091s
```

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/client.go b/client.go
index 5f63f09..7665312 100644
--- a/client.go
+++ b/client.go
@@ -97,6 +97,30 @@ func DoDeadline(req *Request, resp *Response, deadline time.Time) error {
 	return defaultClient.DoDeadline(req, resp, deadline)
 }
 
+// DoRedirects performs the given http request and fills the given http response,
+// following up to maxRedirectsCount redirects. When the redirect count exceeds
+// maxRedirectsCount, ErrTooManyRedirects is returned.
+//
+// Request must contain at least non-zero RequestURI with full url (including
+// scheme and host) or non-zero Host header + RequestURI.
+//
+// Client determines the server to be requested in the following order:
+//
+//   - from RequestURI if it contains full url with scheme and host;
+//   - from Host header otherwise.
+//
+// Response is ignored if resp is nil.
+//
+// ErrNoFreeConns is returned if all DefaultMaxConnsPerHost connections
+// to the requested host are busy.
+//
+// It is recommended obtaining req and resp via AcquireRequest
+// and AcquireResponse in performance-critical code.
+func DoRedirects(req *Request, resp *Response, maxRedirectsCount int) error {
+	_, _, err := doRequestFollowRedirects(req, resp, req.URI().String(), maxRedirectsCount, &defaultClient)
+	return err
+}
+
 // Get returns the status code and body of url.
 //
 // The contents of dst will be replaced by the body and returned, if the dst
@@ -260,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
@@ -372,6 +401,30 @@ func (c *Client) DoDeadline(req *Request, resp *Response, deadline time.Time) er
 	return clientDoDeadline(req, resp, deadline, c)
 }
 
+// DoRedirects performs the given http request and fills the given http response,
+// following up to maxRedirectsCount redirects. When the redirect count exceeds
+// maxRedirectsCount, ErrTooManyRedirects is returned.
+//
+// Request must contain at least non-zero RequestURI with full url (including
+// scheme and host) or non-zero Host header + RequestURI.
+//
+// Client determines the server to be requested in the following order:
+//
+//   - from RequestURI if it contains full url with scheme and host;
+//   - from Host header otherwise.
+//
+// Response is ignored if resp is nil.
+//
+// ErrNoFreeConns is returned if all DefaultMaxConnsPerHost connections
+// to the requested host are busy.
+//
+// It is recommended obtaining req and resp via AcquireRequest
+// and AcquireResponse in performance-critical code.
+func (c *Client) DoRedirects(req *Request, resp *Response, maxRedirectsCount int) error {
+	_, _, err := doRequestFollowRedirects(req, resp, req.URI().String(), maxRedirectsCount, c)
+	return err
+}
+
 // Do performs the given http request and fills the given http response.
 //
 // Request must contain at least non-zero RequestURI with full url (including
@@ -439,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 {
@@ -639,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
@@ -731,7 +791,7 @@ type clientDoer interface {
 func clientGetURL(dst []byte, url string, c clientDoer) (statusCode int, body []byte, err error) {
 	req := AcquireRequest()
 
-	statusCode, body, err = doRequestFollowRedirects(req, dst, url, c)
+	statusCode, body, err = doRequestFollowRedirectsBuffer(req, dst, url, c)
 
 	ReleaseRequest(req)
 	return statusCode, body, err
@@ -771,7 +831,7 @@ func clientGetURLDeadline(dst []byte, url string, deadline time.Time, c clientDo
 	// concurrent requests, since timed out requests on client side
 	// usually continue execution on the host.
 	go func() {
-		statusCodeCopy, bodyCopy, errCopy := doRequestFollowRedirects(req, dst, url, c)
+		statusCodeCopy, bodyCopy, errCopy := doRequestFollowRedirectsBuffer(req, dst, url, c)
 		ch <- clientURLResponse{
 			statusCode: statusCodeCopy,
 			body:       bodyCopy,
@@ -808,29 +868,45 @@ func clientPostURL(dst []byte, url string, postArgs *Args, c clientDoer) (status
 		}
 	}
 
-	statusCode, body, err = doRequestFollowRedirects(req, dst, url, c)
+	statusCode, body, err = doRequestFollowRedirectsBuffer(req, dst, url, c)
 
 	ReleaseRequest(req)
 	return statusCode, body, err
 }
 
 var (
-	errMissingLocation  = errors.New("missing Location header for http redirect")
-	errTooManyRedirects = errors.New("too many redirects detected when doing the request")
+	// ErrMissingLocation is returned by clients when the Location header is missing on
+	// an HTTP response with a redirect status code.
+	ErrMissingLocation = errors.New("missing Location header for http redirect")
+	// ErrTooManyRedirects is returned by clients when the number of redirects followed
+	// exceed the max count.
+	ErrTooManyRedirects = errors.New("too many redirects detected when doing the request")
 )
 
-const maxRedirectsCount = 16
+const defaultMaxRedirectsCount = 16
 
-func doRequestFollowRedirects(req *Request, dst []byte, url string, c clientDoer) (statusCode int, body []byte, err error) {
+func doRequestFollowRedirectsBuffer(req *Request, dst []byte, url string, c clientDoer) (statusCode int, body []byte, err error) {
 	resp := AcquireResponse()
 	bodyBuf := resp.bodyBuffer()
 	resp.keepBodyBuffer = true
 	oldBody := bodyBuf.B
 	bodyBuf.B = dst
+
+	statusCode, body, err = doRequestFollowRedirects(req, resp, url, defaultMaxRedirectsCount, c)
+
+	body = bodyBuf.B
+	bodyBuf.B = oldBody
+	resp.keepBodyBuffer = false
+	ReleaseResponse(resp)
+
+	return statusCode, body, err
+}
+
+func doRequestFollowRedirects(req *Request, resp *Response, url string, maxRedirectsCount int, c clientDoer) (statusCode int, body []byte, err error) {
 	scheme := req.uri.Scheme()
 	req.schemaUpdate = false
-
 	redirectsCount := 0
+
 	for {
 		// In case redirect to different scheme
 		if redirectsCount > 0 && !bytes.Equal(scheme, req.uri.Scheme()) {
@@ -859,22 +935,17 @@ func doRequestFollowRedirects(req *Request, dst []byte, url string, c clientDoer
 
 		redirectsCount++
 		if redirectsCount > maxRedirectsCount {
-			err = errTooManyRedirects
+			err = ErrTooManyRedirects
 			break
 		}
 		location := resp.Header.peek(strLocation)
 		if len(location) == 0 {
-			err = errMissingLocation
+			err = ErrMissingLocation
 			break
 		}
 		url = getRedirectURL(url, location)
 	}
 
-	body = bodyBuf.B
-	bodyBuf.B = oldBody
-	resp.keepBodyBuffer = false
-	ReleaseResponse(resp)
-
 	return statusCode, body, err
 }
 
@@ -994,6 +1065,30 @@ func (c *HostClient) DoDeadline(req *Request, resp *Response, deadline time.Time
 	return clientDoDeadline(req, resp, deadline, c)
 }
 
+// DoRedirects performs the given http request and fills the given http response,
+// following up to maxRedirectsCount redirects. When the redirect count exceeds
+// maxRedirectsCount, ErrTooManyRedirects is returned.
+//
+// Request must contain at least non-zero RequestURI with full url (including
+// scheme and host) or non-zero Host header + RequestURI.
+//
+// Client determines the server to be requested in the following order:
+//
+//   - from RequestURI if it contains full url with scheme and host;
+//   - from Host header otherwise.
+//
+// Response is ignored if resp is nil.
+//
+// ErrNoFreeConns is returned if all DefaultMaxConnsPerHost connections
+// to the requested host are busy.
+//
+// It is recommended obtaining req and resp via AcquireRequest
+// and AcquireResponse in performance-critical code.
+func (c *HostClient) DoRedirects(req *Request, resp *Response, maxRedirectsCount int) error {
+	_, _, err := doRequestFollowRedirects(req, resp, req.URI().String(), maxRedirectsCount, c)
+	return err
+}
+
 func clientDoTimeout(req *Request, resp *Response, timeout time.Duration, c clientDoer) error {
 	deadline := time.Now().Add(timeout)
 	return clientDoDeadline(req, resp, deadline, c)
@@ -1036,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()
 		{
@@ -1101,6 +1197,7 @@ func (c *HostClient) Do(req *Request, resp *Response) error {
 		maxAttempts = DefaultMaxIdemponentCallAttempts
 	}
 	attempts := 0
+	hasBodyStream := req.IsBodyStream()
 
 	atomic.AddInt32(&c.pendingRequests, 1)
 	for {
@@ -1109,6 +1206,9 @@ func (c *HostClient) Do(req *Request, resp *Response) error {
 			break
 		}
 
+		if hasBodyStream {
+			break
+		}
 		if !isIdempotent(req) {
 			// Retry non-idempotent requests if the server closes
 			// the connection before sending the response.
@@ -1195,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
 	}
@@ -1322,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
 
@@ -1355,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 {
@@ -1372,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
@@ -1435,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 {
@@ -1461,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 {
@@ -1689,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 841f1ee..59bc413 100644
--- a/client_test.go
+++ b/client_test.go
@@ -2,6 +2,7 @@ package fasthttp
 
 import (
 	"bufio"
+	"bytes"
 	"crypto/tls"
 	"fmt"
 	"io"
@@ -1227,6 +1228,42 @@ func TestClientFollowRedirects(t *testing.T) {
 			t.Fatalf("unexpected response %q. Expecting %q", body, "/aaab/sss")
 		}
 	}
+
+	for i := 0; i < 10; i++ {
+		req := AcquireRequest()
+		resp := AcquireResponse()
+
+		req.SetRequestURI("http://xxx/foo")
+
+		err := c.DoRedirects(req, resp, 16)
+		if err != nil {
+			t.Fatalf("unexpected error: %s", err)
+		}
+
+		if statusCode := resp.StatusCode(); statusCode != StatusOK {
+			t.Fatalf("unexpected status code: %d", statusCode)
+		}
+
+		if body := string(resp.Body()); body != "/bar" {
+			t.Fatalf("unexpected response %q. Expecting %q", body, "/bar")
+		}
+
+		ReleaseRequest(req)
+		ReleaseResponse(resp)
+	}
+
+	req := AcquireRequest()
+	resp := AcquireResponse()
+
+	req.SetRequestURI("http://xxx/foo")
+
+	err := c.DoRedirects(req, resp, 0)
+	if have, want := err, ErrTooManyRedirects; have != want {
+		t.Fatalf("want error: %v, have %v", want, have)
+	}
+
+	ReleaseRequest(req)
+	ReleaseResponse(resp)
 }
 
 func TestClientGetTimeoutSuccess(t *testing.T) {
@@ -1457,6 +1494,43 @@ func TestClientNonIdempotentRetry(t *testing.T) {
 	}
 }
 
+func TestClientNonIdempotentRetry_BodyStream(t *testing.T) {
+	t.Parallel()
+
+	dialsCount := 0
+	c := &Client{
+		Dial: func(addr string) (net.Conn, error) {
+			dialsCount++
+			switch dialsCount {
+			case 1, 2:
+				return &readErrorConn{}, nil
+			case 3:
+				return &singleEchoConn{
+					b: []byte("HTTP/1.1 345 OK\r\nContent-Type: foobar\r\n\r\n"),
+				}, nil
+			default:
+				t.Fatalf("unexpected number of dials: %d", dialsCount)
+			}
+			panic("unreachable")
+		},
+	}
+
+	dialsCount = 0
+
+	req := Request{}
+	res := Response{}
+
+	req.SetRequestURI("http://foobar/a/b")
+	req.Header.SetMethod("POST")
+	body := bytes.NewBufferString("test")
+	req.SetBodyStream(body, body.Len())
+
+	err := c.Do(&req, &res)
+	if err == nil {
+		t.Fatal("expected error from being unable to retry a bodyStream")
+	}
+}
+
 func TestClientIdempotentRequest(t *testing.T) {
 	t.Parallel()
 
@@ -1588,6 +1662,71 @@ func (r *singleReadConn) RemoteAddr() net.Addr {
 	return nil
 }
 
+type singleEchoConn struct {
+	net.Conn
+	b []byte
+	n int
+}
+
+func (r *singleEchoConn) Read(p []byte) (int, error) {
+	if len(r.b) == r.n {
+		return 0, io.EOF
+	}
+	n := copy(p, r.b[r.n:])
+	r.n += n
+	return n, nil
+}
+
+func (r *singleEchoConn) Write(p []byte) (int, error) {
+	r.b = append(r.b, p...)
+	return len(p), nil
+}
+
+func (r *singleEchoConn) Close() error {
+	return nil
+}
+
+func (r *singleEchoConn) LocalAddr() net.Addr {
+	return nil
+}
+
+func (r *singleEchoConn) RemoteAddr() net.Addr {
+	return nil
+}
+
+func TestSingleEchoConn(t *testing.T) {
+	t.Parallel()
+
+	c := &Client{
+		Dial: func(addr string) (net.Conn, error) {
+			return &singleEchoConn{
+				b: []byte("HTTP/1.1 345 OK\r\nContent-Type: foobar\r\n\r\n"),
+			}, nil
+		},
+	}
+
+	req := Request{}
+	res := Response{}
+
+	req.SetRequestURI("http://foobar/a/b")
+	req.Header.SetMethod("POST")
+	req.Header.Set("Content-Type", "text/plain")
+	body := bytes.NewBufferString("test")
+	req.SetBodyStream(body, body.Len())
+
+	err := c.Do(&req, &res)
+	if err != nil {
+		t.Fatalf("unexpected error: %s", err)
+	}
+	if res.StatusCode() != 345 {
+		t.Fatalf("unexpected status code: %d. Expecting 345", res.StatusCode())
+	}
+	expected := "POST /a/b HTTP/1.1\r\nUser-Agent: fasthttp\r\nHost: foobar\r\nContent-Type: text/plain\r\nContent-Length: 4\r\n\r\ntest"
+	if string(res.Body()) != expected {
+		t.Fatalf("unexpected body: %q. Expecting %q", res.Body(), expected)
+	}
+}
+
 func TestClientHTTPSInvalidServerName(t *testing.T) {
 	t.Parallel()
 
@@ -1981,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")
+	}
+}