optimize: the high CPU usage issue of the rpcMergeMessageSend thread

[PleaseGiveMeTheCoke]

I have registered the PR changes.
Ⅰ. Describe what this PR did

1. no longer wait every 1ms, directly wait with 10ms or greater, anyway, sendSyncRequest come in to put the message into the queue will wake up the thread
2. tm and rm threads are merged off, do not need a separate thread

Ⅱ. Does this pull request fix one issue?
#6041
Ⅲ. Why don't you add test cases (unit test/integration test)?
Ⅳ. Describe how to verify it
Ⅴ. Special notes for reviews

[CLAassistant]

All committers have signed the CLA.

[codecov]

Codecov Report

Merging #6061 (3b4faf2) into 2.x (7894473) will increase coverage by 0.04%.
The diff coverage is 36.84%.

Additional details and impacted files

@@             Coverage Diff              @@
##                2.x    #6061      +/-   ##
============================================
+ Coverage     49.55%   49.59%   +0.04%     
- Complexity     4759     4766       +7     
============================================
  Files           908      908              
  Lines         31362    31375      +13     
  Branches       3778     3782       +4     
============================================
+ Hits          15540    15561      +21     
+ Misses        14285    14279       -6     
+ Partials       1537     1535       -2     

Files	Coverage Δ
...io/seata/core/rpc/netty/AbstractNettyRemoting.java	14.28% <100.00%> (ø)
...ta/core/rpc/netty/AbstractNettyRemotingClient.java	18.51% <35.13%> (-0.21%)	⬇️

... and 10 files with indirect coverage changes

[funky-eyes]

请登记pr和作者信息至 https://github.com/seata/seata/tree/2.x/changes 中的2.x.md中
Please register the PR and author information in the 2.x.md file located at https://github.com/seata/seata/tree/2.x/changes

[funky-eyes]

Thank you for your submission! We really appreciate it. Like many open source projects, we ask that you all sign our Contributor License Agreement before we can accept your contribution.
1 out of 2 committers have signed the CLA.

[funky-eyes]

May I ask for your email address? I will send you my contact information so we can further communicate.

[PleaseGiveMeTheCoke]

@funky-eyes
Sure. I'd love to communicate with you further. [email protected]

[funky-eyes]

@funky-eyes Sure. I'd love to communicate with you further. [email protected]

I have sent my contact information to this email address, please check it out.

[slievrly]

Performance optimization PR needs sufficient data support.

[laywin]

in MergedSendRunnable synchronized block code . check queue's status first seems better, so it can be direct invoke wait. because synchronized is mutual exclusion，there would be two situation, first invoke notify, There must be data. so no wait happens. first invoke wait, then it will be awakened by notify.

  synchronized (mergeLock) {
      try {
          if (basketMap.values().stream().allMatch(Collection::isEmpty)) {
              mergeLock.wait();
          }
      } catch (InterruptedException e) {
      }
  }

[PleaseGiveMeTheCoke]

Performance optimization PR needs sufficient data support.

Ok, I'll try to find data to support this optimization, thanks for the heads up!

[PleaseGiveMeTheCoke]

in MergedSendRunnable synchronized block code . check queue's status first seems better, so it can be direct invoke wait. because synchronized is mutual exclusion，there would be two situation, first invoke notify, There must be data. so no wait happens. first invoke wait, then it will be awakened by notify.

  synchronized (mergeLock) {
      try {
          if (basketMap.values().stream().allMatch(Collection::isEmpty)) {
              mergeLock.wait();
          }
      } catch (InterruptedException e) {
      }
  }

Very great idea! I will try to modify it this way and verify the feasibility

[PleaseGiveMeTheCoke]

@funky-eyes @slievrly @laywin Please help me to see the reasonableness of the test code

Test Tools

Performance test using JMH, CPU occupancy test using arthas

Test Code

@Test
void start() throws Exception {
    Options opt = new OptionsBuilder()
            .include(MergedThreadTest.class.getSimpleName())
            .forks(1)
            .build();
    new Runner(opt).run();
}

@Benchmark
@Threads(16)
@OutputTimeUnit(TimeUnit.NANOSECONDS)
@Measurement(iterations = 5, time = 10, timeUnit = TimeUnit.SECONDS)
@BenchmarkMode(Mode.AverageTime)
public void sendRequest(Client client) throws TimeoutException {
    GlobalReportRequest globalReport = new GlobalReportRequest();
    globalReport.setXid("123");
    globalReport.setGlobalStatus(GlobalStatus.UnKnown);
    client.tmClient.sendSyncRequest(globalReport);

    GlobalLockQueryRequest request = new GlobalLockQueryRequest();
    request.setXid("123");
    request.setLockKey("lockKeys");
    request.setResourceId("resourceId");
    client.rmClient.sendSyncRequest(request);
}


@State(Scope.Benchmark)
public static class Client {

    static {
        System.setProperty("registry.type", "nacos");
        System.setProperty("registry.nacos.application", "seata-server1");
        System.setProperty("registry.nacos.serverAddr", "localhost:8848");
        System.setProperty("registry.nacos.group", "SEATA_GROUP");
        System.setProperty("registry.nacos.username", "nacos");
        System.setProperty("registry.nacos.password", "nacos");
    }

    public RmNettyRemotingClient rmClient = initRm();

    public TmNettyRemotingClient tmClient = initTM();

    private static RmNettyRemotingClient initRm(){
        RmNettyRemotingClient rmNettyRemotingClient = RmNettyRemotingClient.getInstance("testApplication",
                DefaultValues.DEFAULT_TX_GROUP);
        rmNettyRemotingClient.setTransactionMessageHandler(DefaultRMHandler.get());
        rmNettyRemotingClient.setResourceManager(DefaultResourceManager.get());
        TCCResource tccResource = new TCCResource();
        tccResource.setActionName("action");
        DefaultResourceManager.get().registerResource(tccResource);
        rmNettyRemotingClient.init();

        return rmNettyRemotingClient;
    }

    private static TmNettyRemotingClient initTM() {
        TmNettyRemotingClient tmNettyRemotingClient = TmNettyRemotingClient.getInstance("testApplication",
                DefaultValues.DEFAULT_TX_GROUP, "", "");
        tmNettyRemotingClient.init();
        return tmNettyRemotingClient;
    }
}

Test Data

Before thread merging

Performance Testing Data

8 thread
Warmup Iteration 2: 12139329.543 ±(99.9%) 848561.019 ns/op
Warmup Iteration 3: 11963078.561 ±(99.9%) 960887.166 ns/op
Warmup Iteration 4: 12981660.479 ±(99.9%) 1189936.352 ns/op
Warmup Iteration 5: 12492056.457 ±(99.9%) 883896.984 ns/op
Iteration 1: 13048335.217 ±(99.9%) 1128320.673 ns/op
Iteration 2: 13633428.156 ±(99.9%) 1217889.924 ns/op
Iteration 3: 17895714.077 ±(99.9%) 906983.698 ns/op
Iteration 4: 14793398.420 ±(99.9%) 1110027.805 ns/op
Iteration 5: 13455755.155 ±(99.9%) 658577.878 ns/op
Result "io.seata.core.rpc.netty.v1.MergedThreadTest.sendRequest":
14565326.205 ±(99.9%) 7590835.199 ns/op [Average]
(min, avg, max) = (13048335.217, 14565326.205, 17895714.077), stdev = 1971315.795
CI (99.9%): [6974491.006, 22156161.404] (assumes normal distribution)

16 thread
Warmup Iteration 2: 19813489.277 ±(99.9%) 968142.331 ns/op
Warmup Iteration 3: 20198656.297 ±(99.9%) 749479.682 ns/op
Warmup Iteration 4: 20573418.230 ±(99.9%) 923624.110 ns/op
Warmup Iteration 5: 20253840.605 ±(99.9%) 1177409.617 ns/op
Iteration 1: 20197239.104 ±(99.9%) 888597.560 ns/op
Iteration 2: 20102682.120 ±(99.9%) 985867.184 ns/op
Iteration 3: 21468587.136 ±(99.9%) 664266.590 ns/op
Iteration 4: 20359510.680 ±(99.9%) 548540.336 ns/op
Iteration 5: 19561127.870 ±(99.9%) 980520.992 ns/op
Result "io.seata.core.rpc.netty.v1.MergedThreadTest.sendRequest":
20337829.382 ±(99.9%) 2693668.053 ns/op [Average]
(min, avg, max) = (19561127.870, 20337829.382, 21468587.136), stdev = 699537.039
CI (99.9%): [17644161.329, 23031497.435] (assumes normal distribution)

CPU Occupancy Testing Data

After the thread merge

Performance Testing Data

8 thread
Warmup Iteration 2: 11032606.896 ±(99.9%) 510359.137 ns/op
Warmup Iteration 3: 11430048.542 ±(99.9%) 878519.234 ns/op
Warmup Iteration 4: 10928250.943 ±(99.9%) 862919.770 ns/op
Warmup Iteration 5: 10542404.005 ±(99.9%) 1162601.368 ns/op
Iteration 1: 10176809.317 ±(99.9%) 491029.968 ns/op
Iteration 2: 10377410.748 ±(99.9%) 704883.549 ns/op
Iteration 3: 9895530.796 ±(99.9%) 1226249.260 ns/op
Iteration 4: 11388840.466 ±(99.9%) 928430.523 ns/op
Iteration 5: 11275567.639 ±(99.9%) 929608.316 ns/op
Result "io.seata.core.rpc.netty.v1.MergedThreadTest.sendRequest":
10622831.793 ±(99.9%) 2583784.694 ns/op [Average]
(min, avg, max) = (9895530.796, 10622831.793, 11388840.466), stdev = 671000.680
CI (99.9%): [8039047.099, 13206616.487] (assumes normal distribution)

16 thread
Warmup Iteration 2: 16809090.598 ±(99.9%) 788713.429 ns/op
Warmup Iteration 3: 16926928.220 ±(99.9%) 1060596.141 ns/op
Warmup Iteration 4: 17194963.678 ±(99.9%) 1115132.284 ns/op
Warmup Iteration 5: 17232139.011 ±(99.9%) 1122411.667 ns/op
Iteration 1: 17220331.141 ±(99.9%) 678882.820 ns/op
Iteration 2: 16812748.717 ±(99.9%) 980338.756 ns/op
Iteration 3: 16119182.067 ±(99.9%) 758937.643 ns/op
Iteration 4: 15774596.784 ±(99.9%) 790845.394 ns/op
Iteration 5: 16058259.592 ±(99.9%) 972153.339 ns/op
Result "io.seata.core.rpc.netty.v1.MergedThreadTest.sendRequest":
16397023.660 ±(99.9%) 2302378.216 ns/op [Average]
(min, avg, max) = (15774596.784, 16397023.660, 17220331.141), stdev = 597920.311
CI (99.9%): [14094645.445, 18699401.876] (assumes normal distribution)

CPU Occupancy Testing Data

Test Conclusion

Thread merging has little effect on CPU utilization, but has a more significant effect on request delivery time, reducing request time by about 25%.

[funky-eyes]

Error: /home/runner/work/seata/seata/core/src/main/java/io/seata/core/rpc/netty/AbstractNettyRemotingClient.java:347:17: 'if' is not followed by whitespace. [WhitespaceAround]
Error: /home/runner/work/seata/seata/core/src/main/java/io/seata/core/rpc/netty/AbstractNettyRemotingClient.java:347:118: '{' is not preceded with whitespace. [WhitespaceAround]

[funky-eyes]

LGTM

[funky-eyes]

LGTM