View Javadoc
1   /*
2    * Copyright 2014 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.util.Recycler;
19  import io.netty.util.ReferenceCountUtil;
20  import io.netty.util.concurrent.PromiseCombiner;
21  import io.netty.util.internal.ObjectUtil;
22  import io.netty.util.internal.SystemPropertyUtil;
23  import io.netty.util.internal.logging.InternalLogger;
24  import io.netty.util.internal.logging.InternalLoggerFactory;
25  
26  /**
27   * A queue of write operations which are pending for later execution. It also updates the
28   * {@linkplain Channel#isWritable() writability} of the associated {@link Channel}, so that
29   * the pending write operations are also considered to determine the writability.
30   */
31  public final class PendingWriteQueue {
32      private static final InternalLogger logger = InternalLoggerFactory.getInstance(PendingWriteQueue.class);
33      // Assuming a 64-bit JVM:
34      //  - 16 bytes object header
35      //  - 4 reference fields
36      //  - 1 long fields
37      private static final int PENDING_WRITE_OVERHEAD =
38              SystemPropertyUtil.getInt("io.netty.transport.pendingWriteSizeOverhead", 64);
39  
40      private final ChannelHandlerContext ctx;
41      private final PendingBytesTracker tracker;
42  
43      // head and tail pointers for the linked-list structure. If empty head and tail are null.
44      private PendingWrite head;
45      private PendingWrite tail;
46      private int size;
47      private long bytes;
48  
49      public PendingWriteQueue(ChannelHandlerContext ctx) {
50          tracker = PendingBytesTracker.newTracker(ctx.channel());
51          this.ctx = ctx;
52      }
53  
54      /**
55       * Returns {@code true} if there are no pending write operations left in this queue.
56       */
57      public boolean isEmpty() {
58          assert ctx.executor().inEventLoop();
59          return head == null;
60      }
61  
62      /**
63       * Returns the number of pending write operations.
64       */
65      public int size() {
66          assert ctx.executor().inEventLoop();
67          return size;
68      }
69  
70      /**
71       * Returns the total number of bytes that are pending because of pending messages. This is only an estimate so
72       * it should only be treated as a hint.
73       */
74      public long bytes() {
75          assert ctx.executor().inEventLoop();
76          return bytes;
77      }
78  
79      private int size(Object msg) {
80          // It is possible for writes to be triggered from removeAndFailAll(). To preserve ordering,
81          // we should add them to the queue and let removeAndFailAll() fail them later.
82          int messageSize = tracker.size(msg);
83          if (messageSize < 0) {
84              // Size may be unknown so just use 0
85              messageSize = 0;
86          }
87          return messageSize + PENDING_WRITE_OVERHEAD;
88      }
89  
90      /**
91       * Add the given {@code msg} and {@link ChannelPromise}.
92       */
93      public void add(Object msg, ChannelPromise promise) {
94          assert ctx.executor().inEventLoop();
95          if (msg == null) {
96              throw new NullPointerException("msg");
97          }
98          if (promise == null) {
99              throw new NullPointerException("promise");
100         }
101         // It is possible for writes to be triggered from removeAndFailAll(). To preserve ordering,
102         // we should add them to the queue and let removeAndFailAll() fail them later.
103         int messageSize = size(msg);
104 
105         PendingWrite write = PendingWrite.newInstance(msg, messageSize, promise);
106         PendingWrite currentTail = tail;
107         if (currentTail == null) {
108             tail = head = write;
109         } else {
110             currentTail.next = write;
111             tail = write;
112         }
113         size ++;
114         bytes += messageSize;
115         tracker.incrementPendingOutboundBytes(write.size);
116     }
117 
118     /**
119      * Remove all pending write operation and performs them via
120      * {@link ChannelHandlerContext#write(Object, ChannelPromise)}.
121      *
122      * @return  {@link ChannelFuture} if something was written and {@code null}
123      *          if the {@link PendingWriteQueue} is empty.
124      */
125     public ChannelFuture removeAndWriteAll() {
126         assert ctx.executor().inEventLoop();
127 
128         if (isEmpty()) {
129             return null;
130         }
131 
132         ChannelPromise p = ctx.newPromise();
133         PromiseCombiner combiner = new PromiseCombiner();
134         try {
135             // It is possible for some of the written promises to trigger more writes. The new writes
136             // will "revive" the queue, so we need to write them up until the queue is empty.
137             for (PendingWrite write = head; write != null; write = head) {
138                 head = tail = null;
139                 size = 0;
140                 bytes = 0;
141 
142                 while (write != null) {
143                     PendingWrite next = write.next;
144                     Object msg = write.msg;
145                     ChannelPromise promise = write.promise;
146                     recycle(write, false);
147                     combiner.add(promise);
148                     ctx.write(msg, promise);
149                     write = next;
150                 }
151             }
152             combiner.finish(p);
153         } catch (Throwable cause) {
154             p.setFailure(cause);
155         }
156         assertEmpty();
157         return p;
158     }
159 
160     /**
161      * Remove all pending write operation and fail them with the given {@link Throwable}. The message will be released
162      * via {@link ReferenceCountUtil#safeRelease(Object)}.
163      */
164     public void removeAndFailAll(Throwable cause) {
165         assert ctx.executor().inEventLoop();
166         if (cause == null) {
167             throw new NullPointerException("cause");
168         }
169         // It is possible for some of the failed promises to trigger more writes. The new writes
170         // will "revive" the queue, so we need to clean them up until the queue is empty.
171         for (PendingWrite write = head; write != null; write = head) {
172             head = tail = null;
173             size = 0;
174             bytes = 0;
175             while (write != null) {
176                 PendingWrite next = write.next;
177                 ReferenceCountUtil.safeRelease(write.msg);
178                 ChannelPromise promise = write.promise;
179                 recycle(write, false);
180                 safeFail(promise, cause);
181                 write = next;
182             }
183         }
184         assertEmpty();
185     }
186 
187     /**
188      * Remove a pending write operation and fail it with the given {@link Throwable}. The message will be released via
189      * {@link ReferenceCountUtil#safeRelease(Object)}.
190      */
191     public void removeAndFail(Throwable cause) {
192         assert ctx.executor().inEventLoop();
193         if (cause == null) {
194             throw new NullPointerException("cause");
195         }
196         PendingWrite write = head;
197 
198         if (write == null) {
199             return;
200         }
201         ReferenceCountUtil.safeRelease(write.msg);
202         ChannelPromise promise = write.promise;
203         safeFail(promise, cause);
204         recycle(write, true);
205     }
206 
207     private void assertEmpty() {
208         assert tail == null && head == null && size == 0;
209     }
210 
211     /**
212      * Removes a pending write operation and performs it via
213      * {@link ChannelHandlerContext#write(Object, ChannelPromise)}.
214      *
215      * @return  {@link ChannelFuture} if something was written and {@code null}
216      *          if the {@link PendingWriteQueue} is empty.
217      */
218     public ChannelFuture removeAndWrite() {
219         assert ctx.executor().inEventLoop();
220         PendingWrite write = head;
221         if (write == null) {
222             return null;
223         }
224         Object msg = write.msg;
225         ChannelPromise promise = write.promise;
226         recycle(write, true);
227         return ctx.write(msg, promise);
228     }
229 
230     /**
231      * Removes a pending write operation and release it's message via {@link ReferenceCountUtil#safeRelease(Object)}.
232      *
233      * @return  {@link ChannelPromise} of the pending write or {@code null} if the queue is empty.
234      *
235      */
236     public ChannelPromise remove() {
237         assert ctx.executor().inEventLoop();
238         PendingWrite write = head;
239         if (write == null) {
240             return null;
241         }
242         ChannelPromise promise = write.promise;
243         ReferenceCountUtil.safeRelease(write.msg);
244         recycle(write, true);
245         return promise;
246     }
247 
248     /**
249      * Return the current message or {@code null} if empty.
250      */
251     public Object current() {
252         assert ctx.executor().inEventLoop();
253         PendingWrite write = head;
254         if (write == null) {
255             return null;
256         }
257         return write.msg;
258     }
259 
260     private void recycle(PendingWrite write, boolean update) {
261         final PendingWrite next = write.next;
262         final long writeSize = write.size;
263 
264         if (update) {
265             if (next == null) {
266                 // Handled last PendingWrite so rest head and tail
267                 // Guard against re-entrance by directly reset
268                 head = tail = null;
269                 size = 0;
270                 bytes = 0;
271             } else {
272                 head = next;
273                 size --;
274                 bytes -= writeSize;
275                 assert size > 0 && bytes >= 0;
276             }
277         }
278 
279         write.recycle();
280         tracker.decrementPendingOutboundBytes(writeSize);
281     }
282 
283     private static void safeFail(ChannelPromise promise, Throwable cause) {
284         if (!(promise instanceof VoidChannelPromise) && !promise.tryFailure(cause)) {
285             logger.warn("Failed to mark a promise as failure because it's done already: {}", promise, cause);
286         }
287     }
288 
289     /**
290      * Holds all meta-data and construct the linked-list structure.
291      */
292     static final class PendingWrite {
293         private static final Recycler<PendingWrite> RECYCLER = new Recycler<PendingWrite>() {
294             @Override
295             protected PendingWrite newObject(Handle<PendingWrite> handle) {
296                 return new PendingWrite(handle);
297             }
298         };
299 
300         private final Recycler.Handle<PendingWrite> handle;
301         private PendingWrite next;
302         private long size;
303         private ChannelPromise promise;
304         private Object msg;
305 
306         private PendingWrite(Recycler.Handle<PendingWrite> handle) {
307             this.handle = handle;
308         }
309 
310         static PendingWrite newInstance(Object msg, int size, ChannelPromise promise) {
311             PendingWrite write = RECYCLER.get();
312             write.size = size;
313             write.msg = msg;
314             write.promise = promise;
315             return write;
316         }
317 
318         private void recycle() {
319             size = 0;
320             next = null;
321             msg = null;
322             promise = null;
323             handle.recycle(this);
324         }
325     }
326 }