writesched.go 7.2 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242
  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. package http2
  5. import "fmt"
  6. // WriteScheduler is the interface implemented by HTTP/2 write schedulers.
  7. // Methods are never called concurrently.
  8. type WriteScheduler interface {
  9. // OpenStream opens a new stream in the write scheduler.
  10. // It is illegal to call this with streamID=0 or with a streamID that is
  11. // already open -- the call may panic.
  12. OpenStream(streamID uint32, options OpenStreamOptions)
  13. // CloseStream closes a stream in the write scheduler. Any frames queued on
  14. // this stream should be discarded. It is illegal to call this on a stream
  15. // that is not open -- the call may panic.
  16. CloseStream(streamID uint32)
  17. // AdjustStream adjusts the priority of the given stream. This may be called
  18. // on a stream that has not yet been opened or has been closed. Note that
  19. // RFC 7540 allows PRIORITY frames to be sent on streams in any state. See:
  20. // https://tools.ietf.org/html/rfc7540#section-5.1
  21. AdjustStream(streamID uint32, priority PriorityParam)
  22. // Push queues a frame in the scheduler. In most cases, this will not be
  23. // called with wr.StreamID()!=0 unless that stream is currently open. The one
  24. // exception is RST_STREAM frames, which may be sent on idle or closed streams.
  25. Push(wr FrameWriteRequest)
  26. // Pop dequeues the next frame to write. Returns false if no frames can
  27. // be written. Frames with a given wr.StreamID() are Pop'd in the same
  28. // order they are Push'd.
  29. Pop() (wr FrameWriteRequest, ok bool)
  30. }
  31. // OpenStreamOptions specifies extra options for WriteScheduler.OpenStream.
  32. type OpenStreamOptions struct {
  33. // PusherID is zero if the stream was initiated by the client. Otherwise,
  34. // PusherID names the stream that pushed the newly opened stream.
  35. PusherID uint32
  36. }
  37. // FrameWriteRequest is a request to write a frame.
  38. type FrameWriteRequest struct {
  39. // write is the interface value that does the writing, once the
  40. // WriteScheduler has selected this frame to write. The write
  41. // functions are all defined in write.go.
  42. write writeFramer
  43. // stream is the stream on which this frame will be written.
  44. // nil for non-stream frames like PING and SETTINGS.
  45. stream *stream
  46. // done, if non-nil, must be a buffered channel with space for
  47. // 1 message and is sent the return value from write (or an
  48. // earlier error) when the frame has been written.
  49. done chan error
  50. }
  51. // StreamID returns the id of the stream this frame will be written to.
  52. // 0 is used for non-stream frames such as PING and SETTINGS.
  53. func (wr FrameWriteRequest) StreamID() uint32 {
  54. if wr.stream == nil {
  55. if se, ok := wr.write.(StreamError); ok {
  56. // (*serverConn).resetStream doesn't set
  57. // stream because it doesn't necessarily have
  58. // one. So special case this type of write
  59. // message.
  60. return se.StreamID
  61. }
  62. return 0
  63. }
  64. return wr.stream.id
  65. }
  66. // DataSize returns the number of flow control bytes that must be consumed
  67. // to write this entire frame. This is 0 for non-DATA frames.
  68. func (wr FrameWriteRequest) DataSize() int {
  69. if wd, ok := wr.write.(*writeData); ok {
  70. return len(wd.p)
  71. }
  72. return 0
  73. }
  74. // Consume consumes min(n, available) bytes from this frame, where available
  75. // is the number of flow control bytes available on the stream. Consume returns
  76. // 0, 1, or 2 frames, where the integer return value gives the number of frames
  77. // returned.
  78. //
  79. // If flow control prevents consuming any bytes, this returns (_, _, 0). If
  80. // the entire frame was consumed, this returns (wr, _, 1). Otherwise, this
  81. // returns (consumed, rest, 2), where 'consumed' contains the consumed bytes and
  82. // 'rest' contains the remaining bytes. The consumed bytes are deducted from the
  83. // underlying stream's flow control budget.
  84. func (wr FrameWriteRequest) Consume(n int32) (FrameWriteRequest, FrameWriteRequest, int) {
  85. var empty FrameWriteRequest
  86. // Non-DATA frames are always consumed whole.
  87. wd, ok := wr.write.(*writeData)
  88. if !ok || len(wd.p) == 0 {
  89. return wr, empty, 1
  90. }
  91. // Might need to split after applying limits.
  92. allowed := wr.stream.flow.available()
  93. if n < allowed {
  94. allowed = n
  95. }
  96. if wr.stream.sc.maxFrameSize < allowed {
  97. allowed = wr.stream.sc.maxFrameSize
  98. }
  99. if allowed <= 0 {
  100. return empty, empty, 0
  101. }
  102. if len(wd.p) > int(allowed) {
  103. wr.stream.flow.take(allowed)
  104. consumed := FrameWriteRequest{
  105. stream: wr.stream,
  106. write: &writeData{
  107. streamID: wd.streamID,
  108. p: wd.p[:allowed],
  109. // Even if the original had endStream set, there
  110. // are bytes remaining because len(wd.p) > allowed,
  111. // so we know endStream is false.
  112. endStream: false,
  113. },
  114. // Our caller is blocking on the final DATA frame, not
  115. // this intermediate frame, so no need to wait.
  116. done: nil,
  117. }
  118. rest := FrameWriteRequest{
  119. stream: wr.stream,
  120. write: &writeData{
  121. streamID: wd.streamID,
  122. p: wd.p[allowed:],
  123. endStream: wd.endStream,
  124. },
  125. done: wr.done,
  126. }
  127. return consumed, rest, 2
  128. }
  129. // The frame is consumed whole.
  130. // NB: This cast cannot overflow because allowed is <= math.MaxInt32.
  131. wr.stream.flow.take(int32(len(wd.p)))
  132. return wr, empty, 1
  133. }
  134. // String is for debugging only.
  135. func (wr FrameWriteRequest) String() string {
  136. var des string
  137. if s, ok := wr.write.(fmt.Stringer); ok {
  138. des = s.String()
  139. } else {
  140. des = fmt.Sprintf("%T", wr.write)
  141. }
  142. return fmt.Sprintf("[FrameWriteRequest stream=%d, ch=%v, writer=%v]", wr.StreamID(), wr.done != nil, des)
  143. }
  144. // replyToWriter sends err to wr.done and panics if the send must block
  145. // This does nothing if wr.done is nil.
  146. func (wr *FrameWriteRequest) replyToWriter(err error) {
  147. if wr.done == nil {
  148. return
  149. }
  150. select {
  151. case wr.done <- err:
  152. default:
  153. panic(fmt.Sprintf("unbuffered done channel passed in for type %T", wr.write))
  154. }
  155. wr.write = nil // prevent use (assume it's tainted after wr.done send)
  156. }
  157. // writeQueue is used by implementations of WriteScheduler.
  158. type writeQueue struct {
  159. s []FrameWriteRequest
  160. }
  161. func (q *writeQueue) empty() bool { return len(q.s) == 0 }
  162. func (q *writeQueue) push(wr FrameWriteRequest) {
  163. q.s = append(q.s, wr)
  164. }
  165. func (q *writeQueue) shift() FrameWriteRequest {
  166. if len(q.s) == 0 {
  167. panic("invalid use of queue")
  168. }
  169. wr := q.s[0]
  170. // TODO: less copy-happy queue.
  171. copy(q.s, q.s[1:])
  172. q.s[len(q.s)-1] = FrameWriteRequest{}
  173. q.s = q.s[:len(q.s)-1]
  174. return wr
  175. }
  176. // consume consumes up to n bytes from q.s[0]. If the frame is
  177. // entirely consumed, it is removed from the queue. If the frame
  178. // is partially consumed, the frame is kept with the consumed
  179. // bytes removed. Returns true iff any bytes were consumed.
  180. func (q *writeQueue) consume(n int32) (FrameWriteRequest, bool) {
  181. if len(q.s) == 0 {
  182. return FrameWriteRequest{}, false
  183. }
  184. consumed, rest, numresult := q.s[0].Consume(n)
  185. switch numresult {
  186. case 0:
  187. return FrameWriteRequest{}, false
  188. case 1:
  189. q.shift()
  190. case 2:
  191. q.s[0] = rest
  192. }
  193. return consumed, true
  194. }
  195. type writeQueuePool []*writeQueue
  196. // put inserts an unused writeQueue into the pool.
  197. func (p *writeQueuePool) put(q *writeQueue) {
  198. for i := range q.s {
  199. q.s[i] = FrameWriteRequest{}
  200. }
  201. q.s = q.s[:0]
  202. *p = append(*p, q)
  203. }
  204. // get returns an empty writeQueue.
  205. func (p *writeQueuePool) get() *writeQueue {
  206. ln := len(*p)
  207. if ln == 0 {
  208. return new(writeQueue)
  209. }
  210. x := ln - 1
  211. q := (*p)[x]
  212. (*p)[x] = nil
  213. *p = (*p)[:x]
  214. return q
  215. }