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