[#34] Registry-based routing pool

Author: cabol

lib/nebulex_redis_adapter.ex

@@ -168,9 +168,9 @@ defmodule NebulexRedisAdapter do
           ]
         ]
 
-  By default, the adapter uses `NebulexRedisAdapter.ClientCluster.Keyslot` for the
-  keyslot. Besides, if `:jchash` is defined as dependency, the adapter will use
-  consistent-hashing automatically. However, you can also provide your own
+  By default, the adapter uses `NebulexRedisAdapter.ClientCluster.Keyslot` for
+  the keyslot. Besides, if `:jchash` is defined as dependency, the adapter will
+  use consistent-hashing automatically. However, you can also provide your own
   implementation by implementing the `Nebulex.Adapter.Keyslot` and set it into
   the `:keyslot` option. For example:
 
@@ -194,12 +194,12 @@ defmodule NebulexRedisAdapter do
 
   ### Client-side cluster options
 
-  In addition to shared options, `:client_side_cluster` mode supports the following
-  options:
+  In addition to shared options, `:client_side_cluster` mode supports the
+  following options:
 
     * `:nodes` - The list of nodes the adapter will setup the cluster with;
-      a pool of connections is established per node. The `:client_side_cluster` mode
-      enables resilience to be able to survive in case any node(s) gets
+      a pool of connections is established per node. The `:client_side_cluster`
+      mode enables resilience to be able to survive in case any node(s) gets
       unreachable. For each element of the list, we set the configuration
       for each node, such as `:conn_opts`, `:pool_size`, etc.
 
@@ -333,13 +333,19 @@ defmodule NebulexRedisAdapter do
 
   @impl true
   def init(opts) do
-    # required cache name
+    # Required cache name
     name = opts[:name] || Keyword.fetch!(opts, :cache)
 
-    # adapter mode
+    # Init stats
+    stats_counter = Stats.init(opts)
+
+    # Adapter mode
     mode = Keyword.get(opts, :mode, :standalone)
 
-    # pool size
+    # Local registry
+    registry = normalize_module_name([name, Registry])
+
+    # Resolve the pool size
     pool_size =
       get_option(
         opts,
@@ -349,60 +355,62 @@ defmodule NebulexRedisAdapter do
         System.schedulers_online()
       )
 
-    # init the specs according to the adapter mode
-    {children, default_keyslot} = do_init(mode, name, pool_size, opts)
-
-    # keyslot module for selecting nodes
+    # Keyslot module for selecting nodes
     keyslot =
-      opts
-      |> Keyword.get(:keyslot, default_keyslot)
-      |> assert_behaviour(Nebulex.Adapter.Keyslot, "keyslot")
+      if keyslot = Keyword.get(opts, :keyslot) do
+        assert_behaviour(keyslot, Nebulex.Adapter.Keyslot, "keyslot")
+      end
 
-    # cluster nodes
+    # Cluster nodes
     nodes =
       for {node_name, node_opts} <- Keyword.get(opts, :nodes, []) do
-        {node_name, Keyword.get(node_opts, :pool_size, System.schedulers_online())}
+        {node_name, Keyword.get(node_opts, :pool_size, pool_size)}
       end
 
-    # init stats
-    stats_counter = Stats.init(opts)
-
+    # Init adapter metadata
     adapter_meta = %{
+      cache_pid: self(),
       name: name,
       mode: mode,
       keyslot: keyslot,
       nodes: nodes,
       pool_size: pool_size,
       stats_counter: stats_counter,
+      registry: registry,
       started_at: DateTime.utc_now(),
       default_dt: Keyword.get(opts, :default_data_type, :object),
       telemetry: Keyword.fetch!(opts, :telemetry),
       telemetry_prefix: Keyword.fetch!(opts, :telemetry_prefix)
     }
 
+    # Init the connections child spec according to the adapter mode
+    {conn_child_spec, adapter_meta} = do_init(adapter_meta, opts)
+
+    # Build the child spec
     child_spec =
       Nebulex.Adapters.Supervisor.child_spec(
         name: normalize_module_name([name, Supervisor]),
         strategy: :one_for_all,
-        children: [{NebulexRedisAdapter.BootstrapServer, adapter_meta} | children]
+        children: [
+          {NebulexRedisAdapter.BootstrapServer, adapter_meta},
+          {Registry, name: registry, keys: :unique},
+          conn_child_spec
+        ]
       )
 
     {:ok, child_spec, adapter_meta}
   end
 
-  defp do_init(:standalone, name, pool_size, opts) do
-    {:ok, children} = Connection.init(name, pool_size, opts)
-    {children, ClientCluster.Keyslot}
+  defp do_init(%{mode: :standalone} = adapter_meta, opts) do
+    Connection.init(adapter_meta, opts)
   end
 
-  defp do_init(:client_side_cluster, _name, _pool_size, opts) do
-    {:ok, children} = ClientCluster.init(opts)
-    {children, ClientCluster.Keyslot}
+  defp do_init(%{mode: :redis_cluster} = adapter_meta, opts) do
+    RedisCluster.init(adapter_meta, opts)
   end
 
-  defp do_init(:redis_cluster, name, pool_size, opts) do
-    {:ok, children} = RedisCluster.init(name, pool_size, opts)
-    {children, RedisCluster.Keyslot}
+  defp do_init(%{mode: :client_side_cluster} = adapter_meta, opts) do
+    ClientCluster.init(adapter_meta, opts)
   end
 
   ## Nebulex.Adapter.Entry

lib/nebulex_redis_adapter/client_cluster.ex

@@ -2,28 +2,46 @@ defmodule NebulexRedisAdapter.ClientCluster do
   # Client-side Cluster
   @moduledoc false
 
-  alias NebulexRedisAdapter.ClientCluster.Supervisor, as: ClusterSupervisor
-  alias NebulexRedisAdapter.{Connection, Pool}
+  alias NebulexRedisAdapter.ClientCluster.Keyslot, as: ClientClusterKeyslot
+  alias NebulexRedisAdapter.ClientCluster.Supervisor, as: ClientClusterSupervisor
+  alias NebulexRedisAdapter.Pool
+
+  @typedoc "Proxy type to the adapter meta"
+  @type adapter_meta :: Nebulex.Adapter.metadata()
 
   @type hash_slot :: {:"$hash_slot", term}
   @type node_entry :: {node_name :: atom, pool_size :: pos_integer}
   @type nodes_config :: [node_entry]
 
-  @compile {:inline, pool_name: 2}
-
   ## API
 
-  @spec init(Keyword.t()) :: {:ok, [:supervisor.child_spec() | {module(), term()} | module()]}
-  def init(opts) do
-    cache = Keyword.fetch!(opts, :cache)
-
-    children =
+  @spec init(adapter_meta, Keyword.t()) :: {Supervisor.child_spec(), adapter_meta}
+  def init(%{name: name, registry: registry, pool_size: pool_size} = adapter_meta, opts) do
+    node_connections_specs =
       for {node_name, node_opts} <- Keyword.get(opts, :nodes, []) do
-        arg = {Connection, [name: pool_name(cache, node_name)] ++ node_opts}
-        Supervisor.child_spec({ClusterSupervisor, arg}, type: :supervisor, id: {cache, node_name})
+        node_opts =
+          node_opts
+          |> Keyword.put(:name, name)
+          |> Keyword.put(:registry, registry)
+          |> Keyword.put(:node, node_name)
+          |> Keyword.put_new(:pool_size, pool_size)
+
+        Supervisor.child_spec({ClientClusterSupervisor, node_opts},
+          type: :supervisor,
+          id: {name, node_name}
+        )
       end
 
-    {:ok, children}
+    node_connections_supervisor_spec = %{
+      id: :node_connections_supervisor,
+      type: :supervisor,
+      start: {Supervisor, :start_link, [node_connections_specs, [strategy: :one_for_one]]}
+    }
+
+    adapter_meta =
+      Map.update(adapter_meta, :keyslot, ClientClusterKeyslot, &(&1 || ClientClusterKeyslot))
+
+    {node_connections_supervisor_spec, adapter_meta}
   end
 
   @spec exec!(
@@ -33,37 +51,31 @@ defmodule NebulexRedisAdapter.ClientCluster do
           reducer :: (any, any -> any)
         ) :: any | no_return
   def exec!(
-        %{name: name, nodes: nodes},
+        %{name: name, registry: registry, nodes: nodes},
         command,
         init_acc \\ nil,
         reducer \\ fn res, _ -> res end
       ) do
-    # TODO: Perhaps this should be performed in parallel
     Enum.reduce(nodes, init_acc, fn {node_name, pool_size}, acc ->
-      name
-      |> pool_name(node_name)
-      |> Pool.get_conn(pool_size)
+      registry
+      |> Pool.get_conn({name, node_name}, pool_size)
       |> Redix.command!(command)
       |> reducer.(acc)
     end)
   end
 
-  @spec get_conn(Nebulex.Cache.t(), nodes_config, atom) :: atom
-  def get_conn(cache, nodes, node_name) do
+  @spec get_conn(atom, atom, nodes_config, atom) :: pid
+  def get_conn(registry, name, nodes, node_name) do
     pool_size = Keyword.fetch!(nodes, node_name)
 
-    cache
-    |> pool_name(node_name)
-    |> Pool.get_conn(pool_size)
+    Pool.get_conn(registry, {name, node_name}, pool_size)
   end
 
-  @spec get_conn(Nebulex.Cache.t(), nodes_config, term, module) :: atom
-  def get_conn(cache, nodes, key, module) do
+  @spec get_conn(atom, atom, nodes_config, term, module) :: pid
+  def get_conn(registry, name, nodes, key, module) do
     {node_name, pool_size} = get_node(module, nodes, key)
 
-    cache
-    |> pool_name(node_name)
-    |> Pool.get_conn(pool_size)
+    Pool.get_conn(registry, {name, node_name}, pool_size)
   end
 
   @spec group_keys_by_hash_slot(Enum.t(), nodes_config, module) :: map
@@ -79,9 +91,6 @@ defmodule NebulexRedisAdapter.ClientCluster do
     end)
   end
 
-  @spec pool_name(Nebulex.Cache.t(), atom) :: atom
-  def pool_name(cache, node_name), do: :"#{cache}.#{node_name}"
-
   ## Private Functions
 
   defp get_node(module, nodes, key) do

lib/nebulex_redis_adapter/client_cluster/supervisor.ex

@@ -7,17 +7,24 @@ defmodule NebulexRedisAdapter.ClientCluster.Supervisor do
   ## API
 
   @doc false
-  def start_link({module, opts}) do
-    name = Keyword.fetch!(opts, :name)
-    Supervisor.start_link(__MODULE__, {module, opts}, name: name)
+  def start_link(opts) do
+    Supervisor.start_link(__MODULE__, opts)
   end
 
   ## Supervisor Callbacks
 
   @impl true
-  def init({module, opts}) do
-    module
-    |> Pool.children(opts)
-    |> Supervisor.init(strategy: :one_for_one)
+  def init(opts) do
+    name = Keyword.fetch!(opts, :name)
+    registry = Keyword.fetch!(opts, :registry)
+    node = Keyword.fetch!(opts, :node)
+    pool_size = Keyword.fetch!(opts, :pool_size)
+
+    children =
+      Pool.register_names(registry, {name, node}, pool_size, fn conn_name ->
+        {NebulexRedisAdapter.Connection, Keyword.put(opts, :name, conn_name)}
+      end)
+
+    Supervisor.init(children, strategy: :one_for_one)
   end
 end

lib/nebulex_redis_adapter/command.ex

@@ -31,20 +31,32 @@ defmodule NebulexRedisAdapter.Command do
 
   ## Private Functions
 
-  defp conn(%{mode: :standalone, name: name, pool_size: pool_size}, _key) do
-    Pool.get_conn(name, pool_size)
+  defp conn(%{mode: :standalone, name: name, registry: registry, pool_size: pool_size}, _key) do
+    Pool.get_conn(registry, name, pool_size)
   end
 
-  defp conn(%{mode: :client_side_cluster, name: name, nodes: nodes}, {:"$hash_slot", node_name}) do
-    ClientCluster.get_conn(name, nodes, node_name)
+  defp conn(%{mode: :redis_cluster} = meta, key) do
+    RedisCluster.get_conn(meta, key)
   end
 
-  defp conn(%{mode: :client_side_cluster, name: name, nodes: nodes, keyslot: keyslot}, key) do
-    ClientCluster.get_conn(name, nodes, key, keyslot)
+  defp conn(
+         %{mode: :client_side_cluster, name: name, registry: registry, nodes: nodes},
+         {:"$hash_slot", node_name}
+       ) do
+    ClientCluster.get_conn(registry, name, nodes, node_name)
   end
 
-  defp conn(%{mode: :redis_cluster} = meta, key) do
-    RedisCluster.get_conn(meta, key)
+  defp conn(
+         %{
+           mode: :client_side_cluster,
+           name: name,
+           registry: registry,
+           nodes: nodes,
+           keyslot: keyslot
+         },
+         key
+       ) do
+    ClientCluster.get_conn(registry, name, nodes, key, keyslot)
   end
 
   defp handle_command_response({:ok, response}) do

lib/nebulex_redis_adapter/connection.ex

@@ -1,16 +1,29 @@
 defmodule NebulexRedisAdapter.Connection do
   @moduledoc false
 
+  alias NebulexRedisAdapter.Pool
+
+  @typedoc "Proxy type to the adapter meta"
+  @type adapter_meta :: Nebulex.Adapter.metadata()
+
   ## API
 
-  @spec init(atom, pos_integer, Keyword.t()) :: {:ok, [Supervisor.child_spec()]}
-  def init(name, pool_size, opts) do
-    children =
-      for i <- 0..(pool_size - 1) do
-        child_spec([name: :"#{name}.#{i}"] ++ opts)
-      end
+  @spec init(adapter_meta, Keyword.t()) :: {Supervisor.child_spec(), adapter_meta}
+  def init(%{name: name, registry: registry, pool_size: pool_size} = adapter_meta, opts) do
+    connections_specs =
+      Pool.register_names(registry, name, pool_size, fn conn_name ->
+        opts
+        |> Keyword.put(:name, conn_name)
+        |> child_spec()
+      end)
+
+    connections_supervisor_spec = %{
+      id: :connections_supervisor,
+      type: :supervisor,
+      start: {Supervisor, :start_link, [connections_specs, [strategy: :one_for_one]]}
+    }
 
-    {:ok, children}
+    {connections_supervisor_spec, adapter_meta}
   end
 
   @spec child_spec(Keyword.t()) :: Supervisor.child_spec()

lib/nebulex_redis_adapter/pool.ex

@@ -3,20 +3,21 @@ defmodule NebulexRedisAdapter.Pool do
 
   ## API
 
-  @spec children(module, Keyword.t()) :: [Supervisor.child_spec()]
-  def children(module, opts) do
-    name = Keyword.fetch!(opts, :name)
-    pool_size = opts[:pool_size] || System.schedulers_online()
-
+  @spec register_names(atom, term, pos_integer, ({:via, module, term} -> term)) :: [term]
+  def register_names(registry, key, pool_size, fun) do
     for index <- 0..(pool_size - 1) do
-      opts = Keyword.put(opts, :name, :"#{name}.#{index}")
-      {module, opts}
+      fun.({:via, Registry, {registry, {key, index}}})
     end
   end
 
-  @spec get_conn(atom, pos_integer) :: atom
-  def get_conn(name, pool_size) do
-    # ensure to select the same connection based on the caller PID
-    :"#{name}.#{:erlang.phash2(self(), pool_size)}"
+  @spec get_conn(atom, term, pos_integer) :: pid
+  def get_conn(registry, key, pool_size) do
+    # Ensure selecting the same connection based on the caller PID
+    index = :erlang.phash2(self(), pool_size)
+
+    registry
+    |> Registry.lookup({key, index})
+    |> hd()
+    |> elem(0)
   end
 end