Source file src/net/http/h2_bundle.go
package http
import (
"bufio"
"bytes"
"compress/gzip"
"context"
"crypto/rand"
"crypto/tls"
"encoding/binary"
"errors"
"fmt"
"io"
"io/ioutil"
"log"
"math"
"net"
"net/http/httptrace"
"net/textproto"
"net/url"
"os"
"reflect"
"runtime"
"sort"
"strconv"
"strings"
"sync"
"time"
"golang_org/x/net/http2/hpack"
"golang_org/x/net/idna"
"golang_org/x/net/lex/httplex"
)
type http2ClientConnPool interface {
GetClientConn(req *Request, addr string) (*http2ClientConn, error)
MarkDead(*http2ClientConn)
}
type http2clientConnPoolIdleCloser interface {
http2ClientConnPool
closeIdleConnections()
}
var (
_ http2clientConnPoolIdleCloser = (*http2clientConnPool)(nil)
_ http2clientConnPoolIdleCloser = http2noDialClientConnPool{}
)
type http2clientConnPool struct {
t *http2Transport
mu sync.Mutex
conns map[string][]*http2ClientConn
dialing map[string]*http2dialCall
keys map[*http2ClientConn][]string
addConnCalls map[string]*http2addConnCall
}
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(req *Request, addr string, dialOnMiss bool) (*http2ClientConn, error) {
if http2isConnectionCloseRequest(req) && dialOnMiss {
const singleUse = true
cc, err := p.t.dialClientConn(addr, singleUse)
if err != nil {
return nil, err
}
return cc, nil
}
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
}
type http2dialCall struct {
p *http2clientConnPool
done chan struct{}
res *http2ClientConn
err error
}
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
}
func (c *http2dialCall) dial(addr string) {
const singleUse = false
c.res, c.err = c.p.t.dialClientConn(addr, singleUse)
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()
}
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{}
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()
}
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
}
type http2noDialClientConnPool struct{ *http2clientConnPool }
func (p http2noDialClientConnPool) GetClientConn(req *Request, addr string) (*http2ClientConn, error) {
return p.getClientConn(req, addr, http2noDialOnMiss)
}
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("https", 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
}
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
}
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
}
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))
}
type http2ConnectionError http2ErrCode
func (e http2ConnectionError) Error() string {
return fmt.Sprintf("connection error: %s", http2ErrCode(e))
}
type http2StreamError struct {
StreamID uint32
Code http2ErrCode
Cause error
}
func http2streamError(id uint32, code http2ErrCode) http2StreamError {
return http2StreamError{StreamID: id, Code: code}
}
func (e http2StreamError) Error() string {
if e.Cause != nil {
return fmt.Sprintf("stream error: stream ID %d; %v; %v", e.StreamID, e.Code, e.Cause)
}
return fmt.Sprintf("stream error: stream ID %d; %v", e.StreamID, e.Code)
}
type http2goAwayFlowError struct{}
func (http2goAwayFlowError) Error() string { return "connection exceeded flow control window size" }
type http2connError struct {
Code http2ErrCode
Reason string
}
func (e http2connError) Error() string {
return fmt.Sprintf("http2: connection error: %v: %v", e.Code, e.Reason)
}
type http2pseudoHeaderError string
func (e http2pseudoHeaderError) Error() string {
return fmt.Sprintf("invalid pseudo-header %q", string(e))
}
type http2duplicatePseudoHeaderError string
func (e http2duplicatePseudoHeaderError) Error() string {
return fmt.Sprintf("duplicate pseudo-header %q", string(e))
}
type http2headerFieldNameError string
func (e http2headerFieldNameError) Error() string {
return fmt.Sprintf("invalid header field name %q", string(e))
}
type http2headerFieldValueError string
func (e http2headerFieldValueError) Error() string {
return fmt.Sprintf("invalid header field value %q", string(e))
}
var (
http2errMixPseudoHeaderTypes = errors.New("mix of request and response pseudo headers")
http2errPseudoAfterRegular = errors.New("pseudo header field after regular")
)
type http2fixedBuffer struct {
buf []byte
r, w int
}
var (
http2errReadEmpty = errors.New("read from empty fixedBuffer")
http2errWriteFull = errors.New("write on full fixedBuffer")
)
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
}
func (b *http2fixedBuffer) Len() int {
return b.w - b.r
}
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
}
type http2flow struct {
n int32
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
}
}
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)
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))
}
type http2Flags uint8
func (f http2Flags) Has(v http2Flags) bool {
return (f & v) == v
}
const (
http2FlagDataEndStream http2Flags = 0x1
http2FlagDataPadded http2Flags = 0x8
http2FlagHeadersEndStream http2Flags = 0x1
http2FlagHeadersEndHeaders http2Flags = 0x4
http2FlagHeadersPadded http2Flags = 0x8
http2FlagHeadersPriority http2Flags = 0x20
http2FlagSettingsAck http2Flags = 0x1
http2FlagPingAck http2Flags = 0x1
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",
},
}
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
}
type http2FrameHeader struct {
valid bool
Type http2FrameType
Flags http2Flags
Length uint32
StreamID uint32
}
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<<i) == 0 {
continue
}
set++
if set > 1 {
buf.WriteByte('|')
}
name := http2flagName[h.Type][http2Flags(1<<i)]
if name != "" {
buf.WriteString(name)
} else {
fmt.Fprintf(buf, "0x%x", 1<<i)
}
}
}
if h.StreamID != 0 {
fmt.Fprintf(buf, " stream=%d", h.StreamID)
}
fmt.Fprintf(buf, " len=%d", h.Length)
}
func (h *http2FrameHeader) checkValid() {
if !h.valid {
panic("Frame accessor called on non-owned Frame")
}
}
func (h *http2FrameHeader) invalidate() { h.valid = false }
var http2fhBytes = sync.Pool{
New: func() interface{} {
buf := make([]byte, http2frameHeaderLen)
return &buf
},
}
func http2ReadFrameHeader(r io.Reader) (http2FrameHeader, error) {
bufp := http2fhBytes.Get().(*[]byte)
defer http2fhBytes.Put(bufp)
return http2readFrameHeader(*bufp, r)
}
func http2readFrameHeader(buf []byte, r io.Reader) (http2FrameHeader, error) {
_, err := io.ReadFull(r, buf[:http2frameHeaderLen])
if err != nil {
return http2FrameHeader{}, err
}
return http2FrameHeader{
Length: (uint32(buf[0])<<16 | uint32(buf[1])<<8 | uint32(buf[2])),
Type: http2FrameType(buf[3]),
Flags: http2Flags(buf[4]),
StreamID: binary.BigEndian.Uint32(buf[5:]) & (1<<31 - 1),
valid: true,
}, nil
}
type http2Frame interface {
Header() http2FrameHeader
invalidate()
}
type http2Framer struct {
r io.Reader
lastFrame http2Frame
errDetail error
lastHeaderStream uint32
maxReadSize uint32
headerBuf [http2frameHeaderLen]byte
getReadBuf func(size uint32) []byte
readBuf []byte
maxWriteSize uint32
w io.Writer
wbuf []byte
AllowIllegalWrites bool
AllowIllegalReads bool
ReadMetaHeaders *hpack.Decoder
MaxHeaderListSize uint32
logReads, logWrites bool
debugFramer *http2Framer
debugFramerBuf *bytes.Buffer
debugReadLoggerf func(string, ...interface{})
debugWriteLoggerf func(string, ...interface{})
}
func (fr *http2Framer) maxHeaderListSize() uint32 {
if fr.MaxHeaderListSize == 0 {
return 16 << 20
}
return fr.MaxHeaderListSize
}
func (f *http2Framer) startWrite(ftype http2FrameType, flags http2Flags, streamID uint32) {
f.wbuf = append(f.wbuf[:0],
0,
0,
0,
byte(ftype),
byte(flags),
byte(streamID>>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 f.logWrites {
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 {
f.debugWriteLoggerf("http2: Framer %p: failed to decode just-written frame", f)
return
}
f.debugWriteLoggerf("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
)
func http2NewFramer(w io.Writer, r io.Reader) *http2Framer {
fr := &http2Framer{
w: w,
r: r,
logReads: http2logFrameReads,
logWrites: http2logFrameWrites,
debugReadLoggerf: log.Printf,
debugWriteLoggerf: log.Printf,
}
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
}
func (fr *http2Framer) SetMaxReadFrameSize(v uint32) {
if v > http2maxFrameSize {
v = http2maxFrameSize
}
fr.maxReadSize = v
}
func (fr *http2Framer) ErrorDetail() error {
return fr.errDetail
}
var http2ErrFrameTooLarge = errors.New("http2: frame too large")
func http2terminalReadFrameError(err error) bool {
if _, ok := err.(http2StreamError); ok {
return false
}
return err != nil
}
func (fr *http2Framer) ReadFrame() (http2Frame, error) {
fr.errDetail = nil
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 {
fr.debugReadLoggerf("http2: Framer %p: read %v", fr, http2summarizeFrame(f))
}
if fh.Type == http2FrameHeaders && fr.ReadMetaHeaders != nil {
return fr.readMetaFrame(f.(*http2HeadersFrame))
}
return f, nil
}
func (fr *http2Framer) connError(code http2ErrCode, reason string) error {
fr.errDetail = errors.New(reason)
return http2ConnectionError(code)
}
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
}
type http2DataFrame struct {
http2FrameHeader
data []byte
}
func (f *http2DataFrame) StreamEnded() bool {
return f.http2FrameHeader.Flags.Has(http2FlagDataEndStream)
}
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 stream ID")
http2errDepStreamID = errors.New("invalid dependent stream ID")
http2errPadLength = errors.New("pad length too large")
)
func http2validStreamIDOrZero(streamID uint32) bool {
return streamID&(1<<31) == 0
}
func http2validStreamID(streamID uint32) bool {
return streamID != 0 && streamID&(1<<31) == 0
}
func (f *http2Framer) WriteData(streamID uint32, endStream bool, data []byte) error {
return f.WriteDataPadded(streamID, endStream, data, nil)
}
func (f *http2Framer) WriteDataPadded(streamID uint32, endStream bool, data, pad []byte) error {
if !http2validStreamID(streamID) && !f.AllowIllegalWrites {
return http2errStreamID
}
if len(pad) > 255 {
return http2errPadLength
}
var flags http2Flags
if endStream {
flags |= http2FlagDataEndStream
}
if pad != nil {
flags |= http2FlagDataPadded
}
f.startWrite(http2FrameData, flags, streamID)
if pad != nil {
f.wbuf = append(f.wbuf, byte(len(pad)))
}
f.wbuf = append(f.wbuf, data...)
f.wbuf = append(f.wbuf, pad...)
return f.endWrite()
}
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
}
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
}
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()
}
func (f *http2Framer) WriteSettingsAck() error {
f.startWrite(http2FrameSettings, http2FlagSettingsAck, 0)
return f.endWrite()
}
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()
}
type http2GoAwayFrame struct {
http2FrameHeader
LastStreamID uint32
ErrCode http2ErrCode
debugData []byte
}
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()
}
type http2UnknownFrame struct {
http2FrameHeader
p []byte
}
func (f *http2UnknownFrame) Payload() []byte {
f.checkValid()
return f.p
}
func http2parseUnknownFrame(fh http2FrameHeader, p []byte) (http2Frame, error) {
return &http2UnknownFrame{fh, p}, nil
}
type http2WindowUpdateFrame struct {
http2FrameHeader
Increment uint32
}
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
}
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()
}
type http2HeadersFrame struct {
http2FrameHeader
Priority http2PriorityParam
headerFragBuf []byte
}
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
}
type http2HeadersFrameParam struct {
StreamID uint32
BlockFragment []byte
EndStream bool
EndHeaders bool
PadLength uint8
Priority http2PriorityParam
}
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 !http2validStreamIDOrZero(v) && !f.AllowIllegalWrites {
return http2errDepStreamID
}
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()
}
type http2PriorityFrame struct {
http2FrameHeader
http2PriorityParam
}
type http2PriorityParam struct {
StreamDep uint32
Exclusive bool
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
}
func (f *http2Framer) WritePriority(streamID uint32, p http2PriorityParam) error {
if !http2validStreamID(streamID) && !f.AllowIllegalWrites {
return http2errStreamID
}
if !http2validStreamIDOrZero(p.StreamDep) {
return http2errDepStreamID
}
f.startWrite(http2FramePriority, 0, streamID)
v := p.StreamDep
if p.Exclusive {
v |= 1 << 31
}
f.writeUint32(v)
f.writeByte(p.Weight)
return f.endWrite()
}
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
}
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()
}
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)
}
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()
}
type http2PushPromiseFrame struct {
http2FrameHeader
PromiseID uint32
headerFragBuf []byte
}
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)
}
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
}
type http2PushPromiseParam struct {
StreamID uint32
PromiseID uint32
BlockFragment []byte
EndHeaders bool
PadLength uint8
}
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()
}
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
}
type http2headersOrContinuation interface {
http2headersEnder
HeaderBlockFragment() []byte
}
type http2MetaHeadersFrame struct {
*http2HeadersFrame
Fields []hpack.HeaderField
Truncated bool
}
func (mh *http2MetaHeadersFrame) PseudoValue(pseudo string) string {
for _, hf := range mh.Fields {
if !hf.IsPseudo() {
return ""
}
if hf.Name[1:] == pseudo {
return hf.Value
}
}
return ""
}
func (mh *http2MetaHeadersFrame) RegularFields() []hpack.HeaderField {
for i, hf := range mh.Fields {
if !hf.IsPseudo() {
return mh.Fields[i:]
}
}
return nil
}
func (mh *http2MetaHeadersFrame) PseudoFields() []hpack.HeaderField {
for i, hf := range mh.Fields {
if !hf.IsPseudo() {
return mh.Fields[:i]
}
}
return mh.Fields
}
func (mh *http2MetaHeadersFrame) checkPseudos() error {
var isRequest, isResponse bool
pf := mh.PseudoFields()
for i, hf := range pf {
switch hf.Name {
case ":method", ":path", ":scheme", ":authority":
isRequest = true
case ":status":
isResponse = true
default:
return http2pseudoHeaderError(hf.Name)
}
for _, hf2 := range pf[:i] {
if hf.Name == hf2.Name {
return http2duplicatePseudoHeaderError(hf.Name)
}
}
}
if isRequest && isResponse {
return http2errMixPseudoHeaderTypes
}
return nil
}
func (fr *http2Framer) maxHeaderStringLen() int {
v := fr.maxHeaderListSize()
if uint32(int(v)) == v {
return int(v)
}
return 0
}
func (fr *http2Framer) readMetaFrame(hf *http2HeadersFrame) (*http2MetaHeadersFrame, error) {
if fr.AllowIllegalReads {
return nil, errors.New("illegal use of AllowIllegalReads with ReadMetaHeaders")
}
mh := &http2MetaHeadersFrame{
http2HeadersFrame: hf,
}
var remainSize = fr.maxHeaderListSize()
var sawRegular bool
var invalid error
hdec := fr.ReadMetaHeaders
hdec.SetEmitEnabled(true)
hdec.SetMaxStringLength(fr.maxHeaderStringLen())
hdec.SetEmitFunc(func(hf hpack.HeaderField) {
if http2VerboseLogs && fr.logReads {
fr.debugReadLoggerf("http2: decoded hpack field %+v", hf)
}
if !httplex.ValidHeaderFieldValue(hf.Value) {
invalid = http2headerFieldValueError(hf.Value)
}
isPseudo := strings.HasPrefix(hf.Name, ":")
if isPseudo {
if sawRegular {
invalid = http2errPseudoAfterRegular
}
} else {
sawRegular = true
if !http2validWireHeaderFieldName(hf.Name) {
invalid = http2headerFieldNameError(hf.Name)
}
}
if invalid != nil {
hdec.SetEmitEnabled(false)
return
}
size := hf.Size()
if size > remainSize {
hdec.SetEmitEnabled(false)
mh.Truncated = true
return
}
remainSize -= size
mh.Fields = append(mh.Fields, hf)
})
defer hdec.SetEmitFunc(func(hf hpack.HeaderField) {})
var hc http2headersOrContinuation = hf
for {
frag := hc.HeaderBlockFragment()
if _, err := hdec.Write(frag); err != nil {
return nil, http2ConnectionError(http2ErrCodeCompression)
}
if hc.HeadersEnded() {
break
}
if f, err := fr.ReadFrame(); err != nil {
return nil, err
} else {
hc = f.(*http2ContinuationFrame)
}
}
mh.http2HeadersFrame.headerFragBuf = nil
mh.http2HeadersFrame.invalidate()
if err := hdec.Close(); err != nil {
return nil, http2ConnectionError(http2ErrCodeCompression)
}
if invalid != nil {
fr.errDetail = invalid
if http2VerboseLogs {
log.Printf("http2: invalid header: %v", invalid)
}
return nil, http2StreamError{mh.StreamID, http2ErrCodeProtocol, invalid}
}
if err := mh.checkPseudos(); err != nil {
fr.errDetail = err
if http2VerboseLogs {
log.Printf("http2: invalid pseudo headers: %v", err)
}
return nil, http2StreamError{mh.StreamID, http2ErrCodeProtocol, err}
}
return mh, nil
}
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 http2transportExpectContinueTimeout(t1 *Transport) time.Duration {
return t1.ExpectContinueTimeout
}
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_RSA_WITH_AES_128_GCM_SHA256,
tls.TLS_RSA_WITH_AES_256_GCM_SHA384,
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
}
}
type http2contextContext interface {
context.Context
}
func http2serverConnBaseContext(c net.Conn, opts *http2ServeConnOpts) (ctx http2contextContext, cancel func()) {
ctx, cancel = context.WithCancel(context.Background())
ctx = context.WithValue(ctx, LocalAddrContextKey, c.LocalAddr())
if hs := opts.baseConfig(); hs != nil {
ctx = context.WithValue(ctx, ServerContextKey, hs)
}
return
}
func http2contextWithCancel(ctx http2contextContext) (_ http2contextContext, cancel func()) {
return context.WithCancel(ctx)
}
func http2requestWithContext(req *Request, ctx http2contextContext) *Request {
return req.WithContext(ctx)
}
type http2clientTrace httptrace.ClientTrace
func http2reqContext(r *Request) context.Context { return r.Context() }
func (t *http2Transport) idleConnTimeout() time.Duration {
if t.t1 != nil {
return t.t1.IdleConnTimeout
}
return 0
}
func http2setResponseUncompressed(res *Response) { res.Uncompressed = true }
func http2traceGotConn(req *Request, cc *http2ClientConn) {
trace := httptrace.ContextClientTrace(req.Context())
if trace == nil || trace.GotConn == nil {
return
}
ci := httptrace.GotConnInfo{Conn: cc.tconn}
cc.mu.Lock()
ci.Reused = cc.nextStreamID > 1
ci.WasIdle = len(cc.streams) == 0 && ci.Reused
if ci.WasIdle && !cc.lastActive.IsZero() {
ci.IdleTime = time.Now().Sub(cc.lastActive)
}
cc.mu.Unlock()
trace.GotConn(ci)
}
func http2traceWroteHeaders(trace *http2clientTrace) {
if trace != nil && trace.WroteHeaders != nil {
trace.WroteHeaders()
}
}
func http2traceGot100Continue(trace *http2clientTrace) {
if trace != nil && trace.Got100Continue != nil {
trace.Got100Continue()
}
}
func http2traceWait100Continue(trace *http2clientTrace) {
if trace != nil && trace.Wait100Continue != nil {
trace.Wait100Continue()
}
}
func http2traceWroteRequest(trace *http2clientTrace, err error) {
if trace != nil && trace.WroteRequest != nil {
trace.WroteRequest(httptrace.WroteRequestInfo{Err: err})
}
}
func http2traceFirstResponseByte(trace *http2clientTrace) {
if trace != nil && trace.GotFirstResponseByte != nil {
trace.GotFirstResponseByte()
}
}
func http2requestTrace(req *Request) *http2clientTrace {
trace := httptrace.ContextClientTrace(req.Context())
return (*http2clientTrace)(trace)
}
func (cc *http2ClientConn) Ping(ctx context.Context) error {
return cc.ping(ctx)
}
func http2cloneTLSConfig(c *tls.Config) *tls.Config { return c.Clone() }
var _ Pusher = (*http2responseWriter)(nil)
func (w *http2responseWriter) Push(target string, opts *PushOptions) error {
internalOpts := http2pushOptions{}
if opts != nil {
internalOpts.Method = opts.Method
internalOpts.Header = opts.Header
}
return w.push(target, internalOpts)
}
func http2configureServer18(h1 *Server, h2 *http2Server) error {
if h2.IdleTimeout == 0 {
if h1.IdleTimeout != 0 {
h2.IdleTimeout = h1.IdleTimeout
} else {
h2.IdleTimeout = h1.ReadTimeout
}
}
return nil
}
func http2shouldLogPanic(panicValue interface{}) bool {
return panicValue != nil && panicValue != ErrAbortHandler
}
func http2reqGetBody(req *Request) func() (io.ReadCloser, error) {
return req.GetBody
}
func http2reqBodyIsNoBody(body io.ReadCloser) bool {
return body == NoBody
}
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
},
}
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<<uint(bitSize) - 1
for i := 0; i < len(s); i++ {
var v byte
d := s[i]
switch {
case '0' <= d && d <= '9':
v = d - '0'
case 'a' <= d && d <= 'z':
v = d - 'a' + 10
case 'A' <= d && d <= 'Z':
v = d - 'A' + 10
default:
n = 0
err = strconv.ErrSyntax
goto Error
}
if int(v) >= 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}
}
func http2cutoff64(base int) uint64 {
if base < 2 {
return 0
}
return (1<<64-1)/uint64(base) + 1
}
var (
http2commonLowerHeader = map[string]string{}
http2commonCanonHeader = map[string]string{}
)
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
http2inTests 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 (
http2ClientPreface = "PRI * HTTP/2.0\r\n\r\nSM\r\n\r\n"
http2initialMaxFrameSize = 16384
http2NextProtoTLS = "h2"
http2initialHeaderTableSize = 4096
http2initialWindowSize = 65535
http2defaultMaxReadFrameSize = 1 << 20
)
var (
http2clientPreface = []byte(http2ClientPreface)
)
type http2streamState int
const (
http2stateIdle http2streamState = iota
http2stateOpen
http2stateHalfClosedLocal
http2stateHalfClosedRemote
http2stateClosed
)
var http2stateName = [...]string{
http2stateIdle: "Idle",
http2stateOpen: "Open",
http2stateHalfClosedLocal: "HalfClosedLocal",
http2stateHalfClosedRemote: "HalfClosedRemote",
http2stateClosed: "Closed",
}
func (st http2streamState) String() string {
return http2stateName[st]
}
type http2Setting struct {
ID http2SettingID
Val uint32
}
func (s http2Setting) String() string {
return fmt.Sprintf("[%v = %d]", s.ID, s.Val)
}
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
}
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")
)
func http2validWireHeaderFieldName(v string) bool {
if len(v) == 0 {
return false
}
for _, r := range v {
if !httplex.IsTokenRune(r) {
return false
}
if 'A' <= r && r <= 'Z' {
return false
}
}
return true
}
var http2httpCodeStringCommon = map[int]string{}
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)
}
type http2stringWriter interface {
WriteString(s string) (n int, err error)
}
type http2gate chan struct{}
func (g http2gate) Done() { g <- struct{}{} }
func (g http2gate) Wait() { <-g }
type http2closeWaiter chan struct{}
func (cw *http2closeWaiter) Init() {
*cw = make(chan struct{})
}
func (cw http2closeWaiter) Close() {
close(cw)
}
func (cw http2closeWaiter) Wait() {
<-cw
}
type http2bufferedWriter struct {
w io.Writer
bw *bufio.Writer
}
func http2newBufferedWriter(w io.Writer) *http2bufferedWriter {
return &http2bufferedWriter{w: w}
}
const http2bufWriterPoolBufferSize = 4 << 10
var http2bufWriterPool = sync.Pool{
New: func() interface{} {
return bufio.NewWriterSize(nil, http2bufWriterPoolBufferSize)
},
}
func (w *http2bufferedWriter) Available() int {
if w.bw == nil {
return http2bufWriterPoolBufferSize
}
return w.bw.Available()
}
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)
}
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}
type http2connectionStater interface {
ConnectionState() tls.ConnectionState
}
var http2sorterPool = sync.Pool{New: func() interface{} { return new(http2sorter) }}
type http2sorter struct {
v []string
}
func (s *http2sorter) Len() int { return len(s.v) }
func (s *http2sorter) Swap(i, j int) { s.v[i], s.v[j] = s.v[j], s.v[i] }
func (s *http2sorter) Less(i, j int) bool { return s.v[i] < s.v[j] }
func (s *http2sorter) Keys(h Header) []string {
keys := s.v[:0]
for k := range h {
keys = append(keys, k)
}
s.v = keys
sort.Sort(s)
return keys
}
func (s *http2sorter) SortStrings(ss []string) {
save := s.v
s.v = ss
sort.Sort(s)
s.v = save
}
func http2validPseudoPath(v string) bool {
return (len(v) > 0 && v[0] == '/' && (len(v) == 1 || v[1] != '/')) || v == "*"
}
type http2pipe struct {
mu sync.Mutex
c sync.Cond
b http2pipeBuffer
err error
breakErr error
donec chan struct{}
readFn func()
}
type http2pipeBuffer interface {
Len() int
io.Writer
io.Reader
}
func (p *http2pipe) Len() int {
p.mu.Lock()
defer p.mu.Unlock()
return p.b.Len()
}
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")
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)
}
func (p *http2pipe) CloseWithError(err error) { p.closeWithError(&p.err, err, nil) }
func (p *http2pipe) BreakWithError(err error) { p.closeWithError(&p.breakErr, err, nil) }
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()
}
func (p *http2pipe) closeDoneLocked() {
if p.donec == nil {
return
}
select {
case <-p.donec:
default:
close(p.donec)
}
}
func (p *http2pipe) Err() error {
p.mu.Lock()
defer p.mu.Unlock()
if p.breakErr != nil {
return p.breakErr
}
return p.err
}
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
http2handlerChunkWriteSize = 4 << 10
http2defaultMaxStreams = 250
)
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
},
}
var (
http2testHookOnConn func()
http2testHookGetServerConn func(*http2serverConn)
http2testHookOnPanicMu *sync.Mutex
http2testHookOnPanic func(sc *http2serverConn, panicVal interface{}) (rePanic bool)
)
type http2Server struct {
MaxHandlers int
MaxConcurrentStreams uint32
MaxReadFrameSize uint32
PermitProhibitedCipherSuites bool
IdleTimeout time.Duration
NewWriteScheduler func() http2WriteScheduler
}
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
}
func http2ConfigureServer(s *Server, conf *http2Server) error {
if s == nil {
panic("nil *http.Server")
}
if conf == nil {
conf = new(http2Server)
}
if err := http2configureServer18(s, conf); err != nil {
return err
}
if s.TLSConfig == nil {
s.TLSConfig = new(tls.Config)
} else if s.TLSConfig.CipherSuites != nil {
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)
}
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
return nil
}
type http2ServeConnOpts struct {
BaseConfig *Server
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
}
func (s *http2Server) ServeConn(c net.Conn, opts *http2ServeConnOpts) {
baseCtx, cancel := http2serverConnBaseContext(c, opts)
defer cancel()
sc := &http2serverConn{
srv: s,
hs: opts.baseConfig(),
conn: c,
baseCtx: baseCtx,
remoteAddrStr: c.RemoteAddr().String(),
bw: http2newBufferedWriter(c),
handler: opts.handler(),
streams: make(map[uint32]*http2stream),
readFrameCh: make(chan http2readFrameResult),
wantWriteFrameCh: make(chan http2FrameWriteRequest, 8),
wantStartPushCh: make(chan http2startPushRequest, 8),
wroteFrameCh: make(chan http2frameWriteResult, 1),
bodyReadCh: make(chan http2bodyReadMsg),
doneServing: make(chan struct{}),
clientMaxStreams: math.MaxUint32,
advMaxStreams: s.maxConcurrentStreams(),
initialWindowSize: http2initialWindowSize,
maxFrameSize: http2initialMaxFrameSize,
headerTableSize: http2initialHeaderTableSize,
serveG: http2newGoroutineLock(),
pushEnabled: true,
}
if sc.hs.WriteTimeout != 0 {
sc.conn.SetWriteDeadline(time.Time{})
}
if s.NewWriteScheduler != nil {
sc.writeSched = s.NewWriteScheduler()
} else {
sc.writeSched = http2NewRandomWriteScheduler()
}
sc.flow.add(http2initialWindowSize)
sc.inflow.add(http2initialWindowSize)
sc.hpackEncoder = hpack.NewEncoder(&sc.headerWriteBuf)
fr := http2NewFramer(sc.bw, c)
fr.ReadMetaHeaders = hpack.NewDecoder(http2initialHeaderTableSize, nil)
fr.MaxHeaderListSize = sc.maxHeaderListSize()
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()
}
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 {
srv *http2Server
hs *Server
conn net.Conn
bw *http2bufferedWriter
handler Handler
baseCtx http2contextContext
framer *http2Framer
doneServing chan struct{}
readFrameCh chan http2readFrameResult
wantWriteFrameCh chan http2FrameWriteRequest
wantStartPushCh chan http2startPushRequest
wroteFrameCh chan http2frameWriteResult
bodyReadCh chan http2bodyReadMsg
testHookCh chan func(int)
flow http2flow
inflow http2flow
tlsState *tls.ConnectionState
remoteAddrStr string
writeSched http2WriteScheduler
serveG http2goroutineLock
pushEnabled bool
sawFirstSettings bool
needToSendSettingsAck bool
unackedSettings int
clientMaxStreams uint32
advMaxStreams uint32
curClientStreams uint32
curPushedStreams uint32
maxClientStreamID uint32
maxPushPromiseID uint32
streams map[uint32]*http2stream
initialWindowSize int32
maxFrameSize int32
headerTableSize uint32
peerMaxHeaderListSize uint32
canonHeader map[string]string
writingFrame bool
writingFrameAsync bool
needsFrameFlush bool
inGoAway bool
inFrameScheduleLoop bool
needToSendGoAway bool
goAwayCode http2ErrCode
shutdownTimerCh <-chan time.Time
shutdownTimer *time.Timer
idleTimer *time.Timer
idleTimerCh <-chan time.Time
headerWriteBuf bytes.Buffer
hpackEncoder *hpack.Encoder
}
func (sc *http2serverConn) maxHeaderListSize() uint32 {
n := sc.hs.MaxHeaderBytes
if n <= 0 {
n = DefaultMaxHeaderBytes
}
const perFieldOverhead = 32
const typicalHeaders = 10
return uint32(n + typicalHeaders*perFieldOverhead)
}
func (sc *http2serverConn) curOpenStreams() uint32 {
sc.serveG.check()
return sc.curClientStreams + sc.curPushedStreams
}
type http2stream struct {
sc *http2serverConn
id uint32
body *http2pipe
cw http2closeWaiter
ctx http2contextContext
cancelCtx func()
bodyBytes int64
declBodyBytes int64
flow http2flow
inflow http2flow
parent *http2stream
numTrailerValues int64
weight uint8
state http2streamState
resetQueued bool
gotTrailerHeader bool
wroteHeaders bool
reqBuf []byte
trailer Header
reqTrailer Header
}
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%2 == 1 {
if streamID <= sc.maxClientStreamID {
return http2stateClosed, nil
}
} else {
if streamID <= sc.maxPushPromiseID {
return http2stateClosed, nil
}
}
return http2stateIdle, nil
}
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...)
}
}
func http2errno(v error) uintptr {
if rv := reflect.ValueOf(v); rv.Kind() == reflect.Uintptr {
return uintptr(rv.Uint())
}
return 0
}
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) 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
err error
readMore func()
}
func (sc *http2serverConn) readFrames() {
gate := make(http2gate)
gateDone := gate.Done
for {
f, err := sc.framer.ReadFrame()
select {
case sc.readFrameCh <- http2readFrameResult{f, err, gateDone}:
case <-sc.doneServing:
return
}
select {
case <-gate:
case <-sc.doneServing:
return
}
if http2terminalReadFrameError(err) {
return
}
}
}
type http2frameWriteResult struct {
wr http2FrameWriteRequest
err error
}
func (sc *http2serverConn) writeFrameAsync(wr http2FrameWriteRequest) {
err := wr.write.writeFrame(sc)
sc.wroteFrameCh <- http2frameWriteResult{wr, 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(http2FrameWriteRequest{
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)
if sc.srv.IdleTimeout != 0 {
sc.idleTimer = time.NewTimer(sc.srv.IdleTimeout)
defer sc.idleTimer.Stop()
sc.idleTimerCh = sc.idleTimer.C
}
var gracefulShutdownCh chan struct{}
if sc.hs != nil {
ch := http2h1ServerShutdownChan(sc.hs)
if ch != nil {
gracefulShutdownCh = make(chan struct{})
go sc.awaitGracefulShutdown(ch, gracefulShutdownCh)
}
}
go sc.readFrames()
settingsTimer := time.NewTimer(http2firstSettingsTimeout)
loopNum := 0
for {
loopNum++
select {
case wr := <-sc.wantWriteFrameCh:
sc.writeFrame(wr)
case spr := <-sc.wantStartPushCh:
sc.startPush(spr)
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 <-gracefulShutdownCh:
gracefulShutdownCh = nil
sc.startGracefulShutdown()
case <-sc.shutdownTimerCh:
sc.vlogf("GOAWAY close timer fired; closing conn from %v", sc.conn.RemoteAddr())
return
case <-sc.idleTimerCh:
sc.vlogf("connection is idle")
sc.goAway(http2ErrCodeNo)
case fn := <-sc.testHookCh:
fn(loopNum)
}
if sc.inGoAway && sc.curOpenStreams() == 0 && !sc.needToSendGoAway && !sc.writingFrame {
return
}
}
}
func (sc *http2serverConn) awaitGracefulShutdown(sharedCh <-chan struct{}, privateCh chan struct{}) {
select {
case <-sc.doneServing:
case <-sharedCh:
close(privateCh)
}
}
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) },
}
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(http2FrameWriteRequest{
write: writeArg,
stream: stream,
done: ch,
})
if err != nil {
return err
}
var frameWriteDone bool
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
}
func (sc *http2serverConn) writeFrameFromHandler(wr http2FrameWriteRequest) error {
sc.serveG.checkNotOn()
select {
case sc.wantWriteFrameCh <- wr:
return nil
case <-sc.doneServing:
return http2errClientDisconnected
}
}
func (sc *http2serverConn) writeFrame(wr http2FrameWriteRequest) {
sc.serveG.check()
var ignoreWrite bool
if wr.StreamID() != 0 {
_, isReset := wr.write.(http2StreamError)
if state, _ := sc.state(wr.StreamID()); state == http2stateClosed && !isReset {
ignoreWrite = true
}
}
switch wr.write.(type) {
case *http2writeResHeaders:
wr.stream.wroteHeaders = true
case http2write100ContinueHeadersFrame:
if wr.stream.wroteHeaders {
if wr.done != nil {
panic("wr.done != nil for write100ContinueHeadersFrame")
}
ignoreWrite = true
}
}
if !ignoreWrite {
sc.writeSched.Push(wr)
}
sc.scheduleFrameWrite()
}
func (sc *http2serverConn) startFrameWrite(wr http2FrameWriteRequest) {
sc.serveG.check()
if sc.writingFrame {
panic("internal error: can only be writing one frame at a time")
}
st := wr.stream
if st != nil {
switch st.state {
case http2stateHalfClosedLocal:
switch wr.write.(type) {
case http2StreamError, http2handlerPanicRST, http2writeWindowUpdate:
default:
panic(fmt.Sprintf("internal error: attempt to send frame on a half-closed-local stream: %v", wr))
}
case http2stateClosed:
panic(fmt.Sprintf("internal error: attempt to send frame on a closed stream: %v", wr))
}
}
if wpp, ok := wr.write.(*http2writePushPromise); ok {
var err error
wpp.promisedID, err = wpp.allocatePromisedID()
if err != nil {
sc.writingFrameAsync = false
wr.replyToWriter(err)
return
}
}
sc.writingFrame = true
sc.needsFrameFlush = true
if wr.write.staysWithinBuffer(sc.bw.Available()) {
sc.writingFrameAsync = false
err := wr.write.writeFrame(sc)
sc.wroteFrame(http2frameWriteResult{wr, err})
} else {
sc.writingFrameAsync = true
go sc.writeFrameAsync(wr)
}
}
var http2errHandlerPanicked = errors.New("http2: handler panicked")
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
sc.writingFrameAsync = false
wr := res.wr
if http2writeEndsStream(wr.write) {
st := wr.stream
if st == nil {
panic("internal error: expecting non-nil stream")
}
switch st.state {
case http2stateOpen:
st.state = http2stateHalfClosedLocal
sc.resetStream(http2streamError(st.id, http2ErrCodeCancel))
case http2stateHalfClosedRemote:
sc.closeStream(st, http2errHandlerComplete)
}
} else {
switch v := wr.write.(type) {
case http2StreamError:
if st, ok := sc.streams[v.StreamID]; ok {
sc.closeStream(st, v)
}
case http2handlerPanicRST:
sc.closeStream(wr.stream, http2errHandlerPanicked)
}
}
wr.replyToWriter(res.err)
sc.scheduleFrameWrite()
}
func (sc *http2serverConn) scheduleFrameWrite() {
sc.serveG.check()
if sc.writingFrame || sc.inFrameScheduleLoop {
return
}
sc.inFrameScheduleLoop = true
for !sc.writingFrameAsync {
if sc.needToSendGoAway {
sc.needToSendGoAway = false
sc.startFrameWrite(http2FrameWriteRequest{
write: &http2writeGoAway{
maxStreamID: sc.maxClientStreamID,
code: sc.goAwayCode,
},
})
continue
}
if sc.needToSendSettingsAck {
sc.needToSendSettingsAck = false
sc.startFrameWrite(http2FrameWriteRequest{write: http2writeSettingsAck{}})
continue
}
if !sc.inGoAway || sc.goAwayCode == http2ErrCodeNo {
if wr, ok := sc.writeSched.Pop(); ok {
sc.startFrameWrite(wr)
continue
}
}
if sc.needsFrameFlush {
sc.startFrameWrite(http2FrameWriteRequest{write: http2flushFrameWriter{}})
sc.needsFrameFlush = false
continue
}
break
}
sc.inFrameScheduleLoop = false
}
func (sc *http2serverConn) startGracefulShutdown() {
sc.goAwayIn(http2ErrCodeNo, 0)
}
func (sc *http2serverConn) goAway(code http2ErrCode) {
sc.serveG.check()
var forceCloseIn time.Duration
if code != http2ErrCodeNo {
forceCloseIn = 250 * time.Millisecond
} else {
forceCloseIn = 1 * time.Second
}
sc.goAwayIn(code, forceCloseIn)
}
func (sc *http2serverConn) goAwayIn(code http2ErrCode, forceCloseIn time.Duration) {
sc.serveG.check()
if sc.inGoAway {
return
}
if forceCloseIn != 0 {
sc.shutDownIn(forceCloseIn)
}
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(http2FrameWriteRequest{write: se})
if st, ok := sc.streams[se.StreamID]; ok {
st.resetQueued = true
}
}
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 *http2MetaHeadersFrame:
return sc.processHeaders(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 *http2GoAwayFrame:
return sc.processGoAway(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)
}
if sc.inGoAway && sc.goAwayCode != http2ErrCodeNo {
return nil
}
sc.writeFrame(http2FrameWriteRequest{write: http2writePingAck{f}})
return nil
}
func (sc *http2serverConn) processWindowUpdate(f *http2WindowUpdateFrame) error {
sc.serveG.check()
switch {
case f.StreamID != 0:
state, st := sc.state(f.StreamID)
if state == http2stateIdle {
return http2ConnectionError(http2ErrCodeProtocol)
}
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.cancelCtx()
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
if st.isPushed() {
sc.curPushedStreams--
} else {
sc.curClientStreams--
}
delete(sc.streams, st.id)
if len(sc.streams) == 0 {
sc.setConnState(StateIdle)
if sc.srv.IdleTimeout != 0 {
sc.idleTimer.Reset(sc.srv.IdleTimeout)
}
if http2h1ServerKeepAlivesDisabled(sc.hs) {
sc.startGracefulShutdown()
}
}
if p := st.body; p != nil {
sc.sendWindowUpdate(nil, p.Len())
p.CloseWithError(err)
}
st.cw.Close()
sc.writeSched.CloseStream(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.maxFrameSize = int32(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()
if sc.inGoAway && sc.goAwayCode != http2ErrCodeNo {
return nil
}
data := f.Data()
id := f.Header().StreamID
state, st := sc.state(id)
if id == 0 || state == http2stateIdle {
return http2ConnectionError(http2ErrCodeProtocol)
}
if st == nil || state != http2stateOpen || st.gotTrailerHeader || st.resetQueued {
if sc.inflow.available() < int32(f.Length) {
return http2streamError(id, http2ErrCodeFlowControl)
}
sc.inflow.take(int32(f.Length))
sc.sendWindowUpdate(nil, int(f.Length))
if st != nil && st.resetQueued {
return nil
}
return http2streamError(id, http2ErrCodeStreamClosed)
}
if st.body == nil {
panic("internal error: should have a body in this state")
}
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 f.Length > 0 {
if st.inflow.available() < int32(f.Length) {
return http2streamError(id, http2ErrCodeFlowControl)
}
st.inflow.take(int32(f.Length))
if len(data) > 0 {
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 pad := int32(f.Length) - int32(len(data)); pad > 0 {
sc.sendWindowUpdate32(nil, pad)
sc.sendWindowUpdate32(st, pad)
}
}
if f.StreamEnded() {
st.endStream()
}
return nil
}
func (sc *http2serverConn) processGoAway(f *http2GoAwayFrame) error {
sc.serveG.check()
if f.ErrCode != http2ErrCodeNo {
sc.logf("http2: received GOAWAY %+v, starting graceful shutdown", f)
} else {
sc.vlogf("http2: received GOAWAY %+v, starting graceful shutdown", f)
}
sc.startGracefulShutdown()
sc.pushEnabled = false
return nil
}
func (st *http2stream) isPushed() bool {
return st.id%2 == 0
}
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
}
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 *http2MetaHeadersFrame) error {
sc.serveG.check()
id := f.StreamID
if sc.inGoAway {
return nil
}
if id%2 != 1 {
return http2ConnectionError(http2ErrCodeProtocol)
}
if st := sc.streams[f.StreamID]; st != nil {
if st.resetQueued {
return nil
}
return st.processTrailerHeaders(f)
}
if id <= sc.maxClientStreamID {
return http2ConnectionError(http2ErrCodeProtocol)
}
sc.maxClientStreamID = id
if sc.idleTimer != nil {
sc.idleTimer.Stop()
}
if sc.curClientStreams+1 > sc.advMaxStreams {
if sc.unackedSettings == 0 {
return http2streamError(id, http2ErrCodeProtocol)
}
return http2streamError(id, http2ErrCodeRefusedStream)
}
initialState := http2stateOpen
if f.StreamEnded() {
initialState = http2stateHalfClosedRemote
}
st := sc.newStream(id, 0, initialState)
if f.HasPriority() {
if err := http2checkPriority(f.StreamID, f.Priority); err != nil {
return err
}
sc.writeSched.AdjustStream(st.id, f.Priority)
}
rw, req, err := sc.newWriterAndRequest(st, f)
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 f.Truncated {
handler = http2handleHeaderListTooLong
} else if err := http2checkValidHTTP2RequestHeaders(req.Header); err != nil {
handler = http2new400Handler(err)
}
if sc.hs.ReadTimeout != 0 {
sc.conn.SetReadDeadline(time.Time{})
}
go sc.runHandler(rw, req, handler)
return nil
}
func (st *http2stream) processTrailerHeaders(f *http2MetaHeadersFrame) error {
sc := st.sc
sc.serveG.check()
if st.gotTrailerHeader {
return http2ConnectionError(http2ErrCodeProtocol)
}
st.gotTrailerHeader = true
if !f.StreamEnded() {
return http2streamError(st.id, http2ErrCodeProtocol)
}
if len(f.PseudoFields()) > 0 {
return http2streamError(st.id, http2ErrCodeProtocol)
}
if st.trailer != nil {
for _, hf := range f.RegularFields() {
key := sc.canonicalHeader(hf.Name)
if !http2ValidTrailerHeader(key) {
return http2streamError(st.id, http2ErrCodeProtocol)
}
st.trailer[key] = append(st.trailer[key], hf.Value)
}
}
st.endStream()
return nil
}
func http2checkPriority(streamID uint32, p http2PriorityParam) error {
if streamID == p.StreamDep {
return http2streamError(streamID, http2ErrCodeProtocol)
}
return nil
}
func (sc *http2serverConn) processPriority(f *http2PriorityFrame) error {
if sc.inGoAway {
return nil
}
if err := http2checkPriority(f.StreamID, f.http2PriorityParam); err != nil {
return err
}
sc.writeSched.AdjustStream(f.StreamID, f.http2PriorityParam)
return nil
}
func (sc *http2serverConn) newStream(id, pusherID uint32, state http2streamState) *http2stream {
sc.serveG.check()
if id == 0 {
panic("internal error: cannot create stream with id 0")
}
ctx, cancelCtx := http2contextWithCancel(sc.baseCtx)
st := &http2stream{
sc: sc,
id: id,
state: state,
ctx: ctx,
cancelCtx: cancelCtx,
}
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
sc.writeSched.OpenStream(st.id, http2OpenStreamOptions{PusherID: pusherID})
if st.isPushed() {
sc.curPushedStreams++
} else {
sc.curClientStreams++
}
if sc.curOpenStreams() == 1 {
sc.setConnState(StateActive)
}
return st
}
func (sc *http2serverConn) newWriterAndRequest(st *http2stream, f *http2MetaHeadersFrame) (*http2responseWriter, *Request, error) {
sc.serveG.check()
rp := http2requestParam{
method: f.PseudoValue("method"),
scheme: f.PseudoValue("scheme"),
authority: f.PseudoValue("authority"),
path: f.PseudoValue("path"),
}
isConnect := rp.method == "CONNECT"
if isConnect {
if rp.path != "" || rp.scheme != "" || rp.authority == "" {
return nil, nil, http2streamError(f.StreamID, http2ErrCodeProtocol)
}
} else if rp.method == "" || rp.path == "" || (rp.scheme != "https" && rp.scheme != "http") {
return nil, nil, http2streamError(f.StreamID, http2ErrCodeProtocol)
}
bodyOpen := !f.StreamEnded()
if rp.method == "HEAD" && bodyOpen {
return nil, nil, http2streamError(f.StreamID, http2ErrCodeProtocol)
}
rp.header = make(Header)
for _, hf := range f.RegularFields() {
rp.header.Add(sc.canonicalHeader(hf.Name), hf.Value)
}
if rp.authority == "" {
rp.authority = rp.header.Get("Host")
}
rw, req, err := sc.newWriterAndRequestNoBody(st, rp)
if err != nil {
return nil, nil, err
}
if bodyOpen {
st.reqBuf = http2getRequestBodyBuf()
req.Body.(*http2requestBody).pipe = &http2pipe{
b: &http2fixedBuffer{buf: st.reqBuf},
}
if vv, ok := rp.header["Content-Length"]; ok {
req.ContentLength, _ = strconv.ParseInt(vv[0], 10, 64)
} else {
req.ContentLength = -1
}
}
return rw, req, nil
}
type http2requestParam struct {
method string
scheme, authority, path string
header Header
}
func (sc *http2serverConn) newWriterAndRequestNoBody(st *http2stream, rp http2requestParam) (*http2responseWriter, *Request, error) {
sc.serveG.check()
var tlsState *tls.ConnectionState
if rp.scheme == "https" {
tlsState = sc.tlsState
}
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, "; "))
}
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")
var url_ *url.URL
var requestURI string
if rp.method == "CONNECT" {
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(st.id, http2ErrCodeProtocol)
}
requestURI = rp.path
}
body := &http2requestBody{
conn: sc,
stream: st,
needsContinue: needsContinue,
}
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: rp.authority,
Body: body,
Trailer: trailer,
}
req = http2requestWithContext(req, st.ctx)
rws := http2responseWriterStatePool.Get().(*http2responseWriterState)
bwSave := rws.bw
*rws = http2responseWriterState{}
rws.conn = sc
rws.bw = bwSave
rws.bw.Reset(http2chunkWriter{rws})
rws.stream = st
rws.req = req
rws.body = body
rw := &http2responseWriter{rws: rws}
return rw, req, nil
}
var http2reqBodyCache = make(chan []byte, 8)
func http2getRequestBodyBuf() []byte {
select {
case b := <-http2reqBodyCache:
return b
default:
return make([]byte, http2initialWindowSize)
}
}
func http2putRequestBodyBuf(b []byte) {
select {
case http2reqBodyCache <- b:
default:
}
}
func (sc *http2serverConn) runHandler(rw *http2responseWriter, req *Request, handler func(ResponseWriter, *Request)) {
didPanic := true
defer func() {
rw.rws.stream.cancelCtx()
if didPanic {
e := recover()
sc.writeFrameFromHandler(http2FrameWriteRequest{
write: http2handlerPanicRST{rw.rws.stream.id},
stream: rw.rws.stream,
})
if http2shouldLogPanic(e) {
const size = 64 << 10
buf := make([]byte, size)
buf = buf[:runtime.Stack(buf, false)]
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) {
const statusRequestHeaderFieldsTooLarge = 431
w.WriteHeader(statusRequestHeaderFieldsTooLarge)
io.WriteString(w, "<h1>HTTP Error 431</h1><p>Request Header Field(s) Too Large</p>")
}
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(http2FrameWriteRequest{
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
}
func (sc *http2serverConn) write100ContinueHeaders(st *http2stream) {
sc.writeFrameFromHandler(http2FrameWriteRequest{
write: http2write100ContinueHeadersFrame{st.id},
stream: st,
})
}
type http2bodyReadMsg struct {
st *http2stream
n int
}
func (sc *http2serverConn) noteBodyReadFromHandler(st *http2stream, n int, err error) {
sc.serveG.checkNotOn()
if n > 0 {
select {
case sc.bodyReadCh <- http2bodyReadMsg{st, n}:
case <-sc.doneServing:
}
}
if err == io.EOF {
if buf := st.reqBuf; buf != nil {
st.reqBuf = nil
http2putRequestBodyBuf(buf)
}
}
}
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)
}
}
func (sc *http2serverConn) sendWindowUpdate(st *http2stream, n int) {
sc.serveG.check()
const maxUint31 = 1<<31 - 1
for n >= maxUint31 {
sc.sendWindowUpdate32(st, maxUint31)
n -= maxUint31
}
sc.sendWindowUpdate32(st, int32(n))
}
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(http2FrameWriteRequest{
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
sawEOF bool
pipe *http2pipe
needsContinue bool
}
func (b *http2requestBody) Close() error {
if b.pipe != nil && !b.closed {
b.pipe.BreakWithError(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 || b.sawEOF {
return 0, io.EOF
}
n, err = b.pipe.Read(p)
if err == io.EOF {
b.sawEOF = true
}
if b.conn == nil && http2inTests {
return
}
b.conn.noteBodyReadFromHandler(b.stream, n, err)
return
}
type http2responseWriter struct {
rws *http2responseWriterState
}
var (
_ CloseNotifier = (*http2responseWriter)(nil)
_ Flusher = (*http2responseWriter)(nil)
_ http2stringWriter = (*http2responseWriter)(nil)
)
type http2responseWriterState struct {
stream *http2stream
req *Request
body *http2requestBody
conn *http2serverConn
bw *bufio.Writer
handlerHeader Header
snapHeader Header
trailers []string
status int
wroteHeader bool
sentHeader bool
handlerDone bool
sentContentLen int64
wroteBytes int64
closeNotifierMu sync.Mutex
closeNotifierCh chan bool
}
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 }
func (rws *http2responseWriterState) declareTrailer(k string) {
k = CanonicalHeaderKey(k)
if !http2ValidTrailerHeader(k) {
rws.conn.logf("ignoring invalid trailer %q", k)
return
}
if !http2strSliceContains(rws.trailers, k) {
rws.trailers = append(rws.trailers, k)
}
}
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
}
const http2TrailerPrefix = "Trailer:"
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
}
if len(rws.trailers) > 1 {
sorter := http2sorterPool.Get().(*http2sorter)
sorter.SortStrings(rws.trailers)
http2sorterPool.Put(sorter)
}
}
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
cw := rws.stream.cw
go func() {
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
}
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)
}
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)
}
var (
http2ErrRecursivePush = errors.New("http2: recursive push not allowed")
http2ErrPushLimitReached = errors.New("http2: push would exceed peer's SETTINGS_MAX_CONCURRENT_STREAMS")
)
type http2pushOptions struct {
Method string
Header Header
}
func (w *http2responseWriter) push(target string, opts http2pushOptions) error {
st := w.rws.stream
sc := st.sc
sc.serveG.checkNotOn()
if st.isPushed() {
return http2ErrRecursivePush
}
if opts.Method == "" {
opts.Method = "GET"
}
if opts.Header == nil {
opts.Header = Header{}
}
wantScheme := "http"
if w.rws.req.TLS != nil {
wantScheme = "https"
}
u, err := url.Parse(target)
if err != nil {
return err
}
if u.Scheme == "" {
if !strings.HasPrefix(target, "/") {
return fmt.Errorf("target must be an absolute URL or an absolute path: %q", target)
}
u.Scheme = wantScheme
u.Host = w.rws.req.Host
} else {
if u.Scheme != wantScheme {
return fmt.Errorf("cannot push URL with scheme %q from request with scheme %q", u.Scheme, wantScheme)
}
if u.Host == "" {
return errors.New("URL must have a host")
}
}
for k := range opts.Header {
if strings.HasPrefix(k, ":") {
return fmt.Errorf("promised request headers cannot include pseudo header %q", k)
}
switch strings.ToLower(k) {
case "content-length", "content-encoding", "trailer", "te", "expect", "host":
return fmt.Errorf("promised request headers cannot include %q", k)
}
}
if err := http2checkValidHTTP2RequestHeaders(opts.Header); err != nil {
return err
}
if opts.Method != "GET" && opts.Method != "HEAD" {
return fmt.Errorf("method %q must be GET or HEAD", opts.Method)
}
msg := http2startPushRequest{
parent: st,
method: opts.Method,
url: u,
header: http2cloneHeader(opts.Header),
done: http2errChanPool.Get().(chan error),
}
select {
case <-sc.doneServing:
return http2errClientDisconnected
case <-st.cw:
return http2errStreamClosed
case sc.wantStartPushCh <- msg:
}
select {
case <-sc.doneServing:
return http2errClientDisconnected
case <-st.cw:
return http2errStreamClosed
case err := <-msg.done:
http2errChanPool.Put(msg.done)
return err
}
}
type http2startPushRequest struct {
parent *http2stream
method string
url *url.URL
header Header
done chan error
}
func (sc *http2serverConn) startPush(msg http2startPushRequest) {
sc.serveG.check()
if msg.parent.state != http2stateOpen && msg.parent.state != http2stateHalfClosedRemote {
msg.done <- http2errStreamClosed
return
}
if !sc.pushEnabled {
msg.done <- ErrNotSupported
return
}
allocatePromisedID := func() (uint32, error) {
sc.serveG.check()
if !sc.pushEnabled {
return 0, ErrNotSupported
}
if sc.curPushedStreams+1 > sc.clientMaxStreams {
return 0, http2ErrPushLimitReached
}
if sc.maxPushPromiseID+2 >= 1<<31 {
sc.startGracefulShutdown()
return 0, http2ErrPushLimitReached
}
sc.maxPushPromiseID += 2
promisedID := sc.maxPushPromiseID
promised := sc.newStream(promisedID, msg.parent.id, http2stateHalfClosedRemote)
rw, req, err := sc.newWriterAndRequestNoBody(promised, http2requestParam{
method: msg.method,
scheme: msg.url.Scheme,
authority: msg.url.Host,
path: msg.url.RequestURI(),
header: http2cloneHeader(msg.header),
})
if err != nil {
panic(fmt.Sprintf("newWriterAndRequestNoBody(%+v): %v", msg.url, err))
}
go sc.runHandler(rw, req, sc.handler.ServeHTTP)
return promisedID, nil
}
sc.writeFrame(http2FrameWriteRequest{
write: &http2writePushPromise{
streamID: msg.parent.id,
method: msg.method,
url: msg.url,
h: msg.header,
allocatePromisedID: allocatePromisedID,
},
stream: msg.parent,
done: msg.done,
})
}
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)
}
}
}
var http2connHeaders = []string{
"Connection",
"Keep-Alive",
"Proxy-Connection",
"Transfer-Encoding",
"Upgrade",
}
func http2checkValidHTTP2RequestHeaders(h Header) error {
for _, k := range http2connHeaders {
if _, ok := h[k]; ok {
return fmt.Errorf("request header %q is not valid in HTTP/2", k)
}
}
te := h["Te"]
if len(te) > 0 && (len(te) > 1 || (te[0] != "trailers" && te[0] != "")) {
return errors.New(`request header "TE" may only be "trailers" in HTTP/2`)
}
return nil
}
func http2new400Handler(err error) HandlerFunc {
return func(w ResponseWriter, r *Request) {
Error(w, err.Error(), StatusBadRequest)
}
}
func http2ValidTrailerHeader(name string) bool {
name = CanonicalHeaderKey(name)
if strings.HasPrefix(name, "If-") || http2badTrailer[name] {
return false
}
return true
}
var http2badTrailer = map[string]bool{
"Authorization": true,
"Cache-Control": true,
"Connection": true,
"Content-Encoding": true,
"Content-Length": true,
"Content-Range": true,
"Content-Type": true,
"Expect": true,
"Host": true,
"Keep-Alive": true,
"Max-Forwards": true,
"Pragma": true,
"Proxy-Authenticate": true,
"Proxy-Authorization": true,
"Proxy-Connection": true,
"Range": true,
"Realm": true,
"Te": true,
"Trailer": true,
"Transfer-Encoding": true,
"Www-Authenticate": true,
}
func http2h1ServerShutdownChan(hs *Server) <-chan struct{} {
if fn := http2testh1ServerShutdownChan; fn != nil {
return fn(hs)
}
var x interface{} = hs
type I interface {
getDoneChan() <-chan struct{}
}
if hs, ok := x.(I); ok {
return hs.getDoneChan()
}
return nil
}
var http2testh1ServerShutdownChan func(hs *Server) <-chan struct{}
func http2h1ServerKeepAlivesDisabled(hs *Server) bool {
var x interface{} = hs
type I interface {
doKeepAlives() bool
}
if hs, ok := x.(I); ok {
return !hs.doKeepAlives()
}
return false
}
const (
http2transportDefaultConnFlow = 1 << 30
http2transportDefaultStreamFlow = 4 << 20
http2transportDefaultStreamMinRefresh = 4 << 10
http2defaultUserAgent = "Go-http-client/2.0"
)
type http2Transport struct {
DialTLS func(network, addr string, cfg *tls.Config) (net.Conn, error)
TLSClientConfig *tls.Config
ConnPool http2ClientConnPool
DisableCompression bool
AllowHTTP bool
MaxHeaderListSize uint32
t1 *Transport
connPoolOnce sync.Once
connPoolOrDef http2ClientConnPool
}
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")
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}
}
}
type http2ClientConn struct {
t *http2Transport
tconn net.Conn
tlsState *tls.ConnectionState
singleUse bool
readerDone chan struct{}
readerErr error
idleTimeout time.Duration
idleTimer *time.Timer
mu sync.Mutex
cond *sync.Cond
flow http2flow
inflow http2flow
closed bool
wantSettingsAck bool
goAway *http2GoAwayFrame
goAwayDebug string
streams map[uint32]*http2clientStream
nextStreamID uint32
pings map[[8]byte]chan struct{}
bw *bufio.Writer
br *bufio.Reader
fr *http2Framer
lastActive time.Time
maxFrameSize uint32
maxConcurrentStreams uint32
initialWindowSize uint32
hbuf bytes.Buffer
henc *hpack.Encoder
freeBuf [][]byte
wmu sync.Mutex
werr error
}
type http2clientStream struct {
cc *http2ClientConn
req *Request
trace *http2clientTrace
ID uint32
resc chan http2resAndError
bufPipe http2pipe
startedWrite bool
requestedGzip bool
on100 func()
flow http2flow
inflow http2flow
bytesRemain int64
readErr error
stopReqBody error
didReset bool
peerReset chan struct{}
resetErr error
done chan struct{}
firstByte bool
pastHeaders bool
pastTrailers bool
trailer Header
resTrailer *Header
}
func (cs *http2clientStream) awaitRequestCancel(req *Request) {
ctx := http2reqContext(req)
if req.Cancel == nil && ctx.Done() == nil {
return
}
select {
case <-req.Cancel:
cs.cancelStream()
cs.bufPipe.CloseWithError(http2errRequestCanceled)
case <-ctx.Done():
cs.cancelStream()
cs.bufPipe.CloseWithError(ctx.Err())
case <-cs.done:
}
}
func (cs *http2clientStream) cancelStream() {
cs.cc.mu.Lock()
didReset := cs.didReset
cs.didReset = true
cs.cc.mu.Unlock()
if !didReset {
cs.cc.writeStreamReset(cs.ID, http2ErrCodeCancel, nil)
}
}
func (cs *http2clientStream) checkResetOrDone() error {
select {
case <-cs.peerReset:
return cs.resetErr
case <-cs.done:
return http2errStreamClosed
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")
type http2RoundTripOpt struct {
OnlyCachedConn bool
}
func (t *http2Transport) RoundTrip(req *Request) (*Response, error) {
return t.RoundTripOpt(req, http2RoundTripOpt{})
}
func http2authorityAddr(scheme string, authority string) (addr string) {
host, port, err := net.SplitHostPort(authority)
if err != nil {
port = "443"
if scheme == "http" {
port = "80"
}
host = authority
}
if a, err := idna.ToASCII(host); err == nil {
host = a
}
if strings.HasPrefix(host, "[") && strings.HasSuffix(host, "]") {
return host + ":" + port
}
return net.JoinHostPort(host, port)
}
func (t *http2Transport) RoundTripOpt(req *Request, opt http2RoundTripOpt) (*Response, error) {
if !(req.URL.Scheme == "https" || (req.URL.Scheme == "http" && t.AllowHTTP)) {
return nil, errors.New("http2: unsupported scheme")
}
addr := http2authorityAddr(req.URL.Scheme, 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
}
http2traceGotConn(req, cc)
res, err := cc.RoundTrip(req)
if err != nil {
if req, err = http2shouldRetryRequest(req, err); err == nil {
continue
}
}
if err != nil {
t.vlogf("RoundTrip failure: %v", err)
return nil, err
}
return res, nil
}
}
func (t *http2Transport) CloseIdleConnections() {
if cp, ok := t.connPool().(http2clientConnPoolIdleCloser); ok {
cp.closeIdleConnections()
}
}
var (
http2errClientConnClosed = errors.New("http2: client conn is closed")
http2errClientConnUnusable = errors.New("http2: client conn not usable")
http2errClientConnGotGoAway = errors.New("http2: Transport received Server's graceful shutdown GOAWAY")
http2errClientConnGotGoAwayAfterSomeReqBody = errors.New("http2: Transport received Server's graceful shutdown GOAWAY; some request body already written")
)
func http2shouldRetryRequest(req *Request, err error) (*Request, error) {
switch err {
default:
return nil, err
case http2errClientConnUnusable, http2errClientConnGotGoAway:
return req, nil
case http2errClientConnGotGoAwayAfterSomeReqBody:
if req.Body == nil || http2reqBodyIsNoBody(req.Body) {
return req, nil
}
getBody := http2reqGetBody(req)
if getBody == nil {
return nil, errors.New("http2: Transport: peer server initiated graceful shutdown after some of Request.Body was written; define Request.GetBody to avoid this error")
}
body, err := getBody()
if err != nil {
return nil, err
}
newReq := *req
newReq.Body = body
return &newReq, nil
}
}
func (t *http2Transport) dialClientConn(addr string, singleUse bool) (*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, singleUse)
}
func (t *http2Transport) newTLSConfig(host string) *tls.Config {
cfg := new(tls.Config)
if t.TLSClientConfig != nil {
*cfg = *http2cloneTLSConfig(t.TLSClientConfig)
}
if !http2strSliceContains(cfg.NextProtos, http2NextProtoTLS) {
cfg.NextProtos = append([]string{http2NextProtoTLS}, cfg.NextProtos...)
}
if cfg.ServerName == "" {
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
}
func (t *http2Transport) disableKeepAlives() bool {
return t.t1 != nil && t.t1.DisableKeepAlives
}
func (t *http2Transport) expectContinueTimeout() time.Duration {
if t.t1 == nil {
return 0
}
return http2transportExpectContinueTimeout(t.t1)
}
func (t *http2Transport) NewClientConn(c net.Conn) (*http2ClientConn, error) {
return t.newClientConn(c, false)
}
func (t *http2Transport) newClientConn(c net.Conn, singleUse bool) (*http2ClientConn, error) {
cc := &http2ClientConn{
t: t,
tconn: c,
readerDone: make(chan struct{}),
nextStreamID: 1,
maxFrameSize: 16 << 10,
initialWindowSize: 65535,
maxConcurrentStreams: 1000,
streams: make(map[uint32]*http2clientStream),
singleUse: singleUse,
wantSettingsAck: true,
pings: make(map[[8]byte]chan struct{}),
}
if d := t.idleConnTimeout(); d != 0 {
cc.idleTimeout = d
cc.idleTimer = time.AfterFunc(d, cc.onIdleTimeout)
}
if http2VerboseLogs {
t.vlogf("http2: Transport creating client conn %p to %v", cc, c.RemoteAddr())
}
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.fr.ReadMetaHeaders = hpack.NewDecoder(http2initialHeaderTableSize, nil)
cc.fr.MaxHeaderListSize = t.maxHeaderListSize()
cc.henc = hpack.NewEncoder(&cc.hbuf)
if cs, ok := c.(http2connectionStater); ok {
state := cs.ConnectionState()
cc.tlsState = &state
}
initialSettings := []http2Setting{
{ID: http2SettingEnablePush, Val: 0},
{ID: http2SettingInitialWindowSize, Val: http2transportDefaultStreamFlow},
}
if max := t.maxHeaderListSize(); max != 0 {
initialSettings = append(initialSettings, http2Setting{ID: http2SettingMaxHeaderListSize, Val: max})
}
cc.bw.Write(http2clientPreface)
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
}
go cc.readLoop()
return cc, nil
}
func (cc *http2ClientConn) setGoAway(f *http2GoAwayFrame) {
cc.mu.Lock()
defer cc.mu.Unlock()
old := cc.goAway
cc.goAway = f
if cc.goAwayDebug == "" {
cc.goAwayDebug = string(f.DebugData())
}
if old != nil && old.ErrCode != http2ErrCodeNo {
cc.goAway.ErrCode = old.ErrCode
}
last := f.LastStreamID
for streamID, cs := range cc.streams {
if streamID > last {
select {
case cs.resc <- http2resAndError{err: http2errClientConnGotGoAway}:
default:
}
}
}
}
func (cc *http2ClientConn) CanTakeNewRequest() bool {
cc.mu.Lock()
defer cc.mu.Unlock()
return cc.canTakeNewRequestLocked()
}
func (cc *http2ClientConn) canTakeNewRequestLocked() bool {
if cc.singleUse && cc.nextStreamID > 1 {
return false
}
return cc.goAway == nil && !cc.closed &&
int64(len(cc.streams)+1) < int64(cc.maxConcurrentStreams) &&
cc.nextStreamID < math.MaxInt32
}
func (cc *http2ClientConn) onIdleTimeout() {
cc.closeIfIdle()
}
func (cc *http2ClientConn) closeIfIdle() {
cc.mu.Lock()
if len(cc.streams) > 0 {
cc.mu.Unlock()
return
}
cc.closed = true
nextID := cc.nextStreamID
cc.mu.Unlock()
if http2VerboseLogs {
cc.vlogf("http2: Transport closing idle conn %p (forSingleUse=%v, maxStream=%v)", cc, cc.singleUse, nextID-2)
}
cc.tconn.Close()
}
const http2maxAllocFrameSize = 512 << 10
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
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
}
}
}
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
}
func http2checkConnHeaders(req *Request) error {
if v := req.Header.Get("Upgrade"); v != "" {
return fmt.Errorf("http2: invalid Upgrade request header: %q", req.Header["Upgrade"])
}
if vv := req.Header["Transfer-Encoding"]; len(vv) > 0 && (len(vv) > 1 || vv[0] != "" && vv[0] != "chunked") {
return fmt.Errorf("http2: invalid Transfer-Encoding request header: %q", vv)
}
if vv := req.Header["Connection"]; len(vv) > 0 && (len(vv) > 1 || vv[0] != "" && vv[0] != "close" && vv[0] != "keep-alive") {
return fmt.Errorf("http2: invalid Connection request header: %q", vv)
}
return nil
}
func http2actualContentLength(req *Request) int64 {
if req.Body == nil {
return 0
}
if req.ContentLength != 0 {
return req.ContentLength
}
return -1
}
func (cc *http2ClientConn) RoundTrip(req *Request) (*Response, error) {
if err := http2checkConnHeaders(req); err != nil {
return nil, err
}
if cc.idleTimer != nil {
cc.idleTimer.Stop()
}
trailers, err := http2commaSeparatedTrailers(req)
if err != nil {
return nil, err
}
hasTrailers := trailers != ""
cc.mu.Lock()
cc.lastActive = time.Now()
if cc.closed || !cc.canTakeNewRequestLocked() {
cc.mu.Unlock()
return nil, http2errClientConnUnusable
}
body := req.Body
hasBody := body != nil
contentLen := http2actualContentLength(req)
var requestedGzip bool
if !cc.t.disableCompression() &&
req.Header.Get("Accept-Encoding") == "" &&
req.Header.Get("Range") == "" &&
req.Method != "HEAD" {
requestedGzip = true
}
hdrs, err := cc.encodeHeaders(req, requestedGzip, trailers, contentLen)
if err != nil {
cc.mu.Unlock()
return nil, err
}
cs := cc.newStream()
cs.req = req
cs.trace = http2requestTrace(req)
cs.requestedGzip = requestedGzip
bodyWriter := cc.t.getBodyWriterState(cs, body)
cs.on100 = bodyWriter.on100
cc.wmu.Lock()
endStream := !hasBody && !hasTrailers
werr := cc.writeHeaders(cs.ID, endStream, hdrs)
cc.wmu.Unlock()
http2traceWroteHeaders(cs.trace)
cc.mu.Unlock()
if werr != nil {
if hasBody {
req.Body.Close()
bodyWriter.cancel()
}
cc.forgetStreamID(cs.ID)
http2traceWroteRequest(cs.trace, werr)
return nil, werr
}
var respHeaderTimer <-chan time.Time
if hasBody {
bodyWriter.scheduleBodyWrite()
} else {
http2traceWroteRequest(cs.trace, nil)
if d := cc.responseHeaderTimeout(); d != 0 {
timer := time.NewTimer(d)
defer timer.Stop()
respHeaderTimer = timer.C
}
}
readLoopResCh := cs.resc
bodyWritten := false
ctx := http2reqContext(req)
handleReadLoopResponse := func(re http2resAndError) (*Response, error) {
res := re.res
if re.err != nil || res.StatusCode > 299 {
bodyWriter.cancel()
cs.abortRequestBodyWrite(http2errStopReqBodyWrite)
}
if re.err != nil {
if re.err == http2errClientConnGotGoAway {
cc.mu.Lock()
if cs.startedWrite {
re.err = http2errClientConnGotGoAwayAfterSomeReqBody
}
cc.mu.Unlock()
}
cc.forgetStreamID(cs.ID)
return nil, re.err
}
res.Request = req
res.TLS = cc.tlsState
return res, nil
}
for {
select {
case re := <-readLoopResCh:
return handleReadLoopResponse(re)
case <-respHeaderTimer:
cc.forgetStreamID(cs.ID)
if !hasBody || bodyWritten {
cc.writeStreamReset(cs.ID, http2ErrCodeCancel, nil)
} else {
bodyWriter.cancel()
cs.abortRequestBodyWrite(http2errStopReqBodyWriteAndCancel)
}
return nil, http2errTimeout
case <-ctx.Done():
cc.forgetStreamID(cs.ID)
if !hasBody || bodyWritten {
cc.writeStreamReset(cs.ID, http2ErrCodeCancel, nil)
} else {
bodyWriter.cancel()
cs.abortRequestBodyWrite(http2errStopReqBodyWriteAndCancel)
}
return nil, ctx.Err()
case <-req.Cancel:
cc.forgetStreamID(cs.ID)
if !hasBody || bodyWritten {
cc.writeStreamReset(cs.ID, http2ErrCodeCancel, nil)
} else {
bodyWriter.cancel()
cs.abortRequestBodyWrite(http2errStopReqBodyWriteAndCancel)
}
return nil, http2errRequestCanceled
case <-cs.peerReset:
return nil, cs.resetErr
case err := <-bodyWriter.resc:
select {
case re := <-readLoopResCh:
return handleReadLoopResponse(re)
default:
}
if err != nil {
return nil, err
}
bodyWritten = true
if d := cc.responseHeaderTimeout(); d != 0 {
timer := time.NewTimer(d)
defer timer.Stop()
respHeaderTimer = timer.C
}
}
}
}
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
}
var (
http2errStopReqBodyWrite = errors.New("http2: aborting request body write")
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() {
http2traceWroteRequest(cs.trace, err)
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
}
}
if sentEnd {
return nil
}
var trls []byte
if hasTrailers {
cc.mu.Lock()
defer cc.mu.Unlock()
trls = cc.encodeTrailers(req)
}
cc.wmu.Lock()
defer cc.wmu.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
}
return err
}
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.checkResetOrDone(); 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) }
func (cc *http2ClientConn) encodeHeaders(req *Request, addGzipHeader bool, trailers string, contentLength int64) ([]byte, error) {
cc.hbuf.Reset()
host := req.Host
if host == "" {
host = req.URL.Host
}
host, err := httplex.PunycodeHostPort(host)
if err != nil {
return nil, err
}
var path string
if req.Method != "CONNECT" {
path = req.URL.RequestURI()
if !http2validPseudoPath(path) {
orig := path
path = strings.TrimPrefix(path, req.URL.Scheme+"://"+host)
if !http2validPseudoPath(path) {
if req.URL.Opaque != "" {
return nil, fmt.Errorf("invalid request :path %q from URL.Opaque = %q", orig, req.URL.Opaque)
} else {
return nil, fmt.Errorf("invalid request :path %q", orig)
}
}
}
}
for k, vv := range req.Header {
if !httplex.ValidHeaderFieldName(k) {
return nil, fmt.Errorf("invalid HTTP header name %q", k)
}
for _, v := range vv {
if !httplex.ValidHeaderFieldValue(v) {
return nil, fmt.Errorf("invalid HTTP header value %q for header %q", v, k)
}
}
}
cc.writeHeader(":authority", host)
cc.writeHeader(":method", req.Method)
if req.Method != "CONNECT" {
cc.writeHeader(":path", path)
cc.writeHeader(":scheme", req.URL.Scheme)
}
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", "keep-alive":
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(), nil
}
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
}
}
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
}
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 {
cc.lastActive = time.Now()
delete(cc.streams, id)
if len(cc.streams) == 0 && cc.idleTimer != nil {
cc.idleTimer.Reset(cc.idleTimeout)
}
close(cs.done)
cc.cond.Broadcast()
}
return cs
}
type http2clientConnReadLoop struct {
cc *http2ClientConn
activeRes map[uint32]*http2clientStream
closeWhenIdle bool
}
func (cc *http2ClientConn) readLoop() {
rl := &http2clientConnReadLoop{
cc: cc,
activeRes: make(map[uint32]*http2clientStream),
}
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()
}
}
type http2GoAwayError struct {
LastStreamID uint32
ErrCode http2ErrCode
DebugData string
}
func (e http2GoAwayError) Error() string {
return fmt.Sprintf("http2: server sent GOAWAY and closed the connection; LastStreamID=%v, ErrCode=%v, debug=%q",
e.LastStreamID, e.ErrCode, e.DebugData)
}
func http2isEOFOrNetReadError(err error) bool {
if err == io.EOF {
return true
}
ne, ok := err.(*net.OpError)
return ok && ne.Op == "read"
}
func (rl *http2clientConnReadLoop) cleanup() {
cc := rl.cc
defer cc.tconn.Close()
defer cc.t.connPool().MarkDead(cc)
defer close(cc.readerDone)
if cc.idleTimer != nil {
cc.idleTimer.Stop()
}
err := cc.readerErr
cc.mu.Lock()
if cc.goAway != nil && http2isEOFOrNetReadError(err) {
err = http2GoAwayError{
LastStreamID: cc.goAway.LastStreamID,
ErrCode: cc.goAway.ErrCode,
DebugData: cc.goAwayDebug,
}
} else if err == io.EOF {
err = io.ErrUnexpectedEOF
}
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() || cc.singleUse
gotReply := false
gotSettings := false
for {
f, err := cc.fr.ReadFrame()
if err != nil {
cc.vlogf("http2: Transport readFrame error on conn %p: (%T) %v", cc, err, err)
}
if se, ok := err.(http2StreamError); ok {
if cs := cc.streamByID(se.StreamID, true); cs != nil {
cs.cc.writeStreamReset(cs.ID, se.Code, err)
if se.Cause == nil {
se.Cause = cc.fr.errDetail
}
rl.endStreamError(cs, se)
}
continue
} else if err != nil {
return err
}
if http2VerboseLogs {
cc.vlogf("http2: Transport received %s", http2summarizeFrame(f))
}
if !gotSettings {
if _, ok := f.(*http2SettingsFrame); !ok {
cc.logf("protocol error: received %T before a SETTINGS frame", f)
return http2ConnectionError(http2ErrCodeProtocol)
}
gotSettings = true
}
maybeIdle := false
switch f := f.(type) {
case *http2MetaHeadersFrame:
err = rl.processHeaders(f)
maybeIdle = true
gotReply = 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 {
if http2VerboseLogs {
cc.vlogf("http2: Transport conn %p received error from processing frame %v: %v", cc, http2summarizeFrame(f), err)
}
return err
}
if rl.closeWhenIdle && gotReply && maybeIdle && len(rl.activeRes) == 0 {
cc.closeIfIdle()
}
}
}
func (rl *http2clientConnReadLoop) processHeaders(f *http2MetaHeadersFrame) error {
cc := rl.cc
cs := cc.streamByID(f.StreamID, f.StreamEnded())
if cs == nil {
return nil
}
if !cs.firstByte {
if cs.trace != nil {
http2traceFirstResponseByte(cs.trace)
}
cs.firstByte = true
}
if !cs.pastHeaders {
cs.pastHeaders = true
} else {
return rl.processTrailers(cs, f)
}
res, err := rl.handleResponse(cs, f)
if err != nil {
if _, ok := err.(http2ConnectionError); ok {
return err
}
cs.cc.writeStreamReset(f.StreamID, http2ErrCodeProtocol, err)
cs.resc <- http2resAndError{err: err}
return nil
}
if res == nil {
return nil
}
if res.Body != http2noBody {
rl.activeRes[cs.ID] = cs
}
cs.resTrailer = &res.Trailer
cs.resc <- http2resAndError{res: res}
return nil
}
func (rl *http2clientConnReadLoop) handleResponse(cs *http2clientStream, f *http2MetaHeadersFrame) (*Response, error) {
if f.Truncated {
return nil, http2errResponseHeaderListSize
}
status := f.PseudoValue("status")
if status == "" {
return nil, errors.New("missing status pseudo header")
}
statusCode, err := strconv.Atoi(status)
if err != nil {
return nil, errors.New("malformed non-numeric status pseudo header")
}
if statusCode == 100 {
http2traceGot100Continue(cs.trace)
if cs.on100 != nil {
cs.on100()
}
cs.pastHeaders = false
return nil, nil
}
header := make(Header)
res := &Response{
Proto: "HTTP/2.0",
ProtoMajor: 2,
Header: header,
StatusCode: statusCode,
Status: status + " " + StatusText(statusCode),
}
for _, hf := range f.RegularFields() {
key := CanonicalHeaderKey(hf.Name)
if key == "Trailer" {
t := res.Trailer
if t == nil {
t = make(Header)
res.Trailer = t
}
http2foreachHeaderElement(hf.Value, func(v string) {
t[CanonicalHeaderKey(v)] = nil
})
} else {
header[key] = append(header[key], hf.Value)
}
}
streamEnded := f.StreamEnded()
isHead := cs.req.Method == "HEAD"
if !streamEnded || isHead {
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 || isHead {
res.Body = http2noBody
return res, nil
}
buf := new(bytes.Buffer)
cs.bufPipe = http2pipe{b: buf}
cs.bytesRemain = res.ContentLength
res.Body = http2transportResponseBody{cs}
go cs.awaitRequestCancel(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}
http2setResponseUncompressed(res)
}
return res, nil
}
func (rl *http2clientConnReadLoop) processTrailers(cs *http2clientStream, f *http2MetaHeadersFrame) error {
if cs.pastTrailers {
return http2ConnectionError(http2ErrCodeProtocol)
}
cs.pastTrailers = true
if !f.StreamEnded() {
return http2ConnectionError(http2ErrCodeProtocol)
}
if len(f.PseudoFields()) > 0 {
return http2ConnectionError(http2ErrCodeProtocol)
}
trailer := make(Header)
for _, hf := range f.RegularFields() {
key := CanonicalHeaderKey(hf.Name)
trailer[key] = append(trailer[key], hf.Value)
}
cs.trailer = trailer
rl.endStream(cs)
return nil
}
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 {
v := int(cs.inflow.available()) + cs.bufPipe.Len()
if v < http2transportDefaultStreamFlow-http2transportDefaultStreamMinRefresh {
streamAdd = int32(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
cc := cs.cc
serverSentStreamEnd := cs.bufPipe.Err() == io.EOF
unread := cs.bufPipe.Len()
if unread > 0 || !serverSentStreamEnd {
cc.mu.Lock()
cc.wmu.Lock()
if !serverSentStreamEnd {
cc.fr.WriteRSTStream(cs.ID, http2ErrCodeCancel)
}
if unread > 0 {
cc.inflow.add(int32(unread))
cc.fr.WriteWindowUpdate(0, uint32(unread))
}
cc.bw.Flush()
cc.wmu.Unlock()
cc.mu.Unlock()
}
cs.bufPipe.BreakWithError(http2errClosedResponseBody)
return nil
}
func (rl *http2clientConnReadLoop) processData(f *http2DataFrame) error {
cc := rl.cc
cs := cc.streamByID(f.StreamID, f.StreamEnded())
data := f.Data()
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)
}
if f.Length > 0 {
cc.mu.Lock()
cc.inflow.add(int32(f.Length))
cc.mu.Unlock()
cc.wmu.Lock()
cc.fr.WriteWindowUpdate(0, uint32(f.Length))
cc.bw.Flush()
cc.wmu.Unlock()
}
return nil
}
if f.Length > 0 {
if len(data) > 0 && 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(f.Length) {
cs.inflow.take(int32(f.Length))
} else {
cc.mu.Unlock()
return http2ConnectionError(http2ErrCodeFlowControl)
}
if pad := int32(f.Length) - int32(len(data)); pad > 0 {
cs.inflow.add(pad)
cc.inflow.add(pad)
cc.wmu.Lock()
cc.fr.WriteWindowUpdate(0, uint32(pad))
cc.fr.WriteWindowUpdate(cs.ID, uint32(pad))
cc.bw.Flush()
cc.wmu.Unlock()
}
didReset := cs.didReset
cc.mu.Unlock()
if len(data) > 0 && !didReset {
if _, err := cs.bufPipe.Write(data); err != nil {
rl.endStreamError(cs, err)
return err
}
}
}
if f.StreamEnded() {
rl.endStream(cs)
}
return nil
}
var http2errInvalidTrailers = errors.New("http2: invalid trailers")
func (rl *http2clientConnReadLoop) endStream(cs *http2clientStream) {
rl.endStreamError(cs, nil)
}
func (rl *http2clientConnReadLoop) endStreamError(cs *http2clientStream, err error) {
var code func()
if err == nil {
err = io.EOF
code = cs.copyTrailers
}
cs.bufPipe.closeWithErrorAndCode(err, code)
delete(rl.activeRes, cs.ID)
if http2isConnectionCloseRequest(cs.req) {
rl.closeWhenIdle = true
}
select {
case cs.resc <- http2resAndError{err: err}:
default:
}
}
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()
if f.IsAck() {
if cc.wantSettingsAck {
cc.wantSettingsAck = false
return nil
}
return http2ConnectionError(http2ErrCodeProtocol)
}
err := f.ForeachSetting(func(s http2Setting) error {
switch s.ID {
case http2SettingMaxFrameSize:
cc.maxFrameSize = s.Val
case http2SettingMaxConcurrentStreams:
cc.maxConcurrentStreams = s.Val
case http2SettingInitialWindowSize:
if s.Val > math.MaxInt32 {
return http2ConnectionError(http2ErrCodeFlowControl)
}
delta := int32(s.Val) - int32(cc.initialWindowSize)
for _, cs := range cc.streams {
cs.flow.add(delta)
}
cc.cond.Broadcast()
cc.initialWindowSize = s.Val
default:
cc.vlogf("Unhandled Setting: %v", s)
}
return nil
})
if err != nil {
return err
}
cc.wmu.Lock()
defer cc.wmu.Unlock()
cc.fr.WriteSettingsAck()
cc.bw.Flush()
return cc.werr
}
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 (cc *http2ClientConn) ping(ctx http2contextContext) error {
c := make(chan struct{})
var p [8]byte
for {
if _, err := rand.Read(p[:]); err != nil {
return err
}
cc.mu.Lock()
if _, found := cc.pings[p]; !found {
cc.pings[p] = c
cc.mu.Unlock()
break
}
cc.mu.Unlock()
}
cc.wmu.Lock()
if err := cc.fr.WritePing(false, p); err != nil {
cc.wmu.Unlock()
return err
}
if err := cc.bw.Flush(); err != nil {
cc.wmu.Unlock()
return err
}
cc.wmu.Unlock()
select {
case <-c:
return nil
case <-ctx.Done():
return ctx.Err()
case <-cc.readerDone:
return cc.readerErr
}
}
func (rl *http2clientConnReadLoop) processPing(f *http2PingFrame) error {
if f.IsAck() {
cc := rl.cc
cc.mu.Lock()
defer cc.mu.Unlock()
if c, ok := cc.pings[f.Data]; ok {
close(c)
delete(cc.pings, f.Data)
}
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 (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 }
type http2gzipReader struct {
body io.ReadCloser
zr *gzip.Reader
zerr 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 }
type http2bodyWriterState struct {
cs *http2clientStream
timer *time.Timer
fnonce *sync.Once
fn func()
resc chan error
delay time.Duration
}
func (t *http2Transport) getBodyWriterState(cs *http2clientStream, body io.Reader) (s http2bodyWriterState) {
s.cs = cs
if body == nil {
return
}
resc := make(chan error, 1)
s.resc = resc
s.fn = func() {
cs.cc.mu.Lock()
cs.startedWrite = true
cs.cc.mu.Unlock()
resc <- cs.writeRequestBody(body, cs.req.Body)
}
s.delay = t.expectContinueTimeout()
if s.delay == 0 ||
!httplex.HeaderValuesContainsToken(
cs.req.Header["Expect"],
"100-continue") {
return
}
s.fnonce = new(sync.Once)
const hugeDuration = 365 * 24 * time.Hour
s.timer = time.AfterFunc(hugeDuration, func() {
s.fnonce.Do(s.fn)
})
return
}
func (s http2bodyWriterState) cancel() {
if s.timer != nil {
s.timer.Stop()
}
}
func (s http2bodyWriterState) on100() {
if s.timer == nil {
return
}
s.timer.Stop()
go func() { s.fnonce.Do(s.fn) }()
}
func (s http2bodyWriterState) scheduleBodyWrite() {
if s.timer == nil {
go s.fn()
return
}
http2traceWait100Continue(s.cs.trace)
if s.timer.Stop() {
s.timer.Reset(s.delay)
}
}
func http2isConnectionCloseRequest(req *Request) bool {
return req.Close || httplex.HeaderValuesContainsToken(req.Header["Connection"], "close")
}
type http2writeFramer interface {
writeFrame(http2writeContext) error
staysWithinBuffer(size int) bool
}
type http2writeContext interface {
Framer() *http2Framer
Flush() error
CloseConn() error
HeaderEncoder() (*hpack.Encoder, *bytes.Buffer)
}
func http2writeEndsStream(w http2writeFramer) bool {
switch v := w.(type) {
case *http2writeData:
return v.endStream
case *http2writeResHeaders:
return v.endStream
case nil:
panic("writeEndsStream called on nil writeFramer")
}
return false
}
type http2flushFrameWriter struct{}
func (http2flushFrameWriter) writeFrame(ctx http2writeContext) error {
return ctx.Flush()
}
func (http2flushFrameWriter) staysWithinBuffer(max int) bool { return false }
type http2writeSettings []http2Setting
func (s http2writeSettings) staysWithinBuffer(max int) bool {
const settingSize = 6
return http2frameHeaderLen+settingSize*len(s) <= max
}
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
}
func (*http2writeGoAway) staysWithinBuffer(max int) bool { return false }
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)
}
func (w *http2writeData) staysWithinBuffer(max int) bool {
return http2frameHeaderLen+len(w.p) <= max
}
type http2handlerPanicRST struct {
StreamID uint32
}
func (hp http2handlerPanicRST) writeFrame(ctx http2writeContext) error {
return ctx.Framer().WriteRSTStream(hp.StreamID, http2ErrCodeInternal)
}
func (hp http2handlerPanicRST) staysWithinBuffer(max int) bool { return http2frameHeaderLen+4 <= max }
func (se http2StreamError) writeFrame(ctx http2writeContext) error {
return ctx.Framer().WriteRSTStream(se.StreamID, se.Code)
}
func (se http2StreamError) staysWithinBuffer(max int) bool { return http2frameHeaderLen+4 <= max }
type http2writePingAck struct{ pf *http2PingFrame }
func (w http2writePingAck) writeFrame(ctx http2writeContext) error {
return ctx.Framer().WritePing(true, w.pf.Data)
}
func (w http2writePingAck) staysWithinBuffer(max int) bool {
return http2frameHeaderLen+len(w.pf.Data) <= max
}
type http2writeSettingsAck struct{}
func (http2writeSettingsAck) writeFrame(ctx http2writeContext) error {
return ctx.Framer().WriteSettingsAck()
}
func (http2writeSettingsAck) staysWithinBuffer(max int) bool { return http2frameHeaderLen <= max }
func http2splitHeaderBlock(ctx http2writeContext, headerBlock []byte, fn func(ctx http2writeContext, frag []byte, firstFrag, lastFrag bool) error) error {
const maxFrameSize = 16384
first := true
for len(headerBlock) > 0 {
frag := headerBlock
if len(frag) > maxFrameSize {
frag = frag[:maxFrameSize]
}
headerBlock = headerBlock[len(frag):]
if err := fn(ctx, frag, first, len(headerBlock) == 0); err != nil {
return err
}
first = false
}
return nil
}
type http2writeResHeaders struct {
streamID uint32
httpResCode int
h Header
trailers []string
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) staysWithinBuffer(max int) bool {
return false
}
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")
}
return http2splitHeaderBlock(ctx, headerBlock, w.writeHeaderBlock)
}
func (w *http2writeResHeaders) writeHeaderBlock(ctx http2writeContext, frag []byte, firstFrag, lastFrag bool) error {
if firstFrag {
return ctx.Framer().WriteHeaders(http2HeadersFrameParam{
StreamID: w.streamID,
BlockFragment: frag,
EndStream: w.endStream,
EndHeaders: lastFrag,
})
} else {
return ctx.Framer().WriteContinuation(w.streamID, lastFrag, frag)
}
}
type http2writePushPromise struct {
streamID uint32
method string
url *url.URL
h Header
allocatePromisedID func() (uint32, error)
promisedID uint32
}
func (w *http2writePushPromise) staysWithinBuffer(max int) bool {
return false
}
func (w *http2writePushPromise) writeFrame(ctx http2writeContext) error {
enc, buf := ctx.HeaderEncoder()
buf.Reset()
http2encKV(enc, ":method", w.method)
http2encKV(enc, ":scheme", w.url.Scheme)
http2encKV(enc, ":authority", w.url.Host)
http2encKV(enc, ":path", w.url.RequestURI())
http2encodeHeaders(enc, w.h, nil)
headerBlock := buf.Bytes()
if len(headerBlock) == 0 {
panic("unexpected empty hpack")
}
return http2splitHeaderBlock(ctx, headerBlock, w.writeHeaderBlock)
}
func (w *http2writePushPromise) writeHeaderBlock(ctx http2writeContext, frag []byte, firstFrag, lastFrag bool) error {
if firstFrag {
return ctx.Framer().WritePushPromise(http2PushPromiseParam{
StreamID: w.streamID,
PromiseID: w.promisedID,
BlockFragment: frag,
EndHeaders: lastFrag,
})
} else {
return ctx.Framer().WriteContinuation(w.streamID, lastFrag, frag)
}
}
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,
})
}
func (w http2write100ContinueHeadersFrame) staysWithinBuffer(max int) bool {
return 9+2*(len(":status")+len("100")) <= max
}
type http2writeWindowUpdate struct {
streamID uint32
n uint32
}
func (wu http2writeWindowUpdate) staysWithinBuffer(max int) bool { return http2frameHeaderLen+4 <= max }
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 {
sorter := http2sorterPool.Get().(*http2sorter)
defer http2sorterPool.Put(sorter)
keys = sorter.Keys(h)
}
for _, k := range keys {
vv := h[k]
k = http2lowerHeader(k)
if !http2validWireHeaderFieldName(k) {
continue
}
isTE := k == "transfer-encoding"
for _, v := range vv {
if !httplex.ValidHeaderFieldValue(v) {
continue
}
if isTE && v != "trailers" {
continue
}
http2encKV(enc, k, v)
}
}
}
type http2WriteScheduler interface {
OpenStream(streamID uint32, options http2OpenStreamOptions)
CloseStream(streamID uint32)
AdjustStream(streamID uint32, priority http2PriorityParam)
Push(wr http2FrameWriteRequest)
Pop() (wr http2FrameWriteRequest, ok bool)
}
type http2OpenStreamOptions struct {
PusherID uint32
}
type http2FrameWriteRequest struct {
write http2writeFramer
stream *http2stream
done chan error
}
func (wr http2FrameWriteRequest) StreamID() uint32 {
if wr.stream == nil {
if se, ok := wr.write.(http2StreamError); ok {
return se.StreamID
}
return 0
}
return wr.stream.id
}
func (wr http2FrameWriteRequest) DataSize() int {
if wd, ok := wr.write.(*http2writeData); ok {
return len(wd.p)
}
return 0
}
func (wr http2FrameWriteRequest) Consume(n int32) (http2FrameWriteRequest, http2FrameWriteRequest, int) {
var empty http2FrameWriteRequest
wd, ok := wr.write.(*http2writeData)
if !ok || len(wd.p) == 0 {
return wr, empty, 1
}
allowed := wr.stream.flow.available()
if n < allowed {
allowed = n
}
if wr.stream.sc.maxFrameSize < allowed {
allowed = wr.stream.sc.maxFrameSize
}
if allowed <= 0 {
return empty, empty, 0
}
if len(wd.p) > int(allowed) {
wr.stream.flow.take(allowed)
consumed := http2FrameWriteRequest{
stream: wr.stream,
write: &http2writeData{
streamID: wd.streamID,
p: wd.p[:allowed],
endStream: false,
},
done: nil,
}
rest := http2FrameWriteRequest{
stream: wr.stream,
write: &http2writeData{
streamID: wd.streamID,
p: wd.p[allowed:],
endStream: wd.endStream,
},
done: wr.done,
}
return consumed, rest, 2
}
wr.stream.flow.take(int32(len(wd.p)))
return wr, empty, 1
}
func (wr http2FrameWriteRequest) String() string {
var des string
if s, ok := wr.write.(fmt.Stringer); ok {
des = s.String()
} else {
des = fmt.Sprintf("%T", wr.write)
}
return fmt.Sprintf("[FrameWriteRequest stream=%d, ch=%v, writer=%v]", wr.StreamID(), wr.done != nil, des)
}
func (wr *http2FrameWriteRequest) replyToWriter(err error) {
if wr.done == nil {
return
}
select {
case wr.done <- err:
default:
panic(fmt.Sprintf("unbuffered done channel passed in for type %T", wr.write))
}
wr.write = nil
}
type http2writeQueue struct {
s []http2FrameWriteRequest
}
func (q *http2writeQueue) empty() bool { return len(q.s) == 0 }
func (q *http2writeQueue) push(wr http2FrameWriteRequest) {
q.s = append(q.s, wr)
}
func (q *http2writeQueue) shift() http2FrameWriteRequest {
if len(q.s) == 0 {
panic("invalid use of queue")
}
wr := q.s[0]
copy(q.s, q.s[1:])
q.s[len(q.s)-1] = http2FrameWriteRequest{}
q.s = q.s[:len(q.s)-1]
return wr
}
func (q *http2writeQueue) consume(n int32) (http2FrameWriteRequest, bool) {
if len(q.s) == 0 {
return http2FrameWriteRequest{}, false
}
consumed, rest, numresult := q.s[0].Consume(n)
switch numresult {
case 0:
return http2FrameWriteRequest{}, false
case 1:
q.shift()
case 2:
q.s[0] = rest
}
return consumed, true
}
type http2writeQueuePool []*http2writeQueue
func (p *http2writeQueuePool) put(q *http2writeQueue) {
for i := range q.s {
q.s[i] = http2FrameWriteRequest{}
}
q.s = q.s[:0]
*p = append(*p, q)
}
func (p *http2writeQueuePool) get() *http2writeQueue {
ln := len(*p)
if ln == 0 {
return new(http2writeQueue)
}
x := ln - 1
q := (*p)[x]
(*p)[x] = nil
*p = (*p)[:x]
return q
}
const http2priorityDefaultWeight = 15
type http2PriorityWriteSchedulerConfig struct {
MaxClosedNodesInTree int
MaxIdleNodesInTree int
ThrottleOutOfOrderWrites bool
}
func http2NewPriorityWriteScheduler(cfg *http2PriorityWriteSchedulerConfig) http2WriteScheduler {
if cfg == nil {
cfg = &http2PriorityWriteSchedulerConfig{
MaxClosedNodesInTree: 10,
MaxIdleNodesInTree: 10,
ThrottleOutOfOrderWrites: false,
}
}
ws := &http2priorityWriteScheduler{
nodes: make(map[uint32]*http2priorityNode),
maxClosedNodesInTree: cfg.MaxClosedNodesInTree,
maxIdleNodesInTree: cfg.MaxIdleNodesInTree,
enableWriteThrottle: cfg.ThrottleOutOfOrderWrites,
}
ws.nodes[0] = &ws.root
if cfg.ThrottleOutOfOrderWrites {
ws.writeThrottleLimit = 1024
} else {
ws.writeThrottleLimit = math.MaxInt32
}
return ws
}
type http2priorityNodeState int
const (
http2priorityNodeOpen http2priorityNodeState = iota
http2priorityNodeClosed
http2priorityNodeIdle
)
type http2priorityNode struct {
q http2writeQueue
id uint32
weight uint8
state http2priorityNodeState
bytes int64
subtreeBytes int64
parent *http2priorityNode
kids *http2priorityNode
prev, next *http2priorityNode
}
func (n *http2priorityNode) setParent(parent *http2priorityNode) {
if n == parent {
panic("setParent to self")
}
if n.parent == parent {
return
}
if parent := n.parent; parent != nil {
if n.prev == nil {
parent.kids = n.next
} else {
n.prev.next = n.next
}
if n.next != nil {
n.next.prev = n.prev
}
}
n.parent = parent
if parent == nil {
n.next = nil
n.prev = nil
} else {
n.next = parent.kids
n.prev = nil
if n.next != nil {
n.next.prev = n
}
parent.kids = n
}
}
func (n *http2priorityNode) addBytes(b int64) {
n.bytes += b
for ; n != nil; n = n.parent {
n.subtreeBytes += b
}
}
func (n *http2priorityNode) walkReadyInOrder(openParent bool, tmp *[]*http2priorityNode, f func(*http2priorityNode, bool) bool) bool {
if !n.q.empty() && f(n, openParent) {
return true
}
if n.kids == nil {
return false
}
if n.id != 0 {
openParent = openParent || (n.state == http2priorityNodeOpen)
}
w := n.kids.weight
needSort := false
for k := n.kids.next; k != nil; k = k.next {
if k.weight != w {
needSort = true
break
}
}
if !needSort {
for k := n.kids; k != nil; k = k.next {
if k.walkReadyInOrder(openParent, tmp, f) {
return true
}
}
return false
}
*tmp = (*tmp)[:0]
for n.kids != nil {
*tmp = append(*tmp, n.kids)
n.kids.setParent(nil)
}
sort.Sort(http2sortPriorityNodeSiblings(*tmp))
for i := len(*tmp) - 1; i >= 0; i-- {
(*tmp)[i].setParent(n)
}
for k := n.kids; k != nil; k = k.next {
if k.walkReadyInOrder(openParent, tmp, f) {
return true
}
}
return false
}
type http2sortPriorityNodeSiblings []*http2priorityNode
func (z http2sortPriorityNodeSiblings) Len() int { return len(z) }
func (z http2sortPriorityNodeSiblings) Swap(i, k int) { z[i], z[k] = z[k], z[i] }
func (z http2sortPriorityNodeSiblings) Less(i, k int) bool {
wi, bi := float64(z[i].weight+1), float64(z[i].subtreeBytes)
wk, bk := float64(z[k].weight+1), float64(z[k].subtreeBytes)
if bi == 0 && bk == 0 {
return wi >= wk
}
if bk == 0 {
return false
}
return bi/bk <= wi/wk
}
type http2priorityWriteScheduler struct {
root http2priorityNode
nodes map[uint32]*http2priorityNode
maxID uint32
closedNodes, idleNodes []*http2priorityNode
maxClosedNodesInTree int
maxIdleNodesInTree int
writeThrottleLimit int32
enableWriteThrottle bool
tmp []*http2priorityNode
queuePool http2writeQueuePool
}
func (ws *http2priorityWriteScheduler) OpenStream(streamID uint32, options http2OpenStreamOptions) {
if curr := ws.nodes[streamID]; curr != nil {
if curr.state != http2priorityNodeIdle {
panic(fmt.Sprintf("stream %d already opened", streamID))
}
curr.state = http2priorityNodeOpen
return
}
parent := ws.nodes[options.PusherID]
if parent == nil {
parent = &ws.root
}
n := &http2priorityNode{
q: *ws.queuePool.get(),
id: streamID,
weight: http2priorityDefaultWeight,
state: http2priorityNodeOpen,
}
n.setParent(parent)
ws.nodes[streamID] = n
if streamID > ws.maxID {
ws.maxID = streamID
}
}
func (ws *http2priorityWriteScheduler) CloseStream(streamID uint32) {
if streamID == 0 {
panic("violation of WriteScheduler interface: cannot close stream 0")
}
if ws.nodes[streamID] == nil {
panic(fmt.Sprintf("violation of WriteScheduler interface: unknown stream %d", streamID))
}
if ws.nodes[streamID].state != http2priorityNodeOpen {
panic(fmt.Sprintf("violation of WriteScheduler interface: stream %d already closed", streamID))
}
n := ws.nodes[streamID]
n.state = http2priorityNodeClosed
n.addBytes(-n.bytes)
q := n.q
ws.queuePool.put(&q)
n.q.s = nil
if ws.maxClosedNodesInTree > 0 {
ws.addClosedOrIdleNode(&ws.closedNodes, ws.maxClosedNodesInTree, n)
} else {
ws.removeNode(n)
}
}
func (ws *http2priorityWriteScheduler) AdjustStream(streamID uint32, priority http2PriorityParam) {
if streamID == 0 {
panic("adjustPriority on root")
}
n := ws.nodes[streamID]
if n == nil {
if streamID <= ws.maxID || ws.maxIdleNodesInTree == 0 {
return
}
ws.maxID = streamID
n = &http2priorityNode{
q: *ws.queuePool.get(),
id: streamID,
weight: http2priorityDefaultWeight,
state: http2priorityNodeIdle,
}
n.setParent(&ws.root)
ws.nodes[streamID] = n
ws.addClosedOrIdleNode(&ws.idleNodes, ws.maxIdleNodesInTree, n)
}
parent := ws.nodes[priority.StreamDep]
if parent == nil {
n.setParent(&ws.root)
n.weight = http2priorityDefaultWeight
return
}
if n == parent {
return
}
for x := parent.parent; x != nil; x = x.parent {
if x == n {
parent.setParent(n.parent)
break
}
}
if priority.Exclusive {
k := parent.kids
for k != nil {
next := k.next
if k != n {
k.setParent(n)
}
k = next
}
}
n.setParent(parent)
n.weight = priority.Weight
}
func (ws *http2priorityWriteScheduler) Push(wr http2FrameWriteRequest) {
var n *http2priorityNode
if id := wr.StreamID(); id == 0 {
n = &ws.root
} else {
n = ws.nodes[id]
if n == nil {
if wr.DataSize() > 0 {
panic("add DATA on non-open stream")
}
n = &ws.root
}
}
n.q.push(wr)
}
func (ws *http2priorityWriteScheduler) Pop() (wr http2FrameWriteRequest, ok bool) {
ws.root.walkReadyInOrder(false, &ws.tmp, func(n *http2priorityNode, openParent bool) bool {
limit := int32(math.MaxInt32)
if openParent {
limit = ws.writeThrottleLimit
}
wr, ok = n.q.consume(limit)
if !ok {
return false
}
n.addBytes(int64(wr.DataSize()))
if openParent {
ws.writeThrottleLimit += 1024
if ws.writeThrottleLimit < 0 {
ws.writeThrottleLimit = math.MaxInt32
}
} else if ws.enableWriteThrottle {
ws.writeThrottleLimit = 1024
}
return true
})
return wr, ok
}
func (ws *http2priorityWriteScheduler) addClosedOrIdleNode(list *[]*http2priorityNode, maxSize int, n *http2priorityNode) {
if maxSize == 0 {
return
}
if len(*list) == maxSize {
ws.removeNode((*list)[0])
x := (*list)[1:]
copy(*list, x)
*list = (*list)[:len(x)]
}
*list = append(*list, n)
}
func (ws *http2priorityWriteScheduler) removeNode(n *http2priorityNode) {
for k := n.kids; k != nil; k = k.next {
k.setParent(n.parent)
}
n.setParent(nil)
delete(ws.nodes, n.id)
}
func http2NewRandomWriteScheduler() http2WriteScheduler {
return &http2randomWriteScheduler{sq: make(map[uint32]*http2writeQueue)}
}
type http2randomWriteScheduler struct {
zero http2writeQueue
sq map[uint32]*http2writeQueue
queuePool http2writeQueuePool
}
func (ws *http2randomWriteScheduler) OpenStream(streamID uint32, options http2OpenStreamOptions) {
}
func (ws *http2randomWriteScheduler) CloseStream(streamID uint32) {
q, ok := ws.sq[streamID]
if !ok {
return
}
delete(ws.sq, streamID)
ws.queuePool.put(q)
}
func (ws *http2randomWriteScheduler) AdjustStream(streamID uint32, priority http2PriorityParam) {
}
func (ws *http2randomWriteScheduler) Push(wr http2FrameWriteRequest) {
id := wr.StreamID()
if id == 0 {
ws.zero.push(wr)
return
}
q, ok := ws.sq[id]
if !ok {
q = ws.queuePool.get()
ws.sq[id] = q
}
q.push(wr)
}
func (ws *http2randomWriteScheduler) Pop() (http2FrameWriteRequest, bool) {
if !ws.zero.empty() {
return ws.zero.shift(), true
}
for _, q := range ws.sq {
if wr, ok := q.consume(math.MaxInt32); ok {
return wr, true
}
}
return http2FrameWriteRequest{}, false
}
View as plain text