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 "compress/flate"
15 "compress/gzip"
16 "container/list"
17 "context"
18 "crypto/tls"
19 "errors"
20 "fmt"
21 "internal/godebug"
22 "io"
23 "log"
24 "maps"
25 "net"
26 "net/http/httptrace"
27 "net/http/internal"
28 "net/http/internal/ascii"
29 "net/textproto"
30 "net/url"
31 "reflect"
32 "strings"
33 "sync"
34 "sync/atomic"
35 "time"
36 _ "unsafe"
37
38 "golang.org/x/net/http/httpguts"
39 "golang.org/x/net/http/httpproxy"
40 )
41
42
43
44
45
46
47
48 var DefaultTransport RoundTripper = &Transport{
49 Proxy: ProxyFromEnvironment,
50 DialContext: defaultTransportDialContext(&net.Dialer{
51 Timeout: 30 * time.Second,
52 KeepAlive: 30 * time.Second,
53 }),
54 ForceAttemptHTTP2: true,
55 MaxIdleConns: 100,
56 IdleConnTimeout: 90 * time.Second,
57 TLSHandshakeTimeout: 10 * time.Second,
58 ExpectContinueTimeout: 1 * time.Second,
59 }
60
61
62
63 const DefaultMaxIdleConnsPerHost = 2
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99 type Transport struct {
100 idleMu sync.Mutex
101 closeIdle bool
102 idleConn map[connectMethodKey][]*persistConn
103 idleConnWait map[connectMethodKey]wantConnQueue
104 idleLRU connLRU
105
106 altMu sync.Mutex
107 altProto atomic.Value
108
109 connsPerHostMu sync.Mutex
110 connsPerHost map[connectMethodKey]int
111 connsPerHostWait map[connectMethodKey]wantConnQueue
112 dialsInProgress wantConnQueue
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128 Proxy func(*Request) (*url.URL, error)
129
130
131
132
133 OnProxyConnectResponse func(ctx context.Context, proxyURL *url.URL, connectReq *Request, connectRes *Response) error
134
135
136
137
138
139
140
141
142
143 DialContext func(ctx context.Context, network, addr string) (net.Conn, error)
144
145
146
147
148
149
150
151
152
153
154
155 Dial func(network, addr string) (net.Conn, error)
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170 DialTLSContext func(ctx context.Context, network, addr string) (net.Conn, error)
171
172
173
174
175
176
177
178 DialTLS func(network, addr string) (net.Conn, error)
179
180
181
182
183
184 TLSClientConfig *tls.Config
185
186
187
188 TLSHandshakeTimeout time.Duration
189
190
191
192
193
194
195 DisableKeepAlives bool
196
197
198
199
200
201
202
203
204
205 DisableCompression bool
206
207
208
209 MaxIdleConns int
210
211
212
213
214 MaxIdleConnsPerHost int
215
216
217
218
219
220
221 MaxConnsPerHost int
222
223
224
225
226
227 IdleConnTimeout time.Duration
228
229
230
231
232
233 ResponseHeaderTimeout time.Duration
234
235
236
237
238
239
240
241
242 ExpectContinueTimeout time.Duration
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257 TLSNextProto map[string]func(authority string, c *tls.Conn) RoundTripper
258
259
260
261
262 ProxyConnectHeader Header
263
264
265
266
267
268
269
270
271 GetProxyConnectHeader func(ctx context.Context, proxyURL *url.URL, target string) (Header, error)
272
273
274
275
276
277
278 MaxResponseHeaderBytes int64
279
280
281
282
283 WriteBufferSize int
284
285
286
287
288 ReadBufferSize int
289
290
291
292 nextProtoOnce sync.Once
293 closeIdleFunc closeIdleConnectionser
294 h2Transport *http2Transport
295 h2Config http2ExternalTransportConfig
296 h3Transport dialClientConner
297 tlsNextProtoWasNil bool
298
299
300
301
302
303
304 ForceAttemptHTTP2 bool
305
306
307 HTTP2 *HTTP2Config
308
309
310
311
312
313
314
315
316
317 Protocols *Protocols
318 }
319
320 func (t *Transport) writeBufferSize() int {
321 if t.WriteBufferSize > 0 {
322 return t.WriteBufferSize
323 }
324 return 4 << 10
325 }
326
327 func (t *Transport) readBufferSize() int {
328 if t.ReadBufferSize > 0 {
329 return t.ReadBufferSize
330 }
331 return 4 << 10
332 }
333
334 func (t *Transport) maxHeaderResponseSize() int64 {
335 if t.MaxResponseHeaderBytes > 0 {
336 return t.MaxResponseHeaderBytes
337 }
338 return 10 << 20
339 }
340
341
342 func (t *Transport) Clone() *Transport {
343 t.nextProtoOnce.Do(t.onceSetNextProtoDefaults)
344 t2 := &Transport{
345 Proxy: t.Proxy,
346 OnProxyConnectResponse: t.OnProxyConnectResponse,
347 DialContext: t.DialContext,
348 Dial: t.Dial,
349 DialTLS: t.DialTLS,
350 DialTLSContext: t.DialTLSContext,
351 TLSHandshakeTimeout: t.TLSHandshakeTimeout,
352 DisableKeepAlives: t.DisableKeepAlives,
353 DisableCompression: t.DisableCompression,
354 MaxIdleConns: t.MaxIdleConns,
355 MaxIdleConnsPerHost: t.MaxIdleConnsPerHost,
356 MaxConnsPerHost: t.MaxConnsPerHost,
357 IdleConnTimeout: t.IdleConnTimeout,
358 ResponseHeaderTimeout: t.ResponseHeaderTimeout,
359 ExpectContinueTimeout: t.ExpectContinueTimeout,
360 ProxyConnectHeader: t.ProxyConnectHeader.Clone(),
361 GetProxyConnectHeader: t.GetProxyConnectHeader,
362 MaxResponseHeaderBytes: t.MaxResponseHeaderBytes,
363 ForceAttemptHTTP2: t.ForceAttemptHTTP2,
364 WriteBufferSize: t.WriteBufferSize,
365 ReadBufferSize: t.ReadBufferSize,
366 }
367 if t.TLSClientConfig != nil {
368 t2.TLSClientConfig = t.TLSClientConfig.Clone()
369 }
370 if t.HTTP2 != nil {
371 t2.HTTP2 = &HTTP2Config{}
372 *t2.HTTP2 = *t.HTTP2
373 }
374 if t.Protocols != nil {
375 t2.Protocols = &Protocols{}
376 *t2.Protocols = *t.Protocols
377 }
378 if !t.tlsNextProtoWasNil {
379 npm := maps.Clone(t.TLSNextProto)
380 if npm == nil {
381 npm = make(map[string]func(authority string, c *tls.Conn) RoundTripper)
382 }
383 t2.TLSNextProto = npm
384 }
385 return t2
386 }
387
388 type dialClientConner interface {
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414 DialClientConn(ctx context.Context, address string, proxy *url.URL, tlsConfig *tls.Config, internalStateHook func()) (RoundTripper, error)
415 }
416
417 type closeIdleConnectionser interface {
418
419
420
421
422
423
424
425
426 CloseIdleConnections()
427 }
428
429 func (t *Transport) hasCustomTLSDialer() bool {
430 return t.DialTLS != nil || t.DialTLSContext != nil
431 }
432
433 var http2client = godebug.New("http2client")
434
435
436
437 func (t *Transport) onceSetNextProtoDefaults() {
438 t.tlsNextProtoWasNil = (t.TLSNextProto == nil)
439 if http2client.Value() == "0" {
440 http2client.IncNonDefault()
441 return
442 }
443
444
445
446
447
448
449 altProto, _ := t.altProto.Load().(map[string]RoundTripper)
450 if rv := reflect.ValueOf(altProto["https"]); rv.IsValid() && rv.Type().Kind() == reflect.Struct && rv.Type().NumField() == 1 {
451 if v := rv.Field(0); v.CanInterface() {
452 if h2i, ok := v.Interface().(closeIdleConnectionser); ok {
453 t.closeIdleFunc = h2i
454 return
455 }
456 }
457 }
458
459 if _, ok := t.TLSNextProto["h2"]; ok {
460
461 return
462 }
463 protocols := t.protocols()
464 if !protocols.HTTP2() && !protocols.UnencryptedHTTP2() {
465 return
466 }
467 if omitBundledHTTP2 {
468 return
469 }
470
471 t.configureHTTP2(protocols)
472 }
473
474 func (t *Transport) protocols() Protocols {
475 if t.Protocols != nil {
476 return *t.Protocols
477 }
478 var p Protocols
479 p.SetHTTP1(true)
480 switch {
481 case t.TLSNextProto != nil:
482
483
484 if t.TLSNextProto["h2"] != nil {
485 p.SetHTTP2(true)
486 }
487 case !t.ForceAttemptHTTP2 && (t.TLSClientConfig != nil || t.Dial != nil || t.DialContext != nil || t.hasCustomTLSDialer()):
488
489
490
491
492
493
494 case http2client.Value() == "0":
495 default:
496 p.SetHTTP2(true)
497 }
498 return p
499 }
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518 func ProxyFromEnvironment(req *Request) (*url.URL, error) {
519 return envProxyFunc()(req.URL)
520 }
521
522
523
524 func ProxyURL(fixedURL *url.URL) func(*Request) (*url.URL, error) {
525 return func(*Request) (*url.URL, error) {
526 return fixedURL, nil
527 }
528 }
529
530
531
532
533 type transportRequest struct {
534 *Request
535 extra Header
536 trace *httptrace.ClientTrace
537
538 ctx context.Context
539 cancel context.CancelCauseFunc
540
541 mu sync.Mutex
542 err error
543 }
544
545 func (tr *transportRequest) extraHeaders() Header {
546 if tr.extra == nil {
547 tr.extra = make(Header)
548 }
549 return tr.extra
550 }
551
552 func (tr *transportRequest) setError(err error) {
553 tr.mu.Lock()
554 if tr.err == nil {
555 tr.err = err
556 }
557 tr.mu.Unlock()
558 }
559
560
561
562 func (t *Transport) useRegisteredProtocol(req *Request) bool {
563 if req.URL.Scheme == "https" && req.requiresHTTP1() {
564
565
566
567
568 return false
569 }
570 return true
571 }
572
573
574
575
576 func (t *Transport) alternateRoundTripper(req *Request) RoundTripper {
577 if !t.useRegisteredProtocol(req) {
578 return nil
579 }
580 if req.URL.Scheme == "https" && t.h2Config != nil && t.h2Config.ExternalRoundTrip() {
581
582
583
584
585
586
587 return t.h2Config
588 }
589 altProto, _ := t.altProto.Load().(map[string]RoundTripper)
590 return altProto[req.URL.Scheme]
591 }
592
593 func validateHeaders(hdrs Header) string {
594 for k, vv := range hdrs {
595 if !httpguts.ValidHeaderFieldName(k) {
596 return fmt.Sprintf("field name %q", k)
597 }
598 for _, v := range vv {
599 if !httpguts.ValidHeaderFieldValue(v) {
600
601
602 return fmt.Sprintf("field value for %q", k)
603 }
604 }
605 }
606 return ""
607 }
608
609
610 func (t *Transport) roundTrip(req *Request) (_ *Response, err error) {
611 t.nextProtoOnce.Do(t.onceSetNextProtoDefaults)
612 ctx := req.Context()
613 trace := httptrace.ContextClientTrace(ctx)
614
615 if req.URL == nil {
616 req.closeBody()
617 return nil, errors.New("http: nil Request.URL")
618 }
619 if req.Header == nil {
620 req.closeBody()
621 return nil, errors.New("http: nil Request.Header")
622 }
623 scheme := req.URL.Scheme
624 isHTTP := scheme == "http" || scheme == "https"
625 if isHTTP {
626
627 if err := validateHeaders(req.Header); err != "" {
628 req.closeBody()
629 return nil, fmt.Errorf("net/http: invalid header %s", err)
630 }
631
632
633 if err := validateHeaders(req.Trailer); err != "" {
634 req.closeBody()
635 return nil, fmt.Errorf("net/http: invalid trailer %s", err)
636 }
637 }
638
639 origReq := req
640 req = setupRewindBody(req)
641
642 if altRT := t.alternateRoundTripper(req); altRT != nil {
643 if resp, err := altRT.RoundTrip(req); err != ErrSkipAltProtocol {
644 return resp, err
645 }
646 var err error
647 req, err = rewindBody(req)
648 if err != nil {
649 return nil, err
650 }
651 }
652 if !isHTTP {
653 req.closeBody()
654 return nil, badStringError("unsupported protocol scheme", scheme)
655 }
656 if req.Method != "" && !validMethod(req.Method) {
657 req.closeBody()
658 return nil, fmt.Errorf("net/http: invalid method %q", req.Method)
659 }
660 if req.URL.Host == "" {
661 req.closeBody()
662 return nil, errors.New("http: no Host in request URL")
663 }
664
665
666
667
668
669
670
671
672
673
674 ctx, cancel := context.WithCancelCause(req.Context())
675
676
677 if origReq.Cancel != nil {
678 go awaitLegacyCancel(ctx, cancel, origReq)
679 }
680
681 defer func() {
682 if err != nil {
683 cancel(err)
684 }
685 }()
686
687 for {
688 select {
689 case <-ctx.Done():
690 req.closeBody()
691 return nil, context.Cause(ctx)
692 default:
693 }
694
695
696 treq := &transportRequest{Request: req, trace: trace, ctx: ctx, cancel: cancel}
697 cm, err := t.connectMethodForRequest(treq)
698 if err != nil {
699 req.closeBody()
700 return nil, err
701 }
702
703
704
705
706
707 pconn, err := t.getConn(treq, cm)
708 if err != nil {
709 req.closeBody()
710 return nil, err
711 }
712
713 var resp *Response
714 if pconn.alt != nil {
715
716 resp, err = pconn.alt.RoundTrip(req)
717 } else {
718 resp, err = pconn.roundTrip(treq)
719 }
720 if err == nil {
721 if pconn.alt != nil {
722
723
724
725
726 cancel(errRequestDone)
727 }
728 resp.Request = origReq
729 return resp, nil
730 }
731
732
733 if http2isNoCachedConnError(err) {
734 if t.removeIdleConn(pconn) {
735 t.decConnsPerHost(pconn.cacheKey)
736 }
737 } else if !pconn.shouldRetryRequest(req, err) {
738
739
740 if e, ok := err.(nothingWrittenError); ok {
741 err = e.error
742 }
743 if e, ok := err.(transportReadFromServerError); ok {
744 err = e.err
745 }
746 if b, ok := req.Body.(*readTrackingBody); ok && !b.didClose.Load() {
747
748
749
750 req.closeBody()
751 }
752 return nil, err
753 }
754 testHookRoundTripRetried()
755
756
757 req, err = rewindBody(req)
758 if err != nil {
759 return nil, err
760 }
761 }
762 }
763
764 func http2isNoCachedConnError(err error) bool {
765 _, ok := err.(interface{ IsHTTP2NoCachedConnError() })
766 return ok
767 }
768
769 func awaitLegacyCancel(ctx context.Context, cancel context.CancelCauseFunc, req *Request) {
770 select {
771 case <-req.Cancel:
772 cancel(errRequestCanceled)
773 case <-ctx.Done():
774 }
775 }
776
777 var errCannotRewind = errors.New("net/http: cannot rewind body after connection loss")
778
779 type readTrackingBody struct {
780 io.ReadCloser
781 didRead bool
782 didClose atomic.Bool
783 }
784
785 func (r *readTrackingBody) Read(data []byte) (int, error) {
786 r.didRead = true
787 return r.ReadCloser.Read(data)
788 }
789
790 func (r *readTrackingBody) Close() error {
791 if !r.didClose.CompareAndSwap(false, true) {
792 return nil
793 }
794 return r.ReadCloser.Close()
795 }
796
797
798
799
800
801 func setupRewindBody(req *Request) *Request {
802 if req.Body == nil || req.Body == NoBody {
803 return req
804 }
805 newReq := *req
806 newReq.Body = &readTrackingBody{ReadCloser: req.Body}
807 return &newReq
808 }
809
810
811
812
813
814 func rewindBody(req *Request) (rewound *Request, err error) {
815 if req.Body == nil || req.Body == NoBody || (!req.Body.(*readTrackingBody).didRead && !req.Body.(*readTrackingBody).didClose.Load()) {
816 return req, nil
817 }
818 if !req.Body.(*readTrackingBody).didClose.Load() {
819 req.closeBody()
820 }
821 if req.GetBody == nil {
822 return nil, errCannotRewind
823 }
824 body, err := req.GetBody()
825 if err != nil {
826 return nil, err
827 }
828 newReq := *req
829 newReq.Body = &readTrackingBody{ReadCloser: body}
830 return &newReq, nil
831 }
832
833
834
835
836 func (pc *persistConn) shouldRetryRequest(req *Request, err error) bool {
837 if http2isNoCachedConnError(err) {
838
839
840
841
842
843
844 return true
845 }
846 if err == errMissingHost {
847
848 return false
849 }
850 if !pc.isReused() {
851
852
853
854
855
856
857
858 return false
859 }
860 if _, ok := err.(nothingWrittenError); ok {
861
862
863 return req.outgoingLength() == 0 || req.GetBody != nil
864 }
865 if !req.isReplayable() {
866
867 return false
868 }
869 if _, ok := err.(transportReadFromServerError); ok {
870
871
872 return true
873 }
874 if err == errServerClosedIdle {
875
876
877
878 return true
879 }
880 return false
881 }
882
883
884 var ErrSkipAltProtocol = internal.ErrSkipAltProtocol
885
886
887
888
889
890
891
892
893
894
895
896 func (t *Transport) RegisterProtocol(scheme string, rt RoundTripper) {
897 if err := t.registerProtocol(scheme, rt); err != nil {
898 panic(err)
899 }
900 }
901
902 func (t *Transport) registerProtocol(scheme string, rt RoundTripper) error {
903 t.altMu.Lock()
904 defer t.altMu.Unlock()
905
906 if scheme == "http/2" {
907 if t.h2Config != nil {
908 panic("http: HTTP/2 Transport already registered")
909 }
910 var ok bool
911 if t.h2Config, ok = rt.(http2ExternalTransportConfig); !ok {
912 panic("http: HTTP/2 configuration does not implement ExternalTransportConfig")
913 }
914 t.h2Config.Registered(t)
915 }
916
917 if scheme == "http/3" {
918 if t.h3Transport != nil {
919 panic("http: HTTP/3 Transport already registered")
920 }
921 var ok bool
922 if t.h3Transport, ok = rt.(dialClientConner); !ok {
923 panic("http: HTTP/3 RoundTripper does not implement DialClientConn")
924 }
925
926
927 if r, ok := rt.(interface {
928 Registered(*Transport)
929 }); ok {
930 r.Registered(t)
931 }
932 return nil
933 }
934
935 oldMap, _ := t.altProto.Load().(map[string]RoundTripper)
936 if _, exists := oldMap[scheme]; exists {
937 return errors.New("protocol " + scheme + " already registered")
938 }
939 newMap := maps.Clone(oldMap)
940 if newMap == nil {
941 newMap = make(map[string]RoundTripper)
942 }
943 newMap[scheme] = rt
944 t.altProto.Store(newMap)
945 return nil
946 }
947
948
949
950
951
952 func (t *Transport) CloseIdleConnections() {
953 t.nextProtoOnce.Do(t.onceSetNextProtoDefaults)
954 t.idleMu.Lock()
955 m := t.idleConn
956 t.idleConn = nil
957 t.closeIdle = true
958 t.idleLRU = connLRU{}
959 t.idleMu.Unlock()
960 for _, conns := range m {
961 for _, pconn := range conns {
962 pconn.close(errCloseIdleConns)
963 }
964 }
965 t.connsPerHostMu.Lock()
966 t.dialsInProgress.all(func(w *wantConn) {
967 if w.cancelCtx != nil && !w.waiting() {
968 w.cancelCtx()
969 }
970 })
971 t.connsPerHostMu.Unlock()
972
973
974
975
976 if tr2 := t.h2Transport; tr2 != nil {
977 tr2.CloseIdleConnections()
978 }
979
980
981
982
983 if t2 := t.closeIdleFunc; t2 != nil {
984 t2.CloseIdleConnections()
985 }
986
987 if cc, ok := t.h3Transport.(closeIdleConnectionser); ok {
988 cc.CloseIdleConnections()
989 }
990 }
991
992
993
994
995
996 func (t *Transport) CancelRequest(req *Request) {
997 }
998
999
1000
1001
1002
1003 var (
1004 envProxyOnce sync.Once
1005 envProxyFuncValue func(*url.URL) (*url.URL, error)
1006 )
1007
1008
1009
1010 func envProxyFunc() func(*url.URL) (*url.URL, error) {
1011 envProxyOnce.Do(func() {
1012 envProxyFuncValue = httpproxy.FromEnvironment().ProxyFunc()
1013 })
1014 return envProxyFuncValue
1015 }
1016
1017
1018 func resetProxyConfig() {
1019 envProxyOnce = sync.Once{}
1020 envProxyFuncValue = nil
1021 }
1022
1023 func (t *Transport) connectMethodForRequest(treq *transportRequest) (cm connectMethod, err error) {
1024 cm.targetScheme = treq.URL.Scheme
1025 cm.targetAddr = canonicalAddr(treq.URL)
1026 if t.Proxy != nil {
1027 cm.proxyURL, err = t.Proxy(treq.Request)
1028 }
1029 cm.onlyH1 = treq.requiresHTTP1()
1030 return cm, err
1031 }
1032
1033
1034
1035 func (cm *connectMethod) proxyAuth() string {
1036 if cm.proxyURL == nil {
1037 return ""
1038 }
1039 if u := cm.proxyURL.User; u != nil {
1040 username := u.Username()
1041 password, _ := u.Password()
1042 return "Basic " + basicAuth(username, password)
1043 }
1044 return ""
1045 }
1046
1047
1048 var (
1049 errKeepAlivesDisabled = errors.New("http: putIdleConn: keep alives disabled")
1050 errConnBroken = errors.New("http: putIdleConn: connection is in bad state")
1051 errCloseIdle = errors.New("http: putIdleConn: CloseIdleConnections was called")
1052 errTooManyIdle = errors.New("http: putIdleConn: too many idle connections")
1053 errTooManyIdleHost = errors.New("http: putIdleConn: too many idle connections for host")
1054 errCloseIdleConns = errors.New("http: CloseIdleConnections called")
1055 errReadLoopExiting = errors.New("http: persistConn.readLoop exiting")
1056 errIdleConnTimeout = errors.New("http: idle connection timeout")
1057
1058
1059
1060
1061
1062 errServerClosedIdle = errors.New("http: server closed idle connection")
1063 )
1064
1065
1066
1067
1068
1069
1070
1071
1072
1073 type transportReadFromServerError struct {
1074 err error
1075 }
1076
1077 func (e transportReadFromServerError) Unwrap() error { return e.err }
1078
1079 func (e transportReadFromServerError) Error() string {
1080 return fmt.Sprintf("net/http: Transport failed to read from server: %v", e.err)
1081 }
1082
1083 func (t *Transport) putOrCloseIdleConn(pconn *persistConn) {
1084 if err := t.tryPutIdleConn(pconn); err != nil {
1085 pconn.close(err)
1086 }
1087 }
1088
1089 func (t *Transport) maxIdleConnsPerHost() int {
1090 if v := t.MaxIdleConnsPerHost; v != 0 {
1091 return v
1092 }
1093 return DefaultMaxIdleConnsPerHost
1094 }
1095
1096
1097
1098
1099
1100
1101 func (t *Transport) tryPutIdleConn(pconn *persistConn) error {
1102 if t.DisableKeepAlives || t.MaxIdleConnsPerHost < 0 {
1103 return errKeepAlivesDisabled
1104 }
1105 if pconn.isBroken() {
1106 return errConnBroken
1107 }
1108 pconn.markReused()
1109 if pconn.isClientConn {
1110
1111 defer pconn.internalStateHook()
1112 pconn.mu.Lock()
1113 defer pconn.mu.Unlock()
1114 if !pconn.inFlight {
1115 panic("pconn is not in flight")
1116 }
1117 pconn.inFlight = false
1118 select {
1119 case pconn.availch <- struct{}{}:
1120 default:
1121 panic("unable to make pconn available")
1122 }
1123 return nil
1124 }
1125
1126 t.idleMu.Lock()
1127 defer t.idleMu.Unlock()
1128
1129
1130
1131
1132 if pconn.alt != nil && t.idleLRU.m[pconn] != nil {
1133 return nil
1134 }
1135
1136
1137
1138
1139
1140 key := pconn.cacheKey
1141 if q, ok := t.idleConnWait[key]; ok {
1142 done := false
1143 if pconn.alt == nil {
1144
1145
1146 for q.len() > 0 {
1147 w := q.popFront()
1148 if w.tryDeliver(pconn, nil, time.Time{}) {
1149 done = true
1150 break
1151 }
1152 }
1153 } else {
1154
1155
1156
1157
1158 for q.len() > 0 {
1159 w := q.popFront()
1160 w.tryDeliver(pconn, nil, time.Time{})
1161 }
1162 }
1163 if q.len() == 0 {
1164 delete(t.idleConnWait, key)
1165 } else {
1166 t.idleConnWait[key] = q
1167 }
1168 if done {
1169 return nil
1170 }
1171 }
1172
1173 if t.closeIdle {
1174 return errCloseIdle
1175 }
1176 if t.idleConn == nil {
1177 t.idleConn = make(map[connectMethodKey][]*persistConn)
1178 }
1179 idles := t.idleConn[key]
1180 if len(idles) >= t.maxIdleConnsPerHost() {
1181 return errTooManyIdleHost
1182 }
1183 for _, exist := range idles {
1184 if exist == pconn {
1185 log.Fatalf("dup idle pconn %p in freelist", pconn)
1186 }
1187 }
1188 t.idleConn[key] = append(idles, pconn)
1189 t.idleLRU.add(pconn)
1190 if t.MaxIdleConns != 0 && t.idleLRU.len() > t.MaxIdleConns {
1191 oldest := t.idleLRU.removeOldest()
1192 oldest.close(errTooManyIdle)
1193 t.removeIdleConnLocked(oldest)
1194 }
1195
1196
1197
1198
1199 if t.IdleConnTimeout > 0 && pconn.alt == nil {
1200 if pconn.idleTimer != nil {
1201 pconn.idleTimer.Reset(t.IdleConnTimeout)
1202 } else {
1203 pconn.idleTimer = time.AfterFunc(t.IdleConnTimeout, pconn.closeConnIfStillIdle)
1204 }
1205 }
1206 pconn.idleAt = time.Now()
1207 return nil
1208 }
1209
1210
1211
1212
1213 func (t *Transport) queueForIdleConn(w *wantConn) (delivered bool) {
1214 if t.DisableKeepAlives {
1215 return false
1216 }
1217
1218 t.idleMu.Lock()
1219 defer t.idleMu.Unlock()
1220
1221
1222
1223 t.closeIdle = false
1224
1225 if w == nil {
1226
1227 return false
1228 }
1229
1230
1231
1232
1233 var oldTime time.Time
1234 if t.IdleConnTimeout > 0 {
1235 oldTime = time.Now().Add(-t.IdleConnTimeout)
1236 }
1237
1238
1239 if list, ok := t.idleConn[w.key]; ok {
1240 stop := false
1241 delivered := false
1242 for len(list) > 0 && !stop {
1243 pconn := list[len(list)-1]
1244
1245
1246
1247
1248 tooOld := !oldTime.IsZero() && pconn.idleAt.Round(0).Before(oldTime)
1249 if tooOld {
1250
1251
1252
1253 go pconn.closeConnIfStillIdle()
1254 }
1255 if pconn.isBroken() || tooOld {
1256
1257
1258
1259
1260
1261 list = list[:len(list)-1]
1262 continue
1263 }
1264 delivered = w.tryDeliver(pconn, nil, pconn.idleAt)
1265 if delivered {
1266 if pconn.alt != nil {
1267
1268
1269 } else {
1270
1271
1272 t.idleLRU.remove(pconn)
1273 list = list[:len(list)-1]
1274 }
1275 }
1276 stop = true
1277 }
1278 if len(list) > 0 {
1279 t.idleConn[w.key] = list
1280 } else {
1281 delete(t.idleConn, w.key)
1282 }
1283 if stop {
1284 return delivered
1285 }
1286 }
1287
1288
1289 if t.idleConnWait == nil {
1290 t.idleConnWait = make(map[connectMethodKey]wantConnQueue)
1291 }
1292 q := t.idleConnWait[w.key]
1293 q.cleanFrontNotWaiting()
1294 q.pushBack(w)
1295 t.idleConnWait[w.key] = q
1296 return false
1297 }
1298
1299
1300 func (t *Transport) removeIdleConn(pconn *persistConn) bool {
1301 if pconn.isClientConn {
1302 return true
1303 }
1304 t.idleMu.Lock()
1305 defer t.idleMu.Unlock()
1306 return t.removeIdleConnLocked(pconn)
1307 }
1308
1309
1310 func (t *Transport) removeIdleConnLocked(pconn *persistConn) bool {
1311 if pconn.idleTimer != nil {
1312 pconn.idleTimer.Stop()
1313 }
1314 t.idleLRU.remove(pconn)
1315 key := pconn.cacheKey
1316 pconns := t.idleConn[key]
1317 var removed bool
1318 switch len(pconns) {
1319 case 0:
1320
1321 case 1:
1322 if pconns[0] == pconn {
1323 delete(t.idleConn, key)
1324 removed = true
1325 }
1326 default:
1327 for i, v := range pconns {
1328 if v != pconn {
1329 continue
1330 }
1331
1332
1333 copy(pconns[i:], pconns[i+1:])
1334 t.idleConn[key] = pconns[:len(pconns)-1]
1335 removed = true
1336 break
1337 }
1338 }
1339 return removed
1340 }
1341
1342 var zeroDialer net.Dialer
1343
1344 func (t *Transport) dial(ctx context.Context, network, addr string) (net.Conn, error) {
1345 if t.DialContext != nil {
1346 c, err := t.DialContext(ctx, network, addr)
1347 if c == nil && err == nil {
1348 err = errors.New("net/http: Transport.DialContext hook returned (nil, nil)")
1349 }
1350 return c, err
1351 }
1352 if t.Dial != nil {
1353 c, err := t.Dial(network, addr)
1354 if c == nil && err == nil {
1355 err = errors.New("net/http: Transport.Dial hook returned (nil, nil)")
1356 }
1357 return c, err
1358 }
1359 return zeroDialer.DialContext(ctx, network, addr)
1360 }
1361
1362
1363
1364
1365
1366
1367
1368 type wantConn struct {
1369 cm connectMethod
1370 key connectMethodKey
1371
1372
1373
1374
1375 beforeDial func()
1376 afterDial func()
1377
1378 mu sync.Mutex
1379 ctx context.Context
1380 cancelCtx context.CancelFunc
1381 done bool
1382 result chan connOrError
1383 }
1384
1385 type connOrError struct {
1386 pc *persistConn
1387 err error
1388 idleAt time.Time
1389 }
1390
1391
1392 func (w *wantConn) waiting() bool {
1393 w.mu.Lock()
1394 defer w.mu.Unlock()
1395
1396 return !w.done
1397 }
1398
1399
1400 func (w *wantConn) getCtxForDial() context.Context {
1401 w.mu.Lock()
1402 defer w.mu.Unlock()
1403
1404 return w.ctx
1405 }
1406
1407
1408 func (w *wantConn) tryDeliver(pc *persistConn, err error, idleAt time.Time) bool {
1409 w.mu.Lock()
1410 defer w.mu.Unlock()
1411
1412 if w.done {
1413 return false
1414 }
1415 if (pc == nil) == (err == nil) {
1416 panic("net/http: internal error: misuse of tryDeliver")
1417 }
1418 w.ctx = nil
1419 w.done = true
1420
1421 w.result <- connOrError{pc: pc, err: err, idleAt: idleAt}
1422 close(w.result)
1423
1424 return true
1425 }
1426
1427
1428
1429 func (w *wantConn) cancel(t *Transport) {
1430 w.mu.Lock()
1431 var pc *persistConn
1432 if w.done {
1433 if r, ok := <-w.result; ok {
1434 pc = r.pc
1435 }
1436 } else {
1437 close(w.result)
1438 }
1439 w.ctx = nil
1440 w.done = true
1441 w.mu.Unlock()
1442
1443
1444
1445
1446 if pc != nil && pc.alt == nil {
1447 t.putOrCloseIdleConn(pc)
1448 }
1449 }
1450
1451
1452 type wantConnQueue struct {
1453
1454
1455
1456
1457
1458
1459
1460
1461
1462
1463 head []*wantConn
1464 headPos int
1465 tail []*wantConn
1466 }
1467
1468
1469 func (q *wantConnQueue) len() int {
1470 return len(q.head) - q.headPos + len(q.tail)
1471 }
1472
1473
1474 func (q *wantConnQueue) pushBack(w *wantConn) {
1475 q.tail = append(q.tail, w)
1476 }
1477
1478
1479 func (q *wantConnQueue) popFront() *wantConn {
1480 if q.headPos >= len(q.head) {
1481 if len(q.tail) == 0 {
1482 return nil
1483 }
1484
1485 q.head, q.headPos, q.tail = q.tail, 0, q.head[:0]
1486 }
1487 w := q.head[q.headPos]
1488 q.head[q.headPos] = nil
1489 q.headPos++
1490 return w
1491 }
1492
1493
1494 func (q *wantConnQueue) peekFront() *wantConn {
1495 if q.headPos < len(q.head) {
1496 return q.head[q.headPos]
1497 }
1498 if len(q.tail) > 0 {
1499 return q.tail[0]
1500 }
1501 return nil
1502 }
1503
1504
1505
1506 func (q *wantConnQueue) cleanFrontNotWaiting() (cleaned bool) {
1507 for {
1508 w := q.peekFront()
1509 if w == nil || w.waiting() {
1510 return cleaned
1511 }
1512 q.popFront()
1513 cleaned = true
1514 }
1515 }
1516
1517
1518 func (q *wantConnQueue) cleanFrontCanceled() {
1519 for {
1520 w := q.peekFront()
1521 if w == nil || w.cancelCtx != nil {
1522 return
1523 }
1524 q.popFront()
1525 }
1526 }
1527
1528
1529
1530 func (q *wantConnQueue) all(f func(*wantConn)) {
1531 for _, w := range q.head[q.headPos:] {
1532 f(w)
1533 }
1534 for _, w := range q.tail {
1535 f(w)
1536 }
1537 }
1538
1539 func (t *Transport) customDialTLS(ctx context.Context, network, addr string) (conn net.Conn, err error) {
1540 if t.DialTLSContext != nil {
1541 conn, err = t.DialTLSContext(ctx, network, addr)
1542 } else {
1543 conn, err = t.DialTLS(network, addr)
1544 }
1545 if conn == nil && err == nil {
1546 err = errors.New("net/http: Transport.DialTLS or DialTLSContext returned (nil, nil)")
1547 }
1548 return
1549 }
1550
1551
1552
1553
1554
1555 func (t *Transport) getConn(treq *transportRequest, cm connectMethod) (_ *persistConn, err error) {
1556 req := treq.Request
1557 trace := treq.trace
1558 ctx := req.Context()
1559 if trace != nil && trace.GetConn != nil {
1560 trace.GetConn(cm.addr())
1561 }
1562
1563
1564
1565
1566
1567
1568 dialCtx, dialCancel := context.WithCancel(context.WithoutCancel(ctx))
1569
1570 w := &wantConn{
1571 cm: cm,
1572 key: cm.key(),
1573 ctx: dialCtx,
1574 cancelCtx: dialCancel,
1575 result: make(chan connOrError, 1),
1576 beforeDial: testHookPrePendingDial,
1577 afterDial: testHookPostPendingDial,
1578 }
1579 defer func() {
1580 if err != nil {
1581 w.cancel(t)
1582 }
1583 }()
1584
1585
1586 if delivered := t.queueForIdleConn(w); !delivered {
1587 t.queueForDial(w)
1588 }
1589
1590
1591 select {
1592 case r := <-w.result:
1593
1594
1595 if r.pc != nil && r.pc.alt == nil && trace != nil && trace.GotConn != nil {
1596 info := httptrace.GotConnInfo{
1597 Conn: r.pc.conn,
1598 Reused: r.pc.isReused(),
1599 }
1600 if !r.idleAt.IsZero() {
1601 info.WasIdle = true
1602 info.IdleTime = time.Since(r.idleAt)
1603 }
1604 trace.GotConn(info)
1605 }
1606 if r.err != nil {
1607
1608
1609
1610 select {
1611 case <-treq.ctx.Done():
1612 err := context.Cause(treq.ctx)
1613 if err == errRequestCanceled {
1614 err = errRequestCanceledConn
1615 }
1616 return nil, err
1617 default:
1618
1619 }
1620 }
1621 return r.pc, r.err
1622 case <-treq.ctx.Done():
1623 err := context.Cause(treq.ctx)
1624 if err == errRequestCanceled {
1625 err = errRequestCanceledConn
1626 }
1627 return nil, err
1628 }
1629 }
1630
1631
1632
1633 func (t *Transport) queueForDial(w *wantConn) {
1634 w.beforeDial()
1635
1636 t.connsPerHostMu.Lock()
1637 defer t.connsPerHostMu.Unlock()
1638
1639 if t.MaxConnsPerHost <= 0 {
1640 t.startDialConnForLocked(w)
1641 return
1642 }
1643
1644 if n := t.connsPerHost[w.key]; n < t.MaxConnsPerHost {
1645 if t.connsPerHost == nil {
1646 t.connsPerHost = make(map[connectMethodKey]int)
1647 }
1648 t.connsPerHost[w.key] = n + 1
1649 t.startDialConnForLocked(w)
1650 return
1651 }
1652
1653 if t.connsPerHostWait == nil {
1654 t.connsPerHostWait = make(map[connectMethodKey]wantConnQueue)
1655 }
1656 q := t.connsPerHostWait[w.key]
1657 q.cleanFrontNotWaiting()
1658 q.pushBack(w)
1659 t.connsPerHostWait[w.key] = q
1660 }
1661
1662
1663
1664 func (t *Transport) startDialConnForLocked(w *wantConn) {
1665 t.dialsInProgress.cleanFrontCanceled()
1666 t.dialsInProgress.pushBack(w)
1667 go func() {
1668 t.dialConnFor(w)
1669 t.connsPerHostMu.Lock()
1670 defer t.connsPerHostMu.Unlock()
1671 w.cancelCtx = nil
1672 }()
1673 }
1674
1675
1676
1677
1678 func (t *Transport) dialConnFor(w *wantConn) {
1679 defer w.afterDial()
1680 ctx := w.getCtxForDial()
1681 if ctx == nil {
1682 t.decConnsPerHost(w.key)
1683 return
1684 }
1685
1686 const isClientConn = false
1687 pc, err := t.dialConn(ctx, w.cm, isClientConn, nil)
1688 if err == nil && pc.alt != nil {
1689
1690
1691 t.putOrCloseIdleConn(pc)
1692 }
1693 delivered := w.tryDeliver(pc, err, time.Time{})
1694 if err == nil && !delivered && pc.alt == nil {
1695
1696
1697 t.putOrCloseIdleConn(pc)
1698 }
1699 if err != nil {
1700 t.decConnsPerHost(w.key)
1701 }
1702 }
1703
1704
1705
1706 func (t *Transport) decConnsPerHost(key connectMethodKey) {
1707 if t.MaxConnsPerHost <= 0 {
1708 return
1709 }
1710
1711 t.connsPerHostMu.Lock()
1712 defer t.connsPerHostMu.Unlock()
1713 n := t.connsPerHost[key]
1714 if n == 0 {
1715
1716
1717 panic("net/http: internal error: connCount underflow")
1718 }
1719
1720
1721
1722
1723
1724 if q := t.connsPerHostWait[key]; q.len() > 0 {
1725 done := false
1726 for q.len() > 0 {
1727 w := q.popFront()
1728 if w.waiting() {
1729 t.startDialConnForLocked(w)
1730 done = true
1731 break
1732 }
1733 }
1734 if q.len() == 0 {
1735 delete(t.connsPerHostWait, key)
1736 } else {
1737
1738
1739 t.connsPerHostWait[key] = q
1740 }
1741 if done {
1742 return
1743 }
1744 }
1745
1746
1747 if n--; n == 0 {
1748 delete(t.connsPerHost, key)
1749 } else {
1750 t.connsPerHost[key] = n
1751 }
1752 }
1753
1754 func (t *Transport) tlsConfigForDial(host string) (*tls.Config, error) {
1755 firstTLSHost, _, err := net.SplitHostPort(host)
1756 if err != nil {
1757 return nil, err
1758 }
1759 cfg := cloneTLSConfig(t.TLSClientConfig)
1760 if cfg.ServerName == "" {
1761 cfg.ServerName = firstTLSHost
1762 }
1763 return cfg, nil
1764 }
1765
1766
1767
1768
1769 func (pconn *persistConn) addTLS(ctx context.Context, addr string, trace *httptrace.ClientTrace) error {
1770 cfg, err := pconn.t.tlsConfigForDial(addr)
1771 if err != nil {
1772 pconn.conn.Close()
1773 return err
1774 }
1775 if pconn.cacheKey.onlyH1 {
1776 cfg.NextProtos = nil
1777 }
1778 plainConn := pconn.conn
1779 tlsConn := tls.Client(plainConn, cfg)
1780 errc := make(chan error, 2)
1781 var timer *time.Timer
1782 if d := pconn.t.TLSHandshakeTimeout; d != 0 {
1783 timer = time.AfterFunc(d, func() {
1784 errc <- tlsHandshakeTimeoutError{}
1785 })
1786 }
1787 go func() {
1788 if trace != nil && trace.TLSHandshakeStart != nil {
1789 trace.TLSHandshakeStart()
1790 }
1791 err := tlsConn.HandshakeContext(ctx)
1792 if timer != nil {
1793 timer.Stop()
1794 }
1795 errc <- err
1796 }()
1797 if err := <-errc; err != nil {
1798 plainConn.Close()
1799 if err == (tlsHandshakeTimeoutError{}) {
1800
1801
1802 <-errc
1803 }
1804 if trace != nil && trace.TLSHandshakeDone != nil {
1805 trace.TLSHandshakeDone(tls.ConnectionState{}, err)
1806 }
1807 return err
1808 }
1809 cs := tlsConn.ConnectionState()
1810 if trace != nil && trace.TLSHandshakeDone != nil {
1811 trace.TLSHandshakeDone(cs, nil)
1812 }
1813 pconn.tlsState = &cs
1814 pconn.conn = tlsConn
1815 return nil
1816 }
1817
1818 type erringRoundTripper interface {
1819 RoundTripErr() error
1820 }
1821
1822 var testHookProxyConnectTimeout = context.WithTimeout
1823
1824 func (t *Transport) dialConn(ctx context.Context, cm connectMethod, isClientConn bool, internalStateHook func()) (pconn *persistConn, err error) {
1825
1826
1827
1828
1829 if p := t.protocols(); p.http3() {
1830 if p.HTTP1() || p.HTTP2() || p.UnencryptedHTTP2() {
1831 return nil, errors.New("http: when using HTTP3, Transport.Protocols must contain only HTTP3")
1832 }
1833 if t.h3Transport == nil {
1834 return nil, errors.New("http: Transport.Protocols contains HTTP3, but Transport does not support HTTP/3")
1835 }
1836 tlsConfig, err := t.tlsConfigForDial(cm.addr())
1837 if err != nil {
1838 return nil, err
1839 }
1840 tlsConfig.NextProtos = []string{"h3"}
1841 rt, err := t.h3Transport.DialClientConn(ctx, cm.addr(), cm.proxyURL, tlsConfig, internalStateHook)
1842 if err != nil {
1843 return nil, err
1844 }
1845 return &persistConn{
1846 t: t,
1847 cacheKey: cm.key(),
1848 alt: rt,
1849 }, nil
1850 }
1851
1852 pconn = &persistConn{
1853 t: t,
1854 cacheKey: cm.key(),
1855 reqch: make(chan requestAndChan, 1),
1856 writech: make(chan writeRequest, 1),
1857 closech: make(chan struct{}),
1858 writeErrCh: make(chan error, 1),
1859 writeLoopDone: make(chan struct{}),
1860 isClientConn: isClientConn,
1861 internalStateHook: internalStateHook,
1862 }
1863 trace := httptrace.ContextClientTrace(ctx)
1864 wrapErr := func(err error) error {
1865 if cm.proxyURL != nil {
1866
1867 return &net.OpError{Op: "proxyconnect", Net: "tcp", Err: err}
1868 }
1869 return err
1870 }
1871
1872 if rt, err := t.http2ExternalDial(ctx, cm); err != errors.ErrUnsupported {
1873 if err != nil {
1874 return nil, err
1875 }
1876 return &persistConn{t: t, cacheKey: pconn.cacheKey, alt: rt}, nil
1877 }
1878
1879 if cm.scheme() == "https" && t.hasCustomTLSDialer() {
1880 var err error
1881 pconn.conn, err = t.customDialTLS(ctx, "tcp", cm.addr())
1882 if err != nil {
1883 return nil, wrapErr(err)
1884 }
1885 type connectionStater interface {
1886 ConnectionState() tls.ConnectionState
1887 }
1888 type handshaker interface {
1889 HandshakeContext(context.Context) error
1890 }
1891 if cstater, ok := pconn.conn.(connectionStater); ok {
1892 if trace != nil && trace.TLSHandshakeStart != nil {
1893 trace.TLSHandshakeStart()
1894 }
1895 if handshaker, ok := cstater.(handshaker); ok {
1896
1897
1898 if err := handshaker.HandshakeContext(ctx); err != nil {
1899 go pconn.conn.Close()
1900 if trace != nil && trace.TLSHandshakeDone != nil {
1901 trace.TLSHandshakeDone(tls.ConnectionState{}, err)
1902 }
1903 return nil, err
1904 }
1905 }
1906 cs := cstater.ConnectionState()
1907 if trace != nil && trace.TLSHandshakeDone != nil {
1908 trace.TLSHandshakeDone(cs, nil)
1909 }
1910 pconn.tlsState = &cs
1911 }
1912 } else {
1913 conn, err := t.dial(ctx, "tcp", cm.addr())
1914 if err != nil {
1915 return nil, wrapErr(err)
1916 }
1917 pconn.conn = conn
1918 if cm.scheme() == "https" {
1919 if err = pconn.addTLS(ctx, cm.addr(), trace); err != nil {
1920 return nil, wrapErr(err)
1921 }
1922 }
1923 }
1924
1925
1926 switch {
1927 case cm.proxyURL == nil:
1928
1929 case cm.proxyURL.Scheme == "socks5" || cm.proxyURL.Scheme == "socks5h":
1930 conn := pconn.conn
1931 d := socksNewDialer("tcp", conn.RemoteAddr().String())
1932 if u := cm.proxyURL.User; u != nil {
1933 auth := &socksUsernamePassword{
1934 Username: u.Username(),
1935 }
1936 auth.Password, _ = u.Password()
1937 d.AuthMethods = []socksAuthMethod{
1938 socksAuthMethodNotRequired,
1939 socksAuthMethodUsernamePassword,
1940 }
1941 d.Authenticate = auth.Authenticate
1942 }
1943 if _, err := d.DialWithConn(ctx, conn, "tcp", cm.targetAddr); err != nil {
1944 conn.Close()
1945 return nil, err
1946 }
1947 case cm.targetScheme == "http":
1948 pconn.isProxy = true
1949 if pa := cm.proxyAuth(); pa != "" {
1950 pconn.mutateHeaderFunc = func(h Header) {
1951 h.Set("Proxy-Authorization", pa)
1952 }
1953 }
1954 case cm.targetScheme == "https":
1955 conn := pconn.conn
1956 var hdr Header
1957 if t.GetProxyConnectHeader != nil {
1958 var err error
1959 hdr, err = t.GetProxyConnectHeader(ctx, cm.proxyURL, cm.targetAddr)
1960 if err != nil {
1961 conn.Close()
1962 return nil, err
1963 }
1964 } else {
1965 hdr = t.ProxyConnectHeader
1966 }
1967 if hdr == nil {
1968 hdr = make(Header)
1969 }
1970 if pa := cm.proxyAuth(); pa != "" {
1971 hdr = hdr.Clone()
1972 hdr.Set("Proxy-Authorization", pa)
1973 }
1974 connectReq := &Request{
1975 Method: "CONNECT",
1976 URL: &url.URL{Opaque: cm.targetAddr},
1977 Host: cm.targetAddr,
1978 Header: hdr,
1979 }
1980
1981
1982
1983
1984 connectCtx, cancel := testHookProxyConnectTimeout(ctx, 1*time.Minute)
1985 defer cancel()
1986
1987 didReadResponse := make(chan struct{})
1988 var (
1989 resp *Response
1990 err error
1991 )
1992
1993 go func() {
1994 defer close(didReadResponse)
1995 err = connectReq.Write(conn)
1996 if err != nil {
1997 return
1998 }
1999
2000
2001 br := bufio.NewReader(&io.LimitedReader{R: conn, N: t.maxHeaderResponseSize()})
2002 resp, err = ReadResponse(br, connectReq)
2003 }()
2004 select {
2005 case <-connectCtx.Done():
2006 conn.Close()
2007 <-didReadResponse
2008 return nil, connectCtx.Err()
2009 case <-didReadResponse:
2010
2011 }
2012 if err != nil {
2013 conn.Close()
2014 return nil, err
2015 }
2016
2017 if t.OnProxyConnectResponse != nil {
2018 err = t.OnProxyConnectResponse(ctx, cm.proxyURL, connectReq, resp)
2019 if err != nil {
2020 conn.Close()
2021 return nil, err
2022 }
2023 }
2024
2025 if resp.StatusCode != 200 {
2026 _, text, ok := strings.Cut(resp.Status, " ")
2027 conn.Close()
2028 if !ok {
2029 return nil, errors.New("unknown status code")
2030 }
2031 return nil, errors.New(text)
2032 }
2033 }
2034
2035 if cm.proxyURL != nil && cm.targetScheme == "https" {
2036 if err := pconn.addTLS(ctx, cm.targetAddr, trace); err != nil {
2037 return nil, err
2038 }
2039 }
2040
2041
2042 unencryptedHTTP2 := pconn.tlsState == nil &&
2043 t.Protocols != nil &&
2044 t.Protocols.UnencryptedHTTP2() &&
2045 !t.Protocols.HTTP1()
2046
2047 http2 := unencryptedHTTP2 ||
2048 (pconn.tlsState != nil && pconn.tlsState.NegotiatedProtocol == "h2")
2049
2050 if http2 && t.h2Transport != nil {
2051 if isClientConn {
2052 cc, err := t.http2NewClientConn(pconn.conn, internalStateHook)
2053 if err == nil {
2054 return &persistConn{t: t, cacheKey: pconn.cacheKey, alt: cc, isClientConn: true}, nil
2055 }
2056 if err != errors.ErrUnsupported {
2057 return nil, err
2058 }
2059 } else {
2060 rt, err := t.http2AddConn(cm.targetScheme, cm.targetAddr, pconn.conn)
2061 if err == nil {
2062 return &persistConn{t: t, cacheKey: pconn.cacheKey, alt: rt}, nil
2063 }
2064 if err != errors.ErrUnsupported {
2065 return nil, err
2066 }
2067 }
2068 }
2069
2070 if isClientConn && (unencryptedHTTP2 || (pconn.tlsState != nil && pconn.tlsState.NegotiatedProtocol == "h2")) {
2071 altProto, _ := t.altProto.Load().(map[string]RoundTripper)
2072 h2, ok := altProto["https"].(newClientConner)
2073 if !ok {
2074 return nil, errors.New("http: HTTP/2 implementation does not support NewClientConn (update golang.org/x/net?)")
2075 }
2076 alt, err := h2.NewClientConn(pconn.conn, internalStateHook)
2077 if err != nil {
2078 pconn.conn.Close()
2079 return nil, err
2080 }
2081 return &persistConn{t: t, cacheKey: pconn.cacheKey, alt: alt, isClientConn: true}, nil
2082 }
2083
2084 if unencryptedHTTP2 {
2085 next, ok := t.TLSNextProto[nextProtoUnencryptedHTTP2]
2086 if !ok {
2087 return nil, errors.New("http: Transport does not support unencrypted HTTP/2")
2088 }
2089 alt := next(cm.targetAddr, unencryptedTLSConn(pconn.conn))
2090 if e, ok := alt.(erringRoundTripper); ok {
2091
2092 return nil, e.RoundTripErr()
2093 }
2094 return &persistConn{t: t, cacheKey: pconn.cacheKey, alt: alt}, nil
2095 }
2096
2097 if s := pconn.tlsState; s != nil && s.NegotiatedProtocolIsMutual && s.NegotiatedProtocol != "" {
2098 tlsConn, tlsConnOK := pconn.conn.(*tls.Conn)
2099 if next, ok := t.TLSNextProto[s.NegotiatedProtocol]; tlsConnOK && ok {
2100 alt := next(cm.targetAddr, tlsConn)
2101 if e, ok := alt.(erringRoundTripper); ok {
2102
2103 return nil, e.RoundTripErr()
2104 }
2105 return &persistConn{t: t, cacheKey: pconn.cacheKey, alt: alt}, nil
2106 }
2107 }
2108
2109 pconn.br = bufio.NewReaderSize(pconn, t.readBufferSize())
2110 pconn.bw = bufio.NewWriterSize(persistConnWriter{pconn}, t.writeBufferSize())
2111
2112 go pconn.readLoop()
2113 go pconn.writeLoop()
2114 return pconn, nil
2115 }
2116
2117
2118
2119
2120
2121
2122
2123 type persistConnWriter struct {
2124 pc *persistConn
2125 }
2126
2127 func (w persistConnWriter) Write(p []byte) (n int, err error) {
2128 n, err = w.pc.conn.Write(p)
2129 w.pc.nwrite += int64(n)
2130 return
2131 }
2132
2133
2134
2135
2136 func (w persistConnWriter) ReadFrom(r io.Reader) (n int64, err error) {
2137 n, err = io.Copy(w.pc.conn, r)
2138 w.pc.nwrite += n
2139 return
2140 }
2141
2142 var _ io.ReaderFrom = (*persistConnWriter)(nil)
2143
2144
2145
2146
2147
2148
2149
2150
2151
2152
2153
2154
2155
2156
2157
2158
2159
2160 type connectMethod struct {
2161 _ incomparable
2162 proxyURL *url.URL
2163 targetScheme string
2164
2165
2166
2167 targetAddr string
2168 onlyH1 bool
2169 }
2170
2171 func (cm *connectMethod) key() connectMethodKey {
2172 proxyStr := ""
2173 targetAddr := cm.targetAddr
2174 if cm.proxyURL != nil {
2175 proxyStr = cm.proxyURL.String()
2176 if (cm.proxyURL.Scheme == "http" || cm.proxyURL.Scheme == "https") && cm.targetScheme == "http" {
2177 targetAddr = ""
2178 }
2179 }
2180 return connectMethodKey{
2181 proxy: proxyStr,
2182 scheme: cm.targetScheme,
2183 addr: targetAddr,
2184 onlyH1: cm.onlyH1,
2185 }
2186 }
2187
2188
2189 func (cm *connectMethod) scheme() string {
2190 if cm.proxyURL != nil {
2191 return cm.proxyURL.Scheme
2192 }
2193 return cm.targetScheme
2194 }
2195
2196
2197 func (cm *connectMethod) addr() string {
2198 if cm.proxyURL != nil {
2199 return canonicalAddr(cm.proxyURL)
2200 }
2201 return cm.targetAddr
2202 }
2203
2204
2205
2206
2207 type connectMethodKey struct {
2208 proxy, scheme, addr string
2209 onlyH1 bool
2210 }
2211
2212 func (k connectMethodKey) String() string {
2213
2214 var h1 string
2215 if k.onlyH1 {
2216 h1 = ",h1"
2217 }
2218 return fmt.Sprintf("%s|%s%s|%s", k.proxy, k.scheme, h1, k.addr)
2219 }
2220
2221
2222
2223 type persistConn struct {
2224
2225
2226
2227 alt RoundTripper
2228
2229 t *Transport
2230 cacheKey connectMethodKey
2231 conn net.Conn
2232 tlsState *tls.ConnectionState
2233 br *bufio.Reader
2234 bw *bufio.Writer
2235 nwrite int64
2236 reqch chan requestAndChan
2237 writech chan writeRequest
2238 closech chan struct{}
2239 availch chan struct{}
2240 isProxy bool
2241 sawEOF bool
2242 isClientConn bool
2243 readLimit int64
2244
2245
2246
2247
2248 writeErrCh chan error
2249
2250 writeLoopDone chan struct{}
2251
2252
2253 idleAt time.Time
2254 idleTimer *time.Timer
2255
2256 mu sync.Mutex
2257 numExpectedResponses int
2258 closed error
2259 canceledErr error
2260 reused bool
2261 reserved bool
2262 inFlight bool
2263 internalStateHook func()
2264
2265
2266
2267
2268 mutateHeaderFunc func(Header)
2269 }
2270
2271 func (pc *persistConn) maxHeaderResponseSize() int64 {
2272 return pc.t.maxHeaderResponseSize()
2273 }
2274
2275 func (pc *persistConn) Read(p []byte) (n int, err error) {
2276 if pc.readLimit <= 0 {
2277 return 0, fmt.Errorf("read limit of %d bytes exhausted", pc.maxHeaderResponseSize())
2278 }
2279 if int64(len(p)) > pc.readLimit {
2280 p = p[:pc.readLimit]
2281 }
2282 n, err = pc.conn.Read(p)
2283 if err == io.EOF {
2284 pc.sawEOF = true
2285 }
2286 pc.readLimit -= int64(n)
2287 return
2288 }
2289
2290
2291 func (pc *persistConn) isBroken() bool {
2292 pc.mu.Lock()
2293 b := pc.closed != nil
2294 pc.mu.Unlock()
2295 return b
2296 }
2297
2298
2299 func (pc *persistConn) canceled() error {
2300 pc.mu.Lock()
2301 defer pc.mu.Unlock()
2302 return pc.canceledErr
2303 }
2304
2305
2306 func (pc *persistConn) isReused() bool {
2307 pc.mu.Lock()
2308 r := pc.reused
2309 pc.mu.Unlock()
2310 return r
2311 }
2312
2313 func (pc *persistConn) cancelRequest(err error) {
2314 pc.mu.Lock()
2315 defer pc.mu.Unlock()
2316 pc.canceledErr = err
2317 pc.closeLocked(errRequestCanceled)
2318 }
2319
2320
2321
2322
2323 func (pc *persistConn) closeConnIfStillIdle() {
2324 t := pc.t
2325 t.idleMu.Lock()
2326 defer t.idleMu.Unlock()
2327 if _, ok := t.idleLRU.m[pc]; !ok {
2328
2329 return
2330 }
2331 t.removeIdleConnLocked(pc)
2332 pc.close(errIdleConnTimeout)
2333 }
2334
2335
2336
2337
2338
2339
2340
2341
2342
2343 func (pc *persistConn) mapRoundTripError(req *transportRequest, startBytesWritten int64, err error) error {
2344 if err == nil {
2345 return nil
2346 }
2347
2348
2349
2350
2351
2352
2353
2354
2355 <-pc.writeLoopDone
2356
2357
2358
2359
2360 if cerr := pc.canceled(); cerr != nil {
2361 return cerr
2362 }
2363
2364
2365 req.mu.Lock()
2366 reqErr := req.err
2367 req.mu.Unlock()
2368 if reqErr != nil {
2369 return reqErr
2370 }
2371
2372 if err == errServerClosedIdle {
2373
2374 return err
2375 }
2376
2377 if _, ok := err.(transportReadFromServerError); ok {
2378 if pc.nwrite == startBytesWritten {
2379 return nothingWrittenError{err}
2380 }
2381
2382 return err
2383 }
2384 if pc.isBroken() {
2385 if pc.nwrite == startBytesWritten {
2386 return nothingWrittenError{err}
2387 }
2388 return fmt.Errorf("net/http: HTTP/1.x transport connection broken: %w", err)
2389 }
2390 return err
2391 }
2392
2393
2394
2395
2396 var errCallerOwnsConn = errors.New("read loop ending; caller owns writable underlying conn")
2397
2398
2399
2400
2401 const maxPostCloseReadBytes = 256 << 10
2402
2403
2404
2405
2406 const maxPostCloseReadTime = 50 * time.Millisecond
2407
2408 func maybeDrainBody(body io.Reader) bool {
2409 drainedCh := make(chan bool, 1)
2410 go func() {
2411 if _, err := io.CopyN(io.Discard, body, maxPostCloseReadBytes+1); err == io.EOF {
2412 drainedCh <- true
2413 } else {
2414 drainedCh <- false
2415 }
2416 }()
2417 select {
2418 case drained := <-drainedCh:
2419 return drained
2420 case <-time.After(maxPostCloseReadTime):
2421 return false
2422 }
2423 }
2424
2425 func (pc *persistConn) readLoop() {
2426 closeErr := errReadLoopExiting
2427 defer func() {
2428 pc.close(closeErr)
2429 pc.t.removeIdleConn(pc)
2430 if pc.internalStateHook != nil {
2431 pc.internalStateHook()
2432 }
2433 }()
2434
2435 tryPutIdleConn := func(treq *transportRequest) bool {
2436 trace := treq.trace
2437 if err := pc.t.tryPutIdleConn(pc); err != nil {
2438 closeErr = err
2439 if trace != nil && trace.PutIdleConn != nil && err != errKeepAlivesDisabled {
2440 trace.PutIdleConn(err)
2441 }
2442 return false
2443 }
2444 if trace != nil && trace.PutIdleConn != nil {
2445 trace.PutIdleConn(nil)
2446 }
2447 return true
2448 }
2449
2450
2451
2452
2453 eofc := make(chan struct{})
2454 defer close(eofc)
2455
2456
2457 testHookMu.Lock()
2458 testHookReadLoopBeforeNextRead := testHookReadLoopBeforeNextRead
2459 testHookMu.Unlock()
2460
2461 alive := true
2462 for alive {
2463 pc.readLimit = pc.maxHeaderResponseSize()
2464 _, err := pc.br.Peek(1)
2465
2466 pc.mu.Lock()
2467 if pc.numExpectedResponses == 0 {
2468 pc.readLoopPeekFailLocked(err)
2469 pc.mu.Unlock()
2470 return
2471 }
2472 pc.mu.Unlock()
2473
2474 rc := <-pc.reqch
2475 trace := rc.treq.trace
2476
2477 var resp *Response
2478 if err == nil {
2479 resp, err = pc.readResponse(rc, trace)
2480 } else {
2481 err = transportReadFromServerError{err}
2482 closeErr = err
2483 }
2484
2485 if err != nil {
2486 if pc.readLimit <= 0 {
2487 err = fmt.Errorf("net/http: server response headers exceeded %d bytes; aborted", pc.maxHeaderResponseSize())
2488 }
2489
2490 select {
2491 case rc.ch <- responseAndError{err: err}:
2492 case <-rc.callerGone:
2493 return
2494 }
2495 return
2496 }
2497 pc.readLimit = maxInt64
2498
2499 pc.mu.Lock()
2500 pc.numExpectedResponses--
2501 pc.mu.Unlock()
2502
2503 bodyWritable := resp.bodyIsWritable()
2504 hasBody := rc.treq.Request.Method != "HEAD" && resp.ContentLength != 0
2505
2506 if resp.Close || rc.treq.Request.Close || resp.StatusCode <= 199 || bodyWritable {
2507
2508
2509
2510 alive = false
2511 }
2512
2513 if !hasBody || bodyWritable {
2514
2515
2516
2517
2518
2519 alive = alive &&
2520 !pc.sawEOF &&
2521 pc.wroteRequest() &&
2522 tryPutIdleConn(rc.treq)
2523
2524 if bodyWritable {
2525 closeErr = errCallerOwnsConn
2526 }
2527
2528 select {
2529 case rc.ch <- responseAndError{res: resp}:
2530 case <-rc.callerGone:
2531 return
2532 }
2533
2534 rc.treq.cancel(errRequestDone)
2535
2536
2537
2538
2539 testHookReadLoopBeforeNextRead()
2540 continue
2541 }
2542
2543 waitForBodyRead := make(chan bool, 2)
2544 body := &bodyEOFSignal{
2545 body: resp.Body,
2546 earlyCloseFn: func() error {
2547 waitForBodyRead <- false
2548 <-eofc
2549 return nil
2550
2551 },
2552 fn: func(err error) error {
2553 isEOF := err == io.EOF
2554 waitForBodyRead <- isEOF
2555 if isEOF {
2556 <-eofc
2557 } else if err != nil {
2558 if cerr := pc.canceled(); cerr != nil {
2559 return cerr
2560 }
2561 }
2562 return err
2563 },
2564 }
2565
2566 resp.Body = body
2567 if rc.addedGzip && ascii.EqualFold(resp.Header.Get("Content-Encoding"), "gzip") {
2568 resp.Body = &gzipReader{body: body}
2569 resp.Header.Del("Content-Encoding")
2570 resp.Header.Del("Content-Length")
2571 resp.ContentLength = -1
2572 resp.Uncompressed = true
2573 }
2574
2575 select {
2576 case rc.ch <- responseAndError{res: resp}:
2577 case <-rc.callerGone:
2578 return
2579 }
2580
2581
2582
2583
2584 select {
2585 case bodyEOF := <-waitForBodyRead:
2586 tryDrain := !bodyEOF && resp.ContentLength <= maxPostCloseReadBytes
2587 if tryDrain {
2588 eofc <- struct{}{}
2589 bodyEOF = maybeDrainBody(body.body)
2590 }
2591 alive = alive &&
2592 bodyEOF &&
2593 !pc.sawEOF &&
2594 pc.wroteRequest() &&
2595 tryPutIdleConn(rc.treq)
2596 if !tryDrain && bodyEOF {
2597 eofc <- struct{}{}
2598 }
2599 case <-rc.treq.ctx.Done():
2600 alive = false
2601 pc.cancelRequest(context.Cause(rc.treq.ctx))
2602 case <-pc.closech:
2603 alive = false
2604 }
2605
2606 rc.treq.cancel(errRequestDone)
2607 testHookReadLoopBeforeNextRead()
2608 }
2609 }
2610
2611 func (pc *persistConn) readLoopPeekFailLocked(peekErr error) {
2612 if pc.closed != nil {
2613 return
2614 }
2615 if n := pc.br.Buffered(); n > 0 {
2616 buf, _ := pc.br.Peek(n)
2617 if is408Message(buf) {
2618 pc.closeLocked(errServerClosedIdle)
2619 return
2620 } else {
2621 log.Printf("Unsolicited response received on idle HTTP channel starting with %q; err=%v", buf, peekErr)
2622 }
2623 }
2624 if peekErr == io.EOF {
2625
2626 pc.closeLocked(errServerClosedIdle)
2627 } else {
2628 pc.closeLocked(fmt.Errorf("readLoopPeekFailLocked: %w", peekErr))
2629 }
2630 }
2631
2632
2633
2634
2635 func is408Message(buf []byte) bool {
2636 if len(buf) < len("HTTP/1.x 408") {
2637 return false
2638 }
2639 if string(buf[:7]) != "HTTP/1." {
2640 return false
2641 }
2642 return string(buf[8:12]) == " 408"
2643 }
2644
2645
2646
2647
2648 func (pc *persistConn) readResponse(rc requestAndChan, trace *httptrace.ClientTrace) (resp *Response, err error) {
2649 if trace != nil && trace.GotFirstResponseByte != nil {
2650 if peek, err := pc.br.Peek(1); err == nil && len(peek) == 1 {
2651 trace.GotFirstResponseByte()
2652 }
2653 }
2654
2655 continueCh := rc.continueCh
2656 for {
2657 resp, err = ReadResponse(pc.br, rc.treq.Request)
2658 if err != nil {
2659 return
2660 }
2661 resCode := resp.StatusCode
2662 if continueCh != nil && resCode == StatusContinue {
2663 if trace != nil && trace.Got100Continue != nil {
2664 trace.Got100Continue()
2665 }
2666 continueCh <- struct{}{}
2667 continueCh = nil
2668 }
2669 is1xx := 100 <= resCode && resCode <= 199
2670
2671 is1xxNonTerminal := is1xx && resCode != StatusSwitchingProtocols
2672 if is1xxNonTerminal {
2673 if trace != nil && trace.Got1xxResponse != nil {
2674 if err := trace.Got1xxResponse(resCode, textproto.MIMEHeader(resp.Header)); err != nil {
2675 return nil, err
2676 }
2677
2678
2679
2680
2681
2682
2683
2684 pc.readLimit = pc.maxHeaderResponseSize()
2685 }
2686 continue
2687 }
2688 break
2689 }
2690 if resp.isProtocolSwitch() {
2691 resp.Body = newReadWriteCloserBody(pc.br, pc.conn)
2692 }
2693 if continueCh != nil {
2694
2695
2696
2697
2698
2699
2700
2701
2702
2703
2704
2705
2706 if resp.Close || rc.treq.Request.Close {
2707 close(continueCh)
2708 } else {
2709 continueCh <- struct{}{}
2710 }
2711 }
2712
2713 resp.TLS = pc.tlsState
2714 return
2715 }
2716
2717
2718
2719
2720 func (pc *persistConn) waitForContinue(continueCh <-chan struct{}) func() bool {
2721 if continueCh == nil {
2722 return nil
2723 }
2724 return func() bool {
2725 timer := time.NewTimer(pc.t.ExpectContinueTimeout)
2726 defer timer.Stop()
2727
2728 select {
2729 case _, ok := <-continueCh:
2730 return ok
2731 case <-timer.C:
2732 return true
2733 case <-pc.closech:
2734 return false
2735 }
2736 }
2737 }
2738
2739 func newReadWriteCloserBody(br *bufio.Reader, rwc io.ReadWriteCloser) io.ReadWriteCloser {
2740 body := &readWriteCloserBody{ReadWriteCloser: rwc}
2741 if br.Buffered() != 0 {
2742 body.br = br
2743 }
2744 return body
2745 }
2746
2747
2748
2749
2750
2751
2752 type readWriteCloserBody struct {
2753 _ incomparable
2754 br *bufio.Reader
2755 io.ReadWriteCloser
2756 }
2757
2758 func (b *readWriteCloserBody) Read(p []byte) (n int, err error) {
2759 if b.br != nil {
2760 if n := b.br.Buffered(); len(p) > n {
2761 p = p[:n]
2762 }
2763 n, err = b.br.Read(p)
2764 if b.br.Buffered() == 0 {
2765 b.br = nil
2766 }
2767 return n, err
2768 }
2769 return b.ReadWriteCloser.Read(p)
2770 }
2771
2772 func (b *readWriteCloserBody) CloseWrite() error {
2773 if cw, ok := b.ReadWriteCloser.(interface{ CloseWrite() error }); ok {
2774 return cw.CloseWrite()
2775 }
2776 return fmt.Errorf("CloseWrite: %w", ErrNotSupported)
2777 }
2778
2779
2780 type nothingWrittenError struct {
2781 error
2782 }
2783
2784 func (nwe nothingWrittenError) Unwrap() error {
2785 return nwe.error
2786 }
2787
2788 func (pc *persistConn) writeLoop() {
2789 defer close(pc.writeLoopDone)
2790 for {
2791 select {
2792 case wr := <-pc.writech:
2793 startBytesWritten := pc.nwrite
2794 err := wr.req.Request.write(pc.bw, pc.isProxy, wr.req.extra, pc.waitForContinue(wr.continueCh))
2795 if bre, ok := err.(requestBodyReadError); ok {
2796 err = bre.error
2797
2798
2799
2800
2801
2802
2803
2804 wr.req.setError(err)
2805 }
2806 if err == nil {
2807 err = pc.bw.Flush()
2808 }
2809 if err != nil {
2810 if pc.nwrite == startBytesWritten {
2811 err = nothingWrittenError{err}
2812 }
2813 }
2814 pc.writeErrCh <- err
2815 wr.ch <- err
2816 if err != nil {
2817 pc.close(err)
2818 return
2819 }
2820 case <-pc.closech:
2821 return
2822 }
2823 }
2824 }
2825
2826
2827
2828
2829
2830
2831
2832 var maxWriteWaitBeforeConnReuse = 50 * time.Millisecond
2833
2834
2835
2836 func (pc *persistConn) wroteRequest() bool {
2837 select {
2838 case err := <-pc.writeErrCh:
2839
2840
2841 return err == nil
2842 default:
2843
2844
2845
2846
2847
2848
2849
2850
2851
2852
2853 t := time.NewTimer(maxWriteWaitBeforeConnReuse)
2854 defer t.Stop()
2855 select {
2856 case err := <-pc.writeErrCh:
2857 return err == nil
2858 case <-t.C:
2859 return false
2860 }
2861 }
2862 }
2863
2864
2865
2866 type responseAndError struct {
2867 _ incomparable
2868 res *Response
2869 err error
2870 }
2871
2872 type requestAndChan struct {
2873 _ incomparable
2874 treq *transportRequest
2875 ch chan responseAndError
2876
2877
2878
2879
2880 addedGzip bool
2881
2882
2883
2884
2885
2886 continueCh chan<- struct{}
2887
2888 callerGone <-chan struct{}
2889 }
2890
2891
2892
2893
2894
2895 type writeRequest struct {
2896 req *transportRequest
2897 ch chan<- error
2898
2899
2900
2901
2902 continueCh <-chan struct{}
2903 }
2904
2905
2906
2907 type timeoutError struct {
2908 err string
2909 }
2910
2911 func (e *timeoutError) Error() string { return e.err }
2912 func (e *timeoutError) Timeout() bool { return true }
2913 func (e *timeoutError) Temporary() bool { return true }
2914 func (e *timeoutError) Is(err error) bool { return err == context.DeadlineExceeded }
2915
2916 var errTimeout error = &timeoutError{"net/http: timeout awaiting response headers"}
2917
2918
2919
2920 var errRequestCanceled = internal.ErrRequestCanceled
2921 var errRequestCanceledConn = errors.New("net/http: request canceled while waiting for connection")
2922
2923
2924
2925 var errRequestDone = errors.New("net/http: request completed")
2926
2927 func nop() {}
2928
2929
2930 var (
2931 testHookEnterRoundTrip = nop
2932 testHookWaitResLoop = nop
2933 testHookRoundTripRetried = nop
2934 testHookPrePendingDial = nop
2935 testHookPostPendingDial = nop
2936
2937 testHookMu sync.Locker = fakeLocker{}
2938 testHookReadLoopBeforeNextRead = nop
2939 )
2940
2941 func (pc *persistConn) waitForAvailability(ctx context.Context) error {
2942 select {
2943 case <-pc.availch:
2944 return nil
2945 case <-pc.closech:
2946 return pc.closed
2947 case <-ctx.Done():
2948 return ctx.Err()
2949 }
2950 }
2951
2952 func (pc *persistConn) roundTrip(req *transportRequest) (resp *Response, err error) {
2953 testHookEnterRoundTrip()
2954
2955 pc.mu.Lock()
2956 if pc.isClientConn {
2957 if !pc.reserved {
2958 pc.mu.Unlock()
2959 if err := pc.waitForAvailability(req.ctx); err != nil {
2960 return nil, err
2961 }
2962 pc.mu.Lock()
2963 }
2964 pc.reserved = false
2965 pc.inFlight = true
2966 }
2967 pc.numExpectedResponses++
2968 headerFn := pc.mutateHeaderFunc
2969 pc.mu.Unlock()
2970
2971 if headerFn != nil {
2972 headerFn(req.extraHeaders())
2973 }
2974
2975
2976
2977
2978
2979 requestedGzip := false
2980 if !pc.t.DisableCompression &&
2981 req.Header.Get("Accept-Encoding") == "" &&
2982 req.Header.Get("Range") == "" &&
2983 req.Method != "HEAD" {
2984
2985
2986
2987
2988
2989
2990
2991
2992
2993
2994
2995
2996 requestedGzip = true
2997 req.extraHeaders().Set("Accept-Encoding", "gzip")
2998 }
2999
3000 var continueCh chan struct{}
3001 if req.ProtoAtLeast(1, 1) && req.Body != nil && req.expectsContinue() {
3002 continueCh = make(chan struct{}, 1)
3003 }
3004
3005 if pc.t.DisableKeepAlives &&
3006 !req.wantsClose() &&
3007 !isProtocolSwitchHeader(req.Header) {
3008 req.extraHeaders().Set("Connection", "close")
3009 }
3010
3011 gone := make(chan struct{})
3012 defer close(gone)
3013
3014 const debugRoundTrip = false
3015
3016
3017
3018
3019 startBytesWritten := pc.nwrite
3020 writeErrCh := make(chan error, 1)
3021 pc.writech <- writeRequest{req, writeErrCh, continueCh}
3022
3023 resc := make(chan responseAndError)
3024 pc.reqch <- requestAndChan{
3025 treq: req,
3026 ch: resc,
3027 addedGzip: requestedGzip,
3028 continueCh: continueCh,
3029 callerGone: gone,
3030 }
3031
3032 handleResponse := func(re responseAndError) (*Response, error) {
3033 if (re.res == nil) == (re.err == nil) {
3034 panic(fmt.Sprintf("internal error: exactly one of res or err should be set; nil=%v", re.res == nil))
3035 }
3036 if debugRoundTrip {
3037 req.logf("resc recv: %p, %T/%#v", re.res, re.err, re.err)
3038 }
3039 if re.err != nil {
3040 return nil, pc.mapRoundTripError(req, startBytesWritten, re.err)
3041 }
3042 return re.res, nil
3043 }
3044
3045 var respHeaderTimer <-chan time.Time
3046 ctxDoneChan := req.ctx.Done()
3047 pcClosed := pc.closech
3048 for {
3049 testHookWaitResLoop()
3050 select {
3051 case err := <-writeErrCh:
3052 if debugRoundTrip {
3053 req.logf("writeErrCh recv: %T/%#v", err, err)
3054 }
3055 if err != nil {
3056 pc.close(fmt.Errorf("write error: %w", err))
3057 return nil, pc.mapRoundTripError(req, startBytesWritten, err)
3058 }
3059 if d := pc.t.ResponseHeaderTimeout; d > 0 {
3060 if debugRoundTrip {
3061 req.logf("starting timer for %v", d)
3062 }
3063 timer := time.NewTimer(d)
3064 defer timer.Stop()
3065 respHeaderTimer = timer.C
3066 }
3067 case <-pcClosed:
3068 select {
3069 case re := <-resc:
3070
3071
3072
3073 return handleResponse(re)
3074 default:
3075 }
3076 if debugRoundTrip {
3077 req.logf("closech recv: %T %#v", pc.closed, pc.closed)
3078 }
3079 return nil, pc.mapRoundTripError(req, startBytesWritten, pc.closed)
3080 case <-respHeaderTimer:
3081 if debugRoundTrip {
3082 req.logf("timeout waiting for response headers.")
3083 }
3084 pc.close(errTimeout)
3085 return nil, errTimeout
3086 case re := <-resc:
3087 return handleResponse(re)
3088 case <-ctxDoneChan:
3089 select {
3090 case re := <-resc:
3091
3092
3093
3094 return handleResponse(re)
3095 default:
3096 }
3097 pc.cancelRequest(context.Cause(req.ctx))
3098 }
3099 }
3100 }
3101
3102
3103
3104 type tLogKey struct{}
3105
3106 func (tr *transportRequest) logf(format string, args ...any) {
3107 if logf, ok := tr.Request.Context().Value(tLogKey{}).(func(string, ...any)); ok {
3108 logf(time.Now().Format(time.RFC3339Nano)+": "+format, args...)
3109 }
3110 }
3111
3112
3113
3114 func (pc *persistConn) markReused() {
3115 pc.mu.Lock()
3116 pc.reused = true
3117 pc.mu.Unlock()
3118 }
3119
3120
3121
3122
3123
3124
3125 func (pc *persistConn) close(err error) {
3126 pc.mu.Lock()
3127 defer pc.mu.Unlock()
3128 pc.closeLocked(err)
3129 }
3130
3131 func (pc *persistConn) closeLocked(err error) {
3132 if err == nil {
3133 panic("nil error")
3134 }
3135 if pc.closed == nil {
3136 pc.closed = err
3137 pc.t.decConnsPerHost(pc.cacheKey)
3138
3139
3140
3141 if pc.alt == nil {
3142 if err != errCallerOwnsConn {
3143 pc.conn.Close()
3144 }
3145 close(pc.closech)
3146 } else {
3147 if cc, ok := pc.alt.(io.Closer); ok {
3148 cc.Close()
3149 }
3150 }
3151 }
3152 pc.mutateHeaderFunc = nil
3153 }
3154
3155 func schemePort(scheme string) string {
3156 switch scheme {
3157 case "http":
3158 return "80"
3159 case "https":
3160 return "443"
3161 case "socks5", "socks5h":
3162 return "1080"
3163 default:
3164 return ""
3165 }
3166 }
3167
3168 func idnaASCIIFromURL(url *url.URL) string {
3169 addr := url.Hostname()
3170 if v, err := idnaASCII(addr); err == nil && v != "" {
3171 addr = v
3172 }
3173 return addr
3174 }
3175
3176
3177 func canonicalAddr(url *url.URL) string {
3178 port := url.Port()
3179 if port == "" {
3180 port = schemePort(url.Scheme)
3181 }
3182 return net.JoinHostPort(idnaASCIIFromURL(url), port)
3183 }
3184
3185
3186
3187
3188
3189
3190
3191
3192
3193
3194
3195
3196 type bodyEOFSignal struct {
3197 body io.ReadCloser
3198 mu sync.Mutex
3199 closed bool
3200 rerr error
3201 fn func(error) error
3202 earlyCloseFn func() error
3203 }
3204
3205 var errReadOnClosedResBody = errors.New("http: read on closed response body")
3206 var errConcurrentReadOnResBody = errors.New("http: concurrent read on response body")
3207
3208 func (es *bodyEOFSignal) Read(p []byte) (n int, err error) {
3209 es.mu.Lock()
3210 closed, rerr := es.closed, es.rerr
3211 es.mu.Unlock()
3212 if closed {
3213 return 0, errReadOnClosedResBody
3214 }
3215 if rerr != nil {
3216 return 0, rerr
3217 }
3218
3219 n, err = es.body.Read(p)
3220 if err != nil {
3221 es.mu.Lock()
3222 defer es.mu.Unlock()
3223 if es.rerr == nil {
3224 es.rerr = err
3225 }
3226 err = es.condfn(err)
3227 }
3228 return
3229 }
3230
3231 func (es *bodyEOFSignal) Close() error {
3232 es.mu.Lock()
3233 defer es.mu.Unlock()
3234 if es.closed {
3235 return nil
3236 }
3237 es.closed = true
3238 if es.earlyCloseFn != nil && es.rerr != io.EOF {
3239 return es.earlyCloseFn()
3240 }
3241 err := es.body.Close()
3242 return es.condfn(err)
3243 }
3244
3245
3246 func (es *bodyEOFSignal) condfn(err error) error {
3247 if es.fn == nil {
3248 return err
3249 }
3250 err = es.fn(err)
3251 es.fn = nil
3252 return err
3253 }
3254
3255
3256
3257
3258
3259 type gzipReader struct {
3260 _ incomparable
3261 body *bodyEOFSignal
3262 mu sync.Mutex
3263 zr *gzip.Reader
3264 zerr error
3265 }
3266
3267 type eofReader struct{}
3268
3269 func (eofReader) Read([]byte) (int, error) { return 0, io.EOF }
3270 func (eofReader) ReadByte() (byte, error) { return 0, io.EOF }
3271
3272 var gzipPool = sync.Pool{New: func() any { return new(gzip.Reader) }}
3273
3274
3275 func gzipPoolGet(r io.Reader) (*gzip.Reader, error) {
3276 zr := gzipPool.Get().(*gzip.Reader)
3277 if err := zr.Reset(r); err != nil {
3278 gzipPoolPut(zr)
3279 return nil, err
3280 }
3281 return zr, nil
3282 }
3283
3284
3285 func gzipPoolPut(zr *gzip.Reader) {
3286
3287
3288 var r flate.Reader = eofReader{}
3289 zr.Reset(r)
3290 gzipPool.Put(zr)
3291 }
3292
3293
3294
3295 func (gz *gzipReader) acquire() (*gzip.Reader, error) {
3296 gz.mu.Lock()
3297 defer gz.mu.Unlock()
3298 if gz.zerr != nil {
3299 return nil, gz.zerr
3300 }
3301 if gz.zr == nil {
3302
3303
3304
3305
3306 gz.zerr = errConcurrentReadOnResBody
3307 gz.mu.Unlock()
3308 zr, err := gzipPoolGet(gz.body)
3309 gz.mu.Lock()
3310
3311 if gz.zerr != errConcurrentReadOnResBody {
3312 if zr != nil {
3313 gzipPoolPut(zr)
3314 }
3315 return nil, gz.zerr
3316 }
3317 gz.zr, gz.zerr = zr, err
3318 if gz.zerr != nil {
3319 return nil, gz.zerr
3320 }
3321 }
3322 ret := gz.zr
3323 gz.zr, gz.zerr = nil, errConcurrentReadOnResBody
3324 return ret, nil
3325 }
3326
3327
3328 func (gz *gzipReader) release(zr *gzip.Reader) {
3329 gz.mu.Lock()
3330 defer gz.mu.Unlock()
3331 if gz.zerr == errConcurrentReadOnResBody {
3332 gz.zr, gz.zerr = zr, nil
3333 } else {
3334 gzipPoolPut(zr)
3335 }
3336 }
3337
3338
3339
3340 func (gz *gzipReader) close() {
3341 gz.mu.Lock()
3342 defer gz.mu.Unlock()
3343 if gz.zerr == nil && gz.zr != nil {
3344 gzipPoolPut(gz.zr)
3345 gz.zr = nil
3346 }
3347 gz.zerr = errReadOnClosedResBody
3348 }
3349
3350 func (gz *gzipReader) Read(p []byte) (n int, err error) {
3351 zr, err := gz.acquire()
3352 if err != nil {
3353 return 0, err
3354 }
3355 defer gz.release(zr)
3356
3357 return zr.Read(p)
3358 }
3359
3360 func (gz *gzipReader) Close() error {
3361 gz.close()
3362
3363 return gz.body.Close()
3364 }
3365
3366 type tlsHandshakeTimeoutError struct{}
3367
3368 func (tlsHandshakeTimeoutError) Timeout() bool { return true }
3369 func (tlsHandshakeTimeoutError) Temporary() bool { return true }
3370 func (tlsHandshakeTimeoutError) Error() string { return "net/http: TLS handshake timeout" }
3371
3372
3373
3374
3375 type fakeLocker struct{}
3376
3377 func (fakeLocker) Lock() {}
3378 func (fakeLocker) Unlock() {}
3379
3380
3381
3382
3383
3384
3385
3386
3387
3388
3389
3390
3391
3392
3393 func cloneTLSConfig(cfg *tls.Config) *tls.Config {
3394 if cfg == nil {
3395 return &tls.Config{}
3396 }
3397 return cfg.Clone()
3398 }
3399
3400 type connLRU struct {
3401 ll *list.List
3402 m map[*persistConn]*list.Element
3403 }
3404
3405
3406 func (cl *connLRU) add(pc *persistConn) {
3407 if cl.ll == nil {
3408 cl.ll = list.New()
3409 cl.m = make(map[*persistConn]*list.Element)
3410 }
3411 ele := cl.ll.PushFront(pc)
3412 if _, ok := cl.m[pc]; ok {
3413 panic("persistConn was already in LRU")
3414 }
3415 cl.m[pc] = ele
3416 }
3417
3418 func (cl *connLRU) removeOldest() *persistConn {
3419 ele := cl.ll.Back()
3420 pc := ele.Value.(*persistConn)
3421 cl.ll.Remove(ele)
3422 delete(cl.m, pc)
3423 return pc
3424 }
3425
3426
3427 func (cl *connLRU) remove(pc *persistConn) {
3428 if ele, ok := cl.m[pc]; ok {
3429 cl.ll.Remove(ele)
3430 delete(cl.m, pc)
3431 }
3432 }
3433
3434
3435 func (cl *connLRU) len() int {
3436 return len(cl.m)
3437 }
3438
View as plain text