+++ /dev/null
-// Copyright 2014 The Go Authors. All rights reserved.
-// Use of this source code is governed by a BSD-style
-// license that can be found in the LICENSE file.
-
-package http2
-
-import "fmt"
-
-// WriteScheduler is the interface implemented by HTTP/2 write schedulers.
-// Methods are never called concurrently.
-type WriteScheduler interface {
- // 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.
- Pop() (wr FrameWriteRequest, ok bool)
-}
-
-// OpenStreamOptions specifies extra options for WriteScheduler.OpenStream.
-type OpenStreamOptions struct {
- // 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.
-type FrameWriteRequest struct {
- // 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 {
- if se, ok := wr.write.(StreamError); ok {
- // (*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
-}
-
-// 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,
- p: wd.p[:allowed],
- // 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")
- }
- wr := q.s[0]
- // 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
-}
OSDN Git Service
