Source: writesched.go in package golang.org/x/net/http2


package http2

import "fmt"

WriteScheduler is the interface implemented by HTTP/2 write schedulers. Methods are never called concurrently.

OpenStream opens a new stream in the write scheduler. It is illegal to call this with streamID=0 or with a streamID that is already open -- the call may panic.

	OpenStream(streamID uint32, options OpenStreamOptions)

CloseStream closes a stream in the write scheduler. Any frames queued on this stream should be discarded. It is illegal to call this on a stream that is not open -- the call may panic.

	CloseStream(streamID uint32)

AdjustStream adjusts the priority of the given stream. This may be called on a stream that has not yet been opened or has been closed. Note that RFC 7540 allows PRIORITY frames to be sent on streams in any state. See: https://tools.ietf.org/html/rfc7540#section-5.1

	AdjustStream(streamID uint32, priority PriorityParam)

Push queues a frame in the scheduler. In most cases, this will not be called with wr.StreamID()!=0 unless that stream is currently open. The one exception is RST_STREAM frames, which may be sent on idle or closed streams.

	Push(wr FrameWriteRequest)

Pop dequeues the next frame to write. Returns false if no frames can be written. Frames with a given wr.StreamID() are Pop'd in the same order they are Push'd. No frames should be discarded except by CloseStream.

	Pop() (wr FrameWriteRequest, ok bool)
}

OpenStreamOptions specifies extra options for WriteScheduler.OpenStream.

PusherID is zero if the stream was initiated by the client. Otherwise, PusherID names the stream that pushed the newly opened stream.

	PusherID uint32
}

FrameWriteRequest is a request to write a frame.

write is the interface value that does the writing, once the WriteScheduler has selected this frame to write. The write functions are all defined in write.go.

	write writeFramer

stream is the stream on which this frame will be written. nil for non-stream frames like PING and SETTINGS.

	stream *stream

done, if non-nil, must be a buffered channel with space for 1 message and is sent the return value from write (or an earlier error) when the frame has been written.

	done chan error
}

StreamID returns the id of the stream this frame will be written to. 0 is used for non-stream frames such as PING and SETTINGS.

func (wr FrameWriteRequest) StreamID() uint32 {
	if wr.stream == nil {

(*serverConn).resetStream doesn't set stream because it doesn't necessarily have one. So special case this type of write message.

			return se.StreamID
		}
		return 0
	}
	return wr.stream.id
}

isControl reports whether wr is a control frame for MaxQueuedControlFrames purposes. That includes non-stream frames and RST_STREAM frames.

func (wr FrameWriteRequest) isControl() bool {
	return wr.stream == nil
}

DataSize returns the number of flow control bytes that must be consumed to write this entire frame. This is 0 for non-DATA frames.

func (wr FrameWriteRequest) DataSize() int {
	if wd, ok := wr.write.(*writeData); ok {
		return len(wd.p)
	}
	return 0
}

Consume consumes min(n, available) bytes from this frame, where available is the number of flow control bytes available on the stream. Consume returns 0, 1, or 2 frames, where the integer return value gives the number of frames returned. If flow control prevents consuming any bytes, this returns (_, _, 0). If the entire frame was consumed, this returns (wr, _, 1). Otherwise, this returns (consumed, rest, 2), where 'consumed' contains the consumed bytes and 'rest' contains the remaining bytes. The consumed bytes are deducted from the underlying stream's flow control budget.

func (wr FrameWriteRequest) Consume(n int32) (FrameWriteRequest, FrameWriteRequest, int) {
	var empty FrameWriteRequest

Non-DATA frames are always consumed whole.

	wd, ok := wr.write.(*writeData)
	if !ok || len(wd.p) == 0 {
		return wr, empty, 1
	}

Might need to split after applying limits.

	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 := FrameWriteRequest{
			stream: wr.stream,
			write: &writeData{
				streamID: wd.streamID,

Even if the original had endStream set, there are bytes remaining because len(wd.p) > allowed, so we know endStream is false.

				endStream: false,

Our caller is blocking on the final DATA frame, not this intermediate frame, so no need to wait.

			done: nil,
		}
		rest := FrameWriteRequest{
			stream: wr.stream,
			write: &writeData{
				streamID:  wd.streamID,
				p:         wd.p[allowed:],
				endStream: wd.endStream,
			},
			done: wr.done,
		}
		return consumed, rest, 2
	}

The frame is consumed whole. NB: This cast cannot overflow because allowed is <= math.MaxInt32.

	wr.stream.flow.take(int32(len(wd.p)))
	return wr, empty, 1
}

String is for debugging only.

func (wr FrameWriteRequest) 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)
}

replyToWriter sends err to wr.done and panics if the send must block This does nothing if wr.done is nil.

func (wr *FrameWriteRequest) 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 // prevent use (assume it's tainted after wr.done send)
}

writeQueue is used by implementations of WriteScheduler.

type writeQueue struct {
	s []FrameWriteRequest
}

func (q *writeQueue) empty() bool { return len(q.s) == 0 }

func (q *writeQueue) push(wr FrameWriteRequest) {
	q.s = append(q.s, wr)
}

func (q *writeQueue) shift() FrameWriteRequest {
	if len(q.s) == 0 {
		panic("invalid use of queue")
	}

TODO: less copy-happy queue.

	copy(q.s, q.s[1:])
	q.s[len(q.s)-1] = FrameWriteRequest{}
	q.s = q.s[:len(q.s)-1]
	return wr
}

consume consumes up to n bytes from q.s[0]. If the frame is entirely consumed, it is removed from the queue. If the frame is partially consumed, the frame is kept with the consumed bytes removed. Returns true iff any bytes were consumed.

func (q *writeQueue) consume(n int32) (FrameWriteRequest, bool) {
	if len(q.s) == 0 {
		return FrameWriteRequest{}, false
	}
	consumed, rest, numresult := q.s[0].Consume(n)
	switch numresult {
	case 0:
		return FrameWriteRequest{}, false
	case 1:
		q.shift()
	case 2:
		q.s[0] = rest
	}
	return consumed, true
}

type writeQueuePool []*writeQueue

put inserts an unused writeQueue into the pool.

func (p *writeQueuePool) put(q *writeQueue) {
	for i := range q.s {
		q.s[i] = FrameWriteRequest{}
	}
	q.s = q.s[:0]
	*p = append(*p, q)
}

get returns an empty writeQueue.

func (p *writeQueuePool) get() *writeQueue {
	ln := len(*p)
	if ln == 0 {
		return new(writeQueue)
	}
	x := ln - 1
	q := (*p)[x]
	(*p)[x] = nil
	*p = (*p)[:x]
	return q