clset-example

Introduction

This repository contains a simple proof-of-concept of a distributed, multi-master and eventually consistent key-value store. Multiple peers can do local operations (insertions, removals and updates), and they can sync keys from other peers. Conflicting updates to the same key are automatically resolved.

The implementation is based on the Causal-Length Set (CLSet) CRDT, introduced in the paper by Weihai Yu and Sigbjørn Rostad.

This project draws inspiration from CR-SQLite (and its fork https://github.com/superfly/cr-sqlite/), which also implement a CLSet.

Quickstart

Start a peer

Run a peer named peer1:

go run ./cmd/srv peer1 "" 9000 8080

In the above example, peer1 listens on P2P port 9000 and exposes its HTTP API on port 8080.

To also run a peer named peer2, you can run:

go run ./cmd/srv peer1 "" 9001 8081

In the above example, peer2 listens on P2P port 9001 and exposes its HTTP API on port 8081.

Both peers should automatically discover each other via mDNS. In order to explicitly connect to peer1, you can also run

go run ./cmd/srv peer2 /ip4/127.0.0.1/tcp/9000/p2p/<peer1-id> 9001 8081

where <peer1-id> should be replaced with the Peer ID value printed by peer1 on startup.

HTTP API endpoints

Method	Endpoint	Description
PUT	/key/{key}	Create or update a key
GET	/key/{key}	Retrieve a key
DELETE	/key/{key}	Delete a key
GET	/count	Get total number of keys

Example HTTP API requests

Create or update a key

curl -X PUT -H "Content-Type: application/json" -d '{"value":"hello"}' http://localhost:8080/key/mykey

Retrieve the value of a key

curl http://localhost:8080/key/mykey

Response:

{"key":"mykey","value":"hello"}

Delete a key

curl -X DELETE http://localhost:8080/key/mykey

Get key count

curl http://localhost:8080/count

Response:

{"count": 0}

Detailed description

In this system:

• Each peer is a CRDT instance.
• Each peer must have a unique peer ID.
• Each peer has a per-peer sequence number. Every time the peer does a local operation (an insertion, update or deletion of a key-value pair), the peer's sequence number is increased by 1, and the key that was changed by the operation is tagged with this number (see below).

Each key is associated to the following metadata, all described in more detail in the subsequent paragraphs:

• Causal length: a per-key increasing number that globally tracks the latest state of each key (inserted or removed). This value is per key, i.e., each key has its own independent causal length.
• Value version: an additional counter used for conflict resolution in the case of value updates.
• Peer ID: the peer who authored the latest change to this key.
• Peer seq: the peer sequence number associated with the latest change to this key. As already mentioned above, this number is per peer, and is increaesd on every local insertion/update/removal.

The causal length and the value version are used as follows:

• When a key is first created, its causal length is 1.
• When an existing key is removed, its causal length increases by 1.
• When a removed key is reinserted, its causal length increases by 1.
• Therefore, the causal length is odd if and only if the key exists, and it's even if and only if the key was removed.
• For handling value updates, an additional counter per key is used: the value version. It is 1 whenever a key is created or reinserted, and is increased by 1 whenever the value is changed.
• When peers merge keys from other peers, conflict resolution works as follows: if two peers have the same key, the one with higher causal length wins. Otherwise, if causal lengths are equal, the one with higher value version wins. Otherwise, if both causal length and value version are equal, the one whose value is lexicographically higher wins. Otherwise, as a last resort, the lexicographically highest peer ID wins.

The sync mechanism works as follows:

• The main metadata used for syncing are the key's peer ID and peer seq. As already mentioned, whenever a peer locally creates, updates or removes a key, the key's peer ID is set to the local peer's ID, and the key's peer seq is set to the next available sequence number of the local peer.
• Each peer keeps a "tracked peers" map: a mapping from peer ID to the latest sequence number "known" from remote peers. This is used to keep track of what changes they already have from all peers and is updated whenever peers merge changes from remote peers. It also contains the local peer's latest sequence number.
• Each peer has a GetLatestChanges function that lets remote peers fetch changes that they don't already have. This function receives two arguments: requestorTrackedPeers (a copy of the tracked peers map of the requesting peer), and requestorPeerID (the peer ID of the requesting peer). Its return value is a pair of two items: changes (the list of changes that the requestor doesn't already have) and trackedPeers (a copy of the called peer's tracked peers map). More specifically, changes contains all keys from the called peer such that (1) the key's peerID is not equal to requestorPeerID and (2) either the key's peerID is not present in requestorTrackedPeers or the key's peer seq is greater than the sequence number recorded in requestorTrackedPeers.
• A peer A can merge changes from another peer B. For that, A calls B's GetLatestChanges, then A merges the changes to its local CRDT and merges the tracked peer information provided by B into its own local tracked peers map.
• When a peer A updates its tracked peer information from a remote peer B, the merge is done as follows:
  • For peers that A already has, the recorded sequence number is updated to the maximum between the peer's local value and the value gotten from B.
  • Any peer that A doesn't already know is simply added to A's tracked peers map.

Load testing

A simple load testing tool can be found in cmd/load. It uses the f1 library.

Currently, it has only one scenario. This scenario tests putting a given number of keys via the HTTP API.

In order to run this load test, you first have to run two peers listening on HTTP ports 8080 and 8081:

go run ./cmd/srv peer1 "" 9000 8080

go run ./cmd/srv peer2 "" 9001 8081

Now, you can run the test:

go run ./cmd/load run constant -r 2000 -i 10000 -d 4000s -c 4000 crdt-load

In the above example, we run 10000 key put requests (-i 10000) at a rate of 2000 requests/second (-r 2000). These requests are sent in round-robin. That is, the first request is sent to the peer on :8080, the second one is sent to the peer on :8081, the third one is sent to :8080, and so on.

The rate is constant, which is done using f1's constant command.

The -d parameter from f1 is the maximum duration of the test, and -c is the maximum concurrency. The values chosen for -d and -c in the above example are arbitrary and merely illustrative.

Run go run ./cmd/load run constant --help for help regarding the constant command, and go run ./cmd/load run --help for the list of available commands.