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