client_conn_pool.go 6.4 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256
  1. // Copyright 2015 The Go Authors. All rights reserved.
  2. // Use of this source code is governed by a BSD-style
  3. // license that can be found in the LICENSE file.
  4. // Transport code's client connection pooling.
  5. package http2
  6. import (
  7. "crypto/tls"
  8. "net/http"
  9. "sync"
  10. )
  11. // ClientConnPool manages a pool of HTTP/2 client connections.
  12. type ClientConnPool interface {
  13. GetClientConn(req *http.Request, addr string) (*ClientConn, error)
  14. MarkDead(*ClientConn)
  15. }
  16. // clientConnPoolIdleCloser is the interface implemented by ClientConnPool
  17. // implementations which can close their idle connections.
  18. type clientConnPoolIdleCloser interface {
  19. ClientConnPool
  20. closeIdleConnections()
  21. }
  22. var (
  23. _ clientConnPoolIdleCloser = (*clientConnPool)(nil)
  24. _ clientConnPoolIdleCloser = noDialClientConnPool{}
  25. )
  26. // TODO: use singleflight for dialing and addConnCalls?
  27. type clientConnPool struct {
  28. t *Transport
  29. mu sync.Mutex // TODO: maybe switch to RWMutex
  30. // TODO: add support for sharing conns based on cert names
  31. // (e.g. share conn for googleapis.com and appspot.com)
  32. conns map[string][]*ClientConn // key is host:port
  33. dialing map[string]*dialCall // currently in-flight dials
  34. keys map[*ClientConn][]string
  35. addConnCalls map[string]*addConnCall // in-flight addConnIfNeede calls
  36. }
  37. func (p *clientConnPool) GetClientConn(req *http.Request, addr string) (*ClientConn, error) {
  38. return p.getClientConn(req, addr, dialOnMiss)
  39. }
  40. const (
  41. dialOnMiss = true
  42. noDialOnMiss = false
  43. )
  44. func (p *clientConnPool) getClientConn(req *http.Request, addr string, dialOnMiss bool) (*ClientConn, error) {
  45. if isConnectionCloseRequest(req) && dialOnMiss {
  46. // It gets its own connection.
  47. const singleUse = true
  48. cc, err := p.t.dialClientConn(addr, singleUse)
  49. if err != nil {
  50. return nil, err
  51. }
  52. return cc, nil
  53. }
  54. p.mu.Lock()
  55. for _, cc := range p.conns[addr] {
  56. if cc.CanTakeNewRequest() {
  57. p.mu.Unlock()
  58. return cc, nil
  59. }
  60. }
  61. if !dialOnMiss {
  62. p.mu.Unlock()
  63. return nil, ErrNoCachedConn
  64. }
  65. call := p.getStartDialLocked(addr)
  66. p.mu.Unlock()
  67. <-call.done
  68. return call.res, call.err
  69. }
  70. // dialCall is an in-flight Transport dial call to a host.
  71. type dialCall struct {
  72. p *clientConnPool
  73. done chan struct{} // closed when done
  74. res *ClientConn // valid after done is closed
  75. err error // valid after done is closed
  76. }
  77. // requires p.mu is held.
  78. func (p *clientConnPool) getStartDialLocked(addr string) *dialCall {
  79. if call, ok := p.dialing[addr]; ok {
  80. // A dial is already in-flight. Don't start another.
  81. return call
  82. }
  83. call := &dialCall{p: p, done: make(chan struct{})}
  84. if p.dialing == nil {
  85. p.dialing = make(map[string]*dialCall)
  86. }
  87. p.dialing[addr] = call
  88. go call.dial(addr)
  89. return call
  90. }
  91. // run in its own goroutine.
  92. func (c *dialCall) dial(addr string) {
  93. const singleUse = false // shared conn
  94. c.res, c.err = c.p.t.dialClientConn(addr, singleUse)
  95. close(c.done)
  96. c.p.mu.Lock()
  97. delete(c.p.dialing, addr)
  98. if c.err == nil {
  99. c.p.addConnLocked(addr, c.res)
  100. }
  101. c.p.mu.Unlock()
  102. }
  103. // addConnIfNeeded makes a NewClientConn out of c if a connection for key doesn't
  104. // already exist. It coalesces concurrent calls with the same key.
  105. // This is used by the http1 Transport code when it creates a new connection. Because
  106. // the http1 Transport doesn't de-dup TCP dials to outbound hosts (because it doesn't know
  107. // the protocol), it can get into a situation where it has multiple TLS connections.
  108. // This code decides which ones live or die.
  109. // The return value used is whether c was used.
  110. // c is never closed.
  111. func (p *clientConnPool) addConnIfNeeded(key string, t *Transport, c *tls.Conn) (used bool, err error) {
  112. p.mu.Lock()
  113. for _, cc := range p.conns[key] {
  114. if cc.CanTakeNewRequest() {
  115. p.mu.Unlock()
  116. return false, nil
  117. }
  118. }
  119. call, dup := p.addConnCalls[key]
  120. if !dup {
  121. if p.addConnCalls == nil {
  122. p.addConnCalls = make(map[string]*addConnCall)
  123. }
  124. call = &addConnCall{
  125. p: p,
  126. done: make(chan struct{}),
  127. }
  128. p.addConnCalls[key] = call
  129. go call.run(t, key, c)
  130. }
  131. p.mu.Unlock()
  132. <-call.done
  133. if call.err != nil {
  134. return false, call.err
  135. }
  136. return !dup, nil
  137. }
  138. type addConnCall struct {
  139. p *clientConnPool
  140. done chan struct{} // closed when done
  141. err error
  142. }
  143. func (c *addConnCall) run(t *Transport, key string, tc *tls.Conn) {
  144. cc, err := t.NewClientConn(tc)
  145. p := c.p
  146. p.mu.Lock()
  147. if err != nil {
  148. c.err = err
  149. } else {
  150. p.addConnLocked(key, cc)
  151. }
  152. delete(p.addConnCalls, key)
  153. p.mu.Unlock()
  154. close(c.done)
  155. }
  156. func (p *clientConnPool) addConn(key string, cc *ClientConn) {
  157. p.mu.Lock()
  158. p.addConnLocked(key, cc)
  159. p.mu.Unlock()
  160. }
  161. // p.mu must be held
  162. func (p *clientConnPool) addConnLocked(key string, cc *ClientConn) {
  163. for _, v := range p.conns[key] {
  164. if v == cc {
  165. return
  166. }
  167. }
  168. if p.conns == nil {
  169. p.conns = make(map[string][]*ClientConn)
  170. }
  171. if p.keys == nil {
  172. p.keys = make(map[*ClientConn][]string)
  173. }
  174. p.conns[key] = append(p.conns[key], cc)
  175. p.keys[cc] = append(p.keys[cc], key)
  176. }
  177. func (p *clientConnPool) MarkDead(cc *ClientConn) {
  178. p.mu.Lock()
  179. defer p.mu.Unlock()
  180. for _, key := range p.keys[cc] {
  181. vv, ok := p.conns[key]
  182. if !ok {
  183. continue
  184. }
  185. newList := filterOutClientConn(vv, cc)
  186. if len(newList) > 0 {
  187. p.conns[key] = newList
  188. } else {
  189. delete(p.conns, key)
  190. }
  191. }
  192. delete(p.keys, cc)
  193. }
  194. func (p *clientConnPool) closeIdleConnections() {
  195. p.mu.Lock()
  196. defer p.mu.Unlock()
  197. // TODO: don't close a cc if it was just added to the pool
  198. // milliseconds ago and has never been used. There's currently
  199. // a small race window with the HTTP/1 Transport's integration
  200. // where it can add an idle conn just before using it, and
  201. // somebody else can concurrently call CloseIdleConns and
  202. // break some caller's RoundTrip.
  203. for _, vv := range p.conns {
  204. for _, cc := range vv {
  205. cc.closeIfIdle()
  206. }
  207. }
  208. }
  209. func filterOutClientConn(in []*ClientConn, exclude *ClientConn) []*ClientConn {
  210. out := in[:0]
  211. for _, v := range in {
  212. if v != exclude {
  213. out = append(out, v)
  214. }
  215. }
  216. // If we filtered it out, zero out the last item to prevent
  217. // the GC from seeing it.
  218. if len(in) != len(out) {
  219. in[len(in)-1] = nil
  220. }
  221. return out
  222. }
  223. // noDialClientConnPool is an implementation of http2.ClientConnPool
  224. // which never dials. We let the HTTP/1.1 client dial and use its TLS
  225. // connection instead.
  226. type noDialClientConnPool struct{ *clientConnPool }
  227. func (p noDialClientConnPool) GetClientConn(req *http.Request, addr string) (*ClientConn, error) {
  228. return p.getClientConn(req, addr, noDialOnMiss)
  229. }