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  	switch scheme {
   117  	case "http", "https":
   118  	default:
   119  		return nil, fmt.Errorf("net/http: invalid scheme %q", scheme)
   120  	}
   121  
   122  	host, port, err := net.SplitHostPort(address)
   123  	if err != nil {
   124  		return nil, err
   125  	}
   126  	if port == "" {
   127  		port = schemePort(scheme)
   128  	}
   129  
   130  	var proxyURL *url.URL
   131  	if t.Proxy != nil {
   132  		// Transport.Proxy takes a *Request, so create a fake one to pass it.
   133  		req := &Request{
   134  			ctx:    ctx,
   135  			Method: "GET",
   136  			URL: &url.URL{
   137  				Scheme: scheme,
   138  				Host:   host,
   139  				Path:   "/",
   140  			},
   141  			Proto:      "HTTP/1.1",
   142  			ProtoMajor: 1,
   143  			ProtoMinor: 1,
   144  			Header:     make(Header),
   145  			Body:       NoBody,
   146  			Host:       host,
   147  		}
   148  		var err error
   149  		proxyURL, err = t.Proxy(req)
   150  		if err != nil {
   151  			return nil, err
   152  		}
   153  	}
   154  
   155  	cm := connectMethod{
   156  		targetScheme: scheme,
   157  		targetAddr:   net.JoinHostPort(host, port),
   158  		proxyURL:     proxyURL,
   159  	}
   160  
   161  	// The state hook is a bit tricky:
   162  	// The persistConn has a state hook which calls ClientConn.maybeRunStateHook,
   163  	// which in turn calls the user-provided state hook (if any).
   164  	//
   165  	// ClientConn.maybeRunStateHook handles debouncing hook calls for both
   166  	// HTTP/1 and HTTP/2.
   167  	//
   168  	// Since there's no need to change the persistConn's hook, we set it at creation time.
   169  	cc := &ClientConn{}
   170  	const isClientConn = true
   171  	pconn, err := t.dialConn(ctx, cm, isClientConn, cc.maybeRunStateHook)
   172  	if err != nil {
   173  		return nil, err
   174  	}
   175  
   176  	// Note that cc.maybeRunStateHook may have been called
   177  	// in the short window between dialConn and now.
   178  	// This is fine.
   179  	cc.stateHookMu.Lock()
   180  	defer cc.stateHookMu.Unlock()
   181  	if pconn.alt != nil {
   182  		// If pconn.alt is set, this is a connection implemented in another package
   183  		// (probably x/net/http2) or the bundled copy in h2_bundle.go.
   184  		gc, ok := pconn.alt.(genericClientConn)
   185  		if !ok {
   186  			return nil, errors.New("http: NewClientConn returned something that is not a ClientConn")
   187  		}
   188  		cc.cc = gc
   189  		cc.lastAvailable = gc.Available()
   190  	} else {
   191  		// This is an HTTP/1 connection.
   192  		pconn.availch = make(chan struct{}, 1)
   193  		pconn.availch <- struct{}{}
   194  		cc.cc = http1ClientConn{pconn}
   195  		cc.lastAvailable = 1
   196  	}
   197  	return cc, nil
   198  }
   199  
   200  // Close closes the connection.
   201  // Outstanding RoundTrip calls are interrupted.
   202  func (cc *ClientConn) Close() error {
   203  	defer cc.maybeRunStateHook()
   204  	return cc.cc.Close()
   205  }
   206  
   207  // Err reports any fatal connection errors.
   208  // It returns nil if the connection is usable.
   209  // If it returns non-nil, the connection can no longer be used.
   210  func (cc *ClientConn) Err() error {
   211  	return cc.cc.Err()
   212  }
   213  
   214  func validateClientConnRequest(req *Request) error {
   215  	if req.URL == nil {
   216  		return errors.New("http: nil Request.URL")
   217  	}
   218  	if req.Header == nil {
   219  		return errors.New("http: nil Request.Header")
   220  	}
   221  	// Validate the outgoing headers.
   222  	if err := validateHeaders(req.Header); err != "" {
   223  		return fmt.Errorf("http: invalid header %s", err)
   224  	}
   225  	// Validate the outgoing trailers too.
   226  	if err := validateHeaders(req.Trailer); err != "" {
   227  		return fmt.Errorf("http: invalid trailer %s", err)
   228  	}
   229  	if req.Method != "" && !validMethod(req.Method) {
   230  		return fmt.Errorf("http: invalid method %q", req.Method)
   231  	}
   232  	if req.URL.Host == "" {
   233  		return errors.New("http: no Host in request URL")
   234  	}
   235  	return nil
   236  }
   237  
   238  // RoundTrip implements the [RoundTripper] interface.
   239  //
   240  // The request is sent on the client connection,
   241  // regardless of the URL being requested or any proxy settings.
   242  //
   243  // If the connection is at its concurrency limit,
   244  // RoundTrip waits for the connection to become available
   245  // before sending the request.
   246  func (cc *ClientConn) RoundTrip(req *Request) (*Response, error) {
   247  	defer cc.maybeRunStateHook()
   248  	if err := validateClientConnRequest(req); err != nil {
   249  		cc.Release()
   250  		return nil, err
   251  	}
   252  	return cc.cc.RoundTrip(req)
   253  }
   254  
   255  // Available reports the number of requests that may be sent
   256  // to the connection without blocking.
   257  // It returns 0 if the connection is closed.
   258  func (cc *ClientConn) Available() int {
   259  	return cc.cc.Available()
   260  }
   261  
   262  // InFlight reports the number of requests in flight,
   263  // including reserved requests.
   264  // It returns 0 if the connection is closed.
   265  func (cc *ClientConn) InFlight() int {
   266  	return cc.cc.InFlight()
   267  }
   268  
   269  // Reserve reserves a concurrency slot on the connection.
   270  // If Reserve returns nil, one additional RoundTrip call may be made
   271  // without waiting for an existing request to complete.
   272  //
   273  // The reserved concurrency slot is accounted as an in-flight request.
   274  // A successful call to RoundTrip will decrement the Available count
   275  // and increment the InFlight count.
   276  //
   277  // Each successful call to Reserve should be followed by exactly one call
   278  // to RoundTrip or Release, which will consume or release the reservation.
   279  //
   280  // If the connection is closed or at its concurrency limit,
   281  // Reserve returns an error.
   282  func (cc *ClientConn) Reserve() error {
   283  	defer cc.maybeRunStateHook()
   284  	return cc.cc.Reserve()
   285  }
   286  
   287  // Release releases an unused concurrency slot reserved by Reserve.
   288  // If there are no reserved concurrency slots, it has no effect.
   289  func (cc *ClientConn) Release() {
   290  	defer cc.maybeRunStateHook()
   291  	cc.cc.Release()
   292  }
   293  
   294  // shouldRunStateHook returns the user's state hook if we should call it,
   295  // or nil if we don't need to call it at this time.
   296  func (cc *ClientConn) shouldRunStateHook(stopRunning bool) func(*ClientConn) {
   297  	cc.stateHookMu.Lock()
   298  	defer cc.stateHookMu.Unlock()
   299  	if cc.cc == nil {
   300  		return nil
   301  	}
   302  	if stopRunning {
   303  		cc.stateHookRunning = false
   304  	}
   305  	if cc.userStateHook == nil {
   306  		return nil
   307  	}
   308  	if cc.stateHookRunning {
   309  		return nil
   310  	}
   311  	var (
   312  		available = cc.Available()
   313  		inFlight  = cc.InFlight()
   314  		closed    = cc.Err() != nil
   315  	)
   316  	var hook func(*ClientConn)
   317  	if available > cc.lastAvailable || inFlight < cc.lastInFlight || closed != cc.lastClosed {
   318  		hook = cc.userStateHook
   319  		cc.stateHookRunning = true
   320  	}
   321  	cc.lastAvailable = available
   322  	cc.lastInFlight = inFlight
   323  	cc.lastClosed = closed
   324  	return hook
   325  }
   326  
   327  func (cc *ClientConn) maybeRunStateHook() {
   328  	hook := cc.shouldRunStateHook(false)
   329  	if hook == nil {
   330  		return
   331  	}
   332  	// Run the hook synchronously.
   333  	//
   334  	// This means that if, for example, the user calls resp.Body.Close to finish a request,
   335  	// the Close call will synchronously run the hook, giving the hook the chance to
   336  	// return the ClientConn to a connection pool before the next request is made.
   337  	hook(cc)
   338  	// The connection state may have changed while the hook was running,
   339  	// in which case we need to run it again.
   340  	//
   341  	// If we do need to run the hook again, do so in a new goroutine to avoid blocking
   342  	// the current goroutine indefinitely.
   343  	hook = cc.shouldRunStateHook(true)
   344  	if hook != nil {
   345  		go func() {
   346  			for hook != nil {
   347  				hook(cc)
   348  				hook = cc.shouldRunStateHook(true)
   349  			}
   350  		}()
   351  	}
   352  }
   353  
   354  // SetStateHook arranges for f to be called when the state of the connection changes.
   355  // At most one call to f is made at a time.
   356  // If the connection's state has changed since it was created,
   357  // f is called immediately in a separate goroutine.
   358  // f may be called synchronously from RoundTrip or Response.Body.Close.
   359  //
   360  // If SetStateHook is called multiple times, the new hook replaces the old one.
   361  // If f is nil, no further calls will be made to f after SetStateHook returns.
   362  //
   363  // f is called when Available increases (more requests may be sent on the connection),
   364  // InFlight decreases (existing requests complete), or Err begins returning non-nil
   365  // (the connection is no longer usable).
   366  func (cc *ClientConn) SetStateHook(f func(*ClientConn)) {
   367  	cc.stateHookMu.Lock()
   368  	cc.userStateHook = f
   369  	cc.stateHookMu.Unlock()
   370  	cc.maybeRunStateHook()
   371  }
   372  
   373  // http1ClientConn is a genericClientConn implementation backed by
   374  // an HTTP/1 *persistConn (pconn.alt is nil).
   375  type http1ClientConn struct {
   376  	pconn *persistConn
   377  }
   378  
   379  func (cc http1ClientConn) RoundTrip(req *Request) (*Response, error) {
   380  	ctx := req.Context()
   381  	trace := httptrace.ContextClientTrace(ctx)
   382  
   383  	// Convert Request.Cancel into context cancelation.
   384  	ctx, cancel := context.WithCancelCause(req.Context())
   385  	if req.Cancel != nil {
   386  		go awaitLegacyCancel(ctx, cancel, req)
   387  	}
   388  
   389  	treq := &transportRequest{Request: req, trace: trace, ctx: ctx, cancel: cancel}
   390  	resp, err := cc.pconn.roundTrip(treq)
   391  	if err != nil {
   392  		return nil, err
   393  	}
   394  	resp.Request = req
   395  	return resp, nil
   396  }
   397  
   398  func (cc http1ClientConn) Close() error {
   399  	cc.pconn.close(errors.New("ClientConn closed"))
   400  	return nil
   401  }
   402  
   403  func (cc http1ClientConn) Err() error {
   404  	select {
   405  	case <-cc.pconn.closech:
   406  		return cc.pconn.closed
   407  	default:
   408  		return nil
   409  	}
   410  }
   411  
   412  func (cc http1ClientConn) Available() int {
   413  	cc.pconn.mu.Lock()
   414  	defer cc.pconn.mu.Unlock()
   415  	if cc.pconn.closed != nil || cc.pconn.reserved || cc.pconn.inFlight {
   416  		return 0
   417  	}
   418  	return 1
   419  }
   420  
   421  func (cc http1ClientConn) InFlight() int {
   422  	cc.pconn.mu.Lock()
   423  	defer cc.pconn.mu.Unlock()
   424  	if cc.pconn.closed == nil && (cc.pconn.reserved || cc.pconn.inFlight) {
   425  		return 1
   426  	}
   427  	return 0
   428  }
   429  
   430  func (cc http1ClientConn) Reserve() error {
   431  	cc.pconn.mu.Lock()
   432  	defer cc.pconn.mu.Unlock()
   433  	if cc.pconn.closed != nil {
   434  		return cc.pconn.closed
   435  	}
   436  	select {
   437  	case <-cc.pconn.availch:
   438  	default:
   439  		return errors.New("connection is unavailable")
   440  	}
   441  	cc.pconn.reserved = true
   442  	return nil
   443  }
   444  
   445  func (cc http1ClientConn) Release() {
   446  	cc.pconn.mu.Lock()
   447  	defer cc.pconn.mu.Unlock()
   448  	if cc.pconn.reserved {
   449  		select {
   450  		case cc.pconn.availch <- struct{}{}:
   451  		default:
   452  			panic("cannot release reservation")
   453  		}
   454  		cc.pconn.reserved = false
   455  	}
   456  }
   457
View as plain text