Repository: netty/netty
Issue: Support gathering/scattering in EpollDatagramChannel
Labels: feature

As @pcarrier pointed out on monday it is possible to do gathering writes even when using datagram/udp on linux. We should add support for it:

See http://man7.org/linux/man-pages/man2/send.2.html


Issue comments:
Added to 4.0, 4.1 and master 


Pull request review comments:

Patch that resolved the issue:
diff --git a/transport-native-epoll/src/main/c/io_netty_channel_epoll_Native.c b/transport-native-epoll/src/main/c/io_netty_channel_epoll_Native.c
index 52be419d943..5dc7eb5825d 100644
--- a/transport-native-epoll/src/main/c/io_netty_channel_epoll_Native.c
+++ b/transport-native-epoll/src/main/c/io_netty_channel_epoll_Native.c
@@ -34,6 +34,15 @@
 // optional
 extern int accept4(int sockFd, struct sockaddr *addr, socklen_t *addrlen, int flags) __attribute__((weak));
 extern int epoll_create1(int flags) __attribute__((weak));
+extern int sendmmsg(int sockfd, struct mmsghdr *msgvec, unsigned int vlen, unsigned int flags) __attribute__((weak));
+
+// Just define it here and NOT use #define _GNU_SOURCE as we also want to be able to build on systems that not support
+// sendmmsg yet. The problem is if we use _GNU_SOURCE we will not be able to declare sendmmsg as extern
+struct mmsghdr {
+    struct msghdr msg_hdr;  /* Message header */
+    unsigned int  msg_len;  /* Number of bytes transmitted */
+};
+
 
 // Those are initialized in the init(...) method and cached for performance reasons
 jmethodID updatePosId = NULL;
@@ -44,7 +53,14 @@ jfieldID limitFieldId = NULL;
 jfieldID fileChannelFieldId = NULL;
 jfieldID transferedFieldId = NULL;
 jfieldID fdFieldId = NULL;
-jfieldID fileDescriptorFieldId = NULL;;
+jfieldID fileDescriptorFieldId = NULL;
+
+jfieldID packetAddrFieldId = NULL;
+jfieldID packetScopeIdFieldId = NULL;
+jfieldID packetPortFieldId = NULL;
+jfieldID packetMemoryAddressFieldId = NULL;
+jfieldID packetCountFieldId = NULL;
+
 jmethodID inetSocketAddrMethodId = NULL;
 jmethodID datagramSocketAddrMethodId = NULL;
 jclass runtimeExceptionClass = NULL;
@@ -53,6 +69,7 @@ jclass closedChannelExceptionClass = NULL;
 jmethodID closedChannelExceptionMethodId = NULL;
 jclass inetSocketAddressClass = NULL;
 jclass datagramSocketAddressClass = NULL;
+jclass nativeDatagramPacketClass = NULL;
 
 static int socketType;
 static const char *ip4prefix = "::ffff:";
@@ -414,6 +431,38 @@ jint JNI_OnLoad(JavaVM* vm, void* reserved) {
             throwRuntimeException(env, "Unable to obtain constructor of DatagramSocketAddress");
             return JNI_ERR;
         }
+        jclass nativeDatagramPacketCls = (*env)->FindClass(env, "io/netty/channel/epoll/NativeDatagramPacketArray$NativeDatagramPacket");
+        if (nativeDatagramPacketCls == NULL) {
+            // pending exception...
+            return JNI_ERR;
+        }
+
+        packetAddrFieldId = (*env)->GetFieldID(env, nativeDatagramPacketCls, "addr", "[B");
+        if (packetAddrFieldId == NULL) {
+            throwRuntimeException(env, "Unable to obtain addr field for NativeDatagramPacket");
+            return JNI_ERR;
+        }
+        packetScopeIdFieldId = (*env)->GetFieldID(env, nativeDatagramPacketCls, "scopeId", "I");
+        if (packetScopeIdFieldId == NULL) {
+            throwRuntimeException(env, "Unable to obtain scopeId field for NativeDatagramPacket");
+            return JNI_ERR;
+        }
+        packetPortFieldId = (*env)->GetFieldID(env, nativeDatagramPacketCls, "port", "I");
+        if (packetPortFieldId == NULL) {
+            throwRuntimeException(env, "Unable to obtain port field for NativeDatagramPacket");
+            return JNI_ERR;
+        }
+        packetMemoryAddressFieldId = (*env)->GetFieldID(env, nativeDatagramPacketCls, "memoryAddress", "J");
+        if (packetMemoryAddressFieldId == NULL) {
+            throwRuntimeException(env, "Unable to obtain memoryAddress field for NativeDatagramPacket");
+            return JNI_ERR;
+        }
+
+        packetCountFieldId = (*env)->GetFieldID(env, nativeDatagramPacketCls, "count", "I");
+        if (packetCountFieldId == NULL) {
+            throwRuntimeException(env, "Unable to obtain count field for NativeDatagramPacket");
+            return JNI_ERR;
+        }
         return JNI_VERSION_1_6;
     }
 }
@@ -655,6 +704,88 @@ JNIEXPORT jint JNICALL Java_io_netty_channel_epoll_Native_sendToAddress(JNIEnv *
     return sendTo0(env, fd, (void*) memoryAddress, pos, limit, address, scopeId, port);
 }
 
+JNIEXPORT jint JNICALL Java_io_netty_channel_epoll_Native_sendToAddresses(JNIEnv * env, jclass clazz, jint fd, jlong memoryAddress, jint length, jbyteArray address, jint scopeId, jint port) {
+    struct sockaddr_storage addr;
+
+    if (init_sockaddr(env, address, scopeId, port, &addr) == -1) {
+        return -1;
+    }
+
+    struct msghdr m;
+    m.msg_name = (void*) &addr;
+    m.msg_namelen = (socklen_t) sizeof(struct sockaddr_storage);
+    m.msg_iov = (struct iovec *) memoryAddress;
+    m.msg_iovlen = length;
+
+    ssize_t res;
+    int err;
+    do {
+       res = sendmsg(fd, &m, 0);
+       // keep on writing if it was interrupted
+    } while(res == -1 && ((err = errno) == EINTR));
+
+    if (res < 0) {
+        // network stack saturated... try again later
+        if (err == EAGAIN || err == EWOULDBLOCK) {
+            return 0;
+        }
+        if (err == EBADF) {
+            throwClosedChannelException(env);
+            return -1;
+        }
+        throwIOException(env, exceptionMessage("Error while sendto(...): ", err));
+        return -1;
+    }
+    return (jint) res;
+}
+
+JNIEXPORT jint JNICALL Java_io_netty_channel_epoll_Native_sendmmsg(JNIEnv * env, jclass clazz, jint fd, jobjectArray packets, jint offset, jint len) {
+    struct mmsghdr msg[len];
+    int i;
+
+    memset(msg, 0, sizeof(msg));
+
+    for (i = 0; i < len; i++) {
+        struct sockaddr_storage addr;
+
+        jobject packet = (*env)->GetObjectArrayElement(env, packets, i + offset);
+        jbyteArray address = (jbyteArray) (*env)->GetObjectField(env, packet, packetAddrFieldId);
+        jint scopeId = (*env)->GetIntField(env, packet, packetScopeIdFieldId);
+        jint port = (*env)->GetIntField(env, packet, packetPortFieldId);
+
+        if (init_sockaddr(env, address, scopeId, port, &addr) == -1) {
+            return -1;
+        }
+
+        msg[i].msg_hdr.msg_name = &addr;
+        msg[i].msg_hdr.msg_namelen = sizeof(addr);
+
+        msg[i].msg_hdr.msg_iov = (struct iovec *) (*env)->GetLongField(env, packet, packetMemoryAddressFieldId);
+        msg[i].msg_hdr.msg_iovlen = (*env)->GetIntField(env, packet, packetCountFieldId);;
+    }
+
+    ssize_t res;
+    int err;
+    do {
+       res = sendmmsg(fd, msg, len, 0);
+       // keep on writing if it was interrupted
+    } while(res == -1 && ((err = errno) == EINTR));
+
+    if (res < 0) {
+        // network stack saturated... try again later
+        if (err == EAGAIN || err == EWOULDBLOCK) {
+            return 0;
+        }
+        if (err == EBADF) {
+            throwClosedChannelException(env);
+            return -1;
+        }
+        throwIOException(env, exceptionMessage("Error while sendmmsg(...): ", err));
+        return -1;
+    }
+    return (jint) res;
+}
+
 jobject recvFrom0(JNIEnv * env, jint fd, void* buffer, jint pos, jint limit) {
     struct sockaddr_storage addr;
     socklen_t addrlen = sizeof(addr);
@@ -1191,3 +1322,16 @@ JNIEXPORT jstring JNICALL Java_io_netty_channel_epoll_Native_kernelVersion(JNIEn
 JNIEXPORT jint JNICALL Java_io_netty_channel_epoll_Native_iovMax(JNIEnv *env, jclass clazz) {
     return IOV_MAX;
 }
+
+JNIEXPORT jint JNICALL Java_io_netty_channel_epoll_Native_uioMaxIov(JNIEnv *env, jclass clazz) {
+    return UIO_MAXIOV;
+}
+
+
+JNIEXPORT jboolean JNICALL Java_io_netty_channel_epoll_Native_isSupportingSendmmsg(JNIEnv *env, jclass clazz) {
+    if (sendmmsg) {
+        return JNI_TRUE;
+    }
+    return JNI_FALSE;
+}
+
diff --git a/transport-native-epoll/src/main/c/io_netty_channel_epoll_Native.h b/transport-native-epoll/src/main/c/io_netty_channel_epoll_Native.h
index ac0677f97b5..a6e9fb04aae 100644
--- a/transport-native-epoll/src/main/c/io_netty_channel_epoll_Native.h
+++ b/transport-native-epoll/src/main/c/io_netty_channel_epoll_Native.h
@@ -33,6 +33,11 @@
 #define IOV_MAX 1024
 #endif /* IOV_MAX */
 
+// Define UIO_MAXIOV if not found
+#ifndef UIO_MAXIOV
+#define UIO_MAXIOV 1024
+#endif /* UIO_MAXIOV */
+
 jint Java_io_netty_channel_epoll_Native_eventFd(JNIEnv * env, jclass clazz);
 void Java_io_netty_channel_epoll_Native_eventFdWrite(JNIEnv * env, jclass clazz, jint fd, jlong value);
 void Java_io_netty_channel_epoll_Native_eventFdRead(JNIEnv * env, jclass clazz, jint fd);
@@ -47,6 +52,8 @@ jlong Java_io_netty_channel_epoll_Native_writev(JNIEnv * env, jclass clazz, jint
 jlong Java_io_netty_channel_epoll_Native_writevAddresses(JNIEnv * env, jclass clazz, jint fd, jlong memoryAddress, jint length);
 jint Java_io_netty_channel_epoll_Native_sendTo(JNIEnv * env, jclass clazz, jint fd, jobject jbuffer, jint pos, jint limit, jbyteArray address, jint scopeId, jint port);
 jint Java_io_netty_channel_epoll_Native_sendToAddress(JNIEnv * env, jclass clazz, jint fd, jlong memoryAddress, jint pos, jint limit, jbyteArray address, jint scopeId, jint port);
+jint Java_io_netty_channel_epoll_Native_sendToAddresses(JNIEnv * env, jclass clazz, jint fd, jlong memoryAddress, jint length, jbyteArray address, jint scopeId, jint port);
+jint Java_io_netty_channel_epoll_Native_sendmmsg(JNIEnv * env, jclass clazz, jint fd, jobjectArray packets, jint offset, jint len);
 
 jint Java_io_netty_channel_epoll_Native_read(JNIEnv * env, jclass clazz, jint fd, jobject jbuffer, jint pos, jint limit);
 jint Java_io_netty_channel_epoll_Native_readAddress(JNIEnv * env, jclass clazz, jint fd, jlong address, jint pos, jint limit);
@@ -94,3 +101,5 @@ jint Java_io_netty_channel_epoll_Native_getTcpKeepCnt(JNIEnv *env, jclass clazz,
 
 jstring Java_io_netty_channel_epoll_Native_kernelVersion(JNIEnv *env, jclass clazz);
 jint Java_io_netty_channel_epoll_Native_iovMax(JNIEnv *env, jclass clazz);
+jint Java_io_netty_channel_epoll_Native_uioMaxIov(JNIEnv *env, jclass clazz);
+jboolean Java_io_netty_channel_epoll_Native_isSupportingSendmmsg(JNIEnv *env, jclass clazz);
diff --git a/transport-native-epoll/src/main/java/io/netty/channel/epoll/EpollDatagramChannel.java b/transport-native-epoll/src/main/java/io/netty/channel/epoll/EpollDatagramChannel.java
index 26cc442a73f..75d34150965 100644
--- a/transport-native-epoll/src/main/java/io/netty/channel/epoll/EpollDatagramChannel.java
+++ b/transport-native-epoll/src/main/java/io/netty/channel/epoll/EpollDatagramChannel.java
@@ -16,6 +16,7 @@
 package io.netty.channel.epoll;
 
 import io.netty.buffer.ByteBuf;
+import io.netty.buffer.CompositeByteBuf;
 import io.netty.channel.AddressedEnvelope;
 import io.netty.channel.ChannelFuture;
 import io.netty.channel.ChannelMetadata;
@@ -28,6 +29,7 @@
 import io.netty.channel.socket.DatagramChannel;
 import io.netty.channel.socket.DatagramChannelConfig;
 import io.netty.channel.socket.DatagramPacket;
+import io.netty.util.internal.PlatformDependent;
 import io.netty.util.internal.StringUtil;
 
 import java.io.IOException;
@@ -264,6 +266,32 @@ protected void doWrite(ChannelOutboundBuffer in) throws Exception {
             }
 
             try {
+                // Check if sendmmsg(...) is supported which is only the case for GLIBC 2.14+
+                if (Native.IS_SUPPORTING_SENDMMSG && in.size() > 1) {
+                    NativeDatagramPacketArray array = NativeDatagramPacketArray.getInstance(in);
+                    int cnt = array.count();
+
+                    if (cnt >= 1) {
+                        // Try to use gathering writes via sendmmsg(...) syscall.
+                        int offset = 0;
+                        NativeDatagramPacketArray.NativeDatagramPacket[] packets = array.packets();
+
+                        while (cnt > 0) {
+                            int send = Native.sendmmsg(fd, packets, offset, cnt);
+                            if (send == 0) {
+                                // Did not write all messages.
+                                setEpollOut();
+                                return;
+                            }
+                            for (int i = 0; i < send; i++) {
+                                in.remove();
+                            }
+                            cnt -= send;
+                            offset += send;
+                        }
+                        continue;
+                    }
+                }
                 boolean done = false;
                 for (int i = config().getWriteSpinCount() - 1; i >= 0; i--) {
                     if (doWriteMessage(msg)) {
@@ -288,7 +316,7 @@ protected void doWrite(ChannelOutboundBuffer in) throws Exception {
         }
     }
 
-    private boolean doWriteMessage(Object msg) throws IOException {
+    private boolean doWriteMessage(Object msg) throws Exception {
         final ByteBuf data;
         InetSocketAddress remoteAddress;
         if (msg instanceof AddressedEnvelope) {
@@ -319,6 +347,13 @@ private boolean doWriteMessage(Object msg) throws IOException {
             long memoryAddress = data.memoryAddress();
             writtenBytes = Native.sendToAddress(fd, memoryAddress, data.readerIndex(), data.writerIndex(),
                     remoteAddress.getAddress(), remoteAddress.getPort());
+        } else if (data instanceof CompositeByteBuf) {
+            IovArray array = IovArrayThreadLocal.get((CompositeByteBuf) data);
+            int cnt = array.count();
+            assert cnt != 0;
+
+            writtenBytes = Native.sendToAddresses(fd, array.memoryAddress(0),
+                    cnt, remoteAddress.getAddress(), remoteAddress.getPort());
         } else  {
             ByteBuffer nioData = data.internalNioBuffer(data.readerIndex(), data.readableBytes());
             writtenBytes = Native.sendTo(fd, nioData, nioData.position(), nioData.limit(),
@@ -344,13 +379,24 @@ protected Object filterOutboundMessage(Object msg) {
 
         if (msg instanceof ByteBuf) {
             ByteBuf buf = (ByteBuf) msg;
-            if (buf.hasMemoryAddress()) {
-                return msg;
+            if (!buf.hasMemoryAddress() && (PlatformDependent.hasUnsafe() || !buf.isDirect())) {
+                if (buf instanceof CompositeByteBuf) {
+                    // Special handling of CompositeByteBuf to reduce memory copies if some of the Components
+                    // in the CompositeByteBuf are backed by a memoryAddress.
+                    CompositeByteBuf comp = (CompositeByteBuf) buf;
+                    if (!comp.isDirect() || comp.nioBufferCount() > Native.IOV_MAX) {
+                        // more then 1024 buffers for gathering writes so just do a memory copy.
+                        buf = newDirectBuffer(buf);
+                        assert buf.hasMemoryAddress();
+                    }
+                } else {
+                    // We can only handle buffers with memory address so we need to copy if a non direct is
+                    // passed to write.
+                    buf = newDirectBuffer(buf);
+                    assert buf.hasMemoryAddress();
+                }
             }
-
-            // We can only handle direct buffers so we need to copy if a non direct is
-            // passed to write.
-            return newDirectBuffer(buf);
+            return buf;
         }
 
         if (msg instanceof AddressedEnvelope) {
@@ -363,7 +409,14 @@ protected Object filterOutboundMessage(Object msg) {
                 if (content.hasMemoryAddress()) {
                     return e;
                 }
-
+                if (content instanceof CompositeByteBuf) {
+                    // Special handling of CompositeByteBuf to reduce memory copies if some of the Components
+                    // in the CompositeByteBuf are backed by a memoryAddress.
+                    CompositeByteBuf comp = (CompositeByteBuf) content;
+                    if (comp.isDirect() && comp.nioBufferCount() <= Native.IOV_MAX) {
+                        return e;
+                    }
+                }
                 // We can only handle direct buffers so we need to copy if a non direct is
                 // passed to write.
                 return new DefaultAddressedEnvelope<ByteBuf, InetSocketAddress>(
diff --git a/transport-native-epoll/src/main/java/io/netty/channel/epoll/EpollSocketChannel.java b/transport-native-epoll/src/main/java/io/netty/channel/epoll/EpollSocketChannel.java
index d38bef175e3..975178ac8bc 100644
--- a/transport-native-epoll/src/main/java/io/netty/channel/epoll/EpollSocketChannel.java
+++ b/transport-native-epoll/src/main/java/io/netty/channel/epoll/EpollSocketChannel.java
@@ -358,7 +358,7 @@ private boolean doWriteSingle(ChannelOutboundBuffer in) throws Exception {
     private boolean doWriteMultiple(ChannelOutboundBuffer in) throws Exception {
         if (PlatformDependent.hasUnsafe()) {
             // this means we can cast to IovArray and write the IovArray directly.
-            IovArray array = IovArray.get(in);
+            IovArray array = IovArrayThreadLocal.get(in);
             int cnt = array.count();
             if (cnt >= 1) {
                 // TODO: Handle the case where cnt == 1 specially.
diff --git a/transport-native-epoll/src/main/java/io/netty/channel/epoll/IovArray.java b/transport-native-epoll/src/main/java/io/netty/channel/epoll/IovArray.java
index e685d9760b4..e489837f868 100644
--- a/transport-native-epoll/src/main/java/io/netty/channel/epoll/IovArray.java
+++ b/transport-native-epoll/src/main/java/io/netty/channel/epoll/IovArray.java
@@ -17,9 +17,7 @@
 
 import io.netty.buffer.ByteBuf;
 import io.netty.buffer.CompositeByteBuf;
-import io.netty.channel.ChannelOutboundBuffer;
 import io.netty.channel.ChannelOutboundBuffer.MessageProcessor;
-import io.netty.util.concurrent.FastThreadLocal;
 import io.netty.util.internal.PlatformDependent;
 
 import java.nio.ByteBuffer;
@@ -52,37 +50,30 @@ final class IovArray implements MessageProcessor {
      */
     private static final int IOV_SIZE = 2 * ADDRESS_SIZE;
 
-    /** The needed memory to hold up to {@link Native#IOV_MAX} iov entries, where {@link Native#IOV_MAX} signified
+    /**
+     * The needed memory to hold up to {@link Native#IOV_MAX} iov entries, where {@link Native#IOV_MAX} signified
      * the maximum number of {@code iovec} structs that can be passed to {@code writev(...)}.
      */
     private static final int CAPACITY = Native.IOV_MAX * IOV_SIZE;
 
-    private static final FastThreadLocal<IovArray> ARRAY = new FastThreadLocal<IovArray>() {
-        @Override
-        protected IovArray initialValue() throws Exception {
-            return new IovArray();
-        }
-
-        @Override
-        protected void onRemoval(IovArray value) throws Exception {
-            // free the direct memory now
-            PlatformDependent.freeMemory(value.memoryAddress);
-        }
-    };
-
     private final long memoryAddress;
     private int count;
     private long size;
 
-    private IovArray() {
+    IovArray() {
         memoryAddress = PlatformDependent.allocateMemory(CAPACITY);
     }
 
+    void clear() {
+        count = 0;
+        size = 0;
+    }
+
     /**
      * Try to add the given {@link ByteBuf}. Returns {@code true} on success,
      * {@code false} otherwise.
      */
-    private boolean add(ByteBuf buf) {
+    boolean add(ByteBuf buf) {
         if (count == Native.IOV_MAX) {
             // No more room!
             return false;
@@ -124,7 +115,11 @@ private void add(long addr, int offset, int len) {
         size += len;
     }
 
-    private boolean add(CompositeByteBuf buf) {
+    /**
+     * Try to add the given {@link CompositeByteBuf}. Returns {@code true} on success,
+     * {@code false} otherwise.
+     */
+    boolean add(CompositeByteBuf buf) {
         ByteBuffer[] buffers = buf.nioBuffers();
         if (count + buffers.length >= Native.IOV_MAX) {
             // No more room!
@@ -196,6 +191,13 @@ long memoryAddress(int offset) {
         return memoryAddress + IOV_SIZE * offset;
     }
 
+    /**
+     * Release the {@link IovArray}. Once release further using of it may crash the JVM!
+     */
+    void release() {
+        PlatformDependent.freeMemory(memoryAddress);
+    }
+
     @Override
     public boolean processMessage(Object msg) throws Exception {
         if (msg instanceof ByteBuf) {
@@ -207,15 +209,4 @@ public boolean processMessage(Object msg) throws Exception {
         }
         return false;
     }
-
-    /**
-     * Returns a {@link IovArray} which is filled with the flushed messages of {@link ChannelOutboundBuffer}.
-     */
-    static IovArray get(ChannelOutboundBuffer buffer) throws Exception {
-        IovArray array = ARRAY.get();
-        array.size = 0;
-        array.count = 0;
-        buffer.forEachFlushedMessage(array);
-        return array;
-    }
 }
diff --git a/transport-native-epoll/src/main/java/io/netty/channel/epoll/IovArrayThreadLocal.java b/transport-native-epoll/src/main/java/io/netty/channel/epoll/IovArrayThreadLocal.java
new file mode 100644
index 00000000000..b7f66dcb42b
--- /dev/null
+++ b/transport-native-epoll/src/main/java/io/netty/channel/epoll/IovArrayThreadLocal.java
@@ -0,0 +1,61 @@
+/*
+ * Copyright 2014 The Netty Project
+ *
+ * The Netty Project licenses this file to you under the Apache License,
+ * version 2.0 (the "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at:
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations
+ * under the License.
+ */
+package io.netty.channel.epoll;
+
+import io.netty.buffer.CompositeByteBuf;
+import io.netty.channel.ChannelOutboundBuffer;
+import io.netty.util.concurrent.FastThreadLocal;
+
+/**
+ * Allow to obtain {@link IovArray} instances.
+ */
+final class IovArrayThreadLocal {
+
+    private static final FastThreadLocal<IovArray> ARRAY = new FastThreadLocal<IovArray>() {
+        @Override
+        protected IovArray initialValue() throws Exception {
+            return new IovArray();
+        }
+
+        @Override
+        protected void onRemoval(IovArray value) throws Exception {
+            // free the direct memory now
+            value.release();
+        }
+    };
+
+    /**
+     * Returns a {@link IovArray} which is filled with the flushed messages of {@link ChannelOutboundBuffer}.
+     */
+    static IovArray get(ChannelOutboundBuffer buffer) throws Exception {
+        IovArray array = ARRAY.get();
+        array.clear();
+        buffer.forEachFlushedMessage(array);
+        return array;
+    }
+
+    /**
+     * Returns a {@link IovArray} which is filled with the {@link CompositeByteBuf}.
+     */
+    static IovArray get(CompositeByteBuf buf) throws Exception {
+        IovArray array = ARRAY.get();
+        array.clear();
+        array.add(buf);
+        return array;
+    }
+
+    private IovArrayThreadLocal() { }
+}
diff --git a/transport-native-epoll/src/main/java/io/netty/channel/epoll/Native.java b/transport-native-epoll/src/main/java/io/netty/channel/epoll/Native.java
index e06f12477b9..5bbd890692b 100644
--- a/transport-native-epoll/src/main/java/io/netty/channel/epoll/Native.java
+++ b/transport-native-epoll/src/main/java/io/netty/channel/epoll/Native.java
@@ -52,6 +52,8 @@ final class Native {
     public static final int EPOLLACCEPT = 0x04;
     public static final int EPOLLRDHUP = 0x08;
     public static final int IOV_MAX = iovMax();
+    public static final int UIO_MAX_IOV = uioMaxIov();
+    public static final boolean IS_SUPPORTING_SENDMMSG = isSupportingSendmmsg();
 
     public static native int eventFd();
     public static native void eventFdWrite(int fd, long value);
@@ -118,12 +120,37 @@ public static int sendToAddress(
     private static native int sendToAddress(
             int fd, long memoryAddress, int pos, int limit, byte[] address, int scopeId, int port) throws IOException;
 
+    public static int sendToAddresses(
+            int fd, long memoryAddress, int length, InetAddress addr, int port) throws IOException {
+        // just duplicate the toNativeInetAddress code here to minimize object creation as this method is expected
+        // to be called frequently
+        byte[] address;
+        int scopeId;
+        if (addr instanceof Inet6Address) {
+            address = addr.getAddress();
+            scopeId = ((Inet6Address) addr).getScopeId();
+        } else {
+            // convert to ipv4 mapped ipv6 address;
+            scopeId = 0;
+            address = ipv4MappedIpv6Address(addr.getAddress());
+        }
+        return sendToAddresses(fd, memoryAddress, length, address, scopeId, port);
+    }
+
+    private static native int sendToAddresses(
+            int fd, long memoryAddress, int length, byte[] address, int scopeId, int port) throws IOException;
+
     public static native EpollDatagramChannel.DatagramSocketAddress recvFrom(
             int fd, ByteBuffer buf, int pos, int limit) throws IOException;
 
     public static native EpollDatagramChannel.DatagramSocketAddress recvFromAddress(
             int fd, long memoryAddress, int pos, int limit) throws IOException;
 
+    public static native int sendmmsg(
+            int fd, NativeDatagramPacketArray.NativeDatagramPacket[] msgs, int offset, int len) throws IOException;
+
+    private static native boolean isSupportingSendmmsg();
+
     // socket operations
     public static int socketStreamFd() {
         try {
@@ -148,7 +175,7 @@ public static void bind(int fd, InetAddress addr, int port) throws IOException {
         bind(fd, address.address, address.scopeId, port);
     }
 
-    private static byte[] ipv4MappedIpv6Address(byte[] ipv4) {
+    static byte[] ipv4MappedIpv6Address(byte[] ipv4) {
         byte[] address = new byte[16];
         System.arraycopy(IPV4_MAPPED_IPV6_PREFIX, 0, address, 0, IPV4_MAPPED_IPV6_PREFIX.length);
         System.arraycopy(ipv4, 0, address, 12, ipv4.length);
@@ -226,6 +253,8 @@ private static class NativeInetAddress {
 
     private static native int iovMax();
 
+    private static native int uioMaxIov();
+
     private Native() {
         // utility
     }
diff --git a/transport-native-epoll/src/main/java/io/netty/channel/epoll/NativeDatagramPacketArray.java b/transport-native-epoll/src/main/java/io/netty/channel/epoll/NativeDatagramPacketArray.java
new file mode 100644
index 00000000000..2861e51032d
--- /dev/null
+++ b/transport-native-epoll/src/main/java/io/netty/channel/epoll/NativeDatagramPacketArray.java
@@ -0,0 +1,158 @@
+/*
+ * Copyright 2014 The Netty Project
+ *
+ * The Netty Project licenses this file to you under the Apache License,
+ * version 2.0 (the "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at:
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations
+ * under the License.
+ */
+package io.netty.channel.epoll;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.channel.ChannelOutboundBuffer;
+import io.netty.channel.socket.DatagramPacket;
+import io.netty.util.concurrent.FastThreadLocal;
+
+import java.net.Inet6Address;
+import java.net.InetAddress;
+import java.net.InetSocketAddress;
+
+/**
+ * Support <a href="http://linux.die.net/man/2/sendmmsg">sendmmsg(...)</a> on linux with GLIBC 2.14+
+ */
+final class NativeDatagramPacketArray implements ChannelOutboundBuffer.MessageProcessor {
+
+    private static final FastThreadLocal<NativeDatagramPacketArray> ARRAY =
+            new FastThreadLocal<NativeDatagramPacketArray>() {
+                @Override
+                protected NativeDatagramPacketArray initialValue() throws Exception {
+                    return new NativeDatagramPacketArray();
+                }
+
+                @Override
+                protected void onRemoval(NativeDatagramPacketArray value) throws Exception {
+                    NativeDatagramPacket[] array = value.packets;
+                    // Release all packets
+                    for (int i = 0; i < array.length; i++) {
+                        array[i].release();
+                    }
+                }
+            };
+
+    // Use UIO_MAX_IOV as this is the maximum number we can write with one sendmmsg(...) call.
+    private final NativeDatagramPacket[] packets = new NativeDatagramPacket[Native.UIO_MAX_IOV];
+    private int count;
+
+    private NativeDatagramPacketArray() {
+        for (int i = 0; i < packets.length; i++) {
+            packets[i] = new NativeDatagramPacket();
+        }
+    }
+
+    /**
+     * Try to add the given {@link DatagramPacket}. Returns {@code true} on success,
+     * {@code false} otherwise.
+     */
+    boolean add(DatagramPacket packet) {
+        if (count == packets.length) {
+            return false;
+        }
+        ByteBuf content = packet.content();
+        int len = content.readableBytes();
+        if (len == 0) {
+            return true;
+        }
+        NativeDatagramPacket p = packets[count];
+        InetSocketAddress recipient = packet.recipient();
+        if (!p.init(content, recipient)) {
+            return false;
+        }
+
+        count++;
+        return true;
+    }
+
+    @Override
+    public boolean processMessage(Object msg) throws Exception {
+        return msg instanceof DatagramPacket && add((DatagramPacket) msg);
+    }
+
+    /**
+     * Returns the count
+     */
+    int count() {
+        return count;
+    }
+
+    /**
+     * Returns an array with {@link #count()} {@link NativeDatagramPacket}s filled.
+     */
+    NativeDatagramPacket[] packets() {
+        return packets;
+    }
+
+    /**
+     * Returns a {@link NativeDatagramPacketArray} which is filled with the flushed messages of
+     * {@link ChannelOutboundBuffer}.
+     */
+    static NativeDatagramPacketArray getInstance(ChannelOutboundBuffer buffer) throws Exception {
+        NativeDatagramPacketArray array = ARRAY.get();
+        array.count = 0;
+        buffer.forEachFlushedMessage(array);
+        return array;
+    }
+
+    /**
+     * Used to pass needed data to JNI.
+     */
+    @SuppressWarnings("unused")
+    static final class NativeDatagramPacket {
+        // Each NativeDatagramPackets holds a IovArray which is used for gathering writes.
+        // This is ok as NativeDatagramPacketArray is always obtained via a FastThreadLocal and
+        // so the memory needed is quite small anyway.
+        private final IovArray array = new IovArray();
+
+        // This is the actual struct iovec*
+        private long memoryAddress;
+        private int count;
+
+        private byte[] addr;
+        private int scopeId;
+        private int port;
+
+        private void release() {
+            array.release();
+        }
+
+        /**
+         * Init this instance and return {@code true} if the init was successful.
+         */
+        private boolean init(ByteBuf buf, InetSocketAddress recipient) {
+            array.clear();
+            if (!array.add(buf)) {
+                return false;
+            }
+            // always start from offset 0
+            memoryAddress = array.memoryAddress(0);
+            count = array.count();
+
+            InetAddress address = recipient.getAddress();
+            if (address instanceof Inet6Address) {
+                addr = address.getAddress();
+                scopeId = ((Inet6Address) address).getScopeId();
+            } else {
+                addr = Native.ipv4MappedIpv6Address(address.getAddress());
+                scopeId = 0;
+            }
+            port = recipient.getPort();
+            return true;
+        }
+    }
+}


Test patch:
diff --git a/testsuite/src/test/java/io/netty/testsuite/transport/socket/DatagramUnicastTest.java b/testsuite/src/test/java/io/netty/testsuite/transport/socket/DatagramUnicastTest.java
index 0ab73a96136..07dd9ded4ca 100644
--- a/testsuite/src/test/java/io/netty/testsuite/transport/socket/DatagramUnicastTest.java
+++ b/testsuite/src/test/java/io/netty/testsuite/transport/socket/DatagramUnicastTest.java
@@ -16,6 +16,8 @@
 package io.netty.testsuite.transport.socket;
 
 import io.netty.bootstrap.Bootstrap;
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.CompositeByteBuf;
 import io.netty.buffer.Unpooled;
 import io.netty.channel.Channel;
 import io.netty.channel.ChannelHandlerContext;
@@ -32,36 +34,74 @@
 public class DatagramUnicastTest extends AbstractDatagramTest {
 
     @Test
-    public void testSimpleSend() throws Throwable {
+    public void testSimpleSendDirectByteBuf() throws Throwable {
         run();
     }
 
-    public void testSimpleSend(Bootstrap sb, Bootstrap cb) throws Throwable {
-        final CountDownLatch latch = new CountDownLatch(1);
+    public void testSimpleSendDirectByteBuf(Bootstrap sb, Bootstrap cb) throws Throwable {
+        testSimpleSend0(sb, cb, Unpooled.directBuffer(), true, 1);
+        testSimpleSend0(sb, cb, Unpooled.directBuffer(), true, 4);
+    }
 
-        sb.handler(new SimpleChannelInboundHandler<DatagramPacket>() {
-            @Override
-            public void channelRead0(ChannelHandlerContext ctx, DatagramPacket msg) throws Exception {
-                assertEquals(1, msg.content().readInt());
-                latch.countDown();
-            }
-        });
+    @Test
+    public void testSimpleSendHeapByteBuf() throws Throwable {
+        run();
+    }
 
-        cb.handler(new SimpleChannelInboundHandler<Object>() {
-            @Override
-            public void channelRead0(ChannelHandlerContext ctx, Object msgs) throws Exception {
-                // Nothing will be sent.
-            }
-        });
+    public void testSimpleSendHeapByteBuf(Bootstrap sb, Bootstrap cb) throws Throwable {
+        testSimpleSend0(sb, cb, Unpooled.directBuffer(), true, 1);
+        testSimpleSend0(sb, cb, Unpooled.directBuffer(), true, 4);
+    }
 
-        Channel sc = sb.bind().sync().channel();
-        Channel cc = cb.bind().sync().channel();
+    @Test
+    public void testSimpleSendCompositeDirectByteBuf() throws Throwable {
+        run();
+    }
 
-        cc.writeAndFlush(new DatagramPacket(Unpooled.copyInt(1), addr)).sync();
-        assertTrue(latch.await(10, TimeUnit.SECONDS));
+    public void testSimpleSendCompositeDirectByteBuf(Bootstrap sb, Bootstrap cb) throws Throwable {
+        CompositeByteBuf buf = Unpooled.compositeBuffer();
+        buf.addComponent(Unpooled.directBuffer(2, 2));
+        buf.addComponent(Unpooled.directBuffer(2, 2));
+        testSimpleSend0(sb, cb, buf, true, 1);
 
-        sc.close().sync();
-        cc.close().sync();
+        CompositeByteBuf buf2 = Unpooled.compositeBuffer();
+        buf2.addComponent(Unpooled.directBuffer(2, 2));
+        buf2.addComponent(Unpooled.directBuffer(2, 2));
+        testSimpleSend0(sb, cb, buf2, true, 4);
+    }
+
+    @Test
+    public void testSimpleSendCompositeHeapByteBuf() throws Throwable {
+        run();
+    }
+
+    public void testSimpleSendCompositeHeapByteBuf(Bootstrap sb, Bootstrap cb) throws Throwable {
+        CompositeByteBuf buf = Unpooled.compositeBuffer();
+        buf.addComponent(Unpooled.buffer(2, 2));
+        buf.addComponent(Unpooled.buffer(2, 2));
+        testSimpleSend0(sb, cb, buf, true, 1);
+
+        CompositeByteBuf buf2 = Unpooled.compositeBuffer();
+        buf2.addComponent(Unpooled.buffer(2, 2));
+        buf2.addComponent(Unpooled.buffer(2, 2));
+        testSimpleSend0(sb, cb, buf2, true, 4);
+    }
+
+    @Test
+    public void testSimpleSendCompositeMixedByteBuf() throws Throwable {
+        run();
+    }
+
+    public void testSimpleSendCompositeMixedByteBuf(Bootstrap sb, Bootstrap cb) throws Throwable {
+        CompositeByteBuf buf = Unpooled.compositeBuffer();
+        buf.addComponent(Unpooled.directBuffer(2, 2));
+        buf.addComponent(Unpooled.buffer(2, 2));
+        testSimpleSend0(sb, cb, buf, true, 1);
+
+        CompositeByteBuf buf2 = Unpooled.compositeBuffer();
+        buf2.addComponent(Unpooled.directBuffer(2, 2));
+        buf2.addComponent(Unpooled.buffer(2, 2));
+        testSimpleSend0(sb, cb, buf2, true, 4);
     }
 
     @Test
@@ -69,9 +109,16 @@ public void testSimpleSendWithoutBind() throws Throwable {
         run();
     }
 
-    @SuppressWarnings("deprecation")
     public void testSimpleSendWithoutBind(Bootstrap sb, Bootstrap cb) throws Throwable {
-        final CountDownLatch latch = new CountDownLatch(1);
+        testSimpleSend0(sb, cb, Unpooled.directBuffer(), false, 1);
+        testSimpleSend0(sb, cb, Unpooled.directBuffer(), false, 4);
+    }
+
+    @SuppressWarnings("deprecation")
+    private void testSimpleSend0(Bootstrap sb, Bootstrap cb, ByteBuf buf, boolean bindClient, int count)
+            throws Throwable {
+        buf.writeInt(1);
+        final CountDownLatch latch = new CountDownLatch(count);
 
         sb.handler(new SimpleChannelInboundHandler<DatagramPacket>() {
             @Override
@@ -87,12 +134,22 @@ public void channelRead0(ChannelHandlerContext ctx, Object msgs) throws Exceptio
                 // Nothing will be sent.
             }
         });
-        cb.option(ChannelOption.DATAGRAM_CHANNEL_ACTIVE_ON_REGISTRATION, true);
 
         Channel sc = sb.bind().sync().channel();
-        Channel cc = cb.register().sync().channel();
-
-        cc.writeAndFlush(new DatagramPacket(Unpooled.copyInt(1), addr)).sync();
+        Channel cc;
+        if (bindClient) {
+            cc = cb.bind().sync().channel();
+        } else {
+            cb.option(ChannelOption.DATAGRAM_CHANNEL_ACTIVE_ON_REGISTRATION, true);
+            cc = cb.register().sync().channel();
+        }
+
+        for (int i = 0; i < count; i++) {
+            cc.write(new DatagramPacket(buf.retain().duplicate(), addr));
+        }
+        // release as we used buf.retain() before
+        buf.release();
+        cc.flush();
         assertTrue(latch.await(10, TimeUnit.SECONDS));
 
         sc.close().sync();