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