partition_map

defmodule PartitionMap do
  @moduledoc """
  PartitionMap maps arbitrary terms to a set of contiguous and dynamic (but deterministic) partitions. It's intended as a hash table for distributed systems.

  The general goal is to assign every possible term to a set of known buckets, without the O(N) size requirement of a Map.

  You can think of PartitionMap like [Consistent Hashing](https://en.wikipedia.org/wiki/Consistent_hashing) but with a couple of key advantages:
    - dynamic number of partitions
    - dynamically sized partitions
    - pluggable slicing strategies for application-specific optimizations
    - doesn't conflate partition assignment with replication policies

  There's a great article by @slfritchie on the topic: [A Critique of Resizable Hash Tables: Riak Core & Random Slicing](https://www.infoq.com/articles/dynamo-riak-random-slicing)

  PartitionMap works by hashing the given term into a fixed-size range of integer keys, the hash lands inside one of many contiguous intervals of integers (partitions).

  Each partition belongs to an "owner", each owner owns a configurable proportion of the total hash space.

  A graphical example:

    owner 1    o2           o1           o1       o2        o1             o2
      v        v            v            v        v         v              v
  <- p1 ->|<- p2 ->|<----- p3 ----->|<- p4 ->|<- p5 ->|<-- p6 -->|<------ p7 ------>|
  |-------|--------|----------------|--------|--------|----------|------------------|
  ^  ^        ^        ^       ^           ^               ^                        ^
  0  |        |        |       |           |               |                   2**32 (phash2)
     |        |        |       |           |               |
   "abcd"  <<1,2>>    "z"  %{a: 123}      500        {:ok, [1, 2, 3]}

  So, the terms `"z"` and `%{a: 123}` are both mapped to the same partition.

  PartitionMap takes heavy inspiration and uses the "CutShift" algorithm from:
    ["Random slicing: Efficient and scalable data placement for large-scale storage systems."](docs/miranda-tos14.pdf)
    Alberto Miranda, Sascha Effert, Yangwook Kang, Ethan L. Miller, Ivan Popov, Andre Brinkmann, TomFriedetzky, and Toni Cortes. 2014.
    ACM Trans. Storage 10, 3, Article 9 (July 2014), 35 pages.
    ftp://ftp.cse.ucsc.edu/pub/darrell/miranda-tos14.pdf
  """

  alias PartitionMap.Partition
  alias PartitionMap.Diff

  alias PartitionMap.Util
  import PartitionMap.Util, only: [sum_by: 2, hash_range: 0, deterministic_integer_for_colors: 1]

  defstruct [
    strategy: :unset,
    interval_map: IntervalMap.new(),
    private: :unset, # the strategy's private state
    next_id: 0
  ]

  @type key :: any
  @type partition_id :: integer
  @type partitions :: [Partition.t()]

  @type strategy :: module
  @type strategy_args :: list

  @type weight :: number
  @type owner :: any
  @type owner_names :: [owner]
  @type owners_with_weights :: %{required(owner) => weight}
  @type owners :: owner_names | owners_with_weights

  @type t :: %__MODULE__{
    strategy: strategy,
    interval_map: IntervalMap.t(),
    private: atom,
    next_id: integer
  }

  @visualization_delimeter "|"

  @doc """
    Creates a new PartitionMap with the specified strategy and strategy arguments.

    iex> PartitionMap.new(PartitionMap.SplitLargestN, n: 8,owners: [:a, :b, :c, :d])
    # => %PartitionMap{}

  """
  @spec new(strategy, strategy_args) :: t
  def new(strategy, strategy_args) when is_atom(strategy) and is_list(strategy_args) do
    {partitions, private} = apply(strategy, :new, [strategy_args])

    {partitions, next_id} = assign_new_ids(partitions, 0)

    interval_map = partitions_to_interval_map(partitions)

    %__MODULE__{interval_map: interval_map, strategy: strategy, private: private, next_id: next_id}
  end

  @doc """
  """
  @spec get(t, key) :: Partition.t()
  def get(%__MODULE__{} = partition_map, term) do
    get_with_digested_key(partition_map, digest_key(term))
  end

  @doc """
  """
  @spec get_with_digested_key(t, key) :: Partition.t()
  def get_with_digested_key(%__MODULE__{interval_map: interval_map}, term) do
    interval_map
    |> IntervalMap.get(term)
    |> Partition.from_interval
  end

  defdelegate digest_key(key), to: Util, as: :hash

  @doc """
  Adds a number of owners to the given `partition_map`, partitions are then assigned to the new owners
  using the partition_map's strategy.
  """
  @spec add_owners(t, owners) :: t
  def add_owners(%__MODULE__{strategy: strategy, private: private, next_id: next_id} = partition_map, owners) do
    partitions = to_list(partition_map)

    {partitions, private} = apply(strategy, :add_owners, [partitions, owners, private])

    {partitions, next_id} = assign_new_ids(partitions, next_id)

    interval_map = partitions_to_interval_map(partitions)

    %__MODULE__{partition_map | interval_map: interval_map, private: private, next_id: next_id}
  end

  @doc """
  Returns a list of differences between `partition_map` and `other_partition_map` as a `Diff`.
  """
  defdelegate diff(partition_map, other_partition_map), to: Diff

  @doc """
  Converts the given `partition_map` to a list.
  """
  @spec to_list(t) :: partitions
  def to_list(%__MODULE__{interval_map: interval_map}) do
    interval_map
    |> IntervalMap.to_list
    |> Enum.map(&Partition.from_interval/1)
  end

  @doc """
  Returns a list of owners in the given `partition_map`
  """
  @spec owners(t) :: owner_names
  def owners(%__MODULE__{} = partition_map) do
    partition_map
    |> to_list
    |> Enum.map(fn %Partition{owner: owner} -> owner end)
  end

  @doc """
  The size of the given partition
  """
  defdelegate size(partition), to: Partition

  # for debugging/testing
  @doc false
  @spec calculate_relative_owner_weights(t) :: owners_with_weights
  def calculate_relative_owner_weights(%__MODULE__{} = partition_map) do
    partition_map
    |> to_list
    |> Enum.group_by(fn %Partition{owner: owner} -> owner end)
    |> Enum.into(%{}, fn {owner, partitions} ->
      num_keys = sum_by(partitions, &Partition.size/1)
      weight = num_keys / hash_range()
      {owner, weight}
    end)
  end

  @doc """
  Visually inspect a PartionMap.

  The wider you make your terminal, the more accurate it'll be.
  """
  def inspect(%__MODULE__{} = partition_map, width \\ iex_columns()) do
    partitions = to_list(partition_map)

    total_pips = width - 1 # due to initial delimeter

    # owner colors are calculated in a deterministic manner to ensure that their colors stay the same
    # between invocations, so we can see a clear timeline
    owner_colors =
      partitions
      |> Enum.map(fn %Partition{owner: owner} -> owner end)
      |> Enum.uniq
      |> Enum.into(%{}, fn owner ->
        r = deterministic_integer_for_colors({owner, :red})
        g = deterministic_integer_for_colors({owner, :green})
        b = deterministic_integer_for_colors({owner, :blue})
        {owner, IO.ANSI.color(r, g, b)}
      end)

    num_keys_per_pip = hash_range() / total_pips

    # this looks convoluted, but it avoids cascading round() errors by calculating the absolute positions
    # first, then rounding. (rather than adding pre-rounded numbers in series)
    {sections, tags} =
      Enum.reduce(partitions, {[], 0}, fn partition, {keys_so_far, sum} ->
        sum = sum + Partition.size(partition)
        {[sum | keys_so_far], sum}
      end)
      |> elem(0)
      |> Enum.reverse
      |> Enum.map(fn num_keys ->
        num_keys / num_keys_per_pip |> round()
      end)
      |> Enum.reduce({[], 0}, fn absolute_pip_index, {pip_lengths, last_pip_index} ->
        {[absolute_pip_index - last_pip_index | pip_lengths], absolute_pip_index}
      end)
      |> elem(0)
      |> Enum.reverse
      |> Enum.map(&Kernel.-(&1, 1)) # subtract one from each length for its delimeter
      |> Enum.zip(partitions)
      |> Enum.map(fn {num_pips, %Partition{owner: owner, id: id}} ->
        owner_color = Map.get(owner_colors, owner)

        section = String.pad_leading("", num_pips, "-")

        tag =
          "#{id}(#{owner})"
          |> String.pad_leading(div(num_pips, 2) + 3, " ")
          |> String.pad_trailing(num_pips + 1, " ")

        {
          [owner_color, section, IO.ANSI.reset, @visualization_delimeter],
          [owner_color, tag, IO.ANSI.reset]
        }
      end)
      |> Enum.unzip

    [@visualization_delimeter, sections, "\n", tags] |> IO.puts

    partition_map
  end

  defp partitions_to_interval_map(partitions) do
    Enum.reduce(partitions, IntervalMap.new(), fn %Partition{left: left, right: right} = partition, interval_map ->
      IntervalMap.put(interval_map, {left, right}, Partition.marshal(partition))
    end)
  end

  defp assign_new_ids(partitions, next_id) do
    {without_ids, with_ids} = Enum.split_with(partitions, fn %Partition{id: id} -> is_nil(id) end)

    newly_assigned =
      without_ids
      |> Enum.with_index
      |> Enum.map(fn {partition, index} -> %Partition{partition | id: next_id + index} end)

    {newly_assigned ++ with_ids, next_id + length(newly_assigned)}
  end

  defp iex_columns do
    {:ok, columns} = :io.columns
    columns
  end
end