Fix consumer performance bottleneck

[maharifu]

The readUint8Array function calls the encoding/binary.Read function with a *[]uint8 slice. However, that function will fallback to using reflection on point to slice types, which is much slower. By using a []uint8 directly, performance increases drastically for large message sizes.

For 400 byte message, here's the results before the fix:

$ go run perftest.go --time 10 --consumer-offset next --fixed-body 400
2023/03/31 18:11:35 [info] - Silent (1.1.0) Simulation, url: [rabbitmq-stream://guest:guest@localhost:5552/%2f] publishers: 1 consumers: 1 streams: [perf-test-go]
2023/03/31 18:11:35 [info] - Declaring streams: [perf-test-go]
2023/03/31 18:11:35 [info] - stream perf-test-go, meta data: leader lc-desktop:5552, followers
2023/03/31 18:11:35 [info] - End Init streams :[perf-test-go]
2023/03/31 18:11:35 [info] - Starting 1 consumers...
2023/03/31 18:11:35 [info] - Starting consumer number: perf-test-go-0, form next
2023/03/31 18:11:35 [info] - Starting 1 publishers...
2023/03/31 18:11:35 [info] - Starting publisher number: 1
2023/03/31 18:11:37 [info] - Published 712400.0 msg/s | Confirmed 666939.0 msg/s |  Consumed 240955.0 msg/s |  Full rate  |  Fixed Body: 408  |  msg sent: 712600  |
2023/03/31 18:11:38 [info] - Published 697900.0 msg/s | Confirmed 677533.5 msg/s |  Consumed 243357.5 msg/s |  Full rate  |  Fixed Body: 408  |  msg sent: 1395900  |
2023/03/31 18:11:39 [info] - Published 708800.0 msg/s | Confirmed 691987.7 msg/s |  Consumed 246889.0 msg/s |  Full rate  |  Fixed Body: 408  |  msg sent: 2126500  |
2023/03/31 18:11:40 [info] - Published 705725.0 msg/s | Confirmed 695118.8 msg/s |  Consumed 248654.8 msg/s |  Full rate  |  Fixed Body: 408  |  msg sent: 2823000  |
2023/03/31 18:11:41 [info] - Published 704220.0 msg/s | Confirmed 696997.4 msg/s |  Consumed 249714.2 msg/s |  Full rate  |  Fixed Body: 408  |  msg sent: 3521200  |
2023/03/31 18:11:42 [info] - Published 703816.7 msg/s | Confirmed 695519.2 msg/s |  Consumed 249055.2 msg/s |  Full rate  |  Fixed Body: 408  |  msg sent: 4223000  |
2023/03/31 18:11:43 [info] - Published 699357.1 msg/s | Confirmed 693293.0 msg/s |  Consumed 248584.4 msg/s |  Full rate  |  Fixed Body: 408  |  msg sent: 4895600  |
2023/03/31 18:11:44 [info] - Published 703037.5 msg/s | Confirmed 699285.5 msg/s |  Consumed 249255.4 msg/s |  Full rate  |  Fixed Body: 408  |  msg sent: 5624400  |
2023/03/31 18:11:45 [info] - Published 704200.0 msg/s | Confirmed 699866.2 msg/s |  Consumed 248867.0 msg/s |  Full rate  |  Fixed Body: 408  |  msg sent: 6337900  |
2023/03/31 18:11:46 [info] - Stopping after 10 seconds

And here's the results after the fix:

$ go run perftest.go --time 10 --consumer-offset next --fixed-body 400
2023/03/31 18:11:49 [info] - Silent (1.1.0) Simulation, url: [rabbitmq-stream://guest:guest@localhost:5552/%2f] publishers: 1 consumers: 1 streams: [perf-test-go]
2023/03/31 18:11:49 [info] - Declaring streams: [perf-test-go]
2023/03/31 18:11:49 [info] - stream perf-test-go, meta data: leader lc-desktop:5552, followers
2023/03/31 18:11:49 [info] - End Init streams :[perf-test-go]
2023/03/31 18:11:49 [info] - Starting 1 consumers...
2023/03/31 18:11:49 [info] - Starting consumer number: perf-test-go-0, form next
2023/03/31 18:11:49 [info] - Starting 1 publishers...
2023/03/31 18:11:49 [info] - Starting publisher number: 1
2023/03/31 18:11:50 [info] - Published 744200.0 msg/s | Confirmed 701837.0 msg/s |  Consumed 685453.0 msg/s |  Full rate  |  Fixed Body: 408  |  msg sent: 744300  |
2023/03/31 18:11:51 [info] - Published 688050.0 msg/s | Confirmed 670392.0 msg/s |  Consumed 666296.0 msg/s |  Full rate  |  Fixed Body: 408  |  msg sent: 1376200  |
2023/03/31 18:11:52 [info] - Published 688000.0 msg/s | Confirmed 676292.3 msg/s |  Consumed 670831.7 msg/s |  Full rate  |  Fixed Body: 408  |  msg sent: 2064100  |
2023/03/31 18:11:53 [info] - Published 684875.0 msg/s | Confirmed 677196.0 msg/s |  Consumed 675148.0 msg/s |  Full rate  |  Fixed Body: 408  |  msg sent: 2739600  |
2023/03/31 18:11:54 [info] - Published 693180.0 msg/s | Confirmed 684290.0 msg/s |  Consumed 677737.0 msg/s |  Full rate  |  Fixed Body: 408  |  msg sent: 3466000  |
2023/03/31 18:11:55 [info] - Published 686483.3 msg/s | Confirmed 679463.7 msg/s |  Consumed 678098.3 msg/s |  Full rate  |  Fixed Body: 408  |  msg sent: 4119000  |
2023/03/31 18:11:56 [info] - Published 694957.1 msg/s | Confirmed 688889.0 msg/s |  Consumed 687718.7 msg/s |  Full rate  |  Fixed Body: 408  |  msg sent: 4864800  |
2023/03/31 18:11:57 [info] - Published 688475.0 msg/s | Confirmed 683670.0 msg/s |  Consumed 681622.2 msg/s |  Full rate  |  Fixed Body: 408  |  msg sent: 5507900  |
2023/03/31 18:11:58 [info] - Published 687977.8 msg/s | Confirmed 683251.6 msg/s |  Consumed 681431.1 msg/s |  Full rate  |  Fixed Body: 408  |  msg sent: 6191900  |
2023/03/31 18:11:59 [info] - Published 687590.0 msg/s | Confirmed 682917.0 msg/s |  Consumed 682097.8 msg/s |  Full rate  |  Fixed Body: 408  |  msg sent: 6876000  |
2023/03/31 18:11:59 [info] - Stopping after 10 seconds

Can you guys take a look a this and see if it makes sense?

[Gsantomaggio]

That's interesting. We use the pointers everywhere in binary.Read(readerStream, ... which ended up in reflections.

We will investigate. Thank you for researching and for the. PR.

cc @Zerpet

[michaelklishin]

@maharifu good catch, thanks.

[Gsantomaggio]

Thank you @maharifu