clientconn.go

     1  // Copyright 2025 The Go Authors. All rights reserved.
     2  // Use of this source code is governed by a BSD-style
     3  // license that can be found in the LICENSE file.
     4  
     5  package http
     6  
     7  import (
     8  	"context"
     9  	"errors"
    10  	"fmt"
    11  	"net"
    12  	"net/http/httptrace"
    13  	"net/url"
    14  	"sync"
    15  )
    16  
    17  // A ClientConn is a client connection to an HTTP server.
    18  //
    19  // Unlike a [Transport], a ClientConn represents a single connection.
    20  // Most users should use a Transport rather than creating client connections directly.
    21  type ClientConn struct {
    22  	cc genericClientConn
    23  
    24  	stateHookMu      sync.Mutex
    25  	userStateHook    func(*ClientConn)
    26  	stateHookRunning bool
    27  	lastAvailable    int
    28  	lastInFlight     int
    29  	lastClosed       bool
    30  }
    31  
    32  // newClientConner is the interface implemented by HTTP/2 transports to create new client conns.
    33  //
    34  // The http package (this package) needs a way to ask the http2 package to
    35  // create a client connection.
    36  //
    37  // Transport.TLSNextProto["h2"] contains a function which appears to do this,
    38  // but for historical reasons it does not: The TLSNextProto function adds a
    39  // *tls.Conn to the http2.Transport's connection pool and returns a RoundTripper
    40  // which is backed by that connection pool. NewClientConn needs a way to get a
    41  // single client connection out of the http2 package.
    42  //
    43  // The http2 package registers a RoundTripper with Transport.RegisterProtocol.
    44  // If this RoundTripper implements newClientConner, then Transport.NewClientConn will use
    45  // it to create new HTTP/2 client connections.
    46  type newClientConner interface {
    47  	// NewClientConn creates a new client connection from a net.Conn.
    48  	//
    49  	// The RoundTripper returned by NewClientConn must implement genericClientConn.
    50  	// (We don't define NewClientConn as returning genericClientConn,
    51  	// because either we'd need to make genericClientConn an exported type
    52  	// or define it as a type alias. Neither is particularly appealing.)
    53  	//
    54  	// The state hook passed here is the internal state hook
    55  	// (ClientConn.maybeRunStateHook). The internal state hook calls
    56  	// the user state hook (if any), which is set by the user with
    57  	// ClientConn.SetStateHook.
    58  	//
    59  	// The client connection should arrange to call the internal state hook
    60  	// when the connection closes, when requests complete, and when the
    61  	// connection concurrency limit changes.
    62  	//
    63  	// The client connection must call the internal state hook when the connection state
    64  	// changes asynchronously, such as when a request completes.
    65  	//
    66  	// The internal state hook need not be called after synchronous changes to the state:
    67  	// Close, Reserve, Release, and RoundTrip calls which don't start a request
    68  	// do not need to call the hook.
    69  	//
    70  	// The general idea is that if we call (for example) Close,
    71  	// we know that the connection state has probably changed and we
    72  	// don't need the state hook to tell us that.
    73  	// However, if the connection closes asynchronously
    74  	// (because, for example, the other end of the conn closed it),
    75  	// the state hook needs to inform us.
    76  	NewClientConn(nc net.Conn, internalStateHook func()) (RoundTripper, error)
    77  }
    78  
    79  // genericClientConn is an interface implemented by HTTP/2 client conns
    80  // returned from newClientConner.NewClientConn.
    81  //
    82  // See the newClientConner doc comment for more information.
    83  type genericClientConn interface {
    84  	Close() error
    85  	Err() error
    86  	RoundTrip(req *Request) (*Response, error)
    87  	Reserve() error
    88  	Release()
    89  	Available() int
    90  	InFlight() int
    91  }
    92  
    93  // NewClientConn creates a new client connection to the given address.
    94  //
    95  // If scheme is "http", the connection is unencrypted.
    96  // If scheme is "https", the connection uses TLS.
    97  //
    98  // The protocol used for the new connection is determined by the scheme,
    99  // Transport.Protocols configuration field, and protocols supported by the
   100  // server. See Transport.Protocols for more details.
   101  //
   102  // If Transport.Proxy is set and indicates that a request sent to the given
   103  // address should use a proxy, the new connection uses that proxy.
   104  //
   105  // NewClientConn always creates a new connection,
   106  // even if the Transport has an existing cached connection to the given host.
   107  //
   108  // The new connection is not added to the Transport's connection cache,
   109  // and will not be used by [Transport.RoundTrip].
   110  // It does not count against the MaxIdleConns and MaxConnsPerHost limits.
   111  //
   112  // The caller is responsible for closing the new connection.
   113  func (t *Transport) NewClientConn(ctx context.Context, scheme, address string) (*ClientConn, error) {
   114  	t.nextProtoOnce.Do(t.onceSetNextProtoDefaults)
   115  
   116  	if t.h2Config != nil {
   117  		// Handle x/net/http2.Transport.NewClientConn passing us a net.Conn
   118  		// to create a ClientConn from.
   119  		if cc, err := t.http2NewClientConnFromContext(ctx); err != errors.ErrUnsupported {
   120  			return cc, err
   121  		}
   122  	}
   123  
   124  	switch scheme {
   125  	case "http", "https":
   126  	default:
   127  		return nil, fmt.Errorf("net/http: invalid scheme %q", scheme)
   128  	}
   129  
   130  	host, port, err := net.SplitHostPort(address)
   131  	if err != nil {
   132  		return nil, err
   133  	}
   134  	if port == "" {
   135  		port = schemePort(scheme)
   136  	}
   137  
   138  	var proxyURL *url.URL
   139  	if t.Proxy != nil {
   140  		// Transport.Proxy takes a *Request, so create a fake one to pass it.
   141  		req := &Request{
   142  			ctx:    ctx,
   143  			Method: "GET",
   144  			URL: &url.URL{
   145  				Scheme: scheme,
   146  				Host:   host,
   147  				Path:   "/",
   148  			},
   149  			Proto:      "HTTP/1.1",
   150  			ProtoMajor: 1,
   151  			ProtoMinor: 1,
   152  			Header:     make(Header),
   153  			Body:       NoBody,
   154  			Host:       host,
   155  		}
   156  		var err error
   157  		proxyURL, err = t.Proxy(req)
   158  		if err != nil {
   159  			return nil, err
   160  		}
   161  	}
   162  
   163  	cm := connectMethod{
   164  		targetScheme: scheme,
   165  		targetAddr:   net.JoinHostPort(host, port),
   166  		proxyURL:     proxyURL,
   167  	}
   168  
   169  	// The state hook is a bit tricky:
   170  	// The persistConn has a state hook which calls ClientConn.maybeRunStateHook,
   171  	// which in turn calls the user-provided state hook (if any).
   172  	//
   173  	// ClientConn.maybeRunStateHook handles debouncing hook calls for both
   174  	// HTTP/1 and HTTP/2.
   175  	//
   176  	// Since there's no need to change the persistConn's hook, we set it at creation time.
   177  	cc := &ClientConn{}
   178  	const isClientConn = true
   179  	pconn, err := t.dialConn(ctx, cm, isClientConn, cc.maybeRunStateHook)
   180  	if err != nil {
   181  		return nil, err
   182  	}
   183  
   184  	// Note that cc.maybeRunStateHook may have been called
   185  	// in the short window between dialConn and now.
   186  	// This is fine.
   187  	cc.stateHookMu.Lock()
   188  	defer cc.stateHookMu.Unlock()
   189  	if pconn.alt != nil {
   190  		// If pconn.alt is set, this is a connection implemented in another package
   191  		// (probably x/net/http2) or the bundled copy in h2_bundle.go.
   192  		gc, ok := pconn.alt.(genericClientConn)
   193  		if !ok {
   194  			return nil, errors.New("http: NewClientConn returned something that is not a ClientConn")
   195  		}
   196  		cc.cc = gc
   197  		cc.lastAvailable = gc.Available()
   198  	} else {
   199  		// This is an HTTP/1 connection.
   200  		pconn.availch = make(chan struct{}, 1)
   201  		pconn.availch <- struct{}{}
   202  		cc.cc = http1ClientConn{pconn}
   203  		cc.lastAvailable = 1
   204  	}
   205  	return cc, nil
   206  }
   207  
   208  // Close closes the connection.
   209  // Outstanding RoundTrip calls are interrupted.
   210  func (cc *ClientConn) Close() error {
   211  	defer cc.maybeRunStateHook()
   212  	return cc.cc.Close()
   213  }
   214  
   215  // Err reports any fatal connection errors.
   216  // It returns nil if the connection is usable.
   217  // If it returns non-nil, the connection can no longer be used.
   218  func (cc *ClientConn) Err() error {
   219  	return cc.cc.Err()
   220  }
   221  
   222  func validateClientConnRequest(req *Request) error {
   223  	if req.URL == nil {
   224  		return errors.New("http: nil Request.URL")
   225  	}
   226  	if req.Header == nil {
   227  		return errors.New("http: nil Request.Header")
   228  	}
   229  	// Validate the outgoing headers.
   230  	if err := validateHeaders(req.Header); err != "" {
   231  		return fmt.Errorf("http: invalid header %s", err)
   232  	}
   233  	// Validate the outgoing trailers too.
   234  	if err := validateHeaders(req.Trailer); err != "" {
   235  		return fmt.Errorf("http: invalid trailer %s", err)
   236  	}
   237  	if req.Method != "" && !validMethod(req.Method) {
   238  		return fmt.Errorf("http: invalid method %q", req.Method)
   239  	}
   240  	if req.URL.Host == "" {
   241  		return errors.New("http: no Host in request URL")
   242  	}
   243  	return nil
   244  }
   245  
   246  // RoundTrip implements the [RoundTripper] interface.
   247  //
   248  // The request is sent on the client connection,
   249  // regardless of the URL being requested or any proxy settings.
   250  //
   251  // If the connection is at its concurrency limit,
   252  // RoundTrip waits for the connection to become available
   253  // before sending the request.
   254  func (cc *ClientConn) RoundTrip(req *Request) (*Response, error) {
   255  	defer cc.maybeRunStateHook()
   256  	if req.URL == nil && req.Method == ":ping" {
   257  		// Undocumented feature for sending a PING frame to a HTTP/2 connection,
   258  		// included to support x/net/http2.ClientConn.Ping.
   259  		pinger, ok := cc.cc.(interface {
   260  			Ping(context.Context) error
   261  		})
   262  		if !ok {
   263  			return nil, errors.New("http: ClientConn does not support PING")
   264  		}
   265  		return nil, pinger.Ping(req.Context())
   266  	}
   267  	if err := validateClientConnRequest(req); err != nil {
   268  		cc.Release()
   269  		return nil, err
   270  	}
   271  	return cc.cc.RoundTrip(req)
   272  }
   273  
   274  // Available reports the number of requests that may be sent
   275  // to the connection without blocking.
   276  // It returns 0 if the connection is closed.
   277  func (cc *ClientConn) Available() int {
   278  	return cc.cc.Available()
   279  }
   280  
   281  // InFlight reports the number of requests in flight,
   282  // including reserved requests.
   283  // It returns 0 if the connection is closed.
   284  func (cc *ClientConn) InFlight() int {
   285  	return cc.cc.InFlight()
   286  }
   287  
   288  // Reserve reserves a concurrency slot on the connection.
   289  // If Reserve returns nil, one additional RoundTrip call may be made
   290  // without waiting for an existing request to complete.
   291  //
   292  // The reserved concurrency slot is accounted as an in-flight request.
   293  // A successful call to RoundTrip will decrement the Available count
   294  // and increment the InFlight count.
   295  //
   296  // Each successful call to Reserve should be followed by exactly one call
   297  // to RoundTrip or Release, which will consume or release the reservation.
   298  //
   299  // If the connection is closed or at its concurrency limit,
   300  // Reserve returns an error.
   301  func (cc *ClientConn) Reserve() error {
   302  	defer cc.maybeRunStateHook()
   303  	return cc.cc.Reserve()
   304  }
   305  
   306  // Release releases an unused concurrency slot reserved by Reserve.
   307  // If there are no reserved concurrency slots, it has no effect.
   308  func (cc *ClientConn) Release() {
   309  	defer cc.maybeRunStateHook()
   310  	cc.cc.Release()
   311  }
   312  
   313  // shouldRunStateHook returns the user's state hook if we should call it,
   314  // or nil if we don't need to call it at this time.
   315  func (cc *ClientConn) shouldRunStateHook(stopRunning bool) func(*ClientConn) {
   316  	cc.stateHookMu.Lock()
   317  	defer cc.stateHookMu.Unlock()
   318  	if cc.cc == nil {
   319  		return nil
   320  	}
   321  	if stopRunning {
   322  		cc.stateHookRunning = false
   323  	}
   324  	if cc.userStateHook == nil {
   325  		return nil
   326  	}
   327  	if cc.stateHookRunning {
   328  		return nil
   329  	}
   330  	var (
   331  		available = cc.Available()
   332  		inFlight  = cc.InFlight()
   333  		closed    = cc.Err() != nil
   334  	)
   335  	var hook func(*ClientConn)
   336  	if available > cc.lastAvailable || inFlight < cc.lastInFlight || closed != cc.lastClosed {
   337  		hook = cc.userStateHook
   338  		cc.stateHookRunning = true
   339  	}
   340  	cc.lastAvailable = available
   341  	cc.lastInFlight = inFlight
   342  	cc.lastClosed = closed
   343  	return hook
   344  }
   345  
   346  func (cc *ClientConn) maybeRunStateHook() {
   347  	hook := cc.shouldRunStateHook(false)
   348  	if hook == nil {
   349  		return
   350  	}
   351  	// Run the hook synchronously.
   352  	//
   353  	// This means that if, for example, the user calls resp.Body.Close to finish a request,
   354  	// the Close call will synchronously run the hook, giving the hook the chance to
   355  	// return the ClientConn to a connection pool before the next request is made.
   356  	hook(cc)
   357  	// The connection state may have changed while the hook was running,
   358  	// in which case we need to run it again.
   359  	//
   360  	// If we do need to run the hook again, do so in a new goroutine to avoid blocking
   361  	// the current goroutine indefinitely.
   362  	hook = cc.shouldRunStateHook(true)
   363  	if hook != nil {
   364  		go func() {
   365  			for hook != nil {
   366  				hook(cc)
   367  				hook = cc.shouldRunStateHook(true)
   368  			}
   369  		}()
   370  	}
   371  }
   372  
   373  // SetStateHook arranges for f to be called when the state of the connection changes.
   374  // At most one call to f is made at a time.
   375  // If the connection's state has changed since it was created,
   376  // f is called immediately in a separate goroutine.
   377  // f may be called synchronously from RoundTrip or Response.Body.Close.
   378  //
   379  // If SetStateHook is called multiple times, the new hook replaces the old one.
   380  // If f is nil, no further calls will be made to f after SetStateHook returns.
   381  //
   382  // f is called when Available increases (more requests may be sent on the connection),
   383  // InFlight decreases (existing requests complete), or Err begins returning non-nil
   384  // (the connection is no longer usable).
   385  func (cc *ClientConn) SetStateHook(f func(*ClientConn)) {
   386  	cc.stateHookMu.Lock()
   387  	cc.userStateHook = f
   388  	cc.stateHookMu.Unlock()
   389  	cc.maybeRunStateHook()
   390  }
   391  
   392  // http1ClientConn is a genericClientConn implementation backed by
   393  // an HTTP/1 *persistConn (pconn.alt is nil).
   394  type http1ClientConn struct {
   395  	pconn *persistConn
   396  }
   397  
   398  func (cc http1ClientConn) RoundTrip(req *Request) (*Response, error) {
   399  	ctx := req.Context()
   400  	trace := httptrace.ContextClientTrace(ctx)
   401  
   402  	// Convert Request.Cancel into context cancellation.
   403  	ctx, cancel := context.WithCancelCause(req.Context())
   404  	if req.Cancel != nil {
   405  		go awaitLegacyCancel(ctx, cancel, req)
   406  	}
   407  
   408  	treq := &transportRequest{Request: req, trace: trace, ctx: ctx, cancel: cancel}
   409  	resp, err := cc.pconn.roundTrip(treq)
   410  	if err != nil {
   411  		return nil, err
   412  	}
   413  	resp.Request = req
   414  	return resp, nil
   415  }
   416  
   417  func (cc http1ClientConn) Close() error {
   418  	cc.pconn.close(errors.New("ClientConn closed"))
   419  	return nil
   420  }
   421  
   422  func (cc http1ClientConn) Err() error {
   423  	select {
   424  	case <-cc.pconn.closech:
   425  		return cc.pconn.closed
   426  	default:
   427  		return nil
   428  	}
   429  }
   430  
   431  func (cc http1ClientConn) Available() int {
   432  	cc.pconn.mu.Lock()
   433  	defer cc.pconn.mu.Unlock()
   434  	if cc.pconn.closed != nil || cc.pconn.reserved || cc.pconn.inFlight {
   435  		return 0
   436  	}
   437  	return 1
   438  }
   439  
   440  func (cc http1ClientConn) InFlight() int {
   441  	cc.pconn.mu.Lock()
   442  	defer cc.pconn.mu.Unlock()
   443  	if cc.pconn.closed == nil && (cc.pconn.reserved || cc.pconn.inFlight) {
   444  		return 1
   445  	}
   446  	return 0
   447  }
   448  
   449  func (cc http1ClientConn) Reserve() error {
   450  	cc.pconn.mu.Lock()
   451  	defer cc.pconn.mu.Unlock()
   452  	if cc.pconn.closed != nil {
   453  		return cc.pconn.closed
   454  	}
   455  	select {
   456  	case <-cc.pconn.availch:
   457  	default:
   458  		return errors.New("connection is unavailable")
   459  	}
   460  	cc.pconn.reserved = true
   461  	return nil
   462  }
   463  
   464  func (cc http1ClientConn) Release() {
   465  	cc.pconn.mu.Lock()
   466  	defer cc.pconn.mu.Unlock()
   467  	if cc.pconn.reserved {
   468  		select {
   469  		case cc.pconn.availch <- struct{}{}:
   470  		default:
   471  			panic("cannot release reservation")
   472  		}
   473  		cc.pconn.reserved = false
   474  	}
   475  }
   476
View as plain text