pool: handle composite grizzly buffers in NFS mover

fixed commit 08b70a5aaf

Motivation:
When allocation a buffer that is bigger that max chunk size of grizzly
memory manager the caller will get a composite Buffer. Calling `toByteBuffer`
on such composite buffer will create a new buffer, which is disconnected
from the original buffer. To utilize ByteBuffer chunks of composite buffer
the application must iterate over those chunks manually.

Modification:
Update EDSOperationREAD#process to iterate over ByteBuffer chinks if
Buffer is composite.

Result:
correct behavior in case of composite grizzly buffers.

Acked-by: Lea Morschel
Target: master
Require-book: no
Require-notes: no

modules/dcache-nfs/src/main/java/org/dcache/chimera/nfsv41/mover/EDSOperationREAD.java

@@ -61,23 +61,51 @@ public void process(CompoundContext context, nfs_resop4 result) {
                 return;
             }
 
-            var gBuffer = POOLED_BUFFER_ALLOCATOR.allocate(count);
-            gBuffer.allowBufferDispose(true);
-            ByteBuffer bb = gBuffer.toByteBuffer();
-            bb.clear().limit(count);
-
             RepositoryChannel fc = mover.getMoverChannel();
-            int bytesRead = fc.read(bb, offset);
-
-            if (bytesRead > 0) {
-                // the positions of Buffer and ByteBuffer are independent, thus keep it in sync manually
-                gBuffer.position(bytesRead);
+            var gBuffer = POOLED_BUFFER_ALLOCATOR.allocate(count);
+            int bytesToRead = count;
+
+            int rc = -1;
+            if (gBuffer.isComposite()) {
+                // composite buffers built out of array of simple buffers, thus we have to fill
+                // each buffer manually
+                var bArray = gBuffer.toBufferArray();
+                Buffer[] bufs = bArray.getArray();
+                int size = bArray.size();
+
+                for(int i = 0; i < size; i++) {
+
+                    ByteBuffer directChunk = bufs[i].toByteBuffer();
+                    directChunk.clear().limit(Math.min(directChunk.capacity(), bytesToRead));
+                    rc = fc.read(directChunk, offset + count - bytesToRead);
+                    if (rc < 0) {
+                        break;
+                    }
+
+                    // the positions of Buffer and ByteBuffer are independent, thus keep it in sync manually
+                    gBuffer.position(gBuffer.position() + directChunk.position());
+                    directChunk.flip();
+
+                    bytesToRead -= rc;
+                }
+            } else {
+
+                ByteBuffer directBuffer = gBuffer.toByteBuffer();
+                directBuffer.clear().limit(count);
+                rc = fc.read(directBuffer, offset);
+                if (rc > 0) {
+                    // the positions of Buffer and ByteBuffer are independent, thus keep it in sync manually
+                    gBuffer.position(directBuffer.position());
+                    bytesToRead -= rc;
+                }
             }
+
+            int bytesRead = count - bytesToRead;
             gBuffer.flip();
 
             res.status = nfsstat.NFS_OK;
             res.resok4 = new ShallowREAD4resok(gBuffer);
-            if (bytesRead == -1 || offset + bytesRead == fc.size()) {
+            if (rc == -1 || offset + bytesRead == fc.size()) {
                 res.resok4.eof = true;
             }