cloudflare/cloudflared
Publicmirrored from https://github.com/cloudflare/cloudflaredAvailable
h2mux/muxwriter.go
287lines · modecode
| 1 | package h2mux |
| 2 | |
| 3 | import ( |
| 4 | "bytes" |
| 5 | "encoding/binary" |
| 6 | "io" |
| 7 | "time" |
| 8 | |
| 9 | log "github.com/sirupsen/logrus" |
| 10 | "golang.org/x/net/http2" |
| 11 | "golang.org/x/net/http2/hpack" |
| 12 | ) |
| 13 | |
| 14 | type MuxWriter struct { |
| 15 | // f is used to write HTTP2 frames. |
| 16 | f *http2.Framer |
| 17 | // streams tracks currently-open streams. |
| 18 | streams *activeStreamMap |
| 19 | // streamErrors receives stream errors raised by the MuxReader. |
| 20 | streamErrors *StreamErrorMap |
| 21 | // readyStreamChan is used to multiplex writable streams onto the single connection. |
| 22 | // When a stream becomes writable its ID is sent on this channel. |
| 23 | readyStreamChan <-chan uint32 |
| 24 | // newStreamChan is used to create new streams with a given set of headers. |
| 25 | newStreamChan <-chan MuxedStreamRequest |
| 26 | // goAwayChan is used to send a single GOAWAY message to the peer. The element received |
| 27 | // is the HTTP/2 error code to send. |
| 28 | goAwayChan <-chan http2.ErrCode |
| 29 | // abortChan is used when shutting down ungracefully. When this becomes readable, all activity should stop. |
| 30 | abortChan <-chan struct{} |
| 31 | // pingTimestamp is an atomic value containing the latest received ping timestamp. |
| 32 | pingTimestamp *PingTimestamp |
| 33 | // A timer used to measure idle connection time. Reset after sending data. |
| 34 | idleTimer *IdleTimer |
| 35 | // connActiveChan receives a signal that the connection received some (read) activity. |
| 36 | connActiveChan <-chan struct{} |
| 37 | // Maximum size of all frames that can be sent on this connection. |
| 38 | maxFrameSize uint32 |
| 39 | // headerEncoder is the stateful header encoder for this connection |
| 40 | headerEncoder *hpack.Encoder |
| 41 | // headerBuffer is the temporary buffer used by headerEncoder. |
| 42 | headerBuffer bytes.Buffer |
| 43 | // updateReceiveWindowChan is the channel to update receiveWindow size to muxerMetricsUpdater |
| 44 | updateReceiveWindowChan chan<- uint32 |
| 45 | // updateSendWindowChan is the channel to update sendWindow size to muxerMetricsUpdater |
| 46 | updateSendWindowChan chan<- uint32 |
| 47 | // bytesWrote is the amount of bytes wrote to data frame since the last time we send bytes wrote to metrics |
| 48 | bytesWrote *AtomicCounter |
| 49 | // updateOutBoundBytesChan is the channel to send bytesWrote to muxerMetricsUpdater |
| 50 | updateOutBoundBytesChan chan<- uint64 |
| 51 | |
| 52 | useDictChan <-chan useDictRequest |
| 53 | } |
| 54 | |
| 55 | type MuxedStreamRequest struct { |
| 56 | stream *MuxedStream |
| 57 | body io.Reader |
| 58 | } |
| 59 | |
| 60 | func (r *MuxedStreamRequest) flushBody() { |
| 61 | io.Copy(r.stream, r.body) |
| 62 | r.stream.CloseWrite() |
| 63 | } |
| 64 | |
| 65 | func tsToPingData(ts int64) [8]byte { |
| 66 | pingData := [8]byte{} |
| 67 | binary.LittleEndian.PutUint64(pingData[:], uint64(ts)) |
| 68 | return pingData |
| 69 | } |
| 70 | |
| 71 | func (w *MuxWriter) run(parentLogger *log.Entry) error { |
| 72 | logger := parentLogger.WithFields(log.Fields{ |
| 73 | "subsystem": "mux", |
| 74 | "dir": "write", |
| 75 | }) |
| 76 | defer logger.Debug("event loop finished") |
| 77 | |
| 78 | // routine to periodically communicate bytesWrote |
| 79 | go func() { |
| 80 | tickC := time.Tick(updateFreq) |
| 81 | for { |
| 82 | select { |
| 83 | case <-w.abortChan: |
| 84 | return |
| 85 | case <-tickC: |
| 86 | w.updateOutBoundBytesChan <- w.bytesWrote.Count() |
| 87 | } |
| 88 | } |
| 89 | }() |
| 90 | |
| 91 | for { |
| 92 | select { |
| 93 | case <-w.abortChan: |
| 94 | logger.Debug("aborting writer thread") |
| 95 | return nil |
| 96 | case errCode := <-w.goAwayChan: |
| 97 | logger.Debug("sending GOAWAY code ", errCode) |
| 98 | err := w.f.WriteGoAway(w.streams.LastPeerStreamID(), errCode, []byte{}) |
| 99 | if err != nil { |
| 100 | return err |
| 101 | } |
| 102 | w.idleTimer.MarkActive() |
| 103 | case <-w.pingTimestamp.GetUpdateChan(): |
| 104 | logger.Debug("sending PING ACK") |
| 105 | err := w.f.WritePing(true, tsToPingData(w.pingTimestamp.Get())) |
| 106 | if err != nil { |
| 107 | return err |
| 108 | } |
| 109 | w.idleTimer.MarkActive() |
| 110 | case <-w.idleTimer.C: |
| 111 | if !w.idleTimer.Retry() { |
| 112 | return ErrConnectionDropped |
| 113 | } |
| 114 | logger.Debug("sending PING") |
| 115 | err := w.f.WritePing(false, tsToPingData(time.Now().UnixNano())) |
| 116 | if err != nil { |
| 117 | return err |
| 118 | } |
| 119 | w.idleTimer.ResetTimer() |
| 120 | case <-w.connActiveChan: |
| 121 | w.idleTimer.MarkActive() |
| 122 | case <-w.streamErrors.GetSignalChan(): |
| 123 | for streamID, errCode := range w.streamErrors.GetErrors() { |
| 124 | logger.WithField("stream", streamID).WithField("code", errCode).Debug("resetting stream") |
| 125 | err := w.f.WriteRSTStream(streamID, errCode) |
| 126 | if err != nil { |
| 127 | return err |
| 128 | } |
| 129 | } |
| 130 | w.idleTimer.MarkActive() |
| 131 | case streamRequest := <-w.newStreamChan: |
| 132 | streamID := w.streams.AcquireLocalID() |
| 133 | streamRequest.stream.streamID = streamID |
| 134 | if !w.streams.Set(streamRequest.stream) { |
| 135 | // Race between OpenStream and Shutdown, and Shutdown won. Let Shutdown (and the eventual abort) take |
| 136 | // care of this stream. Ideally we'd pass the error directly to the stream object somehow so the |
| 137 | // caller can be unblocked sooner, but the value of that optimisation is minimal for most of the |
| 138 | // reasons why you'd call Shutdown anyway. |
| 139 | continue |
| 140 | } |
| 141 | if streamRequest.body != nil { |
| 142 | go streamRequest.flushBody() |
| 143 | } |
| 144 | streamLogger := logger.WithField("stream", streamID) |
| 145 | err := w.writeStreamData(streamRequest.stream, streamLogger) |
| 146 | if err != nil { |
| 147 | return err |
| 148 | } |
| 149 | w.idleTimer.MarkActive() |
| 150 | case streamID := <-w.readyStreamChan: |
| 151 | streamLogger := logger.WithField("stream", streamID) |
| 152 | stream, ok := w.streams.Get(streamID) |
| 153 | if !ok { |
| 154 | continue |
| 155 | } |
| 156 | err := w.writeStreamData(stream, streamLogger) |
| 157 | if err != nil { |
| 158 | return err |
| 159 | } |
| 160 | w.idleTimer.MarkActive() |
| 161 | case useDict := <-w.useDictChan: |
| 162 | err := w.writeUseDictionary(useDict) |
| 163 | if err != nil { |
| 164 | logger.WithError(err).Warn("error writing use dictionary") |
| 165 | return err |
| 166 | } |
| 167 | w.idleTimer.MarkActive() |
| 168 | } |
| 169 | } |
| 170 | } |
| 171 | |
| 172 | func (w *MuxWriter) writeStreamData(stream *MuxedStream, logger *log.Entry) error { |
| 173 | logger.Debug("writable") |
| 174 | chunk := stream.getChunk() |
| 175 | w.updateReceiveWindowChan <- stream.getReceiveWindow() |
| 176 | w.updateSendWindowChan <- stream.getSendWindow() |
| 177 | if chunk.sendHeadersFrame() { |
| 178 | err := w.writeHeaders(chunk.streamID, chunk.headers) |
| 179 | if err != nil { |
| 180 | logger.WithError(err).Warn("error writing headers") |
| 181 | return err |
| 182 | } |
| 183 | logger.Debug("output headers") |
| 184 | } |
| 185 | |
| 186 | if chunk.sendWindowUpdateFrame() { |
| 187 | // Send a WINDOW_UPDATE frame to update our receive window. |
| 188 | // If the Stream ID is zero, the window update applies to the connection as a whole |
| 189 | // RFC7540 section-6.9.1 "A receiver that receives a flow-controlled frame MUST |
| 190 | // always account for its contribution against the connection flow-control |
| 191 | // window, unless the receiver treats this as a connection error" |
| 192 | err := w.f.WriteWindowUpdate(chunk.streamID, chunk.windowUpdate) |
| 193 | if err != nil { |
| 194 | logger.WithError(err).Warn("error writing window update") |
| 195 | return err |
| 196 | } |
| 197 | logger.Debugf("increment receive window by %d", chunk.windowUpdate) |
| 198 | } |
| 199 | |
| 200 | for chunk.sendDataFrame() { |
| 201 | payload, sentEOF := chunk.nextDataFrame(int(w.maxFrameSize)) |
| 202 | err := w.f.WriteData(chunk.streamID, sentEOF, payload) |
| 203 | if err != nil { |
| 204 | logger.WithError(err).Warn("error writing data") |
| 205 | return err |
| 206 | } |
| 207 | // update the amount of data wrote |
| 208 | w.bytesWrote.IncrementBy(uint64(len(payload))) |
| 209 | logger.WithField("len", len(payload)).Debug("output data") |
| 210 | |
| 211 | if sentEOF { |
| 212 | if stream.readBuffer.Closed() { |
| 213 | // transition into closed state |
| 214 | if !stream.gotReceiveEOF() { |
| 215 | // the peer may send data that we no longer want to receive. Force them into the |
| 216 | // closed state. |
| 217 | logger.Debug("resetting stream") |
| 218 | w.f.WriteRSTStream(chunk.streamID, http2.ErrCodeNo) |
| 219 | } else { |
| 220 | // Half-open stream transitioned into closed |
| 221 | logger.Debug("closing stream") |
| 222 | } |
| 223 | w.streams.Delete(chunk.streamID) |
| 224 | } else { |
| 225 | logger.Debug("closing stream write side") |
| 226 | } |
| 227 | } |
| 228 | } |
| 229 | return nil |
| 230 | } |
| 231 | |
| 232 | func (w *MuxWriter) encodeHeaders(headers []Header) ([]byte, error) { |
| 233 | w.headerBuffer.Reset() |
| 234 | for _, header := range headers { |
| 235 | err := w.headerEncoder.WriteField(hpack.HeaderField{ |
| 236 | Name: header.Name, |
| 237 | Value: header.Value, |
| 238 | }) |
| 239 | if err != nil { |
| 240 | return nil, err |
| 241 | } |
| 242 | } |
| 243 | return w.headerBuffer.Bytes(), nil |
| 244 | } |
| 245 | |
| 246 | // writeHeaders writes a block of encoded headers, splitting it into multiple frames if necessary. |
| 247 | func (w *MuxWriter) writeHeaders(streamID uint32, headers []Header) error { |
| 248 | encodedHeaders, err := w.encodeHeaders(headers) |
| 249 | if err != nil { |
| 250 | return err |
| 251 | } |
| 252 | blockSize := int(w.maxFrameSize) |
| 253 | endHeaders := len(encodedHeaders) == 0 |
| 254 | for !endHeaders && err == nil { |
| 255 | blockFragment := encodedHeaders |
| 256 | if len(encodedHeaders) > blockSize { |
| 257 | blockFragment = blockFragment[:blockSize] |
| 258 | encodedHeaders = encodedHeaders[blockSize:] |
| 259 | // Send CONTINUATION frame if the headers can't be fit into 1 frame |
| 260 | err = w.f.WriteContinuation(streamID, endHeaders, blockFragment) |
| 261 | } else { |
| 262 | endHeaders = true |
| 263 | err = w.f.WriteHeaders(http2.HeadersFrameParam{ |
| 264 | StreamID: streamID, |
| 265 | EndHeaders: endHeaders, |
| 266 | BlockFragment: blockFragment, |
| 267 | }) |
| 268 | } |
| 269 | } |
| 270 | return err |
| 271 | } |
| 272 | |
| 273 | func (w *MuxWriter) writeUseDictionary(dictRequest useDictRequest) error { |
| 274 | err := w.f.WriteRawFrame(FrameUseDictionary, 0, dictRequest.streamID, []byte{byte(dictRequest.dictID)}) |
| 275 | if err != nil { |
| 276 | return err |
| 277 | } |
| 278 | payload := make([]byte, 0, 64) |
| 279 | for _, set := range dictRequest.setDict { |
| 280 | payload = append(payload, byte(set.dictID)) |
| 281 | payload = appendVarInt(payload, 7, uint64(set.dictSZ)) |
| 282 | payload = append(payload, 0x80) // E = 1, D = 0, Truncate = 0 |
| 283 | } |
| 284 | |
| 285 | err = w.f.WriteRawFrame(FrameSetDictionary, FlagSetDictionaryAppend, dictRequest.streamID, payload) |
| 286 | return err |
| 287 | } |
| 288 | |