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