idleMu sync.Mutex

idleConn map[string][]*persistConn

idleConnCh map[string]chan *persistConn

reqMu sync.Mutex

reqConn map[*Request]*persistConn

altMu sync.RWMutex

altProto map[string]RoundTripper // nil or map of URI scheme => RoundTripper

// Proxy specifies a function to return a proxy for a given

// Request. If the function returns a non-nil error, the

// request is aborted with the provided error.

// If Proxy is nil or returns a nil *URL, no proxy is used.

Proxy func(*Request) (*url.URL, error)

// Dial specifies the dial function for creating TCP

// connections.

// If Dial is nil, net.Dial is used.

Dial func(network, addr string) (net.Conn, error)

// TLSClientConfig specifies the TLS configuration to use with

// tls.Client. If nil, the default configuration is used.

TLSClientConfig *bfe_tls.Config

// DisableKeepAlives, if true, prevents re-use of TCP connections

// between different HTTP requests.

DisableKeepAlives bool

// DisableCompression, if true, prevents the Transport from

// requesting compression with an "Accept-Encoding: gzip"

// request header when the Request contains no existing

// Accept-Encoding value. If the Transport requests gzip on

// its own and gets a gzipped response, it's transparently

// decoded in the Response.Body. However, if the user

// explicitly requested gzip it is not automatically

// uncompressed.

DisableCompression bool

// MaxIdleConnsPerHost, if non-zero, controls the maximum idle

// (keep-alive) to keep per-host. If zero,

// DefaultMaxIdleConnsPerHost is used.

MaxIdleConnsPerHost int

// ResponseHeaderTimeout, if non-zero, specifies the amount of

// time to wait for a server's response headers after fully

// writing the request (including its body, if any). This

// time does not include the time to read the response body.

ResponseHeaderTimeout time.Duration

// ReqWriteBufferSize specifies writer buffer size for request.

// If zero, default buffer size is used.

ReqWriteBufferSize int

// FlushInterval specifies the flush interval to flush to

// the backend while copying the request body.

// If zero, no periodic flushing is done.

ReqFlushInterval time.Duration

// TODO: tunable on global max cached connections

// TODO: tunable on timeout on cached connections

proxy := getenvEitherCase("HTTP_PROXY")

if proxy == "" {

return nil, nil

}

if !useProxy(canonicalAddr(req.URL)) {

return nil, nil

}

proxyURL, err := url.Parse(proxy)

if err != nil || !strings.HasPrefix(proxyURL.Scheme, "http") {

// proxy was bogus. Try prepending "http://" to it and

// see if that parses correctly. If not, we fall

// through and complain about the original one.

if proxyURL, err := url.Parse("http://" + proxy); err == nil {

return proxyURL, nil

}

}

if err != nil {

return nil, fmt.Errorf("invalid proxy address %q: %v", proxy, err)

}

return proxyURL, nil

return func(*Request) (*url.URL, error) {

return fixedURL, nil

}

if tr.extra == nil {

tr.extra = make(Header)

}

return tr.extra

state.HttpBackendReqAll.Inc(1)

if req.URL == nil {

return nil, errors.New("http: nil Request.URL")

}

if req.Header == nil {

return nil, errors.New("http: nil Request.Header")

}

if req.URL.Scheme != "http" && req.URL.Scheme != "https" {

t.altMu.RLock()

var rt RoundTripper

if t.altProto != nil {

rt = t.altProto[req.URL.Scheme]

}

t.altMu.RUnlock()

if rt == nil {

return nil, &badStringError{"unsupported protocol scheme", req.URL.Scheme}

}

return rt.RoundTrip(req)

}

if req.URL.Host == "" {

return nil, errors.New("http: no Host in request URL")

}

treq := &transportRequest{Request: req}

cm, err := t.connectMethodForRequest(treq)

if err != nil {

return nil, err

}

// record start connect backend time

connectStartTime := time.Now()

// Get the cached or newly-created connection to either the

// host (for http or https), the http proxy, or the http proxy

// pre-CONNECTed to https server. In any case, we'll be ready

// to send it requests.

pconn, err := t.getConn(cm)

if err != nil {

return nil, ConnectError{Err: err, Addr: cm.addr()}

}

// set connect backend time stat

req.State.ConnectBackendStart = connectStartTime

req.State.ConnectBackendEnd = time.Now()

resp, err = pconn.roundTrip(treq)

if err == nil {

state.HttpBackendReqSucc.Inc(1)

}

return

if scheme == "http" || scheme == "https" {

panic("protocol " + scheme + " already registered")

}

t.altMu.Lock()

defer t.altMu.Unlock()

if t.altProto == nil {

t.altProto = make(map[string]RoundTripper)

}

if _, exists := t.altProto[scheme]; exists {

panic("protocol " + scheme + " already registered")

}

t.altProto[scheme] = rt

t.idleMu.Lock()

m := t.idleConn

t.idleConn = nil

t.idleConnCh = nil

t.idleMu.Unlock()

if m == nil {

return

}

for _, conns := range m {

for _, pconn := range conns {

pconn.close()

}

}

t.reqMu.Lock()

pc := t.reqConn[req]

t.reqMu.Unlock()

if pc != nil {

pc.conn.Close()

}

if v := os.Getenv(strings.ToUpper(k)); v != "" {

return v

}

return os.Getenv(strings.ToLower(k))

cm := &connectMethod{

targetScheme: treq.URL.Scheme,

targetAddr: canonicalAddr(treq.URL),

}

if t.Proxy != nil {

var err error

cm.proxyURL, err = t.Proxy(treq.Request)

if err != nil {

return nil, err

}

}

return cm, nil

if cm.proxyURL == nil {

return ""

}

if u := cm.proxyURL.User; u != nil {

username := u.Username()

password, _ := u.Password()

return "Basic " + basicAuth(username, password)

}

return ""

if t.DisableKeepAlives || t.MaxIdleConnsPerHost < 0 {

pconn.close()

return false

}

if pconn.isBroken() {

return false

}

key := pconn.cacheKey

max := t.MaxIdleConnsPerHost

if max == 0 {

max = DefaultMaxIdleConnsPerHost

}

t.idleMu.Lock()

waitingDialer := t.idleConnCh[key]

select {

case waitingDialer <- pconn:

// We're done with this pconn and somebody else is

// currently waiting for a conn of this type (they're

// actively dialing, but this conn is ready

// first). Chrome calls this socket late binding. See

// https://insouciant.org/tech/connection-management-in-chromium/

t.idleMu.Unlock()

return true

default:

if waitingDialer != nil {

// They had populated this, but their dial won

// first, so we can clean up this map entry.

delete(t.idleConnCh, key)

}

}

if t.idleConn == nil {

t.idleConn = make(map[string][]*persistConn)

}

if len(t.idleConn[key]) >= max {

t.idleMu.Unlock()

pconn.close()

return false

}

for _, exist := range t.idleConn[key] {

if exist == pconn {

log.Logger.Error("dup idle pconn %p in freelist", pconn)

}

}

t.idleConn[key] = append(t.idleConn[key], pconn)

t.idleMu.Unlock()

return true

if t.DisableKeepAlives {

return nil

}

key := cm.key()

t.idleMu.Lock()

defer t.idleMu.Unlock()

if t.idleConnCh == nil {

t.idleConnCh = make(map[string]chan *persistConn)

}

ch, ok := t.idleConnCh[key]

if !ok {

ch = make(chan *persistConn)

t.idleConnCh[key] = ch

}

return ch

key := cm.key()

t.idleMu.Lock()

defer t.idleMu.Unlock()

if t.idleConn == nil {

return nil

}

for {

pconns, ok := t.idleConn[key]

if !ok {

return nil

}

if len(pconns) == 1 {

pconn = pconns[0]

delete(t.idleConn, key)

} else {

// 2 or more cached connections; pop last

// TODO: queue?

pconn = pconns[len(pconns)-1]

t.idleConn[key] = pconns[0 : len(pconns)-1]

}

if !pconn.isBroken() {

return

}

}

t.reqMu.Lock()

defer t.reqMu.Unlock()

if t.reqConn == nil {

t.reqConn = make(map[*Request]*persistConn)

}

if pc != nil {

t.reqConn[r] = pc

} else {

delete(t.reqConn, r)

}

if t.Dial != nil {

return t.Dial(network, addr)

}

return net.Dial(network, addr)

if pc := t.getIdleConn(cm); pc != nil {

return pc, nil

}

type dialRes struct {

pc *persistConn

err error

}

dialc := make(chan dialRes)

go func() {

pc, err := t.dialConn(cm)

state.HttpBackendConnAll.Inc(1)

if err == nil {

state.HttpBackendConnSucc.Inc(1)

}

dialc <- dialRes{pc, err}

}()

idleConnCh := t.getIdleConnCh(cm)

select {

case v := <-dialc:

// Our dial finished.

return v.pc, v.err

case pc := <-idleConnCh:

// Another request finished first and its net.Conn

// became available before our dial. Or somebody

// else's dial that they didn't use.

// But our dial is still going, so give it away

// when it finishes:

go func() {

if v := <-dialc; v.err == nil {

t.putIdleConn(v.pc)

}

}()

return pc, nil

}

conn, err := t.dial("tcp", cm.addr())

if err != nil {

if cm.proxyURL != nil {

err = fmt.Errorf("http: error connecting to proxy %s: %v", cm.proxyURL, err)

}

return nil, err

}

pa := cm.proxyAuth()

pconn := &persistConn{

t: t,

cacheKey: cm.key(),

conn: conn,

reqch: make(chan requestAndChan, 1),

writech: make(chan writeRequest, 1),

closech: make(chan struct{}),

}

switch {

case cm.proxyURL == nil:

// Do nothing.

case cm.targetScheme == "http":

pconn.isProxy = true

if pa != "" {

pconn.mutateHeaderFunc = func(h Header) {

h.Set("Proxy-Authorization", pa)

}

}

case cm.targetScheme == "https":

connectReq := &Request{

Method: "CONNECT",

URL: &url.URL{Opaque: cm.targetAddr},

Host: cm.targetAddr,

Header: make(Header),

}

if pa != "" {

connectReq.Header.Set("Proxy-Authorization", pa)

}

connectReq.Write(conn)

// Read response.

// Okay to use and discard buffered reader here, because

// TLS server will not speak until spoken to.

br := bfe_bufio.NewReader(conn)

resp, err := ReadResponse(br, connectReq)

if err != nil {

conn.Close()

return nil, err

}

if resp.StatusCode != 200 {

f := strings.SplitN(resp.Status, " ", 2)

conn.Close()

return nil, errors.New(f[1])

}

}

if cm.targetScheme == "https" {

// Initiate TLS and check remote host name against certificate.

cfg := t.TLSClientConfig

if cfg == nil || cfg.ServerName == "" {

host := cm.tlsHost()

if cfg == nil {

cfg = &bfe_tls.Config{ServerName: host}

} else {

clone := cfg.Clone() // shallow clone

clone.ServerName = host

cfg = clone

}

}

conn = bfe_tls.Client(conn, cfg)

if err = conn.(*bfe_tls.Conn).Handshake(); err != nil {

return nil, err

}

if !cfg.InsecureSkipVerify {

if err = conn.(*bfe_tls.Conn).VerifyHostname(cfg.ServerName); err != nil {

return nil, err

}

}

pconn.conn = conn

}

pconn.br = bfe_bufio.NewReader(pconn.conn)

if t.ReqWriteBufferSize > 0 {

pconn.bw = bfe_bufio.NewWriterSize(pconn.conn, t.ReqWriteBufferSize)

} else {

pconn.bw = bfe_bufio.NewWriter(pconn.conn)

}

if t.ReqFlushInterval > 0 {

pconn.bw = NewMaxLatencyWriter(pconn.bw, t.ReqFlushInterval, nil)

}

go pconn.readLoop()

go pconn.writeLoop()

return pconn, nil

if len(addr) == 0 {

return true

}

host, _, err := net.SplitHostPort(addr)

if err != nil {

return false

}

if host == "localhost" {

return false

}

if ip := net.ParseIP(host); ip != nil {

if ip.IsLoopback() {

return false

}

}

no_proxy := getenvEitherCase("NO_PROXY")

if no_proxy == "*" {

return false

}

addr = strings.ToLower(strings.TrimSpace(addr))

if hasPort(addr) {

addr = addr[:strings.LastIndex(addr, ":")]

}

for _, p := range strings.Split(no_proxy, ",") {

p = strings.ToLower(strings.TrimSpace(p))

if len(p) == 0 {

continue

}

if hasPort(p) {

p = p[:strings.LastIndex(p, ":")]

}

if addr == p {

return false

}

if p[0] == '.' && (strings.HasSuffix(addr, p) || addr == p[1:]) {

// no_proxy ".foo.com" matches "bar.foo.com" or "foo.com"

return false

}

if p[0] != '.' && strings.HasSuffix(addr, p) && addr[len(addr)-len(p)-1] == '.' {

// no_proxy "foo.com" matches "bar.foo.com"

return false

}

}

return true

proxyURL *url.URL // nil for no proxy, else full proxy URL

targetScheme string // "http" or "https"

targetAddr string // Not used if proxy + http targetScheme (4th example in table)

proxyStr := ""

targetAddr := ck.targetAddr

if ck.proxyURL != nil {

proxyStr = ck.proxyURL.String()

if ck.targetScheme == "http" {

targetAddr = ""

}

}

return strings.Join([]string{proxyStr, ck.targetScheme, targetAddr}, "|")

if cm.proxyURL != nil {

return canonicalAddr(cm.proxyURL)

}

return cm.targetAddr

h := cm.targetAddr

if hasPort(h) {

h = h[:strings.LastIndex(h, ":")]

}

return h

t *Transport

cacheKey string // its connectMethod.String()

conn net.Conn

closed bool // whether conn has been closed

br *bfe_bufio.Reader // from conn

bw WriteFlusher // to conn

reqch chan requestAndChan // written by roundTrip; read by readLoop

writech chan writeRequest // written by roundTrip; read by writeLoop

closech chan struct{} // broadcast close when readLoop (TCP connection) closes

isProxy bool

lk sync.Mutex // guards following 3 fields

numExpectedResponses int

broken bool // an error has happened on this connection; marked broken so it's not reused.

// mutateHeaderFunc is an optional func to modify extra

// headers on each outbound request before it's written. (the

// original Request given to RoundTrip is not modified)

mutateHeaderFunc func(Header)

pc.lk.Lock()

b := pc.broken

pc.lk.Unlock()

return b

if err == io.EOF {

return true

}

if remoteSideClosedFunc != nil {

return remoteSideClosedFunc(err)

}

return false

defer close(pc.closech)

defer func() {

if err := recover(); err != nil {

log.Logger.Warn("panic:persistConn.readLoop():%v\n%s", err, gotrack.CurrentStackTrace(0))

pc.close()

state.HttpPanicBackendRead.Inc(1)

}

}()

alive := true

for alive {

pb, err := pc.br.Peek(1)

pc.lk.Lock()

if pc.numExpectedResponses == 0 {

pc.closeLocked()

pc.lk.Unlock()

if len(pb) > 0 {

log.Logger.Info("Unsolicited response received on idle HTTP channel starting with %q; err=%v",

string(pb), err)

}

return

}

pc.lk.Unlock()

rc := <-pc.reqch

var resp *Response

if err == nil {

resp, err = ReadResponse(pc.br, rc.req)

if err == nil && resp.StatusCode == 100 {

// Skip any 100-continue for now.

// TODO(bradfitz): if rc.req had "Expect: 100-continue",

// actually block the request body write and signal the

// writeLoop now to begin sending it. (Issue 2184) For now we

// eat it, since we're never expecting one.

resp, err = ReadResponse(pc.br, rc.req)

}

}

hasBody := resp != nil && rc.req.Method != "HEAD" && resp.ContentLength != 0

if err != nil {

pc.close()

} else {

if rc.addedGzip && hasBody && resp.Header.Get("Content-Encoding") == "gzip" {

resp.Header.Del("Content-Encoding")

resp.Header.Del("Content-Length")

resp.ContentLength = -1

gzReader, zerr := gzip.NewReader(resp.Body)

if zerr != nil {

pc.close()

err = zerr

} else {

resp.Body = &readerAndCloser{gzReader, resp.Body}

}

}

resp.Body = &bodyEOFSignal{body: resp.Body}

}

if err != nil || resp.Close || rc.req.Close || resp.StatusCode <= 199 {

// Don't do keep-alive on error if either party requested a close

// or we get an unexpected informational (1xx) response.

// StatusCode 100 is already handled above.

alive = false

}

var waitForBodyRead chan bool

if hasBody {

waitForBodyRead = make(chan bool, 2)

resp.Body.(*bodyEOFSignal).earlyCloseFn = func() error {

// Sending false here sets alive to

// false and closes the connection

// below.

waitForBodyRead <- false

return nil

}

resp.Body.(*bodyEOFSignal).fn = func(err error) {

alive1 := alive

if err != nil {

alive1 = false

}

if alive1 && !pc.t.putIdleConn(pc) {

alive1 = false

}

if !alive1 || pc.isBroken() {

pc.close()

}

waitForBodyRead <- alive1

}

}

if alive && !hasBody {

if !pc.t.putIdleConn(pc) {

alive = false

}

}

if err != nil {

err = ReadRespHeaderError{Err: err}

}

rc.ch <- responseAndError{resp, err}

// Wait for the just-returned response body to be fully consumed

// before we race and peek on the underlying bufio reader.

if waitForBodyRead != nil {

alive = <-waitForBodyRead

}

pc.t.setReqConn(rc.req, nil)

if !alive {

pc.close()

}

}

defer func() {

if err := recover(); err != nil {

log.Logger.Warn("panic:persistConn.writeLoop():%v\n%s", err, gotrack.CurrentStackTrace(0))

// close connection

pc.close()

state.HttpPanicBackendWrite.Inc(1)

}

}()

// Start flush loop for request

if mlw, ok := pc.bw.(*MaxLatencyWriter); ok {

go mlw.FlushLoop()

defer mlw.Stop()

}

for {

select {

case wr := <-pc.writech:

if pc.isBroken() {

wr.ch <- WriteRequestError{Err: errors.New("http: can't write HTTP request on broken connection")}

continue

}

err := wr.req.Request.write(pc.bw, pc.isProxy, wr.req.extra)

if err == nil {

err = pc.bw.Flush()

}

if err != nil {

err = WriteRequestError{Err: err}

pc.markBroken()

}

wr.ch <- err

case <-pc.closech:

return

}

}

req *Request

ch chan responseAndError

// did the Transport (as opposed to the client code) add an

// Accept-Encoding gzip header? only if it we set it do

// we transparently decode the gzip.

addedGzip bool

pc.t.setReqConn(req.Request, pc)

pc.lk.Lock()

pc.numExpectedResponses++

headerFn := pc.mutateHeaderFunc

pc.lk.Unlock()

if headerFn != nil {

headerFn(req.extraHeaders())

}

// Ask for a compressed version if the caller didn't set their

// own value for Accept-Encoding. We only attempted to

// uncompress the gzip stream if we were the layer that

// requested it.

requestedGzip := false

if !pc.t.DisableCompression && req.Header.Get("Accept-Encoding") == "" && req.Method != "HEAD" {

// Request gzip only, not deflate. Deflate is ambiguous and

// not as universally supported anyway.

// See: http://www.gzip.org/zlib/zlib_faq.html#faq38

//

// Note that we don't request this for HEAD requests,

// due to a bug in nginx:

// http://trac.nginx.org/nginx/ticket/358

// http://golang.org/issue/5522

requestedGzip = true

req.extraHeaders().Set("Accept-Encoding", "gzip")

}

// Write the request concurrently with waiting for a response,

// in case the server decides to reply before reading our full

// request body.

writeErrCh := make(chan error, 1)

pc.writech <- writeRequest{req, writeErrCh}

resc := make(chan responseAndError, 1)

pc.reqch <- requestAndChan{req.Request, resc, requestedGzip}

var re responseAndError

var pconnDeadCh = pc.closech

var failTicker <-chan time.Time

var respHeaderTicker <-chan time.Time

WaitResponse:

for {

select {

case err := <-writeErrCh:

if err != nil {

re = responseAndError{nil, err}

pc.close()

break WaitResponse

}

if d := pc.t.ResponseHeaderTimeout; d > 0 {

// Note: The underlying timer created by time.After() is not

// recovered by the garbage collector until the timer fires.

// For efficiency, we use time.NewTimer() instead of time.After()

// and stop the timer if no longer needed.

t := time.NewTimer(d)

defer t.Stop()

respHeaderTicker = t.C

}

case <-pconnDeadCh:

// The persist connection is dead. This shouldn't

// usually happen (only with Connection: close responses

// with no response bodies), but if it does happen it

// means either a) the remote server hung up on us

// prematurely, or b) the readLoop sent us a response &

// closed its closech at roughly the same time, and we

// selected this case first, in which case a response

// might still be coming soon.

//

// We can't avoid the select race in b) by using a unbuffered

// resc channel instead, because then goroutines can

// leak if we exit due to other errors.

pconnDeadCh = nil // avoid spinning

failTicker = time.After(100 * time.Millisecond) // arbitrary time to wait for resc

case <-failTicker:

re = responseAndError{err: TransportBrokenError{}}

break WaitResponse

case <-respHeaderTicker:

pc.close()