http2: add a MinConcurrentConns field to Transport for improving throughput

ochanism · ochanism · commit b2cc93d427fb · 2020-02-22T19:31:18.000+09:00
diff --git a/http2/client_conn_pool.go b/http2/client_conn_pool.go
@@ -10,6 +10,8 @@ import (
 	"crypto/tls"
 	"net/http"
 	"sync"
+	"math/rand"
+	"time"
 )
 
 // ClientConnPool manages a pool of HTTP/2 client connections.
@@ -84,27 +86,52 @@ func (p *clientConnPool) getClientConn(req *http.Request, addr string, dialOnMis
 		}
 		return cc, nil
 	}
+
 	p.mu.Lock()
-	for _, cc := range p.conns[addr] {
-		if st := cc.idleState(); st.canTakeNewRequest {
-			if p.shouldTraceGetConn(st) {
-				traceGetConn(req, addr)
+	defer p.mu.Unlock()
+
+	ccList := p.conns[addr]
+	minCon := int(p.t.MinConcurrentConns)
+
+	if len(ccList) >= minCon {
+		if minCon > 0 {
+			rndSrc := rand.NewSource(time.Now().UnixNano())
+			rnd := rand.New(rndSrc)
+			startOffset := rnd.Intn(minCon)
+			// find an available connection randomly among MinConcurrentConns connections for spreading streams
+			for i := 0; i < minCon; i++ {
+				idx := (i + startOffset) % minCon
+				if p.takeNewRequest(req, addr, ccList[idx]) {
+					return ccList[idx], nil
+				}
+			}
+		}
+		// find an available connection sequentially among the rests
+		for _, cc := range ccList[minCon:] {
+			if p.takeNewRequest(req, addr, cc) {
+				return cc, nil
 			}
-			p.mu.Unlock()
-			return cc, nil
 		}
 	}
 	if !dialOnMiss {
-		p.mu.Unlock()
 		return nil, ErrNoCachedConn
 	}
 	traceGetConn(req, addr)
 	call := p.getStartDialLocked(addr)
-	p.mu.Unlock()
 	<-call.done
 	return call.res, call.err
 }
 
+func (p *clientConnPool) takeNewRequest(req *http.Request, addr string, cc *ClientConn) bool {
+	if st := cc.idleState(); st.canTakeNewRequest {
+		if p.shouldTraceGetConn(st) {
+			traceGetConn(req, addr)
+		}
+		return true
+	}
+	return false
+}
+
 // dialCall is an in-flight Transport dial call to a host.
 type dialCall struct {
 	p    *clientConnPool
diff --git a/http2/transport.go b/http2/transport.go
@@ -108,6 +108,13 @@ type Transport struct {
 	// waiting for their turn.
 	StrictMaxConcurrentStreams bool
 
+	// MinConcurrentConns is the minimum number of TCP connections.
+	// ClientConnPool tries to maintain  MinConcurrentConns
+	// TCP connections at least. If 0, ClientConnPool creates a
+	// TCP connection only when needed. The default value is 0.
+	// Increase this value if you need high throughput.
+	MinConcurrentConns uint32
+
 	// t1, if non-nil, is the standard library Transport using
 	// this transport. Its settings are used (but not its
 	// RoundTrip method, etc).
diff --git a/http2/transport_test.go b/http2/transport_test.go
@@ -4557,3 +4557,56 @@ func TestClientConnTooIdle(t *testing.T) {
 		}
 	}
 }
+
+func TestTransportMinConcurrentConns(t *testing.T) {
+	st := newServerTester(t,
+		func(w http.ResponseWriter, r *http.Request) {
+			io.WriteString(w, "test")
+		},
+		func(s *Server) {
+			s.MaxConcurrentStreams = 100
+			s.IdleTimeout = 30 * time.Second
+		},
+		optOnlyServer,
+	)
+	defer st.Close()
+
+	req, err := http.NewRequest("GET", st.ts.URL, nil)
+	if err != nil {
+		t.Fatalf("NewRequest: %v", err)
+	}
+	addr := authorityAddr(req.URL.Scheme, req.URL.Host)
+
+	MinConcurrentConns := 5
+
+	tr := &Transport{
+		TLSClientConfig: tlsConfigInsecure,
+		MinConcurrentConns: uint32(MinConcurrentConns),
+	}
+	defer tr.CloseIdleConnections()
+
+	// execute a dummy request
+	for i := 0; i < 10; i++ {
+		_, err := tr.RoundTrip(req)
+		if err != nil {
+			t.Fatalf("%v", err)
+		}
+	}
+
+	// check client connection pool
+	cp, ok := tr.connPool().(*clientConnPool)
+	if !ok {
+		t.Fatalf("Conn pool is %T; want *clientConnPool", tr.connPool())
+	}
+
+	cp.mu.Lock()
+	defer cp.mu.Unlock()
+
+	conns, ok := cp.conns[addr]
+	if !ok {
+		t.Fatalf("cannot find connections for %s", addr)
+	}
+	if len(conns) != MinConcurrentConns {
+		t.Fatalf("the number of connections should be %d, but got %d", MinConcurrentConns, len(conns))
+	}
+}
