View Javadoc
1   /*
2    * Copyright 2011 The Netty Project
3    *
4    * The Netty Project licenses this file to you under the Apache License,
5    * version 2.0 (the "License"); you may not use this file except in compliance
6    * with the License. You may obtain a copy of the License at:
7    *
8    * http://www.apache.org/licenses/LICENSE-2.0
9    *
10   * Unless required by applicable law or agreed to in writing, software
11   * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
12   * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
13   * License for the specific language governing permissions and limitations
14   * under the License.
15   */
16  package io.netty.handler.traffic;
17  
18  import io.netty.buffer.ByteBuf;
19  import io.netty.buffer.ByteBufHolder;
20  import io.netty.channel.Channel;
21  import io.netty.channel.ChannelDuplexHandler;
22  import io.netty.channel.ChannelConfig;
23  import io.netty.channel.ChannelHandlerContext;
24  import io.netty.channel.ChannelOutboundBuffer;
25  import io.netty.channel.ChannelPromise;
26  import io.netty.util.Attribute;
27  import io.netty.util.AttributeKey;
28  import io.netty.util.internal.logging.InternalLogger;
29  import io.netty.util.internal.logging.InternalLoggerFactory;
30  
31  import java.util.concurrent.TimeUnit;
32  
33  /**
34   * <p>AbstractTrafficShapingHandler allows to limit the global bandwidth
35   * (see {@link GlobalTrafficShapingHandler}) or per session
36   * bandwidth (see {@link ChannelTrafficShapingHandler}), as traffic shaping.
37   * It allows you to implement an almost real time monitoring of the bandwidth using
38   * the monitors from {@link TrafficCounter} that will call back every checkInterval
39   * the method doAccounting of this handler.</p>
40   *
41   * <p>If you want for any particular reasons to stop the monitoring (accounting) or to change
42   * the read/write limit or the check interval, several methods allow that for you:</p>
43   * <ul>
44   * <li><tt>configure</tt> allows you to change read or write limits, or the checkInterval</li>
45   * <li><tt>getTrafficCounter</tt> allows you to have access to the TrafficCounter and so to stop
46   * or start the monitoring, to change the checkInterval directly, or to have access to its values.</li>
47   * </ul>
48   */
49  public abstract class AbstractTrafficShapingHandler extends ChannelDuplexHandler {
50      private static final InternalLogger logger =
51              InternalLoggerFactory.getInstance(AbstractTrafficShapingHandler.class);
52      /**
53       * Default delay between two checks: 1s
54       */
55      public static final long DEFAULT_CHECK_INTERVAL = 1000;
56  
57     /**
58      * Default max delay in case of traffic shaping
59      * (during which no communication will occur).
60      * Shall be less than TIMEOUT. Here half of "standard" 30s
61      */
62      public static final long DEFAULT_MAX_TIME = 15000;
63  
64      /**
65       * Default max size to not exceed in buffer (write only).
66       */
67      static final long DEFAULT_MAX_SIZE = 4 * 1024 * 1024L;
68  
69      /**
70       * Default minimal time to wait
71       */
72      static final long MINIMAL_WAIT = 10;
73  
74      /**
75       * Traffic Counter
76       */
77      protected TrafficCounter trafficCounter;
78  
79      /**
80       * Limit in B/s to apply to write
81       */
82      private volatile long writeLimit;
83  
84      /**
85       * Limit in B/s to apply to read
86       */
87      private volatile long readLimit;
88  
89      /**
90       * Max delay in wait
91       */
92      protected volatile long maxTime = DEFAULT_MAX_TIME; // default 15 s
93  
94      /**
95       * Delay between two performance snapshots
96       */
97      protected volatile long checkInterval = DEFAULT_CHECK_INTERVAL; // default 1 s
98  
99      static final AttributeKey<Boolean> READ_SUSPENDED = AttributeKey
100             .valueOf(AbstractTrafficShapingHandler.class.getName() + ".READ_SUSPENDED");
101     static final AttributeKey<Runnable> REOPEN_TASK = AttributeKey.valueOf(AbstractTrafficShapingHandler.class
102             .getName() + ".REOPEN_TASK");
103 
104     /**
105      * Max time to delay before proposing to stop writing new objects from next handlers
106      */
107     volatile long maxWriteDelay = 4 * DEFAULT_CHECK_INTERVAL; // default 4 s
108     /**
109      * Max size in the list before proposing to stop writing new objects from next handlers
110      */
111     volatile long maxWriteSize = DEFAULT_MAX_SIZE; // default 4MB
112 
113     /**
114      * Rank in UserDefinedWritability (1 for Channel, 2 for Global TrafficShapingHandler).
115      * Set in final constructor. Must be between 1 and 31
116      */
117     final int userDefinedWritabilityIndex;
118 
119     /**
120      * Default value for Channel UserDefinedWritability index
121      */
122     static final int CHANNEL_DEFAULT_USER_DEFINED_WRITABILITY_INDEX = 1;
123 
124     /**
125      * Default value for Global UserDefinedWritability index
126      */
127     static final int GLOBAL_DEFAULT_USER_DEFINED_WRITABILITY_INDEX = 2;
128 
129     /**
130      * Default value for GlobalChannel UserDefinedWritability index
131      */
132     static final int GLOBALCHANNEL_DEFAULT_USER_DEFINED_WRITABILITY_INDEX = 3;
133 
134     /**
135      * @param newTrafficCounter
136      *            the TrafficCounter to set
137      */
138     void setTrafficCounter(TrafficCounter newTrafficCounter) {
139         trafficCounter = newTrafficCounter;
140     }
141 
142     /**
143      * @return the index to be used by the TrafficShapingHandler to manage the user defined writability.
144      *              For Channel TSH it is defined as {@value #CHANNEL_DEFAULT_USER_DEFINED_WRITABILITY_INDEX},
145      *              for Global TSH it is defined as {@value #GLOBAL_DEFAULT_USER_DEFINED_WRITABILITY_INDEX},
146      *              for GlobalChannel TSH it is defined as
147      *              {@value #GLOBALCHANNEL_DEFAULT_USER_DEFINED_WRITABILITY_INDEX}.
148      */
149     protected int userDefinedWritabilityIndex() {
150         return CHANNEL_DEFAULT_USER_DEFINED_WRITABILITY_INDEX;
151     }
152 
153     /**
154      * @param writeLimit
155      *          0 or a limit in bytes/s
156      * @param readLimit
157      *          0 or a limit in bytes/s
158      * @param checkInterval
159      *            The delay between two computations of performances for
160      *            channels or 0 if no stats are to be computed.
161      * @param maxTime
162      *            The maximum delay to wait in case of traffic excess.
163      *            Must be positive.
164      */
165     protected AbstractTrafficShapingHandler(long writeLimit, long readLimit, long checkInterval, long maxTime) {
166         if (maxTime <= 0) {
167             throw new IllegalArgumentException("maxTime must be positive");
168         }
169 
170         userDefinedWritabilityIndex = userDefinedWritabilityIndex();
171         this.writeLimit = writeLimit;
172         this.readLimit = readLimit;
173         this.checkInterval = checkInterval;
174         this.maxTime = maxTime;
175     }
176 
177     /**
178      * Constructor using default max time as delay allowed value of {@value #DEFAULT_MAX_TIME} ms.
179      * @param writeLimit
180      *            0 or a limit in bytes/s
181      * @param readLimit
182      *            0 or a limit in bytes/s
183      * @param checkInterval
184      *            The delay between two computations of performances for
185      *            channels or 0 if no stats are to be computed.
186      */
187     protected AbstractTrafficShapingHandler(long writeLimit, long readLimit, long checkInterval) {
188         this(writeLimit, readLimit, checkInterval, DEFAULT_MAX_TIME);
189     }
190 
191     /**
192      * Constructor using default Check Interval value of {@value #DEFAULT_CHECK_INTERVAL} ms and
193      * default max time as delay allowed value of {@value #DEFAULT_MAX_TIME} ms.
194      *
195      * @param writeLimit
196      *          0 or a limit in bytes/s
197      * @param readLimit
198      *          0 or a limit in bytes/s
199      */
200     protected AbstractTrafficShapingHandler(long writeLimit, long readLimit) {
201         this(writeLimit, readLimit, DEFAULT_CHECK_INTERVAL, DEFAULT_MAX_TIME);
202     }
203 
204     /**
205      * Constructor using NO LIMIT, default Check Interval value of {@value #DEFAULT_CHECK_INTERVAL} ms and
206      * default max time as delay allowed value of {@value #DEFAULT_MAX_TIME} ms.
207      */
208     protected AbstractTrafficShapingHandler() {
209         this(0, 0, DEFAULT_CHECK_INTERVAL, DEFAULT_MAX_TIME);
210     }
211 
212     /**
213      * Constructor using NO LIMIT and
214      * default max time as delay allowed value of {@value #DEFAULT_MAX_TIME} ms.
215      *
216      * @param checkInterval
217      *            The delay between two computations of performances for
218      *            channels or 0 if no stats are to be computed.
219      */
220     protected AbstractTrafficShapingHandler(long checkInterval) {
221         this(0, 0, checkInterval, DEFAULT_MAX_TIME);
222     }
223 
224     /**
225      * Change the underlying limitations and check interval.
226      * <p>Note the change will be taken as best effort, meaning
227      * that all already scheduled traffics will not be
228      * changed, but only applied to new traffics.</p>
229      * <p>So the expected usage of this method is to be used not too often,
230      * accordingly to the traffic shaping configuration.</p>
231      *
232      * @param newWriteLimit The new write limit (in bytes)
233      * @param newReadLimit The new read limit (in bytes)
234      * @param newCheckInterval The new check interval (in milliseconds)
235      */
236     public void configure(long newWriteLimit, long newReadLimit,
237             long newCheckInterval) {
238         configure(newWriteLimit, newReadLimit);
239         configure(newCheckInterval);
240     }
241 
242     /**
243      * Change the underlying limitations.
244      * <p>Note the change will be taken as best effort, meaning
245      * that all already scheduled traffics will not be
246      * changed, but only applied to new traffics.</p>
247      * <p>So the expected usage of this method is to be used not too often,
248      * accordingly to the traffic shaping configuration.</p>
249      *
250      * @param newWriteLimit The new write limit (in bytes)
251      * @param newReadLimit The new read limit (in bytes)
252      */
253     public void configure(long newWriteLimit, long newReadLimit) {
254         writeLimit = newWriteLimit;
255         readLimit = newReadLimit;
256         if (trafficCounter != null) {
257             trafficCounter.resetAccounting(TrafficCounter.milliSecondFromNano());
258         }
259     }
260 
261     /**
262      * Change the check interval.
263      *
264      * @param newCheckInterval The new check interval (in milliseconds)
265      */
266     public void configure(long newCheckInterval) {
267         checkInterval = newCheckInterval;
268         if (trafficCounter != null) {
269             trafficCounter.configure(checkInterval);
270         }
271     }
272 
273     /**
274      * @return the writeLimit
275      */
276     public long getWriteLimit() {
277         return writeLimit;
278     }
279 
280     /**
281      * <p>Note the change will be taken as best effort, meaning
282      * that all already scheduled traffics will not be
283      * changed, but only applied to new traffics.</p>
284      * <p>So the expected usage of this method is to be used not too often,
285      * accordingly to the traffic shaping configuration.</p>
286      *
287      * @param writeLimit the writeLimit to set
288      */
289     public void setWriteLimit(long writeLimit) {
290         this.writeLimit = writeLimit;
291         if (trafficCounter != null) {
292             trafficCounter.resetAccounting(TrafficCounter.milliSecondFromNano());
293         }
294     }
295 
296     /**
297      * @return the readLimit
298      */
299     public long getReadLimit() {
300         return readLimit;
301     }
302 
303     /**
304      * <p>Note the change will be taken as best effort, meaning
305      * that all already scheduled traffics will not be
306      * changed, but only applied to new traffics.</p>
307      * <p>So the expected usage of this method is to be used not too often,
308      * accordingly to the traffic shaping configuration.</p>
309      *
310      * @param readLimit the readLimit to set
311      */
312     public void setReadLimit(long readLimit) {
313         this.readLimit = readLimit;
314         if (trafficCounter != null) {
315             trafficCounter.resetAccounting(TrafficCounter.milliSecondFromNano());
316         }
317     }
318 
319     /**
320      * @return the checkInterval
321      */
322     public long getCheckInterval() {
323         return checkInterval;
324     }
325 
326     /**
327      * @param checkInterval the interval in ms between each step check to set, default value being 1000 ms.
328      */
329     public void setCheckInterval(long checkInterval) {
330         this.checkInterval = checkInterval;
331         if (trafficCounter != null) {
332             trafficCounter.configure(checkInterval);
333         }
334     }
335 
336     /**
337      * <p>Note the change will be taken as best effort, meaning
338      * that all already scheduled traffics will not be
339      * changed, but only applied to new traffics.</p>
340      * <p>So the expected usage of this method is to be used not too often,
341      * accordingly to the traffic shaping configuration.</p>
342      *
343      * @param maxTime
344      *            Max delay in wait, shall be less than TIME OUT in related protocol.
345      *            Must be positive.
346      */
347     public void setMaxTimeWait(long maxTime) {
348         if (maxTime <= 0) {
349             throw new IllegalArgumentException("maxTime must be positive");
350         }
351         this.maxTime = maxTime;
352     }
353 
354     /**
355      * @return the max delay in wait to prevent TIME OUT
356      */
357     public long getMaxTimeWait() {
358         return maxTime;
359     }
360 
361     /**
362      * @return the maxWriteDelay
363      */
364     public long getMaxWriteDelay() {
365         return maxWriteDelay;
366     }
367 
368     /**
369      * <p>Note the change will be taken as best effort, meaning
370      * that all already scheduled traffics will not be
371      * changed, but only applied to new traffics.</p>
372      * <p>So the expected usage of this method is to be used not too often,
373      * accordingly to the traffic shaping configuration.</p>
374      *
375      * @param maxWriteDelay the maximum Write Delay in ms in the buffer allowed before write suspension is set.
376      *              Must be positive.
377      */
378     public void setMaxWriteDelay(long maxWriteDelay) {
379         if (maxWriteDelay <= 0) {
380             throw new IllegalArgumentException("maxWriteDelay must be positive");
381         }
382         this.maxWriteDelay = maxWriteDelay;
383     }
384 
385     /**
386      * @return the maxWriteSize default being {@value #DEFAULT_MAX_SIZE} bytes.
387      */
388     public long getMaxWriteSize() {
389         return maxWriteSize;
390     }
391 
392     /**
393      * <p>Note that this limit is a best effort on memory limitation to prevent Out Of
394      * Memory Exception. To ensure it works, the handler generating the write should
395      * use one of the way provided by Netty to handle the capacity:</p>
396      * <p>- the {@code Channel.isWritable()} property and the corresponding
397      * {@code channelWritabilityChanged()}</p>
398      * <p>- the {@code ChannelFuture.addListener(new GenericFutureListener())}</p>
399      *
400      * @param maxWriteSize the maximum Write Size allowed in the buffer
401      *            per channel before write suspended is set,
402      *            default being {@value #DEFAULT_MAX_SIZE} bytes.
403      */
404     public void setMaxWriteSize(long maxWriteSize) {
405         this.maxWriteSize = maxWriteSize;
406     }
407 
408     /**
409      * Called each time the accounting is computed from the TrafficCounters.
410      * This method could be used for instance to implement almost real time accounting.
411      *
412      * @param counter
413      *            the TrafficCounter that computes its performance
414      */
415     protected void doAccounting(TrafficCounter counter) {
416         // NOOP by default
417     }
418 
419     /**
420      * Class to implement setReadable at fix time
421      */
422     static final class ReopenReadTimerTask implements Runnable {
423         final ChannelHandlerContext ctx;
424         ReopenReadTimerTask(ChannelHandlerContext ctx) {
425             this.ctx = ctx;
426         }
427 
428         @Override
429         public void run() {
430             Channel channel = ctx.channel();
431             ChannelConfig config = channel.config();
432             if (!config.isAutoRead() && isHandlerActive(ctx)) {
433                 // If AutoRead is False and Active is True, user make a direct setAutoRead(false)
434                 // Then Just reset the status
435                 if (logger.isDebugEnabled()) {
436                     logger.debug("Not unsuspend: " + config.isAutoRead() + ':' +
437                             isHandlerActive(ctx));
438                 }
439                 channel.attr(READ_SUSPENDED).set(false);
440             } else {
441                 // Anything else allows the handler to reset the AutoRead
442                 if (logger.isDebugEnabled()) {
443                     if (config.isAutoRead() && !isHandlerActive(ctx)) {
444                         if (logger.isDebugEnabled()) {
445                             logger.debug("Unsuspend: " + config.isAutoRead() + ':' +
446                                     isHandlerActive(ctx));
447                         }
448                     } else {
449                         if (logger.isDebugEnabled()) {
450                             logger.debug("Normal unsuspend: " + config.isAutoRead() + ':'
451                                     + isHandlerActive(ctx));
452                         }
453                     }
454                 }
455                 channel.attr(READ_SUSPENDED).set(false);
456                 config.setAutoRead(true);
457                 channel.read();
458             }
459             if (logger.isDebugEnabled()) {
460                 logger.debug("Unsuspend final status => " + config.isAutoRead() + ':'
461                         + isHandlerActive(ctx));
462             }
463         }
464     }
465 
466     /**
467      * Release the Read suspension
468      */
469     void releaseReadSuspended(ChannelHandlerContext ctx) {
470         Channel channel = ctx.channel();
471         channel.attr(READ_SUSPENDED).set(false);
472         channel.config().setAutoRead(true);
473     }
474 
475     @Override
476     public void channelRead(final ChannelHandlerContext ctx, final Object msg) throws Exception {
477         long size = calculateSize(msg);
478         long now = TrafficCounter.milliSecondFromNano();
479         if (size > 0) {
480             // compute the number of ms to wait before reopening the channel
481             long wait = trafficCounter.readTimeToWait(size, readLimit, maxTime, now);
482             wait = checkWaitReadTime(ctx, wait, now);
483             if (wait >= MINIMAL_WAIT) { // At least 10ms seems a minimal
484                 // time in order to try to limit the traffic
485                 // Only AutoRead AND HandlerActive True means Context Active
486                 Channel channel = ctx.channel();
487                 ChannelConfig config = channel.config();
488                 if (logger.isDebugEnabled()) {
489                     logger.debug("Read suspend: " + wait + ':' + config.isAutoRead() + ':'
490                             + isHandlerActive(ctx));
491                 }
492                 if (config.isAutoRead() && isHandlerActive(ctx)) {
493                     config.setAutoRead(false);
494                     channel.attr(READ_SUSPENDED).set(true);
495                     // Create a Runnable to reactive the read if needed. If one was create before it will just be
496                     // reused to limit object creation
497                     Attribute<Runnable> attr = channel.attr(REOPEN_TASK);
498                     Runnable reopenTask = attr.get();
499                     if (reopenTask == null) {
500                         reopenTask = new ReopenReadTimerTask(ctx);
501                         attr.set(reopenTask);
502                     }
503                     ctx.executor().schedule(reopenTask, wait, TimeUnit.MILLISECONDS);
504                     if (logger.isDebugEnabled()) {
505                         logger.debug("Suspend final status => " + config.isAutoRead() + ':'
506                                 + isHandlerActive(ctx) + " will reopened at: " + wait);
507                     }
508                 }
509             }
510         }
511         informReadOperation(ctx, now);
512         ctx.fireChannelRead(msg);
513     }
514 
515     /**
516      * Method overridden in GTSH to take into account specific timer for the channel.
517      * @param wait the wait delay computed in ms
518      * @param now the relative now time in ms
519      * @return the wait to use according to the context
520      */
521     long checkWaitReadTime(final ChannelHandlerContext ctx, long wait, final long now) {
522         // no change by default
523         return wait;
524     }
525 
526     /**
527      * Method overridden in GTSH to take into account specific timer for the channel.
528      * @param now the relative now time in ms
529      */
530     void informReadOperation(final ChannelHandlerContext ctx, final long now) {
531         // default noop
532     }
533 
534     protected static boolean isHandlerActive(ChannelHandlerContext ctx) {
535         Boolean suspended = ctx.channel().attr(READ_SUSPENDED).get();
536         return suspended == null || Boolean.FALSE.equals(suspended);
537     }
538 
539     @Override
540     public void read(ChannelHandlerContext ctx) {
541         if (isHandlerActive(ctx)) {
542             // For Global Traffic (and Read when using EventLoop in pipeline) : check if READ_SUSPENDED is False
543             ctx.read();
544         }
545     }
546 
547     @Override
548     public void write(final ChannelHandlerContext ctx, final Object msg, final ChannelPromise promise)
549             throws Exception {
550         long size = calculateSize(msg);
551         long now = TrafficCounter.milliSecondFromNano();
552         if (size > 0) {
553             // compute the number of ms to wait before continue with the channel
554             long wait = trafficCounter.writeTimeToWait(size, writeLimit, maxTime, now);
555             if (wait >= MINIMAL_WAIT) {
556                 if (logger.isDebugEnabled()) {
557                     logger.debug("Write suspend: " + wait + ':' + ctx.channel().config().isAutoRead() + ':'
558                             + isHandlerActive(ctx));
559                 }
560                 submitWrite(ctx, msg, size, wait, now, promise);
561                 return;
562             }
563         }
564         // to maintain order of write
565         submitWrite(ctx, msg, size, 0, now, promise);
566     }
567 
568     @Deprecated
569     protected void submitWrite(final ChannelHandlerContext ctx, final Object msg,
570             final long delay, final ChannelPromise promise) {
571         submitWrite(ctx, msg, calculateSize(msg),
572                 delay, TrafficCounter.milliSecondFromNano(), promise);
573     }
574 
575     abstract void submitWrite(
576             ChannelHandlerContext ctx, Object msg, long size, long delay, long now, ChannelPromise promise);
577 
578     @Override
579     public void channelRegistered(ChannelHandlerContext ctx) throws Exception {
580         setUserDefinedWritability(ctx, true);
581         super.channelRegistered(ctx);
582     }
583 
584     void setUserDefinedWritability(ChannelHandlerContext ctx, boolean writable) {
585         ChannelOutboundBuffer cob = ctx.channel().unsafe().outboundBuffer();
586         if (cob != null) {
587             cob.setUserDefinedWritability(userDefinedWritabilityIndex, writable);
588         }
589     }
590 
591     /**
592      * Check the writability according to delay and size for the channel.
593      * Set if necessary setUserDefinedWritability status.
594      * @param delay the computed delay
595      * @param queueSize the current queueSize
596      */
597     void checkWriteSuspend(ChannelHandlerContext ctx, long delay, long queueSize) {
598         if (queueSize > maxWriteSize || delay > maxWriteDelay) {
599             setUserDefinedWritability(ctx, false);
600         }
601     }
602     /**
603      * Explicitly release the Write suspended status.
604      */
605     void releaseWriteSuspended(ChannelHandlerContext ctx) {
606         setUserDefinedWritability(ctx, true);
607     }
608 
609     /**
610      * @return the current TrafficCounter (if
611      *         channel is still connected)
612      */
613     public TrafficCounter trafficCounter() {
614         return trafficCounter;
615     }
616 
617     @Override
618     public String toString() {
619         StringBuilder builder = new StringBuilder(290)
620             .append("TrafficShaping with Write Limit: ").append(writeLimit)
621             .append(" Read Limit: ").append(readLimit)
622             .append(" CheckInterval: ").append(checkInterval)
623             .append(" maxDelay: ").append(maxWriteDelay)
624             .append(" maxSize: ").append(maxWriteSize)
625             .append(" and Counter: ");
626         if (trafficCounter != null) {
627             builder.append(trafficCounter);
628         } else {
629             builder.append("none");
630         }
631         return builder.toString();
632     }
633 
634     /**
635      * Calculate the size of the given {@link Object}.
636      *
637      * This implementation supports {@link ByteBuf} and {@link ByteBufHolder}. Sub-classes may override this.
638      * @param msg the msg for which the size should be calculated.
639      * @return size the size of the msg or {@code -1} if unknown.
640      */
641     protected long calculateSize(Object msg) {
642         if (msg instanceof ByteBuf) {
643             return ((ByteBuf) msg).readableBytes();
644         }
645         if (msg instanceof ByteBufHolder) {
646             return ((ByteBufHolder) msg).content().readableBytes();
647         }
648         return -1;
649     }
650 }