Add key-based partitioning to duplicate registries (#14654)
This commit is contained in:
Rafał Studnicki
committed by
José Valim
José Valim
parent
9fc3708980
commit
e29c40a7e9
+119
-35
@@ -187,7 +187,7 @@ defmodule Registry do
|
||||
Note that the registry uses one ETS table plus two ETS tables per partition.
|
||||
"""
|
||||
|
||||
@keys [:unique, :duplicate]
|
||||
@keys [:unique, :duplicate, {:duplicate, :key}, {:duplicate, :pid}]
|
||||
@all_info -1
|
||||
@key_info -2
|
||||
|
||||
@@ -195,7 +195,7 @@ defmodule Registry do
|
||||
@type registry :: atom
|
||||
|
||||
@typedoc "The type of the registry"
|
||||
@type keys :: :unique | :duplicate
|
||||
@type keys :: :unique | :duplicate | {:duplicate, :key} | {:duplicate, :pid}
|
||||
|
||||
@typedoc "The type of keys allowed on registration"
|
||||
@type key :: term
|
||||
@@ -266,8 +266,8 @@ defmodule Registry do
|
||||
:undefined
|
||||
end
|
||||
|
||||
{kind, _, _} ->
|
||||
raise ArgumentError, ":via is not supported for #{kind} registries"
|
||||
{{:duplicate, _}, _, _} ->
|
||||
raise ArgumentError, ":via is not supported for duplicate registries"
|
||||
end
|
||||
end
|
||||
|
||||
@@ -329,11 +329,24 @@ defmodule Registry do
|
||||
{Registry, keys: :unique, name: MyApp.Registry, partitions: System.schedulers_online()}
|
||||
], strategy: :one_for_one)
|
||||
|
||||
For `:duplicate` registries with many different keys (e.g., many topics with
|
||||
few subscribers each), you can optimize key-based lookups by partitioning by key:
|
||||
|
||||
Registry.start_link(
|
||||
keys: {:duplicate, :key},
|
||||
name: MyApp.TopicRegistry,
|
||||
partitions: System.schedulers_online()
|
||||
)
|
||||
|
||||
This allows key-based lookups to check only a single partition instead of
|
||||
searching all partitions. Use the default `:pid` partitioning when you have
|
||||
fewer keys with many entries each (e.g., one topic with many subscribers).
|
||||
|
||||
## Options
|
||||
|
||||
The registry requires the following keys:
|
||||
|
||||
* `:keys` - chooses if keys are `:unique` or `:duplicate`
|
||||
* `:keys` - chooses if keys are `:unique`, `:duplicate`, `{:duplicate, :key}`, or `{:duplicate, :pid}`
|
||||
* `:name` - the name of the registry and its tables
|
||||
|
||||
The following keys are optional:
|
||||
@@ -345,16 +358,40 @@ defmodule Registry do
|
||||
crashes. Messages sent to listeners are of type `t:listener_message/0`.
|
||||
* `:meta` - a keyword list of metadata to be attached to the registry.
|
||||
|
||||
For `:duplicate` registries, you can specify the partitioning strategy
|
||||
directly in the `:keys` option:
|
||||
|
||||
* `:duplicate` or `{:duplicate, :pid}` - Use `:pid` partitioning (default)
|
||||
when you have keys with many entries (e.g., one topic with many subscribers).
|
||||
This is the traditional behavior and groups all entries from the same process together.
|
||||
|
||||
* `{:duplicate, :key}` - Use `:key` partitioning when entries are spread across
|
||||
many different keys (e.g., many topics with few subscribers each). This makes
|
||||
key-based lookups more efficient as they only need to check a single partition
|
||||
instead of all partitions.
|
||||
|
||||
"""
|
||||
@doc since: "1.5.0"
|
||||
@spec start_link([start_option]) :: {:ok, pid} | {:error, term}
|
||||
def start_link(options) do
|
||||
keys = Keyword.get(options, :keys)
|
||||
|
||||
if keys not in @keys do
|
||||
raise ArgumentError,
|
||||
"expected :keys to be given and be one of :unique or :duplicate, got: #{inspect(keys)}"
|
||||
end
|
||||
# Validate and normalize keys format
|
||||
kind =
|
||||
case keys do
|
||||
{:duplicate, partition_strategy} when partition_strategy in [:key, :pid] ->
|
||||
{:duplicate, partition_strategy}
|
||||
|
||||
:unique ->
|
||||
:unique
|
||||
|
||||
:duplicate ->
|
||||
{:duplicate, :pid}
|
||||
|
||||
_ ->
|
||||
raise ArgumentError,
|
||||
"expected :keys to be given and be one of :unique, :duplicate, {:duplicate, :key}, or {:duplicate, :pid}, got: #{inspect(keys)}"
|
||||
end
|
||||
|
||||
name =
|
||||
case Keyword.fetch(options, :name) do
|
||||
@@ -397,11 +434,18 @@ defmodule Registry do
|
||||
|
||||
# The @info format must be kept in sync with Registry.Partition optimization.
|
||||
entries = [
|
||||
{@all_info, {keys, partitions, nil, nil, listeners}},
|
||||
{@key_info, {keys, partitions, nil}} | meta
|
||||
{@all_info, {kind, partitions, nil, nil, listeners}},
|
||||
{@key_info, {kind, partitions, nil}} | meta
|
||||
]
|
||||
|
||||
Registry.Supervisor.start_link(keys, name, partitions, listeners, entries, compressed)
|
||||
Registry.Supervisor.start_link(
|
||||
kind,
|
||||
name,
|
||||
partitions,
|
||||
listeners,
|
||||
entries,
|
||||
compressed
|
||||
)
|
||||
end
|
||||
|
||||
@doc false
|
||||
@@ -468,7 +512,8 @@ defmodule Registry do
|
||||
end
|
||||
|
||||
{kind, _, _} ->
|
||||
raise ArgumentError, "Registry.update_value/3 is not supported for #{kind} registries"
|
||||
raise ArgumentError,
|
||||
"Registry.update_value/3 is not supported for #{inspect(kind)} registries"
|
||||
end
|
||||
end
|
||||
|
||||
@@ -508,12 +553,12 @@ defmodule Registry do
|
||||
|> List.wrap()
|
||||
|> apply_non_empty_to_mfa_or_fun(mfa_or_fun)
|
||||
|
||||
{:duplicate, 1, key_ets} ->
|
||||
{{:duplicate, _}, 1, key_ets} ->
|
||||
key_ets
|
||||
|> safe_lookup_second(key)
|
||||
|> apply_non_empty_to_mfa_or_fun(mfa_or_fun)
|
||||
|
||||
{:duplicate, partitions, _} ->
|
||||
{{:duplicate, _}, partitions, _} ->
|
||||
if Keyword.get(opts, :parallel, false) do
|
||||
registry
|
||||
|> dispatch_parallel(key, mfa_or_fun, partitions)
|
||||
@@ -625,10 +670,14 @@ defmodule Registry do
|
||||
[]
|
||||
end
|
||||
|
||||
{:duplicate, 1, key_ets} ->
|
||||
{{:duplicate, _}, 1, key_ets} ->
|
||||
safe_lookup_second(key_ets, key)
|
||||
|
||||
{:duplicate, partitions, _key_ets} ->
|
||||
{{:duplicate, :key}, partitions, _key_ets} ->
|
||||
partition = hash(key, partitions)
|
||||
safe_lookup_second(key_ets!(registry, partition), key)
|
||||
|
||||
{{:duplicate, :pid}, partitions, _key_ets} ->
|
||||
for partition <- 0..(partitions - 1),
|
||||
pair <- safe_lookup_second(key_ets!(registry, partition), key),
|
||||
do: pair
|
||||
@@ -749,10 +798,10 @@ defmodule Registry do
|
||||
key_ets = key_ets || key_ets!(registry, key, partitions)
|
||||
:ets.select(key_ets, spec)
|
||||
|
||||
{:duplicate, 1, key_ets} ->
|
||||
{{:duplicate, _}, 1, key_ets} ->
|
||||
:ets.select(key_ets, spec)
|
||||
|
||||
{:duplicate, partitions, _key_ets} ->
|
||||
{{:duplicate, _}, partitions, _key_ets} ->
|
||||
for partition <- 0..(partitions - 1),
|
||||
pair <- :ets.select(key_ets!(registry, partition), spec),
|
||||
do: pair
|
||||
@@ -795,15 +844,34 @@ defmodule Registry do
|
||||
@spec keys(registry, pid) :: [key]
|
||||
def keys(registry, pid) when is_atom(registry) and is_pid(pid) do
|
||||
{kind, partitions, _, pid_ets, _} = info!(registry)
|
||||
{_, pid_ets} = pid_ets || pid_ets!(registry, pid, partitions)
|
||||
|
||||
pid_etses =
|
||||
if pid_ets do
|
||||
{_, pid_ets} = pid_ets
|
||||
[pid_ets]
|
||||
else
|
||||
case kind do
|
||||
{:duplicate, :key} ->
|
||||
for partition <- 0..(partitions - 1) do
|
||||
{_, pid_ets} = pid_ets!(registry, partition)
|
||||
pid_ets
|
||||
end
|
||||
|
||||
_ ->
|
||||
{_, pid_ets} = pid_ets!(registry, pid, partitions)
|
||||
[pid_ets]
|
||||
end
|
||||
end
|
||||
|
||||
keys =
|
||||
try do
|
||||
spec = [{{pid, :"$1", :"$2", :_}, [], [{{:"$1", :"$2"}}]}]
|
||||
:ets.select(pid_ets, spec)
|
||||
catch
|
||||
:error, :badarg -> []
|
||||
end
|
||||
Enum.flat_map(pid_etses, fn pid_ets ->
|
||||
try do
|
||||
spec = [{{pid, :"$1", :"$2", :_}, [], [{{:"$1", :"$2"}}]}]
|
||||
:ets.select(pid_ets, spec)
|
||||
catch
|
||||
:error, :badarg -> []
|
||||
end
|
||||
end)
|
||||
|
||||
# Handle the possibility of fake keys
|
||||
keys = gather_keys(keys, [], false)
|
||||
@@ -882,8 +950,17 @@ defmodule Registry do
|
||||
[]
|
||||
end
|
||||
|
||||
{:duplicate, partitions, key_ets} ->
|
||||
key_ets = key_ets || key_ets!(registry, pid, partitions)
|
||||
{{:duplicate, _}, 1, key_ets} ->
|
||||
for {^pid, value} <- safe_lookup_second(key_ets, key), do: value
|
||||
|
||||
{{:duplicate, :key}, partitions, _key_ets} ->
|
||||
partition = hash(key, partitions)
|
||||
key_ets = key_ets!(registry, partition)
|
||||
for {^pid, value} <- safe_lookup_second(key_ets, key), do: value
|
||||
|
||||
{{:duplicate, :pid}, partitions, _key_ets} ->
|
||||
partition = hash(pid, partitions)
|
||||
key_ets = key_ets!(registry, partition)
|
||||
for {^pid, value} <- safe_lookup_second(key_ets, key), do: value
|
||||
end
|
||||
end
|
||||
@@ -1121,7 +1198,7 @@ defmodule Registry do
|
||||
end
|
||||
end
|
||||
|
||||
defp register_key(:duplicate, key_ets, _key, entry) do
|
||||
defp register_key({:duplicate, _}, key_ets, _key, entry) do
|
||||
true = :ets.insert(key_ets, entry)
|
||||
:ok
|
||||
end
|
||||
@@ -1339,10 +1416,10 @@ defmodule Registry do
|
||||
key_ets = key_ets || key_ets!(registry, key, partitions)
|
||||
:ets.select_count(key_ets, spec)
|
||||
|
||||
{:duplicate, 1, key_ets} ->
|
||||
{{:duplicate, _}, 1, key_ets} ->
|
||||
:ets.select_count(key_ets, spec)
|
||||
|
||||
{:duplicate, partitions, _key_ets} ->
|
||||
{{:duplicate, _}, partitions, _key_ets} ->
|
||||
Enum.sum_by(0..(partitions - 1), fn partition_index ->
|
||||
:ets.select_count(key_ets!(registry, partition_index), spec)
|
||||
end)
|
||||
@@ -1512,7 +1589,12 @@ defmodule Registry do
|
||||
{hash(key, partitions), hash(pid, partitions)}
|
||||
end
|
||||
|
||||
defp partitions(:duplicate, _key, pid, partitions) do
|
||||
defp partitions({:duplicate, :key}, key, _pid, partitions) do
|
||||
partition = hash(key, partitions)
|
||||
{partition, partition}
|
||||
end
|
||||
|
||||
defp partitions({:duplicate, :pid}, _key, pid, partitions) do
|
||||
partition = hash(pid, partitions)
|
||||
{partition, partition}
|
||||
end
|
||||
@@ -1576,9 +1658,10 @@ defmodule Registry.Supervisor do
|
||||
defp strategy_for_kind(:unique), do: :one_for_all
|
||||
|
||||
# Duplicate registries have both key and pid partitions hashed
|
||||
# by pid. This means that, if a PID partition crashes, all of
|
||||
# by key ({:duplicate, :key}) or pid ({:duplicate, :pid}).
|
||||
# This means that, if a PID or key partition crashes, all of
|
||||
# its associated entries are in its sibling table, so we crash one.
|
||||
defp strategy_for_kind(:duplicate), do: :one_for_one
|
||||
defp strategy_for_kind({:duplicate, _}), do: :one_for_one
|
||||
end
|
||||
|
||||
defmodule Registry.Partition do
|
||||
@@ -1633,6 +1716,7 @@ defmodule Registry.Partition do
|
||||
|
||||
def init({kind, registry, i, partitions, key_partition, pid_partition, listeners, compressed}) do
|
||||
Process.flag(:trap_exit, true)
|
||||
|
||||
key_ets = init_key_ets(kind, key_partition, compressed)
|
||||
pid_ets = init_pid_ets(kind, pid_partition)
|
||||
|
||||
@@ -1659,7 +1743,7 @@ defmodule Registry.Partition do
|
||||
:ets.new(key_partition, compression_opt(opts, compressed))
|
||||
end
|
||||
|
||||
defp init_key_ets(:duplicate, key_partition, compressed) do
|
||||
defp init_key_ets({:duplicate, _}, key_partition, compressed) do
|
||||
opts = [:duplicate_bag, :public, read_concurrency: true, write_concurrency: true]
|
||||
:ets.new(key_partition, compression_opt(opts, compressed))
|
||||
end
|
||||
|
||||
@@ -0,0 +1,504 @@
|
||||
# SPDX-License-Identifier: Apache-2.0
|
||||
# SPDX-FileCopyrightText: 2021 The Elixir Team
|
||||
# SPDX-FileCopyrightText: 2012 Plataformatec
|
||||
|
||||
Code.require_file("../test_helper.exs", __DIR__)
|
||||
|
||||
defmodule Registry.DuplicateTest do
|
||||
use ExUnit.Case,
|
||||
async: true,
|
||||
parameterize:
|
||||
for(
|
||||
keys <- [:duplicate, {:duplicate, :pid}, {:duplicate, :key}],
|
||||
partitions <- [1, 8],
|
||||
do: %{keys: keys, partitions: partitions}
|
||||
)
|
||||
|
||||
setup config do
|
||||
keys = config.keys
|
||||
partitions = config.partitions
|
||||
|
||||
listeners =
|
||||
List.wrap(config[:base_listener]) |> Enum.map(&:"#{&1}_#{partitions}_#{inspect(keys)}")
|
||||
|
||||
name = :"#{config.test}_#{partitions}_#{inspect(keys)}"
|
||||
opts = [keys: config.keys, name: name, partitions: partitions, listeners: listeners]
|
||||
{:ok, _} = start_supervised({Registry, opts})
|
||||
%{registry: name, listeners: listeners}
|
||||
end
|
||||
|
||||
test "starts configured number of partitions", %{registry: registry, partitions: partitions} do
|
||||
assert length(Supervisor.which_children(registry)) == partitions
|
||||
end
|
||||
|
||||
test "counts 0 keys in an empty registry", %{registry: registry} do
|
||||
assert 0 == Registry.count(registry)
|
||||
end
|
||||
|
||||
test "counts the number of keys in a registry", %{registry: registry} do
|
||||
{:ok, _} = Registry.register(registry, "hello", :value)
|
||||
{:ok, _} = Registry.register(registry, "hello", :value)
|
||||
|
||||
assert 2 == Registry.count(registry)
|
||||
end
|
||||
|
||||
test "has duplicate registrations", %{registry: registry} do
|
||||
{:ok, pid} = Registry.register(registry, "hello", :value)
|
||||
assert is_pid(pid)
|
||||
assert Registry.keys(registry, self()) == ["hello"]
|
||||
assert Registry.values(registry, "hello", self()) == [:value]
|
||||
|
||||
assert {:ok, pid} = Registry.register(registry, "hello", :value)
|
||||
assert is_pid(pid)
|
||||
assert Registry.keys(registry, self()) == ["hello", "hello"]
|
||||
assert Registry.values(registry, "hello", self()) == [:value, :value]
|
||||
|
||||
{:ok, pid} = Registry.register(registry, "world", :value)
|
||||
assert is_pid(pid)
|
||||
assert Registry.keys(registry, self()) |> Enum.sort() == ["hello", "hello", "world"]
|
||||
end
|
||||
|
||||
test "has duplicate registrations across processes", %{registry: registry} do
|
||||
{_, task} = register_task(registry, "hello", :world)
|
||||
assert Registry.keys(registry, self()) == []
|
||||
assert Registry.keys(registry, task) == ["hello"]
|
||||
assert Registry.values(registry, "hello", self()) == []
|
||||
assert Registry.values(registry, "hello", task) == [:world]
|
||||
|
||||
assert {:ok, _pid} = Registry.register(registry, "hello", :value)
|
||||
assert Registry.keys(registry, self()) == ["hello"]
|
||||
assert Registry.values(registry, "hello", self()) == [:value]
|
||||
end
|
||||
|
||||
test "compares using matches", %{registry: registry} do
|
||||
{:ok, _} = Registry.register(registry, 1.0, :value)
|
||||
{:ok, _} = Registry.register(registry, 1, :value)
|
||||
assert Registry.keys(registry, self()) |> Enum.sort() == [1, 1.0]
|
||||
end
|
||||
|
||||
test "dispatches to multiple keys in serial", %{registry: registry} do
|
||||
Process.flag(:trap_exit, true)
|
||||
parent = self()
|
||||
|
||||
fun = fn _ -> raise "will never be invoked" end
|
||||
assert Registry.dispatch(registry, "hello", fun, parallel: false) == :ok
|
||||
|
||||
{:ok, _} = Registry.register(registry, "hello", :value1)
|
||||
{:ok, _} = Registry.register(registry, "hello", :value2)
|
||||
{:ok, _} = Registry.register(registry, "world", :value3)
|
||||
|
||||
fun = fn entries ->
|
||||
assert parent == self()
|
||||
for {pid, value} <- entries, do: send(pid, {:dispatch, value})
|
||||
end
|
||||
|
||||
assert Registry.dispatch(registry, "hello", fun, parallel: false)
|
||||
|
||||
assert_received {:dispatch, :value1}
|
||||
assert_received {:dispatch, :value2}
|
||||
refute_received {:dispatch, :value3}
|
||||
|
||||
fun = fn entries ->
|
||||
assert parent == self()
|
||||
for {pid, value} <- entries, do: send(pid, {:dispatch, value})
|
||||
end
|
||||
|
||||
assert Registry.dispatch(registry, "world", fun, parallel: false)
|
||||
|
||||
refute_received {:dispatch, :value1}
|
||||
refute_received {:dispatch, :value2}
|
||||
assert_received {:dispatch, :value3}
|
||||
|
||||
refute_received {:EXIT, _, _}
|
||||
end
|
||||
|
||||
test "dispatches to multiple keys in parallel", context do
|
||||
%{registry: registry, partitions: partitions} = context
|
||||
Process.flag(:trap_exit, true)
|
||||
parent = self()
|
||||
|
||||
fun = fn _ -> raise "will never be invoked" end
|
||||
assert Registry.dispatch(registry, "hello", fun, parallel: true) == :ok
|
||||
|
||||
{:ok, _} = Registry.register(registry, "hello", :value1)
|
||||
{:ok, _} = Registry.register(registry, "hello", :value2)
|
||||
{:ok, _} = Registry.register(registry, "world", :value3)
|
||||
|
||||
fun = fn entries ->
|
||||
if partitions == 8 do
|
||||
assert parent != self()
|
||||
else
|
||||
assert parent == self()
|
||||
end
|
||||
|
||||
for {pid, value} <- entries, do: send(pid, {:dispatch, value})
|
||||
end
|
||||
|
||||
assert Registry.dispatch(registry, "hello", fun, parallel: true)
|
||||
|
||||
assert_received {:dispatch, :value1}
|
||||
assert_received {:dispatch, :value2}
|
||||
refute_received {:dispatch, :value3}
|
||||
|
||||
fun = fn entries ->
|
||||
if partitions == 8 do
|
||||
assert parent != self()
|
||||
else
|
||||
assert parent == self()
|
||||
end
|
||||
|
||||
for {pid, value} <- entries, do: send(pid, {:dispatch, value})
|
||||
end
|
||||
|
||||
assert Registry.dispatch(registry, "world", fun, parallel: true)
|
||||
|
||||
refute_received {:dispatch, :value1}
|
||||
refute_received {:dispatch, :value2}
|
||||
assert_received {:dispatch, :value3}
|
||||
|
||||
refute_received {:EXIT, _, _}
|
||||
end
|
||||
|
||||
test "unregisters by key", %{registry: registry} do
|
||||
{:ok, _} = Registry.register(registry, "hello", :value)
|
||||
{:ok, _} = Registry.register(registry, "hello", :value)
|
||||
{:ok, _} = Registry.register(registry, "world", :value)
|
||||
assert Registry.keys(registry, self()) |> Enum.sort() == ["hello", "hello", "world"]
|
||||
|
||||
:ok = Registry.unregister(registry, "hello")
|
||||
assert Registry.keys(registry, self()) == ["world"]
|
||||
|
||||
:ok = Registry.unregister(registry, "world")
|
||||
assert Registry.keys(registry, self()) == []
|
||||
end
|
||||
|
||||
test "unregisters with no entries", %{registry: registry} do
|
||||
assert Registry.unregister(registry, "hello") == :ok
|
||||
end
|
||||
|
||||
test "unregisters with tricky keys", %{registry: registry} do
|
||||
{:ok, _} = Registry.register(registry, :_, :foo)
|
||||
{:ok, _} = Registry.register(registry, :_, :bar)
|
||||
{:ok, _} = Registry.register(registry, "hello", "a")
|
||||
{:ok, _} = Registry.register(registry, "hello", "b")
|
||||
|
||||
Registry.unregister(registry, :_)
|
||||
assert Registry.keys(registry, self()) |> Enum.sort() == ["hello", "hello"]
|
||||
end
|
||||
|
||||
test "supports match patterns", %{registry: registry} do
|
||||
value1 = {1, :atom, 1}
|
||||
value2 = {2, :atom, 2}
|
||||
|
||||
{:ok, _} = Registry.register(registry, "hello", value1)
|
||||
{:ok, _} = Registry.register(registry, "hello", value2)
|
||||
|
||||
assert Registry.match(registry, "hello", {1, :_, :_}) == [{self(), value1}]
|
||||
assert Registry.match(registry, "hello", {1.0, :_, :_}) == []
|
||||
|
||||
assert Registry.match(registry, "hello", {:_, :atom, :_}) |> Enum.sort() ==
|
||||
[{self(), value1}, {self(), value2}]
|
||||
|
||||
assert Registry.match(registry, "hello", {:"$1", :_, :"$1"}) |> Enum.sort() ==
|
||||
[{self(), value1}, {self(), value2}]
|
||||
|
||||
assert Registry.match(registry, "hello", {2, :_, :_}) == [{self(), value2}]
|
||||
assert Registry.match(registry, "hello", {2.0, :_, :_}) == []
|
||||
end
|
||||
|
||||
test "supports guards", %{registry: registry} do
|
||||
value1 = {1, :atom, 1}
|
||||
value2 = {2, :atom, 2}
|
||||
|
||||
{:ok, _} = Registry.register(registry, "hello", value1)
|
||||
{:ok, _} = Registry.register(registry, "hello", value2)
|
||||
|
||||
assert Registry.match(registry, "hello", {:"$1", :_, :_}, [{:<, :"$1", 2}]) ==
|
||||
[{self(), value1}]
|
||||
|
||||
assert Registry.match(registry, "hello", {:"$1", :_, :_}, [{:>, :"$1", 3}]) == []
|
||||
|
||||
assert Registry.match(registry, "hello", {:"$1", :_, :_}, [{:<, :"$1", 3}]) |> Enum.sort() ==
|
||||
[{self(), value1}, {self(), value2}]
|
||||
|
||||
assert Registry.match(registry, "hello", {:_, :"$1", :_}, [{:is_atom, :"$1"}])
|
||||
|> Enum.sort() == [{self(), value1}, {self(), value2}]
|
||||
end
|
||||
|
||||
test "count_match supports match patterns", %{registry: registry} do
|
||||
value = {1, :atom, 1}
|
||||
{:ok, _} = Registry.register(registry, "hello", value)
|
||||
assert 1 == Registry.count_match(registry, "hello", {1, :_, :_})
|
||||
assert 0 == Registry.count_match(registry, "hello", {1.0, :_, :_})
|
||||
assert 1 == Registry.count_match(registry, "hello", {:_, :atom, :_})
|
||||
assert 1 == Registry.count_match(registry, "hello", {:"$1", :_, :"$1"})
|
||||
assert 1 == Registry.count_match(registry, "hello", :_)
|
||||
assert 0 == Registry.count_match(registry, :_, :_)
|
||||
|
||||
value2 = %{a: "a", b: "b"}
|
||||
{:ok, _} = Registry.register(registry, "world", value2)
|
||||
assert 1 == Registry.count_match(registry, "world", %{b: "b"})
|
||||
end
|
||||
|
||||
test "count_match supports guard conditions", %{registry: registry} do
|
||||
value = {1, :atom, 2}
|
||||
{:ok, _} = Registry.register(registry, "hello", value)
|
||||
|
||||
assert 1 == Registry.count_match(registry, "hello", {:_, :_, :"$1"}, [{:>, :"$1", 1}])
|
||||
assert 0 == Registry.count_match(registry, "hello", {:_, :_, :"$1"}, [{:>, :"$1", 2}])
|
||||
assert 1 == Registry.count_match(registry, "hello", {:_, :"$1", :_}, [{:is_atom, :"$1"}])
|
||||
end
|
||||
|
||||
test "unregister_match supports patterns", %{registry: registry} do
|
||||
value1 = {1, :atom, 1}
|
||||
value2 = {2, :atom, 2}
|
||||
|
||||
{:ok, _} = Registry.register(registry, "hello", value1)
|
||||
{:ok, _} = Registry.register(registry, "hello", value2)
|
||||
|
||||
Registry.unregister_match(registry, "hello", {2, :_, :_})
|
||||
assert Registry.lookup(registry, "hello") == [{self(), value1}]
|
||||
|
||||
{:ok, _} = Registry.register(registry, "hello", value2)
|
||||
Registry.unregister_match(registry, "hello", {2.0, :_, :_})
|
||||
assert Registry.lookup(registry, "hello") == [{self(), value1}, {self(), value2}]
|
||||
Registry.unregister_match(registry, "hello", {:_, :atom, :_})
|
||||
assert Registry.lookup(registry, "hello") == []
|
||||
end
|
||||
|
||||
test "unregister_match supports guards", %{registry: registry} do
|
||||
value1 = {1, :atom, 1}
|
||||
value2 = {2, :atom, 2}
|
||||
|
||||
{:ok, _} = Registry.register(registry, "hello", value1)
|
||||
{:ok, _} = Registry.register(registry, "hello", value2)
|
||||
|
||||
Registry.unregister_match(registry, "hello", {:"$1", :_, :_}, [{:<, :"$1", 2}])
|
||||
assert Registry.lookup(registry, "hello") == [{self(), value2}]
|
||||
end
|
||||
|
||||
test "unregister_match supports tricky keys", %{registry: registry} do
|
||||
{:ok, _} = Registry.register(registry, :_, :foo)
|
||||
{:ok, _} = Registry.register(registry, :_, :bar)
|
||||
{:ok, _} = Registry.register(registry, "hello", "a")
|
||||
{:ok, _} = Registry.register(registry, "hello", "b")
|
||||
|
||||
Registry.unregister_match(registry, :_, :foo)
|
||||
assert Registry.lookup(registry, :_) == [{self(), :bar}]
|
||||
|
||||
assert Registry.keys(registry, self()) |> Enum.sort() == [:_, "hello", "hello"]
|
||||
end
|
||||
|
||||
@tag base_listener: :unique_listener
|
||||
test "allows listeners", %{registry: registry, listeners: [listener]} do
|
||||
Process.register(self(), listener)
|
||||
{_, task} = register_task(registry, "hello", :world)
|
||||
assert_received {:register, ^registry, "hello", ^task, :world}
|
||||
|
||||
self = self()
|
||||
{:ok, _} = Registry.register(registry, "hello", :value)
|
||||
assert_received {:register, ^registry, "hello", ^self, :value}
|
||||
|
||||
:ok = Registry.unregister(registry, "hello")
|
||||
assert_received {:unregister, ^registry, "hello", ^self}
|
||||
after
|
||||
Process.unregister(listener)
|
||||
end
|
||||
|
||||
test "links and unlinks on register/unregister", %{registry: registry} do
|
||||
{:ok, pid} = Registry.register(registry, "hello", :value)
|
||||
{:links, links} = Process.info(self(), :links)
|
||||
assert pid in links
|
||||
|
||||
{:ok, pid} = Registry.register(registry, "world", :value)
|
||||
{:links, links} = Process.info(self(), :links)
|
||||
assert pid in links
|
||||
|
||||
:ok = Registry.unregister(registry, "hello")
|
||||
{:links, links} = Process.info(self(), :links)
|
||||
assert pid in links
|
||||
|
||||
:ok = Registry.unregister(registry, "world")
|
||||
{:links, links} = Process.info(self(), :links)
|
||||
refute pid in links
|
||||
end
|
||||
|
||||
test "raises on unknown registry name" do
|
||||
assert_raise ArgumentError, ~r/unknown registry/, fn ->
|
||||
Registry.register(:unknown, "hello", :value)
|
||||
end
|
||||
end
|
||||
|
||||
test "raises if attempt to be used on via", %{registry: registry} do
|
||||
assert_raise ArgumentError, ":via is not supported for duplicate registries", fn ->
|
||||
name = {:via, Registry, {registry, "hello"}}
|
||||
Agent.start_link(fn -> 0 end, name: name)
|
||||
end
|
||||
end
|
||||
|
||||
test "empty list for empty registry", %{registry: registry} do
|
||||
assert Registry.select(registry, [{{:_, :_, :_}, [], [:"$_"]}]) == []
|
||||
end
|
||||
|
||||
test "select all", %{registry: registry} do
|
||||
{:ok, _} = Registry.register(registry, "hello", :value)
|
||||
{:ok, _} = Registry.register(registry, "hello", :value)
|
||||
|
||||
assert Registry.select(registry, [{{:"$1", :"$2", :"$3"}, [], [{{:"$1", :"$2", :"$3"}}]}])
|
||||
|> Enum.sort() ==
|
||||
[{"hello", self(), :value}, {"hello", self(), :value}]
|
||||
end
|
||||
|
||||
test "select supports full match specs", %{registry: registry} do
|
||||
value = {1, :atom, 1}
|
||||
{:ok, _} = Registry.register(registry, "hello", value)
|
||||
|
||||
assert [{"hello", self(), value}] ==
|
||||
Registry.select(registry, [
|
||||
{{"hello", :"$2", :"$3"}, [], [{{"hello", :"$2", :"$3"}}]}
|
||||
])
|
||||
|
||||
assert [{"hello", self(), value}] ==
|
||||
Registry.select(registry, [
|
||||
{{:"$1", self(), :"$3"}, [], [{{:"$1", self(), :"$3"}}]}
|
||||
])
|
||||
|
||||
assert [{"hello", self(), value}] ==
|
||||
Registry.select(registry, [
|
||||
{{:"$1", :"$2", value}, [], [{{:"$1", :"$2", {value}}}]}
|
||||
])
|
||||
|
||||
assert [] ==
|
||||
Registry.select(registry, [
|
||||
{{"world", :"$2", :"$3"}, [], [{{"world", :"$2", :"$3"}}]}
|
||||
])
|
||||
|
||||
assert [] == Registry.select(registry, [{{:"$1", :"$2", {1.0, :_, :_}}, [], [:"$_"]}])
|
||||
|
||||
assert [{"hello", self(), value}] ==
|
||||
Registry.select(registry, [
|
||||
{{:"$1", :"$2", {:"$3", :atom, :"$4"}}, [],
|
||||
[{{:"$1", :"$2", {{:"$3", :atom, :"$4"}}}}]}
|
||||
])
|
||||
|
||||
assert [{"hello", self(), {1, :atom, 1}}] ==
|
||||
Registry.select(registry, [
|
||||
{{:"$1", :"$2", {:"$3", :"$4", :"$3"}}, [],
|
||||
[{{:"$1", :"$2", {{:"$3", :"$4", :"$3"}}}}]}
|
||||
])
|
||||
|
||||
value2 = %{a: "a", b: "b"}
|
||||
{:ok, _} = Registry.register(registry, "world", value2)
|
||||
|
||||
assert [:match] ==
|
||||
Registry.select(registry, [{{"world", self(), %{b: "b"}}, [], [:match]}])
|
||||
|
||||
assert ["hello", "world"] ==
|
||||
Registry.select(registry, [{{:"$1", :_, :_}, [], [:"$1"]}]) |> Enum.sort()
|
||||
end
|
||||
|
||||
test "select supports guard conditions", %{registry: registry} do
|
||||
value = {1, :atom, 2}
|
||||
{:ok, _} = Registry.register(registry, "hello", value)
|
||||
|
||||
assert [{"hello", self(), {1, :atom, 2}}] ==
|
||||
Registry.select(registry, [
|
||||
{{:"$1", :"$2", {:"$3", :"$4", :"$5"}}, [{:>, :"$5", 1}],
|
||||
[{{:"$1", :"$2", {{:"$3", :"$4", :"$5"}}}}]}
|
||||
])
|
||||
|
||||
assert [] ==
|
||||
Registry.select(registry, [
|
||||
{{:_, :_, {:_, :_, :"$1"}}, [{:>, :"$1", 2}], [:"$_"]}
|
||||
])
|
||||
|
||||
assert ["hello"] ==
|
||||
Registry.select(registry, [
|
||||
{{:"$1", :_, {:_, :"$2", :_}}, [{:is_atom, :"$2"}], [:"$1"]}
|
||||
])
|
||||
end
|
||||
|
||||
test "select allows multiple specs", %{registry: registry} do
|
||||
{:ok, _} = Registry.register(registry, "hello", :value)
|
||||
{:ok, _} = Registry.register(registry, "world", :value)
|
||||
|
||||
assert ["hello", "world"] ==
|
||||
Registry.select(registry, [
|
||||
{{"hello", :_, :_}, [], [{:element, 1, :"$_"}]},
|
||||
{{"world", :_, :_}, [], [{:element, 1, :"$_"}]}
|
||||
])
|
||||
|> Enum.sort()
|
||||
end
|
||||
|
||||
test "count_select supports match specs", %{registry: registry} do
|
||||
value = {1, :atom, 1}
|
||||
{:ok, _} = Registry.register(registry, "hello", value)
|
||||
assert 1 == Registry.count_select(registry, [{{:_, :_, value}, [], [true]}])
|
||||
assert 1 == Registry.count_select(registry, [{{"hello", :_, :_}, [], [true]}])
|
||||
assert 1 == Registry.count_select(registry, [{{:_, :_, {1, :atom, :_}}, [], [true]}])
|
||||
assert 1 == Registry.count_select(registry, [{{:_, :_, {:"$1", :_, :"$1"}}, [], [true]}])
|
||||
assert 0 == Registry.count_select(registry, [{{"hello", :_, nil}, [], [true]}])
|
||||
|
||||
value2 = %{a: "a", b: "b"}
|
||||
{:ok, _} = Registry.register(registry, "world", value2)
|
||||
assert 1 == Registry.count_select(registry, [{{"world", :_, :_}, [], [true]}])
|
||||
end
|
||||
|
||||
test "count_select supports guard conditions", %{registry: registry} do
|
||||
value = {1, :atom, 2}
|
||||
{:ok, _} = Registry.register(registry, "hello", value)
|
||||
|
||||
assert 1 ==
|
||||
Registry.count_select(registry, [
|
||||
{{:_, :_, {:_, :"$1", :_}}, [{:is_atom, :"$1"}], [true]}
|
||||
])
|
||||
|
||||
assert 1 ==
|
||||
Registry.count_select(registry, [
|
||||
{{:_, :_, {:_, :_, :"$1"}}, [{:>, :"$1", 1}], [true]}
|
||||
])
|
||||
|
||||
assert 0 ==
|
||||
Registry.count_select(registry, [
|
||||
{{:_, :_, {:_, :_, :"$1"}}, [{:>, :"$1", 2}], [true]}
|
||||
])
|
||||
end
|
||||
|
||||
test "count_select allows multiple specs", %{registry: registry} do
|
||||
{:ok, _} = Registry.register(registry, "hello", :value)
|
||||
{:ok, _} = Registry.register(registry, "world", :value)
|
||||
|
||||
assert 2 ==
|
||||
Registry.count_select(registry, [
|
||||
{{"hello", :_, :_}, [], [true]},
|
||||
{{"world", :_, :_}, [], [true]}
|
||||
])
|
||||
end
|
||||
|
||||
test "rejects invalid tuple syntax", %{partitions: partitions} do
|
||||
name = :"test_invalid_tuple_#{partitions}"
|
||||
|
||||
assert_raise ArgumentError, ~r/expected :keys to be given and be one of/, fn ->
|
||||
Registry.start_link(keys: {:duplicate, :invalid}, name: name, partitions: partitions)
|
||||
end
|
||||
end
|
||||
|
||||
test "update_value is not supported", %{registry: registry} do
|
||||
assert_raise ArgumentError, ~r/Registry.update_value\/3 is not supported/, fn ->
|
||||
Registry.update_value(registry, "hello", fn val -> val end)
|
||||
end
|
||||
end
|
||||
|
||||
defp register_task(registry, key, value) do
|
||||
parent = self()
|
||||
|
||||
{:ok, task} =
|
||||
Task.start(fn ->
|
||||
send(parent, Registry.register(registry, key, value))
|
||||
Process.sleep(:infinity)
|
||||
end)
|
||||
|
||||
assert_receive {:ok, owner}
|
||||
{owner, task}
|
||||
end
|
||||
end
|
||||
@@ -0,0 +1,507 @@
|
||||
# SPDX-License-Identifier: Apache-2.0
|
||||
# SPDX-FileCopyrightText: 2021 The Elixir Team
|
||||
# SPDX-FileCopyrightText: 2012 Plataformatec
|
||||
|
||||
Code.require_file("../test_helper.exs", __DIR__)
|
||||
|
||||
defmodule Registry.UniqueTest do
|
||||
use ExUnit.Case,
|
||||
async: true,
|
||||
parameterize: [
|
||||
%{partitions: 1},
|
||||
%{partitions: 8}
|
||||
]
|
||||
|
||||
@keys :unique
|
||||
|
||||
setup config do
|
||||
partitions = config.partitions
|
||||
listeners = List.wrap(config[:base_listener]) |> Enum.map(&:"#{&1}_#{partitions}")
|
||||
name = :"#{config.test}_#{partitions}"
|
||||
opts = [keys: @keys, name: name, partitions: partitions, listeners: listeners]
|
||||
{:ok, _} = start_supervised({Registry, opts})
|
||||
%{registry: name, listeners: listeners}
|
||||
end
|
||||
|
||||
test "starts configured number of partitions", %{registry: registry, partitions: partitions} do
|
||||
assert length(Supervisor.which_children(registry)) == partitions
|
||||
end
|
||||
|
||||
test "counts 0 keys in an empty registry", %{registry: registry} do
|
||||
assert 0 == Registry.count(registry)
|
||||
end
|
||||
|
||||
test "counts the number of keys in a registry", %{registry: registry} do
|
||||
{:ok, _} = Registry.register(registry, "hello", :value)
|
||||
{:ok, _} = Registry.register(registry, "world", :value)
|
||||
|
||||
assert 2 == Registry.count(registry)
|
||||
end
|
||||
|
||||
test "has unique registrations", %{registry: registry} do
|
||||
{:ok, pid} = Registry.register(registry, "hello", :value)
|
||||
assert is_pid(pid)
|
||||
assert Registry.keys(registry, self()) == ["hello"]
|
||||
assert Registry.values(registry, "hello", self()) == [:value]
|
||||
|
||||
assert {:error, {:already_registered, pid}} = Registry.register(registry, "hello", :value)
|
||||
assert pid == self()
|
||||
assert Registry.keys(registry, self()) == ["hello"]
|
||||
assert Registry.values(registry, "hello", self()) == [:value]
|
||||
|
||||
{:ok, pid} = Registry.register(registry, "world", :value)
|
||||
assert is_pid(pid)
|
||||
assert Registry.keys(registry, self()) |> Enum.sort() == ["hello", "world"]
|
||||
end
|
||||
|
||||
test "has unique registrations across processes", %{registry: registry} do
|
||||
{_, task} = register_task(registry, "hello", :value)
|
||||
Process.link(Process.whereis(registry))
|
||||
assert Registry.keys(registry, task) == ["hello"]
|
||||
assert Registry.values(registry, "hello", task) == [:value]
|
||||
|
||||
assert {:error, {:already_registered, ^task}} =
|
||||
Registry.register(registry, "hello", :recent)
|
||||
|
||||
assert Registry.keys(registry, self()) == []
|
||||
assert Registry.values(registry, "hello", self()) == []
|
||||
|
||||
{:links, links} = Process.info(self(), :links)
|
||||
assert Process.whereis(registry) in links
|
||||
end
|
||||
|
||||
test "has unique registrations even if partition is delayed", %{registry: registry} do
|
||||
{owner, task} = register_task(registry, "hello", :value)
|
||||
|
||||
assert Registry.register(registry, "hello", :other) ==
|
||||
{:error, {:already_registered, task}}
|
||||
|
||||
:sys.suspend(owner)
|
||||
kill_and_assert_down(task)
|
||||
Registry.register(registry, "hello", :other)
|
||||
assert Registry.lookup(registry, "hello") == [{self(), :other}]
|
||||
end
|
||||
|
||||
test "supports match patterns", %{registry: registry} do
|
||||
value = {1, :atom, 1}
|
||||
{:ok, _} = Registry.register(registry, "hello", value)
|
||||
assert Registry.match(registry, "hello", {1, :_, :_}) == [{self(), value}]
|
||||
assert Registry.match(registry, "hello", {1.0, :_, :_}) == []
|
||||
assert Registry.match(registry, "hello", {:_, :atom, :_}) == [{self(), value}]
|
||||
assert Registry.match(registry, "hello", {:"$1", :_, :"$1"}) == [{self(), value}]
|
||||
assert Registry.match(registry, "hello", :_) == [{self(), value}]
|
||||
assert Registry.match(registry, :_, :_) == []
|
||||
|
||||
value2 = %{a: "a", b: "b"}
|
||||
{:ok, _} = Registry.register(registry, "world", value2)
|
||||
assert Registry.match(registry, "world", %{b: "b"}) == [{self(), value2}]
|
||||
end
|
||||
|
||||
test "supports guard conditions", %{registry: registry} do
|
||||
value = {1, :atom, 2}
|
||||
{:ok, _} = Registry.register(registry, "hello", value)
|
||||
|
||||
assert Registry.match(registry, "hello", {:_, :_, :"$1"}, [{:>, :"$1", 1}]) ==
|
||||
[{self(), value}]
|
||||
|
||||
assert Registry.match(registry, "hello", {:_, :_, :"$1"}, [{:>, :"$1", 2}]) == []
|
||||
|
||||
assert Registry.match(registry, "hello", {:_, :"$1", :_}, [{:is_atom, :"$1"}]) ==
|
||||
[{self(), value}]
|
||||
end
|
||||
|
||||
test "count_match supports match patterns", %{registry: registry} do
|
||||
value = {1, :atom, 1}
|
||||
{:ok, _} = Registry.register(registry, "hello", value)
|
||||
assert 1 == Registry.count_match(registry, "hello", {1, :_, :_})
|
||||
assert 0 == Registry.count_match(registry, "hello", {1.0, :_, :_})
|
||||
assert 1 == Registry.count_match(registry, "hello", {:_, :atom, :_})
|
||||
assert 1 == Registry.count_match(registry, "hello", {:"$1", :_, :"$1"})
|
||||
assert 1 == Registry.count_match(registry, "hello", :_)
|
||||
assert 0 == Registry.count_match(registry, :_, :_)
|
||||
|
||||
value2 = %{a: "a", b: "b"}
|
||||
{:ok, _} = Registry.register(registry, "world", value2)
|
||||
assert 1 == Registry.count_match(registry, "world", %{b: "b"})
|
||||
end
|
||||
|
||||
test "count_match supports guard conditions", %{registry: registry} do
|
||||
value = {1, :atom, 2}
|
||||
{:ok, _} = Registry.register(registry, "hello", value)
|
||||
|
||||
assert 1 == Registry.count_match(registry, "hello", {:_, :_, :"$1"}, [{:>, :"$1", 1}])
|
||||
assert 0 == Registry.count_match(registry, "hello", {:_, :_, :"$1"}, [{:>, :"$1", 2}])
|
||||
assert 1 == Registry.count_match(registry, "hello", {:_, :"$1", :_}, [{:is_atom, :"$1"}])
|
||||
end
|
||||
|
||||
test "unregister_match supports patterns", %{registry: registry} do
|
||||
value = {1, :atom, 1}
|
||||
{:ok, _} = Registry.register(registry, "hello", value)
|
||||
|
||||
Registry.unregister_match(registry, "hello", {2, :_, :_})
|
||||
assert Registry.lookup(registry, "hello") == [{self(), value}]
|
||||
Registry.unregister_match(registry, "hello", {1.0, :_, :_})
|
||||
assert Registry.lookup(registry, "hello") == [{self(), value}]
|
||||
Registry.unregister_match(registry, "hello", {:_, :atom, :_})
|
||||
assert Registry.lookup(registry, "hello") == []
|
||||
end
|
||||
|
||||
test "unregister_match supports guards", %{registry: registry} do
|
||||
value = {1, :atom, 1}
|
||||
{:ok, _} = Registry.register(registry, "hello", value)
|
||||
|
||||
Registry.unregister_match(registry, "hello", {:"$1", :_, :_}, [{:<, :"$1", 2}])
|
||||
assert Registry.lookup(registry, "hello") == []
|
||||
end
|
||||
|
||||
test "unregister_match supports tricky keys", %{registry: registry} do
|
||||
{:ok, _} = Registry.register(registry, :_, :foo)
|
||||
{:ok, _} = Registry.register(registry, "hello", "b")
|
||||
|
||||
Registry.unregister_match(registry, :_, :foo)
|
||||
assert Registry.lookup(registry, :_) == []
|
||||
assert Registry.keys(registry, self()) |> Enum.sort() == ["hello"]
|
||||
end
|
||||
|
||||
test "compares using ===", %{registry: registry} do
|
||||
{:ok, _} = Registry.register(registry, 1.0, :value)
|
||||
{:ok, _} = Registry.register(registry, 1, :value)
|
||||
assert Registry.keys(registry, self()) |> Enum.sort() == [1, 1.0]
|
||||
end
|
||||
|
||||
test "updates current process value", %{registry: registry} do
|
||||
assert Registry.update_value(registry, "hello", &raise/1) == :error
|
||||
register_task(registry, "hello", :value)
|
||||
assert Registry.update_value(registry, "hello", &raise/1) == :error
|
||||
|
||||
Registry.register(registry, "world", 1)
|
||||
assert Registry.lookup(registry, "world") == [{self(), 1}]
|
||||
assert Registry.update_value(registry, "world", &(&1 + 1)) == {2, 1}
|
||||
assert Registry.lookup(registry, "world") == [{self(), 2}]
|
||||
end
|
||||
|
||||
test "dispatches to a single key", %{registry: registry} do
|
||||
fun = fn _ -> raise "will never be invoked" end
|
||||
assert Registry.dispatch(registry, "hello", fun) == :ok
|
||||
|
||||
{:ok, _} = Registry.register(registry, "hello", :value)
|
||||
|
||||
fun = fn [{pid, value}] -> send(pid, {:dispatch, value}) end
|
||||
assert Registry.dispatch(registry, "hello", fun)
|
||||
|
||||
assert_received {:dispatch, :value}
|
||||
end
|
||||
|
||||
test "unregisters process by key", %{registry: registry} do
|
||||
:ok = Registry.unregister(registry, "hello")
|
||||
|
||||
{:ok, _} = Registry.register(registry, "hello", :value)
|
||||
{:ok, _} = Registry.register(registry, "world", :value)
|
||||
assert Registry.keys(registry, self()) |> Enum.sort() == ["hello", "world"]
|
||||
|
||||
:ok = Registry.unregister(registry, "hello")
|
||||
assert Registry.keys(registry, self()) == ["world"]
|
||||
|
||||
:ok = Registry.unregister(registry, "world")
|
||||
assert Registry.keys(registry, self()) == []
|
||||
end
|
||||
|
||||
test "unregisters with no entries", %{registry: registry} do
|
||||
assert Registry.unregister(registry, "hello") == :ok
|
||||
end
|
||||
|
||||
test "unregisters with tricky keys", %{registry: registry} do
|
||||
{:ok, _} = Registry.register(registry, :_, :foo)
|
||||
{:ok, _} = Registry.register(registry, "hello", "b")
|
||||
|
||||
Registry.unregister(registry, :_)
|
||||
assert Registry.lookup(registry, :_) == []
|
||||
assert Registry.keys(registry, self()) |> Enum.sort() == ["hello"]
|
||||
end
|
||||
|
||||
@tag base_listener: :unique_listener
|
||||
test "allows listeners", %{registry: registry, listeners: [listener]} do
|
||||
Process.register(self(), listener)
|
||||
{_, task} = register_task(registry, "hello", :world)
|
||||
assert_received {:register, ^registry, "hello", ^task, :world}
|
||||
|
||||
self = self()
|
||||
{:ok, _} = Registry.register(registry, "world", :value)
|
||||
assert_received {:register, ^registry, "world", ^self, :value}
|
||||
|
||||
:ok = Registry.unregister(registry, "world")
|
||||
assert_received {:unregister, ^registry, "world", ^self}
|
||||
after
|
||||
Process.unregister(listener)
|
||||
end
|
||||
|
||||
test "links and unlinks on register/unregister", %{registry: registry} do
|
||||
{:ok, pid} = Registry.register(registry, "hello", :value)
|
||||
{:links, links} = Process.info(self(), :links)
|
||||
assert pid in links
|
||||
|
||||
{:ok, pid} = Registry.register(registry, "world", :value)
|
||||
{:links, links} = Process.info(self(), :links)
|
||||
assert pid in links
|
||||
|
||||
:ok = Registry.unregister(registry, "hello")
|
||||
{:links, links} = Process.info(self(), :links)
|
||||
assert pid in links
|
||||
|
||||
:ok = Registry.unregister(registry, "world")
|
||||
{:links, links} = Process.info(self(), :links)
|
||||
refute pid in links
|
||||
end
|
||||
|
||||
test "raises on unknown registry name" do
|
||||
assert_raise ArgumentError, ~r/unknown registry/, fn ->
|
||||
Registry.register(:unknown, "hello", :value)
|
||||
end
|
||||
end
|
||||
|
||||
test "via callbacks", %{registry: registry} do
|
||||
name = {:via, Registry, {registry, "hello"}}
|
||||
|
||||
# register_name
|
||||
{:ok, pid} = Agent.start_link(fn -> 0 end, name: name)
|
||||
|
||||
# send
|
||||
assert Agent.update(name, &(&1 + 1)) == :ok
|
||||
|
||||
# whereis_name
|
||||
assert Agent.get(name, & &1) == 1
|
||||
|
||||
# unregister_name
|
||||
assert {:error, _} = Agent.start(fn -> raise "oops" end)
|
||||
|
||||
# errors
|
||||
assert {:error, {:already_started, ^pid}} = Agent.start(fn -> 0 end, name: name)
|
||||
end
|
||||
|
||||
test "uses value provided in via", %{registry: registry} do
|
||||
name = {:via, Registry, {registry, "hello", :value}}
|
||||
{:ok, pid} = Agent.start_link(fn -> 0 end, name: name)
|
||||
assert Registry.lookup(registry, "hello") == [{pid, :value}]
|
||||
end
|
||||
|
||||
test "empty list for empty registry", %{registry: registry} do
|
||||
assert Registry.select(registry, [{{:_, :_, :_}, [], [:"$_"]}]) == []
|
||||
end
|
||||
|
||||
test "select all", %{registry: registry} do
|
||||
name = {:via, Registry, {registry, "hello"}}
|
||||
{:ok, pid} = Agent.start_link(fn -> 0 end, name: name)
|
||||
{:ok, _} = Registry.register(registry, "world", :value)
|
||||
|
||||
assert Registry.select(registry, [{{:"$1", :"$2", :"$3"}, [], [{{:"$1", :"$2", :"$3"}}]}])
|
||||
|> Enum.sort() ==
|
||||
[{"hello", pid, nil}, {"world", self(), :value}]
|
||||
end
|
||||
|
||||
test "select supports full match specs", %{registry: registry} do
|
||||
value = {1, :atom, 1}
|
||||
{:ok, _} = Registry.register(registry, "hello", value)
|
||||
|
||||
assert [{"hello", self(), value}] ==
|
||||
Registry.select(registry, [
|
||||
{{"hello", :"$2", :"$3"}, [], [{{"hello", :"$2", :"$3"}}]}
|
||||
])
|
||||
|
||||
assert [{"hello", self(), value}] ==
|
||||
Registry.select(registry, [
|
||||
{{:"$1", self(), :"$3"}, [], [{{:"$1", self(), :"$3"}}]}
|
||||
])
|
||||
|
||||
assert [{"hello", self(), value}] ==
|
||||
Registry.select(registry, [
|
||||
{{:"$1", :"$2", value}, [], [{{:"$1", :"$2", {value}}}]}
|
||||
])
|
||||
|
||||
assert [] ==
|
||||
Registry.select(registry, [
|
||||
{{"world", :"$2", :"$3"}, [], [{{"world", :"$2", :"$3"}}]}
|
||||
])
|
||||
|
||||
assert [] == Registry.select(registry, [{{:"$1", :"$2", {1.0, :_, :_}}, [], [:"$_"]}])
|
||||
|
||||
assert [{"hello", self(), value}] ==
|
||||
Registry.select(registry, [
|
||||
{{:"$1", :"$2", {:"$3", :atom, :"$4"}}, [],
|
||||
[{{:"$1", :"$2", {{:"$3", :atom, :"$4"}}}}]}
|
||||
])
|
||||
|
||||
assert [{"hello", self(), {1, :atom, 1}}] ==
|
||||
Registry.select(registry, [
|
||||
{{:"$1", :"$2", {:"$3", :"$4", :"$3"}}, [],
|
||||
[{{:"$1", :"$2", {{:"$3", :"$4", :"$3"}}}}]}
|
||||
])
|
||||
|
||||
value2 = %{a: "a", b: "b"}
|
||||
{:ok, _} = Registry.register(registry, "world", value2)
|
||||
|
||||
assert [:match] ==
|
||||
Registry.select(registry, [{{"world", self(), %{b: "b"}}, [], [:match]}])
|
||||
|
||||
assert ["hello", "world"] ==
|
||||
Registry.select(registry, [{{:"$1", :_, :_}, [], [:"$1"]}]) |> Enum.sort()
|
||||
end
|
||||
|
||||
test "select supports guard conditions", %{registry: registry} do
|
||||
value = {1, :atom, 2}
|
||||
{:ok, _} = Registry.register(registry, "hello", value)
|
||||
|
||||
assert [{"hello", self(), {1, :atom, 2}}] ==
|
||||
Registry.select(registry, [
|
||||
{{:"$1", :"$2", {:"$3", :"$4", :"$5"}}, [{:>, :"$5", 1}],
|
||||
[{{:"$1", :"$2", {{:"$3", :"$4", :"$5"}}}}]}
|
||||
])
|
||||
|
||||
assert [] ==
|
||||
Registry.select(registry, [
|
||||
{{:_, :_, {:_, :_, :"$1"}}, [{:>, :"$1", 2}], [:"$_"]}
|
||||
])
|
||||
|
||||
assert ["hello"] ==
|
||||
Registry.select(registry, [
|
||||
{{:"$1", :_, {:_, :"$2", :_}}, [{:is_atom, :"$2"}], [:"$1"]}
|
||||
])
|
||||
end
|
||||
|
||||
test "select allows multiple specs", %{registry: registry} do
|
||||
{:ok, _} = Registry.register(registry, "hello", :value)
|
||||
{:ok, _} = Registry.register(registry, "world", :value)
|
||||
|
||||
assert ["hello", "world"] ==
|
||||
Registry.select(registry, [
|
||||
{{"hello", :_, :_}, [], [{:element, 1, :"$_"}]},
|
||||
{{"world", :_, :_}, [], [{:element, 1, :"$_"}]}
|
||||
])
|
||||
|> Enum.sort()
|
||||
end
|
||||
|
||||
test "select raises on incorrect shape of match spec", %{registry: registry} do
|
||||
assert_raise ArgumentError, fn ->
|
||||
Registry.select(registry, [{:_, [], []}])
|
||||
end
|
||||
end
|
||||
|
||||
test "count_select supports match specs", %{registry: registry} do
|
||||
value = {1, :atom, 1}
|
||||
{:ok, _} = Registry.register(registry, "hello", value)
|
||||
assert 1 == Registry.count_select(registry, [{{:_, :_, value}, [], [true]}])
|
||||
assert 1 == Registry.count_select(registry, [{{"hello", :_, :_}, [], [true]}])
|
||||
assert 1 == Registry.count_select(registry, [{{:_, :_, {1, :atom, :_}}, [], [true]}])
|
||||
assert 1 == Registry.count_select(registry, [{{:_, :_, {:"$1", :_, :"$1"}}, [], [true]}])
|
||||
assert 0 == Registry.count_select(registry, [{{"hello", :_, nil}, [], [true]}])
|
||||
|
||||
value2 = %{a: "a", b: "b"}
|
||||
{:ok, _} = Registry.register(registry, "world", value2)
|
||||
assert 1 == Registry.count_select(registry, [{{"world", :_, :_}, [], [true]}])
|
||||
end
|
||||
|
||||
test "count_select supports guard conditions", %{registry: registry} do
|
||||
value = {1, :atom, 2}
|
||||
{:ok, _} = Registry.register(registry, "hello", value)
|
||||
|
||||
assert 1 ==
|
||||
Registry.count_select(registry, [
|
||||
{{:_, :_, {:_, :"$1", :_}}, [{:is_atom, :"$1"}], [true]}
|
||||
])
|
||||
|
||||
assert 1 ==
|
||||
Registry.count_select(registry, [
|
||||
{{:_, :_, {:_, :_, :"$1"}}, [{:>, :"$1", 1}], [true]}
|
||||
])
|
||||
|
||||
assert 0 ==
|
||||
Registry.count_select(registry, [
|
||||
{{:_, :_, {:_, :_, :"$1"}}, [{:>, :"$1", 2}], [true]}
|
||||
])
|
||||
end
|
||||
|
||||
test "count_select allows multiple specs", %{registry: registry} do
|
||||
{:ok, _} = Registry.register(registry, "hello", :value)
|
||||
{:ok, _} = Registry.register(registry, "world", :value)
|
||||
|
||||
assert 2 ==
|
||||
Registry.count_select(registry, [
|
||||
{{"hello", :_, :_}, [], [true]},
|
||||
{{"world", :_, :_}, [], [true]}
|
||||
])
|
||||
end
|
||||
|
||||
test "count_select raises on incorrect shape of match spec", %{registry: registry} do
|
||||
assert_raise ArgumentError, fn ->
|
||||
Registry.count_select(registry, [{:_, [], []}])
|
||||
end
|
||||
end
|
||||
|
||||
test "doesn't grow ets on already_registered",
|
||||
%{registry: registry, partitions: partitions} do
|
||||
assert sum_pid_entries(registry, partitions) == 0
|
||||
|
||||
{:ok, pid} = Registry.register(registry, "hello", :value)
|
||||
assert is_pid(pid)
|
||||
assert sum_pid_entries(registry, partitions) == 1
|
||||
|
||||
{:ok, pid} = Registry.register(registry, "world", :value)
|
||||
assert is_pid(pid)
|
||||
assert sum_pid_entries(registry, partitions) == 2
|
||||
|
||||
assert {:error, {:already_registered, _pid}} =
|
||||
Registry.register(registry, "hello", :value)
|
||||
|
||||
assert sum_pid_entries(registry, partitions) == 2
|
||||
end
|
||||
|
||||
test "doesn't grow ets on already_registered across processes",
|
||||
%{registry: registry, partitions: partitions} do
|
||||
assert sum_pid_entries(registry, partitions) == 0
|
||||
|
||||
{_, task} = register_task(registry, "hello", :value)
|
||||
Process.link(Process.whereis(registry))
|
||||
|
||||
assert sum_pid_entries(registry, partitions) == 1
|
||||
|
||||
{:ok, pid} = Registry.register(registry, "world", :value)
|
||||
assert is_pid(pid)
|
||||
assert sum_pid_entries(registry, partitions) == 2
|
||||
|
||||
assert {:error, {:already_registered, ^task}} =
|
||||
Registry.register(registry, "hello", :recent)
|
||||
|
||||
assert sum_pid_entries(registry, partitions) == 2
|
||||
end
|
||||
|
||||
defp register_task(registry, key, value) do
|
||||
parent = self()
|
||||
|
||||
{:ok, task} =
|
||||
Task.start(fn ->
|
||||
send(parent, Registry.register(registry, key, value))
|
||||
Process.sleep(:infinity)
|
||||
end)
|
||||
|
||||
assert_receive {:ok, owner}
|
||||
{owner, task}
|
||||
end
|
||||
|
||||
defp kill_and_assert_down(pid) do
|
||||
ref = Process.monitor(pid)
|
||||
Process.exit(pid, :kill)
|
||||
assert_receive {:DOWN, ^ref, _, _, _}
|
||||
end
|
||||
|
||||
defp sum_pid_entries(registry, partitions) do
|
||||
Enum.sum_by(0..(partitions - 1), fn partition ->
|
||||
registry
|
||||
|> Module.concat("PIDPartition#{partition}")
|
||||
|> ets_entries()
|
||||
end)
|
||||
end
|
||||
|
||||
defp ets_entries(table_name) do
|
||||
:ets.all()
|
||||
|> Enum.find_value(fn id -> :ets.info(id, :name) == table_name and :ets.info(id, :size) end)
|
||||
end
|
||||
end
|
||||
File diff suppressed because it is too large
Load Diff
Reference in New Issue
Block a user