New RPC protocol #1276

ujjboy · 2019-07-07T15:42:55Z

Ⅰ. Describe what this PR did

Add new protocol codec.
Change msgId to int
The response uses the same codec and compressor with the request.

Ⅱ. Does this pull request fix one issue?

#893

Ⅲ. Why don't you add test cases (unit test/integration test)?

Ⅳ. Describe how to verify it

Ⅴ. Special notes for reviews

codecov-io · 2019-07-07T15:46:59Z

Codecov Report

Merging #1276 into develop will decrease coverage by 1.13%.
The diff coverage is 7.05%.

@@              Coverage Diff              @@
##             develop    #1276      +/-   ##
=============================================
- Coverage      49.44%   48.31%   -1.14%     
+ Complexity      1652     1643       -9     
=============================================
  Files            326      334       +8     
  Lines          11446    11600     +154     
  Branches        1412     1432      +20     
=============================================
- Hits            5659     5604      -55     
- Misses          5157     5376     +219     
+ Partials         630      620      -10

Impacted Files	Coverage Δ	Complexity Δ
...ava/io/seata/core/constants/ConfigurationKeys.java	`0% <ø> (ø)`	`0 <0> (ø)`	⬇️
...c/main/java/io/seata/core/rpc/netty/RpcServer.java	`0% <0%> (ø)`	`0 <0> (ø)`	⬇️
.../java/io/seata/core/protocol/HeartbeatMessage.java	`0% <0%> (ø)`	`0 <0> (ø)`	⬇️
...ata/core/rpc/DefaultServerMessageListenerImpl.java	`0% <0%> (ø)`	`0 <0> (ø)`	⬇️
...ava/io/seata/core/rpc/netty/RmMessageListener.java	`0% <0%> (ø)`	`0 <0> (ø)`	⬇️
.../io/seata/core/rpc/netty/v1/ProtocolV1Encoder.java	`0% <0%> (ø)`	`0 <0> (?)`
...o/seata/common/io/UnsafeByteArrayOutputStream.java	`0% <0%> (ø)`	`0 <0> (?)`
.../io/seata/core/rpc/netty/v1/HeadMapSerializer.java	`0% <0%> (ø)`	`0 <0> (?)`
.../main/java/io/seata/core/codec/CompressorType.java	`0% <0%> (ø)`	`0 <0> (?)`
.../io/seata/core/rpc/netty/v1/ProtocolV1Decoder.java	`0% <0%> (ø)`	`0 <0> (?)`
... and 26 more

Continue to review full report at Codecov.

Legend - Click here to learn more
Δ = absolute <relative> (impact), ø = not affected, ? = missing data
Powered by Codecov. Last update a125233...85c37d1. Read the comment docs.

slievrly · 2019-07-08T08:58:43Z

common/src/main/java/io/seata/common/thread/PositiveAtomicCounter.java

+ * @author Geng Zhang
+ */
+public class PositiveAtomicCounter {
+    private static final int    MASK = 0x7FFFFFFF;


range:0~2^31-1 , how about 0~2^32-1 ?

slievrly · 2019-07-08T09:07:03Z

common/src/main/java/io/seata/common/io/UnsafeByteArrayInputStream.java

@@ -0,0 +1,120 @@
+/*
+ *  Copyright 1999-2019 Seata.io Group.


should use apache/dubbo license header.

slievrly · 2019-07-08T09:08:12Z

common/src/main/java/io/seata/common/io/UnsafeByteArrayOutputStream.java

@@ -0,0 +1,103 @@
+/*
+ *  Copyright 1999-2019 Seata.io Group.


should use apache/dubbo license header.

slievrly · 2019-07-08T09:17:51Z

core/src/main/java/io/seata/core/rpc/netty/AbstractRpcRemoting.java

-        rpcMessage.setRequest(true);
+        rpcMessage.setMessageType(msg instanceof HeartbeatMessage ?
+                ProtocolConstants.MSGTYPE_RESQUEST
+                : ProtocolConstants.MSGTYPE_HEARTBEAT);


slievrly · 2019-07-08T09:23:58Z

common/src/main/java/io/seata/common/util/CodecUtils.java

+     * @param b 原始byte
+     * @return byte数组{<16,<16}
+     */
+    public static byte[] parseHigh4Low4Bytes(byte b) {


The method name is a bit confusing, remove all Chinese comments.

slievrly · 2019-07-08T09:42:38Z

core/src/main/java/io/seata/core/rpc/netty/AbstractRpcRemoting.java

     * @param channel the channel
     * @param msg     the msg
     */
-    protected void sendResponse(long msgId, Channel channel, Object msg) {
+    protected void sendResponse(RpcMessage request, Channel channel, Object msg) {


just use request.getId why wrap as RpcMessage. if use RpcMessage ,it is best to wrap msg in the body

slievrly · 2019-07-08T09:49:09Z

core/src/main/java/io/seata/core/rpc/ServerMessageListener.java

     * @param ctx     the ctx
     * @param message the message
     * @param sender  the sender
     */
-    void onTrxMessage(long msgId, ChannelHandlerContext ctx, Object message, ServerMessageSender sender);
+    void onTrxMessage(RpcMessage request, ChannelHandlerContext ctx, Object message, ServerMessageSender sender);


RpcMessage contains message.

slievrly · 2019-07-08T09:49:32Z

core/src/main/java/io/seata/core/rpc/ServerMessageListener.java

     * @param ctx              the ctx
     * @param message          the message
     * @param sender           the sender
     * @param checkAuthHandler the check auth handler
     */
-    void onRegRmMessage(long msgId, ChannelHandlerContext ctx, RegisterRMRequest message,
+    void onRegRmMessage(RpcMessage request, ChannelHandlerContext ctx, RegisterRMRequest message,


same as above.

slievrly · 2019-07-08T09:49:39Z

core/src/main/java/io/seata/core/rpc/ServerMessageListener.java

     * @param ctx              the ctx
     * @param message          the message
     * @param sender           the sender
     * @param checkAuthHandler the check auth handler
     */
-    void onRegTmMessage(long msgId, ChannelHandlerContext ctx, RegisterTMRequest message,
+    void onRegTmMessage(RpcMessage request, ChannelHandlerContext ctx, RegisterTMRequest message,


same as above

slievrly · 2019-07-08T09:49:54Z

core/src/main/java/io/seata/core/rpc/ServerMessageSender.java

     * @param channel the channel
     * @param msg     the msg
     */
-    void sendResponse(long msgId, Channel channel, Object msg);
+    void sendResponse(RpcMessage request, Channel channel, Object msg);


same as above

slievrly · 2019-07-08T09:51:11Z

core/src/main/java/io/seata/core/rpc/netty/AbstractRpcRemoting.java

@@ -68,10 +70,21 @@
     * The Message executor.
     */
    protected final ThreadPoolExecutor messageExecutor;
+
+    private static PositiveAtomicCounter NEXT_ID = new PositiveAtomicCounter();


slievrly · 2019-07-08T10:04:58Z

core/src/main/java/io/seata/core/protocol/ProtocolConstants.java

+    // TODO add compressor like CodecFactory
+    public static final byte COMPRESS_NONE = 0;
+
+    public static final byte DEFAULT_CODEC = CodecType.valueOf(ConfigurationFactory.getInstance().getConfig(


not default.

slievrly · 2019-07-08T10:05:53Z

core/src/main/java/io/seata/core/protocol/ProtocolConstants.java

+    public static final byte DEFAULT_CODEC = CodecType.valueOf(ConfigurationFactory.getInstance().getConfig(
+            ConfigurationKeys.SERIALIZE_FOR_RPC, CodecType.SEATA.name()).toUpperCase()).getCode();
+
+    public static final byte DEFAULT_COMPRESSOR = COMPRESS_NONE;


not default

slievrly · 2019-07-08T10:09:17Z

core/src/main/java/io/seata/core/rpc/netty/AbstractRpcRemoting.java

-        rpcMessage.setHeartbeat(false);
-        rpcMessage.setRequest(true);
+        rpcMessage.setId(getNextMessageId());
+        rpcMessage.setMessageType(ProtocolConstants.MSGTYPE_RESQUEST_ONEWAY);


slievrly · 2019-07-08T11:29:01Z

core/src/test/java/io/seata/core/protocol/RpcMessageTest.java

-    private final boolean HEART_BEAT_FIELD = true;
-    private final boolean REQUEST_FIELD = false;
-    private final long ID_FIELD = 100L;
+    private final int ID_FIELD = 100;


slievrly · 2019-07-08T13:09:16Z

core/src/main/java/io/seata/core/rpc/netty/v1/ProtocolV1Encoder.java

+                RpcMessage rpcMessage = (RpcMessage) msg;
+
+                int fullLength = 13;
+                int headLength = 0;


headLength is only the size of the headmap?

slievrly · 2019-07-08T13:19:07Z

core/src/main/java/io/seata/core/rpc/netty/v1/ProtocolV1Encoder.java

+    @Override
+    public void encode(ChannelHandlerContext ctx, Object msg, ByteBuf out) {
+        if (out == null) {
+            out = ctx.alloc().buffer();


Thread-safe, why the buffer is null?

slievrly

AT mode function verification passed(seata)

slievrly

protobuf:

leizhiyuan · 2019-07-11T04:15:53Z

It seems ok in my local machine.

slievrly

AT mode function verification passed(protobuf)

slievrly

Some other minor modifications will be in another pr.

leizhiyuan

LGTM

* New rpc protocol. * remove debug log, encode body of heartbeat. * bugfix: pass txId into TCC interceptor (apache#1278) * Fix same code as review. * Fix protobuf and add default conf.

New rpc protocol.

73c333b

zhangthen added this to the 0.7.0 milestone Jul 8, 2019

zhangthen requested review from leizhiyuan and slievrly July 8, 2019 02:49

remove debug log, encode body of heartbeat.

0207297

slievrly reviewed Jul 8, 2019

View reviewed changes

zhangthen and others added 3 commits July 8, 2019 23:13

bugfix: pass txId into TCC interceptor (apache#1278)

891dd33

Fix same code as review.

c422d73

Merge branch 'develop' into new_rpc_protocol

77340c0

slievrly reviewed Jul 10, 2019

View reviewed changes

Fix protobuf and add default conf.

8bc660c

ujjboy force-pushed the new_rpc_protocol branch from cb69247 to 8bc660c Compare July 10, 2019 12:21

slievrly reviewed Jul 11, 2019

View reviewed changes

slievrly approved these changes Jul 11, 2019

View reviewed changes

slievrly changed the title ~~[WIP] New RPC protocol.~~ New RPC protocol Jul 11, 2019

leizhiyuan approved these changes Jul 11, 2019

View reviewed changes

Merge branch 'develop' into new_rpc_protocol

85c37d1

leizhiyuan merged commit 7505dea into apache:develop Jul 11, 2019

ujjboy deleted the new_rpc_protocol branch July 11, 2019 12:01

fescar-robot mentioned this pull request Jul 12, 2019

[WeeklyReport] Weekly report for Seata 2019-7-5 to 2019-7-12 #1292

Closed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

New RPC protocol #1276

New RPC protocol #1276

ujjboy commented Jul 7, 2019 •

edited

codecov-io commented Jul 7, 2019 •

edited

slievrly Jul 8, 2019

slievrly Jul 8, 2019

slievrly Jul 8, 2019

slievrly Jul 8, 2019

slievrly Jul 8, 2019

slievrly Jul 8, 2019

slievrly Jul 8, 2019

slievrly Jul 8, 2019

slievrly Jul 8, 2019

slievrly Jul 8, 2019

slievrly Jul 8, 2019

slievrly Jul 8, 2019

slievrly Jul 8, 2019

slievrly Jul 8, 2019

slievrly Jul 8, 2019

slievrly Jul 8, 2019

slievrly Jul 8, 2019

slievrly left a comment •

edited

slievrly left a comment

leizhiyuan commented Jul 11, 2019

slievrly left a comment

slievrly left a comment

leizhiyuan left a comment

New RPC protocol #1276

New RPC protocol #1276

Conversation

ujjboy commented Jul 7, 2019 • edited

Ⅰ. Describe what this PR did

Ⅱ. Does this pull request fix one issue?

Ⅲ. Why don't you add test cases (unit test/integration test)?

Ⅳ. Describe how to verify it

Ⅴ. Special notes for reviews

codecov-io commented Jul 7, 2019 • edited

Codecov Report

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

slievrly left a comment • edited

Choose a reason for hiding this comment

slievrly left a comment

Choose a reason for hiding this comment

leizhiyuan commented Jul 11, 2019

slievrly left a comment

Choose a reason for hiding this comment

slievrly left a comment

Choose a reason for hiding this comment

leizhiyuan left a comment

Choose a reason for hiding this comment

ujjboy commented Jul 7, 2019 •

edited

codecov-io commented Jul 7, 2019 •

edited

slievrly left a comment •

edited