tcrain
commented
8 years ago For information I did some tests on antidote to see the impact of the materialiser, all experiments are done using counter CRDT types, 2000 keys, single operation transactions either performing a read or an update:
This first benchmark is single DC, eight antidote machines, and two workload generation machines each running 80 client threads. Average latencies for reads are around 1 - 2 ms, updates around 3 - 4 ms
This second benchmark is the same as the previous, except without the materialiser. By "without the materialiser" I mean that no list of versioned snapshots are created and no lists of updates are calculated and kept in memory. So it is more like a single version database. Everything else is the same, all updates are still logged, all messages sent, locks, etc... What we see here is just a small increase in performance, so it seems like the materialiser is doing a good job, and since it is single DC, likely most the time we are just dealing with the most recent snapshot (i.e. we are not really using the multi-versioning anyway).
This graph show what happens when we increase the number of DCs to 3. So each update is replicated 3 times, there is a total of 24 antidote machines (8 per DC) and 6 workload generation machines (2 per DC). What we see here is a huge performance drop as soon as we increase the update rate. Average read latencies go up to around 5 - 10 ms, updates between 10 - 100. In the 99% update workload we see around 5000 ops/sec.
This is the same 3 DC test as above except without the materialiser. Again we are still doing the same amount of logging, network traffic, locks, etc.. We are just avoiding the multi versioning and materialisation. Maybe the first thing to notice is that Antidote scales more as expected, we get 3x increased read throughput vs the single DC and around 2x write throughput (since each update is replicated 3x we are actually performing 6x more work, but most of it asynchronous). Average latencies for reads are around 1 - 2 ms, updates around 3 - 4 ms.
So then the question is, is why does the materialiser work well for 1 DC and so poorly for 3 DCs? Well it appears to be the tradeoff of multi versioning and staleness. In the 1 DC case, we are pretty much always operating on the latest version of the object, as our snapshot is based on our local physical clock. With multiple DCs we are accessing older versions based on the stable time which can vary from node to node and are using a vector clock to find and calculate the version (i.e. there could be a large number of versions). This still shouldn't account for a 20x difference in performance, so I plan to look deeper into the issue (how many snapshots we are creating, do we go to the log etc..), but I see two possible issues, first we risk over-engenieering trying to choose the proper cache size for the number of DCs/update rate, and second that there isn't really much support in Erlang for these kind of shared data structures that these multi-version objects really want to be.
marc-shapiro
bieniusa
cmeiklejohn
aletomsic
Current Implementation:
The are two per-key caches: an operation cache where the most recent operations are stored, and a snapshot cache which stores the latest materialised views (snapshots) of a CRDT. When the materialiser receives an operation, it stores in its operation cache. When it receives a read request, it checks whether there are suitable materialized views available in the snapshot cache and whether there are operations to be applied in the operation cache. If so, it materializes a new version of the CRDT that satisfies the request by applying the potentially missing operations to the corresponding snapshot, thus creating a new snapshot. In the event that cached snapshots or operations exist for a requested key, but none satisfying the read request parameters (i.e. snapshot time), the materialiser replies with an {error, no_snapshot} message.
GC As there are object views (snapshots) and operations, the growing number of any of them needs to trigger the GC mechanism.
Please contribute to the following with your ideas:
Future Caching The idea is to generate some policies to define how each key will be cached, depending on its type or maybe its size. It might be even possible to change policy dynamically, e.g., according to an element’s size. The goal is to achieve a more efficient handling of the cache, in both terms of size and speed on handling requests. There should be an implementation of the materialiser read and update functions for each policy.
NoOps: for an object cached respecting this policy, there will be no operations stored, only snapshots. This policy could be particularly handy for CRDTs which state is small compared to the size of an operation: e.g. counters, where the cost in terms of size of storing the value (a snapshot) is similar to storing the operation. In this case, for each update, a snapshot should be created. On read, the snapshots would be ready to serve requests. This would improve both size and read time for counter/like CRDTs. DeltaCheckpoints: in the snapshot cache, there will be full snapshots and deltas (from those snapshots). There have to be parameters like how many full snapshots and how many “delta snapshots” are stored. This could be useful for big sets. Ops&Snaps: the current implementation.
GC Also, the idea here is to implement GC policies. The following are possible ideas: NumberLimit: the current implementation. If the number of snapshots or operations in the cache for a given key grows to a certain threshold, perform GC. CacheSize: when the overall cache reaches a memory limit, trigger GC. TimeBased: perform GC every certain period of time. ActiveSnapshot: all partitions should periodically exchange the latest snapshot time an active tx took and exchange it. That way, a partition can GC all state which is older than the ActiveSnapshot. In order to implement this, we should have an efficient broadcast mechanism. Could this possibly be piggybacked with group membership information in riak_core?