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  	avail, ok := wr.stream.flow.available()
   119  	if !ok {
   120  		return empty, empty, 0
   121  	}
   122  	allowed := min(n, avail)
   123  	if wr.stream.sc.maxFrameSize < allowed {
   124  		allowed = wr.stream.sc.maxFrameSize
   125  	}
   126  	if allowed <= 0 {
   127  		return empty, empty, 0
   128  	}
   129  	if len(wd.p) > int(allowed) {
   130  		wr.stream.flow.take(allowed)
   131  		consumed := FrameWriteRequest{
   132  			stream: wr.stream,
   133  			write: &writeData{
   134  				streamID: wd.streamID,
   135  				p:        wd.p[:allowed],
   136  				// Even if the original had endStream set, there
   137  				// are bytes remaining because len(wd.p) > allowed,
   138  				// so we know endStream is false.
   139  				endStream: false,
   140  			},
   141  			// Our caller is blocking on the final DATA frame, not
   142  			// this intermediate frame, so no need to wait.
   143  			done: nil,
   144  		}
   145  		rest := FrameWriteRequest{
   146  			stream: wr.stream,
   147  			write: &writeData{
   148  				streamID:  wd.streamID,
   149  				p:         wd.p[allowed:],
   150  				endStream: wd.endStream,
   151  			},
   152  			done: wr.done,
   153  		}
   154  		return consumed, rest, 2
   155  	}
   156  
   157  	// The frame is consumed whole.
   158  	// NB: This cast cannot overflow because allowed is <= math.MaxInt32.
   159  	wr.stream.flow.take(int32(len(wd.p)))
   160  	return wr, empty, 1
   161  }
   162  
   163  // String is for debugging only.
   164  func (wr FrameWriteRequest) String() string {
   165  	var des string
   166  	if s, ok := wr.write.(fmt.Stringer); ok {
   167  		des = s.String()
   168  	} else {
   169  		des = fmt.Sprintf("%T", wr.write)
   170  	}
   171  	return fmt.Sprintf("[FrameWriteRequest stream=%d, ch=%v, writer=%v]", wr.StreamID(), wr.done != nil, des)
   172  }
   173  
   174  // replyToWriter sends err to wr.done and panics if the send must block
   175  // This does nothing if wr.done is nil.
   176  func (wr *FrameWriteRequest) replyToWriter(err error) {
   177  	if wr.done == nil {
   178  		return
   179  	}
   180  	select {
   181  	case wr.done <- err:
   182  	default:
   183  		panic(fmt.Sprintf("unbuffered done channel passed in for type %T", wr.write))
   184  	}
   185  	wr.write = nil // prevent use (assume it's tainted after wr.done send)
   186  }
   187  
   188  // writeQueue is used by implementations of WriteScheduler.
   189  //
   190  // Each writeQueue contains a queue of FrameWriteRequests, meant to store all
   191  // FrameWriteRequests associated with a given stream. This is implemented as a
   192  // two-stage queue: currQueue[currPos:] and nextQueue. Removing an item is done
   193  // by incrementing currPos of currQueue. Adding an item is done by appending it
   194  // to the nextQueue. If currQueue is empty when trying to remove an item, we
   195  // can swap currQueue and nextQueue to remedy the situation.
   196  // This two-stage queue is analogous to the use of two lists in Okasaki's
   197  // purely functional queue but without the overhead of reversing the list when
   198  // swapping stages.
   199  //
   200  // writeQueue also contains prev and next, this can be used by implementations
   201  // of WriteScheduler to construct data structures that represent the order of
   202  // writing between different streams (e.g. circular linked list).
   203  type writeQueue struct {
   204  	currQueue []FrameWriteRequest
   205  	nextQueue []FrameWriteRequest
   206  	currPos   int
   207  
   208  	prev, next *writeQueue
   209  }
   210  
   211  func (q *writeQueue) empty() bool {
   212  	return (len(q.currQueue) - q.currPos + len(q.nextQueue)) == 0
   213  }
   214  
   215  func (q *writeQueue) push(wr FrameWriteRequest) {
   216  	q.nextQueue = append(q.nextQueue, wr)
   217  }
   218  
   219  func (q *writeQueue) shift() FrameWriteRequest {
   220  	if q.empty() {
   221  		panic("invalid use of queue")
   222  	}
   223  	if q.currPos >= len(q.currQueue) {
   224  		q.currQueue, q.currPos, q.nextQueue = q.nextQueue, 0, q.currQueue[:0]
   225  	}
   226  	wr := q.currQueue[q.currPos]
   227  	q.currQueue[q.currPos] = FrameWriteRequest{}
   228  	q.currPos++
   229  	return wr
   230  }
   231  
   232  func (q *writeQueue) peek() *FrameWriteRequest {
   233  	if q.currPos < len(q.currQueue) {
   234  		return &q.currQueue[q.currPos]
   235  	}
   236  	if len(q.nextQueue) > 0 {
   237  		return &q.nextQueue[0]
   238  	}
   239  	return nil
   240  }
   241  
   242  // consume consumes up to n bytes from q.s[0]. If the frame is
   243  // entirely consumed, it is removed from the queue. If the frame
   244  // is partially consumed, the frame is kept with the consumed
   245  // bytes removed. Returns true iff any bytes were consumed.
   246  func (q *writeQueue) consume(n int32) (FrameWriteRequest, bool) {
   247  	if q.empty() {
   248  		return FrameWriteRequest{}, false
   249  	}
   250  	consumed, rest, numresult := q.peek().Consume(n)
   251  	switch numresult {
   252  	case 0:
   253  		return FrameWriteRequest{}, false
   254  	case 1:
   255  		q.shift()
   256  	case 2:
   257  		*q.peek() = rest
   258  	}
   259  	return consumed, true
   260  }
   261  
   262  type writeQueuePool []*writeQueue
   263  
   264  // put inserts an unused writeQueue into the pool.
   265  func (p *writeQueuePool) put(q *writeQueue) {
   266  	for i := range q.currQueue {
   267  		q.currQueue[i] = FrameWriteRequest{}
   268  	}
   269  	for i := range q.nextQueue {
   270  		q.nextQueue[i] = FrameWriteRequest{}
   271  	}
   272  	q.currQueue = q.currQueue[:0]
   273  	q.nextQueue = q.nextQueue[:0]
   274  	q.currPos = 0
   275  	*p = append(*p, q)
   276  }
   277  
   278  // get returns an empty writeQueue.
   279  func (p *writeQueuePool) get() *writeQueue {
   280  	ln := len(*p)
   281  	if ln == 0 {
   282  		return new(writeQueue)
   283  	}
   284  	x := ln - 1
   285  	q := (*p)[x]
   286  	(*p)[x] = nil
   287  	*p = (*p)[:x]
   288  	return q
   289  }
   290
View as plain text