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