defmodule Globals do use GenServer require Logger @moduledoc """ Globals is a module that provides ETS backed global varaibles with PubSub capabilities. How this is different from other PubSub implementations: - Support PubSub patterns and value storage together - keys can be any() term such as atoms, strings, tuples, etc. - Application starts `Globals` automatically no additional Supervision tree entry required - Support for `depends_on` for variable dependencies - `await` and `update_await` for awaiting async updates """ defstruct [ # subscribers is a map of {key => {pid => monitor_ref}} subscribers: %{}, # refreshers is a map of {key => fun} refreshers: %{}, # dependencies is a map of {key => [key]} dependencies: %{}, # monitors is a map of {pid => {monitor_ref, [key]}} monitors: %{}, # waiting is a map of {key => [{generation, pid, timer}]} waiting: %{}, # Generation tracks update_await() call origin times. So that a update_await() call # can only be responded to using an update that has be started at or after the update_await() call, # but not from an previous update that is still running. generation: 0 ] @timeout 30_000 def start_link(_args) do GenServer.start_link(__MODULE__, [], name: __MODULE__, hibernate_after: 5000) end @impl true def init([]) do :ets.new(__MODULE__, [:public, :named_table, read_concurrency: true]) {:ok, %Globals{}} end @doc """ get returns the value of the key - key is the name of the key and can be any() term - default is the default value to return if the key is not found Example: iex> Globals.set(:key11, "value") iex> Globals.get(:key11) "value" iex> Globals.get(:key12) nil iex> Globals.get(:key13, "default") "default" """ def get(key, default \\ nil) do case :ets.lookup(__MODULE__, key) do [] -> default [{_, value}] -> value end end @doc """ get_lazy returns the value of the key or calls the function to get the value if the key is not found Example: iex> Globals.get_lazy(:key14, fn -> "value" end) "value" iex> Globals.set(:key14, false) false iex> Globals.get_lazy(:key14, fn -> "value" end) false iex> Globals.set(:key14, nil) nil iex> Globals.get_lazy(:key14, fn -> "value" end) "value" """ def get_lazy(key, fun) do case get(key) do nil -> fun.() value -> value end end @doc """ get_g returns the value of the key and the generation The generation is a monotonically increasing number that is incremented every time a value is updated. If g has changed then there was a change to the value. """ def get_g(key, default \\ nil) do call({:get_g, key, default}) end def update_await(key, opts \\ []) do timeout = Keyword.get(opts, :timeout) call({:update_await, key}, timeout) end @doc """ await is similiar to get() but will block for timeout until another process defines the key or the timeout expires. Example: iex> Globals.set(:key, "value") iex> Globals.await(:key) "value" iex> Globals.update(:key2, fn -> ...> Process.sleep(100) ...> "value2" ...> end) iex> Globals.await(:key2) "value2" """ def await(key, timeout \\ :infinity) do get_lazy(key, fn -> call({:await, key}, timeout) end) end @doc """ pop returns the value of the key and deletes the key Example: iex> Globals.set(:key, "value") iex> Globals.pop(:key) "value" iex> Globals.get(:key) nil """ def pop(key, default \\ nil) do case get(key) do nil -> default value -> :ets.delete(__MODULE__, key) GenServer.cast(__MODULE__, :pop) value end end @doc """ incr increments the value of the key and returns the previous value Example: iex> Globals.incr(:key) 0 iex> Globals.incr(:key) 1 iex> Globals.get(:key) 2 """ def incr(key, default \\ 0) when is_number(default) do call({:incr, key, default}) end @doc """ put_new sets the new value and returns true if the key was not already set Example: iex> Globals.put_new(:key5, "value") true iex> Globals.put_new(:key6, "value1") true iex> Globals.put_new(:key6, "value2") false iex> Globals.get(:key6) "value1" """ def put_new(key, value) do if :ets.insert_new(__MODULE__, {key, value}) do GenServer.cast(__MODULE__, {:put, key, nil, value, nil}) true else false end end @doc """ put sets the new value and returns the OLD value Example: iex> Globals.set(:key, "value") iex> Globals.put(:key, "new_value") "value" """ def put(key, value, src_gen \\ nil) do old = get(key, nil) :ets.insert(__MODULE__, {key, value}) GenServer.cast(__MODULE__, {:put, key, old, value, src_gen}) old end @doc """ pushes a new value to all subscribers without setting it if a value is already set, it is deleted returns the value Example: iex> Globals.set(:key, "value") iex> Globals.push(:key, "new_value") "new_value" iex> Globals.get(:key) nil """ def push(key, value) do call({:push, key, value}) end @doc """ set sets the new value and returns the NEW value Example: iex> Globals.set(:key, "old_value") "old_value" iex> Globals.set(:key, "new_value") "new_value" """ def set(key, value) do put(key, value) value end @doc """ update triggers an update to the key with the registered function in a background process Example: iex> Globals.register(:key, fn -> "new_value" end) iex> Globals.update_await(:key) iex> Globals.get(:key) "new_value" """ def update(key) do Debouncer.immediate( {__MODULE__, key}, fn -> {fun, gen} = call({:update, key}) if fun != nil, do: do_update(key, fun, gen) end, update_timeout() ) end @doc """ update updates the key with the given function in a background process Example: iex> Globals.update(:key23, fn -> "new_value" end) iex> Process.sleep(100) iex> Globals.get(:key23) "new_value" """ def update(key, fun) do Debouncer.immediate( {__MODULE__, key}, fn -> do_update(key, fun) end, update_timeout() ) end @doc """ Update_timeout returns the debounce timeout for the update function. When calling `Globals.update()` or `Globals.update_await()` the function will be debounced for the timeout. Example: iex> Globals.update_timeout() 1000 """ def update_timeout() do :persistent_term.get(:globals_update_timeout, 1000) end @doc """ Set_update_timeout sets the debounce timeout for the update function. Example: iex> Globals.set_update_timeout(2000) :ok """ def set_update_timeout(timeout) when is_integer(timeout) do :persistent_term.put(:globals_update_timeout, timeout) end defp do_update(key, fun, gen \\ nil) do {old, other_gen} = get_g(key) gen = gen || other_gen new = case fun do {module, fun, args} -> apply(module, fun, args) no_args when is_function(fun, 0) -> no_args.() one_arg when is_function(fun, 1) -> one_arg.(old) end old2 = put(key, new, gen) if old2 != old do Logger.warning("update conflict in #{inspect(key)}") end new end @doc """ Register a key and function to be called when the key is updated. If the function is not provided, the key must be a tuple of {module, function, args}. - key is the name of the key and can be any() term - fun is the function to call to get the value, can be a tuple of {module, function, args}, a function of 0 arguments, or a function of 1 argument (the old value) - opts is a keyword list of options - opts.depends_on is a list of keys that the key depends on - opts.changes is a list of keys that the key changes - opts.await is a boolean to wait for the key to be updated Example: # Registering key with a function iex> Globals.register(:one, fn -> "one" end) iex> Globals.register(:time, fn -> System.os_time() end) iex> Globals.register(:time, {System, :os_time, []}) # Registering where key == {module, function, args} and acts as update function as well: iex> Globals.register({System, :os_time, []}) # Depends on and changes: iex> Globals.register(:a, fn -> "a" end) iex> Globals.register(:b, fn -> Globals.await(:a) <> "b" end) iex> Globals.await(:a) "a" iex> Globals.await(:b) "ab" """ def register(key, fun \\ nil, opts \\ []) do {opts, fun} = if opts == [] and is_list(fun) do {fun, nil} else {opts, fun} end fun = if fun == nil do if not is_tuple(key) and tuple_size(key) != 3 do raise "Need to provide a function for key: #{inspect(key)} or key must be a tuple of {module, function, args}" end key else fun end if !call({:register, key, fun, opts}) do update(key, fun) end if Keyword.get(opts, :await, false) do await(key) end end @doc """ subscribe subscribes to the given keys - key(s) can be a single key or a list of keys - whom is the pid that will receive the updates. defaults to self() The `whom` pid will receives updates in the form of `{:update, key, value}` Example: iex> Globals.subscribe(:key21) :ok iex> Globals.subscribe([:key22, :key23]) :ok iex> Globals.push(:key21, "value") "value" iex> receive do ...> {:update, :key21, "value"} -> :ok ...> end :ok """ def subscribe(keys, whom \\ self()) do List.wrap(keys) |> Enum.each(fn key -> GenServer.cast(__MODULE__, {:subscribe, whom, key}) end) :ok end @doc """ subscribed? returns true if the given key is subscribed to by the given whom - key is the name of the key and can be any() term - whom is the pid that will receive the updates Example: iex> Globals.subscribe(:key24) iex> Globals.subscribed?(:key24) true iex> Globals.subscribed?(:key25) false """ def subscribed?(key, whom \\ self()) do call({:subscribed?, key, whom}) end @doc """ unsubscribe unsubscribes from the given keys - key(s) can be a single key or a list of keys - whom is the pid that will receive the updates """ def unsubscribe(keys, whom \\ self()) do call({:unsubscribe, List.wrap(keys), whom}) end defp do_unsubscribe(g = %Globals{subscribers: subs, monitors: mons}, :all, whom) do {ref, remove_keys} = Map.get(mons, whom) Process.demonitor(ref) mons = Map.delete(mons, whom) subs = do_unsubscribe_key_subs(subs, Enum.to_list(remove_keys), whom) %Globals{g | subscribers: subs, monitors: mons} end defp do_unsubscribe(g = %Globals{subscribers: subs, monitors: mons}, remove_keys, whom) do {ref, keys} = Map.get(mons, whom) new_keys = Enum.reduce(remove_keys, keys, &MapSet.delete(&2, &1)) mons = if MapSet.size(new_keys) == 0 do Process.demonitor(ref) Map.delete(mons, whom) else Map.put(mons, whom, new_keys) end subs = do_unsubscribe_key_subs(subs, Enum.to_list(remove_keys), whom) %Globals{g | subscribers: subs, monitors: mons} end defp do_unsubscribe_key_subs(subs, [], _whom) do subs end defp do_unsubscribe_key_subs(subs, [key | rest], whom) do {_ref, key_subs} = Map.get(subs, key, %{}) |> Map.pop(whom) if map_size(key_subs) == 0 do Map.delete(subs, key) else Map.put(subs, key, key_subs) end |> do_unsubscribe_key_subs(rest, whom) end @impl true def handle_cast({:put, key, old, value, src_gen}, g = %Globals{generation: generation}) do g = %Globals{g | generation: generation + 1} {:noreply, publish(g, key, value, old, src_gen)} end def handle_cast(:pop, g = %Globals{generation: generation}) do {:noreply, %Globals{g | generation: generation + 1}} end def handle_cast({:subscribe, whom, key}, g = %Globals{subscribers: subs, monitors: mons}) when is_pid(whom) do {ref, keys} = case Map.get(mons, whom) do nil -> {Process.monitor(whom), MapSet.new([key])} {ref, keys} -> {ref, MapSet.put(keys, key)} end mons = Map.put(mons, whom, {ref, keys}) key_subs = Map.get(subs, key, %{}) |> Map.put_new(whom, ref) subs = Map.put(subs, key, key_subs) {:noreply, %Globals{g | subscribers: subs, monitors: mons}} end def handle_cast({:subscribe, mfa, key}, g = %Globals{subscribers: subs}) do key_subs = Map.get(subs, key, %{}) |> Map.put_new(mfa, mfa) subs = Map.put(subs, key, key_subs) {:noreply, %Globals{g | subscribers: subs}} end @impl true def handle_call({:unsubscribe, keys, whom}, _from, g = %Globals{}) do {:reply, :ok, do_unsubscribe(g, keys, whom)} end def handle_call({:subscribed?, key, whom}, _from, g = %Globals{}) do {:reply, Map.has_key?(Map.get(g.subscribers, key, %{}), whom), g} end def handle_call( {:register, key, fun, opts}, _from, g = %Globals{refreshers: refs, dependencies: deps} ) do depends_on = List.wrap(Keyword.get(opts, :depends_on, [])) changes = List.wrap(Keyword.get(opts, :changes, [])) deps = Enum.reduce(depends_on, deps, fn dep, deps -> Map.put(deps, dep, Map.get(deps, dep, []) ++ [key]) end) deps = Enum.reduce(changes, deps, fn dst, deps -> Map.put(deps, key, Map.get(deps, key, []) ++ [dst]) end) {:reply, Map.has_key?(refs, key), %Globals{g | dependencies: deps, refreshers: Map.put(refs, key, fun)}} end def handle_call( {:await, key}, from, g = %Globals{waiting: waiting, refreshers: refs} ) do case get(key) do nil -> timer = Process.send_after(self(), {:timeout, key, from}, @timeout - 2000) waiting = Map.update(waiting, key, [{nil, from, timer}], fn pids -> [{nil, from, timer} | pids] end) if Map.has_key?(refs, key) do update(key) else Logger.info("Awaiting undefined key: #{inspect(key)}") end {:noreply, %Globals{g | waiting: waiting}} value -> {:reply, value, g} end end def handle_call({:incr, key, default}, _from, g = %Globals{generation: generation}) do prev_value = case get(key, default) do prev_value when is_number(prev_value) -> prev_value _other -> default end value = prev_value + 1 :ets.insert(__MODULE__, {key, value}) g = %Globals{g | generation: generation + 1} {:reply, prev_value, publish(g, key, value, nil, nil)} end def handle_call({:push, key, value}, _from, g = %Globals{generation: generation}) do :ets.delete(__MODULE__, key) g = %Globals{g | generation: generation + 1} {:reply, value, publish(g, key, value, nil, nil)} end def handle_call({:get_g, key, default}, _from, g = %Globals{generation: generation}) do value = get(key, default) {:reply, {value, generation}, g} end def handle_call({:update_await, key}, from, g = %Globals{waiting: waiting, generation: gen}) do timer = Process.send_after(self(), {:timeout, key, from}, @timeout - 2000) waiting = Map.update(waiting, key, [{gen, from, timer}], fn pids -> [{gen, from, timer} | pids] end) update(key) {:noreply, %Globals{g | waiting: waiting}} end def handle_call({:update, key}, _from, g = %Globals{refreshers: refs, generation: gen}) do {:reply, {Map.get(refs, key), gen}, g} end @impl true def handle_info({:DOWN, _ref, :process, pid, _reason}, g = %Globals{}) do {:noreply, do_unsubscribe(g, :all, pid)} end def handle_info({:timeout, key, from}, g = %Globals{waiting: waiting}) do froms = Map.get(waiting, key, []) zombie? = Enum.all?(froms, fn {_gen, f, _timer} -> f != from end) # Spawn a new process to handle the timeout reporting # this protects Globals from waiting for the worker() and stracktrace() calls spawn(fn -> pid = Debouncer.worker({__MODULE__, key}) if pid == nil do Logger.error("Timeout waiting for #{inspect(key)} (zombie=#{zombie?}) - No worker found") else Logger.error("Timeout waiting for #{inspect(key)} (zombie=#{zombie?}) - \ Worker stuck in: #{inspect(Profiler.stacktrace(pid))}") end end) {:noreply, g} end defp call(what, timeout \\ @timeout) do GenServer.call(__MODULE__, what, timeout || @timeout) end defp publish( g = %Globals{subscribers: subs, waiting: waiting, dependencies: deps}, key, value, old, src_gen ) do if old == nil or value != old do pids = Map.keys(Map.get(subs, key, %{})) Debouncer.immediate( {__MODULE__, :publish, key, value}, fn -> {pids, mfas} = Enum.split_with(pids, &is_pid/1) for pid <- pids do send(pid, {:update, key, value}) end for fun <- mfas do case fun do {module, fun, args} -> apply(module, fun, args) no_args when is_function(fun, 0) -> no_args.() one_arg when is_function(fun, 1) -> one_arg.(value) two_arg when is_function(fun, 2) -> two_arg.(key, value) end end end, 500 ) for dep_key <- Map.get(deps, key, []) do update(dep_key) end end list = Map.get(waiting, key, []) |> Enum.reject(fn {gen, from, timer} -> if src_gen == nil or gen == nil or gen <= src_gen do Process.cancel_timer(timer) GenServer.reply(from, value) true end end) waiting = if list == [], do: Map.delete(waiting, key), else: Map.put(waiting, key, list) %Globals{g | waiting: waiting} end @doc """ Locked locks the given lockname and calls the given function with the lock held until the function returns. This is useful for ensuring that only one process can update the key at a time. - lockname is the name of the lock and can be any() term Example: iex> Globals.locked(:lockname, fn -> "new_value" end) "new_value" """ def locked(lockname, fun) do :global.trans({lockname, self()}, fun) end @doc """ caches the given key with the given function. If timeout is not provided, the value is cached for 5 seconds. - key is the name of the key and can be any() term - fun is the function to call to get the value - timeout is the timeout in milliseconds or :infinity to cache forever Example: iex> Globals.cache(:cache_key, fn -> "new_value" end, 1000) "new_value" iex> Globals.get(:cache_key) "new_value" iex> Process.sleep(2000) :ok iex> Globals.get(:cache_key) nil """ def cache(key, fun, timeout \\ 5_000) do value = get_lazy(key, fn -> locked({__MODULE__, :cache, key}, fn -> get(key) || set(key, fun.()) end) end) if timeout != :infinity do Debouncer.apply( {__MODULE__, :cache, key}, fn -> pop(key) end, timeout ) end value end end