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                         // The code exists in an extra method to ensure the method is not too big to inline as this
438                         // branch is not very likely to get hit very frequently.
439                         nioBufferCount = nioBuffers(entry, buf, nioBuffers, nioBufferCount, maxCount);
440                     }
441                     if (nioBufferCount == maxCount) {
442                         break;
443                     }
444                 }
445             }
446             entry = entry.next;
447         }
448         this.nioBufferCount = nioBufferCount;
449         this.nioBufferSize = nioBufferSize;
450 
451         return nioBuffers;
452     }
453 
454     private static int nioBuffers(Entry entry, ByteBuf buf, ByteBuffer[] nioBuffers, int nioBufferCount, int maxCount) {
455         ByteBuffer[] nioBufs = entry.bufs;
456         if (nioBufs == null) {
457             // cached ByteBuffers as they may be expensive to create in terms
458             // of Object allocation
459             entry.bufs = nioBufs = buf.nioBuffers();
460         }
461         for (int i = 0; i < nioBufs.length && nioBufferCount < maxCount; ++i) {
462             ByteBuffer nioBuf = nioBufs[i];
463             if (nioBuf == null) {
464                 break;
465             } else if (!nioBuf.hasRemaining()) {
466                 continue;
467             }
468             nioBuffers[nioBufferCount++] = nioBuf;
469         }
470         return nioBufferCount;
471     }
472 
473     private static ByteBuffer[] expandNioBufferArray(ByteBuffer[] array, int neededSpace, int size) {
474         int newCapacity = array.length;
475         do {
476             // double capacity until it is big enough
477             // See https://github.com/netty/netty/issues/1890
478             newCapacity <<= 1;
479 
480             if (newCapacity < 0) {
481                 throw new IllegalStateException();
482             }
483 
484         } while (neededSpace > newCapacity);
485 
486         ByteBuffer[] newArray = new ByteBuffer[newCapacity];
487         System.arraycopy(array, 0, newArray, 0, size);
488 
489         return newArray;
490     }
491 
492     /**
493      * Returns the number of {@link ByteBuffer} that can be written out of the {@link ByteBuffer} array that was
494      * obtained via {@link #nioBuffers()}. This method <strong>MUST</strong> be called after {@link #nioBuffers()}
495      * was called.
496      */
497     public int nioBufferCount() {
498         return nioBufferCount;
499     }
500 
501     /**
502      * Returns the number of bytes that can be written out of the {@link ByteBuffer} array that was
503      * obtained via {@link #nioBuffers()}. This method <strong>MUST</strong> be called after {@link #nioBuffers()}
504      * was called.
505      */
506     public long nioBufferSize() {
507         return nioBufferSize;
508     }
509 
510     /**
511      * Returns {@code true} if and only if {@linkplain #totalPendingWriteBytes() the total number of pending bytes} did
512      * not exceed the write watermark of the {@link Channel} and
513      * no {@linkplain #setUserDefinedWritability(int, boolean) user-defined writability flag} has been set to
514      * {@code false}.
515      */
516     public boolean isWritable() {
517         return unwritable == 0;
518     }
519 
520     /**
521      * Returns {@code true} if and only if the user-defined writability flag at the specified index is set to
522      * {@code true}.
523      */
524     public boolean getUserDefinedWritability(int index) {
525         return (unwritable & writabilityMask(index)) == 0;
526     }
527 
528     /**
529      * Sets a user-defined writability flag at the specified index.
530      */
531     public void setUserDefinedWritability(int index, boolean writable) {
532         if (writable) {
533             setUserDefinedWritability(index);
534         } else {
535             clearUserDefinedWritability(index);
536         }
537     }
538 
539     private void setUserDefinedWritability(int index) {
540         final int mask = ~writabilityMask(index);
541         for (;;) {
542             final int oldValue = unwritable;
543             final int newValue = oldValue & mask;
544             if (UNWRITABLE_UPDATER.compareAndSet(this, oldValue, newValue)) {
545                 if (oldValue != 0 && newValue == 0) {
546                     fireChannelWritabilityChanged(true);
547                 }
548                 break;
549             }
550         }
551     }
552 
553     private void clearUserDefinedWritability(int index) {
554         final int mask = writabilityMask(index);
555         for (;;) {
556             final int oldValue = unwritable;
557             final int newValue = oldValue | mask;
558             if (UNWRITABLE_UPDATER.compareAndSet(this, oldValue, newValue)) {
559                 if (oldValue == 0 && newValue != 0) {
560                     fireChannelWritabilityChanged(true);
561                 }
562                 break;
563             }
564         }
565     }
566 
567     private static int writabilityMask(int index) {
568         if (index < 1 || index > 31) {
569             throw new IllegalArgumentException("index: " + index + " (expected: 1~31)");
570         }
571         return 1 << index;
572     }
573 
574     private void setWritable(boolean invokeLater) {
575         for (;;) {
576             final int oldValue = unwritable;
577             final int newValue = oldValue & ~1;
578             if (UNWRITABLE_UPDATER.compareAndSet(this, oldValue, newValue)) {
579                 if (oldValue != 0 && newValue == 0) {
580                     fireChannelWritabilityChanged(invokeLater);
581                 }
582                 break;
583             }
584         }
585     }
586 
587     private void setUnwritable(boolean invokeLater) {
588         for (;;) {
589             final int oldValue = unwritable;
590             final int newValue = oldValue | 1;
591             if (UNWRITABLE_UPDATER.compareAndSet(this, oldValue, newValue)) {
592                 if (oldValue == 0 && newValue != 0) {
593                     fireChannelWritabilityChanged(invokeLater);
594                 }
595                 break;
596             }
597         }
598     }
599 
600     private void fireChannelWritabilityChanged(boolean invokeLater) {
601         final ChannelPipeline pipeline = channel.pipeline();
602         if (invokeLater) {
603             Runnable task = fireChannelWritabilityChangedTask;
604             if (task == null) {
605                 fireChannelWritabilityChangedTask = task = new Runnable() {
606                     @Override
607                     public void run() {
608                         pipeline.fireChannelWritabilityChanged();
609                     }
610                 };
611             }
612             channel.eventLoop().execute(task);
613         } else {
614             pipeline.fireChannelWritabilityChanged();
615         }
616     }
617 
618     /**
619      * Returns the number of flushed messages in this {@link ChannelOutboundBuffer}.
620      */
621     public int size() {
622         return flushed;
623     }
624 
625     /**
626      * Returns {@code true} if there are flushed messages in this {@link ChannelOutboundBuffer} or {@code false}
627      * otherwise.
628      */
629     public boolean isEmpty() {
630         return flushed == 0;
631     }
632 
633     void failFlushed(Throwable cause, boolean notify) {
634         // Make sure that this method does not reenter.  A listener added to the current promise can be notified by the
635         // current thread in the tryFailure() call of the loop below, and the listener can trigger another fail() call
636         // indirectly (usually by closing the channel.)
637         //
638         // See https://github.com/netty/netty/issues/1501
639         if (inFail) {
640             return;
641         }
642 
643         try {
644             inFail = true;
645             for (;;) {
646                 if (!remove0(cause, notify)) {
647                     break;
648                 }
649             }
650         } finally {
651             inFail = false;
652         }
653     }
654 
655     void close(final Throwable cause, final boolean allowChannelOpen) {
656         if (inFail) {
657             channel.eventLoop().execute(new Runnable() {
658                 @Override
659                 public void run() {
660                     close(cause, allowChannelOpen);
661                 }
662             });
663             return;
664         }
665 
666         inFail = true;
667 
668         if (!allowChannelOpen && channel.isOpen()) {
669             throw new IllegalStateException("close() must be invoked after the channel is closed.");
670         }
671 
672         if (!isEmpty()) {
673             throw new IllegalStateException("close() must be invoked after all flushed writes are handled.");
674         }
675 
676         // Release all unflushed messages.
677         try {
678             Entry e = unflushedEntry;
679             while (e != null) {
680                 // Just decrease; do not trigger any events via decrementPendingOutboundBytes()
681                 int size = e.pendingSize;
682                 TOTAL_PENDING_SIZE_UPDATER.addAndGet(this, -size);
683 
684                 if (!e.cancelled) {
685                     ReferenceCountUtil.safeRelease(e.msg);
686                     safeFail(e.promise, cause);
687                 }
688                 e = e.recycleAndGetNext();
689             }
690         } finally {
691             inFail = false;
692         }
693         clearNioBuffers();
694     }
695 
696     void close(ClosedChannelException cause) {
697         close(cause, false);
698     }
699 
700     private static void safeSuccess(ChannelPromise promise) {
701         // Only log if the given promise is not of type VoidChannelPromise as trySuccess(...) is expected to return
702         // false.
703         PromiseNotificationUtil.trySuccess(promise, null, promise instanceof VoidChannelPromise ? null : logger);
704     }
705 
706     private static void safeFail(ChannelPromise promise, Throwable cause) {
707         // Only log if the given promise is not of type VoidChannelPromise as tryFailure(...) is expected to return
708         // false.
709         PromiseNotificationUtil.tryFailure(promise, cause, promise instanceof VoidChannelPromise ? null : logger);
710     }
711 
712     @Deprecated
713     public void recycle() {
714         // NOOP
715     }
716 
717     public long totalPendingWriteBytes() {
718         return totalPendingSize;
719     }
720 
721     /**
722      * Get how many bytes can be written until {@link #isWritable()} returns {@code false}.
723      * This quantity will always be non-negative. If {@link #isWritable()} is {@code false} then 0.
724      */
725     public long bytesBeforeUnwritable() {
726         long bytes = channel.config().getWriteBufferHighWaterMark() - totalPendingSize;
727         // If bytes is negative we know we are not writable, but if bytes is non-negative we have to check writability.
728         // Note that totalPendingSize and isWritable() use different volatile variables that are not synchronized
729         // together. totalPendingSize will be updated before isWritable().
730         if (bytes > 0) {
731             return isWritable() ? bytes : 0;
732         }
733         return 0;
734     }
735 
736     /**
737      * Get how many bytes must be drained from the underlying buffer until {@link #isWritable()} returns {@code true}.
738      * This quantity will always be non-negative. If {@link #isWritable()} is {@code true} then 0.
739      */
740     public long bytesBeforeWritable() {
741         long bytes = totalPendingSize - channel.config().getWriteBufferLowWaterMark();
742         // If bytes is negative we know we are writable, but if bytes is non-negative we have to check writability.
743         // Note that totalPendingSize and isWritable() use different volatile variables that are not synchronized
744         // together. totalPendingSize will be updated before isWritable().
745         if (bytes > 0) {
746             return isWritable() ? 0 : bytes;
747         }
748         return 0;
749     }
750 
751     /**
752      * Call {@link MessageProcessor#processMessage(Object)} for each flushed message
753      * in this {@link ChannelOutboundBuffer} until {@link MessageProcessor#processMessage(Object)}
754      * returns {@code false} or there are no more flushed messages to process.
755      */
756     public void forEachFlushedMessage(MessageProcessor processor) throws Exception {
757         if (processor == null) {
758             throw new NullPointerException("processor");
759         }
760 
761         Entry entry = flushedEntry;
762         if (entry == null) {
763             return;
764         }
765 
766         do {
767             if (!entry.cancelled) {
768                 if (!processor.processMessage(entry.msg)) {
769                     return;
770                 }
771             }
772             entry = entry.next;
773         } while (isFlushedEntry(entry));
774     }
775 
776     private boolean isFlushedEntry(Entry e) {
777         return e != null && e != unflushedEntry;
778     }
779 
780     public interface MessageProcessor {
781         /**
782          * Will be called for each flushed message until it either there are no more flushed messages or this
783          * method returns {@code false}.
784          */
785         boolean processMessage(Object msg) throws Exception;
786     }
787 
788     static final class Entry {
789         private static final Recycler<Entry> RECYCLER = new Recycler<Entry>() {
790             @Override
791             protected Entry newObject(Handle<Entry> handle) {
792                 return new Entry(handle);
793             }
794         };
795 
796         private final Handle<Entry> handle;
797         Entry next;
798         Object msg;
799         ByteBuffer[] bufs;
800         ByteBuffer buf;
801         ChannelPromise promise;
802         long progress;
803         long total;
804         int pendingSize;
805         int count = -1;
806         boolean cancelled;
807 
808         private Entry(Handle<Entry> handle) {
809             this.handle = handle;
810         }
811 
812         static Entry newInstance(Object msg, int size, long total, ChannelPromise promise) {
813             Entry entry = RECYCLER.get();
814             entry.msg = msg;
815             entry.pendingSize = size + CHANNEL_OUTBOUND_BUFFER_ENTRY_OVERHEAD;
816             entry.total = total;
817             entry.promise = promise;
818             return entry;
819         }
820 
821         int cancel() {
822             if (!cancelled) {
823                 cancelled = true;
824                 int pSize = pendingSize;
825 
826                 // release message and replace with an empty buffer
827                 ReferenceCountUtil.safeRelease(msg);
828                 msg = Unpooled.EMPTY_BUFFER;
829 
830                 pendingSize = 0;
831                 total = 0;
832                 progress = 0;
833                 bufs = null;
834                 buf = null;
835                 return pSize;
836             }
837             return 0;
838         }
839 
840         void recycle() {
841             next = null;
842             bufs = null;
843             buf = null;
844             msg = null;
845             promise = null;
846             progress = 0;
847             total = 0;
848             pendingSize = 0;
849             count = -1;
850             cancelled = false;
851             handle.recycle(this);
852         }
853 
854         Entry recycleAndGetNext() {
855             Entry next = this.next;
856             recycle();
857             return next;
858         }
859     }
860 }