« back to all changes in this revision
Viewing changes to http13client/transport.go
- browse files at revision 81.1.1
- compare with another revision
- download diff
- download tarball
- view history from revision 81.1.1
- Committer: Samuele Pedroni (Canonical Services Ltd.)
- Date: 2014-03-19 20:20:19 UTC
- mto: This revision was merged to the branch mainline in revision 82.
- Revision ID: samuele.pedroni@canonical.com-20140319202019-p0w8krshj1098f82
grab go 1.3 dev net/http and massage it so that the test run on 1.2
- files added:
- http13client
- http13client/_patches
- http13client/httptest
- http13client/httputil
- http13client/testdata
added
removed
http13client/transport.go
1
// Copyright 2011 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
5
// HTTP client implementation. See RFC 2616.
6
//
7
// This is the low-level Transport implementation of RoundTripper.
8
// The high-level interface is in client.go.
9
10
package http
11
12
import (
13
"bufio"
14
"compress/gzip"
15
"crypto/tls"
16
"errors"
17
"fmt"
18
"io"
19
"log"
20
"net"
21
"net/url"
22
"os"
23
"strings"
24
"sync"
25
"time"
26
)
27
28
// DefaultTransport is the default implementation of Transport and is
29
// used by DefaultClient. It establishes network connections as needed
30
// and caches them for reuse by subsequent calls. It uses HTTP proxies
31
// as directed by the $HTTP_PROXY and $NO_PROXY (or $http_proxy and
32
// $no_proxy) environment variables.
33
var DefaultTransport RoundTripper = &Transport{
34
Proxy: ProxyFromEnvironment,
35
Dial: (&net.Dialer{
36
Timeout: 30 * time.Second,
37
// KeepAlive: 30 * time.Second,
38
}).Dial,
39
TLSHandshakeTimeout: 10 * time.Second,
40
}
41
42
// DefaultMaxIdleConnsPerHost is the default value of Transport's
43
// MaxIdleConnsPerHost.
44
const DefaultMaxIdleConnsPerHost = 2
45
46
// Transport is an implementation of RoundTripper that supports http,
47
// https, and http proxies (for either http or https with CONNECT).
48
// Transport can also cache connections for future re-use.
49
type Transport struct {
50
idleMu sync.Mutex
51
idleConn map[connectMethodKey][]*persistConn
52
idleConnCh map[connectMethodKey]chan *persistConn
53
reqMu sync.Mutex
54
reqCanceler map[*Request]func()
55
altMu sync.RWMutex
56
altProto map[string]RoundTripper // nil or map of URI scheme => RoundTripper
57
58
// Proxy specifies a function to return a proxy for a given
59
// Request. If the function returns a non-nil error, the
60
// request is aborted with the provided error.
61
// If Proxy is nil or returns a nil *URL, no proxy is used.
62
Proxy func(*Request) (*url.URL, error)
63
64
// Dial specifies the dial function for creating TCP
65
// connections.
66
// If Dial is nil, net.Dial is used.
67
Dial func(network, addr string) (net.Conn, error)
68
69
// TLSClientConfig specifies the TLS configuration to use with
70
// tls.Client. If nil, the default configuration is used.
71
TLSClientConfig *tls.Config
72
73
// TLSHandshakeTimeout specifies the maximum amount of time waiting to
74
// wait for a TLS handshake. Zero means no timeout.
75
TLSHandshakeTimeout time.Duration
76
77
// DisableKeepAlives, if true, prevents re-use of TCP connections
78
// between different HTTP requests.
79
DisableKeepAlives bool
80
81
// DisableCompression, if true, prevents the Transport from
82
// requesting compression with an "Accept-Encoding: gzip"
83
// request header when the Request contains no existing
84
// Accept-Encoding value. If the Transport requests gzip on
85
// its own and gets a gzipped response, it's transparently
86
// decoded in the Response.Body. However, if the user
87
// explicitly requested gzip it is not automatically
88
// uncompressed.
89
DisableCompression bool
90
91
// MaxIdleConnsPerHost, if non-zero, controls the maximum idle
92
// (keep-alive) to keep per-host. If zero,
93
// DefaultMaxIdleConnsPerHost is used.
94
MaxIdleConnsPerHost int
95
96
// ResponseHeaderTimeout, if non-zero, specifies the amount of
97
// time to wait for a server's response headers after fully
98
// writing the request (including its body, if any). This
99
// time does not include the time to read the response body.
100
ResponseHeaderTimeout time.Duration
101
102
// TODO: tunable on global max cached connections
103
// TODO: tunable on timeout on cached connections
104
}
105
106
// ProxyFromEnvironment returns the URL of the proxy to use for a
107
// given request, as indicated by the environment variables
108
// $HTTP_PROXY and $NO_PROXY (or $http_proxy and $no_proxy).
109
// An error is returned if the proxy environment is invalid.
110
// A nil URL and nil error are returned if no proxy is defined in the
111
// environment, or a proxy should not be used for the given request.
112
func ProxyFromEnvironment(req *Request) (*url.URL, error) {
113
proxy := httpProxyEnv.Get()
114
if proxy == "" {
115
return nil, nil
116
}
117
if !useProxy(canonicalAddr(req.URL)) {
118
return nil, nil
119
}
120
proxyURL, err := url.Parse(proxy)
121
if err != nil || !strings.HasPrefix(proxyURL.Scheme, "http") {
122
// proxy was bogus. Try prepending "http://" to it and
123
// see if that parses correctly. If not, we fall
124
// through and complain about the original one.
125
if proxyURL, err := url.Parse("http://" + proxy); err == nil {
126
return proxyURL, nil
127
}
128
}
129
if err != nil {
130
return nil, fmt.Errorf("invalid proxy address %q: %v", proxy, err)
131
}
132
return proxyURL, nil
133
}
134
135
// ProxyURL returns a proxy function (for use in a Transport)
136
// that always returns the same URL.
137
func ProxyURL(fixedURL *url.URL) func(*Request) (*url.URL, error) {
138
return func(*Request) (*url.URL, error) {
139
return fixedURL, nil
140
}
141
}
142
143
// transportRequest is a wrapper around a *Request that adds
144
// optional extra headers to write.
145
type transportRequest struct {
146
*Request // original request, not to be mutated
147
extra Header // extra headers to write, or nil
148
}
149
150
func (tr *transportRequest) extraHeaders() Header {
151
if tr.extra == nil {
152
tr.extra = make(Header)
153
}
154
return tr.extra
155
}
156
157
// RoundTrip implements the RoundTripper interface.
158
//
159
// For higher-level HTTP client support (such as handling of cookies
160
// and redirects), see Get, Post, and the Client type.
161
func (t *Transport) RoundTrip(req *Request) (resp *Response, err error) {
162
if req.URL == nil {
163
return nil, errors.New("http: nil Request.URL")
164
}
165
if req.Header == nil {
166
return nil, errors.New("http: nil Request.Header")
167
}
168
if req.URL.Scheme != "http" && req.URL.Scheme != "https" {
169
t.altMu.RLock()
170
var rt RoundTripper
171
if t.altProto != nil {
172
rt = t.altProto[req.URL.Scheme]
173
}
174
t.altMu.RUnlock()
175
if rt == nil {
176
return nil, &badStringError{"unsupported protocol scheme", req.URL.Scheme}
177
}
178
return rt.RoundTrip(req)
179
}
180
if req.URL.Host == "" {
181
return nil, errors.New("http: no Host in request URL")
182
}
183
treq := &transportRequest{Request: req}
184
cm, err := t.connectMethodForRequest(treq)
185
if err != nil {
186
return nil, err
187
}
188
189
// Get the cached or newly-created connection to either the
190
// host (for http or https), the http proxy, or the http proxy
191
// pre-CONNECTed to https server. In any case, we'll be ready
192
// to send it requests.
193
pconn, err := t.getConn(req, cm)
194
if err != nil {
195
t.setReqCanceler(req, nil)
196
return nil, err
197
}
198
199
return pconn.roundTrip(treq)
200
}
201
202
// RegisterProtocol registers a new protocol with scheme.
203
// The Transport will pass requests using the given scheme to rt.
204
// It is rt's responsibility to simulate HTTP request semantics.
205
//
206
// RegisterProtocol can be used by other packages to provide
207
// implementations of protocol schemes like "ftp" or "file".
208
func (t *Transport) RegisterProtocol(scheme string, rt RoundTripper) {
209
if scheme == "http" || scheme == "https" {
210
panic("protocol " + scheme + " already registered")
211
}
212
t.altMu.Lock()
213
defer t.altMu.Unlock()
214
if t.altProto == nil {
215
t.altProto = make(map[string]RoundTripper)
216
}
217
if _, exists := t.altProto[scheme]; exists {
218
panic("protocol " + scheme + " already registered")
219
}
220
t.altProto[scheme] = rt
221
}
222
223
// CloseIdleConnections closes any connections which were previously
224
// connected from previous requests but are now sitting idle in
225
// a "keep-alive" state. It does not interrupt any connections currently
226
// in use.
227
func (t *Transport) CloseIdleConnections() {
228
t.idleMu.Lock()
229
m := t.idleConn
230
t.idleConn = nil
231
t.idleConnCh = nil
232
t.idleMu.Unlock()
233
if m == nil {
234
return
235
}
236
for _, conns := range m {
237
for _, pconn := range conns {
238
pconn.close()
239
}
240
}
241
}
242
243
// CancelRequest cancels an in-flight request by closing its
244
// connection.
245
func (t *Transport) CancelRequest(req *Request) {
246
t.reqMu.Lock()
247
cancel := t.reqCanceler[req]
248
t.reqMu.Unlock()
249
if cancel != nil {
250
cancel()
251
}
252
}
253
254
//
255
// Private implementation past this point.
256
//
257
258
var (
259
httpProxyEnv = &envOnce{
260
names: []string{"HTTP_PROXY", "http_proxy"},
261
}
262
noProxyEnv = &envOnce{
263
names: []string{"NO_PROXY", "no_proxy"},
264
}
265
)
266
267
// envOnce looks up an environment variable (optionally by multiple
268
// names) once. It mitigates expensive lookups on some platforms
269
// (e.g. Windows).
270
type envOnce struct {
271
names []string
272
once sync.Once
273
val string
274
}
275
276
func (e *envOnce) Get() string {
277
e.once.Do(e.init)
278
return e.val
279
}
280
281
func (e *envOnce) init() {
282
for _, n := range e.names {
283
e.val = os.Getenv(n)
284
if e.val != "" {
285
return
286
}
287
}
288
}
289
290
// reset is used by tests
291
func (e *envOnce) reset() {
292
e.once = sync.Once{}
293
e.val = ""
294
}
295
296
func (t *Transport) connectMethodForRequest(treq *transportRequest) (cm connectMethod, err error) {
297
cm.targetScheme = treq.URL.Scheme
298
cm.targetAddr = canonicalAddr(treq.URL)
299
if t.Proxy != nil {
300
cm.proxyURL, err = t.Proxy(treq.Request)
301
}
302
return cm, nil
303
}
304
305
// proxyAuth returns the Proxy-Authorization header to set
306
// on requests, if applicable.
307
func (cm *connectMethod) proxyAuth() string {
308
if cm.proxyURL == nil {
309
return ""
310
}
311
if u := cm.proxyURL.User; u != nil {
312
username := u.Username()
313
password, _ := u.Password()
314
return "Basic " + basicAuth(username, password)
315
}
316
return ""
317
}
318
319
// putIdleConn adds pconn to the list of idle persistent connections awaiting
320
// a new request.
321
// If pconn is no longer needed or not in a good state, putIdleConn
322
// returns false.
323
func (t *Transport) putIdleConn(pconn *persistConn) bool {
324
if t.DisableKeepAlives || t.MaxIdleConnsPerHost < 0 {
325
pconn.close()
326
return false
327
}
328
if pconn.isBroken() {
329
return false
330
}
331
key := pconn.cacheKey
332
max := t.MaxIdleConnsPerHost
333
if max == 0 {
334
max = DefaultMaxIdleConnsPerHost
335
}
336
t.idleMu.Lock()
337
338
waitingDialer := t.idleConnCh[key]
339
select {
340
case waitingDialer <- pconn:
341
// We're done with this pconn and somebody else is
342
// currently waiting for a conn of this type (they're
343
// actively dialing, but this conn is ready
344
// first). Chrome calls this socket late binding. See
345
// https://insouciant.org/tech/connection-management-in-chromium/
346
t.idleMu.Unlock()
347
return true
348
default:
349
if waitingDialer != nil {
350
// They had populated this, but their dial won
351
// first, so we can clean up this map entry.
352
delete(t.idleConnCh, key)
353
}
354
}
355
if t.idleConn == nil {
356
t.idleConn = make(map[connectMethodKey][]*persistConn)
357
}
358
if len(t.idleConn[key]) >= max {
359
t.idleMu.Unlock()
360
pconn.close()
361
return false
362
}
363
for _, exist := range t.idleConn[key] {
364
if exist == pconn {
365
log.Fatalf("dup idle pconn %p in freelist", pconn)
366
}
367
}
368
t.idleConn[key] = append(t.idleConn[key], pconn)
369
t.idleMu.Unlock()
370
return true
371
}
372
373
// getIdleConnCh returns a channel to receive and return idle
374
// persistent connection for the given connectMethod.
375
// It may return nil, if persistent connections are not being used.
376
func (t *Transport) getIdleConnCh(cm connectMethod) chan *persistConn {
377
if t.DisableKeepAlives {
378
return nil
379
}
380
key := cm.key()
381
t.idleMu.Lock()
382
defer t.idleMu.Unlock()
383
if t.idleConnCh == nil {
384
t.idleConnCh = make(map[connectMethodKey]chan *persistConn)
385
}
386
ch, ok := t.idleConnCh[key]
387
if !ok {
388
ch = make(chan *persistConn)
389
t.idleConnCh[key] = ch
390
}
391
return ch
392
}
393
394
func (t *Transport) getIdleConn(cm connectMethod) (pconn *persistConn) {
395
key := cm.key()
396
t.idleMu.Lock()
397
defer t.idleMu.Unlock()
398
if t.idleConn == nil {
399
return nil
400
}
401
for {
402
pconns, ok := t.idleConn[key]
403
if !ok {
404
return nil
405
}
406
if len(pconns) == 1 {
407
pconn = pconns[0]
408
delete(t.idleConn, key)
409
} else {
410
// 2 or more cached connections; pop last
411
// TODO: queue?
412
pconn = pconns[len(pconns)-1]
413
t.idleConn[key] = pconns[:len(pconns)-1]
414
}
415
if !pconn.isBroken() {
416
return
417
}
418
}
419
}
420
421
func (t *Transport) setReqCanceler(r *Request, fn func()) {
422
t.reqMu.Lock()
423
defer t.reqMu.Unlock()
424
if t.reqCanceler == nil {
425
t.reqCanceler = make(map[*Request]func())
426
}
427
if fn != nil {
428
t.reqCanceler[r] = fn
429
} else {
430
delete(t.reqCanceler, r)
431
}
432
}
433
434
func (t *Transport) dial(network, addr string) (c net.Conn, err error) {
435
if t.Dial != nil {
436
return t.Dial(network, addr)
437
}
438
return net.Dial(network, addr)
439
}
440
441
// getConn dials and creates a new persistConn to the target as
442
// specified in the connectMethod. This includes doing a proxy CONNECT
443
// and/or setting up TLS. If this doesn't return an error, the persistConn
444
// is ready to write requests to.
445
func (t *Transport) getConn(req *Request, cm connectMethod) (*persistConn, error) {
446
if pc := t.getIdleConn(cm); pc != nil {
447
return pc, nil
448
}
449
450
type dialRes struct {
451
pc *persistConn
452
err error
453
}
454
dialc := make(chan dialRes)
455
456
handlePendingDial := func() {
457
if v := <-dialc; v.err == nil {
458
t.putIdleConn(v.pc)
459
}
460
}
461
462
cancelc := make(chan struct{})
463
t.setReqCanceler(req, func() { close(cancelc) })
464
465
go func() {
466
pc, err := t.dialConn(cm)
467
dialc <- dialRes{pc, err}
468
}()
469
470
idleConnCh := t.getIdleConnCh(cm)
471
select {
472
case v := <-dialc:
473
// Our dial finished.
474
return v.pc, v.err
475
case pc := <-idleConnCh:
476
// Another request finished first and its net.Conn
477
// became available before our dial. Or somebody
478
// else's dial that they didn't use.
479
// But our dial is still going, so give it away
480
// when it finishes:
481
go handlePendingDial()
482
return pc, nil
483
case <-cancelc:
484
go handlePendingDial()
485
return nil, errors.New("net/http: request canceled while waiting for connection")
486
}
487
}
488
489
func (t *Transport) dialConn(cm connectMethod) (*persistConn, error) {
490
conn, err := t.dial("tcp", cm.addr())
491
if err != nil {
492
if cm.proxyURL != nil {
493
err = fmt.Errorf("http: error connecting to proxy %s: %v", cm.proxyURL, err)
494
}
495
return nil, err
496
}
497
498
pa := cm.proxyAuth()
499
500
pconn := &persistConn{
501
t: t,
502
cacheKey: cm.key(),
503
conn: conn,
504
reqch: make(chan requestAndChan, 50),
505
writech: make(chan writeRequest, 50),
506
closech: make(chan struct{}),
507
}
508
509
switch {
510
case cm.proxyURL == nil:
511
// Do nothing.
512
case cm.targetScheme == "http":
513
pconn.isProxy = true
514
if pa != "" {
515
pconn.mutateHeaderFunc = func(h Header) {
516
h.Set("Proxy-Authorization", pa)
517
}
518
}
519
case cm.targetScheme == "https":
520
connectReq := &Request{
521
Method: "CONNECT",
522
URL: &url.URL{Opaque: cm.targetAddr},
523
Host: cm.targetAddr,
524
Header: make(Header),
525
}
526
if pa != "" {
527
connectReq.Header.Set("Proxy-Authorization", pa)
528
}
529
connectReq.Write(conn)
530
531
// Read response.
532
// Okay to use and discard buffered reader here, because
533
// TLS server will not speak until spoken to.
534
br := bufio.NewReader(conn)
535
resp, err := ReadResponse(br, connectReq)
536
if err != nil {
537
conn.Close()
538
return nil, err
539
}
540
if resp.StatusCode != 200 {
541
f := strings.SplitN(resp.Status, " ", 2)
542
conn.Close()
543
return nil, errors.New(f[1])
544
}
545
}
546
547
if cm.targetScheme == "https" {
548
// Initiate TLS and check remote host name against certificate.
549
cfg := t.TLSClientConfig
550
if cfg == nil || cfg.ServerName == "" {
551
host := cm.tlsHost()
552
if cfg == nil {
553
cfg = &tls.Config{ServerName: host}
554
} else {
555
clone := *cfg // shallow clone
556
clone.ServerName = host
557
cfg = &clone
558
}
559
}
560
plainConn := conn
561
tlsConn := tls.Client(plainConn, cfg)
562
errc := make(chan error, 2)
563
var timer *time.Timer // for canceling TLS handshake
564
if d := t.TLSHandshakeTimeout; d != 0 {
565
timer = time.AfterFunc(d, func() {
566
errc <- tlsHandshakeTimeoutError{}
567
})
568
}
569
go func() {
570
err := tlsConn.Handshake()
571
if timer != nil {
572
timer.Stop()
573
}
574
errc <- err
575
}()
576
if err := <-errc; err != nil {
577
plainConn.Close()
578
return nil, err
579
}
580
if !cfg.InsecureSkipVerify {
581
if err := tlsConn.VerifyHostname(cfg.ServerName); err != nil {
582
plainConn.Close()
583
return nil, err
584
}
585
}
586
cs := tlsConn.ConnectionState()
587
pconn.tlsState = &cs
588
pconn.conn = tlsConn
589
}
590
591
pconn.br = bufio.NewReader(pconn.conn)
592
pconn.bw = bufio.NewWriter(pconn.conn)
593
go pconn.readLoop()
594
go pconn.writeLoop()
595
return pconn, nil
596
}
597
598
// useProxy returns true if requests to addr should use a proxy,
599
// according to the NO_PROXY or no_proxy environment variable.
600
// addr is always a canonicalAddr with a host and port.
601
func useProxy(addr string) bool {
602
if len(addr) == 0 {
603
return true
604
}
605
host, _, err := net.SplitHostPort(addr)
606
if err != nil {
607
return false
608
}
609
if host == "localhost" {
610
return false
611
}
612
if ip := net.ParseIP(host); ip != nil {
613
if ip.IsLoopback() {
614
return false
615
}
616
}
617
618
no_proxy := noProxyEnv.Get()
619
if no_proxy == "*" {
620
return false
621
}
622
623
addr = strings.ToLower(strings.TrimSpace(addr))
624
if hasPort(addr) {
625
addr = addr[:strings.LastIndex(addr, ":")]
626
}
627
628
for _, p := range strings.Split(no_proxy, ",") {
629
p = strings.ToLower(strings.TrimSpace(p))
630
if len(p) == 0 {
631
continue
632
}
633
if hasPort(p) {
634
p = p[:strings.LastIndex(p, ":")]
635
}
636
if addr == p {
637
return false
638
}
639
if p[0] == '.' && (strings.HasSuffix(addr, p) || addr == p[1:]) {
640
// no_proxy ".foo.com" matches "bar.foo.com" or "foo.com"
641
return false
642
}
643
if p[0] != '.' && strings.HasSuffix(addr, p) && addr[len(addr)-len(p)-1] == '.' {
644
// no_proxy "foo.com" matches "bar.foo.com"
645
return false
646
}
647
}
648
return true
649
}
650
651
// connectMethod is the map key (in its String form) for keeping persistent
652
// TCP connections alive for subsequent HTTP requests.
653
//
654
// A connect method may be of the following types:
655
//
656
// Cache key form Description
657
// ----------------- -------------------------
658
// |http|foo.com http directly to server, no proxy
659
// |https|foo.com https directly to server, no proxy
660
// http://proxy.com|https|foo.com http to proxy, then CONNECT to foo.com
661
// http://proxy.com|http http to proxy, http to anywhere after that
662
//
663
// Note: no support to https to the proxy yet.
664
//
665
type connectMethod struct {
666
proxyURL *url.URL // nil for no proxy, else full proxy URL
667
targetScheme string // "http" or "https"
668
targetAddr string // Not used if proxy + http targetScheme (4th example in table)
669
}
670
671
func (cm *connectMethod) key() connectMethodKey {
672
proxyStr := ""
673
targetAddr := cm.targetAddr
674
if cm.proxyURL != nil {
675
proxyStr = cm.proxyURL.String()
676
if cm.targetScheme == "http" {
677
targetAddr = ""
678
}
679
}
680
return connectMethodKey{
681
proxy: proxyStr,
682
scheme: cm.targetScheme,
683
addr: targetAddr,
684
}
685
}
686
687
// addr returns the first hop "host:port" to which we need to TCP connect.
688
func (cm *connectMethod) addr() string {
689
if cm.proxyURL != nil {
690
return canonicalAddr(cm.proxyURL)
691
}
692
return cm.targetAddr
693
}
694
695
// tlsHost returns the host name to match against the peer's
696
// TLS certificate.
697
func (cm *connectMethod) tlsHost() string {
698
h := cm.targetAddr
699
if hasPort(h) {
700
h = h[:strings.LastIndex(h, ":")]
701
}
702
return h
703
}
704
705
// connectMethodKey is the map key version of connectMethod, with a
706
// stringified proxy URL (or the empty string) instead of a pointer to
707
// a URL.
708
type connectMethodKey struct {
709
proxy, scheme, addr string
710
}
711
712
func (k connectMethodKey) String() string {
713
// Only used by tests.
714
return fmt.Sprintf("%s|%s|%s", k.proxy, k.scheme, k.addr)
715
}
716
717
// persistConn wraps a connection, usually a persistent one
718
// (but may be used for non-keep-alive requests as well)
719
type persistConn struct {
720
t *Transport
721
cacheKey connectMethodKey
722
conn net.Conn
723
tlsState *tls.ConnectionState
724
closed bool // whether conn has been closed
725
br *bufio.Reader // from conn
726
bw *bufio.Writer // to conn
727
reqch chan requestAndChan // written by roundTrip; read by readLoop
728
writech chan writeRequest // written by roundTrip; read by writeLoop
729
closech chan struct{} // broadcast close when readLoop (TCP connection) closes
730
isProxy bool
731
732
lk sync.Mutex // guards following 3 fields
733
numExpectedResponses int
734
broken bool // an error has happened on this connection; marked broken so it's not reused.
735
// mutateHeaderFunc is an optional func to modify extra
736
// headers on each outbound request before it's written. (the
737
// original Request given to RoundTrip is not modified)
738
mutateHeaderFunc func(Header)
739
}
740
741
func (pc *persistConn) isBroken() bool {
742
pc.lk.Lock()
743
b := pc.broken
744
pc.lk.Unlock()
745
return b
746
}
747
748
func (pc *persistConn) cancelRequest() {
749
pc.conn.Close()
750
}
751
752
var remoteSideClosedFunc func(error) bool // or nil to use default
753
754
func remoteSideClosed(err error) bool {
755
if err == io.EOF {
756
return true
757
}
758
if remoteSideClosedFunc != nil {
759
return remoteSideClosedFunc(err)
760
}
761
return false
762
}
763
764
func (pc *persistConn) readLoop() {
765
defer close(pc.closech)
766
alive := true
767
768
for alive {
769
pb, err := pc.br.Peek(1)
770
771
pc.lk.Lock()
772
if pc.numExpectedResponses == 0 {
773
pc.closeLocked()
774
pc.lk.Unlock()
775
if len(pb) > 0 {
776
log.Printf("Unsolicited response received on idle HTTP channel starting with %q; err=%v",
777
string(pb), err)
778
}
779
return
780
}
781
pc.lk.Unlock()
782
783
rc := <-pc.reqch
784
785
var resp *Response
786
if err == nil {
787
resp, err = ReadResponse(pc.br, rc.req)
788
if err == nil && resp.StatusCode == 100 {
789
// Skip any 100-continue for now.
790
// TODO(bradfitz): if rc.req had "Expect: 100-continue",
791
// actually block the request body write and signal the
792
// writeLoop now to begin sending it. (Issue 2184) For now we
793
// eat it, since we're never expecting one.
794
resp, err = ReadResponse(pc.br, rc.req)
795
}
796
}
797
798
if resp != nil {
799
resp.TLS = pc.tlsState
800
}
801
802
hasBody := resp != nil && rc.req.Method != "HEAD" && resp.ContentLength != 0
803
804
if err != nil {
805
pc.close()
806
} else {
807
if rc.addedGzip && hasBody && resp.Header.Get("Content-Encoding") == "gzip" {
808
resp.Header.Del("Content-Encoding")
809
resp.Header.Del("Content-Length")
810
resp.ContentLength = -1
811
gzReader, zerr := gzip.NewReader(resp.Body)
812
if zerr != nil {
813
pc.close()
814
err = zerr
815
} else {
816
resp.Body = &readerAndCloser{gzReader, resp.Body}
817
}
818
}
819
resp.Body = &bodyEOFSignal{body: resp.Body}
820
}
821
822
if err != nil || resp.Close || rc.req.Close || resp.StatusCode <= 199 {
823
// Don't do keep-alive on error if either party requested a close
824
// or we get an unexpected informational (1xx) response.
825
// StatusCode 100 is already handled above.
826
alive = false
827
}
828
829
var waitForBodyRead chan bool
830
if hasBody {
831
waitForBodyRead = make(chan bool, 2)
832
resp.Body.(*bodyEOFSignal).earlyCloseFn = func() error {
833
// Sending false here sets alive to
834
// false and closes the connection
835
// below.
836
waitForBodyRead <- false
837
return nil
838
}
839
resp.Body.(*bodyEOFSignal).fn = func(err error) {
840
alive1 := alive
841
if err != nil {
842
alive1 = false
843
}
844
if alive1 && !pc.t.putIdleConn(pc) {
845
alive1 = false
846
}
847
if !alive1 || pc.isBroken() {
848
pc.close()
849
}
850
waitForBodyRead <- alive1
851
}
852
}
853
854
if alive && !hasBody {
855
if !pc.t.putIdleConn(pc) {
856
alive = false
857
}
858
}
859
860
rc.ch <- responseAndError{resp, err}
861
862
// Wait for the just-returned response body to be fully consumed
863
// before we race and peek on the underlying bufio reader.
864
if waitForBodyRead != nil {
865
alive = <-waitForBodyRead
866
}
867
868
pc.t.setReqCanceler(rc.req, nil)
869
870
if !alive {
871
pc.close()
872
}
873
}
874
}
875
876
func (pc *persistConn) writeLoop() {
877
for {
878
select {
879
case wr := <-pc.writech:
880
if pc.isBroken() {
881
wr.ch <- errors.New("http: can't write HTTP request on broken connection")
882
continue
883
}
884
err := wr.req.Request.write(pc.bw, pc.isProxy, wr.req.extra)
885
if err == nil {
886
err = pc.bw.Flush()
887
}
888
if err != nil {
889
pc.markBroken()
890
}
891
wr.ch <- err
892
case <-pc.closech:
893
return
894
}
895
}
896
}
897
898
type responseAndError struct {
899
res *Response
900
err error
901
}
902
903
type requestAndChan struct {
904
req *Request
905
ch chan responseAndError
906
907
// did the Transport (as opposed to the client code) add an
908
// Accept-Encoding gzip header? only if it we set it do
909
// we transparently decode the gzip.
910
addedGzip bool
911
}
912
913
// A writeRequest is sent by the readLoop's goroutine to the
914
// writeLoop's goroutine to write a request while the read loop
915
// concurrently waits on both the write response and the server's
916
// reply.
917
type writeRequest struct {
918
req *transportRequest
919
ch chan<- error
920
}
921
922
type httpError struct {
923
err string
924
timeout bool
925
}
926
927
func (e *httpError) Error() string { return e.err }
928
func (e *httpError) Timeout() bool { return e.timeout }
929
func (e *httpError) Temporary() bool { return true }
930
931
var errTimeout error = &httpError{err: "net/http: timeout awaiting response headers", timeout: true}
932
var errClosed error = &httpError{err: "net/http: transport closed before response was received"}
933
934
func (pc *persistConn) roundTrip(req *transportRequest) (resp *Response, err error) {
935
pc.t.setReqCanceler(req.Request, pc.cancelRequest)
936
pc.lk.Lock()
937
pc.numExpectedResponses++
938
headerFn := pc.mutateHeaderFunc
939
pc.lk.Unlock()
940
941
if headerFn != nil {
942
headerFn(req.extraHeaders())
943
}
944
945
// Ask for a compressed version if the caller didn't set their
946
// own value for Accept-Encoding. We only attempted to
947
// uncompress the gzip stream if we were the layer that
948
// requested it.
949
requestedGzip := false
950
if !pc.t.DisableCompression && req.Header.Get("Accept-Encoding") == "" && req.Method != "HEAD" {
951
// Request gzip only, not deflate. Deflate is ambiguous and
952
// not as universally supported anyway.
953
// See: http://www.gzip.org/zlib/zlib_faq.html#faq38
954
//
955
// Note that we don't request this for HEAD requests,
956
// due to a bug in nginx:
957
// http://trac.nginx.org/nginx/ticket/358
958
// http://golang.org/issue/5522
959
requestedGzip = true
960
req.extraHeaders().Set("Accept-Encoding", "gzip")
961
}
962
963
// Write the request concurrently with waiting for a response,
964
// in case the server decides to reply before reading our full
965
// request body.
966
writeErrCh := make(chan error, 1)
967
pc.writech <- writeRequest{req, writeErrCh}
968
969
resc := make(chan responseAndError, 1)
970
pc.reqch <- requestAndChan{req.Request, resc, requestedGzip}
971
972
var re responseAndError
973
var pconnDeadCh = pc.closech
974
var failTicker <-chan time.Time
975
var respHeaderTimer <-chan time.Time
976
WaitResponse:
977
for {
978
select {
979
case err := <-writeErrCh:
980
if err != nil {
981
re = responseAndError{nil, err}
982
pc.close()
983
break WaitResponse
984
}
985
if d := pc.t.ResponseHeaderTimeout; d > 0 {
986
respHeaderTimer = time.After(d)
987
}
988
case <-pconnDeadCh:
989
// The persist connection is dead. This shouldn't
990
// usually happen (only with Connection: close responses
991
// with no response bodies), but if it does happen it
992
// means either a) the remote server hung up on us
993
// prematurely, or b) the readLoop sent us a response &
994
// closed its closech at roughly the same time, and we
995
// selected this case first, in which case a response
996
// might still be coming soon.
997
//
998
// We can't avoid the select race in b) by using a unbuffered
999
// resc channel instead, because then goroutines can
1000
// leak if we exit due to other errors.
1001
pconnDeadCh = nil // avoid spinning
1002
failTicker = time.After(100 * time.Millisecond) // arbitrary time to wait for resc
1003
case <-failTicker:
1004
re = responseAndError{err: errClosed}
1005
break WaitResponse
1006
case <-respHeaderTimer:
1007
pc.close()
1008
re = responseAndError{err: errTimeout}
1009
break WaitResponse
1010
case re = <-resc:
1011
break WaitResponse
1012
}
1013
}
1014
1015
pc.lk.Lock()
1016
pc.numExpectedResponses--
1017
pc.lk.Unlock()
1018
1019
if re.err != nil {
1020
pc.t.setReqCanceler(req.Request, nil)
1021
}
1022
return re.res, re.err
1023
}
1024
1025
// markBroken marks a connection as broken (so it's not reused).
1026
// It differs from close in that it doesn't close the underlying
1027
// connection for use when it's still being read.
1028
func (pc *persistConn) markBroken() {
1029
pc.lk.Lock()
1030
defer pc.lk.Unlock()
1031
pc.broken = true
1032
}
1033
1034
func (pc *persistConn) close() {
1035
pc.lk.Lock()
1036
defer pc.lk.Unlock()
1037
pc.closeLocked()
1038
}
1039
1040
func (pc *persistConn) closeLocked() {
1041
pc.broken = true
1042
if !pc.closed {
1043
pc.conn.Close()
1044
pc.closed = true
1045
}
1046
pc.mutateHeaderFunc = nil
1047
}
1048
1049
var portMap = map[string]string{
1050
"http": "80",
1051
"https": "443",
1052
}
1053
1054
// canonicalAddr returns url.Host but always with a ":port" suffix
1055
func canonicalAddr(url *url.URL) string {
1056
addr := url.Host
1057
if !hasPort(addr) {
1058
return addr + ":" + portMap[url.Scheme]
1059
}
1060
return addr
1061
}
1062
1063
// bodyEOFSignal wraps a ReadCloser but runs fn (if non-nil) at most
1064
// once, right before its final (error-producing) Read or Close call
1065
// returns. If earlyCloseFn is non-nil and Close is called before
1066
// io.EOF is seen, earlyCloseFn is called instead of fn, and its
1067
// return value is the return value from Close.
1068
type bodyEOFSignal struct {
1069
body io.ReadCloser
1070
mu sync.Mutex // guards following 4 fields
1071
closed bool // whether Close has been called
1072
rerr error // sticky Read error
1073
fn func(error) // error will be nil on Read io.EOF
1074
earlyCloseFn func() error // optional alt Close func used if io.EOF not seen
1075
}
1076
1077
func (es *bodyEOFSignal) Read(p []byte) (n int, err error) {
1078
es.mu.Lock()
1079
closed, rerr := es.closed, es.rerr
1080
es.mu.Unlock()
1081
if closed {
1082
return 0, errors.New("http: read on closed response body")
1083
}
1084
if rerr != nil {
1085
return 0, rerr
1086
}
1087
1088
n, err = es.body.Read(p)
1089
if err != nil {
1090
es.mu.Lock()
1091
defer es.mu.Unlock()
1092
if es.rerr == nil {
1093
es.rerr = err
1094
}
1095
es.condfn(err)
1096
}
1097
return
1098
}
1099
1100
func (es *bodyEOFSignal) Close() error {
1101
es.mu.Lock()
1102
defer es.mu.Unlock()
1103
if es.closed {
1104
return nil
1105
}
1106
es.closed = true
1107
if es.earlyCloseFn != nil && es.rerr != io.EOF {
1108
return es.earlyCloseFn()
1109
}
1110
err := es.body.Close()
1111
es.condfn(err)
1112
return err
1113
}
1114
1115
// caller must hold es.mu.
1116
func (es *bodyEOFSignal) condfn(err error) {
1117
if es.fn == nil {
1118
return
1119
}
1120
if err == io.EOF {
1121
err = nil
1122
}
1123
es.fn(err)
1124
es.fn = nil
1125
}
1126
1127
type readerAndCloser struct {
1128
io.Reader
1129
io.Closer
1130
}
1131
1132
type tlsHandshakeTimeoutError struct{}
1133
1134
func (tlsHandshakeTimeoutError) Timeout() bool { return true }
1135
func (tlsHandshakeTimeoutError) Temporary() bool { return true }
1136
func (tlsHandshakeTimeoutError) Error() string { return "net/http: TLS handshake timeout" }
Loggerhead is a web-based interface for Breezy
Version: 2.0.1