Source file
src/net/http/transport.go
1
2
3
4
5
6
7
8
9
10 package http
11
12 import (
13 "bufio"
14 "container/list"
15 "context"
16 "crypto/tls"
17 "errors"
18 "fmt"
19 "internal/godebug"
20 "io"
21 "log"
22 "maps"
23 "net"
24 "net/http/httptrace"
25 "net/http/internal"
26 "net/http/internal/ascii"
27 "net/http/internal/httpcommon"
28 "net/textproto"
29 "net/url"
30 "reflect"
31 "strings"
32 "sync"
33 "sync/atomic"
34 "time"
35 _ "unsafe"
36
37 "golang.org/x/net/http/httpguts"
38 "golang.org/x/net/http/httpproxy"
39 )
40
41
42
43
44
45
46
47 var DefaultTransport RoundTripper = &Transport{
48 Proxy: ProxyFromEnvironment,
49 DialContext: defaultTransportDialContext(&net.Dialer{
50 Timeout: 30 * time.Second,
51 KeepAlive: 30 * time.Second,
52 }),
53 ForceAttemptHTTP2: true,
54 MaxIdleConns: 100,
55 IdleConnTimeout: 90 * time.Second,
56 TLSHandshakeTimeout: 10 * time.Second,
57 ExpectContinueTimeout: 1 * time.Second,
58 }
59
60
61
62 const DefaultMaxIdleConnsPerHost = 2
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98 type Transport struct {
99 idleMu sync.Mutex
100 closeIdle bool
101 idleConn map[connectMethodKey][]*persistConn
102 idleConnWait map[connectMethodKey]wantConnQueue
103 idleLRU connLRU
104
105 altMu sync.Mutex
106 altProto atomic.Value
107
108 connsPerHostMu sync.Mutex
109 connsPerHost map[connectMethodKey]int
110 connsPerHostWait map[connectMethodKey]wantConnQueue
111 dialsInProgress wantConnQueue
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127 Proxy func(*Request) (*url.URL, error)
128
129
130
131
132 OnProxyConnectResponse func(ctx context.Context, proxyURL *url.URL, connectReq *Request, connectRes *Response) error
133
134
135
136
137
138
139
140
141
142 DialContext func(ctx context.Context, network, addr string) (net.Conn, error)
143
144
145
146
147
148
149
150
151
152
153
154 Dial func(network, addr string) (net.Conn, error)
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169 DialTLSContext func(ctx context.Context, network, addr string) (net.Conn, error)
170
171
172
173
174
175
176
177 DialTLS func(network, addr string) (net.Conn, error)
178
179
180
181
182
183 TLSClientConfig *tls.Config
184
185
186
187 TLSHandshakeTimeout time.Duration
188
189
190
191
192
193
194 DisableKeepAlives bool
195
196
197
198
199
200
201
202
203
204 DisableCompression bool
205
206
207
208 MaxIdleConns int
209
210
211
212
213 MaxIdleConnsPerHost int
214
215
216
217
218
219
220 MaxConnsPerHost int
221
222
223
224
225
226 IdleConnTimeout time.Duration
227
228
229
230
231
232 ResponseHeaderTimeout time.Duration
233
234
235
236
237
238
239
240
241 ExpectContinueTimeout time.Duration
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256 TLSNextProto map[string]func(authority string, c *tls.Conn) RoundTripper
257
258
259
260
261 ProxyConnectHeader Header
262
263
264
265
266
267
268
269
270 GetProxyConnectHeader func(ctx context.Context, proxyURL *url.URL, target string) (Header, error)
271
272
273
274
275
276
277 MaxResponseHeaderBytes int64
278
279
280
281
282 WriteBufferSize int
283
284
285
286
287 ReadBufferSize int
288
289
290
291 nextProtoOnce sync.Once
292 closeIdleFunc closeIdleConnectionser
293 h2Transport *http2Transport
294 h2Config http2ExternalTransportConfig
295 h3Transport dialClientConner
296 tlsNextProtoWasNil bool
297
298
299
300
301
302
303 ForceAttemptHTTP2 bool
304
305
306 HTTP2 *HTTP2Config
307
308
309
310
311
312
313
314
315
316 Protocols *Protocols
317 }
318
319 func (t *Transport) writeBufferSize() int {
320 if t.WriteBufferSize > 0 {
321 return t.WriteBufferSize
322 }
323 return 4 << 10
324 }
325
326 func (t *Transport) readBufferSize() int {
327 if t.ReadBufferSize > 0 {
328 return t.ReadBufferSize
329 }
330 return 4 << 10
331 }
332
333 func (t *Transport) maxHeaderResponseSize() int64 {
334 if t.MaxResponseHeaderBytes > 0 {
335 return t.MaxResponseHeaderBytes
336 }
337 return 10 << 20
338 }
339
340
341 func (t *Transport) Clone() *Transport {
342 t.nextProtoOnce.Do(t.onceSetNextProtoDefaults)
343 t2 := &Transport{
344 Proxy: t.Proxy,
345 OnProxyConnectResponse: t.OnProxyConnectResponse,
346 DialContext: t.DialContext,
347 Dial: t.Dial,
348 DialTLS: t.DialTLS,
349 DialTLSContext: t.DialTLSContext,
350 TLSHandshakeTimeout: t.TLSHandshakeTimeout,
351 DisableKeepAlives: t.DisableKeepAlives,
352 DisableCompression: t.DisableCompression,
353 MaxIdleConns: t.MaxIdleConns,
354 MaxIdleConnsPerHost: t.MaxIdleConnsPerHost,
355 MaxConnsPerHost: t.MaxConnsPerHost,
356 IdleConnTimeout: t.IdleConnTimeout,
357 ResponseHeaderTimeout: t.ResponseHeaderTimeout,
358 ExpectContinueTimeout: t.ExpectContinueTimeout,
359 ProxyConnectHeader: t.ProxyConnectHeader.Clone(),
360 GetProxyConnectHeader: t.GetProxyConnectHeader,
361 MaxResponseHeaderBytes: t.MaxResponseHeaderBytes,
362 ForceAttemptHTTP2: t.ForceAttemptHTTP2,
363 WriteBufferSize: t.WriteBufferSize,
364 ReadBufferSize: t.ReadBufferSize,
365 }
366 if t.TLSClientConfig != nil {
367 t2.TLSClientConfig = t.TLSClientConfig.Clone()
368 }
369 if t.HTTP2 != nil {
370 t2.HTTP2 = &HTTP2Config{}
371 *t2.HTTP2 = *t.HTTP2
372 }
373 if t.Protocols != nil {
374 t2.Protocols = &Protocols{}
375 *t2.Protocols = *t.Protocols
376 }
377 if !t.tlsNextProtoWasNil {
378 npm := maps.Clone(t.TLSNextProto)
379 if npm == nil {
380 npm = make(map[string]func(authority string, c *tls.Conn) RoundTripper)
381 }
382 t2.TLSNextProto = npm
383 }
384 return t2
385 }
386
387 type dialClientConner interface {
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413 DialClientConn(ctx context.Context, address string, proxy *url.URL, tlsConfig *tls.Config, internalStateHook func()) (RoundTripper, error)
414 }
415
416 type closeIdleConnectionser interface {
417
418
419
420
421
422
423
424
425
426 CloseIdleConnections()
427 }
428
429 func (t *Transport) hasCustomTLSDialer() bool {
430 return t.DialTLS != nil || t.DialTLSContext != nil
431 }
432
433 var http2client = godebug.New("http2client")
434
435
436
437 func (t *Transport) onceSetNextProtoDefaults() {
438 t.tlsNextProtoWasNil = (t.TLSNextProto == nil)
439 if http2client.Value() == "0" {
440 http2client.IncNonDefault()
441 return
442 }
443
444
445
446
447
448
449 altProto, _ := t.altProto.Load().(map[string]RoundTripper)
450 if rv := reflect.ValueOf(altProto["https"]); rv.IsValid() && rv.Type().Kind() == reflect.Struct && rv.Type().NumField() == 1 {
451 if v := rv.Field(0); v.CanInterface() {
452 if h2i, ok := v.Interface().(closeIdleConnectionser); ok {
453 t.closeIdleFunc = h2i
454 return
455 }
456 }
457 }
458
459 if _, ok := t.TLSNextProto["h2"]; ok {
460
461 return
462 }
463 protocols := t.protocols()
464 if !protocols.HTTP2() && !protocols.UnencryptedHTTP2() {
465 return
466 }
467 if omitBundledHTTP2 {
468 return
469 }
470
471 t.configureHTTP2(protocols)
472 }
473
474 func (t *Transport) protocols() Protocols {
475 if t.Protocols != nil {
476 return *t.Protocols
477 }
478 var p Protocols
479 p.SetHTTP1(true)
480 switch {
481 case t.TLSNextProto != nil:
482
483
484 if t.TLSNextProto["h2"] != nil {
485 p.SetHTTP2(true)
486 }
487 case !t.ForceAttemptHTTP2 && (t.TLSClientConfig != nil || t.Dial != nil || t.DialContext != nil || t.hasCustomTLSDialer()):
488
489
490
491
492
493
494 case http2client.Value() == "0":
495 default:
496 p.SetHTTP2(true)
497 }
498 return p
499 }
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518 func ProxyFromEnvironment(req *Request) (*url.URL, error) {
519 return envProxyFunc()(req.URL)
520 }
521
522
523
524 func ProxyURL(fixedURL *url.URL) func(*Request) (*url.URL, error) {
525 return func(*Request) (*url.URL, error) {
526 return fixedURL, nil
527 }
528 }
529
530
531
532
533 type transportRequest struct {
534 *Request
535 extra Header
536 trace *httptrace.ClientTrace
537
538 ctx context.Context
539 cancel context.CancelCauseFunc
540
541 mu sync.Mutex
542 err error
543 }
544
545 func (tr *transportRequest) extraHeaders() Header {
546 if tr.extra == nil {
547 tr.extra = make(Header)
548 }
549 return tr.extra
550 }
551
552 func (tr *transportRequest) setError(err error) {
553 tr.mu.Lock()
554 if tr.err == nil {
555 tr.err = err
556 }
557 tr.mu.Unlock()
558 }
559
560
561
562 func (t *Transport) useRegisteredProtocol(req *Request) bool {
563 if req.URL.Scheme == "https" && req.requiresHTTP1() {
564
565
566
567
568 return false
569 }
570 return true
571 }
572
573
574
575
576 func (t *Transport) alternateRoundTripper(req *Request) RoundTripper {
577 if !t.useRegisteredProtocol(req) {
578 return nil
579 }
580 if req.URL.Scheme == "https" && t.h2Config != nil && t.h2Config.ExternalRoundTrip() {
581
582
583
584
585
586
587 return t.h2Config
588 }
589 altProto, _ := t.altProto.Load().(map[string]RoundTripper)
590 return altProto[req.URL.Scheme]
591 }
592
593 func validateHeaders(hdrs Header) string {
594 for k, vv := range hdrs {
595 if !httpguts.ValidHeaderFieldName(k) {
596 return fmt.Sprintf("field name %q", k)
597 }
598 for _, v := range vv {
599 if !httpguts.ValidHeaderFieldValue(v) {
600
601
602 return fmt.Sprintf("field value for %q", k)
603 }
604 }
605 }
606 return ""
607 }
608
609
610 func (t *Transport) roundTrip(req *Request) (_ *Response, err error) {
611 t.nextProtoOnce.Do(t.onceSetNextProtoDefaults)
612 ctx := req.Context()
613 trace := httptrace.ContextClientTrace(ctx)
614
615 if req.URL == nil {
616 req.closeBody()
617 return nil, errors.New("http: nil Request.URL")
618 }
619 if req.Header == nil {
620 req.closeBody()
621 return nil, errors.New("http: nil Request.Header")
622 }
623 scheme := req.URL.Scheme
624 isHTTP := scheme == "http" || scheme == "https"
625 if isHTTP {
626
627 if err := validateHeaders(req.Header); err != "" {
628 req.closeBody()
629 return nil, fmt.Errorf("net/http: invalid header %s", err)
630 }
631
632
633 if err := validateHeaders(req.Trailer); err != "" {
634 req.closeBody()
635 return nil, fmt.Errorf("net/http: invalid trailer %s", err)
636 }
637 }
638
639 origReq := req
640 req = setupRewindBody(req)
641
642 if altRT := t.alternateRoundTripper(req); altRT != nil {
643 if resp, err := altRT.RoundTrip(req); err != ErrSkipAltProtocol {
644 return resp, err
645 }
646 var err error
647 req, err = rewindBody(req)
648 if err != nil {
649 return nil, err
650 }
651 }
652 if !isHTTP {
653 req.closeBody()
654 return nil, badStringError("unsupported protocol scheme", scheme)
655 }
656 if req.Method != "" && !validMethod(req.Method) {
657 req.closeBody()
658 return nil, fmt.Errorf("net/http: invalid method %q", req.Method)
659 }
660 if req.URL.Host == "" {
661 req.closeBody()
662 return nil, errors.New("http: no Host in request URL")
663 }
664
665
666
667
668
669
670
671
672
673
674 ctx, cancel := context.WithCancelCause(req.Context())
675
676
677 if origReq.Cancel != nil {
678 go awaitLegacyCancel(ctx, cancel, origReq)
679 }
680
681 defer func() {
682 if err != nil {
683 cancel(err)
684 }
685 }()
686
687 for {
688 select {
689 case <-ctx.Done():
690 req.closeBody()
691 return nil, context.Cause(ctx)
692 default:
693 }
694
695
696 treq := &transportRequest{Request: req, trace: trace, ctx: ctx, cancel: cancel}
697 cm, err := t.connectMethodForRequest(treq)
698 if err != nil {
699 req.closeBody()
700 return nil, err
701 }
702
703
704
705
706
707 pconn, err := t.getConn(treq, cm)
708 if err != nil {
709 req.closeBody()
710 return nil, err
711 }
712
713 var resp *Response
714 if pconn.alt != nil {
715
716 resp, err = pconn.alt.RoundTrip(req)
717 } else {
718 resp, err = pconn.roundTrip(treq)
719 }
720 if err == nil {
721 if pconn.alt != nil {
722
723
724
725
726 cancel(errRequestDone)
727 }
728 resp.Request = origReq
729 return resp, nil
730 }
731
732
733 if http2isNoCachedConnError(err) {
734 if t.removeIdleConn(pconn) {
735 t.decConnsPerHost(pconn.cacheKey)
736 }
737 } else if !pconn.shouldRetryRequest(req, err) {
738
739
740 if e, ok := err.(nothingWrittenError); ok {
741 err = e.error
742 }
743 if e, ok := err.(transportReadFromServerError); ok {
744 err = e.err
745 }
746 if b, ok := req.Body.(*readTrackingBody); ok && !b.didClose.Load() {
747
748
749
750 req.closeBody()
751 }
752 return nil, err
753 }
754 testHookRoundTripRetried()
755
756
757 req, err = rewindBody(req)
758 if err != nil {
759 return nil, err
760 }
761 }
762 }
763
764 func http2isNoCachedConnError(err error) bool {
765 _, ok := err.(interface{ IsHTTP2NoCachedConnError() })
766 return ok
767 }
768
769 func awaitLegacyCancel(ctx context.Context, cancel context.CancelCauseFunc, req *Request) {
770 select {
771 case <-req.Cancel:
772 cancel(errRequestCanceled)
773 case <-ctx.Done():
774 }
775 }
776
777 var errCannotRewind = errors.New("net/http: cannot rewind body after connection loss")
778
779 type readTrackingBody struct {
780 io.ReadCloser
781 didRead bool
782 didClose atomic.Bool
783 }
784
785 func (r *readTrackingBody) Read(data []byte) (int, error) {
786 r.didRead = true
787 return r.ReadCloser.Read(data)
788 }
789
790 func (r *readTrackingBody) Close() error {
791 if !r.didClose.CompareAndSwap(false, true) {
792 return nil
793 }
794 return r.ReadCloser.Close()
795 }
796
797
798
799
800
801 func setupRewindBody(req *Request) *Request {
802 if req.Body == nil || req.Body == NoBody {
803 return req
804 }
805 newReq := *req
806 newReq.Body = &readTrackingBody{ReadCloser: req.Body}
807 return &newReq
808 }
809
810
811
812
813
814 func rewindBody(req *Request) (rewound *Request, err error) {
815 if req.Body == nil || req.Body == NoBody || (!req.Body.(*readTrackingBody).didRead && !req.Body.(*readTrackingBody).didClose.Load()) {
816 return req, nil
817 }
818 if !req.Body.(*readTrackingBody).didClose.Load() {
819 req.closeBody()
820 }
821 if req.GetBody == nil {
822 return nil, errCannotRewind
823 }
824 body, err := req.GetBody()
825 if err != nil {
826 return nil, err
827 }
828 newReq := *req
829 newReq.Body = &readTrackingBody{ReadCloser: body}
830 return &newReq, nil
831 }
832
833
834
835
836 func (pc *persistConn) shouldRetryRequest(req *Request, err error) bool {
837 if http2isNoCachedConnError(err) {
838
839
840
841
842
843
844 return true
845 }
846 if err == errMissingHost {
847
848 return false
849 }
850 if !pc.isReused() {
851
852
853
854
855
856
857
858 return false
859 }
860 if _, ok := err.(nothingWrittenError); ok {
861
862
863 return req.outgoingLength() == 0 || req.GetBody != nil
864 }
865 if !req.isReplayable() {
866
867 return false
868 }
869 if _, ok := err.(transportReadFromServerError); ok {
870
871
872 return true
873 }
874 if err == errServerClosedIdle {
875
876
877
878 return true
879 }
880 return false
881 }
882
883
884 var ErrSkipAltProtocol = internal.ErrSkipAltProtocol
885
886
887
888
889
890
891
892
893
894
895
896 func (t *Transport) RegisterProtocol(scheme string, rt RoundTripper) {
897 if err := t.registerProtocol(scheme, rt); err != nil {
898 panic(err)
899 }
900 }
901
902 func (t *Transport) registerProtocol(scheme string, rt RoundTripper) error {
903 t.altMu.Lock()
904 defer t.altMu.Unlock()
905
906 if scheme == "http/2" {
907 if t.h2Config != nil {
908 panic("http: HTTP/2 Transport already registered")
909 }
910 var ok bool
911 if t.h2Config, ok = rt.(http2ExternalTransportConfig); !ok {
912 panic("http: HTTP/2 configuration does not implement ExternalTransportConfig")
913 }
914 t.h2Config.Registered(t)
915 }
916
917 if scheme == "http/3" {
918 if t.h3Transport != nil {
919 panic("http: HTTP/3 Transport already registered")
920 }
921 var ok bool
922 if t.h3Transport, ok = rt.(dialClientConner); !ok {
923 panic("http: HTTP/3 RoundTripper does not implement DialClientConn")
924 }
925
926
927 if r, ok := rt.(interface {
928 Registered(*Transport)
929 }); ok {
930 r.Registered(t)
931 }
932 return nil
933 }
934
935 oldMap, _ := t.altProto.Load().(map[string]RoundTripper)
936 if _, exists := oldMap[scheme]; exists {
937 return errors.New("protocol " + scheme + " already registered")
938 }
939 newMap := maps.Clone(oldMap)
940 if newMap == nil {
941 newMap = make(map[string]RoundTripper)
942 }
943 newMap[scheme] = rt
944 t.altProto.Store(newMap)
945 return nil
946 }
947
948
949
950
951
952 func (t *Transport) CloseIdleConnections() {
953 t.nextProtoOnce.Do(t.onceSetNextProtoDefaults)
954 t.idleMu.Lock()
955 m := t.idleConn
956 t.idleConn = nil
957 t.closeIdle = true
958 t.idleLRU = connLRU{}
959 t.idleMu.Unlock()
960 for _, conns := range m {
961 for _, pconn := range conns {
962 pconn.close(errCloseIdleConns)
963 }
964 }
965 t.connsPerHostMu.Lock()
966 t.dialsInProgress.all(func(w *wantConn) {
967 if w.cancelCtx != nil && !w.waiting() {
968 w.cancelCtx()
969 }
970 })
971 t.connsPerHostMu.Unlock()
972
973
974
975
976 if tr2 := t.h2Transport; tr2 != nil {
977 tr2.CloseIdleConnections()
978 }
979
980
981
982
983 if t2 := t.closeIdleFunc; t2 != nil {
984 t2.CloseIdleConnections()
985 }
986
987 if cc, ok := t.h3Transport.(closeIdleConnectionser); ok {
988 cc.CloseIdleConnections()
989 }
990 }
991
992
993
994
995
996 func (t *Transport) CancelRequest(req *Request) {
997 }
998
999
1000
1001
1002
1003 var (
1004 envProxyOnce sync.Once
1005 envProxyFuncValue func(*url.URL) (*url.URL, error)
1006 )
1007
1008
1009
1010 func envProxyFunc() func(*url.URL) (*url.URL, error) {
1011 envProxyOnce.Do(func() {
1012 envProxyFuncValue = httpproxy.FromEnvironment().ProxyFunc()
1013 })
1014 return envProxyFuncValue
1015 }
1016
1017
1018 func resetProxyConfig() {
1019 envProxyOnce = sync.Once{}
1020 envProxyFuncValue = nil
1021 }
1022
1023 func (t *Transport) connectMethodForRequest(treq *transportRequest) (cm connectMethod, err error) {
1024 cm.targetScheme = treq.URL.Scheme
1025 cm.targetAddr = canonicalAddr(treq.URL)
1026 if t.Proxy != nil {
1027 cm.proxyURL, err = t.Proxy(treq.Request)
1028 }
1029 cm.onlyH1 = treq.requiresHTTP1()
1030 return cm, err
1031 }
1032
1033
1034
1035 func (cm *connectMethod) proxyAuth() string {
1036 if cm.proxyURL == nil {
1037 return ""
1038 }
1039 if u := cm.proxyURL.User; u != nil {
1040 username := u.Username()
1041 password, _ := u.Password()
1042 return "Basic " + basicAuth(username, password)
1043 }
1044 return ""
1045 }
1046
1047
1048 var (
1049 errKeepAlivesDisabled = errors.New("http: putIdleConn: keep alives disabled")
1050 errConnBroken = errors.New("http: putIdleConn: connection is in bad state")
1051 errCloseIdle = errors.New("http: putIdleConn: CloseIdleConnections was called")
1052 errTooManyIdle = errors.New("http: putIdleConn: too many idle connections")
1053 errTooManyIdleHost = errors.New("http: putIdleConn: too many idle connections for host")
1054 errCloseIdleConns = errors.New("http: CloseIdleConnections called")
1055 errReadLoopExiting = errors.New("http: persistConn.readLoop exiting")
1056 errIdleConnTimeout = errors.New("http: idle connection timeout")
1057
1058
1059
1060
1061
1062 errServerClosedIdle = errors.New("http: server closed idle connection")
1063 )
1064
1065
1066
1067
1068
1069
1070
1071
1072
1073 type transportReadFromServerError struct {
1074 err error
1075 }
1076
1077 func (e transportReadFromServerError) Unwrap() error { return e.err }
1078
1079 func (e transportReadFromServerError) Error() string {
1080 return fmt.Sprintf("net/http: Transport failed to read from server: %v", e.err)
1081 }
1082
1083 func (t *Transport) putOrCloseIdleConn(pconn *persistConn) {
1084 if err := t.tryPutIdleConn(pconn); err != nil {
1085 pconn.close(err)
1086 }
1087 }
1088
1089 func (t *Transport) maxIdleConnsPerHost() int {
1090 if v := t.MaxIdleConnsPerHost; v != 0 {
1091 return v
1092 }
1093 return DefaultMaxIdleConnsPerHost
1094 }
1095
1096 func (t *Transport) keepAlivesDisabled() bool {
1097 return t.DisableKeepAlives || t.MaxIdleConnsPerHost < 0
1098 }
1099
1100
1101
1102
1103
1104
1105 func (t *Transport) tryPutIdleConn(pconn *persistConn) error {
1106 if t.keepAlivesDisabled() {
1107 return errKeepAlivesDisabled
1108 }
1109 if pconn.isBroken() {
1110 return errConnBroken
1111 }
1112 pconn.markReused()
1113 if pconn.isClientConn {
1114
1115 defer pconn.internalStateHook()
1116 pconn.mu.Lock()
1117 defer pconn.mu.Unlock()
1118 if !pconn.inFlight {
1119 panic("pconn is not in flight")
1120 }
1121 pconn.inFlight = false
1122 select {
1123 case pconn.availch <- struct{}{}:
1124 default:
1125 panic("unable to make pconn available")
1126 }
1127 return nil
1128 }
1129
1130 t.idleMu.Lock()
1131 defer t.idleMu.Unlock()
1132
1133
1134
1135
1136 if pconn.alt != nil && t.idleLRU.m[pconn] != nil {
1137 return nil
1138 }
1139
1140
1141
1142
1143
1144 key := pconn.cacheKey
1145 if q, ok := t.idleConnWait[key]; ok {
1146 done := false
1147 if pconn.alt == nil {
1148
1149
1150 for q.len() > 0 {
1151 w := q.popFront()
1152 if w.tryDeliver(pconn, nil, time.Time{}) {
1153 done = true
1154 break
1155 }
1156 }
1157 } else {
1158
1159
1160
1161
1162 for q.len() > 0 {
1163 w := q.popFront()
1164 w.tryDeliver(pconn, nil, time.Time{})
1165 }
1166 }
1167 if q.len() == 0 {
1168 delete(t.idleConnWait, key)
1169 } else {
1170 t.idleConnWait[key] = q
1171 }
1172 if done {
1173 return nil
1174 }
1175 }
1176
1177 if t.closeIdle {
1178 return errCloseIdle
1179 }
1180 if t.idleConn == nil {
1181 t.idleConn = make(map[connectMethodKey][]*persistConn)
1182 }
1183 idles := t.idleConn[key]
1184 if len(idles) >= t.maxIdleConnsPerHost() {
1185 return errTooManyIdleHost
1186 }
1187 for _, exist := range idles {
1188 if exist == pconn {
1189 log.Fatalf("dup idle pconn %p in freelist", pconn)
1190 }
1191 }
1192 t.idleConn[key] = append(idles, pconn)
1193 t.idleLRU.add(pconn)
1194 if t.MaxIdleConns != 0 && t.idleLRU.len() > t.MaxIdleConns {
1195 oldest := t.idleLRU.removeOldest()
1196 oldest.close(errTooManyIdle)
1197 t.removeIdleConnLocked(oldest)
1198 }
1199
1200
1201
1202
1203 if t.IdleConnTimeout > 0 && pconn.alt == nil {
1204 if pconn.idleTimer != nil {
1205 pconn.idleTimer.Reset(t.IdleConnTimeout)
1206 } else {
1207 pconn.idleTimer = time.AfterFunc(t.IdleConnTimeout, pconn.closeConnIfStillIdle)
1208 }
1209 }
1210 pconn.idleAt = time.Now()
1211 return nil
1212 }
1213
1214
1215
1216
1217 func (t *Transport) queueForIdleConn(w *wantConn) (delivered bool) {
1218 if t.DisableKeepAlives {
1219 return false
1220 }
1221
1222 t.idleMu.Lock()
1223 defer t.idleMu.Unlock()
1224
1225
1226
1227 t.closeIdle = false
1228
1229 if w == nil {
1230
1231 return false
1232 }
1233
1234
1235
1236
1237 var oldTime time.Time
1238 if t.IdleConnTimeout > 0 {
1239 oldTime = time.Now().Add(-t.IdleConnTimeout)
1240 }
1241
1242
1243 if list, ok := t.idleConn[w.key]; ok {
1244 stop := false
1245 delivered := false
1246 for len(list) > 0 && !stop {
1247 pconn := list[len(list)-1]
1248
1249
1250
1251
1252 tooOld := !oldTime.IsZero() && pconn.idleAt.Round(0).Before(oldTime)
1253 if tooOld {
1254
1255
1256
1257 go pconn.closeConnIfStillIdle()
1258 }
1259 if pconn.isBroken() || tooOld {
1260
1261
1262
1263
1264
1265 list = list[:len(list)-1]
1266 continue
1267 }
1268 delivered = w.tryDeliver(pconn, nil, pconn.idleAt)
1269 if delivered {
1270 if pconn.alt != nil {
1271
1272
1273 } else {
1274
1275
1276 t.idleLRU.remove(pconn)
1277 list = list[:len(list)-1]
1278 }
1279 }
1280 stop = true
1281 }
1282 if len(list) > 0 {
1283 t.idleConn[w.key] = list
1284 } else {
1285 delete(t.idleConn, w.key)
1286 }
1287 if stop {
1288 return delivered
1289 }
1290 }
1291
1292
1293 if t.idleConnWait == nil {
1294 t.idleConnWait = make(map[connectMethodKey]wantConnQueue)
1295 }
1296 q := t.idleConnWait[w.key]
1297 q.cleanFrontNotWaiting()
1298 q.pushBack(w)
1299 t.idleConnWait[w.key] = q
1300 return false
1301 }
1302
1303
1304 func (t *Transport) removeIdleConn(pconn *persistConn) bool {
1305 if pconn.isClientConn {
1306 return true
1307 }
1308 t.idleMu.Lock()
1309 defer t.idleMu.Unlock()
1310 return t.removeIdleConnLocked(pconn)
1311 }
1312
1313
1314 func (t *Transport) removeIdleConnLocked(pconn *persistConn) bool {
1315 if pconn.idleTimer != nil {
1316 pconn.idleTimer.Stop()
1317 }
1318 t.idleLRU.remove(pconn)
1319 key := pconn.cacheKey
1320 pconns := t.idleConn[key]
1321 var removed bool
1322 switch len(pconns) {
1323 case 0:
1324
1325 case 1:
1326 if pconns[0] == pconn {
1327 delete(t.idleConn, key)
1328 removed = true
1329 }
1330 default:
1331 for i, v := range pconns {
1332 if v != pconn {
1333 continue
1334 }
1335
1336
1337 copy(pconns[i:], pconns[i+1:])
1338 t.idleConn[key] = pconns[:len(pconns)-1]
1339 removed = true
1340 break
1341 }
1342 }
1343 return removed
1344 }
1345
1346 var zeroDialer net.Dialer
1347
1348 func (t *Transport) dial(ctx context.Context, network, addr string) (net.Conn, error) {
1349 if t.DialContext != nil {
1350 c, err := t.DialContext(ctx, network, addr)
1351 if c == nil && err == nil {
1352 err = errors.New("net/http: Transport.DialContext hook returned (nil, nil)")
1353 }
1354 return c, err
1355 }
1356 if t.Dial != nil {
1357 c, err := t.Dial(network, addr)
1358 if c == nil && err == nil {
1359 err = errors.New("net/http: Transport.Dial hook returned (nil, nil)")
1360 }
1361 return c, err
1362 }
1363 return zeroDialer.DialContext(ctx, network, addr)
1364 }
1365
1366
1367
1368
1369
1370
1371
1372 type wantConn struct {
1373 cm connectMethod
1374 key connectMethodKey
1375
1376
1377
1378
1379 beforeDial func()
1380 afterDial func()
1381
1382 mu sync.Mutex
1383 ctx context.Context
1384 cancelCtx context.CancelFunc
1385 done bool
1386 result chan connOrError
1387 }
1388
1389 type connOrError struct {
1390 pc *persistConn
1391 err error
1392 idleAt time.Time
1393 }
1394
1395
1396 func (w *wantConn) waiting() bool {
1397 w.mu.Lock()
1398 defer w.mu.Unlock()
1399
1400 return !w.done
1401 }
1402
1403
1404 func (w *wantConn) getCtxForDial() context.Context {
1405 w.mu.Lock()
1406 defer w.mu.Unlock()
1407
1408 return w.ctx
1409 }
1410
1411
1412 func (w *wantConn) tryDeliver(pc *persistConn, err error, idleAt time.Time) bool {
1413 w.mu.Lock()
1414 defer w.mu.Unlock()
1415
1416 if w.done {
1417 return false
1418 }
1419 if (pc == nil) == (err == nil) {
1420 panic("net/http: internal error: misuse of tryDeliver")
1421 }
1422 w.ctx = nil
1423 w.done = true
1424
1425 w.result <- connOrError{pc: pc, err: err, idleAt: idleAt}
1426 close(w.result)
1427
1428 return true
1429 }
1430
1431
1432
1433 func (w *wantConn) cancel(t *Transport) {
1434 w.mu.Lock()
1435 var pc *persistConn
1436 if w.done {
1437 if r, ok := <-w.result; ok {
1438 pc = r.pc
1439 }
1440 } else {
1441 close(w.result)
1442 }
1443 w.ctx = nil
1444 w.done = true
1445 w.mu.Unlock()
1446
1447
1448
1449
1450 if pc != nil && pc.alt == nil {
1451 t.putOrCloseIdleConn(pc)
1452 }
1453 }
1454
1455
1456 type wantConnQueue struct {
1457
1458
1459
1460
1461
1462
1463
1464
1465
1466
1467 head []*wantConn
1468 headPos int
1469 tail []*wantConn
1470 }
1471
1472
1473 func (q *wantConnQueue) len() int {
1474 return len(q.head) - q.headPos + len(q.tail)
1475 }
1476
1477
1478 func (q *wantConnQueue) pushBack(w *wantConn) {
1479 q.tail = append(q.tail, w)
1480 }
1481
1482
1483 func (q *wantConnQueue) popFront() *wantConn {
1484 if q.headPos >= len(q.head) {
1485 if len(q.tail) == 0 {
1486 return nil
1487 }
1488
1489 q.head, q.headPos, q.tail = q.tail, 0, q.head[:0]
1490 }
1491 w := q.head[q.headPos]
1492 q.head[q.headPos] = nil
1493 q.headPos++
1494 return w
1495 }
1496
1497
1498 func (q *wantConnQueue) peekFront() *wantConn {
1499 if q.headPos < len(q.head) {
1500 return q.head[q.headPos]
1501 }
1502 if len(q.tail) > 0 {
1503 return q.tail[0]
1504 }
1505 return nil
1506 }
1507
1508
1509
1510 func (q *wantConnQueue) cleanFrontNotWaiting() (cleaned bool) {
1511 for {
1512 w := q.peekFront()
1513 if w == nil || w.waiting() {
1514 return cleaned
1515 }
1516 q.popFront()
1517 cleaned = true
1518 }
1519 }
1520
1521
1522 func (q *wantConnQueue) cleanFrontCanceled() {
1523 for {
1524 w := q.peekFront()
1525 if w == nil || w.cancelCtx != nil {
1526 return
1527 }
1528 q.popFront()
1529 }
1530 }
1531
1532
1533
1534 func (q *wantConnQueue) all(f func(*wantConn)) {
1535 for _, w := range q.head[q.headPos:] {
1536 f(w)
1537 }
1538 for _, w := range q.tail {
1539 f(w)
1540 }
1541 }
1542
1543 func (t *Transport) customDialTLS(ctx context.Context, network, addr string) (conn net.Conn, err error) {
1544 if t.DialTLSContext != nil {
1545 conn, err = t.DialTLSContext(ctx, network, addr)
1546 } else {
1547 conn, err = t.DialTLS(network, addr)
1548 }
1549 if conn == nil && err == nil {
1550 err = errors.New("net/http: Transport.DialTLS or DialTLSContext returned (nil, nil)")
1551 }
1552 return
1553 }
1554
1555
1556
1557
1558
1559 func (t *Transport) getConn(treq *transportRequest, cm connectMethod) (_ *persistConn, err error) {
1560 req := treq.Request
1561 trace := treq.trace
1562 ctx := req.Context()
1563 if trace != nil && trace.GetConn != nil {
1564 trace.GetConn(cm.addr())
1565 }
1566
1567
1568
1569
1570
1571
1572 dialCtx, dialCancel := context.WithCancel(context.WithoutCancel(ctx))
1573
1574 w := &wantConn{
1575 cm: cm,
1576 key: cm.key(),
1577 ctx: dialCtx,
1578 cancelCtx: dialCancel,
1579 result: make(chan connOrError, 1),
1580 beforeDial: testHookPrePendingDial,
1581 afterDial: testHookPostPendingDial,
1582 }
1583 defer func() {
1584 if err != nil {
1585 w.cancel(t)
1586 }
1587 }()
1588
1589
1590 if delivered := t.queueForIdleConn(w); !delivered {
1591 t.queueForDial(w)
1592 }
1593
1594
1595 select {
1596 case r := <-w.result:
1597
1598
1599 if r.pc != nil && r.pc.alt == nil && trace != nil && trace.GotConn != nil {
1600 info := httptrace.GotConnInfo{
1601 Conn: r.pc.conn,
1602 Reused: r.pc.isReused(),
1603 }
1604 if !r.idleAt.IsZero() {
1605 info.WasIdle = true
1606 info.IdleTime = time.Since(r.idleAt)
1607 }
1608 trace.GotConn(info)
1609 }
1610 if r.err != nil {
1611
1612
1613
1614 select {
1615 case <-treq.ctx.Done():
1616 err := context.Cause(treq.ctx)
1617 if err == errRequestCanceled {
1618 err = errRequestCanceledConn
1619 }
1620 return nil, err
1621 default:
1622
1623 }
1624 }
1625 return r.pc, r.err
1626 case <-treq.ctx.Done():
1627 err := context.Cause(treq.ctx)
1628 if err == errRequestCanceled {
1629 err = errRequestCanceledConn
1630 }
1631 return nil, err
1632 }
1633 }
1634
1635
1636
1637 func (t *Transport) queueForDial(w *wantConn) {
1638 w.beforeDial()
1639
1640 t.connsPerHostMu.Lock()
1641 defer t.connsPerHostMu.Unlock()
1642
1643 if t.MaxConnsPerHost <= 0 {
1644 t.startDialConnForLocked(w)
1645 return
1646 }
1647
1648 if n := t.connsPerHost[w.key]; n < t.MaxConnsPerHost {
1649 if t.connsPerHost == nil {
1650 t.connsPerHost = make(map[connectMethodKey]int)
1651 }
1652 t.connsPerHost[w.key] = n + 1
1653 t.startDialConnForLocked(w)
1654 return
1655 }
1656
1657 if t.connsPerHostWait == nil {
1658 t.connsPerHostWait = make(map[connectMethodKey]wantConnQueue)
1659 }
1660 q := t.connsPerHostWait[w.key]
1661 q.cleanFrontNotWaiting()
1662 q.pushBack(w)
1663 t.connsPerHostWait[w.key] = q
1664 }
1665
1666
1667
1668 func (t *Transport) startDialConnForLocked(w *wantConn) {
1669 t.dialsInProgress.cleanFrontCanceled()
1670 t.dialsInProgress.pushBack(w)
1671 go func() {
1672 t.dialConnFor(w)
1673 t.connsPerHostMu.Lock()
1674 defer t.connsPerHostMu.Unlock()
1675 w.cancelCtx = nil
1676 }()
1677 }
1678
1679
1680
1681
1682 func (t *Transport) dialConnFor(w *wantConn) {
1683 defer w.afterDial()
1684 ctx := w.getCtxForDial()
1685 if ctx == nil {
1686 t.decConnsPerHost(w.key)
1687 return
1688 }
1689
1690 const isClientConn = false
1691 pc, err := t.dialConn(ctx, w.cm, isClientConn, nil)
1692 if err == nil && pc.alt != nil {
1693
1694
1695 t.putOrCloseIdleConn(pc)
1696 }
1697 delivered := w.tryDeliver(pc, err, time.Time{})
1698 if err == nil && !delivered && pc.alt == nil {
1699
1700
1701 t.putOrCloseIdleConn(pc)
1702 }
1703 if err != nil {
1704 t.decConnsPerHost(w.key)
1705 }
1706 }
1707
1708
1709
1710 func (t *Transport) decConnsPerHost(key connectMethodKey) {
1711 if t.MaxConnsPerHost <= 0 {
1712 return
1713 }
1714
1715 t.connsPerHostMu.Lock()
1716 defer t.connsPerHostMu.Unlock()
1717 n := t.connsPerHost[key]
1718 if n == 0 {
1719
1720
1721 panic("net/http: internal error: connCount underflow")
1722 }
1723
1724
1725
1726
1727
1728 if q := t.connsPerHostWait[key]; q.len() > 0 {
1729 done := false
1730 for q.len() > 0 {
1731 w := q.popFront()
1732 if w.waiting() {
1733 t.startDialConnForLocked(w)
1734 done = true
1735 break
1736 }
1737 }
1738 if q.len() == 0 {
1739 delete(t.connsPerHostWait, key)
1740 } else {
1741
1742
1743 t.connsPerHostWait[key] = q
1744 }
1745 if done {
1746 return
1747 }
1748 }
1749
1750
1751 if n--; n == 0 {
1752 delete(t.connsPerHost, key)
1753 } else {
1754 t.connsPerHost[key] = n
1755 }
1756 }
1757
1758 func (t *Transport) tlsConfigForDial(host string) (*tls.Config, error) {
1759 firstTLSHost, _, err := net.SplitHostPort(host)
1760 if err != nil {
1761 return nil, err
1762 }
1763 cfg := cloneTLSConfig(t.TLSClientConfig)
1764 if cfg.ServerName == "" {
1765 cfg.ServerName = firstTLSHost
1766 }
1767 return cfg, nil
1768 }
1769
1770
1771
1772
1773 func (pconn *persistConn) addTLS(ctx context.Context, addr string, trace *httptrace.ClientTrace) error {
1774 cfg, err := pconn.t.tlsConfigForDial(addr)
1775 if err != nil {
1776 pconn.conn.Close()
1777 return err
1778 }
1779 if pconn.cacheKey.onlyH1 {
1780 cfg.NextProtos = nil
1781 }
1782 plainConn := pconn.conn
1783 tlsConn := tls.Client(plainConn, cfg)
1784 errc := make(chan error, 2)
1785 var timer *time.Timer
1786 if d := pconn.t.TLSHandshakeTimeout; d != 0 {
1787 timer = time.AfterFunc(d, func() {
1788 errc <- tlsHandshakeTimeoutError{}
1789 })
1790 }
1791 go func() {
1792 if trace != nil && trace.TLSHandshakeStart != nil {
1793 trace.TLSHandshakeStart()
1794 }
1795 err := tlsConn.HandshakeContext(ctx)
1796 if timer != nil {
1797 timer.Stop()
1798 }
1799 errc <- err
1800 }()
1801 if err := <-errc; err != nil {
1802 plainConn.Close()
1803 if err == (tlsHandshakeTimeoutError{}) {
1804
1805
1806 <-errc
1807 }
1808 if trace != nil && trace.TLSHandshakeDone != nil {
1809 trace.TLSHandshakeDone(tls.ConnectionState{}, err)
1810 }
1811 return err
1812 }
1813 cs := tlsConn.ConnectionState()
1814 if trace != nil && trace.TLSHandshakeDone != nil {
1815 trace.TLSHandshakeDone(cs, nil)
1816 }
1817 pconn.tlsState = &cs
1818 pconn.conn = tlsConn
1819 return nil
1820 }
1821
1822 type erringRoundTripper interface {
1823 RoundTripErr() error
1824 }
1825
1826 var testHookProxyConnectTimeout = context.WithTimeout
1827
1828 func (t *Transport) dialConn(ctx context.Context, cm connectMethod, isClientConn bool, internalStateHook func()) (pconn *persistConn, err error) {
1829
1830
1831
1832
1833 if p := t.protocols(); p.http3() {
1834 if p.HTTP1() || p.HTTP2() || p.UnencryptedHTTP2() {
1835 return nil, errors.New("http: when using HTTP3, Transport.Protocols must contain only HTTP3")
1836 }
1837 if t.h3Transport == nil {
1838 return nil, errors.New("http: Transport.Protocols contains HTTP3, but Transport does not support HTTP/3")
1839 }
1840 tlsConfig, err := t.tlsConfigForDial(cm.addr())
1841 if err != nil {
1842 return nil, err
1843 }
1844 tlsConfig.NextProtos = []string{"h3"}
1845 rt, err := t.h3Transport.DialClientConn(ctx, cm.addr(), cm.proxyURL, tlsConfig, internalStateHook)
1846 if err != nil {
1847 return nil, err
1848 }
1849 return &persistConn{
1850 t: t,
1851 cacheKey: cm.key(),
1852 alt: rt,
1853 }, nil
1854 }
1855
1856 pconn = &persistConn{
1857 t: t,
1858 cacheKey: cm.key(),
1859 reqch: make(chan requestAndChan, 1),
1860 writech: make(chan writeRequest, 1),
1861 closech: make(chan struct{}),
1862 writeErrCh: make(chan error, 1),
1863 writeLoopDone: make(chan struct{}),
1864 isClientConn: isClientConn,
1865 internalStateHook: internalStateHook,
1866 }
1867 trace := httptrace.ContextClientTrace(ctx)
1868 wrapErr := func(err error) error {
1869 if cm.proxyURL != nil {
1870
1871 return &net.OpError{Op: "proxyconnect", Net: "tcp", Err: err}
1872 }
1873 return err
1874 }
1875
1876 if rt, err := t.http2ExternalDial(ctx, cm); err != errors.ErrUnsupported {
1877 if err != nil {
1878 return nil, err
1879 }
1880 return &persistConn{t: t, cacheKey: pconn.cacheKey, alt: rt}, nil
1881 }
1882
1883 if cm.scheme() == "https" && t.hasCustomTLSDialer() {
1884 var err error
1885 pconn.conn, err = t.customDialTLS(ctx, "tcp", cm.addr())
1886 if err != nil {
1887 return nil, wrapErr(err)
1888 }
1889 type connectionStater interface {
1890 ConnectionState() tls.ConnectionState
1891 }
1892 type handshaker interface {
1893 HandshakeContext(context.Context) error
1894 }
1895 if cstater, ok := pconn.conn.(connectionStater); ok {
1896 if trace != nil && trace.TLSHandshakeStart != nil {
1897 trace.TLSHandshakeStart()
1898 }
1899 if handshaker, ok := cstater.(handshaker); ok {
1900
1901
1902 if err := handshaker.HandshakeContext(ctx); err != nil {
1903 go pconn.conn.Close()
1904 if trace != nil && trace.TLSHandshakeDone != nil {
1905 trace.TLSHandshakeDone(tls.ConnectionState{}, err)
1906 }
1907 return nil, err
1908 }
1909 }
1910 cs := cstater.ConnectionState()
1911 if trace != nil && trace.TLSHandshakeDone != nil {
1912 trace.TLSHandshakeDone(cs, nil)
1913 }
1914 pconn.tlsState = &cs
1915 }
1916 } else {
1917 conn, err := t.dial(ctx, "tcp", cm.addr())
1918 if err != nil {
1919 return nil, wrapErr(err)
1920 }
1921 pconn.conn = conn
1922 if cm.scheme() == "https" {
1923 if err = pconn.addTLS(ctx, cm.addr(), trace); err != nil {
1924 return nil, wrapErr(err)
1925 }
1926 }
1927 }
1928
1929
1930 switch {
1931 case cm.proxyURL == nil:
1932
1933 case cm.proxyURL.Scheme == "socks5" || cm.proxyURL.Scheme == "socks5h":
1934 conn := pconn.conn
1935 d := socksNewDialer("tcp", conn.RemoteAddr().String())
1936 if u := cm.proxyURL.User; u != nil {
1937 auth := &socksUsernamePassword{
1938 Username: u.Username(),
1939 }
1940 auth.Password, _ = u.Password()
1941 d.AuthMethods = []socksAuthMethod{
1942 socksAuthMethodNotRequired,
1943 socksAuthMethodUsernamePassword,
1944 }
1945 d.Authenticate = auth.Authenticate
1946 }
1947 if _, err := d.DialWithConn(ctx, conn, "tcp", cm.targetAddr); err != nil {
1948 conn.Close()
1949 return nil, err
1950 }
1951 case cm.targetScheme == "http":
1952 pconn.isProxy = true
1953 if pa := cm.proxyAuth(); pa != "" {
1954 pconn.mutateHeaderFunc = func(h Header) {
1955 h.Set("Proxy-Authorization", pa)
1956 }
1957 }
1958 case cm.targetScheme == "https":
1959 conn := pconn.conn
1960 var hdr Header
1961 if t.GetProxyConnectHeader != nil {
1962 var err error
1963 hdr, err = t.GetProxyConnectHeader(ctx, cm.proxyURL, cm.targetAddr)
1964 if err != nil {
1965 conn.Close()
1966 return nil, err
1967 }
1968 } else {
1969 hdr = t.ProxyConnectHeader
1970 }
1971 if hdr == nil {
1972 hdr = make(Header)
1973 }
1974 if pa := cm.proxyAuth(); pa != "" {
1975 hdr = hdr.Clone()
1976 hdr.Set("Proxy-Authorization", pa)
1977 }
1978 connectReq := &Request{
1979 Method: "CONNECT",
1980 URL: &url.URL{Opaque: cm.targetAddr},
1981 Host: cm.targetAddr,
1982 Header: hdr,
1983 }
1984
1985
1986
1987
1988 connectCtx, cancel := testHookProxyConnectTimeout(ctx, 1*time.Minute)
1989 defer cancel()
1990
1991 didReadResponse := make(chan struct{})
1992 var (
1993 resp *Response
1994 err error
1995 )
1996
1997 go func() {
1998 defer close(didReadResponse)
1999 err = connectReq.Write(conn)
2000 if err != nil {
2001 return
2002 }
2003
2004
2005 br := bufio.NewReader(&io.LimitedReader{R: conn, N: t.maxHeaderResponseSize()})
2006 resp, err = ReadResponse(br, connectReq)
2007 }()
2008 select {
2009 case <-connectCtx.Done():
2010 conn.Close()
2011 <-didReadResponse
2012 return nil, connectCtx.Err()
2013 case <-didReadResponse:
2014
2015 }
2016 if err != nil {
2017 conn.Close()
2018 return nil, err
2019 }
2020
2021 if t.OnProxyConnectResponse != nil {
2022 err = t.OnProxyConnectResponse(ctx, cm.proxyURL, connectReq, resp)
2023 if err != nil {
2024 conn.Close()
2025 return nil, err
2026 }
2027 }
2028
2029 if resp.StatusCode != 200 {
2030 _, text, ok := strings.Cut(resp.Status, " ")
2031 conn.Close()
2032 if !ok {
2033 return nil, errors.New("unknown status code")
2034 }
2035 return nil, errors.New(text)
2036 }
2037 }
2038
2039 if cm.proxyURL != nil && cm.targetScheme == "https" {
2040 if err := pconn.addTLS(ctx, cm.targetAddr, trace); err != nil {
2041 return nil, err
2042 }
2043 }
2044
2045
2046 unencryptedHTTP2 := pconn.tlsState == nil &&
2047 t.Protocols != nil &&
2048 t.Protocols.UnencryptedHTTP2() &&
2049 !t.Protocols.HTTP1()
2050
2051 http2 := unencryptedHTTP2 ||
2052 (pconn.tlsState != nil && pconn.tlsState.NegotiatedProtocol == "h2")
2053
2054 if http2 && t.h2Transport != nil {
2055 if isClientConn {
2056 cc, err := t.http2NewClientConn(pconn.conn, internalStateHook)
2057 if err == nil {
2058 return &persistConn{t: t, cacheKey: pconn.cacheKey, alt: cc, isClientConn: true}, nil
2059 }
2060 if err != errors.ErrUnsupported {
2061 return nil, err
2062 }
2063 } else {
2064 rt, err := t.http2AddConn(cm.targetScheme, cm.targetAddr, pconn.conn)
2065 if err == nil {
2066 return &persistConn{t: t, cacheKey: pconn.cacheKey, alt: rt}, nil
2067 }
2068 if err != errors.ErrUnsupported {
2069 return nil, err
2070 }
2071 }
2072 }
2073
2074 if isClientConn && (unencryptedHTTP2 || (pconn.tlsState != nil && pconn.tlsState.NegotiatedProtocol == "h2")) {
2075 altProto, _ := t.altProto.Load().(map[string]RoundTripper)
2076 h2, ok := altProto["https"].(newClientConner)
2077 if !ok {
2078 return nil, errors.New("http: HTTP/2 implementation does not support NewClientConn (update golang.org/x/net?)")
2079 }
2080 alt, err := h2.NewClientConn(pconn.conn, internalStateHook)
2081 if err != nil {
2082 pconn.conn.Close()
2083 return nil, err
2084 }
2085 return &persistConn{t: t, cacheKey: pconn.cacheKey, alt: alt, isClientConn: true}, nil
2086 }
2087
2088 if unencryptedHTTP2 {
2089 next, ok := t.TLSNextProto[nextProtoUnencryptedHTTP2]
2090 if !ok {
2091 return nil, errors.New("http: Transport does not support unencrypted HTTP/2")
2092 }
2093 alt := next(cm.targetAddr, unencryptedTLSConn(pconn.conn))
2094 if e, ok := alt.(erringRoundTripper); ok {
2095
2096 return nil, e.RoundTripErr()
2097 }
2098 return &persistConn{t: t, cacheKey: pconn.cacheKey, alt: alt}, nil
2099 }
2100
2101 if s := pconn.tlsState; s != nil && s.NegotiatedProtocolIsMutual && s.NegotiatedProtocol != "" {
2102 tlsConn, tlsConnOK := pconn.conn.(*tls.Conn)
2103 if next, ok := t.TLSNextProto[s.NegotiatedProtocol]; tlsConnOK && ok {
2104 alt := next(cm.targetAddr, tlsConn)
2105 if e, ok := alt.(erringRoundTripper); ok {
2106
2107 return nil, e.RoundTripErr()
2108 }
2109 return &persistConn{t: t, cacheKey: pconn.cacheKey, alt: alt}, nil
2110 }
2111 }
2112
2113 pconn.br = bufio.NewReaderSize(pconn, t.readBufferSize())
2114 pconn.bw = bufio.NewWriterSize(persistConnWriter{pconn}, t.writeBufferSize())
2115
2116 go pconn.readLoop()
2117 go pconn.writeLoop()
2118 return pconn, nil
2119 }
2120
2121
2122
2123
2124
2125
2126
2127 type persistConnWriter struct {
2128 pc *persistConn
2129 }
2130
2131 func (w persistConnWriter) Write(p []byte) (n int, err error) {
2132 n, err = w.pc.conn.Write(p)
2133 w.pc.nwrite += int64(n)
2134 return
2135 }
2136
2137
2138
2139
2140 func (w persistConnWriter) ReadFrom(r io.Reader) (n int64, err error) {
2141 n, err = io.Copy(w.pc.conn, r)
2142 w.pc.nwrite += n
2143 return
2144 }
2145
2146 var _ io.ReaderFrom = (*persistConnWriter)(nil)
2147
2148
2149
2150
2151
2152
2153
2154
2155
2156
2157
2158
2159
2160
2161
2162
2163
2164 type connectMethod struct {
2165 _ incomparable
2166 proxyURL *url.URL
2167 targetScheme string
2168
2169
2170
2171 targetAddr string
2172 onlyH1 bool
2173 }
2174
2175 func (cm *connectMethod) key() connectMethodKey {
2176 proxyStr := ""
2177 targetAddr := cm.targetAddr
2178 if cm.proxyURL != nil {
2179 proxyStr = cm.proxyURL.String()
2180 if (cm.proxyURL.Scheme == "http" || cm.proxyURL.Scheme == "https") && cm.targetScheme == "http" {
2181 targetAddr = ""
2182 }
2183 }
2184 return connectMethodKey{
2185 proxy: proxyStr,
2186 scheme: cm.targetScheme,
2187 addr: targetAddr,
2188 onlyH1: cm.onlyH1,
2189 }
2190 }
2191
2192
2193 func (cm *connectMethod) scheme() string {
2194 if cm.proxyURL != nil {
2195 return cm.proxyURL.Scheme
2196 }
2197 return cm.targetScheme
2198 }
2199
2200
2201 func (cm *connectMethod) addr() string {
2202 if cm.proxyURL != nil {
2203 return canonicalAddr(cm.proxyURL)
2204 }
2205 return cm.targetAddr
2206 }
2207
2208
2209
2210
2211 type connectMethodKey struct {
2212 proxy, scheme, addr string
2213 onlyH1 bool
2214 }
2215
2216 func (k connectMethodKey) String() string {
2217
2218 var h1 string
2219 if k.onlyH1 {
2220 h1 = ",h1"
2221 }
2222 return fmt.Sprintf("%s|%s%s|%s", k.proxy, k.scheme, h1, k.addr)
2223 }
2224
2225
2226
2227 type persistConn struct {
2228
2229
2230
2231 alt RoundTripper
2232
2233 t *Transport
2234 cacheKey connectMethodKey
2235 conn net.Conn
2236 tlsState *tls.ConnectionState
2237 br *bufio.Reader
2238 bw *bufio.Writer
2239 nwrite int64
2240 reqch chan requestAndChan
2241 writech chan writeRequest
2242 closech chan struct{}
2243 availch chan struct{}
2244 isProxy bool
2245 sawEOF bool
2246 isClientConn bool
2247 readLimit int64
2248
2249
2250
2251
2252 writeErrCh chan error
2253
2254 writeLoopDone chan struct{}
2255
2256
2257 idleAt time.Time
2258 idleTimer *time.Timer
2259
2260 mu sync.Mutex
2261 numExpectedResponses int
2262 closed error
2263 canceledErr error
2264 reused bool
2265 reserved bool
2266 inFlight bool
2267 internalStateHook func()
2268
2269
2270
2271
2272 mutateHeaderFunc func(Header)
2273 }
2274
2275 func (pc *persistConn) maxHeaderResponseSize() int64 {
2276 return pc.t.maxHeaderResponseSize()
2277 }
2278
2279 func (pc *persistConn) Read(p []byte) (n int, err error) {
2280 if pc.readLimit <= 0 {
2281 return 0, fmt.Errorf("read limit of %d bytes exhausted", pc.maxHeaderResponseSize())
2282 }
2283 if int64(len(p)) > pc.readLimit {
2284 p = p[:pc.readLimit]
2285 }
2286 n, err = pc.conn.Read(p)
2287 if err == io.EOF {
2288 pc.sawEOF = true
2289 }
2290 pc.readLimit -= int64(n)
2291 return
2292 }
2293
2294
2295 func (pc *persistConn) isBroken() bool {
2296 pc.mu.Lock()
2297 b := pc.closed != nil
2298 pc.mu.Unlock()
2299 return b
2300 }
2301
2302
2303 func (pc *persistConn) canceled() error {
2304 pc.mu.Lock()
2305 defer pc.mu.Unlock()
2306 return pc.canceledErr
2307 }
2308
2309
2310 func (pc *persistConn) isReused() bool {
2311 pc.mu.Lock()
2312 r := pc.reused
2313 pc.mu.Unlock()
2314 return r
2315 }
2316
2317 func (pc *persistConn) cancelRequest(err error) {
2318 pc.mu.Lock()
2319 defer pc.mu.Unlock()
2320 pc.canceledErr = err
2321 pc.closeLocked(errRequestCanceled)
2322 }
2323
2324
2325
2326
2327 func (pc *persistConn) closeConnIfStillIdle() {
2328 t := pc.t
2329 t.idleMu.Lock()
2330 defer t.idleMu.Unlock()
2331 if _, ok := t.idleLRU.m[pc]; !ok {
2332
2333 return
2334 }
2335 t.removeIdleConnLocked(pc)
2336 pc.close(errIdleConnTimeout)
2337 }
2338
2339
2340
2341
2342
2343
2344
2345
2346
2347 func (pc *persistConn) mapRoundTripError(req *transportRequest, startBytesWritten int64, err error) error {
2348 if err == nil {
2349 return nil
2350 }
2351
2352
2353
2354
2355
2356
2357
2358
2359 <-pc.writeLoopDone
2360
2361
2362
2363
2364 if cerr := pc.canceled(); cerr != nil {
2365 return cerr
2366 }
2367
2368
2369 req.mu.Lock()
2370 reqErr := req.err
2371 req.mu.Unlock()
2372 if reqErr != nil {
2373 return reqErr
2374 }
2375
2376 if err == errServerClosedIdle {
2377
2378 return err
2379 }
2380
2381 if _, ok := err.(transportReadFromServerError); ok {
2382 if pc.nwrite == startBytesWritten {
2383 return nothingWrittenError{err}
2384 }
2385
2386 return err
2387 }
2388 if pc.isBroken() {
2389 if pc.nwrite == startBytesWritten {
2390 return nothingWrittenError{err}
2391 }
2392 return fmt.Errorf("net/http: HTTP/1.x transport connection broken: %w", err)
2393 }
2394 return err
2395 }
2396
2397
2398
2399
2400 var errCallerOwnsConn = errors.New("read loop ending; caller owns writable underlying conn")
2401
2402
2403
2404
2405 const maxPostCloseReadBytes = 256 << 10
2406
2407
2408
2409
2410 const maxPostCloseReadTime = 50 * time.Millisecond
2411
2412 func maybeDrainBody(r io.Reader) bool {
2413 drainedCh := make(chan bool, 1)
2414 go func() {
2415
2416
2417
2418 if b, ok := r.(*body); ok {
2419 b.discardTrailer()
2420 }
2421 if _, err := io.CopyN(io.Discard, r, maxPostCloseReadBytes+1); err == io.EOF {
2422 drainedCh <- true
2423 } else {
2424 drainedCh <- false
2425 }
2426 }()
2427 select {
2428 case drained := <-drainedCh:
2429 return drained
2430 case <-time.After(maxPostCloseReadTime):
2431 return false
2432 }
2433 }
2434
2435
2436
2437 var errClosedEarly = errors.New("net/http: response body closed early")
2438
2439 func (pc *persistConn) readLoop() {
2440 closeErr := errReadLoopExiting
2441 defer func() {
2442 pc.close(closeErr)
2443 pc.t.removeIdleConn(pc)
2444 if pc.internalStateHook != nil {
2445 pc.internalStateHook()
2446 }
2447 }()
2448
2449 tryPutIdleConn := func(treq *transportRequest) bool {
2450 trace := treq.trace
2451 if err := pc.t.tryPutIdleConn(pc); err != nil {
2452 closeErr = err
2453 if trace != nil && trace.PutIdleConn != nil && err != errKeepAlivesDisabled {
2454 trace.PutIdleConn(err)
2455 }
2456 return false
2457 }
2458 if trace != nil && trace.PutIdleConn != nil {
2459 trace.PutIdleConn(nil)
2460 }
2461 return true
2462 }
2463
2464
2465
2466
2467 eofc := make(chan struct{})
2468 defer close(eofc)
2469
2470
2471 testHookMu.Lock()
2472 testHookReadLoopBeforeNextRead := testHookReadLoopBeforeNextRead
2473 testHookMu.Unlock()
2474
2475 alive := true
2476 for alive {
2477 pc.readLimit = pc.maxHeaderResponseSize()
2478 _, err := pc.br.Peek(1)
2479
2480 pc.mu.Lock()
2481 if pc.numExpectedResponses == 0 {
2482 pc.readLoopPeekFailLocked(err)
2483 pc.mu.Unlock()
2484 return
2485 }
2486 pc.mu.Unlock()
2487
2488 rc := <-pc.reqch
2489 trace := rc.treq.trace
2490
2491 var resp *Response
2492 if err == nil {
2493 resp, err = pc.readResponse(rc, trace)
2494 } else {
2495 err = transportReadFromServerError{err}
2496 closeErr = err
2497 }
2498
2499 if err != nil {
2500 if pc.readLimit <= 0 {
2501 err = fmt.Errorf("net/http: server response headers exceeded %d bytes; aborted", pc.maxHeaderResponseSize())
2502 }
2503
2504 select {
2505 case rc.ch <- responseAndError{err: err}:
2506 case <-rc.callerGone:
2507 return
2508 }
2509 return
2510 }
2511 pc.readLimit = maxInt64
2512
2513 pc.mu.Lock()
2514 pc.numExpectedResponses--
2515 pc.mu.Unlock()
2516
2517 bodyWritable := resp.bodyIsWritable()
2518 hasBody := rc.treq.Request.Method != "HEAD" && resp.ContentLength != 0
2519
2520 if resp.Close || rc.treq.Request.Close || resp.StatusCode <= 199 || bodyWritable {
2521
2522
2523
2524 alive = false
2525 }
2526
2527 if !hasBody || bodyWritable {
2528
2529
2530
2531
2532
2533 alive = alive &&
2534 !pc.sawEOF &&
2535 pc.wroteRequest() &&
2536 tryPutIdleConn(rc.treq)
2537
2538 if bodyWritable {
2539 closeErr = errCallerOwnsConn
2540 }
2541
2542 select {
2543 case rc.ch <- responseAndError{res: resp}:
2544 case <-rc.callerGone:
2545 return
2546 }
2547
2548 rc.treq.cancel(errRequestDone)
2549
2550
2551
2552
2553 testHookReadLoopBeforeNextRead()
2554 continue
2555 }
2556
2557 waitForBodyRead := make(chan error, 1)
2558 body := &bodyEOFSignal{
2559 body: resp.Body,
2560 earlyCloseFn: func() error {
2561 waitForBodyRead <- errClosedEarly
2562 <-eofc
2563 return nil
2564 },
2565 fn: func(err error) error {
2566 waitForBodyRead <- err
2567 if err == io.EOF {
2568 <-eofc
2569 } else if err != nil {
2570 if cerr := pc.canceled(); cerr != nil {
2571 return cerr
2572 }
2573 }
2574 return err
2575 },
2576 }
2577
2578 resp.Body = body
2579 if rc.addedGzip && ascii.EqualFold(resp.Header.Get("Content-Encoding"), "gzip") {
2580 resp.Body = &httpcommon.GzipReader{Body: body}
2581 resp.Header.Del("Content-Encoding")
2582 resp.Header.Del("Content-Length")
2583 resp.ContentLength = -1
2584 resp.Uncompressed = true
2585 }
2586
2587 select {
2588 case rc.ch <- responseAndError{res: resp}:
2589 case <-rc.callerGone:
2590 return
2591 }
2592
2593
2594
2595
2596 select {
2597 case err := <-waitForBodyRead:
2598 tryPutIdle := func() {
2599 alive = alive &&
2600 !pc.sawEOF &&
2601 pc.wroteRequest() &&
2602 tryPutIdleConn(rc.treq)
2603 }
2604 switch err {
2605 case io.EOF:
2606 tryPutIdle()
2607 eofc <- struct{}{}
2608 case errClosedEarly:
2609
2610
2611 tryDrain := alive && !pc.t.keepAlivesDisabled() && resp.ContentLength <= maxPostCloseReadBytes
2612 eofc <- struct{}{}
2613 if tryDrain && maybeDrainBody(body.body) {
2614 tryPutIdle()
2615 } else {
2616 alive = false
2617 }
2618 default:
2619 alive = false
2620 }
2621 case <-rc.treq.ctx.Done():
2622 alive = false
2623 pc.cancelRequest(context.Cause(rc.treq.ctx))
2624 case <-pc.closech:
2625 alive = false
2626 }
2627
2628 rc.treq.cancel(errRequestDone)
2629 testHookReadLoopBeforeNextRead()
2630 }
2631 }
2632
2633 func (pc *persistConn) readLoopPeekFailLocked(peekErr error) {
2634 if pc.closed != nil {
2635 return
2636 }
2637 if n := pc.br.Buffered(); n > 0 {
2638 buf, _ := pc.br.Peek(n)
2639 if is408Message(buf) {
2640 pc.closeLocked(errServerClosedIdle)
2641 return
2642 } else {
2643 log.Printf("Unsolicited response received on idle HTTP channel starting with %q; err=%v", buf, peekErr)
2644 }
2645 }
2646 if peekErr == io.EOF {
2647
2648 pc.closeLocked(errServerClosedIdle)
2649 } else {
2650 pc.closeLocked(fmt.Errorf("readLoopPeekFailLocked: %w", peekErr))
2651 }
2652 }
2653
2654
2655
2656
2657 func is408Message(buf []byte) bool {
2658 if len(buf) < len("HTTP/1.x 408") {
2659 return false
2660 }
2661 if string(buf[:7]) != "HTTP/1." {
2662 return false
2663 }
2664 return string(buf[8:12]) == " 408"
2665 }
2666
2667
2668
2669
2670 func (pc *persistConn) readResponse(rc requestAndChan, trace *httptrace.ClientTrace) (resp *Response, err error) {
2671 if trace != nil && trace.GotFirstResponseByte != nil {
2672 if peek, err := pc.br.Peek(1); err == nil && len(peek) == 1 {
2673 trace.GotFirstResponseByte()
2674 }
2675 }
2676
2677 continueCh := rc.continueCh
2678 for {
2679 resp, err = ReadResponse(pc.br, rc.treq.Request)
2680 if err != nil {
2681 return
2682 }
2683 resCode := resp.StatusCode
2684 if continueCh != nil && resCode == StatusContinue {
2685 if trace != nil && trace.Got100Continue != nil {
2686 trace.Got100Continue()
2687 }
2688 continueCh <- struct{}{}
2689 continueCh = nil
2690 }
2691 is1xx := 100 <= resCode && resCode <= 199
2692
2693 is1xxNonTerminal := is1xx && resCode != StatusSwitchingProtocols
2694 if is1xxNonTerminal {
2695 if trace != nil && trace.Got1xxResponse != nil {
2696 if err := trace.Got1xxResponse(resCode, textproto.MIMEHeader(resp.Header)); err != nil {
2697 return nil, err
2698 }
2699
2700
2701
2702
2703
2704
2705
2706 pc.readLimit = pc.maxHeaderResponseSize()
2707 }
2708 continue
2709 }
2710 break
2711 }
2712 if resp.isProtocolSwitch() {
2713 resp.Body = newReadWriteCloserBody(pc.br, pc.conn)
2714 }
2715 if continueCh != nil {
2716
2717
2718
2719
2720
2721
2722
2723
2724
2725
2726
2727
2728 if resp.Close || rc.treq.Request.Close {
2729 close(continueCh)
2730 } else {
2731 continueCh <- struct{}{}
2732 }
2733 }
2734
2735 resp.TLS = pc.tlsState
2736 return
2737 }
2738
2739
2740
2741
2742 func (pc *persistConn) waitForContinue(continueCh <-chan struct{}) func() bool {
2743 if continueCh == nil {
2744 return nil
2745 }
2746 return func() bool {
2747 timer := time.NewTimer(pc.t.ExpectContinueTimeout)
2748 defer timer.Stop()
2749
2750 select {
2751 case _, ok := <-continueCh:
2752 return ok
2753 case <-timer.C:
2754 return true
2755 case <-pc.closech:
2756 return false
2757 }
2758 }
2759 }
2760
2761 func newReadWriteCloserBody(br *bufio.Reader, rwc io.ReadWriteCloser) io.ReadWriteCloser {
2762 body := &readWriteCloserBody{ReadWriteCloser: rwc}
2763 if br.Buffered() != 0 {
2764 body.br = br
2765 }
2766 return body
2767 }
2768
2769
2770
2771
2772
2773
2774 type readWriteCloserBody struct {
2775 _ incomparable
2776 br *bufio.Reader
2777 io.ReadWriteCloser
2778 }
2779
2780 func (b *readWriteCloserBody) Read(p []byte) (n int, err error) {
2781 if b.br != nil {
2782 if n := b.br.Buffered(); len(p) > n {
2783 p = p[:n]
2784 }
2785 n, err = b.br.Read(p)
2786 if b.br.Buffered() == 0 {
2787 b.br = nil
2788 }
2789 return n, err
2790 }
2791 return b.ReadWriteCloser.Read(p)
2792 }
2793
2794 func (b *readWriteCloserBody) CloseWrite() error {
2795 if cw, ok := b.ReadWriteCloser.(interface{ CloseWrite() error }); ok {
2796 return cw.CloseWrite()
2797 }
2798 return fmt.Errorf("CloseWrite: %w", ErrNotSupported)
2799 }
2800
2801
2802 type nothingWrittenError struct {
2803 error
2804 }
2805
2806 func (nwe nothingWrittenError) Unwrap() error {
2807 return nwe.error
2808 }
2809
2810 func (pc *persistConn) writeLoop() {
2811 defer close(pc.writeLoopDone)
2812 for {
2813 select {
2814 case wr := <-pc.writech:
2815 startBytesWritten := pc.nwrite
2816 err := wr.req.Request.write(pc.bw, pc.isProxy, wr.req.extra, pc.waitForContinue(wr.continueCh))
2817 if bre, ok := err.(requestBodyReadError); ok {
2818 err = bre.error
2819
2820
2821
2822
2823
2824
2825
2826 wr.req.setError(err)
2827 }
2828 if err == nil {
2829 err = pc.bw.Flush()
2830 }
2831 if err != nil {
2832 if pc.nwrite == startBytesWritten {
2833 err = nothingWrittenError{err}
2834 }
2835 }
2836 pc.writeErrCh <- err
2837 wr.ch <- err
2838 if err != nil {
2839 pc.close(err)
2840 return
2841 }
2842 case <-pc.closech:
2843 return
2844 }
2845 }
2846 }
2847
2848
2849
2850
2851
2852
2853
2854 var maxWriteWaitBeforeConnReuse = 50 * time.Millisecond
2855
2856
2857
2858 func (pc *persistConn) wroteRequest() bool {
2859 select {
2860 case err := <-pc.writeErrCh:
2861
2862
2863 return err == nil
2864 default:
2865
2866
2867
2868
2869
2870
2871
2872
2873
2874
2875 t := time.NewTimer(maxWriteWaitBeforeConnReuse)
2876 defer t.Stop()
2877 select {
2878 case err := <-pc.writeErrCh:
2879 return err == nil
2880 case <-t.C:
2881 return false
2882 }
2883 }
2884 }
2885
2886
2887
2888 type responseAndError struct {
2889 _ incomparable
2890 res *Response
2891 err error
2892 }
2893
2894 type requestAndChan struct {
2895 _ incomparable
2896 treq *transportRequest
2897 ch chan responseAndError
2898
2899
2900
2901
2902 addedGzip bool
2903
2904
2905
2906
2907
2908 continueCh chan<- struct{}
2909
2910 callerGone <-chan struct{}
2911 }
2912
2913
2914
2915
2916
2917 type writeRequest struct {
2918 req *transportRequest
2919 ch chan<- error
2920
2921
2922
2923
2924 continueCh <-chan struct{}
2925 }
2926
2927
2928
2929 type timeoutError struct {
2930 err string
2931 }
2932
2933 func (e *timeoutError) Error() string { return e.err }
2934 func (e *timeoutError) Timeout() bool { return true }
2935 func (e *timeoutError) Temporary() bool { return true }
2936 func (e *timeoutError) Is(err error) bool { return err == context.DeadlineExceeded }
2937
2938 var errTimeout error = &timeoutError{"net/http: timeout awaiting response headers"}
2939
2940
2941
2942 var errRequestCanceled = internal.ErrRequestCanceled
2943 var errRequestCanceledConn = errors.New("net/http: request canceled while waiting for connection")
2944
2945
2946
2947 var errRequestDone = errors.New("net/http: request completed")
2948
2949 func nop() {}
2950
2951
2952 var (
2953 testHookEnterRoundTrip = nop
2954 testHookWaitResLoop = nop
2955 testHookRoundTripRetried = nop
2956 testHookPrePendingDial = nop
2957 testHookPostPendingDial = nop
2958
2959 testHookMu sync.Locker = fakeLocker{}
2960 testHookReadLoopBeforeNextRead = nop
2961 )
2962
2963 func (pc *persistConn) waitForAvailability(ctx context.Context) error {
2964 select {
2965 case <-pc.availch:
2966 return nil
2967 case <-pc.closech:
2968 return pc.closed
2969 case <-ctx.Done():
2970 return ctx.Err()
2971 }
2972 }
2973
2974 func (pc *persistConn) roundTrip(req *transportRequest) (resp *Response, err error) {
2975 testHookEnterRoundTrip()
2976
2977 pc.mu.Lock()
2978 if pc.isClientConn {
2979 if !pc.reserved {
2980 pc.mu.Unlock()
2981 if err := pc.waitForAvailability(req.ctx); err != nil {
2982 return nil, err
2983 }
2984 pc.mu.Lock()
2985 }
2986 pc.reserved = false
2987 pc.inFlight = true
2988 }
2989 pc.numExpectedResponses++
2990 headerFn := pc.mutateHeaderFunc
2991 pc.mu.Unlock()
2992
2993 if headerFn != nil {
2994 headerFn(req.extraHeaders())
2995 }
2996
2997
2998
2999
3000
3001 requestedGzip := false
3002 if !pc.t.DisableCompression &&
3003 req.Header.Get("Accept-Encoding") == "" &&
3004 req.Header.Get("Range") == "" &&
3005 req.Method != "HEAD" {
3006
3007
3008
3009
3010
3011
3012
3013
3014
3015
3016
3017
3018 requestedGzip = true
3019 req.extraHeaders().Set("Accept-Encoding", "gzip")
3020 }
3021
3022 var continueCh chan struct{}
3023 if req.ProtoAtLeast(1, 1) && req.Body != nil && req.expectsContinue() {
3024 continueCh = make(chan struct{}, 1)
3025 }
3026
3027 if pc.t.DisableKeepAlives &&
3028 !req.wantsClose() &&
3029 !isProtocolSwitchHeader(req.Header) {
3030 req.extraHeaders().Set("Connection", "close")
3031 }
3032
3033 gone := make(chan struct{})
3034 defer close(gone)
3035
3036 const debugRoundTrip = false
3037
3038
3039
3040
3041 startBytesWritten := pc.nwrite
3042 writeErrCh := make(chan error, 1)
3043 pc.writech <- writeRequest{req, writeErrCh, continueCh}
3044
3045 resc := make(chan responseAndError)
3046 pc.reqch <- requestAndChan{
3047 treq: req,
3048 ch: resc,
3049 addedGzip: requestedGzip,
3050 continueCh: continueCh,
3051 callerGone: gone,
3052 }
3053
3054 handleResponse := func(re responseAndError) (*Response, error) {
3055 if (re.res == nil) == (re.err == nil) {
3056 panic(fmt.Sprintf("internal error: exactly one of res or err should be set; nil=%v", re.res == nil))
3057 }
3058 if debugRoundTrip {
3059 req.logf("resc recv: %p, %T/%#v", re.res, re.err, re.err)
3060 }
3061 if re.err != nil {
3062 return nil, pc.mapRoundTripError(req, startBytesWritten, re.err)
3063 }
3064 return re.res, nil
3065 }
3066
3067 var respHeaderTimer <-chan time.Time
3068 ctxDoneChan := req.ctx.Done()
3069 pcClosed := pc.closech
3070 for {
3071 testHookWaitResLoop()
3072 select {
3073 case err := <-writeErrCh:
3074 if debugRoundTrip {
3075 req.logf("writeErrCh recv: %T/%#v", err, err)
3076 }
3077 if err != nil {
3078 pc.close(fmt.Errorf("write error: %w", err))
3079 return nil, pc.mapRoundTripError(req, startBytesWritten, err)
3080 }
3081 if d := pc.t.ResponseHeaderTimeout; d > 0 {
3082 if debugRoundTrip {
3083 req.logf("starting timer for %v", d)
3084 }
3085 timer := time.NewTimer(d)
3086 defer timer.Stop()
3087 respHeaderTimer = timer.C
3088 }
3089 case <-pcClosed:
3090 select {
3091 case re := <-resc:
3092
3093
3094
3095 return handleResponse(re)
3096 default:
3097 }
3098 if debugRoundTrip {
3099 req.logf("closech recv: %T %#v", pc.closed, pc.closed)
3100 }
3101 return nil, pc.mapRoundTripError(req, startBytesWritten, pc.closed)
3102 case <-respHeaderTimer:
3103 if debugRoundTrip {
3104 req.logf("timeout waiting for response headers.")
3105 }
3106 pc.close(errTimeout)
3107 return nil, errTimeout
3108 case re := <-resc:
3109 return handleResponse(re)
3110 case <-ctxDoneChan:
3111 select {
3112 case re := <-resc:
3113
3114
3115
3116 return handleResponse(re)
3117 default:
3118 }
3119 pc.cancelRequest(context.Cause(req.ctx))
3120 }
3121 }
3122 }
3123
3124
3125
3126 type tLogKey struct{}
3127
3128 func (tr *transportRequest) logf(format string, args ...any) {
3129 if logf, ok := tr.Request.Context().Value(tLogKey{}).(func(string, ...any)); ok {
3130 logf(time.Now().Format(time.RFC3339Nano)+": "+format, args...)
3131 }
3132 }
3133
3134
3135
3136 func (pc *persistConn) markReused() {
3137 pc.mu.Lock()
3138 pc.reused = true
3139 pc.mu.Unlock()
3140 }
3141
3142
3143
3144
3145
3146
3147 func (pc *persistConn) close(err error) {
3148 pc.mu.Lock()
3149 defer pc.mu.Unlock()
3150 pc.closeLocked(err)
3151 }
3152
3153 func (pc *persistConn) closeLocked(err error) {
3154 if err == nil {
3155 panic("nil error")
3156 }
3157 if pc.closed == nil {
3158 pc.closed = err
3159 pc.t.decConnsPerHost(pc.cacheKey)
3160
3161
3162
3163 if pc.alt == nil {
3164 if err != errCallerOwnsConn {
3165 pc.conn.Close()
3166 }
3167 close(pc.closech)
3168 } else {
3169 if cc, ok := pc.alt.(io.Closer); ok {
3170 cc.Close()
3171 }
3172 }
3173 }
3174 pc.mutateHeaderFunc = nil
3175 }
3176
3177 func schemePort(scheme string) string {
3178 switch scheme {
3179 case "http":
3180 return "80"
3181 case "https":
3182 return "443"
3183 case "socks5", "socks5h":
3184 return "1080"
3185 default:
3186 return ""
3187 }
3188 }
3189
3190 func idnaASCIIFromURL(url *url.URL) string {
3191 addr := url.Hostname()
3192 if v, err := idnaASCII(addr); err == nil && v != "" {
3193 addr = v
3194 }
3195 return addr
3196 }
3197
3198
3199 func canonicalAddr(url *url.URL) string {
3200 port := url.Port()
3201 if port == "" {
3202 port = schemePort(url.Scheme)
3203 }
3204 return net.JoinHostPort(idnaASCIIFromURL(url), port)
3205 }
3206
3207
3208
3209
3210
3211
3212
3213
3214
3215
3216
3217
3218 type bodyEOFSignal struct {
3219 body io.ReadCloser
3220 mu sync.Mutex
3221 closed bool
3222 rerr error
3223 fn func(error) error
3224 earlyCloseFn func() error
3225 }
3226
3227 var errReadOnClosedResBody = errors.New("http: read on closed response body")
3228
3229 func (es *bodyEOFSignal) Read(p []byte) (n int, err error) {
3230 es.mu.Lock()
3231 closed, rerr := es.closed, es.rerr
3232 es.mu.Unlock()
3233 if closed {
3234 return 0, errReadOnClosedResBody
3235 }
3236 if rerr != nil {
3237 return 0, rerr
3238 }
3239
3240 n, err = es.body.Read(p)
3241 if err != nil {
3242 es.mu.Lock()
3243 defer es.mu.Unlock()
3244 if es.rerr == nil {
3245 es.rerr = err
3246 }
3247 err = es.condfn(err)
3248 }
3249 return
3250 }
3251
3252 func (es *bodyEOFSignal) Close() error {
3253 es.mu.Lock()
3254 defer es.mu.Unlock()
3255 if es.closed {
3256 return nil
3257 }
3258 es.closed = true
3259 if es.earlyCloseFn != nil && es.rerr == nil {
3260 earlyCloseFn := es.earlyCloseFn
3261 es.earlyCloseFn = nil
3262 es.fn = nil
3263 return earlyCloseFn()
3264 }
3265 if es.rerr != nil && es.rerr != io.EOF {
3266
3267
3268 return nil
3269 }
3270 err := es.body.Close()
3271 return es.condfn(err)
3272 }
3273
3274
3275 func (es *bodyEOFSignal) condfn(err error) error {
3276 if es.fn == nil {
3277 return err
3278 }
3279 fn := es.fn
3280 es.fn = nil
3281 es.earlyCloseFn = nil
3282 return fn(err)
3283 }
3284
3285 type tlsHandshakeTimeoutError struct{}
3286
3287 func (tlsHandshakeTimeoutError) Timeout() bool { return true }
3288 func (tlsHandshakeTimeoutError) Temporary() bool { return true }
3289 func (tlsHandshakeTimeoutError) Error() string { return "net/http: TLS handshake timeout" }
3290
3291
3292
3293
3294 type fakeLocker struct{}
3295
3296 func (fakeLocker) Lock() {}
3297 func (fakeLocker) Unlock() {}
3298
3299
3300
3301
3302
3303
3304
3305
3306
3307
3308
3309
3310
3311
3312 func cloneTLSConfig(cfg *tls.Config) *tls.Config {
3313 if cfg == nil {
3314 return &tls.Config{}
3315 }
3316 return cfg.Clone()
3317 }
3318
3319 type connLRU struct {
3320 ll *list.List
3321 m map[*persistConn]*list.Element
3322 }
3323
3324
3325 func (cl *connLRU) add(pc *persistConn) {
3326 if cl.ll == nil {
3327 cl.ll = list.New()
3328 cl.m = make(map[*persistConn]*list.Element)
3329 }
3330 ele := cl.ll.PushFront(pc)
3331 if _, ok := cl.m[pc]; ok {
3332 panic("persistConn was already in LRU")
3333 }
3334 cl.m[pc] = ele
3335 }
3336
3337 func (cl *connLRU) removeOldest() *persistConn {
3338 ele := cl.ll.Back()
3339 pc := ele.Value.(*persistConn)
3340 cl.ll.Remove(ele)
3341 delete(cl.m, pc)
3342 return pc
3343 }
3344
3345
3346 func (cl *connLRU) remove(pc *persistConn) {
3347 if ele, ok := cl.m[pc]; ok {
3348 cl.ll.Remove(ele)
3349 delete(cl.m, pc)
3350 }
3351 }
3352
3353
3354 func (cl *connLRU) len() int {
3355 return len(cl.m)
3356 }
3357
View as plain text