View Javadoc
1   /*
2    * Copyright 2013 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.channel;
17  
18  import io.netty.buffer.ByteBuf;
19  import io.netty.buffer.ByteBufHolder;
20  import io.netty.buffer.Unpooled;
21  import io.netty.channel.socket.nio.NioSocketChannel;
22  import io.netty.util.Recycler;
23  import io.netty.util.Recycler.Handle;
24  import io.netty.util.ReferenceCountUtil;
25  import io.netty.util.concurrent.FastThreadLocal;
26  import io.netty.util.internal.InternalThreadLocalMap;
27  import io.netty.util.internal.PromiseNotificationUtil;
28  import io.netty.util.internal.SystemPropertyUtil;
29  import io.netty.util.internal.logging.InternalLogger;
30  import io.netty.util.internal.logging.InternalLoggerFactory;
31  
32  import java.nio.ByteBuffer;
33  import java.nio.channels.ClosedChannelException;
34  import java.util.Arrays;
35  import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
36  import java.util.concurrent.atomic.AtomicLongFieldUpdater;
37  
38  import static java.lang.Math.min;
39  
40  /**
41   * (Transport implementors only) an internal data structure used by {@link AbstractChannel} to store its pending
42   * outbound write requests.
43   * <p>
44   * All methods must be called by a transport implementation from an I/O thread, except the following ones:
45   * <ul>
46   * <li>{@link #size()} and {@link #isEmpty()}</li>
47   * <li>{@link #isWritable()}</li>
48   * <li>{@link #getUserDefinedWritability(int)} and {@link #setUserDefinedWritability(int, boolean)}</li>
49   * </ul>
50   * </p>
51   */
52  public final class ChannelOutboundBuffer {
53      // Assuming a 64-bit JVM:
54      //  - 16 bytes object header
55      //  - 8 reference fields
56      //  - 2 long fields
57      //  - 2 int fields
58      //  - 1 boolean field
59      //  - padding
60      static final int CHANNEL_OUTBOUND_BUFFER_ENTRY_OVERHEAD =
61              SystemPropertyUtil.getInt("io.netty.transport.outboundBufferEntrySizeOverhead", 96);
62  
63      private static final InternalLogger logger = InternalLoggerFactory.getInstance(ChannelOutboundBuffer.class);
64  
65      private static final FastThreadLocal<ByteBuffer[]> NIO_BUFFERS = new FastThreadLocal<ByteBuffer[]>() {
66          @Override
67          protected ByteBuffer[] initialValue() throws Exception {
68              return new ByteBuffer[1024];
69          }
70      };
71  
72      private final Channel channel;
73  
74      // Entry(flushedEntry) --> ... Entry(unflushedEntry) --> ... Entry(tailEntry)
75      //
76      // The Entry that is the first in the linked-list structure that was flushed
77      private Entry flushedEntry;
78      // The Entry which is the first unflushed in the linked-list structure
79      private Entry unflushedEntry;
80      // The Entry which represents the tail of the buffer
81      private Entry tailEntry;
82      // The number of flushed entries that are not written yet
83      private int flushed;
84  
85      private int nioBufferCount;
86      private long nioBufferSize;
87  
88      private boolean inFail;
89  
90      private static final AtomicLongFieldUpdater<ChannelOutboundBuffer> TOTAL_PENDING_SIZE_UPDATER =
91              AtomicLongFieldUpdater.newUpdater(ChannelOutboundBuffer.class, "totalPendingSize");
92  
93      @SuppressWarnings("UnusedDeclaration")
94      private volatile long totalPendingSize;
95  
96      private static final AtomicIntegerFieldUpdater<ChannelOutboundBuffer> UNWRITABLE_UPDATER =
97              AtomicIntegerFieldUpdater.newUpdater(ChannelOutboundBuffer.class, "unwritable");
98  
99      @SuppressWarnings("UnusedDeclaration")
100     private volatile int unwritable;
101 
102     private volatile Runnable fireChannelWritabilityChangedTask;
103 
104     ChannelOutboundBuffer(AbstractChannel channel) {
105         this.channel = channel;
106     }
107 
108     /**
109      * Add given message to this {@link ChannelOutboundBuffer}. The given {@link ChannelPromise} will be notified once
110      * the message was written.
111      */
112     public void addMessage(Object msg, int size, ChannelPromise promise) {
113         Entry entry = Entry.newInstance(msg, size, total(msg), promise);
114         if (tailEntry == null) {
115             flushedEntry = null;
116         } else {
117             Entry tail = tailEntry;
118             tail.next = entry;
119         }
120         tailEntry = entry;
121         if (unflushedEntry == null) {
122             unflushedEntry = entry;
123         }
124 
125         // increment pending bytes after adding message to the unflushed arrays.
126         // See https://github.com/netty/netty/issues/1619
127         incrementPendingOutboundBytes(entry.pendingSize, false);
128     }
129 
130     /**
131      * Add a flush to this {@link ChannelOutboundBuffer}. This means all previous added messages are marked as flushed
132      * and so you will be able to handle them.
133      */
134     public void addFlush() {
135         // There is no need to process all entries if there was already a flush before and no new messages
136         // where added in the meantime.
137         //
138         // See https://github.com/netty/netty/issues/2577
139         Entry entry = unflushedEntry;
140         if (entry != null) {
141             if (flushedEntry == null) {
142                 // there is no flushedEntry yet, so start with the entry
143                 flushedEntry = entry;
144             }
145             do {
146                 flushed ++;
147                 if (!entry.promise.setUncancellable()) {
148                     // Was cancelled so make sure we free up memory and notify about the freed bytes
149                     int pending = entry.cancel();
150                     decrementPendingOutboundBytes(pending, false, true);
151                 }
152                 entry = entry.next;
153             } while (entry != null);
154 
155             // All flushed so reset unflushedEntry
156             unflushedEntry = null;
157         }
158     }
159 
160     /**
161      * Increment the pending bytes which will be written at some point.
162      * This method is thread-safe!
163      */
164     void incrementPendingOutboundBytes(long size) {
165         incrementPendingOutboundBytes(size, true);
166     }
167 
168     private void incrementPendingOutboundBytes(long size, boolean invokeLater) {
169         if (size == 0) {
170             return;
171         }
172 
173         long newWriteBufferSize = TOTAL_PENDING_SIZE_UPDATER.addAndGet(this, size);
174         if (newWriteBufferSize > channel.config().getWriteBufferHighWaterMark()) {
175             setUnwritable(invokeLater);
176         }
177     }
178 
179     /**
180      * Decrement the pending bytes which will be written at some point.
181      * This method is thread-safe!
182      */
183     void decrementPendingOutboundBytes(long size) {
184         decrementPendingOutboundBytes(size, true, true);
185     }
186 
187     private void decrementPendingOutboundBytes(long size, boolean invokeLater, boolean notifyWritability) {
188         if (size == 0) {
189             return;
190         }
191 
192         long newWriteBufferSize = TOTAL_PENDING_SIZE_UPDATER.addAndGet(this, -size);
193         if (notifyWritability && newWriteBufferSize < channel.config().getWriteBufferLowWaterMark()) {
194             setWritable(invokeLater);
195         }
196     }
197 
198     private static long total(Object msg) {
199         if (msg instanceof ByteBuf) {
200             return ((ByteBuf) msg).readableBytes();
201         }
202         if (msg instanceof FileRegion) {
203             return ((FileRegion) msg).count();
204         }
205         if (msg instanceof ByteBufHolder) {
206             return ((ByteBufHolder) msg).content().readableBytes();
207         }
208         return -1;
209     }
210 
211     /**
212      * Return the current message to write or {@code null} if nothing was flushed before and so is ready to be written.
213      */
214     public Object current() {
215         Entry entry = flushedEntry;
216         if (entry == null) {
217             return null;
218         }
219 
220         return entry.msg;
221     }
222 
223     /**
224      * Notify the {@link ChannelPromise} of the current message about writing progress.
225      */
226     public void progress(long amount) {
227         Entry e = flushedEntry;
228         assert e != null;
229         ChannelPromise p = e.promise;
230         if (p instanceof ChannelProgressivePromise) {
231             long progress = e.progress + amount;
232             e.progress = progress;
233             ((ChannelProgressivePromise) p).tryProgress(progress, e.total);
234         }
235     }
236 
237     /**
238      * Will remove the current message, mark its {@link ChannelPromise} as success and return {@code true}. If no
239      * flushed message exists at the time this method is called it will return {@code false} to signal that no more
240      * messages are ready to be handled.
241      */
242     public boolean remove() {
243         Entry e = flushedEntry;
244         if (e == null) {
245             clearNioBuffers();
246             return false;
247         }
248         Object msg = e.msg;
249 
250         ChannelPromise promise = e.promise;
251         int size = e.pendingSize;
252 
253         removeEntry(e);
254 
255         if (!e.cancelled) {
256             // only release message, notify and decrement if it was not canceled before.
257             ReferenceCountUtil.safeRelease(msg);
258             safeSuccess(promise);
259             decrementPendingOutboundBytes(size, false, true);
260         }
261 
262         // recycle the entry
263         e.recycle();
264 
265         return true;
266     }
267 
268     /**
269      * Will remove the current message, mark its {@link ChannelPromise} as failure using the given {@link Throwable}
270      * and return {@code true}. If no   flushed message exists at the time this method is called it will return
271      * {@code false} to signal that no more messages are ready to be handled.
272      */
273     public boolean remove(Throwable cause) {
274         return remove0(cause, true);
275     }
276 
277     private boolean remove0(Throwable cause, boolean notifyWritability) {
278         Entry e = flushedEntry;
279         if (e == null) {
280             clearNioBuffers();
281             return false;
282         }
283         Object msg = e.msg;
284 
285         ChannelPromise promise = e.promise;
286         int size = e.pendingSize;
287 
288         removeEntry(e);
289 
290         if (!e.cancelled) {
291             // only release message, fail and decrement if it was not canceled before.
292             ReferenceCountUtil.safeRelease(msg);
293 
294             safeFail(promise, cause);
295             decrementPendingOutboundBytes(size, false, notifyWritability);
296         }
297 
298         // recycle the entry
299         e.recycle();
300 
301         return true;
302     }
303 
304     private void removeEntry(Entry e) {
305         if (-- flushed == 0) {
306             // processed everything
307             flushedEntry = null;
308             if (e == tailEntry) {
309                 tailEntry = null;
310                 unflushedEntry = null;
311             }
312         } else {
313             flushedEntry = e.next;
314         }
315     }
316 
317     /**
318      * Removes the fully written entries and update the reader index of the partially written entry.
319      * This operation assumes all messages in this buffer is {@link ByteBuf}.
320      */
321     public void removeBytes(long writtenBytes) {
322         for (;;) {
323             Object msg = current();
324             if (!(msg instanceof ByteBuf)) {
325                 assert writtenBytes == 0;
326                 break;
327             }
328 
329             final ByteBuf buf = (ByteBuf) msg;
330             final int readerIndex = buf.readerIndex();
331             final int readableBytes = buf.writerIndex() - readerIndex;
332 
333             if (readableBytes <= writtenBytes) {
334                 if (writtenBytes != 0) {
335                     progress(readableBytes);
336                     writtenBytes -= readableBytes;
337                 }
338                 remove();
339             } else { // readableBytes > writtenBytes
340                 if (writtenBytes != 0) {
341                     buf.readerIndex(readerIndex + (int) writtenBytes);
342                     progress(writtenBytes);
343                 }
344                 break;
345             }
346         }
347         clearNioBuffers();
348     }
349 
350     // Clear all ByteBuffer from the array so these can be GC'ed.
351     // See https://github.com/netty/netty/issues/3837
352     private void clearNioBuffers() {
353         int count = nioBufferCount;
354         if (count > 0) {
355             nioBufferCount = 0;
356             Arrays.fill(NIO_BUFFERS.get(), 0, count, null);
357         }
358     }
359 
360     /**
361      * Returns an array of direct NIO buffers if the currently pending messages are made of {@link ByteBuf} only.
362      * {@link #nioBufferCount()} and {@link #nioBufferSize()} will return the number of NIO buffers in the returned
363      * array and the total number of readable bytes of the NIO buffers respectively.
364      * <p>
365      * Note that the returned array is reused and thus should not escape
366      * {@link AbstractChannel#doWrite(ChannelOutboundBuffer)}.
367      * Refer to {@link NioSocketChannel#doWrite(ChannelOutboundBuffer)} for an example.
368      * </p>
369      */
370     public ByteBuffer[] nioBuffers() {
371         return nioBuffers(Integer.MAX_VALUE, Integer.MAX_VALUE);
372     }
373 
374     /**
375      * Returns an array of direct NIO buffers if the currently pending messages are made of {@link ByteBuf} only.
376      * {@link #nioBufferCount()} and {@link #nioBufferSize()} will return the number of NIO buffers in the returned
377      * array and the total number of readable bytes of the NIO buffers respectively.
378      * <p>
379      * Note that the returned array is reused and thus should not escape
380      * {@link AbstractChannel#doWrite(ChannelOutboundBuffer)}.
381      * Refer to {@link NioSocketChannel#doWrite(ChannelOutboundBuffer)} for an example.
382      * </p>
383      * @param maxCount The maximum amount of buffers that will be added to the return value.
384      * @param maxBytes A hint toward the maximum number of bytes to include as part of the return value. Note that this
385      *                 value maybe exceeded because we make a best effort to include at least 1 {@link ByteBuffer}
386      *                 in the return value to ensure write progress is made.
387      */
388     public ByteBuffer[] nioBuffers(int maxCount, long maxBytes) {
389         assert maxCount > 0;
390         assert maxBytes > 0;
391         long nioBufferSize = 0;
392         int nioBufferCount = 0;
393         final InternalThreadLocalMap threadLocalMap = InternalThreadLocalMap.get();
394         ByteBuffer[] nioBuffers = NIO_BUFFERS.get(threadLocalMap);
395         Entry entry = flushedEntry;
396         while (isFlushedEntry(entry) && entry.msg instanceof ByteBuf) {
397             if (!entry.cancelled) {
398                 ByteBuf buf = (ByteBuf) entry.msg;
399                 final int readerIndex = buf.readerIndex();
400                 final int readableBytes = buf.writerIndex() - readerIndex;
401 
402                 if (readableBytes > 0) {
403                     if (maxBytes - readableBytes < nioBufferSize && nioBufferCount != 0) {
404                         // If the nioBufferSize + readableBytes will overflow maxBytes, and there is at least one entry
405                         // we stop populate the ByteBuffer array. This is done for 2 reasons:
406                         // 1. bsd/osx don't allow to write more bytes then Integer.MAX_VALUE with one writev(...) call
407                         // and so will return 'EINVAL', which will raise an IOException. On Linux it may work depending
408                         // on the architecture and kernel but to be safe we also enforce the limit here.
409                         // 2. There is no sense in putting more data in the array than is likely to be accepted by the
410                         // OS.
411                         //
412                         // See also:
413                         // - https://www.freebsd.org/cgi/man.cgi?query=write&sektion=2
414                         // - http://linux.die.net/man/2/writev
415                         break;
416                     }
417                     nioBufferSize += readableBytes;
418                     int count = entry.count;
419                     if (count == -1) {
420                         //noinspection ConstantValueVariableUse
421                         entry.count = count = buf.nioBufferCount();
422                     }
423                     int neededSpace = min(maxCount, nioBufferCount + count);
424                     if (neededSpace > nioBuffers.length) {
425                         nioBuffers = expandNioBufferArray(nioBuffers, neededSpace, nioBufferCount);
426                         NIO_BUFFERS.set(threadLocalMap, nioBuffers);
427                     }
428                     if (count == 1) {
429                         ByteBuffer nioBuf = entry.buf;
430                         if (nioBuf == null) {
431                             // cache ByteBuffer as it may need to create a new ByteBuffer instance if its a
432                             // derived buffer
433                             entry.buf = nioBuf = buf.internalNioBuffer(readerIndex, readableBytes);
434                         }
435                         nioBuffers[nioBufferCount++] = nioBuf;
436                     } else {
437                         ByteBuffer[] nioBufs = entry.bufs;
438                         if (nioBufs == null) {
439                             // cached ByteBuffers as they may be expensive to create in terms
440                             // of Object allocation
441                             entry.bufs = nioBufs = buf.nioBuffers();
442                         }
443                         for (int i = 0; i < nioBufs.length && nioBufferCount < maxCount; ++i) {
444                             ByteBuffer nioBuf = nioBufs[i];
445                             if (nioBuf == null) {
446                                 break;
447                             } else if (!nioBuf.hasRemaining()) {
448                                 continue;
449                             }
450                             nioBuffers[nioBufferCount++] = nioBuf;
451                         }
452                     }
453                     if (nioBufferCount == maxCount) {
454                         break;
455                     }
456                 }
457             }
458             entry = entry.next;
459         }
460         this.nioBufferCount = nioBufferCount;
461         this.nioBufferSize = nioBufferSize;
462 
463         return nioBuffers;
464     }
465 
466     private static ByteBuffer[] expandNioBufferArray(ByteBuffer[] array, int neededSpace, int size) {
467         int newCapacity = array.length;
468         do {
469             // double capacity until it is big enough
470             // See https://github.com/netty/netty/issues/1890
471             newCapacity <<= 1;
472 
473             if (newCapacity < 0) {
474                 throw new IllegalStateException();
475             }
476 
477         } while (neededSpace > newCapacity);
478 
479         ByteBuffer[] newArray = new ByteBuffer[newCapacity];
480         System.arraycopy(array, 0, newArray, 0, size);
481 
482         return newArray;
483     }
484 
485     /**
486      * Returns the number of {@link ByteBuffer} that can be written out of the {@link ByteBuffer} array that was
487      * obtained via {@link #nioBuffers()}. This method <strong>MUST</strong> be called after {@link #nioBuffers()}
488      * was called.
489      */
490     public int nioBufferCount() {
491         return nioBufferCount;
492     }
493 
494     /**
495      * Returns the number of bytes that can be written out of the {@link ByteBuffer} array that was
496      * obtained via {@link #nioBuffers()}. This method <strong>MUST</strong> be called after {@link #nioBuffers()}
497      * was called.
498      */
499     public long nioBufferSize() {
500         return nioBufferSize;
501     }
502 
503     /**
504      * Returns {@code true} if and only if {@linkplain #totalPendingWriteBytes() the total number of pending bytes} did
505      * not exceed the write watermark of the {@link Channel} and
506      * no {@linkplain #setUserDefinedWritability(int, boolean) user-defined writability flag} has been set to
507      * {@code false}.
508      */
509     public boolean isWritable() {
510         return unwritable == 0;
511     }
512 
513     /**
514      * Returns {@code true} if and only if the user-defined writability flag at the specified index is set to
515      * {@code true}.
516      */
517     public boolean getUserDefinedWritability(int index) {
518         return (unwritable & writabilityMask(index)) == 0;
519     }
520 
521     /**
522      * Sets a user-defined writability flag at the specified index.
523      */
524     public void setUserDefinedWritability(int index, boolean writable) {
525         if (writable) {
526             setUserDefinedWritability(index);
527         } else {
528             clearUserDefinedWritability(index);
529         }
530     }
531 
532     private void setUserDefinedWritability(int index) {
533         final int mask = ~writabilityMask(index);
534         for (;;) {
535             final int oldValue = unwritable;
536             final int newValue = oldValue & mask;
537             if (UNWRITABLE_UPDATER.compareAndSet(this, oldValue, newValue)) {
538                 if (oldValue != 0 && newValue == 0) {
539                     fireChannelWritabilityChanged(true);
540                 }
541                 break;
542             }
543         }
544     }
545 
546     private void clearUserDefinedWritability(int index) {
547         final int mask = writabilityMask(index);
548         for (;;) {
549             final int oldValue = unwritable;
550             final int newValue = oldValue | mask;
551             if (UNWRITABLE_UPDATER.compareAndSet(this, oldValue, newValue)) {
552                 if (oldValue == 0 && newValue != 0) {
553                     fireChannelWritabilityChanged(true);
554                 }
555                 break;
556             }
557         }
558     }
559 
560     private static int writabilityMask(int index) {
561         if (index < 1 || index > 31) {
562             throw new IllegalArgumentException("index: " + index + " (expected: 1~31)");
563         }
564         return 1 << index;
565     }
566 
567     private void setWritable(boolean invokeLater) {
568         for (;;) {
569             final int oldValue = unwritable;
570             final int newValue = oldValue & ~1;
571             if (UNWRITABLE_UPDATER.compareAndSet(this, oldValue, newValue)) {
572                 if (oldValue != 0 && newValue == 0) {
573                     fireChannelWritabilityChanged(invokeLater);
574                 }
575                 break;
576             }
577         }
578     }
579 
580     private void setUnwritable(boolean invokeLater) {
581         for (;;) {
582             final int oldValue = unwritable;
583             final int newValue = oldValue | 1;
584             if (UNWRITABLE_UPDATER.compareAndSet(this, oldValue, newValue)) {
585                 if (oldValue == 0 && newValue != 0) {
586                     fireChannelWritabilityChanged(invokeLater);
587                 }
588                 break;
589             }
590         }
591     }
592 
593     private void fireChannelWritabilityChanged(boolean invokeLater) {
594         final ChannelPipeline pipeline = channel.pipeline();
595         if (invokeLater) {
596             Runnable task = fireChannelWritabilityChangedTask;
597             if (task == null) {
598                 fireChannelWritabilityChangedTask = task = new Runnable() {
599                     @Override
600                     public void run() {
601                         pipeline.fireChannelWritabilityChanged();
602                     }
603                 };
604             }
605             channel.eventLoop().execute(task);
606         } else {
607             pipeline.fireChannelWritabilityChanged();
608         }
609     }
610 
611     /**
612      * Returns the number of flushed messages in this {@link ChannelOutboundBuffer}.
613      */
614     public int size() {
615         return flushed;
616     }
617 
618     /**
619      * Returns {@code true} if there are flushed messages in this {@link ChannelOutboundBuffer} or {@code false}
620      * otherwise.
621      */
622     public boolean isEmpty() {
623         return flushed == 0;
624     }
625 
626     void failFlushed(Throwable cause, boolean notify) {
627         // Make sure that this method does not reenter.  A listener added to the current promise can be notified by the
628         // current thread in the tryFailure() call of the loop below, and the listener can trigger another fail() call
629         // indirectly (usually by closing the channel.)
630         //
631         // See https://github.com/netty/netty/issues/1501
632         if (inFail) {
633             return;
634         }
635 
636         try {
637             inFail = true;
638             for (;;) {
639                 if (!remove0(cause, notify)) {
640                     break;
641                 }
642             }
643         } finally {
644             inFail = false;
645         }
646     }
647 
648     void close(final Throwable cause, final boolean allowChannelOpen) {
649         if (inFail) {
650             channel.eventLoop().execute(new Runnable() {
651                 @Override
652                 public void run() {
653                     close(cause, allowChannelOpen);
654                 }
655             });
656             return;
657         }
658 
659         inFail = true;
660 
661         if (!allowChannelOpen && channel.isOpen()) {
662             throw new IllegalStateException("close() must be invoked after the channel is closed.");
663         }
664 
665         if (!isEmpty()) {
666             throw new IllegalStateException("close() must be invoked after all flushed writes are handled.");
667         }
668 
669         // Release all unflushed messages.
670         try {
671             Entry e = unflushedEntry;
672             while (e != null) {
673                 // Just decrease; do not trigger any events via decrementPendingOutboundBytes()
674                 int size = e.pendingSize;
675                 TOTAL_PENDING_SIZE_UPDATER.addAndGet(this, -size);
676 
677                 if (!e.cancelled) {
678                     ReferenceCountUtil.safeRelease(e.msg);
679                     safeFail(e.promise, cause);
680                 }
681                 e = e.recycleAndGetNext();
682             }
683         } finally {
684             inFail = false;
685         }
686         clearNioBuffers();
687     }
688 
689     void close(ClosedChannelException cause) {
690         close(cause, false);
691     }
692 
693     private static void safeSuccess(ChannelPromise promise) {
694         // Only log if the given promise is not of type VoidChannelPromise as trySuccess(...) is expected to return
695         // false.
696         PromiseNotificationUtil.trySuccess(promise, null, promise instanceof VoidChannelPromise ? null : logger);
697     }
698 
699     private static void safeFail(ChannelPromise promise, Throwable cause) {
700         // Only log if the given promise is not of type VoidChannelPromise as tryFailure(...) is expected to return
701         // false.
702         PromiseNotificationUtil.tryFailure(promise, cause, promise instanceof VoidChannelPromise ? null : logger);
703     }
704 
705     @Deprecated
706     public void recycle() {
707         // NOOP
708     }
709 
710     public long totalPendingWriteBytes() {
711         return totalPendingSize;
712     }
713 
714     /**
715      * Get how many bytes can be written until {@link #isWritable()} returns {@code false}.
716      * This quantity will always be non-negative. If {@link #isWritable()} is {@code false} then 0.
717      */
718     public long bytesBeforeUnwritable() {
719         long bytes = channel.config().getWriteBufferHighWaterMark() - totalPendingSize;
720         // If bytes is negative we know we are not writable, but if bytes is non-negative we have to check writability.
721         // Note that totalPendingSize and isWritable() use different volatile variables that are not synchronized
722         // together. totalPendingSize will be updated before isWritable().
723         if (bytes > 0) {
724             return isWritable() ? bytes : 0;
725         }
726         return 0;
727     }
728 
729     /**
730      * Get how many bytes must be drained from the underlying buffer until {@link #isWritable()} returns {@code true}.
731      * This quantity will always be non-negative. If {@link #isWritable()} is {@code true} then 0.
732      */
733     public long bytesBeforeWritable() {
734         long bytes = totalPendingSize - channel.config().getWriteBufferLowWaterMark();
735         // If bytes is negative we know we are writable, but if bytes is non-negative we have to check writability.
736         // Note that totalPendingSize and isWritable() use different volatile variables that are not synchronized
737         // together. totalPendingSize will be updated before isWritable().
738         if (bytes > 0) {
739             return isWritable() ? 0 : bytes;
740         }
741         return 0;
742     }
743 
744     /**
745      * Call {@link MessageProcessor#processMessage(Object)} for each flushed message
746      * in this {@link ChannelOutboundBuffer} until {@link MessageProcessor#processMessage(Object)}
747      * returns {@code false} or there are no more flushed messages to process.
748      */
749     public void forEachFlushedMessage(MessageProcessor processor) throws Exception {
750         if (processor == null) {
751             throw new NullPointerException("processor");
752         }
753 
754         Entry entry = flushedEntry;
755         if (entry == null) {
756             return;
757         }
758 
759         do {
760             if (!entry.cancelled) {
761                 if (!processor.processMessage(entry.msg)) {
762                     return;
763                 }
764             }
765             entry = entry.next;
766         } while (isFlushedEntry(entry));
767     }
768 
769     private boolean isFlushedEntry(Entry e) {
770         return e != null && e != unflushedEntry;
771     }
772 
773     public interface MessageProcessor {
774         /**
775          * Will be called for each flushed message until it either there are no more flushed messages or this
776          * method returns {@code false}.
777          */
778         boolean processMessage(Object msg) throws Exception;
779     }
780 
781     static final class Entry {
782         private static final Recycler<Entry> RECYCLER = new Recycler<Entry>() {
783             @Override
784             protected Entry newObject(Handle<Entry> handle) {
785                 return new Entry(handle);
786             }
787         };
788 
789         private final Handle<Entry> handle;
790         Entry next;
791         Object msg;
792         ByteBuffer[] bufs;
793         ByteBuffer buf;
794         ChannelPromise promise;
795         long progress;
796         long total;
797         int pendingSize;
798         int count = -1;
799         boolean cancelled;
800 
801         private Entry(Handle<Entry> handle) {
802             this.handle = handle;
803         }
804 
805         static Entry newInstance(Object msg, int size, long total, ChannelPromise promise) {
806             Entry entry = RECYCLER.get();
807             entry.msg = msg;
808             entry.pendingSize = size + CHANNEL_OUTBOUND_BUFFER_ENTRY_OVERHEAD;
809             entry.total = total;
810             entry.promise = promise;
811             return entry;
812         }
813 
814         int cancel() {
815             if (!cancelled) {
816                 cancelled = true;
817                 int pSize = pendingSize;
818 
819                 // release message and replace with an empty buffer
820                 ReferenceCountUtil.safeRelease(msg);
821                 msg = Unpooled.EMPTY_BUFFER;
822 
823                 pendingSize = 0;
824                 total = 0;
825                 progress = 0;
826                 bufs = null;
827                 buf = null;
828                 return pSize;
829             }
830             return 0;
831         }
832 
833         void recycle() {
834             next = null;
835             bufs = null;
836             buf = null;
837             msg = null;
838             promise = null;
839             progress = 0;
840             total = 0;
841             pendingSize = 0;
842             count = -1;
843             cancelled = false;
844             handle.recycle(this);
845         }
846 
847         Entry recycleAndGetNext() {
848             Entry next = this.next;
849             recycle();
850             return next;
851         }
852     }
853 }