Playing with CRDTs

crdt_counter.py

class Counter(object):
    """Increment-only counter"""
    def __init__(self, peer_id):
        self._peer_id = peer_id
        self._counters = {peer_id: 0}
    
    def increment(self):
        self._counters[self._peer_id] += 1

    def value(self):
        return sum(self._counters.values())

    def merge(self, peer_counter):
        # peer_counter can be immutable - its read, never written
        keys = set(self._counters.keys()).union(peer_counter._counters.keys())
        counters = {}
        for key in keys:
            counters[key] = max(self._counters[key], peer_counter._counters[key])
        self._counters = counters

# Usage
a = Counter(1) # peer 1, in real life these are probably on different machines/threads
b = Counter(2) # peer 2

a.increment()
a.increment()
a.value() # => 2

b.value() # => 0
b.increment()
b.value() # => 1

# At some point a will be propagated to b, we use the merge operation to make it consistent 
b.merge(a) 
b.value() # => 3, our counter now reflects the correct consistent state, no increments were lost

crdt_counter2.py

class PNCounter(object):
  """Counter with increment and decrement"""
  def __init__(self, peer_id):
    self._increments = Counter(peer_id)
    self._decrements = Counter(peer_id)
  
  def increment(self):
    self._increments.increment()
  
  def decrement(self):
    self._decrements.increment()
  
  def value(self):
    return self._incrments.value() - self._decrments.value()
  
  def merge(self, peer_counter):
    self._increments.merge(peer_counter._incrments)
    self._decrments.merge(peer_counter._decrements)

crdt_set.py

class Set(object):
  """An add-only set"""
  def __init__(self, peer_id):
    self._set = set()
  
  def add(self, value):
    self._set.add(value)
  
  def value(self):
    return self._set
  
  def merge(self, peer_set):
    # set union already has CRDT merge properties, so we can just use it as is
    self._set = self._set.union(peer_set._set)

crdt_set2.py

class Set2P(object):
  """A set with add and remove.
  A value can only be added once, if it is removed it cannot be added again"""
  def __init__(self, peer_id):
    self._added = Set(peer_id)
    self._removed = Set(peer_id)
  
  def add(self, value):
    self._added.add(value)
  
  def remove(self, value):
    self.removed.add(value)
  
  def value(self):
    return self._added.value() - self._removed.value()
  
  def merge(self, peer_set):
    self._added.merge(peer_set._added)
    self._removed.merge(peer_set._removed)