Repository: netty/netty
Issue: Transferring a ByteBuf to Outputstream without duplicating memory

(running io.netty.buffer:4.1.21.Final)

I have a Java Output stream (java.io.OutputStream) I want to incrementally copy a ByteBuffer too.

The byte buffer is off heap and big.
I naively used:
    public abstract ByteBuf getBytes(int index, OutputStream out, int length) throws IOException;

assuming that it would copy all the data to the output stream.

It does, but to do so it copies the whole native buffer into the heap! I do not have that much heap, resulting in:

```
java.lang.OutOfMemoryError: Java heap space 
at io.netty.util.internal.PlatformDependent.allocateUninitializedArray(PlatformDependent.java:200) ~[40:io.netty.common:4.1.21.Final] 
at io.netty.buffer.UnpooledUnsafeHeapByteBuf.allocateArray(UnpooledUnsafeHeapByteBuf.java:34) ~[34:io.netty.buffer:4.1.21.Final] 
at io.netty.buffer.UnpooledByteBufAllocator$InstrumentedUnpooledUnsafeHeapByteBuf.allocateArray(UnpooledByteBufAllocator.java:144) ~[34:io.netty.buffer:4.1.21.Final] 
at io.netty.buffer.UnpooledHeapByteBuf.<init>(UnpooledHeapByteBuf.java:60) ~[34:io.netty.buffer:4.1.21.Final] 
at io.netty.buffer.UnpooledUnsafeHeapByteBuf.<init>(UnpooledUnsafeHeapByteBuf.java:29) ~[34:io.netty.buffer:4.1.21.Final] 
at io.netty.buffer.UnpooledByteBufAllocator$InstrumentedUnpooledUnsafeHeapByteBuf.<init>(UnpooledByteBufAllocator.java:139) ~[34:io.netty.buffer:4.1.21.Final] 
at io.netty.buffer.UnpooledByteBufAllocator.newHeapBuffer(UnpooledByteBufAllocator.java:82) ~[34:io.netty.buffer:4.1.21.Final] 
at io.netty.buffer.AbstractByteBufAllocator.heapBuffer(AbstractByteBufAllocator.java:166) ~[34:io.netty.buffer:4.1.21.Final] 
at io.netty.buffer.AbstractByteBufAllocator.heapBuffer(AbstractByteBufAllocator.java:157) ~[34:io.netty.buffer:4.1.21.Final] 
at io.netty.buffer.UnsafeByteBufUtil.getBytes(UnsafeByteBufUtil.java:572) ~[34:io.netty.buffer:4.1.21.Final] 
at io.netty.buffer.UnpooledUnsafeDirectByteBuf.getBytes(UnpooledUnsafeDirectByteBuf.java:365) ~[34:io.netty.buffer:4.1.21.Final] 
```

I checked io.netty.buffer.UnsafeByteBufUtil.getBytes(AbstractByteBuf, long, int, OutputStream, int)
and behold. It creates a heap bytebuffer of that size to copy the array into it and then to copy that to the output stream.

Bummer!

Shouldn't netty do this better by itself? it seems that I need to manually do this by using a (reasonably sized) byte array and read from byteBuf into that and flush that to my stream in a loop

Issue comments:
You are right...
Want to do a pr?

> Am 22.03.2018 um 17:58 schrieb Fabian Lange <notifications@github.com>:
> 
> (running io.netty.buffer:4.1.21.Final)
> 
> I have a Java Output stream (java.io.OutputStream) I want to incrementally copy a ByteBuffer too.
> 
> The byte buffer is off heap and big.
> I naively used:
> public abstract ByteBuf getBytes(int index, OutputStream out, int length) throws IOException;
> 
> assuming that it would copy all the data to the output stream.
> 
> It does, but to do so it copies the whole native buffer into the heap! I do not have that much heap, resulting in:
> 
> java.lang.OutOfMemoryError: Java heap space 
> at io.netty.util.internal.PlatformDependent.allocateUninitializedArray(PlatformDependent.java:200) ~[40:io.netty.common:4.1.21.Final] 
> at io.netty.buffer.UnpooledUnsafeHeapByteBuf.allocateArray(UnpooledUnsafeHeapByteBuf.java:34) ~[34:io.netty.buffer:4.1.21.Final] 
> at io.netty.buffer.UnpooledByteBufAllocator$InstrumentedUnpooledUnsafeHeapByteBuf.allocateArray(UnpooledByteBufAllocator.java:144) ~[34:io.netty.buffer:4.1.21.Final] 
> at io.netty.buffer.UnpooledHeapByteBuf.<init>(UnpooledHeapByteBuf.java:60) ~[34:io.netty.buffer:4.1.21.Final] 
> at io.netty.buffer.UnpooledUnsafeHeapByteBuf.<init>(UnpooledUnsafeHeapByteBuf.java:29) ~[34:io.netty.buffer:4.1.21.Final] 
> at io.netty.buffer.UnpooledByteBufAllocator$InstrumentedUnpooledUnsafeHeapByteBuf.<init>(UnpooledByteBufAllocator.java:139) ~[34:io.netty.buffer:4.1.21.Final] 
> at io.netty.buffer.UnpooledByteBufAllocator.newHeapBuffer(UnpooledByteBufAllocator.java:82) ~[34:io.netty.buffer:4.1.21.Final] 
> at io.netty.buffer.AbstractByteBufAllocator.heapBuffer(AbstractByteBufAllocator.java:166) ~[34:io.netty.buffer:4.1.21.Final] 
> at io.netty.buffer.AbstractByteBufAllocator.heapBuffer(AbstractByteBufAllocator.java:157) ~[34:io.netty.buffer:4.1.21.Final] 
> at io.netty.buffer.UnsafeByteBufUtil.getBytes(UnsafeByteBufUtil.java:572) ~[34:io.netty.buffer:4.1.21.Final] 
> at io.netty.buffer.UnpooledUnsafeDirectByteBuf.getBytes(UnpooledUnsafeDirectByteBuf.java:365) ~[34:io.netty.buffer:4.1.21.Final] 
> I checked io.netty.buffer.UnsafeByteBufUtil.getBytes(AbstractByteBuf, long, int, OutputStream, int)
> and behold. It creates a heap bytebuffer of that size to copy the array into it and then to copy that to the output stream.
> 
> Bummer!
> 
> Shouldn't netty do this better by itself? it seems that I need to manually do this by using a (reasonably sized) byte array and read from byteBuf into that and flush that to my stream in a loop
> 
> —
> You are receiving this because you are subscribed to this thread.
> Reply to this email directly, view it on GitHub, or mute the thread.
> 

Doing a userspace fix for my immediate need right now. 
The question for a netty internal implementation would be how big that buffer should be. right now its full size, so it can do that in one pass.
The typical buffer size is 8k. do we want this configurable (yet another method overload?)
I think just using 8k (or less if the ByteBuf is smaller ) is ok for now 

> Am 22.03.2018 um 18:05 schrieb Fabian Lange <notifications@github.com>:
> 
> Doing a userspace fix for my immediate need right now.
> The question for a netty internal implementation would be how big that buffer should be. right now its full size, so it can do that in one pass.
> The typical buffer size is 8k. do we want this configurable (yet another method overload?)
> 
> —
> You are receiving this because you commented.
> Reply to this email directly, view it on GitHub, or mute the thread.
> 


Pull request review comments:
changes -> chances
How about:
- addr -> inAddr
- array -> in
- offset -> inOffset
- chunkLen -> inLen
- length -> outLen
Can we just do: `Math.min(Math.min(length, ByteBufUtil.WRITE_CHUNK_SIZE), length)`? This would eliminate the reason to pass 2 length parameters to the `getBytes()` below.
hmm wouldn't we still need it ? I am confused why we not need it with this change ... Can you explain
done
Because I'm an idiot? Didn't notice `outLen` is updated at the end of the loop. :-)
when I tried to improve this I was wondering if we can avoid that duplicate call?
@CodingFabian nope we can't... `internal` will be false if you use `get*`. This is needed as it may be called from multiple threads at the same time which should be fine as you not expect to do any internal updates on the indices. If you use a `get*` method there will be no duplicate invoked 
lol ok that explains why =P
so you decided against using alloc() to get this temp array?
let me check if its possible here as well (in the `Unsafe` portion it is.
fixed @CodingFabian 

Patch that resolved the issue:
diff --git a/buffer/src/main/java/io/netty/buffer/ByteBufUtil.java b/buffer/src/main/java/io/netty/buffer/ByteBufUtil.java
index b0ecb35cace..768f5587146 100644
--- a/buffer/src/main/java/io/netty/buffer/ByteBufUtil.java
+++ b/buffer/src/main/java/io/netty/buffer/ByteBufUtil.java
@@ -27,6 +27,8 @@
 import io.netty.util.internal.logging.InternalLogger;
 import io.netty.util.internal.logging.InternalLoggerFactory;
 
+import java.io.IOException;
+import java.io.OutputStream;
 import java.nio.ByteBuffer;
 import java.nio.ByteOrder;
 import java.nio.CharBuffer;
@@ -64,6 +66,7 @@ protected CharBuffer initialValue() throws Exception {
     private static final int MAX_BYTES_PER_CHAR_UTF8 =
             (int) CharsetUtil.encoder(CharsetUtil.UTF_8).maxBytesPerChar();
 
+    static final int WRITE_CHUNK_SIZE = 8192;
     static final ByteBufAllocator DEFAULT_ALLOCATOR;
 
     static {
@@ -1392,5 +1395,43 @@ private static boolean isUtf8(ByteBuf buf, int index, int length) {
         return true;
     }
 
+    /**
+     * Read bytes from the given {@link ByteBuffer} into the given {@link OutputStream} using the {@code position} and
+     * {@code length}. The position and limit of the given {@link ByteBuffer} may be adjusted.
+     */
+    static void readBytes(ByteBufAllocator allocator, ByteBuffer buffer, int position, int length, OutputStream out)
+            throws IOException {
+        if (buffer.hasArray()) {
+            out.write(buffer.array(), position + buffer.arrayOffset(), length);
+        } else {
+            int chunkLen = Math.min(length, WRITE_CHUNK_SIZE);
+            buffer.clear().position(position);
+
+            if (allocator.isDirectBufferPooled()) {
+                // if direct buffers are pooled chances are good that heap buffers are pooled as well.
+                ByteBuf tmpBuf = allocator.heapBuffer(chunkLen);
+                try {
+                    byte[] tmp = tmpBuf.array();
+                    int offset = tmpBuf.arrayOffset();
+                    getBytes(buffer, tmp, offset, chunkLen, out, length);
+                } finally {
+                    tmpBuf.release();
+                }
+            } else {
+                getBytes(buffer, new byte[chunkLen], 0, chunkLen, out, length);
+            }
+        }
+    }
+
+    private static void getBytes(ByteBuffer inBuffer, byte[] in, int inOffset, int inLen, OutputStream out, int outLen)
+            throws IOException {
+        do {
+            int len = Math.min(inLen, outLen);
+            inBuffer.get(in, inOffset, len);
+            out.write(in, inOffset, len);
+            outLen -= len;
+        } while (outLen > 0);
+    }
+
     private ByteBufUtil() { }
 }
diff --git a/buffer/src/main/java/io/netty/buffer/PooledDirectByteBuf.java b/buffer/src/main/java/io/netty/buffer/PooledDirectByteBuf.java
index 7c6192bf076..3c43509c054 100644
--- a/buffer/src/main/java/io/netty/buffer/PooledDirectByteBuf.java
+++ b/buffer/src/main/java/io/netty/buffer/PooledDirectByteBuf.java
@@ -190,17 +190,7 @@ private void getBytes(int index, OutputStream out, int length, boolean internal)
         if (length == 0) {
             return;
         }
-
-        byte[] tmp = new byte[length];
-        ByteBuffer tmpBuf;
-        if (internal) {
-            tmpBuf = internalNioBuffer();
-        } else {
-            tmpBuf = memory.duplicate();
-        }
-        tmpBuf.clear().position(idx(index));
-        tmpBuf.get(tmp);
-        out.write(tmp);
+        ByteBufUtil.readBytes(alloc(), internal ? internalNioBuffer() : memory.duplicate(), idx(index), length, out);
     }
 
     @Override
diff --git a/buffer/src/main/java/io/netty/buffer/UnpooledDirectByteBuf.java b/buffer/src/main/java/io/netty/buffer/UnpooledDirectByteBuf.java
index 3ce75d1d637..49960b37641 100644
--- a/buffer/src/main/java/io/netty/buffer/UnpooledDirectByteBuf.java
+++ b/buffer/src/main/java/io/netty/buffer/UnpooledDirectByteBuf.java
@@ -491,21 +491,7 @@ private void getBytes(int index, OutputStream out, int length, boolean internal)
         if (length == 0) {
             return;
         }
-
-        if (buffer.hasArray()) {
-            out.write(buffer.array(), index + buffer.arrayOffset(), length);
-        } else {
-            byte[] tmp = new byte[length];
-            ByteBuffer tmpBuf;
-            if (internal) {
-                tmpBuf = internalNioBuffer();
-            } else {
-                tmpBuf = buffer.duplicate();
-            }
-            tmpBuf.clear().position(index);
-            tmpBuf.get(tmp);
-            out.write(tmp);
-        }
+        ByteBufUtil.readBytes(alloc(), internal ? internalNioBuffer() : buffer.duplicate(), index, length, out);
     }
 
     @Override
diff --git a/buffer/src/main/java/io/netty/buffer/UnsafeByteBufUtil.java b/buffer/src/main/java/io/netty/buffer/UnsafeByteBufUtil.java
index 1016cc4ab93..7d866d5b433 100644
--- a/buffer/src/main/java/io/netty/buffer/UnsafeByteBufUtil.java
+++ b/buffer/src/main/java/io/netty/buffer/UnsafeByteBufUtil.java
@@ -583,18 +583,34 @@ private static void setSingleBytes(final AbstractByteBuf buf, final long addr, f
     static void getBytes(AbstractByteBuf buf, long addr, int index, OutputStream out, int length) throws IOException {
         buf.checkIndex(index, length);
         if (length != 0) {
-            ByteBuf tmpBuf = buf.alloc().heapBuffer(length);
-            try {
-                byte[] tmp = tmpBuf.array();
-                int offset = tmpBuf.arrayOffset();
-                PlatformDependent.copyMemory(addr, tmp, offset, length);
-                out.write(tmp, offset, length);
-            } finally {
-                tmpBuf.release();
+            int len = Math.min(length, ByteBufUtil.WRITE_CHUNK_SIZE);
+            if (buf.alloc().isDirectBufferPooled()) {
+                // if direct buffers are pooled chances are good that heap buffers are pooled as well.
+                ByteBuf tmpBuf = buf.alloc().heapBuffer(len);
+                try {
+                    byte[] tmp = tmpBuf.array();
+                    int offset = tmpBuf.arrayOffset();
+                    getBytes(addr, tmp, offset, len, out, length);
+                } finally {
+                    tmpBuf.release();
+                }
+            } else {
+                getBytes(addr, new byte[len], 0, len, out, length);
             }
         }
     }
 
+    private static void getBytes(long inAddr, byte[] in, int inOffset, int inLen, OutputStream out, int outLen)
+            throws IOException {
+        do {
+            int len = Math.min(inLen, outLen);
+            PlatformDependent.copyMemory(inAddr, in, inOffset, len);
+            out.write(in, inOffset, len);
+            outLen -= len;
+            inAddr += len;
+        } while (outLen > 0);
+    }
+
     static void setZero(long addr, int length) {
         if (length == 0) {
             return;


Test patch: