xdr: pass grizzly memory managet to Xdr

Motivation:
when a Xdr stream is created for bigger requests the RPC layer might
require to grow the under laying grizzly buffer. As grizzly memory
manager can reallocate only buffers that ware allocated by the same
memory manager, the Xdr should know which memory manager is used by RPC
layer.

Modification:
Update Xdr to accept the corresponding memory manager. Update TPC an UDP
filters to pass memory manager used by under laying connection. Added
test to demonstrate the behavior.

Result:
Correct behavior when direct byte buffers are used.

Acked-by: Paul Millar
Target: master

oncrpc4j-core/pom.xml

@@ -52,7 +52,11 @@
           <artifactId>mockito-core</artifactId>
           <scope>test</scope>
       </dependency>
-
+      <dependency>
+          <groupId>org.hamcrest</groupId>
+          <artifactId>hamcrest</artifactId>
+          <scope>test</scope>
+      </dependency>
       <dependency>
           <groupId>org.bouncycastle</groupId>
           <artifactId>bcprov-jdk15on</artifactId>

oncrpc4j-core/src/main/java/org/dcache/oncrpc4j/grizzly/GrizzlyMemoryManager.java

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2009 - 2018 Deutsches Elektronen-Synchroton,
+ * Copyright (c) 2009 - 2022 Deutsches Elektronen-Synchroton,
  * Member of the Helmholtz Association, (DESY), HAMBURG, GERMANY
  *
  * This library is free software; you can redistribute it and/or modify
@@ -21,7 +21,6 @@
 
 import org.glassfish.grizzly.Buffer;
 import org.glassfish.grizzly.memory.Buffers;
-import org.glassfish.grizzly.memory.BuffersBuffer;
 import org.glassfish.grizzly.memory.CompositeBuffer;
 import org.glassfish.grizzly.memory.MemoryManager;
 
@@ -36,28 +35,24 @@ public class GrizzlyMemoryManager {
     // Utility class
     private GrizzlyMemoryManager() {}
 
+    public static MemoryManager getDefaultMemoryManager() {
+        return GRIZZLY_MM;
+    }
+
     public static Buffer allocate(int size) {
         return GRIZZLY_MM.allocate(size);
     }
 
-    public static Buffer reallocate(Buffer oldBuffer, int newSize) {
+    public static Buffer reallocate(MemoryManager memoryManager, Buffer oldBuffer, int newSize) {
         if (oldBuffer.isComposite()) {
-            Buffer addon = allocate(newSize-oldBuffer.capacity());
+            Buffer addon = memoryManager.allocate(newSize-oldBuffer.capacity());
             ((CompositeBuffer)oldBuffer).append(addon);
             return oldBuffer;
         }
-        return GRIZZLY_MM.reallocate(oldBuffer, newSize);
+        return memoryManager.reallocate(oldBuffer, newSize);
     }
 
     public static Buffer wrap(byte[] bytes) {
         return Buffers.wrap(GRIZZLY_MM, bytes);
     }
-
-    public static Buffer createComposite(Buffer...buffers) {
-        return BuffersBuffer.create(GRIZZLY_MM, buffers);
-    }
-
-    public static BuffersBuffer create() {
-        return BuffersBuffer.create();
-    }
 }

oncrpc4j-core/src/main/java/org/dcache/oncrpc4j/rpc/RpcMessageParserTCP.java

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2009 - 2020 Deutsches Elektronen-Synchroton,
+ * Copyright (c) 2009 - 2022 Deutsches Elektronen-Synchroton,
  * Member of the Helmholtz Association, (DESY), HAMBURG, GERMANY
  *
  * This library is free software; you can redistribute it and/or modify
@@ -23,12 +23,12 @@
 import java.io.IOException;
 
 import java.nio.ByteOrder;
-import org.dcache.oncrpc4j.grizzly.GrizzlyMemoryManager;
 import org.glassfish.grizzly.Buffer;
 import org.glassfish.grizzly.filterchain.BaseFilter;
 import org.glassfish.grizzly.filterchain.FilterChainContext;
 import org.glassfish.grizzly.filterchain.NextAction;
 import org.glassfish.grizzly.memory.BuffersBuffer;
+import org.glassfish.grizzly.memory.MemoryManager;
 
 public class RpcMessageParserTCP extends BaseFilter {
 
@@ -53,7 +53,7 @@ public NextAction handleRead(FilterChainContext ctx) throws IOException {
             return ctx.getStopAction(messageBuffer);
         }
 
-        ctx.setMessage(assembleXdr(messageBuffer));
+        ctx.setMessage(assembleXdr(messageBuffer, ctx.getMemoryManager()));
 
         final Buffer reminder = messageBuffer.hasRemaining()
                 ? messageBuffer.split(messageBuffer.position()) : null;
@@ -67,19 +67,19 @@ public NextAction handleWrite(FilterChainContext ctx) throws IOException {
         Buffer b = ctx.getMessage();
         int len = b.remaining() | RPC_LAST_FRAG;
 
-        Buffer marker = GrizzlyMemoryManager.allocate(4);
+        Buffer marker = ctx.getMemoryManager().allocate(4);
         marker.order(ByteOrder.BIG_ENDIAN);
         marker.putInt(len);
         marker.flip();
         marker.allowBufferDispose(true);
         b.allowBufferDispose(true);
-        Buffer composite = GrizzlyMemoryManager.createComposite(marker, b);
+        Buffer composite = BuffersBuffer.create(ctx.getMemoryManager(), marker, b);
         composite.allowBufferDispose(true);
         ctx.setMessage(composite);
         return ctx.getInvokeAction();
     }
 
-    private boolean isAllFragmentsArrived(Buffer messageBuffer) throws IOException {
+    private boolean isAllFragmentsArrived(Buffer messageBuffer) {
         final Buffer buffer = messageBuffer.duplicate();
         buffer.order(ByteOrder.BIG_ENDIAN);
 
@@ -119,7 +119,7 @@ private static boolean isLastFragment(int marker) {
         return (marker & RPC_LAST_FRAG) != 0;
     }
 
-    private Xdr assembleXdr(Buffer messageBuffer) {
+    private Xdr assembleXdr(Buffer messageBuffer, MemoryManager memoryManager) {
 
         Buffer currentFragment;
         BuffersBuffer multipleFragments = null;
@@ -141,14 +141,14 @@ private Xdr assembleXdr(Buffer messageBuffer) {
                  * we use composite buffer only if required
                  * as they not for free.
                  */
-                multipleFragments = GrizzlyMemoryManager.create();
+                multipleFragments = BuffersBuffer.create(memoryManager);
             }
 
             if (multipleFragments != null) {
                 multipleFragments.append(currentFragment);
             }
         } while (!messageComplete);
 
-        return new Xdr(multipleFragments == null ? currentFragment : multipleFragments);
+        return new Xdr(multipleFragments == null ? currentFragment : multipleFragments, memoryManager);
     }
 }

oncrpc4j-core/src/main/java/org/dcache/oncrpc4j/rpc/RpcMessageParserUDP.java

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2009 - 2018 Deutsches Elektronen-Synchroton,
+ * Copyright (c) 2009 - 2022 Deutsches Elektronen-Synchroton,
  * Member of the Helmholtz Association, (DESY), HAMBURG, GERMANY
  *
  * This library is free software; you can redistribute it and/or modify
@@ -37,7 +37,7 @@ public class RpcMessageParserUDP extends BaseFilter {
     public NextAction handleRead(FilterChainContext ctx) throws IOException {
         Buffer messageBuffer = ctx.getMessage();
 
-        Xdr xdr = new Xdr(messageBuffer);
+        Xdr xdr = new Xdr(messageBuffer, ctx.getMemoryManager());
         ctx.setMessage(xdr);
 
         return ctx.getInvokeAction();

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

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2009 - 2020 Deutsches Elektronen-Synchroton,
+ * Copyright (c) 2009 - 2022 Deutsches Elektronen-Synchroton,
  * Member of the Helmholtz Association, (DESY), HAMBURG, GERMANY
  *
  * This library is free software; you can redistribute it and/or modify
@@ -24,6 +24,7 @@
 import java.nio.charset.StandardCharsets;
 import org.dcache.oncrpc4j.grizzly.GrizzlyMemoryManager;
 import org.glassfish.grizzly.Buffer;
+import org.glassfish.grizzly.memory.MemoryManager;
 
 import static com.google.common.base.Preconditions.checkState;
 
@@ -49,13 +50,17 @@ public class Xdr implements XdrDecodingStream, XdrEncodingStream, AutoCloseable
      */
     private boolean _inUse;
 
+    /**
+     * Memory manager used to allocate, resize buffers.
+     */
+    private final MemoryManager _memoryManager;
     /**
      * Create a new Xdr object with a buffer of given size.
      *
      * @param size of the buffer in bytes
      */
     public Xdr(int size) {
-        this(GrizzlyMemoryManager.allocate(size));
+        this(GrizzlyMemoryManager.allocate(size), GrizzlyMemoryManager.getDefaultMemoryManager());
     }
 
 
@@ -69,14 +74,26 @@ public Xdr(int size) {
      * @param  bytes The array that will back this Xdr.
      */
     public Xdr(byte[] bytes) {
-        this(GrizzlyMemoryManager.wrap(bytes));
+        this(GrizzlyMemoryManager.wrap(bytes), GrizzlyMemoryManager.getDefaultMemoryManager());
     }
 
     /**
      * Create a new XDR back ended with given {@link ByteBuffer}.
      * @param body buffer to use
+     * @deprecated this constructor shouldn't be ued
      */
+    @Deprecated
     public Xdr(Buffer body) {
+        this(body, GrizzlyMemoryManager.getDefaultMemoryManager());
+    }
+
+    /**
+     * Create a new XDR back ended with given {@link ByteBuffer}.
+     * @param body buffer to use
+     * @param memoryManager memory manager used to allocate provided buffer.
+     */
+    public Xdr(Buffer body, MemoryManager memoryManager) {
+        _memoryManager = memoryManager;
         _buffer = body;
         _buffer.order(ByteOrder.BIG_ENDIAN);
     }
@@ -873,7 +890,7 @@ private void ensureCapacity(int size) {
         if(_buffer.remaining() < size) {
             int oldCapacity = _buffer.capacity();
             int newCapacity = Math.max((oldCapacity * 3) / 2 + 1, oldCapacity + size);
-            _buffer = GrizzlyMemoryManager.reallocate(_buffer, newCapacity);
+            _buffer = GrizzlyMemoryManager.reallocate(_memoryManager, _buffer, newCapacity);
         }
     }
 

oncrpc4j-core/src/test/java/org/dcache/oncrpc4j/xdr/XdrTest.java

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2009 - 2018 Deutsches Elektronen-Synchroton,
+ * Copyright (c) 2009 - 2022 Deutsches Elektronen-Synchroton,
  * Member of the Helmholtz Association, (DESY), HAMBURG, GERMANY
  *
  * This library is free software; you can redistribute it and/or modify
@@ -22,16 +22,22 @@
 import java.nio.ByteBuffer;
 import org.dcache.oncrpc4j.util.Bytes;
 import java.nio.ByteOrder;
-import java.util.Arrays;
 import org.dcache.oncrpc4j.grizzly.GrizzlyMemoryManager;
 import org.glassfish.grizzly.Buffer;
 import org.glassfish.grizzly.memory.BuffersBuffer;
+import org.glassfish.grizzly.memory.ByteBufferManager;
 import org.glassfish.grizzly.memory.CompositeBuffer;
+import org.glassfish.grizzly.memory.MemoryManager;
 import org.junit.Before;
 import org.junit.Test;
 
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.greaterThan;
+import static org.junit.Assert.assertArrayEquals;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
 import static org.mockito.Mockito.*;
-import static org.junit.Assert.*;
 
 public class XdrTest {
 
@@ -638,8 +644,35 @@ public void testByteBuffer() throws BadXdrOncRpcException {
         assertEquals(decoded.getChar(), 'a');
     }
 
+
+    // this test demonstrates that grizzly memory manager can grow only buffers of the same type
+    @Test(expected = Exception.class)
+    public void testGrowDirectBufferDefaultMM() throws BadXdrOncRpcException {
+        // memory manager with direct buffers, max cached buffer 512 and min allocation 0
+        MemoryManager mm = new ByteBufferManager(true, 512, 0);
+        Buffer b = mm.allocate(4);
+
+        Xdr xdr = new Xdr(b);
+        xdr.beginEncoding();
+        xdr.xdrEncodeLong(1L);
+    }
+
+    @Test
+    public void testGrowDirectBuffer() throws BadXdrOncRpcException {
+        // memory manager with direct buffers, max cached buffer 512 and min allocation 0
+        MemoryManager mm = new ByteBufferManager(true, 512, 0);
+        Buffer b = mm.allocate(4);
+
+        Xdr xdr = new Xdr(b, mm);
+        xdr.beginEncoding();
+        xdr.xdrEncodeLong(1L);
+
+        Buffer newBuffer = xdr.asBuffer();
+        assertThat("Underlying buffer is not grown", newBuffer.capacity(), greaterThan(b.capacity()));
+    }
+
     @Test
-    public void testRelaseBufferOnClose() {
+    public void testReleaseBufferOnClose() {
 
         Buffer b = mock(Buffer.class);
         Xdr xdr = new Xdr(b);

pom.xml

@@ -75,7 +75,7 @@
 
     <!--  version of test dependencies  -->
     <junit.version>4.13.1</junit.version>
-    <mockito-core.version>2.22.0</mockito-core.version>
+    <mockito-core.version>3.6.0</mockito-core.version>
     <logback-classic.version>1.2.3</logback-classic.version>
     <jmh.version>1.21</jmh.version>
   </properties>
@@ -212,6 +212,12 @@
               <version>${mockito-core.version}</version>
               <scope>test</scope>
           </dependency>
+          <dependency>
+              <groupId>org.hamcrest</groupId>
+              <artifactId>hamcrest</artifactId>
+              <version>2.2</version>
+              <scope>test</scope>
+          </dependency>
           <dependency>
               <groupId>com.google.guava</groupId>
               <artifactId>guava</artifactId>