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 omitHTTP2Client {
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 type bodyReadStatus uint8
2436
2437 const (
2438 bodyReadEOF bodyReadStatus = iota
2439 bodyReadClosedEarly
2440 bodyReadError
2441 )
2442
2443 func (pc *persistConn) readLoop() {
2444 closeErr := errReadLoopExiting
2445 defer func() {
2446 pc.close(closeErr)
2447 pc.t.removeIdleConn(pc)
2448 if pc.internalStateHook != nil {
2449 pc.internalStateHook()
2450 }
2451 }()
2452
2453 tryPutIdleConn := func(treq *transportRequest) bool {
2454 trace := treq.trace
2455 if err := pc.t.tryPutIdleConn(pc); err != nil {
2456 closeErr = err
2457 if trace != nil && trace.PutIdleConn != nil && err != errKeepAlivesDisabled {
2458 trace.PutIdleConn(err)
2459 }
2460 return false
2461 }
2462 if trace != nil && trace.PutIdleConn != nil {
2463 trace.PutIdleConn(nil)
2464 }
2465 return true
2466 }
2467
2468
2469
2470
2471 eofc := make(chan struct{})
2472 defer close(eofc)
2473
2474
2475 testHookMu.Lock()
2476 testHookReadLoopBeforeNextRead := testHookReadLoopBeforeNextRead
2477 testHookMu.Unlock()
2478
2479 alive := true
2480 for alive {
2481 pc.readLimit = pc.maxHeaderResponseSize()
2482 _, err := pc.br.Peek(1)
2483
2484 pc.mu.Lock()
2485 if pc.numExpectedResponses == 0 {
2486 pc.readLoopPeekFailLocked(err)
2487 pc.mu.Unlock()
2488 return
2489 }
2490 pc.mu.Unlock()
2491
2492 rc := <-pc.reqch
2493 trace := rc.treq.trace
2494
2495 var resp *Response
2496 if err == nil {
2497 resp, err = pc.readResponse(rc, trace)
2498 } else {
2499 err = transportReadFromServerError{err}
2500 closeErr = err
2501 }
2502
2503 if err != nil {
2504 if pc.readLimit <= 0 {
2505 err = fmt.Errorf("net/http: server response headers exceeded %d bytes; aborted", pc.maxHeaderResponseSize())
2506 }
2507
2508 select {
2509 case rc.ch <- responseAndError{err: err}:
2510 case <-rc.callerGone:
2511 return
2512 }
2513 return
2514 }
2515 pc.readLimit = maxInt64
2516
2517 pc.mu.Lock()
2518 pc.numExpectedResponses--
2519 pc.mu.Unlock()
2520
2521 bodyWritable := resp.bodyIsWritable()
2522 hasBody := rc.treq.Request.Method != "HEAD" && resp.ContentLength != 0
2523
2524 if resp.Close || rc.treq.Request.Close || resp.StatusCode <= 199 || bodyWritable {
2525
2526
2527
2528 alive = false
2529 }
2530
2531 if !hasBody || bodyWritable {
2532
2533
2534
2535
2536
2537 alive = alive &&
2538 !pc.sawEOF &&
2539 pc.wroteRequest() &&
2540 tryPutIdleConn(rc.treq)
2541
2542 if bodyWritable {
2543 closeErr = errCallerOwnsConn
2544 }
2545
2546 select {
2547 case rc.ch <- responseAndError{res: resp}:
2548 case <-rc.callerGone:
2549 return
2550 }
2551
2552 rc.treq.cancel(errRequestDone)
2553
2554
2555
2556
2557 testHookReadLoopBeforeNextRead()
2558 continue
2559 }
2560
2561 waitForBodyRead := make(chan bodyReadStatus, 1)
2562 body := &bodyEOFSignal{
2563 body: resp.Body,
2564 earlyCloseFn: func() error {
2565 waitForBodyRead <- bodyReadClosedEarly
2566 <-eofc
2567 return nil
2568 },
2569 fn: func(err error) error {
2570 if err == io.EOF {
2571 waitForBodyRead <- bodyReadEOF
2572 <-eofc
2573 } else {
2574 waitForBodyRead <- bodyReadError
2575 if err != nil {
2576 if cerr := pc.canceled(); cerr != nil {
2577 return cerr
2578 }
2579 }
2580 }
2581 return err
2582 },
2583 }
2584
2585 resp.Body = body
2586 if rc.addedGzip && ascii.EqualFold(resp.Header.Get("Content-Encoding"), "gzip") {
2587 resp.Body = &httpcommon.GzipReader{Body: body}
2588 resp.Header.Del("Content-Encoding")
2589 resp.Header.Del("Content-Length")
2590 resp.ContentLength = -1
2591 resp.Uncompressed = true
2592 }
2593
2594 select {
2595 case rc.ch <- responseAndError{res: resp}:
2596 case <-rc.callerGone:
2597 return
2598 }
2599
2600
2601
2602
2603 select {
2604 case status := <-waitForBodyRead:
2605 tryPutIdle := func() {
2606 alive = alive &&
2607 !pc.sawEOF &&
2608 pc.wroteRequest() &&
2609 tryPutIdleConn(rc.treq)
2610 }
2611 switch status {
2612 case bodyReadEOF:
2613 tryPutIdle()
2614 eofc <- struct{}{}
2615 case bodyReadClosedEarly:
2616
2617
2618 tryDrain := alive && !pc.t.keepAlivesDisabled() && resp.ContentLength <= maxPostCloseReadBytes
2619 eofc <- struct{}{}
2620 if tryDrain && maybeDrainBody(body.body) {
2621 tryPutIdle()
2622 } else {
2623 alive = false
2624 }
2625 case bodyReadError:
2626 alive = false
2627 }
2628 case <-rc.treq.ctx.Done():
2629 alive = false
2630 pc.cancelRequest(context.Cause(rc.treq.ctx))
2631 case <-pc.closech:
2632 alive = false
2633 }
2634
2635 rc.treq.cancel(errRequestDone)
2636 testHookReadLoopBeforeNextRead()
2637 }
2638 }
2639
2640 func (pc *persistConn) readLoopPeekFailLocked(peekErr error) {
2641 if pc.closed != nil {
2642 return
2643 }
2644 if n := pc.br.Buffered(); n > 0 {
2645 buf, _ := pc.br.Peek(n)
2646 if is408Message(buf) {
2647 pc.closeLocked(errServerClosedIdle)
2648 return
2649 } else {
2650 log.Printf("Unsolicited response received on idle HTTP channel starting with %q; err=%v", buf, peekErr)
2651 }
2652 }
2653 if peekErr == io.EOF {
2654
2655 pc.closeLocked(errServerClosedIdle)
2656 } else {
2657 pc.closeLocked(fmt.Errorf("readLoopPeekFailLocked: %w", peekErr))
2658 }
2659 }
2660
2661
2662
2663
2664 func is408Message(buf []byte) bool {
2665 if len(buf) < len("HTTP/1.x 408") {
2666 return false
2667 }
2668 if string(buf[:7]) != "HTTP/1." {
2669 return false
2670 }
2671 return string(buf[8:12]) == " 408"
2672 }
2673
2674
2675
2676
2677 func (pc *persistConn) readResponse(rc requestAndChan, trace *httptrace.ClientTrace) (resp *Response, err error) {
2678 if trace != nil && trace.GotFirstResponseByte != nil {
2679 if peek, err := pc.br.Peek(1); err == nil && len(peek) == 1 {
2680 trace.GotFirstResponseByte()
2681 }
2682 }
2683
2684 continueCh := rc.continueCh
2685 for {
2686 resp, err = ReadResponse(pc.br, rc.treq.Request)
2687 if err != nil {
2688 return
2689 }
2690 resCode := resp.StatusCode
2691 if continueCh != nil && resCode == StatusContinue {
2692 if trace != nil && trace.Got100Continue != nil {
2693 trace.Got100Continue()
2694 }
2695 continueCh <- struct{}{}
2696 continueCh = nil
2697 }
2698 is1xx := 100 <= resCode && resCode <= 199
2699
2700 is1xxNonTerminal := is1xx && resCode != StatusSwitchingProtocols
2701 if is1xxNonTerminal {
2702 if trace != nil && trace.Got1xxResponse != nil {
2703 if err := trace.Got1xxResponse(resCode, textproto.MIMEHeader(resp.Header)); err != nil {
2704 return nil, err
2705 }
2706
2707
2708
2709
2710
2711
2712
2713 pc.readLimit = pc.maxHeaderResponseSize()
2714 }
2715 continue
2716 }
2717 break
2718 }
2719 if resp.isProtocolSwitch() {
2720 resp.Body = newReadWriteCloserBody(pc.br, pc.conn)
2721 }
2722 if continueCh != nil {
2723
2724
2725
2726
2727
2728
2729
2730
2731
2732
2733
2734
2735 if resp.Close || rc.treq.Request.Close {
2736 close(continueCh)
2737 } else {
2738 continueCh <- struct{}{}
2739 }
2740 }
2741
2742 resp.TLS = pc.tlsState
2743 return
2744 }
2745
2746
2747
2748
2749 func (pc *persistConn) waitForContinue(continueCh <-chan struct{}) func() bool {
2750 if continueCh == nil {
2751 return nil
2752 }
2753 return func() bool {
2754 timer := time.NewTimer(pc.t.ExpectContinueTimeout)
2755 defer timer.Stop()
2756
2757 select {
2758 case _, ok := <-continueCh:
2759 return ok
2760 case <-timer.C:
2761 return true
2762 case <-pc.closech:
2763 return false
2764 }
2765 }
2766 }
2767
2768 func newReadWriteCloserBody(br *bufio.Reader, rwc io.ReadWriteCloser) io.ReadWriteCloser {
2769 body := &readWriteCloserBody{ReadWriteCloser: rwc}
2770 if br.Buffered() != 0 {
2771 body.br = br
2772 }
2773 return body
2774 }
2775
2776
2777
2778
2779
2780
2781 type readWriteCloserBody struct {
2782 _ incomparable
2783 br *bufio.Reader
2784 io.ReadWriteCloser
2785 }
2786
2787 func (b *readWriteCloserBody) Read(p []byte) (n int, err error) {
2788 if b.br != nil {
2789 if n := b.br.Buffered(); len(p) > n {
2790 p = p[:n]
2791 }
2792 n, err = b.br.Read(p)
2793 if b.br.Buffered() == 0 {
2794 b.br = nil
2795 }
2796 return n, err
2797 }
2798 return b.ReadWriteCloser.Read(p)
2799 }
2800
2801 func (b *readWriteCloserBody) CloseWrite() error {
2802 if cw, ok := b.ReadWriteCloser.(interface{ CloseWrite() error }); ok {
2803 return cw.CloseWrite()
2804 }
2805 return fmt.Errorf("CloseWrite: %w", ErrNotSupported)
2806 }
2807
2808
2809 type nothingWrittenError struct {
2810 error
2811 }
2812
2813 func (nwe nothingWrittenError) Unwrap() error {
2814 return nwe.error
2815 }
2816
2817 func (pc *persistConn) writeLoop() {
2818 defer close(pc.writeLoopDone)
2819 for {
2820 select {
2821 case wr := <-pc.writech:
2822 startBytesWritten := pc.nwrite
2823 err := wr.req.Request.write(pc.bw, pc.isProxy, wr.req.extra, pc.waitForContinue(wr.continueCh))
2824 if bre, ok := err.(requestBodyReadError); ok {
2825 err = bre.error
2826
2827
2828
2829
2830
2831
2832
2833 wr.req.setError(err)
2834 }
2835 if err == nil {
2836 err = pc.bw.Flush()
2837 }
2838 if err != nil {
2839 if pc.nwrite == startBytesWritten {
2840 err = nothingWrittenError{err}
2841 }
2842 }
2843 pc.writeErrCh <- err
2844 wr.ch <- err
2845 if err != nil {
2846 pc.close(err)
2847 return
2848 }
2849 case <-pc.closech:
2850 return
2851 }
2852 }
2853 }
2854
2855
2856
2857
2858
2859
2860
2861 var maxWriteWaitBeforeConnReuse = 50 * time.Millisecond
2862
2863
2864
2865 func (pc *persistConn) wroteRequest() bool {
2866 select {
2867 case err := <-pc.writeErrCh:
2868
2869
2870 return err == nil
2871 default:
2872
2873
2874
2875
2876
2877
2878
2879
2880
2881
2882 t := time.NewTimer(maxWriteWaitBeforeConnReuse)
2883 defer t.Stop()
2884 select {
2885 case err := <-pc.writeErrCh:
2886 return err == nil
2887 case <-t.C:
2888 return false
2889 }
2890 }
2891 }
2892
2893
2894
2895 type responseAndError struct {
2896 _ incomparable
2897 res *Response
2898 err error
2899 }
2900
2901 type requestAndChan struct {
2902 _ incomparable
2903 treq *transportRequest
2904 ch chan responseAndError
2905
2906
2907
2908
2909 addedGzip bool
2910
2911
2912
2913
2914
2915 continueCh chan<- struct{}
2916
2917 callerGone <-chan struct{}
2918 }
2919
2920
2921
2922
2923
2924 type writeRequest struct {
2925 req *transportRequest
2926 ch chan<- error
2927
2928
2929
2930
2931 continueCh <-chan struct{}
2932 }
2933
2934
2935
2936 type timeoutError struct {
2937 err string
2938 }
2939
2940 func (e *timeoutError) Error() string { return e.err }
2941 func (e *timeoutError) Timeout() bool { return true }
2942 func (e *timeoutError) Temporary() bool { return true }
2943 func (e *timeoutError) Is(err error) bool { return err == context.DeadlineExceeded }
2944
2945 var errTimeout error = &timeoutError{"net/http: timeout awaiting response headers"}
2946
2947
2948
2949 var errRequestCanceled = internal.ErrRequestCanceled
2950 var errRequestCanceledConn = errors.New("net/http: request canceled while waiting for connection")
2951
2952
2953
2954 var errRequestDone = errors.New("net/http: request completed")
2955
2956 func nop() {}
2957
2958
2959 var (
2960 testHookEnterRoundTrip = nop
2961 testHookWaitResLoop = nop
2962 testHookRoundTripRetried = nop
2963 testHookPrePendingDial = nop
2964 testHookPostPendingDial = nop
2965
2966 testHookMu sync.Locker = fakeLocker{}
2967 testHookReadLoopBeforeNextRead = nop
2968 )
2969
2970 func (pc *persistConn) waitForAvailability(ctx context.Context) error {
2971 select {
2972 case <-pc.availch:
2973 return nil
2974 case <-pc.closech:
2975 return pc.closed
2976 case <-ctx.Done():
2977 return ctx.Err()
2978 }
2979 }
2980
2981 func (pc *persistConn) roundTrip(req *transportRequest) (resp *Response, err error) {
2982 testHookEnterRoundTrip()
2983
2984 pc.mu.Lock()
2985 if pc.isClientConn {
2986 if !pc.reserved {
2987 pc.mu.Unlock()
2988 if err := pc.waitForAvailability(req.ctx); err != nil {
2989 return nil, err
2990 }
2991 pc.mu.Lock()
2992 }
2993 pc.reserved = false
2994 pc.inFlight = true
2995 }
2996 pc.numExpectedResponses++
2997 headerFn := pc.mutateHeaderFunc
2998 pc.mu.Unlock()
2999
3000 if headerFn != nil {
3001 headerFn(req.extraHeaders())
3002 }
3003
3004
3005
3006
3007
3008 requestedGzip := false
3009 if !pc.t.DisableCompression &&
3010 req.Header.Get("Accept-Encoding") == "" &&
3011 req.Header.Get("Range") == "" &&
3012 req.Method != "HEAD" {
3013
3014
3015
3016
3017
3018
3019
3020
3021
3022
3023
3024
3025 requestedGzip = true
3026 req.extraHeaders().Set("Accept-Encoding", "gzip")
3027 }
3028
3029 var continueCh chan struct{}
3030 if req.ProtoAtLeast(1, 1) && req.Body != nil && req.expectsContinue() {
3031 continueCh = make(chan struct{}, 1)
3032 }
3033
3034 if pc.t.DisableKeepAlives &&
3035 !req.wantsClose() &&
3036 !isProtocolSwitchHeader(req.Header) {
3037 req.extraHeaders().Set("Connection", "close")
3038 }
3039
3040 gone := make(chan struct{})
3041 defer close(gone)
3042
3043 const debugRoundTrip = false
3044
3045
3046
3047
3048 startBytesWritten := pc.nwrite
3049 writeErrCh := make(chan error, 1)
3050 pc.writech <- writeRequest{req, writeErrCh, continueCh}
3051
3052 resc := make(chan responseAndError)
3053 pc.reqch <- requestAndChan{
3054 treq: req,
3055 ch: resc,
3056 addedGzip: requestedGzip,
3057 continueCh: continueCh,
3058 callerGone: gone,
3059 }
3060
3061 handleResponse := func(re responseAndError) (*Response, error) {
3062 if (re.res == nil) == (re.err == nil) {
3063 panic(fmt.Sprintf("internal error: exactly one of res or err should be set; nil=%v", re.res == nil))
3064 }
3065 if debugRoundTrip {
3066 req.logf("resc recv: %p, %T/%#v", re.res, re.err, re.err)
3067 }
3068 if re.err != nil {
3069 return nil, pc.mapRoundTripError(req, startBytesWritten, re.err)
3070 }
3071 return re.res, nil
3072 }
3073
3074 var respHeaderTimer <-chan time.Time
3075 ctxDoneChan := req.ctx.Done()
3076 pcClosed := pc.closech
3077 for {
3078 testHookWaitResLoop()
3079 select {
3080 case err := <-writeErrCh:
3081 if debugRoundTrip {
3082 req.logf("writeErrCh recv: %T/%#v", err, err)
3083 }
3084 if err != nil {
3085 pc.close(fmt.Errorf("write error: %w", err))
3086 return nil, pc.mapRoundTripError(req, startBytesWritten, err)
3087 }
3088 if d := pc.t.ResponseHeaderTimeout; d > 0 {
3089 if debugRoundTrip {
3090 req.logf("starting timer for %v", d)
3091 }
3092 timer := time.NewTimer(d)
3093 defer timer.Stop()
3094 respHeaderTimer = timer.C
3095 }
3096 case <-pcClosed:
3097 select {
3098 case re := <-resc:
3099
3100
3101
3102 return handleResponse(re)
3103 default:
3104 }
3105 if debugRoundTrip {
3106 req.logf("closech recv: %T %#v", pc.closed, pc.closed)
3107 }
3108 return nil, pc.mapRoundTripError(req, startBytesWritten, pc.closed)
3109 case <-respHeaderTimer:
3110 if debugRoundTrip {
3111 req.logf("timeout waiting for response headers.")
3112 }
3113 pc.close(errTimeout)
3114 return nil, errTimeout
3115 case re := <-resc:
3116 return handleResponse(re)
3117 case <-ctxDoneChan:
3118 select {
3119 case re := <-resc:
3120
3121
3122
3123 return handleResponse(re)
3124 default:
3125 }
3126 pc.cancelRequest(context.Cause(req.ctx))
3127 }
3128 }
3129 }
3130
3131
3132
3133 type tLogKey struct{}
3134
3135 func (tr *transportRequest) logf(format string, args ...any) {
3136 if logf, ok := tr.Request.Context().Value(tLogKey{}).(func(string, ...any)); ok {
3137 logf(time.Now().Format(time.RFC3339Nano)+": "+format, args...)
3138 }
3139 }
3140
3141
3142
3143 func (pc *persistConn) markReused() {
3144 pc.mu.Lock()
3145 pc.reused = true
3146 pc.mu.Unlock()
3147 }
3148
3149
3150
3151
3152
3153
3154 func (pc *persistConn) close(err error) {
3155 pc.mu.Lock()
3156 defer pc.mu.Unlock()
3157 pc.closeLocked(err)
3158 }
3159
3160 func (pc *persistConn) closeLocked(err error) {
3161 if err == nil {
3162 panic("nil error")
3163 }
3164 if pc.closed == nil {
3165 pc.closed = err
3166 pc.t.decConnsPerHost(pc.cacheKey)
3167
3168
3169
3170 if pc.alt == nil {
3171 if err != errCallerOwnsConn {
3172 pc.conn.Close()
3173 }
3174 close(pc.closech)
3175 } else {
3176 if cc, ok := pc.alt.(io.Closer); ok {
3177 cc.Close()
3178 }
3179 }
3180 }
3181 pc.mutateHeaderFunc = nil
3182 }
3183
3184 func schemePort(scheme string) string {
3185 switch scheme {
3186 case "http":
3187 return "80"
3188 case "https":
3189 return "443"
3190 case "socks5", "socks5h":
3191 return "1080"
3192 default:
3193 return ""
3194 }
3195 }
3196
3197 func idnaASCIIFromURL(url *url.URL) string {
3198 addr := url.Hostname()
3199 if v, err := idnaASCII(addr); err == nil && v != "" {
3200 addr = v
3201 }
3202 return addr
3203 }
3204
3205
3206 func canonicalAddr(url *url.URL) string {
3207 port := url.Port()
3208 if port == "" {
3209 port = schemePort(url.Scheme)
3210 }
3211 return net.JoinHostPort(idnaASCIIFromURL(url), port)
3212 }
3213
3214
3215
3216
3217
3218
3219
3220
3221
3222
3223
3224
3225 type bodyEOFSignal struct {
3226 body io.ReadCloser
3227 mu sync.Mutex
3228 closed bool
3229 rerr error
3230 fn func(error) error
3231 earlyCloseFn func() error
3232 }
3233
3234 var errReadOnClosedResBody = errors.New("http: read on closed response body")
3235
3236 func (es *bodyEOFSignal) Read(p []byte) (n int, err error) {
3237 es.mu.Lock()
3238 closed, rerr := es.closed, es.rerr
3239 es.mu.Unlock()
3240 if closed {
3241 return 0, errReadOnClosedResBody
3242 }
3243 if rerr != nil {
3244 return 0, rerr
3245 }
3246
3247 n, err = es.body.Read(p)
3248 if err != nil {
3249 es.mu.Lock()
3250 defer es.mu.Unlock()
3251 if es.rerr == nil {
3252 es.rerr = err
3253 }
3254 err = es.condfn(err)
3255 }
3256 return
3257 }
3258
3259 func (es *bodyEOFSignal) Close() error {
3260 es.mu.Lock()
3261 defer es.mu.Unlock()
3262 if es.closed {
3263 return nil
3264 }
3265 es.closed = true
3266 if es.earlyCloseFn != nil && es.rerr == nil {
3267 earlyCloseFn := es.earlyCloseFn
3268 es.earlyCloseFn = nil
3269 es.fn = nil
3270 return earlyCloseFn()
3271 }
3272 if es.rerr != nil && es.rerr != io.EOF {
3273
3274
3275 return nil
3276 }
3277 err := es.body.Close()
3278 return es.condfn(err)
3279 }
3280
3281
3282 func (es *bodyEOFSignal) condfn(err error) error {
3283 if es.fn == nil {
3284 return err
3285 }
3286 fn := es.fn
3287 es.fn = nil
3288 es.earlyCloseFn = nil
3289 return fn(err)
3290 }
3291
3292 type tlsHandshakeTimeoutError struct{}
3293
3294 func (tlsHandshakeTimeoutError) Timeout() bool { return true }
3295 func (tlsHandshakeTimeoutError) Temporary() bool { return true }
3296 func (tlsHandshakeTimeoutError) Error() string { return "net/http: TLS handshake timeout" }
3297
3298
3299
3300
3301 type fakeLocker struct{}
3302
3303 func (fakeLocker) Lock() {}
3304 func (fakeLocker) Unlock() {}
3305
3306
3307
3308
3309
3310
3311
3312
3313
3314
3315
3316
3317
3318
3319 func cloneTLSConfig(cfg *tls.Config) *tls.Config {
3320 if cfg == nil {
3321 return &tls.Config{}
3322 }
3323 return cfg.Clone()
3324 }
3325
3326 type connLRU struct {
3327 ll *list.List
3328 m map[*persistConn]*list.Element
3329 }
3330
3331
3332 func (cl *connLRU) add(pc *persistConn) {
3333 if cl.ll == nil {
3334 cl.ll = list.New()
3335 cl.m = make(map[*persistConn]*list.Element)
3336 }
3337 ele := cl.ll.PushFront(pc)
3338 if _, ok := cl.m[pc]; ok {
3339 panic("persistConn was already in LRU")
3340 }
3341 cl.m[pc] = ele
3342 }
3343
3344 func (cl *connLRU) removeOldest() *persistConn {
3345 ele := cl.ll.Back()
3346 pc := ele.Value.(*persistConn)
3347 cl.ll.Remove(ele)
3348 delete(cl.m, pc)
3349 return pc
3350 }
3351
3352
3353 func (cl *connLRU) remove(pc *persistConn) {
3354 if ele, ok := cl.m[pc]; ok {
3355 cl.ll.Remove(ele)
3356 delete(cl.m, pc)
3357 }
3358 }
3359
3360
3361 func (cl *connLRU) len() int {
3362 return len(cl.m)
3363 }
3364
View as plain text