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.epoll;
17  
18  import io.netty.buffer.ByteBuf;
19  import io.netty.buffer.CompositeByteBuf;
20  import io.netty.channel.ChannelOutboundBuffer.MessageProcessor;
21  import io.netty.util.internal.PlatformDependent;
22  
23  import java.nio.ByteBuffer;
24  
25  /**
26   * Represent an array of struct array and so can be passed directly over via JNI without the need to do any more
27   * array copies.
28   *
29   * The buffers are written out directly into direct memory to match the struct iov. See also {@code man writev}.
30   *
31   * <pre>
32   * struct iovec {
33   *   void  *iov_base;
34   *   size_t iov_len;
35   * };
36   * </pre>
37   *
38   * See also
39   * <a href="http://rkennke.wordpress.com/2007/07/30/efficient-jni-programming-iv-wrapping-native-data-objects/"
40   * >Efficient JNI programming IV: Wrapping native data objects</a>.
41   */
42  final class IovArray implements MessageProcessor {
43  
44      /** The size of an address which should be 8 for 64 bits and 4 for 32 bits. */
45      private static final int ADDRESS_SIZE = PlatformDependent.addressSize();
46  
47      /**
48       * The size of an {@code iovec} struct in bytes. This is calculated as we have 2 entries each of the size of the
49       * address.
50       */
51      private static final int IOV_SIZE = 2 * ADDRESS_SIZE;
52  
53      /**
54       * The needed memory to hold up to {@link Native#IOV_MAX} iov entries, where {@link Native#IOV_MAX} signified
55       * the maximum number of {@code iovec} structs that can be passed to {@code writev(...)}.
56       */
57      private static final int CAPACITY = Native.IOV_MAX * IOV_SIZE;
58  
59      private final long memoryAddress;
60      private int count;
61      private long size;
62  
63      IovArray() {
64          memoryAddress = PlatformDependent.allocateMemory(CAPACITY);
65      }
66  
67      void clear() {
68          count = 0;
69          size = 0;
70      }
71  
72      /**
73       * Try to add the given {@link ByteBuf}. Returns {@code true} on success,
74       * {@code false} otherwise.
75       */
76      boolean add(ByteBuf buf) {
77          if (count == Native.IOV_MAX) {
78              // No more room!
79              return false;
80          }
81  
82          final int len = buf.readableBytes();
83          if (len == 0) {
84              // No need to add an empty buffer.
85              // We return true here because we want ChannelOutboundBuffer.forEachFlushedMessage() to continue
86              // fetching the next buffers.
87              return true;
88          }
89  
90          final long addr = buf.memoryAddress();
91          final int offset = buf.readerIndex();
92          add(addr, offset, len);
93          return true;
94      }
95  
96      private void add(long addr, int offset, int len) {
97          if (len == 0) {
98              // No need to add an empty buffer.
99              return;
100         }
101 
102         final long baseOffset = memoryAddress(count++);
103         final long lengthOffset = baseOffset + ADDRESS_SIZE;
104 
105         if (ADDRESS_SIZE == 8) {
106             // 64bit
107             PlatformDependent.putLong(baseOffset, addr + offset);
108             PlatformDependent.putLong(lengthOffset, len);
109         } else {
110             assert ADDRESS_SIZE == 4;
111             PlatformDependent.putInt(baseOffset, (int) addr + offset);
112             PlatformDependent.putInt(lengthOffset, len);
113         }
114 
115         size += len;
116     }
117 
118     /**
119      * Try to add the given {@link CompositeByteBuf}. Returns {@code true} on success,
120      * {@code false} otherwise.
121      */
122     boolean add(CompositeByteBuf buf) {
123         ByteBuffer[] buffers = buf.nioBuffers();
124         if (count + buffers.length >= Native.IOV_MAX) {
125             // No more room!
126             return false;
127         }
128         for (int i = 0; i < buffers.length; i++) {
129             ByteBuffer nioBuffer = buffers[i];
130             int offset = nioBuffer.position();
131             int len = nioBuffer.limit() - nioBuffer.position();
132             if (len == 0) {
133                 // No need to add an empty buffer so just continue
134                 continue;
135             }
136             long addr = PlatformDependent.directBufferAddress(nioBuffer);
137 
138             add(addr, offset, len);
139         }
140         return true;
141     }
142 
143     /**
144      * Process the written iov entries. This will return the length of the iov entry on the given index if it is
145      * smaller then the given {@code written} value. Otherwise it returns {@code -1}.
146      */
147     long processWritten(int index, long written) {
148         long baseOffset = memoryAddress(index);
149         long lengthOffset = baseOffset + ADDRESS_SIZE;
150         if (ADDRESS_SIZE == 8) {
151             // 64bit
152             long len = PlatformDependent.getLong(lengthOffset);
153             if (len > written) {
154                 long offset = PlatformDependent.getLong(baseOffset);
155                 PlatformDependent.putLong(baseOffset, offset + written);
156                 PlatformDependent.putLong(lengthOffset, len - written);
157                 return -1;
158             }
159             return len;
160         } else {
161             assert ADDRESS_SIZE == 4;
162             long len = PlatformDependent.getInt(lengthOffset);
163             if (len > written) {
164                 int offset = PlatformDependent.getInt(baseOffset);
165                 PlatformDependent.putInt(baseOffset, (int) (offset + written));
166                 PlatformDependent.putInt(lengthOffset, (int) (len - written));
167                 return -1;
168             }
169             return len;
170         }
171     }
172 
173     /**
174      * Returns the number if iov entries.
175      */
176     int count() {
177         return count;
178     }
179 
180     /**
181      * Returns the size in bytes
182      */
183     long size() {
184         return size;
185     }
186 
187     /**
188      * Returns the {@code memoryAddress} for the given {@code offset}.
189      */
190     long memoryAddress(int offset) {
191         return memoryAddress + IOV_SIZE * offset;
192     }
193 
194     /**
195      * Release the {@link IovArray}. Once release further using of it may crash the JVM!
196      */
197     void release() {
198         PlatformDependent.freeMemory(memoryAddress);
199     }
200 
201     @Override
202     public boolean processMessage(Object msg) throws Exception {
203         if (msg instanceof ByteBuf) {
204             if (msg instanceof CompositeByteBuf) {
205                 return add((CompositeByteBuf) msg);
206             } else {
207                 return add((ByteBuf) msg);
208             }
209         }
210         return false;
211     }
212 }