cloudflare/cloudflared
Publicmirrored from https://github.com/cloudflare/cloudflaredAvailable
h2mux/h2_dictionaries.go
593lines · modecode
| 1 | package h2mux |
| 2 | |
| 3 | import ( |
| 4 | "bytes" |
| 5 | "io" |
| 6 | "strings" |
| 7 | "sync" |
| 8 | |
| 9 | "golang.org/x/net/http2" |
| 10 | ) |
| 11 | |
| 12 | /* This is an implementation of https://github.com/vkrasnov/h2-compression-dictionaries |
| 13 | but modified for tunnels in a few key ways: |
| 14 | Since tunnels is a server-to-server service, some aspects of the spec would cause |
| 15 | unnessasary head-of-line blocking on the CPU and on the network, hence this implementation |
| 16 | allows for parallel compression on the "client", and buffering on the "server" to solve |
| 17 | this problem. */ |
| 18 | |
| 19 | // Assign temporary values |
| 20 | const SettingCompression http2.SettingID = 0xff20 |
| 21 | |
| 22 | const ( |
| 23 | FrameSetCompressionContext http2.FrameType = 0xf0 |
| 24 | FrameUseDictionary http2.FrameType = 0xf1 |
| 25 | FrameSetDictionary http2.FrameType = 0xf2 |
| 26 | ) |
| 27 | |
| 28 | const ( |
| 29 | FlagSetDictionaryAppend http2.Flags = 0x1 |
| 30 | FlagSetDictionaryOffset http2.Flags = 0x2 |
| 31 | ) |
| 32 | |
| 33 | const compressionVersion = uint8(1) |
| 34 | const compressionFormat = uint8(2) |
| 35 | |
| 36 | type CompressionSetting uint |
| 37 | |
| 38 | const ( |
| 39 | CompressionNone CompressionSetting = iota |
| 40 | CompressionLow |
| 41 | CompressionMedium |
| 42 | CompressionMax |
| 43 | ) |
| 44 | |
| 45 | type CompressionPreset struct { |
| 46 | nDicts, dictSize, quality uint8 |
| 47 | } |
| 48 | |
| 49 | type compressor interface { |
| 50 | Write([]byte) (int, error) |
| 51 | Flush() error |
| 52 | SetDictionary([]byte) |
| 53 | Close() error |
| 54 | } |
| 55 | |
| 56 | type decompressor interface { |
| 57 | Read([]byte) (int, error) |
| 58 | SetDictionary([]byte) |
| 59 | Close() error |
| 60 | } |
| 61 | |
| 62 | var compressionPresets = map[CompressionSetting]CompressionPreset{ |
| 63 | CompressionNone: {0, 0, 0}, |
| 64 | CompressionLow: {32, 17, 5}, |
| 65 | CompressionMedium: {64, 18, 6}, |
| 66 | CompressionMax: {255, 19, 9}, |
| 67 | } |
| 68 | |
| 69 | func compressionSettingVal(version, fmt, sz, nd uint8) uint32 { |
| 70 | // Currently the compression settings are inlcude: |
| 71 | // * version: only 1 is supported |
| 72 | // * fmt: only 2 for brotli is supported |
| 73 | // * sz: log2 of the maximal allowed dictionary size |
| 74 | // * nd: max allowed number of dictionaries |
| 75 | return uint32(version)<<24 + uint32(fmt)<<16 + uint32(sz)<<8 + uint32(nd) |
| 76 | } |
| 77 | |
| 78 | func parseCompressionSettingVal(setting uint32) (version, fmt, sz, nd uint8) { |
| 79 | version = uint8(setting >> 24) |
| 80 | fmt = uint8(setting >> 16) |
| 81 | sz = uint8(setting >> 8) |
| 82 | nd = uint8(setting) |
| 83 | return |
| 84 | } |
| 85 | |
| 86 | func (c CompressionSetting) toH2Setting() uint32 { |
| 87 | p, ok := compressionPresets[c] |
| 88 | if !ok { |
| 89 | return 0 |
| 90 | } |
| 91 | return compressionSettingVal(compressionVersion, compressionFormat, p.dictSize, p.nDicts) |
| 92 | } |
| 93 | |
| 94 | func (c CompressionSetting) getPreset() CompressionPreset { |
| 95 | return compressionPresets[c] |
| 96 | } |
| 97 | |
| 98 | type dictUpdate struct { |
| 99 | reader *h2DictionaryReader |
| 100 | dictionary *h2ReadDictionary |
| 101 | buff []byte |
| 102 | isReady bool |
| 103 | isUse bool |
| 104 | s setDictRequest |
| 105 | } |
| 106 | |
| 107 | type h2ReadDictionary struct { |
| 108 | dictionary []byte |
| 109 | queue []*dictUpdate |
| 110 | maxSize int |
| 111 | } |
| 112 | |
| 113 | type h2ReadDictionaries struct { |
| 114 | d []h2ReadDictionary |
| 115 | maxSize int |
| 116 | } |
| 117 | |
| 118 | type h2DictionaryReader struct { |
| 119 | *SharedBuffer // Propagate the decompressed output into the original buffer |
| 120 | decompBuffer *bytes.Buffer // Intermediate buffer for the brotli compressor |
| 121 | dictionary []byte // The content of the dictionary being used by this reader |
| 122 | internalBuffer []byte |
| 123 | s, e int // Start and end of the buffer |
| 124 | decomp decompressor // The brotli compressor |
| 125 | isClosed bool // Indicates that Close was called for this reader |
| 126 | queue []*dictUpdate // List of dictionaries to update, when the data is available |
| 127 | } |
| 128 | |
| 129 | type h2WriteDictionary []byte |
| 130 | |
| 131 | type setDictRequest struct { |
| 132 | streamID uint32 |
| 133 | dictID uint8 |
| 134 | dictSZ uint64 |
| 135 | truncate, offset uint64 |
| 136 | P, E, D bool |
| 137 | } |
| 138 | |
| 139 | type useDictRequest struct { |
| 140 | dictID uint8 |
| 141 | streamID uint32 |
| 142 | setDict []setDictRequest |
| 143 | } |
| 144 | |
| 145 | type h2WriteDictionaries struct { |
| 146 | dictLock sync.Mutex |
| 147 | dictChan chan useDictRequest |
| 148 | dictionaries []h2WriteDictionary |
| 149 | nextAvail int // next unused dictionary slot |
| 150 | maxAvail int // max ID, defined by SETTINGS |
| 151 | maxSize int // max size, defined by SETTINGS |
| 152 | typeToDict map[string]uint8 // map from content type to dictionary that encodes it |
| 153 | pathToDict map[string]uint8 // map from path to dictionary that encodes it |
| 154 | quality int |
| 155 | window int |
| 156 | compIn, compOut *AtomicCounter |
| 157 | } |
| 158 | |
| 159 | type h2DictWriter struct { |
| 160 | *bytes.Buffer |
| 161 | comp compressor |
| 162 | dicts *h2WriteDictionaries |
| 163 | writerLock sync.Mutex |
| 164 | |
| 165 | streamID uint32 |
| 166 | path string |
| 167 | contentType string |
| 168 | } |
| 169 | |
| 170 | type h2Dictionaries struct { |
| 171 | write *h2WriteDictionaries |
| 172 | read *h2ReadDictionaries |
| 173 | } |
| 174 | |
| 175 | func (o *dictUpdate) update(buff []byte) { |
| 176 | o.buff = make([]byte, len(buff)) |
| 177 | copy(o.buff, buff) |
| 178 | o.isReady = true |
| 179 | } |
| 180 | |
| 181 | func (d *h2ReadDictionary) update() { |
| 182 | for len(d.queue) > 0 { |
| 183 | o := d.queue[0] |
| 184 | if !o.isReady { |
| 185 | break |
| 186 | } |
| 187 | if o.isUse { |
| 188 | reader := o.reader |
| 189 | reader.dictionary = make([]byte, len(d.dictionary)) |
| 190 | copy(reader.dictionary, d.dictionary) |
| 191 | reader.decomp = newDecompressor(reader.decompBuffer) |
| 192 | if len(reader.dictionary) > 0 { |
| 193 | reader.decomp.SetDictionary(reader.dictionary) |
| 194 | } |
| 195 | reader.Write([]byte{}) |
| 196 | } else { |
| 197 | d.dictionary = adjustDictionary(d.dictionary, o.buff, o.s, d.maxSize) |
| 198 | } |
| 199 | d.queue = d.queue[1:] |
| 200 | } |
| 201 | } |
| 202 | |
| 203 | func newH2ReadDictionaries(nd, sz uint8) h2ReadDictionaries { |
| 204 | d := make([]h2ReadDictionary, int(nd)) |
| 205 | for i := range d { |
| 206 | d[i].maxSize = 1 << uint(sz) |
| 207 | } |
| 208 | return h2ReadDictionaries{d: d, maxSize: 1 << uint(sz)} |
| 209 | } |
| 210 | |
| 211 | func (dicts *h2ReadDictionaries) getDictByID(dictID uint8) (*h2ReadDictionary, error) { |
| 212 | if int(dictID) > len(dicts.d) { |
| 213 | return nil, MuxerStreamError{"dictID too big", http2.ErrCodeProtocol} |
| 214 | } |
| 215 | |
| 216 | return &dicts.d[dictID], nil |
| 217 | } |
| 218 | |
| 219 | func (dicts *h2ReadDictionaries) newReader(b *SharedBuffer, dictID uint8) *h2DictionaryReader { |
| 220 | if int(dictID) > len(dicts.d) { |
| 221 | return nil |
| 222 | } |
| 223 | |
| 224 | dictionary := &dicts.d[dictID] |
| 225 | reader := &h2DictionaryReader{SharedBuffer: b, decompBuffer: &bytes.Buffer{}, internalBuffer: make([]byte, dicts.maxSize)} |
| 226 | |
| 227 | if len(dictionary.queue) == 0 { |
| 228 | reader.dictionary = make([]byte, len(dictionary.dictionary)) |
| 229 | copy(reader.dictionary, dictionary.dictionary) |
| 230 | reader.decomp = newDecompressor(reader.decompBuffer) |
| 231 | if len(reader.dictionary) > 0 { |
| 232 | reader.decomp.SetDictionary(reader.dictionary) |
| 233 | } |
| 234 | } else { |
| 235 | dictionary.queue = append(dictionary.queue, &dictUpdate{isUse: true, isReady: true, reader: reader}) |
| 236 | } |
| 237 | return reader |
| 238 | } |
| 239 | |
| 240 | func (r *h2DictionaryReader) updateWaitingDictionaries() { |
| 241 | // Update all the waiting dictionaries |
| 242 | for _, o := range r.queue { |
| 243 | if o.isReady { |
| 244 | continue |
| 245 | } |
| 246 | if r.isClosed || uint64(r.e) >= o.s.dictSZ { |
| 247 | o.update(r.internalBuffer[:r.e]) |
| 248 | if o == o.dictionary.queue[0] { |
| 249 | defer o.dictionary.update() |
| 250 | } |
| 251 | } |
| 252 | } |
| 253 | } |
| 254 | |
| 255 | // Write actually happens when reading from network, this is therefore the stage where we decompress the buffer |
| 256 | func (r *h2DictionaryReader) Write(p []byte) (n int, err error) { |
| 257 | // Every write goes into brotli buffer first |
| 258 | n, err = r.decompBuffer.Write(p) |
| 259 | if err != nil { |
| 260 | return |
| 261 | } |
| 262 | |
| 263 | if r.decomp == nil { |
| 264 | return |
| 265 | } |
| 266 | |
| 267 | for { |
| 268 | m, err := r.decomp.Read(r.internalBuffer[r.e:]) |
| 269 | if err != nil && err != io.EOF { |
| 270 | r.SharedBuffer.Close() |
| 271 | r.decomp.Close() |
| 272 | return n, err |
| 273 | } |
| 274 | |
| 275 | r.SharedBuffer.Write(r.internalBuffer[r.e : r.e+m]) |
| 276 | r.e += m |
| 277 | |
| 278 | if m == 0 { |
| 279 | break |
| 280 | } |
| 281 | |
| 282 | if r.e == len(r.internalBuffer) { |
| 283 | r.updateWaitingDictionaries() |
| 284 | r.e = 0 |
| 285 | } |
| 286 | } |
| 287 | |
| 288 | r.updateWaitingDictionaries() |
| 289 | |
| 290 | if r.isClosed { |
| 291 | r.SharedBuffer.Close() |
| 292 | r.decomp.Close() |
| 293 | } |
| 294 | |
| 295 | return |
| 296 | } |
| 297 | |
| 298 | func (r *h2DictionaryReader) Close() error { |
| 299 | if r.isClosed { |
| 300 | return nil |
| 301 | } |
| 302 | r.isClosed = true |
| 303 | r.Write([]byte{}) |
| 304 | return nil |
| 305 | } |
| 306 | |
| 307 | var compressibleTypes = map[string]bool{ |
| 308 | "application/atom+xml": true, |
| 309 | "application/javascript": true, |
| 310 | "application/json": true, |
| 311 | "application/ld+json": true, |
| 312 | "application/manifest+json": true, |
| 313 | "application/rss+xml": true, |
| 314 | "application/vnd.geo+json": true, |
| 315 | "application/vnd.ms-fontobject": true, |
| 316 | "application/x-font-ttf": true, |
| 317 | "application/x-yaml": true, |
| 318 | "application/x-web-app-manifest+json": true, |
| 319 | "application/xhtml+xml": true, |
| 320 | "application/xml": true, |
| 321 | "font/opentype": true, |
| 322 | "image/bmp": true, |
| 323 | "image/svg+xml": true, |
| 324 | "image/x-icon": true, |
| 325 | "text/cache-manifest": true, |
| 326 | "text/css": true, |
| 327 | "text/html": true, |
| 328 | "text/plain": true, |
| 329 | "text/vcard": true, |
| 330 | "text/vnd.rim.location.xloc": true, |
| 331 | "text/vtt": true, |
| 332 | "text/x-component": true, |
| 333 | "text/x-cross-domain-policy": true, |
| 334 | "text/x-yaml": true, |
| 335 | } |
| 336 | |
| 337 | func getContentType(headers []Header) string { |
| 338 | for _, h := range headers { |
| 339 | if strings.ToLower(h.Name) == "content-type" { |
| 340 | val := strings.ToLower(h.Value) |
| 341 | sep := strings.IndexRune(val, ';') |
| 342 | if sep != -1 { |
| 343 | return val[:sep] |
| 344 | } |
| 345 | return val |
| 346 | } |
| 347 | } |
| 348 | |
| 349 | return "" |
| 350 | } |
| 351 | |
| 352 | func newH2WriteDictionaries(nd, sz, quality uint8, compIn, compOut *AtomicCounter) (*h2WriteDictionaries, chan useDictRequest) { |
| 353 | useDictChan := make(chan useDictRequest) |
| 354 | return &h2WriteDictionaries{ |
| 355 | dictionaries: make([]h2WriteDictionary, nd), |
| 356 | nextAvail: 0, |
| 357 | maxAvail: int(nd), |
| 358 | maxSize: 1 << uint(sz), |
| 359 | dictChan: useDictChan, |
| 360 | typeToDict: make(map[string]uint8), |
| 361 | pathToDict: make(map[string]uint8), |
| 362 | quality: int(quality), |
| 363 | window: 1 << uint(sz+1), |
| 364 | compIn: compIn, |
| 365 | compOut: compOut, |
| 366 | }, useDictChan |
| 367 | } |
| 368 | |
| 369 | func adjustDictionary(currentDictionary, newData []byte, set setDictRequest, maxSize int) []byte { |
| 370 | currentDictionary = append(currentDictionary, newData[:set.dictSZ]...) |
| 371 | |
| 372 | if len(currentDictionary) > maxSize { |
| 373 | currentDictionary = currentDictionary[len(currentDictionary)-maxSize:] |
| 374 | } |
| 375 | |
| 376 | return currentDictionary |
| 377 | } |
| 378 | |
| 379 | func (h2d *h2WriteDictionaries) getNextDictID() (dictID uint8, ok bool) { |
| 380 | if h2d.nextAvail < h2d.maxAvail { |
| 381 | dictID, ok = uint8(h2d.nextAvail), true |
| 382 | h2d.nextAvail++ |
| 383 | return |
| 384 | } |
| 385 | |
| 386 | return 0, false |
| 387 | } |
| 388 | |
| 389 | func (h2d *h2WriteDictionaries) getGenericDictID() (dictID uint8, ok bool) { |
| 390 | if h2d.maxAvail == 0 { |
| 391 | return 0, false |
| 392 | } |
| 393 | return uint8(h2d.maxAvail - 1), true |
| 394 | } |
| 395 | |
| 396 | func (h2d *h2WriteDictionaries) getDictWriter(s *MuxedStream, headers []Header) *h2DictWriter { |
| 397 | w := s.writeBuffer |
| 398 | |
| 399 | if w == nil { |
| 400 | return nil |
| 401 | } |
| 402 | |
| 403 | if s.method != "GET" && s.method != "POST" { |
| 404 | return nil |
| 405 | } |
| 406 | |
| 407 | s.contentType = getContentType(headers) |
| 408 | if _, ok := compressibleTypes[s.contentType]; !ok && !strings.HasPrefix(s.contentType, "text") { |
| 409 | return nil |
| 410 | } |
| 411 | |
| 412 | return &h2DictWriter{ |
| 413 | Buffer: w.(*bytes.Buffer), |
| 414 | path: s.path, |
| 415 | contentType: s.contentType, |
| 416 | streamID: s.streamID, |
| 417 | dicts: h2d, |
| 418 | } |
| 419 | } |
| 420 | |
| 421 | func assignDictToStream(s *MuxedStream, p []byte) bool { |
| 422 | |
| 423 | // On first write to stream: |
| 424 | // * assign the right dictionary |
| 425 | // * update relevant dictionaries |
| 426 | // * send the required USE_DICT and SET_DICT frames |
| 427 | |
| 428 | h2d := s.dictionaries.write |
| 429 | if h2d == nil { |
| 430 | return false |
| 431 | } |
| 432 | |
| 433 | w, ok := s.writeBuffer.(*h2DictWriter) |
| 434 | if !ok || w.comp != nil { |
| 435 | return false |
| 436 | } |
| 437 | |
| 438 | h2d.dictLock.Lock() |
| 439 | |
| 440 | if w.comp != nil { |
| 441 | // Check again with lock, in therory the inteface allows for unordered writes |
| 442 | h2d.dictLock.Unlock() |
| 443 | return false |
| 444 | } |
| 445 | |
| 446 | // The logic of dictionary generation is below |
| 447 | |
| 448 | // Is there a dictionary for the exact path or content-type? |
| 449 | var useID uint8 |
| 450 | pathID, pathFound := h2d.pathToDict[w.path] |
| 451 | typeID, typeFound := h2d.typeToDict[w.contentType] |
| 452 | |
| 453 | if pathFound { |
| 454 | // Use dictionary for path as top priority |
| 455 | useID = pathID |
| 456 | if !typeFound { // Shouldn't really happen, unless type changes between requests |
| 457 | typeID, typeFound = h2d.getNextDictID() |
| 458 | if typeFound { |
| 459 | h2d.typeToDict[w.contentType] = typeID |
| 460 | } |
| 461 | } |
| 462 | } else if typeFound { |
| 463 | // Use dictionary for same content type as second priority |
| 464 | useID = typeID |
| 465 | pathID, pathFound = h2d.getNextDictID() |
| 466 | if pathFound { // If a slot is available, generate new dictionary for path |
| 467 | h2d.pathToDict[w.path] = pathID |
| 468 | } |
| 469 | } else { |
| 470 | // Use the overflow dictionary as last resort |
| 471 | // If slots are availabe generate new dictioanries for path and content-type |
| 472 | useID, _ = h2d.getGenericDictID() |
| 473 | pathID, pathFound = h2d.getNextDictID() |
| 474 | if pathFound { |
| 475 | h2d.pathToDict[w.path] = pathID |
| 476 | } |
| 477 | typeID, typeFound = h2d.getNextDictID() |
| 478 | if typeFound { |
| 479 | h2d.typeToDict[w.contentType] = typeID |
| 480 | } |
| 481 | } |
| 482 | |
| 483 | useLen := h2d.maxSize |
| 484 | if len(p) < useLen { |
| 485 | useLen = len(p) |
| 486 | } |
| 487 | |
| 488 | // Update all the dictionaries using the new data |
| 489 | setDicts := make([]setDictRequest, 0, 3) |
| 490 | setDict := setDictRequest{ |
| 491 | streamID: w.streamID, |
| 492 | dictID: useID, |
| 493 | dictSZ: uint64(useLen), |
| 494 | } |
| 495 | setDicts = append(setDicts, setDict) |
| 496 | if pathID != useID { |
| 497 | setDict.dictID = pathID |
| 498 | setDicts = append(setDicts, setDict) |
| 499 | } |
| 500 | if typeID != useID { |
| 501 | setDict.dictID = typeID |
| 502 | setDicts = append(setDicts, setDict) |
| 503 | } |
| 504 | |
| 505 | h2d.dictChan <- useDictRequest{streamID: w.streamID, dictID: uint8(useID), setDict: setDicts} |
| 506 | |
| 507 | dict := h2d.dictionaries[useID] |
| 508 | |
| 509 | // Brolti requires the dictionary to be immutable |
| 510 | copyDict := make([]byte, len(dict)) |
| 511 | copy(copyDict, dict) |
| 512 | |
| 513 | for _, set := range setDicts { |
| 514 | h2d.dictionaries[set.dictID] = adjustDictionary(h2d.dictionaries[set.dictID], p, set, h2d.maxSize) |
| 515 | } |
| 516 | |
| 517 | w.comp = newCompressor(w.Buffer, h2d.quality, h2d.window) |
| 518 | |
| 519 | s.writeLock.Lock() |
| 520 | h2d.dictLock.Unlock() |
| 521 | |
| 522 | if len(copyDict) > 0 { |
| 523 | w.comp.SetDictionary(copyDict) |
| 524 | } |
| 525 | |
| 526 | return true |
| 527 | } |
| 528 | |
| 529 | func (w *h2DictWriter) Write(p []byte) (n int, err error) { |
| 530 | bufLen := w.Buffer.Len() |
| 531 | if w.comp != nil { |
| 532 | n, err = w.comp.Write(p) |
| 533 | if err != nil { |
| 534 | return |
| 535 | } |
| 536 | err = w.comp.Flush() |
| 537 | w.dicts.compIn.IncrementBy(uint64(n)) |
| 538 | w.dicts.compOut.IncrementBy(uint64(w.Buffer.Len() - bufLen)) |
| 539 | return |
| 540 | } |
| 541 | return w.Buffer.Write(p) |
| 542 | } |
| 543 | |
| 544 | func (w *h2DictWriter) Close() error { |
| 545 | return w.comp.Close() |
| 546 | } |
| 547 | |
| 548 | // From http2/hpack |
| 549 | func http2ReadVarInt(n byte, p []byte) (remain []byte, v uint64, err error) { |
| 550 | if n < 1 || n > 8 { |
| 551 | panic("bad n") |
| 552 | } |
| 553 | if len(p) == 0 { |
| 554 | return nil, 0, MuxerStreamError{"unexpected EOF", http2.ErrCodeProtocol} |
| 555 | } |
| 556 | v = uint64(p[0]) |
| 557 | if n < 8 { |
| 558 | v &= (1 << uint64(n)) - 1 |
| 559 | } |
| 560 | if v < (1<<uint64(n))-1 { |
| 561 | return p[1:], v, nil |
| 562 | } |
| 563 | |
| 564 | origP := p |
| 565 | p = p[1:] |
| 566 | var m uint64 |
| 567 | for len(p) > 0 { |
| 568 | b := p[0] |
| 569 | p = p[1:] |
| 570 | v += uint64(b&127) << m |
| 571 | if b&128 == 0 { |
| 572 | return p, v, nil |
| 573 | } |
| 574 | m += 7 |
| 575 | if m >= 63 { |
| 576 | return origP, 0, MuxerStreamError{"invalid integer", http2.ErrCodeProtocol} |
| 577 | } |
| 578 | } |
| 579 | return nil, 0, MuxerStreamError{"unexpected EOF", http2.ErrCodeProtocol} |
| 580 | } |
| 581 | |
| 582 | func appendVarInt(dst []byte, n byte, i uint64) []byte { |
| 583 | k := uint64((1 << n) - 1) |
| 584 | if i < k { |
| 585 | return append(dst, byte(i)) |
| 586 | } |
| 587 | dst = append(dst, byte(k)) |
| 588 | i -= k |
| 589 | for ; i >= 128; i >>= 7 { |
| 590 | dst = append(dst, byte(0x80|(i&0x7f))) |
| 591 | } |
| 592 | return append(dst, byte(i)) |
| 593 | } |