ExHashRing for distributing the keys across the cluster members

Author: cabol

.github/workflows/ci.yml

@@ -33,7 +33,7 @@ jobs:
             otp: 25.x
             os: 'ubuntu-latest'
           - elixir: 1.14.x
-            otp: 24.x
+            otp: 25.x
             os: 'ubuntu-latest'
 
     env:

CHANGELOG.md

@@ -12,13 +12,22 @@
 - [Nebulex.Adapters.Partitioned] The adapter implements the new Nebulex v3 API.
 - [Nebulex.Adapters.Partitioned] The adapter supports the `:timeout` option from
   `Nebulex.Cache`.
+- [Nebulex.Adapters.Partitioned] The adapter uses `ExHashRing` for distributing
+  the keys across the cluster members.
+- [Nebulex.Adapters.Partitioned] he adapter supports the `:hash_ring` option to
+  configute `ExHashRing`.
 
 ### Backwards incompatible changes
 
 - [Nebulex.Adapters.Multilevel] Option `:model` is no longer supported. Please
   use the option `:inclusion_policy` instead.
 - [Nebulex.Adapters.Multilevel] The previous extended function `model/0,1` has
   been removed. Please use `inclusion_policy/0,1` instead.
+- [Nebulex.Adapters.Partitioned] The option `:keyslot` is no longer supported,
+  so the `Nebulex.Adapter.Keyslot` behaviour has been removed. The partitioned
+  adapter uses `ExHashRing` for key distribution under-the-hood.
+- [Nebulex.Adapters.Partitioned] The option `:join_timeout` is no longer
+  supported.
 - [Nebulex.Distributed.RPC] The usage of async tasks for the RPC calls has been
   removed (for OTP < 23). The new `Nebulex.Distributed.RPC` module uses `:erpc`.
 

README.md

@@ -25,7 +25,8 @@ the primary cache storage with the option `:primary_storage_adapter`.
 
 ## Installation
 
-Add `:nebulex_distributed` to your list of dependencies in `mix.exs`:
+`:nebulex_distributed` requires Erlang/OTP 25 or later. Then add
+`:nebulex_distributed` to your list of dependencies in `mix.exs`:
 
 ```elixir
 def deps do

coveralls.json

@@ -2,8 +2,8 @@
   "coverage_options": {
     "minimum_coverage": 100
   },
-
   "skip_files": [
+    "lib/nebulex/distributed.ex",
     "test/*"
   ]
-}
+}

lib/nebulex/adapter/keyslot.ex

@@ -561,7 +561,7 @@ defmodule Nebulex.Adapters.Multilevel do
       |> levels(ml_opts)
       |> Enum.reduce([node()], fn %{name: name, cache: cache}, acc ->
         if cache.__adapter__() in [Nebulex.Adapters.Partitioned, Nebulex.Adapters.Replicated] do
-          Cluster.get_nodes(name || cache) ++ acc
+          Cluster.pg_nodes(name || cache) ++ acc
         else
           acc
         end

lib/nebulex/adapters/multilevel.ex

@@ -6,7 +6,7 @@ defmodule Nebulex.Adapters.Partitioned do
 
     * Partitioned cache topology (Sharding Distribution Model).
     * Configurable primary storage adapter.
-    * Configurable Keyslot to distributed the keys across the cluster members.
+    * `ExHashRing` for distributing the keys across the cluster members.
     * Support for transactions via Erlang global name registration facility.
 
   ## Partitioned Cache Topology
@@ -102,29 +102,10 @@ defmodule Nebulex.Adapters.Partitioned do
           primary_storage_adapter: Nebulex.Adapters.Local
       end
 
-  Also, you can provide a custom keyslot function:
-
-      defmodule MyApp.PartitionedCache do
-        use Nebulex.Cache,
-          otp_app: :my_app,
-          adapter: Nebulex.Adapters.Partitioned,
-          primary_storage_adapter: Nebulex.Adapters.Local
-
-        @behaviour Nebulex.Adapter.Keyslot
-
-        @impl true
-        def hash_slot(key, range) do
-          key
-          |> :erlang.phash2()
-          |> :jchash.compute(range)
-        end
-      end
-
   Where the configuration for the cache must be in your application environment,
   usually defined in your `config/config.exs`:
 
       config :my_app, MyApp.PartitionedCache,
-        keyslot: MyApp.PartitionedCache,
         primary: [
           gc_interval: 3_600_000,
           backend: :shards
@@ -315,7 +296,9 @@ defmodule Nebulex.Adapters.Partitioned do
 
   Get a cluster node based on the given `key`:
 
-      MyCache.get_node("mykey")
+      MyCache.find_node("mykey")
+
+      MyCache.find_node!("mykey")
 
   Joining the cache to the cluster:
 
@@ -348,9 +331,6 @@ defmodule Nebulex.Adapters.Partitioned do
   # Inherit default observable implementation
   use Nebulex.Adapter.Observable
 
-  # Inherit default keyslot implementation
-  use Nebulex.Adapter.Keyslot
-
   import Nebulex.Adapter
   import Nebulex.Utils
 
@@ -388,14 +368,30 @@ defmodule Nebulex.Adapters.Partitioned do
       A convenience function for getting the cluster nodes.
       """
       def nodes(name \\ get_dynamic_cache()) do
-        Cluster.get_nodes(name)
+        name
+        |> lookup_meta()
+        |> get_in([:hash_ring, :name])
+        |> Cluster.ring_nodes()
       end
 
       @doc """
       A convenience function to get the node of the given `key`.
       """
-      def get_node(name \\ get_dynamic_cache(), key) do
-        Cluster.get_node(name, key, lookup_meta(name).keyslot)
+      def find_node(name \\ get_dynamic_cache(), key) do
+        name
+        |> lookup_meta()
+        |> get_in([:hash_ring, :name])
+        |> Cluster.find_node(key)
+      end
+
+      @doc """
+      Same as `find_node/2` but raises an error if an error occurs.
+      """
+      def find_node!(name \\ get_dynamic_cache(), key) do
+        case find_node(name, key) do
+          {:ok, node} -> node
+          {:error, _} = error -> raise error
+        end
       end
 
       @doc """
@@ -440,16 +436,19 @@ defmodule Nebulex.Adapters.Partitioned do
         do: [name: camelize_and_concat([name, Primary])] ++ primary_opts,
         else: primary_opts
 
-    # Keyslot module for selecting nodes
-    keyslot = Keyword.fetch!(opts, :keyslot)
+    # Hash ring options
+    hash_ring =
+      opts
+      |> Keyword.fetch!(:hash_ring)
+      |> Keyword.put_new_lazy(:name, fn -> camelize_and_concat([name, Ring]) end)
 
     # Prepare metadata
     adapter_meta = %{
       telemetry_prefix: telemetry_prefix,
       telemetry: telemetry,
       name: name,
       primary_name: primary_opts[:name],
-      keyslot: keyslot
+      hash_ring: hash_ring
     }
 
     # Prepare child spec
@@ -525,6 +524,9 @@ defmodule Nebulex.Adapters.Partitioned do
     end
 
     case map_reduce(entries, adapter_meta, action, [opts], timeout, {true, []}, reducer) do
+      {:error, _} = error ->
+        error
+
       {true, _} ->
         {:ok, true}
 
@@ -638,12 +640,13 @@ defmodule Nebulex.Adapters.Partitioned do
     end
   end
 
-  defp do_execute(adapter_meta, %{op: op} = query, opts) do
+  defp do_execute(%{hash_ring: hash_ring} = adapter_meta, %{op: op} = query, opts) do
+    ring = Keyword.fetch!(hash_ring, :name)
     timeout = Keyword.fetch!(opts, :timeout)
     query = build_query(query)
 
     RPC.multicall(
-      Cluster.get_nodes(adapter_meta.name),
+      Cluster.ring_nodes(ring),
       __MODULE__,
       :with_dynamic_cache,
       [adapter_meta, op, [query, opts]],
@@ -687,11 +690,13 @@ defmodule Nebulex.Adapters.Partitioned do
   ## Nebulex.Adapter.Transaction
 
   @impl true
-  def transaction(adapter_meta, fun, opts) do
+  def transaction(%{hash_ring: hash_ring} = adapter_meta, fun, opts) do
+    ring = Keyword.fetch!(hash_ring, :name)
+
     opts =
       opts
       |> Options.validate_common_runtime_opts!()
-      |> Keyword.put(:nodes, Cluster.get_nodes(adapter_meta.name))
+      |> Keyword.put(:nodes, Cluster.ring_nodes(ring))
 
     super(adapter_meta, fun, opts)
   end
@@ -722,8 +727,10 @@ defmodule Nebulex.Adapters.Partitioned do
     super(adapter_meta, :server, opts)
   end
 
-  def info(adapter_meta, :nodes, _opts) do
-    {:ok, Cluster.get_nodes(adapter_meta.name)}
+  def info(%{hash_ring: hash_ring}, :nodes, _opts) do
+    ring = Keyword.fetch!(hash_ring, :name)
+
+    {:ok, Cluster.ring_nodes(ring)}
   end
 
   def info(adapter_meta, :nodes_info, opts) do
@@ -778,11 +785,12 @@ defmodule Nebulex.Adapters.Partitioned do
     end
   end
 
-  defp fetch_nodes_info(adapter_meta, spec, opts) do
+  defp fetch_nodes_info(%{hash_ring: hash_ring} = adapter_meta, spec, opts) do
     opts = Options.validate_common_runtime_opts!(opts)
+    ring = Keyword.fetch!(hash_ring, :name)
 
     RPC.multicall(
-      Cluster.get_nodes(adapter_meta.name),
+      Cluster.ring_nodes(ring),
       __MODULE__,
       :with_dynamic_cache,
       [adapter_meta, :info, [spec, opts]],
@@ -839,16 +847,18 @@ defmodule Nebulex.Adapters.Partitioned do
     end
   end
 
-  defp get_node(%{name: name, keyslot: keyslot}, key) do
-    Cluster.get_node(name, key, keyslot)
+  defp find_node(%{hash_ring: hash_ring}, key) do
+    hash_ring
+    |> Keyword.fetch!(:name)
+    |> Cluster.find_node(key)
   end
 
   defp call(adapter_meta, key, action, args, opts) do
     timeout = Keyword.fetch!(opts, :timeout)
 
-    adapter_meta
-    |> get_node(key)
-    |> RPC.call(__MODULE__, :with_dynamic_cache, [adapter_meta, action, args], timeout)
+    with {:ok, node} <- find_node(adapter_meta, key) do
+      RPC.call(node, __MODULE__, :with_dynamic_cache, [adapter_meta, action, args], timeout)
+    end
   end
 
   defp map_reduce(enum, meta, action, args, timeout, acc, reducer, group_fun \\ & &1) do
@@ -858,14 +868,23 @@ defmodule Nebulex.Adapters.Partitioned do
       {node, {__MODULE__, :with_dynamic_cache, [meta, action, [group_fun.(group) | args]]}}
     end)
     |> RPC.multi_mfa_call(timeout, acc, reducer)
+  catch
+    error -> error
   end
 
   defp group_by_node(enum, adapter_meta, action) when action in [:put_all, :put_new_all] do
-    Enum.group_by(enum, &get_node(adapter_meta, elem(&1, 0)))
+    Enum.group_by(enum, &find_node_or_throw(adapter_meta, elem(&1, 0)))
   end
 
   defp group_by_node(enum, adapter_meta, _action) do
-    Enum.group_by(enum, &get_node(adapter_meta, &1))
+    Enum.group_by(enum, &find_node_or_throw(adapter_meta, &1))
+  end
+
+  defp find_node_or_throw(adapter_meta, key) do
+    case find_node(adapter_meta, key) do
+      {:ok, node} -> node
+      {:error, _} = error -> throw(error)
+    end
   end
 
   defp handle_rpc_multi_call({res, []}, _action, fun) do
@@ -879,9 +898,11 @@ defmodule Nebulex.Adapters.Partitioned do
       action: action
   end
 
+  ## Error formatting
+
   @doc false
-  def format_error({:rpc, {:unexpected_errors, errors}}, opts) do
-    action = Keyword.fetch!(opts, :action)
+  def format_error({:rpc, {:unexpected_errors, errors}}, metadata) do
+    action = Keyword.fetch!(metadata, :action)
 
     formatted_errors =
       Enum.map_join(errors, "\n\n", fn {{:error, reason}, node} ->