From 174e76e00b536be88e2477f4afea24a40c0ddb09 Mon Sep 17 00:00:00 2001 From: zhouyuan1 Date: Tue, 25 Oct 2022 16:37:57 +0800 Subject: [PATCH 1/2] v1 --- .../nfsclient/network/RPCRecordDecoder.java | 22 ++++++++++++++----- 1 file changed, 16 insertions(+), 6 deletions(-) diff --git a/src/main/java/com/emc/ecs/nfsclient/network/RPCRecordDecoder.java b/src/main/java/com/emc/ecs/nfsclient/network/RPCRecordDecoder.java index e1e94f0..7a144e3 100644 --- a/src/main/java/com/emc/ecs/nfsclient/network/RPCRecordDecoder.java +++ b/src/main/java/com/emc/ecs/nfsclient/network/RPCRecordDecoder.java @@ -1,12 +1,12 @@ /** * Copyright 2016-2018 Dell Inc. or its subsidiaries. All rights reserved. - * + *

* Licensed under the Apache License, Version 2.0 (the "License"). * You may not use this file except in compliance with the License. * A copy of the License is located at - * + *

* http://www.apache.org/licenses/LICENSE-2.0.txt - * + *

* or in the "license" file accompanying this file. This file 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 @@ -22,7 +22,7 @@ /** * To receive the entire response. We do not actually decode the rpc packet here. * Just get the size from the packet and then put them in internal buffer until all data arrive. - * + * * @author seibed */ public class RPCRecordDecoder extends FrameDecoder { @@ -33,6 +33,9 @@ public class RPCRecordDecoder extends FrameDecoder { */ private int _recordLength = 0; + // To hold the real position of fragment starts + private int _realReaderIndex = 0; + /* (non-Javadoc) * @see org.jboss.netty.handler.codec.frame.FrameDecoder#decode(org.jboss.netty.channel.ChannelHandlerContext, org.jboss.netty.channel.Channel, org.jboss.netty.buffer.ChannelBuffer) */ @@ -46,7 +49,7 @@ protected Object decode(ChannelHandlerContext channelHandlerContext, Channel cha //marking the current reading position channelBuffer.markReaderIndex(); - + channelBuffer.skipBytes(_realReaderIndex); //get the fragment size and wait until the entire fragment is available. long fragSize = channelBuffer.readUnsignedInt(); boolean lastFragment = RecordMarkingUtil.isLastFragment(fragSize); @@ -61,17 +64,24 @@ protected Object decode(ChannelHandlerContext channelHandlerContext, Channel cha _recordLength += 4 + (int) fragSize; + System.out.println("[test]current realindex=" + _realReaderIndex + " readerIndex=" + channelBuffer.readerIndex() + " length=" + _recordLength + " fragSize=" + fragSize + " readable=" + channelBuffer.readableBytes()); + //check the last fragment if (!lastFragment) { + channelBuffer.resetReaderIndex(); + _realReaderIndex += 4 + (int) fragSize; //not the last fragment, the data is put in an internally maintained cumulative buffer return null; } - + System.out.println("[test]is last"); byte[] rpcResponse = new byte[_recordLength]; + channelBuffer.readerIndex(channelBuffer.readerIndex() - _recordLength); + channelBuffer.readBytes(rpcResponse, 0, _recordLength); _recordLength = 0; + _realReaderIndex = 0; return rpcResponse; } } From 94febe984da48b5e68b8ad9b9303872090687944 Mon Sep 17 00:00:00 2001 From: zhouyuan1 Date: Tue, 25 Oct 2022 19:10:50 +0800 Subject: [PATCH 2/2] Fix bug of handle large rpc call response If the response of rpc call is separate into serval fragments, will trying to hold them and merge into ONE buffer. This bug will happen when using this client to download large files. --- .../java/com/emc/ecs/nfsclient/network/RPCRecordDecoder.java | 3 --- .../java/com/emc/ecs/nfsclient/network/RecordMarkingUtil.java | 2 +- 2 files changed, 1 insertion(+), 4 deletions(-) diff --git a/src/main/java/com/emc/ecs/nfsclient/network/RPCRecordDecoder.java b/src/main/java/com/emc/ecs/nfsclient/network/RPCRecordDecoder.java index 7a144e3..7109d1c 100644 --- a/src/main/java/com/emc/ecs/nfsclient/network/RPCRecordDecoder.java +++ b/src/main/java/com/emc/ecs/nfsclient/network/RPCRecordDecoder.java @@ -64,8 +64,6 @@ protected Object decode(ChannelHandlerContext channelHandlerContext, Channel cha _recordLength += 4 + (int) fragSize; - System.out.println("[test]current realindex=" + _realReaderIndex + " readerIndex=" + channelBuffer.readerIndex() + " length=" + _recordLength + " fragSize=" + fragSize + " readable=" + channelBuffer.readableBytes()); - //check the last fragment if (!lastFragment) { channelBuffer.resetReaderIndex(); @@ -73,7 +71,6 @@ protected Object decode(ChannelHandlerContext channelHandlerContext, Channel cha //not the last fragment, the data is put in an internally maintained cumulative buffer return null; } - System.out.println("[test]is last"); byte[] rpcResponse = new byte[_recordLength]; channelBuffer.readerIndex(channelBuffer.readerIndex() - _recordLength); diff --git a/src/main/java/com/emc/ecs/nfsclient/network/RecordMarkingUtil.java b/src/main/java/com/emc/ecs/nfsclient/network/RecordMarkingUtil.java index da0dd41..8e6aff7 100644 --- a/src/main/java/com/emc/ecs/nfsclient/network/RecordMarkingUtil.java +++ b/src/main/java/com/emc/ecs/nfsclient/network/RecordMarkingUtil.java @@ -139,7 +139,7 @@ static Xdr removeRecordMarking(byte[] bytes) { fragSize = maskFragmentSize(fragSize); toReturn.putBytes(input.getBuffer(), input.getOffset(), (int) fragSize); - inputOff += fragSize; + inputOff += 4 + fragSize; input.setOffset(inputOff); }