xdr: optimize encoding of ByteBuffer

Motivation:
When a ByteBuffer is encoded into XdrStream then a implicit copy of
underlaying data is performed. By wrapping ByteBuffer with and building
a composite buffer the additional copy can be avoid.

Modification:
Add Xdr#xdrEncodeShallowByteBuffer that creates a composite buffer to avoid
extra copy.

Result:
Encoding of a ByteBuffer is an O(1) operation, if possible (original
buffer is not modified).

                                     Buffer size
XdrBenchmark.encodeByteBufferShallow        1024  thrpt    5   9402670.599 ±  26808.413  ops/s
XdrBenchmark.encodeByteBufferShallow        8192  thrpt    5   9102908.516 ±  64365.443  ops/s
XdrBenchmark.encodeByteBufferShallow      262144  thrpt    5   9182060.856 ±  83327.650  ops/s
XdrBenchmark.encodeByteBufferShallow     1048576  thrpt    5   9052320.542 ±  23289.002  ops/s

XdrBenchmark.encodeByteBuffer               1024  thrpt    5  28773875.725 ± 250222.745  ops/s
XdrBenchmark.encodeByteBuffer               8192  thrpt    5   6439241.956 ±  24534.514  ops/s
XdrBenchmark.encodeByteBuffer             262144  thrpt    5     88237.372 ±    704.050  ops/s
XdrBenchmark.encodeByteBuffer            1048576  thrpt    5     21276.243 ±    116.393  ops/s

Acked-by: Albert Rossi
Target: master

oncrpc4j-benchmark/src/main/java/org/dcache/oncrpc4j/benchmarks/XdrBenchmark.java

@@ -0,0 +1,60 @@
+package org.dcache.oncrpc4j.benchmarks;
+
+import java.nio.ByteBuffer;
+import java.util.concurrent.ThreadLocalRandom;
+import org.dcache.oncrpc4j.xdr.Xdr;
+import org.openjdk.jmh.annotations.Benchmark;
+import org.openjdk.jmh.annotations.BenchmarkMode;
+import org.openjdk.jmh.annotations.Mode;
+import org.openjdk.jmh.annotations.Param;
+import org.openjdk.jmh.annotations.Scope;
+import org.openjdk.jmh.annotations.Setup;
+import org.openjdk.jmh.annotations.State;
+import org.openjdk.jmh.infra.Blackhole;
+
+@State(Scope.Benchmark)
+@BenchmarkMode(Mode.Throughput)
+public class XdrBenchmark {
+
+
+    @Param({"1024", "8192", "262144", "1048576"})
+    private String size;
+
+    private Xdr xdr;
+
+    private ByteBuffer bb;
+
+    @Setup
+    public void setUp() {
+
+        byte[] buf = new byte[Integer.parseInt(size)];
+        ThreadLocalRandom.current().nextBytes(buf);
+
+        bb = ByteBuffer.wrap(buf);
+        xdr = new Xdr(256);
+    }
+
+
+    @Benchmark
+    public void encodeByteBuffer(Blackhole blackhole) {
+
+        xdr.beginEncoding();
+        bb.clear().limit(bb.capacity());
+        xdr.xdrEncodeByteBuffer(bb);
+        xdr.endEncoding();
+
+        blackhole.consume(xdr);
+    }
+
+    @Benchmark
+    public void encodeByteBufferShallow(Blackhole blackhole) {
+
+        xdr.beginEncoding();
+        bb.clear().limit(bb.capacity());
+        xdr.xdrEncodeShallowByteBuffer(bb);
+        xdr.endEncoding();
+
+        blackhole.consume(xdr);
+    }
+
+}

oncrpc4j-core/src/main/java/org/dcache/oncrpc4j/xdr/Xdr.java

@@ -24,6 +24,8 @@
 import java.nio.charset.StandardCharsets;
 import org.dcache.oncrpc4j.grizzly.GrizzlyMemoryManager;
 import org.glassfish.grizzly.Buffer;
+import org.glassfish.grizzly.memory.BuffersBuffer;
+import org.glassfish.grizzly.memory.ByteBufferWrapper;
 import org.glassfish.grizzly.memory.MemoryManager;
 
 import static com.google.common.base.Preconditions.checkState;
@@ -759,6 +761,29 @@ public void xdrEncodeByteBuffer(ByteBuffer buf) {
         _buffer.position(_buffer.position() + padding);
     }
 
+    /**
+     * A version of {@link #xdrEncodeByteBuffer(ByteBuffer)} which avoids internal copy.
+     * Note: any change to the {@code buf} will cause unpredicted behavior.
+     *
+     * @param buf The buffer from which bytes are to be retrieved.
+     */
+    public void xdrEncodeShallowByteBuffer(ByteBuffer buf) {
+        int len = buf.remaining();
+        int padding = (4 - (len & 3)) & 3;
+        xdrEncodeInt(len);
+        int ep = _buffer.position() + buf.remaining();
+        var b = new ByteBufferWrapper(buf);
+        b.allowBufferDispose(true);
+        var composite = BuffersBuffer.create(_memoryManager);
+        composite.append(_buffer.slice(0, _buffer.position())).append(b);
+        composite.position(ep);
+        composite.limit(ep);
+
+        _buffer = composite;
+        ensureCapacity(padding);
+        _buffer.position(_buffer.position() + padding);
+    }
+
     /**
      * Encodes (aka "serializes") a vector of bytes, which is nothing more than
      * a series of octets (or 8 bits wide bytes), each packed into its very own