From 933c3503ccebe4717ac036cdf368653cdde20eaf Mon Sep 17 00:00:00 2001 From: DualFroz Date: Tue, 1 Sep 2026 12:48:56 +0200 Subject: [PATCH] feat: extend application with trackerSpec and Excache.Blocking --- lib/excache/application.ex | 15 +++- lib/excache/blocking.ex | 157 +++++++++++++++++++++++++++++++++++++ 2 files changed, 171 insertions(+), 1 deletion(-) create mode 100644 lib/excache/blocking.ex diff --git a/lib/excache/application.ex b/lib/excache/application.ex index 58e3792..bdb9b0f 100644 --- a/lib/excache/application.ex +++ b/lib/excache/application.ex @@ -11,8 +11,12 @@ defmodule Excache.Application do {Registry, keys: :duplicate, name: Excache.PubSub.Registry}, Excache.Store.Supervisor, Excache.Aof, + Excache.Blocking, Excache.Expiry, - {DynamicSupervisor, strategy: :one_for_one, name: Excache.Server.ConnectionSupervisor} + trackerSpec(), + Excache.Eviction.Runner, + {DynamicSupervisor, strategy: :one_for_one, name: Excache.Server.ConnectionSupervisor}, + Excache.Clients ] case Supervisor.start_link(children, strategy: :one_for_one, name: Excache.Supervisor) do @@ -25,6 +29,15 @@ defmodule Excache.Application do end end + def trackerSpec do + {Excache.Eviction.Tracker, + [ + name: Excache.Eviction.Tracker, + logFactor: Excache.Config.lfuLogFactor(), + decayTime: Excache.Config.lfuDecayTime() + ]} + end + defp start_acceptor(supervisor, port) when is_integer(port) do case Supervisor.start_child(supervisor, {Excache.Server.Acceptor, [port: port]}) do {:ok, _pid} -> {:ok, supervisor} diff --git a/lib/excache/blocking.ex b/lib/excache/blocking.ex new file mode 100644 index 0000000..eaf913c --- /dev/null +++ b/lib/excache/blocking.ex @@ -0,0 +1,157 @@ +defmodule Excache.Blocking do + @moduledoc false + + use GenServer + + alias Excache.Store + alias Excache.Command.Reply + + @name __MODULE__ + + def start_link(_arg), do: GenServer.start_link(__MODULE__, nil, name: @name) + + def acquire(request), do: GenServer.call(@name, {:acquire, request, :block}) + + def attempt(request) do + {:served, term} = GenServer.call(@name, {:acquire, request, :immediate}) + term + end + + def signal([]), do: [] + def signal(keys), do: GenServer.call(@name, {:ready, keys}) + + def takeWrites, do: GenServer.call(@name, :takeWrites) + + def waitingCount(key), do: GenServer.call(@name, {:waitingCount, key}) + + @impl true + def init(_arg), do: {:ok, %{queues: %{}, waiters: %{}, monitors: %{}, writes: []}} + + @impl true + def handle_call({:acquire, request, mode}, {pid, _tag}, state) do + waiter = %{ + ref: make_ref(), + pid: pid, + keys: request.keys, + op: request.op, + monitor: nil, + timer: nil + } + + state = enqueue(state, waiter) + {state, served} = drain(state, request.keys, []) + + case {deliver(served, waiter.ref), mode} do + {nil, :immediate} -> {:reply, {:served, missReply(request.op)}, dropWaiter(state, waiter)} + {nil, :block} -> {:reply, {:blocked, waiter.ref}, arm(state, waiter, request.timeout)} + {term, _mode} -> {:reply, {:served, term}, state} + end + end + + def handle_call({:ready, keys}, _from, state) do + {state, served} = drain(state, keys, []) + deliver(served, nil) + takeWrites(state) + end + + def handle_call(:takeWrites, _from, state), do: takeWrites(state) + + def handle_call({:waitingCount, key}, _from, state) do + count = + case Map.fetch(state.queues, key) do + {:ok, queue} -> :queue.len(queue) + :error -> 0 + end + + {:reply, count, state} + end + + @impl true + def handle_info({:blockingTimeout, ref}, state) do + case Map.fetch(state.waiters, ref) do + {:ok, waiter} -> + send(waiter.pid, {:excacheUnblocked, ref, {:array, nil}}) + {:noreply, dropWaiter(state, waiter)} + + :error -> + {:noreply, state} + end + end + + def handle_info({:DOWN, monitor, :process, _pid, _reason}, state) do + ref = Map.fetch!(state.monitors, monitor) + {:noreply, dropWaiter(state, Map.fetch!(state.waiters, ref))} + end + + defp drain(state, [], served), do: {state, served} + + defp drain(state, [key | rest], served) do + {state, served, chained} = drainKey(state, key, served, []) + drain(state, chained ++ rest, served) + end + + defp drainKey(state, key, served, chained) do + case headWaiter(state, key) do + nil -> + {state, served, chained} + + waiter -> + case serveWaiter(waiter) do + :empty -> + {state, served, chained} + + {:ok, term, ready, writes} -> + state = recordWrites(dropWaiter(state, waiter), writes) + drainKey(state, key, [{waiter, term} | served], ready ++ chained) + + {:error, term} -> + drainKey(dropWaiter(state, waiter), key, [{waiter, term} | served], chained) + end + end + end + + defp recordWrites(state, writes), do: %{state | writes: state.writes ++ writes} + + defp takeWrites(state), do: {:reply, state.writes, %{state | writes: []}} + + defp serveWaiter(waiter) do + Enum.reduce_while(waiter.keys, :empty, fn key, _pending -> + case serve(waiter.op, key) do + :empty -> {:cont, :empty} + outcome -> {:halt, outcome} + end + end) + end + + defp serve({:pop, side}, key) do + case popOne(key, side) do + {:ok, value} -> + {:ok, {:array, [{:bulk, key}, {:bulk, value}]}, [], [{popName(side), [key]}]} + + nil -> + :empty + + {:error, reason} -> + {:error, errorTerm(reason)} + end + end + + defp serve({:mpop, side, count}, key) do + case popMany(key, side, count) do + {:ok, [_ | _] = values} -> + taken = Integer.to_string(length(values)) + + {:ok, {:array, [{:bulk, key}, {:array, Reply.bulks(values)}]}, [], + [{popName(side), [key, taken]}]} + + {:ok, []} -> + :empty + + nil -> + :empty + + {:error, reason} -> + {:error, errorTerm(reason)} + end + end +