flow.go

     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  // Flow control
     6  
     7  package http2
     8  
     9  // inflowMinRefresh is the minimum number of bytes we'll send for a
    10  // flow control window update.
    11  const inflowMinRefresh = 4 << 10
    12  
    13  // maxFlowWindow is the maximum size of a flow control window.
    14  const maxFlowWindow = (1 << 31) - 1
    15  
    16  // inflow accounts for an inbound flow control window.
    17  // It tracks both the latest window sent to the peer (used for enforcement)
    18  // and the accumulated unsent window.
    19  type inflow struct {
    20  	avail  int32
    21  	unsent int32
    22  }
    23  
    24  // init sets the initial window.
    25  func (f *inflow) init(n int32) {
    26  	f.avail = n
    27  }
    28  
    29  // add adds n bytes to the window, with a maximum window size of max,
    30  // indicating that the peer can now send us more data.
    31  // For example, the user read from a {Request,Response} body and consumed
    32  // some of the buffered data, so the peer can now send more.
    33  // It returns the number of bytes to send in a WINDOW_UPDATE frame to the peer.
    34  // Window updates are accumulated and sent when the unsent capacity
    35  // is at least inflowMinRefresh or will at least double the peer's available window.
    36  func (f *inflow) add(n int) (connAdd int32) {
    37  	if n < 0 {
    38  		panic("negative update")
    39  	}
    40  	unsent := int64(f.unsent) + int64(n)
    41  	// "A sender MUST NOT allow a flow-control window to exceed 2^31-1 octets."
    42  	// RFC 7540 Section 6.9.1.
    43  	const maxWindow = 1<<31 - 1
    44  	if unsent+int64(f.avail) > maxWindow {
    45  		panic("flow control update exceeds maximum window size")
    46  	}
    47  	f.unsent = int32(unsent)
    48  	if f.unsent < inflowMinRefresh && f.unsent < f.avail {
    49  		// If there aren't at least inflowMinRefresh bytes of window to send,
    50  		// and this update won't at least double the window, buffer the update for later.
    51  		return 0
    52  	}
    53  	f.avail += f.unsent
    54  	f.unsent = 0
    55  	return int32(unsent)
    56  }
    57  
    58  // take attempts to take n bytes from the peer's flow control window.
    59  // It reports whether the window has available capacity.
    60  func (f *inflow) take(n uint32) bool {
    61  	if n > uint32(f.avail) {
    62  		return false
    63  	}
    64  	f.avail -= int32(n)
    65  	return true
    66  }
    67  
    68  // takeInflows attempts to take n bytes from two inflows,
    69  // typically connection-level and stream-level flows.
    70  // It reports whether both windows have available capacity.
    71  func takeInflows(f1, f2 *inflow, n uint32) bool {
    72  	if n > uint32(f1.avail) || n > uint32(f2.avail) {
    73  		return false
    74  	}
    75  	f1.avail -= int32(n)
    76  	f2.avail -= int32(n)
    77  	return true
    78  }
    79  
    80  // connOutflow is connection-level outbound flow control.
    81  type connOutflow struct {
    82  	initial int32 // SETTINGS_INITIAL_WINDOW_SIZE, changes with settings updates
    83  	n       int32 // connection-level flow control window
    84  	flowErr bool  // set when a flow control error is encountered
    85  }
    86  
    87  func (f *connOutflow) init() {
    88  	f.initial = initialWindowSize // initial stream window size
    89  	f.n = initialWindowSize       // current connection window size
    90  }
    91  
    92  func (f *connOutflow) changeInitialWindowSize(size int64) bool {
    93  	if size > maxFlowWindow {
    94  		f.flowErr = true
    95  		return false
    96  	}
    97  	f.initial = int32(size)
    98  	return true
    99  }
   100  
   101  func (f *connOutflow) add(n int32) bool {
   102  	sum := int64(f.n) + int64(n)
   103  	if sum > maxFlowWindow {
   104  		f.flowErr = true
   105  		return false
   106  	}
   107  	f.n += n
   108  	return true
   109  }
   110  
   111  // outflow is the stream-level outbound flow control window's size.
   112  type outflow struct {
   113  	_ incomparable
   114  
   115  	// delta is the difference between the stream's flow control window and
   116  	// the connection's initial window size (conn.initial).
   117  	//
   118  	// Another view is that delta is the number of flow control bytes provided to this
   119  	// stream in WINDOW_UPDATE frames, less the number of bytes sent on the stream.
   120  	delta int32
   121  
   122  	// conn points to the shared connection-level outflow that is
   123  	// shared by all streams on that conn.
   124  	conn *connOutflow
   125  }
   126  
   127  func (f *outflow) available() (int32, bool) {
   128  	if f.conn == nil {
   129  		return maxFlowWindow, true // only happens in tests
   130  	}
   131  	if f.conn.flowErr {
   132  		// Block all sending once any stream observes a flow control error.
   133  		return 0, false
   134  	}
   135  	n := int64(f.conn.initial) + int64(f.delta)
   136  	if n > maxFlowWindow {
   137  		f.conn.flowErr = true
   138  		return 0, false
   139  	}
   140  	return min(int32(n), f.conn.n), true
   141  }
   142  
   143  func (f *outflow) take(n int32) {
   144  	if f.conn == nil {
   145  		return // only happens in tests
   146  	}
   147  	avail, _ := f.available()
   148  	if n > avail {
   149  		panic("internal error: took too much")
   150  	}
   151  	f.delta -= n
   152  	f.conn.n -= n
   153  }
   154  
   155  // add adds n bytes (positive or negative) to the flow control window.
   156  // It returns false if the sum would exceed 2^31-1.
   157  func (f *outflow) add(n int32) bool {
   158  	if f.conn == nil {
   159  		return true // only happens in tests
   160  	}
   161  	avail := int64(f.conn.initial) + int64(f.delta)
   162  	if avail > maxFlowWindow {
   163  		// An earlier change to the initial window pushed this stream over the limit.
   164  		// This is a connection-level flow control error.
   165  		f.conn.flowErr = true
   166  		return false
   167  	}
   168  	if avail+int64(n) > maxFlowWindow {
   169  		// This update would push the stream over the limit.
   170  		// This is a stream-level flow control error.
   171  		return false
   172  	}
   173  	f.delta += n
   174  	return true
   175  }
   176
View as plain text