ecql

%%==============================================================================
%% Copyright (c) Exosite LLC
%%
%% ecql_cache.erl - Cache
%%==============================================================================
-module(ecql_cache).
-behaviour(gen_server).

%% Public API
-export([
   cache_size/0
  ,cluster_module/0
  ,clear/0
  ,get/2
  ,dirty/1
  ,local_match_clear/1
  ,match_clear/1
  ,set/2
  ,set_cache_size/1
  ,set_cluster_module/1
  ,stats/0
  ,clear_stats/0
]).

%% OTP gen_server
-export([
   init/1
  ,start_link/0
  ,stop/0
  ,handle_call/3
  ,handle_cast/2
  ,handle_info/2
  ,code_change/3
  ,terminate/2
]).

%% Includes
-include("ecql.hrl").

%% Defines
%-define(stats, false).
-define(DEFAULT_CACHESIZE, 1000000).
-define(DEFAULT_CLUSTER_MODULE, erlang).
-define(seconds(X), X*1000000).
-define(CACHE_RETRY_LIMIT, 10).

%%-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=
%% Public API
%%-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=

%%------------------------------------------------------------------------------
cache_size() ->
  private_get(cache_size, ?DEFAULT_CACHESIZE)
.

%%------------------------------------------------------------------------------
cluster_module() ->
  private_get(cluster_module, ?DEFAULT_CLUSTER_MODULE)
.

%%------------------------------------------------------------------------------
current_slice() ->
  element(private_get(current_slice, 1), ?CACHE_SLICES_TUPLE)
.

%%------------------------------------------------------------------------------
clear() ->
  gen_server:abcast(?MODULE, clear)
.

%%------------------------------------------------------------------------------
get(Key, FunResult) ->
  case find(Key) of
    {_Slice, Value} ->
      Value
    ;
    {_Slice, dirty, _} ->
      do_get(Key, FunResult)
    ;
    undefined ->
      do_get(Key, FunResult)
    %~
  end
.
do_get(Key, FunResult) ->
  do_get(Key, FunResult, 0)
.
do_get(_Key, FunResult, ?CACHE_RETRY_LIMIT) ->
   incr_stat(empty)
  ,error_logger:error_msg(
      "~p: reach retry limit of ~p.~n"
     ,[?MODULE, ?CACHE_RETRY_LIMIT]
   )
  ,FunResult()
;
do_get(Key, FunResult, AttemptCount) ->
   incr_stat(empty)
  ,Ts = system_time_in_micro_seconds()
  ,Result = FunResult()
  ,case is_cache_dirty_since(Key, Ts) of
    {no, Slice} ->
       ets:insert(Slice, {Key, Result})
      ,Result
    ;
    no ->
       cache_insert_new({Key, Result})
      ,Result
    ;
    yes -> do_get(Key, FunResult, AttemptCount + 1)
   end
.

%%------------------------------------------------------------------------------
system_time_in_micro_seconds() ->
   {A, B, C} = os:timestamp()
  ,(A * 1000000 + B) * 1000000 + C
.

%%------------------------------------------------------------------------------
is_cache_dirty_since(Key, Timestamp) ->
  case find(Key) of
    {Slice, dirty, DirtyTimestamp} ->
      if DirtyTimestamp > Timestamp ->
        % dirty mark is newer than direct read result - read again
        yes
      ;
      true ->
        % dirty but result is newer than dirty timestamp - overwrite dirty mark
        {no, Slice}
      end
    ;
    {Slice, _Value} ->
      {no, Slice}
    ;
    undefined ->
      no
    %~
  end
.

%%------------------------------------------------------------------------------
get_stat(Key) ->
  private_get(Key, 0)
.

%%------------------------------------------------------------------------------
stats() ->
  [{Key, get_stat(Key)} || Key <- [undef, empty, old, conflict]]
.

%%------------------------------------------------------------------------------
clear_stats() ->
  [set_stat(Key, 0) || {Key, _} <- stats()]
.

%%------------------------------------------------------------------------------
-ifdef(stats).
incr_stat(Key) ->
  set_stat(Key, get_stat(Key) + 1)
.
-else.
incr_stat(_Key) ->
  ok
.
-endif.

%%------------------------------------------------------------------------------
set_stat(Key, Num) when is_integer(Num) ->
  private_set(Key, Num)
.


%%------------------------------------------------------------------------------
dirty(Key) ->
   Module = cluster_module()
  ,gen_server:abcast(Module:nodes(), ?MODULE, {dirty, Key})
  ,do_dirty(Key)
  ,ok
.
do_dirty(Key) ->
   Ts = system_time_in_micro_seconds()
  ,Record = {Key, dirty, Ts}
  ,case find(Key) of
    {Slice, _Value} ->
      ets:insert(Slice, Record)
    ;
    {Slice, dirty, _} ->
      ets:insert(Slice, Record)
    ;
    undefined ->
      cache_insert_new(Record)
    %~
   end
.

%%------------------------------------------------------------------------------
local_match_clear(Pattern) when is_atom(Pattern); is_tuple(Pattern) ->
  lists:foreach(
    fun (Slice) ->
      case catch ets:match_delete(Slice, Pattern) of
        {'EXIT', Error} ->
           error_logger:error_msg(
             "~p: local_match_clear error. pattern: ~p, error: ~p~n"
            ,[?MODULE, Pattern, Error]
           )
        ;
        _ ->
           true
        %~
      end
    end
   ,?CACHE_SLICES_LIST
  )
;
local_match_clear(_) ->
  ok
.

%%------------------------------------------------------------------------------
match_clear(Pattern) ->
   ClusterModule = cluster_module()
  ,gen_server:abcast(ClusterModule:nodes(), ?MODULE, {match_clear, Pattern})
  ,local_match_clear(Pattern)
.

%%------------------------------------------------------------------------------
set(Key, Result) ->
   Record = {Key, Result}
  ,case find(Key) of
    {Slice, _Value} ->
      ets:insert(Slice, Record)
    ;
    {Slice, dirty, _Ts} ->
      ets:insert(Slice, Record)
    ;
    undefined ->
      cache_insert_new(Record)
    %~
   end
  ,Result
.

%%------------------------------------------------------------------------------
cache_insert_new(Object) ->
   Slice = current_slice()
  ,ets:insert(Slice, Object)
  ,Size = ets:info(Slice, size)
  ,SliceCount = tuple_size(?CACHE_SLICES_TUPLE)
  ,Limit = cache_size() / SliceCount
  ,(Size > Limit) andalso begin
     Index = private_get(current_slice, 1) + 1
    ,case (Index > SliceCount) of
      true ->
        private_set(current_slice, 1)
      ;
      false ->
        private_set(current_slice, Index)
      %~
    end
    ,ets:delete_all_objects(current_slice())
   end
  ,ok
.

%%------------------------------------------------------------------------------
set_cache_size(CacheSize) when is_integer(CacheSize) ->
  private_set(cache_size, CacheSize)
.

%%------------------------------------------------------------------------------
set_cluster_module(Module) when is_atom(Module) ->
  private_set(cluster_module, Module)
.

%%-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=
%% OTP gen_server API
%%-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=

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

%%------------------------------------------------------------------------------
init(_) ->
   Configuration = application:get_all_env()
  ,ok = net_kernel:monitor_nodes(true)
  ,set_cache_size(proplists:get_value(cache_size, Configuration, ?DEFAULT_CACHESIZE))
  ,set_cluster_module(proplists:get_value(cluster_module, Configuration, ?DEFAULT_CLUSTER_MODULE))
  ,{ok, Configuration}
.

%%------------------------------------------------------------------------------
stop() ->
  gen_server:call(?MODULE, stop)
.

%%------------------------------------------------------------------------------
handle_call(stop, _From, State) ->
  {stop, normal, ok, State}
;
handle_call(_, _From, State) ->
  {reply, unknown, State}
.

%%------------------------------------------------------------------------------
handle_cast(clear, State) ->
   CacheSize = cache_size()
  ,Module = cluster_module()
  ,ets:delete_all_objects(?MODULE)
  ,[ets:delete_all_objects(Table) || Table <- ?CACHE_SLICES_LIST]
  ,set_cache_size(CacheSize)
  ,set_cluster_module(Module)
  ,{noreply, State}
;
handle_cast({dirty, Key}, State) ->
  do_dirty(Key)
 ,{noreply, State}
;
handle_cast({match_clear, Pattern}, State) ->
  local_match_clear(Pattern)
 ,{noreply, State}
;
handle_cast(terminate ,State) ->
  {stop ,terminated ,State}
;
handle_cast(_, State) ->
  {noreply, State}
.

%%------------------------------------------------------------------------------
handle_info({nodeup, _}, State) ->
   % Can ignore
   {noreply, State}
;
handle_info({nodedown, _}, State) ->
   % Nodedown means we likely lost messages
   clear()
  ,{noreply, State}
;
handle_info(timeout, State) ->
   % Who timed out?
   error_logger:error_msg("ecql_cache: Timeout occured~n")
  ,{noreply, State}
;
handle_info(_, State) ->
   {noreply, State}
.

%%------------------------------------------------------------------------------
terminate(_Reason, State) ->
  {shutdown, State}
.

%%------------------------------------------------------------------------------
code_change(_ ,State ,_) ->
  {ok ,State}
.

%%-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=
%% Private API
%%-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=

%%------------------------------------------------------------------------------
find(Key) ->
  find(Key, ?CACHE_SLICES_LIST)
.

%%------------------------------------------------------------------------------
find(Key, [Slice | Rest]) ->
  case ets:lookup(Slice, Key) of
    [{Key, Value}] ->
      {Slice, Value}
    ;
    [{Key, dirty, Timestamp}] ->
      {Slice, dirty, Timestamp}
    ;
    [] -> % [] or [Index, OtherKey, Value, Time]
      find(Key, Rest)
    %~
  end
;
find(_Key, []) ->
  undefined
.

%%------------------------------------------------------------------------------
private_get(Key, Default) ->
  case ets:lookup(?MODULE, Key) of
    [{Key, Value}] ->
      Value
    ;
    _Other ->
      Default
    %~
  end
.

%%------------------------------------------------------------------------------
private_set(Key, Value) ->
  ets:insert(?MODULE, {Key, Value})
.


%%==============================================================================
%% END OF FILE