CRDT Performance

[martinsumner]

In the bigsets repo, one of the starting points is a single-client test to show escalating PUT times as the set grows:

In order to re-calibrate this, a riak_test is needed (like general_api_perf) to run this same test. This should at first be able to establish what the difference is with modern hardware, and with different backends.

An eprof profile can be run in parallel with the test in the Riak node, to better establish where time is being lost. Are there simple improvements which can be made?

[martinsumner]

The test - OpenRiak/riak_test#11

[martinsumner]

Running the test with openriak-3.4, OTP 24, on a fast (relative to a machine being used a decade ago) Macbook, the performance is better than previously experienced - but the same order of magnitude:

2025-08-27 16:06:10.690861+01:00 [info] 100 updates with size up to 9000 took mean_ms=35 with max_ms=37 fetch mean_ms=7
2025-08-27 16:06:15.114539+01:00 [info] 100 updates with size up to 9100 took mean_ms=36 with max_ms=38 fetch mean_ms=7
2025-08-27 16:06:19.583299+01:00 [info] 100 updates with size up to 9200 took mean_ms=36 with max_ms=38 fetch mean_ms=8
2025-08-27 16:06:24.114194+01:00 [info] 100 updates with size up to 9300 took mean_ms=37 with max_ms=39 fetch mean_ms=8
2025-08-27 16:06:28.672488+01:00 [info] 100 updates with size up to 9400 took mean_ms=37 with max_ms=42 fetch mean_ms=8
2025-08-27 16:06:33.283243+01:00 [info] 100 updates with size up to 9500 took mean_ms=37 with max_ms=40 fetch mean_ms=8
2025-08-27 16:06:37.937691+01:00 [info] 100 updates with size up to 9600 took mean_ms=38 with max_ms=42 fetch mean_ms=8
2025-08-27 16:06:42.632374+01:00 [info] 100 updates with size up to 9700 took mean_ms=38 with max_ms=42 fetch mean_ms=8
2025-08-27 16:06:47.352901+01:00 [info] 100 updates with size up to 9800 took mean_ms=39 with max_ms=40 fetch mean_ms=8
2025-08-27 16:06:52.122877+01:00 [info] 100 updates with size up to 9900 took mean_ms=39 with max_ms=40 fetch mean_ms=8

A sample eprof profile, during the test:

FUNCTION                                 CALLS        %    TIME  [uS / CALLS]
--------                                 -----  -------    ----  [----------]
gen_statem:loop_receive/3                    5     0.48     530  [    106.00]
riak_dt_orswot:'-value/1-lc$^1/1-1-'/1    8102     0.50     554  [      0.07]
lists:merge3_1/6                          9702     0.51     563  [      0.06]
riak_dt_vclock:get_counter/2              8102     0.51     568  [      0.07]
riak_dt_pb:e_field_dtvalue_set_value/3    8102     0.55     615  [      0.08]
lists:merge3_2/6                         10946     0.56     625  [      0.06]
riak_dt_pb:e_type_bytes/3                 8102     0.59     660  [      0.08]
dict:maybe_expand_aux/2                   8149     0.71     790  [      0.10]
riak_dt:from_binary/1                        4     0.72     802  [    200.50]
riak_dt_vclock:drop_dots/3               16202     0.78     871  [      0.05]
dict:find/2                               8114     0.79     878  [      0.11]
dict:store/3                              8149     0.81     899  [      0.11]
dict:put_bucket_s/3                       3210     0.90     998  [      0.31]
dict:get_slot/2                          16290     1.03    1145  [      0.07]
riak_dt_orswot:'-merge/2-fun-0-'/6        8101     1.03    1146  [      0.14]
erts_internal:trace_pattern/3                2     1.04    1158  [    579.00]
dict:fold_bucket/3                       20426     1.19    1322  [      0.06]
dict:rehash/4                            12802     1.36    1521  [      0.12]
erlang:setelement/3                      37841     1.70    1894  [      0.05]
erlang:phash/2                           27509     1.80    2011  [      0.07]
dict:on_bucket/3                          8156     2.30    2565  [      0.31]
dict:store_bkt_val/3                     49701     3.28    3651  [      0.07]
erlang:binary_to_term/1                     31     4.54    5055  [    163.06]
gen_fsm:handle_msg/8                        31     5.70    6354  [    204.97]
zlib:deflate_nif/4                         105     9.39   10468  [     99.70]
erlang:term_to_binary/2                     10    10.94   12195  [   1219.50]
gen_fsm:loop/8                              31    11.94   13310  [    429.35]
gen_server:loop/7                           39    26.39   29411  [    754.13]
--------------------------------------  ------  -------  ------  [----------]
Total:                                  422464  100.00%  111463  [      0.26]

[martinsumner]

In the initial run zlib is prominent. The leveled backend will use zlib compression by default, so changing the leveled configuration (within the test runner to the faster zstd:

{
  leveled,
    [
      {compaction_runs_perday, 48},
      {journal_objectcount, 20000},
      {compression_method, zstd}
    ]
}

... leads to the following performance in the test:

2025-08-27 16:17:20.073812+01:00 [info] 100 updates with size up to 9000 took mean_ms=26 with max_ms=35 fetch mean_ms=7
2025-08-27 16:17:23.543198+01:00 [info] 100 updates with size up to 9100 took mean_ms=26 with max_ms=32 fetch mean_ms=7
2025-08-27 16:17:27.058353+01:00 [info] 100 updates with size up to 9200 took mean_ms=27 with max_ms=32 fetch mean_ms=7
2025-08-27 16:17:30.610276+01:00 [info] 100 updates with size up to 9300 took mean_ms=27 with max_ms=30 fetch mean_ms=8
2025-08-27 16:17:34.189535+01:00 [info] 100 updates with size up to 9400 took mean_ms=27 with max_ms=30 fetch mean_ms=8
2025-08-27 16:17:37.796782+01:00 [info] 100 updates with size up to 9500 took mean_ms=27 with max_ms=29 fetch mean_ms=8
2025-08-27 16:17:41.444816+01:00 [info] 100 updates with size up to 9600 took mean_ms=28 with max_ms=29 fetch mean_ms=8
2025-08-27 16:17:45.120073+01:00 [info] 100 updates with size up to 9700 took mean_ms=28 with max_ms=29 fetch mean_ms=8
2025-08-27 16:17:48.807770+01:00 [info] 100 updates with size up to 9800 took mean_ms=28 with max_ms=30 fetch mean_ms=8
2025-08-27 16:17:52.544438+01:00 [info] 100 updates with size up to 9900 took mean_ms=29 with max_ms=30 fetch mean_ms=8

The update time at o(10K) members is now down to 29ms from 39ms.

This is not a "real" improvement, as it simply brings performance in leveled in-line with other backends that either don't compress (bitcask) or already use fast compression (eleveldb).

[martinsumner]

Leveled isn't the only place where there is, by default, compression.

With the riak_dt application, each object is serialised using term_to_binary, but with compression:

https://github.com/OpenRiak/riak_dt/blob/c2a3e1b8e3bc6f586f181558396683137ebe8c87/src/riak_dt.erl#L71-L78

If the backend is also compressing the object, this compression serves little purpose (other than a reduction in network bandwidth). Experiments with compressing example large dicts indicate multiple milliseconds can be lost both serialising and deserialising.

The compression is configurable, and so can be disabled in the test:

{
  riak_dt,
    [
      {binary_compression, false}
    ]
}

Now, when running the test performance improves again:

2025-08-27 16:25:58.211724+01:00 [info] 100 updates with size up to 9000 took mean_ms=15 with max_ms=19 fetch mean_ms=7
2025-08-27 16:26:00.583720+01:00 [info] 100 updates with size up to 9100 took mean_ms=15 with max_ms=28 fetch mean_ms=7
2025-08-27 16:26:02.933302+01:00 [info] 100 updates with size up to 9200 took mean_ms=15 with max_ms=18 fetch mean_ms=7
2025-08-27 16:26:05.326525+01:00 [info] 100 updates with size up to 9300 took mean_ms=15 with max_ms=17 fetch mean_ms=8
2025-08-27 16:26:07.763007+01:00 [info] 100 updates with size up to 9400 took mean_ms=16 with max_ms=17 fetch mean_ms=8
2025-08-27 16:26:10.297843+01:00 [info] 100 updates with size up to 9500 took mean_ms=16 with max_ms=32 fetch mean_ms=8
2025-08-27 16:26:12.775658+01:00 [info] 100 updates with size up to 9600 took mean_ms=16 with max_ms=18 fetch mean_ms=8
2025-08-27 16:26:15.327692+01:00 [info] 100 updates with size up to 9700 took mean_ms=16 with max_ms=23 fetch mean_ms=8
2025-08-27 16:26:17.883456+01:00 [info] 100 updates with size up to 9800 took mean_ms=16 with max_ms=19 fetch mean_ms=8
2025-08-27 16:26:20.455761+01:00 [info] 100 updates with size up to 9900 took mean_ms=17 with max_ms=19 fetch mean_ms=8

[martinsumner]

Looking at the profile, one can seen a set of lines associated with dict operations (note that erlang:phash/2 is a deprecated function used for backwards compatibility within the dict module) - where the number of calls is multiple orders of magnitude higher than the number of updates being performed during the profile window:

dict:maybe_expand_aux/2                   8149     0.71     790  [      0.10]
dict:find/2                               8114     0.79     878  [      0.11]
dict:store/3                              8149     0.81     899  [      0.11]
dict:put_bucket_s/3                       3210     0.90     998  [      0.31]
dict:get_slot/2                          16290     1.03    1145  [      0.07]
riak_dt_orswot:'-merge/2-fun-0-'/6        8101     1.03    1146  [      0.14]
dict:fold_bucket/3                       20426     1.19    1322  [      0.06]
dict:rehash/4                            12802     1.36    1521  [      0.12]
erlang:phash/2                           27509     1.80    2011  [      0.07]
dict:on_bucket/3                          8156     2.30    2565  [      0.31]
dict:store_bkt_val/3                     49701     3.28    3651  [      0.07]

Why are there so many calls to riak_dt_orswot:'-merge/2-fun-0-'/6 - and are the dict operations related?

The issue is the way updates are made to CRDTs. To fit into the general operation of the PUT_FSM & vnode interaction the code design expects an updated object to apply - but the PUT operation is not supplied an updated object via the API, just a set of update operations.

This is handled for all CRDTs by creating an empty CRDT of the expected type, and making this the updated object. At the coordinator the empty CRDT updates the existing CRDT (which is safe because CRDTs!), the update operations are applied - and there is now a genuine updated object for the PUT_FSM to forward.

So every CRDT operation includes a merge between an existing object and an empty CRDT, on the vnode coordinating the PUT.

In the riak_dt application though, a merge with an empty set requires a fold over the whole of the existing (now large set) and a re-write of every element to a new dict:

https://github.com/OpenRiak/riak_dt/blob/c2a3e1b8e3bc6f586f181558396683137ebe8c87/src/riak_dt_orswot.erl#L288-L342

[EDIT - this commit has en error in that it doesn't check that the RHSEntries is empty due to a typo - this is corrected in a subsequent commit - https://github.com/OpenRiak/riak_dt/commit/05b1d5725699f5e5b40536a718961deaa69d2187]

So instead, it would be better if we shortcut this process when the RHS is empty - the outcome should just be the LHS:

OpenRiak/riak_dt@6ea0397

Rerunning the test with this optimised code, yields the following results:

2025-08-27 16:44:08.939951+01:00 [info] 100 updates with size up to 9000 took mean_ms=7 with max_ms=8 fetch mean_ms=8
2025-08-27 16:44:10.477853+01:00 [info] 100 updates with size up to 9100 took mean_ms=7 with max_ms=9 fetch mean_ms=8
2025-08-27 16:44:12.048574+01:00 [info] 100 updates with size up to 9200 took mean_ms=7 with max_ms=9 fetch mean_ms=8
2025-08-27 16:44:13.617982+01:00 [info] 100 updates with size up to 9300 took mean_ms=7 with max_ms=10 fetch mean_ms=8
2025-08-27 16:44:15.200284+01:00 [info] 100 updates with size up to 9400 took mean_ms=7 with max_ms=8 fetch mean_ms=8
2025-08-27 16:44:16.812916+01:00 [info] 100 updates with size up to 9500 took mean_ms=7 with max_ms=8 fetch mean_ms=8
2025-08-27 16:44:18.447098+01:00 [info] 100 updates with size up to 9600 took mean_ms=7 with max_ms=8 fetch mean_ms=8
2025-08-27 16:44:20.127296+01:00 [info] 100 updates with size up to 9700 took mean_ms=8 with max_ms=12 fetch mean_ms=8
2025-08-27 16:44:21.794892+01:00 [info] 100 updates with size up to 9800 took mean_ms=8 with max_ms=13 fetch mean_ms=8
2025-08-27 16:44:23.461083+01:00 [info] 100 updates with size up to 9900 took mean_ms=8 with max_ms=15 fetch mean_ms=8

[martinsumner]

There may be further improvements possible. The most obvious is to rewrite riak_dt_orswot using a map not a dict. There may be a downside to this as calling term_to_binary/1 on a large map takes slightly longer than with a large dict (but the binary form produced is more compact).

However, given the performance improvements made through the later OTP versions in maps, and the availability of useful functions like merge/2, map/2 and keys/1 - the non-happy day merge (i.e. a genuine merge with a non-empty set) may be improved.

[martinsumner]

As well as sets, it is interesting to look a the performance of counters.

As counters are small updates to small objects, it is not obvious they will perform well in Riak.

A test has been written similar to the set test. This test can be run in two modes, one mode to flat-out test high volume counter updates:

%% Settings for full perf test
-define(DEFAULT_RING_SIZE, 32).
-define(COUNTER_COUNT, 20000).
-define(UPDATE_COUNT, 20000).
-define(CLIENT_COUNT, 100).
-define(LOG_EVERY, 1000).
-define(PROFILE_TEST, false).

This tests 2M counter updates spread across 100 clients updating 20K counters.

The second mode is a single client variant, to measure the speed of an individual counter update.

-define(DEFAULT_RING_SIZE, 8).
-define(COUNTER_COUNT, 10000).
-define(UPDATE_COUNT, 500000).
-define(CLIENT_COUNT, 1).
-define(LOG_EVERY, 10000).
-define(PROFILE_TEST, false).

In the single client mode, its is also practical to enable PROFILE_TEST and take eprof snapshots.

These are n_val=3 tests, but for convenience of measurement I'm testing on a single node (an M1 Macbook Pro).

The results are shown below, with the throughput (counter updates persec) achieved in the 100 client test (and the latency seen in the single client test - in microseconds). We tested on bitcask, leveled and memory backends for comparison. From the baseline, all changes are incremental (i.e. lower rows contain the changes in rows above).

Test	Bitcask	Leveled	Memory
Control - Riak 3.4 with OTP 24, no data-type compression	11,219 (453.7)	10,158 (559.6)	17,290 (349.7)
OTP 26.2.5.12	13,039 (356.6)	11,937 (444.4)	22,933 (229.5)
Set bucket to {aae_tree_exclude, true}	13.424 (340.3)	12,326 (433.7)	24,030 (223.8)
Improve efficiency of bucket property fetches	13,693 (323.2)	12,496 (420.9)	24,647 (211.2)
Use map not dict for riak object metadata	14,067 (317.1)	12,652 (416.8)	26,172 (205.2)
Refactor high-frequency functions in riak_core	14,084 (317.4)	12,717 (410.0)	26,618 (207.7)
Adding a vnode object cache - by bucket property [1]	15,878 (319.7)	14,849 (369.1)	26,725 (199.3)
Async PUT bucket property [2]	15,866 (316.4)	15,609 (292.2)	26,674 (198.9)
Flo Nightingale - Stats Collator	17,417 (302.7)	16,954 (287.9)	28,919 (191.0)

[1] - vnode object cache disabled for memory test
[2] - capability check in feature, and only leveled has capability, so only applied in leveled

All of these tests were run with tictac_aae enabled - so that this will work within a cluster that needs either intra-cluster or inter-cluster reconciliation.

Re-running the final code version, but with AAE disabled, gives the following results

Test	Bitcask	Leveled	Memory
AAE Disabled	19,925 (261.8)	17,634 (279.9)	37,201 (163.9)

[martinsumner]

Stats updated for each counter update (which don't seem unreasonable)...

Stats added at API (once per update - if and only if return_body = true):

<<"counter_actor_counts_100">>,
<<"counter_actor_counts_95">>,
<<"counter_actor_counts_99">>,
<<"counter_actor_counts_mean">>,
<<"counter_actor_counts_median">>.

Stats added at PUT_FSM (once per update):

<<"node_puts_counter">>,
<<"node_puts_counter_total">>,
<<"node_put_fsm_counter_time_100">>,
<<"node_put_fsm_counter_time_95">>,
<<"node_put_fsm_counter_time_99">>,
<<"node_put_fsm_counter_time_mean">>,
<<"node_put_fsm_counter_time_median">>.

Stats added at vnode (but only if merge required i.e conflict at non-coordinator):

<<"object_counter_merge">>,
<<"object_counter_merge_time_100">>,
<<"object_counter_merge_time_95">>,
<<"object_counter_merge_time_99">>,
<<"object_counter_merge_time_mean">>,
<<"object_counter_merge_time_median">>,
<<"object_counter_merge_total">>.

Stats added at vnode (one per PUT - at coordinator only):

<<"vnode_counter_update">>,
<<"vnode_counter_update_time_100">>,
<<"vnode_counter_update_time_95">>,
<<"vnode_counter_update_time_99">>,
<<"vnode_counter_update_time_mean">>,
<<"vnode_counter_update_time_median">>,
<<"vnode_counter_update_total">>.

Stats added at vnode (N times per PUT):

<<"vnode_put_fsm_time_100">>,
<<"vnode_put_fsm_time_95">>,
<<"vnode_put_fsm_time_99">>,
<<"vnode_put_fsm_time_mean">>,
<<"vnode_put_fsm_time_median">>,
<<"vnode_puts">>,
<<"vnode_puts_total">>.