writesched.go

     1  // Copyright 2014 The Go Authors. All rights reserved.
     2  // Use of this source code is governed by a BSD-style
     3  // license that can be found in the LICENSE file.
     4  
     5  package http2
     6  
     7  import "fmt"
     8  
     9  // WriteScheduler is the interface implemented by HTTP/2 write schedulers.
    10  // Methods are never called concurrently.
    11  type WriteScheduler interface {
    12  	// OpenStream opens a new stream in the write scheduler.
    13  	// It is illegal to call this with streamID=0 or with a streamID that is
    14  	// already open -- the call may panic.
    15  	OpenStream(streamID uint32, options OpenStreamOptions)
    16  
    17  	// CloseStream closes a stream in the write scheduler. Any frames queued on
    18  	// this stream should be discarded. It is illegal to call this on a stream
    19  	// that is not open -- the call may panic.
    20  	CloseStream(streamID uint32)
    21  
    22  	// AdjustStream adjusts the priority of the given stream. This may be called
    23  	// on a stream that has not yet been opened or has been closed. Note that
    24  	// RFC 7540 allows PRIORITY frames to be sent on streams in any state. See:
    25  	// https://tools.ietf.org/html/rfc7540#section-5.1
    26  	AdjustStream(streamID uint32, priority PriorityParam)
    27  
    28  	// Push queues a frame in the scheduler. In most cases, this will not be
    29  	// called with wr.StreamID()!=0 unless that stream is currently open. The one
    30  	// exception is RST_STREAM frames, which may be sent on idle or closed streams.
    31  	Push(wr FrameWriteRequest)
    32  
    33  	// Pop dequeues the next frame to write. Returns false if no frames can
    34  	// be written. Frames with a given wr.StreamID() are Pop'd in the same
    35  	// order they are Push'd, except RST_STREAM frames. No frames should be
    36  	// discarded except by CloseStream.
    37  	Pop() (wr FrameWriteRequest, ok bool)
    38  }
    39  
    40  // OpenStreamOptions specifies extra options for WriteScheduler.OpenStream.
    41  type OpenStreamOptions struct {
    42  	// PusherID is zero if the stream was initiated by the client. Otherwise,
    43  	// PusherID names the stream that pushed the newly opened stream.
    44  	PusherID uint32
    45  	// priority is used to set the priority of the newly opened stream.
    46  	priority PriorityParam
    47  }
    48  
    49  // FrameWriteRequest is a request to write a frame.
    50  type FrameWriteRequest struct {
    51  	// write is the interface value that does the writing, once the
    52  	// WriteScheduler has selected this frame to write. The write
    53  	// functions are all defined in write.go.
    54  	write writeFramer
    55  
    56  	// stream is the stream on which this frame will be written.
    57  	// nil for non-stream frames like PING and SETTINGS.
    58  	// nil for RST_STREAM streams, which use the StreamError.StreamID field instead.
    59  	stream *stream
    60  
    61  	// done, if non-nil, must be a buffered channel with space for
    62  	// 1 message and is sent the return value from write (or an
    63  	// earlier error) when the frame has been written.
    64  	done chan error
    65  }
    66  
    67  // StreamID returns the id of the stream this frame will be written to.
    68  // 0 is used for non-stream frames such as PING and SETTINGS.
    69  func (wr FrameWriteRequest) StreamID() uint32 {
    70  	if wr.stream == nil {
    71  		if se, ok := wr.write.(StreamError); ok {
    72  			// (*serverConn).resetStream doesn't set
    73  			// stream because it doesn't necessarily have
    74  			// one. So special case this type of write
    75  			// message.
    76  			return se.StreamID
    77  		}
    78  		return 0
    79  	}
    80  	return wr.stream.id
    81  }
    82  
    83  // isControl reports whether wr is a control frame for MaxQueuedControlFrames
    84  // purposes. That includes non-stream frames and RST_STREAM frames.
    85  func (wr FrameWriteRequest) isControl() bool {
    86  	return wr.stream == nil
    87  }
    88  
    89  // DataSize returns the number of flow control bytes that must be consumed
    90  // to write this entire frame. This is 0 for non-DATA frames.
    91  func (wr FrameWriteRequest) DataSize() int {
    92  	if wd, ok := wr.write.(*writeData); ok {
    93  		return len(wd.p)
    94  	}
    95  	return 0
    96  }
    97  
    98  // Consume consumes min(n, available) bytes from this frame, where available
    99  // is the number of flow control bytes available on the stream. Consume returns
   100  // 0, 1, or 2 frames, where the integer return value gives the number of frames
   101  // returned.
   102  //
   103  // If flow control prevents consuming any bytes, this returns (_, _, 0). If
   104  // the entire frame was consumed, this returns (wr, _, 1). Otherwise, this
   105  // returns (consumed, rest, 2), where 'consumed' contains the consumed bytes and
   106  // 'rest' contains the remaining bytes. The consumed bytes are deducted from the
   107  // underlying stream's flow control budget.
   108  func (wr FrameWriteRequest) Consume(n int32) (FrameWriteRequest, FrameWriteRequest, int) {
   109  	var empty FrameWriteRequest
   110  
   111  	// Non-DATA frames are always consumed whole.
   112  	wd, ok := wr.write.(*writeData)
   113  	if !ok || len(wd.p) == 0 {
   114  		return wr, empty, 1
   115  	}
   116  
   117  	// Might need to split after applying limits.
   118  	allowed := wr.stream.flow.available()
   119  	if n < allowed {
   120  		allowed = n
   121  	}
   122  	if wr.stream.sc.maxFrameSize < allowed {
   123  		allowed = wr.stream.sc.maxFrameSize
   124  	}
   125  	if allowed <= 0 {
   126  		return empty, empty, 0
   127  	}
   128  	if len(wd.p) > int(allowed) {
   129  		wr.stream.flow.take(allowed)
   130  		consumed := FrameWriteRequest{
   131  			stream: wr.stream,
   132  			write: &writeData{
   133  				streamID: wd.streamID,
   134  				p:        wd.p[:allowed],
   135  				// Even if the original had endStream set, there
   136  				// are bytes remaining because len(wd.p) > allowed,
   137  				// so we know endStream is false.
   138  				endStream: false,
   139  			},
   140  			// Our caller is blocking on the final DATA frame, not
   141  			// this intermediate frame, so no need to wait.
   142  			done: nil,
   143  		}
   144  		rest := FrameWriteRequest{
   145  			stream: wr.stream,
   146  			write: &writeData{
   147  				streamID:  wd.streamID,
   148  				p:         wd.p[allowed:],
   149  				endStream: wd.endStream,
   150  			},
   151  			done: wr.done,
   152  		}
   153  		return consumed, rest, 2
   154  	}
   155  
   156  	// The frame is consumed whole.
   157  	// NB: This cast cannot overflow because allowed is <= math.MaxInt32.
   158  	wr.stream.flow.take(int32(len(wd.p)))
   159  	return wr, empty, 1
   160  }
   161  
   162  // String is for debugging only.
   163  func (wr FrameWriteRequest) String() string {
   164  	var des string
   165  	if s, ok := wr.write.(fmt.Stringer); ok {
   166  		des = s.String()
   167  	} else {
   168  		des = fmt.Sprintf("%T", wr.write)
   169  	}
   170  	return fmt.Sprintf("[FrameWriteRequest stream=%d, ch=%v, writer=%v]", wr.StreamID(), wr.done != nil, des)
   171  }
   172  
   173  // replyToWriter sends err to wr.done and panics if the send must block
   174  // This does nothing if wr.done is nil.
   175  func (wr *FrameWriteRequest) replyToWriter(err error) {
   176  	if wr.done == nil {
   177  		return
   178  	}
   179  	select {
   180  	case wr.done <- err:
   181  	default:
   182  		panic(fmt.Sprintf("unbuffered done channel passed in for type %T", wr.write))
   183  	}
   184  	wr.write = nil // prevent use (assume it's tainted after wr.done send)
   185  }
   186  
   187  // writeQueue is used by implementations of WriteScheduler.
   188  //
   189  // Each writeQueue contains a queue of FrameWriteRequests, meant to store all
   190  // FrameWriteRequests associated with a given stream. This is implemented as a
   191  // two-stage queue: currQueue[currPos:] and nextQueue. Removing an item is done
   192  // by incrementing currPos of currQueue. Adding an item is done by appending it
   193  // to the nextQueue. If currQueue is empty when trying to remove an item, we
   194  // can swap currQueue and nextQueue to remedy the situation.
   195  // This two-stage queue is analogous to the use of two lists in Okasaki's
   196  // purely functional queue but without the overhead of reversing the list when
   197  // swapping stages.
   198  //
   199  // writeQueue also contains prev and next, this can be used by implementations
   200  // of WriteScheduler to construct data structures that represent the order of
   201  // writing between different streams (e.g. circular linked list).
   202  type writeQueue struct {
   203  	currQueue []FrameWriteRequest
   204  	nextQueue []FrameWriteRequest
   205  	currPos   int
   206  
   207  	prev, next *writeQueue
   208  }
   209  
   210  func (q *writeQueue) empty() bool {
   211  	return (len(q.currQueue) - q.currPos + len(q.nextQueue)) == 0
   212  }
   213  
   214  func (q *writeQueue) push(wr FrameWriteRequest) {
   215  	q.nextQueue = append(q.nextQueue, wr)
   216  }
   217  
   218  func (q *writeQueue) shift() FrameWriteRequest {
   219  	if q.empty() {
   220  		panic("invalid use of queue")
   221  	}
   222  	if q.currPos >= len(q.currQueue) {
   223  		q.currQueue, q.currPos, q.nextQueue = q.nextQueue, 0, q.currQueue[:0]
   224  	}
   225  	wr := q.currQueue[q.currPos]
   226  	q.currQueue[q.currPos] = FrameWriteRequest{}
   227  	q.currPos++
   228  	return wr
   229  }
   230  
   231  func (q *writeQueue) peek() *FrameWriteRequest {
   232  	if q.currPos < len(q.currQueue) {
   233  		return &q.currQueue[q.currPos]
   234  	}
   235  	if len(q.nextQueue) > 0 {
   236  		return &q.nextQueue[0]
   237  	}
   238  	return nil
   239  }
   240  
   241  // consume consumes up to n bytes from q.s[0]. If the frame is
   242  // entirely consumed, it is removed from the queue. If the frame
   243  // is partially consumed, the frame is kept with the consumed
   244  // bytes removed. Returns true iff any bytes were consumed.
   245  func (q *writeQueue) consume(n int32) (FrameWriteRequest, bool) {
   246  	if q.empty() {
   247  		return FrameWriteRequest{}, false
   248  	}
   249  	consumed, rest, numresult := q.peek().Consume(n)
   250  	switch numresult {
   251  	case 0:
   252  		return FrameWriteRequest{}, false
   253  	case 1:
   254  		q.shift()
   255  	case 2:
   256  		*q.peek() = rest
   257  	}
   258  	return consumed, true
   259  }
   260  
   261  type writeQueuePool []*writeQueue
   262  
   263  // put inserts an unused writeQueue into the pool.
   264  func (p *writeQueuePool) put(q *writeQueue) {
   265  	for i := range q.currQueue {
   266  		q.currQueue[i] = FrameWriteRequest{}
   267  	}
   268  	for i := range q.nextQueue {
   269  		q.nextQueue[i] = FrameWriteRequest{}
   270  	}
   271  	q.currQueue = q.currQueue[:0]
   272  	q.nextQueue = q.nextQueue[:0]
   273  	q.currPos = 0
   274  	*p = append(*p, q)
   275  }
   276  
   277  // get returns an empty writeQueue.
   278  func (p *writeQueuePool) get() *writeQueue {
   279  	ln := len(*p)
   280  	if ln == 0 {
   281  		return new(writeQueue)
   282  	}
   283  	x := ln - 1
   284  	q := (*p)[x]
   285  	(*p)[x] = nil
   286  	*p = (*p)[:x]
   287  	return q
   288  }
   289
View as plain text