%% Copyright (c) 2026 Benoit Chesneau. Licensed under the MIT License.
%% See the LICENSE file at the project root.
%%
-module(erllama_cache_meta_srv).
-moduledoc """
Sole writer for the cache meta and LRU ETS tables; arbitrates
claim/release and the reservation state machine for save
publication.
Two read-mostly ETS tables, owned by this process and `protected`
so any caller can read them without a server hop:
erllama_cache_meta : set, key = cache_key, row layout per
include/erllama_cache.hrl ?POS_* constants
erllama_cache_lru : ordered_set, key = {LastUsedNs, cache_key},
value = []
Two server-internal maps in process state:
holders : MonRef -> {Pid, Key}; one entry per active claim
reservations : Key -> #reservation{}; one entry per in-flight save
Plus a waiters map for `lookup_exact_or_wait/2` which defers replies
until the in-flight save publishes (or the per-call deadline fires).
The reservation state machine has two stages, `pre_link` and
`post_link`, to make crash cleanup correct: a writer that died
before `link/2` leaves no file; a writer that died after `link/2`
may have left a valid `.kvc` we can validate-and-adopt.
""".
-behaviour(gen_server).
-include("erllama_cache.hrl").
-export([
start_link/0,
%% Read-only (no server hop)
lookup_exact/1,
lookup_longest_prefix/4,
%% Read with bounded wait for an in-flight save
lookup_exact_or_wait/2,
%% Claim/release (active reader of a slab)
checkout/2,
checkin/1,
%% Reservation state machine (writer)
reserve_save/3,
check_reservation/2,
mark_published/3,
announce_saved/4,
announce_saved/5,
cancel_reservation/2,
%% Operator/test helpers
gc/0,
evict_bytes/1,
evict_bytes/2,
dump/0,
dump/1,
insert_available/5,
insert_available/6
]).
-export([init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2]).
-define(SERVER, ?MODULE).
-define(TBL_META, erllama_cache_meta).
-define(TBL_LRU, erllama_cache_lru).
%% Default time-to-live for a reservation when the writer is silent.
%% Refreshed on each successful state transition (reserve / check /
%% mark_published / announce_saved).
-define(DEFAULT_TTL_NS, 60 * 1000 * 1000 * 1000).
%% Periodic sweep for stale reservations and expired waiters.
-define(SWEEP_INTERVAL_MS, 30 * 1000).
-record(reservation, {
writer :: pid(),
token :: reference(),
monref :: reference(),
expires_ns :: integer(),
stage :: pre_link | post_link,
tier :: erllama_cache:tier(),
path :: file:name() | undefined
}).
-record(state, {
holders :: #{reference() => {pid(), erllama_cache:cache_key()}},
reservations :: #{erllama_cache:cache_key() => #reservation{}},
waiters ::
#{erllama_cache:cache_key() => [{gen_server:from(), integer(), reference()}]},
sweep_timer :: reference() | undefined
}).
-type state() :: #state{}.
%% =============================================================================
%% Public API
%% =============================================================================
-spec start_link() -> {ok, pid()} | {error, term()}.
start_link() ->
gen_server:start_link({local, ?SERVER}, ?MODULE, [], []).
-spec lookup_exact(erllama_cache:cache_key()) ->
{ok, tuple()} | miss.
lookup_exact(Key) ->
%% Single ets:lookup; an eviction between the historical two-call
%% pattern (lookup_element + lookup) would crash the match.
try ets:lookup(?TBL_META, Key) of
[Row] when element(?POS_STATUS, Row) =:= available ->
{ok, Row};
_ ->
miss
catch
error:badarg -> miss
end.
-spec lookup_exact_or_wait(erllama_cache:cache_key(), non_neg_integer()) ->
{ok, tuple()} | miss.
lookup_exact_or_wait(Key, MaxWaitMs) ->
gen_server:call(?SERVER, {lookup_or_wait, Key, MaxWaitMs}, infinity).
-doc """
Walk Tokens backward in Stride steps and return the row for the
longest cached prefix. Pure ETS reads, no server hop. Used by
stateless callers (HTTP front-end, agent loops) that resend a full
conversation each turn and don't have a parent_key to thread.
Stops at MinTokens floor and returns `miss` if nothing matches.
Walks at most `length(Tokens) / Stride` rows; with the default
2048-token stride that's ~15 lookups for a 30k-token prompt.
Encodes the full token list to a binary once at entry, then passes
`binary:part(TokensBin, 0, N*4)` sub-binaries (O(1) views) to
`erllama_cache_key:make/4` per probe. Avoids re-traversing the list
and re-allocating the binary on every step; the only per-probe cost
is the SHA-256 over N*4 bytes plus the ETS lookup.
""".
-spec lookup_longest_prefix(map(), [erllama_nif:token_id()], pos_integer(), pos_integer()) ->
{ok, pos_integer(), tuple()} | miss.
lookup_longest_prefix(KeyMeta, Tokens, Stride, MinTokens) when
is_integer(Stride), Stride > 0, is_integer(MinTokens), MinTokens > 0
->
T0 = erlang:monotonic_time(nanosecond),
Len = length(Tokens),
Start = (Len div Stride) * Stride,
Floor = max(MinTokens, Stride),
AllTokensBin = erllama_cache_key:encode_tokens(Tokens),
#{fingerprint := Fp, quant_type := QT, ctx_params_hash := CtxHash} = KeyMeta,
Result = walk_prefix(Fp, QT, CtxHash, AllTokensBin, Start, Stride, Floor, 0),
Elapsed = erlang:monotonic_time(nanosecond) - T0,
erllama_cache_counters:add(?C_LONGEST_PREFIX_NS, max(Elapsed, 0)),
Result.
walk_prefix(_Fp, _QT, _CtxHash, _Bin, N, _Stride, Floor, Probes) when N < Floor ->
erllama_cache_counters:add(?C_LONGEST_PREFIX_PROBES, Probes),
miss;
walk_prefix(Fp, QT, CtxHash, Bin, N, Stride, Floor, Probes) ->
PrefixBin = binary:part(Bin, 0, N * 4),
Key = erllama_cache_key:make(Fp, QT, CtxHash, PrefixBin),
case lookup_exact(Key) of
{ok, Row} ->
erllama_cache_counters:add(?C_LONGEST_PREFIX_PROBES, Probes + 1),
{ok, N, Row};
miss ->
walk_prefix(Fp, QT, CtxHash, Bin, N - Stride, Stride, Floor, Probes + 1)
end.
-spec checkout(erllama_cache:cache_key(), pid()) ->
{ok, reference(), erllama_cache:tier(), term(), binary(), term()}
| {error, busy}
| miss.
checkout(Key, Pid) when is_pid(Pid) ->
gen_server:call(?SERVER, {checkout, Key, Pid}).
-spec checkin(reference()) -> ok.
checkin(MonRef) when is_reference(MonRef) ->
gen_server:call(?SERVER, {checkin, MonRef}).
-spec reserve_save(erllama_cache:cache_key(), erllama_cache:tier(), pid()) ->
{ok, reference()} | {error, already_present | conflict}.
reserve_save(Key, Tier, Pid) when is_pid(Pid) ->
gen_server:call(?SERVER, {reserve_save, Key, Tier, Pid}).
-spec check_reservation(erllama_cache:cache_key(), reference()) ->
ok | {error, expired}.
check_reservation(Key, Token) when is_reference(Token) ->
gen_server:call(?SERVER, {check_reservation, Key, Token}).
-spec mark_published(erllama_cache:cache_key(), reference(), file:name()) ->
ok | {error, expired}.
mark_published(Key, Token, Path) when is_reference(Token) ->
gen_server:call(?SERVER, {mark_published, Key, Token, Path}).
-spec announce_saved(
erllama_cache:cache_key(), reference(), non_neg_integer(), binary()
) -> ok | {error, expired}.
announce_saved(Key, Token, Size, Header) ->
announce_saved(Key, Token, Size, Header, undefined).
-spec announce_saved(
erllama_cache:cache_key(),
reference(),
non_neg_integer(),
binary(),
binary() | undefined
) -> ok | {error, expired}.
announce_saved(Key, Token, Size, Header, TokensBin) when
is_reference(Token), is_binary(Header)
->
gen_server:call(?SERVER, {announce_saved, Key, Token, Size, Header, TokensBin}).
-spec cancel_reservation(erllama_cache:cache_key(), reference()) -> ok.
cancel_reservation(Key, Token) when is_reference(Token) ->
gen_server:call(?SERVER, {cancel_reservation, Key, Token}).
%% Direct insertion of an `available` row. Used by the RAM tier
%% (which has no on-disk publish step) and by the disk tier on-start
%% scan to register pre-existing valid files.
-spec insert_available(
erllama_cache:cache_key(),
erllama_cache:tier(),
non_neg_integer(),
binary(),
term()
) -> ok.
insert_available(Key, Tier, Size, Header, Location) ->
insert_available(Key, Tier, Size, Header, Location, undefined).
-spec insert_available(
erllama_cache:cache_key(),
erllama_cache:tier(),
non_neg_integer(),
binary(),
term(),
binary() | undefined
) -> ok.
insert_available(Key, Tier, Size, Header, Location, TokensBin) ->
gen_server:call(
?SERVER, {insert_available, Key, Tier, Size, Header, Location, TokensBin}
).
-spec gc() -> {evicted, non_neg_integer()}.
gc() ->
gen_server:call(?SERVER, gc).
-doc """
Evict oldest available rows until at least TargetBytes have been
freed (or no more candidates remain). Returns the number of rows
evicted and the bytes actually freed.
""".
-spec evict_bytes(non_neg_integer()) ->
{evicted, non_neg_integer(), non_neg_integer()}.
evict_bytes(TargetBytes) ->
evict_bytes(TargetBytes, all).
-doc """
Evict oldest available rows whose tier is in Tiers until at least
TargetBytes have been freed. `Tiers = all` matches every tier;
otherwise it must be a list drawn from `[ram, ram_file, disk]`.
""".
-spec evict_bytes(non_neg_integer(), all | [erllama_cache:tier()]) ->
{evicted, non_neg_integer(), non_neg_integer()}.
evict_bytes(TargetBytes, Tiers) when is_integer(TargetBytes), TargetBytes >= 0 ->
gen_server:call(?SERVER, {evict_bytes, TargetBytes, Tiers}).
-spec dump() -> [tuple()].
dump() ->
ets:tab2list(?TBL_META).
-spec dump(erllama_cache:cache_key()) -> {ok, tuple()} | miss.
dump(Key) ->
lookup_row(Key).
%% Note: dump/1 wants the full row, so the lookup/2 form is appropriate
%% here. lookup_element + lookup would be two ETS ops on the hit path
%% with no benefit.
%% Shared ETS row lookup used by dump/1 and by notify_waiters/2.
%% Returns the raw row tuple as `{ok, Row}` or `miss`.
-spec lookup_row(erllama_cache:cache_key()) -> {ok, tuple()} | miss.
lookup_row(Key) ->
case ets:lookup(?TBL_META, Key) of
[Row] -> {ok, Row};
[] -> miss
end.
%% =============================================================================
%% gen_server callbacks
%% =============================================================================
-spec init([]) -> {ok, state()}.
init([]) ->
EtsOpts = [
named_table,
protected,
{keypos, 1},
{read_concurrency, true}
],
ets:new(?TBL_META, [set | EtsOpts]),
ets:new(?TBL_LRU, [ordered_set | EtsOpts]),
Timer = erlang:send_after(?SWEEP_INTERVAL_MS, self(), sweep),
State = #state{
holders = #{},
reservations = #{},
waiters = #{},
sweep_timer = Timer
},
{ok, State}.
handle_call({checkout, Key, Pid}, _From, S) ->
try ets:lookup_element(?TBL_META, Key, ?POS_STATUS) of
available ->
[Row] = ets:lookup(?TBL_META, Key),
do_checkout(Key, Pid, Row, S);
_ ->
{reply, {error, busy}, S}
catch
error:badarg -> {reply, miss, S}
end;
handle_call({checkin, MonRef}, _From, S) ->
case maps:take(MonRef, S#state.holders) of
{{_Pid, Key}, Holders1} ->
erlang:demonitor(MonRef, [flush]),
decrement_refcount(Key),
{reply, ok, S#state{holders = Holders1}};
error ->
{reply, ok, S}
end;
handle_call({lookup_or_wait, Key, MaxWaitMs}, From, S) ->
try ets:lookup_element(?TBL_META, Key, ?POS_STATUS) of
available ->
[Row] = ets:lookup(?TBL_META, Key),
{reply, {ok, Row}, S};
writing when MaxWaitMs > 0 ->
{noreply, add_waiter(Key, From, MaxWaitMs, S)};
_ ->
{reply, miss, S}
catch
error:badarg -> {reply, miss, S}
end;
handle_call({reserve_save, Key, Tier, Pid}, _From, S) ->
do_reserve_save(Key, Tier, Pid, S);
handle_call({check_reservation, Key, Token}, _From, S) ->
case maps:get(Key, S#state.reservations, undefined) of
#reservation{token = Token} = R ->
R1 = R#reservation{expires_ns = monotonic_ns() + ?DEFAULT_TTL_NS},
{reply, ok, S#state{reservations = (S#state.reservations)#{Key => R1}}};
_ ->
{reply, {error, expired}, S}
end;
handle_call({mark_published, Key, Token, Path}, _From, S) ->
case maps:get(Key, S#state.reservations, undefined) of
#reservation{token = Token} = R ->
R1 = R#reservation{
stage = post_link,
path = Path,
expires_ns = monotonic_ns() + ?DEFAULT_TTL_NS
},
{reply, ok, S#state{reservations = (S#state.reservations)#{Key => R1}}};
_ ->
{reply, {error, expired}, S}
end;
handle_call({announce_saved, Key, Token, Size, Header, TokensBin}, _From, S) ->
case maps:get(Key, S#state.reservations, undefined) of
#reservation{token = Token, monref = MonRef, tier = Tier, path = Path} ->
erlang:demonitor(MonRef, [flush]),
install_available_row(
Key, Tier, Size, Header, location_for(Tier, Path), TokensBin
),
S1 = S#state{reservations = maps:remove(Key, S#state.reservations)},
S2 = notify_waiters(Key, S1),
{reply, ok, S2};
_ ->
{reply, {error, expired}, S}
end;
handle_call({cancel_reservation, Key, Token}, _From, S) ->
case maps:get(Key, S#state.reservations, undefined) of
#reservation{token = Token} = R ->
erlang:demonitor(R#reservation.monref, [flush]),
ets:delete(?TBL_META, Key),
S1 = S#state{reservations = maps:remove(Key, S#state.reservations)},
{reply, ok, S1};
_ ->
{reply, ok, S}
end;
handle_call({insert_available, Key, Tier, Size, Header, Location, TokensBin}, _From, S) ->
install_available_row(Key, Tier, Size, Header, Location, TokensBin),
S1 = notify_waiters(Key, S),
{reply, ok, S1};
handle_call(gc, _From, S) ->
Evicted = run_eviction(),
{reply, {evicted, Evicted}, S};
handle_call({evict_bytes, 0, _Tiers}, _From, S) ->
%% "Evict at least 0 bytes" is a no-op. Use gc/0 for full GC.
{reply, {evicted, 0, 0}, S};
handle_call({evict_bytes, Target, Tiers}, _From, S) when Target > 0 ->
{N, Bytes} = run_eviction_bytes(Target, tier_pred(Tiers)),
{reply, {evicted, N, Bytes}, S};
handle_call(_Msg, _From, S) ->
{reply, {error, unknown_call}, S}.
-spec handle_cast(term(), state()) -> {noreply, state()}.
handle_cast(_Msg, S) ->
{noreply, S}.
handle_info({'DOWN', Ref, process, _DownPid, _Reason}, S) ->
case maps:take(Ref, S#state.holders) of
{{_HolderPid, Key}, Holders1} ->
decrement_refcount(Key),
{noreply, S#state{holders = Holders1}};
error ->
{noreply, on_writer_down(Ref, S)}
end;
handle_info({waiter_expire, Key, From}, S) ->
{noreply, expire_waiter(Key, From, S)};
handle_info(sweep, S) ->
S1 = sweep_reservations(S),
Timer = erlang:send_after(?SWEEP_INTERVAL_MS, self(), sweep),
{noreply, S1#state{sweep_timer = Timer}};
handle_info(_Msg, S) ->
{noreply, S}.
terminate(_Reason, S) ->
case S#state.sweep_timer of
undefined ->
ok;
TRef ->
_ = erlang:cancel_timer(TRef),
ok
end.
%% =============================================================================
%% Internal: checkout / refcount / LRU
%% =============================================================================
do_checkout(Key, Pid, Row, S) ->
MonRef = erlang:monitor(process, Pid),
NowNs = monotonic_ns(),
OldLastUsed = element(?POS_LAST_USED, Row),
ets:delete(?TBL_LRU, {OldLastUsed, Key}),
ets:insert(?TBL_LRU, {{NowNs, Key}, []}),
[NewHits] = ets:update_counter(?TBL_META, Key, [{?POS_HITS, +1}]),
_ = ets:update_counter(?TBL_META, Key, {?POS_REFCOUNT, +1}),
ets:update_element(?TBL_META, Key, {?POS_LAST_USED, NowNs}),
Tier = element(?POS_TIER, Row),
Loc = element(?POS_LOCATION, Row),
Header = element(?POS_HEADER_BIN, Row),
Tokens = element(?POS_TOKENS_REF, Row),
%% Persist the bumped hit count for restart-survival. Best-effort:
%% a failed write only loses the increment, with no behavioural
%% impact this run. RAM tier has no persistent file.
persist_hits(Loc, NewHits),
Holders1 = (S#state.holders)#{MonRef => {Pid, Key}},
{reply, {ok, MonRef, Tier, Loc, Header, Tokens}, S#state{holders = Holders1}}.
persist_hits({disk, Path}, Hits) ->
erllama_cache_disk_srv:touch_hits(Path, Hits);
persist_hits({ram_file, Path}, Hits) ->
erllama_cache_disk_srv:touch_hits(Path, Hits);
persist_hits(_, _) ->
ok.
decrement_refcount(Key) ->
try
_ = ets:update_counter(?TBL_META, Key, {?POS_REFCOUNT, -1, 0, 0}),
ok
catch
error:badarg -> ok
end.
install_available_row(Key, Tier, Size, Header, Location, TokensBin) ->
NowNs = monotonic_ns(),
try ets:lookup_element(?TBL_META, Key, ?POS_LAST_USED) of
OldLastUsed -> ets:delete(?TBL_LRU, {OldLastUsed, Key})
catch
error:badarg -> ok
end,
Hits = hits_from_header(Header),
%% Bias last_used by accumulated hits so a high-hit row survives
%% an LRU walk on a freshly-restarted server, where every row's
%% natural last_used would otherwise collapse to ~NowNs and leave
%% the order effectively random. Each accumulated hit pushes the
%% row 1 second forward in the LRU. Once a row is actively
%% checked out at runtime, recency takes over.
LastUsed = NowNs + Hits * 1_000_000_000,
Row =
{Key, Tier, Size, LastUsed, 0, available, Header, Location, TokensBin, Hits},
ets:insert(?TBL_META, Row),
ets:insert(?TBL_LRU, {{LastUsed, Key}, []}),
ok.
%% Extract the u32 hit_count from the on-disk header. RAM tier saves
%% pass a placeholder header so we treat a too-short header as "no
%% prior hits".
hits_from_header(<<_:?KVC_HEADER_HITS_OFFSET/binary, Hits:32/little, _/binary>>) ->
Hits;
hits_from_header(_) ->
0.
%% =============================================================================
%% Internal: reservation
%% =============================================================================
do_reserve_save(Key, Tier, Pid, S) ->
try ets:lookup_element(?TBL_META, Key, ?POS_STATUS) of
available ->
{reply, {error, already_present}, S};
writing ->
handle_existing_writing_row(Key, Tier, Pid, S);
evicting ->
{reply, {error, conflict}, S}
catch
error:badarg -> create_reservation(Key, Tier, Pid, S)
end.
handle_existing_writing_row(Key, Tier, Pid, S) ->
case maps:get(Key, S#state.reservations, undefined) of
undefined ->
ets:delete(?TBL_META, Key),
create_reservation(Key, Tier, Pid, S);
#reservation{} = Old ->
case reservation_is_live(Old) of
true ->
{reply, {error, conflict}, S};
false ->
S1 = cleanup_stale_reservation(Key, Old, S),
create_reservation(Key, Tier, Pid, S1)
end
end.
reservation_is_live(#reservation{writer = Pid, expires_ns = E}) ->
is_process_alive(Pid) andalso E > monotonic_ns().
create_reservation(Key, Tier, Pid, S) ->
MonRef = erlang:monitor(process, Pid),
Token = erlang:make_ref(),
NowNs = monotonic_ns(),
R = #reservation{
writer = Pid,
token = Token,
monref = MonRef,
expires_ns = NowNs + ?DEFAULT_TTL_NS,
stage = pre_link,
tier = Tier,
path = undefined
},
Placeholder = {Key, Tier, 0, NowNs, 0, writing, <<>>, undefined, undefined},
ets:insert(?TBL_META, Placeholder),
{reply, {ok, Token}, S#state{reservations = (S#state.reservations)#{Key => R}}}.
cleanup_stale_reservation(Key, Old, S) ->
erlang:demonitor(Old#reservation.monref, [flush]),
cleanup_by_stage(Key, Old, S).
cleanup_by_stage(Key, #reservation{stage = pre_link}, S) ->
ets:delete(?TBL_META, Key),
S#state{reservations = maps:remove(Key, S#state.reservations)};
cleanup_by_stage(Key, #reservation{stage = post_link, path = Path, tier = Tier}, S) ->
case validate_and_adopt(Key, Path) of
{ok, Size, Header, TokensBin} ->
install_available_row(
Key, Tier, Size, Header, location_for(Tier, Path), TokensBin
),
S1 = S#state{reservations = maps:remove(Key, S#state.reservations)},
notify_waiters(Key, S1);
{error, _Reason} ->
_ = file:delete(Path),
ets:delete(?TBL_META, Key),
S#state{reservations = maps:remove(Key, S#state.reservations)}
end.
validate_and_adopt(Key, Path) ->
case file:read_file(Path) of
{ok, Bin} ->
case erllama_cache_kvc:parse(Bin, Key) of
{ok, Info, _Payload} ->
Tokens = maps:get(tokens, Info, []),
TokensBin = erllama_cache_key:encode_tokens(Tokens),
{ok, byte_size(Bin), header_slice(Bin), TokensBin};
{error, R} ->
{error, R}
end;
{error, R} ->
{error, R}
end.
header_slice(Bin) ->
HeaderSize = 48,
binary:part(Bin, 0, HeaderSize).
location_for(ram, _) -> {ram};
location_for(ram_file, Path) -> {ram_file, Path};
location_for(disk, Path) -> {disk, Path}.
%% =============================================================================
%% Internal: writer DOWN handling
%% =============================================================================
on_writer_down(Ref, S) ->
case find_reservation_by_monref(Ref, S#state.reservations) of
{Key, R} -> cleanup_by_stage(Key, R, S);
none -> S
end.
find_reservation_by_monref(Ref, Reservations) ->
Found = maps:fold(
fun(K, R = #reservation{monref = M}, Acc) ->
case M of
Ref -> [{K, R} | Acc];
_ -> Acc
end
end,
[],
Reservations
),
case Found of
[] -> none;
[{K, R} | _] -> {K, R}
end.
%% =============================================================================
%% Internal: waiters (lookup_exact_or_wait)
%% =============================================================================
add_waiter(Key, From, MaxWaitMs, S) ->
Expires = monotonic_ns() + MaxWaitMs * 1_000_000,
TRef = erlang:send_after(MaxWaitMs, self(), {waiter_expire, Key, From}),
Existing = maps:get(Key, S#state.waiters, []),
Waiters1 = (S#state.waiters)#{Key => [{From, Expires, TRef} | Existing]},
S#state{waiters = Waiters1}.
notify_waiters(Key, S) ->
case maps:take(Key, S#state.waiters) of
{Waiters, W1} ->
Reply = lookup_row(Key),
lists:foreach(
fun({From, _Exp, TRef}) ->
_ = erlang:cancel_timer(TRef),
gen_server:reply(From, Reply)
end,
Waiters
),
S#state{waiters = W1};
error ->
S
end.
expire_waiter(Key, From, S) ->
case maps:get(Key, S#state.waiters, []) of
[] ->
S;
Waiters ->
case lists:keytake(From, 1, Waiters) of
{value, _, []} ->
gen_server:reply(From, miss),
S#state{waiters = maps:remove(Key, S#state.waiters)};
{value, _, Rest} ->
gen_server:reply(From, miss),
S#state{waiters = (S#state.waiters)#{Key => Rest}};
false ->
%% Already replied (e.g. a save published first) and the
%% timer was cancelled in notify_waiters; this message is
%% the one that lost the race between cancel and fire.
S
end
end.
%% =============================================================================
%% Internal: sweep
%% =============================================================================
sweep_reservations(S) ->
Now = monotonic_ns(),
Stale = maps:fold(
fun(K, R = #reservation{expires_ns = E, writer = Pid}, Acc) ->
case E =< Now orelse not is_process_alive(Pid) of
true -> [{K, R} | Acc];
false -> Acc
end
end,
[],
S#state.reservations
),
lists:foldl(
fun({Key, R}, Acc) -> cleanup_stale_reservation(Key, R, Acc) end,
S,
Stale
).
%% =============================================================================
%% Internal: eviction
%% =============================================================================
run_eviction() ->
run_eviction(ets:first(?TBL_LRU), 0).
run_eviction('$end_of_table', N) ->
N;
run_eviction({_LastUsed, Key} = LruKey, N) ->
Next = ets:next(?TBL_LRU, LruKey),
case try_evict_one(LruKey, Key) of
{ok, _Bytes} -> run_eviction(Next, N + 1);
skip -> run_eviction(Next, N);
gone -> run_eviction(Next, N)
end.
run_eviction_bytes(Target, TierPred) ->
run_eviction_bytes(ets:first(?TBL_LRU), Target, TierPred, 0, 0).
run_eviction_bytes('$end_of_table', _Target, _TierPred, N, Bytes) ->
{N, Bytes};
run_eviction_bytes(_LruKey, Target, _TierPred, N, Bytes) when
Bytes >= Target, Target > 0
->
{N, Bytes};
run_eviction_bytes({_LastUsed, Key} = LruKey, Target, TierPred, N, Bytes) ->
Next = ets:next(?TBL_LRU, LruKey),
case try_evict_one(LruKey, Key, TierPred) of
{ok, B} -> run_eviction_bytes(Next, Target, TierPred, N + 1, Bytes + B);
skip -> run_eviction_bytes(Next, Target, TierPred, N, Bytes);
gone -> run_eviction_bytes(Next, Target, TierPred, N, Bytes)
end.
try_evict_one(LruKey, Key) ->
try_evict_one(LruKey, Key, fun(_) -> true end).
try_evict_one(LruKey, Key, TierPred) ->
case ets:lookup(?TBL_META, Key) of
[Row] ->
Tier = element(?POS_TIER, Row),
case {element(?POS_STATUS, Row), element(?POS_REFCOUNT, Row), TierPred(Tier)} of
{available, 0, true} ->
Location = element(?POS_LOCATION, Row),
Size = element(?POS_SIZE, Row),
delete_from_tier(Tier, Key, Location),
ets:delete(?TBL_LRU, LruKey),
ets:delete(?TBL_META, Key),
erllama_cache_counters:incr(?C_EVICTIONS),
{ok, Size};
_ ->
skip
end;
[] ->
ets:delete(?TBL_LRU, LruKey),
gone
end.
tier_pred(all) ->
fun(_) -> true end;
tier_pred(Tiers) when is_list(Tiers) ->
fun(T) -> lists:member(T, Tiers) end.
delete_from_tier(ram, Key, _Loc) ->
erllama_cache_ram:delete(Key);
delete_from_tier(disk, _Key, {disk, Path}) ->
_ = file:delete(Path),
ok;
delete_from_tier(ram_file, _Key, {ram_file, Path}) ->
_ = file:delete(Path),
ok;
delete_from_tier(_, _, _) ->
ok.
%% =============================================================================
%% Internal: helpers
%% =============================================================================
-spec monotonic_ns() -> integer().
monotonic_ns() ->
erlang:monotonic_time(nanosecond).
