You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@uniffle.apache.org by "leixm (via GitHub)" <gi...@apache.org> on 2023/03/19 15:47:21 UTC
[GitHub] [incubator-uniffle] leixm opened a new pull request, #742: [#133][Netty] Add Encoder and Decoder.
leixm opened a new pull request, #742:
URL: https://github.com/apache/incubator-uniffle/pull/742
### What changes were proposed in this pull request?
Add Netty Encoder and Decoder.
### Why are the changes needed?
For #133
### Does this PR introduce _any_ user-facing change?
No.
### How was this patch tested?
UTs.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] leixm commented on pull request #742: [#133] feat(netty): Add Encoder and Decoder.
Posted by "leixm (via GitHub)" <gi...@apache.org>.
leixm commented on PR #742:
URL: https://github.com/apache/incubator-uniffle/pull/742#issuecomment-1479733738
cc @jerqi All done.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] jerqi commented on a diff in pull request #742: [#133] feat(netty): Add Encoder and Decoder.
Posted by "jerqi (via GitHub)" <gi...@apache.org>.
jerqi commented on code in PR #742:
URL: https://github.com/apache/incubator-uniffle/pull/742#discussion_r1141937131
##########
common/src/main/java/org/apache/uniffle/common/netty/protocol/SendShuffleDataRequest.java:
##########
@@ -0,0 +1,147 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License 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 permissions and
+ * limitations under the License.
+ */
+
+package org.apache.uniffle.common.netty.protocol;
Review Comment:
Could we put them together and remove this protocol in this pr?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] jerqi commented on a diff in pull request #742: [#133] feat(netty): Add Encoder and Decoder.
Posted by "jerqi (via GitHub)" <gi...@apache.org>.
jerqi commented on code in PR #742:
URL: https://github.com/apache/incubator-uniffle/pull/742#discussion_r1145015942
##########
common/src/main/java/org/apache/uniffle/common/netty/MessageEncoder.java:
##########
@@ -0,0 +1,49 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License 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 permissions and
+ * limitations under the License.
+ */
+
+package org.apache.uniffle.common.netty;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelOutboundHandlerAdapter;
+import io.netty.channel.ChannelPromise;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.uniffle.common.netty.protocol.Message;
+
+public class MessageEncoder extends ChannelOutboundHandlerAdapter {
Review Comment:
Could we add some comments to explain our encode plan?
##########
common/src/main/java/org/apache/uniffle/common/netty/MessageEncoder.java:
##########
@@ -0,0 +1,49 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License 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 permissions and
+ * limitations under the License.
+ */
+
+package org.apache.uniffle.common.netty;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelOutboundHandlerAdapter;
+import io.netty.channel.ChannelPromise;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.uniffle.common.netty.protocol.Message;
+
+public class MessageEncoder extends ChannelOutboundHandlerAdapter {
+
+ private static final Logger LOG = LoggerFactory.getLogger(MessageEncoder.class);
+
+ @Override
+ public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) {
+ // todo: support zero copy
Review Comment:
If this is a `todo`, we should set up an issue, too.
##########
common/src/main/java/org/apache/uniffle/common/netty/TransportFrameDecoder.java:
##########
@@ -0,0 +1,191 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License 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 permissions and
+ * limitations under the License.
+ */
+
+package org.apache.uniffle.common.netty;
+
+import java.util.LinkedList;
+
+import com.google.common.base.Preconditions;
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.CompositeByteBuf;
+import io.netty.buffer.Unpooled;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelInboundHandlerAdapter;
+
+import org.apache.uniffle.common.netty.protocol.Message;
+
+/**
+ * A customized frame decoder that allows intercepting raw data.
+ *
+ * <p>This behaves like Netty's frame decoder (with hard coded parameters that match this library's
+ * needs), except it allows an interceptor to be installed to read data directly before it's framed.
+ *
+ * <p>Unlike Netty's frame decoder, each frame is dispatched to child handlers as soon as it's
+ * decoded, instead of building as many frames as the current buffer allows and dispatching all of
+ * them. This allows a child handler to install an interceptor if needed.
+ *
+ * <p>If an interceptor is installed, framing stops, and data is instead fed directly to the
+ * interceptor. When the interceptor indicates that it doesn't need to read any more data, framing
+ * resumes. Interceptors should not hold references to the data buffers provided to their handle()
+ * method.
+ */
+public class TransportFrameDecoder extends ChannelInboundHandlerAdapter implements FrameDecoder {
+ private int msgSize = -1;
+ private Message.Type curType = Message.Type.UNKNOWN_TYPE;
+ private ByteBuf headerBuf = Unpooled.buffer(HEADER_SIZE, HEADER_SIZE);
+ private static final int MAX_FRAME_SIZE = Integer.MAX_VALUE;
+ private static final int UNKNOWN_FRAME_SIZE = -1;
+
+ private final LinkedList<ByteBuf> buffers = new LinkedList<>();
+
+ private long totalSize = 0;
+ private long nextFrameSize = UNKNOWN_FRAME_SIZE;
+
+ @Override
+ public void channelRead(ChannelHandlerContext ctx, Object data) {
+ ByteBuf in = (ByteBuf) data;
+ buffers.add(in);
+ totalSize += in.readableBytes();
+
+ while (!buffers.isEmpty()) {
+ ByteBuf frame = decodeNext();
+ if (frame == null) {
+ break;
+ }
+ Message msg = Message.decode(curType, frame);
Review Comment:
Will the method `decode` throw some runtime exception? It may cause that frame don't release memory.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] jerqi commented on a diff in pull request #742: [#133] feat(netty): Add Encoder and Decoder.
Posted by "jerqi (via GitHub)" <gi...@apache.org>.
jerqi commented on code in PR #742:
URL: https://github.com/apache/incubator-uniffle/pull/742#discussion_r1142438755
##########
common/src/main/java/org/apache/uniffle/common/netty/protocol/Encoders.java:
##########
@@ -0,0 +1,69 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License 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 permissions and
+ * limitations under the License.
+ */
+
+package org.apache.uniffle.common.netty.protocol;
+
+import java.util.List;
+
+import io.netty.buffer.ByteBuf;
+
+import org.apache.uniffle.common.ShuffleBlockInfo;
+import org.apache.uniffle.common.ShuffleServerInfo;
+import org.apache.uniffle.common.util.ByteBufUtils;
+
+public class Encoders {
+ public static void encodeShuffleServerInfo(ShuffleServerInfo shuffleServerInfo, ByteBuf byteBuf) {
+ ByteBufUtils.writeLengthAndString(byteBuf, shuffleServerInfo.getId());
+ ByteBufUtils.writeLengthAndString(byteBuf, shuffleServerInfo.getHost());
+ byteBuf.writeInt(shuffleServerInfo.getGrpcPort());
+ byteBuf.writeInt(shuffleServerInfo.getNettyPort());
+ }
+
+ public static void encodeShuffleBlockInfo(ShuffleBlockInfo shuffleBlockInfo, ByteBuf byteBuf) {
+ byteBuf.writeInt(shuffleBlockInfo.getPartitionId());
+ byteBuf.writeLong(shuffleBlockInfo.getBlockId());
+ byteBuf.writeInt(shuffleBlockInfo.getLength());
+ byteBuf.writeInt(shuffleBlockInfo.getShuffleId());
+ byteBuf.writeLong(shuffleBlockInfo.getCrc());
+ byteBuf.writeLong(shuffleBlockInfo.getTaskAttemptId());
+ ByteBufUtils.copyByteBuf(shuffleBlockInfo.getData(), byteBuf);
Review Comment:
Could we use CompositeByteBuf to avoid data copy here?
##########
common/src/main/java/org/apache/uniffle/common/netty/protocol/Decoders.java:
##########
@@ -0,0 +1,58 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License 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 permissions and
+ * limitations under the License.
+ */
+
+package org.apache.uniffle.common.netty.protocol;
+
+import java.util.List;
+
+import com.google.common.collect.Lists;
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.Unpooled;
+
+import org.apache.uniffle.common.ShuffleBlockInfo;
+import org.apache.uniffle.common.ShuffleServerInfo;
+import org.apache.uniffle.common.util.ByteBufUtils;
+
+public class Decoders {
+ public static ShuffleServerInfo decodeShuffleServerInfo(ByteBuf byteBuf) {
+ String id = ByteBufUtils.readLengthAndString(byteBuf);
+ String host = ByteBufUtils.readLengthAndString(byteBuf);
+ int grpcPort = byteBuf.readInt();
+ int nettyPort = byteBuf.readInt();
+ return new ShuffleServerInfo(id, host, grpcPort, nettyPort);
+ }
+
+ public static ShuffleBlockInfo decodeShuffleBlockInfo(ByteBuf byteBuf) {
+ int partId = byteBuf.readInt();
+ long blockId = byteBuf.readLong();
+ int length = byteBuf.readInt();
+ int shuffleId = byteBuf.readInt();
+ long crc = byteBuf.readLong();
+ long taskAttemptId = byteBuf.readLong();
+ // todo: we can readSlice here, but it needs to be released manually after use, or it will cause a memory leak
+ byte[] data = ByteBufUtils.readByteArray(byteBuf);
Review Comment:
Could we use `ByteBuf` instead of `byte[]` here?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] leixm commented on a diff in pull request #742: [#133] feat(netty): Add Encoder and Decoder.
Posted by "leixm (via GitHub)" <gi...@apache.org>.
leixm commented on code in PR #742:
URL: https://github.com/apache/incubator-uniffle/pull/742#discussion_r1142837805
##########
common/src/main/java/org/apache/uniffle/common/netty/protocol/SendShuffleDataRequest.java:
##########
@@ -0,0 +1,153 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License 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 permissions and
+ * limitations under the License.
+ */
+
+package org.apache.uniffle.common.netty.protocol;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+
+import com.google.common.collect.Lists;
+import com.google.common.collect.Maps;
+import io.netty.buffer.ByteBuf;
+
+import org.apache.uniffle.common.ShuffleBlockInfo;
+import org.apache.uniffle.common.util.ByteBufUtils;
+
+public class SendShuffleDataRequest extends Message {
+ public long requestId;
+ private String appId;
+ private int shuffleId;
+ private long requireId;
+ private Map<Integer, List<ShuffleBlockInfo>> partitionToBlocks;
+ private long timestamp;
+
+ public SendShuffleDataRequest(long requestId, String appId, int shuffleId, long requireId,
+ Map<Integer, List<ShuffleBlockInfo>> partitionToBlocks, long timestamp) {
+ this.requestId = requestId;
+ this.appId = appId;
+ this.shuffleId = shuffleId;
+ this.requireId = requireId;
+ this.partitionToBlocks = partitionToBlocks;
+ this.timestamp = timestamp;
+ }
+
+ @Override
+ public Type type() {
+ return Type.SEND_SHUFFLE_DATA_REQUEST;
+ }
+
+ @Override
+ public int encodedLength() {
+ int encodeLength = Long.BYTES + ByteBufUtils.encodedLength(appId) + Integer.BYTES + Long.BYTES + Integer.BYTES;
+ for (Map.Entry<Integer, List<ShuffleBlockInfo>> entry : partitionToBlocks.entrySet()) {
+ encodeLength += 2 * Integer.BYTES;
+ for (ShuffleBlockInfo sbi : entry.getValue()) {
+ encodeLength += Encoders.encodeLengthOfShuffleBlockInfo(sbi);
+ }
+ }
+ return encodeLength + Long.BYTES;
+ }
+
+ @Override
+ public void encode(ByteBuf buf) {
+ buf.writeLong(requestId);
+ ByteBufUtils.writeLengthAndString(buf, appId);
+ buf.writeInt(shuffleId);
+ buf.writeLong(requireId);
+ encodePartitionData(buf);
+ buf.writeLong(timestamp);
+ }
+
+ private static Map<Integer, List<ShuffleBlockInfo>> decodePartitionData(ByteBuf byteBuf) {
+ Map<Integer, List<ShuffleBlockInfo>> partitionToBlocks = Maps.newHashMap();
+ int lengthOfPartitionData = byteBuf.readInt();
+ for (int i = 0; i < lengthOfPartitionData; i++) {
+ int partitionId = byteBuf.readInt();
+ int lengthOfShuffleBlocks = byteBuf.readInt();
+ List<ShuffleBlockInfo> shuffleBlockInfoList = Lists.newArrayList();
+ for (int j = 0; j < lengthOfShuffleBlocks; j++) {
+ shuffleBlockInfoList.add(Decoders.decodeShuffleBlockInfo(byteBuf));
+ }
+ partitionToBlocks.put(partitionId, shuffleBlockInfoList);
+ }
+ return partitionToBlocks;
+ }
+
+ public static SendShuffleDataRequest decode(ByteBuf byteBuf) {
+ long requestId = byteBuf.readLong();
+ String appId = ByteBufUtils.readLengthAndString(byteBuf);
+ int shuffleId = byteBuf.readInt();
+ long requireId = byteBuf.readLong();
+ Map<Integer, List<ShuffleBlockInfo>> partitionToBlocks = decodePartitionData(byteBuf);
+ long timestamp = byteBuf.readLong();
+ return new SendShuffleDataRequest(requestId, appId, shuffleId, requireId, partitionToBlocks, timestamp);
+ }
+
+ private void encodePartitionData(ByteBuf buf) {
+ buf.writeInt(partitionToBlocks.size());
+ for (Map.Entry<Integer, List<ShuffleBlockInfo>> entry : partitionToBlocks.entrySet()) {
+ buf.writeInt(entry.getKey());
+ buf.writeInt(entry.getValue().size());
+ for (ShuffleBlockInfo sbi : entry.getValue()) {
+ Encoders.encodeShuffleBlockInfo(sbi, buf);
+ }
+ }
+ }
+
+ public long getRequestId() {
+ return requestId;
+ }
+
+ public String getAppId() {
+ return appId;
+ }
+
+ public int getShuffleId() {
+ return shuffleId;
+ }
+
+ public long getRequireId() {
+ return requireId;
+ }
+
+ public Map<Integer, List<ShuffleBlockInfo>> getPartitionToBlocks() {
+ return partitionToBlocks;
+ }
+
+ @Override
+ public boolean equals(Object o) {
Review Comment:
Ok.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] jerqi commented on a diff in pull request #742: [#133] feat(netty): Add Encoder and Decoder.
Posted by "jerqi (via GitHub)" <gi...@apache.org>.
jerqi commented on code in PR #742:
URL: https://github.com/apache/incubator-uniffle/pull/742#discussion_r1142851301
##########
common/src/main/java/org/apache/uniffle/common/netty/protocol/Encoders.java:
##########
@@ -0,0 +1,69 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License 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 permissions and
+ * limitations under the License.
+ */
+
+package org.apache.uniffle.common.netty.protocol;
+
+import java.util.List;
+
+import io.netty.buffer.ByteBuf;
+
+import org.apache.uniffle.common.ShuffleBlockInfo;
+import org.apache.uniffle.common.ShuffleServerInfo;
+import org.apache.uniffle.common.util.ByteBufUtils;
+
+public class Encoders {
+ public static void encodeShuffleServerInfo(ShuffleServerInfo shuffleServerInfo, ByteBuf byteBuf) {
+ ByteBufUtils.writeLengthAndString(byteBuf, shuffleServerInfo.getId());
+ ByteBufUtils.writeLengthAndString(byteBuf, shuffleServerInfo.getHost());
+ byteBuf.writeInt(shuffleServerInfo.getGrpcPort());
+ byteBuf.writeInt(shuffleServerInfo.getNettyPort());
+ }
+
+ public static void encodeShuffleBlockInfo(ShuffleBlockInfo shuffleBlockInfo, ByteBuf byteBuf) {
+ byteBuf.writeInt(shuffleBlockInfo.getPartitionId());
+ byteBuf.writeLong(shuffleBlockInfo.getBlockId());
+ byteBuf.writeInt(shuffleBlockInfo.getLength());
+ byteBuf.writeInt(shuffleBlockInfo.getShuffleId());
+ byteBuf.writeLong(shuffleBlockInfo.getCrc());
+ byteBuf.writeLong(shuffleBlockInfo.getTaskAttemptId());
+ ByteBufUtils.copyByteBuf(shuffleBlockInfo.getData(), byteBuf);
Review Comment:
If it's difficult to implement, we can add a `todo` here.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] zuston commented on a diff in pull request #742: [#133] feat(netty): Add Encoder and Decoder.
Posted by "zuston (via GitHub)" <gi...@apache.org>.
zuston commented on code in PR #742:
URL: https://github.com/apache/incubator-uniffle/pull/742#discussion_r1145585978
##########
common/src/main/java/org/apache/uniffle/common/netty/protocol/Message.java:
##########
@@ -53,6 +73,46 @@ public static Type decode(ByteBuf buf) {
switch (id) {
case 0:
return RPC_RESPONSE;
+ case 1:
+ return SHUFFLE_REGISTER_REQUEST;
Review Comment:
We will only replace the send/get data grpc api, right? If so, these different types are unnecessary?
##########
common/src/main/java/org/apache/uniffle/common/netty/protocol/Encoders.java:
##########
@@ -0,0 +1,71 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License 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 permissions and
+ * limitations under the License.
+ */
+
+package org.apache.uniffle.common.netty.protocol;
+
+import java.util.List;
+
+import io.netty.buffer.ByteBuf;
+
+import org.apache.uniffle.common.ShuffleBlockInfo;
+import org.apache.uniffle.common.ShuffleServerInfo;
+import org.apache.uniffle.common.util.ByteBufUtils;
+
+public class Encoders {
Review Comment:
From this implementation of `Encoders` and `Decoders`, I think they are bound with each other. How about introducing a `codec` ,which will include encode and decode method?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] leixm commented on a diff in pull request #742: [#133] feat(netty): Add Encoder and Decoder.
Posted by "leixm (via GitHub)" <gi...@apache.org>.
leixm commented on code in PR #742:
URL: https://github.com/apache/incubator-uniffle/pull/742#discussion_r1145091131
##########
common/src/main/java/org/apache/uniffle/common/netty/TransportFrameDecoder.java:
##########
@@ -0,0 +1,191 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License 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 permissions and
+ * limitations under the License.
+ */
+
+package org.apache.uniffle.common.netty;
+
+import java.util.LinkedList;
+
+import com.google.common.base.Preconditions;
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.CompositeByteBuf;
+import io.netty.buffer.Unpooled;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelInboundHandlerAdapter;
+
+import org.apache.uniffle.common.netty.protocol.Message;
+
+/**
+ * A customized frame decoder that allows intercepting raw data.
+ *
+ * <p>This behaves like Netty's frame decoder (with hard coded parameters that match this library's
+ * needs), except it allows an interceptor to be installed to read data directly before it's framed.
+ *
+ * <p>Unlike Netty's frame decoder, each frame is dispatched to child handlers as soon as it's
+ * decoded, instead of building as many frames as the current buffer allows and dispatching all of
+ * them. This allows a child handler to install an interceptor if needed.
+ *
+ * <p>If an interceptor is installed, framing stops, and data is instead fed directly to the
+ * interceptor. When the interceptor indicates that it doesn't need to read any more data, framing
+ * resumes. Interceptors should not hold references to the data buffers provided to their handle()
+ * method.
+ */
+public class TransportFrameDecoder extends ChannelInboundHandlerAdapter implements FrameDecoder {
+ private int msgSize = -1;
+ private Message.Type curType = Message.Type.UNKNOWN_TYPE;
+ private ByteBuf headerBuf = Unpooled.buffer(HEADER_SIZE, HEADER_SIZE);
+ private static final int MAX_FRAME_SIZE = Integer.MAX_VALUE;
+ private static final int UNKNOWN_FRAME_SIZE = -1;
+
+ private final LinkedList<ByteBuf> buffers = new LinkedList<>();
+
+ private long totalSize = 0;
+ private long nextFrameSize = UNKNOWN_FRAME_SIZE;
+
+ @Override
+ public void channelRead(ChannelHandlerContext ctx, Object data) {
+ ByteBuf in = (ByteBuf) data;
+ buffers.add(in);
+ totalSize += in.readableBytes();
+
+ while (!buffers.isEmpty()) {
+ ByteBuf frame = decodeNext();
+ if (frame == null) {
+ break;
+ }
+ Message msg = Message.decode(curType, frame);
Review Comment:
This happens only if there is a bug in decode, right? this should not exist.
##########
common/src/main/java/org/apache/uniffle/common/netty/MessageEncoder.java:
##########
@@ -0,0 +1,49 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License 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 permissions and
+ * limitations under the License.
+ */
+
+package org.apache.uniffle.common.netty;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelOutboundHandlerAdapter;
+import io.netty.channel.ChannelPromise;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.uniffle.common.netty.protocol.Message;
+
+public class MessageEncoder extends ChannelOutboundHandlerAdapter {
Review Comment:
sure.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] leixm commented on a diff in pull request #742: [#133] feat(netty): Add Encoder and Decoder.
Posted by "leixm (via GitHub)" <gi...@apache.org>.
leixm commented on code in PR #742:
URL: https://github.com/apache/incubator-uniffle/pull/742#discussion_r1145092044
##########
common/src/main/java/org/apache/uniffle/common/netty/MessageEncoder.java:
##########
@@ -0,0 +1,49 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License 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 permissions and
+ * limitations under the License.
+ */
+
+package org.apache.uniffle.common.netty;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelOutboundHandlerAdapter;
+import io.netty.channel.ChannelPromise;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.uniffle.common.netty.protocol.Message;
+
+public class MessageEncoder extends ChannelOutboundHandlerAdapter {
+
+ private static final Logger LOG = LoggerFactory.getLogger(MessageEncoder.class);
+
+ @Override
+ public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) {
+ // todo: support zero copy
Review Comment:
I will set up an issue later.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] jerqi commented on a diff in pull request #742: [#133] feat(netty): Add Encoder and Decoder.
Posted by "jerqi (via GitHub)" <gi...@apache.org>.
jerqi commented on code in PR #742:
URL: https://github.com/apache/incubator-uniffle/pull/742#discussion_r1141407306
##########
common/src/main/java/org/apache/uniffle/common/netty/protocol/SendShuffleDataRequest.java:
##########
@@ -0,0 +1,147 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License 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 permissions and
+ * limitations under the License.
+ */
+
+package org.apache.uniffle.common.netty.protocol;
Review Comment:
Why do we only have this class?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] leixm commented on a diff in pull request #742: [#133] feat(netty): Add Encoder and Decoder.
Posted by "leixm (via GitHub)" <gi...@apache.org>.
leixm commented on code in PR #742:
URL: https://github.com/apache/incubator-uniffle/pull/742#discussion_r1142850185
##########
common/src/main/java/org/apache/uniffle/common/netty/protocol/Encoders.java:
##########
@@ -0,0 +1,69 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License 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 permissions and
+ * limitations under the License.
+ */
+
+package org.apache.uniffle.common.netty.protocol;
+
+import java.util.List;
+
+import io.netty.buffer.ByteBuf;
+
+import org.apache.uniffle.common.ShuffleBlockInfo;
+import org.apache.uniffle.common.ShuffleServerInfo;
+import org.apache.uniffle.common.util.ByteBufUtils;
+
+public class Encoders {
+ public static void encodeShuffleServerInfo(ShuffleServerInfo shuffleServerInfo, ByteBuf byteBuf) {
+ ByteBufUtils.writeLengthAndString(byteBuf, shuffleServerInfo.getId());
+ ByteBufUtils.writeLengthAndString(byteBuf, shuffleServerInfo.getHost());
+ byteBuf.writeInt(shuffleServerInfo.getGrpcPort());
+ byteBuf.writeInt(shuffleServerInfo.getNettyPort());
+ }
+
+ public static void encodeShuffleBlockInfo(ShuffleBlockInfo shuffleBlockInfo, ByteBuf byteBuf) {
+ byteBuf.writeInt(shuffleBlockInfo.getPartitionId());
+ byteBuf.writeLong(shuffleBlockInfo.getBlockId());
+ byteBuf.writeInt(shuffleBlockInfo.getLength());
+ byteBuf.writeInt(shuffleBlockInfo.getShuffleId());
+ byteBuf.writeLong(shuffleBlockInfo.getCrc());
+ byteBuf.writeLong(shuffleBlockInfo.getTaskAttemptId());
+ ByteBufUtils.copyByteBuf(shuffleBlockInfo.getData(), byteBuf);
Review Comment:
When to use copy? When to use readSlice/CompositeByteBuf?
When we need to pass the content of ByteBuf A to ByteBuf B, `if we can manage ByteBuf B, then we can avoid copying` (such as through the readSlice method), When B will not be accessed anywhere, and its content will not passed to another ByteBuf, we can call the release method of B, and A will also be released at this time
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] leixm commented on a diff in pull request #742: [#133] feat(netty): Add Encoder and Decoder.
Posted by "leixm (via GitHub)" <gi...@apache.org>.
leixm commented on code in PR #742:
URL: https://github.com/apache/incubator-uniffle/pull/742#discussion_r1144900530
##########
common/src/test/java/org/apache/uniffle/common/netty/protocol/NettyProtocolTestUtils.java:
##########
@@ -17,128 +17,71 @@
package org.apache.uniffle.common.netty.protocol;
+import java.util.Iterator;
import java.util.List;
import java.util.Map;
-import java.util.Objects;
-
-import com.google.common.collect.Lists;
-import com.google.common.collect.Maps;
-import io.netty.buffer.ByteBuf;
import org.apache.uniffle.common.ShuffleBlockInfo;
-import org.apache.uniffle.common.ShuffleServerInfo;
public class NettyProtocolTestUtils {
- static class SendShuffleDataRequestTest extends SendShuffleDataRequest {
- SendShuffleDataRequestTest(long requestId, String appId, int shuffleId, long requireId,
- Map<Integer, List<ShuffleBlockInfo>> partitionToBlocks, long timestamp) {
- super(requestId, appId, shuffleId, requireId, partitionToBlocks, timestamp);
- }
-
- @Override
- public boolean equals(Object o) {
- if (this == o) {
- return true;
- }
- if (o == null || getClass() != o.getClass()) {
- return false;
- }
- SendShuffleDataRequestTest that = (SendShuffleDataRequestTest) o;
- Map<Integer, List<ShuffleBlockInfoTest>> map1 = Maps.newHashMap();
- Map<Integer, List<ShuffleBlockInfoTest>> map2 = Maps.newHashMap();
- for (Map.Entry<Integer, List<ShuffleBlockInfo>> entry : this.getPartitionToBlocks().entrySet()) {
- map1.put(entry.getKey(), toShuffleBlockInfoTestList(entry.getValue()));
- }
- for (Map.Entry<Integer, List<ShuffleBlockInfo>> entry : that.getPartitionToBlocks().entrySet()) {
- map2.put(entry.getKey(), toShuffleBlockInfoTestList(entry.getValue()));
- }
- return this.requestId == that.requestId
- && this.getShuffleId() == that.getShuffleId()
- && this.getRequireId() == that.getRequireId()
- && this.getTimestamp() == that.getTimestamp()
- && this.getAppId().equals(that.getAppId())
- && map1.equals(map2);
- }
-
- @Override
- public int hashCode() {
- return Objects.hash(requestId, this.getAppId(), this.getShuffleId(), this.getRequireId(),
- this.getPartitionToBlocks(), this.getTimestamp());
- }
+ private static boolean compareShuffleBlockInfo(ShuffleBlockInfo blockInfo1, ShuffleBlockInfo blockInfo2) {
+ return blockInfo1.getPartitionId() == blockInfo2.getPartitionId()
+ && blockInfo1.getBlockId() == blockInfo2.getBlockId()
+ && blockInfo1.getLength() == blockInfo2.getLength()
+ && blockInfo1.getShuffleId() == blockInfo2.getShuffleId()
+ && blockInfo1.getCrc() == blockInfo2.getCrc()
+ && blockInfo1.getTaskAttemptId() == blockInfo2.getTaskAttemptId()
+ && blockInfo1.getUncompressLength() == blockInfo2.getUncompressLength()
+ && blockInfo1.getFreeMemory() == blockInfo2.getFreeMemory()
+ && blockInfo1.getData().equals(blockInfo2.getData())
+ && blockInfo1.getShuffleServerInfos().equals(blockInfo2.getShuffleServerInfos());
}
- static class ShuffleBlockInfoTest extends ShuffleBlockInfo {
- ShuffleBlockInfoTest(int shuffleId, int partitionId, long blockId, int length, long crc,
- ByteBuf data, List<ShuffleServerInfo> shuffleServerInfos, int uncompressLength,
- long freeMemory, long taskAttemptId) {
- super(shuffleId, partitionId, blockId, length, crc, data, shuffleServerInfos, uncompressLength,
- freeMemory, taskAttemptId);
- }
- @Override
- public boolean equals(Object o) {
- if (this == o) {
- return true;
- }
- if (o == null || getClass() != o.getClass()) {
+ private static boolean compareBlockList(List<ShuffleBlockInfo> list1, List<ShuffleBlockInfo> list2) {
+ if (list1 == null || list2 == null || list1.size() != list2.size()) {
+ return false;
+ }
+ for (int i = 0; i < list1.size(); i++) {
+ if (!compareShuffleBlockInfo(list1.get(i), list2.get(i))) {
return false;
}
- ShuffleBlockInfoTest that = (ShuffleBlockInfoTest) o;
- return this.getPartitionId() == that.getPartitionId()
- && this.getBlockId() == that.getBlockId()
- && this.getLength() == that.getLength()
- && this.getShuffleId() == that.getShuffleId()
- && this.getCrc() == that.getCrc()
- && this.getTaskAttemptId() == that.getTaskAttemptId()
- && this.getUncompressLength() == that.getUncompressLength()
- && this.getFreeMemory() == that.getFreeMemory()
- && this.getData().equals(that.getData())
- && this.getShuffleServerInfos().equals(that.getShuffleServerInfos());
- }
-
- @Override
- public int hashCode() {
- return Objects.hash(this.getPartitionId(), this.getBlockId(), this.getLength(), this.getShuffleId(),
- this.getCrc(), this.getTaskAttemptId(), this.getData(), this.getShuffleServerInfos(),
- this.getUncompressLength(), this.getFreeMemory());
}
+ return true;
}
- private static SendShuffleDataRequestTest toSendShuffleDataRequestTest(
- SendShuffleDataRequest sendShuffleDataRequest) {
- return new SendShuffleDataRequestTest(sendShuffleDataRequest.requestId,
- sendShuffleDataRequest.getAppId(),
- sendShuffleDataRequest.getShuffleId(),
- sendShuffleDataRequest.getRequireId(),
- sendShuffleDataRequest.getPartitionToBlocks(),
- sendShuffleDataRequest.getTimestamp());
- }
+ private static boolean comparePartitionToBlockList(Map<Integer, List<ShuffleBlockInfo>> m1,
+ Map<Integer, List<ShuffleBlockInfo>> m2) {
+ Iterator<Map.Entry<Integer, List<ShuffleBlockInfo>>> iter1 = m1.entrySet().iterator();
+ while (iter1.hasNext()) {
+ Map.Entry<Integer, List<ShuffleBlockInfo>> entry1 = iter1.next();
+ List<ShuffleBlockInfo> m1value = entry1.getValue() == null ? null : entry1.getValue();
Review Comment:
that's a mistake.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] jerqi commented on pull request #742: [#133] feat(netty): Add Encoder and Decoder.
Posted by "jerqi (via GitHub)" <gi...@apache.org>.
jerqi commented on PR #742:
URL: https://github.com/apache/incubator-uniffle/pull/742#issuecomment-1479790015
cc @zuston Could help me review this pr, too?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] leixm commented on a diff in pull request #742: [#133] feat(netty): Add Encoder and Decoder.
Posted by "leixm (via GitHub)" <gi...@apache.org>.
leixm commented on code in PR #742:
URL: https://github.com/apache/incubator-uniffle/pull/742#discussion_r1141938675
##########
internal-client/src/main/java/org/apache/uniffle/client/impl/grpc/ShuffleServerGrpcClient.java:
##########
@@ -319,7 +319,7 @@ public RssSendShuffleDataResponse sendShuffleData(RssSendShuffleDataRequest requ
.setLength(sbi.getLength())
.setTaskAttemptId(sbi.getTaskAttemptId())
.setUncompressLength(sbi.getUncompressLength())
- .setData(ByteString.copyFrom(sbi.getData()))
+ .setData(ByteString.copyFrom(sbi.getData().nioBuffer()))
Review Comment:
setData needs to pass in a parameter value of type ByteString, which is a type of protobuf, which doesn’t seem to work.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] leixm commented on pull request #742: [#133] feat(netty): Add Encoder and Decoder.
Posted by "leixm (via GitHub)" <gi...@apache.org>.
leixm commented on PR #742:
URL: https://github.com/apache/incubator-uniffle/pull/742#issuecomment-1475999150
I will fix the CI issue later.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] jerqi commented on a diff in pull request #742: [#133] feat(netty): Add Encoder and Decoder.
Posted by "jerqi (via GitHub)" <gi...@apache.org>.
jerqi commented on code in PR #742:
URL: https://github.com/apache/incubator-uniffle/pull/742#discussion_r1141406412
##########
common/src/main/java/org/apache/uniffle/common/ShuffleBlockInfo.java:
##########
@@ -27,14 +29,14 @@ public class ShuffleBlockInfo {
private int shuffleId;
private long crc;
private long taskAttemptId;
- private byte[] data;
+ private ByteBuf data;
private List<ShuffleServerInfo> shuffleServerInfos;
private int uncompressLength;
private long freeMemory;
public ShuffleBlockInfo(int shuffleId, int partitionId, long blockId, int length, long crc,
- byte[] data, List<ShuffleServerInfo> shuffleServerInfos,
- int uncompressLength, int freeMemory, long taskAttemptId) {
+ ByteBuf data, List<ShuffleServerInfo> shuffleServerInfos,
+ int uncompressLength, long freeMemory, long taskAttemptId) {
this.partitionId = partitionId;
Review Comment:
We can keep the original constructor for tests and add a new constructor for new implement.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] jerqi commented on a diff in pull request #742: [#133] feat(netty): Add Encoder and Decoder.
Posted by "jerqi (via GitHub)" <gi...@apache.org>.
jerqi commented on code in PR #742:
URL: https://github.com/apache/incubator-uniffle/pull/742#discussion_r1145098193
##########
common/src/main/java/org/apache/uniffle/common/netty/TransportFrameDecoder.java:
##########
@@ -0,0 +1,191 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License 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 permissions and
+ * limitations under the License.
+ */
+
+package org.apache.uniffle.common.netty;
+
+import java.util.LinkedList;
+
+import com.google.common.base.Preconditions;
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.CompositeByteBuf;
+import io.netty.buffer.Unpooled;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelInboundHandlerAdapter;
+
+import org.apache.uniffle.common.netty.protocol.Message;
+
+/**
+ * A customized frame decoder that allows intercepting raw data.
+ *
+ * <p>This behaves like Netty's frame decoder (with hard coded parameters that match this library's
+ * needs), except it allows an interceptor to be installed to read data directly before it's framed.
+ *
+ * <p>Unlike Netty's frame decoder, each frame is dispatched to child handlers as soon as it's
+ * decoded, instead of building as many frames as the current buffer allows and dispatching all of
+ * them. This allows a child handler to install an interceptor if needed.
+ *
+ * <p>If an interceptor is installed, framing stops, and data is instead fed directly to the
+ * interceptor. When the interceptor indicates that it doesn't need to read any more data, framing
+ * resumes. Interceptors should not hold references to the data buffers provided to their handle()
+ * method.
+ */
+public class TransportFrameDecoder extends ChannelInboundHandlerAdapter implements FrameDecoder {
+ private int msgSize = -1;
+ private Message.Type curType = Message.Type.UNKNOWN_TYPE;
+ private ByteBuf headerBuf = Unpooled.buffer(HEADER_SIZE, HEADER_SIZE);
+ private static final int MAX_FRAME_SIZE = Integer.MAX_VALUE;
+ private static final int UNKNOWN_FRAME_SIZE = -1;
+
+ private final LinkedList<ByteBuf> buffers = new LinkedList<>();
+
+ private long totalSize = 0;
+ private long nextFrameSize = UNKNOWN_FRAME_SIZE;
+
+ @Override
+ public void channelRead(ChannelHandlerContext ctx, Object data) {
+ ByteBuf in = (ByteBuf) data;
+ buffers.add(in);
+ totalSize += in.readableBytes();
+
+ while (!buffers.isEmpty()) {
+ ByteBuf frame = decodeNext();
+ if (frame == null) {
+ break;
+ }
+ Message msg = Message.decode(curType, frame);
Review Comment:
We could add a `todo` first. We need more survery.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] leixm commented on a diff in pull request #742: [#133] feat(netty): Add Encoder and Decoder.
Posted by "leixm (via GitHub)" <gi...@apache.org>.
leixm commented on code in PR #742:
URL: https://github.com/apache/incubator-uniffle/pull/742#discussion_r1144178568
##########
common/src/main/java/org/apache/uniffle/common/netty/protocol/Encoders.java:
##########
@@ -0,0 +1,69 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License 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 permissions and
+ * limitations under the License.
+ */
+
+package org.apache.uniffle.common.netty.protocol;
+
+import java.util.List;
+
+import io.netty.buffer.ByteBuf;
+
+import org.apache.uniffle.common.ShuffleBlockInfo;
+import org.apache.uniffle.common.ShuffleServerInfo;
+import org.apache.uniffle.common.util.ByteBufUtils;
+
+public class Encoders {
+ public static void encodeShuffleServerInfo(ShuffleServerInfo shuffleServerInfo, ByteBuf byteBuf) {
+ ByteBufUtils.writeLengthAndString(byteBuf, shuffleServerInfo.getId());
+ ByteBufUtils.writeLengthAndString(byteBuf, shuffleServerInfo.getHost());
+ byteBuf.writeInt(shuffleServerInfo.getGrpcPort());
+ byteBuf.writeInt(shuffleServerInfo.getNettyPort());
+ }
+
+ public static void encodeShuffleBlockInfo(ShuffleBlockInfo shuffleBlockInfo, ByteBuf byteBuf) {
+ byteBuf.writeInt(shuffleBlockInfo.getPartitionId());
+ byteBuf.writeLong(shuffleBlockInfo.getBlockId());
+ byteBuf.writeInt(shuffleBlockInfo.getLength());
+ byteBuf.writeInt(shuffleBlockInfo.getShuffleId());
+ byteBuf.writeLong(shuffleBlockInfo.getCrc());
+ byteBuf.writeLong(shuffleBlockInfo.getTaskAttemptId());
+ ByteBufUtils.copyByteBuf(shuffleBlockInfo.getData(), byteBuf);
Review Comment:
> > Do we need to release memory after copying here?
> > ```
> > shuffleBlockInfo.getData().release().
> > ```
>
> If ByteBuf uses heap memory, I think there is no need to manually call release and this part of memory will be reclaimed during gc. Secondly, there may be retries. The current encode does not mean that the request must be successful.
I think we better call `release()` method here, see https://netty.io/wiki/reference-counted-objects.html.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] leixm commented on a diff in pull request #742: [#133] feat(netty): Add Encoder and Decoder.
Posted by "leixm (via GitHub)" <gi...@apache.org>.
leixm commented on code in PR #742:
URL: https://github.com/apache/incubator-uniffle/pull/742#discussion_r1141932555
##########
common/src/main/java/org/apache/uniffle/common/ShuffleBlockInfo.java:
##########
@@ -27,14 +29,14 @@ public class ShuffleBlockInfo {
private int shuffleId;
private long crc;
private long taskAttemptId;
- private byte[] data;
+ private ByteBuf data;
private List<ShuffleServerInfo> shuffleServerInfos;
private int uncompressLength;
private long freeMemory;
public ShuffleBlockInfo(int shuffleId, int partitionId, long blockId, int length, long crc,
- byte[] data, List<ShuffleServerInfo> shuffleServerInfos,
- int uncompressLength, int freeMemory, long taskAttemptId) {
+ ByteBuf data, List<ShuffleServerInfo> shuffleServerInfos,
+ int uncompressLength, long freeMemory, long taskAttemptId) {
this.partitionId = partitionId;
Review Comment:
OK, I got it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] jerqi commented on a diff in pull request #742: [#133] feat(netty): Add Encoder and Decoder.
Posted by "jerqi (via GitHub)" <gi...@apache.org>.
jerqi commented on code in PR #742:
URL: https://github.com/apache/incubator-uniffle/pull/742#discussion_r1144293643
##########
common/src/test/java/org/apache/uniffle/common/netty/protocol/NettyProtocolTestUtils.java:
##########
@@ -0,0 +1,119 @@
+package org.apache.uniffle.common.netty.protocol;
+
+import com.google.common.collect.Lists;
+import com.google.common.collect.Maps;
+import io.netty.buffer.ByteBuf;
+import org.apache.uniffle.common.ShuffleBlockInfo;
+import org.apache.uniffle.common.ShuffleServerInfo;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+
+public class NettyProtocolTestUtils {
+
+ static class SendShuffleDataRequestTest extends SendShuffleDataRequest {
+ public SendShuffleDataRequestTest(long requestId, String appId, int shuffleId, long requireId, Map<Integer, List<ShuffleBlockInfo>> partitionToBlocks, long timestamp) {
+ super(requestId, appId, shuffleId, requireId, partitionToBlocks, timestamp);
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ SendShuffleDataRequestTest that = (SendShuffleDataRequestTest) o;
+ Map<Integer, List<ShuffleBlockInfoTest>> map1 = Maps.newHashMap();
+ Map<Integer, List<ShuffleBlockInfoTest>> map2 = Maps.newHashMap();
+ for (Map.Entry<Integer, List<ShuffleBlockInfo>> entry : this.getPartitionToBlocks().entrySet()) {
+ map1.put(entry.getKey(), toShuffleBlockInfoTestList(entry.getValue()));
+ }
+ for (Map.Entry<Integer, List<ShuffleBlockInfo>> entry : that.getPartitionToBlocks().entrySet()) {
+ map2.put(entry.getKey(), toShuffleBlockInfoTestList(entry.getValue()));
+ }
+ return this.requestId == that.requestId
+ && this.getShuffleId() == that.getShuffleId()
+ && this.getRequireId() == that.getRequireId()
+ && this.getTimestamp() == that.getTimestamp()
+ && this.getAppId().equals(that.getAppId())
+ && map1.equals(map2);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(requestId, this.getAppId(), this.getShuffleId(), this.getRequireId(),
+ this.getPartitionToBlocks(), this.getTimestamp());
+ }
+ }
+
+ static class ShuffleBlockInfoTest extends ShuffleBlockInfo {
+ public ShuffleBlockInfoTest(int shuffleId, int partitionId, long blockId, int length, long crc, ByteBuf data, List<ShuffleServerInfo> shuffleServerInfos, int uncompressLength, long freeMemory, long taskAttemptId) {
+ super(shuffleId, partitionId, blockId, length, crc, data, shuffleServerInfos, uncompressLength, freeMemory, taskAttemptId);
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ ShuffleBlockInfoTest that = (ShuffleBlockInfoTest) o;
+ return this.getPartitionId() == that.getPartitionId()
+ && this.getBlockId() == that.getBlockId()
+ && this.getLength() == that.getLength()
+ && this.getShuffleId() == that.getShuffleId()
+ && this.getCrc() == that.getCrc()
+ && this.getTaskAttemptId() == that.getTaskAttemptId()
+ && this.getUncompressLength() == that.getUncompressLength()
+ && this.getFreeMemory() == that.getFreeMemory()
+ && this.getData().equals(that.getData())
+ && this.getShuffleServerInfos().equals(that.getShuffleServerInfos());
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(this.getPartitionId(), this.getBlockId(), this.getLength(), this.getShuffleId(),
+ this.getCrc(), this.getTaskAttemptId(), this.getData(), this.getShuffleServerInfos(),
+ this.getUncompressLength(), this.getFreeMemory());
+ }
+ }
+
+ private static SendShuffleDataRequestTest toSendShuffleDataRequestTest(SendShuffleDataRequest sendShuffleDataRequest) {
+ return new SendShuffleDataRequestTest(sendShuffleDataRequest.requestId,
+ sendShuffleDataRequest.getAppId(),
+ sendShuffleDataRequest.getShuffleId(),
+ sendShuffleDataRequest.getRequireId(),
+ sendShuffleDataRequest.getPartitionToBlocks(),
+ sendShuffleDataRequest.getTimestamp());
+ }
+
+ private static List<ShuffleBlockInfoTest> toShuffleBlockInfoTestList(List<ShuffleBlockInfo> shuffleBlockInfoList) {
+ List<ShuffleBlockInfoTest> res = Lists.newArrayList();
+ for (ShuffleBlockInfo shuffleBlockInfo : shuffleBlockInfoList) {
+ res.add(toShuffleBlockInfoTest(shuffleBlockInfo));
+ }
+ return res;
+ }
+
+ private static ShuffleBlockInfoTest toShuffleBlockInfoTest(ShuffleBlockInfo shuffleBlockInfo) {
+ return new ShuffleBlockInfoTest(shuffleBlockInfo.getShuffleId(),
+ shuffleBlockInfo.getPartitionId(),
+ shuffleBlockInfo.getBlockId(),
+ shuffleBlockInfo.getLength(),
+ shuffleBlockInfo.getCrc(),
+ shuffleBlockInfo.getData(),
+ shuffleBlockInfo.getShuffleServerInfos(),
+ shuffleBlockInfo.getUncompressLength(),
+ shuffleBlockInfo.getFreeMemory(),
+ shuffleBlockInfo.getTaskAttemptId());
+ }
+
+ public static boolean compareSendShuffleDataRequest(SendShuffleDataRequest req1, SendShuffleDataRequest req2) {
Review Comment:
It's ok to add the compare method of list and map.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] leixm commented on a diff in pull request #742: [#133] feat(netty): Add Encoder and Decoder.
Posted by "leixm (via GitHub)" <gi...@apache.org>.
leixm commented on code in PR #742:
URL: https://github.com/apache/incubator-uniffle/pull/742#discussion_r1141930285
##########
internal-client/src/main/java/org/apache/uniffle/client/impl/grpc/ShuffleServerGrpcClient.java:
##########
@@ -319,7 +319,7 @@ public RssSendShuffleDataResponse sendShuffleData(RssSendShuffleDataRequest requ
.setLength(sbi.getLength())
.setTaskAttemptId(sbi.getTaskAttemptId())
.setUncompressLength(sbi.getUncompressLength())
- .setData(ByteString.copyFrom(sbi.getData()))
+ .setData(ByteString.copyFrom(sbi.getData().nioBuffer()))
Review Comment:
ByteBuf.nioBuffer() will not cause a copy of the byte array, So there is no difference from before the modification.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] jerqi commented on a diff in pull request #742: [#133] feat(netty): Add Encoder and Decoder.
Posted by "jerqi (via GitHub)" <gi...@apache.org>.
jerqi commented on code in PR #742:
URL: https://github.com/apache/incubator-uniffle/pull/742#discussion_r1141406940
##########
internal-client/src/main/java/org/apache/uniffle/client/impl/grpc/ShuffleServerGrpcClient.java:
##########
@@ -319,7 +319,7 @@ public RssSendShuffleDataResponse sendShuffleData(RssSendShuffleDataRequest requ
.setLength(sbi.getLength())
.setTaskAttemptId(sbi.getTaskAttemptId())
.setUncompressLength(sbi.getUncompressLength())
- .setData(ByteString.copyFrom(sbi.getData()))
+ .setData(ByteString.copyFrom(sbi.getData().nioBuffer()))
Review Comment:
Could we reduce data copy here?
##########
common/src/main/java/org/apache/uniffle/common/netty/protocol/SendShuffleDataRequest.java:
##########
@@ -0,0 +1,147 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License 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 permissions and
+ * limitations under the License.
+ */
+
+package org.apache.uniffle.common.netty.protocol;
Review Comment:
Why do we only have this method?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] jerqi commented on a diff in pull request #742: [#133] feat(netty): Add Encoder and Decoder.
Posted by "jerqi (via GitHub)" <gi...@apache.org>.
jerqi commented on code in PR #742:
URL: https://github.com/apache/incubator-uniffle/pull/742#discussion_r1142837168
##########
common/src/main/java/org/apache/uniffle/common/netty/protocol/SendShuffleDataRequest.java:
##########
@@ -0,0 +1,153 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License 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 permissions and
+ * limitations under the License.
+ */
+
+package org.apache.uniffle.common.netty.protocol;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+
+import com.google.common.collect.Lists;
+import com.google.common.collect.Maps;
+import io.netty.buffer.ByteBuf;
+
+import org.apache.uniffle.common.ShuffleBlockInfo;
+import org.apache.uniffle.common.util.ByteBufUtils;
+
+public class SendShuffleDataRequest extends Message {
+ public long requestId;
+ private String appId;
+ private int shuffleId;
+ private long requireId;
+ private Map<Integer, List<ShuffleBlockInfo>> partitionToBlocks;
+ private long timestamp;
+
+ public SendShuffleDataRequest(long requestId, String appId, int shuffleId, long requireId,
+ Map<Integer, List<ShuffleBlockInfo>> partitionToBlocks, long timestamp) {
+ this.requestId = requestId;
+ this.appId = appId;
+ this.shuffleId = shuffleId;
+ this.requireId = requireId;
+ this.partitionToBlocks = partitionToBlocks;
+ this.timestamp = timestamp;
+ }
+
+ @Override
+ public Type type() {
+ return Type.SEND_SHUFFLE_DATA_REQUEST;
+ }
+
+ @Override
+ public int encodedLength() {
+ int encodeLength = Long.BYTES + ByteBufUtils.encodedLength(appId) + Integer.BYTES + Long.BYTES + Integer.BYTES;
+ for (Map.Entry<Integer, List<ShuffleBlockInfo>> entry : partitionToBlocks.entrySet()) {
+ encodeLength += 2 * Integer.BYTES;
+ for (ShuffleBlockInfo sbi : entry.getValue()) {
+ encodeLength += Encoders.encodeLengthOfShuffleBlockInfo(sbi);
+ }
+ }
+ return encodeLength + Long.BYTES;
+ }
+
+ @Override
+ public void encode(ByteBuf buf) {
+ buf.writeLong(requestId);
+ ByteBufUtils.writeLengthAndString(buf, appId);
+ buf.writeInt(shuffleId);
+ buf.writeLong(requireId);
+ encodePartitionData(buf);
+ buf.writeLong(timestamp);
+ }
+
+ private static Map<Integer, List<ShuffleBlockInfo>> decodePartitionData(ByteBuf byteBuf) {
+ Map<Integer, List<ShuffleBlockInfo>> partitionToBlocks = Maps.newHashMap();
+ int lengthOfPartitionData = byteBuf.readInt();
+ for (int i = 0; i < lengthOfPartitionData; i++) {
+ int partitionId = byteBuf.readInt();
+ int lengthOfShuffleBlocks = byteBuf.readInt();
+ List<ShuffleBlockInfo> shuffleBlockInfoList = Lists.newArrayList();
+ for (int j = 0; j < lengthOfShuffleBlocks; j++) {
+ shuffleBlockInfoList.add(Decoders.decodeShuffleBlockInfo(byteBuf));
+ }
+ partitionToBlocks.put(partitionId, shuffleBlockInfoList);
+ }
+ return partitionToBlocks;
+ }
+
+ public static SendShuffleDataRequest decode(ByteBuf byteBuf) {
+ long requestId = byteBuf.readLong();
+ String appId = ByteBufUtils.readLengthAndString(byteBuf);
+ int shuffleId = byteBuf.readInt();
+ long requireId = byteBuf.readLong();
+ Map<Integer, List<ShuffleBlockInfo>> partitionToBlocks = decodePartitionData(byteBuf);
+ long timestamp = byteBuf.readLong();
+ return new SendShuffleDataRequest(requestId, appId, shuffleId, requireId, partitionToBlocks, timestamp);
+ }
+
+ private void encodePartitionData(ByteBuf buf) {
+ buf.writeInt(partitionToBlocks.size());
+ for (Map.Entry<Integer, List<ShuffleBlockInfo>> entry : partitionToBlocks.entrySet()) {
+ buf.writeInt(entry.getKey());
+ buf.writeInt(entry.getValue().size());
+ for (ShuffleBlockInfo sbi : entry.getValue()) {
+ Encoders.encodeShuffleBlockInfo(sbi, buf);
+ }
+ }
+ }
+
+ public long getRequestId() {
+ return requestId;
+ }
+
+ public String getAppId() {
+ return appId;
+ }
+
+ public int getShuffleId() {
+ return shuffleId;
+ }
+
+ public long getRequireId() {
+ return requireId;
+ }
+
+ public Map<Integer, List<ShuffleBlockInfo>> getPartitionToBlocks() {
+ return partitionToBlocks;
+ }
+
+ @Override
+ public boolean equals(Object o) {
Review Comment:
> for test, see SendShuffleDataRequestTest.java.
We can create another method like `compareSendShuffleDataRequest` in the test code.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] leixm commented on a diff in pull request #742: [#133] feat(netty): Add Encoder and Decoder.
Posted by "leixm (via GitHub)" <gi...@apache.org>.
leixm commented on code in PR #742:
URL: https://github.com/apache/incubator-uniffle/pull/742#discussion_r1144199304
##########
common/src/test/java/org/apache/uniffle/common/netty/protocol/NettyProtocolTestUtils.java:
##########
@@ -0,0 +1,119 @@
+package org.apache.uniffle.common.netty.protocol;
+
+import com.google.common.collect.Lists;
+import com.google.common.collect.Maps;
+import io.netty.buffer.ByteBuf;
+import org.apache.uniffle.common.ShuffleBlockInfo;
+import org.apache.uniffle.common.ShuffleServerInfo;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+
+public class NettyProtocolTestUtils {
+
+ static class SendShuffleDataRequestTest extends SendShuffleDataRequest {
+ public SendShuffleDataRequestTest(long requestId, String appId, int shuffleId, long requireId, Map<Integer, List<ShuffleBlockInfo>> partitionToBlocks, long timestamp) {
+ super(requestId, appId, shuffleId, requireId, partitionToBlocks, timestamp);
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ SendShuffleDataRequestTest that = (SendShuffleDataRequestTest) o;
+ Map<Integer, List<ShuffleBlockInfoTest>> map1 = Maps.newHashMap();
+ Map<Integer, List<ShuffleBlockInfoTest>> map2 = Maps.newHashMap();
+ for (Map.Entry<Integer, List<ShuffleBlockInfo>> entry : this.getPartitionToBlocks().entrySet()) {
+ map1.put(entry.getKey(), toShuffleBlockInfoTestList(entry.getValue()));
+ }
+ for (Map.Entry<Integer, List<ShuffleBlockInfo>> entry : that.getPartitionToBlocks().entrySet()) {
+ map2.put(entry.getKey(), toShuffleBlockInfoTestList(entry.getValue()));
+ }
+ return this.requestId == that.requestId
+ && this.getShuffleId() == that.getShuffleId()
+ && this.getRequireId() == that.getRequireId()
+ && this.getTimestamp() == that.getTimestamp()
+ && this.getAppId().equals(that.getAppId())
+ && map1.equals(map2);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(requestId, this.getAppId(), this.getShuffleId(), this.getRequireId(),
+ this.getPartitionToBlocks(), this.getTimestamp());
+ }
+ }
+
+ static class ShuffleBlockInfoTest extends ShuffleBlockInfo {
+ public ShuffleBlockInfoTest(int shuffleId, int partitionId, long blockId, int length, long crc, ByteBuf data, List<ShuffleServerInfo> shuffleServerInfos, int uncompressLength, long freeMemory, long taskAttemptId) {
+ super(shuffleId, partitionId, blockId, length, crc, data, shuffleServerInfos, uncompressLength, freeMemory, taskAttemptId);
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ ShuffleBlockInfoTest that = (ShuffleBlockInfoTest) o;
+ return this.getPartitionId() == that.getPartitionId()
+ && this.getBlockId() == that.getBlockId()
+ && this.getLength() == that.getLength()
+ && this.getShuffleId() == that.getShuffleId()
+ && this.getCrc() == that.getCrc()
+ && this.getTaskAttemptId() == that.getTaskAttemptId()
+ && this.getUncompressLength() == that.getUncompressLength()
+ && this.getFreeMemory() == that.getFreeMemory()
+ && this.getData().equals(that.getData())
+ && this.getShuffleServerInfos().equals(that.getShuffleServerInfos());
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(this.getPartitionId(), this.getBlockId(), this.getLength(), this.getShuffleId(),
+ this.getCrc(), this.getTaskAttemptId(), this.getData(), this.getShuffleServerInfos(),
+ this.getUncompressLength(), this.getFreeMemory());
+ }
+ }
+
+ private static SendShuffleDataRequestTest toSendShuffleDataRequestTest(SendShuffleDataRequest sendShuffleDataRequest) {
+ return new SendShuffleDataRequestTest(sendShuffleDataRequest.requestId,
+ sendShuffleDataRequest.getAppId(),
+ sendShuffleDataRequest.getShuffleId(),
+ sendShuffleDataRequest.getRequireId(),
+ sendShuffleDataRequest.getPartitionToBlocks(),
+ sendShuffleDataRequest.getTimestamp());
+ }
+
+ private static List<ShuffleBlockInfoTest> toShuffleBlockInfoTestList(List<ShuffleBlockInfo> shuffleBlockInfoList) {
+ List<ShuffleBlockInfoTest> res = Lists.newArrayList();
+ for (ShuffleBlockInfo shuffleBlockInfo : shuffleBlockInfoList) {
+ res.add(toShuffleBlockInfoTest(shuffleBlockInfo));
+ }
+ return res;
+ }
+
+ private static ShuffleBlockInfoTest toShuffleBlockInfoTest(ShuffleBlockInfo shuffleBlockInfo) {
+ return new ShuffleBlockInfoTest(shuffleBlockInfo.getShuffleId(),
+ shuffleBlockInfo.getPartitionId(),
+ shuffleBlockInfo.getBlockId(),
+ shuffleBlockInfo.getLength(),
+ shuffleBlockInfo.getCrc(),
+ shuffleBlockInfo.getData(),
+ shuffleBlockInfo.getShuffleServerInfos(),
+ shuffleBlockInfo.getUncompressLength(),
+ shuffleBlockInfo.getFreeMemory(),
+ shuffleBlockInfo.getTaskAttemptId());
+ }
+
+ public static boolean compareSendShuffleDataRequest(SendShuffleDataRequest req1, SendShuffleDataRequest req2) {
Review Comment:
> Class SendShuffleDataRequestTest seems unnecessary.
>
> ```
> return req1.getPartitionId() == req2.getPartitionId()
> && req1.getBlockId() == req2.getBlockId()
> && req1.getLength() == req2.getLength()
> && req1.getShuffleId() == req2.getShuffleId()
> && req1.getCrc() == req2.getCrc()
> && req1.getTaskAttemptId() == req2.getTaskAttemptId()
> && req1.getUncompressLength() == req2.getUncompressLength()
> && req1.getFreeMemory() == req2.getFreeMemory()
> && req1.getData().equals(req2.getData())
> && CompareShuffleInfos(req1.getShuffleServerInfos(), req2.getShuffleServerInfos());
> ```
If we don't rewrite the equals method, we need to write a bunch of redundant methods, such as list, compare of map content.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] jerqi commented on a diff in pull request #742: [#133] feat(netty): Add Encoder and Decoder.
Posted by "jerqi (via GitHub)" <gi...@apache.org>.
jerqi commented on code in PR #742:
URL: https://github.com/apache/incubator-uniffle/pull/742#discussion_r1141942533
##########
common/src/main/java/org/apache/uniffle/common/netty/protocol/SendShuffleDataRequest.java:
##########
@@ -0,0 +1,147 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License 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 permissions and
+ * limitations under the License.
+ */
+
+package org.apache.uniffle.common.netty.protocol;
Review Comment:
OK, I got it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] leixm commented on a diff in pull request #742: [#133] feat(netty): Add Encoder and Decoder.
Posted by "leixm (via GitHub)" <gi...@apache.org>.
leixm commented on code in PR #742:
URL: https://github.com/apache/incubator-uniffle/pull/742#discussion_r1142834272
##########
common/src/main/java/org/apache/uniffle/common/netty/MessageEncoder.java:
##########
@@ -0,0 +1,49 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License 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 permissions and
+ * limitations under the License.
+ */
+
+package org.apache.uniffle.common.netty;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelOutboundHandlerAdapter;
+import io.netty.channel.ChannelPromise;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.uniffle.common.netty.protocol.Message;
+
+public class MessageEncoder extends ChannelOutboundHandlerAdapter {
+
+ private static final Logger LOG = LoggerFactory.getLogger(MessageEncoder.class);
+
+ @Override
+ public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) {
+ // todo: support zero copy
+ Message message = (Message) msg;
+ int encodeLength = message.encodedLength();
+ ByteBuf byteBuf = ctx.alloc().buffer(FrameDecoder.HEADER_SIZE + encodeLength);
+ try {
+ byteBuf.writeInt(encodeLength);
+ byteBuf.writeByte(message.type().id());
+ message.encode(byteBuf);
+ } catch (Exception e) {
+ LOG.error("Unexpected exception during process encode!", e);
+ byteBuf.release();
+ }
+ ctx.writeAndFlush(byteBuf);
Review Comment:
writeAndFlush will release byteBuf.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] jerqi commented on a diff in pull request #742: [#133] feat(netty): Add Encoder and Decoder.
Posted by "jerqi (via GitHub)" <gi...@apache.org>.
jerqi commented on code in PR #742:
URL: https://github.com/apache/incubator-uniffle/pull/742#discussion_r1141935430
##########
internal-client/src/main/java/org/apache/uniffle/client/impl/grpc/ShuffleServerGrpcClient.java:
##########
@@ -319,7 +319,7 @@ public RssSendShuffleDataResponse sendShuffleData(RssSendShuffleDataRequest requ
.setLength(sbi.getLength())
.setTaskAttemptId(sbi.getTaskAttemptId())
.setUncompressLength(sbi.getUncompressLength())
- .setData(ByteString.copyFrom(sbi.getData()))
+ .setData(ByteString.copyFrom(sbi.getData().nioBuffer()))
Review Comment:
Could we remove `ByteString.copyFrom`?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] leixm commented on pull request #742: [#133] feat(netty): Add Encoder and Decoder.
Posted by "leixm (via GitHub)" <gi...@apache.org>.
leixm commented on PR #742:
URL: https://github.com/apache/incubator-uniffle/pull/742#issuecomment-1482171829
@smallzhongfeng @zuston @jerqi All done!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] zuston commented on a diff in pull request #742: [#133] feat(netty): Add Encoder and Decoder.
Posted by "zuston (via GitHub)" <gi...@apache.org>.
zuston commented on code in PR #742:
URL: https://github.com/apache/incubator-uniffle/pull/742#discussion_r1145579974
##########
common/src/main/java/org/apache/uniffle/common/ShuffleBlockInfo.java:
##########
@@ -27,14 +31,21 @@ public class ShuffleBlockInfo {
private int shuffleId;
private long crc;
private long taskAttemptId;
- private byte[] data;
+ private ByteBuf data;
Review Comment:
This is OK for me, let's make netty first
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] leixm commented on a diff in pull request #742: [#133] feat(netty): Add Encoder and Decoder.
Posted by "leixm (via GitHub)" <gi...@apache.org>.
leixm commented on code in PR #742:
URL: https://github.com/apache/incubator-uniffle/pull/742#discussion_r1142863019
##########
common/src/main/java/org/apache/uniffle/common/netty/protocol/Encoders.java:
##########
@@ -0,0 +1,69 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License 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 permissions and
+ * limitations under the License.
+ */
+
+package org.apache.uniffle.common.netty.protocol;
+
+import java.util.List;
+
+import io.netty.buffer.ByteBuf;
+
+import org.apache.uniffle.common.ShuffleBlockInfo;
+import org.apache.uniffle.common.ShuffleServerInfo;
+import org.apache.uniffle.common.util.ByteBufUtils;
+
+public class Encoders {
+ public static void encodeShuffleServerInfo(ShuffleServerInfo shuffleServerInfo, ByteBuf byteBuf) {
+ ByteBufUtils.writeLengthAndString(byteBuf, shuffleServerInfo.getId());
+ ByteBufUtils.writeLengthAndString(byteBuf, shuffleServerInfo.getHost());
+ byteBuf.writeInt(shuffleServerInfo.getGrpcPort());
+ byteBuf.writeInt(shuffleServerInfo.getNettyPort());
+ }
+
+ public static void encodeShuffleBlockInfo(ShuffleBlockInfo shuffleBlockInfo, ByteBuf byteBuf) {
+ byteBuf.writeInt(shuffleBlockInfo.getPartitionId());
+ byteBuf.writeLong(shuffleBlockInfo.getBlockId());
+ byteBuf.writeInt(shuffleBlockInfo.getLength());
+ byteBuf.writeInt(shuffleBlockInfo.getShuffleId());
+ byteBuf.writeLong(shuffleBlockInfo.getCrc());
+ byteBuf.writeLong(shuffleBlockInfo.getTaskAttemptId());
+ ByteBufUtils.copyByteBuf(shuffleBlockInfo.getData(), byteBuf);
Review Comment:
> You can see that Spark copy `Bytebuf` to `Channel`. It's ok. We copy `ByteBuf` to `ByteBuf` here. This should be optimized.
We will implement similar logic when supporting zero-copy.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] advancedxy commented on a diff in pull request #742: [#133] feat(netty): Add Encoder and Decoder.
Posted by "advancedxy (via GitHub)" <gi...@apache.org>.
advancedxy commented on code in PR #742:
URL: https://github.com/apache/incubator-uniffle/pull/742#discussion_r1142859818
##########
common/src/main/java/org/apache/uniffle/common/ShuffleBlockInfo.java:
##########
@@ -27,14 +31,21 @@ public class ShuffleBlockInfo {
private int shuffleId;
private long crc;
private long taskAttemptId;
- private byte[] data;
+ private ByteBuf data;
Review Comment:
This pr is kind of related to @zuston's #717. I think you could help review each other, and coordinator which pr should be merged first. It will help reduce info gap.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] jerqi commented on a diff in pull request #742: [#133] feat(netty): Add Encoder and Decoder.
Posted by "jerqi (via GitHub)" <gi...@apache.org>.
jerqi commented on code in PR #742:
URL: https://github.com/apache/incubator-uniffle/pull/742#discussion_r1142425988
##########
common/src/main/java/org/apache/uniffle/common/netty/protocol/Encoders.java:
##########
@@ -0,0 +1,69 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License 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 permissions and
+ * limitations under the License.
+ */
+
+package org.apache.uniffle.common.netty.protocol;
+
+import java.util.List;
+
+import io.netty.buffer.ByteBuf;
+
+import org.apache.uniffle.common.ShuffleBlockInfo;
+import org.apache.uniffle.common.ShuffleServerInfo;
+import org.apache.uniffle.common.util.ByteBufUtils;
+
+public class Encoders {
+ public static void encodeShuffleServerInfo(ShuffleServerInfo shuffleServerInfo, ByteBuf byteBuf) {
+ ByteBufUtils.writeLengthAndString(byteBuf, shuffleServerInfo.getId());
+ ByteBufUtils.writeLengthAndString(byteBuf, shuffleServerInfo.getHost());
+ byteBuf.writeInt(shuffleServerInfo.getGrpcPort());
+ byteBuf.writeInt(shuffleServerInfo.getNettyPort());
+ }
+
+ public static void encodeShuffleBlockInfo(ShuffleBlockInfo shuffleBlockInfo, ByteBuf byteBuf) {
+ byteBuf.writeInt(shuffleBlockInfo.getPartitionId());
+ byteBuf.writeLong(shuffleBlockInfo.getBlockId());
+ byteBuf.writeInt(shuffleBlockInfo.getLength());
+ byteBuf.writeInt(shuffleBlockInfo.getShuffleId());
+ byteBuf.writeLong(shuffleBlockInfo.getCrc());
+ byteBuf.writeLong(shuffleBlockInfo.getTaskAttemptId());
+ ByteBufUtils.copyByteBuf(shuffleBlockInfo.getData(), byteBuf);
Review Comment:
Do we need to release memory after copying here?
```
shuffleBlockInfo.getData().release().
```
##########
common/src/main/java/org/apache/uniffle/common/netty/protocol/SendShuffleDataRequest.java:
##########
@@ -0,0 +1,153 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License 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 permissions and
+ * limitations under the License.
+ */
+
+package org.apache.uniffle.common.netty.protocol;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+
+import com.google.common.collect.Lists;
+import com.google.common.collect.Maps;
+import io.netty.buffer.ByteBuf;
+
+import org.apache.uniffle.common.ShuffleBlockInfo;
+import org.apache.uniffle.common.util.ByteBufUtils;
+
+public class SendShuffleDataRequest extends Message {
+ public long requestId;
+ private String appId;
+ private int shuffleId;
+ private long requireId;
+ private Map<Integer, List<ShuffleBlockInfo>> partitionToBlocks;
+ private long timestamp;
+
+ public SendShuffleDataRequest(long requestId, String appId, int shuffleId, long requireId,
+ Map<Integer, List<ShuffleBlockInfo>> partitionToBlocks, long timestamp) {
+ this.requestId = requestId;
+ this.appId = appId;
+ this.shuffleId = shuffleId;
+ this.requireId = requireId;
+ this.partitionToBlocks = partitionToBlocks;
+ this.timestamp = timestamp;
+ }
+
+ @Override
+ public Type type() {
+ return Type.SEND_SHUFFLE_DATA_REQUEST;
+ }
+
+ @Override
+ public int encodedLength() {
+ int encodeLength = Long.BYTES + ByteBufUtils.encodedLength(appId) + Integer.BYTES + Long.BYTES + Integer.BYTES;
+ for (Map.Entry<Integer, List<ShuffleBlockInfo>> entry : partitionToBlocks.entrySet()) {
+ encodeLength += 2 * Integer.BYTES;
+ for (ShuffleBlockInfo sbi : entry.getValue()) {
+ encodeLength += Encoders.encodeLengthOfShuffleBlockInfo(sbi);
+ }
+ }
+ return encodeLength + Long.BYTES;
+ }
+
+ @Override
+ public void encode(ByteBuf buf) {
+ buf.writeLong(requestId);
+ ByteBufUtils.writeLengthAndString(buf, appId);
+ buf.writeInt(shuffleId);
+ buf.writeLong(requireId);
+ encodePartitionData(buf);
+ buf.writeLong(timestamp);
+ }
+
+ private static Map<Integer, List<ShuffleBlockInfo>> decodePartitionData(ByteBuf byteBuf) {
+ Map<Integer, List<ShuffleBlockInfo>> partitionToBlocks = Maps.newHashMap();
+ int lengthOfPartitionData = byteBuf.readInt();
+ for (int i = 0; i < lengthOfPartitionData; i++) {
+ int partitionId = byteBuf.readInt();
+ int lengthOfShuffleBlocks = byteBuf.readInt();
+ List<ShuffleBlockInfo> shuffleBlockInfoList = Lists.newArrayList();
+ for (int j = 0; j < lengthOfShuffleBlocks; j++) {
+ shuffleBlockInfoList.add(Decoders.decodeShuffleBlockInfo(byteBuf));
+ }
+ partitionToBlocks.put(partitionId, shuffleBlockInfoList);
+ }
+ return partitionToBlocks;
+ }
+
+ public static SendShuffleDataRequest decode(ByteBuf byteBuf) {
+ long requestId = byteBuf.readLong();
+ String appId = ByteBufUtils.readLengthAndString(byteBuf);
+ int shuffleId = byteBuf.readInt();
+ long requireId = byteBuf.readLong();
+ Map<Integer, List<ShuffleBlockInfo>> partitionToBlocks = decodePartitionData(byteBuf);
+ long timestamp = byteBuf.readLong();
+ return new SendShuffleDataRequest(requestId, appId, shuffleId, requireId, partitionToBlocks, timestamp);
+ }
+
+ private void encodePartitionData(ByteBuf buf) {
+ buf.writeInt(partitionToBlocks.size());
+ for (Map.Entry<Integer, List<ShuffleBlockInfo>> entry : partitionToBlocks.entrySet()) {
+ buf.writeInt(entry.getKey());
+ buf.writeInt(entry.getValue().size());
+ for (ShuffleBlockInfo sbi : entry.getValue()) {
+ Encoders.encodeShuffleBlockInfo(sbi, buf);
+ }
+ }
+ }
+
+ public long getRequestId() {
+ return requestId;
+ }
+
+ public String getAppId() {
+ return appId;
+ }
+
+ public int getShuffleId() {
+ return shuffleId;
+ }
+
+ public long getRequireId() {
+ return requireId;
+ }
+
+ public Map<Integer, List<ShuffleBlockInfo>> getPartitionToBlocks() {
+ return partitionToBlocks;
+ }
+
+ @Override
+ public boolean equals(Object o) {
Review Comment:
Why do we need to override the method ``equal?
##########
common/src/main/java/org/apache/uniffle/common/netty/MessageEncoder.java:
##########
@@ -0,0 +1,49 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License 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 permissions and
+ * limitations under the License.
+ */
+
+package org.apache.uniffle.common.netty;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelOutboundHandlerAdapter;
+import io.netty.channel.ChannelPromise;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.uniffle.common.netty.protocol.Message;
+
+public class MessageEncoder extends ChannelOutboundHandlerAdapter {
+
+ private static final Logger LOG = LoggerFactory.getLogger(MessageEncoder.class);
+
+ @Override
+ public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) {
+ // todo: support zero copy
+ Message message = (Message) msg;
+ int encodeLength = message.encodedLength();
+ ByteBuf byteBuf = ctx.alloc().buffer(FrameDecoder.HEADER_SIZE + encodeLength);
+ try {
+ byteBuf.writeInt(encodeLength);
+ byteBuf.writeByte(message.type().id());
+ message.encode(byteBuf);
+ } catch (Exception e) {
+ LOG.error("Unexpected exception during process encode!", e);
+ byteBuf.release();
+ }
+ ctx.writeAndFlush(byteBuf);
Review Comment:
When do we release byteBuf?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] leixm commented on a diff in pull request #742: [#133] feat(netty): Add Encoder and Decoder.
Posted by "leixm (via GitHub)" <gi...@apache.org>.
leixm commented on code in PR #742:
URL: https://github.com/apache/incubator-uniffle/pull/742#discussion_r1146188950
##########
common/src/main/java/org/apache/uniffle/common/netty/protocol/Encoders.java:
##########
@@ -0,0 +1,71 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License 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 permissions and
+ * limitations under the License.
+ */
+
+package org.apache.uniffle.common.netty.protocol;
+
+import java.util.List;
+
+import io.netty.buffer.ByteBuf;
+
+import org.apache.uniffle.common.ShuffleBlockInfo;
+import org.apache.uniffle.common.ShuffleServerInfo;
+import org.apache.uniffle.common.util.ByteBufUtils;
+
+public class Encoders {
Review Comment:
I will optimize in the followup PRs.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] leixm commented on a diff in pull request #742: [#133] feat(netty): Add Encoder and Decoder.
Posted by "leixm (via GitHub)" <gi...@apache.org>.
leixm commented on code in PR #742:
URL: https://github.com/apache/incubator-uniffle/pull/742#discussion_r1146177566
##########
common/src/main/java/org/apache/uniffle/common/netty/protocol/Message.java:
##########
@@ -53,6 +73,46 @@ public static Type decode(ByteBuf buf) {
switch (id) {
case 0:
return RPC_RESPONSE;
+ case 1:
+ return SHUFFLE_REGISTER_REQUEST;
Review Comment:
We will implement all protocols of ShuffleServer.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] smallzhongfeng merged pull request #742: [#133] feat(netty): Add Encoder and Decoder.
Posted by "smallzhongfeng (via GitHub)" <gi...@apache.org>.
smallzhongfeng merged PR #742:
URL: https://github.com/apache/incubator-uniffle/pull/742
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] smallzhongfeng commented on a diff in pull request #742: [#133] feat(netty): Add Encoder and Decoder.
Posted by "smallzhongfeng (via GitHub)" <gi...@apache.org>.
smallzhongfeng commented on code in PR #742:
URL: https://github.com/apache/incubator-uniffle/pull/742#discussion_r1145784004
##########
common/src/main/java/org/apache/uniffle/common/netty/protocol/Encoders.java:
##########
@@ -0,0 +1,71 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License 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 permissions and
+ * limitations under the License.
+ */
+
+package org.apache.uniffle.common.netty.protocol;
+
+import java.util.List;
+
+import io.netty.buffer.ByteBuf;
+
+import org.apache.uniffle.common.ShuffleBlockInfo;
+import org.apache.uniffle.common.ShuffleServerInfo;
+import org.apache.uniffle.common.util.ByteBufUtils;
+
+public class Encoders {
+ public static void encodeShuffleServerInfo(ShuffleServerInfo shuffleServerInfo, ByteBuf byteBuf) {
+ ByteBufUtils.writeLengthAndString(byteBuf, shuffleServerInfo.getId());
+ ByteBufUtils.writeLengthAndString(byteBuf, shuffleServerInfo.getHost());
+ byteBuf.writeInt(shuffleServerInfo.getGrpcPort());
+ byteBuf.writeInt(shuffleServerInfo.getNettyPort());
+ }
+
+ public static void encodeShuffleBlockInfo(ShuffleBlockInfo shuffleBlockInfo, ByteBuf byteBuf) {
+ byteBuf.writeInt(shuffleBlockInfo.getPartitionId());
+ byteBuf.writeLong(shuffleBlockInfo.getBlockId());
+ byteBuf.writeInt(shuffleBlockInfo.getLength());
+ byteBuf.writeInt(shuffleBlockInfo.getShuffleId());
+ byteBuf.writeLong(shuffleBlockInfo.getCrc());
+ byteBuf.writeLong(shuffleBlockInfo.getTaskAttemptId());
+ // todo: avoid copy
+ ByteBufUtils.copyByteBuf(shuffleBlockInfo.getData(), byteBuf);
+ shuffleBlockInfo.getData().release();
+ List<ShuffleServerInfo> shuffleServerInfoList = shuffleBlockInfo.getShuffleServerInfos();
+ byteBuf.writeInt(shuffleServerInfoList.size());
+ for (ShuffleServerInfo shuffleServerInfo : shuffleServerInfoList) {
+ Encoders.encodeShuffleServerInfo(shuffleServerInfo, byteBuf);
+ }
+ byteBuf.writeInt(shuffleBlockInfo.getUncompressLength());
+ byteBuf.writeLong(shuffleBlockInfo.getFreeMemory());
+ }
+
+
+ public static int encodeLengthOfShuffleServerInfo(ShuffleServerInfo shuffleServerInfo) {
+ return ByteBufUtils.encodedLength(shuffleServerInfo.getId())
+ + ByteBufUtils.encodedLength(shuffleServerInfo.getHost())
+ + 2 * Integer.BYTES;
+ }
+
+ public static int encodeLengthOfShuffleBlockInfo(ShuffleBlockInfo shuffleBlockInfo) {
+ int encodeLength = 4 * Long.BYTES + 4 * Integer.BYTES + ByteBufUtils.encodedLength(shuffleBlockInfo.getData())
+ + Integer.BYTES;
Review Comment:
nit: control Indentation.
##########
common/src/test/java/org/apache/uniffle/common/netty/protocol/NettyProtocolTestUtils.java:
##########
@@ -0,0 +1,86 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License 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 permissions and
+ * limitations under the License.
+ */
+
+package org.apache.uniffle.common.netty.protocol;
+
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+
+import org.apache.uniffle.common.ShuffleBlockInfo;
+
+public class NettyProtocolTestUtils {
+
+ private static boolean compareShuffleBlockInfo(ShuffleBlockInfo blockInfo1, ShuffleBlockInfo blockInfo2) {
+ return blockInfo1.getPartitionId() == blockInfo2.getPartitionId()
+ && blockInfo1.getBlockId() == blockInfo2.getBlockId()
+ && blockInfo1.getLength() == blockInfo2.getLength()
+ && blockInfo1.getShuffleId() == blockInfo2.getShuffleId()
+ && blockInfo1.getCrc() == blockInfo2.getCrc()
+ && blockInfo1.getTaskAttemptId() == blockInfo2.getTaskAttemptId()
+ && blockInfo1.getUncompressLength() == blockInfo2.getUncompressLength()
+ && blockInfo1.getFreeMemory() == blockInfo2.getFreeMemory()
+ && blockInfo1.getData().equals(blockInfo2.getData())
+ && blockInfo1.getShuffleServerInfos().equals(blockInfo2.getShuffleServerInfos());
+ }
+
+ private static boolean compareBlockList(List<ShuffleBlockInfo> list1, List<ShuffleBlockInfo> list2) {
+ if (list1 == null || list2 == null || list1.size() != list2.size()) {
+ return false;
+ }
+ for (int i = 0; i < list1.size(); i++) {
+ if (!compareShuffleBlockInfo(list1.get(i), list2.get(i))) {
+ return false;
+ }
+ }
+ return true;
+ }
+
+ private static boolean comparePartitionToBlockList(Map<Integer, List<ShuffleBlockInfo>> m1,
+ Map<Integer, List<ShuffleBlockInfo>> m2) {
+ if (m1 == null || m2 == null || m1.size() != m2.size()) {
+ return false;
+ }
+ Iterator<Map.Entry<Integer, List<ShuffleBlockInfo>>> iter1 = m1.entrySet().iterator();
+ while (iter1.hasNext()) {
+ Map.Entry<Integer, List<ShuffleBlockInfo>> entry1 = iter1.next();
+ if (!compareBlockList(entry1.getValue(), m2.get(entry1.getKey()))) {
+ return false;
+ }
+ }
+ return true;
+ }
+
+ public static boolean compareSendShuffleDataRequest(SendShuffleDataRequest req1,
+ SendShuffleDataRequest req2) {
+ if (req1 == req2) {
+ return true;
+ }
+ if (req1 == null || req2 == null) {
+ return false;
+ }
+ boolean isEqual = req1.requestId == req2.requestId
+ && req1.getShuffleId() == req2.getShuffleId()
Review Comment:
nit: control Indentation.
##########
common/src/test/java/org/apache/uniffle/common/netty/protocol/NettyProtocolTestUtils.java:
##########
@@ -0,0 +1,86 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License 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 permissions and
+ * limitations under the License.
+ */
+
+package org.apache.uniffle.common.netty.protocol;
+
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+
+import org.apache.uniffle.common.ShuffleBlockInfo;
+
+public class NettyProtocolTestUtils {
+
+ private static boolean compareShuffleBlockInfo(ShuffleBlockInfo blockInfo1, ShuffleBlockInfo blockInfo2) {
+ return blockInfo1.getPartitionId() == blockInfo2.getPartitionId()
+ && blockInfo1.getBlockId() == blockInfo2.getBlockId()
Review Comment:
nit: control Indentation.
##########
common/src/main/java/org/apache/uniffle/common/netty/protocol/Encoders.java:
##########
@@ -0,0 +1,71 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License 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 permissions and
+ * limitations under the License.
+ */
+
+package org.apache.uniffle.common.netty.protocol;
+
+import java.util.List;
+
+import io.netty.buffer.ByteBuf;
+
+import org.apache.uniffle.common.ShuffleBlockInfo;
+import org.apache.uniffle.common.ShuffleServerInfo;
+import org.apache.uniffle.common.util.ByteBufUtils;
+
+public class Encoders {
+ public static void encodeShuffleServerInfo(ShuffleServerInfo shuffleServerInfo, ByteBuf byteBuf) {
+ ByteBufUtils.writeLengthAndString(byteBuf, shuffleServerInfo.getId());
+ ByteBufUtils.writeLengthAndString(byteBuf, shuffleServerInfo.getHost());
+ byteBuf.writeInt(shuffleServerInfo.getGrpcPort());
+ byteBuf.writeInt(shuffleServerInfo.getNettyPort());
+ }
+
+ public static void encodeShuffleBlockInfo(ShuffleBlockInfo shuffleBlockInfo, ByteBuf byteBuf) {
+ byteBuf.writeInt(shuffleBlockInfo.getPartitionId());
+ byteBuf.writeLong(shuffleBlockInfo.getBlockId());
+ byteBuf.writeInt(shuffleBlockInfo.getLength());
+ byteBuf.writeInt(shuffleBlockInfo.getShuffleId());
+ byteBuf.writeLong(shuffleBlockInfo.getCrc());
+ byteBuf.writeLong(shuffleBlockInfo.getTaskAttemptId());
+ // todo: avoid copy
+ ByteBufUtils.copyByteBuf(shuffleBlockInfo.getData(), byteBuf);
+ shuffleBlockInfo.getData().release();
+ List<ShuffleServerInfo> shuffleServerInfoList = shuffleBlockInfo.getShuffleServerInfos();
+ byteBuf.writeInt(shuffleServerInfoList.size());
+ for (ShuffleServerInfo shuffleServerInfo : shuffleServerInfoList) {
+ Encoders.encodeShuffleServerInfo(shuffleServerInfo, byteBuf);
+ }
+ byteBuf.writeInt(shuffleBlockInfo.getUncompressLength());
+ byteBuf.writeLong(shuffleBlockInfo.getFreeMemory());
+ }
+
+
+ public static int encodeLengthOfShuffleServerInfo(ShuffleServerInfo shuffleServerInfo) {
+ return ByteBufUtils.encodedLength(shuffleServerInfo.getId())
+ + ByteBufUtils.encodedLength(shuffleServerInfo.getHost())
Review Comment:
nit: control Indentation.
##########
common/src/test/java/org/apache/uniffle/common/netty/EncoderAndDecoderTest.java:
##########
@@ -0,0 +1,174 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License 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 permissions and
+ * limitations under the License.
+ */
+
+package org.apache.uniffle.common.netty;
+
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
+
+import com.google.common.collect.Maps;
+import io.netty.bootstrap.Bootstrap;
+import io.netty.bootstrap.ServerBootstrap;
+import io.netty.buffer.PooledByteBufAllocator;
+import io.netty.buffer.Unpooled;
+import io.netty.channel.Channel;
+import io.netty.channel.ChannelFuture;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelInboundHandlerAdapter;
+import io.netty.channel.ChannelInitializer;
+import io.netty.channel.ChannelOption;
+import io.netty.channel.EventLoopGroup;
+import io.netty.channel.socket.SocketChannel;
+import io.netty.channel.socket.nio.NioServerSocketChannel;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import org.apache.uniffle.common.ShuffleBlockInfo;
+import org.apache.uniffle.common.ShuffleServerInfo;
+import org.apache.uniffle.common.exception.RssException;
+import org.apache.uniffle.common.netty.protocol.NettyProtocolTestUtils;
+import org.apache.uniffle.common.netty.protocol.RpcResponse;
+import org.apache.uniffle.common.netty.protocol.SendShuffleDataRequest;
+import org.apache.uniffle.common.rpc.StatusCode;
+import org.apache.uniffle.common.util.NettyUtils;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+public class EncoderAndDecoderTest {
+ private EventLoopGroup bossGroup;
+ private EventLoopGroup workerGroup;
+ private ChannelFuture channelFuture;
+ private static final SendShuffleDataRequest DATA_REQUEST = generateShuffleDataRequest();
+ private static final String EXPECTED_MESSAGE = "test_message";
+ private static final long REQUEST_ID = 1;
+ private static final StatusCode STATUS_CODE = StatusCode.SUCCESS;
+ private static final int PORT = 12345;
+
+ static class MockResponseHandler extends ChannelInboundHandlerAdapter {
+ @Override
+ public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
+ if (msg instanceof RpcResponse) {
+ RpcResponse rpcResponse = (RpcResponse) msg;
+ assertEquals(REQUEST_ID, rpcResponse.getRequestId());
+ assertEquals(STATUS_CODE, rpcResponse.getStatusCode());
+ assertEquals(EXPECTED_MESSAGE, rpcResponse.getRetMessage());
+ } else if (msg instanceof SendShuffleDataRequest) {
+ SendShuffleDataRequest sendShuffleDataRequest = (SendShuffleDataRequest) msg;
+ assertTrue(NettyProtocolTestUtils.compareSendShuffleDataRequest(sendShuffleDataRequest, DATA_REQUEST));
+ sendShuffleDataRequest.getPartitionToBlocks().values().stream().flatMap(Collection::stream)
+ .forEach(shuffleBlockInfo -> shuffleBlockInfo.getData().release());
+ sendShuffleDataRequest.getPartitionToBlocks().values().stream().flatMap(Collection::stream)
+ .forEach(shuffleBlockInfo -> assertEquals(0,shuffleBlockInfo.getData().refCnt()));
+ RpcResponse rpcResponse = new RpcResponse(REQUEST_ID, STATUS_CODE, EXPECTED_MESSAGE);
+ ctx.writeAndFlush(rpcResponse);
+ } else {
+ throw new RssException("receive unexpected message!");
Review Comment:
I think we can add a class of netty transport exceptions in the followup pr, which may better distinguish these exceptions.
##########
common/src/main/java/org/apache/uniffle/common/netty/TransportFrameDecoder.java:
##########
@@ -0,0 +1,191 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License 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 permissions and
+ * limitations under the License.
+ */
+
+package org.apache.uniffle.common.netty;
+
+import java.util.LinkedList;
+
+import com.google.common.base.Preconditions;
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.CompositeByteBuf;
+import io.netty.buffer.Unpooled;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelInboundHandlerAdapter;
+
+import org.apache.uniffle.common.netty.protocol.Message;
+
+/**
+ * A customized frame decoder that allows intercepting raw data.
+ *
+ * <p>This behaves like Netty's frame decoder (with hard coded parameters that match this library's
+ * needs), except it allows an interceptor to be installed to read data directly before it's framed.
+ *
+ * <p>Unlike Netty's frame decoder, each frame is dispatched to child handlers as soon as it's
+ * decoded, instead of building as many frames as the current buffer allows and dispatching all of
+ * them. This allows a child handler to install an interceptor if needed.
+ *
+ * <p>If an interceptor is installed, framing stops, and data is instead fed directly to the
+ * interceptor. When the interceptor indicates that it doesn't need to read any more data, framing
+ * resumes. Interceptors should not hold references to the data buffers provided to their handle()
+ * method.
+ */
+public class TransportFrameDecoder extends ChannelInboundHandlerAdapter implements FrameDecoder {
+ private int msgSize = -1;
+ private Message.Type curType = Message.Type.UNKNOWN_TYPE;
+ private ByteBuf headerBuf = Unpooled.buffer(HEADER_SIZE, HEADER_SIZE);
+ private static final int MAX_FRAME_SIZE = Integer.MAX_VALUE;
+ private static final int UNKNOWN_FRAME_SIZE = -1;
+
+ private final LinkedList<ByteBuf> buffers = new LinkedList<>();
+
+ private long totalSize = 0;
+ private long nextFrameSize = UNKNOWN_FRAME_SIZE;
+
+ @Override
+ public void channelRead(ChannelHandlerContext ctx, Object data) {
+ ByteBuf in = (ByteBuf) data;
+ buffers.add(in);
+ totalSize += in.readableBytes();
+
+ while (!buffers.isEmpty()) {
+ ByteBuf frame = decodeNext();
+ if (frame == null) {
+ break;
+ }
+ Message msg = Message.decode(curType, frame);
+ frame.release();
+ ctx.fireChannelRead(msg);
+ clear();
+ }
+ }
+
+ private void clear() {
+ curType = Message.Type.UNKNOWN_TYPE;
+ msgSize = -1;
+ headerBuf.clear();
+ }
+
+ private long decodeFrameSize() {
+ if (nextFrameSize != UNKNOWN_FRAME_SIZE || totalSize < HEADER_SIZE) {
+ return nextFrameSize;
+ }
+
+ // We know there's enough data. If the first buffer contains all the data, great. Otherwise,
+ // hold the bytes for the frame length in a composite buffer until we have enough data to read
+ // the frame size. Normally, it should be rare to need more than one buffer to read the frame
+ // size.
+ ByteBuf first = buffers.getFirst();
+ if (first.readableBytes() >= HEADER_SIZE) {
+ msgSize = first.readInt();
+ curType = Message.Type.decode(first);
+ nextFrameSize = msgSize;
+ totalSize -= HEADER_SIZE;
+ if (!first.isReadable()) {
+ buffers.removeFirst().release();
+ }
+ return nextFrameSize;
+ }
+
+ while (headerBuf.readableBytes() < HEADER_SIZE) {
+ ByteBuf next = buffers.getFirst();
+ int toRead = Math.min(next.readableBytes(), HEADER_SIZE - headerBuf.readableBytes());
+ headerBuf.writeBytes(next, toRead);
+ if (!next.isReadable()) {
+ buffers.removeFirst().release();
+ }
+ }
+
+ msgSize = headerBuf.readInt();
+ curType = Message.Type.decode(headerBuf);
+ nextFrameSize = msgSize;
+ totalSize -= HEADER_SIZE;
+ return nextFrameSize;
+ }
+
+ private ByteBuf decodeNext() {
+ long frameSize = decodeFrameSize();
+ if (frameSize == UNKNOWN_FRAME_SIZE || totalSize < frameSize) {
+ return null;
+ }
+
+ // Reset size for next frame.
+ nextFrameSize = UNKNOWN_FRAME_SIZE;
+
+ Preconditions.checkArgument(frameSize < MAX_FRAME_SIZE, "Too large frame: %s", frameSize);
Review Comment:
It seems that we can `frameSize <= MAX_FRAME_SIZE`?
##########
common/src/main/java/org/apache/uniffle/common/netty/protocol/Encoders.java:
##########
@@ -0,0 +1,71 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License 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 permissions and
+ * limitations under the License.
+ */
+
+package org.apache.uniffle.common.netty.protocol;
+
+import java.util.List;
+
+import io.netty.buffer.ByteBuf;
+
+import org.apache.uniffle.common.ShuffleBlockInfo;
+import org.apache.uniffle.common.ShuffleServerInfo;
+import org.apache.uniffle.common.util.ByteBufUtils;
+
+public class Encoders {
Review Comment:
Yes, but I think it can be optimized in the followup PR.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] leixm commented on a diff in pull request #742: [#133] feat(netty): Add Encoder and Decoder.
Posted by "leixm (via GitHub)" <gi...@apache.org>.
leixm commented on code in PR #742:
URL: https://github.com/apache/incubator-uniffle/pull/742#discussion_r1141931113
##########
common/src/main/java/org/apache/uniffle/common/netty/protocol/SendShuffleDataRequest.java:
##########
@@ -0,0 +1,147 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License 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 permissions and
+ * limitations under the License.
+ */
+
+package org.apache.uniffle.common.netty.protocol;
Review Comment:
I will rewrite all protocol in subsequent PRs.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] zuston commented on a diff in pull request #742: [#133] feat(netty): Add Encoder and Decoder.
Posted by "zuston (via GitHub)" <gi...@apache.org>.
zuston commented on code in PR #742:
URL: https://github.com/apache/incubator-uniffle/pull/742#discussion_r1147047766
##########
common/src/main/java/org/apache/uniffle/common/netty/protocol/Encoders.java:
##########
@@ -0,0 +1,71 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License 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 permissions and
+ * limitations under the License.
+ */
+
+package org.apache.uniffle.common.netty.protocol;
+
+import java.util.List;
+
+import io.netty.buffer.ByteBuf;
+
+import org.apache.uniffle.common.ShuffleBlockInfo;
+import org.apache.uniffle.common.ShuffleServerInfo;
+import org.apache.uniffle.common.util.ByteBufUtils;
+
+public class Encoders {
Review Comment:
Nice! Let's push forward quickly
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] jerqi commented on a diff in pull request #742: [#133] feat(netty): Add Encoder and Decoder.
Posted by "jerqi (via GitHub)" <gi...@apache.org>.
jerqi commented on code in PR #742:
URL: https://github.com/apache/incubator-uniffle/pull/742#discussion_r1144710543
##########
common/src/test/java/org/apache/uniffle/common/netty/protocol/NettyProtocolTestUtils.java:
##########
@@ -17,128 +17,71 @@
package org.apache.uniffle.common.netty.protocol;
+import java.util.Iterator;
import java.util.List;
import java.util.Map;
-import java.util.Objects;
-
-import com.google.common.collect.Lists;
-import com.google.common.collect.Maps;
-import io.netty.buffer.ByteBuf;
import org.apache.uniffle.common.ShuffleBlockInfo;
-import org.apache.uniffle.common.ShuffleServerInfo;
public class NettyProtocolTestUtils {
- static class SendShuffleDataRequestTest extends SendShuffleDataRequest {
- SendShuffleDataRequestTest(long requestId, String appId, int shuffleId, long requireId,
- Map<Integer, List<ShuffleBlockInfo>> partitionToBlocks, long timestamp) {
- super(requestId, appId, shuffleId, requireId, partitionToBlocks, timestamp);
- }
-
- @Override
- public boolean equals(Object o) {
- if (this == o) {
- return true;
- }
- if (o == null || getClass() != o.getClass()) {
- return false;
- }
- SendShuffleDataRequestTest that = (SendShuffleDataRequestTest) o;
- Map<Integer, List<ShuffleBlockInfoTest>> map1 = Maps.newHashMap();
- Map<Integer, List<ShuffleBlockInfoTest>> map2 = Maps.newHashMap();
- for (Map.Entry<Integer, List<ShuffleBlockInfo>> entry : this.getPartitionToBlocks().entrySet()) {
- map1.put(entry.getKey(), toShuffleBlockInfoTestList(entry.getValue()));
- }
- for (Map.Entry<Integer, List<ShuffleBlockInfo>> entry : that.getPartitionToBlocks().entrySet()) {
- map2.put(entry.getKey(), toShuffleBlockInfoTestList(entry.getValue()));
- }
- return this.requestId == that.requestId
- && this.getShuffleId() == that.getShuffleId()
- && this.getRequireId() == that.getRequireId()
- && this.getTimestamp() == that.getTimestamp()
- && this.getAppId().equals(that.getAppId())
- && map1.equals(map2);
- }
-
- @Override
- public int hashCode() {
- return Objects.hash(requestId, this.getAppId(), this.getShuffleId(), this.getRequireId(),
- this.getPartitionToBlocks(), this.getTimestamp());
- }
+ private static boolean compareShuffleBlockInfo(ShuffleBlockInfo blockInfo1, ShuffleBlockInfo blockInfo2) {
+ return blockInfo1.getPartitionId() == blockInfo2.getPartitionId()
+ && blockInfo1.getBlockId() == blockInfo2.getBlockId()
+ && blockInfo1.getLength() == blockInfo2.getLength()
+ && blockInfo1.getShuffleId() == blockInfo2.getShuffleId()
+ && blockInfo1.getCrc() == blockInfo2.getCrc()
+ && blockInfo1.getTaskAttemptId() == blockInfo2.getTaskAttemptId()
+ && blockInfo1.getUncompressLength() == blockInfo2.getUncompressLength()
+ && blockInfo1.getFreeMemory() == blockInfo2.getFreeMemory()
+ && blockInfo1.getData().equals(blockInfo2.getData())
+ && blockInfo1.getShuffleServerInfos().equals(blockInfo2.getShuffleServerInfos());
}
- static class ShuffleBlockInfoTest extends ShuffleBlockInfo {
- ShuffleBlockInfoTest(int shuffleId, int partitionId, long blockId, int length, long crc,
- ByteBuf data, List<ShuffleServerInfo> shuffleServerInfos, int uncompressLength,
- long freeMemory, long taskAttemptId) {
- super(shuffleId, partitionId, blockId, length, crc, data, shuffleServerInfos, uncompressLength,
- freeMemory, taskAttemptId);
- }
- @Override
- public boolean equals(Object o) {
- if (this == o) {
- return true;
- }
- if (o == null || getClass() != o.getClass()) {
+ private static boolean compareBlockList(List<ShuffleBlockInfo> list1, List<ShuffleBlockInfo> list2) {
+ if (list1 == null || list2 == null || list1.size() != list2.size()) {
+ return false;
+ }
+ for (int i = 0; i < list1.size(); i++) {
+ if (!compareShuffleBlockInfo(list1.get(i), list2.get(i))) {
return false;
}
- ShuffleBlockInfoTest that = (ShuffleBlockInfoTest) o;
- return this.getPartitionId() == that.getPartitionId()
- && this.getBlockId() == that.getBlockId()
- && this.getLength() == that.getLength()
- && this.getShuffleId() == that.getShuffleId()
- && this.getCrc() == that.getCrc()
- && this.getTaskAttemptId() == that.getTaskAttemptId()
- && this.getUncompressLength() == that.getUncompressLength()
- && this.getFreeMemory() == that.getFreeMemory()
- && this.getData().equals(that.getData())
- && this.getShuffleServerInfos().equals(that.getShuffleServerInfos());
- }
-
- @Override
- public int hashCode() {
- return Objects.hash(this.getPartitionId(), this.getBlockId(), this.getLength(), this.getShuffleId(),
- this.getCrc(), this.getTaskAttemptId(), this.getData(), this.getShuffleServerInfos(),
- this.getUncompressLength(), this.getFreeMemory());
}
+ return true;
}
- private static SendShuffleDataRequestTest toSendShuffleDataRequestTest(
- SendShuffleDataRequest sendShuffleDataRequest) {
- return new SendShuffleDataRequestTest(sendShuffleDataRequest.requestId,
- sendShuffleDataRequest.getAppId(),
- sendShuffleDataRequest.getShuffleId(),
- sendShuffleDataRequest.getRequireId(),
- sendShuffleDataRequest.getPartitionToBlocks(),
- sendShuffleDataRequest.getTimestamp());
- }
+ private static boolean comparePartitionToBlockList(Map<Integer, List<ShuffleBlockInfo>> m1,
+ Map<Integer, List<ShuffleBlockInfo>> m2) {
+ Iterator<Map.Entry<Integer, List<ShuffleBlockInfo>>> iter1 = m1.entrySet().iterator();
+ while (iter1.hasNext()) {
+ Map.Entry<Integer, List<ShuffleBlockInfo>> entry1 = iter1.next();
+ List<ShuffleBlockInfo> m1value = entry1.getValue() == null ? null : entry1.getValue();
Review Comment:
Is it right?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] leixm commented on a diff in pull request #742: [#133] feat(netty): Add Encoder and Decoder.
Posted by "leixm (via GitHub)" <gi...@apache.org>.
leixm commented on code in PR #742:
URL: https://github.com/apache/incubator-uniffle/pull/742#discussion_r1141941955
##########
internal-client/src/main/java/org/apache/uniffle/client/impl/grpc/ShuffleServerGrpcClient.java:
##########
@@ -319,7 +319,7 @@ public RssSendShuffleDataResponse sendShuffleData(RssSendShuffleDataRequest requ
.setLength(sbi.getLength())
.setTaskAttemptId(sbi.getTaskAttemptId())
.setUncompressLength(sbi.getUncompressLength())
- .setData(ByteString.copyFrom(sbi.getData()))
+ .setData(ByteString.copyFrom(sbi.getData().nioBuffer()))
Review Comment:
Looks like we can use NioByteString, I'll test it out.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] jerqi commented on a diff in pull request #742: [#133] feat(netty): Add Encoder and Decoder.
Posted by "jerqi (via GitHub)" <gi...@apache.org>.
jerqi commented on code in PR #742:
URL: https://github.com/apache/incubator-uniffle/pull/742#discussion_r1144194839
##########
common/src/test/java/org/apache/uniffle/common/netty/protocol/NettyProtocolTestUtils.java:
##########
@@ -0,0 +1,119 @@
+package org.apache.uniffle.common.netty.protocol;
+
+import com.google.common.collect.Lists;
+import com.google.common.collect.Maps;
+import io.netty.buffer.ByteBuf;
+import org.apache.uniffle.common.ShuffleBlockInfo;
+import org.apache.uniffle.common.ShuffleServerInfo;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+
+public class NettyProtocolTestUtils {
+
+ static class SendShuffleDataRequestTest extends SendShuffleDataRequest {
+ public SendShuffleDataRequestTest(long requestId, String appId, int shuffleId, long requireId, Map<Integer, List<ShuffleBlockInfo>> partitionToBlocks, long timestamp) {
+ super(requestId, appId, shuffleId, requireId, partitionToBlocks, timestamp);
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ SendShuffleDataRequestTest that = (SendShuffleDataRequestTest) o;
+ Map<Integer, List<ShuffleBlockInfoTest>> map1 = Maps.newHashMap();
+ Map<Integer, List<ShuffleBlockInfoTest>> map2 = Maps.newHashMap();
+ for (Map.Entry<Integer, List<ShuffleBlockInfo>> entry : this.getPartitionToBlocks().entrySet()) {
+ map1.put(entry.getKey(), toShuffleBlockInfoTestList(entry.getValue()));
+ }
+ for (Map.Entry<Integer, List<ShuffleBlockInfo>> entry : that.getPartitionToBlocks().entrySet()) {
+ map2.put(entry.getKey(), toShuffleBlockInfoTestList(entry.getValue()));
+ }
+ return this.requestId == that.requestId
+ && this.getShuffleId() == that.getShuffleId()
+ && this.getRequireId() == that.getRequireId()
+ && this.getTimestamp() == that.getTimestamp()
+ && this.getAppId().equals(that.getAppId())
+ && map1.equals(map2);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(requestId, this.getAppId(), this.getShuffleId(), this.getRequireId(),
+ this.getPartitionToBlocks(), this.getTimestamp());
+ }
+ }
+
+ static class ShuffleBlockInfoTest extends ShuffleBlockInfo {
+ public ShuffleBlockInfoTest(int shuffleId, int partitionId, long blockId, int length, long crc, ByteBuf data, List<ShuffleServerInfo> shuffleServerInfos, int uncompressLength, long freeMemory, long taskAttemptId) {
+ super(shuffleId, partitionId, blockId, length, crc, data, shuffleServerInfos, uncompressLength, freeMemory, taskAttemptId);
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ ShuffleBlockInfoTest that = (ShuffleBlockInfoTest) o;
+ return this.getPartitionId() == that.getPartitionId()
+ && this.getBlockId() == that.getBlockId()
+ && this.getLength() == that.getLength()
+ && this.getShuffleId() == that.getShuffleId()
+ && this.getCrc() == that.getCrc()
+ && this.getTaskAttemptId() == that.getTaskAttemptId()
+ && this.getUncompressLength() == that.getUncompressLength()
+ && this.getFreeMemory() == that.getFreeMemory()
+ && this.getData().equals(that.getData())
+ && this.getShuffleServerInfos().equals(that.getShuffleServerInfos());
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(this.getPartitionId(), this.getBlockId(), this.getLength(), this.getShuffleId(),
+ this.getCrc(), this.getTaskAttemptId(), this.getData(), this.getShuffleServerInfos(),
+ this.getUncompressLength(), this.getFreeMemory());
+ }
+ }
+
+ private static SendShuffleDataRequestTest toSendShuffleDataRequestTest(SendShuffleDataRequest sendShuffleDataRequest) {
+ return new SendShuffleDataRequestTest(sendShuffleDataRequest.requestId,
+ sendShuffleDataRequest.getAppId(),
+ sendShuffleDataRequest.getShuffleId(),
+ sendShuffleDataRequest.getRequireId(),
+ sendShuffleDataRequest.getPartitionToBlocks(),
+ sendShuffleDataRequest.getTimestamp());
+ }
+
+ private static List<ShuffleBlockInfoTest> toShuffleBlockInfoTestList(List<ShuffleBlockInfo> shuffleBlockInfoList) {
+ List<ShuffleBlockInfoTest> res = Lists.newArrayList();
+ for (ShuffleBlockInfo shuffleBlockInfo : shuffleBlockInfoList) {
+ res.add(toShuffleBlockInfoTest(shuffleBlockInfo));
+ }
+ return res;
+ }
+
+ private static ShuffleBlockInfoTest toShuffleBlockInfoTest(ShuffleBlockInfo shuffleBlockInfo) {
+ return new ShuffleBlockInfoTest(shuffleBlockInfo.getShuffleId(),
+ shuffleBlockInfo.getPartitionId(),
+ shuffleBlockInfo.getBlockId(),
+ shuffleBlockInfo.getLength(),
+ shuffleBlockInfo.getCrc(),
+ shuffleBlockInfo.getData(),
+ shuffleBlockInfo.getShuffleServerInfos(),
+ shuffleBlockInfo.getUncompressLength(),
+ shuffleBlockInfo.getFreeMemory(),
+ shuffleBlockInfo.getTaskAttemptId());
+ }
+
+ public static boolean compareSendShuffleDataRequest(SendShuffleDataRequest req1, SendShuffleDataRequest req2) {
Review Comment:
Class SendShuffleDataRequestTest seems unnecessary.
```
return req1.getPartitionId() == req2.getPartitionId()
&& req1.getBlockId() == req2.getBlockId()
&& req1.getLength() == req2.getLength()
&& req1.getShuffleId() == req2.getShuffleId()
&& req1.getCrc() == req2.getCrc()
&& req1.getTaskAttemptId() == req2.getTaskAttemptId()
&& req1.getUncompressLength() == req2.getUncompressLength()
&& req1.getFreeMemory() == req2.getFreeMemory()
&& req1.getData().equals(req2.getData())
&& CompareShuffleInfos(req1.getShuffleServerInfos(), req2.getShuffleServerInfos());
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] jerqi commented on a diff in pull request #742: [#133] feat(netty): Add Encoder and Decoder.
Posted by "jerqi (via GitHub)" <gi...@apache.org>.
jerqi commented on code in PR #742:
URL: https://github.com/apache/incubator-uniffle/pull/742#discussion_r1141406481
##########
common/src/main/java/org/apache/uniffle/common/ShuffleBlockInfo.java:
##########
@@ -27,14 +29,14 @@ public class ShuffleBlockInfo {
private int shuffleId;
private long crc;
private long taskAttemptId;
- private byte[] data;
+ private ByteBuf data;
private List<ShuffleServerInfo> shuffleServerInfos;
private int uncompressLength;
private long freeMemory;
public ShuffleBlockInfo(int shuffleId, int partitionId, long blockId, int length, long crc,
- byte[] data, List<ShuffleServerInfo> shuffleServerInfos,
- int uncompressLength, int freeMemory, long taskAttemptId) {
+ ByteBuf data, List<ShuffleServerInfo> shuffleServerInfos,
+ int uncompressLength, long freeMemory, long taskAttemptId) {
this.partitionId = partitionId;
Review Comment:
We can keep the original constructor for tests and add a new constructor for new implement.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] jerqi commented on a diff in pull request #742: [#133] feat(netty): Add Encoder and Decoder.
Posted by "jerqi (via GitHub)" <gi...@apache.org>.
jerqi commented on code in PR #742:
URL: https://github.com/apache/incubator-uniffle/pull/742#discussion_r1142428425
##########
common/src/main/java/org/apache/uniffle/common/netty/protocol/SendShuffleDataRequest.java:
##########
@@ -0,0 +1,153 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License 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 permissions and
+ * limitations under the License.
+ */
+
+package org.apache.uniffle.common.netty.protocol;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+
+import com.google.common.collect.Lists;
+import com.google.common.collect.Maps;
+import io.netty.buffer.ByteBuf;
+
+import org.apache.uniffle.common.ShuffleBlockInfo;
+import org.apache.uniffle.common.util.ByteBufUtils;
+
+public class SendShuffleDataRequest extends Message {
+ public long requestId;
+ private String appId;
+ private int shuffleId;
+ private long requireId;
+ private Map<Integer, List<ShuffleBlockInfo>> partitionToBlocks;
+ private long timestamp;
+
+ public SendShuffleDataRequest(long requestId, String appId, int shuffleId, long requireId,
+ Map<Integer, List<ShuffleBlockInfo>> partitionToBlocks, long timestamp) {
+ this.requestId = requestId;
+ this.appId = appId;
+ this.shuffleId = shuffleId;
+ this.requireId = requireId;
+ this.partitionToBlocks = partitionToBlocks;
+ this.timestamp = timestamp;
+ }
+
+ @Override
+ public Type type() {
+ return Type.SEND_SHUFFLE_DATA_REQUEST;
+ }
+
+ @Override
+ public int encodedLength() {
+ int encodeLength = Long.BYTES + ByteBufUtils.encodedLength(appId) + Integer.BYTES + Long.BYTES + Integer.BYTES;
+ for (Map.Entry<Integer, List<ShuffleBlockInfo>> entry : partitionToBlocks.entrySet()) {
+ encodeLength += 2 * Integer.BYTES;
+ for (ShuffleBlockInfo sbi : entry.getValue()) {
+ encodeLength += Encoders.encodeLengthOfShuffleBlockInfo(sbi);
+ }
+ }
+ return encodeLength + Long.BYTES;
+ }
+
+ @Override
+ public void encode(ByteBuf buf) {
+ buf.writeLong(requestId);
+ ByteBufUtils.writeLengthAndString(buf, appId);
+ buf.writeInt(shuffleId);
+ buf.writeLong(requireId);
+ encodePartitionData(buf);
+ buf.writeLong(timestamp);
+ }
+
+ private static Map<Integer, List<ShuffleBlockInfo>> decodePartitionData(ByteBuf byteBuf) {
+ Map<Integer, List<ShuffleBlockInfo>> partitionToBlocks = Maps.newHashMap();
+ int lengthOfPartitionData = byteBuf.readInt();
+ for (int i = 0; i < lengthOfPartitionData; i++) {
+ int partitionId = byteBuf.readInt();
+ int lengthOfShuffleBlocks = byteBuf.readInt();
+ List<ShuffleBlockInfo> shuffleBlockInfoList = Lists.newArrayList();
+ for (int j = 0; j < lengthOfShuffleBlocks; j++) {
+ shuffleBlockInfoList.add(Decoders.decodeShuffleBlockInfo(byteBuf));
+ }
+ partitionToBlocks.put(partitionId, shuffleBlockInfoList);
+ }
+ return partitionToBlocks;
+ }
+
+ public static SendShuffleDataRequest decode(ByteBuf byteBuf) {
+ long requestId = byteBuf.readLong();
+ String appId = ByteBufUtils.readLengthAndString(byteBuf);
+ int shuffleId = byteBuf.readInt();
+ long requireId = byteBuf.readLong();
+ Map<Integer, List<ShuffleBlockInfo>> partitionToBlocks = decodePartitionData(byteBuf);
+ long timestamp = byteBuf.readLong();
+ return new SendShuffleDataRequest(requestId, appId, shuffleId, requireId, partitionToBlocks, timestamp);
+ }
+
+ private void encodePartitionData(ByteBuf buf) {
+ buf.writeInt(partitionToBlocks.size());
+ for (Map.Entry<Integer, List<ShuffleBlockInfo>> entry : partitionToBlocks.entrySet()) {
+ buf.writeInt(entry.getKey());
+ buf.writeInt(entry.getValue().size());
+ for (ShuffleBlockInfo sbi : entry.getValue()) {
+ Encoders.encodeShuffleBlockInfo(sbi, buf);
+ }
+ }
+ }
+
+ public long getRequestId() {
+ return requestId;
+ }
+
+ public String getAppId() {
+ return appId;
+ }
+
+ public int getShuffleId() {
+ return shuffleId;
+ }
+
+ public long getRequireId() {
+ return requireId;
+ }
+
+ public Map<Integer, List<ShuffleBlockInfo>> getPartitionToBlocks() {
+ return partitionToBlocks;
+ }
+
+ @Override
+ public boolean equals(Object o) {
Review Comment:
Why do we need to override the method `equa`l?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] codecov-commenter commented on pull request #742: [#133] feat(netty): Add Encoder and Decoder.
Posted by "codecov-commenter (via GitHub)" <gi...@apache.org>.
codecov-commenter commented on PR #742:
URL: https://github.com/apache/incubator-uniffle/pull/742#issuecomment-1475302259
## [Codecov](https://codecov.io/gh/apache/incubator-uniffle/pull/742?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) Report
> Merging [#742](https://codecov.io/gh/apache/incubator-uniffle/pull/742?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) (2dca710) into [master](https://codecov.io/gh/apache/incubator-uniffle/commit/d60d675d38c833b99b012a1f4c726a012ce93463?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) (d60d675) will **increase** coverage by `3.49%`.
> The diff coverage is `66.90%`.
```diff
@@ Coverage Diff @@
## master #742 +/- ##
============================================
+ Coverage 60.63% 64.13% +3.49%
+ Complexity 1893 1806 -87
============================================
Files 238 216 -22
Lines 13000 10037 -2963
Branches 1090 1002 -88
============================================
- Hits 7883 6437 -1446
+ Misses 4679 3227 -1452
+ Partials 438 373 -65
```
| [Impacted Files](https://codecov.io/gh/apache/incubator-uniffle/pull/742?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) | Coverage Δ | |
|---|---|---|
| [...che/uniffle/common/netty/protocol/RpcResponse.java](https://codecov.io/gh/apache/incubator-uniffle/pull/742?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-Y29tbW9uL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS91bmlmZmxlL2NvbW1vbi9uZXR0eS9wcm90b2NvbC9ScGNSZXNwb25zZS5qYXZh) | `65.51% <12.50%> (-20.20%)` | :arrow_down: |
| [...va/org/apache/uniffle/common/ShuffleBlockInfo.java](https://codecov.io/gh/apache/incubator-uniffle/pull/742?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-Y29tbW9uL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS91bmlmZmxlL2NvbW1vbi9TaHVmZmxlQmxvY2tJbmZvLmphdmE=) | `82.97% <22.22%> (-4.21%)` | :arrow_down: |
| [...he/uniffle/common/netty/TransportFrameDecoder.java](https://codecov.io/gh/apache/incubator-uniffle/pull/742?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-Y29tbW9uL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS91bmlmZmxlL2NvbW1vbi9uZXR0eS9UcmFuc3BvcnRGcmFtZURlY29kZXIuamF2YQ==) | `50.60% <50.60%> (ø)` | |
| [.../apache/uniffle/common/netty/protocol/Message.java](https://codecov.io/gh/apache/incubator-uniffle/pull/742?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-Y29tbW9uL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS91bmlmZmxlL2NvbW1vbi9uZXR0eS9wcm90b2NvbC9NZXNzYWdlLmphdmE=) | `52.45% <54.76%> (+2.45%)` | :arrow_up: |
| [...rg/apache/uniffle/common/netty/MessageEncoder.java](https://codecov.io/gh/apache/incubator-uniffle/pull/742?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-Y29tbW9uL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS91bmlmZmxlL2NvbW1vbi9uZXR0eS9NZXNzYWdlRW5jb2Rlci5qYXZh) | `78.57% <78.57%> (ø)` | |
| [.../common/netty/protocol/SendShuffleDataRequest.java](https://codecov.io/gh/apache/incubator-uniffle/pull/742?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-Y29tbW9uL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS91bmlmZmxlL2NvbW1vbi9uZXR0eS9wcm90b2NvbC9TZW5kU2h1ZmZsZURhdGFSZXF1ZXN0LmphdmE=) | `80.95% <80.95%> (ø)` | |
| [...a/org/apache/uniffle/common/util/ByteBufUtils.java](https://codecov.io/gh/apache/incubator-uniffle/pull/742?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-Y29tbW9uL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS91bmlmZmxlL2NvbW1vbi91dGlsL0J5dGVCdWZVdGlscy5qYXZh) | `89.65% <81.81%> (-4.79%)` | :arrow_down: |
| [...apache/uniffle/common/netty/protocol/Decoders.java](https://codecov.io/gh/apache/incubator-uniffle/pull/742?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-Y29tbW9uL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS91bmlmZmxlL2NvbW1vbi9uZXR0eS9wcm90b2NvbC9EZWNvZGVycy5qYXZh) | `95.23% <95.23%> (ø)` | |
| [...apache/uniffle/common/netty/protocol/Encoders.java](https://codecov.io/gh/apache/incubator-uniffle/pull/742?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-Y29tbW9uL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS91bmlmZmxlL2NvbW1vbi9uZXR0eS9wcm90b2NvbC9FbmNvZGVycy5qYXZh) | `96.42% <96.42%> (ø)` | |
| [...pache/spark/shuffle/writer/WriteBufferManager.java](https://codecov.io/gh/apache/incubator-uniffle/pull/742?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-Y2xpZW50LXNwYXJrL2NvbW1vbi9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvc3Bhcmsvc2h1ZmZsZS93cml0ZXIvV3JpdGVCdWZmZXJNYW5hZ2VyLmphdmE=) | `83.21% <100.00%> (+0.11%)` | :arrow_up: |
... and [28 files with indirect coverage changes](https://codecov.io/gh/apache/incubator-uniffle/pull/742/indirect-changes?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation)
:mega: We’re building smart automated test selection to slash your CI/CD build times. [Learn more](https://about.codecov.io/iterative-testing/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] smallzhongfeng commented on pull request #742: [#133] feat(netty): Add Encoder and Decoder.
Posted by "smallzhongfeng (via GitHub)" <gi...@apache.org>.
smallzhongfeng commented on PR #742:
URL: https://github.com/apache/incubator-uniffle/pull/742#issuecomment-1484049142
Merged to master. Thanks @leixm
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] leixm commented on a diff in pull request #742: [#133] feat(netty): Add Encoder and Decoder.
Posted by "leixm (via GitHub)" <gi...@apache.org>.
leixm commented on code in PR #742:
URL: https://github.com/apache/incubator-uniffle/pull/742#discussion_r1141941223
##########
common/src/main/java/org/apache/uniffle/common/netty/protocol/SendShuffleDataRequest.java:
##########
@@ -0,0 +1,147 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License 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 permissions and
+ * limitations under the License.
+ */
+
+package org.apache.uniffle.common.netty.protocol;
Review Comment:
This protocol can help us test the Encoder and Decoder, SendShuffleDataRequest is the most complex protocol.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] leixm commented on a diff in pull request #742: [#133] feat(netty): Add Encoder and Decoder.
Posted by "leixm (via GitHub)" <gi...@apache.org>.
leixm commented on code in PR #742:
URL: https://github.com/apache/incubator-uniffle/pull/742#discussion_r1141963571
##########
internal-client/src/main/java/org/apache/uniffle/client/impl/grpc/ShuffleServerGrpcClient.java:
##########
@@ -319,7 +319,7 @@ public RssSendShuffleDataResponse sendShuffleData(RssSendShuffleDataRequest requ
.setLength(sbi.getLength())
.setTaskAttemptId(sbi.getTaskAttemptId())
.setUncompressLength(sbi.getUncompressLength())
- .setData(ByteString.copyFrom(sbi.getData()))
+ .setData(ByteString.copyFrom(sbi.getData().nioBuffer()))
Review Comment:
Okay, I will replace it with UnsafeByteOperations.unsafeWrap().
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] leixm commented on a diff in pull request #742: [#133] feat(netty): Add Encoder and Decoder.
Posted by "leixm (via GitHub)" <gi...@apache.org>.
leixm commented on code in PR #742:
URL: https://github.com/apache/incubator-uniffle/pull/742#discussion_r1142835083
##########
common/src/main/java/org/apache/uniffle/common/netty/protocol/SendShuffleDataRequest.java:
##########
@@ -0,0 +1,153 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License 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 permissions and
+ * limitations under the License.
+ */
+
+package org.apache.uniffle.common.netty.protocol;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+
+import com.google.common.collect.Lists;
+import com.google.common.collect.Maps;
+import io.netty.buffer.ByteBuf;
+
+import org.apache.uniffle.common.ShuffleBlockInfo;
+import org.apache.uniffle.common.util.ByteBufUtils;
+
+public class SendShuffleDataRequest extends Message {
+ public long requestId;
+ private String appId;
+ private int shuffleId;
+ private long requireId;
+ private Map<Integer, List<ShuffleBlockInfo>> partitionToBlocks;
+ private long timestamp;
+
+ public SendShuffleDataRequest(long requestId, String appId, int shuffleId, long requireId,
+ Map<Integer, List<ShuffleBlockInfo>> partitionToBlocks, long timestamp) {
+ this.requestId = requestId;
+ this.appId = appId;
+ this.shuffleId = shuffleId;
+ this.requireId = requireId;
+ this.partitionToBlocks = partitionToBlocks;
+ this.timestamp = timestamp;
+ }
+
+ @Override
+ public Type type() {
+ return Type.SEND_SHUFFLE_DATA_REQUEST;
+ }
+
+ @Override
+ public int encodedLength() {
+ int encodeLength = Long.BYTES + ByteBufUtils.encodedLength(appId) + Integer.BYTES + Long.BYTES + Integer.BYTES;
+ for (Map.Entry<Integer, List<ShuffleBlockInfo>> entry : partitionToBlocks.entrySet()) {
+ encodeLength += 2 * Integer.BYTES;
+ for (ShuffleBlockInfo sbi : entry.getValue()) {
+ encodeLength += Encoders.encodeLengthOfShuffleBlockInfo(sbi);
+ }
+ }
+ return encodeLength + Long.BYTES;
+ }
+
+ @Override
+ public void encode(ByteBuf buf) {
+ buf.writeLong(requestId);
+ ByteBufUtils.writeLengthAndString(buf, appId);
+ buf.writeInt(shuffleId);
+ buf.writeLong(requireId);
+ encodePartitionData(buf);
+ buf.writeLong(timestamp);
+ }
+
+ private static Map<Integer, List<ShuffleBlockInfo>> decodePartitionData(ByteBuf byteBuf) {
+ Map<Integer, List<ShuffleBlockInfo>> partitionToBlocks = Maps.newHashMap();
+ int lengthOfPartitionData = byteBuf.readInt();
+ for (int i = 0; i < lengthOfPartitionData; i++) {
+ int partitionId = byteBuf.readInt();
+ int lengthOfShuffleBlocks = byteBuf.readInt();
+ List<ShuffleBlockInfo> shuffleBlockInfoList = Lists.newArrayList();
+ for (int j = 0; j < lengthOfShuffleBlocks; j++) {
+ shuffleBlockInfoList.add(Decoders.decodeShuffleBlockInfo(byteBuf));
+ }
+ partitionToBlocks.put(partitionId, shuffleBlockInfoList);
+ }
+ return partitionToBlocks;
+ }
+
+ public static SendShuffleDataRequest decode(ByteBuf byteBuf) {
+ long requestId = byteBuf.readLong();
+ String appId = ByteBufUtils.readLengthAndString(byteBuf);
+ int shuffleId = byteBuf.readInt();
+ long requireId = byteBuf.readLong();
+ Map<Integer, List<ShuffleBlockInfo>> partitionToBlocks = decodePartitionData(byteBuf);
+ long timestamp = byteBuf.readLong();
+ return new SendShuffleDataRequest(requestId, appId, shuffleId, requireId, partitionToBlocks, timestamp);
+ }
+
+ private void encodePartitionData(ByteBuf buf) {
+ buf.writeInt(partitionToBlocks.size());
+ for (Map.Entry<Integer, List<ShuffleBlockInfo>> entry : partitionToBlocks.entrySet()) {
+ buf.writeInt(entry.getKey());
+ buf.writeInt(entry.getValue().size());
+ for (ShuffleBlockInfo sbi : entry.getValue()) {
+ Encoders.encodeShuffleBlockInfo(sbi, buf);
+ }
+ }
+ }
+
+ public long getRequestId() {
+ return requestId;
+ }
+
+ public String getAppId() {
+ return appId;
+ }
+
+ public int getShuffleId() {
+ return shuffleId;
+ }
+
+ public long getRequireId() {
+ return requireId;
+ }
+
+ public Map<Integer, List<ShuffleBlockInfo>> getPartitionToBlocks() {
+ return partitionToBlocks;
+ }
+
+ @Override
+ public boolean equals(Object o) {
Review Comment:
for test, see SendShuffleDataRequestTest.java.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] jerqi commented on a diff in pull request #742: [#133] feat(netty): Add Encoder and Decoder.
Posted by "jerqi (via GitHub)" <gi...@apache.org>.
jerqi commented on code in PR #742:
URL: https://github.com/apache/incubator-uniffle/pull/742#discussion_r1142837538
##########
common/src/main/java/org/apache/uniffle/common/netty/protocol/SendShuffleDataRequest.java:
##########
@@ -0,0 +1,153 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License 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 permissions and
+ * limitations under the License.
+ */
+
+package org.apache.uniffle.common.netty.protocol;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+
+import com.google.common.collect.Lists;
+import com.google.common.collect.Maps;
+import io.netty.buffer.ByteBuf;
+
+import org.apache.uniffle.common.ShuffleBlockInfo;
+import org.apache.uniffle.common.util.ByteBufUtils;
+
+public class SendShuffleDataRequest extends Message {
+ public long requestId;
+ private String appId;
+ private int shuffleId;
+ private long requireId;
+ private Map<Integer, List<ShuffleBlockInfo>> partitionToBlocks;
+ private long timestamp;
+
+ public SendShuffleDataRequest(long requestId, String appId, int shuffleId, long requireId,
+ Map<Integer, List<ShuffleBlockInfo>> partitionToBlocks, long timestamp) {
+ this.requestId = requestId;
+ this.appId = appId;
+ this.shuffleId = shuffleId;
+ this.requireId = requireId;
+ this.partitionToBlocks = partitionToBlocks;
+ this.timestamp = timestamp;
+ }
+
+ @Override
+ public Type type() {
+ return Type.SEND_SHUFFLE_DATA_REQUEST;
+ }
+
+ @Override
+ public int encodedLength() {
+ int encodeLength = Long.BYTES + ByteBufUtils.encodedLength(appId) + Integer.BYTES + Long.BYTES + Integer.BYTES;
+ for (Map.Entry<Integer, List<ShuffleBlockInfo>> entry : partitionToBlocks.entrySet()) {
+ encodeLength += 2 * Integer.BYTES;
+ for (ShuffleBlockInfo sbi : entry.getValue()) {
+ encodeLength += Encoders.encodeLengthOfShuffleBlockInfo(sbi);
+ }
+ }
+ return encodeLength + Long.BYTES;
+ }
+
+ @Override
+ public void encode(ByteBuf buf) {
+ buf.writeLong(requestId);
+ ByteBufUtils.writeLengthAndString(buf, appId);
+ buf.writeInt(shuffleId);
+ buf.writeLong(requireId);
+ encodePartitionData(buf);
+ buf.writeLong(timestamp);
+ }
+
+ private static Map<Integer, List<ShuffleBlockInfo>> decodePartitionData(ByteBuf byteBuf) {
+ Map<Integer, List<ShuffleBlockInfo>> partitionToBlocks = Maps.newHashMap();
+ int lengthOfPartitionData = byteBuf.readInt();
+ for (int i = 0; i < lengthOfPartitionData; i++) {
+ int partitionId = byteBuf.readInt();
+ int lengthOfShuffleBlocks = byteBuf.readInt();
+ List<ShuffleBlockInfo> shuffleBlockInfoList = Lists.newArrayList();
+ for (int j = 0; j < lengthOfShuffleBlocks; j++) {
+ shuffleBlockInfoList.add(Decoders.decodeShuffleBlockInfo(byteBuf));
+ }
+ partitionToBlocks.put(partitionId, shuffleBlockInfoList);
+ }
+ return partitionToBlocks;
+ }
+
+ public static SendShuffleDataRequest decode(ByteBuf byteBuf) {
+ long requestId = byteBuf.readLong();
+ String appId = ByteBufUtils.readLengthAndString(byteBuf);
+ int shuffleId = byteBuf.readInt();
+ long requireId = byteBuf.readLong();
+ Map<Integer, List<ShuffleBlockInfo>> partitionToBlocks = decodePartitionData(byteBuf);
+ long timestamp = byteBuf.readLong();
+ return new SendShuffleDataRequest(requestId, appId, shuffleId, requireId, partitionToBlocks, timestamp);
+ }
+
+ private void encodePartitionData(ByteBuf buf) {
+ buf.writeInt(partitionToBlocks.size());
+ for (Map.Entry<Integer, List<ShuffleBlockInfo>> entry : partitionToBlocks.entrySet()) {
+ buf.writeInt(entry.getKey());
+ buf.writeInt(entry.getValue().size());
+ for (ShuffleBlockInfo sbi : entry.getValue()) {
+ Encoders.encodeShuffleBlockInfo(sbi, buf);
+ }
+ }
+ }
+
+ public long getRequestId() {
+ return requestId;
+ }
+
+ public String getAppId() {
+ return appId;
+ }
+
+ public int getShuffleId() {
+ return shuffleId;
+ }
+
+ public long getRequireId() {
+ return requireId;
+ }
+
+ public Map<Integer, List<ShuffleBlockInfo>> getPartitionToBlocks() {
+ return partitionToBlocks;
+ }
+
+ @Override
+ public boolean equals(Object o) {
Review Comment:
I worried about performance of this method `equal`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] jerqi commented on a diff in pull request #742: [#133] feat(netty): Add Encoder and Decoder.
Posted by "jerqi (via GitHub)" <gi...@apache.org>.
jerqi commented on code in PR #742:
URL: https://github.com/apache/incubator-uniffle/pull/742#discussion_r1145096325
##########
common/src/main/java/org/apache/uniffle/common/netty/TransportFrameDecoder.java:
##########
@@ -0,0 +1,191 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License 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 permissions and
+ * limitations under the License.
+ */
+
+package org.apache.uniffle.common.netty;
+
+import java.util.LinkedList;
+
+import com.google.common.base.Preconditions;
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.CompositeByteBuf;
+import io.netty.buffer.Unpooled;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelInboundHandlerAdapter;
+
+import org.apache.uniffle.common.netty.protocol.Message;
+
+/**
+ * A customized frame decoder that allows intercepting raw data.
+ *
+ * <p>This behaves like Netty's frame decoder (with hard coded parameters that match this library's
+ * needs), except it allows an interceptor to be installed to read data directly before it's framed.
+ *
+ * <p>Unlike Netty's frame decoder, each frame is dispatched to child handlers as soon as it's
+ * decoded, instead of building as many frames as the current buffer allows and dispatching all of
+ * them. This allows a child handler to install an interceptor if needed.
+ *
+ * <p>If an interceptor is installed, framing stops, and data is instead fed directly to the
+ * interceptor. When the interceptor indicates that it doesn't need to read any more data, framing
+ * resumes. Interceptors should not hold references to the data buffers provided to their handle()
+ * method.
+ */
+public class TransportFrameDecoder extends ChannelInboundHandlerAdapter implements FrameDecoder {
+ private int msgSize = -1;
+ private Message.Type curType = Message.Type.UNKNOWN_TYPE;
+ private ByteBuf headerBuf = Unpooled.buffer(HEADER_SIZE, HEADER_SIZE);
+ private static final int MAX_FRAME_SIZE = Integer.MAX_VALUE;
+ private static final int UNKNOWN_FRAME_SIZE = -1;
+
+ private final LinkedList<ByteBuf> buffers = new LinkedList<>();
+
+ private long totalSize = 0;
+ private long nextFrameSize = UNKNOWN_FRAME_SIZE;
+
+ @Override
+ public void channelRead(ChannelHandlerContext ctx, Object data) {
+ ByteBuf in = (ByteBuf) data;
+ buffers.add(in);
+ totalSize += in.readableBytes();
+
+ while (!buffers.isEmpty()) {
+ ByteBuf frame = decodeNext();
+ if (frame == null) {
+ break;
+ }
+ Message msg = Message.decode(curType, frame);
Review Comment:
No, if some data is corrupted, it may throw IllegalArgumentException or IllegalStateException. I don't ensure that. Maybe we should have more discussion and refer to other systems.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] zuston commented on a diff in pull request #742: [#133] feat(netty): Add Encoder and Decoder.
Posted by "zuston (via GitHub)" <gi...@apache.org>.
zuston commented on code in PR #742:
URL: https://github.com/apache/incubator-uniffle/pull/742#discussion_r1145585978
##########
common/src/main/java/org/apache/uniffle/common/netty/protocol/Message.java:
##########
@@ -53,6 +73,46 @@ public static Type decode(ByteBuf buf) {
switch (id) {
case 0:
return RPC_RESPONSE;
+ case 1:
+ return SHUFFLE_REGISTER_REQUEST;
Review Comment:
We will only replace the send/get data grpc api, right? If so, these different types are necessary?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] leixm commented on a diff in pull request #742: [#133] feat(netty): Add Encoder and Decoder.
Posted by "leixm (via GitHub)" <gi...@apache.org>.
leixm commented on code in PR #742:
URL: https://github.com/apache/incubator-uniffle/pull/742#discussion_r1146171772
##########
common/src/main/java/org/apache/uniffle/common/netty/protocol/Encoders.java:
##########
@@ -0,0 +1,71 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License 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 permissions and
+ * limitations under the License.
+ */
+
+package org.apache.uniffle.common.netty.protocol;
+
+import java.util.List;
+
+import io.netty.buffer.ByteBuf;
+
+import org.apache.uniffle.common.ShuffleBlockInfo;
+import org.apache.uniffle.common.ShuffleServerInfo;
+import org.apache.uniffle.common.util.ByteBufUtils;
+
+public class Encoders {
Review Comment:
> From this implementation of `Encoders` and `Decoders`, I think they are bound with each other. How about introducing a `codec` ,which will include encode and decode method?
Great idea!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] leixm commented on a diff in pull request #742: [#133] feat(netty): Add Encoder and Decoder.
Posted by "leixm (via GitHub)" <gi...@apache.org>.
leixm commented on code in PR #742:
URL: https://github.com/apache/incubator-uniffle/pull/742#discussion_r1142846642
##########
common/src/main/java/org/apache/uniffle/common/netty/protocol/Encoders.java:
##########
@@ -0,0 +1,69 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License 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 permissions and
+ * limitations under the License.
+ */
+
+package org.apache.uniffle.common.netty.protocol;
+
+import java.util.List;
+
+import io.netty.buffer.ByteBuf;
+
+import org.apache.uniffle.common.ShuffleBlockInfo;
+import org.apache.uniffle.common.ShuffleServerInfo;
+import org.apache.uniffle.common.util.ByteBufUtils;
+
+public class Encoders {
+ public static void encodeShuffleServerInfo(ShuffleServerInfo shuffleServerInfo, ByteBuf byteBuf) {
+ ByteBufUtils.writeLengthAndString(byteBuf, shuffleServerInfo.getId());
+ ByteBufUtils.writeLengthAndString(byteBuf, shuffleServerInfo.getHost());
+ byteBuf.writeInt(shuffleServerInfo.getGrpcPort());
+ byteBuf.writeInt(shuffleServerInfo.getNettyPort());
+ }
+
+ public static void encodeShuffleBlockInfo(ShuffleBlockInfo shuffleBlockInfo, ByteBuf byteBuf) {
+ byteBuf.writeInt(shuffleBlockInfo.getPartitionId());
+ byteBuf.writeLong(shuffleBlockInfo.getBlockId());
+ byteBuf.writeInt(shuffleBlockInfo.getLength());
+ byteBuf.writeInt(shuffleBlockInfo.getShuffleId());
+ byteBuf.writeLong(shuffleBlockInfo.getCrc());
+ byteBuf.writeLong(shuffleBlockInfo.getTaskAttemptId());
+ ByteBufUtils.copyByteBuf(shuffleBlockInfo.getData(), byteBuf);
Review Comment:
> Could we use CompositeByteBuf to avoid data copy here?
It doesn’t seem to work. The byteBuf here comes from MessageEncoder. I checked the source code of Spark. How does it handle it? Spark will divide the Message into Header and body. The body is usually a ByteBuf and implements a MessageWithHeader (link: https://github.com/apache/spark/blob/master/common/network-common/src/main/java/org/apache/spark/network/protocol/MessageWithHeader.java), but also cannot avoid copy, see line 123: `writtenBody = copyByteBuf((ByteBuf) body, target);`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] leixm commented on a diff in pull request #742: [#133] feat(netty): Add Encoder and Decoder.
Posted by "leixm (via GitHub)" <gi...@apache.org>.
leixm commented on code in PR #742:
URL: https://github.com/apache/incubator-uniffle/pull/742#discussion_r1146178309
##########
common/src/main/java/org/apache/uniffle/common/netty/TransportFrameDecoder.java:
##########
@@ -0,0 +1,191 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License 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 permissions and
+ * limitations under the License.
+ */
+
+package org.apache.uniffle.common.netty;
+
+import java.util.LinkedList;
+
+import com.google.common.base.Preconditions;
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.CompositeByteBuf;
+import io.netty.buffer.Unpooled;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelInboundHandlerAdapter;
+
+import org.apache.uniffle.common.netty.protocol.Message;
+
+/**
+ * A customized frame decoder that allows intercepting raw data.
+ *
+ * <p>This behaves like Netty's frame decoder (with hard coded parameters that match this library's
+ * needs), except it allows an interceptor to be installed to read data directly before it's framed.
+ *
+ * <p>Unlike Netty's frame decoder, each frame is dispatched to child handlers as soon as it's
+ * decoded, instead of building as many frames as the current buffer allows and dispatching all of
+ * them. This allows a child handler to install an interceptor if needed.
+ *
+ * <p>If an interceptor is installed, framing stops, and data is instead fed directly to the
+ * interceptor. When the interceptor indicates that it doesn't need to read any more data, framing
+ * resumes. Interceptors should not hold references to the data buffers provided to their handle()
+ * method.
+ */
+public class TransportFrameDecoder extends ChannelInboundHandlerAdapter implements FrameDecoder {
+ private int msgSize = -1;
+ private Message.Type curType = Message.Type.UNKNOWN_TYPE;
+ private ByteBuf headerBuf = Unpooled.buffer(HEADER_SIZE, HEADER_SIZE);
+ private static final int MAX_FRAME_SIZE = Integer.MAX_VALUE;
+ private static final int UNKNOWN_FRAME_SIZE = -1;
+
+ private final LinkedList<ByteBuf> buffers = new LinkedList<>();
+
+ private long totalSize = 0;
+ private long nextFrameSize = UNKNOWN_FRAME_SIZE;
+
+ @Override
+ public void channelRead(ChannelHandlerContext ctx, Object data) {
+ ByteBuf in = (ByteBuf) data;
+ buffers.add(in);
+ totalSize += in.readableBytes();
+
+ while (!buffers.isEmpty()) {
+ ByteBuf frame = decodeNext();
+ if (frame == null) {
+ break;
+ }
+ Message msg = Message.decode(curType, frame);
+ frame.release();
+ ctx.fireChannelRead(msg);
+ clear();
+ }
+ }
+
+ private void clear() {
+ curType = Message.Type.UNKNOWN_TYPE;
+ msgSize = -1;
+ headerBuf.clear();
+ }
+
+ private long decodeFrameSize() {
+ if (nextFrameSize != UNKNOWN_FRAME_SIZE || totalSize < HEADER_SIZE) {
+ return nextFrameSize;
+ }
+
+ // We know there's enough data. If the first buffer contains all the data, great. Otherwise,
+ // hold the bytes for the frame length in a composite buffer until we have enough data to read
+ // the frame size. Normally, it should be rare to need more than one buffer to read the frame
+ // size.
+ ByteBuf first = buffers.getFirst();
+ if (first.readableBytes() >= HEADER_SIZE) {
+ msgSize = first.readInt();
+ curType = Message.Type.decode(first);
+ nextFrameSize = msgSize;
+ totalSize -= HEADER_SIZE;
+ if (!first.isReadable()) {
+ buffers.removeFirst().release();
+ }
+ return nextFrameSize;
+ }
+
+ while (headerBuf.readableBytes() < HEADER_SIZE) {
+ ByteBuf next = buffers.getFirst();
+ int toRead = Math.min(next.readableBytes(), HEADER_SIZE - headerBuf.readableBytes());
+ headerBuf.writeBytes(next, toRead);
+ if (!next.isReadable()) {
+ buffers.removeFirst().release();
+ }
+ }
+
+ msgSize = headerBuf.readInt();
+ curType = Message.Type.decode(headerBuf);
+ nextFrameSize = msgSize;
+ totalSize -= HEADER_SIZE;
+ return nextFrameSize;
+ }
+
+ private ByteBuf decodeNext() {
+ long frameSize = decodeFrameSize();
+ if (frameSize == UNKNOWN_FRAME_SIZE || totalSize < frameSize) {
+ return null;
+ }
+
+ // Reset size for next frame.
+ nextFrameSize = UNKNOWN_FRAME_SIZE;
+
+ Preconditions.checkArgument(frameSize < MAX_FRAME_SIZE, "Too large frame: %s", frameSize);
Review Comment:
Maybe not very important, Spark does the same.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] leixm commented on a diff in pull request #742: [#133] feat(netty): Add Encoder and Decoder.
Posted by "leixm (via GitHub)" <gi...@apache.org>.
leixm commented on code in PR #742:
URL: https://github.com/apache/incubator-uniffle/pull/742#discussion_r1145103751
##########
common/src/main/java/org/apache/uniffle/common/netty/TransportFrameDecoder.java:
##########
@@ -0,0 +1,191 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License 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 permissions and
+ * limitations under the License.
+ */
+
+package org.apache.uniffle.common.netty;
+
+import java.util.LinkedList;
+
+import com.google.common.base.Preconditions;
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.CompositeByteBuf;
+import io.netty.buffer.Unpooled;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelInboundHandlerAdapter;
+
+import org.apache.uniffle.common.netty.protocol.Message;
+
+/**
+ * A customized frame decoder that allows intercepting raw data.
+ *
+ * <p>This behaves like Netty's frame decoder (with hard coded parameters that match this library's
+ * needs), except it allows an interceptor to be installed to read data directly before it's framed.
+ *
+ * <p>Unlike Netty's frame decoder, each frame is dispatched to child handlers as soon as it's
+ * decoded, instead of building as many frames as the current buffer allows and dispatching all of
+ * them. This allows a child handler to install an interceptor if needed.
+ *
+ * <p>If an interceptor is installed, framing stops, and data is instead fed directly to the
+ * interceptor. When the interceptor indicates that it doesn't need to read any more data, framing
+ * resumes. Interceptors should not hold references to the data buffers provided to their handle()
+ * method.
+ */
+public class TransportFrameDecoder extends ChannelInboundHandlerAdapter implements FrameDecoder {
+ private int msgSize = -1;
+ private Message.Type curType = Message.Type.UNKNOWN_TYPE;
+ private ByteBuf headerBuf = Unpooled.buffer(HEADER_SIZE, HEADER_SIZE);
+ private static final int MAX_FRAME_SIZE = Integer.MAX_VALUE;
+ private static final int UNKNOWN_FRAME_SIZE = -1;
+
+ private final LinkedList<ByteBuf> buffers = new LinkedList<>();
+
+ private long totalSize = 0;
+ private long nextFrameSize = UNKNOWN_FRAME_SIZE;
+
+ @Override
+ public void channelRead(ChannelHandlerContext ctx, Object data) {
+ ByteBuf in = (ByteBuf) data;
+ buffers.add(in);
+ totalSize += in.readableBytes();
+
+ while (!buffers.isEmpty()) {
+ ByteBuf frame = decodeNext();
+ if (frame == null) {
+ break;
+ }
+ Message msg = Message.decode(curType, frame);
Review Comment:
ok.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] jerqi commented on pull request #742: [#133] feat(netty): Add Encoder and Decoder.
Posted by "jerqi (via GitHub)" <gi...@apache.org>.
jerqi commented on PR #742:
URL: https://github.com/apache/incubator-uniffle/pull/742#issuecomment-1479891875
Let @smallzhongfeng and @zuston take another look. LGTM except some minors.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] jerqi commented on a diff in pull request #742: [#133] feat(netty): Add Encoder and Decoder.
Posted by "jerqi (via GitHub)" <gi...@apache.org>.
jerqi commented on code in PR #742:
URL: https://github.com/apache/incubator-uniffle/pull/742#discussion_r1141406412
##########
common/src/main/java/org/apache/uniffle/common/ShuffleBlockInfo.java:
##########
@@ -27,14 +29,14 @@ public class ShuffleBlockInfo {
private int shuffleId;
private long crc;
private long taskAttemptId;
- private byte[] data;
+ private ByteBuf data;
private List<ShuffleServerInfo> shuffleServerInfos;
private int uncompressLength;
private long freeMemory;
public ShuffleBlockInfo(int shuffleId, int partitionId, long blockId, int length, long crc,
- byte[] data, List<ShuffleServerInfo> shuffleServerInfos,
- int uncompressLength, int freeMemory, long taskAttemptId) {
+ ByteBuf data, List<ShuffleServerInfo> shuffleServerInfos,
+ int uncompressLength, long freeMemory, long taskAttemptId) {
this.partitionId = partitionId;
Review Comment:
We can keep the original constructor for tests and add a new constructor for new implement.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] leixm commented on a diff in pull request #742: [#133] feat(netty): Add Encoder and Decoder.
Posted by "leixm (via GitHub)" <gi...@apache.org>.
leixm commented on code in PR #742:
URL: https://github.com/apache/incubator-uniffle/pull/742#discussion_r1142862693
##########
common/src/main/java/org/apache/uniffle/common/ShuffleBlockInfo.java:
##########
@@ -27,14 +31,21 @@ public class ShuffleBlockInfo {
private int shuffleId;
private long crc;
private long taskAttemptId;
- private byte[] data;
+ private ByteBuf data;
Review Comment:
Ok, thank you @advancedxy .
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] leixm commented on a diff in pull request #742: [#133] feat(netty): Add Encoder and Decoder.
Posted by "leixm (via GitHub)" <gi...@apache.org>.
leixm commented on code in PR #742:
URL: https://github.com/apache/incubator-uniffle/pull/742#discussion_r1142836611
##########
common/src/main/java/org/apache/uniffle/common/netty/protocol/Encoders.java:
##########
@@ -0,0 +1,69 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License 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 permissions and
+ * limitations under the License.
+ */
+
+package org.apache.uniffle.common.netty.protocol;
+
+import java.util.List;
+
+import io.netty.buffer.ByteBuf;
+
+import org.apache.uniffle.common.ShuffleBlockInfo;
+import org.apache.uniffle.common.ShuffleServerInfo;
+import org.apache.uniffle.common.util.ByteBufUtils;
+
+public class Encoders {
+ public static void encodeShuffleServerInfo(ShuffleServerInfo shuffleServerInfo, ByteBuf byteBuf) {
+ ByteBufUtils.writeLengthAndString(byteBuf, shuffleServerInfo.getId());
+ ByteBufUtils.writeLengthAndString(byteBuf, shuffleServerInfo.getHost());
+ byteBuf.writeInt(shuffleServerInfo.getGrpcPort());
+ byteBuf.writeInt(shuffleServerInfo.getNettyPort());
+ }
+
+ public static void encodeShuffleBlockInfo(ShuffleBlockInfo shuffleBlockInfo, ByteBuf byteBuf) {
+ byteBuf.writeInt(shuffleBlockInfo.getPartitionId());
+ byteBuf.writeLong(shuffleBlockInfo.getBlockId());
+ byteBuf.writeInt(shuffleBlockInfo.getLength());
+ byteBuf.writeInt(shuffleBlockInfo.getShuffleId());
+ byteBuf.writeLong(shuffleBlockInfo.getCrc());
+ byteBuf.writeLong(shuffleBlockInfo.getTaskAttemptId());
+ ByteBufUtils.copyByteBuf(shuffleBlockInfo.getData(), byteBuf);
Review Comment:
> Do we need to release memory after copying here?
>
> ```
> shuffleBlockInfo.getData().release().
> ```
If ByteBuf uses heap memory, I think there is no need to manually call release and this part of memory will be reclaimed during gc. Secondly, there may be retries. The current encode does not mean that the request must be successful.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] leixm commented on a diff in pull request #742: [#133] feat(netty): Add Encoder and Decoder.
Posted by "leixm (via GitHub)" <gi...@apache.org>.
leixm commented on code in PR #742:
URL: https://github.com/apache/incubator-uniffle/pull/742#discussion_r1142834409
##########
common/src/main/java/org/apache/uniffle/common/netty/protocol/Decoders.java:
##########
@@ -0,0 +1,58 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License 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 permissions and
+ * limitations under the License.
+ */
+
+package org.apache.uniffle.common.netty.protocol;
+
+import java.util.List;
+
+import com.google.common.collect.Lists;
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.Unpooled;
+
+import org.apache.uniffle.common.ShuffleBlockInfo;
+import org.apache.uniffle.common.ShuffleServerInfo;
+import org.apache.uniffle.common.util.ByteBufUtils;
+
+public class Decoders {
+ public static ShuffleServerInfo decodeShuffleServerInfo(ByteBuf byteBuf) {
+ String id = ByteBufUtils.readLengthAndString(byteBuf);
+ String host = ByteBufUtils.readLengthAndString(byteBuf);
+ int grpcPort = byteBuf.readInt();
+ int nettyPort = byteBuf.readInt();
+ return new ShuffleServerInfo(id, host, grpcPort, nettyPort);
+ }
+
+ public static ShuffleBlockInfo decodeShuffleBlockInfo(ByteBuf byteBuf) {
+ int partId = byteBuf.readInt();
+ long blockId = byteBuf.readLong();
+ int length = byteBuf.readInt();
+ int shuffleId = byteBuf.readInt();
+ long crc = byteBuf.readLong();
+ long taskAttemptId = byteBuf.readLong();
+ // todo: we can readSlice here, but it needs to be released manually after use, or it will cause a memory leak
+ byte[] data = ByteBufUtils.readByteArray(byteBuf);
Review Comment:
sure.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] jerqi commented on a diff in pull request #742: [#133] feat(netty): Add Encoder and Decoder.
Posted by "jerqi (via GitHub)" <gi...@apache.org>.
jerqi commented on code in PR #742:
URL: https://github.com/apache/incubator-uniffle/pull/742#discussion_r1142848373
##########
common/src/main/java/org/apache/uniffle/common/netty/protocol/Encoders.java:
##########
@@ -0,0 +1,69 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License 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 permissions and
+ * limitations under the License.
+ */
+
+package org.apache.uniffle.common.netty.protocol;
+
+import java.util.List;
+
+import io.netty.buffer.ByteBuf;
+
+import org.apache.uniffle.common.ShuffleBlockInfo;
+import org.apache.uniffle.common.ShuffleServerInfo;
+import org.apache.uniffle.common.util.ByteBufUtils;
+
+public class Encoders {
+ public static void encodeShuffleServerInfo(ShuffleServerInfo shuffleServerInfo, ByteBuf byteBuf) {
+ ByteBufUtils.writeLengthAndString(byteBuf, shuffleServerInfo.getId());
+ ByteBufUtils.writeLengthAndString(byteBuf, shuffleServerInfo.getHost());
+ byteBuf.writeInt(shuffleServerInfo.getGrpcPort());
+ byteBuf.writeInt(shuffleServerInfo.getNettyPort());
+ }
+
+ public static void encodeShuffleBlockInfo(ShuffleBlockInfo shuffleBlockInfo, ByteBuf byteBuf) {
+ byteBuf.writeInt(shuffleBlockInfo.getPartitionId());
+ byteBuf.writeLong(shuffleBlockInfo.getBlockId());
+ byteBuf.writeInt(shuffleBlockInfo.getLength());
+ byteBuf.writeInt(shuffleBlockInfo.getShuffleId());
+ byteBuf.writeLong(shuffleBlockInfo.getCrc());
+ byteBuf.writeLong(shuffleBlockInfo.getTaskAttemptId());
+ ByteBufUtils.copyByteBuf(shuffleBlockInfo.getData(), byteBuf);
Review Comment:
You can see that
Spark copy `Bytebuf` to `Channel`.
It's ok.
We copy `ByteBuf` to `ByteBuf`.
This should be optimized.
##########
common/src/main/java/org/apache/uniffle/common/netty/protocol/Encoders.java:
##########
@@ -0,0 +1,69 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License 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 permissions and
+ * limitations under the License.
+ */
+
+package org.apache.uniffle.common.netty.protocol;
+
+import java.util.List;
+
+import io.netty.buffer.ByteBuf;
+
+import org.apache.uniffle.common.ShuffleBlockInfo;
+import org.apache.uniffle.common.ShuffleServerInfo;
+import org.apache.uniffle.common.util.ByteBufUtils;
+
+public class Encoders {
+ public static void encodeShuffleServerInfo(ShuffleServerInfo shuffleServerInfo, ByteBuf byteBuf) {
+ ByteBufUtils.writeLengthAndString(byteBuf, shuffleServerInfo.getId());
+ ByteBufUtils.writeLengthAndString(byteBuf, shuffleServerInfo.getHost());
+ byteBuf.writeInt(shuffleServerInfo.getGrpcPort());
+ byteBuf.writeInt(shuffleServerInfo.getNettyPort());
+ }
+
+ public static void encodeShuffleBlockInfo(ShuffleBlockInfo shuffleBlockInfo, ByteBuf byteBuf) {
+ byteBuf.writeInt(shuffleBlockInfo.getPartitionId());
+ byteBuf.writeLong(shuffleBlockInfo.getBlockId());
+ byteBuf.writeInt(shuffleBlockInfo.getLength());
+ byteBuf.writeInt(shuffleBlockInfo.getShuffleId());
+ byteBuf.writeLong(shuffleBlockInfo.getCrc());
+ byteBuf.writeLong(shuffleBlockInfo.getTaskAttemptId());
+ ByteBufUtils.copyByteBuf(shuffleBlockInfo.getData(), byteBuf);
Review Comment:
You can see that
Spark copy `Bytebuf` to `Channel`.
It's ok.
We copy `ByteBuf` to `ByteBuf` here.
This should be optimized.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org