filezcache

%% @author Marc Worrell
%% @copyright 2013-2014 Marc Worrell

%% Copyright 2013-2014 Marc Worrell
%%
%% Licensed under the Apache License, Version 2.0 (the "License");
%% you may not use this file except in compliance with the License.
%% You may obtain a copy of the License at
%%
%%     http://www.apache.org/licenses/LICENSE-2.0
%%
%% Unless required by applicable law or agreed to in writing, software
%% distributed under the License is distributed on an "AS IS" BASIS,
%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
%% See the License for the specific language governing permissions and
%% limitations under the License.

%%% @doc Manage all entries.

-module(filezcache_entry_manager).

-behaviour(gen_server).

-include_lib("kernel/include/file.hrl").

-export([
        start_link/0,
        insert/2,
        lookup/1,
        lookup/2,
        delete/1,
        gc/1,
        stats/0,
        log_access/1,
        log_access/2,
        log_ready/5
    ]).

-export([
    init/1,
    handle_call/3,
    handle_cast/2,
    handle_info/2,
    terminate/2,
    code_change/3
    ]).

-export([
    ensure_tables/0
    ]).

-record(filezcache_log_entry, {
        key,
        filename,
        size,
        checksum
    }).


-record(state, {
    % Monitors for cache write entries
    monitors,
    sizes,

    % Monitors for key entries referring to cache keys
    key2referrers,
    referrer2keys,

    iterator = start,
    recent :: list(),
    gc_candidate_pool = [] :: list(),
    bytes = 0 :: integer(),
    max_bytes :: integer() 
    }).


-define(TIMEOUT, infinity).

% Garbage collection setings
-define(GC_INTERVAL, 1000). 
-define(GC_MAX_BYTES, 10737418240).
-define(GC_POOL_SIZE, 100).
-define(GC_CHANCE_1_IN_N, 20).

% Every 10 minutes we empty the oldest table with recently used items
-define(RECENT_INTERVAL, 60000).


%% API

start_link() ->
    ok = ensure_tables(),
    gen_server:start_link({local, ?MODULE}, ?MODULE, [], []).

stats() ->
    gen_server:call(?MODULE, stats).

insert(Key, Opts) ->
    gen_server:call(?MODULE, {insert, Key, self(), Opts}, ?TIMEOUT).

lookup(Key) ->
    lookup(Key, undefined).

lookup(Key, MonitorPid) ->
    case mnesia:dirty_read(filezcache_log_entry, Key) of
        [] ->
            {error, enoent};
        [#filezcache_log_entry{size=undefined}] ->
            {error, enoent};
        [#filezcache_log_entry{size=Size, filename=Filename}] ->
            case filelib:is_regular(Filename) of
                true ->
                    log_access(Key, MonitorPid),
                    {ok, {file, Size, Filename}};
                false ->
                    delete(Key, Filename),
                    {error, enoent}
            end 
    end.

gc(Key) ->
    gen_server:cast(?MODULE, {gc, Key}).

log_access(Key) ->
    log_access(Key, undefined).

log_access(Key, MonitorPid) ->
    gen_server:cast(?MODULE, {log_access, Key, MonitorPid}).

log_ready(EntryPid, Key, Filename, Size, Checksum) ->
    gen_server:cast(?MODULE, {log_ready, EntryPid, Key, Filename, Size, Checksum}).


%% gen_server callbacks

init([]) ->
    filezcache_store:init(),
    gen_server:cast(self(), log_init),
    timer:send_after(?GC_INTERVAL, gc),
    timer:send_after(?RECENT_INTERVAL, recent_rotate),
    {ok, #state{
            sizes = ets:new(?MODULE, [set, private]),
            monitors = gb_trees:empty(),
            recent = [ 
                ets:new(filezcache_recent_1, [set]), 
                ets:new(filezcache_recent_2, [set]),
                ets:new(filezcache_recent_3, [set])
            ],
            key2referrers = dict:new(),
            referrer2keys = dict:new(),
            max_bytes = max_bytes()}}.

handle_call({insert, Key, WriterPid, Opts}, _From, State) ->
    case filezcache_store:lookup(Key) of
        {ok, Pid} ->
            {reply, {error, {already_started, Pid}}, State};
        {error, enoent} ->
            {ok, Pid} = filezcache_entry_sup:start_child(Key, WriterPid, Opts),
            filezcache_store:insert(Key, Pid),
            filezcache_event:insert(Key),
            Mon = erlang:monitor(process, Pid),
            State1 = State#state{monitors=gb_trees:enter(Mon, {Key, Pid}, State#state.monitors)},
            State2 = case proplists:get_value(monitor, Opts) of
                true -> 
                    maybe_add_key_referrer(Key, WriterPid, State1);
                _ ->
                    State1
            end,
            {reply, {ok, Pid}, State2}
    end;
handle_call(stats, _From, State) ->
    Stats = [
        {bytes, State#state.bytes},
        {max_bytes, max_bytes()},
        {processes, gb_trees:size(State#state.monitors)},
        {entries, mnesia:table_info(filezcache_log_entry, size)},
        {referrers, dict:size(State#state.referrer2keys)},
        {gc_candidate_pool, State#state.gc_candidate_pool}
    ],
    {reply, Stats, State}.

handle_cast({repop, Key, Filename, undefined, _Checksum}, State) ->
    case filezcache_store:lookup(Key) of
        {ok, _Pid} ->
            ok;
        {error, enoent} ->
            delete(Key, Filename)
    end,
    {noreply, State};
handle_cast({repop, Key, _Filename, Size, _Checksum}, State) ->
    case filezcache_store:lookup(Key) of
        {ok, _Pid} ->
            {noreply, State};
        {error, enoent} ->
            {noreply, State#state{bytes=State#state.bytes+Size}}
    end;
handle_cast({delete_if_inactive, Key, Filename}, State) ->
    case filezcache_store:lookup(Key) of
        {ok, _Pid} ->
            nop;
        {error, enoent} ->
            case is_recently_used(Key, State) orelse is_referred(Key, State) of
                true ->
                    nop;
                false ->
                    delete(Key, Filename)
            end
    end,
    {noreply, State};
handle_cast({gc, Key}, State) ->
    case filezcache_store:lookup(Key) of
        {ok, _Pid} ->
            {noreply, State};
        {error, enoent} ->
            case is_recently_used(Key, State) orelse is_referred(Key, State) of
                true ->
                    {noreply, State};
                false ->
                    case delete(Key) of
                        {ok, Size} ->
                            {noreply, State#state{bytes=State#state.bytes - Size}};
                        {error, _} ->
                            {noreply, State}
                    end
            end
    end;

handle_cast(log_init, State) ->
    proc_lib:spawn_link(fun() -> repopulate() end),
    {noreply, State};

%% Log a cache entry to the disk log
handle_cast({log_ready, Pid, Key, Filename, Size, Checksum}, #state{bytes=Bytes} = State) ->
    F = fun() ->
            mnesia:write(#filezcache_log_entry{key=Key, filename=Filename, size=Size, checksum=Checksum})
        end,
    mnesia:activity(transaction, F),
    filezcache_event:insert_ready(Key, Size, Filename),
    case Size of
        undefined -> ok;
        _ -> filezcache_entry:logged(Pid)
    end,
    {noreply, State#state{bytes=Bytes + case Size of undefined -> 0; _ -> Size end}};

%% Remove recently used keys from the eviction pool
handle_cast({log_access, Key, MonitorPid}, #state{gc_candidate_pool=Pool, recent=[Table|_]} = State) ->
    ets:insert(Table, {Key, os:timestamp()}),
    State1 = State#state{gc_candidate_pool=lists:delete(Key, Pool)},
    {noreply, maybe_add_key_referrer(Key, MonitorPid, State1)};

handle_cast(_Msg, State) ->
    {noreply, State}.

handle_info({'DOWN', MRef, process, Pid, _Reason}, #state{monitors=Monitors, bytes=Bytes} = State) ->
    State1 = case gb_trees:lookup(MRef, State#state.monitors) of
        {value, {Key, Pid}} ->
            filezcache_store:delete(Key),
            Bytes1 = case is_logged(Key) of
                true ->
                    Bytes;
                false ->
                    {ok, FileSize} = delete(Key),
                    filezcache_event:delete(Key),
                    Bytes - FileSize
            end,
            State#state{monitors = gb_trees:delete(MRef, Monitors), bytes=Bytes1};
        none ->
            State
    end,
    State2 = case dict:find(Pid, State1#state.referrer2keys) of
        {ok, RefKeys} ->
            State1#state{
                referrer2keys = dict:erase(Pid, State#state.referrer2keys), 
                key2referrers = remove_referrer(Pid, RefKeys, State#state.key2referrers)
            };
        error -> 
            State1
    end,
    {noreply, State2};
handle_info(gc, State) ->
    State1 = do_gc(State),
    timer:send_after(?GC_INTERVAL, gc),
    {noreply, State1};
handle_info(recent_rotate, #state{recent=Tables} = State) ->
    timer:send_after(?RECENT_INTERVAL, recent_rotate),
    ets:delete_all_objects(lists:last(Tables)),
    Tables1 = [ lists:last(Tables) | lists:sublist(Tables, length(Tables)-1) ],
    {noreply, State#state{recent=Tables1}};
handle_info(_Info, State) ->
    {noreply, State}.

terminate(_Reason, _State) ->
    ok.

code_change(_OldVsn, State, _Extra) ->
    {ok, State}.


%% @doc Check if the key is in the list of recently used entries
is_recently_used(Key, #state{recent=Tables}) ->
    lists:any(fun(Table) ->
                 ets:lookup(Table, Key) =/= []
              end,
              Tables).

is_logged(Key) ->
    F = fun() ->
            mnesia:read(filezcache_log_entry, Key)
        end,
    case mnesia:activity(transaction, F) of
        [] ->
            false;
        [#filezcache_log_entry{size=undefined}] ->
            false;
        [#filezcache_log_entry{}] ->
            true
    end.

%% @doc Add a monitor to a process that caches a key - this prevents garbage collection of the entry
maybe_add_key_referrer(_Key, undefined, State) ->
    State;
maybe_add_key_referrer(Key, Pid, State) ->
    case dict:find(Pid, State#state.referrer2keys) of
        {ok, RefKeys} ->
            case lists:member(Key, RefKeys) of
                true ->
                    State;
                false ->
                    do_add_key_referrer(Key, Pid, State)
            end;
        error ->
            do_add_key_referrer(Key, Pid, State)
    end.

do_add_key_referrer(Key, Pid, State) ->
    _ = erlang:monitor(process, Pid),
    State#state{
        key2referrers = dict:append(Key, Pid, State#state.key2referrers), 
        referrer2keys = dict:append(Pid, Key, State#state.referrer2keys)
    }.

is_referred(Key, State) ->
    dict:is_key(Key, State#state.key2referrers).

remove_referrer(_Pid, [], Key2Pids) ->
    Key2Pids;
remove_referrer(Pid, [Key|Keys], Key2Pids) ->
    case dict:find(Key, Key2Pids) of
        {ok, Pids} ->
            case lists:delete(Pid, Pids) of
                [] ->
                    remove_referrer(Pid, Keys, dict:erase(Key, Key2Pids));
                Pids1 ->
                    remove_referrer(Pid, Keys, dict:store(Key, Pids1, Key2Pids))
            end;
        error ->
            remove_referrer(Pid, Keys, Key2Pids)
    end. 

%% @doc Ensure that the proper filezcache_log table has been created
ensure_tables() ->
    TabDef = [
        {type, set},
        {record_name, filezcache_log_entry},
        {index, [#filezcache_log_entry.filename]},
        {attributes, record_info(fields, filezcache_log_entry)}
        | case application:get_env(mnesia, dir) of
             {ok, _} -> [ {disc_copies, [node()]} ];
             undefined -> []
          end
    ],
    case mnesia:create_table(filezcache_log_entry, TabDef) of
        {atomic, ok} -> ok;
        {aborted, {already_exists, filezcache_log_entry}} -> ok
    end.


%% @doc Repopulates the cache using the log
repopulate() ->
    Keys = mnesia:dirty_all_keys(filezcache_log_entry), 
    error_logger:info_msg("filezcache: repopulating cache with ~p keys", [length(Keys)]),
    repopulate(Keys),
    error_logger:info_msg("filezcache: scanning cache directory for unknown files."),
    scan_cache().

repopulate([]) ->
    ok;
repopulate([Key|Keys]) ->
    F = fun() ->
            mnesia:read(filezcache_log_entry, Key)
        end,
    Rs = mnesia:activity(transaction, F),
    lists:foreach(fun repop_term/1, Rs),
    repopulate(Keys).

repop_term(#filezcache_log_entry{key=Key, filename=Filename, size=Size, checksum=Checksum}) ->
    case filezcache_store:lookup(Key) of
        {error, enoent} ->
            case file:read_file_info(Filename) of
                {ok, #file_info{type=regular, size=Size}} ->
                    gen_server:cast(?MODULE, {repop, Key, Filename, Size, Checksum});
                {ok, #file_info{type=regular, size=_OtherSize}} ->
                    gen_server:cast(?MODULE, {delete_if_inactive, Key});
                _Other ->
                    nop
            end;
        {ok, _Pid} ->
            nop
    end;
repop_term(_Term) ->
    ok.

%% @doc Scan the cache directories, remove all files not in the log
scan_cache() ->
    scan_dir(filezcache:data_dir()).

scan_dir(Dir) ->
    case file:list_dir(Dir) of
        {ok, Files} ->
            scan_files(Dir, Files);
        {error, _} ->
            ok
    end. 

scan_files(_Dir, []) ->
    ok;
scan_files(Dir, ["."++_|Fs]) ->
    scan_files(Dir, Fs);
scan_files(Dir, [F|Fs]) ->
    Path = filename:join(Dir, F),
    case filelib:is_regular(Path) of
        true ->
            case find_by_filename(Path) of
                [] -> file:delete(Path);
                [_|_] -> ok
            end;
        false ->
            scan_dir(Path)
    end,
    scan_files(Dir, Fs).


%% @doc Delete an entry by key
delete(Key) ->
    F = fun() ->
            mnesia:read(filezcache_log_entry, Key)
        end,
    case mnesia:activity(transaction, F) of
        [] -> 
            {ok, 0};
        [#filezcache_log_entry{key=Key,filename=Filename, size=Size}] ->
            ok = delete(Key,Filename),
            case Size of
                undefined -> {ok, 0};
                _ -> {ok, Size}
            end
    end.

%% @doc Delete an entry and its associated cache file.
delete(Key, Filename) ->
    F = fun() ->
            mnesia:delete({filezcache_log_entry, Key})
        end,
    mnesia:activity(transaction, F),
    _ = file:delete(Filename),
    ok.

%% @doc Find an entry by the cached file
find_by_filename(Path) ->
    F = fun() ->
            mnesia:index_read(filezcache_log_entry, Path, #filezcache_log_entry.filename)
        end,
    mnesia:activity(transaction, F).


%% @doc Perform a gc step, keep eviction pool with gc-candidates populated
do_gc(State) ->
    State1 = maybe_evict(State),
    fill_pool(State1, normal, 1).

maybe_evict(#state{gc_candidate_pool=[], bytes=Bytes} = State) ->
    case Bytes > max_bytes() of
        true -> do_gc(fill_pool(State, eager, 1));
        false -> State
    end;
maybe_evict(#state{gc_candidate_pool=Pool, bytes=Bytes} = State) ->
    case Bytes > max_bytes() of
        true -> State#state{gc_candidate_pool=random_evict(Pool)};
        false -> State
    end;
maybe_evict(State) ->
    State.


fill_pool(State, normal, N) when N >= 20 ->
    State;
fill_pool(State, eager, N) when N >= 100 ->
    State;
fill_pool(#state{gc_candidate_pool=Pool, iterator=Iterator} = State, Method, N) ->
    case length(Pool) < ?GC_POOL_SIZE of
        true ->
            {Candidates, Iterator1} = iterate(Iterator), 
            Pool1 = fill_pool_1(Pool, Candidates, Method),
            fill_pool(State#state{gc_candidate_pool=Pool1, iterator=Iterator1}, Method, N+1);
        false ->
            State
    end.

fill_pool_1(Pool, [], _Method) ->
    Pool;
fill_pool_1(Pool, [#filezcache_log_entry{key=Key, filename=Filename}|Cs], Method) ->
    case not lists:member(Key, Pool) andalso do_select(Method) of
        true ->
            case filelib:is_regular(Filename) of
                true ->
                    fill_pool_1([Key|Pool], Cs, Method);
                false ->
                    delete(Key),
                    fill_pool_1(Pool, Cs, Method)
            end;
        false ->
            fill_pool_1(Pool, Cs, Method)
    end.

do_select(eager) ->
    true;
do_select(normal) ->
    rand_uniform(?GC_CHANCE_1_IN_N) =:= 1.

random_evict([]) -> 
    [];
random_evict(Pool) ->
    Key = lists:nth(rand_uniform(length(Pool)), Pool),
    gc(Key),
    lists:delete(Key, Pool).

max_bytes() ->
    case application:get_env(filezcache, max_bytes) of
        undefined -> ?GC_MAX_BYTES;
        {ok, N} when is_integer(N) -> N
    end.


%% @doc Iterate over mnesia
iterate(start) ->
    iterate(0);
iterate(SlotNr) ->
    case get_slot(SlotNr) of
        '$end_of_table' -> 
            {[], start};
        Entries -> 
            {Entries, SlotNr+1}
    end.

get_slot(SlotNr) ->
    try 
        mnesia:dirty_slot(filezcache_log_entry, SlotNr)
    catch
        error:badarg -> '$end_of_table'
    end.

-spec rand_uniform( pos_integer() ) -> pos_integer().
-ifdef(rand_only).
rand_uniform(N) ->
    rand:uniform(N).
-else.
rand_uniform(N) ->
    crypto:rand_uniform(1,N+1).
-endif.