dvvset.ex

defmodule Ravel.DVVSet do
  @moduledoc """
  An observed-remove DVVSet.
  
  Doesn't use tombstones, but has 1 vclock for each entry and a vclock for the
  set. When sets are merged, the set vclocks can be used to make sure that
  deletes don't reappear.
  """
  
  defrecordp :s, __MODULE__, [clock: nil, values: nil]
  
  @type vclock :: Ravel.VClock.vclock
  @type actor :: term
  
  @opaque dvvset :: {__MODULE__, clock :: vclock, values :: Dict.t}
  
  @doc "Creates a new DVVSet"
  @spec new() :: dvvset
  def new() do
    s(clock: Ravel.VClock.new(), values: HashDict.new())
  end
  
  @doc "Get the value of a DVVSet. In this case, it's a list."
  @spec value(dvvset) :: [term()]
  def value(s(values: values)) do
    Dict.keys(values)
  end
  
  @doc "Get the size of the values"
  @spec size(dvvset) :: non_neg_integer
  def size(s(values: values)) do
    Dict.size(values)
  end
  
  @doc "Check if the set contains an element"
  @spec contains(dvvset, term) :: boolean
  def contains(s(values: values), elem) do
    values[elem] !== nil
  end
  
  @doc "Add an element to the set"
  @spec add(dvvset, actor, term) :: {:ok, dvvset}
  def add(s(clock: setclock, values: values)=dvvset, actor, elem) do
    new_clock = Ravel.VClock.increment(setclock, actor)
    new_value_clock = Ravel.VClock.new([{actor, Ravel.VClock.get_counter(setclock, actor)}])
    new_values = HashDict.update(values, elem, new_value_clock, fn(old_value_clock) ->
      Ravel.VClock.merge(new_value_clock, old_value_clock)
    end)
    {:ok, s(dvvset, clock: new_clock, values: new_values)}
  end
  
  @doc "Add an element to the set"
  @spec add!(dvvset, actor, term) :: dvvset
  def add!(dvvset, actor, elem) do
    {:ok, new_dvvset} = add(dvvset, actor, elem)
    new_dvvset
  end
  
  @doc "Remove an element from the set"
  @spec remove(dvvset, term) :: {:ok, dvvset} | {:error, {:precondition, {:not_present, term() }}}
  def remove(s(values: values)=dvvset, elem) do
    case values[elem] do
      nil -> {:error, {:precondition, {:not_present, elem}}}
      _ -> {:ok, s(dvvset, values: Dict.delete(values, elem))}
    end
  end
  
  @doc "Remove an element from the set"
  @spec remove!(dvvset, term) :: dvvset
  def remove!(dvvset, elem) do
    {:ok, new_dvvset} = remove(dvvset, elem)
    new_dvvset
  end
  
  @doc "Merge two dvvsets"
  @spec merge(dvvset, dvvset) :: dvvset
  def merge(v1, v2) when v1 == v2 do
    v1
  end
  def merge(s(clock: clock1, values: values1)=v1, s(clock: clock2, values: values2)=v2) do
    case either_dominates(clock1, clock2) do
      clock when clock==clock1 -> v1
      clock when clock==clock2 -> v2
      :concurrent ->
        newclock = Ravel.VClock.merge(clock1, clock2)
        keys1 = HashSet.new(Dict.keys(values1))
        keys2 = HashSet.new(Dict.keys(values2))
        
        common_keys = Set.intersection(keys1, keys2)
        uniquekeys1 = Set.difference(keys1, common_keys)
        uniquekeys2 = Set.difference(keys2, common_keys)
        
        newvalues = merge_common_keys(common_keys, values1, values2) |>
        merge_disjoint_keys(uniquekeys1, values1, clock2) |>
        merge_disjoint_keys(uniquekeys2, values2, clock1)
        
        s(v1, clock: newclock, values: newvalues)
    end
  end
  
  # See which clock wins when merging
  @spec either_dominates(vclock, vclock) :: vclock | :concurrent
  defp either_dominates(clock1, clock2) do
    case Ravel.VClock.descends(clock1, clock2) do
      true -> clock1
      false ->
        case Ravel.VClock.descends(clock2, clock1) do
          true -> clock2
          false -> :concurrent
        end
    end
  end
  
  # Merge common keys into a new values dictionary
  @spec merge_common_keys(Set.t, Dict.t, Dict.t) :: Dict.t
  defp merge_common_keys(common_keys, values1, values2) do
    Enum.map(common_keys, fn(k) ->
      {k, Ravel.VClock.merge(values1[k], values2[k])}
    end) |> HashDict.new
  end
  
  # Merge disjoint keys into the values dictionary
  @spec merge_disjoint_keys(Dict.t, Set.t, Dict.t, vclock) :: Dict.t
  defp merge_disjoint_keys(new_values, keys, values, setclock) do
    Enum.reduce(keys, new_values, fn(k, acc) ->
      vclock = values[k]
      case Ravel.VClock.descends(setclock, vclock) do
        false ->
          new_clock = Ravel.VClock.subtract_dots(vclock, setclock)
          Dict.put(acc, k, new_clock)
        true ->
          acc
      end
    end)
  end
  
end