defmodule Phoenix.SocketClient.RouteCache do @moduledoc """ High-performance ETS-based caching for message routing. This module provides an ETS-based cache for storing and retrieving channel process PIDs by topic, significantly improving message routing performance compared to Registry lookups. ## Features - ETS-based O(1) lookups for channel routing - Automatic cache invalidation and cleanup - Memory-efficient storage with TTL support - Statistics and monitoring capabilities - Cache warming strategies ## Usage The route cache is automatically started by the socket supervisor and provides fast channel-to-process mapping for message routing. ## Performance Benefits - Eliminates Registry overhead for frequent lookups - Provides constant-time routing operations - Reduces process discovery latency - Handles high-frequency message routing efficiently """ use GenServer require Logger @default_cache_size 1000 # 5 minutes @default_ttl 300_000 # 1 minute @default_cleanup_interval 60_000 @typedoc """ Cache configuration options """ @type opts :: [ cache_size: non_neg_integer(), ttl: non_neg_integer(), cleanup_interval: non_neg_integer(), registry_name: atom() ] @typedoc """ Cache entry with metadata """ @type cache_entry :: %{ pid: pid(), timestamp: integer(), access_count: non_neg_integer(), last_access: integer() } @typedoc """ Route cache state """ @type t :: %__MODULE__{ cache_table: :ets.tid(), cache_size: non_neg_integer(), ttl: non_neg_integer(), cleanup_interval: non_neg_integer(), registry_name: atom() | nil, cleanup_timer: reference() | nil, stats: %{ hits: non_neg_integer(), misses: non_neg_integer(), evictions: non_neg_integer(), insertions: non_neg_integer(), current_size: non_neg_integer() } } defstruct [ :cache_table, :cache_size, :ttl, :cleanup_interval, :registry_name, :cleanup_timer, stats: %{ hits: 0, misses: 0, evictions: 0, insertions: 0, current_size: 0 } ] @doc """ Starts the route cache server. ## Options * `:cache_size` - Maximum number of cached routes (default: 1000) * `:ttl` - Time-to-live for cache entries in milliseconds (default: 300000) * `:cleanup_interval` - Cleanup interval for expired entries (default: 60000) * `:registry_name` - Registry name for fallback lookups """ @spec start_link(opts()) :: GenServer.on_start() def start_link(opts \\ []) do opts = if Keyword.keyword?(opts), do: opts, else: Enum.into(opts, []) registry_name = Keyword.get(opts, :registry_name) name = route_cache_name(registry_name) GenServer.start_link(__MODULE__, opts, name: name) end defp route_cache_name(nil), do: nil defp route_cache_name(registry_name) do if Process.whereis(registry_name) do {:via, Registry, {registry_name, :route_cache}} end end @doc """ Gets a channel PID from the cache. Returns {:ok, pid} if found and valid, :error otherwise. """ @spec get(pid(), String.t()) :: {:ok, pid()} | :error def get(cache_pid, topic) when is_pid(cache_pid) and is_binary(topic) do GenServer.call(cache_pid, {:get, topic}, 1000) end @doc """ Puts a route mapping in the cache. Associates a topic with a channel PID. """ @spec put(pid(), String.t(), pid()) :: :ok def put(cache_pid, topic, channel_pid) when is_pid(cache_pid) and is_binary(topic) and is_pid(channel_pid) do GenServer.cast(cache_pid, {:put, topic, channel_pid}) end @doc """ Removes a route from the cache. Useful when a channel leaves or terminates. """ @spec delete(pid(), String.t()) :: :ok def delete(cache_pid, topic) when is_pid(cache_pid) and is_binary(topic) do GenServer.cast(cache_pid, {:delete, topic}) end @doc """ Clears all cached routes. Useful for cache invalidation or testing. """ @spec clear(pid()) :: :ok def clear(cache_pid) when is_pid(cache_pid) do GenServer.cast(cache_pid, :clear) end @doc """ Gets cache statistics for monitoring. Returns detailed performance and usage statistics. """ @spec stats(pid()) :: map() def stats(cache_pid) when is_pid(cache_pid) do GenServer.call(cache_pid, :stats, 1000) end @doc """ Warms up the cache with common routes. Pre-populates the cache with known channel mappings. """ @spec warm_up(pid(), [{String.t(), pid()}]) :: :ok def warm_up(cache_pid, routes) when is_pid(cache_pid) and is_list(routes) do GenServer.call(cache_pid, {:warm_up, routes}, 10_000) end @doc """ Checks if a topic is cached. Returns true if the topic exists in cache, false otherwise. """ @spec cached?(pid(), String.t()) :: boolean() def cached?(cache_pid, topic) when is_pid(cache_pid) and is_binary(topic) do GenServer.call(cache_pid, {:cached?, topic}, 1000) end # GenServer callbacks @impl true def init(opts) do cache_size = Keyword.get(opts, :cache_size, @default_cache_size) ttl = Keyword.get(opts, :ttl, @default_ttl) cleanup_interval = Keyword.get(opts, :cleanup_interval, @default_cleanup_interval) registry_name = Keyword.get(opts, :registry_name) # Create ETS table for caching with unique name table_name = :"phoenix_route_cache_#{System.unique_integer([:positive])}" cache_table = :ets.new(table_name, [ :set, :public, :named_table, {:read_concurrency, true}, {:write_concurrency, true} ]) # Start cleanup timer {:ok, timer} = :timer.send_interval(cleanup_interval, :cleanup) state = %__MODULE__{ cache_table: cache_table, cache_size: cache_size, ttl: ttl, cleanup_interval: cleanup_interval, registry_name: registry_name, cleanup_timer: timer } Phoenix.SocketClient.Telemetry.optimization(:route_cache_started, %{ cache_size: cache_size, ttl: ttl }) {:ok, state} end @impl true def handle_call({:get, topic}, _from, state) do current_time = System.monotonic_time(:millisecond) case :ets.lookup(state.cache_table, topic) do [{^topic, entry}] -> # Check if entry is still valid if current_time - entry.timestamp <= state.ttl and Process.alive?(entry.pid) do # Update access statistics updated_entry = %{ entry | access_count: entry.access_count + 1, last_access: current_time } :ets.insert(state.cache_table, {topic, updated_entry}) new_stats = %{state.stats | hits: state.stats.hits + 1} new_state = %{state | stats: new_stats} {:reply, {:ok, entry.pid}, new_state} else # Entry expired or process dead, remove it :ets.delete(state.cache_table, topic) current_size = :ets.info(state.cache_table, :size) new_stats = %{state.stats | misses: state.stats.misses + 1, current_size: current_size} new_state = %{state | stats: new_stats} {:reply, :error, new_state} end [] -> # Not found in cache new_stats = %{state.stats | misses: state.stats.misses + 1} new_state = %{state | stats: new_stats} {:reply, :error, new_state} end end @impl true def handle_call(:stats, _from, state) do current_size = :ets.info(state.cache_table, :size) memory = :ets.info(state.cache_table, :memory) stats = Map.merge(state.stats, %{ current_size: current_size, # ETS memory is in words memory_bytes: memory * 8, cache_table_info: %{ size: current_size, memory: memory, owner: :ets.info(state.cache_table, :owner), protection: :ets.info(state.cache_table, :protection) } }) {:reply, stats, state} end @impl true def handle_call({:warm_up, routes}, _from, state) do current_time = System.monotonic_time(:millisecond) _insertions_count = 0 {inserted_count, new_state} = Enum.reduce(routes, {0, state}, fn {topic, pid}, {count, acc_state} -> if Process.alive?(pid) and count < acc_state.cache_size do entry = %{ pid: pid, timestamp: current_time, access_count: 0, last_access: current_time } :ets.insert(acc_state.cache_table, {topic, entry}) {count + 1, acc_state} else {count, acc_state} end end) current_size = :ets.info(state.cache_table, :size) new_stats = %{ state.stats | insertions: state.stats.insertions + inserted_count, current_size: current_size } final_state = %{new_state | stats: new_stats} Phoenix.SocketClient.Telemetry.optimization(:route_cache_warmup, %{ routes_inserted: inserted_count }) {:reply, :ok, final_state} end @impl true def handle_call({:cached?, topic}, _from, state) do cached = case :ets.lookup(state.cache_table, topic) do [{^topic, _entry}] -> true [] -> false end {:reply, cached, state} end @impl true def handle_cast({:put, topic, channel_pid}, state) do current_time = System.monotonic_time(:millisecond) current_size = :ets.info(state.cache_table, :size) # Check if we need to evict entries new_state = if current_size >= state.cache_size do evict_lru_entries(state) else state end entry = %{ pid: channel_pid, timestamp: current_time, access_count: 0, last_access: current_time } :ets.insert(new_state.cache_table, {topic, entry}) final_size = :ets.info(new_state.cache_table, :size) new_stats = %{ new_state.stats | insertions: new_state.stats.insertions + 1, current_size: final_size } final_state = %{new_state | stats: new_stats} {:noreply, final_state} end @impl true def handle_cast({:delete, topic}, state) do :ets.delete(state.cache_table, topic) current_size = :ets.info(state.cache_table, :size) new_stats = %{state.stats | current_size: current_size} new_state = %{state | stats: new_stats} {:noreply, new_state} end @impl true def handle_cast(:clear, state) do :ets.delete_all_objects(state.cache_table) new_stats = %{state.stats | current_size: 0} new_state = %{state | stats: new_stats} Phoenix.SocketClient.Telemetry.optimization(:route_cache_cleared, %{}) {:noreply, new_state} end @impl true def handle_info(:cleanup, state) do current_time = System.monotonic_time(:millisecond) ttl = state.ttl # Remove expired entries and dead processes {removed_count, evicted_count} = :ets.tab2list(state.cache_table) |> Enum.reduce({0, 0}, fn {topic, entry}, {expired, dead} -> age = current_time - entry.timestamp cond do age > ttl -> :ets.delete(state.cache_table, topic) {expired + 1, dead} not Process.alive?(entry.pid) -> :ets.delete(state.cache_table, topic) {expired, dead + 1} true -> {expired, dead} end end) total_removed = removed_count + evicted_count current_size = :ets.info(state.cache_table, :size) if total_removed > 0 do Phoenix.SocketClient.Telemetry.optimization(:route_cache_cleanup, %{ total_removed: total_removed, expired_removed: removed_count, dead_removed: evicted_count, cache_size: :ets.info(state.cache_table, :size) }) end new_stats = %{ state.stats | evictions: state.stats.evictions + evicted_count, current_size: current_size } new_state = %{state | stats: new_stats} {:noreply, new_state} end @impl true def terminate(_reason, state) do if state.cleanup_timer do :timer.cancel(state.cleanup_timer) end entries_cleared = :ets.info(state.cache_table, :size) :ets.delete(state.cache_table) Phoenix.SocketClient.Telemetry.optimization(:route_cache_terminating, %{ entries_cleared: entries_cleared }) :ok end # Private helper functions defp evict_lru_entries(state) do # Evict least recently used entries # Evict to 50% capacity target_size = div(state.cache_size, 2) entries = :ets.tab2list(state.cache_table) |> Enum.sort_by(fn {_topic, entry} -> entry.last_access end) {to_evict, _to_keep} = Enum.split(entries, length(entries) - target_size) evicted_count = Enum.reduce(to_evict, 0, fn {topic, _entry}, count -> :ets.delete(state.cache_table, topic) count + 1 end) Phoenix.SocketClient.Telemetry.optimization(:route_cache_lr_eviction, %{ evicted_count: evicted_count, target_size: target_size, cache_size: :ets.info(state.cache_table, :size) }) new_stats = %{state.stats | evictions: state.stats.evictions + evicted_count} %{state | stats: new_stats} end end