1 //
2 // ========================================================================
3 // Copyright (c) 1995-2016 Mort Bay Consulting Pty. Ltd.
4 // ------------------------------------------------------------------------
5 // All rights reserved. This program and the accompanying materials
6 // are made available under the terms of the Eclipse Public License v1.0
7 // and Apache License v2.0 which accompanies this distribution.
8 //
9 // The Eclipse Public License is available at
10 // http://www.eclipse.org/legal/epl-v10.html
11 //
12 // The Apache License v2.0 is available at
13 // http://www.opensource.org/licenses/apache2.0.php
14 //
15 // You may elect to redistribute this code under either of these licenses.
16 // ========================================================================
17 //
18
19 package org.eclipse.jetty.http2;
20
21 import java.util.Map;
22 import java.util.concurrent.ConcurrentHashMap;
23 import java.util.concurrent.atomic.AtomicInteger;
24
25 import org.eclipse.jetty.http2.frames.Frame;
26 import org.eclipse.jetty.http2.frames.WindowUpdateFrame;
27 import org.eclipse.jetty.util.Atomics;
28 import org.eclipse.jetty.util.Callback;
29 import org.eclipse.jetty.util.annotation.ManagedAttribute;
30 import org.eclipse.jetty.util.annotation.ManagedObject;
31
32 /**
33 * <p>A flow control strategy that accumulates updates and emits window control
34 * frames when the accumulated value reaches a threshold.</p>
35 * <p>The sender flow control window is represented in the receiver as two
36 * buckets: a bigger bucket, initially full, that is drained when data is
37 * received, and a smaller bucket, initially empty, that is filled when data is
38 * consumed. Only the smaller bucket can refill the bigger bucket.</p>
39 * <p>The smaller bucket is defined as a fraction of the bigger bucket.</p>
40 * <p>For a more visual representation, see the
41 * <a href="http://en.wikipedia.org/wiki/Shishi-odoshi">rocking bamboo fountain</a>.</p>
42 * <p>The algorithm works in this way.</p>
43 * <p>The initial bigger bucket (BB) capacity is 100, and let's imagine the smaller
44 * bucket (SB) being 40% of the bigger bucket: 40.</p>
45 * <p>The receiver receives a data frame of 60, so now BB=40; the data frame is
46 * passed to the application that consumes 25, so now SB=25. Since SB is not full,
47 * no window control frames are emitted.</p>
48 * <p>The application consumes other 20, so now SB=45. Since SB is full, its 45
49 * are transferred to BB, which is now BB=85, and a window control frame is sent
50 * with delta=45.</p>
51 * <p>The application consumes the remaining 15, so now SB=15, and no window
52 * control frame is emitted.</p>
53 */
54 @ManagedObject
55 public class BufferingFlowControlStrategy extends AbstractFlowControlStrategy
56 {
57 private final AtomicInteger maxSessionRecvWindow = new AtomicInteger(DEFAULT_WINDOW_SIZE);
58 private final AtomicInteger sessionLevel = new AtomicInteger();
59 private final Map<IStream, AtomicInteger> streamLevels = new ConcurrentHashMap<>();
60 private float bufferRatio;
61
62 public BufferingFlowControlStrategy(float bufferRatio)
63 {
64 this(DEFAULT_WINDOW_SIZE, bufferRatio);
65 }
66
67 public BufferingFlowControlStrategy(int initialStreamSendWindow, float bufferRatio)
68 {
69 super(initialStreamSendWindow);
70 this.bufferRatio = bufferRatio;
71 }
72
73 @ManagedAttribute("The ratio between the receive buffer and the consume buffer")
74 public float getBufferRatio()
75 {
76 return bufferRatio;
77 }
78
79 public void setBufferRatio(float bufferRatio)
80 {
81 this.bufferRatio = bufferRatio;
82 }
83
84 @Override
85 public void onStreamCreated(IStream stream)
86 {
87 super.onStreamCreated(stream);
88 streamLevels.put(stream, new AtomicInteger());
89 }
90
91 @Override
92 public void onStreamDestroyed(IStream stream)
93 {
94 streamLevels.remove(stream);
95 super.onStreamDestroyed(stream);
96 }
97
98 @Override
99 public void onDataConsumed(ISession session, IStream stream, int length)
100 {
101 if (length <= 0)
102 return;
103
104 float ratio = bufferRatio;
105
106 WindowUpdateFrame windowFrame = null;
107 int level = sessionLevel.addAndGet(length);
108 int maxLevel = (int)(maxSessionRecvWindow.get() * ratio);
109 if (level > maxLevel)
110 {
111 if (sessionLevel.compareAndSet(level, 0))
112 {
113 session.updateRecvWindow(level);
114 if (LOG.isDebugEnabled())
115 LOG.debug("Data consumed, {} bytes, updated session recv window by {}/{} for {}", length, level, maxLevel, session);
116 windowFrame = new WindowUpdateFrame(0, level);
117 }
118 else
119 {
120 if (LOG.isDebugEnabled())
121 LOG.debug("Data consumed, {} bytes, concurrent session recv window level {}/{} for {}", length, sessionLevel, maxLevel, session);
122 }
123 }
124 else
125 {
126 if (LOG.isDebugEnabled())
127 LOG.debug("Data consumed, {} bytes, session recv window level {}/{} for {}", length, level, maxLevel, session);
128 }
129
130 Frame[] windowFrames = Frame.EMPTY_ARRAY;
131 if (stream != null)
132 {
133 if (stream.isClosed())
134 {
135 if (LOG.isDebugEnabled())
136 LOG.debug("Data consumed, {} bytes, ignoring update stream recv window for closed {}", length, stream);
137 }
138 else
139 {
140 AtomicInteger streamLevel = streamLevels.get(stream);
141 if (streamLevel != null)
142 {
143 level = streamLevel.addAndGet(length);
144 maxLevel = (int)(getInitialStreamRecvWindow() * ratio);
145 if (level > maxLevel)
146 {
147 level = streamLevel.getAndSet(0);
148 stream.updateRecvWindow(level);
149 if (LOG.isDebugEnabled())
150 LOG.debug("Data consumed, {} bytes, updated stream recv window by {}/{} for {}", length, level, maxLevel, stream);
151 WindowUpdateFrame frame = new WindowUpdateFrame(stream.getId(), level);
152 if (windowFrame == null)
153 windowFrame = frame;
154 else
155 windowFrames = new Frame[]{frame};
156 }
157 else
158 {
159 if (LOG.isDebugEnabled())
160 LOG.debug("Data consumed, {} bytes, stream recv window level {}/{} for {}", length, level, maxLevel, stream);
161 }
162 }
163 }
164 }
165
166 if (windowFrame != null)
167 session.frames(stream, Callback.NOOP, windowFrame, windowFrames);
168 }
169
170 @Override
171 public void windowUpdate(ISession session, IStream stream, WindowUpdateFrame frame)
172 {
173 super.windowUpdate(session, stream, frame);
174
175 // Window updates cannot be negative.
176 // The SettingsFrame.INITIAL_WINDOW_SIZE setting
177 // only influences the *stream* window size.
178 // Therefore the session window can only be enlarged,
179 // and here we keep track of its max value.
180
181 // Updating the max session recv window is done here
182 // so that if a peer decides to send an unilateral
183 // window update to enlarge the session window,
184 // without the corresponding data consumption, here
185 // we can track it.
186 // Note that it is not perfect, since there is a time
187 // window between the session recv window being updated
188 // before the window update frame is sent, and the
189 // invocation of this method: in between data may arrive
190 // and reduce the session recv window size.
191 // But eventually the max value will be seen.
192
193 // Note that we cannot avoid the time window described
194 // above by updating the session recv window from here
195 // because there is a race between the sender and the
196 // receiver: the sender may receive a window update and
197 // send more data, while this method has not yet been
198 // invoked; when the data is received the session recv
199 // window may become negative and the connection will
200 // be closed (per specification).
201
202 if (frame.getStreamId() == 0)
203 {
204 int sessionWindow = session.updateRecvWindow(0);
205 Atomics.updateMax(maxSessionRecvWindow, sessionWindow);
206 }
207 }
208
209 @Override
210 public String toString()
211 {
212 return String.format("%s@%x[ratio=%.2f,sessionLevel=%s,sessionStallTime=%dms,streamsStallTime=%dms]",
213 getClass().getSimpleName(),
214 hashCode(),
215 bufferRatio,
216 sessionLevel,
217 getSessionStallTime(),
218 getStreamsStallTime());
219 }
220 }