Restore use of SpmcArrayQueue in RxRingBuffer

[benjchristensen]

• Modification of SpmcArrayQueue with fix from Offer returns false occasionally when space should exist JCTools/JCTools#21
  • I expect updates to this code over time but wanted to move forward with a working solution for now
• Restore RxRingBuffer to use SpmcArrayQueue
  • this reduces object allocation significantly and increases performance in some use cases

Performance Diff

0.20.0-RC3

../gradlew benchmarks '-Pjmh=-f 1 -wi 5 -i 5 -r 2 .*rx.internal.RxRingBufferPerf.*'

Benchmark                                            Mode   Samples        Score  Score error    Units
r.i.RxRingBufferPerf.spmcCreateUseAndDestroy1       thrpt         5 33224158.580   399867.898    ops/s
r.i.RxRingBufferPerf.spmcCreateUseAndDestroy1000    thrpt         5    73346.934      853.989    ops/s
r.i.RxRingBufferPerf.spmcRingBufferAddRemove1       thrpt         5 22034846.226   579563.777    ops/s
r.i.RxRingBufferPerf.spmcRingBufferAddRemove1000    thrpt         5    21612.517      567.086    ops/s
r.i.RxRingBufferPerf.spscCreateUseAndDestroy1       thrpt         5 32277699.473  2481022.929    ops/s
r.i.RxRingBufferPerf.spscCreateUseAndDestroy1000    thrpt         5    72782.657     3836.229    ops/s
r.i.RxRingBufferPerf.spscRingBufferAddRemove1       thrpt         5 21625427.399   555584.413    ops/s
r.i.RxRingBufferPerf.spscRingBufferAddRemove1000    thrpt         5    21588.532      875.871    ops/s

../gradlew benchmarks '-Pjmh=-f 1 -wi 5 -i 5 -r 2 .*rx.operators.OperatorMergePerf.*'

Benchmark                                          (size)   Mode   Samples        Score  Score error    Units
r.o.OperatorMergePerf.merge1SyncStreamOfN               1  thrpt         5  5066495.356    73296.083    ops/s
r.o.OperatorMergePerf.merge1SyncStreamOfN            1000  thrpt         5    55094.237     1067.519    ops/s
r.o.OperatorMergePerf.merge1SyncStreamOfN         1000000  thrpt         5       57.981        1.324    ops/s
r.o.OperatorMergePerf.mergeNAsyncStreamsOfN             1  thrpt         5    97273.839     9582.028    ops/s
r.o.OperatorMergePerf.mergeNAsyncStreamsOfN          1000  thrpt         5        4.754        0.115    ops/s
r.o.OperatorMergePerf.mergeNSyncStreamsOf1              1  thrpt         5  4384357.616   302930.385    ops/s
r.o.OperatorMergePerf.mergeNSyncStreamsOf1            100  thrpt         5   278124.421    36626.222    ops/s
r.o.OperatorMergePerf.mergeNSyncStreamsOf1           1000  thrpt         5    35179.120     1059.293    ops/s
r.o.OperatorMergePerf.mergeNSyncStreamsOfN              1  thrpt         5  5383548.483   125915.654    ops/s
r.o.OperatorMergePerf.mergeNSyncStreamsOfN           1000  thrpt         5       54.551        3.245    ops/s
r.o.OperatorMergePerf.mergeTwoAsyncStreamsOfN           1  thrpt         5    66546.808    30474.772    ops/s
r.o.OperatorMergePerf.mergeTwoAsyncStreamsOfN        1000  thrpt         5     2828.244      103.470    ops/s
r.o.OperatorMergePerf.oneStreamOfNthatMergesIn1         1  thrpt         5  5030721.519   181734.814    ops/s
r.o.OperatorMergePerf.oneStreamOfNthatMergesIn1      1000  thrpt         5    24489.183      759.520    ops/s
r.o.OperatorMergePerf.oneStreamOfNthatMergesIn1   1000000  thrpt         5       30.101        0.802    ops/s

Memory allocation capture with Java Flight Recorder while executing:

../gradlew benchmarks '-Pjmh=-f 1 -wi 5 -i 5 -r 20 .*rx.operators.OperatorMergePerf.mergeNAsyncStreamsOfN.*'

with JCTools impl

../gradlew benchmarks '-Pjmh=-f 1 -wi 5 -i 5 -r 2 .*rx.internal.RxRingBufferPerf.*'

Benchmark                                            Mode   Samples        Score  Score error    Units
r.i.RxRingBufferPerf.spmcCreateUseAndDestroy1       thrpt         5 27028267.871  2547216.752    ops/s
r.i.RxRingBufferPerf.spmcCreateUseAndDestroy1000    thrpt         5    78238.446     2539.165    ops/s
r.i.RxRingBufferPerf.spmcRingBufferAddRemove1       thrpt         5 43396574.546   918454.062    ops/s
r.i.RxRingBufferPerf.spmcRingBufferAddRemove1000    thrpt         5    64074.506      531.873    ops/s
r.i.RxRingBufferPerf.spscCreateUseAndDestroy1       thrpt         5 27163028.506   506868.149    ops/s
r.i.RxRingBufferPerf.spscCreateUseAndDestroy1000    thrpt         5    77472.474     2875.597    ops/s
r.i.RxRingBufferPerf.spscRingBufferAddRemove1       thrpt         5 43309504.983   337590.502    ops/s
r.i.RxRingBufferPerf.spscRingBufferAddRemove1000    thrpt         5    64140.916     1292.265    ops/s

../gradlew benchmarks '-Pjmh=-f 1 -wi 5 -i 5 -r 2 .*rx.operators.OperatorMergePerf.*'

Benchmark                                          (size)   Mode   Samples        Score  Score error    Units
r.o.OperatorMergePerf.merge1SyncStreamOfN               1  thrpt         5  5146417.279   168011.480    ops/s
r.o.OperatorMergePerf.merge1SyncStreamOfN            1000  thrpt         5    56640.224      789.709    ops/s
r.o.OperatorMergePerf.merge1SyncStreamOfN         1000000  thrpt         5       60.610        1.022    ops/s
r.o.OperatorMergePerf.mergeNAsyncStreamsOfN             1  thrpt         5    97759.963     1897.592    ops/s
r.o.OperatorMergePerf.mergeNAsyncStreamsOfN          1000  thrpt         5        5.937        0.404    ops/s
r.o.OperatorMergePerf.mergeNSyncStreamsOf1              1  thrpt         5  4511685.053   136414.240    ops/s
r.o.OperatorMergePerf.mergeNSyncStreamsOf1            100  thrpt         5   285644.030     9325.626    ops/s
r.o.OperatorMergePerf.mergeNSyncStreamsOf1           1000  thrpt         5    32789.901     1143.208    ops/s
r.o.OperatorMergePerf.mergeNSyncStreamsOfN              1  thrpt         5  5415231.898   107538.428    ops/s
r.o.OperatorMergePerf.mergeNSyncStreamsOfN           1000  thrpt         5       55.918        0.660    ops/s
r.o.OperatorMergePerf.mergeTwoAsyncStreamsOfN           1  thrpt         5    74341.892     3274.847    ops/s
r.o.OperatorMergePerf.mergeTwoAsyncStreamsOfN        1000  thrpt         5     4138.087      329.935    ops/s
r.o.OperatorMergePerf.oneStreamOfNthatMergesIn1         1  thrpt         5  5026384.994   242183.595    ops/s
r.o.OperatorMergePerf.oneStreamOfNthatMergesIn1      1000  thrpt         5    24700.473      334.516    ops/s
r.o.OperatorMergePerf.oneStreamOfNthatMergesIn1   1000000  thrpt         5       30.458        0.603    ops/s

Memory allocation capture with Java Flight Recorder while executing:

../gradlew benchmarks '-Pjmh=-f 1 -wi 5 -i 5 -r 20 .*rx.operators.OperatorMergePerf.mergeNAsyncStreamsOfN.*'

[benjchristensen]

Note that in this comparison (1 minute test of mergeNAsyncStreamsOfN), over 6GB of object allocation is eliminated by changing the data structure.

[cloudbees-pull-request-builder]

RxJava-pull-requests #1446 SUCCESS
This pull request looks good