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