Use Disruptor to implement our lightweight queue model

[CalvinKirs]

• Why do you submit this issue?

• Question or discussion

• Feature or performance improvement

Requirement or improvement

Hi, guys, in #6744, the @wu-sheng teacher suggested that we can use Disruptor to implement our lightweight queue model, the project is widely used and the performance looks great.

Everyone is very welcome to discuss
If everyone agrees on ， I would like to finish this work. ( I will provide JMH-Test after the implementation)

disruptor
GitHub: https://github.com/LMAX-Exchange/disruptor

License: Apache-2.0

[wu-sheng]

@CalvinKirs With months passed, any update? If not, it is better to close this issue, or we at least should convert this to discussion.

[CalvinKirs]

@CalvinKirs With months passed, any update? If not, it is better to close this issue, or we at least should convert this to discussion.

I'm sorry I've been so busy lately, I'll close it for now and I'll sync here when I've made progress.

[CalvinKirs]

I did a simple test comparison and it looks like Disruptor does serve our purpose, I will provide the full JMH Test shortly.

[wu-sheng]

I did a simple test comparison and it looks like Disruptor does serve our purpose, I will provide the full JMH Test shortly.

Disruptor has better performance for sure. We have tested years ago. The challenge is thread-model. We are not using one thread per logic queue, because we created one logic queue per metrics, which could be changed and keep increasing.

[CalvinKirs]

I did a simple test comparison and it looks like Disruptor does serve our purpose, I will provide the full JMH Test shortly.

Disruptor has better performance for sure. We have tested years ago. The challenge is thread-model. We are not using one thread per logic queue, because we created one logic queue per metrics, which could be changed and keep increasing.

Yes, this is a clever design by Channel,

For Disruptor, at present, I can only think of providing a more flexible configuration to achieve an ideal effect.

[CalvinKirs]

testSingleThreadDataCarrier:
Single-threaded message delivery
2 threads consumption
Data volume=100*dataSize
channelSize=32 bufferSize=4

testSingleThreadMiniDataCarrier:
Single-threaded message delivery
2 threads consumption
Data volume=100*dataSize
channelSize=4 bufferSize=32

testSingleThreadDisruptor
Single-threaded message delivery
Single producer
2 threads consumption
Data volume=100*dataSize
ringBufferSize=128

testMultiThreadDataCarrier
2 threads to deliver messages
2 threads consumption
Data volume=100*2*dataSize
channelSize=32 bufferSize=4
testMultiThreadMiniChannelDataCarrier

2 threads to deliver messages
2 threads consumption
Data volume=100*2*dataSize
channelSize=4 bufferSize=32

testMultiThreadDisruptor
Multi-producer mode (only one thread is used)
2 threads to deliver messages
2 threads consumption
Data volume=100*2*dataSize
ringBufferSize=128

Benchmark                                                                                             (dataSize)   Mode  Cnt         Score   Error   Units
DisruptorDataCarrierBenchmark.testMultiThreadDataCarrier                                                       1  thrpt    2        21.587           ops/s
DisruptorDataCarrierBenchmark.testMultiThreadDataCarrier:·gc.alloc.rate                                        1  thrpt    2         0.026          MB/sec
DisruptorDataCarrierBenchmark.testMultiThreadDataCarrier:·gc.alloc.rate.norm                                   1  thrpt    2      1331.032            B/op
DisruptorDataCarrierBenchmark.testMultiThreadDataCarrier:·gc.churn.PS_Eden_Space                               1  thrpt    2        27.020          MB/sec
DisruptorDataCarrierBenchmark.testMultiThreadDataCarrier:·gc.churn.PS_Eden_Space.norm                          1  thrpt    2   1378235.838            B/op
DisruptorDataCarrierBenchmark.testMultiThreadDataCarrier:·gc.churn.PS_Survivor_Space                           1  thrpt    2         0.162          MB/sec
DisruptorDataCarrierBenchmark.testMultiThreadDataCarrier:·gc.churn.PS_Survivor_Space.norm                      1  thrpt    2      8275.731            B/op
DisruptorDataCarrierBenchmark.testMultiThreadDataCarrier:·gc.count                                             1  thrpt    2         4.000          counts
DisruptorDataCarrierBenchmark.testMultiThreadDataCarrier:·gc.time                                              1  thrpt    2         9.000              ms
DisruptorDataCarrierBenchmark.testMultiThreadDataCarrier                                                      10  thrpt    2         2.616           ops/s
DisruptorDataCarrierBenchmark.testMultiThreadDataCarrier:·gc.alloc.rate                                       10  thrpt    2         0.003          MB/sec
DisruptorDataCarrierBenchmark.testMultiThreadDataCarrier:·gc.alloc.rate.norm                                  10  thrpt    2      1363.219            B/op
DisruptorDataCarrierBenchmark.testMultiThreadDataCarrier:·gc.churn.PS_Eden_Space                              10  thrpt    2         6.106          MB/sec
DisruptorDataCarrierBenchmark.testMultiThreadDataCarrier:·gc.churn.PS_Eden_Space.norm                         10  thrpt    2   2591192.615            B/op
DisruptorDataCarrierBenchmark.testMultiThreadDataCarrier:·gc.churn.PS_Survivor_Space                          10  thrpt    2         0.170          MB/sec
DisruptorDataCarrierBenchmark.testMultiThreadDataCarrier:·gc.churn.PS_Survivor_Space.norm                     10  thrpt    2     72070.769            B/op
DisruptorDataCarrierBenchmark.testMultiThreadDataCarrier:·gc.count                                            10  thrpt    2         1.000          counts
DisruptorDataCarrierBenchmark.testMultiThreadDataCarrier:·gc.time                                             10  thrpt    2         3.000              ms
DisruptorDataCarrierBenchmark.testMultiThreadDataCarrier                                                     100  thrpt    2         0.273           ops/s
DisruptorDataCarrierBenchmark.testMultiThreadDataCarrier:·gc.alloc.rate                                      100  thrpt    2        ≈ 10⁻³          MB/sec
DisruptorDataCarrierBenchmark.testMultiThreadDataCarrier:·gc.alloc.rate.norm                                 100  thrpt    2      1632.000            B/op
DisruptorDataCarrierBenchmark.testMultiThreadDataCarrier:·gc.count                                           100  thrpt    2           ≈ 0          counts
DisruptorDataCarrierBenchmark.testMultiThreadDataCarrier                                                    1000  thrpt    2         0.027           ops/s
DisruptorDataCarrierBenchmark.testMultiThreadDataCarrier:·gc.alloc.rate                                     1000  thrpt    2        ≈ 10⁻⁴          MB/sec
DisruptorDataCarrierBenchmark.testMultiThreadDataCarrier:·gc.alloc.rate.norm                                1000  thrpt    2      4300.000            B/op
DisruptorDataCarrierBenchmark.testMultiThreadDataCarrier:·gc.count                                          1000  thrpt    2           ≈ 0          counts
DisruptorDataCarrierBenchmark.testMultiThreadDisruptor                                                         1  thrpt    2      8933.128           ops/s
DisruptorDataCarrierBenchmark.testMultiThreadDisruptor:·gc.alloc.rate                                          1  thrpt    2        11.479          MB/sec
DisruptorDataCarrierBenchmark.testMultiThreadDisruptor:·gc.alloc.rate.norm                                     1  thrpt    2      1416.457            B/op
DisruptorDataCarrierBenchmark.testMultiThreadDisruptor:·gc.churn.PS_Eden_Space                                 1  thrpt    2     14422.861          MB/sec
DisruptorDataCarrierBenchmark.testMultiThreadDisruptor:·gc.churn.PS_Eden_Space.norm                            1  thrpt    2   1778182.066            B/op
DisruptorDataCarrierBenchmark.testMultiThreadDisruptor:·gc.churn.PS_Survivor_Space                             1  thrpt    2         1.161          MB/sec
DisruptorDataCarrierBenchmark.testMultiThreadDisruptor:·gc.churn.PS_Survivor_Space.norm                        1  thrpt    2       143.073            B/op
DisruptorDataCarrierBenchmark.testMultiThreadDisruptor:·gc.count                                               1  thrpt    2      1788.000          counts
DisruptorDataCarrierBenchmark.testMultiThreadDisruptor:·gc.time                                                1  thrpt    2       604.000              ms
DisruptorDataCarrierBenchmark.testMultiThreadDisruptor                                                        10  thrpt    2      2444.081           ops/s
DisruptorDataCarrierBenchmark.testMultiThreadDisruptor:·gc.alloc.rate                                         10  thrpt    2         4.281          MB/sec
DisruptorDataCarrierBenchmark.testMultiThreadDisruptor:·gc.alloc.rate.norm                                    10  thrpt    2      1928.139            B/op
DisruptorDataCarrierBenchmark.testMultiThreadDisruptor:·gc.churn.PS_Eden_Space                                10  thrpt    2      3154.202          MB/sec
DisruptorDataCarrierBenchmark.testMultiThreadDisruptor:·gc.churn.PS_Eden_Space.norm                           10  thrpt    2   1421635.320            B/op
DisruptorDataCarrierBenchmark.testMultiThreadDisruptor:·gc.churn.PS_Survivor_Space                            10  thrpt    2         0.256          MB/sec
DisruptorDataCarrierBenchmark.testMultiThreadDisruptor:·gc.churn.PS_Survivor_Space.norm                       10  thrpt    2       115.265            B/op
DisruptorDataCarrierBenchmark.testMultiThreadDisruptor:·gc.count                                              10  thrpt    2       391.000          counts
DisruptorDataCarrierBenchmark.testMultiThreadDisruptor:·gc.time                                               10  thrpt    2       171.000              ms
DisruptorDataCarrierBenchmark.testMultiThreadDisruptor                                                       100  thrpt    2       320.904           ops/s
DisruptorDataCarrierBenchmark.testMultiThreadDisruptor:·gc.alloc.rate                                        100  thrpt    2         2.078          MB/sec
DisruptorDataCarrierBenchmark.testMultiThreadDisruptor:·gc.alloc.rate.norm                                   100  thrpt    2      7115.277            B/op
DisruptorDataCarrierBenchmark.testMultiThreadDisruptor:·gc.churn.PS_Eden_Space                               100  thrpt    2       378.837          MB/sec
DisruptorDataCarrierBenchmark.testMultiThreadDisruptor:·gc.churn.PS_Eden_Space.norm                          100  thrpt    2   1300370.489            B/op
DisruptorDataCarrierBenchmark.testMultiThreadDisruptor:·gc.churn.PS_Survivor_Space                           100  thrpt    2         0.024          MB/sec
DisruptorDataCarrierBenchmark.testMultiThreadDisruptor:·gc.churn.PS_Survivor_Space.norm                      100  thrpt    2        81.640            B/op
DisruptorDataCarrierBenchmark.testMultiThreadDisruptor:·gc.count                                             100  thrpt    2        47.000          counts
DisruptorDataCarrierBenchmark.testMultiThreadDisruptor:·gc.time                                              100  thrpt    2        21.000              ms
DisruptorDataCarrierBenchmark.testMultiThreadDisruptor                                                      1000  thrpt    2        32.394           ops/s
DisruptorDataCarrierBenchmark.testMultiThreadDisruptor:·gc.alloc.rate                                       1000  thrpt    2         1.741          MB/sec
DisruptorDataCarrierBenchmark.testMultiThreadDisruptor:·gc.alloc.rate.norm                                  1000  thrpt    2     59387.583            B/op
DisruptorDataCarrierBenchmark.testMultiThreadDisruptor:·gc.churn.PS_Eden_Space                              1000  thrpt    2       387.073          MB/sec
DisruptorDataCarrierBenchmark.testMultiThreadDisruptor:·gc.churn.PS_Eden_Space.norm                         1000  thrpt    2  13159086.907            B/op
DisruptorDataCarrierBenchmark.testMultiThreadDisruptor:·gc.churn.PS_Survivor_Space                          1000  thrpt    2         0.031          MB/sec
DisruptorDataCarrierBenchmark.testMultiThreadDisruptor:·gc.churn.PS_Survivor_Space.norm                     1000  thrpt    2      1061.780            B/op
DisruptorDataCarrierBenchmark.testMultiThreadDisruptor:·gc.count                                            1000  thrpt    2        48.000          counts
DisruptorDataCarrierBenchmark.testMultiThreadDisruptor:·gc.time                                             1000  thrpt    2        23.000              ms
DisruptorDataCarrierBenchmark.testMultiThreadMiniChannelDataCarrier                                            1  thrpt    2        19.972           ops/s
DisruptorDataCarrierBenchmark.testMultiThreadMiniChannelDataCarrier:·gc.alloc.rate                             1  thrpt    2         0.024          MB/sec
DisruptorDataCarrierBenchmark.testMultiThreadMiniChannelDataCarrier:·gc.alloc.rate.norm                        1  thrpt    2      1331.280            B/op
DisruptorDataCarrierBenchmark.testMultiThreadMiniChannelDataCarrier:·gc.churn.PS_Eden_Space                    1  thrpt    2        25.725          MB/sec
DisruptorDataCarrierBenchmark.testMultiThreadMiniChannelDataCarrier:·gc.churn.PS_Eden_Space.norm               1  thrpt    2   1419509.760            B/op
DisruptorDataCarrierBenchmark.testMultiThreadMiniChannelDataCarrier:·gc.churn.PS_Survivor_Space                1  thrpt    2         0.162          MB/sec
DisruptorDataCarrierBenchmark.testMultiThreadMiniChannelDataCarrier:·gc.churn.PS_Survivor_Space.norm           1  thrpt    2      8937.580            B/op
DisruptorDataCarrierBenchmark.testMultiThreadMiniChannelDataCarrier:·gc.count                                  1  thrpt    2         4.000          counts
DisruptorDataCarrierBenchmark.testMultiThreadMiniChannelDataCarrier:·gc.time                                   1  thrpt    2        11.000              ms
DisruptorDataCarrierBenchmark.testMultiThreadMiniChannelDataCarrier                                           10  thrpt    2         2.499           ops/s
DisruptorDataCarrierBenchmark.testMultiThreadMiniChannelDataCarrier:·gc.alloc.rate                            10  thrpt    2         0.003          MB/sec
DisruptorDataCarrierBenchmark.testMultiThreadMiniChannelDataCarrier:·gc.alloc.rate.norm                       10  thrpt    2      1358.763            B/op
DisruptorDataCarrierBenchmark.testMultiThreadMiniChannelDataCarrier:·gc.churn.PS_Eden_Space                   10  thrpt    2         5.905          MB/sec
DisruptorDataCarrierBenchmark.testMultiThreadMiniChannelDataCarrier:·gc.churn.PS_Eden_Space.norm              10  thrpt    2   2591192.615            B/op
DisruptorDataCarrierBenchmark.testMultiThreadMiniChannelDataCarrier:·gc.churn.PS_Survivor_Space               10  thrpt    2         0.162          MB/sec
DisruptorDataCarrierBenchmark.testMultiThreadMiniChannelDataCarrier:·gc.churn.PS_Survivor_Space.norm          10  thrpt    2     70975.385            B/op
DisruptorDataCarrierBenchmark.testMultiThreadMiniChannelDataCarrier:·gc.count                                 10  thrpt    2         1.000          counts
DisruptorDataCarrierBenchmark.testMultiThreadMiniChannelDataCarrier:·gc.time                                  10  thrpt    2         3.000              ms
DisruptorDataCarrierBenchmark.testMultiThreadMiniChannelDataCarrier                                          100  thrpt    2         0.256           ops/s
DisruptorDataCarrierBenchmark.testMultiThreadMiniChannelDataCarrier:·gc.alloc.rate                           100  thrpt    2        ≈ 10⁻³          MB/sec
DisruptorDataCarrierBenchmark.testMultiThreadMiniChannelDataCarrier:·gc.alloc.rate.norm                      100  thrpt    2      1557.333            B/op
DisruptorDataCarrierBenchmark.testMultiThreadMiniChannelDataCarrier:·gc.count                                100  thrpt    2           ≈ 0          counts
DisruptorDataCarrierBenchmark.testMultiThreadMiniChannelDataCarrier                                         1000  thrpt    2         0.028           ops/s
DisruptorDataCarrierBenchmark.testMultiThreadMiniChannelDataCarrier:·gc.alloc.rate                          1000  thrpt    2        ≈ 10⁻⁴          MB/sec
DisruptorDataCarrierBenchmark.testMultiThreadMiniChannelDataCarrier:·gc.alloc.rate.norm                     1000  thrpt    2      2092.000            B/op
DisruptorDataCarrierBenchmark.testMultiThreadMiniChannelDataCarrier:·gc.count                               1000  thrpt    2           ≈ 0          counts
DisruptorDataCarrierBenchmark.testSingleThreadDataCarrier                                                      1  thrpt    2        43.171           ops/s
DisruptorDataCarrierBenchmark.testSingleThreadDataCarrier:·gc.alloc.rate                                       1  thrpt    2         0.220          MB/sec
DisruptorDataCarrierBenchmark.testSingleThreadDataCarrier:·gc.alloc.rate.norm                                  1  thrpt    2      5601.517            B/op
DisruptorDataCarrierBenchmark.testSingleThreadDataCarrier:·gc.count                                            1  thrpt    2           ≈ 0          counts
DisruptorDataCarrierBenchmark.testSingleThreadDataCarrier                                                     10  thrpt    2         5.243           ops/s
DisruptorDataCarrierBenchmark.testSingleThreadDataCarrier:·gc.alloc.rate                                      10  thrpt    2         0.268          MB/sec
DisruptorDataCarrierBenchmark.testSingleThreadDataCarrier:·gc.alloc.rate.norm                                 10  thrpt    2     56349.585            B/op
DisruptorDataCarrierBenchmark.testSingleThreadDataCarrier:·gc.count                                           10  thrpt    2           ≈ 0          counts
DisruptorDataCarrierBenchmark.testSingleThreadDataCarrier                                                    100  thrpt    2         0.530           ops/s
DisruptorDataCarrierBenchmark.testSingleThreadDataCarrier:·gc.alloc.rate                                     100  thrpt    2         0.307          MB/sec
DisruptorDataCarrierBenchmark.testSingleThreadDataCarrier:·gc.alloc.rate.norm                                100  thrpt    2    635533.333            B/op
DisruptorDataCarrierBenchmark.testSingleThreadDataCarrier:·gc.count                                          100  thrpt    2           ≈ 0          counts
DisruptorDataCarrierBenchmark.testSingleThreadDataCarrier                                                   1000  thrpt    2         0.053           ops/s
DisruptorDataCarrierBenchmark.testSingleThreadDataCarrier:·gc.alloc.rate                                    1000  thrpt    2         0.316          MB/sec
DisruptorDataCarrierBenchmark.testSingleThreadDataCarrier:·gc.alloc.rate.norm                               1000  thrpt    2   6426032.000            B/op
DisruptorDataCarrierBenchmark.testSingleThreadDataCarrier:·gc.count                                         1000  thrpt    2           ≈ 0          counts
DisruptorDataCarrierBenchmark.testSingleThreadDisruptor                                                        1  thrpt    2     23499.217           ops/s
DisruptorDataCarrierBenchmark.testSingleThreadDisruptor:·gc.alloc.rate                                         1  thrpt    2       120.315          MB/sec
DisruptorDataCarrierBenchmark.testSingleThreadDisruptor:·gc.alloc.rate.norm                                    1  thrpt    2      5640.267            B/op
DisruptorDataCarrierBenchmark.testSingleThreadDisruptor:·gc.churn.PS_Eden_Space                                1  thrpt    2       122.342          MB/sec
DisruptorDataCarrierBenchmark.testSingleThreadDisruptor:·gc.churn.PS_Eden_Space.norm                           1  thrpt    2      5735.384            B/op
DisruptorDataCarrierBenchmark.testSingleThreadDisruptor:·gc.churn.PS_Survivor_Space                            1  thrpt    2         0.015          MB/sec
DisruptorDataCarrierBenchmark.testSingleThreadDisruptor:·gc.churn.PS_Survivor_Space.norm                       1  thrpt    2         0.694            B/op
DisruptorDataCarrierBenchmark.testSingleThreadDisruptor:·gc.count                                              1  thrpt    2        16.000          counts
DisruptorDataCarrierBenchmark.testSingleThreadDisruptor:·gc.time                                               1  thrpt    2        11.000              ms
DisruptorDataCarrierBenchmark.testSingleThreadDisruptor                                                       10  thrpt    2      5626.929           ops/s
DisruptorDataCarrierBenchmark.testSingleThreadDisruptor:·gc.alloc.rate                                        10  thrpt    2       287.682          MB/sec
DisruptorDataCarrierBenchmark.testSingleThreadDisruptor:·gc.alloc.rate.norm                                   10  thrpt    2     56325.927            B/op
DisruptorDataCarrierBenchmark.testSingleThreadDisruptor:·gc.churn.PS_Eden_Space                               10  thrpt    2       289.150          MB/sec
DisruptorDataCarrierBenchmark.testSingleThreadDisruptor:·gc.churn.PS_Eden_Space.norm                          10  thrpt    2     56627.705            B/op
DisruptorDataCarrierBenchmark.testSingleThreadDisruptor:·gc.churn.PS_Survivor_Space                           10  thrpt    2         0.012          MB/sec
DisruptorDataCarrierBenchmark.testSingleThreadDisruptor:·gc.churn.PS_Survivor_Space.norm                      10  thrpt    2         2.343            B/op
DisruptorDataCarrierBenchmark.testSingleThreadDisruptor:·gc.count                                             10  thrpt    2        36.000          counts
DisruptorDataCarrierBenchmark.testSingleThreadDisruptor:·gc.time                                              10  thrpt    2        16.000              ms
DisruptorDataCarrierBenchmark.testSingleThreadDisruptor                                                      100  thrpt    2       750.957           ops/s
DisruptorDataCarrierBenchmark.testSingleThreadDisruptor:·gc.alloc.rate                                       100  thrpt    2       383.967          MB/sec
DisruptorDataCarrierBenchmark.testSingleThreadDisruptor:·gc.alloc.rate.norm                                  100  thrpt    2    563272.376            B/op
DisruptorDataCarrierBenchmark.testSingleThreadDisruptor:·gc.churn.PS_Eden_Space                              100  thrpt    2       386.852          MB/sec
DisruptorDataCarrierBenchmark.testSingleThreadDisruptor:·gc.churn.PS_Eden_Space.norm                         100  thrpt    2    567547.589            B/op
DisruptorDataCarrierBenchmark.testSingleThreadDisruptor:·gc.churn.PS_Survivor_Space                          100  thrpt    2         0.019          MB/sec
DisruptorDataCarrierBenchmark.testSingleThreadDisruptor:·gc.churn.PS_Survivor_Space.norm                     100  thrpt    2        28.311            B/op
DisruptorDataCarrierBenchmark.testSingleThreadDisruptor:·gc.count                                            100  thrpt    2        48.000          counts
DisruptorDataCarrierBenchmark.testSingleThreadDisruptor:·gc.time                                             100  thrpt    2        22.000              ms
DisruptorDataCarrierBenchmark.testSingleThreadDisruptor                                                     1000  thrpt    2        82.218           ops/s
DisruptorDataCarrierBenchmark.testSingleThreadDisruptor:·gc.alloc.rate                                      1000  thrpt    2       473.891          MB/sec
DisruptorDataCarrierBenchmark.testSingleThreadDisruptor:·gc.alloc.rate.norm                                 1000  thrpt    2   6348470.312            B/op
DisruptorDataCarrierBenchmark.testSingleThreadDisruptor:·gc.churn.PS_Eden_Space                             1000  thrpt    2       475.561          MB/sec
DisruptorDataCarrierBenchmark.testSingleThreadDisruptor:·gc.churn.PS_Eden_Space.norm                        1000  thrpt    2   6371246.361            B/op
DisruptorDataCarrierBenchmark.testSingleThreadDisruptor:·gc.churn.PS_Survivor_Space                         1000  thrpt    2         0.040          MB/sec
DisruptorDataCarrierBenchmark.testSingleThreadDisruptor:·gc.churn.PS_Survivor_Space.norm                    1000  thrpt    2       537.732            B/op
DisruptorDataCarrierBenchmark.testSingleThreadDisruptor:·gc.count                                           1000  thrpt    2        59.000          counts
DisruptorDataCarrierBenchmark.testSingleThreadDisruptor:·gc.time                                            1000  thrpt    2        27.000              ms
DisruptorDataCarrierBenchmark.testSingleThreadMiniDataCarrier                                                  1  thrpt    2        43.311           ops/s
DisruptorDataCarrierBenchmark.testSingleThreadMiniDataCarrier:·gc.alloc.rate                                   1  thrpt    2         0.220          MB/sec
DisruptorDataCarrierBenchmark.testSingleThreadMiniDataCarrier:·gc.alloc.rate.norm                              1  thrpt    2      5601.513            B/op
DisruptorDataCarrierBenchmark.testSingleThreadMiniDataCarrier:·gc.count                                        1  thrpt    2           ≈ 0          counts
DisruptorDataCarrierBenchmark.testSingleThreadMiniDataCarrier                                                 10  thrpt    2         5.194           ops/s
DisruptorDataCarrierBenchmark.testSingleThreadMiniDataCarrier:·gc.alloc.rate                                  10  thrpt    2         0.266          MB/sec
DisruptorDataCarrierBenchmark.testSingleThreadMiniDataCarrier:·gc.alloc.rate.norm                             10  thrpt    2     56316.615            B/op
DisruptorDataCarrierBenchmark.testSingleThreadMiniDataCarrier:·gc.count                                       10  thrpt    2           ≈ 0          counts
DisruptorDataCarrierBenchmark.testSingleThreadMiniDataCarrier                                                100  thrpt    2         0.527           ops/s
DisruptorDataCarrierBenchmark.testSingleThreadMiniDataCarrier:·gc.alloc.rate                                 100  thrpt    2         0.306          MB/sec
DisruptorDataCarrierBenchmark.testSingleThreadMiniDataCarrier:·gc.alloc.rate.norm                            100  thrpt    2    635616.000            B/op
DisruptorDataCarrierBenchmark.testSingleThreadMiniDataCarrier:·gc.count                                      100  thrpt    2           ≈ 0          counts
DisruptorDataCarrierBenchmark.testSingleThreadMiniDataCarrier                                               1000  thrpt    2         0.054           ops/s
DisruptorDataCarrierBenchmark.testSingleThreadMiniDataCarrier:·gc.alloc.rate                                1000  thrpt    2         0.325          MB/sec
DisruptorDataCarrierBenchmark.testSingleThreadMiniDataCarrier:·gc.alloc.rate.norm                           1000  thrpt    2   6426080.000            B/op
DisruptorDataCarrierBenchmark.testSingleThreadMiniDataCarrier:·gc.count                                     1000  thrpt    2           ≈ 0          counts

[wu-sheng]

Surprising to see this update. Could you be clear about which cases are you using?

[CalvinKirs]

Sorry for the late reply.
I have some questions, our current consumption is to pull the list data in the queue, and the Disruptor is to process the data one by one.
For some operations designed to DB, do we need to take some measures to do batch processing and consumption?

[wu-sheng]

Yes, batch-mode is important. But I remember disruptor has end-of-batch concept to cover this case.

Also, you should notice, in the latest OAP side DataCarrier(right now, OAP and Agent DataCarrier are different) has a concept of IConsumer#nothingToConsume event, which I think missed in Disruptor. It is important to build batch(not to database) as it is based on both duration and size.

[CalvinKirs]

Surprising to see this update. Could you be clear about which cases are you using?

Thanks for your reply, am I missing something?

 public static AtomicInteger oldQueueConsumerIndex = new AtomicInteger(0);
    public static AtomicInteger disruptorConsumerIndex = new AtomicInteger(0);
    DataCarrier<String> dataCarrier = new DataCarrier<>(32, 4);
    DataCarrier<String> miniChannelDataCarrier = new DataCarrier<>(4, 32);
    DataCarrierDisruptor disruptor = new DataCarrierDisruptor("Message", TestWorkHandler.class, ProducerType.SINGLE);
    DataCarrierDisruptor multiProducerDisruptor = new DataCarrierDisruptor("Message", TestWorkHandler.class, ProducerType.MULTI);


    @Param(value = {"1", "10", "100", "1000"})
    private int dataSize;


    @Setup
    public void setUp() {
        dataCarrier.consume(TestOldConsumer.class, 2);
        miniChannelDataCarrier.consume(TestOldConsumer.class, 2);
    }

    @Benchmark
    public void testSingleThreadDataCarrier() {
        for (int i = 0; i < 100 * dataSize; i++) {
            dataCarrier.produce("Message:" + i);
        }
        while (oldQueueConsumerIndex.get() < 100 * dataSize) {
            Thread.yield();
        }
        oldQueueConsumerIndex.set(0);
    }

public class TestOldConsumer implements IConsumer<String> {


    @Override
    public void init(Properties properties) {

    }

    @Override
    public void consume(List<String> data) {
        if (CollectionUtils.isEmpty(data)) {
            return;
        }
        data.forEach(d -> {
            DisruptorDataCarrierBenchmark.oldQueueConsumerIndex.getAndIncrement();
        });
    }
}

[wu-sheng]

Could you draw a graph to visualize how do you implement DataCarrier multiple consumers through Disruptor?

[CalvinKirs]

I use this method to set up multiple consumers
com.lmax.disruptor.dsl.Disruptor#handleEventsWithWorkerPool

[wu-sheng]

Let's move to discussion panel and continue.

In the case, you should set the benchmark in this way. If you have more than 8 core CPU

• 6 threads for producing messages
• 2 threads for consuming messages
• Size of queue is 20k
• The total size of the testing should target on 2m(at least should run in 3-5 min)

The measure way of this is checking

• The CPU costing during the time
• The time of processing the whole data set

Then we have size-of-processing / CPU(producing + consuming) metric, which should be better to determine whether Disruptor is better and how much more it is.

Please provide a github repo or gist to host the benchmark codes

[wu-sheng]

There is no chance for don't provide. You are providing a DataCarrier alternative implementation, it should be implemented one way or another.
Again, DataCarrier is not a simple queue, it is an important async mechanism/core for the whole SkyWalking OAP process.

This should not be an argument about what features is a queue provided, instead, this is a SkyWalking mechanism replacement.

[CalvinKirs]

Thanks for your reply, I'll try to do as you suggested first.：）

[CalvinKirs]

If we do this: there is a register, all Disruptor startups will register with it (how often to notify), and then there is a thread inside it that will deliver "NOTIFY" messages to Disruptor according to the time, and then we trigger noThingToConsumer based on this message type. What do you think?

[wu-sheng]

Disruptor is used inside the DataCarrier(OAP side) I think. We don't need to worry about this register mechanism.

[CalvinKirs]

I will submit the design later, thank you for your advice:)

[lujiajing1126]

I had experience with Disruptor several years ago, but I found it may not be suitable for a relatively small instances, e.g. 4 cores or less. See the discussion thread, LMAX-Exchange/disruptor#204 (comment)

I am not sure whether they have a big progress in the recent years for the WaitStrategy. Probably we may also provide benchmark results with limited resources?

PS. the current lightweight queue model is awesome!

[wu-sheng]

Typically, we should not provide a thing so-called default strategy. We need to pick the best suitable one and use it.

I think I mentioned last month, in this context, we are not just replacing a queue. This should be a reasonable update for end users out of the box. We only need to take an action when the PoC tests show positive results. The DataCarriers work well, we are still at the evaluating phase.

[CalvinKirs]

Thanks for the reply, I'll keep an eye on this, the default strategy provided should be what works best in most cases. POC testing I may not have a server, I can only do it locally, is this ok?

[wu-sheng]

Yes, it should be fine.

[lujiajing1126]

Hi, deeply thanks for your attention, I read this issue, but what I want to say is that Disruptor itself provides a variety of strategies, users can choose according to their own needs, this may not be regarded as a problem, this is a kind of tradeoff. Back to our current problem, we only need to ensure that the default strategy has more obvious performance advantages under the same resources.

Exactly. As mentioned by Wu-Sheng above, I suppose we have to provide a default strategy for the majority of SW users (with the consideration of limited resources, e.g. ~4 cores) and prove that it performs better than the current lw queue model.

Besides, if we can give a full performance insight, e.g. throughput/latency/CPU usage vs. the number of cores for various WaitStrategy, we can then probably automatically optimize the choice based on our investigation.

[CalvinKirs]

Thanks to both of you for your suggestions, I will pay special attention to these points. Thank you again.

[wu-sheng]

@CalvinKirs Please be advised, I have finished an enhancement to OAP side DataCarrier through #8664.

Please evaluate whether this potential change is still reasonable. With this kind of SkyWalking-native features, which do not exist in any other queue, I assume you would be hard to follow on this path.

[CalvinKirs]

Thanks for the reminder, I looked at your design and as it stands, it seems like it's hard to achieve this with a disruptor. 😂

[wu-sheng]

Should we call an end to this discussion? As more and more enhancements are added, it seems more reasonable to keep on the original design.

[CalvinKirs]

Sorry, I'll close it, thanks for your guidance.

[CalvinKirs]

I can't close it, please help to mark it.

[wu-sheng]

I labeled won't fix