1/* 2 * 3 * Copyright 2014 gRPC authors. 4 * 5 * Licensed under the Apache License, Version 2.0 (the "License"); 6 * you may not use this file except in compliance with the License. 7 * You may obtain a copy of the License at 8 * 9 * http://www.apache.org/licenses/LICENSE-2.0 10 * 11 * Unless required by applicable law or agreed to in writing, software 12 * distributed under the License is distributed on an "AS IS" BASIS, 13 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 14 * See the License for the specific language governing permissions and 15 * limitations under the License. 16 * 17 */ 18 19package transport 20 21import ( 22 "bytes" 23 "errors" 24 "fmt" 25 "runtime" 26 "strconv" 27 "sync" 28 "sync/atomic" 29 30 "golang.org/x/net/http2" 31 "golang.org/x/net/http2/hpack" 32 "google.golang.org/grpc/internal/grpcutil" 33 "google.golang.org/grpc/status" 34) 35 36var updateHeaderTblSize = func(e *hpack.Encoder, v uint32) { 37 e.SetMaxDynamicTableSizeLimit(v) 38} 39 40type itemNode struct { 41 it interface{} 42 next *itemNode 43} 44 45type itemList struct { 46 head *itemNode 47 tail *itemNode 48} 49 50func (il *itemList) enqueue(i interface{}) { 51 n := &itemNode{it: i} 52 if il.tail == nil { 53 il.head, il.tail = n, n 54 return 55 } 56 il.tail.next = n 57 il.tail = n 58} 59 60// peek returns the first item in the list without removing it from the 61// list. 62func (il *itemList) peek() interface{} { 63 return il.head.it 64} 65 66func (il *itemList) dequeue() interface{} { 67 if il.head == nil { 68 return nil 69 } 70 i := il.head.it 71 il.head = il.head.next 72 if il.head == nil { 73 il.tail = nil 74 } 75 return i 76} 77 78func (il *itemList) dequeueAll() *itemNode { 79 h := il.head 80 il.head, il.tail = nil, nil 81 return h 82} 83 84func (il *itemList) isEmpty() bool { 85 return il.head == nil 86} 87 88// The following defines various control items which could flow through 89// the control buffer of transport. They represent different aspects of 90// control tasks, e.g., flow control, settings, streaming resetting, etc. 91 92// maxQueuedTransportResponseFrames is the most queued "transport response" 93// frames we will buffer before preventing new reads from occurring on the 94// transport. These are control frames sent in response to client requests, 95// such as RST_STREAM due to bad headers or settings acks. 96const maxQueuedTransportResponseFrames = 50 97 98type cbItem interface { 99 isTransportResponseFrame() bool 100} 101 102// registerStream is used to register an incoming stream with loopy writer. 103type registerStream struct { 104 streamID uint32 105 wq *writeQuota 106} 107 108func (*registerStream) isTransportResponseFrame() bool { return false } 109 110// headerFrame is also used to register stream on the client-side. 111type headerFrame struct { 112 streamID uint32 113 hf []hpack.HeaderField 114 endStream bool // Valid on server side. 115 initStream func(uint32) error // Used only on the client side. 116 onWrite func() 117 wq *writeQuota // write quota for the stream created. 118 cleanup *cleanupStream // Valid on the server side. 119 onOrphaned func(error) // Valid on client-side 120} 121 122func (h *headerFrame) isTransportResponseFrame() bool { 123 return h.cleanup != nil && h.cleanup.rst // Results in a RST_STREAM 124} 125 126type cleanupStream struct { 127 streamID uint32 128 rst bool 129 rstCode http2.ErrCode 130 onWrite func() 131} 132 133func (c *cleanupStream) isTransportResponseFrame() bool { return c.rst } // Results in a RST_STREAM 134 135type earlyAbortStream struct { 136 streamID uint32 137 contentSubtype string 138 status *status.Status 139} 140 141func (*earlyAbortStream) isTransportResponseFrame() bool { return false } 142 143type dataFrame struct { 144 streamID uint32 145 endStream bool 146 h []byte 147 d []byte 148 // onEachWrite is called every time 149 // a part of d is written out. 150 onEachWrite func() 151} 152 153func (*dataFrame) isTransportResponseFrame() bool { return false } 154 155type incomingWindowUpdate struct { 156 streamID uint32 157 increment uint32 158} 159 160func (*incomingWindowUpdate) isTransportResponseFrame() bool { return false } 161 162type outgoingWindowUpdate struct { 163 streamID uint32 164 increment uint32 165} 166 167func (*outgoingWindowUpdate) isTransportResponseFrame() bool { 168 return false // window updates are throttled by thresholds 169} 170 171type incomingSettings struct { 172 ss []http2.Setting 173} 174 175func (*incomingSettings) isTransportResponseFrame() bool { return true } // Results in a settings ACK 176 177type outgoingSettings struct { 178 ss []http2.Setting 179} 180 181func (*outgoingSettings) isTransportResponseFrame() bool { return false } 182 183type incomingGoAway struct { 184} 185 186func (*incomingGoAway) isTransportResponseFrame() bool { return false } 187 188type goAway struct { 189 code http2.ErrCode 190 debugData []byte 191 headsUp bool 192 closeConn bool 193} 194 195func (*goAway) isTransportResponseFrame() bool { return false } 196 197type ping struct { 198 ack bool 199 data [8]byte 200} 201 202func (*ping) isTransportResponseFrame() bool { return true } 203 204type outFlowControlSizeRequest struct { 205 resp chan uint32 206} 207 208func (*outFlowControlSizeRequest) isTransportResponseFrame() bool { return false } 209 210type outStreamState int 211 212const ( 213 active outStreamState = iota 214 empty 215 waitingOnStreamQuota 216) 217 218type outStream struct { 219 id uint32 220 state outStreamState 221 itl *itemList 222 bytesOutStanding int 223 wq *writeQuota 224 225 next *outStream 226 prev *outStream 227} 228 229func (s *outStream) deleteSelf() { 230 if s.prev != nil { 231 s.prev.next = s.next 232 } 233 if s.next != nil { 234 s.next.prev = s.prev 235 } 236 s.next, s.prev = nil, nil 237} 238 239type outStreamList struct { 240 // Following are sentinel objects that mark the 241 // beginning and end of the list. They do not 242 // contain any item lists. All valid objects are 243 // inserted in between them. 244 // This is needed so that an outStream object can 245 // deleteSelf() in O(1) time without knowing which 246 // list it belongs to. 247 head *outStream 248 tail *outStream 249} 250 251func newOutStreamList() *outStreamList { 252 head, tail := new(outStream), new(outStream) 253 head.next = tail 254 tail.prev = head 255 return &outStreamList{ 256 head: head, 257 tail: tail, 258 } 259} 260 261func (l *outStreamList) enqueue(s *outStream) { 262 e := l.tail.prev 263 e.next = s 264 s.prev = e 265 s.next = l.tail 266 l.tail.prev = s 267} 268 269// remove from the beginning of the list. 270func (l *outStreamList) dequeue() *outStream { 271 b := l.head.next 272 if b == l.tail { 273 return nil 274 } 275 b.deleteSelf() 276 return b 277} 278 279// controlBuffer is a way to pass information to loopy. 280// Information is passed as specific struct types called control frames. 281// A control frame not only represents data, messages or headers to be sent out 282// but can also be used to instruct loopy to update its internal state. 283// It shouldn't be confused with an HTTP2 frame, although some of the control frames 284// like dataFrame and headerFrame do go out on wire as HTTP2 frames. 285type controlBuffer struct { 286 ch chan struct{} 287 done <-chan struct{} 288 mu sync.Mutex 289 consumerWaiting bool 290 list *itemList 291 err error 292 293 // transportResponseFrames counts the number of queued items that represent 294 // the response of an action initiated by the peer. trfChan is created 295 // when transportResponseFrames >= maxQueuedTransportResponseFrames and is 296 // closed and nilled when transportResponseFrames drops below the 297 // threshold. Both fields are protected by mu. 298 transportResponseFrames int 299 trfChan atomic.Value // *chan struct{} 300} 301 302func newControlBuffer(done <-chan struct{}) *controlBuffer { 303 return &controlBuffer{ 304 ch: make(chan struct{}, 1), 305 list: &itemList{}, 306 done: done, 307 } 308} 309 310// throttle blocks if there are too many incomingSettings/cleanupStreams in the 311// controlbuf. 312func (c *controlBuffer) throttle() { 313 ch, _ := c.trfChan.Load().(*chan struct{}) 314 if ch != nil { 315 select { 316 case <-*ch: 317 case <-c.done: 318 } 319 } 320} 321 322func (c *controlBuffer) put(it cbItem) error { 323 _, err := c.executeAndPut(nil, it) 324 return err 325} 326 327func (c *controlBuffer) executeAndPut(f func(it interface{}) bool, it cbItem) (bool, error) { 328 var wakeUp bool 329 c.mu.Lock() 330 if c.err != nil { 331 c.mu.Unlock() 332 return false, c.err 333 } 334 if f != nil { 335 if !f(it) { // f wasn't successful 336 c.mu.Unlock() 337 return false, nil 338 } 339 } 340 if c.consumerWaiting { 341 wakeUp = true 342 c.consumerWaiting = false 343 } 344 c.list.enqueue(it) 345 if it.isTransportResponseFrame() { 346 c.transportResponseFrames++ 347 if c.transportResponseFrames == maxQueuedTransportResponseFrames { 348 // We are adding the frame that puts us over the threshold; create 349 // a throttling channel. 350 ch := make(chan struct{}) 351 c.trfChan.Store(&ch) 352 } 353 } 354 c.mu.Unlock() 355 if wakeUp { 356 select { 357 case c.ch <- struct{}{}: 358 default: 359 } 360 } 361 return true, nil 362} 363 364// Note argument f should never be nil. 365func (c *controlBuffer) execute(f func(it interface{}) bool, it interface{}) (bool, error) { 366 c.mu.Lock() 367 if c.err != nil { 368 c.mu.Unlock() 369 return false, c.err 370 } 371 if !f(it) { // f wasn't successful 372 c.mu.Unlock() 373 return false, nil 374 } 375 c.mu.Unlock() 376 return true, nil 377} 378 379func (c *controlBuffer) get(block bool) (interface{}, error) { 380 for { 381 c.mu.Lock() 382 if c.err != nil { 383 c.mu.Unlock() 384 return nil, c.err 385 } 386 if !c.list.isEmpty() { 387 h := c.list.dequeue().(cbItem) 388 if h.isTransportResponseFrame() { 389 if c.transportResponseFrames == maxQueuedTransportResponseFrames { 390 // We are removing the frame that put us over the 391 // threshold; close and clear the throttling channel. 392 ch := c.trfChan.Load().(*chan struct{}) 393 close(*ch) 394 c.trfChan.Store((*chan struct{})(nil)) 395 } 396 c.transportResponseFrames-- 397 } 398 c.mu.Unlock() 399 return h, nil 400 } 401 if !block { 402 c.mu.Unlock() 403 return nil, nil 404 } 405 c.consumerWaiting = true 406 c.mu.Unlock() 407 select { 408 case <-c.ch: 409 case <-c.done: 410 c.finish() 411 return nil, ErrConnClosing 412 } 413 } 414} 415 416func (c *controlBuffer) finish() { 417 c.mu.Lock() 418 if c.err != nil { 419 c.mu.Unlock() 420 return 421 } 422 c.err = ErrConnClosing 423 // There may be headers for streams in the control buffer. 424 // These streams need to be cleaned out since the transport 425 // is still not aware of these yet. 426 for head := c.list.dequeueAll(); head != nil; head = head.next { 427 hdr, ok := head.it.(*headerFrame) 428 if !ok { 429 continue 430 } 431 if hdr.onOrphaned != nil { // It will be nil on the server-side. 432 hdr.onOrphaned(ErrConnClosing) 433 } 434 } 435 c.mu.Unlock() 436} 437 438type side int 439 440const ( 441 clientSide side = iota 442 serverSide 443) 444 445// Loopy receives frames from the control buffer. 446// Each frame is handled individually; most of the work done by loopy goes 447// into handling data frames. Loopy maintains a queue of active streams, and each 448// stream maintains a queue of data frames; as loopy receives data frames 449// it gets added to the queue of the relevant stream. 450// Loopy goes over this list of active streams by processing one node every iteration, 451// thereby closely resemebling to a round-robin scheduling over all streams. While 452// processing a stream, loopy writes out data bytes from this stream capped by the min 453// of http2MaxFrameLen, connection-level flow control and stream-level flow control. 454type loopyWriter struct { 455 side side 456 cbuf *controlBuffer 457 sendQuota uint32 458 oiws uint32 // outbound initial window size. 459 // estdStreams is map of all established streams that are not cleaned-up yet. 460 // On client-side, this is all streams whose headers were sent out. 461 // On server-side, this is all streams whose headers were received. 462 estdStreams map[uint32]*outStream // Established streams. 463 // activeStreams is a linked-list of all streams that have data to send and some 464 // stream-level flow control quota. 465 // Each of these streams internally have a list of data items(and perhaps trailers 466 // on the server-side) to be sent out. 467 activeStreams *outStreamList 468 framer *framer 469 hBuf *bytes.Buffer // The buffer for HPACK encoding. 470 hEnc *hpack.Encoder // HPACK encoder. 471 bdpEst *bdpEstimator 472 draining bool 473 474 // Side-specific handlers 475 ssGoAwayHandler func(*goAway) (bool, error) 476} 477 478func newLoopyWriter(s side, fr *framer, cbuf *controlBuffer, bdpEst *bdpEstimator) *loopyWriter { 479 var buf bytes.Buffer 480 l := &loopyWriter{ 481 side: s, 482 cbuf: cbuf, 483 sendQuota: defaultWindowSize, 484 oiws: defaultWindowSize, 485 estdStreams: make(map[uint32]*outStream), 486 activeStreams: newOutStreamList(), 487 framer: fr, 488 hBuf: &buf, 489 hEnc: hpack.NewEncoder(&buf), 490 bdpEst: bdpEst, 491 } 492 return l 493} 494 495const minBatchSize = 1000 496 497// run should be run in a separate goroutine. 498// It reads control frames from controlBuf and processes them by: 499// 1. Updating loopy's internal state, or/and 500// 2. Writing out HTTP2 frames on the wire. 501// 502// Loopy keeps all active streams with data to send in a linked-list. 503// All streams in the activeStreams linked-list must have both: 504// 1. Data to send, and 505// 2. Stream level flow control quota available. 506// 507// In each iteration of run loop, other than processing the incoming control 508// frame, loopy calls processData, which processes one node from the activeStreams linked-list. 509// This results in writing of HTTP2 frames into an underlying write buffer. 510// When there's no more control frames to read from controlBuf, loopy flushes the write buffer. 511// As an optimization, to increase the batch size for each flush, loopy yields the processor, once 512// if the batch size is too low to give stream goroutines a chance to fill it up. 513func (l *loopyWriter) run() (err error) { 514 defer func() { 515 if err == ErrConnClosing { 516 // Don't log ErrConnClosing as error since it happens 517 // 1. When the connection is closed by some other known issue. 518 // 2. User closed the connection. 519 // 3. A graceful close of connection. 520 if logger.V(logLevel) { 521 logger.Infof("transport: loopyWriter.run returning. %v", err) 522 } 523 err = nil 524 } 525 }() 526 for { 527 it, err := l.cbuf.get(true) 528 if err != nil { 529 return err 530 } 531 if err = l.handle(it); err != nil { 532 return err 533 } 534 if _, err = l.processData(); err != nil { 535 return err 536 } 537 gosched := true 538 hasdata: 539 for { 540 it, err := l.cbuf.get(false) 541 if err != nil { 542 return err 543 } 544 if it != nil { 545 if err = l.handle(it); err != nil { 546 return err 547 } 548 if _, err = l.processData(); err != nil { 549 return err 550 } 551 continue hasdata 552 } 553 isEmpty, err := l.processData() 554 if err != nil { 555 return err 556 } 557 if !isEmpty { 558 continue hasdata 559 } 560 if gosched { 561 gosched = false 562 if l.framer.writer.offset < minBatchSize { 563 runtime.Gosched() 564 continue hasdata 565 } 566 } 567 l.framer.writer.Flush() 568 break hasdata 569 570 } 571 } 572} 573 574func (l *loopyWriter) outgoingWindowUpdateHandler(w *outgoingWindowUpdate) error { 575 return l.framer.fr.WriteWindowUpdate(w.streamID, w.increment) 576} 577 578func (l *loopyWriter) incomingWindowUpdateHandler(w *incomingWindowUpdate) error { 579 // Otherwise update the quota. 580 if w.streamID == 0 { 581 l.sendQuota += w.increment 582 return nil 583 } 584 // Find the stream and update it. 585 if str, ok := l.estdStreams[w.streamID]; ok { 586 str.bytesOutStanding -= int(w.increment) 587 if strQuota := int(l.oiws) - str.bytesOutStanding; strQuota > 0 && str.state == waitingOnStreamQuota { 588 str.state = active 589 l.activeStreams.enqueue(str) 590 return nil 591 } 592 } 593 return nil 594} 595 596func (l *loopyWriter) outgoingSettingsHandler(s *outgoingSettings) error { 597 return l.framer.fr.WriteSettings(s.ss...) 598} 599 600func (l *loopyWriter) incomingSettingsHandler(s *incomingSettings) error { 601 if err := l.applySettings(s.ss); err != nil { 602 return err 603 } 604 return l.framer.fr.WriteSettingsAck() 605} 606 607func (l *loopyWriter) registerStreamHandler(h *registerStream) error { 608 str := &outStream{ 609 id: h.streamID, 610 state: empty, 611 itl: &itemList{}, 612 wq: h.wq, 613 } 614 l.estdStreams[h.streamID] = str 615 return nil 616} 617 618func (l *loopyWriter) headerHandler(h *headerFrame) error { 619 if l.side == serverSide { 620 str, ok := l.estdStreams[h.streamID] 621 if !ok { 622 if logger.V(logLevel) { 623 logger.Warningf("transport: loopy doesn't recognize the stream: %d", h.streamID) 624 } 625 return nil 626 } 627 // Case 1.A: Server is responding back with headers. 628 if !h.endStream { 629 return l.writeHeader(h.streamID, h.endStream, h.hf, h.onWrite) 630 } 631 // else: Case 1.B: Server wants to close stream. 632 633 if str.state != empty { // either active or waiting on stream quota. 634 // add it str's list of items. 635 str.itl.enqueue(h) 636 return nil 637 } 638 if err := l.writeHeader(h.streamID, h.endStream, h.hf, h.onWrite); err != nil { 639 return err 640 } 641 return l.cleanupStreamHandler(h.cleanup) 642 } 643 // Case 2: Client wants to originate stream. 644 str := &outStream{ 645 id: h.streamID, 646 state: empty, 647 itl: &itemList{}, 648 wq: h.wq, 649 } 650 str.itl.enqueue(h) 651 return l.originateStream(str) 652} 653 654func (l *loopyWriter) originateStream(str *outStream) error { 655 hdr := str.itl.dequeue().(*headerFrame) 656 if err := hdr.initStream(str.id); err != nil { 657 if err == ErrConnClosing { 658 return err 659 } 660 // Other errors(errStreamDrain) need not close transport. 661 return nil 662 } 663 if err := l.writeHeader(str.id, hdr.endStream, hdr.hf, hdr.onWrite); err != nil { 664 return err 665 } 666 l.estdStreams[str.id] = str 667 return nil 668} 669 670func (l *loopyWriter) writeHeader(streamID uint32, endStream bool, hf []hpack.HeaderField, onWrite func()) error { 671 if onWrite != nil { 672 onWrite() 673 } 674 l.hBuf.Reset() 675 for _, f := range hf { 676 if err := l.hEnc.WriteField(f); err != nil { 677 if logger.V(logLevel) { 678 logger.Warningf("transport: loopyWriter.writeHeader encountered error while encoding headers: %v", err) 679 } 680 } 681 } 682 var ( 683 err error 684 endHeaders, first bool 685 ) 686 first = true 687 for !endHeaders { 688 size := l.hBuf.Len() 689 if size > http2MaxFrameLen { 690 size = http2MaxFrameLen 691 } else { 692 endHeaders = true 693 } 694 if first { 695 first = false 696 err = l.framer.fr.WriteHeaders(http2.HeadersFrameParam{ 697 StreamID: streamID, 698 BlockFragment: l.hBuf.Next(size), 699 EndStream: endStream, 700 EndHeaders: endHeaders, 701 }) 702 } else { 703 err = l.framer.fr.WriteContinuation( 704 streamID, 705 endHeaders, 706 l.hBuf.Next(size), 707 ) 708 } 709 if err != nil { 710 return err 711 } 712 } 713 return nil 714} 715 716func (l *loopyWriter) preprocessData(df *dataFrame) error { 717 str, ok := l.estdStreams[df.streamID] 718 if !ok { 719 return nil 720 } 721 // If we got data for a stream it means that 722 // stream was originated and the headers were sent out. 723 str.itl.enqueue(df) 724 if str.state == empty { 725 str.state = active 726 l.activeStreams.enqueue(str) 727 } 728 return nil 729} 730 731func (l *loopyWriter) pingHandler(p *ping) error { 732 if !p.ack { 733 l.bdpEst.timesnap(p.data) 734 } 735 return l.framer.fr.WritePing(p.ack, p.data) 736 737} 738 739func (l *loopyWriter) outFlowControlSizeRequestHandler(o *outFlowControlSizeRequest) error { 740 o.resp <- l.sendQuota 741 return nil 742} 743 744func (l *loopyWriter) cleanupStreamHandler(c *cleanupStream) error { 745 c.onWrite() 746 if str, ok := l.estdStreams[c.streamID]; ok { 747 // On the server side it could be a trailers-only response or 748 // a RST_STREAM before stream initialization thus the stream might 749 // not be established yet. 750 delete(l.estdStreams, c.streamID) 751 str.deleteSelf() 752 } 753 if c.rst { // If RST_STREAM needs to be sent. 754 if err := l.framer.fr.WriteRSTStream(c.streamID, c.rstCode); err != nil { 755 return err 756 } 757 } 758 if l.side == clientSide && l.draining && len(l.estdStreams) == 0 { 759 return ErrConnClosing 760 } 761 return nil 762} 763 764func (l *loopyWriter) earlyAbortStreamHandler(eas *earlyAbortStream) error { 765 if l.side == clientSide { 766 return errors.New("earlyAbortStream not handled on client") 767 } 768 769 headerFields := []hpack.HeaderField{ 770 {Name: ":status", Value: "200"}, 771 {Name: "content-type", Value: grpcutil.ContentType(eas.contentSubtype)}, 772 {Name: "grpc-status", Value: strconv.Itoa(int(eas.status.Code()))}, 773 {Name: "grpc-message", Value: encodeGrpcMessage(eas.status.Message())}, 774 } 775 776 if err := l.writeHeader(eas.streamID, true, headerFields, nil); err != nil { 777 return err 778 } 779 return nil 780} 781 782func (l *loopyWriter) incomingGoAwayHandler(*incomingGoAway) error { 783 if l.side == clientSide { 784 l.draining = true 785 if len(l.estdStreams) == 0 { 786 return ErrConnClosing 787 } 788 } 789 return nil 790} 791 792func (l *loopyWriter) goAwayHandler(g *goAway) error { 793 // Handling of outgoing GoAway is very specific to side. 794 if l.ssGoAwayHandler != nil { 795 draining, err := l.ssGoAwayHandler(g) 796 if err != nil { 797 return err 798 } 799 l.draining = draining 800 } 801 return nil 802} 803 804func (l *loopyWriter) handle(i interface{}) error { 805 switch i := i.(type) { 806 case *incomingWindowUpdate: 807 return l.incomingWindowUpdateHandler(i) 808 case *outgoingWindowUpdate: 809 return l.outgoingWindowUpdateHandler(i) 810 case *incomingSettings: 811 return l.incomingSettingsHandler(i) 812 case *outgoingSettings: 813 return l.outgoingSettingsHandler(i) 814 case *headerFrame: 815 return l.headerHandler(i) 816 case *registerStream: 817 return l.registerStreamHandler(i) 818 case *cleanupStream: 819 return l.cleanupStreamHandler(i) 820 case *earlyAbortStream: 821 return l.earlyAbortStreamHandler(i) 822 case *incomingGoAway: 823 return l.incomingGoAwayHandler(i) 824 case *dataFrame: 825 return l.preprocessData(i) 826 case *ping: 827 return l.pingHandler(i) 828 case *goAway: 829 return l.goAwayHandler(i) 830 case *outFlowControlSizeRequest: 831 return l.outFlowControlSizeRequestHandler(i) 832 default: 833 return fmt.Errorf("transport: unknown control message type %T", i) 834 } 835} 836 837func (l *loopyWriter) applySettings(ss []http2.Setting) error { 838 for _, s := range ss { 839 switch s.ID { 840 case http2.SettingInitialWindowSize: 841 o := l.oiws 842 l.oiws = s.Val 843 if o < l.oiws { 844 // If the new limit is greater make all depleted streams active. 845 for _, stream := range l.estdStreams { 846 if stream.state == waitingOnStreamQuota { 847 stream.state = active 848 l.activeStreams.enqueue(stream) 849 } 850 } 851 } 852 case http2.SettingHeaderTableSize: 853 updateHeaderTblSize(l.hEnc, s.Val) 854 } 855 } 856 return nil 857} 858 859// processData removes the first stream from active streams, writes out at most 16KB 860// of its data and then puts it at the end of activeStreams if there's still more data 861// to be sent and stream has some stream-level flow control. 862func (l *loopyWriter) processData() (bool, error) { 863 if l.sendQuota == 0 { 864 return true, nil 865 } 866 str := l.activeStreams.dequeue() // Remove the first stream. 867 if str == nil { 868 return true, nil 869 } 870 dataItem := str.itl.peek().(*dataFrame) // Peek at the first data item this stream. 871 // A data item is represented by a dataFrame, since it later translates into 872 // multiple HTTP2 data frames. 873 // Every dataFrame has two buffers; h that keeps grpc-message header and d that is acutal data. 874 // As an optimization to keep wire traffic low, data from d is copied to h to make as big as the 875 // maximum possilbe HTTP2 frame size. 876 877 if len(dataItem.h) == 0 && len(dataItem.d) == 0 { // Empty data frame 878 // Client sends out empty data frame with endStream = true 879 if err := l.framer.fr.WriteData(dataItem.streamID, dataItem.endStream, nil); err != nil { 880 return false, err 881 } 882 str.itl.dequeue() // remove the empty data item from stream 883 if str.itl.isEmpty() { 884 str.state = empty 885 } else if trailer, ok := str.itl.peek().(*headerFrame); ok { // the next item is trailers. 886 if err := l.writeHeader(trailer.streamID, trailer.endStream, trailer.hf, trailer.onWrite); err != nil { 887 return false, err 888 } 889 if err := l.cleanupStreamHandler(trailer.cleanup); err != nil { 890 return false, nil 891 } 892 } else { 893 l.activeStreams.enqueue(str) 894 } 895 return false, nil 896 } 897 var ( 898 buf []byte 899 ) 900 // Figure out the maximum size we can send 901 maxSize := http2MaxFrameLen 902 if strQuota := int(l.oiws) - str.bytesOutStanding; strQuota <= 0 { // stream-level flow control. 903 str.state = waitingOnStreamQuota 904 return false, nil 905 } else if maxSize > strQuota { 906 maxSize = strQuota 907 } 908 if maxSize > int(l.sendQuota) { // connection-level flow control. 909 maxSize = int(l.sendQuota) 910 } 911 // Compute how much of the header and data we can send within quota and max frame length 912 hSize := min(maxSize, len(dataItem.h)) 913 dSize := min(maxSize-hSize, len(dataItem.d)) 914 if hSize != 0 { 915 if dSize == 0 { 916 buf = dataItem.h 917 } else { 918 // We can add some data to grpc message header to distribute bytes more equally across frames. 919 // Copy on the stack to avoid generating garbage 920 var localBuf [http2MaxFrameLen]byte 921 copy(localBuf[:hSize], dataItem.h) 922 copy(localBuf[hSize:], dataItem.d[:dSize]) 923 buf = localBuf[:hSize+dSize] 924 } 925 } else { 926 buf = dataItem.d 927 } 928 929 size := hSize + dSize 930 931 // Now that outgoing flow controls are checked we can replenish str's write quota 932 str.wq.replenish(size) 933 var endStream bool 934 // If this is the last data message on this stream and all of it can be written in this iteration. 935 if dataItem.endStream && len(dataItem.h)+len(dataItem.d) <= size { 936 endStream = true 937 } 938 if dataItem.onEachWrite != nil { 939 dataItem.onEachWrite() 940 } 941 if err := l.framer.fr.WriteData(dataItem.streamID, endStream, buf[:size]); err != nil { 942 return false, err 943 } 944 str.bytesOutStanding += size 945 l.sendQuota -= uint32(size) 946 dataItem.h = dataItem.h[hSize:] 947 dataItem.d = dataItem.d[dSize:] 948 949 if len(dataItem.h) == 0 && len(dataItem.d) == 0 { // All the data from that message was written out. 950 str.itl.dequeue() 951 } 952 if str.itl.isEmpty() { 953 str.state = empty 954 } else if trailer, ok := str.itl.peek().(*headerFrame); ok { // The next item is trailers. 955 if err := l.writeHeader(trailer.streamID, trailer.endStream, trailer.hf, trailer.onWrite); err != nil { 956 return false, err 957 } 958 if err := l.cleanupStreamHandler(trailer.cleanup); err != nil { 959 return false, err 960 } 961 } else if int(l.oiws)-str.bytesOutStanding <= 0 { // Ran out of stream quota. 962 str.state = waitingOnStreamQuota 963 } else { // Otherwise add it back to the list of active streams. 964 l.activeStreams.enqueue(str) 965 } 966 return false, nil 967} 968 969func min(a, b int) int { 970 if a < b { 971 return a 972 } 973 return b 974} 975