// Code generated by golang.org/x/tools/cmd/bundle. //go:generate bundle -o h2_bundle.go -prefix http2 -import golang.org/x/net/http2/hpack=internal/golang.org/x/net/http2/hpack golang.org/x/net/http2 // Package http2 implements the HTTP/2 protocol. // // This package is low-level and intended to be used directly by very // few people. Most users will use it indirectly through the automatic // use by the net/http package (from Go 1.6 and later). // For use in earlier Go versions see ConfigureServer. (Transport support // requires Go 1.6 or later) // // See https://http2.github.io/ for more information on HTTP/2. // // See https://http2.golang.org/ for a test server running this code. // package http import ( "bufio" "bytes" "compress/gzip" "crypto/tls" "encoding/binary" "errors" "fmt" "internal/golang.org/x/net/http2/hpack" "io" "io/ioutil" "log" "net" "net/textproto" "net/url" "os" "reflect" "runtime" "sort" "strconv" "strings" "sync" "time" ) // ClientConnPool manages a pool of HTTP/2 client connections. type http2ClientConnPool interface { GetClientConn(req *Request, addr string) (*http2ClientConn, error) MarkDead(*http2ClientConn) } // TODO: use singleflight for dialing and addConnCalls? type http2clientConnPool struct { t *http2Transport mu sync.Mutex // TODO: maybe switch to RWMutex // TODO: add support for sharing conns based on cert names // (e.g. share conn for googleapis.com and appspot.com) conns map[string][]*http2ClientConn // key is host:port dialing map[string]*http2dialCall // currently in-flight dials keys map[*http2ClientConn][]string addConnCalls map[string]*http2addConnCall // in-flight addConnIfNeede calls } func (p *http2clientConnPool) GetClientConn(req *Request, addr string) (*http2ClientConn, error) { return p.getClientConn(req, addr, http2dialOnMiss) } const ( http2dialOnMiss = true http2noDialOnMiss = false ) func (p *http2clientConnPool) getClientConn(_ *Request, addr string, dialOnMiss bool) (*http2ClientConn, error) { p.mu.Lock() for _, cc := range p.conns[addr] { if cc.CanTakeNewRequest() { p.mu.Unlock() return cc, nil } } if !dialOnMiss { p.mu.Unlock() return nil, http2ErrNoCachedConn } call := p.getStartDialLocked(addr) p.mu.Unlock() <-call.done return call.res, call.err } // dialCall is an in-flight Transport dial call to a host. type http2dialCall struct { p *http2clientConnPool done chan struct{} // closed when done res *http2ClientConn // valid after done is closed err error // valid after done is closed } // requires p.mu is held. func (p *http2clientConnPool) getStartDialLocked(addr string) *http2dialCall { if call, ok := p.dialing[addr]; ok { return call } call := &http2dialCall{p: p, done: make(chan struct{})} if p.dialing == nil { p.dialing = make(map[string]*http2dialCall) } p.dialing[addr] = call go call.dial(addr) return call } // run in its own goroutine. func (c *http2dialCall) dial(addr string) { c.res, c.err = c.p.t.dialClientConn(addr) close(c.done) c.p.mu.Lock() delete(c.p.dialing, addr) if c.err == nil { c.p.addConnLocked(addr, c.res) } c.p.mu.Unlock() } // addConnIfNeeded makes a NewClientConn out of c if a connection for key doesn't // already exist. It coalesces concurrent calls with the same key. // This is used by the http1 Transport code when it creates a new connection. Because // the http1 Transport doesn't de-dup TCP dials to outbound hosts (because it doesn't know // the protocol), it can get into a situation where it has multiple TLS connections. // This code decides which ones live or die. // The return value used is whether c was used. // c is never closed. func (p *http2clientConnPool) addConnIfNeeded(key string, t *http2Transport, c *tls.Conn) (used bool, err error) { p.mu.Lock() for _, cc := range p.conns[key] { if cc.CanTakeNewRequest() { p.mu.Unlock() return false, nil } } call, dup := p.addConnCalls[key] if !dup { if p.addConnCalls == nil { p.addConnCalls = make(map[string]*http2addConnCall) } call = &http2addConnCall{ p: p, done: make(chan struct{}), } p.addConnCalls[key] = call go call.run(t, key, c) } p.mu.Unlock() <-call.done if call.err != nil { return false, call.err } return !dup, nil } type http2addConnCall struct { p *http2clientConnPool done chan struct{} // closed when done err error } func (c *http2addConnCall) run(t *http2Transport, key string, tc *tls.Conn) { cc, err := t.NewClientConn(tc) p := c.p p.mu.Lock() if err != nil { c.err = err } else { p.addConnLocked(key, cc) } delete(p.addConnCalls, key) p.mu.Unlock() close(c.done) } func (p *http2clientConnPool) addConn(key string, cc *http2ClientConn) { p.mu.Lock() p.addConnLocked(key, cc) p.mu.Unlock() } // p.mu must be held func (p *http2clientConnPool) addConnLocked(key string, cc *http2ClientConn) { for _, v := range p.conns[key] { if v == cc { return } } if p.conns == nil { p.conns = make(map[string][]*http2ClientConn) } if p.keys == nil { p.keys = make(map[*http2ClientConn][]string) } p.conns[key] = append(p.conns[key], cc) p.keys[cc] = append(p.keys[cc], key) } func (p *http2clientConnPool) MarkDead(cc *http2ClientConn) { p.mu.Lock() defer p.mu.Unlock() for _, key := range p.keys[cc] { vv, ok := p.conns[key] if !ok { continue } newList := http2filterOutClientConn(vv, cc) if len(newList) > 0 { p.conns[key] = newList } else { delete(p.conns, key) } } delete(p.keys, cc) } func (p *http2clientConnPool) closeIdleConnections() { p.mu.Lock() defer p.mu.Unlock() for _, vv := range p.conns { for _, cc := range vv { cc.closeIfIdle() } } } func http2filterOutClientConn(in []*http2ClientConn, exclude *http2ClientConn) []*http2ClientConn { out := in[:0] for _, v := range in { if v != exclude { out = append(out, v) } } if len(in) != len(out) { in[len(in)-1] = nil } return out } func http2configureTransport(t1 *Transport) (*http2Transport, error) { connPool := new(http2clientConnPool) t2 := &http2Transport{ ConnPool: http2noDialClientConnPool{connPool}, t1: t1, } connPool.t = t2 if err := http2registerHTTPSProtocol(t1, http2noDialH2RoundTripper{t2}); err != nil { return nil, err } if t1.TLSClientConfig == nil { t1.TLSClientConfig = new(tls.Config) } if !http2strSliceContains(t1.TLSClientConfig.NextProtos, "h2") { t1.TLSClientConfig.NextProtos = append([]string{"h2"}, t1.TLSClientConfig.NextProtos...) } if !http2strSliceContains(t1.TLSClientConfig.NextProtos, "http/1.1") { t1.TLSClientConfig.NextProtos = append(t1.TLSClientConfig.NextProtos, "http/1.1") } upgradeFn := func(authority string, c *tls.Conn) RoundTripper { addr := http2authorityAddr(authority) if used, err := connPool.addConnIfNeeded(addr, t2, c); err != nil { go c.Close() return http2erringRoundTripper{err} } else if !used { go c.Close() } return t2 } if m := t1.TLSNextProto; len(m) == 0 { t1.TLSNextProto = map[string]func(string, *tls.Conn) RoundTripper{ "h2": upgradeFn, } } else { m["h2"] = upgradeFn } return t2, nil } // registerHTTPSProtocol calls Transport.RegisterProtocol but // convering panics into errors. func http2registerHTTPSProtocol(t *Transport, rt RoundTripper) (err error) { defer func() { if e := recover(); e != nil { err = fmt.Errorf("%v", e) } }() t.RegisterProtocol("https", rt) return nil } // noDialClientConnPool is an implementation of http2.ClientConnPool // which never dials. We let the HTTP/1.1 client dial and use its TLS // connection instead. type http2noDialClientConnPool struct{ *http2clientConnPool } func (p http2noDialClientConnPool) GetClientConn(req *Request, addr string) (*http2ClientConn, error) { return p.getClientConn(req, addr, http2noDialOnMiss) } // noDialH2RoundTripper is a RoundTripper which only tries to complete the request // if there's already has a cached connection to the host. type http2noDialH2RoundTripper struct{ t *http2Transport } func (rt http2noDialH2RoundTripper) RoundTrip(req *Request) (*Response, error) { res, err := rt.t.RoundTrip(req) if err == http2ErrNoCachedConn { return nil, ErrSkipAltProtocol } return res, err } // An ErrCode is an unsigned 32-bit error code as defined in the HTTP/2 spec. type http2ErrCode uint32 const ( http2ErrCodeNo http2ErrCode = 0x0 http2ErrCodeProtocol http2ErrCode = 0x1 http2ErrCodeInternal http2ErrCode = 0x2 http2ErrCodeFlowControl http2ErrCode = 0x3 http2ErrCodeSettingsTimeout http2ErrCode = 0x4 http2ErrCodeStreamClosed http2ErrCode = 0x5 http2ErrCodeFrameSize http2ErrCode = 0x6 http2ErrCodeRefusedStream http2ErrCode = 0x7 http2ErrCodeCancel http2ErrCode = 0x8 http2ErrCodeCompression http2ErrCode = 0x9 http2ErrCodeConnect http2ErrCode = 0xa http2ErrCodeEnhanceYourCalm http2ErrCode = 0xb http2ErrCodeInadequateSecurity http2ErrCode = 0xc http2ErrCodeHTTP11Required http2ErrCode = 0xd ) var http2errCodeName = map[http2ErrCode]string{ http2ErrCodeNo: "NO_ERROR", http2ErrCodeProtocol: "PROTOCOL_ERROR", http2ErrCodeInternal: "INTERNAL_ERROR", http2ErrCodeFlowControl: "FLOW_CONTROL_ERROR", http2ErrCodeSettingsTimeout: "SETTINGS_TIMEOUT", http2ErrCodeStreamClosed: "STREAM_CLOSED", http2ErrCodeFrameSize: "FRAME_SIZE_ERROR", http2ErrCodeRefusedStream: "REFUSED_STREAM", http2ErrCodeCancel: "CANCEL", http2ErrCodeCompression: "COMPRESSION_ERROR", http2ErrCodeConnect: "CONNECT_ERROR", http2ErrCodeEnhanceYourCalm: "ENHANCE_YOUR_CALM", http2ErrCodeInadequateSecurity: "INADEQUATE_SECURITY", http2ErrCodeHTTP11Required: "HTTP_1_1_REQUIRED", } func (e http2ErrCode) String() string { if s, ok := http2errCodeName[e]; ok { return s } return fmt.Sprintf("unknown error code 0x%x", uint32(e)) } // ConnectionError is an error that results in the termination of the // entire connection. type http2ConnectionError http2ErrCode func (e http2ConnectionError) Error() string { return fmt.Sprintf("connection error: %s", http2ErrCode(e)) } // StreamError is an error that only affects one stream within an // HTTP/2 connection. type http2StreamError struct { StreamID uint32 Code http2ErrCode } func (e http2StreamError) Error() string { return fmt.Sprintf("stream error: stream ID %d; %v", e.StreamID, e.Code) } // 6.9.1 The Flow Control Window // "If a sender receives a WINDOW_UPDATE that causes a flow control // window to exceed this maximum it MUST terminate either the stream // or the connection, as appropriate. For streams, [...]; for the // connection, a GOAWAY frame with a FLOW_CONTROL_ERROR code." type http2goAwayFlowError struct{} func (http2goAwayFlowError) Error() string { return "connection exceeded flow control window size" } // Errors of this type are only returned by the frame parser functions // and converted into ConnectionError(ErrCodeProtocol). type http2connError struct { Code http2ErrCode Reason string } func (e http2connError) Error() string { return fmt.Sprintf("http2: connection error: %v: %v", e.Code, e.Reason) } // fixedBuffer is an io.ReadWriter backed by a fixed size buffer. // It never allocates, but moves old data as new data is written. type http2fixedBuffer struct { buf []byte r, w int } var ( http2errReadEmpty = errors.New("read from empty fixedBuffer") http2errWriteFull = errors.New("write on full fixedBuffer") ) // Read copies bytes from the buffer into p. // It is an error to read when no data is available. func (b *http2fixedBuffer) Read(p []byte) (n int, err error) { if b.r == b.w { return 0, http2errReadEmpty } n = copy(p, b.buf[b.r:b.w]) b.r += n if b.r == b.w { b.r = 0 b.w = 0 } return n, nil } // Len returns the number of bytes of the unread portion of the buffer. func (b *http2fixedBuffer) Len() int { return b.w - b.r } // Write copies bytes from p into the buffer. // It is an error to write more data than the buffer can hold. func (b *http2fixedBuffer) Write(p []byte) (n int, err error) { if b.r > 0 && len(p) > len(b.buf)-b.w { copy(b.buf, b.buf[b.r:b.w]) b.w -= b.r b.r = 0 } n = copy(b.buf[b.w:], p) b.w += n if n < len(p) { err = http2errWriteFull } return n, err } // flow is the flow control window's size. type http2flow struct { // n is the number of DATA bytes we're allowed to send. // A flow is kept both on a conn and a per-stream. n int32 // conn points to the shared connection-level flow that is // shared by all streams on that conn. It is nil for the flow // that's on the conn directly. conn *http2flow } func (f *http2flow) setConnFlow(cf *http2flow) { f.conn = cf } func (f *http2flow) available() int32 { n := f.n if f.conn != nil && f.conn.n < n { n = f.conn.n } return n } func (f *http2flow) take(n int32) { if n > f.available() { panic("internal error: took too much") } f.n -= n if f.conn != nil { f.conn.n -= n } } // add adds n bytes (positive or negative) to the flow control window. // It returns false if the sum would exceed 2^31-1. func (f *http2flow) add(n int32) bool { remain := (1<<31 - 1) - f.n if n > remain { return false } f.n += n return true } const http2frameHeaderLen = 9 var http2padZeros = make([]byte, 255) // zeros for padding // A FrameType is a registered frame type as defined in // http://http2.github.io/http2-spec/#rfc.section.11.2 type http2FrameType uint8 const ( http2FrameData http2FrameType = 0x0 http2FrameHeaders http2FrameType = 0x1 http2FramePriority http2FrameType = 0x2 http2FrameRSTStream http2FrameType = 0x3 http2FrameSettings http2FrameType = 0x4 http2FramePushPromise http2FrameType = 0x5 http2FramePing http2FrameType = 0x6 http2FrameGoAway http2FrameType = 0x7 http2FrameWindowUpdate http2FrameType = 0x8 http2FrameContinuation http2FrameType = 0x9 ) var http2frameName = map[http2FrameType]string{ http2FrameData: "DATA", http2FrameHeaders: "HEADERS", http2FramePriority: "PRIORITY", http2FrameRSTStream: "RST_STREAM", http2FrameSettings: "SETTINGS", http2FramePushPromise: "PUSH_PROMISE", http2FramePing: "PING", http2FrameGoAway: "GOAWAY", http2FrameWindowUpdate: "WINDOW_UPDATE", http2FrameContinuation: "CONTINUATION", } func (t http2FrameType) String() string { if s, ok := http2frameName[t]; ok { return s } return fmt.Sprintf("UNKNOWN_FRAME_TYPE_%d", uint8(t)) } // Flags is a bitmask of HTTP/2 flags. // The meaning of flags varies depending on the frame type. type http2Flags uint8 // Has reports whether f contains all (0 or more) flags in v. func (f http2Flags) Has(v http2Flags) bool { return (f & v) == v } // Frame-specific FrameHeader flag bits. const ( // Data Frame http2FlagDataEndStream http2Flags = 0x1 http2FlagDataPadded http2Flags = 0x8 // Headers Frame http2FlagHeadersEndStream http2Flags = 0x1 http2FlagHeadersEndHeaders http2Flags = 0x4 http2FlagHeadersPadded http2Flags = 0x8 http2FlagHeadersPriority http2Flags = 0x20 // Settings Frame http2FlagSettingsAck http2Flags = 0x1 // Ping Frame http2FlagPingAck http2Flags = 0x1 // Continuation Frame http2FlagContinuationEndHeaders http2Flags = 0x4 http2FlagPushPromiseEndHeaders http2Flags = 0x4 http2FlagPushPromisePadded http2Flags = 0x8 ) var http2flagName = map[http2FrameType]map[http2Flags]string{ http2FrameData: { http2FlagDataEndStream: "END_STREAM", http2FlagDataPadded: "PADDED", }, http2FrameHeaders: { http2FlagHeadersEndStream: "END_STREAM", http2FlagHeadersEndHeaders: "END_HEADERS", http2FlagHeadersPadded: "PADDED", http2FlagHeadersPriority: "PRIORITY", }, http2FrameSettings: { http2FlagSettingsAck: "ACK", }, http2FramePing: { http2FlagPingAck: "ACK", }, http2FrameContinuation: { http2FlagContinuationEndHeaders: "END_HEADERS", }, http2FramePushPromise: { http2FlagPushPromiseEndHeaders: "END_HEADERS", http2FlagPushPromisePadded: "PADDED", }, } // a frameParser parses a frame given its FrameHeader and payload // bytes. The length of payload will always equal fh.Length (which // might be 0). type http2frameParser func(fh http2FrameHeader, payload []byte) (http2Frame, error) var http2frameParsers = map[http2FrameType]http2frameParser{ http2FrameData: http2parseDataFrame, http2FrameHeaders: http2parseHeadersFrame, http2FramePriority: http2parsePriorityFrame, http2FrameRSTStream: http2parseRSTStreamFrame, http2FrameSettings: http2parseSettingsFrame, http2FramePushPromise: http2parsePushPromise, http2FramePing: http2parsePingFrame, http2FrameGoAway: http2parseGoAwayFrame, http2FrameWindowUpdate: http2parseWindowUpdateFrame, http2FrameContinuation: http2parseContinuationFrame, } func http2typeFrameParser(t http2FrameType) http2frameParser { if f := http2frameParsers[t]; f != nil { return f } return http2parseUnknownFrame } // A FrameHeader is the 9 byte header of all HTTP/2 frames. // // See http://http2.github.io/http2-spec/#FrameHeader type http2FrameHeader struct { valid bool // caller can access []byte fields in the Frame // Type is the 1 byte frame type. There are ten standard frame // types, but extension frame types may be written by WriteRawFrame // and will be returned by ReadFrame (as UnknownFrame). Type http2FrameType // Flags are the 1 byte of 8 potential bit flags per frame. // They are specific to the frame type. Flags http2Flags // Length is the length of the frame, not including the 9 byte header. // The maximum size is one byte less than 16MB (uint24), but only // frames up to 16KB are allowed without peer agreement. Length uint32 // StreamID is which stream this frame is for. Certain frames // are not stream-specific, in which case this field is 0. StreamID uint32 } // Header returns h. It exists so FrameHeaders can be embedded in other // specific frame types and implement the Frame interface. func (h http2FrameHeader) Header() http2FrameHeader { return h } func (h http2FrameHeader) String() string { var buf bytes.Buffer buf.WriteString("[FrameHeader ") h.writeDebug(&buf) buf.WriteByte(']') return buf.String() } func (h http2FrameHeader) writeDebug(buf *bytes.Buffer) { buf.WriteString(h.Type.String()) if h.Flags != 0 { buf.WriteString(" flags=") set := 0 for i := uint8(0); i < 8; i++ { if h.Flags&(1< 1 { buf.WriteByte('|') } name := http2flagName[h.Type][http2Flags(1<>24), byte(streamID>>16), byte(streamID>>8), byte(streamID)) } func (f *http2Framer) endWrite() error { length := len(f.wbuf) - http2frameHeaderLen if length >= (1 << 24) { return http2ErrFrameTooLarge } _ = append(f.wbuf[:0], byte(length>>16), byte(length>>8), byte(length)) if http2logFrameWrites { f.logWrite() } n, err := f.w.Write(f.wbuf) if err == nil && n != len(f.wbuf) { err = io.ErrShortWrite } return err } func (f *http2Framer) logWrite() { if f.debugFramer == nil { f.debugFramerBuf = new(bytes.Buffer) f.debugFramer = http2NewFramer(nil, f.debugFramerBuf) f.debugFramer.logReads = false f.debugFramer.AllowIllegalReads = true } f.debugFramerBuf.Write(f.wbuf) fr, err := f.debugFramer.ReadFrame() if err != nil { log.Printf("http2: Framer %p: failed to decode just-written frame", f) return } log.Printf("http2: Framer %p: wrote %v", f, http2summarizeFrame(fr)) } func (f *http2Framer) writeByte(v byte) { f.wbuf = append(f.wbuf, v) } func (f *http2Framer) writeBytes(v []byte) { f.wbuf = append(f.wbuf, v...) } func (f *http2Framer) writeUint16(v uint16) { f.wbuf = append(f.wbuf, byte(v>>8), byte(v)) } func (f *http2Framer) writeUint32(v uint32) { f.wbuf = append(f.wbuf, byte(v>>24), byte(v>>16), byte(v>>8), byte(v)) } const ( http2minMaxFrameSize = 1 << 14 http2maxFrameSize = 1<<24 - 1 ) // NewFramer returns a Framer that writes frames to w and reads them from r. func http2NewFramer(w io.Writer, r io.Reader) *http2Framer { fr := &http2Framer{ w: w, r: r, logReads: http2logFrameReads, } fr.getReadBuf = func(size uint32) []byte { if cap(fr.readBuf) >= int(size) { return fr.readBuf[:size] } fr.readBuf = make([]byte, size) return fr.readBuf } fr.SetMaxReadFrameSize(http2maxFrameSize) return fr } // SetMaxReadFrameSize sets the maximum size of a frame // that will be read by a subsequent call to ReadFrame. // It is the caller's responsibility to advertise this // limit with a SETTINGS frame. func (fr *http2Framer) SetMaxReadFrameSize(v uint32) { if v > http2maxFrameSize { v = http2maxFrameSize } fr.maxReadSize = v } // ErrFrameTooLarge is returned from Framer.ReadFrame when the peer // sends a frame that is larger than declared with SetMaxReadFrameSize. var http2ErrFrameTooLarge = errors.New("http2: frame too large") // terminalReadFrameError reports whether err is an unrecoverable // error from ReadFrame and no other frames should be read. func http2terminalReadFrameError(err error) bool { if _, ok := err.(http2StreamError); ok { return false } return err != nil } // ReadFrame reads a single frame. The returned Frame is only valid // until the next call to ReadFrame. // // If the frame is larger than previously set with SetMaxReadFrameSize, the // returned error is ErrFrameTooLarge. Other errors may be of type // ConnectionError, StreamError, or anything else from from the underlying // reader. func (fr *http2Framer) ReadFrame() (http2Frame, error) { if fr.lastFrame != nil { fr.lastFrame.invalidate() } fh, err := http2readFrameHeader(fr.headerBuf[:], fr.r) if err != nil { return nil, err } if fh.Length > fr.maxReadSize { return nil, http2ErrFrameTooLarge } payload := fr.getReadBuf(fh.Length) if _, err := io.ReadFull(fr.r, payload); err != nil { return nil, err } f, err := http2typeFrameParser(fh.Type)(fh, payload) if err != nil { if ce, ok := err.(http2connError); ok { return nil, fr.connError(ce.Code, ce.Reason) } return nil, err } if err := fr.checkFrameOrder(f); err != nil { return nil, err } if fr.logReads { log.Printf("http2: Framer %p: read %v", fr, http2summarizeFrame(f)) } return f, nil } // connError returns ConnectionError(code) but first // stashes away a public reason to the caller can optionally relay it // to the peer before hanging up on them. This might help others debug // their implementations. func (fr *http2Framer) connError(code http2ErrCode, reason string) error { fr.errReason = reason return http2ConnectionError(code) } // checkFrameOrder reports an error if f is an invalid frame to return // next from ReadFrame. Mostly it checks whether HEADERS and // CONTINUATION frames are contiguous. func (fr *http2Framer) checkFrameOrder(f http2Frame) error { last := fr.lastFrame fr.lastFrame = f if fr.AllowIllegalReads { return nil } fh := f.Header() if fr.lastHeaderStream != 0 { if fh.Type != http2FrameContinuation { return fr.connError(http2ErrCodeProtocol, fmt.Sprintf("got %s for stream %d; expected CONTINUATION following %s for stream %d", fh.Type, fh.StreamID, last.Header().Type, fr.lastHeaderStream)) } if fh.StreamID != fr.lastHeaderStream { return fr.connError(http2ErrCodeProtocol, fmt.Sprintf("got CONTINUATION for stream %d; expected stream %d", fh.StreamID, fr.lastHeaderStream)) } } else if fh.Type == http2FrameContinuation { return fr.connError(http2ErrCodeProtocol, fmt.Sprintf("unexpected CONTINUATION for stream %d", fh.StreamID)) } switch fh.Type { case http2FrameHeaders, http2FrameContinuation: if fh.Flags.Has(http2FlagHeadersEndHeaders) { fr.lastHeaderStream = 0 } else { fr.lastHeaderStream = fh.StreamID } } return nil } // A DataFrame conveys arbitrary, variable-length sequences of octets // associated with a stream. // See http://http2.github.io/http2-spec/#rfc.section.6.1 type http2DataFrame struct { http2FrameHeader data []byte } func (f *http2DataFrame) StreamEnded() bool { return f.http2FrameHeader.Flags.Has(http2FlagDataEndStream) } // Data returns the frame's data octets, not including any padding // size byte or padding suffix bytes. // The caller must not retain the returned memory past the next // call to ReadFrame. func (f *http2DataFrame) Data() []byte { f.checkValid() return f.data } func http2parseDataFrame(fh http2FrameHeader, payload []byte) (http2Frame, error) { if fh.StreamID == 0 { return nil, http2connError{http2ErrCodeProtocol, "DATA frame with stream ID 0"} } f := &http2DataFrame{ http2FrameHeader: fh, } var padSize byte if fh.Flags.Has(http2FlagDataPadded) { var err error payload, padSize, err = http2readByte(payload) if err != nil { return nil, err } } if int(padSize) > len(payload) { return nil, http2connError{http2ErrCodeProtocol, "pad size larger than data payload"} } f.data = payload[:len(payload)-int(padSize)] return f, nil } var http2errStreamID = errors.New("invalid streamid") func http2validStreamID(streamID uint32) bool { return streamID != 0 && streamID&(1<<31) == 0 } // WriteData writes a DATA frame. // // It will perform exactly one Write to the underlying Writer. // It is the caller's responsibility to not call other Write methods concurrently. func (f *http2Framer) WriteData(streamID uint32, endStream bool, data []byte) error { if !http2validStreamID(streamID) && !f.AllowIllegalWrites { return http2errStreamID } var flags http2Flags if endStream { flags |= http2FlagDataEndStream } f.startWrite(http2FrameData, flags, streamID) f.wbuf = append(f.wbuf, data...) return f.endWrite() } // A SettingsFrame conveys configuration parameters that affect how // endpoints communicate, such as preferences and constraints on peer // behavior. // // See http://http2.github.io/http2-spec/#SETTINGS type http2SettingsFrame struct { http2FrameHeader p []byte } func http2parseSettingsFrame(fh http2FrameHeader, p []byte) (http2Frame, error) { if fh.Flags.Has(http2FlagSettingsAck) && fh.Length > 0 { return nil, http2ConnectionError(http2ErrCodeFrameSize) } if fh.StreamID != 0 { return nil, http2ConnectionError(http2ErrCodeProtocol) } if len(p)%6 != 0 { return nil, http2ConnectionError(http2ErrCodeFrameSize) } f := &http2SettingsFrame{http2FrameHeader: fh, p: p} if v, ok := f.Value(http2SettingInitialWindowSize); ok && v > (1<<31)-1 { return nil, http2ConnectionError(http2ErrCodeFlowControl) } return f, nil } func (f *http2SettingsFrame) IsAck() bool { return f.http2FrameHeader.Flags.Has(http2FlagSettingsAck) } func (f *http2SettingsFrame) Value(s http2SettingID) (v uint32, ok bool) { f.checkValid() buf := f.p for len(buf) > 0 { settingID := http2SettingID(binary.BigEndian.Uint16(buf[:2])) if settingID == s { return binary.BigEndian.Uint32(buf[2:6]), true } buf = buf[6:] } return 0, false } // ForeachSetting runs fn for each setting. // It stops and returns the first error. func (f *http2SettingsFrame) ForeachSetting(fn func(http2Setting) error) error { f.checkValid() buf := f.p for len(buf) > 0 { if err := fn(http2Setting{ http2SettingID(binary.BigEndian.Uint16(buf[:2])), binary.BigEndian.Uint32(buf[2:6]), }); err != nil { return err } buf = buf[6:] } return nil } // WriteSettings writes a SETTINGS frame with zero or more settings // specified and the ACK bit not set. // // It will perform exactly one Write to the underlying Writer. // It is the caller's responsibility to not call other Write methods concurrently. func (f *http2Framer) WriteSettings(settings ...http2Setting) error { f.startWrite(http2FrameSettings, 0, 0) for _, s := range settings { f.writeUint16(uint16(s.ID)) f.writeUint32(s.Val) } return f.endWrite() } // WriteSettings writes an empty SETTINGS frame with the ACK bit set. // // It will perform exactly one Write to the underlying Writer. // It is the caller's responsibility to not call other Write methods concurrently. func (f *http2Framer) WriteSettingsAck() error { f.startWrite(http2FrameSettings, http2FlagSettingsAck, 0) return f.endWrite() } // A PingFrame is a mechanism for measuring a minimal round trip time // from the sender, as well as determining whether an idle connection // is still functional. // See http://http2.github.io/http2-spec/#rfc.section.6.7 type http2PingFrame struct { http2FrameHeader Data [8]byte } func (f *http2PingFrame) IsAck() bool { return f.Flags.Has(http2FlagPingAck) } func http2parsePingFrame(fh http2FrameHeader, payload []byte) (http2Frame, error) { if len(payload) != 8 { return nil, http2ConnectionError(http2ErrCodeFrameSize) } if fh.StreamID != 0 { return nil, http2ConnectionError(http2ErrCodeProtocol) } f := &http2PingFrame{http2FrameHeader: fh} copy(f.Data[:], payload) return f, nil } func (f *http2Framer) WritePing(ack bool, data [8]byte) error { var flags http2Flags if ack { flags = http2FlagPingAck } f.startWrite(http2FramePing, flags, 0) f.writeBytes(data[:]) return f.endWrite() } // A GoAwayFrame informs the remote peer to stop creating streams on this connection. // See http://http2.github.io/http2-spec/#rfc.section.6.8 type http2GoAwayFrame struct { http2FrameHeader LastStreamID uint32 ErrCode http2ErrCode debugData []byte } // DebugData returns any debug data in the GOAWAY frame. Its contents // are not defined. // The caller must not retain the returned memory past the next // call to ReadFrame. func (f *http2GoAwayFrame) DebugData() []byte { f.checkValid() return f.debugData } func http2parseGoAwayFrame(fh http2FrameHeader, p []byte) (http2Frame, error) { if fh.StreamID != 0 { return nil, http2ConnectionError(http2ErrCodeProtocol) } if len(p) < 8 { return nil, http2ConnectionError(http2ErrCodeFrameSize) } return &http2GoAwayFrame{ http2FrameHeader: fh, LastStreamID: binary.BigEndian.Uint32(p[:4]) & (1<<31 - 1), ErrCode: http2ErrCode(binary.BigEndian.Uint32(p[4:8])), debugData: p[8:], }, nil } func (f *http2Framer) WriteGoAway(maxStreamID uint32, code http2ErrCode, debugData []byte) error { f.startWrite(http2FrameGoAway, 0, 0) f.writeUint32(maxStreamID & (1<<31 - 1)) f.writeUint32(uint32(code)) f.writeBytes(debugData) return f.endWrite() } // An UnknownFrame is the frame type returned when the frame type is unknown // or no specific frame type parser exists. type http2UnknownFrame struct { http2FrameHeader p []byte } // Payload returns the frame's payload (after the header). It is not // valid to call this method after a subsequent call to // Framer.ReadFrame, nor is it valid to retain the returned slice. // The memory is owned by the Framer and is invalidated when the next // frame is read. func (f *http2UnknownFrame) Payload() []byte { f.checkValid() return f.p } func http2parseUnknownFrame(fh http2FrameHeader, p []byte) (http2Frame, error) { return &http2UnknownFrame{fh, p}, nil } // A WindowUpdateFrame is used to implement flow control. // See http://http2.github.io/http2-spec/#rfc.section.6.9 type http2WindowUpdateFrame struct { http2FrameHeader Increment uint32 // never read with high bit set } func http2parseWindowUpdateFrame(fh http2FrameHeader, p []byte) (http2Frame, error) { if len(p) != 4 { return nil, http2ConnectionError(http2ErrCodeFrameSize) } inc := binary.BigEndian.Uint32(p[:4]) & 0x7fffffff if inc == 0 { if fh.StreamID == 0 { return nil, http2ConnectionError(http2ErrCodeProtocol) } return nil, http2StreamError{fh.StreamID, http2ErrCodeProtocol} } return &http2WindowUpdateFrame{ http2FrameHeader: fh, Increment: inc, }, nil } // WriteWindowUpdate writes a WINDOW_UPDATE frame. // The increment value must be between 1 and 2,147,483,647, inclusive. // If the Stream ID is zero, the window update applies to the // connection as a whole. func (f *http2Framer) WriteWindowUpdate(streamID, incr uint32) error { if (incr < 1 || incr > 2147483647) && !f.AllowIllegalWrites { return errors.New("illegal window increment value") } f.startWrite(http2FrameWindowUpdate, 0, streamID) f.writeUint32(incr) return f.endWrite() } // A HeadersFrame is used to open a stream and additionally carries a // header block fragment. type http2HeadersFrame struct { http2FrameHeader // Priority is set if FlagHeadersPriority is set in the FrameHeader. Priority http2PriorityParam headerFragBuf []byte // not owned } func (f *http2HeadersFrame) HeaderBlockFragment() []byte { f.checkValid() return f.headerFragBuf } func (f *http2HeadersFrame) HeadersEnded() bool { return f.http2FrameHeader.Flags.Has(http2FlagHeadersEndHeaders) } func (f *http2HeadersFrame) StreamEnded() bool { return f.http2FrameHeader.Flags.Has(http2FlagHeadersEndStream) } func (f *http2HeadersFrame) HasPriority() bool { return f.http2FrameHeader.Flags.Has(http2FlagHeadersPriority) } func http2parseHeadersFrame(fh http2FrameHeader, p []byte) (_ http2Frame, err error) { hf := &http2HeadersFrame{ http2FrameHeader: fh, } if fh.StreamID == 0 { return nil, http2connError{http2ErrCodeProtocol, "HEADERS frame with stream ID 0"} } var padLength uint8 if fh.Flags.Has(http2FlagHeadersPadded) { if p, padLength, err = http2readByte(p); err != nil { return } } if fh.Flags.Has(http2FlagHeadersPriority) { var v uint32 p, v, err = http2readUint32(p) if err != nil { return nil, err } hf.Priority.StreamDep = v & 0x7fffffff hf.Priority.Exclusive = (v != hf.Priority.StreamDep) p, hf.Priority.Weight, err = http2readByte(p) if err != nil { return nil, err } } if len(p)-int(padLength) <= 0 { return nil, http2StreamError{fh.StreamID, http2ErrCodeProtocol} } hf.headerFragBuf = p[:len(p)-int(padLength)] return hf, nil } // HeadersFrameParam are the parameters for writing a HEADERS frame. type http2HeadersFrameParam struct { // StreamID is the required Stream ID to initiate. StreamID uint32 // BlockFragment is part (or all) of a Header Block. BlockFragment []byte // EndStream indicates that the header block is the last that // the endpoint will send for the identified stream. Setting // this flag causes the stream to enter one of "half closed" // states. EndStream bool // EndHeaders indicates that this frame contains an entire // header block and is not followed by any // CONTINUATION frames. EndHeaders bool // PadLength is the optional number of bytes of zeros to add // to this frame. PadLength uint8 // Priority, if non-zero, includes stream priority information // in the HEADER frame. Priority http2PriorityParam } // WriteHeaders writes a single HEADERS frame. // // This is a low-level header writing method. Encoding headers and // splitting them into any necessary CONTINUATION frames is handled // elsewhere. // // It will perform exactly one Write to the underlying Writer. // It is the caller's responsibility to not call other Write methods concurrently. func (f *http2Framer) WriteHeaders(p http2HeadersFrameParam) error { if !http2validStreamID(p.StreamID) && !f.AllowIllegalWrites { return http2errStreamID } var flags http2Flags if p.PadLength != 0 { flags |= http2FlagHeadersPadded } if p.EndStream { flags |= http2FlagHeadersEndStream } if p.EndHeaders { flags |= http2FlagHeadersEndHeaders } if !p.Priority.IsZero() { flags |= http2FlagHeadersPriority } f.startWrite(http2FrameHeaders, flags, p.StreamID) if p.PadLength != 0 { f.writeByte(p.PadLength) } if !p.Priority.IsZero() { v := p.Priority.StreamDep if !http2validStreamID(v) && !f.AllowIllegalWrites { return errors.New("invalid dependent stream id") } if p.Priority.Exclusive { v |= 1 << 31 } f.writeUint32(v) f.writeByte(p.Priority.Weight) } f.wbuf = append(f.wbuf, p.BlockFragment...) f.wbuf = append(f.wbuf, http2padZeros[:p.PadLength]...) return f.endWrite() } // A PriorityFrame specifies the sender-advised priority of a stream. // See http://http2.github.io/http2-spec/#rfc.section.6.3 type http2PriorityFrame struct { http2FrameHeader http2PriorityParam } // PriorityParam are the stream prioritzation parameters. type http2PriorityParam struct { // StreamDep is a 31-bit stream identifier for the // stream that this stream depends on. Zero means no // dependency. StreamDep uint32 // Exclusive is whether the dependency is exclusive. Exclusive bool // Weight is the stream's zero-indexed weight. It should be // set together with StreamDep, or neither should be set. Per // the spec, "Add one to the value to obtain a weight between // 1 and 256." Weight uint8 } func (p http2PriorityParam) IsZero() bool { return p == http2PriorityParam{} } func http2parsePriorityFrame(fh http2FrameHeader, payload []byte) (http2Frame, error) { if fh.StreamID == 0 { return nil, http2connError{http2ErrCodeProtocol, "PRIORITY frame with stream ID 0"} } if len(payload) != 5 { return nil, http2connError{http2ErrCodeFrameSize, fmt.Sprintf("PRIORITY frame payload size was %d; want 5", len(payload))} } v := binary.BigEndian.Uint32(payload[:4]) streamID := v & 0x7fffffff return &http2PriorityFrame{ http2FrameHeader: fh, http2PriorityParam: http2PriorityParam{ Weight: payload[4], StreamDep: streamID, Exclusive: streamID != v, }, }, nil } // WritePriority writes a PRIORITY frame. // // It will perform exactly one Write to the underlying Writer. // It is the caller's responsibility to not call other Write methods concurrently. func (f *http2Framer) WritePriority(streamID uint32, p http2PriorityParam) error { if !http2validStreamID(streamID) && !f.AllowIllegalWrites { return http2errStreamID } f.startWrite(http2FramePriority, 0, streamID) v := p.StreamDep if p.Exclusive { v |= 1 << 31 } f.writeUint32(v) f.writeByte(p.Weight) return f.endWrite() } // A RSTStreamFrame allows for abnormal termination of a stream. // See http://http2.github.io/http2-spec/#rfc.section.6.4 type http2RSTStreamFrame struct { http2FrameHeader ErrCode http2ErrCode } func http2parseRSTStreamFrame(fh http2FrameHeader, p []byte) (http2Frame, error) { if len(p) != 4 { return nil, http2ConnectionError(http2ErrCodeFrameSize) } if fh.StreamID == 0 { return nil, http2ConnectionError(http2ErrCodeProtocol) } return &http2RSTStreamFrame{fh, http2ErrCode(binary.BigEndian.Uint32(p[:4]))}, nil } // WriteRSTStream writes a RST_STREAM frame. // // It will perform exactly one Write to the underlying Writer. // It is the caller's responsibility to not call other Write methods concurrently. func (f *http2Framer) WriteRSTStream(streamID uint32, code http2ErrCode) error { if !http2validStreamID(streamID) && !f.AllowIllegalWrites { return http2errStreamID } f.startWrite(http2FrameRSTStream, 0, streamID) f.writeUint32(uint32(code)) return f.endWrite() } // A ContinuationFrame is used to continue a sequence of header block fragments. // See http://http2.github.io/http2-spec/#rfc.section.6.10 type http2ContinuationFrame struct { http2FrameHeader headerFragBuf []byte } func http2parseContinuationFrame(fh http2FrameHeader, p []byte) (http2Frame, error) { if fh.StreamID == 0 { return nil, http2connError{http2ErrCodeProtocol, "CONTINUATION frame with stream ID 0"} } return &http2ContinuationFrame{fh, p}, nil } func (f *http2ContinuationFrame) HeaderBlockFragment() []byte { f.checkValid() return f.headerFragBuf } func (f *http2ContinuationFrame) HeadersEnded() bool { return f.http2FrameHeader.Flags.Has(http2FlagContinuationEndHeaders) } // WriteContinuation writes a CONTINUATION frame. // // It will perform exactly one Write to the underlying Writer. // It is the caller's responsibility to not call other Write methods concurrently. func (f *http2Framer) WriteContinuation(streamID uint32, endHeaders bool, headerBlockFragment []byte) error { if !http2validStreamID(streamID) && !f.AllowIllegalWrites { return http2errStreamID } var flags http2Flags if endHeaders { flags |= http2FlagContinuationEndHeaders } f.startWrite(http2FrameContinuation, flags, streamID) f.wbuf = append(f.wbuf, headerBlockFragment...) return f.endWrite() } // A PushPromiseFrame is used to initiate a server stream. // See http://http2.github.io/http2-spec/#rfc.section.6.6 type http2PushPromiseFrame struct { http2FrameHeader PromiseID uint32 headerFragBuf []byte // not owned } func (f *http2PushPromiseFrame) HeaderBlockFragment() []byte { f.checkValid() return f.headerFragBuf } func (f *http2PushPromiseFrame) HeadersEnded() bool { return f.http2FrameHeader.Flags.Has(http2FlagPushPromiseEndHeaders) } func http2parsePushPromise(fh http2FrameHeader, p []byte) (_ http2Frame, err error) { pp := &http2PushPromiseFrame{ http2FrameHeader: fh, } if pp.StreamID == 0 { return nil, http2ConnectionError(http2ErrCodeProtocol) } // The PUSH_PROMISE frame includes optional padding. // Padding fields and flags are identical to those defined for DATA frames var padLength uint8 if fh.Flags.Has(http2FlagPushPromisePadded) { if p, padLength, err = http2readByte(p); err != nil { return } } p, pp.PromiseID, err = http2readUint32(p) if err != nil { return } pp.PromiseID = pp.PromiseID & (1<<31 - 1) if int(padLength) > len(p) { return nil, http2ConnectionError(http2ErrCodeProtocol) } pp.headerFragBuf = p[:len(p)-int(padLength)] return pp, nil } // PushPromiseParam are the parameters for writing a PUSH_PROMISE frame. type http2PushPromiseParam struct { // StreamID is the required Stream ID to initiate. StreamID uint32 // PromiseID is the required Stream ID which this // Push Promises PromiseID uint32 // BlockFragment is part (or all) of a Header Block. BlockFragment []byte // EndHeaders indicates that this frame contains an entire // header block and is not followed by any // CONTINUATION frames. EndHeaders bool // PadLength is the optional number of bytes of zeros to add // to this frame. PadLength uint8 } // WritePushPromise writes a single PushPromise Frame. // // As with Header Frames, This is the low level call for writing // individual frames. Continuation frames are handled elsewhere. // // It will perform exactly one Write to the underlying Writer. // It is the caller's responsibility to not call other Write methods concurrently. func (f *http2Framer) WritePushPromise(p http2PushPromiseParam) error { if !http2validStreamID(p.StreamID) && !f.AllowIllegalWrites { return http2errStreamID } var flags http2Flags if p.PadLength != 0 { flags |= http2FlagPushPromisePadded } if p.EndHeaders { flags |= http2FlagPushPromiseEndHeaders } f.startWrite(http2FramePushPromise, flags, p.StreamID) if p.PadLength != 0 { f.writeByte(p.PadLength) } if !http2validStreamID(p.PromiseID) && !f.AllowIllegalWrites { return http2errStreamID } f.writeUint32(p.PromiseID) f.wbuf = append(f.wbuf, p.BlockFragment...) f.wbuf = append(f.wbuf, http2padZeros[:p.PadLength]...) return f.endWrite() } // WriteRawFrame writes a raw frame. This can be used to write // extension frames unknown to this package. func (f *http2Framer) WriteRawFrame(t http2FrameType, flags http2Flags, streamID uint32, payload []byte) error { f.startWrite(t, flags, streamID) f.writeBytes(payload) return f.endWrite() } func http2readByte(p []byte) (remain []byte, b byte, err error) { if len(p) == 0 { return nil, 0, io.ErrUnexpectedEOF } return p[1:], p[0], nil } func http2readUint32(p []byte) (remain []byte, v uint32, err error) { if len(p) < 4 { return nil, 0, io.ErrUnexpectedEOF } return p[4:], binary.BigEndian.Uint32(p[:4]), nil } type http2streamEnder interface { StreamEnded() bool } type http2headersEnder interface { HeadersEnded() bool } func http2summarizeFrame(f http2Frame) string { var buf bytes.Buffer f.Header().writeDebug(&buf) switch f := f.(type) { case *http2SettingsFrame: n := 0 f.ForeachSetting(func(s http2Setting) error { n++ if n == 1 { buf.WriteString(", settings:") } fmt.Fprintf(&buf, " %v=%v,", s.ID, s.Val) return nil }) if n > 0 { buf.Truncate(buf.Len() - 1) } case *http2DataFrame: data := f.Data() const max = 256 if len(data) > max { data = data[:max] } fmt.Fprintf(&buf, " data=%q", data) if len(f.Data()) > max { fmt.Fprintf(&buf, " (%d bytes omitted)", len(f.Data())-max) } case *http2WindowUpdateFrame: if f.StreamID == 0 { buf.WriteString(" (conn)") } fmt.Fprintf(&buf, " incr=%v", f.Increment) case *http2PingFrame: fmt.Fprintf(&buf, " ping=%q", f.Data[:]) case *http2GoAwayFrame: fmt.Fprintf(&buf, " LastStreamID=%v ErrCode=%v Debug=%q", f.LastStreamID, f.ErrCode, f.debugData) case *http2RSTStreamFrame: fmt.Fprintf(&buf, " ErrCode=%v", f.ErrCode) } return buf.String() } func http2requestCancel(req *Request) <-chan struct{} { return req.Cancel } var http2DebugGoroutines = os.Getenv("DEBUG_HTTP2_GOROUTINES") == "1" type http2goroutineLock uint64 func http2newGoroutineLock() http2goroutineLock { if !http2DebugGoroutines { return 0 } return http2goroutineLock(http2curGoroutineID()) } func (g http2goroutineLock) check() { if !http2DebugGoroutines { return } if http2curGoroutineID() != uint64(g) { panic("running on the wrong goroutine") } } func (g http2goroutineLock) checkNotOn() { if !http2DebugGoroutines { return } if http2curGoroutineID() == uint64(g) { panic("running on the wrong goroutine") } } var http2goroutineSpace = []byte("goroutine ") func http2curGoroutineID() uint64 { bp := http2littleBuf.Get().(*[]byte) defer http2littleBuf.Put(bp) b := *bp b = b[:runtime.Stack(b, false)] b = bytes.TrimPrefix(b, http2goroutineSpace) i := bytes.IndexByte(b, ' ') if i < 0 { panic(fmt.Sprintf("No space found in %q", b)) } b = b[:i] n, err := http2parseUintBytes(b, 10, 64) if err != nil { panic(fmt.Sprintf("Failed to parse goroutine ID out of %q: %v", b, err)) } return n } var http2littleBuf = sync.Pool{ New: func() interface{} { buf := make([]byte, 64) return &buf }, } // parseUintBytes is like strconv.ParseUint, but using a []byte. func http2parseUintBytes(s []byte, base int, bitSize int) (n uint64, err error) { var cutoff, maxVal uint64 if bitSize == 0 { bitSize = int(strconv.IntSize) } s0 := s switch { case len(s) < 1: err = strconv.ErrSyntax goto Error case 2 <= base && base <= 36: case base == 0: switch { case s[0] == '0' && len(s) > 1 && (s[1] == 'x' || s[1] == 'X'): base = 16 s = s[2:] if len(s) < 1 { err = strconv.ErrSyntax goto Error } case s[0] == '0': base = 8 default: base = 10 } default: err = errors.New("invalid base " + strconv.Itoa(base)) goto Error } n = 0 cutoff = http2cutoff64(base) maxVal = 1<= base { n = 0 err = strconv.ErrSyntax goto Error } if n >= cutoff { n = 1<<64 - 1 err = strconv.ErrRange goto Error } n *= uint64(base) n1 := n + uint64(v) if n1 < n || n1 > maxVal { n = 1<<64 - 1 err = strconv.ErrRange goto Error } n = n1 } return n, nil Error: return n, &strconv.NumError{Func: "ParseUint", Num: string(s0), Err: err} } // Return the first number n such that n*base >= 1<<64. func http2cutoff64(base int) uint64 { if base < 2 { return 0 } return (1<<64-1)/uint64(base) + 1 } var ( http2commonLowerHeader = map[string]string{} // Go-Canonical-Case -> lower-case http2commonCanonHeader = map[string]string{} // lower-case -> Go-Canonical-Case ) func init() { for _, v := range []string{ "accept", "accept-charset", "accept-encoding", "accept-language", "accept-ranges", "age", "access-control-allow-origin", "allow", "authorization", "cache-control", "content-disposition", "content-encoding", "content-language", "content-length", "content-location", "content-range", "content-type", "cookie", "date", "etag", "expect", "expires", "from", "host", "if-match", "if-modified-since", "if-none-match", "if-unmodified-since", "last-modified", "link", "location", "max-forwards", "proxy-authenticate", "proxy-authorization", "range", "referer", "refresh", "retry-after", "server", "set-cookie", "strict-transport-security", "trailer", "transfer-encoding", "user-agent", "vary", "via", "www-authenticate", } { chk := CanonicalHeaderKey(v) http2commonLowerHeader[chk] = v http2commonCanonHeader[v] = chk } } func http2lowerHeader(v string) string { if s, ok := http2commonLowerHeader[v]; ok { return s } return strings.ToLower(v) } var ( http2VerboseLogs bool http2logFrameWrites bool http2logFrameReads bool ) func init() { e := os.Getenv("GODEBUG") if strings.Contains(e, "http2debug=1") { http2VerboseLogs = true } if strings.Contains(e, "http2debug=2") { http2VerboseLogs = true http2logFrameWrites = true http2logFrameReads = true } } const ( // ClientPreface is the string that must be sent by new // connections from clients. http2ClientPreface = "PRI * HTTP/2.0\r\n\r\nSM\r\n\r\n" // SETTINGS_MAX_FRAME_SIZE default // http://http2.github.io/http2-spec/#rfc.section.6.5.2 http2initialMaxFrameSize = 16384 // NextProtoTLS is the NPN/ALPN protocol negotiated during // HTTP/2's TLS setup. http2NextProtoTLS = "h2" // http://http2.github.io/http2-spec/#SettingValues http2initialHeaderTableSize = 4096 http2initialWindowSize = 65535 // 6.9.2 Initial Flow Control Window Size http2defaultMaxReadFrameSize = 1 << 20 ) var ( http2clientPreface = []byte(http2ClientPreface) ) type http2streamState int const ( http2stateIdle http2streamState = iota http2stateOpen http2stateHalfClosedLocal http2stateHalfClosedRemote http2stateResvLocal http2stateResvRemote http2stateClosed ) var http2stateName = [...]string{ http2stateIdle: "Idle", http2stateOpen: "Open", http2stateHalfClosedLocal: "HalfClosedLocal", http2stateHalfClosedRemote: "HalfClosedRemote", http2stateResvLocal: "ResvLocal", http2stateResvRemote: "ResvRemote", http2stateClosed: "Closed", } func (st http2streamState) String() string { return http2stateName[st] } // Setting is a setting parameter: which setting it is, and its value. type http2Setting struct { // ID is which setting is being set. // See http://http2.github.io/http2-spec/#SettingValues ID http2SettingID // Val is the value. Val uint32 } func (s http2Setting) String() string { return fmt.Sprintf("[%v = %d]", s.ID, s.Val) } // Valid reports whether the setting is valid. func (s http2Setting) Valid() error { switch s.ID { case http2SettingEnablePush: if s.Val != 1 && s.Val != 0 { return http2ConnectionError(http2ErrCodeProtocol) } case http2SettingInitialWindowSize: if s.Val > 1<<31-1 { return http2ConnectionError(http2ErrCodeFlowControl) } case http2SettingMaxFrameSize: if s.Val < 16384 || s.Val > 1<<24-1 { return http2ConnectionError(http2ErrCodeProtocol) } } return nil } // A SettingID is an HTTP/2 setting as defined in // http://http2.github.io/http2-spec/#iana-settings type http2SettingID uint16 const ( http2SettingHeaderTableSize http2SettingID = 0x1 http2SettingEnablePush http2SettingID = 0x2 http2SettingMaxConcurrentStreams http2SettingID = 0x3 http2SettingInitialWindowSize http2SettingID = 0x4 http2SettingMaxFrameSize http2SettingID = 0x5 http2SettingMaxHeaderListSize http2SettingID = 0x6 ) var http2settingName = map[http2SettingID]string{ http2SettingHeaderTableSize: "HEADER_TABLE_SIZE", http2SettingEnablePush: "ENABLE_PUSH", http2SettingMaxConcurrentStreams: "MAX_CONCURRENT_STREAMS", http2SettingInitialWindowSize: "INITIAL_WINDOW_SIZE", http2SettingMaxFrameSize: "MAX_FRAME_SIZE", http2SettingMaxHeaderListSize: "MAX_HEADER_LIST_SIZE", } func (s http2SettingID) String() string { if v, ok := http2settingName[s]; ok { return v } return fmt.Sprintf("UNKNOWN_SETTING_%d", uint16(s)) } var ( http2errInvalidHeaderFieldName = errors.New("http2: invalid header field name") http2errInvalidHeaderFieldValue = errors.New("http2: invalid header field value") ) // validHeaderFieldName reports whether v is a valid header field name (key). // RFC 7230 says: // header-field = field-name ":" OWS field-value OWS // field-name = token // tchar = "!" / "#" / "$" / "%" / "&" / "'" / "*" / "+" / "-" / "." / // "^" / "_" / " // Further, http2 says: // "Just as in HTTP/1.x, header field names are strings of ASCII // characters that are compared in a case-insensitive // fashion. However, header field names MUST be converted to // lowercase prior to their encoding in HTTP/2. " func http2validHeaderFieldName(v string) bool { if len(v) == 0 { return false } for _, r := range v { if int(r) >= len(http2isTokenTable) || ('A' <= r && r <= 'Z') { return false } if !http2isTokenTable[byte(r)] { return false } } return true } // validHeaderFieldValue reports whether v is a valid header field value. // // RFC 7230 says: // field-value = *( field-content / obs-fold ) // obj-fold = N/A to http2, and deprecated // field-content = field-vchar [ 1*( SP / HTAB ) field-vchar ] // field-vchar = VCHAR / obs-text // obs-text = %x80-FF // VCHAR = "any visible [USASCII] character" // // http2 further says: "Similarly, HTTP/2 allows header field values // that are not valid. While most of the values that can be encoded // will not alter header field parsing, carriage return (CR, ASCII // 0xd), line feed (LF, ASCII 0xa), and the zero character (NUL, ASCII // 0x0) might be exploited by an attacker if they are translated // verbatim. Any request or response that contains a character not // permitted in a header field value MUST be treated as malformed // (Section 8.1.2.6). Valid characters are defined by the // field-content ABNF rule in Section 3.2 of [RFC7230]." // // This function does not (yet?) properly handle the rejection of // strings that begin or end with SP or HTAB. func http2validHeaderFieldValue(v string) bool { for i := 0; i < len(v); i++ { if b := v[i]; b < ' ' && b != '\t' || b == 0x7f { return false } } return true } var http2httpCodeStringCommon = map[int]string{} // n -> strconv.Itoa(n) func init() { for i := 100; i <= 999; i++ { if v := StatusText(i); v != "" { http2httpCodeStringCommon[i] = strconv.Itoa(i) } } } func http2httpCodeString(code int) string { if s, ok := http2httpCodeStringCommon[code]; ok { return s } return strconv.Itoa(code) } // from pkg io type http2stringWriter interface { WriteString(s string) (n int, err error) } // A gate lets two goroutines coordinate their activities. type http2gate chan struct{} func (g http2gate) Done() { g <- struct{}{} } func (g http2gate) Wait() { <-g } // A closeWaiter is like a sync.WaitGroup but only goes 1 to 0 (open to closed). type http2closeWaiter chan struct{} // Init makes a closeWaiter usable. // It exists because so a closeWaiter value can be placed inside a // larger struct and have the Mutex and Cond's memory in the same // allocation. func (cw *http2closeWaiter) Init() { *cw = make(chan struct{}) } // Close marks the closeWaiter as closed and unblocks any waiters. func (cw http2closeWaiter) Close() { close(cw) } // Wait waits for the closeWaiter to become closed. func (cw http2closeWaiter) Wait() { <-cw } // bufferedWriter is a buffered writer that writes to w. // Its buffered writer is lazily allocated as needed, to minimize // idle memory usage with many connections. type http2bufferedWriter struct { w io.Writer // immutable bw *bufio.Writer // non-nil when data is buffered } func http2newBufferedWriter(w io.Writer) *http2bufferedWriter { return &http2bufferedWriter{w: w} } var http2bufWriterPool = sync.Pool{ New: func() interface{} { return bufio.NewWriterSize(nil, 4<<10) }, } func (w *http2bufferedWriter) Write(p []byte) (n int, err error) { if w.bw == nil { bw := http2bufWriterPool.Get().(*bufio.Writer) bw.Reset(w.w) w.bw = bw } return w.bw.Write(p) } func (w *http2bufferedWriter) Flush() error { bw := w.bw if bw == nil { return nil } err := bw.Flush() bw.Reset(nil) http2bufWriterPool.Put(bw) w.bw = nil return err } func http2mustUint31(v int32) uint32 { if v < 0 || v > 2147483647 { panic("out of range") } return uint32(v) } // bodyAllowedForStatus reports whether a given response status code // permits a body. See RFC2616, section 4.4. func http2bodyAllowedForStatus(status int) bool { switch { case status >= 100 && status <= 199: return false case status == 204: return false case status == 304: return false } return true } type http2httpError struct { msg string timeout bool } func (e *http2httpError) Error() string { return e.msg } func (e *http2httpError) Timeout() bool { return e.timeout } func (e *http2httpError) Temporary() bool { return true } var http2errTimeout error = &http2httpError{msg: "http2: timeout awaiting response headers", timeout: true} var http2isTokenTable = [127]bool{ '!': true, '#': true, '$': true, '%': true, '&': true, '\'': true, '*': true, '+': true, '-': true, '.': true, '0': true, '1': true, '2': true, '3': true, '4': true, '5': true, '6': true, '7': true, '8': true, '9': true, 'A': true, 'B': true, 'C': true, 'D': true, 'E': true, 'F': true, 'G': true, 'H': true, 'I': true, 'J': true, 'K': true, 'L': true, 'M': true, 'N': true, 'O': true, 'P': true, 'Q': true, 'R': true, 'S': true, 'T': true, 'U': true, 'W': true, 'V': true, 'X': true, 'Y': true, 'Z': true, '^': true, '_': true, '`': true, 'a': true, 'b': true, 'c': true, 'd': true, 'e': true, 'f': true, 'g': true, 'h': true, 'i': true, 'j': true, 'k': true, 'l': true, 'm': true, 'n': true, 'o': true, 'p': true, 'q': true, 'r': true, 's': true, 't': true, 'u': true, 'v': true, 'w': true, 'x': true, 'y': true, 'z': true, '|': true, '~': true, } type http2connectionStater interface { ConnectionState() tls.ConnectionState } // pipe is a goroutine-safe io.Reader/io.Writer pair. It's like // io.Pipe except there are no PipeReader/PipeWriter halves, and the // underlying buffer is an interface. (io.Pipe is always unbuffered) type http2pipe struct { mu sync.Mutex c sync.Cond // c.L lazily initialized to &p.mu b http2pipeBuffer err error // read error once empty. non-nil means closed. breakErr error // immediate read error (caller doesn't see rest of b) donec chan struct{} // closed on error readFn func() // optional code to run in Read before error } type http2pipeBuffer interface { Len() int io.Writer io.Reader } // Read waits until data is available and copies bytes // from the buffer into p. func (p *http2pipe) Read(d []byte) (n int, err error) { p.mu.Lock() defer p.mu.Unlock() if p.c.L == nil { p.c.L = &p.mu } for { if p.breakErr != nil { return 0, p.breakErr } if p.b.Len() > 0 { return p.b.Read(d) } if p.err != nil { if p.readFn != nil { p.readFn() p.readFn = nil } return 0, p.err } p.c.Wait() } } var http2errClosedPipeWrite = errors.New("write on closed buffer") // Write copies bytes from p into the buffer and wakes a reader. // It is an error to write more data than the buffer can hold. func (p *http2pipe) Write(d []byte) (n int, err error) { p.mu.Lock() defer p.mu.Unlock() if p.c.L == nil { p.c.L = &p.mu } defer p.c.Signal() if p.err != nil { return 0, http2errClosedPipeWrite } return p.b.Write(d) } // CloseWithError causes the next Read (waking up a current blocked // Read if needed) to return the provided err after all data has been // read. // // The error must be non-nil. func (p *http2pipe) CloseWithError(err error) { p.closeWithError(&p.err, err, nil) } // BreakWithError causes the next Read (waking up a current blocked // Read if needed) to return the provided err immediately, without // waiting for unread data. func (p *http2pipe) BreakWithError(err error) { p.closeWithError(&p.breakErr, err, nil) } // closeWithErrorAndCode is like CloseWithError but also sets some code to run // in the caller's goroutine before returning the error. func (p *http2pipe) closeWithErrorAndCode(err error, fn func()) { p.closeWithError(&p.err, err, fn) } func (p *http2pipe) closeWithError(dst *error, err error, fn func()) { if err == nil { panic("err must be non-nil") } p.mu.Lock() defer p.mu.Unlock() if p.c.L == nil { p.c.L = &p.mu } defer p.c.Signal() if *dst != nil { return } p.readFn = fn *dst = err p.closeDoneLocked() } // requires p.mu be held. func (p *http2pipe) closeDoneLocked() { if p.donec == nil { return } select { case <-p.donec: default: close(p.donec) } } // Err returns the error (if any) first set by BreakWithError or CloseWithError. func (p *http2pipe) Err() error { p.mu.Lock() defer p.mu.Unlock() if p.breakErr != nil { return p.breakErr } return p.err } // Done returns a channel which is closed if and when this pipe is closed // with CloseWithError. func (p *http2pipe) Done() <-chan struct{} { p.mu.Lock() defer p.mu.Unlock() if p.donec == nil { p.donec = make(chan struct{}) if p.err != nil || p.breakErr != nil { p.closeDoneLocked() } } return p.donec } const ( http2prefaceTimeout = 10 * time.Second http2firstSettingsTimeout = 2 * time.Second // should be in-flight with preface anyway http2handlerChunkWriteSize = 4 << 10 http2defaultMaxStreams = 250 // TODO: make this 100 as the GFE seems to? ) var ( http2errClientDisconnected = errors.New("client disconnected") http2errClosedBody = errors.New("body closed by handler") http2errHandlerComplete = errors.New("http2: request body closed due to handler exiting") http2errStreamClosed = errors.New("http2: stream closed") ) var http2responseWriterStatePool = sync.Pool{ New: func() interface{} { rws := &http2responseWriterState{} rws.bw = bufio.NewWriterSize(http2chunkWriter{rws}, http2handlerChunkWriteSize) return rws }, } // Test hooks. var ( http2testHookOnConn func() http2testHookGetServerConn func(*http2serverConn) http2testHookOnPanicMu *sync.Mutex // nil except in tests http2testHookOnPanic func(sc *http2serverConn, panicVal interface{}) (rePanic bool) ) // Server is an HTTP/2 server. type http2Server struct { // MaxHandlers limits the number of http.Handler ServeHTTP goroutines // which may run at a time over all connections. // Negative or zero no limit. // TODO: implement MaxHandlers int // MaxConcurrentStreams optionally specifies the number of // concurrent streams that each client may have open at a // time. This is unrelated to the number of http.Handler goroutines // which may be active globally, which is MaxHandlers. // If zero, MaxConcurrentStreams defaults to at least 100, per // the HTTP/2 spec's recommendations. MaxConcurrentStreams uint32 // MaxReadFrameSize optionally specifies the largest frame // this server is willing to read. A valid value is between // 16k and 16M, inclusive. If zero or otherwise invalid, a // default value is used. MaxReadFrameSize uint32 // PermitProhibitedCipherSuites, if true, permits the use of // cipher suites prohibited by the HTTP/2 spec. PermitProhibitedCipherSuites bool } func (s *http2Server) maxReadFrameSize() uint32 { if v := s.MaxReadFrameSize; v >= http2minMaxFrameSize && v <= http2maxFrameSize { return v } return http2defaultMaxReadFrameSize } func (s *http2Server) maxConcurrentStreams() uint32 { if v := s.MaxConcurrentStreams; v > 0 { return v } return http2defaultMaxStreams } // ConfigureServer adds HTTP/2 support to a net/http Server. // // The configuration conf may be nil. // // ConfigureServer must be called before s begins serving. func http2ConfigureServer(s *Server, conf *http2Server) error { if conf == nil { conf = new(http2Server) } if s.TLSConfig == nil { s.TLSConfig = new(tls.Config) } else if s.TLSConfig.CipherSuites != nil { // If they already provided a CipherSuite list, return // an error if it has a bad order or is missing // ECDHE_RSA_WITH_AES_128_GCM_SHA256. const requiredCipher = tls.TLS_ECDHE_RSA_WITH_AES_128_GCM_SHA256 haveRequired := false sawBad := false for i, cs := range s.TLSConfig.CipherSuites { if cs == requiredCipher { haveRequired = true } if http2isBadCipher(cs) { sawBad = true } else if sawBad { return fmt.Errorf("http2: TLSConfig.CipherSuites index %d contains an HTTP/2-approved cipher suite (%#04x), but it comes after unapproved cipher suites. With this configuration, clients that don't support previous, approved cipher suites may be given an unapproved one and reject the connection.", i, cs) } } if !haveRequired { return fmt.Errorf("http2: TLSConfig.CipherSuites is missing HTTP/2-required TLS_ECDHE_RSA_WITH_AES_128_GCM_SHA256") } } s.TLSConfig.PreferServerCipherSuites = true haveNPN := false for _, p := range s.TLSConfig.NextProtos { if p == http2NextProtoTLS { haveNPN = true break } } if !haveNPN { s.TLSConfig.NextProtos = append(s.TLSConfig.NextProtos, http2NextProtoTLS) } s.TLSConfig.NextProtos = append(s.TLSConfig.NextProtos, "h2-14") if s.TLSNextProto == nil { s.TLSNextProto = map[string]func(*Server, *tls.Conn, Handler){} } protoHandler := func(hs *Server, c *tls.Conn, h Handler) { if http2testHookOnConn != nil { http2testHookOnConn() } conf.ServeConn(c, &http2ServeConnOpts{ Handler: h, BaseConfig: hs, }) } s.TLSNextProto[http2NextProtoTLS] = protoHandler s.TLSNextProto["h2-14"] = protoHandler return nil } // ServeConnOpts are options for the Server.ServeConn method. type http2ServeConnOpts struct { // BaseConfig optionally sets the base configuration // for values. If nil, defaults are used. BaseConfig *Server // Handler specifies which handler to use for processing // requests. If nil, BaseConfig.Handler is used. If BaseConfig // or BaseConfig.Handler is nil, http.DefaultServeMux is used. Handler Handler } func (o *http2ServeConnOpts) baseConfig() *Server { if o != nil && o.BaseConfig != nil { return o.BaseConfig } return new(Server) } func (o *http2ServeConnOpts) handler() Handler { if o != nil { if o.Handler != nil { return o.Handler } if o.BaseConfig != nil && o.BaseConfig.Handler != nil { return o.BaseConfig.Handler } } return DefaultServeMux } // ServeConn serves HTTP/2 requests on the provided connection and // blocks until the connection is no longer readable. // // ServeConn starts speaking HTTP/2 assuming that c has not had any // reads or writes. It writes its initial settings frame and expects // to be able to read the preface and settings frame from the // client. If c has a ConnectionState method like a *tls.Conn, the // ConnectionState is used to verify the TLS ciphersuite and to set // the Request.TLS field in Handlers. // // ServeConn does not support h2c by itself. Any h2c support must be // implemented in terms of providing a suitably-behaving net.Conn. // // The opts parameter is optional. If nil, default values are used. func (s *http2Server) ServeConn(c net.Conn, opts *http2ServeConnOpts) { sc := &http2serverConn{ srv: s, hs: opts.baseConfig(), conn: c, remoteAddrStr: c.RemoteAddr().String(), bw: http2newBufferedWriter(c), handler: opts.handler(), streams: make(map[uint32]*http2stream), readFrameCh: make(chan http2readFrameResult), wantWriteFrameCh: make(chan http2frameWriteMsg, 8), wroteFrameCh: make(chan http2frameWriteResult, 1), bodyReadCh: make(chan http2bodyReadMsg), doneServing: make(chan struct{}), advMaxStreams: s.maxConcurrentStreams(), writeSched: http2writeScheduler{ maxFrameSize: http2initialMaxFrameSize, }, initialWindowSize: http2initialWindowSize, headerTableSize: http2initialHeaderTableSize, serveG: http2newGoroutineLock(), pushEnabled: true, } sc.flow.add(http2initialWindowSize) sc.inflow.add(http2initialWindowSize) sc.hpackEncoder = hpack.NewEncoder(&sc.headerWriteBuf) sc.hpackDecoder = hpack.NewDecoder(http2initialHeaderTableSize, nil) sc.hpackDecoder.SetMaxStringLength(sc.maxHeaderStringLen()) fr := http2NewFramer(sc.bw, c) fr.SetMaxReadFrameSize(s.maxReadFrameSize()) sc.framer = fr if tc, ok := c.(http2connectionStater); ok { sc.tlsState = new(tls.ConnectionState) *sc.tlsState = tc.ConnectionState() if sc.tlsState.Version < tls.VersionTLS12 { sc.rejectConn(http2ErrCodeInadequateSecurity, "TLS version too low") return } if sc.tlsState.ServerName == "" { } if !s.PermitProhibitedCipherSuites && http2isBadCipher(sc.tlsState.CipherSuite) { sc.rejectConn(http2ErrCodeInadequateSecurity, fmt.Sprintf("Prohibited TLS 1.2 Cipher Suite: %x", sc.tlsState.CipherSuite)) return } } if hook := http2testHookGetServerConn; hook != nil { hook(sc) } sc.serve() } // isBadCipher reports whether the cipher is blacklisted by the HTTP/2 spec. func http2isBadCipher(cipher uint16) bool { switch cipher { case tls.TLS_RSA_WITH_RC4_128_SHA, tls.TLS_RSA_WITH_3DES_EDE_CBC_SHA, tls.TLS_RSA_WITH_AES_128_CBC_SHA, tls.TLS_RSA_WITH_AES_256_CBC_SHA, tls.TLS_ECDHE_ECDSA_WITH_RC4_128_SHA, tls.TLS_ECDHE_ECDSA_WITH_AES_128_CBC_SHA, tls.TLS_ECDHE_ECDSA_WITH_AES_256_CBC_SHA, tls.TLS_ECDHE_RSA_WITH_RC4_128_SHA, tls.TLS_ECDHE_RSA_WITH_3DES_EDE_CBC_SHA, tls.TLS_ECDHE_RSA_WITH_AES_128_CBC_SHA, tls.TLS_ECDHE_RSA_WITH_AES_256_CBC_SHA: return true default: return false } } func (sc *http2serverConn) rejectConn(err http2ErrCode, debug string) { sc.vlogf("http2: server rejecting conn: %v, %s", err, debug) sc.framer.WriteGoAway(0, err, []byte(debug)) sc.bw.Flush() sc.conn.Close() } type http2serverConn struct { // Immutable: srv *http2Server hs *Server conn net.Conn bw *http2bufferedWriter // writing to conn handler Handler framer *http2Framer hpackDecoder *hpack.Decoder doneServing chan struct{} // closed when serverConn.serve ends readFrameCh chan http2readFrameResult // written by serverConn.readFrames wantWriteFrameCh chan http2frameWriteMsg // from handlers -> serve wroteFrameCh chan http2frameWriteResult // from writeFrameAsync -> serve, tickles more frame writes bodyReadCh chan http2bodyReadMsg // from handlers -> serve testHookCh chan func(int) // code to run on the serve loop flow http2flow // conn-wide (not stream-specific) outbound flow control inflow http2flow // conn-wide inbound flow control tlsState *tls.ConnectionState // shared by all handlers, like net/http remoteAddrStr string // Everything following is owned by the serve loop; use serveG.check(): serveG http2goroutineLock // used to verify funcs are on serve() pushEnabled bool sawFirstSettings bool // got the initial SETTINGS frame after the preface needToSendSettingsAck bool unackedSettings int // how many SETTINGS have we sent without ACKs? clientMaxStreams uint32 // SETTINGS_MAX_CONCURRENT_STREAMS from client (our PUSH_PROMISE limit) advMaxStreams uint32 // our SETTINGS_MAX_CONCURRENT_STREAMS advertised the client curOpenStreams uint32 // client's number of open streams maxStreamID uint32 // max ever seen streams map[uint32]*http2stream initialWindowSize int32 headerTableSize uint32 peerMaxHeaderListSize uint32 // zero means unknown (default) canonHeader map[string]string // http2-lower-case -> Go-Canonical-Case req http2requestParam // non-zero while reading request headers writingFrame bool // started write goroutine but haven't heard back on wroteFrameCh needsFrameFlush bool // last frame write wasn't a flush writeSched http2writeScheduler inGoAway bool // we've started to or sent GOAWAY needToSendGoAway bool // we need to schedule a GOAWAY frame write goAwayCode http2ErrCode shutdownTimerCh <-chan time.Time // nil until used shutdownTimer *time.Timer // nil until used // Owned by the writeFrameAsync goroutine: headerWriteBuf bytes.Buffer hpackEncoder *hpack.Encoder } func (sc *http2serverConn) maxHeaderStringLen() int { v := sc.maxHeaderListSize() if uint32(int(v)) == v { return int(v) } return 0 } func (sc *http2serverConn) maxHeaderListSize() uint32 { n := sc.hs.MaxHeaderBytes if n <= 0 { n = DefaultMaxHeaderBytes } // http2's count is in a slightly different unit and includes 32 bytes per pair. // So, take the net/http.Server value and pad it up a bit, assuming 10 headers. const perFieldOverhead = 32 // per http2 spec const typicalHeaders = 10 // conservative return uint32(n + typicalHeaders*perFieldOverhead) } // requestParam is the state of the next request, initialized over // potentially several frames HEADERS + zero or more CONTINUATION // frames. type http2requestParam struct { // stream is non-nil if we're reading (HEADER or CONTINUATION) // frames for a request (but not DATA). stream *http2stream header Header method, path string scheme, authority string sawRegularHeader bool // saw a non-pseudo header already invalidHeader bool // an invalid header was seen headerListSize int64 // actually uint32, but easier math this way } // stream represents a stream. This is the minimal metadata needed by // the serve goroutine. Most of the actual stream state is owned by // the http.Handler's goroutine in the responseWriter. Because the // responseWriter's responseWriterState is recycled at the end of a // handler, this struct intentionally has no pointer to the // *responseWriter{,State} itself, as the Handler ending nils out the // responseWriter's state field. type http2stream struct { // immutable: sc *http2serverConn id uint32 body *http2pipe // non-nil if expecting DATA frames cw http2closeWaiter // closed wait stream transitions to closed state // owned by serverConn's serve loop: bodyBytes int64 // body bytes seen so far declBodyBytes int64 // or -1 if undeclared flow http2flow // limits writing from Handler to client inflow http2flow // what the client is allowed to POST/etc to us parent *http2stream // or nil numTrailerValues int64 weight uint8 state http2streamState sentReset bool // only true once detached from streams map gotReset bool // only true once detacted from streams map gotTrailerHeader bool // HEADER frame for trailers was seen trailer Header // accumulated trailers reqTrailer Header // handler's Request.Trailer } func (sc *http2serverConn) Framer() *http2Framer { return sc.framer } func (sc *http2serverConn) CloseConn() error { return sc.conn.Close() } func (sc *http2serverConn) Flush() error { return sc.bw.Flush() } func (sc *http2serverConn) HeaderEncoder() (*hpack.Encoder, *bytes.Buffer) { return sc.hpackEncoder, &sc.headerWriteBuf } func (sc *http2serverConn) state(streamID uint32) (http2streamState, *http2stream) { sc.serveG.check() if st, ok := sc.streams[streamID]; ok { return st.state, st } if streamID <= sc.maxStreamID { return http2stateClosed, nil } return http2stateIdle, nil } // setConnState calls the net/http ConnState hook for this connection, if configured. // Note that the net/http package does StateNew and StateClosed for us. // There is currently no plan for StateHijacked or hijacking HTTP/2 connections. func (sc *http2serverConn) setConnState(state ConnState) { if sc.hs.ConnState != nil { sc.hs.ConnState(sc.conn, state) } } func (sc *http2serverConn) vlogf(format string, args ...interface{}) { if http2VerboseLogs { sc.logf(format, args...) } } func (sc *http2serverConn) logf(format string, args ...interface{}) { if lg := sc.hs.ErrorLog; lg != nil { lg.Printf(format, args...) } else { log.Printf(format, args...) } } // errno returns v's underlying uintptr, else 0. // // TODO: remove this helper function once http2 can use build // tags. See comment in isClosedConnError. func http2errno(v error) uintptr { if rv := reflect.ValueOf(v); rv.Kind() == reflect.Uintptr { return uintptr(rv.Uint()) } return 0 } // isClosedConnError reports whether err is an error from use of a closed // network connection. func http2isClosedConnError(err error) bool { if err == nil { return false } str := err.Error() if strings.Contains(str, "use of closed network connection") { return true } if runtime.GOOS == "windows" { if oe, ok := err.(*net.OpError); ok && oe.Op == "read" { if se, ok := oe.Err.(*os.SyscallError); ok && se.Syscall == "wsarecv" { const WSAECONNABORTED = 10053 const WSAECONNRESET = 10054 if n := http2errno(se.Err); n == WSAECONNRESET || n == WSAECONNABORTED { return true } } } } return false } func (sc *http2serverConn) condlogf(err error, format string, args ...interface{}) { if err == nil { return } if err == io.EOF || err == io.ErrUnexpectedEOF || http2isClosedConnError(err) { sc.vlogf(format, args...) } else { sc.logf(format, args...) } } func (sc *http2serverConn) onNewHeaderField(f hpack.HeaderField) { sc.serveG.check() if http2VerboseLogs { sc.vlogf("http2: server decoded %v", f) } switch { case !http2validHeaderFieldValue(f.Value): sc.req.invalidHeader = true case strings.HasPrefix(f.Name, ":"): if sc.req.sawRegularHeader { sc.logf("pseudo-header after regular header") sc.req.invalidHeader = true return } var dst *string switch f.Name { case ":method": dst = &sc.req.method case ":path": dst = &sc.req.path case ":scheme": dst = &sc.req.scheme case ":authority": dst = &sc.req.authority default: sc.logf("invalid pseudo-header %q", f.Name) sc.req.invalidHeader = true return } if *dst != "" { sc.logf("duplicate pseudo-header %q sent", f.Name) sc.req.invalidHeader = true return } *dst = f.Value case !http2validHeaderFieldName(f.Name): sc.req.invalidHeader = true default: sc.req.sawRegularHeader = true sc.req.header.Add(sc.canonicalHeader(f.Name), f.Value) const headerFieldOverhead = 32 // per spec sc.req.headerListSize += int64(len(f.Name)) + int64(len(f.Value)) + headerFieldOverhead if sc.req.headerListSize > int64(sc.maxHeaderListSize()) { sc.hpackDecoder.SetEmitEnabled(false) } } } func (st *http2stream) onNewTrailerField(f hpack.HeaderField) { sc := st.sc sc.serveG.check() if http2VerboseLogs { sc.vlogf("http2: server decoded trailer %v", f) } switch { case strings.HasPrefix(f.Name, ":"): sc.req.invalidHeader = true return case !http2validHeaderFieldName(f.Name) || !http2validHeaderFieldValue(f.Value): sc.req.invalidHeader = true return default: key := sc.canonicalHeader(f.Name) if st.trailer != nil { vv := append(st.trailer[key], f.Value) st.trailer[key] = vv // arbitrary; TODO: read spec about header list size limits wrt trailers const tooBig = 1000 if len(vv) >= tooBig { sc.hpackDecoder.SetEmitEnabled(false) } } } } func (sc *http2serverConn) canonicalHeader(v string) string { sc.serveG.check() cv, ok := http2commonCanonHeader[v] if ok { return cv } cv, ok = sc.canonHeader[v] if ok { return cv } if sc.canonHeader == nil { sc.canonHeader = make(map[string]string) } cv = CanonicalHeaderKey(v) sc.canonHeader[v] = cv return cv } type http2readFrameResult struct { f http2Frame // valid until readMore is called err error // readMore should be called once the consumer no longer needs or // retains f. After readMore, f is invalid and more frames can be // read. readMore func() } // readFrames is the loop that reads incoming frames. // It takes care to only read one frame at a time, blocking until the // consumer is done with the frame. // It's run on its own goroutine. func (sc *http2serverConn) readFrames() { gate := make(http2gate) for { f, err := sc.framer.ReadFrame() select { case sc.readFrameCh <- http2readFrameResult{f, err, gate.Done}: case <-sc.doneServing: return } select { case <-gate: case <-sc.doneServing: return } if http2terminalReadFrameError(err) { return } } } // frameWriteResult is the message passed from writeFrameAsync to the serve goroutine. type http2frameWriteResult struct { wm http2frameWriteMsg // what was written (or attempted) err error // result of the writeFrame call } // writeFrameAsync runs in its own goroutine and writes a single frame // and then reports when it's done. // At most one goroutine can be running writeFrameAsync at a time per // serverConn. func (sc *http2serverConn) writeFrameAsync(wm http2frameWriteMsg) { err := wm.write.writeFrame(sc) sc.wroteFrameCh <- http2frameWriteResult{wm, err} } func (sc *http2serverConn) closeAllStreamsOnConnClose() { sc.serveG.check() for _, st := range sc.streams { sc.closeStream(st, http2errClientDisconnected) } } func (sc *http2serverConn) stopShutdownTimer() { sc.serveG.check() if t := sc.shutdownTimer; t != nil { t.Stop() } } func (sc *http2serverConn) notePanic() { if http2testHookOnPanicMu != nil { http2testHookOnPanicMu.Lock() defer http2testHookOnPanicMu.Unlock() } if http2testHookOnPanic != nil { if e := recover(); e != nil { if http2testHookOnPanic(sc, e) { panic(e) } } } } func (sc *http2serverConn) serve() { sc.serveG.check() defer sc.notePanic() defer sc.conn.Close() defer sc.closeAllStreamsOnConnClose() defer sc.stopShutdownTimer() defer close(sc.doneServing) if http2VerboseLogs { sc.vlogf("http2: server connection from %v on %p", sc.conn.RemoteAddr(), sc.hs) } sc.writeFrame(http2frameWriteMsg{ write: http2writeSettings{ {http2SettingMaxFrameSize, sc.srv.maxReadFrameSize()}, {http2SettingMaxConcurrentStreams, sc.advMaxStreams}, {http2SettingMaxHeaderListSize, sc.maxHeaderListSize()}, }, }) sc.unackedSettings++ if err := sc.readPreface(); err != nil { sc.condlogf(err, "http2: server: error reading preface from client %v: %v", sc.conn.RemoteAddr(), err) return } sc.setConnState(StateActive) sc.setConnState(StateIdle) go sc.readFrames() settingsTimer := time.NewTimer(http2firstSettingsTimeout) loopNum := 0 for { loopNum++ select { case wm := <-sc.wantWriteFrameCh: sc.writeFrame(wm) case res := <-sc.wroteFrameCh: sc.wroteFrame(res) case res := <-sc.readFrameCh: if !sc.processFrameFromReader(res) { return } res.readMore() if settingsTimer.C != nil { settingsTimer.Stop() settingsTimer.C = nil } case m := <-sc.bodyReadCh: sc.noteBodyRead(m.st, m.n) case <-settingsTimer.C: sc.logf("timeout waiting for SETTINGS frames from %v", sc.conn.RemoteAddr()) return case <-sc.shutdownTimerCh: sc.vlogf("GOAWAY close timer fired; closing conn from %v", sc.conn.RemoteAddr()) return case fn := <-sc.testHookCh: fn(loopNum) } } } // readPreface reads the ClientPreface greeting from the peer // or returns an error on timeout or an invalid greeting. func (sc *http2serverConn) readPreface() error { errc := make(chan error, 1) go func() { buf := make([]byte, len(http2ClientPreface)) if _, err := io.ReadFull(sc.conn, buf); err != nil { errc <- err } else if !bytes.Equal(buf, http2clientPreface) { errc <- fmt.Errorf("bogus greeting %q", buf) } else { errc <- nil } }() timer := time.NewTimer(http2prefaceTimeout) defer timer.Stop() select { case <-timer.C: return errors.New("timeout waiting for client preface") case err := <-errc: if err == nil { if http2VerboseLogs { sc.vlogf("http2: server: client %v said hello", sc.conn.RemoteAddr()) } } return err } } var http2errChanPool = sync.Pool{ New: func() interface{} { return make(chan error, 1) }, } var http2writeDataPool = sync.Pool{ New: func() interface{} { return new(http2writeData) }, } // writeDataFromHandler writes DATA response frames from a handler on // the given stream. func (sc *http2serverConn) writeDataFromHandler(stream *http2stream, data []byte, endStream bool) error { ch := http2errChanPool.Get().(chan error) writeArg := http2writeDataPool.Get().(*http2writeData) *writeArg = http2writeData{stream.id, data, endStream} err := sc.writeFrameFromHandler(http2frameWriteMsg{ write: writeArg, stream: stream, done: ch, }) if err != nil { return err } var frameWriteDone bool // the frame write is done (successfully or not) select { case err = <-ch: frameWriteDone = true case <-sc.doneServing: return http2errClientDisconnected case <-stream.cw: select { case err = <-ch: frameWriteDone = true default: return http2errStreamClosed } } http2errChanPool.Put(ch) if frameWriteDone { http2writeDataPool.Put(writeArg) } return err } // writeFrameFromHandler sends wm to sc.wantWriteFrameCh, but aborts // if the connection has gone away. // // This must not be run from the serve goroutine itself, else it might // deadlock writing to sc.wantWriteFrameCh (which is only mildly // buffered and is read by serve itself). If you're on the serve // goroutine, call writeFrame instead. func (sc *http2serverConn) writeFrameFromHandler(wm http2frameWriteMsg) error { sc.serveG.checkNotOn() select { case sc.wantWriteFrameCh <- wm: return nil case <-sc.doneServing: return http2errClientDisconnected } } // writeFrame schedules a frame to write and sends it if there's nothing // already being written. // // There is no pushback here (the serve goroutine never blocks). It's // the http.Handlers that block, waiting for their previous frames to // make it onto the wire // // If you're not on the serve goroutine, use writeFrameFromHandler instead. func (sc *http2serverConn) writeFrame(wm http2frameWriteMsg) { sc.serveG.check() sc.writeSched.add(wm) sc.scheduleFrameWrite() } // startFrameWrite starts a goroutine to write wm (in a separate // goroutine since that might block on the network), and updates the // serve goroutine's state about the world, updated from info in wm. func (sc *http2serverConn) startFrameWrite(wm http2frameWriteMsg) { sc.serveG.check() if sc.writingFrame { panic("internal error: can only be writing one frame at a time") } st := wm.stream if st != nil { switch st.state { case http2stateHalfClosedLocal: panic("internal error: attempt to send frame on half-closed-local stream") case http2stateClosed: if st.sentReset || st.gotReset { sc.scheduleFrameWrite() return } panic(fmt.Sprintf("internal error: attempt to send a write %v on a closed stream", wm)) } } sc.writingFrame = true sc.needsFrameFlush = true go sc.writeFrameAsync(wm) } // errHandlerPanicked is the error given to any callers blocked in a read from // Request.Body when the main goroutine panics. Since most handlers read in the // the main ServeHTTP goroutine, this will show up rarely. var http2errHandlerPanicked = errors.New("http2: handler panicked") // wroteFrame is called on the serve goroutine with the result of // whatever happened on writeFrameAsync. func (sc *http2serverConn) wroteFrame(res http2frameWriteResult) { sc.serveG.check() if !sc.writingFrame { panic("internal error: expected to be already writing a frame") } sc.writingFrame = false wm := res.wm st := wm.stream closeStream := http2endsStream(wm.write) if _, ok := wm.write.(http2handlerPanicRST); ok { sc.closeStream(st, http2errHandlerPanicked) } if ch := wm.done; ch != nil { select { case ch <- res.err: default: panic(fmt.Sprintf("unbuffered done channel passed in for type %T", wm.write)) } } wm.write = nil if closeStream { if st == nil { panic("internal error: expecting non-nil stream") } switch st.state { case http2stateOpen: st.state = http2stateHalfClosedLocal errCancel := http2StreamError{st.id, http2ErrCodeCancel} sc.resetStream(errCancel) case http2stateHalfClosedRemote: sc.closeStream(st, http2errHandlerComplete) } } sc.scheduleFrameWrite() } // scheduleFrameWrite tickles the frame writing scheduler. // // If a frame is already being written, nothing happens. This will be called again // when the frame is done being written. // // If a frame isn't being written we need to send one, the best frame // to send is selected, preferring first things that aren't // stream-specific (e.g. ACKing settings), and then finding the // highest priority stream. // // If a frame isn't being written and there's nothing else to send, we // flush the write buffer. func (sc *http2serverConn) scheduleFrameWrite() { sc.serveG.check() if sc.writingFrame { return } if sc.needToSendGoAway { sc.needToSendGoAway = false sc.startFrameWrite(http2frameWriteMsg{ write: &http2writeGoAway{ maxStreamID: sc.maxStreamID, code: sc.goAwayCode, }, }) return } if sc.needToSendSettingsAck { sc.needToSendSettingsAck = false sc.startFrameWrite(http2frameWriteMsg{write: http2writeSettingsAck{}}) return } if !sc.inGoAway { if wm, ok := sc.writeSched.take(); ok { sc.startFrameWrite(wm) return } } if sc.needsFrameFlush { sc.startFrameWrite(http2frameWriteMsg{write: http2flushFrameWriter{}}) sc.needsFrameFlush = false return } } func (sc *http2serverConn) goAway(code http2ErrCode) { sc.serveG.check() if sc.inGoAway { return } if code != http2ErrCodeNo { sc.shutDownIn(250 * time.Millisecond) } else { sc.shutDownIn(1 * time.Second) } sc.inGoAway = true sc.needToSendGoAway = true sc.goAwayCode = code sc.scheduleFrameWrite() } func (sc *http2serverConn) shutDownIn(d time.Duration) { sc.serveG.check() sc.shutdownTimer = time.NewTimer(d) sc.shutdownTimerCh = sc.shutdownTimer.C } func (sc *http2serverConn) resetStream(se http2StreamError) { sc.serveG.check() sc.writeFrame(http2frameWriteMsg{write: se}) if st, ok := sc.streams[se.StreamID]; ok { st.sentReset = true sc.closeStream(st, se) } } // processFrameFromReader processes the serve loop's read from readFrameCh from the // frame-reading goroutine. // processFrameFromReader returns whether the connection should be kept open. func (sc *http2serverConn) processFrameFromReader(res http2readFrameResult) bool { sc.serveG.check() err := res.err if err != nil { if err == http2ErrFrameTooLarge { sc.goAway(http2ErrCodeFrameSize) return true } clientGone := err == io.EOF || err == io.ErrUnexpectedEOF || http2isClosedConnError(err) if clientGone { return false } } else { f := res.f if http2VerboseLogs { sc.vlogf("http2: server read frame %v", http2summarizeFrame(f)) } err = sc.processFrame(f) if err == nil { return true } } switch ev := err.(type) { case http2StreamError: sc.resetStream(ev) return true case http2goAwayFlowError: sc.goAway(http2ErrCodeFlowControl) return true case http2ConnectionError: sc.logf("http2: server connection error from %v: %v", sc.conn.RemoteAddr(), ev) sc.goAway(http2ErrCode(ev)) return true default: if res.err != nil { sc.vlogf("http2: server closing client connection; error reading frame from client %s: %v", sc.conn.RemoteAddr(), err) } else { sc.logf("http2: server closing client connection: %v", err) } return false } } func (sc *http2serverConn) processFrame(f http2Frame) error { sc.serveG.check() if !sc.sawFirstSettings { if _, ok := f.(*http2SettingsFrame); !ok { return http2ConnectionError(http2ErrCodeProtocol) } sc.sawFirstSettings = true } switch f := f.(type) { case *http2SettingsFrame: return sc.processSettings(f) case *http2HeadersFrame: return sc.processHeaders(f) case *http2ContinuationFrame: return sc.processContinuation(f) case *http2WindowUpdateFrame: return sc.processWindowUpdate(f) case *http2PingFrame: return sc.processPing(f) case *http2DataFrame: return sc.processData(f) case *http2RSTStreamFrame: return sc.processResetStream(f) case *http2PriorityFrame: return sc.processPriority(f) case *http2PushPromiseFrame: return http2ConnectionError(http2ErrCodeProtocol) default: sc.vlogf("http2: server ignoring frame: %v", f.Header()) return nil } } func (sc *http2serverConn) processPing(f *http2PingFrame) error { sc.serveG.check() if f.IsAck() { return nil } if f.StreamID != 0 { return http2ConnectionError(http2ErrCodeProtocol) } sc.writeFrame(http2frameWriteMsg{write: http2writePingAck{f}}) return nil } func (sc *http2serverConn) processWindowUpdate(f *http2WindowUpdateFrame) error { sc.serveG.check() switch { case f.StreamID != 0: st := sc.streams[f.StreamID] if st == nil { return nil } if !st.flow.add(int32(f.Increment)) { return http2StreamError{f.StreamID, http2ErrCodeFlowControl} } default: if !sc.flow.add(int32(f.Increment)) { return http2goAwayFlowError{} } } sc.scheduleFrameWrite() return nil } func (sc *http2serverConn) processResetStream(f *http2RSTStreamFrame) error { sc.serveG.check() state, st := sc.state(f.StreamID) if state == http2stateIdle { return http2ConnectionError(http2ErrCodeProtocol) } if st != nil { st.gotReset = true sc.closeStream(st, http2StreamError{f.StreamID, f.ErrCode}) } return nil } func (sc *http2serverConn) closeStream(st *http2stream, err error) { sc.serveG.check() if st.state == http2stateIdle || st.state == http2stateClosed { panic(fmt.Sprintf("invariant; can't close stream in state %v", st.state)) } st.state = http2stateClosed sc.curOpenStreams-- if sc.curOpenStreams == 0 { sc.setConnState(StateIdle) } delete(sc.streams, st.id) if p := st.body; p != nil { p.CloseWithError(err) } st.cw.Close() sc.writeSched.forgetStream(st.id) } func (sc *http2serverConn) processSettings(f *http2SettingsFrame) error { sc.serveG.check() if f.IsAck() { sc.unackedSettings-- if sc.unackedSettings < 0 { return http2ConnectionError(http2ErrCodeProtocol) } return nil } if err := f.ForeachSetting(sc.processSetting); err != nil { return err } sc.needToSendSettingsAck = true sc.scheduleFrameWrite() return nil } func (sc *http2serverConn) processSetting(s http2Setting) error { sc.serveG.check() if err := s.Valid(); err != nil { return err } if http2VerboseLogs { sc.vlogf("http2: server processing setting %v", s) } switch s.ID { case http2SettingHeaderTableSize: sc.headerTableSize = s.Val sc.hpackEncoder.SetMaxDynamicTableSize(s.Val) case http2SettingEnablePush: sc.pushEnabled = s.Val != 0 case http2SettingMaxConcurrentStreams: sc.clientMaxStreams = s.Val case http2SettingInitialWindowSize: return sc.processSettingInitialWindowSize(s.Val) case http2SettingMaxFrameSize: sc.writeSched.maxFrameSize = s.Val case http2SettingMaxHeaderListSize: sc.peerMaxHeaderListSize = s.Val default: if http2VerboseLogs { sc.vlogf("http2: server ignoring unknown setting %v", s) } } return nil } func (sc *http2serverConn) processSettingInitialWindowSize(val uint32) error { sc.serveG.check() old := sc.initialWindowSize sc.initialWindowSize = int32(val) growth := sc.initialWindowSize - old for _, st := range sc.streams { if !st.flow.add(growth) { return http2ConnectionError(http2ErrCodeFlowControl) } } return nil } func (sc *http2serverConn) processData(f *http2DataFrame) error { sc.serveG.check() id := f.Header().StreamID st, ok := sc.streams[id] if !ok || st.state != http2stateOpen || st.gotTrailerHeader { return http2StreamError{id, http2ErrCodeStreamClosed} } if st.body == nil { panic("internal error: should have a body in this state") } data := f.Data() if st.declBodyBytes != -1 && st.bodyBytes+int64(len(data)) > st.declBodyBytes { st.body.CloseWithError(fmt.Errorf("sender tried to send more than declared Content-Length of %d bytes", st.declBodyBytes)) return http2StreamError{id, http2ErrCodeStreamClosed} } if len(data) > 0 { if int(st.inflow.available()) < len(data) { return http2StreamError{id, http2ErrCodeFlowControl} } st.inflow.take(int32(len(data))) wrote, err := st.body.Write(data) if err != nil { return http2StreamError{id, http2ErrCodeStreamClosed} } if wrote != len(data) { panic("internal error: bad Writer") } st.bodyBytes += int64(len(data)) } if f.StreamEnded() { st.endStream() } return nil } // endStream closes a Request.Body's pipe. It is called when a DATA // frame says a request body is over (or after trailers). func (st *http2stream) endStream() { sc := st.sc sc.serveG.check() if st.declBodyBytes != -1 && st.declBodyBytes != st.bodyBytes { st.body.CloseWithError(fmt.Errorf("request declared a Content-Length of %d but only wrote %d bytes", st.declBodyBytes, st.bodyBytes)) } else { st.body.closeWithErrorAndCode(io.EOF, st.copyTrailersToHandlerRequest) st.body.CloseWithError(io.EOF) } st.state = http2stateHalfClosedRemote } // copyTrailersToHandlerRequest is run in the Handler's goroutine in // its Request.Body.Read just before it gets io.EOF. func (st *http2stream) copyTrailersToHandlerRequest() { for k, vv := range st.trailer { if _, ok := st.reqTrailer[k]; ok { st.reqTrailer[k] = vv } } } func (sc *http2serverConn) processHeaders(f *http2HeadersFrame) error { sc.serveG.check() id := f.Header().StreamID if sc.inGoAway { return nil } if id%2 != 1 { return http2ConnectionError(http2ErrCodeProtocol) } st := sc.streams[f.Header().StreamID] if st != nil { return st.processTrailerHeaders(f) } if id <= sc.maxStreamID || sc.req.stream != nil { return http2ConnectionError(http2ErrCodeProtocol) } if id > sc.maxStreamID { sc.maxStreamID = id } st = &http2stream{ sc: sc, id: id, state: http2stateOpen, } if f.StreamEnded() { st.state = http2stateHalfClosedRemote } st.cw.Init() st.flow.conn = &sc.flow st.flow.add(sc.initialWindowSize) st.inflow.conn = &sc.inflow st.inflow.add(http2initialWindowSize) sc.streams[id] = st if f.HasPriority() { http2adjustStreamPriority(sc.streams, st.id, f.Priority) } sc.curOpenStreams++ if sc.curOpenStreams == 1 { sc.setConnState(StateActive) } sc.req = http2requestParam{ stream: st, header: make(Header), } sc.hpackDecoder.SetEmitFunc(sc.onNewHeaderField) sc.hpackDecoder.SetEmitEnabled(true) return sc.processHeaderBlockFragment(st, f.HeaderBlockFragment(), f.HeadersEnded()) } func (st *http2stream) processTrailerHeaders(f *http2HeadersFrame) error { sc := st.sc sc.serveG.check() if st.gotTrailerHeader { return http2ConnectionError(http2ErrCodeProtocol) } st.gotTrailerHeader = true if !f.StreamEnded() { return http2StreamError{st.id, http2ErrCodeProtocol} } sc.resetPendingRequest() return st.processTrailerHeaderBlockFragment(f.HeaderBlockFragment(), f.HeadersEnded()) } func (sc *http2serverConn) processContinuation(f *http2ContinuationFrame) error { sc.serveG.check() st := sc.streams[f.Header().StreamID] if st.gotTrailerHeader { return st.processTrailerHeaderBlockFragment(f.HeaderBlockFragment(), f.HeadersEnded()) } return sc.processHeaderBlockFragment(st, f.HeaderBlockFragment(), f.HeadersEnded()) } func (sc *http2serverConn) processHeaderBlockFragment(st *http2stream, frag []byte, end bool) error { sc.serveG.check() if _, err := sc.hpackDecoder.Write(frag); err != nil { return http2ConnectionError(http2ErrCodeCompression) } if !end { return nil } if err := sc.hpackDecoder.Close(); err != nil { return http2ConnectionError(http2ErrCodeCompression) } defer sc.resetPendingRequest() if sc.curOpenStreams > sc.advMaxStreams { if sc.unackedSettings == 0 { return http2StreamError{st.id, http2ErrCodeProtocol} } return http2StreamError{st.id, http2ErrCodeRefusedStream} } rw, req, err := sc.newWriterAndRequest() if err != nil { return err } st.reqTrailer = req.Trailer if st.reqTrailer != nil { st.trailer = make(Header) } st.body = req.Body.(*http2requestBody).pipe st.declBodyBytes = req.ContentLength handler := sc.handler.ServeHTTP if !sc.hpackDecoder.EmitEnabled() { handler = http2handleHeaderListTooLong } go sc.runHandler(rw, req, handler) return nil } func (st *http2stream) processTrailerHeaderBlockFragment(frag []byte, end bool) error { sc := st.sc sc.serveG.check() sc.hpackDecoder.SetEmitFunc(st.onNewTrailerField) if _, err := sc.hpackDecoder.Write(frag); err != nil { return http2ConnectionError(http2ErrCodeCompression) } if !end { return nil } rp := &sc.req if rp.invalidHeader { return http2StreamError{rp.stream.id, http2ErrCodeProtocol} } err := sc.hpackDecoder.Close() st.endStream() if err != nil { return http2ConnectionError(http2ErrCodeCompression) } return nil } func (sc *http2serverConn) processPriority(f *http2PriorityFrame) error { http2adjustStreamPriority(sc.streams, f.StreamID, f.http2PriorityParam) return nil } func http2adjustStreamPriority(streams map[uint32]*http2stream, streamID uint32, priority http2PriorityParam) { st, ok := streams[streamID] if !ok { return } st.weight = priority.Weight parent := streams[priority.StreamDep] if parent == st { return } for piter := parent; piter != nil; piter = piter.parent { if piter == st { parent.parent = st.parent break } } st.parent = parent if priority.Exclusive && (st.parent != nil || priority.StreamDep == 0) { for _, openStream := range streams { if openStream != st && openStream.parent == st.parent { openStream.parent = st } } } } // resetPendingRequest zeros out all state related to a HEADERS frame // and its zero or more CONTINUATION frames sent to start a new // request. func (sc *http2serverConn) resetPendingRequest() { sc.serveG.check() sc.req = http2requestParam{} } func (sc *http2serverConn) newWriterAndRequest() (*http2responseWriter, *Request, error) { sc.serveG.check() rp := &sc.req if rp.invalidHeader { return nil, nil, http2StreamError{rp.stream.id, http2ErrCodeProtocol} } isConnect := rp.method == "CONNECT" if isConnect { if rp.path != "" || rp.scheme != "" || rp.authority == "" { return nil, nil, http2StreamError{rp.stream.id, http2ErrCodeProtocol} } } else if rp.method == "" || rp.path == "" || (rp.scheme != "https" && rp.scheme != "http") { return nil, nil, http2StreamError{rp.stream.id, http2ErrCodeProtocol} } bodyOpen := rp.stream.state == http2stateOpen if rp.method == "HEAD" && bodyOpen { return nil, nil, http2StreamError{rp.stream.id, http2ErrCodeProtocol} } var tlsState *tls.ConnectionState // nil if not scheme https if rp.scheme == "https" { tlsState = sc.tlsState } authority := rp.authority if authority == "" { authority = rp.header.Get("Host") } needsContinue := rp.header.Get("Expect") == "100-continue" if needsContinue { rp.header.Del("Expect") } if cookies := rp.header["Cookie"]; len(cookies) > 1 { rp.header.Set("Cookie", strings.Join(cookies, "; ")) } // Setup Trailers var trailer Header for _, v := range rp.header["Trailer"] { for _, key := range strings.Split(v, ",") { key = CanonicalHeaderKey(strings.TrimSpace(key)) switch key { case "Transfer-Encoding", "Trailer", "Content-Length": default: if trailer == nil { trailer = make(Header) } trailer[key] = nil } } } delete(rp.header, "Trailer") body := &http2requestBody{ conn: sc, stream: rp.stream, needsContinue: needsContinue, } var url_ *url.URL var requestURI string if isConnect { url_ = &url.URL{Host: rp.authority} requestURI = rp.authority } else { var err error url_, err = url.ParseRequestURI(rp.path) if err != nil { return nil, nil, http2StreamError{rp.stream.id, http2ErrCodeProtocol} } requestURI = rp.path } req := &Request{ Method: rp.method, URL: url_, RemoteAddr: sc.remoteAddrStr, Header: rp.header, RequestURI: requestURI, Proto: "HTTP/2.0", ProtoMajor: 2, ProtoMinor: 0, TLS: tlsState, Host: authority, Body: body, Trailer: trailer, } if bodyOpen { body.pipe = &http2pipe{ b: &http2fixedBuffer{buf: make([]byte, http2initialWindowSize)}, } if vv, ok := rp.header["Content-Length"]; ok { req.ContentLength, _ = strconv.ParseInt(vv[0], 10, 64) } else { req.ContentLength = -1 } } rws := http2responseWriterStatePool.Get().(*http2responseWriterState) bwSave := rws.bw *rws = http2responseWriterState{} rws.conn = sc rws.bw = bwSave rws.bw.Reset(http2chunkWriter{rws}) rws.stream = rp.stream rws.req = req rws.body = body rw := &http2responseWriter{rws: rws} return rw, req, nil } // Run on its own goroutine. func (sc *http2serverConn) runHandler(rw *http2responseWriter, req *Request, handler func(ResponseWriter, *Request)) { didPanic := true defer func() { if didPanic { e := recover() // Same as net/http: const size = 64 << 10 buf := make([]byte, size) buf = buf[:runtime.Stack(buf, false)] sc.writeFrameFromHandler(http2frameWriteMsg{ write: http2handlerPanicRST{rw.rws.stream.id}, stream: rw.rws.stream, }) sc.logf("http2: panic serving %v: %v\n%s", sc.conn.RemoteAddr(), e, buf) return } rw.handlerDone() }() handler(rw, req) didPanic = false } func http2handleHeaderListTooLong(w ResponseWriter, r *Request) { // 10.5.1 Limits on Header Block Size: // .. "A server that receives a larger header block than it is // willing to handle can send an HTTP 431 (Request Header Fields Too // Large) status code" const statusRequestHeaderFieldsTooLarge = 431 // only in Go 1.6+ w.WriteHeader(statusRequestHeaderFieldsTooLarge) io.WriteString(w, "

HTTP Error 431

Request Header Field(s) Too Large

") } // called from handler goroutines. // h may be nil. func (sc *http2serverConn) writeHeaders(st *http2stream, headerData *http2writeResHeaders) error { sc.serveG.checkNotOn() var errc chan error if headerData.h != nil { errc = http2errChanPool.Get().(chan error) } if err := sc.writeFrameFromHandler(http2frameWriteMsg{ write: headerData, stream: st, done: errc, }); err != nil { return err } if errc != nil { select { case err := <-errc: http2errChanPool.Put(errc) return err case <-sc.doneServing: return http2errClientDisconnected case <-st.cw: return http2errStreamClosed } } return nil } // called from handler goroutines. func (sc *http2serverConn) write100ContinueHeaders(st *http2stream) { sc.writeFrameFromHandler(http2frameWriteMsg{ write: http2write100ContinueHeadersFrame{st.id}, stream: st, }) } // A bodyReadMsg tells the server loop that the http.Handler read n // bytes of the DATA from the client on the given stream. type http2bodyReadMsg struct { st *http2stream n int } // called from handler goroutines. // Notes that the handler for the given stream ID read n bytes of its body // and schedules flow control tokens to be sent. func (sc *http2serverConn) noteBodyReadFromHandler(st *http2stream, n int) { sc.serveG.checkNotOn() select { case sc.bodyReadCh <- http2bodyReadMsg{st, n}: case <-sc.doneServing: } } func (sc *http2serverConn) noteBodyRead(st *http2stream, n int) { sc.serveG.check() sc.sendWindowUpdate(nil, n) if st.state != http2stateHalfClosedRemote && st.state != http2stateClosed { sc.sendWindowUpdate(st, n) } } // st may be nil for conn-level func (sc *http2serverConn) sendWindowUpdate(st *http2stream, n int) { sc.serveG.check() // "The legal range for the increment to the flow control // window is 1 to 2^31-1 (2,147,483,647) octets." // A Go Read call on 64-bit machines could in theory read // a larger Read than this. Very unlikely, but we handle it here // rather than elsewhere for now. const maxUint31 = 1<<31 - 1 for n >= maxUint31 { sc.sendWindowUpdate32(st, maxUint31) n -= maxUint31 } sc.sendWindowUpdate32(st, int32(n)) } // st may be nil for conn-level func (sc *http2serverConn) sendWindowUpdate32(st *http2stream, n int32) { sc.serveG.check() if n == 0 { return } if n < 0 { panic("negative update") } var streamID uint32 if st != nil { streamID = st.id } sc.writeFrame(http2frameWriteMsg{ write: http2writeWindowUpdate{streamID: streamID, n: uint32(n)}, stream: st, }) var ok bool if st == nil { ok = sc.inflow.add(n) } else { ok = st.inflow.add(n) } if !ok { panic("internal error; sent too many window updates without decrements?") } } type http2requestBody struct { stream *http2stream conn *http2serverConn closed bool pipe *http2pipe // non-nil if we have a HTTP entity message body needsContinue bool // need to send a 100-continue } func (b *http2requestBody) Close() error { if b.pipe != nil { b.pipe.CloseWithError(http2errClosedBody) } b.closed = true return nil } func (b *http2requestBody) Read(p []byte) (n int, err error) { if b.needsContinue { b.needsContinue = false b.conn.write100ContinueHeaders(b.stream) } if b.pipe == nil { return 0, io.EOF } n, err = b.pipe.Read(p) if n > 0 { b.conn.noteBodyReadFromHandler(b.stream, n) } return } // responseWriter is the http.ResponseWriter implementation. It's // intentionally small (1 pointer wide) to minimize garbage. The // responseWriterState pointer inside is zeroed at the end of a // request (in handlerDone) and calls on the responseWriter thereafter // simply crash (caller's mistake), but the much larger responseWriterState // and buffers are reused between multiple requests. type http2responseWriter struct { rws *http2responseWriterState } // Optional http.ResponseWriter interfaces implemented. var ( _ CloseNotifier = (*http2responseWriter)(nil) _ Flusher = (*http2responseWriter)(nil) _ http2stringWriter = (*http2responseWriter)(nil) ) type http2responseWriterState struct { // immutable within a request: stream *http2stream req *Request body *http2requestBody // to close at end of request, if DATA frames didn't conn *http2serverConn // TODO: adjust buffer writing sizes based on server config, frame size updates from peer, etc bw *bufio.Writer // writing to a chunkWriter{this *responseWriterState} // mutated by http.Handler goroutine: handlerHeader Header // nil until called snapHeader Header // snapshot of handlerHeader at WriteHeader time trailers []string // set in writeChunk status int // status code passed to WriteHeader wroteHeader bool // WriteHeader called (explicitly or implicitly). Not necessarily sent to user yet. sentHeader bool // have we sent the header frame? handlerDone bool // handler has finished sentContentLen int64 // non-zero if handler set a Content-Length header wroteBytes int64 closeNotifierMu sync.Mutex // guards closeNotifierCh closeNotifierCh chan bool // nil until first used } type http2chunkWriter struct{ rws *http2responseWriterState } func (cw http2chunkWriter) Write(p []byte) (n int, err error) { return cw.rws.writeChunk(p) } func (rws *http2responseWriterState) hasTrailers() bool { return len(rws.trailers) != 0 } // declareTrailer is called for each Trailer header when the // response header is written. It notes that a header will need to be // written in the trailers at the end of the response. func (rws *http2responseWriterState) declareTrailer(k string) { k = CanonicalHeaderKey(k) switch k { case "Transfer-Encoding", "Content-Length", "Trailer": return } if !http2strSliceContains(rws.trailers, k) { rws.trailers = append(rws.trailers, k) } } // writeChunk writes chunks from the bufio.Writer. But because // bufio.Writer may bypass its chunking, sometimes p may be // arbitrarily large. // // writeChunk is also responsible (on the first chunk) for sending the // HEADER response. func (rws *http2responseWriterState) writeChunk(p []byte) (n int, err error) { if !rws.wroteHeader { rws.writeHeader(200) } isHeadResp := rws.req.Method == "HEAD" if !rws.sentHeader { rws.sentHeader = true var ctype, clen string if clen = rws.snapHeader.Get("Content-Length"); clen != "" { rws.snapHeader.Del("Content-Length") clen64, err := strconv.ParseInt(clen, 10, 64) if err == nil && clen64 >= 0 { rws.sentContentLen = clen64 } else { clen = "" } } if clen == "" && rws.handlerDone && http2bodyAllowedForStatus(rws.status) && (len(p) > 0 || !isHeadResp) { clen = strconv.Itoa(len(p)) } _, hasContentType := rws.snapHeader["Content-Type"] if !hasContentType && http2bodyAllowedForStatus(rws.status) { ctype = DetectContentType(p) } var date string if _, ok := rws.snapHeader["Date"]; !ok { date = time.Now().UTC().Format(TimeFormat) } for _, v := range rws.snapHeader["Trailer"] { http2foreachHeaderElement(v, rws.declareTrailer) } endStream := (rws.handlerDone && !rws.hasTrailers() && len(p) == 0) || isHeadResp err = rws.conn.writeHeaders(rws.stream, &http2writeResHeaders{ streamID: rws.stream.id, httpResCode: rws.status, h: rws.snapHeader, endStream: endStream, contentType: ctype, contentLength: clen, date: date, }) if err != nil { return 0, err } if endStream { return 0, nil } } if isHeadResp { return len(p), nil } if len(p) == 0 && !rws.handlerDone { return 0, nil } if rws.handlerDone { rws.promoteUndeclaredTrailers() } endStream := rws.handlerDone && !rws.hasTrailers() if len(p) > 0 || endStream { if err := rws.conn.writeDataFromHandler(rws.stream, p, endStream); err != nil { return 0, err } } if rws.handlerDone && rws.hasTrailers() { err = rws.conn.writeHeaders(rws.stream, &http2writeResHeaders{ streamID: rws.stream.id, h: rws.handlerHeader, trailers: rws.trailers, endStream: true, }) return len(p), err } return len(p), nil } // TrailerPrefix is a magic prefix for ResponseWriter.Header map keys // that, if present, signals that the map entry is actually for // the response trailers, and not the response headers. The prefix // is stripped after the ServeHTTP call finishes and the values are // sent in the trailers. // // This mechanism is intended only for trailers that are not known // prior to the headers being written. If the set of trailers is fixed // or known before the header is written, the normal Go trailers mechanism // is preferred: // https://golang.org/pkg/net/http/#ResponseWriter // https://golang.org/pkg/net/http/#example_ResponseWriter_trailers const http2TrailerPrefix = "Trailer:" // promoteUndeclaredTrailers permits http.Handlers to set trailers // after the header has already been flushed. Because the Go // ResponseWriter interface has no way to set Trailers (only the // Header), and because we didn't want to expand the ResponseWriter // interface, and because nobody used trailers, and because RFC 2616 // says you SHOULD (but not must) predeclare any trailers in the // header, the official ResponseWriter rules said trailers in Go must // be predeclared, and then we reuse the same ResponseWriter.Header() // map to mean both Headers and Trailers. When it's time to write the // Trailers, we pick out the fields of Headers that were declared as // trailers. That worked for a while, until we found the first major // user of Trailers in the wild: gRPC (using them only over http2), // and gRPC libraries permit setting trailers mid-stream without // predeclarnig them. So: change of plans. We still permit the old // way, but we also permit this hack: if a Header() key begins with // "Trailer:", the suffix of that key is a Trailer. Because ':' is an // invalid token byte anyway, there is no ambiguity. (And it's already // filtered out) It's mildly hacky, but not terrible. // // This method runs after the Handler is done and promotes any Header // fields to be trailers. func (rws *http2responseWriterState) promoteUndeclaredTrailers() { for k, vv := range rws.handlerHeader { if !strings.HasPrefix(k, http2TrailerPrefix) { continue } trailerKey := strings.TrimPrefix(k, http2TrailerPrefix) rws.declareTrailer(trailerKey) rws.handlerHeader[CanonicalHeaderKey(trailerKey)] = vv } sort.Strings(rws.trailers) } func (w *http2responseWriter) Flush() { rws := w.rws if rws == nil { panic("Header called after Handler finished") } if rws.bw.Buffered() > 0 { if err := rws.bw.Flush(); err != nil { return } } else { rws.writeChunk(nil) } } func (w *http2responseWriter) CloseNotify() <-chan bool { rws := w.rws if rws == nil { panic("CloseNotify called after Handler finished") } rws.closeNotifierMu.Lock() ch := rws.closeNotifierCh if ch == nil { ch = make(chan bool, 1) rws.closeNotifierCh = ch go func() { rws.stream.cw.Wait() ch <- true }() } rws.closeNotifierMu.Unlock() return ch } func (w *http2responseWriter) Header() Header { rws := w.rws if rws == nil { panic("Header called after Handler finished") } if rws.handlerHeader == nil { rws.handlerHeader = make(Header) } return rws.handlerHeader } func (w *http2responseWriter) WriteHeader(code int) { rws := w.rws if rws == nil { panic("WriteHeader called after Handler finished") } rws.writeHeader(code) } func (rws *http2responseWriterState) writeHeader(code int) { if !rws.wroteHeader { rws.wroteHeader = true rws.status = code if len(rws.handlerHeader) > 0 { rws.snapHeader = http2cloneHeader(rws.handlerHeader) } } } func http2cloneHeader(h Header) Header { h2 := make(Header, len(h)) for k, vv := range h { vv2 := make([]string, len(vv)) copy(vv2, vv) h2[k] = vv2 } return h2 } // The Life Of A Write is like this: // // * Handler calls w.Write or w.WriteString -> // * -> rws.bw (*bufio.Writer) -> // * (Handler migth call Flush) // * -> chunkWriter{rws} // * -> responseWriterState.writeChunk(p []byte) // * -> responseWriterState.writeChunk (most of the magic; see comment there) func (w *http2responseWriter) Write(p []byte) (n int, err error) { return w.write(len(p), p, "") } func (w *http2responseWriter) WriteString(s string) (n int, err error) { return w.write(len(s), nil, s) } // either dataB or dataS is non-zero. func (w *http2responseWriter) write(lenData int, dataB []byte, dataS string) (n int, err error) { rws := w.rws if rws == nil { panic("Write called after Handler finished") } if !rws.wroteHeader { w.WriteHeader(200) } if !http2bodyAllowedForStatus(rws.status) { return 0, ErrBodyNotAllowed } rws.wroteBytes += int64(len(dataB)) + int64(len(dataS)) if rws.sentContentLen != 0 && rws.wroteBytes > rws.sentContentLen { return 0, errors.New("http2: handler wrote more than declared Content-Length") } if dataB != nil { return rws.bw.Write(dataB) } else { return rws.bw.WriteString(dataS) } } func (w *http2responseWriter) handlerDone() { rws := w.rws rws.handlerDone = true w.Flush() w.rws = nil http2responseWriterStatePool.Put(rws) } // foreachHeaderElement splits v according to the "#rule" construction // in RFC 2616 section 2.1 and calls fn for each non-empty element. func http2foreachHeaderElement(v string, fn func(string)) { v = textproto.TrimString(v) if v == "" { return } if !strings.Contains(v, ",") { fn(v) return } for _, f := range strings.Split(v, ",") { if f = textproto.TrimString(f); f != "" { fn(f) } } } const ( // transportDefaultConnFlow is how many connection-level flow control // tokens we give the server at start-up, past the default 64k. http2transportDefaultConnFlow = 1 << 30 // transportDefaultStreamFlow is how many stream-level flow // control tokens we announce to the peer, and how many bytes // we buffer per stream. http2transportDefaultStreamFlow = 4 << 20 // transportDefaultStreamMinRefresh is the minimum number of bytes we'll send // a stream-level WINDOW_UPDATE for at a time. http2transportDefaultStreamMinRefresh = 4 << 10 http2defaultUserAgent = "Go-http-client/2.0" ) // Transport is an HTTP/2 Transport. // // A Transport internally caches connections to servers. It is safe // for concurrent use by multiple goroutines. type http2Transport struct { // DialTLS specifies an optional dial function for creating // TLS connections for requests. // // If DialTLS is nil, tls.Dial is used. // // If the returned net.Conn has a ConnectionState method like tls.Conn, // it will be used to set http.Response.TLS. DialTLS func(network, addr string, cfg *tls.Config) (net.Conn, error) // TLSClientConfig specifies the TLS configuration to use with // tls.Client. If nil, the default configuration is used. TLSClientConfig *tls.Config // ConnPool optionally specifies an alternate connection pool to use. // If nil, the default is used. ConnPool http2ClientConnPool // DisableCompression, if true, prevents the Transport from // requesting compression with an "Accept-Encoding: gzip" // request header when the Request contains no existing // Accept-Encoding value. If the Transport requests gzip on // its own and gets a gzipped response, it's transparently // decoded in the Response.Body. However, if the user // explicitly requested gzip it is not automatically // uncompressed. DisableCompression bool // MaxHeaderListSize is the http2 SETTINGS_MAX_HEADER_LIST_SIZE to // send in the initial settings frame. It is how many bytes // of response headers are allow. Unlike the http2 spec, zero here // means to use a default limit (currently 10MB). If you actually // want to advertise an ulimited value to the peer, Transport // interprets the highest possible value here (0xffffffff or 1<<32-1) // to mean no limit. MaxHeaderListSize uint32 // t1, if non-nil, is the standard library Transport using // this transport. Its settings are used (but not its // RoundTrip method, etc). t1 *Transport connPoolOnce sync.Once connPoolOrDef http2ClientConnPool // non-nil version of ConnPool } func (t *http2Transport) maxHeaderListSize() uint32 { if t.MaxHeaderListSize == 0 { return 10 << 20 } if t.MaxHeaderListSize == 0xffffffff { return 0 } return t.MaxHeaderListSize } func (t *http2Transport) disableCompression() bool { return t.DisableCompression || (t.t1 != nil && t.t1.DisableCompression) } var http2errTransportVersion = errors.New("http2: ConfigureTransport is only supported starting at Go 1.6") // ConfigureTransport configures a net/http HTTP/1 Transport to use HTTP/2. // It requires Go 1.6 or later and returns an error if the net/http package is too old // or if t1 has already been HTTP/2-enabled. func http2ConfigureTransport(t1 *Transport) error { _, err := http2configureTransport(t1) return err } func (t *http2Transport) connPool() http2ClientConnPool { t.connPoolOnce.Do(t.initConnPool) return t.connPoolOrDef } func (t *http2Transport) initConnPool() { if t.ConnPool != nil { t.connPoolOrDef = t.ConnPool } else { t.connPoolOrDef = &http2clientConnPool{t: t} } } // ClientConn is the state of a single HTTP/2 client connection to an // HTTP/2 server. type http2ClientConn struct { t *http2Transport tconn net.Conn // usually *tls.Conn, except specialized impls tlsState *tls.ConnectionState // nil only for specialized impls // readLoop goroutine fields: readerDone chan struct{} // closed on error readerErr error // set before readerDone is closed mu sync.Mutex // guards following cond *sync.Cond // hold mu; broadcast on flow/closed changes flow http2flow // our conn-level flow control quota (cs.flow is per stream) inflow http2flow // peer's conn-level flow control closed bool goAway *http2GoAwayFrame // if non-nil, the GoAwayFrame we received streams map[uint32]*http2clientStream // client-initiated nextStreamID uint32 bw *bufio.Writer br *bufio.Reader fr *http2Framer // Settings from peer: maxFrameSize uint32 maxConcurrentStreams uint32 initialWindowSize uint32 hbuf bytes.Buffer // HPACK encoder writes into this henc *hpack.Encoder freeBuf [][]byte wmu sync.Mutex // held while writing; acquire AFTER mu if holding both werr error // first write error that has occurred } // clientStream is the state for a single HTTP/2 stream. One of these // is created for each Transport.RoundTrip call. type http2clientStream struct { cc *http2ClientConn req *Request ID uint32 resc chan http2resAndError bufPipe http2pipe // buffered pipe with the flow-controlled response payload requestedGzip bool flow http2flow // guarded by cc.mu inflow http2flow // guarded by cc.mu bytesRemain int64 // -1 means unknown; owned by transportResponseBody.Read readErr error // sticky read error; owned by transportResponseBody.Read stopReqBody error // if non-nil, stop writing req body; guarded by cc.mu peerReset chan struct{} // closed on peer reset resetErr error // populated before peerReset is closed done chan struct{} // closed when stream remove from cc.streams map; close calls guarded by cc.mu // owned by clientConnReadLoop: pastHeaders bool // got HEADERS w/ END_HEADERS pastTrailers bool // got second HEADERS frame w/ END_HEADERS trailer Header // accumulated trailers resTrailer *Header // client's Response.Trailer } // awaitRequestCancel runs in its own goroutine and waits for the user // to either cancel a RoundTrip request (using the provided // Request.Cancel channel), or for the request to be done (any way it // might be removed from the cc.streams map: peer reset, successful // completion, TCP connection breakage, etc) func (cs *http2clientStream) awaitRequestCancel(cancel <-chan struct{}) { if cancel == nil { return } select { case <-cancel: cs.bufPipe.CloseWithError(http2errRequestCanceled) cs.cc.writeStreamReset(cs.ID, http2ErrCodeCancel, nil) case <-cs.done: } } // checkReset reports any error sent in a RST_STREAM frame by the // server. func (cs *http2clientStream) checkReset() error { select { case <-cs.peerReset: return cs.resetErr default: return nil } } func (cs *http2clientStream) abortRequestBodyWrite(err error) { if err == nil { panic("nil error") } cc := cs.cc cc.mu.Lock() cs.stopReqBody = err cc.cond.Broadcast() cc.mu.Unlock() } type http2stickyErrWriter struct { w io.Writer err *error } func (sew http2stickyErrWriter) Write(p []byte) (n int, err error) { if *sew.err != nil { return 0, *sew.err } n, err = sew.w.Write(p) *sew.err = err return } var http2ErrNoCachedConn = errors.New("http2: no cached connection was available") // RoundTripOpt are options for the Transport.RoundTripOpt method. type http2RoundTripOpt struct { // OnlyCachedConn controls whether RoundTripOpt may // create a new TCP connection. If set true and // no cached connection is available, RoundTripOpt // will return ErrNoCachedConn. OnlyCachedConn bool } func (t *http2Transport) RoundTrip(req *Request) (*Response, error) { return t.RoundTripOpt(req, http2RoundTripOpt{}) } // authorityAddr returns a given authority (a host/IP, or host:port / ip:port) // and returns a host:port. The port 443 is added if needed. func http2authorityAddr(authority string) (addr string) { if _, _, err := net.SplitHostPort(authority); err == nil { return authority } return net.JoinHostPort(authority, "443") } // RoundTripOpt is like RoundTrip, but takes options. func (t *http2Transport) RoundTripOpt(req *Request, opt http2RoundTripOpt) (*Response, error) { if req.URL.Scheme != "https" { return nil, errors.New("http2: unsupported scheme") } addr := http2authorityAddr(req.URL.Host) for { cc, err := t.connPool().GetClientConn(req, addr) if err != nil { t.vlogf("http2: Transport failed to get client conn for %s: %v", addr, err) return nil, err } res, err := cc.RoundTrip(req) if http2shouldRetryRequest(req, err) { continue } if err != nil { t.vlogf("RoundTrip failure: %v", err) return nil, err } return res, nil } } // CloseIdleConnections closes any connections which were previously // connected from previous requests but are now sitting idle. // It does not interrupt any connections currently in use. func (t *http2Transport) CloseIdleConnections() { if cp, ok := t.connPool().(*http2clientConnPool); ok { cp.closeIdleConnections() } } var ( http2errClientConnClosed = errors.New("http2: client conn is closed") http2errClientConnUnusable = errors.New("http2: client conn not usable") ) func http2shouldRetryRequest(req *Request, err error) bool { return err == http2errClientConnUnusable } func (t *http2Transport) dialClientConn(addr string) (*http2ClientConn, error) { host, _, err := net.SplitHostPort(addr) if err != nil { return nil, err } tconn, err := t.dialTLS()("tcp", addr, t.newTLSConfig(host)) if err != nil { return nil, err } return t.NewClientConn(tconn) } func (t *http2Transport) newTLSConfig(host string) *tls.Config { cfg := new(tls.Config) if t.TLSClientConfig != nil { *cfg = *t.TLSClientConfig } cfg.NextProtos = []string{http2NextProtoTLS} cfg.ServerName = host return cfg } func (t *http2Transport) dialTLS() func(string, string, *tls.Config) (net.Conn, error) { if t.DialTLS != nil { return t.DialTLS } return t.dialTLSDefault } func (t *http2Transport) dialTLSDefault(network, addr string, cfg *tls.Config) (net.Conn, error) { cn, err := tls.Dial(network, addr, cfg) if err != nil { return nil, err } if err := cn.Handshake(); err != nil { return nil, err } if !cfg.InsecureSkipVerify { if err := cn.VerifyHostname(cfg.ServerName); err != nil { return nil, err } } state := cn.ConnectionState() if p := state.NegotiatedProtocol; p != http2NextProtoTLS { return nil, fmt.Errorf("http2: unexpected ALPN protocol %q; want %q", p, http2NextProtoTLS) } if !state.NegotiatedProtocolIsMutual { return nil, errors.New("http2: could not negotiate protocol mutually") } return cn, nil } // disableKeepAlives reports whether connections should be closed as // soon as possible after handling the first request. func (t *http2Transport) disableKeepAlives() bool { return t.t1 != nil && t.t1.DisableKeepAlives } func (t *http2Transport) NewClientConn(c net.Conn) (*http2ClientConn, error) { if http2VerboseLogs { t.vlogf("http2: Transport creating client conn to %v", c.RemoteAddr()) } if _, err := c.Write(http2clientPreface); err != nil { t.vlogf("client preface write error: %v", err) return nil, err } cc := &http2ClientConn{ t: t, tconn: c, readerDone: make(chan struct{}), nextStreamID: 1, maxFrameSize: 16 << 10, initialWindowSize: 65535, maxConcurrentStreams: 1000, streams: make(map[uint32]*http2clientStream), } cc.cond = sync.NewCond(&cc.mu) cc.flow.add(int32(http2initialWindowSize)) cc.bw = bufio.NewWriter(http2stickyErrWriter{c, &cc.werr}) cc.br = bufio.NewReader(c) cc.fr = http2NewFramer(cc.bw, cc.br) cc.henc = hpack.NewEncoder(&cc.hbuf) if cs, ok := c.(http2connectionStater); ok { state := cs.ConnectionState() cc.tlsState = &state } initialSettings := []http2Setting{ http2Setting{ID: http2SettingEnablePush, Val: 0}, http2Setting{ID: http2SettingInitialWindowSize, Val: http2transportDefaultStreamFlow}, } if max := t.maxHeaderListSize(); max != 0 { initialSettings = append(initialSettings, http2Setting{ID: http2SettingMaxHeaderListSize, Val: max}) } cc.fr.WriteSettings(initialSettings...) cc.fr.WriteWindowUpdate(0, http2transportDefaultConnFlow) cc.inflow.add(http2transportDefaultConnFlow + http2initialWindowSize) cc.bw.Flush() if cc.werr != nil { return nil, cc.werr } f, err := cc.fr.ReadFrame() if err != nil { return nil, err } sf, ok := f.(*http2SettingsFrame) if !ok { return nil, fmt.Errorf("expected settings frame, got: %T", f) } cc.fr.WriteSettingsAck() cc.bw.Flush() sf.ForeachSetting(func(s http2Setting) error { switch s.ID { case http2SettingMaxFrameSize: cc.maxFrameSize = s.Val case http2SettingMaxConcurrentStreams: cc.maxConcurrentStreams = s.Val case http2SettingInitialWindowSize: cc.initialWindowSize = s.Val default: t.vlogf("Unhandled Setting: %v", s) } return nil }) go cc.readLoop() return cc, nil } func (cc *http2ClientConn) setGoAway(f *http2GoAwayFrame) { cc.mu.Lock() defer cc.mu.Unlock() cc.goAway = f } func (cc *http2ClientConn) CanTakeNewRequest() bool { cc.mu.Lock() defer cc.mu.Unlock() return cc.canTakeNewRequestLocked() } func (cc *http2ClientConn) canTakeNewRequestLocked() bool { return cc.goAway == nil && !cc.closed && int64(len(cc.streams)+1) < int64(cc.maxConcurrentStreams) && cc.nextStreamID < 2147483647 } func (cc *http2ClientConn) closeIfIdle() { cc.mu.Lock() if len(cc.streams) > 0 { cc.mu.Unlock() return } cc.closed = true cc.mu.Unlock() cc.tconn.Close() } const http2maxAllocFrameSize = 512 << 10 // frameBuffer returns a scratch buffer suitable for writing DATA frames. // They're capped at the min of the peer's max frame size or 512KB // (kinda arbitrarily), but definitely capped so we don't allocate 4GB // bufers. func (cc *http2ClientConn) frameScratchBuffer() []byte { cc.mu.Lock() size := cc.maxFrameSize if size > http2maxAllocFrameSize { size = http2maxAllocFrameSize } for i, buf := range cc.freeBuf { if len(buf) >= int(size) { cc.freeBuf[i] = nil cc.mu.Unlock() return buf[:size] } } cc.mu.Unlock() return make([]byte, size) } func (cc *http2ClientConn) putFrameScratchBuffer(buf []byte) { cc.mu.Lock() defer cc.mu.Unlock() const maxBufs = 4 // arbitrary; 4 concurrent requests per conn? investigate. if len(cc.freeBuf) < maxBufs { cc.freeBuf = append(cc.freeBuf, buf) return } for i, old := range cc.freeBuf { if old == nil { cc.freeBuf[i] = buf return } } } // errRequestCanceled is a copy of net/http's errRequestCanceled because it's not // exported. At least they'll be DeepEqual for h1-vs-h2 comparisons tests. var http2errRequestCanceled = errors.New("net/http: request canceled") func http2commaSeparatedTrailers(req *Request) (string, error) { keys := make([]string, 0, len(req.Trailer)) for k := range req.Trailer { k = CanonicalHeaderKey(k) switch k { case "Transfer-Encoding", "Trailer", "Content-Length": return "", &http2badStringError{"invalid Trailer key", k} } keys = append(keys, k) } if len(keys) > 0 { sort.Strings(keys) return strings.Join(keys, ","), nil } return "", nil } func (cc *http2ClientConn) responseHeaderTimeout() time.Duration { if cc.t.t1 != nil { return cc.t.t1.ResponseHeaderTimeout } return 0 } // checkConnHeaders checks whether req has any invalid connection-level headers. // per RFC 7540 section 8.1.2.2: Connection-Specific Header Fields. // Certain headers are special-cased as okay but not transmitted later. func http2checkConnHeaders(req *Request) error { if v := req.Header.Get("Upgrade"); v != "" { return errors.New("http2: invalid Upgrade request header") } if v := req.Header.Get("Transfer-Encoding"); (v != "" && v != "chunked") || len(req.Header["Transfer-Encoding"]) > 1 { return errors.New("http2: invalid Transfer-Encoding request header") } if v := req.Header.Get("Connection"); (v != "" && v != "close" && v != "keep-alive") || len(req.Header["Connection"]) > 1 { return errors.New("http2: invalid Connection request header") } return nil } func (cc *http2ClientConn) RoundTrip(req *Request) (*Response, error) { if err := http2checkConnHeaders(req); err != nil { return nil, err } trailers, err := http2commaSeparatedTrailers(req) if err != nil { return nil, err } hasTrailers := trailers != "" var body io.Reader = req.Body contentLen := req.ContentLength if req.Body != nil && contentLen == 0 { // Test to see if it's actually zero or just unset. var buf [1]byte n, rerr := io.ReadFull(body, buf[:]) if rerr != nil && rerr != io.EOF { contentLen = -1 body = http2errorReader{rerr} } else if n == 1 { contentLen = -1 body = io.MultiReader(bytes.NewReader(buf[:]), body) } else { body = nil } } cc.mu.Lock() if cc.closed || !cc.canTakeNewRequestLocked() { cc.mu.Unlock() return nil, http2errClientConnUnusable } cs := cc.newStream() cs.req = req hasBody := body != nil if !cc.t.disableCompression() && req.Header.Get("Accept-Encoding") == "" && req.Header.Get("Range") == "" && req.Method != "HEAD" { cs.requestedGzip = true } hdrs := cc.encodeHeaders(req, cs.requestedGzip, trailers, contentLen) cc.wmu.Lock() endStream := !hasBody && !hasTrailers werr := cc.writeHeaders(cs.ID, endStream, hdrs) cc.wmu.Unlock() cc.mu.Unlock() if werr != nil { if hasBody { req.Body.Close() } cc.forgetStreamID(cs.ID) return nil, werr } var respHeaderTimer <-chan time.Time var bodyCopyErrc chan error // result of body copy if hasBody { bodyCopyErrc = make(chan error, 1) go func() { bodyCopyErrc <- cs.writeRequestBody(body, req.Body) }() } else { if d := cc.responseHeaderTimeout(); d != 0 { timer := time.NewTimer(d) defer timer.Stop() respHeaderTimer = timer.C } } readLoopResCh := cs.resc requestCanceledCh := http2requestCancel(req) bodyWritten := false for { select { case re := <-readLoopResCh: res := re.res if re.err != nil || res.StatusCode > 299 { cs.abortRequestBodyWrite(http2errStopReqBodyWrite) } if re.err != nil { cc.forgetStreamID(cs.ID) return nil, re.err } res.Request = req res.TLS = cc.tlsState return res, nil case <-respHeaderTimer: cc.forgetStreamID(cs.ID) if !hasBody || bodyWritten { cc.writeStreamReset(cs.ID, http2ErrCodeCancel, nil) } else { cs.abortRequestBodyWrite(http2errStopReqBodyWriteAndCancel) } return nil, http2errTimeout case <-requestCanceledCh: cc.forgetStreamID(cs.ID) if !hasBody || bodyWritten { cc.writeStreamReset(cs.ID, http2ErrCodeCancel, nil) } else { cs.abortRequestBodyWrite(http2errStopReqBodyWriteAndCancel) } return nil, http2errRequestCanceled case <-cs.peerReset: return nil, cs.resetErr case err := <-bodyCopyErrc: if err != nil { return nil, err } bodyWritten = true if d := cc.responseHeaderTimeout(); d != 0 { timer := time.NewTimer(d) defer timer.Stop() respHeaderTimer = timer.C } } } } // requires cc.wmu be held func (cc *http2ClientConn) writeHeaders(streamID uint32, endStream bool, hdrs []byte) error { first := true frameSize := int(cc.maxFrameSize) for len(hdrs) > 0 && cc.werr == nil { chunk := hdrs if len(chunk) > frameSize { chunk = chunk[:frameSize] } hdrs = hdrs[len(chunk):] endHeaders := len(hdrs) == 0 if first { cc.fr.WriteHeaders(http2HeadersFrameParam{ StreamID: streamID, BlockFragment: chunk, EndStream: endStream, EndHeaders: endHeaders, }) first = false } else { cc.fr.WriteContinuation(streamID, endHeaders, chunk) } } cc.bw.Flush() return cc.werr } // internal error values; they don't escape to callers var ( // abort request body write; don't send cancel http2errStopReqBodyWrite = errors.New("http2: aborting request body write") // abort request body write, but send stream reset of cancel. http2errStopReqBodyWriteAndCancel = errors.New("http2: canceling request") ) func (cs *http2clientStream) writeRequestBody(body io.Reader, bodyCloser io.Closer) (err error) { cc := cs.cc sentEnd := false buf := cc.frameScratchBuffer() defer cc.putFrameScratchBuffer(buf) defer func() { cerr := bodyCloser.Close() if err == nil { err = cerr } }() req := cs.req hasTrailers := req.Trailer != nil var sawEOF bool for !sawEOF { n, err := body.Read(buf) if err == io.EOF { sawEOF = true err = nil } else if err != nil { return err } remain := buf[:n] for len(remain) > 0 && err == nil { var allowed int32 allowed, err = cs.awaitFlowControl(len(remain)) switch { case err == http2errStopReqBodyWrite: return err case err == http2errStopReqBodyWriteAndCancel: cc.writeStreamReset(cs.ID, http2ErrCodeCancel, nil) return err case err != nil: return err } cc.wmu.Lock() data := remain[:allowed] remain = remain[allowed:] sentEnd = sawEOF && len(remain) == 0 && !hasTrailers err = cc.fr.WriteData(cs.ID, sentEnd, data) if err == nil { err = cc.bw.Flush() } cc.wmu.Unlock() } if err != nil { return err } } cc.wmu.Lock() if !sentEnd { var trls []byte if hasTrailers { cc.mu.Lock() trls = cc.encodeTrailers(req) cc.mu.Unlock() } if len(trls) > 0 { err = cc.writeHeaders(cs.ID, true, trls) } else { err = cc.fr.WriteData(cs.ID, true, nil) } } if ferr := cc.bw.Flush(); ferr != nil && err == nil { err = ferr } cc.wmu.Unlock() return err } // awaitFlowControl waits for [1, min(maxBytes, cc.cs.maxFrameSize)] flow // control tokens from the server. // It returns either the non-zero number of tokens taken or an error // if the stream is dead. func (cs *http2clientStream) awaitFlowControl(maxBytes int) (taken int32, err error) { cc := cs.cc cc.mu.Lock() defer cc.mu.Unlock() for { if cc.closed { return 0, http2errClientConnClosed } if cs.stopReqBody != nil { return 0, cs.stopReqBody } if err := cs.checkReset(); err != nil { return 0, err } if a := cs.flow.available(); a > 0 { take := a if int(take) > maxBytes { take = int32(maxBytes) } if take > int32(cc.maxFrameSize) { take = int32(cc.maxFrameSize) } cs.flow.take(take) return take, nil } cc.cond.Wait() } } type http2badStringError struct { what string str string } func (e *http2badStringError) Error() string { return fmt.Sprintf("%s %q", e.what, e.str) } // requires cc.mu be held. func (cc *http2ClientConn) encodeHeaders(req *Request, addGzipHeader bool, trailers string, contentLength int64) []byte { cc.hbuf.Reset() host := req.Host if host == "" { host = req.URL.Host } cc.writeHeader(":authority", host) cc.writeHeader(":method", req.Method) if req.Method != "CONNECT" { cc.writeHeader(":path", req.URL.RequestURI()) cc.writeHeader(":scheme", "https") } if trailers != "" { cc.writeHeader("trailer", trailers) } var didUA bool for k, vv := range req.Header { lowKey := strings.ToLower(k) switch lowKey { case "host", "content-length": continue case "connection", "proxy-connection", "transfer-encoding", "upgrade": continue case "user-agent": didUA = true if len(vv) < 1 { continue } vv = vv[:1] if vv[0] == "" { continue } } for _, v := range vv { cc.writeHeader(lowKey, v) } } if http2shouldSendReqContentLength(req.Method, contentLength) { cc.writeHeader("content-length", strconv.FormatInt(contentLength, 10)) } if addGzipHeader { cc.writeHeader("accept-encoding", "gzip") } if !didUA { cc.writeHeader("user-agent", http2defaultUserAgent) } return cc.hbuf.Bytes() } // shouldSendReqContentLength reports whether the http2.Transport should send // a "content-length" request header. This logic is basically a copy of the net/http // transferWriter.shouldSendContentLength. // The contentLength is the corrected contentLength (so 0 means actually 0, not unknown). // -1 means unknown. func http2shouldSendReqContentLength(method string, contentLength int64) bool { if contentLength > 0 { return true } if contentLength < 0 { return false } switch method { case "POST", "PUT", "PATCH": return true default: return false } } // requires cc.mu be held. func (cc *http2ClientConn) encodeTrailers(req *Request) []byte { cc.hbuf.Reset() for k, vv := range req.Trailer { lowKey := strings.ToLower(k) for _, v := range vv { cc.writeHeader(lowKey, v) } } return cc.hbuf.Bytes() } func (cc *http2ClientConn) writeHeader(name, value string) { if http2VerboseLogs { log.Printf("http2: Transport encoding header %q = %q", name, value) } cc.henc.WriteField(hpack.HeaderField{Name: name, Value: value}) } type http2resAndError struct { res *Response err error } // requires cc.mu be held. func (cc *http2ClientConn) newStream() *http2clientStream { cs := &http2clientStream{ cc: cc, ID: cc.nextStreamID, resc: make(chan http2resAndError, 1), peerReset: make(chan struct{}), done: make(chan struct{}), } cs.flow.add(int32(cc.initialWindowSize)) cs.flow.setConnFlow(&cc.flow) cs.inflow.add(http2transportDefaultStreamFlow) cs.inflow.setConnFlow(&cc.inflow) cc.nextStreamID += 2 cc.streams[cs.ID] = cs return cs } func (cc *http2ClientConn) forgetStreamID(id uint32) { cc.streamByID(id, true) } func (cc *http2ClientConn) streamByID(id uint32, andRemove bool) *http2clientStream { cc.mu.Lock() defer cc.mu.Unlock() cs := cc.streams[id] if andRemove && cs != nil && !cc.closed { delete(cc.streams, id) close(cs.done) } return cs } // clientConnReadLoop is the state owned by the clientConn's frame-reading readLoop. type http2clientConnReadLoop struct { cc *http2ClientConn activeRes map[uint32]*http2clientStream // keyed by streamID closeWhenIdle bool hdec *hpack.Decoder // Fields reset on each HEADERS: nextRes *Response sawRegHeader bool // saw non-pseudo header reqMalformed error // non-nil once known to be malformed lastHeaderEndsStream bool headerListSize int64 // actually uint32, but easier math this way } // readLoop runs in its own goroutine and reads and dispatches frames. func (cc *http2ClientConn) readLoop() { rl := &http2clientConnReadLoop{ cc: cc, activeRes: make(map[uint32]*http2clientStream), } rl.hdec = hpack.NewDecoder(http2initialHeaderTableSize, rl.onNewHeaderField) defer rl.cleanup() cc.readerErr = rl.run() if ce, ok := cc.readerErr.(http2ConnectionError); ok { cc.wmu.Lock() cc.fr.WriteGoAway(0, http2ErrCode(ce), nil) cc.wmu.Unlock() } } func (rl *http2clientConnReadLoop) cleanup() { cc := rl.cc defer cc.tconn.Close() defer cc.t.connPool().MarkDead(cc) defer close(cc.readerDone) err := cc.readerErr if err == io.EOF { err = io.ErrUnexpectedEOF } cc.mu.Lock() for _, cs := range rl.activeRes { cs.bufPipe.CloseWithError(err) } for _, cs := range cc.streams { select { case cs.resc <- http2resAndError{err: err}: default: } close(cs.done) } cc.closed = true cc.cond.Broadcast() cc.mu.Unlock() } func (rl *http2clientConnReadLoop) run() error { cc := rl.cc rl.closeWhenIdle = cc.t.disableKeepAlives() gotReply := false for { f, err := cc.fr.ReadFrame() if err != nil { cc.vlogf("Transport readFrame error: (%T) %v", err, err) } if se, ok := err.(http2StreamError); ok { return se } else if err != nil { return err } if http2VerboseLogs { cc.vlogf("http2: Transport received %s", http2summarizeFrame(f)) } maybeIdle := false switch f := f.(type) { case *http2HeadersFrame: err = rl.processHeaders(f) maybeIdle = true gotReply = true case *http2ContinuationFrame: err = rl.processContinuation(f) maybeIdle = true case *http2DataFrame: err = rl.processData(f) maybeIdle = true case *http2GoAwayFrame: err = rl.processGoAway(f) maybeIdle = true case *http2RSTStreamFrame: err = rl.processResetStream(f) maybeIdle = true case *http2SettingsFrame: err = rl.processSettings(f) case *http2PushPromiseFrame: err = rl.processPushPromise(f) case *http2WindowUpdateFrame: err = rl.processWindowUpdate(f) case *http2PingFrame: err = rl.processPing(f) default: cc.logf("Transport: unhandled response frame type %T", f) } if err != nil { return err } if rl.closeWhenIdle && gotReply && maybeIdle && len(rl.activeRes) == 0 { cc.closeIfIdle() } } } func (rl *http2clientConnReadLoop) processHeaders(f *http2HeadersFrame) error { rl.sawRegHeader = false rl.reqMalformed = nil rl.lastHeaderEndsStream = f.StreamEnded() rl.headerListSize = 0 rl.nextRes = &Response{ Proto: "HTTP/2.0", ProtoMajor: 2, Header: make(Header), } rl.hdec.SetEmitEnabled(true) return rl.processHeaderBlockFragment(f.HeaderBlockFragment(), f.StreamID, f.HeadersEnded()) } func (rl *http2clientConnReadLoop) processContinuation(f *http2ContinuationFrame) error { return rl.processHeaderBlockFragment(f.HeaderBlockFragment(), f.StreamID, f.HeadersEnded()) } func (rl *http2clientConnReadLoop) processHeaderBlockFragment(frag []byte, streamID uint32, finalFrag bool) error { cc := rl.cc streamEnded := rl.lastHeaderEndsStream cs := cc.streamByID(streamID, streamEnded && finalFrag) if cs == nil { return nil } if cs.pastHeaders { rl.hdec.SetEmitFunc(func(f hpack.HeaderField) { rl.onNewTrailerField(cs, f) }) } else { rl.hdec.SetEmitFunc(rl.onNewHeaderField) } _, err := rl.hdec.Write(frag) if err != nil { return http2ConnectionError(http2ErrCodeCompression) } if finalFrag { if err := rl.hdec.Close(); err != nil { return http2ConnectionError(http2ErrCodeCompression) } } if !finalFrag { return nil } if !cs.pastHeaders { cs.pastHeaders = true } else { if cs.pastTrailers { return http2ConnectionError(http2ErrCodeProtocol) } cs.pastTrailers = true if !streamEnded { return http2ConnectionError(http2ErrCodeProtocol) } rl.endStream(cs) return nil } if rl.reqMalformed != nil { cs.resc <- http2resAndError{err: rl.reqMalformed} rl.cc.writeStreamReset(cs.ID, http2ErrCodeProtocol, rl.reqMalformed) return nil } res := rl.nextRes if res.StatusCode == 100 { cs.pastHeaders = false return nil } if !streamEnded || cs.req.Method == "HEAD" { res.ContentLength = -1 if clens := res.Header["Content-Length"]; len(clens) == 1 { if clen64, err := strconv.ParseInt(clens[0], 10, 64); err == nil { res.ContentLength = clen64 } else { } } else if len(clens) > 1 { } } if streamEnded { res.Body = http2noBody } else { buf := new(bytes.Buffer) cs.bufPipe = http2pipe{b: buf} cs.bytesRemain = res.ContentLength res.Body = http2transportResponseBody{cs} go cs.awaitRequestCancel(http2requestCancel(cs.req)) if cs.requestedGzip && res.Header.Get("Content-Encoding") == "gzip" { res.Header.Del("Content-Encoding") res.Header.Del("Content-Length") res.ContentLength = -1 res.Body = &http2gzipReader{body: res.Body} } rl.activeRes[cs.ID] = cs } cs.resTrailer = &res.Trailer cs.resc <- http2resAndError{res: res} rl.nextRes = nil return nil } // transportResponseBody is the concrete type of Transport.RoundTrip's // Response.Body. It is an io.ReadCloser. On Read, it reads from cs.body. // On Close it sends RST_STREAM if EOF wasn't already seen. type http2transportResponseBody struct { cs *http2clientStream } func (b http2transportResponseBody) Read(p []byte) (n int, err error) { cs := b.cs cc := cs.cc if cs.readErr != nil { return 0, cs.readErr } n, err = b.cs.bufPipe.Read(p) if cs.bytesRemain != -1 { if int64(n) > cs.bytesRemain { n = int(cs.bytesRemain) if err == nil { err = errors.New("net/http: server replied with more than declared Content-Length; truncated") cc.writeStreamReset(cs.ID, http2ErrCodeProtocol, err) } cs.readErr = err return int(cs.bytesRemain), err } cs.bytesRemain -= int64(n) if err == io.EOF && cs.bytesRemain > 0 { err = io.ErrUnexpectedEOF cs.readErr = err return n, err } } if n == 0 { return } cc.mu.Lock() defer cc.mu.Unlock() var connAdd, streamAdd int32 if v := cc.inflow.available(); v < http2transportDefaultConnFlow/2 { connAdd = http2transportDefaultConnFlow - v cc.inflow.add(connAdd) } if err == nil { if v := cs.inflow.available(); v < http2transportDefaultStreamFlow-http2transportDefaultStreamMinRefresh { streamAdd = http2transportDefaultStreamFlow - v cs.inflow.add(streamAdd) } } if connAdd != 0 || streamAdd != 0 { cc.wmu.Lock() defer cc.wmu.Unlock() if connAdd != 0 { cc.fr.WriteWindowUpdate(0, http2mustUint31(connAdd)) } if streamAdd != 0 { cc.fr.WriteWindowUpdate(cs.ID, http2mustUint31(streamAdd)) } cc.bw.Flush() } return } var http2errClosedResponseBody = errors.New("http2: response body closed") func (b http2transportResponseBody) Close() error { cs := b.cs if cs.bufPipe.Err() != io.EOF { cs.cc.writeStreamReset(cs.ID, http2ErrCodeCancel, nil) } cs.bufPipe.BreakWithError(http2errClosedResponseBody) return nil } func (rl *http2clientConnReadLoop) processData(f *http2DataFrame) error { cc := rl.cc cs := cc.streamByID(f.StreamID, f.StreamEnded()) if cs == nil { cc.mu.Lock() neverSent := cc.nextStreamID cc.mu.Unlock() if f.StreamID >= neverSent { cc.logf("http2: Transport received unsolicited DATA frame; closing connection") return http2ConnectionError(http2ErrCodeProtocol) } return nil } if data := f.Data(); len(data) > 0 { if cs.bufPipe.b == nil { cc.logf("http2: Transport received DATA frame for closed stream; closing connection") return http2ConnectionError(http2ErrCodeProtocol) } cc.mu.Lock() if cs.inflow.available() >= int32(len(data)) { cs.inflow.take(int32(len(data))) } else { cc.mu.Unlock() return http2ConnectionError(http2ErrCodeFlowControl) } cc.mu.Unlock() if _, err := cs.bufPipe.Write(data); err != nil { return err } } if f.StreamEnded() { rl.endStream(cs) } return nil } var http2errInvalidTrailers = errors.New("http2: invalid trailers") func (rl *http2clientConnReadLoop) endStream(cs *http2clientStream) { err := io.EOF code := cs.copyTrailers if rl.reqMalformed != nil { err = rl.reqMalformed code = nil } cs.bufPipe.closeWithErrorAndCode(err, code) delete(rl.activeRes, cs.ID) if cs.req.Close || cs.req.Header.Get("Connection") == "close" { rl.closeWhenIdle = true } } func (cs *http2clientStream) copyTrailers() { for k, vv := range cs.trailer { t := cs.resTrailer if *t == nil { *t = make(Header) } (*t)[k] = vv } } func (rl *http2clientConnReadLoop) processGoAway(f *http2GoAwayFrame) error { cc := rl.cc cc.t.connPool().MarkDead(cc) if f.ErrCode != 0 { cc.vlogf("transport got GOAWAY with error code = %v", f.ErrCode) } cc.setGoAway(f) return nil } func (rl *http2clientConnReadLoop) processSettings(f *http2SettingsFrame) error { cc := rl.cc cc.mu.Lock() defer cc.mu.Unlock() return f.ForeachSetting(func(s http2Setting) error { switch s.ID { case http2SettingMaxFrameSize: cc.maxFrameSize = s.Val case http2SettingMaxConcurrentStreams: cc.maxConcurrentStreams = s.Val case http2SettingInitialWindowSize: cc.initialWindowSize = s.Val default: cc.vlogf("Unhandled Setting: %v", s) } return nil }) } func (rl *http2clientConnReadLoop) processWindowUpdate(f *http2WindowUpdateFrame) error { cc := rl.cc cs := cc.streamByID(f.StreamID, false) if f.StreamID != 0 && cs == nil { return nil } cc.mu.Lock() defer cc.mu.Unlock() fl := &cc.flow if cs != nil { fl = &cs.flow } if !fl.add(int32(f.Increment)) { return http2ConnectionError(http2ErrCodeFlowControl) } cc.cond.Broadcast() return nil } func (rl *http2clientConnReadLoop) processResetStream(f *http2RSTStreamFrame) error { cs := rl.cc.streamByID(f.StreamID, true) if cs == nil { return nil } select { case <-cs.peerReset: default: err := http2StreamError{cs.ID, f.ErrCode} cs.resetErr = err close(cs.peerReset) cs.bufPipe.CloseWithError(err) cs.cc.cond.Broadcast() } delete(rl.activeRes, cs.ID) return nil } func (rl *http2clientConnReadLoop) processPing(f *http2PingFrame) error { if f.IsAck() { return nil } cc := rl.cc cc.wmu.Lock() defer cc.wmu.Unlock() if err := cc.fr.WritePing(true, f.Data); err != nil { return err } return cc.bw.Flush() } func (rl *http2clientConnReadLoop) processPushPromise(f *http2PushPromiseFrame) error { return http2ConnectionError(http2ErrCodeProtocol) } func (cc *http2ClientConn) writeStreamReset(streamID uint32, code http2ErrCode, err error) { cc.wmu.Lock() cc.fr.WriteRSTStream(streamID, code) cc.bw.Flush() cc.wmu.Unlock() } var ( http2errResponseHeaderListSize = errors.New("http2: response header list larger than advertised limit") http2errPseudoTrailers = errors.New("http2: invalid pseudo header in trailers") ) func (rl *http2clientConnReadLoop) checkHeaderField(f hpack.HeaderField) bool { if rl.reqMalformed != nil { return false } const headerFieldOverhead = 32 // per spec rl.headerListSize += int64(len(f.Name)) + int64(len(f.Value)) + headerFieldOverhead if max := rl.cc.t.maxHeaderListSize(); max != 0 && rl.headerListSize > int64(max) { rl.hdec.SetEmitEnabled(false) rl.reqMalformed = http2errResponseHeaderListSize return false } if !http2validHeaderFieldValue(f.Value) { rl.reqMalformed = http2errInvalidHeaderFieldValue return false } isPseudo := strings.HasPrefix(f.Name, ":") if isPseudo { if rl.sawRegHeader { rl.reqMalformed = errors.New("http2: invalid pseudo header after regular header") return false } } else { if !http2validHeaderFieldName(f.Name) { rl.reqMalformed = http2errInvalidHeaderFieldName return false } rl.sawRegHeader = true } return true } // onNewHeaderField runs on the readLoop goroutine whenever a new // hpack header field is decoded. func (rl *http2clientConnReadLoop) onNewHeaderField(f hpack.HeaderField) { cc := rl.cc if http2VerboseLogs { cc.logf("http2: Transport decoded %v", f) } if !rl.checkHeaderField(f) { return } isPseudo := strings.HasPrefix(f.Name, ":") if isPseudo { switch f.Name { case ":status": code, err := strconv.Atoi(f.Value) if err != nil { rl.reqMalformed = errors.New("http2: invalid :status") return } rl.nextRes.Status = f.Value + " " + StatusText(code) rl.nextRes.StatusCode = code default: rl.reqMalformed = fmt.Errorf("http2: unknown response pseudo header %q", f.Name) } return } key := CanonicalHeaderKey(f.Name) if key == "Trailer" { t := rl.nextRes.Trailer if t == nil { t = make(Header) rl.nextRes.Trailer = t } http2foreachHeaderElement(f.Value, func(v string) { t[CanonicalHeaderKey(v)] = nil }) } else { rl.nextRes.Header.Add(key, f.Value) } } func (rl *http2clientConnReadLoop) onNewTrailerField(cs *http2clientStream, f hpack.HeaderField) { if http2VerboseLogs { rl.cc.logf("http2: Transport decoded trailer %v", f) } if !rl.checkHeaderField(f) { return } if strings.HasPrefix(f.Name, ":") { rl.reqMalformed = http2errPseudoTrailers return } key := CanonicalHeaderKey(f.Name) // The spec says one must predeclare their trailers but in practice // popular users (which is to say the only user we found) do not so we // violate the spec and accept all of them. const acceptAllTrailers = true if _, ok := (*cs.resTrailer)[key]; ok || acceptAllTrailers { if cs.trailer == nil { cs.trailer = make(Header) } cs.trailer[key] = append(cs.trailer[key], f.Value) } } func (cc *http2ClientConn) logf(format string, args ...interface{}) { cc.t.logf(format, args...) } func (cc *http2ClientConn) vlogf(format string, args ...interface{}) { cc.t.vlogf(format, args...) } func (t *http2Transport) vlogf(format string, args ...interface{}) { if http2VerboseLogs { t.logf(format, args...) } } func (t *http2Transport) logf(format string, args ...interface{}) { log.Printf(format, args...) } var http2noBody io.ReadCloser = ioutil.NopCloser(bytes.NewReader(nil)) func http2strSliceContains(ss []string, s string) bool { for _, v := range ss { if v == s { return true } } return false } type http2erringRoundTripper struct{ err error } func (rt http2erringRoundTripper) RoundTrip(*Request) (*Response, error) { return nil, rt.err } // gzipReader wraps a response body so it can lazily // call gzip.NewReader on the first call to Read type http2gzipReader struct { body io.ReadCloser // underlying Response.Body zr *gzip.Reader // lazily-initialized gzip reader zerr error // sticky error } func (gz *http2gzipReader) Read(p []byte) (n int, err error) { if gz.zerr != nil { return 0, gz.zerr } if gz.zr == nil { gz.zr, err = gzip.NewReader(gz.body) if err != nil { gz.zerr = err return 0, err } } return gz.zr.Read(p) } func (gz *http2gzipReader) Close() error { return gz.body.Close() } type http2errorReader struct{ err error } func (r http2errorReader) Read(p []byte) (int, error) { return 0, r.err } // writeFramer is implemented by any type that is used to write frames. type http2writeFramer interface { writeFrame(http2writeContext) error } // writeContext is the interface needed by the various frame writer // types below. All the writeFrame methods below are scheduled via the // frame writing scheduler (see writeScheduler in writesched.go). // // This interface is implemented by *serverConn. // // TODO: decide whether to a) use this in the client code (which didn't // end up using this yet, because it has a simpler design, not // currently implementing priorities), or b) delete this and // make the server code a bit more concrete. type http2writeContext interface { Framer() *http2Framer Flush() error CloseConn() error // HeaderEncoder returns an HPACK encoder that writes to the // returned buffer. HeaderEncoder() (*hpack.Encoder, *bytes.Buffer) } // endsStream reports whether the given frame writer w will locally // close the stream. func http2endsStream(w http2writeFramer) bool { switch v := w.(type) { case *http2writeData: return v.endStream case *http2writeResHeaders: return v.endStream case nil: panic("endsStream called on nil writeFramer") } return false } type http2flushFrameWriter struct{} func (http2flushFrameWriter) writeFrame(ctx http2writeContext) error { return ctx.Flush() } type http2writeSettings []http2Setting func (s http2writeSettings) writeFrame(ctx http2writeContext) error { return ctx.Framer().WriteSettings([]http2Setting(s)...) } type http2writeGoAway struct { maxStreamID uint32 code http2ErrCode } func (p *http2writeGoAway) writeFrame(ctx http2writeContext) error { err := ctx.Framer().WriteGoAway(p.maxStreamID, p.code, nil) if p.code != 0 { ctx.Flush() time.Sleep(50 * time.Millisecond) ctx.CloseConn() } return err } type http2writeData struct { streamID uint32 p []byte endStream bool } func (w *http2writeData) String() string { return fmt.Sprintf("writeData(stream=%d, p=%d, endStream=%v)", w.streamID, len(w.p), w.endStream) } func (w *http2writeData) writeFrame(ctx http2writeContext) error { return ctx.Framer().WriteData(w.streamID, w.endStream, w.p) } // handlerPanicRST is the message sent from handler goroutines when // the handler panics. type http2handlerPanicRST struct { StreamID uint32 } func (hp http2handlerPanicRST) writeFrame(ctx http2writeContext) error { return ctx.Framer().WriteRSTStream(hp.StreamID, http2ErrCodeInternal) } func (se http2StreamError) writeFrame(ctx http2writeContext) error { return ctx.Framer().WriteRSTStream(se.StreamID, se.Code) } type http2writePingAck struct{ pf *http2PingFrame } func (w http2writePingAck) writeFrame(ctx http2writeContext) error { return ctx.Framer().WritePing(true, w.pf.Data) } type http2writeSettingsAck struct{} func (http2writeSettingsAck) writeFrame(ctx http2writeContext) error { return ctx.Framer().WriteSettingsAck() } // writeResHeaders is a request to write a HEADERS and 0+ CONTINUATION frames // for HTTP response headers or trailers from a server handler. type http2writeResHeaders struct { streamID uint32 httpResCode int // 0 means no ":status" line h Header // may be nil trailers []string // if non-nil, which keys of h to write. nil means all. endStream bool date string contentType string contentLength string } func http2encKV(enc *hpack.Encoder, k, v string) { if http2VerboseLogs { log.Printf("http2: server encoding header %q = %q", k, v) } enc.WriteField(hpack.HeaderField{Name: k, Value: v}) } func (w *http2writeResHeaders) writeFrame(ctx http2writeContext) error { enc, buf := ctx.HeaderEncoder() buf.Reset() if w.httpResCode != 0 { http2encKV(enc, ":status", http2httpCodeString(w.httpResCode)) } http2encodeHeaders(enc, w.h, w.trailers) if w.contentType != "" { http2encKV(enc, "content-type", w.contentType) } if w.contentLength != "" { http2encKV(enc, "content-length", w.contentLength) } if w.date != "" { http2encKV(enc, "date", w.date) } headerBlock := buf.Bytes() if len(headerBlock) == 0 && w.trailers == nil { panic("unexpected empty hpack") } // For now we're lazy and just pick the minimum MAX_FRAME_SIZE // that all peers must support (16KB). Later we could care // more and send larger frames if the peer advertised it, but // there's little point. Most headers are small anyway (so we // generally won't have CONTINUATION frames), and extra frames // only waste 9 bytes anyway. const maxFrameSize = 16384 first := true for len(headerBlock) > 0 { frag := headerBlock if len(frag) > maxFrameSize { frag = frag[:maxFrameSize] } headerBlock = headerBlock[len(frag):] endHeaders := len(headerBlock) == 0 var err error if first { first = false err = ctx.Framer().WriteHeaders(http2HeadersFrameParam{ StreamID: w.streamID, BlockFragment: frag, EndStream: w.endStream, EndHeaders: endHeaders, }) } else { err = ctx.Framer().WriteContinuation(w.streamID, endHeaders, frag) } if err != nil { return err } } return nil } type http2write100ContinueHeadersFrame struct { streamID uint32 } func (w http2write100ContinueHeadersFrame) writeFrame(ctx http2writeContext) error { enc, buf := ctx.HeaderEncoder() buf.Reset() http2encKV(enc, ":status", "100") return ctx.Framer().WriteHeaders(http2HeadersFrameParam{ StreamID: w.streamID, BlockFragment: buf.Bytes(), EndStream: false, EndHeaders: true, }) } type http2writeWindowUpdate struct { streamID uint32 // or 0 for conn-level n uint32 } func (wu http2writeWindowUpdate) writeFrame(ctx http2writeContext) error { return ctx.Framer().WriteWindowUpdate(wu.streamID, wu.n) } func http2encodeHeaders(enc *hpack.Encoder, h Header, keys []string) { if keys == nil { keys = make([]string, 0, len(h)) for k := range h { keys = append(keys, k) } sort.Strings(keys) } for _, k := range keys { vv := h[k] k = http2lowerHeader(k) if !http2validHeaderFieldName(k) { continue } isTE := k == "transfer-encoding" for _, v := range vv { if !http2validHeaderFieldValue(v) { continue } if isTE && v != "trailers" { continue } http2encKV(enc, k, v) } } } // frameWriteMsg is a request to write a frame. type http2frameWriteMsg struct { // write is the interface value that does the writing, once the // writeScheduler (below) has decided to select this frame // to write. The write functions are all defined in write.go. write http2writeFramer stream *http2stream // used for prioritization. nil for non-stream frames. // done, if non-nil, must be a buffered channel with space for // 1 message and is sent the return value from write (or an // earlier error) when the frame has been written. done chan error } // for debugging only: func (wm http2frameWriteMsg) String() string { var streamID uint32 if wm.stream != nil { streamID = wm.stream.id } var des string if s, ok := wm.write.(fmt.Stringer); ok { des = s.String() } else { des = fmt.Sprintf("%T", wm.write) } return fmt.Sprintf("[frameWriteMsg stream=%d, ch=%v, type: %v]", streamID, wm.done != nil, des) } // writeScheduler tracks pending frames to write, priorities, and decides // the next one to use. It is not thread-safe. type http2writeScheduler struct { // zero are frames not associated with a specific stream. // They're sent before any stream-specific freams. zero http2writeQueue // maxFrameSize is the maximum size of a DATA frame // we'll write. Must be non-zero and between 16K-16M. maxFrameSize uint32 // sq contains the stream-specific queues, keyed by stream ID. // when a stream is idle, it's deleted from the map. sq map[uint32]*http2writeQueue // canSend is a slice of memory that's reused between frame // scheduling decisions to hold the list of writeQueues (from sq) // which have enough flow control data to send. After canSend is // built, the best is selected. canSend []*http2writeQueue // pool of empty queues for reuse. queuePool []*http2writeQueue } func (ws *http2writeScheduler) putEmptyQueue(q *http2writeQueue) { if len(q.s) != 0 { panic("queue must be empty") } ws.queuePool = append(ws.queuePool, q) } func (ws *http2writeScheduler) getEmptyQueue() *http2writeQueue { ln := len(ws.queuePool) if ln == 0 { return new(http2writeQueue) } q := ws.queuePool[ln-1] ws.queuePool = ws.queuePool[:ln-1] return q } func (ws *http2writeScheduler) empty() bool { return ws.zero.empty() && len(ws.sq) == 0 } func (ws *http2writeScheduler) add(wm http2frameWriteMsg) { st := wm.stream if st == nil { ws.zero.push(wm) } else { ws.streamQueue(st.id).push(wm) } } func (ws *http2writeScheduler) streamQueue(streamID uint32) *http2writeQueue { if q, ok := ws.sq[streamID]; ok { return q } if ws.sq == nil { ws.sq = make(map[uint32]*http2writeQueue) } q := ws.getEmptyQueue() ws.sq[streamID] = q return q } // take returns the most important frame to write and removes it from the scheduler. // It is illegal to call this if the scheduler is empty or if there are no connection-level // flow control bytes available. func (ws *http2writeScheduler) take() (wm http2frameWriteMsg, ok bool) { if ws.maxFrameSize == 0 { panic("internal error: ws.maxFrameSize not initialized or invalid") } if !ws.zero.empty() { return ws.zero.shift(), true } if len(ws.sq) == 0 { return } for id, q := range ws.sq { if q.firstIsNoCost() { return ws.takeFrom(id, q) } } if len(ws.canSend) != 0 { panic("should be empty") } for _, q := range ws.sq { if n := ws.streamWritableBytes(q); n > 0 { ws.canSend = append(ws.canSend, q) } } if len(ws.canSend) == 0 { return } defer ws.zeroCanSend() q := ws.canSend[0] return ws.takeFrom(q.streamID(), q) } // zeroCanSend is defered from take. func (ws *http2writeScheduler) zeroCanSend() { for i := range ws.canSend { ws.canSend[i] = nil } ws.canSend = ws.canSend[:0] } // streamWritableBytes returns the number of DATA bytes we could write // from the given queue's stream, if this stream/queue were // selected. It is an error to call this if q's head isn't a // *writeData. func (ws *http2writeScheduler) streamWritableBytes(q *http2writeQueue) int32 { wm := q.head() ret := wm.stream.flow.available() if ret == 0 { return 0 } if int32(ws.maxFrameSize) < ret { ret = int32(ws.maxFrameSize) } if ret == 0 { panic("internal error: ws.maxFrameSize not initialized or invalid") } wd := wm.write.(*http2writeData) if len(wd.p) < int(ret) { ret = int32(len(wd.p)) } return ret } func (ws *http2writeScheduler) takeFrom(id uint32, q *http2writeQueue) (wm http2frameWriteMsg, ok bool) { wm = q.head() if wd, ok := wm.write.(*http2writeData); ok && len(wd.p) > 0 { allowed := wm.stream.flow.available() if allowed == 0 { return http2frameWriteMsg{}, false } if int32(ws.maxFrameSize) < allowed { allowed = int32(ws.maxFrameSize) } if len(wd.p) > int(allowed) { wm.stream.flow.take(allowed) chunk := wd.p[:allowed] wd.p = wd.p[allowed:] return http2frameWriteMsg{ stream: wm.stream, write: &http2writeData{ streamID: wd.streamID, p: chunk, endStream: false, }, done: nil, }, true } wm.stream.flow.take(int32(len(wd.p))) } q.shift() if q.empty() { ws.putEmptyQueue(q) delete(ws.sq, id) } return wm, true } func (ws *http2writeScheduler) forgetStream(id uint32) { q, ok := ws.sq[id] if !ok { return } delete(ws.sq, id) for i := range q.s { q.s[i] = http2frameWriteMsg{} } q.s = q.s[:0] ws.putEmptyQueue(q) } type http2writeQueue struct { s []http2frameWriteMsg } // streamID returns the stream ID for a non-empty stream-specific queue. func (q *http2writeQueue) streamID() uint32 { return q.s[0].stream.id } func (q *http2writeQueue) empty() bool { return len(q.s) == 0 } func (q *http2writeQueue) push(wm http2frameWriteMsg) { q.s = append(q.s, wm) } // head returns the next item that would be removed by shift. func (q *http2writeQueue) head() http2frameWriteMsg { if len(q.s) == 0 { panic("invalid use of queue") } return q.s[0] } func (q *http2writeQueue) shift() http2frameWriteMsg { if len(q.s) == 0 { panic("invalid use of queue") } wm := q.s[0] copy(q.s, q.s[1:]) q.s[len(q.s)-1] = http2frameWriteMsg{} q.s = q.s[:len(q.s)-1] return wm } func (q *http2writeQueue) firstIsNoCost() bool { if df, ok := q.s[0].write.(*http2writeData); ok { return len(df.p) == 0 } return true }